counters: pass per thread stats to output api

As well as the global (merged) stats.

diff --git a/src/counters.c b/src/counters.c
index 827b1e3aaa2ab538e5dd2b8ba6a09ab6e0b1351c..acf2a1c9baaef29cd34c50538efba03ad02916fb 100644 (file)
--- a/src/counters.c
+++ b/src/counters.c
@@ -86,6 +86,8 @@ typedef struct StatsGlobalContext_ {
     /** list of thread stores: one per thread plus one global */
     StatsThreadStore *sts;
     SCMutex sts_lock;
+    int sts_cnt;
+
     HashTable *counters_id_hash;
 
     SCPerfPublicContext global_counter_ctx;
@@ -104,7 +106,7 @@ static int StatsThreadRegister(const char *thread_name, SCPerfPublicContext *);
 
 /** stats table is filled each interval and passed to the
  *  loggers. Initialized at first use. */
-static StatsTable stats_table = { NULL, 0, 0, {0 , 0}};
+static StatsTable stats_table = { NULL, NULL, 0, 0, 0, {0 , 0}};
 
 static uint16_t counters_global_id = 0;
 
@@ -608,7 +610,7 @@ static int SCPerfOutputCounterFileIface(ThreadVars *tv)
     void *td = stats_thread_data;
 
     if (stats_table.nstats == 0) {
-        StatsThreadRegister("Globals", &sc_perf_op_ctx->global_counter_ctx);
+        StatsThreadRegister("Global", &sc_perf_op_ctx->global_counter_ctx);
 
         uint32_t nstats = counters_global_id;
 
@@ -620,6 +622,15 @@ static int SCPerfOutputCounterFileIface(ThreadVars *tv)
             return -1;
         }
 
+        stats_table.ntstats = sc_perf_op_ctx->sts_cnt;
+        uint32_t array_size = stats_table.nstats * sizeof(StatsRecord);
+        stats_table.tstats = SCCalloc(stats_table.ntstats, array_size);
+        if (stats_table.tstats == NULL) {
+            stats_table.ntstats = 0;
+            SCLogError(SC_ERR_MEM_ALLOC, "could not alloc memory for stats");
+            return -1;
+        }
+
         stats_table.start_time = sc_start_time;
     }
 
@@ -633,6 +644,7 @@ static int SCPerfOutputCounterFileIface(ThreadVars *tv)
     memset(&merge_table, 0x00,
            counters_global_id * sizeof(struct CountersMergeTable));
 
+    int thread = sc_perf_op_ctx->sts_cnt - 1;
     StatsRecord *table = stats_table.stats;
 
     /* Loop through the thread counter stores. The global counters
@@ -640,47 +652,104 @@ static int SCPerfOutputCounterFileIface(ThreadVars *tv)
     sts = sc_perf_op_ctx->sts;
     SCLogDebug("sts %p", sts);
     while (sts != NULL) {
-        SCLogDebug("Thread %s (%s) ctx %p", sts->name,
+        BUG_ON(thread < 0);
+
+        SCLogDebug("Thread %d %s (%s) ctx %p", thread, sts->name,
                 sts->tm_name ? sts->tm_name : "none", sts->ctx);
 
+        /* temporay table for quickly storing the counters for this
+         * thread store, so that we can post process them outside
+         * of the thread store lock */
+        struct CountersMergeTable thread_table[counters_global_id];
+        memset(&thread_table, 0x00,
+                counters_global_id * sizeof(struct CountersMergeTable));
+
         SCMutexLock(&sts->ctx->m);
         pc = sts->ctx->head;
         while (pc != NULL) {
             SCLogDebug("Counter %s (%u:%u) value %"PRIu64,
                     pc->cname, pc->id, pc->gid, pc->value);
 
-            merge_table[pc->gid].type = pc->type;
+            thread_table[pc->gid].type = pc->type;
             switch (pc->type) {
-                case STATS_TYPE_MAXIMUM:
-                    if (pc->value > merge_table[pc->gid].value)
-                        merge_table[pc->gid].value = pc->value;
-                    table[pc->gid].tm_name = "Total";
-                    break;
                 case STATS_TYPE_FUNC:
                     if (pc->Func != NULL)
-                        merge_table[pc->gid].value = pc->Func();
-                    table[pc->gid].tm_name = "Global";
+                        thread_table[pc->gid].value = pc->Func();
                     break;
+                case STATS_TYPE_AVERAGE:
                 default:
-                    merge_table[pc->gid].value += pc->value;
-                    table[pc->gid].tm_name = "Total";
+                    thread_table[pc->gid].value = pc->value;
                     break;
             }
-            merge_table[pc->gid].updates += pc->updates;
-
+            thread_table[pc->gid].updates = pc->updates;
             table[pc->gid].name = pc->cname;
 
             pc = pc->next;
         }
         SCMutexUnlock(&sts->ctx->m);
