Attempt to work around NULL packets we're seeing ending up in queues when the compiler has optimized our code.

src/packet-queue.c

@@ -134,6 +134,9 @@ void PacketQueueValidate(PacketQueue *q) {
 void PacketEnqueue (PacketQueue *q, Packet *p) {
     //PacketQueueValidateDebug(q);
 
+    if (p == NULL)
+        return;
+
     /* more packets in queue */
     if (q->top != NULL) {
         p->prev = NULL;
@@ -164,6 +167,8 @@ Packet *PacketDequeue (PacketQueue *q) {
         return NULL;
     }
 
+    q->len--;
+
     /* pull the bottom packet from the queue */
     p = q->bot;
     /* Weird issue: sometimes it looks that two thread arrive
@@ -183,8 +188,6 @@ Packet *PacketDequeue (PacketQueue *q) {
         q->bot = NULL;
     }
 
-    q->len--;
-
     //PacketQueueValidateDebug(q);
     return p;
 }

src/source-pcap-file.c

@@ -160,24 +160,39 @@ TmEcode ReceivePcapFile(ThreadVars *tv, Packet *p, void *data, PacketQueue *pq,
         SCReturnInt(TM_ECODE_FAILED);
     }
 
+    if (postpq == NULL) {
+        pcap_max_read_packets = 1;
+    }
+
+    ptv->array_idx = 0;
+    ptv->in_p = p;
+
     /* make sure we have at least one packet in the packet pool, to prevent
      * us from alloc'ing packets at line rate */
-    while (packet_q_len == 0) {
+    do {
         packet_q_len = PacketPoolSize();
         if (packet_q_len == 0) {
             PacketPoolWait();
         }
-    }
-
-    if (postpq == NULL)
-        pcap_max_read_packets = 1;
-
-    ptv->array_idx = 0;
-    ptv->in_p = p;
+    } while (packet_q_len == 0);
 
     /* Right now we just support reading packets one at a time. */
     int r = pcap_dispatch(pcap_g.pcap_handle, (pcap_max_read_packets < packet_q_len) ? pcap_max_read_packets : packet_q_len,
             (pcap_handler)PcapFileCallback, (u_char *)ptv);
+    if (r < 0) {
+        SCLogError(SC_ERR_PCAP_DISPATCH, "error code %" PRId32 " %s",
+                r, pcap_geterr(pcap_g.pcap_handle));
+
+        EngineStop();
+        SCReturnInt(TM_ECODE_FAILED);
+    } else if (r == 0) {
+        SCLogInfo("pcap file end of file reached (pcap err code %" PRId32 ")", r);
+
+        EngineStop();
+        ptv->done = 1;
+
+        /* fall through */
+    }
 
     uint16_t cnt = 0;
     for (cnt = 0; cnt < ptv->array_idx; cnt++) {
@@ -193,20 +208,6 @@ TmEcode ReceivePcapFile(ThreadVars *tv, Packet *p, void *data, PacketQueue *pq,
         }
     }
 
-    if (r < 0) {
-        SCLogError(SC_ERR_PCAP_DISPATCH, "error code %" PRId32 " %s",
-                r, pcap_geterr(pcap_g.pcap_handle));
-
-        EngineStop();
-        SCReturnInt(TM_ECODE_FAILED);
-    } else if (r == 0) {
-        SCLogInfo("pcap file end of file reached (pcap err code %" PRId32 ")", r);
-
-        EngineStop();
-        ptv->done = 1;
-        SCReturnInt(TM_ECODE_OK);
-    }
-
     SCReturnInt(TM_ECODE_OK);
 }
 

src/tm-threads.c

@@ -196,17 +196,21 @@ void *TmThreadsSlot1NoIn(void *td) {
         }
 
         /* handle pre queue */
-        while (s->s.slot_pre_pq.len > 0) {
-            Packet *extra = PacketDequeue(&s->s.slot_pre_pq);
-            tv->tmqh_out(tv, extra);
+        while (s->s.slot_pre_pq.top != NULL) {
+            Packet *extra_p = PacketDequeue(&s->s.slot_pre_pq);
+            if (extra_p != NULL) {
+                tv->tmqh_out(tv, extra_p);
+            }
         }
 
         tv->tmqh_out(tv, p);
 
         /* handle post queue */
-        while (s->s.slot_post_pq.len > 0) {
-            Packet *extra = PacketDequeue(&s->s.slot_post_pq);
-            tv->tmqh_out(tv, extra);
+        while (s->s.slot_post_pq.top != NULL) {
+            Packet *extra_p = PacketDequeue(&s->s.slot_post_pq);
+            if (extra_p != NULL) {
+                tv->tmqh_out(tv, extra_p);
+            }
         }
 
         if (TmThreadsCheckFlag(tv, THV_KILL)) {
@@ -413,7 +417,7 @@ void *TmThreadsSlot1(void *td) {
                 break;
             }
 
-            while (s->s.slot_pre_pq.len > 0) {
+            while (s->s.slot_pre_pq.top != NULL) {
                 /* handle new packets from this func */
                 Packet *extra_p = PacketDequeue(&s->s.slot_pre_pq);
                 if (extra_p != NULL) {
@@ -424,7 +428,7 @@ void *TmThreadsSlot1(void *td) {
             /* output the packet */
             tv->tmqh_out(tv, p);
 
-            while (s->s.slot_post_pq.len > 0) {
+            while (s->s.slot_post_pq.top != NULL) {
                 /* handle new packets from this func */
                 Packet *extra_p = PacketDequeue(&s->s.slot_post_pq);
                 if (extra_p != NULL) {
@@ -481,7 +485,7 @@ static inline TmEcode TmThreadsSlotVarRun (ThreadVars *tv, Packet *p, TmSlot *sl
         }
 
         /* handle new packets */
-        while (s->slot_pre_pq.len > 0) {
+        while (s->slot_pre_pq.top != NULL) {
             Packet *extra_p = PacketDequeue(&s->slot_pre_pq);
             if (extra_p == NULL)
                 continue;
@@ -572,7 +576,7 @@ void *TmThreadsSlotVar(void *td) {
             tv->tmqh_out(tv, p);
 
             /* now handle the post_pq packets */
-            while (s->s->slot_post_pq.len > 0) {
+            while (s->s->slot_post_pq.top != NULL) {
                 Packet *extra_p = PacketDequeue(&s->s->slot_post_pq);
                 if (extra_p == NULL)
                     continue;