Changeset 4bab977 for lib

Timestamp:

11/10/17 14:37:19 (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:

d39cd1e

Parents:

0317e3c

Message:

Modifications in response to changes in the nDAG protocol.

• Updated protocol to account for new laststart field in encap. header. Use this field to recognise when the monitor has been restarted.

• When determining sequence number gap, don't count 0 as a valid sequence number -- seqno 0 is reserved in the nDAG protocol and will never be lost.

Location:

lib

Files:

• format_ndag.c (modified) (15 diffs)
• format_ndag.h (modified) (1 diff)

lib/format_ndag.c

@@ -34 +34 @@
 volatile int ndag_paused = 0;
 
+typedef struct monitor {
+        uint16_t monitorid;
+        uint64_t laststart;
+} ndag_monitor_t;
+
+
 typedef struct streamsource {
         uint16_t monitor;
@@ -47 +53 @@
         uint16_t port;
         uint32_t expectedseq;
-        uint16_t monitorid;
-
+        ndag_monitor_t *monitorptr;
         char **saved;
         char *nextread;
@@ -64 +69 @@
         libtrace_message_queue_t mqueue;
         int threadindex;
+        ndag_monitor_t *knownmonitors;
+        uint16_t monitorcount;
 
         uint64_t dropped_upstream;
@@ -83 +90 @@
 enum {
         NDAG_CLIENT_HALT = 0x01,
-        NDAG_CLIENT_RESTARTED = 0x02,
+        NDAG_CLIENT_RESTARTED = 0x02,   // redundant
         NDAG_CLIENT_NEWGROUP = 0x03
 };
@@ -101 +108 @@
                 return (int) (seq_a - seq_b);
         }
-        return (int) (0xffffffff - ((seq_b - seq_a) - 1));
+
+        /* -1 for the wrap and another -1 because we don't use zero */
+        return (int) (0xffffffff - ((seq_b - seq_a) - 2));
 }
 
@@ -329 +338 @@
                         ptr ++;
                 }
-        } else if (msgtype == NDAG_PKT_RESTARTED) {
-                /* If message is a restart, push that to all active message
-                 * queues. */
-                ndag_internal_message_t alert;
-                alert.type = NDAG_CLIENT_RESTARTED;
-                alert.contents.monitor = ntohs(ndaghdr->monitorid);
-                alert.contents.groupaddr = NULL;
-                alert.contents.localiface = NULL;
-                alert.contents.port = 0;
-                for (i = 0; i < libtrace->perpkt_thread_count; i++) {
-                        libtrace_message_queue_put(
-                                        &(FORMAT_DATA->receivers[i].mqueue),
-                                        (void *)&alert);
-                }
         } else {
                 fprintf(stderr,
@@ -446 +441 @@
                 FORMAT_DATA->receivers[i].sources = NULL;
                 FORMAT_DATA->receivers[i].sourcecount = 0;
+                FORMAT_DATA->receivers[i].knownmonitors = NULL;
+                FORMAT_DATA->receivers[i].monitorcount = 0;
                 FORMAT_DATA->receivers[i].threadindex = i;
                 FORMAT_DATA->receivers[i].dropped_upstream = 0;
@@ -495 +492 @@
                 close(src.sock);
         }
+        if (receiver->knownmonitors) {
+                free(receiver->knownmonitors);
+        }
+
         if (receiver->sources) {
                 free(receiver->sources);
@@ -612 +613 @@
 }
 
+static ndag_monitor_t *add_new_knownmonitor(recvstream_t *rt, uint16_t monid) {
+
+        ndag_monitor_t *mon;
+
+        if (rt->monitorcount == 0) {
+                rt->knownmonitors = (ndag_monitor_t *)
+                                malloc(sizeof(ndag_monitor_t) * 5);
+        } else {
+                rt->knownmonitors = (ndag_monitor_t *)
+                            realloc(rt->knownmonitors,
+                            sizeof(ndag_monitor_t) * (rt->monitorcount * 5));
+        }
+
+        mon = &(rt->knownmonitors[rt->monitorcount]);
+        mon->monitorid = monid;
+        mon->laststart = 0;
+
+        rt->monitorcount ++;
+        return mon;
+}
+
 static int add_new_streamsock(recvstream_t *rt, streamsource_t src) {
 
         streamsock_t *ssock = NULL;
+        ndag_monitor_t *mon = NULL;
         int i;
 
@@ -638 +661 @@
         }
 
+        for (i = 0; i < rt->monitorcount; i++) {
+                if (rt->knownmonitors[i].monitorid == src.monitor) {
+                        mon = &(rt->knownmonitors[i]);
+                        break;
+                }
+        }
+
+        if (mon == NULL) {
+                mon = add_new_knownmonitor(rt, src.monitor);
+        }
+
         ssock->port = src.port;
         ssock->groupaddr = src.groupaddr;
         ssock->expectedseq = 0;
-        ssock->monitorid = src.monitor;
+        ssock->monitorptr = mon;
         ssock->saved = (char **)malloc(sizeof(char *) * ENCAP_BUFFERS);
         ssock->startidle = 0;
@@ -673 +707 @@
                                 }
                                 break;
-                        case NDAG_CLIENT_RESTARTED:
-                                /* TODO */
-
-                                break;
                         case NDAG_CLIENT_HALT:
                                 return 0;
@@ -702 +732 @@
 }
 
+static inline void reset_expected_seqs(recvstream_t *rt, ndag_monitor_t *mon) {
+
+        int i;
+        for (i = 0; i < rt->sourcecount; i++) {
+                if (rt->sources[i].monitorptr == mon) {
+                        rt->sources[i].expectedseq = 0;
+                }
+        }
+
+}
+
 static int receive_from_sockets(recvstream_t *rt) {
 
@@ -714 +755 @@
                 int nw;
                 ndag_encap_t *encaphdr;
+                ndag_monitor_t *mon;
 
                 if (rt->sources[i].sock == -1) {
@@ -787 +829 @@
                                 sizeof(ndag_common_t));
 
+                mon = rt->sources[i].monitorptr;
+                assert(mon);
+
+                if (mon->laststart == 0) {
+                        mon->laststart = bswap_be_to_host64(encaphdr->started);
+                } else if (mon->laststart != bswap_be_to_host64(encaphdr->started)) {
+                        mon->laststart = bswap_be_to_host64(encaphdr->started);
+                        reset_expected_seqs(rt, mon);
+
+                        /* TODO what is a good way to indicate this to clients?
+                         * set the loss counter in the ERF header? a bit rude?
+                         * use another bit in the ERF header?
+                         * add a queryable flag to libtrace_packet_t?
+                         */
+
+                }
+
                 if (rt->sources[i].expectedseq != 0) {
                         rt->missing_records += seq_cmp(
@@ -793 +852 @@
                 }
                 rt->sources[i].expectedseq = ntohl(encaphdr->seqno) + 1;
+                if (rt->sources[i].expectedseq == 0) {
+                        rt->sources[i].expectedseq ++;
+                }
 
                 if (rt->sources[i].nextread == NULL) {

lib/format_ndag.h

@@ -39 +39 @@
 /* Encapsulation header -- used by both ENCAPERF and ENCAPRT records */
 typedef struct ndag_encap {
+        uint64_t started;
         uint32_t seqno;
         uint16_t streamid;