HPCC-XXXXX Add option to avoid too many workers waiting on semaphore

Signed-off-by: Richard Chapman <[email protected]>
hpcc-systems · Oct 2, 2024 · 1d20f33 · 1d20f33
1 parent a22655b
commit 1d20f33
Show file tree

Hide file tree

Showing 3 changed files with 5 additions and 0 deletions.
diff --git a/roxie/ccd/ccd.hpp b/roxie/ccd/ccd.hpp
@@ -389,6 +389,7 @@ extern bool ignoreFileDateMismatches;
 extern bool ignoreFileSizeMismatches;
 extern int fileTimeFuzzySeconds;
 extern SinkMode defaultSinkMode;
+extern bool limitWaitingWorkers;
 
 #if defined(_CONTAINERIZED) || defined(SUBCHANNELS_IN_HEADER)
 static constexpr bool roxieMulticastEnabled = false;

diff --git a/roxie/ccd/ccdmain.cpp b/roxie/ccd/ccdmain.cpp
@@ -203,6 +203,7 @@ unsigned maxGraphLoopIterations;
 bool steppingEnabled = true;
 bool simpleLocalKeyedJoins = true;
 bool adhocRoxie = false;
+bool limitWaitingWorkers = false;
 
 unsigned __int64 minFreeDiskSpace = 1024 * 0x100000;  // default to 1 GB
 unsigned socketCheckInterval = 5000;
@@ -1279,6 +1280,7 @@ int CCD_API roxie_main(int argc, const char *argv[], const char * defaultYaml)
         const char *sinkModeText = topology->queryProp("@sinkMode");
         if (sinkModeText)
             defaultSinkMode = getSinkMode(sinkModeText);
+        limitWaitingWorkers = topology->getPropBool("@limitWaitingWorkers", limitWaitingWorkers);
 
         cacheReportPeriodSeconds = topology->getPropInt("@cacheReportPeriodSeconds", 5*60);
         setLegacyAES(topology->getPropBool("expert/@useLegacyAES", false));

diff --git a/roxie/ccd/ccdqueue.cpp b/roxie/ccd/ccdqueue.cpp
@@ -1149,6 +1149,7 @@ class RoxieQueue : public CInterface, implements IThreadFactory
     Owned <IThreadPool> workers;
     QueueOf<ISerializedRoxieQueryPacket, true> waiting;
     Semaphore available;
+    CriticalSection availCrit;    // Semaphore post may be slow with a lot of waiters - this crit may be used to limit to a single waiter
     CriticalSection qcrit;
     unsigned headRegionSize;
     unsigned numWorkers;
@@ -1316,6 +1317,7 @@ class RoxieQueue : public CInterface, implements IThreadFactory
     void wait()
     {
         idle++;
+        CLeavableCriticalBlock b(availCrit, limitWaitingWorkers);
         available.wait();
         idle--;
     }