leak 1

levb · Jul 11, 2024 · cca5f3e · cca5f3e
1 parent d7b458c
commit cca5f3e
Show file tree

Hide file tree

Showing 3 changed files with 10 additions and 12 deletions.
diff --git a/src/dispatch.h b/src/dispatch.h
@@ -35,7 +35,6 @@ typedef struct __natsDispatcher_s
     natsDispatchQueue queue;
 
     // Flags.
-    unsigned running : 1;
     unsigned shutdown : 1;
 } natsDispatcher;
 

diff --git a/src/dispatch_pool.c b/src/dispatch_pool.c
@@ -30,7 +30,7 @@ static natsSharedDispatchers _gReply;
 
 static inline void _cleanupSharedDispatcher(natsDispatcher *d)
 {
-    if ((d == NULL) || !d->running)
+    if ((d == NULL) || d->thread == NULL)
         return;
 
     natsThread_Destroy(d->thread); // has already been joined, no need to detach.
@@ -90,8 +90,6 @@ static natsStatus _runSharedDispatcher(natsDispatcher *d, bool forReplies)
 {
     natsStatus s = NATS_OK;
 
-    d->running = true;
-
     natsLib_Retain();
 
     s = natsMutex_Create(&d->mu);
@@ -182,7 +180,7 @@ nats_assignSubToSharedDispatcher(natsSubscription *sub, bool forReplies)
 
     // Get a running dispatcher.
     d = &g->dispatchers[g->useNext];
-    if (!d->running)
+    if (d->thread == NULL)
     {
         s = _runSharedDispatcher(d, forReplies);
         if (s == NATS_OK)

diff --git a/src/sub.c b/src/sub.c
@@ -124,12 +124,13 @@ _initOwnDispatcher(natsSubscription *sub)
 
 static inline void _cleanupOwnDispatcher(natsSubscription *sub)
 {
-    if (!sub->ownDispatcher.running)
-        return;
-
     printf("<>/<> _cleanupOwnDispatcher\n");
-    natsThread_Detach(sub->ownDispatcher.thread);
-    natsThread_Destroy(sub->ownDispatcher.thread);
+    if (sub->ownDispatcher.thread != NULL)
+    {
+        natsThread_Detach(sub->ownDispatcher.thread);
+        natsThread_Destroy(sub->ownDispatcher.thread);
+        sub->ownDispatcher.thread = NULL;
+    }
 
     nats_destroyQueuedMessages(&sub->ownDispatcher.queue);
     natsCondition_Destroy(sub->ownDispatcher.cond);
@@ -301,7 +302,7 @@ void natsSub_close(natsSubscription *sub, bool connectionClosed)
 static void
 _asyncTimeoutCb(natsTimer *timer, void *closure)
 {
-    natsSubscription *sub = (natsSubscription*) closure;
+    natsSubscription *sub = (natsSubscription *)closure;
 
     // Should not happen, but in case
     if (sub->dispatcher == NULL)
@@ -318,7 +319,7 @@ _asyncTimeoutCb(natsTimer *timer, void *closure)
 
         // Set the timer to a very high value, it will be reset from the
         // worker thread.
-        natsTimer_Reset(sub->timeoutTimer, 60*60*1000);
+        natsTimer_Reset(sub->timeoutTimer, 60 * 60 * 1000);
 
         // Post a control message to the worker thread.
         printf("<>/<> natsSub_timeoutCb - posting TIMEOUT message\n");