Changeset d24b1df for lib

Timestamp:

11/19/18 11:26:13 (2 years ago)

Author:

Shane Alcock <salcock@…>

Branches:

develop

Children:

ffae0a5

Parents:

1e6d795

Message:

Tidy up source tracking in etsilive: input format

• Attempt to re-use any "old" source entries that have been closed.
• Fix misleading source count output when a new source connects.

File:

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

lib/format_etsilive.c

@@ -72 +72 @@
         etsisocket_t *sources;
         uint16_t sourcecount;
+        uint16_t sourcealloc;
+        uint16_t activesources;
         int threadindex;
         wandder_etsispec_t *etsidec;
@@ -269 +271 @@
 
                 FORMAT_DATA->receivers[i].sources = NULL;
+                FORMAT_DATA->receivers[i].sourcealloc = 0;
                 FORMAT_DATA->receivers[i].sourcecount = 0;
+                FORMAT_DATA->receivers[i].activesources = 0;
                 FORMAT_DATA->receivers[i].threadindex = i;
                 FORMAT_DATA->receivers[i].etsidec =
@@ -326 +330 @@
                         != LIBTRACE_MQ_FAILED) {
                 etsisocket_t *esock = NULL;
+                int i;
 
                 if (et->sourcecount == 0) {
                         et->sources = (etsisocket_t *)malloc(
                                         sizeof(etsisocket_t) * 10);
-                } else if ((et->sourcecount % 10) == 0) {
+                        et->sourcealloc = 10;
+
+                        for (i = 0; i < et->sourcealloc; i++) {
+                                et->sources[i].sock = -1;
+                                et->sources[i].srcaddr = NULL;
+                        }
+
+                        esock = &(et->sources[0]);
+                        et->sourcecount = 1;
+                } else {
+                        for (i = 0; i < et->sourcealloc; i++) {
+                                if (et->sources[i].sock == -1) {
+                                        esock = &(et->sources[i]);
+                                        break;
+                                }
+                        }
+                }
+
+                if (esock == NULL) {
                         et->sources = (etsisocket_t *)realloc(et->sources,
-                                sizeof(etsisocket_t) * (et->sourcecount + 10));
-                }
-
-                esock = &(et->sources[et->sourcecount]);
+                                sizeof(etsisocket_t) * (et->sourcealloc + 10));
+
+                        esock = &(et->sources[et->sourcealloc]);
+                        et->sourcealloc += 10;
+                        et->sourcecount += 1;
+
+                }
+
                 esock->sock = msg.recvsock;
                 esock->srcaddr = msg.recvaddr;
@@ -343 +370 @@
                 esock->cached.length = 0;
 
-                et->sourcecount += 1;
+                et->activesources += 1;
 
                 fprintf(stderr, "Thread %d is now handling %u sources.\n",
-                                et->threadindex, et->sourcecount);
+                                et->threadindex, et->activesources);
         }
         return 1;
 }
 
-static void receive_from_single_socket(etsisocket_t *esock) {
+static void receive_from_single_socket(etsisocket_t *esock, etsithread_t *et) {
 
         int ret = 0;
@@ -361 +388 @@
         ret = libtrace_scb_recv_sock(&(esock->recvbuffer), esock->sock,
                         MSG_DONTWAIT);
-        if (ret == -1) {
+        if (ret < 0) {
                 if (errno == EAGAIN || errno == EWOULDBLOCK) {
                         /* Would have blocked, nothing available */
@@ -370 +397 @@
                 close(esock->sock);
                 esock->sock = -1;
+                et->activesources -= 1;
         }
 
@@ -376 +404 @@
                 close(esock->sock);
                 esock->sock = -1;
+                et->activesources -= 1;
         }
 
@@ -394 +423 @@
         }
 
-        if (et->sourcecount == 0) {
+        if (et->activesources == 0) {
                 return 1;
         }
 
         for (i = 0; i < et->sourcecount; i++) {
-                receive_from_single_socket(&(et->sources[i]));
+                receive_from_single_socket(&(et->sources[i]), et);
         }
         return 1;
@@ -407 +436 @@
 static inline void inspect_next_packet(etsisocket_t *sock,
                 etsisocket_t **earliestsock, uint64_t *earliesttime,
-                wandder_etsispec_t *dec) {
+                wandder_etsispec_t *dec, etsithread_t *et) {
 
 
@@ -460 +489 @@
                         close(sock->sock);
                         sock->sock = -1;
+                        et->activesources -= 1;
                         return;
                 }
@@ -467 +497 @@
                         close(sock->sock);
                         sock->sock = -1;
+                        et->activesources -= 1;
                         return;
                 }
@@ -505 +536 @@
         for (i = 0; i < et->sourcecount; i++) {
                 inspect_next_packet(&(et->sources[i]), &esock, &earliest,
-                        et->etsidec);
+                        et->etsidec, et);
         }
         return esock;