HPCC-15404 Optimize roxie standalone queries

Address race condition questions, and default to prestarting threads when
deployed via ConfigMgr. Standalone Roxie programs will NOT typically load
a RoxieTopology file and will default to not prestarting threads.

Signed-off-by: Richard Chapman <rchapman@hpccsystems.com>

initfiles/componentfiles/configxml/roxie.xsd.in

@@ -692,6 +692,13 @@
         </xs:appinfo>
       </xs:annotation>
     </xs:attribute>
+    <xs:attribute name="prestartSlaveThreads" type="xs:boolean" use="optional" default="true">
+      <xs:annotation>
+        <xs:appinfo>
+          <tooltip>Prestart slave worker threads at startup</tooltip>
+        </xs:appinfo>
+      </xs:annotation>
+    </xs:attribute>
     <xs:attribute name="reloadRetriesFailed" type="xs:boolean" use="optional" default="true">
       <xs:annotation>
         <xs:appinfo>

roxie/ccd/ccd.hpp

@@ -368,6 +368,7 @@ extern unsigned minFilesOpen[2];
 extern unsigned maxFilesOpen[2];
 extern unsigned restarts;
 extern bool checkCompleted;
+extern bool prestartSlaveThreads;
 extern unsigned preabortKeyedJoinsThreshold;
 extern unsigned preabortIndexReadsThreshold;
 extern bool traceStartStop;

roxie/ccd/ccdmain.cpp

@@ -55,6 +55,7 @@ unsigned highTimeout = 2000;
 unsigned slaTimeout = 2000;
 unsigned numServerThreads = 30;
 unsigned numSlaveThreads = 30;
+bool prestartSlaveThreads = false;
 unsigned numRequestArrayThreads = 5;
 unsigned headRegionSize;
 unsigned ccdMulticastPort;
@@ -760,6 +761,7 @@ int STARTQUERY_API start_query(int argc, const char *argv[])
         maxLockAttempts = topology->getPropInt("@maxLockAttempts", 5);
         enableHeartBeat = topology->getPropBool("@enableHeartBeat", true);
         checkCompleted = topology->getPropBool("@checkCompleted", true);
+        prestartSlaveThreads = topology->getPropBool("@prestartSlaveThreads", false);
         preabortKeyedJoinsThreshold = topology->getPropInt("@preabortKeyedJoinsThreshold", 100);
         preabortIndexReadsThreshold = topology->getPropInt("@preabortIndexReadsThreshold", 100);
         defaultMemoryLimit = (memsize_t) topology->getPropInt64("@defaultMemoryLimit", 0);

roxie/ccd/ccdqueue.cpp

@@ -728,6 +728,24 @@ class RoxieQueue : public CInterface, implements IThreadFactory
     unsigned numWorkers;
     unsigned started;
     std::atomic<unsigned> idle;
+
+    void noteQueued()
+    {
+        CriticalBlock b(counterCrit);
+        queueLength++;
+        if (queueLength>maxQueueLength)
+            maxQueueLength = queueLength;
+        // NOTE - there is a small race condition here - if idle is 1 but two enqueue's happen
+        // close enough together that the signal has not yet caused idle to come back down to zero, then the
+        // desired new thread may not be created. It's unlikely, and it's benign in that the query is still
+        // processed and the thread will be created next time the HWM is reached.
+        if (started < numWorkers && idle==0)
+        {
+            workers->start(this);
+            started++;
+        }
+    }
+
 public:
     IMPLEMENT_IINTERFACE;
 
@@ -745,7 +763,14 @@ public:
 
     void start()
     {
-        // We start on demand now...
+        if (prestartSlaveThreads)
+        {
+            while (started < numWorkers)
+            {
+                workers->start(this);
+                started++;
+            }
+        }
     }
 
     IPooledThreadIterator *running()
@@ -773,16 +798,7 @@ public:
             header.tick = msTick();
 #endif
             waiting.enqueue(x);
-
-            CriticalBlock b(counterCrit);
-            queueLength++;
-            if (queueLength>maxQueueLength)
-                maxQueueLength = queueLength;
-            if (idle==0 && started < numWorkers)
-            {
-                workers->start(this);
-                started++;
-            }
+            noteQueued();
         }
         available.signal();
     }
@@ -825,10 +841,7 @@ public:
                 l.CTXLOG("enqueued %s", header.toString(xx).str());
             }
             waiting.enqueue(x);
-            CriticalBlock b(counterCrit);
-            queueLength++;
-            if (queueLength>maxQueueLength)
-                maxQueueLength = queueLength;
+            noteQueued();
         }
         available.signal();
     }
@@ -1378,6 +1391,7 @@ IPooledThread *RoxieQueue::createNew()
 
 void RoxieQueue::abortChannel(unsigned channel)
 {
+    CriticalBlock b(counterCrit);
     Owned<IPooledThreadIterator> wi = workers->running();
     ForEach(*wi)
     {