Resolve issues with integrity signing request pipeline

 * replace libtrace queues for sending requests with ZMQ
 * include forwarding thread ID in requests/responses to ensure
   that responses are routed back to the collector thread that
   asked for them
 * use entirely string based keys for identifying integrity
   "chains", as a 4 byte CIN where one of the bytes was 0x00
   would cause problems
 * add extra map for storing the string key for each chain that
   is associated with an LIID, so we can look up the key with a
   uint64_t hash rather than having to construct the string every
   time we need it

Author: salcock

configure.ac

@@ -52,6 +52,7 @@ AC_CHECK_LIB([uuid], [uuid_parse],,libuuid_found=0)
 
 if test "x$libzmq_found" = "x1"; then
         COLLECTOR_LIBS="$COLLECTOR_LIBS -lzmq"
+        MEDIATOR_LIBS="$MEDIATOR_LIBS -lzmq"
 fi
 
 if test "x$libssl11_found" = "x1"; then

src/mediator/coll_recv_thread.c

@@ -331,6 +331,7 @@ static void remove_expired_liid_queues(coll_recv_t *col) {
             etsili_clear_preencoded_fields(known->preencoded_etsi);
             free(known->preencoded_etsi);
         }
+        clear_digest_key_map(known);
         HASH_DELETE(hh, col->known_liids, known);
         free(known);
     }
@@ -677,6 +678,7 @@ static col_known_liid_t *create_new_known_liid(coll_recv_t *col,
     found->provisioner_withdrawn = 0;
     found->preencoded_etsi = calloc(OPENLI_PREENCODE_LAST,
             sizeof(wandder_encode_job_t));
+    found->digest_cin_keys = NULL;
 
     snprintf(qname, 1024, "%s-iri", found->liid);
     found->queuenames[0] = strdup(qname);
@@ -1301,10 +1303,15 @@ static void cleanup_collector_thread(coll_recv_t *col) {
             etsili_clear_preencoded_fields(known->preencoded_etsi);
             free(known->preencoded_etsi);
         }
+        clear_digest_key_map(known);
         HASH_DELETE(hh, col->known_liids, known);
         free(known);
     }
 
