HPCC-32422 Fix lookupjoin regression (introduced by HPCC-32313)

Signed-off-by: Jake Smith <[email protected]>
hpcc-systems · Aug 13, 2024 · 1e38c14 · 1e38c14
1 parent 9a44c2e
commit 1e38c14
Show file tree

Hide file tree

Showing 2 changed files with 11 additions and 2 deletions.
diff --git a/system/jlib/jqueue.tpp b/system/jlib/jqueue.tpp
@@ -358,6 +358,8 @@ public:
         stopped = false;
     }
 
+    bool queryStopped() const { return stopped; }
+
     bool enqueue(BASE *e,unsigned timeout=INFINITE) 
     { 
         unsigned numToSignal = 0;

diff --git a/thorlcr/activities/lookupjoin/thlookupjoinslave.cpp b/thorlcr/activities/lookupjoin/thlookupjoinslave.cpp
@@ -177,7 +177,9 @@ class CBroadcaster : public CSimpleInterface
 
             // in theory each worker could be sending log(n) packets, with the broadcaster on each blocking waiting for acks
             unsigned limit = nodes * std::ceil(std::log2(nodes));
-            broadcastQueue.setLimit(limit);
+            limit = broadcaster.activity.getOptInt("lookupJoinBroadcastQLimit", limit);
+            if (0 != limit)
+                broadcastQueue.setLimit(limit);
         }
         ~CSend()
         {
@@ -203,8 +205,13 @@ class CBroadcaster : public CSimpleInterface
                     return;
                 }
             }
-            while (!broadcastQueue.enqueue(sendItem, 5000)) // will block if queue full
+            while (!broadcastQueue.enqueue(sendItem, 20000)) // will timeout if queue full
             {
+                if (broadcastQueue.queryStopped())
+                {
+                    DBGLOG("broadcastQueue stopped");
+                    break;
+                }
                 DBGLOG("CSend::addBlock() - broadcastQueue full, waiting for space");
             }
         }