Changeset 76291d1 for tools


Ignore:
Timestamp:
03/31/15 16:02:53 (5 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:
4007dbb
Parents:
58bfabf
Message:

Add a quick mockup of shanes interface using functions for each message

I.e. set a function for MESSAGE_PACKET etc. which only receives packets.

Ported tracestats_parallel to use this interface.

Note it is not yet complete, however should give a feel of how it will work.
And provide a good work around to the thread issues on older machines.

File:
1 edited

Legend:

Unmodified
Added
Removed
  • tools/tracestats/tracestats_parallel.c

    rdfbdda7a r76291d1  
    8787
    8888
    89 static void* per_packet(libtrace_t *trace, libtrace_thread_t *t,
    90                         int mesg, libtrace_generic_t data,
     89static void* per_packet(libtrace_t *trace UNUSED, libtrace_thread_t *t UNUSED,
     90                        int mesg UNUSED, libtrace_generic_t data UNUSED,
    9191                        libtrace_thread_t *sender UNUSED)
    9292{
    93         /* Using first entry as total and those after for filter counts */
    94         static __thread statistics_t * results = NULL;
    95         int i, wlen;
    96         libtrace_stat_t *stats;
    97         libtrace_generic_t gen;
    98 
    99         switch (mesg) {
    100         case MESSAGE_PACKET:
    101                 /* Apply filters to every packet note the result */
    102                 wlen = trace_get_wire_length(data.pkt);
    103                 for(i=0;i<filter_count;++i) {
    104                         if (filters[i].filter == NULL)
    105                                 continue;
    106                         if(trace_apply_filter(filters[i].filter,data.pkt) > 0) {
    107                                 results[i+1].count++;
    108                                 results[i+1].bytes+=wlen;
    109                         }
    110                         if (trace_is_err(trace)) {
    111                                 trace_perror(trace, "trace_apply_filter");
    112                                 fprintf(stderr, "Removing filter from filterlist\n");
    113                                 /* This is a race, but will be atomic */
    114                                 filters[i].filter = NULL;
    115                         }
    116                 }
    117                 results[0].count++;
    118                 results[0].bytes +=wlen;
    119                 return data.pkt;
    120         case MESSAGE_STARTING:
    121                 /* Allocate space to hold a total count and one for each filter */
    122                 results = calloc(1, sizeof(statistics_t) * (filter_count + 1));
    123                 break;
    124         case MESSAGE_STOPPING:
    125                 /* We only output one result per thread with the key 0 when the
    126                  * trace is over. */
    127                 gen.ptr = results;
    128                 trace_publish_result(trace, t, 0, gen, RESULT_USER);
    129                 break;
    130         default:
    131                 break;
    132         }
    13393        return NULL;
    13494}
     
    184144}
    185145
     146
     147static void* fn_starting(libtrace_t *trace UNUSED, libtrace_thread_t *t,
     148                     libtrace_generic_t data UNUSED, void *global UNUSED, void*tls UNUSED) {
     149        /* Allocate space to hold a total count and one for each filter */
     150        statistics_t *results = calloc(1, sizeof(statistics_t) * (filter_count + 1));
     151        trace_set_tls(t, results);
     152        return NULL;
     153}
     154
     155
     156static void* fn_stopping(libtrace_t *trace, libtrace_thread_t *t UNUSED,
     157                     libtrace_generic_t data UNUSED, void *global UNUSED, void*tls) {
     158        statistics_t *results = tls;
     159        libtrace_generic_t gen;
     160        /* We only output one result per thread with the key 0 when the
     161         * trace is over. */
     162        gen.ptr = results;
     163        trace_publish_result(trace, t, 0, gen, RESULT_USER);
     164        return NULL;
     165}
     166
     167static void* fn_packet(libtrace_t *trace, libtrace_thread_t *t UNUSED,
     168                   libtrace_generic_t data, void *global UNUSED, void*tls) {
     169        statistics_t *results = tls;
     170        int i, wlen;
     171
     172        /* Apply filters to every packet note the result */
     173        wlen = trace_get_wire_length(data.pkt);
     174        for(i=0;i<filter_count;++i) {
     175                if (filters[i].filter == NULL)
     176                        continue;
     177                if(trace_apply_filter(filters[i].filter,data.pkt) > 0) {
     178                        results[i+1].count++;
     179                        results[i+1].bytes+=wlen;
     180                }
     181                if (trace_is_err(trace)) {
     182                        trace_perror(trace, "trace_apply_filter");
     183                        fprintf(stderr, "Removing filter from filterlist\n");
     184                        /* This is a race, but will be atomic */
     185                        filters[i].filter = NULL;
     186                }
     187        }
     188        results[0].count++;
     189        results[0].bytes +=wlen;
     190        return data.pkt;
     191}
     192
    186193/* Process a trace, counting packets that match filter(s) */
    187194static void run_trace(char *uri, char *config, char *config_file)
     
    212219                }
    213220        }
     221
     222        trace_set_handler(trace, MESSAGE_PACKET, fn_packet);
     223        trace_set_handler(trace, MESSAGE_STARTING, fn_starting);
     224        trace_set_handler(trace, MESSAGE_STOPPING, fn_stopping);
    214225
    215226        /* Start the trace as a parallel trace */
Note: See TracChangeset for help on using the changeset viewer.