Syslog efforts

src/CMakeLists.txt

@@ -1,6 +1,5 @@
 ADD_EXECUTABLE(it2s-itss-facilities
     config.c
-    queue.c
     cam.c
     denm.c
     infrastructure.c

src/cam.c

@@ -19,19 +19,11 @@
 #include <it2s-tender/space.h>
 #include <it2s-tender/time.h>
 #include <it2s-tender/database.h>
+#include <it2s-tender/constants.h>
+#include <it2s-tender/syslog.h>
 
 #include <it2s-obd/shmem.h>
 
-#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);
+            it2s_tender_queue_add(facilities->tx_queue, tr_oer, enc.encoded+1, ITSS_TRANSPORT, id, "TR.packet.btp");
-            pthread_cond_signal(&facilities->tx_queue->trigger);
 
-            queue_add(facilities->tx_queue, fi_oer, enc_fdi.encoded+1, 5);
+            it2s_tender_queue_add(facilities->tx_queue, fi_oer, enc_fdi.encoded+1, ITSS_APPLICATIONS, id, "FI.message");
-            pthread_cond_signal(&facilities->tx_queue->trigger);
 
             lightship_reset_timer(facilities->lightship, &facilities->epv);
         }

src/cpm.c

@@ -21,17 +21,7 @@
 
 #include <it2s-tender/time.h>
 #include <it2s-tender/space.h>
-
+#include <it2s-tender/syslog.h>
-#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 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);
+                it2s_tender_queue_send(facilities->tx_queue, tr_oer, enc_tdr.encoded+1, ITSS_TRANSPORT, id, "TR.packet.btp");
-                pthread_cond_signal(&facilities->tx_queue->trigger);
 
                 /* Create thread to send packet to the Applications Layer (=5) */
-                queue_add(facilities->tx_queue, fi_oer, enc_fdi.encoded+1, 5);
+                it2s_tender_queue_send(facilities->tx_queue, fi_oer, enc_fdi.encoded+1, ITSS_APPLICATIONS, id, "FI.message");
-                pthread_cond_signal(&facilities->tx_queue->trigger);
 
                 /*Reset Timer for dissemination control */ 
                 dissemination_reset_timer(facilities->dissemination, &facilities->epv,1);

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);
+            it2s_tender_queue_send(facilities->tx_queue, buf, enc.encoded+1, ITSS_TRANSPORT, id, "TR.packet.tcp");
-            pthread_cond_signal(&facilities->tx_queue->trigger);
 
             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:
+                case ITSS_TRANSPORT:
-                    syslog_debug("[facilities]-> sending BDR to ->[transport] | size:%dB", stream->packet_len[i]);
+                    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:

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 <it2s-tender/epv.h>
 #include <it2s-tender/syslog.h>
 #include <it2s-tender/database.h>
+#include <it2s-tender/constants.h>
+#include <it2s-tender/queue.h>
 
 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;

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;
         }

src/queue.c

@@ -1,66 +0,0 @@
-#include "queue.h"
-
-#include <stdlib.h>
-#include <string.h>
-
-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);
-}
-

src/queue.h

@@ -1,22 +0,0 @@
-#pragma once
-
-#include <pthread.h>
-#include <stdbool.h>
-#include <stdint.h>
-
-#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);

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);
+        it2s_tender_queue_send(facilities->tx_queue, tr_oer, enc.encoded+1, ITSS_TRANSPORT, id, "TR.packet.btp");
-        pthread_cond_signal(&facilities->tx_queue->trigger);
-
     }
 
 cleanup:

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);
+                    it2s_tender_queue_send(facilities->tx_queue, tr_oer, enc.encoded+1, ITSS_TRANSPORT, bpr->id, "TR.packet.btp");
-                    pthread_cond_signal(&facilities->tx_queue->trigger);
                 }
             }
             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);
+                            it2s_tender_queue_send(facilities->tx_queue, tr_oer, enc.encoded+1, ITSS_TRANSPORT,  tpr_etc->choice.tcp.id, "TR.packet.tcp");
-                            pthread_cond_signal(&facilities->tx_queue->trigger);
                         }
 
                         ASN_STRUCT_FREE(asn_DEF_TransportRequest, tr_etc);

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);