source: examples/skeleton/parallel.c @ c42a2d6

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

Add mostly complete skeleton for parallel libtrace

This should act as a good starting point for anyone developing
parallel libtrace programs.
  • Property mode set to 100644
File size: 9.6 KB
Line 
1/*
2 * This file is part of libtrace
3 *
4 * Copyright (c) 2007-2015 The University of Waikato, Hamilton, New Zealand.
5 * Authors: Daniel Lawson
6 *          Perry Lorier
7 *          Shane Alcock
8 *          Richard Sanger
9 *
10 * All rights reserved.
11 *
12 * This code has been developed by the University of Waikato WAND
13 * research group. For further information please see http://www.wand.net.nz/
14 *
15 * libtrace is free software; you can redistribute it and/or modify
16 * it under the terms of the GNU General Public License as published by
17 * the Free Software Foundation; either version 2 of the License, or
18 * (at your option) any later version.
19 *
20 * libtrace is distributed in the hope that it will be useful,
21 * but WITHOUT ANY WARRANTY; without even the implied warranty of
22 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
23 * GNU General Public License for more details.
24 *
25 * You should have received a copy of the GNU General Public License
26 * along with libtrace; if not, write to the Free Software
27 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
28 *
29 * $Id: test-rtclient.c,v 1.2 2006/02/27 03:41:12 perry Exp $
30 *
31 */
32#ifndef WIN32
33#  include <sys/time.h>
34#  include <netinet/in.h>
35#  include <netinet/in_systm.h>
36#  include <netinet/tcp.h>
37#  include <netinet/ip.h>
38#  include <netinet/ip_icmp.h>
39#  include <arpa/inet.h>
40#  include <sys/socket.h>
41#endif
42#include <stdio.h>
43#include <stdlib.h>
44#include <assert.h>
45#include <string.h>
46#include <sys/types.h>
47#include <time.h>
48#include <string.h>
49#include <signal.h>
50#include <unistd.h>
51
52#include "libtrace_parallel.h"
53
54volatile int done = 0;
55libtrace_t *trace = NULL;
56
57static void cleanup_signal(int sig) {
58        (void)sig;      /* avoid warnings about unused parameter */
59        done = 1;
60        if (trace)
61                trace_pstop(trace);
62}
63
64
65/* Thread local storage for the reporting thread */
66struct rstorage {
67        int replaceme;
68};
69
70/* Thread local storage for each processing thread */
71struct pstorage {
72        uint32_t replaceme;
73};
74
75static void *report_start(libtrace_t *trace,
76                libtrace_thread_t *t,
77                void *global) {
78
79        /* Create any local storage required by the reporter thread and
80         * return it. */
81        struct rstorage *rs = (struct rstorage *)malloc(sizeof(struct rstorage));
82        rs->replaceme = 0;
83
84        assert(trace);
85        assert(t);
86        assert(global);
87
88        return rs;
89}
90
91static void report_cb(libtrace_t *trace,
92                libtrace_thread_t *sender,
93                void *global, void *tls, libtrace_result_t *res) {
94
95        struct rstorage *rs = (struct rstorage *)tls;
96        assert(trace);
97        assert(sender);
98        assert(global);
99        assert(rs);
100        assert(res);
101
102        /* Process the result */
103
104        /* Make sure we free any packets included in the result */
105        if (res->type == RESULT_PACKET)
106                trace_free_packet(trace, res->value.pkt);
107}
108
109static void report_end(libtrace_t *trace, libtrace_thread_t *t,
110                void *global, void *tls) {
111
112        /* Free the local storage and print any final results */
113        struct rstorage *rs = (struct rstorage *)tls;
114        free(rs);
115        assert(trace);
116        assert(t);
117        assert(global);
118}
119
120static libtrace_packet_t *per_packet(libtrace_t *trace,
121                libtrace_thread_t *t,
122                void *global, void *tls, libtrace_packet_t *packet) {
123        struct pstorage *ps = (struct pstorage *)tls;
124        assert(trace);
125        assert(t);
126        assert(global);
127        assert(ps);
128        /* Do something with the packet */
129
130        /* In this example, we are just forwarding the packet to the reporter */
131        trace_publish_result(trace, t, 0, (libtrace_generic_t){.pkt = packet}, RESULT_PACKET);
132        return NULL;
133}
134
135static void *start_processing(libtrace_t *trace, libtrace_thread_t *t UNUSED,
136                void *global) {
137
138        /* Create any local storage required by the reporter thread and
139         * return it. */
140        struct pstorage *ps = (struct pstorage *)malloc(sizeof(struct pstorage));
141        ps->replaceme = 0;
142
143        assert(trace);
144        assert(t);
145        assert(global);
146        return ps;
147}
148
149static void stop_processing(libtrace_t *trace, libtrace_thread_t *t,
150                void *global, void *tls) {
151
152        struct pstorage *ps = (struct pstorage *)tls;
153
154        /* May want to do a final publish here... */
155
156        assert(trace);
157        assert(t);
158        assert(global);
159        free(ps);
160}
161
162static void process_tick(libtrace_t *trace, libtrace_thread_t *t,
163                void *global, void *tls, uint64_t tick) {
164
165        struct pstorage *ps = (struct pstorage *)tls;
166
167        /* Publish or ignore the tick, as appropriate */
168        assert(trace);
169        assert(t);
170        assert(global);
171        assert(ps);
172
173        if (tick) return;
174
175}
176
177static void pause_processing(libtrace_t *trace,
178                libtrace_thread_t *t,
179                void *global, void *tls) {
180
181        struct pstorage *ps = (struct pstorage *)tls;
182        assert(trace);
183        assert(t);
184        assert(global);
185        assert(ps);
186
187}
188
189static void resume_processing(libtrace_t *trace,
190                libtrace_thread_t *t,
191                void *global, void *tls) {
192
193        struct pstorage *ps = (struct pstorage *)tls;
194        assert(trace);
195        assert(t);
196        assert(global);
197        assert(ps);
198
199}
200
201static void custom_msg(libtrace_t *trace, libtrace_thread_t *t, void *global,
202                void *tls, int mesg, libtrace_generic_t data) {
203
204        struct pstorage *ps = (struct pstorage *)tls;
205        assert(trace);
206        assert(t);
207        assert(global);
208        assert(ps);
209
210        assert(mesg >= MESSAGE_USER);
211        assert(sizeof(data) == 8);
212}
213
214static void usage(char *prog) {
215        fprintf(stderr, "Usage for %s\n\n", prog);
216        fprintf(stderr, "%s [options] inputURI [inputURI ...]\n\n", prog);
217        fprintf(stderr, "Options:\n");
218        fprintf(stderr, "\t-t threads Set the number of processing threads\n");
219        fprintf(stderr, "\t-f expr    Discard all packets that do not match the BPF expression\n");
220
221        exit(0);
222}
223
224int main(int argc, char *argv[]) {
225        libtrace_callback_set_t *processing = NULL;
226        libtrace_callback_set_t *reporter = NULL;
227        libtrace_filter_t *filter = NULL;
228        char *filterstring = NULL;
229        int i, opt, retcode=0;
230        struct sigaction sigact;
231        int threads = 4;
232
233        /* TODO replace this with whatever global data your threads are
234         * likely to need. */
235        uint32_t global = 0;
236
237        if (argc<2) {
238                usage(argv[0]);
239        }
240
241        while ((opt = getopt(argc, argv, "f:t:")) != EOF) {
242                switch (opt) {
243                        case 'f':
244                                filterstring = optarg;
245                                break;
246                        case 't':
247                                threads = atoi(optarg);
248                                break;
249                        default:
250                                usage(argv[0]);
251                }
252        }
253
254        if (optind + 1 > argc) {
255                usage(argv[0]);
256                return 1;
257        }
258
259        if (filterstring) {
260                filter = trace_create_filter(filterstring);
261        }
262
263        sigact.sa_handler = cleanup_signal;
264        sigemptyset(&sigact.sa_mask);
265        sigact.sa_flags = SA_RESTART;
266
267        sigaction(SIGINT, &sigact, NULL);
268        sigaction(SIGTERM, &sigact, NULL);
269
270        processing = trace_create_callback_set();
271        trace_set_starting_cb(processing, start_processing);
272        trace_set_stopping_cb(processing, stop_processing);
273        trace_set_packet_cb(processing, per_packet);
274        trace_set_pausing_cb(processing, pause_processing);
275        trace_set_resuming_cb(processing, resume_processing);
276        trace_set_tick_count_cb(processing, process_tick);
277        trace_set_tick_interval_cb(processing, process_tick);
278        trace_set_user_message_cb(processing, custom_msg);
279
280        reporter = trace_create_callback_set();
281        trace_set_starting_cb(reporter, report_start);
282        trace_set_stopping_cb(reporter, report_end);
283        trace_set_result_cb(reporter, report_cb);
284
285        for (i = optind; i < argc; i++) {
286
287                trace = trace_create(argv[i]);
288
289                if (trace_is_err(trace)) {
290                        trace_perror(trace, "Opening trace file");
291                        retcode = -1;
292                        break;
293                }
294
295                if (filter && trace_config(trace, TRACE_OPTION_FILTER, filter) == -1) {
296                        trace_perror(trace, "trace_config(filter)");
297                        retcode = -1;
298                        break;
299                }
300
301                trace_set_perpkt_threads(trace, threads);
302                trace_set_combiner(trace, &combiner_ordered,
303                                (libtrace_generic_t) {0});
304                trace_set_hasher(trace, HASHER_BIDIRECTIONAL, NULL, NULL);
305
306                if (trace_pstart(trace, &global, processing, reporter)) {
307                        trace_perror(trace, "Starting trace");
308                        break;
309                }
310
311                /* Wait for all threads to stop */
312                trace_join(trace);
313
314                if (trace_is_err(trace)) {
315                        trace_perror(trace, "Processing packets");
316                        retcode = -1;
317                        break;
318                }
319
320                if (done)
321                        break;
322        }
323
324        if (filter)
325                trace_destroy_filter(filter);
326        trace_destroy(trace);
327        trace_destroy_callback_set(processing);
328        trace_destroy_callback_set(reporter);
329        return retcode;
330}
Note: See TracBrowser for help on using the repository browser.