slot_data updated as an atomic var no

diff --git a/src/cuda-packet-batcher.c b/src/cuda-packet-batcher.c
index cf1a3cff69918679ef7091ebbe4962f7c8f58e97..054c0217227c64931c3934a195f02e70520b352e 100644 (file)
--- a/src/cuda-packet-batcher.c
+++ b/src/cuda-packet-batcher.c
@@ -323,13 +323,15 @@ void *SCCudaPBTmThreadsSlot1(void *td)
     SCLogDebug("%s starting", tv->name);
 
     if (s->SlotThreadInit != NULL) {
-        r = s->SlotThreadInit(tv, s->slot_initdata, &s->slot_data);
+        void *slot_data = NULL;
+        r = s->SlotThreadInit(tv, s->slot_initdata, &slot_data);
         if (r != TM_ECODE_OK) {
             EngineKill();
 
             TmThreadsSetFlag(tv, THV_CLOSED | THV_RUNNING_DONE);
             pthread_exit((void *) -1);
         }
+        SC_ATOMIC_CAS(&s->slot_data, SC_ATOMIC_GET(s->slot_data), slot_data);
     }
     memset(&s->slot_pre_pq, 0, sizeof(PacketQueue));
     memset(&s->slot_post_pq, 0, sizeof(PacketQueue));
@@ -352,9 +354,9 @@ void *SCCudaPBTmThreadsSlot1(void *td)
              * the Batcher TM(which is waiting on a cond from the previous
              * feeder TM).  Please handle the NULL packet case in the
              * function that you now call */
