Changeset 0ec8a7c for lib

Timestamp:

02/27/15 17:31:03 (7 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:

98dc1ba

Parents:

7718e54

Message:

Change the user interface from a message and a packet to only messages.

Now a packet is a type of message (MESSAGE_PACKET).
I expanded the message structure into seperate arguments such that these
will be passed in registers on x64 systems. As such performance has
remained identical if not better.

I renamed libtrace_generic_types_t to libtrace_generic_t so it was shorter.

Location:

lib

Files:

• libtrace.h.in (modified) (8 diffs)
• trace_parallel.c (modified) (10 diffs)

lib/libtrace.h.in

@@ -3507 +3507 @@
         float rfloat;
         double rdouble;
-} libtrace_generic_types_t;
+} libtrace_generic_t;
+ct_assert(sizeof(libtrace_generic_t) == 8);
 
 typedef struct libtrace_message_t {
         int code;
-        libtrace_generic_types_t additional;
+        libtrace_generic_t additional;
         libtrace_thread_t *sender;
 } libtrace_message_t;
@@ -3518 +3519 @@
 typedef struct libtrace_result_t {
         uint64_t key;
-        libtrace_generic_types_t value;
+        libtrace_generic_t value;
         int type;
 } libtrace_result_t;
@@ -3525 +3526 @@
 #define RESULT_TICK   2
 
-
-typedef void* (*fn_per_pkt)(libtrace_t* trace, libtrace_packet_t *p, libtrace_message_t *m, libtrace_thread_t *thread);
+/**
+ * The definition for the main function that the user supplies to process
+ * packets.
+ *
+ * @param trace The trace the packet is related to.
+ * @param thread The thread the trace is related to.
+ * @param mesg_code The type of data ready, the most important being MESSAGE_PACKET.
+ * In this case data.pkt contains the packet.
+ * @param data A generic union of types that fit into 8 bytes, containing
+ * information dependent upon the mesg_code.
+ * @param sender The thread that the message originated from.
+ *
+ * The values of data and sender depend upon the mesg_code. Please see the
+ * documentation for the message as to what value these will contain.
+ */
+typedef void* (*fn_per_pkt)(libtrace_t* trace,
+                            libtrace_thread_t *thread,
+                            int mesg_code,
+                            libtrace_generic_t data,
+                            libtrace_thread_t *sender);
 typedef void (*fn_reporter)(libtrace_t* trace, libtrace_result_t *r, libtrace_message_t *m);
 typedef uint64_t (*fn_hasher)(const libtrace_packet_t* packet, void *data);
@@ -3537 +3556 @@
 DLLEXPORT void libtrace_result_set_key(libtrace_result_t * result, uint64_t key);
 DLLEXPORT uint64_t libtrace_result_get_key(libtrace_result_t * result);
-DLLEXPORT void libtrace_result_set_value(libtrace_result_t * result, libtrace_generic_types_t value);
-DLLEXPORT libtrace_generic_types_t libtrace_result_get_value(libtrace_result_t * result);
-DLLEXPORT void libtrace_result_set_key_value(libtrace_result_t * result, uint64_t key, libtrace_generic_types_t value);
+DLLEXPORT void libtrace_result_set_value(libtrace_result_t * result, libtrace_generic_t value);
+DLLEXPORT libtrace_generic_t libtrace_result_get_value(libtrace_result_t * result);
+DLLEXPORT void libtrace_result_set_key_value(libtrace_result_t * result, uint64_t key, libtrace_generic_t value);
 DLLEXPORT void trace_destroy_result(libtrace_result_t ** result);
 
@@ -3549 +3568 @@
 
 
-DLLEXPORT void trace_publish_result(libtrace_t *libtrace, libtrace_thread_t *t, uint64_t key, libtrace_generic_types_t value, int type);
+DLLEXPORT void trace_publish_result(libtrace_t *libtrace, libtrace_thread_t *t, uint64_t key, libtrace_generic_t value, int type);
 typedef struct libtrace_vector libtrace_vector_t;
 
@@ -3599 +3618 @@
 
 enum libtrace_messages {
+        MESSAGE_PACKET,
         MESSAGE_STARTING,
         MESSAGE_RESUMING,
@@ -3613 +3633 @@
         MESSAGE_POST_RANGE,
         MESSAGE_TICK,
-        MESSAGE_USER
+        MESSAGE_USER = 1000
 };
 
@@ -3854 +3874 @@
          * chosen.
          */
-        libtrace_generic_types_t configuration;
+        libtrace_generic_t configuration;
 };
 
-DLLEXPORT void trace_set_combiner(libtrace_t *trace, const libtrace_combine_t *combiner, libtrace_generic_types_t config);
+DLLEXPORT void trace_set_combiner(libtrace_t *trace, const libtrace_combine_t *combiner, libtrace_generic_t config);
 
 #define READ_EOF 0

lib/trace_parallel.c

