source: lib/combiner_sorted.c @ 3dd5acc

4.0.1-hotfixescachetimestampsdevelopdpdk-ndagetsilivelibtrace4ndag_formatpfringrc-4.0.1rc-4.0.2rc-4.0.3rc-4.0.4ringdecrementfixringperformanceringtimestampfixes
Last change on this file since 3dd5acc was 3dd5acc, checked in by Shane Alcock <salcock@…>, 6 years ago

Fix problems with combiners and ticks.

  • Each tick should only be passed to the reporter once at most. In the case of a sorted combiner, all ticks are discarded.
  • In the case of an ordered combiner, ticks that do not use the same ordering as packets (e.g. ts ticks vs count ticks) are not used for any ordering comparisons. A read operation will read from each queue until it encounters a tick in that queue -- if it is a new tick, that will be reported, otherwise the tick is discarded and the queue is marked as not "live".

In terms of testing: works OK when mixing timestamp ticks with non-parallel
input. Still needs to be tested the other way around.

  • Property mode set to 100644
File size: 2.8 KB
Line 
1#include "libtrace.h"
2#include "libtrace_int.h"
3#include "data-struct/vector.h"
4#include <assert.h>
5#include <stdlib.h>
6
7static int init_combiner(libtrace_t *t, libtrace_combine_t *c) {
8        int i = 0;
9        assert(libtrace_get_perpkt_count(t) > 0);
10        libtrace_vector_t *queues;
11        c->queues = calloc(sizeof(libtrace_vector_t), libtrace_get_perpkt_count(t));
12        queues = c->queues;
13        for (i = 0; i < libtrace_get_perpkt_count(t); ++i) {
14                libtrace_vector_init(&queues[i], sizeof(libtrace_result_t));
15        }
16        return 0;
17}
18
19static void publish(libtrace_t *trace UNUSED, int t_id, libtrace_combine_t *c, libtrace_result_t *res) {
20        libtrace_vector_t *vec = &((libtrace_vector_t*)c->queues)[t_id];
21        libtrace_vector_push_back(vec, res);
22}
23
24static void read(libtrace_t *trace UNUSED, libtrace_combine_t *c UNUSED){
25        return;
26}
27
28static int compare_result(const void* p1, const void* p2)
29{
30        const libtrace_result_t * r1 = p1;
31        const libtrace_result_t * r2 = p2;
32        if (r1->key < r2->key)
33                return -1;
34        if (r1->key == r2->key)
35                return 0;
36        else
37                return 1;
38}
39
40static void pause(libtrace_t *trace, libtrace_combine_t *c) {
41        libtrace_vector_t *queues = c->queues;
42        int i;
43        for (i = 0; i < libtrace_get_perpkt_count(trace); ++i) {
44                libtrace_vector_apply_function(&queues[i], (vector_data_fn) libtrace_make_result_safe);
45        }
46}
47
48static void read_final(libtrace_t *trace, libtrace_combine_t *c) {
49        libtrace_vector_t *queues = c->queues;
50        int i;
51        size_t a;
52        // Combine all results into queue 1
53        for (i = 1; i < libtrace_get_perpkt_count(trace); ++i)
54        {
55                libtrace_vector_append(&queues[0],&queues[i]);
56        }
57        // Sort them
58        libtrace_vector_qsort(&queues[0], compare_result);
59
60        for (a = 0; a < libtrace_vector_get_size(&queues[0]); ++a) {
61                libtrace_result_t r;
62                libtrace_generic_t gt = {.res = &r};
63                ASSERT_RET (libtrace_vector_get(&queues[0], a, (void *) &r), == 1);
64                if (r.type == RESULT_TICK_INTERVAL ||
65                                r.type == RESULT_TICK_COUNT) {
66                        /* Ticks are essentially useless for this combiner? */
67                        continue;
68                }
69                trace->reporter(trace, MESSAGE_RESULT, gt, &trace->reporter_thread);
70        }
71        libtrace_vector_empty(&queues[0]);
72}
73
74static void destroy(libtrace_t *trace, libtrace_combine_t *c) {
75        int i;
76        libtrace_vector_t *queues = c->queues;
77
78        for (i = 0; i < libtrace_get_perpkt_count(trace); i++) {
79                assert(libtrace_vector_get_size(&queues[i]) == 0);
80                libtrace_vector_destroy(&queues[i]);
81        }
82        free(queues);
83        queues = NULL;
84}
85
86DLLEXPORT const libtrace_combine_t combiner_sorted = {
87    init_combiner,      /* initialise */
88        destroy,                /* destroy */
89        publish,                /* publish */
90    read,                       /* read */
91    read_final,                 /* read_final */
92    pause,                      /* pause */
93    NULL,                       /* queues */
94    0,                          /* last_count_tick */
95    0,                          /* last_ts_tick */
96    {0}                         /* opts */
97};
Note: See TracBrowser for help on using the repository browser.