+
+        /* update merge table */
+        uint16_t c;
+        for (c = 0; c < counters_global_id; c++) {
+            struct CountersMergeTable *e = &thread_table[c];
+            /* thread only sets type if it has a counter
+             * of this type. */
+            if (e->type == 0)
+                continue;
+
+            switch (e->type) {
+                case STATS_TYPE_MAXIMUM:
+                    if (e->value > merge_table[c].value)
+                        merge_table[c].value = e->value;
+                    break;
+                case STATS_TYPE_FUNC:
+                    merge_table[c].value = e->value;
+                    break;
+                case STATS_TYPE_AVERAGE:
+                default:
+                    merge_table[c].value += e->value;
+                    break;
+            }
+            merge_table[c].updates += e->updates;
+            merge_table[c].type = e->type;
+        }
+
+        /* update per thread stats table */
+        for (c = 0; c < counters_global_id; c++) {
+            struct CountersMergeTable *e = &thread_table[c];
+            /* thread only sets type if it has a counter
+             * of this type. */
+            if (e->type == 0)
+                continue;
+
+            uint32_t offset = (thread * stats_table.nstats) + c;
+            StatsRecord *r = &stats_table.tstats[offset];
+            r->name = table[c].name;
+            r->tm_name = sts->name;
+
+            switch (e->type) {
+                case STATS_TYPE_AVERAGE:
+                    if (e->value > 0 && e->updates > 0) {
+                        r->value = (uint64_t)(e->value / e->updates);
+                    }
+                    break;
+                default:
+                    r->value = e->value;
+                    break;
+            }
+        }
+
         sts = sts->next;
+        thread--;
     }
 
+    /* transfer 'merge table' to final stats table */
     uint16_t x;
     for (x = 0; x < counters_global_id; x++) {
         /* xfer previous value to pvalue and reset value */
         table[x].pvalue = table[x].value;
         table[x].value = 0;
+        table[x].tm_name = "Total";
 
         struct CountersMergeTable *m = &merge_table[x];
         switch (m->type) {
@@ -1102,6 +1171,7 @@ static int StatsThreadRegister(const char *thread_name, SCPerfPublicContext *pct
 
     temp->next = sc_perf_op_ctx->sts;
     sc_perf_op_ctx->sts = temp;
+    sc_perf_op_ctx->sts_cnt++;
     SCLogDebug("sc_perf_op_ctx->sts %p", sc_perf_op_ctx->sts);
 
     SCMutexUnlock(&sc_perf_op_ctx->sts_lock);

diff --git a/src/log-stats.c b/src/log-stats.c
index 1ae4246208e937c1b9a67a3ae6aa2fd4101dc2e5..cd35d57800e3b579c07ffcb79a9dc9c855aca555 100644 (file)
--- a/src/log-stats.c
+++ b/src/log-stats.c
@@ -98,10 +98,11 @@ int LogStatsLogger(ThreadVars *tv, void *thread_data, const StatsTable *st)
     MemBufferWriteString(aft->buffer, "----------------------------------------------"
             "---------------------\n");
 
+    /* global stats */
     uint32_t u = 0;
     for (u = 0; u < st->nstats; u++) {
         if (st->stats[u].name == NULL)
-            break;
+            continue;
 
         char line[1024];
         size_t len = snprintf(line, sizeof(line), "%-25s | %-25s | %-" PRIu64 "\n",
@@ -116,6 +117,33 @@ int LogStatsLogger(ThreadVars *tv, void *thread_data, const StatsTable *st)
         MemBufferWriteString(aft->buffer, "%s", line);
     }
 
+    /* per thread stats */
+    if (st->tstats != NULL) {
+        /* for each thread (store) */
+        uint32_t x;
+        for (x = 0; x < st->ntstats; x++) {
+            uint32_t offset = x * st->nstats;
+
+            /* for each counter */
+            for (u = offset; u < (offset + st->nstats); u++) {
+                if (st->tstats[u].name == NULL)
+                    continue;
+
+                char line[1024];
+                size_t len = snprintf(line, sizeof(line), "%-25s | %-25s | %-" PRIu64 "\n",
+                        st->tstats[u].name, st->tstats[u].tm_name, st->tstats[u].value);
+
+                /* since we can have many threads, the buffer might not be big enough.
+                 * Expand if necessary. */
+                if (MEMBUFFER_OFFSET(aft->buffer) + len > MEMBUFFER_SIZE(aft->buffer)) {
+                    MemBufferExpand(&aft->buffer, OUTPUT_BUFFER_SIZE);
+                }
+
+                MemBufferWriteString(aft->buffer, "%s", line);
+            }
+        }
+    }
+
     SCMutexLock(&aft->statslog_ctx->file_ctx->fp_mutex);
     aft->statslog_ctx->file_ctx->Write((const char *)MEMBUFFER_BUFFER(aft->buffer),
         MEMBUFFER_OFFSET(aft->buffer), aft->statslog_ctx->file_ctx);

diff --git a/src/output-stats.h b/src/output-stats.h
index 6ca89f931bedabe83297843e4d5d9a55d4dd79bf..75017ca20c4e7f78ece739c1c40929c93c37dac2 100644 (file)
--- a/src/output-stats.h
+++ b/src/output-stats.h
@@ -34,8 +34,10 @@ typedef struct StatsRecord_ {
 } StatsRecord;
 
 typedef struct StatsTable_ {
-    StatsRecord *stats;
-    uint32_t nstats;
+    StatsRecord *stats;     /**< array of global stats, indexed by counters gid */
+    StatsRecord *tstats;    /**< array of arrays with per thread stats */
+    uint32_t nstats;        /**< size in records of 'stats' */
+    uint32_t ntstats;       /**< number of threads for which tstats stores stats */
     time_t start_time;
     struct timeval ts;
 } StatsTable;