Changeset 6445c52

Timestamp:

02/13/18 13:23:11 (4 years ago)

Author:

Shane Alcock <salcock@…>

Branches:

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

Children:

7ebf768

Parents:

fd1b63e

Message:

Fixed ndag packet corruption bug

Ensure we only mark receive buffers as available *after* we've
finished fetching a batch of packets -- otherwise we run the
risk of overwriting an earlier packet in the batch by reading
fresh data into the "available" buffer.

File:

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

lib/format_ndag.c

@@ -91 +91 @@
 
         int bufavail;
+        int bufwaiting;
 
 #if HAVE_RECVMMSG
@@ -633 +634 @@
         ssock->nextread += ntohs(erfptr->rlen);
 
+        assert(ssock->nextread - ssock->saved[nr] <= ssock->savedsize[nr]);
+
         if (ssock->nextread - ssock->saved[nr] >= ssock->savedsize[nr]) {
                 /* Read everything from this buffer, mark as empty and
                  * move on. */
                 ssock->savedsize[nr] = 0;
-                ssock->bufavail ++;
-
-                assert(ssock->bufavail > 0 && ssock->bufavail <= ENCAP_BUFFERS);
+                ssock->bufwaiting ++;
+
                 nr ++;
                 if (nr == ENCAP_BUFFERS) {
@@ -652 +654 @@
         packet->error = packet->payload ? ntohs(erfptr->rlen) :
                         erf_get_framing_length(packet);
-
         return ntohs(erfptr->rlen);
 }
@@ -731 +732 @@
         ssock->saved = (char **)malloc(sizeof(char *) * ENCAP_BUFFERS);
         ssock->bufavail = ENCAP_BUFFERS;
+        ssock->bufwaiting = 0;
         ssock->startidle = 0;
 
@@ -901 +903 @@
                 rt->missing_records += seq_cmp(
                                 ntohl(encaphdr->seqno), ssock->expectedseq);
+
         }
         ssock->expectedseq = ntohl(encaphdr->seqno) + 1;
@@ -1121 +1124 @@
                         ssock = &(rt->sources[i]);
                 }
-                /*
-                fprintf(stderr, "%d %d %lu %lu %lu\n", rt->threadindex,
-                                i, currentts,
-                                rt->sources[i].recordcount, 
-                                rt->missing_records);
-                */
         }
         return ssock;
@@ -1133 +1130 @@
 static int ndag_read_packet(libtrace_t *libtrace, libtrace_packet_t *packet) {
 
-        int rem;
+        int rem, ret;
         streamsock_t *nextavail = NULL;
         rem = receive_encap_records_block(libtrace, &(FORMAT_DATA->receivers[0]),
@@ -1150 +1147 @@
          * a libtrace ERF packet. */
 
-        return ndag_prepare_packet_stream(libtrace,
+        ret = ndag_prepare_packet_stream(libtrace,
                         &(FORMAT_DATA->receivers[0]), nextavail,
                         packet, TRACE_PREP_DO_NOT_OWN_BUFFER);
+        nextavail->bufavail += nextavail->bufwaiting;
+        nextavail->bufwaiting = 0;
+        return ret;
 }
 
@@ -1159 +1159 @@
 
         recvstream_t *rt;
-        int rem;
+        int rem, i;
         size_t read_packets = 0;
         streamsock_t *nextavail = NULL;
@@ -1199 +1199 @@
         } while (1);
 
+        
+        for (i = 0; i < rt->sourcecount; i++) {
+                streamsock_t *src = &(rt->sources[i]);
+                src->bufavail += src->bufwaiting;
+                src->bufwaiting = 0;
+                assert(src->bufavail >= 0 && src->bufavail <= ENCAP_BUFFERS);
+        }
+
         return read_packets;
 
@@ -1208 +1216 @@
 
         libtrace_eventobj_t event = {0,0,0.0,0};
-        int rem;
+        int rem, i;
         streamsock_t *nextavail = NULL;
 
@@ -1280 +1288 @@
                 break;
         } while (1);
+
+        for (i = 0; i < FORMAT_DATA->receivers[0].sourcecount; i++) {
+                streamsock_t *src = &(FORMAT_DATA->receivers[0].sources[i]);
+                src->bufavail += src->bufwaiting;
+                src->bufwaiting = 0;
+                assert(src->bufavail >= 0 && src->bufavail <= ENCAP_BUFFERS);
+        }
 
         return event;