source: lib/combiner_sorted.c @ 62b3c4e

4.0.1-hotfixescachetimestampsdevelopdpdk-ndagetsilivelibtrace4ndag_formatpfringrc-4.0.1rc-4.0.2rc-4.0.3rc-4.0.4ringdecrementfixringperformanceringtimestampfixes
Last change on this file since 62b3c4e was 62b3c4e, checked in by Richard Sanger <rsangerarj@…>, 6 years ago

Update the reporter method to match with the per_pkt method.

We now treat a result as a type of message.

  • Property mode set to 100644
File size: 2.5 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        if (libtrace_result_get_key((libtrace_result_t *) p1) < libtrace_result_get_key((libtrace_result_t *) p2))
31                return -1;
32        if (libtrace_result_get_key((libtrace_result_t *) p1) == libtrace_result_get_key((libtrace_result_t *) p2))
33                return 0;
34        else
35                return 1;
36}
37
38static void pause(libtrace_t *trace, libtrace_combine_t *c) {
39        libtrace_vector_t *queues = c->queues;
40        int i;
41        for (i = 0; i < libtrace_get_perpkt_count(trace); ++i) {
42                libtrace_vector_apply_function(&queues[i], (vector_data_fn) libtrace_make_result_safe);
43        }
44}
45
46static void read_final(libtrace_t *trace, libtrace_combine_t *c) {
47        libtrace_vector_t *queues = c->queues;
48        int i;
49        size_t a;
50        // Combine all results into queue 1
51        for (i = 1; i < libtrace_get_perpkt_count(trace); ++i)
52        {
53                libtrace_vector_append(&queues[0],&queues[i]);
54        }
55        // Sort them
56        libtrace_vector_qsort(&queues[0], compare_result);
57
58        for (a = 0; a < libtrace_vector_get_size(&queues[0]); ++a) {
59                libtrace_result_t r;
60                libtrace_generic_t gt = {.res = &r};
61                ASSERT_RET (libtrace_vector_get(&queues[0], a, (void *) &r), == 1);
62                trace->reporter(trace, MESSAGE_RESULT, gt, &trace->reporter_thread);
63        }
64        libtrace_vector_empty(&queues[0]);
65}
66
67static void destroy(libtrace_t *trace, libtrace_combine_t *c) {
68        int i;
69        libtrace_vector_t *queues = c->queues;
70
71        for (i = 0; i < libtrace_get_perpkt_count(trace); i++) {
72                assert(libtrace_vector_get_size(&queues[i]) == 0);
73                libtrace_vector_destroy(&queues[i]);
74        }
75        free(queues);
76        queues = NULL;
77}
78
79DLLEXPORT const libtrace_combine_t combiner_sorted = {
80    init_combiner,      /* initialise */
81        destroy,                /* destroy */
82        publish,                /* publish */
83    read,                       /* read */
84    read_final,                 /* read_final */
85    pause,                      /* pause */
86    NULL,                       /* queues */
87    {0}                         /* opts */
88};
Note: See TracBrowser for help on using the repository browser.