Makes event array compression multi-threaded

1 files changed, 95 insertions, 48 deletions

diff --git a/core/salis.c b/core/salis.c
index def8243..6a42bb3 100644
--- a/core/salis.c
+++ b/core/salis.c
@@ -14,6 +14,14 @@
 #define MALL_FLAG 0x80
 #define UINT64_HALF 0x8000000000000000ul
 
+#if defined(DATA_PUSH_PATH)
+#define EVENT_ARRAYS \
+    EVENT_ARRAY(0, aev) /* allocation events array */ \
+    EVENT_ARRAY(1, eev) /* executions events array */ \
+    EVENT_ARRAY(2, bev) /* birth events array */
+#define EVENT_ARRAYS_COUNT 3
+#endif
+
 #if defined(COMPRESS) || defined(DATA_PUSH_PATH)
 #include "compress.c"
 #endif
@@ -37,7 +45,7 @@ struct Core {
     uint64_t psli;
 
     thrd_t thrd;
-    uint64_t thrd_idx;
+    uint64_t thrd_steps;
 
     uint64_t ivpt;
     uint64_t *ivav;
@@ -49,9 +57,9 @@ struct Core {
     uint64_t eliv; // executions within not-owned live code counter (parasites)
     uint64_t edea; // executions within dead code counter
 
-    uint64_t aeva[MVEC_SIZE]; // allocation events array
-    uint64_t eeva[MVEC_SIZE]; // execution events array
-    uint64_t beva[MVEC_SIZE]; // birth events array
+#define EVENT_ARRAY(_, ev) uint64_t ev##a[MVEC_SIZE];
+    EVENT_ARRAYS
+#undef EVENT_ARRAY
 
 #define CORE_DATA_FIELD(type, name, suff) type name suff;
     CORE_DATA_FIELDS
@@ -79,6 +87,8 @@ char g_asav_pbuf[AUTOSAVE_NAME_LEN];
 
 #if defined(DATA_PUSH_PATH)
 sqlite3 *g_sim_data;
+thrd_t g_eva_thrds[CORES][EVENT_ARRAYS_COUNT];
+struct DeflateParams g_eva_deflate_params[CORES][EVENT_ARRAYS_COUNT];
 #endif
 
 // Each UI must install these logger functions before salis_init() gets invoked
@@ -437,9 +447,10 @@ void core_save(FILE *f, const struct Core *core) {
     fwrite(core->pvec, sizeof(struct Proc), core->pcap, f);
     fwrite(core->mvec, sizeof(uint8_t), MVEC_SIZE, f);
 #if defined(DATA_PUSH_PATH)
-    fwrite(core->aeva, sizeof(uint64_t), MVEC_SIZE, f);
-    fwrite(core->eeva, sizeof(uint64_t), MVEC_SIZE, f);
-    fwrite(core->beva, sizeof(uint64_t), MVEC_SIZE, f);
+#define EVENT_ARRAY(_, ev) \
+    fwrite(core->ev##a, sizeof(uint64_t), MVEC_SIZE, f);
+    EVENT_ARRAYS
+#undef EVENT_ARRAY
 #endif
 
     arch_core_save(f, core);
@@ -534,9 +545,10 @@ void core_load(FILE *f, struct Core *core) {
     fread(core->pvec, sizeof(struct Proc), core->pcap, f);
     fread(core->mvec, sizeof(uint8_t), MVEC_SIZE, f);
 #if defined(DATA_PUSH_PATH)
-    fread(core->aeva, sizeof(uint64_t), MVEC_SIZE, f);
-    fread(core->eeva, sizeof(uint64_t), MVEC_SIZE, f);
-    fread(core->beva, sizeof(uint64_t), MVEC_SIZE, f);
+#define EVENT_ARRAY(_, ev) \
+    fread(core->ev##a, sizeof(uint64_t), MVEC_SIZE, f);
+    EVENT_ARRAYS
+#undef EVENT_ARRAY
 #endif
 
     arch_core_load(f, core);
@@ -663,17 +675,22 @@ void salis_save(const char *path) {
     char *out = malloc(size);
     assert(out);
 
-    z_stream strm = { 0 };
-    salis_deflate(&strm, size, (Bytef *)in, (Bytef *)out);
+    struct DeflateParams params = {
+        .size = size,
+        .in = (Bytef *)in,
+        .out = (Bytef *)out,
+    };
+
+    salis_deflate(&params);
 
     FILE *fx = fopen(path, "wb");
     assert(fx);
 
     fwrite(&size, sizeof(size_t), 1, fx);
-    fwrite(out, sizeof(char), strm.total_out, fx);
+    fwrite(out, sizeof(char), params.strm.total_out, fx);
     fclose(fx);
 
-    salis_deflate_end(&strm);
+    salis_deflate_end(&params);
 
     free(in);
     free(out);
@@ -781,12 +798,19 @@ void salis_push_data_header(void) {
     );
 
     // Memory events
-    char *eprefs[] = { "aev", "eev", "bev" };
-    int eprefs_cnt = sizeof(eprefs) / sizeof(eprefs[0]);
-
     for (int i = 0; i < CORES; ++i) {
-        for (int j = 0; j < eprefs_cnt; ++j) {
-            g_info("Creating '%s_%d' table in SQLite database", eprefs[j], i);
+        for (int j = 0; j < EVENT_ARRAYS_COUNT; ++j) {
+            char *pref = NULL;
+
+            switch (j) {
+#define EVENT_ARRAY(idx, ev) \
+            case idx: pref = #ev; break;
+            EVENT_ARRAYS
+#undef EVENT_ARRAY
+            default: assert(false);
+            }
+
+            g_info("Creating '%s_%d' table in SQLite database", pref, i);
             salis_exec_sql(
                 0, NULL, NULL,
                 "create table %s_%d ("
@@ -797,7 +821,7 @@ void salis_push_data_header(void) {
                 "evts blob not null ,"
                 "step int not null"
                 ");",
-                eprefs[j], i
+                pref, i
             );
         }
     }
@@ -868,34 +892,50 @@ void salis_push_data_line(void) {
     );
 
     // Insert generic memory events
-    char *eprefs[] = { "aev", "eev", "bev" };
-    int eprefs_cnt = sizeof(eprefs) / sizeof(eprefs[0]);
+    // Compression (deflation) is CPU intensive so it is done in parallel
+    memset(&g_eva_deflate_params, 0, sizeof(struct DeflateParams) * CORES * EVENT_ARRAYS_COUNT);
 
     for (int i = 0; i < CORES; ++i) {
-        for (int j = 0; j < eprefs_cnt; ++j) {
+        for (int j = 0; j < EVENT_ARRAYS_COUNT; ++j) {
             uint64_t *in = NULL;
 
-            if (!strcmp("aev", eprefs[j])) {
-                in = g_cores[i].aeva;
-            } else if (!strcmp("eev", eprefs[j])) {
-                in = g_cores[i].eeva;
-            } else if (!strcmp("bev", eprefs[j])) {
-                in = g_cores[i].beva;
+            switch (j) {
+#define EVENT_ARRAY(idx, ev) \
+            case idx: in = g_cores[i].ev##a; break;
+            EVENT_ARRAYS
+#undef EVENT_ARRAY
+            default: assert(false);
             }
 
             // Compress event data
-            size_t size = sizeof(uint64_t) * MVEC_SIZE;
-            char *out = malloc(size);
-            assert(out);
+            struct DeflateParams *params = &g_eva_deflate_params[i][j];
+            params->size = EVA_SIZE,
+            params->in = (Bytef *)in,
+            params->out = (Bytef *)malloc(EVA_SIZE),
+            thrd_create(&g_eva_thrds[i][j], (thrd_start_t)salis_deflate, params);
+        }
+    }
 
-            z_stream strm = { 0 };
-            salis_deflate(&strm, size, (Bytef *)in, (Bytef *)out);
+    for (int i = 0; i < CORES; ++i) {
+        for (int j = 0; j < EVENT_ARRAYS_COUNT; ++j) {
+            char *pref = NULL;
+
+            switch (j) {
+#define EVENT_ARRAY(idx, ev) \
+            case idx: pref = #ev; break;
+            EVENT_ARRAYS
+#undef EVENT_ARRAY
+            default: assert(false);
+            }
+
+            thrd_join(g_eva_thrds[i][j], NULL);
 
             // Insert blob
-            const void *blob = out;
-            int blob_size = strm.total_out;
+            struct DeflateParams *params = &g_eva_deflate_params[i][j];
+            const void *blob = params->out;
+            int blob_size = params->strm.total_out;
 
-            g_info("Pushing row to '%s_%d' table in SQLite database", eprefs[j], i);
+            g_info("Pushing row to '%s_%d' table in SQLite database", pref, i);
             salis_exec_sql(
                 1, &blob, &blob_size,
                 "insert into %s_%d ("
@@ -909,15 +949,15 @@ void salis_push_data_line(void) {
 #undef FOR_CORE
                 "%ld, ?, %ld"
                 ");",
-                eprefs[j], i,
+                pref, i,
 #define FOR_CORE(i) g_cores[i].cycl,
                 FOR_CORES
 #undef FOR_CORE
                 blob_size, g_steps
             );
 
-            salis_deflate_end(&strm);
-            free(out);
+            salis_deflate_end(params);
+            free(params->out);
         }
     }
 
@@ -930,9 +970,10 @@ void salis_push_data_line(void) {
         core->eliv = 0;
         core->edea = 0;
 
-        memset(core->aeva, 0, sizeof(uint64_t) * MVEC_SIZE);
-        memset(core->eeva, 0, sizeof(uint64_t) * MVEC_SIZE);
-        memset(core->beva, 0, sizeof(uint64_t) * MVEC_SIZE);
+#define EVENT_ARRAY(_, ev) \
+        memset(core->ev##a, 0, EVA_SIZE);
+        EVENT_ARRAYS
+#undef EVENT_ARRAY
     }
 
     // Push arch-specific data
@@ -996,9 +1037,15 @@ void salis_load(void) {
     char *out = malloc(size);
     assert(out);
 
-    z_stream strm = { 0 };
-    salis_inflate(&strm, x_size, size, (Bytef *)in, (Bytef *)out);
-    salis_inflate_end(&strm);
+    struct InflateParams params = {
+        .avail_in = x_size,
+        .size = size,
+        .in = (Bytef *)in,
+        .out = (Bytef *)out,
+    };
+
+    salis_inflate(&params);
+    salis_inflate_end(&params);
 
     FILE *f = fmemopen(out, size, "rb");
 #else
@@ -1033,7 +1080,7 @@ void salis_load(void) {
 int salis_thread(struct Core *core) {
     assert(core);
 
-    for (uint64_t i = 0; i < core->thrd_idx; ++i) {
+    for (uint64_t i = 0; i < core->thrd_steps; ++i) {
         core_step(core);
     }
 
@@ -1042,7 +1089,7 @@ int salis_thread(struct Core *core) {
 
 void salis_run_thread(uint64_t ns) {
     for (int i = 0; i < CORES; ++i) {
-        g_cores[i].thrd_idx = ns;
+        g_cores[i].thrd_steps = ns;
 
         thrd_create(
             &g_cores[i].thrd,