Changeset e54bd5f

Timestamp:

12/14/17 11:06:48 (3 years ago)

Author:

Shane Alcock <salcock@…>

Branches:

cachetimestamps, develop, dpdk-ndag, etsilive, master, rc-4.0.3, rc-4.0.4, ringdecrementfix, ringperformance

Children:

4f0f93f

Parents:

a6f2d1d

Message:

Tweaks to dpdkndag: format

• Set snap length for dpdk receiver to 9K so that we can receive jumbo frames.
• Enforce a unidirectional hasher on the dpdk receiver, just to ensure ndag streams don't get spread across multiple CPUs.
• Add a mutex around any statistic updates to keep helgrind happy.
• Increase number of mbufs used from 20 to 40.
• Dropped packets by the dpdk receiver are included in stats as 'errors'.

File:

• lib/format_dpdkndag.c (modified) (10 diffs)

lib/format_dpdkndag.c

@@ -42 +42 @@
         uint32_t ndagsize;
 
+        pthread_mutex_t ndag_lock;
         dpdk_per_stream_t *dpdkstreamdata;
         int burstsize;
         int burstoffset;
-        struct rte_mbuf* burstspace[20];
+        struct rte_mbuf* burstspace[40];
 
 } perthread_t;
@@ -156 +157 @@
                 FORMAT_DATA->threaddatas[i].burstoffset = 0;
                 memset(FORMAT_DATA->threaddatas[i].burstspace, 0,
-                                sizeof(struct rte_mbuf *) * 20);
+                                sizeof(struct rte_mbuf *) * 40);
+                pthread_mutex_init(&(FORMAT_DATA->threaddatas[i].ndag_lock),
+                                NULL);
         }
         return maxthreads;
@@ -162 +165 @@
 
 static int dpdkndag_start_input(libtrace_t *libtrace) {
+        enum hasher_types hash = HASHER_UNIDIRECTIONAL;
+        int snaplen = 9000;
+
+        if (dpdk_config_input(FORMAT_DATA->dpdkrecv, TRACE_OPTION_HASHER,
+                                &hash) == -1) {
+                libtrace_err_t err = trace_get_err(FORMAT_DATA->dpdkrecv);
+                trace_set_err(libtrace, TRACE_ERR_INIT_FAILED, "%s", err.problem);
+                return -1;
+        }
+
+        if (dpdk_config_input(FORMAT_DATA->dpdkrecv, TRACE_OPTION_SNAPLEN,
+                                &snaplen) == -1) {
+                libtrace_err_t err = trace_get_err(FORMAT_DATA->dpdkrecv);
+                trace_set_err(libtrace, TRACE_ERR_INIT_FAILED, "%s", err.problem);
+                return -1;
+        }
 
         if (dpdk_start_input(FORMAT_DATA->dpdkrecv) == -1) {
@@ -176 +195 @@
 static int dpdkndag_pstart_input(libtrace_t *libtrace) {
 
+        enum hasher_types hash = HASHER_UNIDIRECTIONAL;
+        int snaplen = 9000;
         FORMAT_DATA->dpdkrecv->perpkt_thread_count = libtrace->perpkt_thread_count;
+        if (dpdk_config_input(FORMAT_DATA->dpdkrecv, TRACE_OPTION_HASHER,
+                                &hash) == -1) {
+                libtrace_err_t err = trace_get_err(FORMAT_DATA->dpdkrecv);
+                trace_set_err(libtrace, TRACE_ERR_INIT_FAILED, "%s", err.problem);
+                return -1;
+        }
+
+        if (dpdk_config_input(FORMAT_DATA->dpdkrecv, TRACE_OPTION_SNAPLEN,
+                                &snaplen) == -1) {
+                libtrace_err_t err = trace_get_err(FORMAT_DATA->dpdkrecv);
+                trace_set_err(libtrace, TRACE_ERR_INIT_FAILED, "%s", err.problem);
+                return -1;
+        }
         if (dpdk_pstart_input(FORMAT_DATA->dpdkrecv) == -1) {
                 libtrace_err_t err = trace_get_err(FORMAT_DATA->dpdkrecv);
@@ -199 +233 @@
         }
 
-        for (i = 0; i < 20; i++) {
+        for (i = 0; i < 40; i++) {
                 if (pt->burstspace[i]) {
                         rte_pktmbuf_free(pt->burstspace[i]);
                 }
         }
+        pthread_mutex_destroy(&(pt->ndag_lock));
 }
 
@@ -291 +326 @@
 static void dpdkndag_get_statistics(libtrace_t *libtrace, libtrace_stat_t *stat) {
         int i;
+
+        libtrace_stat_t *dpdkstat;
 
         stat->dropped_valid = 1;
@@ -299 +336 @@
         stat->missing = 0;
 
+        dpdkstat = trace_create_statistics();
+        dpdk_get_stats(FORMAT_DATA->dpdkrecv, dpdkstat);
+
+        if (dpdkstat->dropped_valid) {
+                stat->errors_valid = 1;
+                stat->errors = dpdkstat->dropped;
+        }
+
         /* TODO Is this thread safe? */
         for (i = 0; i < libtrace->perpkt_thread_count; i++) {
+                pthread_mutex_lock(&(FORMAT_DATA->threaddatas[i].ndag_lock));
                 stat->dropped += FORMAT_DATA->threaddatas[i].dropped_upstream;
                 stat->received += FORMAT_DATA->threaddatas[i].received_packets;
                 stat->missing += FORMAT_DATA->threaddatas[i].missing_records;
-        }
+                pthread_mutex_unlock(&(FORMAT_DATA->threaddatas[i].ndag_lock));
+        }
+        free(dpdkstat);
 }
 
@@ -434 +482 @@
                 }
         }
-
         if (cap->expectedseq != 0) {
+                pthread_mutex_lock(&pt->ndag_lock);
                 pt->missing_records += seq_cmp(
                                 ntohl(encaphdr->seqno), cap->expectedseq);
+                pthread_mutex_unlock(&pt->ndag_lock);
         }
         cap->expectedseq = ntohl(encaphdr->seqno) + 1;
@@ -490 +539 @@
                 /* TODO */
         } else {
+                pthread_mutex_lock(&(pt->ndag_lock));
                 if (pt->received_packets > 0) {
                         pt->dropped_upstream += ntohs(erfptr->lctr);
                 }
-        }
-
+                pthread_mutex_unlock(&(pt->ndag_lock));
+        }
+
+        pthread_mutex_lock(&(pt->ndag_lock));
         pt->received_packets ++;
+        pthread_mutex_unlock(&(pt->ndag_lock));
         encaphdr = (ndag_encap_t *)(pt->ndagheader + sizeof(ndag_common_t));
 
@@ -542 +595 @@
                                         &t->messages,
                                         pt->burstspace,
-                                        20);
+                                        40);
                         if (ret <= 0) {
                                 return ret;
                         }
+
                         pt->dpdkpkt->buffer = pt->burstspace[0];
                         pt->dpdkpkt->trace = FORMAT_DATA->dpdkrecv;