@@ -385 +385 @@
                                   libtrace_packet_t **packet,
                                   bool tracetime) {
+
         if ((*packet)->error > 0) {
                 if (tracetime) {
@@ -391 +392 @@
                 }
                 t->accepted_packets++;
-                *packet = (*trace->per_pkt)(trace, *packet, NULL, t);
+                libtrace_generic_t data = {.pkt = *packet};
+                *packet = (*trace->per_pkt)(trace, t, MESSAGE_PACKET, data, t);
                 trace_fin_packet(*packet);
         } else {
-                libtrace_message_t message;
                 assert((*packet)->error == READ_TICK);
-                message.code = MESSAGE_TICK;
-                message.additional.uint64 = trace_packet_get_order(*packet);
-                message.sender = t;
-                (*trace->per_pkt)(trace, NULL, &message, t);
+                libtrace_generic_t data = {.uint64 = trace_packet_get_order(*packet)};
+                (*trace->per_pkt)(trace, t, MESSAGE_TICK, data, t);
         }
         return 0;
@@ -462 +461 @@
                                      libtrace_packet_t *packets[],
                                      int nb_packets, int *empty, int *offset) {
-        libtrace_message_t message = {0};
         libtrace_packet_t * packet = NULL;
 
         /* Let the user thread know we are going to pause */
-        message.code = MESSAGE_PAUSING;
-        message.sender = t;
-        (*trace->per_pkt)(trace, NULL, &message, t);
+        (*trace->per_pkt)(trace, t, MESSAGE_PAUSING, (libtrace_generic_t){0}, t);
 
         /* Send through any remaining packets (or messages) without delay */
@@ -511 +507 @@
         /* Now we do the actual pause, this returns when we resumed */
         trace_thread_pause(trace, t);
-        message.code = MESSAGE_RESUMING;
-        (*trace->per_pkt)(trace, NULL, &message, t);
+        (*trace->per_pkt)(trace, t, MESSAGE_RESUMING, (libtrace_generic_t){0}, t);
         return 1;
 }
@@ -558 +553 @@
 
         /* Let the per_packet function know we have started */
-        message.code = MESSAGE_STARTING;
-        message.sender = t;
-        (*trace->per_pkt)(trace, NULL, &message, t);
-        message.code = MESSAGE_RESUMING;
-        (*trace->per_pkt)(trace, NULL, &message, t);
-
+        (*trace->per_pkt)(trace, t, MESSAGE_STARTING, (libtrace_generic_t){0}, t);
+        (*trace->per_pkt)(trace, t, MESSAGE_RESUMING, (libtrace_generic_t){0}, t);
 
         for (;;) {
@@ -586 +577 @@
                                         goto eof;
                         }
-                        (*trace->per_pkt)(trace, NULL, &message, t);
+                        (*trace->per_pkt)(trace, t, message.code, message.additional, message.sender);
                         /* Continue and the empty messages out before packets */
                         continue;
@@ -653 +644 @@
 
         // Let the per_packet function know we have stopped
-        message.code = MESSAGE_PAUSING;
-        message.sender = t;
-        (*trace->per_pkt)(trace, NULL, &message, t);
-        message.code = MESSAGE_STOPPING;
-        message.additional.uint64 = 0;
-        (*trace->per_pkt)(trace, NULL, &message, t);
+        (*trace->per_pkt)(trace, t, MESSAGE_PAUSING, (libtrace_generic_t){0}, t);
+        (*trace->per_pkt)(trace, t, MESSAGE_STOPPING, (libtrace_generic_t){0}, t);
 
         // Free any remaining packets
@@ -2101 +2088 @@
         return result->key;
 }
-DLLEXPORT void libtrace_result_set_value(libtrace_result_t * result, libtrace_generic_types_t value) {
+DLLEXPORT void libtrace_result_set_value(libtrace_result_t * result, libtrace_generic_t value) {
         result->value = value;
 }
-DLLEXPORT libtrace_generic_types_t libtrace_result_get_value(libtrace_result_t * result) {
+DLLEXPORT libtrace_generic_t libtrace_result_get_value(libtrace_result_t * result) {
         return result->value;
 }
-DLLEXPORT void libtrace_result_set_key_value(libtrace_result_t * result, uint64_t key, libtrace_generic_types_t value) {
+DLLEXPORT void libtrace_result_set_key_value(libtrace_result_t * result, uint64_t key, libtrace_generic_t value) {
         result->key = key;
         result->value = value;
@@ -2155 +2142 @@
  * Should only be called by a perpkt thread, i.e. from a perpkt handler
  */
-DLLEXPORT void trace_publish_result(libtrace_t *libtrace, libtrace_thread_t *t, uint64_t key, libtrace_generic_types_t value, int type) {
+DLLEXPORT void trace_publish_result(libtrace_t *libtrace, libtrace_thread_t *t, uint64_t key, libtrace_generic_t value, int type) {
         libtrace_result_t res;
         res.type = type;
@@ -2168 +2155 @@
  * Sets a combiner function against the trace.
  */
-DLLEXPORT void trace_set_combiner(libtrace_t *trace, const libtrace_combine_t *combiner, libtrace_generic_types_t config){
+DLLEXPORT void trace_set_combiner(libtrace_t *trace, const libtrace_combine_t *combiner, libtrace_generic_t config){
         if (combiner) {
                 trace->combiner = *combiner;