+    if (col->zmq_requests) {
+        zmq_close(col->zmq_requests);
+    }
+
     if (col->ipaddr && col->forwarder_id >= 0) {
         logger(LOG_INFO, "OpenLI mediator: exiting collector thread %d for %s",
                 col->forwarder_id, col->ipaddr);
@@ -1359,6 +1366,7 @@ static void *start_collector_thread(void *params) {
     int epoll_fd = -1, timerexpired, nfds;
     med_epoll_ev_t *timerev, *queuecheck = NULL;
     struct epoll_event evs[64];
+    int zero = 0;
 
     if (col->ipaddr == NULL) {
         logger(LOG_INFO, "OpenLI Mediator: started collector thread for NULL collector IP??");
@@ -1376,6 +1384,26 @@ static void *start_collector_thread(void *params) {
     }
     unlock_med_collector_config(col->parentconfig);
 
+    /* create ZMQ requests queue for publishing requests */
+    col->zmq_requests = zmq_socket(col->zmq_ctxt, ZMQ_PUSH);
+    if (zmq_connect(col->zmq_requests, "inproc://openlimed_collrecv_requests")
+            < 0) {
+        logger(LOG_INFO,
+                "OpenLI Mediator: collector thread for %s failed to connect to the ZMQ for pushing requests back to the main thread: %s",
+                col->ipaddr, strerror(errno));
+        zmq_close(col->zmq_requests);
+        col->zmq_requests = NULL;
+    }
+
+    if (col->zmq_requests && zmq_setsockopt(col->zmq_requests, ZMQ_LINGER,
+            &zero, sizeof(zero)) != 0) {
+        logger(LOG_INFO,
+                "OpenLI Mediator: collector thread for %s failed to configure the ZMQ for pushing requests back to the main thread: %s",
+                col->ipaddr, strerror(errno));
+        zmq_close(col->zmq_requests);
+        col->zmq_requests = NULL;
+    }
+
     epoll_fd = epoll_create1(0);
 
     timerev = col->colev = col->rmq_colev = NULL;
@@ -1664,8 +1692,10 @@ static void init_new_colrecv_thread(mediator_collector_t *medcol,
 
     libtrace_message_queue_init(&(newcol->in_main),
             sizeof(col_thread_msg_t));
-    libtrace_message_queue_init(&(newcol->out_main),
-            sizeof(col_thread_msg_t));
+
+    newcol->zmq_ctxt = medcol->zmq_ctxt;
+    newcol->zmq_requests = NULL;
+
     pthread_create(&(newcol->tid), NULL, start_collector_thread, newcol);
 }
 
@@ -1743,7 +1773,6 @@ void mediator_disconnect_all_collectors(mediator_collector_t *medcol) {
 
             pthread_join(col->tid, NULL);
             libtrace_message_queue_destroy(&(col->in_main));
-            libtrace_message_queue_destroy(&(col->out_main));
             tofree = col;
             col = col->next;
             free(tofree);
@@ -1834,7 +1863,10 @@ void mediator_clean_collectors(mediator_collector_t *medcol) {
 
             pthread_join(tofree->tid, NULL);
             libtrace_message_queue_destroy(&(tofree->in_main));
-            libtrace_message_queue_destroy(&(tofree->out_main));
+
+            if (tofree->zmq_requests) {
+                zmq_close(tofree->zmq_requests);
+            }
 
             if (tofree == col && newhead != oldhead) {
                 /* we are removing the head, so we need

src/mediator/coll_recv_thread.h

@@ -29,6 +29,7 @@
 
 #include <libtrace/message_queue.h>
 #include <libwandder_etsili.h>
+#include <zmq.h>
 #include <openssl/evp.h>
 #include <openssl/err.h>
 #include "netcomms.h"
@@ -123,6 +124,13 @@ typedef struct col_thread_msg {
 
 typedef struct agency_digest_config agency_digest_config_t;
 
+typedef struct digest_map_key {
+    uint64_t key_cin;
+    const char *keystring;
+
+    UT_hash_handle hh;
+} digest_map_key_t;
+
 /** Structure for keeping track of the LIIDs that a collector receive thread
  *  has seen
  */
@@ -158,6 +166,8 @@ typedef struct col_known_liid {
 
     wandder_encode_job_t *preencoded_etsi;
 
+    digest_map_key_t *digest_cin_keys;
+
     UT_hash_handle hh;
 } col_known_liid_t;
 
@@ -389,10 +399,13 @@ struct single_coll_receiver {
      */
     libtrace_message_queue_t in_main;
 
+    /** Global ZMQ context for the entire mediator process */
+    void *zmq_ctxt;
+
     /** The message queue on which this thread will send requests (e.g.
      *  integrity check signing requests) back to the main mediator thread.
      */
-    libtrace_message_queue_t out_main;
+    void *zmq_requests;
 
     /** Flag that indicates whether RMQ has told us that it is "connection
      *  blocked, i.e. no longer able to accept published messages
@@ -450,6 +463,8 @@ typedef struct mediator_collectors {
     /** A hashmap containing the set of collector receive threads */
     coll_recv_t *threads;
 
+    void *zmq_ctxt;
+
     /** Shared configuration for all collector receive threads */
     mediator_collector_config_t config;
 
@@ -613,6 +628,7 @@ int send_integrity_check_signing_request(coll_recv_t *col,
 void handle_integrity_check_signature_response(coll_recv_t *col,
         struct ics_sign_response_message *resp);
 void destroy_integrity_sign_job(ics_sign_request_t *job);
+void clear_digest_key_map(col_known_liid_t *known);
 
 /* defined in mediator_encryption.c */
 payload_encryption_method_t check_encryption_requirements(

src/mediator/mediator.c

@@ -191,6 +191,13 @@ static void destroy_med_state(mediator_state_t *state) {
 		free(state->col_clean_timerev);
 	}
 
+    if (state->zmq_request_collrecv) {
+        zmq_close(state->zmq_request_collrecv);
+    }
+    if (state->zmq_ctxt) {
+        zmq_ctx_destroy(state->zmq_ctxt);
+    }
+
 }
 
 /** Reads the configuration for a mediator instance and sets the relevant
@@ -281,12 +288,33 @@ static int init_mediator_config(mediator_state_t *state,
  *  @return -1 if an error occurs, 0 otherwise
  */
 static int init_med_state(mediator_state_t *state, char *configfile) {
+    int zero = 0;
+
     state->listenerev = NULL;
     state->timerev = NULL;
     state->col_clean_timerev = NULL;
     state->epoll_fd = -1;
     state->saved_agencies = NULL;
 
+    state->zmq_ctxt = zmq_ctx_new();
+    state->zmq_request_collrecv = zmq_socket(state->zmq_ctxt, ZMQ_PULL);
+    if (zmq_bind(state->zmq_request_collrecv,
+            "inproc://openlimed_collrecv_requests") < 0) {
+        logger(LOG_INFO,
+                "OpenLI mediator: unable to bind to ZMQ socket for receiving requests from the collrecv threads: %s", strerror(errno));
+        zmq_close(state->zmq_request_collrecv);
+        state->zmq_request_collrecv = NULL;
+    }
+
+    if (state->zmq_request_collrecv &&
+            zmq_setsockopt(state->zmq_request_collrecv, ZMQ_LINGER, &zero,
+                    sizeof(zero)) != 0) {
+        logger(LOG_INFO,
+                "OpenLI mediator: unable to configure ZMQ socket for receiving requests from the collrecv threads: %s", strerror(errno));
+        zmq_close(state->zmq_request_collrecv);
+        state->zmq_request_collrecv = NULL;
+    }
+
     init_provisioner_instance(&(state->provisioner), &(state->sslconf.ctx));
     if (init_mediator_config(state, configfile) < 0) {
         return -1;
@@ -306,6 +334,7 @@ static int init_med_state(mediator_state_t *state, char *configfile) {
 
     /* Initialise state and config for the collector receive threads */
     state->collector_threads.threads = NULL;
+    state->collector_threads.zmq_ctxt = state->zmq_ctxt;
     init_med_collector_config(&(state->collector_threads.config),
             state->etsitls,
             &(state->sslconf), &(state->RMQ_conf), state->mediatorid,
@@ -932,11 +961,17 @@ static int receive_ics_signature(mediator_state_t *state, uint8_t *msgbody,
         goto tidyup_err;
     }
 
-    memset(&msg, 0, sizeof(msg));
-    msg.type = MED_COLL_INTEGRITY_SIGN_RESULT;
-    msg.arg = (uint64_t)resp;
+    while (col != NULL) {
+        if (col->forwarder_id >= 0 &&
+                (uint32_t)col->forwarder_id == resp->requestedby_fwd) {
+            memset(&msg, 0, sizeof(msg));
+            msg.type = MED_COLL_INTEGRITY_SIGN_RESULT;
+            msg.arg = (uint64_t)resp;
 
-    libtrace_message_queue_put(&(col->in_main), &msg);
+            libtrace_message_queue_put(&(col->in_main), &msg);
+        }
+        col = col->next;
+    }
 
     return 0;
 
@@ -1593,6 +1628,48 @@ static int reload_mediator_config(mediator_state_t *currstate) {
 
 }
 
+static void process_collector_thread_requests(mediator_state_t *state) {
+
+    int reqs_processed = 0;
+    unsigned char buf[1024];
+    col_thread_msg_t *colmsg;
+    int r;
+
+    while (reqs_processed < 10 && state->zmq_request_collrecv) {
+        if ((r = zmq_recv(state->zmq_request_collrecv, buf, 1024,
+                        ZMQ_DONTWAIT)) < 0) {
+            if (errno == EAGAIN || errno == EINTR) {
+                /* no requests available */
+                break;
+            }
+            logger(LOG_INFO, "OpenLI mediator: error while reading a request from one of the collector threads: %s", strerror(errno));
+            zmq_close(state->zmq_request_collrecv);
+            state->zmq_request_collrecv = NULL;
+            continue;
+        }
+
+        if (r < (int)(sizeof(col_thread_msg_t))) {
+            logger(LOG_INFO, "OpenLI mediator: unexpected message size received from one of the collector threads: %d\n", r);
+            continue;
+        }
+
+        colmsg = (col_thread_msg_t *)(buf);
+
+        if (colmsg->type == MED_COLL_INTEGRITY_SIGN_REQUEST) {
+            struct ics_sign_request_message *signreq;
+            signreq = (struct ics_sign_request_message *)(colmsg->arg);
+            if (send_ics_signing_request_to_provisioner(
+                        &state->provisioner, signreq) < 0) {
+                logger(LOG_INFO, "OpenLI mediator: failed to pass on integrity check signing request to the provisioner");
+            }
+
+        } else {
+            logger(LOG_INFO, "OpenLI mediator: invalid message type received by main thread from collector thread (%u)", colmsg->type);
+        }
+        reqs_processed ++;
+    }
+}
+
 /** The main loop of the mediator process.
  *
  *  Continually checks for registered epoll events, e.g. timers expiring,
@@ -1611,7 +1688,6 @@ static void run(mediator_state_t *state) {
 	struct epoll_event evs[64];
     int provfail = 0;
     med_epoll_ev_t *signalev;
-    coll_recv_t *col_t, *tmp;
 
     /* Register the epoll event for received signals */
     signalev = create_mediator_fdevent(state->epoll_fd, NULL,
@@ -1638,9 +1714,6 @@ static void run(mediator_state_t *state) {
         goto runfailure;
     }
 
-    /* TODO this timer should be longer, but for testing I've set to fire
-     * more frequently
-     */
     if (start_mediator_timer(state->col_clean_timerev, 30) < 0) {
         logger(LOG_INFO,
                 "OpenLI Mediator: failed to start collector cleanup timer");
@@ -1689,23 +1762,7 @@ static void run(mediator_state_t *state) {
         /* Check for integrity check signing requests from the collector
          * threads.
          */
-        HASH_ITER(hh, state->collector_threads.threads, col_t, tmp) {
-            col_thread_msg_t colmsg;
-            while (libtrace_message_queue_try_get(&(col_t->out_main),
-                    (void *)&colmsg) != LIBTRACE_MQ_FAILED) {
-                if (colmsg.type == MED_COLL_INTEGRITY_SIGN_REQUEST) {
-                    struct ics_sign_request_message *signreq;
-                    signreq = (struct ics_sign_request_message *)(colmsg.arg);
-                    if (send_ics_signing_request_to_provisioner(
-                            &state->provisioner, signreq) < 0) {
-                        logger(LOG_INFO, "OpenLI mediator: failed to pass on integrity check signing request to the provisioner");
-                    }
-
-                } else {
-                    logger(LOG_INFO, "OpenLI mediator: invalid message type received by main thread from collector thread (%u)", colmsg.type);
-                }
-            }
-        }
+        process_collector_thread_requests(state);
 
         /* This timer will force us to stop checking epoll and go back
          * to the start of this loop (i.e. checking if we should halt the

src/mediator/mediator.h

@@ -31,6 +31,7 @@
 #include <libwandder.h>
 #include <libwandder_etsili.h>
 #include <uthash.h>
+#include <zmq.h>
 #include "netcomms.h"
 #include "export_buffer.h"
 #include "util.h"
@@ -115,6 +116,13 @@ typedef struct med_state {
     /** The RabbitMQ configuration for the mediator */
     openli_RMQ_config_t RMQ_conf;
 
+    /** ZeroMQ context for creating ZMQ sockets */
+    void *zmq_ctxt;
+
+    /** ZeroMQ socket for receiving requests from collector threads */
+    void *zmq_request_collrecv;
+
+
 } mediator_state_t;
 
 #endif