Reworked nDPIsrvd.h C-API.

 * nDPIsrvd.h: Provide nDPId thread storage.
 * nDPIsrvd.py: Fixed instance cleanup bug.
 * nDPIsrvd.h: Support for instance/thread user data and cleanup callback.
 * nDPIsrvd.h: Most recent flow time stored in thread ht instead of instance ht.
 * nDPId: Moved flow logger out the memory profilier into SIGUSR1 signal handling.
 * nDPId: Added signal fd to be usable within epoll's event handling (live-capture only!)
 * nDPId: Added information about ZLib compressions to daemon status/shutdown events.

Signed-off-by: Toni Uhlig <matzeton@googlemail.com>

3 files changed, 78 insertions, 17 deletions

diff --git a/examples/c-captured/c-captured.c b/examples/c-captured/c-captured.c
index f8b716c53..d87fb0df5 100644
--- a/examples/c-captured/c-captured.c
+++ b/examples/c-captured/c-captured.c
@@ -338,9 +338,11 @@ static enum nDPIsrvd_conversion_return perror_ull(enum nDPIsrvd_conversion_retur
 
 static enum nDPIsrvd_callback_return captured_json_callback(struct nDPIsrvd_socket * const sock,
                                                             struct nDPIsrvd_instance * const instance,
+                                                            struct nDPIsrvd_thread_data * const thread_data,
                                                             struct nDPIsrvd_flow * const flow)
 {
     (void)instance;
+    (void)thread_data;
 
     if (flow == NULL)
     {
@@ -440,8 +442,7 @@ static enum nDPIsrvd_callback_return captured_json_callback(struct nDPIsrvd_sock
                     nDPIsrvd_ull numeric_risk_value = (nDPIsrvd_ull)-1;
 
                     if (TOKEN_KEY_TO_ULL(current, &numeric_risk_value) == CONVERSION_OK &&
-                        numeric_risk_value < NDPI_MAX_RISK &&
-                        has_ndpi_risk(&process_risky, numeric_risk_value) != 0)
+                        numeric_risk_value < NDPI_MAX_RISK && has_ndpi_risk(&process_risky, numeric_risk_value) != 0)
                     {
                         flow_user->risky = 1;
                     }
@@ -491,8 +492,15 @@ static enum nDPIsrvd_callback_return captured_json_callback(struct nDPIsrvd_sock
     return CALLBACK_OK;
 }
 
-static void nDPIsrvd_write_flow_info_cb(struct nDPIsrvd_flow const * const flow, void * user_data)
+static void nDPIsrvd_write_flow_info_cb(struct nDPIsrvd_socket const * sock,
+                                        struct nDPIsrvd_instance const * instance,
+                                        struct nDPIsrvd_thread_data const * thread_data,
+                                        struct nDPIsrvd_flow const * flow,
+                                        void * user_data)
 {
+    (void)sock;
+    (void)instance;
+    (void)thread_data;
     (void)user_data;
 
     struct flow_user_data const * const flow_user = (struct flow_user_data const *)flow->flow_user_data;
@@ -560,11 +568,13 @@ static void sighandler(int signum)
 
 static void captured_flow_cleanup_callback(struct nDPIsrvd_socket * const sock,
                                            struct nDPIsrvd_instance * const instance,
+                                           struct nDPIsrvd_thread_data * const thread_data,
                                            struct nDPIsrvd_flow * const flow,
                                            enum nDPIsrvd_cleanup_reason reason)
 {
     (void)sock;
     (void)instance;
+    (void)thread_data;
     (void)reason;
 
 #ifdef VERBOSE
@@ -777,8 +787,8 @@ static int mainloop(void)
 
 int main(int argc, char ** argv)
 {
-    sock =
-        nDPIsrvd_socket_init(0, sizeof(struct flow_user_data), captured_json_callback, captured_flow_cleanup_callback);
+    sock = nDPIsrvd_socket_init(
+        0, 0, 0, sizeof(struct flow_user_data), captured_json_callback, NULL, captured_flow_cleanup_callback);
     if (sock == NULL)
     {
         fprintf(stderr, "%s: nDPIsrvd socket memory allocation failed!\n", argv[0]);
diff --git a/examples/c-collectd/c-collectd.c b/examples/c-collectd/c-collectd.c
index c16a0847d..aef84239a 100644
--- a/examples/c-collectd/c-collectd.c
+++ b/examples/c-collectd/c-collectd.c
@@ -440,10 +440,12 @@ static uint64_t get_total_flow_bytes(struct nDPIsrvd_socket * const sock)
 
 static enum nDPIsrvd_callback_return captured_json_callback(struct nDPIsrvd_socket * const sock,
                                                             struct nDPIsrvd_instance * const instance,
+                                                            struct nDPIsrvd_thread_data * const thread_data,
                                                             struct nDPIsrvd_flow * const flow)
 {
     (void)sock;
     (void)instance;
+    (void)thread_data;
     (void)flow;
 
     struct nDPIsrvd_json_token const * const flow_event_name = TOKEN_GET_SZ(sock, "flow_event_name");
@@ -682,7 +684,7 @@ int main(int argc, char ** argv)
 
     openlog("nDPIsrvd-collectd", LOG_CONS, LOG_DAEMON);
 
-    struct nDPIsrvd_socket * sock = nDPIsrvd_socket_init(0, 0, captured_json_callback, NULL);
+    struct nDPIsrvd_socket * sock = nDPIsrvd_socket_init(0, 0, 0, 0, captured_json_callback, NULL, NULL);
     if (sock == NULL)
     {
         LOG(LOG_DAEMON | LOG_ERR, "%s", "nDPIsrvd socket memory allocation failed!");
diff --git a/examples/c-simple/c-simple.c b/examples/c-simple/c-simple.c
index 7580d9a2b..6e7d0f66d 100644
--- a/examples/c-simple/c-simple.c
+++ b/examples/c-simple/c-simple.c
@@ -21,18 +21,25 @@ void nDPIsrvd_memprof_log(char const * const format, ...)
 }
 #endif
 
-static void nDPIsrvd_write_flow_info_cb(struct nDPIsrvd_flow const * const flow, void * user_data)
+static void nDPIsrvd_write_flow_info_cb(struct nDPIsrvd_socket const * sock,
+                                        struct nDPIsrvd_instance const * instance,
+                                        struct nDPIsrvd_thread_data const * thread_data,
+                                        struct nDPIsrvd_flow const * flow,
+                                        void * user_data)
 {
+    (void)sock;
+    (void)instance;
     (void)user_data;
 
     fprintf(stderr,
-            "[Flow %4llu][ptr: "
+            "[Thread %2d][Flow %5llu][ptr: "
 #ifdef __LP64__
             "0x%016llx"
 #else
             "0x%08lx"
 #endif
-            "][last-seen: %13llu][idle-time: %13llu]\n",
+            "][last-seen: %13llu][idle-time: %7llu][time-until-timeout: %7llu]\n",
+            flow->thread_id,
             flow->id_as_ull,
 #ifdef __LP64__
             (unsigned long long int)flow,
@@ -40,14 +47,44 @@ static void nDPIsrvd_write_flow_info_cb(struct nDPIsrvd_flow const * const flow,
             (unsigned long int)flow,
 #endif
             flow->last_seen,
-            flow->idle_time);
+            flow->idle_time,
+            (flow->last_seen + flow->idle_time >= thread_data->most_recent_flow_time
+                 ? flow->last_seen + flow->idle_time - thread_data->most_recent_flow_time
+                 : 0));
 }
 
-static void nDPIsrvd_verify_flows_cb(struct nDPIsrvd_flow const * const flow, void * user_data)
+static void nDPIsrvd_verify_flows_cb(struct nDPIsrvd_thread_data const * const thread_data,
+                                     struct nDPIsrvd_flow const * const flow,
+                                     void * user_data)
 {
     (void)user_data;
 
-    fprintf(stderr, "Flow %llu verification failed\n", flow->id_as_ull);
+    if (thread_data != NULL)
+    {
+        if (flow->last_seen + flow->idle_time >= thread_data->most_recent_flow_time)
+        {
+            fprintf(stderr,
+                    "Thread %d / %d, Flow %llu verification failed\n",
+                    thread_data->thread_key,
+                    flow->thread_id,
+                    flow->id_as_ull);
+        }
+        else
+        {
+            fprintf(stderr,
+                    "Thread %d / %d, Flow %llu verification failed, diff: %llu\n",
+                    thread_data->thread_key,
+                    flow->thread_id,
+                    flow->id_as_ull,
+                    thread_data->most_recent_flow_time - flow->last_seen + flow->idle_time);
+        }
+    }
+    else
+    {
+        fprintf(stderr, "Thread [UNKNOWN], Flow %llu verification failed\n", flow->id_as_ull);
+    }
+
+    exit(1);
 }
 
 static void sighandler(int signum)
@@ -81,15 +118,24 @@ static void sighandler(int signum)
 
 static enum nDPIsrvd_callback_return simple_json_callback(struct nDPIsrvd_socket * const sock,
                                                           struct nDPIsrvd_instance * const instance,
+                                                          struct nDPIsrvd_thread_data * const thread_data,
                                                           struct nDPIsrvd_flow * const flow)
 {
     (void)sock;
-    (void)flow;
+    (void)thread_data;
+
+    if (flow == NULL)
+    {
+        return CALLBACK_OK;
+    }
 
     struct nDPIsrvd_json_token const * const flow_event_name = TOKEN_GET_SZ(sock, "flow_event_name");
     if (TOKEN_VALUE_EQUALS_SZ(flow_event_name, "new") != 0)
     {
-        printf("Instance %d, Flow %llu new\n", instance->alias_source_key, flow->id_as_ull);
+        printf("Instance 0x%x, Thread %d, Flow %llu new\n",
+               instance->alias_source_key,
+               flow->thread_id,
+               flow->id_as_ull);
     }
 
     return CALLBACK_OK;
@@ -97,20 +143,23 @@ static enum nDPIsrvd_callback_return simple_json_callback(struct nDPIsrvd_socket
 
 static void simple_flow_cleanup_callback(struct nDPIsrvd_socket * const sock,
                                          struct nDPIsrvd_instance * const instance,
+                                         struct nDPIsrvd_thread_data * const thread_data,
                                          struct nDPIsrvd_flow * const flow,
                                          enum nDPIsrvd_cleanup_reason reason)
 {
     (void)sock;
+    (void)thread_data;
 
     char const * const reason_str = nDPIsrvd_enum_to_string(reason);
-    printf("Instance %d, Flow %llu cleanup, reason: %s\n",
+    printf("Instance 0x%x, Thread %d, Flow %llu cleanup, reason: %s\n",
            instance->alias_source_key,
+           flow->thread_id,
            flow->id_as_ull,
            (reason_str != NULL ? reason_str : "UNKNOWN"));
 
     if (reason == CLEANUP_REASON_FLOW_TIMEOUT)
     {
-        printf("Did an nDPId instance die or was SIGKILL'ed?\n");
+        fprintf(stderr, "Flow timeout occurred, something really bad happened.\n");
         exit(1);
     }
 }
@@ -122,7 +171,7 @@ int main(int argc, char ** argv)
     signal(SIGTERM, sighandler);
     signal(SIGPIPE, sighandler);
 
-    sock = nDPIsrvd_socket_init(0, 0, simple_json_callback, simple_flow_cleanup_callback);
+    sock = nDPIsrvd_socket_init(0, 0, 0, 0, simple_json_callback, NULL, simple_flow_cleanup_callback);
     if (sock == NULL)
     {
         return 1;