From 6ebddecd7be319a5554a96d62c29f53bf47bf4b8 Mon Sep 17 00:00:00 2001 From: emanuel Date: Wed, 2 Mar 2022 16:34:04 +0000 Subject: [PATCH] Syslog efforts --- src/CMakeLists.txt | 1 - src/cam.c | 22 ++++++---------- src/cpm.c | 22 +++++----------- src/facilities.c | 40 +++++++++++++++++----------- src/facilities.h | 5 ++-- src/pcm.c | 12 ++++++--- src/queue.c | 66 ---------------------------------------------- src/queue.h | 22 ---------------- src/requests.c | 4 +-- src/saem.c | 10 ++++--- src/tpm.c | 8 ++++-- 11 files changed, 65 insertions(+), 147 deletions(-) delete mode 100644 src/queue.c delete mode 100644 src/queue.h diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt index 71ed6f1..f3b9c07 100644 --- a/src/CMakeLists.txt +++ b/src/CMakeLists.txt @@ -1,6 +1,5 @@ ADD_EXECUTABLE(it2s-itss-facilities config.c - queue.c cam.c denm.c infrastructure.c diff --git a/src/cam.c b/src/cam.c index 7a1c489..003c008 100644 --- a/src/cam.c +++ b/src/cam.c @@ -19,19 +19,11 @@ #include #include #include +#include +#include #include -#define syslog_info(msg, ...) syslog(LOG_INFO, msg, ##__VA_ARGS__) -#define syslog_emerg(msg, ...) syslog(LOG_EMERG, "%s:%d [" msg "]", __func__, __LINE__, ##__VA_ARGS__) -#define syslog_err(msg, ...) syslog(LOG_ERR, "%s:%d [" msg "]", __func__, __LINE__, ##__VA_ARGS__) - -#ifndef NDEBUG -#define syslog_debug(msg, ...) syslog(LOG_DEBUG, "%s:%d " msg "", __func__, __LINE__, ##__VA_ARGS__) -#else -#define syslog_debug(msg, ...) -#endif - #define LEAP_SECONDS 5 #define EARTH_RADIUS 6369000 @@ -769,6 +761,10 @@ void *ca_service(void *fc) { bpr->gn.communicationProfile = 0; if (facilities->station_type != 15 && check_pz(facilities->lightship, &facilities->epv)) bpr->gn.communicationProfile = 1; + uint64_t id = rand() + 1; + bpr->id = id; + fmi->id = id; + asn_enc_rval_t enc = oer_encode_to_buffer(&asn_DEF_TransportRequest, NULL, tr, tr_oer+1, 1023); if (enc.encoded == -1) { syslog_err("[facilities] encoding TR for cam failed"); @@ -781,11 +777,9 @@ void *ca_service(void *fc) { continue; } - queue_add(facilities->tx_queue, tr_oer, enc.encoded+1, 3); - pthread_cond_signal(&facilities->tx_queue->trigger); + it2s_tender_queue_add(facilities->tx_queue, tr_oer, enc.encoded+1, ITSS_TRANSPORT, id, "TR.packet.btp"); - queue_add(facilities->tx_queue, fi_oer, enc_fdi.encoded+1, 5); - pthread_cond_signal(&facilities->tx_queue->trigger); + it2s_tender_queue_add(facilities->tx_queue, fi_oer, enc_fdi.encoded+1, ITSS_APPLICATIONS, id, "FI.message"); lightship_reset_timer(facilities->lightship, &facilities->epv); } diff --git a/src/cpm.c b/src/cpm.c index a98f986..20e49b3 100644 --- a/src/cpm.c +++ b/src/cpm.c @@ -21,17 +21,7 @@ #include #include - -#define syslog_info(msg, ...) syslog(LOG_INFO, msg, ##__VA_ARGS__) -#define syslog_emerg(msg, ...) syslog(LOG_EMERG, "%s:%d [" msg "]", __func__, __LINE__, ##__VA_ARGS__) -#define syslog_err(msg, ...) syslog(LOG_ERR, "%s:%d [" msg "]", __func__, __LINE__, ##__VA_ARGS__) - - -#ifndef NDEBUG -#define syslog_debug(msg, ...) syslog(LOG_DEBUG, "%s:%d " msg "", __func__, __LINE__, ##__VA_ARGS__) -#else -#define syslog_debug(msg, ...) -#endif +#include #define PI 3.141592654 @@ -727,6 +717,10 @@ void *cp_service(void *fc){ it2s_tender_db_add(&facilities->db, &facilities->epv, true, 14, bpr->data.buf, bpr->data.size); } + uint64_t id = rand() + 1; + bpr->id = id; + fmi->id = id; + /* Encode TransportRequest */ asn_enc_rval_t enc_tdr = oer_encode_to_buffer(&asn_DEF_TransportRequest, NULL, tr, tr_oer+1, 2047); if(enc_tdr.encoded == -1){ @@ -742,12 +736,10 @@ void *cp_service(void *fc){ } /* Create thread to send packet to the Transport Layer (=3) */ - queue_add(facilities->tx_queue, tr_oer, enc_tdr.encoded+1, 3); - pthread_cond_signal(&facilities->tx_queue->trigger); + it2s_tender_queue_send(facilities->tx_queue, tr_oer, enc_tdr.encoded+1, ITSS_TRANSPORT, id, "TR.packet.btp"); /* Create thread to send packet to the Applications Layer (=5) */ - queue_add(facilities->tx_queue, fi_oer, enc_fdi.encoded+1, 5); - pthread_cond_signal(&facilities->tx_queue->trigger); + it2s_tender_queue_send(facilities->tx_queue, fi_oer, enc_fdi.encoded+1, ITSS_APPLICATIONS, id, "FI.message"); /*Reset Timer for dissemination control */ dissemination_reset_timer(facilities->dissemination, &facilities->epv,1); diff --git a/src/facilities.c b/src/facilities.c index e4e2142..e18aeae 100644 --- a/src/facilities.c +++ b/src/facilities.c @@ -311,8 +311,7 @@ static int transport_indication(facilities_t *facilities, void* responder, void* goto cleanup; } - queue_add(facilities->tx_queue, buf, enc.encoded+1, 3); - pthread_cond_signal(&facilities->tx_queue->trigger); + it2s_tender_queue_send(facilities->tx_queue, buf, enc.encoded+1, ITSS_TRANSPORT, id, "TR.packet.tcp"); break; case TransportPacketIndication_PR_udp: @@ -337,11 +336,11 @@ static int transport_indication(facilities_t *facilities, void* responder, void* fmi->data.buf = malloc(tpi->choice.btp.data.size); memcpy(fmi->data.buf, tpi->choice.btp.data.buf, tpi->choice.btp.data.size); - uint8_t buffer[PACKET_MAX_LEN]; + uint8_t buffer[ITSS_SDU_MAX_LEN]; buffer[0] = 4; // Facilities - asn_enc_rval_t enc = oer_encode_to_buffer(&asn_DEF_FacilitiesIndication, NULL, fi, buffer+1, PACKET_MAX_LEN-1); + asn_enc_rval_t enc = oer_encode_to_buffer(&asn_DEF_FacilitiesIndication, NULL, fi, buffer+1, ITSS_SDU_MAX_LEN-1); - queue_add(facilities->tx_queue, buffer, enc.encoded+1, 5); + it2s_tender_queue_add(facilities->tx_queue, buffer, enc.encoded+1, ITSS_APPLICATIONS, id, "FI.message"); pthread_cond_signal(&facilities->tx_queue->trigger); } @@ -628,7 +627,7 @@ void* tx(void* fc) { facilities_t *facilities = (facilities_t*) fc; - queue_t* queue = facilities->tx_queue; + it2s_tender_queue_t* queue = facilities->tx_queue; uint8_t code; int wait_ms = 1000; @@ -641,7 +640,7 @@ void* tx(void* fc) { zmq_setsockopt(transport_socket, ZMQ_RCVTIMEO, &wait_ms, sizeof(int)); zmq_connect(transport_socket, facilities->zmq.transport_address); - queue_t* stream = queue_init(); + it2s_tender_queue_t* stream = it2s_tender_queue_new(); while (!facilities->exit) { pthread_mutex_lock(&queue->lock); @@ -652,6 +651,8 @@ void* tx(void* fc) { memcpy(stream->packet[i], queue->packet[i], queue->packet_len[i]); stream->packet_len[i] = queue->packet_len[i]; stream->destination[i] = queue->destination[i]; + strcpy(stream->info_msg[i], queue->info_msg[i]); + stream->id[i] = queue->id[i]; } stream->len = queue->len; @@ -660,17 +661,26 @@ void* tx(void* fc) { for (int i = 0; i < stream->len; ++i) { switch (stream->destination[i]) { - case 3: - syslog_debug("[facilities]-> sending BDR to ->[transport] | size:%dB", stream->packet_len[i]); + case ITSS_TRANSPORT: + syslog_debug("[facilities]-> sending %s to ->[transport] | id:%ld size:%dB", + stream->info_msg[i], stream->id[i], stream->packet_len[i]); zmq_send(transport_socket, stream->packet[i], stream->packet_len[i], 0); rv = zmq_recv(transport_socket, &code, 1, 0); - if (rv == -1) {syslog_err("[facilities]-> timeout sending BDR to ->[transport]");} + if (rv == -1) { + syslog_debug("[facilities]-> timeout sending %s to ->[transport] | id:%ld size:%dB", + stream->info_msg[i], stream->id[i], stream->packet_len[i]); + } break; - case 5: + case ITSS_APPLICATIONS: syslog_debug("[facilities]-> sending FDI to ->[applications] | size:%dB", stream->packet_len[i]); + syslog_debug("[facilities]-> sending %s to ->[applications] | id:%ld size:%dB", + stream->info_msg[i], stream->id[i], stream->packet_len[i]); zmq_send(applications_socket, stream->packet[i], stream->packet_len[i], 0); rv = zmq_recv(applications_socket, &code, 1, 0); - if (rv == -1) {syslog_err("[facilities]-> timeout sending FDI to ->[applications]");} + if (rv == -1) { + syslog_debug("[facilities]-> timeout sending %s to ->[transport] | id:%ld size:%dB", + stream->info_msg[i], stream->id[i], stream->packet_len[i]); + } break; } } @@ -688,7 +698,7 @@ int main() { memset(&facilities, 0x00, sizeof(facilities_t)); facilities.zmq.ctx = zmq_ctx_new(); - facilities.tx_queue = queue_init(); + facilities.tx_queue = it2s_tender_queue_new(); facilities.lightship = lightship_init(); facilities.den = calloc(1, sizeof(den_t)); facilities.infrastructure = calloc(1, sizeof(infrastructure_t)); @@ -740,7 +750,7 @@ int main() { zmq_setsockopt(security_socket, ZMQ_RCVTIMEO, &wait_ms, sizeof(int)); zmq_connect(security_socket, facilities.zmq.security_address); - uint8_t buffer[PACKET_MAX_LEN]; + uint8_t buffer[ITSS_SDU_MAX_LEN]; syslog_info("[facilities] listening"); uint8_t code; bool in_idchange; @@ -752,7 +762,7 @@ int main() { for (int i = 0; i < facilities.zmq.n_responders; ++i) { if (facilities.zmq.responders[i].revents) { - rl = zmq_recv(facilities.zmq.responders[i].socket, buffer, PACKET_MAX_LEN, 0); + rl = zmq_recv(facilities.zmq.responders[i].socket, buffer, ITSS_SDU_MAX_LEN, 0); switch (buffer[0]) { case 2: diff --git a/src/facilities.h b/src/facilities.h index 6e0cc19..1769b21 100644 --- a/src/facilities.h +++ b/src/facilities.h @@ -8,7 +8,6 @@ #include "cam.h" #include "denm.h" #include "infrastructure.h" -#include "queue.h" #include "cpm.h" #include "saem.h" #include "tpm.h" @@ -17,6 +16,8 @@ #include #include #include +#include +#include enum ID_CHANGE_STAGE { ID_CHANGE_INACTIVE, @@ -47,7 +48,7 @@ typedef struct facilities { } zmq; // Transmitter - queue_t* tx_queue; + it2s_tender_queue_t* tx_queue; // CA lightship_t* lightship; diff --git a/src/pcm.c b/src/pcm.c index 3d979fd..6ba11cb 100644 --- a/src/pcm.c +++ b/src/pcm.c @@ -134,6 +134,8 @@ static int pcm_check_handle_request(facilities_t* facilities, PCM_t* pcm, mc_nei bpr->btpType = BTPType_btpB; + bpr->id = rand() + 1; + bpr->gn.destinationAddress.buf = malloc(6); for (int i = 0; i < 6; ++i) { bpr->gn.destinationAddress.buf[i] = 0xff; @@ -158,7 +160,7 @@ static int pcm_check_handle_request(facilities_t* facilities, PCM_t* pcm, mc_nei goto cleanup; } - queue_send(facilities->tx_queue, buf, enc.encoded+1, 3); + it2s_tender_queue_send(facilities->tx_queue, buf, enc.encoded+1, ITSS_TRANSPORT, bpr->id, "TR.packet.btp"); cleanup: ASN_STRUCT_FREE(asn_DEF_PCM, pcm_rep); @@ -302,6 +304,8 @@ static int pcm_check_intersection_detected(facilities_t* facilities, PCM_t* pcm, tr->choice.packet.present = TransportPacketRequest_PR_btp; BTPPacketRequest_t* bpr = &tr->choice.packet.choice.btp; + bpr->id = rand() + 1; + bpr->btpType = BTPType_btpB; bpr->gn.destinationAddress.buf = malloc(6); @@ -328,7 +332,7 @@ static int pcm_check_intersection_detected(facilities_t* facilities, PCM_t* pcm, goto cleanup; } - queue_send(facilities->tx_queue, buf, enc.encoded+1, 3); + it2s_tender_queue_send(facilities->tx_queue, buf, enc.encoded+1, ITSS_TRANSPORT, bpr->id, "TR.packet.btp"); cleanup: @@ -569,13 +573,15 @@ void* pc_service(void* fc) { continue; } + bpr->id = rand() + 1; + asn_enc_rval_t enc = oer_encode_to_buffer(&asn_DEF_TransportRequest, NULL, tr, tr_oer+1, 1023); if (enc.encoded == -1) { syslog_err("[facilities] encoding TR for PCM failed"); continue; } - queue_send(facilities->tx_queue, tr_oer, enc.encoded+1, 3); + it2s_tender_queue_send(facilities->tx_queue, tr_oer, enc.encoded+1, ITSS_TRANSPORT, bpr->id, "TR.packet.btp"); coordination->t_last_send_pcm = now; } diff --git a/src/queue.c b/src/queue.c deleted file mode 100644 index 6094c6e..0000000 --- a/src/queue.c +++ /dev/null @@ -1,66 +0,0 @@ -#include "queue.h" - -#include -#include - -queue_t* queue_init() { - queue_t* queue = calloc(1, sizeof(queue_t)); - - pthread_mutex_init(&queue->lock, NULL); - pthread_cond_init(&queue->trigger, NULL); - - queue->len = 0; - queue->packet = calloc(QUEUE_MAX_LEN, sizeof(uint8_t *)); - for (int i = 0; i < QUEUE_MAX_LEN; ++i) { - queue->packet[i] = malloc(PACKET_MAX_LEN); - } - queue->packet_len = calloc(QUEUE_MAX_LEN, sizeof(uint16_t)); - queue->destination = malloc(QUEUE_MAX_LEN); - - return queue; -} - -int queue_add(queue_t* queue, uint8_t *packet, uint16_t packet_len, uint8_t destination) { - int rv = 0; - - pthread_mutex_lock(&queue->lock); - - if (queue->len < QUEUE_MAX_LEN) { - queue->packet_len[queue->len] = packet_len; - memcpy(queue->packet[queue->len], packet, packet_len); - queue->destination[queue->len] = destination; - ++queue->len; - } else { - rv = 1; - } - - pthread_mutex_unlock(&queue->lock); - - return rv; -} - -int queue_send(queue_t* queue, uint8_t *packet, uint16_t packet_len, uint8_t destination) { - int rv = 0; - - pthread_mutex_lock(&queue->lock); - - if (queue->len < QUEUE_MAX_LEN) { - queue->packet_len[queue->len] = packet_len; - memcpy(queue->packet[queue->len], packet, packet_len); - queue->destination[queue->len] = destination; - ++queue->len; - } else { - rv = 1; - } - - pthread_mutex_unlock(&queue->lock); - - queue_trigger(queue); - - return rv; -} - -void queue_trigger(queue_t* queue) { - pthread_cond_signal(&queue->trigger); -} - diff --git a/src/queue.h b/src/queue.h deleted file mode 100644 index 0d8d101..0000000 --- a/src/queue.h +++ /dev/null @@ -1,22 +0,0 @@ -#pragma once - -#include -#include -#include - -#define QUEUE_MAX_LEN 32 -#define PACKET_MAX_LEN 32768 - -typedef struct queue { - uint8_t **packet; - uint16_t *packet_len; - uint8_t *destination; - uint16_t len; - pthread_mutex_t lock; - pthread_cond_t trigger; -} queue_t; - -queue_t* queue_init(); -int queue_add(queue_t* queue, uint8_t *packet, uint16_t packet_len, uint8_t destination); -int queue_send(queue_t* queue, uint8_t *packet, uint16_t packet_len, uint8_t destination); -void queue_trigger(queue_t* queue); diff --git a/src/requests.c b/src/requests.c index 32312ce..5ec38d7 100644 --- a/src/requests.c +++ b/src/requests.c @@ -316,9 +316,7 @@ int facilities_request_single_message(facilities_t* facilities, void* responder, goto cleanup; } - queue_add(facilities->tx_queue, tr_oer, enc.encoded+1, 3); - pthread_cond_signal(&facilities->tx_queue->trigger); - + it2s_tender_queue_send(facilities->tx_queue, tr_oer, enc.encoded+1, ITSS_TRANSPORT, id, "TR.packet.btp"); } cleanup: diff --git a/src/saem.c b/src/saem.c index b7e50c8..6f1e0ce 100644 --- a/src/saem.c +++ b/src/saem.c @@ -325,13 +325,14 @@ void *sa_service(void *fc) { it2s_tender_db_add(&facilities->db, &facilities->epv, true, messageID_saem, bpr->data.buf, bpr->data.size); } + bpr->id = rand() + 1; + asn_enc_rval_t enc = oer_encode_to_buffer(&asn_DEF_TransportRequest, NULL, tr, tr_oer+1, 1023); if (enc.encoded == -1) { syslog_err("[facilities] encoding TR for SAEM failed"); continue; } else { - queue_add(facilities->tx_queue, tr_oer, enc.encoded+1, 3); - pthread_cond_signal(&facilities->tx_queue->trigger); + it2s_tender_queue_send(facilities->tx_queue, tr_oer, enc.encoded+1, ITSS_TRANSPORT, bpr->id, "TR.packet.btp"); } } mk_saem_n_sleep = 0; @@ -421,6 +422,8 @@ void *sa_service(void *fc) { tpr_etc->choice.tcp.gn->destinationAddress.buf = calloc(1, 6); tpr_etc->choice.tcp.gn->destinationAddress.size = 6; + tpr_etc->choice.tcp.id = rand() + 1; + ++bulletin->to_consume[a]->n_trigger; enc = oer_encode_to_buffer(&asn_DEF_TransportRequest, NULL, tr_etc, tr_oer+1, 1023); @@ -428,8 +431,7 @@ void *sa_service(void *fc) { syslog_err("[facilities] encoding TR for ETC-Req failed"); continue; } else { - queue_add(facilities->tx_queue, tr_oer, enc.encoded+1, 3); - pthread_cond_signal(&facilities->tx_queue->trigger); + it2s_tender_queue_send(facilities->tx_queue, tr_oer, enc.encoded+1, ITSS_TRANSPORT, tpr_etc->choice.tcp.id, "TR.packet.tcp"); } ASN_STRUCT_FREE(asn_DEF_TransportRequest, tr_etc); diff --git a/src/tpm.c b/src/tpm.c index 94555c5..6112559 100644 --- a/src/tpm.c +++ b/src/tpm.c @@ -133,6 +133,8 @@ int tpm_pay(void* fc, uint8_t* neighbour) { tr->choice.packet.present = TransportPacketRequest_PR_btp; BTPPacketRequest_t* bpr = &tr->choice.packet.choice.btp; + bpr->id = rand() + 1; + bpr->gn.securityProfile.encrypt = true; bpr->gn.securityProfile.sign = true; @@ -168,7 +170,7 @@ int tpm_pay(void* fc, uint8_t* neighbour) { goto cleanup; } - queue_send(facilities->tx_queue, buf, enc.encoded+1, 3); + it2s_tender_queue_send(facilities->tx_queue, buf, enc.encoded+1, ITSS_TRANSPORT, bpr->id, "TR.packet.btp"); cleanup: ASN_STRUCT_FREE(asn_DEF_TPM, tpm); @@ -330,6 +332,8 @@ static void rsu_handle_recv(facilities_t* facilities, TPM_t* tpm_rx, uint8_t* ne tr->choice.packet.present = TransportPacketRequest_PR_btp; BTPPacketRequest_t* bpr = &tr->choice.packet.choice.btp; + bpr->id = rand() + 1; + bpr->gn.securityProfile.encrypt = true; bpr->gn.securityProfile.sign = true; @@ -366,7 +370,7 @@ static void rsu_handle_recv(facilities_t* facilities, TPM_t* tpm_rx, uint8_t* ne goto cleanup; } - queue_send(facilities->tx_queue, buf, enc.encoded+1, 3); + it2s_tender_queue_send(facilities->tx_queue, buf, enc.encoded+1, ITSS_TRANSPORT, bpr->id, "TR.packet.btp"); cleanup: ASN_STRUCT_FREE(asn_DEF_TPM, tpm);