Make sure flow isn't freed while stream msgs are still in use.

src/flow.h

@@ -68,7 +68,8 @@ typedef struct Flow_
     u_int64_t bytecnt;
 
     void *stream;
-    u_int16_t use_cnt; /* how many pkts are using the flow *right now* */
+    u_int16_t use_cnt; /** how many pkts and stream msgs are
+                           using the flow *right now* */
 
     pthread_mutex_t m;
 

src/l7-app-detect.c

@@ -117,9 +117,11 @@ void *L7AppDetectThread(void *td)
                         printf("L7AppDetectThread: smsg not start, but no l7 data? Weird\n");
                     }
                 }
+            }
 
+            /* XXX we need to improve this logic */
+            smsg->flow->use_cnt--;
             mutex_unlock(&smsg->flow->m);
-            }
 
             /* return the used message to the queue */
             StreamMsgReturnToPool(smsg);

src/stream-tcp-reassemble.c

@@ -353,7 +353,8 @@ int StreamTcpReassembleHandleSegmentUpdateACK (TcpSession *ssn, TcpStream *strea
             printf("StreamTcpReassembleHandleSegmentUpdateACK: seg->seq %u, seg->payload_len %u, stream->ra_base_seq %u\n",
                 seg->seq, seg->payload_len, stream->ra_base_seq);
 
-            /* get a message */
+            /* get a message
+               XXX we need a setup function */
             if (smsg == NULL) {
                 smsg = StreamMsgGetFromPool();
                 if (smsg == NULL) {
@@ -369,6 +370,8 @@ int StreamTcpReassembleHandleSegmentUpdateACK (TcpSession *ssn, TcpStream *strea
                 }
                 smsg->data.data_len = 0;
                 smsg->flow = p->flow;
+                if (smsg->flow)
+                    smsg->flow->use_cnt++;
             }
 
             /* handle segments partly before ra_base_seq */
@@ -451,7 +454,8 @@ int StreamTcpReassembleHandleSegmentUpdateACK (TcpSession *ssn, TcpStream *strea
                     printf("StreamTcpReassembleHandleSegmentUpdateACK: "
                            "new msg at offset %u, payload_len %u\n", payload_offset, payload_len);
 
-                    /* get a new message */
+                    /* get a new message
+                       XXX we need a setup function */
                     smsg = StreamMsgGetFromPool();
                     if (smsg == NULL) {
                         printf("StreamTcpReassembleHandleSegmentUpdateACK: "
@@ -461,6 +465,8 @@ int StreamTcpReassembleHandleSegmentUpdateACK (TcpSession *ssn, TcpStream *strea
                     smsg_offset = 0;
                     smsg->data.data_len = 0;
                     smsg->flow = p->flow;
+                    if (smsg->flow)
+                        smsg->flow->use_cnt++;
 
                     copy_size = sizeof(smsg->data.data) - smsg_offset;
                     if (copy_size > (seg->payload_len - payload_offset)) {