Changeset 4e427da for lib

Timestamp:

11/07/17 14:38:27 (3 years ago)

Author:

Shane Alcock <salcock@…>

Branches:

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

Children:

de825b5

Parents:

aa7db84

Message:

Increase number of packets that can be buffered at once.

100 packets per socket seems to work fairly well, but going to
10,000 introduces a severe performance penalty for some reason.

File:

• lib/format_ndag.c (modified) (14 diffs)

lib/format_ndag.c

@@ -25 +25 @@
 #define ENCAP_BUFSIZE (10000)
 #define CTRL_BUF_SIZE (10000)
+#define ENCAP_BUFFERS (100)
 
 #define FORMAT_DATA ((ndag_format_data_t *)libtrace->format_data)
@@ -47 +48 @@
         uint16_t monitorid;
 
-        char *saved;
+        char **saved;
         char *nextread;
-        int savedsize;
-        ndag_encap_t *encaphdr;
+        int nextreadind;
+        int nextwriteind;
+        int savedsize[ENCAP_BUFFERS];
         uint64_t recordcount;
 
@@ -474 +476 @@
 
 static void halt_ndag_receiver(recvstream_t *receiver) {
-        int i;
+        int j, i;
         libtrace_message_queue_destroy(&(receiver->mqueue));
 
@@ -482 +484 @@
                 streamsock_t src = receiver->sources[i];
                 if (src.saved) {
+                        for (j = 0; i < ENCAP_BUFFERS; j++) {
+                                if (src.saved[j]) {
+                                        free(src.saved[j]);
+                                }
+                        }
                         free(src.saved);
                 }
@@ -526 +533 @@
 
         dag_record_t *erfptr;
+        ndag_encap_t *encaphdr;
         uint16_t ndag_reccount = 0;
+        int nr;
 
         if ((flags & TRACE_PREP_OWN_BUFFER) == TRACE_PREP_OWN_BUFFER) {
@@ -562 +571 @@
         ssock->recordcount += 1;
 
-        ndag_reccount = ntohs(ssock->encaphdr->recordcount);
+        nr = ssock->nextreadind;
+        encaphdr = (ndag_encap_t *)(ssock->saved[nr] +
+                        sizeof(ndag_common_t));
+
+        ndag_reccount = ntohs(encaphdr->recordcount);
         if ((ndag_reccount & 0x8000) != 0) {
                 /* Record was truncated -- update rlen appropriately */
-                erfptr->rlen = htons(ssock->savedsize -
-                                (ssock->nextread - ssock->saved));
+                erfptr->rlen = htons(ssock->savedsize[nr] -
+                                (ssock->nextread - ssock->saved[nr]));
         }
         ssock->nextread += ntohs(erfptr->rlen);
+
+        if (ssock->nextread - ssock->saved[nr] >= ssock->savedsize[nr]) {
+                /* Read everything from this buffer, mark as empty and
+                 * move on. */
+                ssock->savedsize[nr] = 0;
+                nr = (nr + 1) % ENCAP_BUFFERS;
+                ssock->nextread = ssock->saved[nr] + sizeof(ndag_common_t) +
+                                sizeof(ndag_encap_t);
+                ssock->nextreadind = nr;
+        }
 
         packet->order = erf_get_erf_timestamp(packet);
@@ -590 +613 @@
 
         streamsock_t *ssock = NULL;
+        int i;
 
         if (rt->sourcecount == 0) {
@@ -611 +635 @@
         ssock->expectedseq = 0;
         ssock->monitorid = src.monitor;
-        ssock->saved = (char *)malloc(ENCAP_BUFSIZE);
-        ssock->nextread = ssock->saved;
-        ssock->savedsize = 0;
-        ssock->encaphdr = NULL;
+        ssock->saved = (char **)malloc(sizeof(char *) * ENCAP_BUFFERS);
+
+
+        for (i = 0; i < ENCAP_BUFFERS; i++) {
+                ssock->saved[i] = (char *)malloc(ENCAP_BUFSIZE);
+                ssock->savedsize[i] = 0;
+        }
+
+        ssock->nextread = NULL;;
+        ssock->nextreadind = 0;
         ssock->recordcount = 0;
         rt->sourcecount += 1;
@@ -648 +678 @@
 }
 
+static inline int readable_data(streamsock_t ssock) {
+
+        if (ssock.sock == -1) {
+                return 0;
+        }
+        if (ssock.savedsize[ssock.nextreadind] == 0) {
+                return 0;
+        }
+        if (ssock.nextread - ssock.saved[ssock.nextreadind] >=
+                        ssock.savedsize[ssock.nextreadind]) {
+                return 0;
+        }
+        return 1;
+
+
+}
+
 static inline int read_required(streamsock_t ssock) {
         if (ssock.sock == -1)
                 return 0;
-        if (ssock.savedsize == 0)
+        if (ssock.savedsize[ssock.nextwriteind] == 0)
                 return 1;
-        if (ssock.nextread - ssock.saved >= ssock.savedsize)
-                return 1;
+        //if (ssock.nextread - ssock.saved >= ssock.savedsize)
+        //        return 1;
         return 0;
 }
@@ -694 +741 @@
          */
         for (i = 0; i < rt->sourcecount; i ++) {
+                int nw;
+                ndag_encap_t *encaphdr;
+
                 if (!read_required(rt->sources[i])) {
                         if (rt->sources[i].sock != -1) {
@@ -701 +751 @@
                 }
 
-                rt->sources[i].savedsize = 0;
-                rt->sources[i].nextread = rt->sources[i].saved;
-
-                ret = recvfrom(rt->sources[i].sock, rt->sources[i].saved,
+                nw = rt->sources[i].nextwriteind;
+
+                ret = recvfrom(rt->sources[i].sock, rt->sources[i].saved[nw],
                                 ENCAP_BUFSIZE, MSG_DONTWAIT,
                                 rt->sources[i].srcaddr->ai_addr,
                                 &(rt->sources[i].srcaddr->ai_addrlen));
                 if (ret < 0) {
+                        /* Nothing to receive right now, but we should still
+                         * count as 'ready' if at least one buffer is full */
                         if (errno == EAGAIN || errno == EWOULDBLOCK) {
+                                if (readable_data(rt->sources[i])) {
+                                        readybufs ++;
+                                }
                                 continue;
                         }
@@ -732 +786 @@
                 }
 
-                rt->sources[i].savedsize = ret;
+                rt->sources[i].savedsize[nw] = ret;
+                rt->sources[i].nextwriteind =
+                                (rt->sources[i].nextwriteind + 1) % ENCAP_BUFFERS;
                 successrecv ++;
 
                 /* Check that we have a valid nDAG encap record */
-                if (check_ndag_header(rt->sources[i].saved, ret) !=
+                if (check_ndag_header(rt->sources[i].saved[nw], ret) !=
                                         NDAG_PKT_ENCAPERF) {
                         fprintf(stderr, "Received invalid record on the channel for %s:%u.\n",
@@ -747 +803 @@
 
                 /* Save the useful info from the encap header */
-                rt->sources[i].encaphdr = (ndag_encap_t *)
-                                (rt->sources[i].saved + sizeof(ndag_common_t));
+                encaphdr = (ndag_encap_t *)(rt->sources[i].saved[nw] +
+                                sizeof(ndag_common_t));
 
                 if (rt->sources[i].expectedseq != 0) {
                         rt->missing_records += seq_cmp(
-                                        ntohl(rt->sources[i].encaphdr->seqno),
+                                        ntohl(encaphdr->seqno),
                                         rt->sources[i].expectedseq);
                 }
-                rt->sources[i].expectedseq =
-                                ntohl(rt->sources[i].encaphdr->seqno) + 1;
+                rt->sources[i].expectedseq = ntohl(encaphdr->seqno) + 1;
 
                 /* If all good, skip past the nDAG headers */
-                rt->sources[i].nextread = rt->sources[i].saved +
-                                sizeof(ndag_common_t) + sizeof(ndag_encap_t);
+                if (rt->sources[i].nextread == NULL) {
+                        rt->sources[i].nextread = rt->sources[i].saved[0] +
+                                        sizeof(ndag_common_t) + sizeof(ndag_encap_t);
+                }
 
                 readybufs += 1;
@@ -835 +892 @@
 
         for (i = 0; i < rt->sourcecount; i ++) {
-                if (read_required(rt->sources[i])) {
+                if (!readable_data(rt->sources[i])) {
                         continue;
                 }