Changeset f01c479 for lib

Timestamp:

03/25/15 14:28:31 (6 years ago)

Author:

Richard Sanger <rsangerarj@…>

Branches:

4.0.1-hotfixes, cachetimestamps, develop, dpdk-ndag, etsilive, libtrace4, master, ndag_format, pfring, rc-4.0.1, rc-4.0.2, rc-4.0.3, rc-4.0.4, ringdecrementfix, ringperformance, ringtimestampfixes

Children:

5e43b8b

Parents:

a330384

Message:

Move reporter threashold check to after adding the next item, so it makes more sense
in the case the threashold is set to 1.

Add logic to the order combiner allowing multiple results with the same key
to be sent even if all queues are not full.

Location:

lib

Files:

• combiner_ordered.c (modified) (5 diffs)
• combiner_unordered.c (modified) (1 diff)

lib/combiner_ordered.c

@@ -21 +21 @@
 static void publish(libtrace_t *trace, int t_id, libtrace_combine_t *c, libtrace_result_t *res) {
         libtrace_queue_t *queue = &((libtrace_queue_t*)c->queues)[t_id];
+        //while (libtrace_deque_get_size(&t->deque) >= 1000)
+        //      sched_yield();
+        libtrace_deque_push_back(queue, res); // Automatically locking for us :)
+
         if (libtrace_deque_get_size(queue) >= trace->config.reporter_thold) {
                 trace_post_reporter(trace);
         }
-        //while (libtrace_deque_get_size(&t->deque) >= 1000)
-        //      sched_yield();
-        libtrace_deque_push_back(queue, res); // Automatically locking for us :)
 }
 
@@ -35 +36 @@
         uint64_t key[libtrace_get_perpkt_count(trace)]; // Cached keys
         uint64_t min_key = UINT64_MAX;
+        uint64_t prev_min = 0;
         int min_queue = -1;
 
@@ -55 +57 @@
         }
 
-        /* Now remove the smallest and loop - special case if all threads have joined we always flush what's left */
-        while ((live_count == libtrace_get_perpkt_count(trace)) || (live_count && final)) {
-                // || (live_count && ((flags & REDUCE_SEQUENTIAL && min_key == trace->expected_key)))
+        /* Now remove the smallest and loop - special case if all threads have
+         * joined we always flush what's left. Or the next smallest is the same
+         * value or less than the previous */
+        while ((live_count == libtrace_get_perpkt_count(trace)) || (live_count && final)
+               || (live_count && prev_min >= min_key)) {
                 /* Get the minimum queue and then do stuff */
                 libtrace_result_t r;
@@ -64 +68 @@
                 ASSERT_RET (libtrace_deque_pop_front(&queues[min_queue], (void *) &r), == 1);
                 trace->reporter(trace, MESSAGE_RESULT, gt, &trace->reporter_thread);
-
-                // We expect the key we read +1 now , todo put expected in our storage area
-                //trace->expected_key = key[min_queue] + 1;
 
                 // Now update the one we just removed
@@ -89 +90 @@
                         live[min_queue] = false;
                         live_count--;
+                        prev_min = min_key;
                         min_key = UINT64_MAX; // Update our minimum
                         // Check all find the smallest again - all are alive
                         for (i = 0; i < libtrace_get_perpkt_count(trace); ++i) {
-                                // Still not 100% TODO (what if order is wrong or not increasing)
                                 if (live[i] && min_key >= key[i]) {
                                         min_key = key[i];

lib/combiner_unordered.c

@@ -19 +19 @@
 static void publish(libtrace_t *trace, int t_id, libtrace_combine_t *c, libtrace_result_t *res) {
         libtrace_queue_t *queue = &((libtrace_queue_t*)c->queues)[t_id];
+        libtrace_deque_push_back(queue, res); // Automatically locking for us :)
+
         if (libtrace_deque_get_size(queue) >= trace->config.reporter_thold) {
                 trace_post_reporter(trace);
         }
-        libtrace_deque_push_back(queue, res); // Automatically locking for us :)
 }