-            r = s->SlotFunc(tv, p, s->slot_data, NULL, NULL);
+            r = s->SlotFunc(tv, p, SC_ATOMIC_GET(s->slot_data), NULL, NULL);
         } else {
-            r = s->SlotFunc(tv, p, s->slot_data, NULL, NULL);
+            r = s->SlotFunc(tv, p, SC_ATOMIC_GET(s->slot_data), NULL, NULL);
             /* handle error */
             if (r == TM_ECODE_FAILED) {
                 TmqhOutputPacketpool(tv, p);
@@ -376,11 +378,11 @@ void *SCCudaPBTmThreadsSlot1(void *td)
     TmThreadWaitForFlag(tv, THV_DEINIT);
 
     if (s->SlotThreadExitPrintStats != NULL) {
-        s->SlotThreadExitPrintStats(tv, s->slot_data);
+        s->SlotThreadExitPrintStats(tv, SC_ATOMIC_GET(s->slot_data));
     }
 
     if (s->SlotThreadDeinit != NULL) {
-        r = s->SlotThreadDeinit(tv, s->slot_data);
+        r = s->SlotThreadDeinit(tv, SC_ATOMIC_GET(s->slot_data));
         if (r != TM_ECODE_OK) {
             TmThreadsSetFlag(tv, THV_CLOSED);
             pthread_exit((void *) -1);

diff --git a/src/flow-timeout.c b/src/flow-timeout.c
index 3b8858443ab36fbcc6ba2dbbef3e85a25342d04b..c481146f968c4c5305ad2277620cdfc869de7ff6 100644 (file)
--- a/src/flow-timeout.c
+++ b/src/flow-timeout.c
@@ -487,7 +487,7 @@ static inline void FlowForceReassemblyForHash(void)
 
             /* ah ah!  We have some unattended toserver segments */
             if ((client_ok = StreamHasUnprocessedSegments(ssn, 0)) == 1) {
-                StreamTcpThread *stt = stream_pseudo_pkt_stream_tm_slot->slot_data;
+                StreamTcpThread *stt = SC_ATOMIC_GET(stream_pseudo_pkt_stream_tm_slot->slot_data);
 
                 ssn->client.last_ack = (ssn->client.seg_list_tail->seq +
                         ssn->client.seg_list_tail->payload_len);
@@ -504,7 +504,7 @@ static inline void FlowForceReassemblyForHash(void)
             }
             /* oh oh!  We have some unattended toclient segments */
             if ((server_ok = StreamHasUnprocessedSegments(ssn, 1)) == 1) {
-                StreamTcpThread *stt = stream_pseudo_pkt_stream_tm_slot->slot_data;
+                StreamTcpThread *stt = SC_ATOMIC_GET(stream_pseudo_pkt_stream_tm_slot->slot_data);
 
                 ssn->server.last_ack = (ssn->server.seg_list_tail->seq +
                         ssn->server.seg_list_tail->payload_len);
@@ -546,8 +546,8 @@ static inline void FlowForceReassemblyForHash(void)
                 } else {
                     TmSlot *s = stream_pseudo_pkt_detect_tm_slot;
                     while (s != NULL) {
-                        s->SlotFunc(NULL, p, s->slot_data, &s->slot_pre_pq,
-                                &s->slot_post_pq);
+                        s->SlotFunc(NULL, p, SC_ATOMIC_GET(s->slot_data), &s->slot_pre_pq,
+                                    &s->slot_post_pq);
                         s = s->slot_next;
                     }
 
@@ -575,8 +575,8 @@ static inline void FlowForceReassemblyForHash(void)
                 } else {
                     TmSlot *s = stream_pseudo_pkt_detect_tm_slot;
                     while (s != NULL) {
-                        s->SlotFunc(NULL, p, s->slot_data, &s->slot_pre_pq,
-                                &s->slot_post_pq);
+                        s->SlotFunc(NULL, p, SC_ATOMIC_GET(s->slot_data), &s->slot_pre_pq,
+                                    &s->slot_post_pq);
                         s = s->slot_next;
                     }
 

diff --git a/src/tm-threads.c b/src/tm-threads.c
index 1f493bb7fbc5fb65febd68f8c8351db04aeadbed..fdf65b46ad9a0bb73fc4e68746365d5080c016e6 100644 (file)
--- a/src/tm-threads.c
+++ b/src/tm-threads.c
@@ -136,13 +136,15 @@ void *TmThreadsSlot1NoIn(void *td)
         TmThreadSetupOptions(tv);
 
     if (s->SlotThreadInit != NULL) {
-        r = s->SlotThreadInit(tv, s->slot_initdata, &s->slot_data);
+        void *slot_data = NULL;
+        r = s->SlotThreadInit(tv, s->slot_initdata, &slot_data);
         if (r != TM_ECODE_OK) {
             EngineKill();
 
             TmThreadsSetFlag(tv, THV_CLOSED | THV_RUNNING_DONE);
             pthread_exit((void *) -1);
         }
+        SC_ATOMIC_CAS(&s->slot_data, SC_ATOMIC_GET(s->slot_data), slot_data);
     }
     memset(&s->slot_pre_pq, 0, sizeof(PacketQueue));
     memset(&s->slot_post_pq, 0, sizeof(PacketQueue));
@@ -152,7 +154,7 @@ void *TmThreadsSlot1NoIn(void *td)
     while (run) {
         TmThreadTestThreadUnPaused(tv);
 
-        r = s->SlotFunc(tv, NULL, s->slot_data, &s->slot_pre_pq, &s->slot_post_pq);
+        r = s->SlotFunc(tv, NULL, SC_ATOMIC_GET(s->slot_data), &s->slot_pre_pq, &s->slot_post_pq);
 
         /* handle error */
         if (r == TM_ECODE_FAILED) {
@@ -194,11 +196,11 @@ void *TmThreadsSlot1NoIn(void *td)
     TmThreadWaitForFlag(tv, THV_DEINIT);
 
     if (s->SlotThreadExitPrintStats != NULL) {
-        s->SlotThreadExitPrintStats(tv, s->slot_data);
+        s->SlotThreadExitPrintStats(tv, SC_ATOMIC_GET(s->slot_data));
     }
 
     if (s->SlotThreadDeinit != NULL) {
-        r = s->SlotThreadDeinit(tv, s->slot_data);
+        r = s->SlotThreadDeinit(tv, SC_ATOMIC_GET(s->slot_data));
         if (r != TM_ECODE_OK) {
             TmThreadsSetFlag(tv, THV_CLOSED);
             pthread_exit((void *) -1);
@@ -227,13 +229,15 @@ void *TmThreadsSlot1NoOut(void *td)
         TmThreadSetupOptions(tv);
 
     if (s->SlotThreadInit != NULL) {
-        r = s->SlotThreadInit(tv, s->slot_initdata, &s->slot_data);
+        void *slot_data = NULL;
+        r = s->SlotThreadInit(tv, s->slot_initdata, &slot_data);
         if (r != TM_ECODE_OK) {
             EngineKill();
 
             TmThreadsSetFlag(tv, THV_CLOSED | THV_RUNNING_DONE);
             pthread_exit((void *) -1);
         }
+        SC_ATOMIC_CAS(&s->slot_data, SC_ATOMIC_GET(s->slot_data), slot_data);
     }
     memset(&s->slot_pre_pq, 0, sizeof(PacketQueue));
     memset(&s->slot_post_pq, 0, sizeof(PacketQueue));
@@ -246,7 +250,7 @@ void *TmThreadsSlot1NoOut(void *td)
         p = tv->tmqh_in(tv);
 
         PACKET_PROFILING_TMM_START(p, s->tm_id);
-        r = s->SlotFunc(tv, p, s->slot_data, /* no outqh no pq */ NULL,
+        r = s->SlotFunc(tv, p, SC_ATOMIC_GET(s->slot_data), /* no outqh no pq */ NULL,
                         /* no outqh no pq */ NULL);
         PACKET_PROFILING_TMM_END(p, s->tm_id);
 
@@ -267,11 +271,11 @@ void *TmThreadsSlot1NoOut(void *td)
     TmThreadWaitForFlag(tv, THV_DEINIT);
 
     if (s->SlotThreadExitPrintStats != NULL) {
-        s->SlotThreadExitPrintStats(tv, s->slot_data);
+        s->SlotThreadExitPrintStats(tv, SC_ATOMIC_GET(s->slot_data));
     }
 
     if (s->SlotThreadDeinit != NULL) {
-        r = s->SlotThreadDeinit(tv, s->slot_data);
+        r = s->SlotThreadDeinit(tv, SC_ATOMIC_GET(s->slot_data));
         if (r != TM_ECODE_OK) {
             TmThreadsSetFlag(tv, THV_CLOSED);
             pthread_exit((void *) -1);
@@ -301,13 +305,15 @@ void *TmThreadsSlot1NoInOut(void *td)
     SCLogDebug("%s starting", tv->name);
 
     if (s->SlotThreadInit != NULL) {
-        r = s->SlotThreadInit(tv, s->slot_initdata, &s->slot_data);
+        void *slot_data = NULL;
+        r = s->SlotThreadInit(tv, s->slot_initdata, &slot_data);
         if (r != TM_ECODE_OK) {
             EngineKill();
 
             TmThreadsSetFlag(tv, THV_CLOSED | THV_RUNNING_DONE);
             pthread_exit((void *) -1);
         }
+        SC_ATOMIC_CAS(&s->slot_data, SC_ATOMIC_GET(s->slot_data), slot_data);
     }
     memset(&s->slot_pre_pq, 0, sizeof(PacketQueue));
     memset(&s->slot_post_pq, 0, sizeof(PacketQueue));
@@ -317,7 +323,7 @@ void *TmThreadsSlot1NoInOut(void *td)
     while (run) {
         TmThreadTestThreadUnPaused(tv);
 
-        r = s->SlotFunc(tv, NULL, s->slot_data, /* no outqh, no pq */NULL, NULL);
+        r = s->SlotFunc(tv, NULL, SC_ATOMIC_GET(s->slot_data), /* no outqh, no pq */NULL, NULL);
 
         /* handle error */
         if (r == TM_ECODE_FAILED) {
@@ -335,11 +341,11 @@ void *TmThreadsSlot1NoInOut(void *td)
     TmThreadWaitForFlag(tv, THV_DEINIT);
 
     if (s->SlotThreadExitPrintStats != NULL) {
-        s->SlotThreadExitPrintStats(tv, s->slot_data);
+        s->SlotThreadExitPrintStats(tv, SC_ATOMIC_GET(s->slot_data));
     }
 
     if (s->SlotThreadDeinit != NULL) {
-        r = s->SlotThreadDeinit(tv, s->slot_data);
+        r = s->SlotThreadDeinit(tv, SC_ATOMIC_GET(s->slot_data));
         if (r != TM_ECODE_OK) {
             TmThreadsSetFlag(tv, THV_CLOSED);
             pthread_exit((void *) -1);
@@ -370,13 +376,15 @@ void *TmThreadsSlot1(void *td)
     SCLogDebug("%s starting", tv->name);
 
     if (s->SlotThreadInit != NULL) {
-        r = s->SlotThreadInit(tv, s->slot_initdata, &s->slot_data);
+        void *slot_data = NULL;
+        r = s->SlotThreadInit(tv, s->slot_initdata, &slot_data);
         if (r != TM_ECODE_OK) {
             EngineKill();
 
             TmThreadsSetFlag(tv, THV_CLOSED | THV_RUNNING_DONE);
             pthread_exit((void *) -1);
         }
+        SC_ATOMIC_CAS(&s->slot_data, SC_ATOMIC_GET(s->slot_data), slot_data);
     }
     memset(&s->slot_pre_pq, 0, sizeof(PacketQueue));
     SCMutexInit(&s->slot_pre_pq.mutex_q, NULL);
@@ -392,7 +400,7 @@ void *TmThreadsSlot1(void *td)
 
         if (p != NULL) {
             PACKET_PROFILING_TMM_START(p, s->tm_id);
-            r = s->SlotFunc(tv, p, s->slot_data, &s->slot_pre_pq,
+            r = s->SlotFunc(tv, p, SC_ATOMIC_GET(s->slot_data), &s->slot_pre_pq,
                             &s->slot_post_pq);
             PACKET_PROFILING_TMM_END(p, s->tm_id);
 
@@ -442,11 +450,11 @@ void *TmThreadsSlot1(void *td)
     TmThreadWaitForFlag(tv, THV_DEINIT);
 
     if (s->SlotThreadExitPrintStats != NULL) {
-        s->SlotThreadExitPrintStats(tv, s->slot_data);
+        s->SlotThreadExitPrintStats(tv, SC_ATOMIC_GET(s->slot_data));
     }
 
     if (s->SlotThreadDeinit != NULL) {
-        r = s->SlotThreadDeinit(tv, s->slot_data);
+        r = s->SlotThreadDeinit(tv, SC_ATOMIC_GET(s->slot_data));
         if (r != TM_ECODE_OK) {
             TmThreadsSetFlag(tv, THV_CLOSED);
             pthread_exit((void *) -1);
@@ -474,9 +482,9 @@ TmEcode TmThreadsSlotVarRun(ThreadVars *tv, Packet *p,
         PACKET_PROFILING_TMM_START(p, s->tm_id);
 
         if (unlikely(s->id == 0)) {
-            r = s->SlotFunc(tv, p, s->slot_data, &s->slot_pre_pq, &s->slot_post_pq);
+            r = s->SlotFunc(tv, p, SC_ATOMIC_GET(s->slot_data), &s->slot_pre_pq, &s->slot_post_pq);
         } else {
-            r = s->SlotFunc(tv, p, s->slot_data, &s->slot_pre_pq, NULL);
+            r = s->SlotFunc(tv, p, SC_ATOMIC_GET(s->slot_data), &s->slot_pre_pq, NULL);
         }
 
         PACKET_PROFILING_TMM_END(p, s->tm_id);
@@ -579,13 +587,15 @@ void *TmThreadsSlotPktAcqLoop(void *td) {
 
     for (slot = s; slot != NULL; slot = slot->slot_next) {
         if (slot->SlotThreadInit != NULL) {
-            r = slot->SlotThreadInit(tv, slot->slot_initdata, &slot->slot_data);
+            void *slot_data = NULL;
+            r = slot->SlotThreadInit(tv, slot->slot_initdata, &slot_data);
             if (r != TM_ECODE_OK) {
                 EngineKill();
 
                 TmThreadsSetFlag(tv, THV_CLOSED | THV_RUNNING_DONE);
                 pthread_exit((void *) -1);
             }
+            SC_ATOMIC_CAS(&slot->slot_data, SC_ATOMIC_GET(slot->slot_data), slot_data);
         }
         memset(&slot->slot_pre_pq, 0, sizeof(PacketQueue));
         SCMutexInit(&slot->slot_pre_pq.mutex_q, NULL);
@@ -598,7 +608,7 @@ void *TmThreadsSlotPktAcqLoop(void *td) {
     while(run) {
         TmThreadTestThreadUnPaused(tv);
 
-        r = s->PktAcqLoop(tv, s->slot_data, s);
+        r = s->PktAcqLoop(tv, SC_ATOMIC_GET(s->slot_data), s);
 
         if (r == TM_ECODE_FAILED || TmThreadsCheckFlag(tv, THV_KILL)) {
             run = 0;
@@ -611,11 +621,11 @@ void *TmThreadsSlotPktAcqLoop(void *td) {
 
     for (slot = s; slot != NULL; slot = slot->slot_next) {
         if (slot->SlotThreadExitPrintStats != NULL) {
-            slot->SlotThreadExitPrintStats(tv, slot->slot_data);
+            slot->SlotThreadExitPrintStats(tv, SC_ATOMIC_GET(slot->slot_data));
         }
 
         if (slot->SlotThreadDeinit != NULL) {
-            r = slot->SlotThreadDeinit(tv, slot->slot_data);
+            r = slot->SlotThreadDeinit(tv, SC_ATOMIC_GET(slot->slot_data));
             if (r != TM_ECODE_OK) {
                 TmThreadsSetFlag(tv, THV_CLOSED);
                 pthread_exit((void *) -1);
@@ -660,13 +670,15 @@ void *TmThreadsSlotVar(void *td)
 
     for (; s != NULL; s = s->slot_next) {
         if (s->SlotThreadInit != NULL) {
-            r = s->SlotThreadInit(tv, s->slot_initdata, &s->slot_data);
+            void *slot_data = NULL;
+            r = s->SlotThreadInit(tv, s->slot_initdata, &slot_data);
             if (r != TM_ECODE_OK) {
                 EngineKill();
 
                 TmThreadsSetFlag(tv, THV_CLOSED | THV_RUNNING_DONE);
                 pthread_exit((void *) -1);
             }
+            SC_ATOMIC_CAS(&s->slot_data, SC_ATOMIC_GET(s->slot_data), slot_data);
         }
         memset(&s->slot_pre_pq, 0, sizeof(PacketQueue));
         SCMutexInit(&s->slot_pre_pq.mutex_q, NULL);
@@ -741,11 +753,11 @@ void *TmThreadsSlotVar(void *td)
 
     for ( ; s != NULL; s = s->slot_next) {
         if (s->SlotThreadExitPrintStats != NULL) {
-            s->SlotThreadExitPrintStats(tv, s->slot_data);
+            s->SlotThreadExitPrintStats(tv, SC_ATOMIC_GET(s->slot_data));
         }
 
         if (s->SlotThreadDeinit != NULL) {
-            r = s->SlotThreadDeinit(tv, s->slot_data);
+            r = s->SlotThreadDeinit(tv, SC_ATOMIC_GET(s->slot_data));
             if (r != TM_ECODE_OK) {
                 TmThreadsSetFlag(tv, THV_CLOSED);
                 pthread_exit((void *) -1);
@@ -850,6 +862,7 @@ void TmSlotSetFuncAppend(ThreadVars *tv, TmModule *tm, void *data)
     if (slot == NULL)
         return;
     memset(slot, 0, sizeof(TmSlot));
+    SC_ATOMIC_INIT(slot->slot_data);
     slot->tv = tv;
     slot->SlotThreadInit = tm->ThreadInit;
     slot->slot_initdata = data;

diff --git a/src/tm-threads.h b/src/tm-threads.h
index ac2804bf69f960926b5b5f7cbb2e4d6eeb3e66c0..eecd96937c8127e08a9a5b20aeb38b4590922ad9 100644 (file)
--- a/src/tm-threads.h
+++ b/src/tm-threads.h
@@ -45,7 +45,7 @@ typedef struct TmSlot_ {
 
     /* data storage */
     void *slot_initdata;
-    void *slot_data;
+    SC_ATOMIC_DECLARE(void *, slot_data);
 
     /* queue filled by the SlotFunc with packets that will
      * be processed futher _before_ the current packet.

diff --git a/src/util-mpm-b2g-cuda.c b/src/util-mpm-b2g-cuda.c
index 37756ae3af0dc256f5c3da78b012e23d5f27e69c..15a944a2139206421d0c64eba824ff56574bd076 100644 (file)
--- a/src/util-mpm-b2g-cuda.c
+++ b/src/util-mpm-b2g-cuda.c
@@ -2237,18 +2237,20 @@ void *CudaMpmB2gThreadsSlot1(void *td)
     SCLogDebug("%s starting", tv->name);
 
     if (s->SlotThreadInit != NULL) {
-        r = s->SlotThreadInit(tv, s->slot_initdata, &s->slot_data);
+        void *slot_data = NULL;
+        r = s->SlotThreadInit(tv, s->slot_initdata, &slot_data);
         if (r != TM_ECODE_OK) {
             EngineKill();
 
             TmThreadsSetFlag(tv, THV_CLOSED | THV_RUNNING_DONE);
             pthread_exit((void *) -1);
         }
+        SC_ATOMIC_CAS(&s->slot_data, SC_ATOMIC_GET(s->slot_data), slot_data);
     }
     memset(&s->slot_pre_pq, 0, sizeof(PacketQueue));
     memset(&s->slot_post_pq, 0, sizeof(PacketQueue));
 
-    tctx = (B2gCudaMpmThreadCtxData *)s->slot_data;
+    tctx = (B2gCudaMpmThreadCtxData *)SC_ATOMIC_GET(s->slot_data);
 
     TmThreadsSetFlag(tv, THV_INIT_DONE);
     while(run) {
@@ -2294,11 +2296,11 @@ void *CudaMpmB2gThreadsSlot1(void *td)
     TmThreadWaitForFlag(tv, THV_DEINIT);
 
     if (s->SlotThreadExitPrintStats != NULL) {
-        s->SlotThreadExitPrintStats(tv, s->slot_data);
+        s->SlotThreadExitPrintStats(tv, SC_ATOMIC_GET(s->slot_data));
     }
 
     if (s->SlotThreadDeinit != NULL) {
-        r = s->SlotThreadDeinit(tv, s->slot_data);
+        r = s->SlotThreadDeinit(tv, SC_ATOMIC_GET(s->slot_data));
         if (r != TM_ECODE_OK) {
             TmThreadsSetFlag(tv, THV_CLOSED);
             pthread_exit((void *) -1);