HPCC-29831 Allow localAgent (and other) setting to be set in workunit…

…/regression suite These options are only applied when Roxie us run in workunit mode in containerized systems. Signed-off-by: Richard Chapman <[email protected]>
hpcc-systems · Jul 5, 2023 · 11f3ecc · 11f3ecc
1 parent 9e395ea
commit 11f3ecc
Show file tree

Hide file tree

Showing 6 changed files with 77 additions and 17 deletions.
diff --git a/common/workunit/workunit.cpp b/common/workunit/workunit.cpp
@@ -8458,7 +8458,9 @@ void CLocalWorkUnit::copyWorkUnit(IConstWorkUnit *cached, bool copyStats, bool a
 bool CLocalWorkUnit::hasDebugValue(const char *propname) const
 {
     StringBuffer lower;
-    lower.append(propname).toLowerCase();
+    lower.append(propname);
+    if (!strchr(lower, ':'))
+        lower.toLowerCase();
     CriticalBlock block(crit);
     StringBuffer prop("Debug/");
     return p->hasProp(prop.append(lower));
@@ -8467,7 +8469,9 @@ bool CLocalWorkUnit::hasDebugValue(const char *propname) const
 IStringVal& CLocalWorkUnit::getDebugValue(const char *propname, IStringVal &str) const
 {
     StringBuffer lower;
-    lower.append(propname).toLowerCase();
+    lower.append(propname);
+    if (!strchr(lower, ':'))
+        lower.toLowerCase();
     CriticalBlock block(crit);
     StringBuffer prop("Debug/");
     str.set(p->queryProp(prop.append(lower).str()));
@@ -8486,7 +8490,9 @@ IStringIterator& CLocalWorkUnit::getDebugValues(const char *prop) const
     if (prop)
     {
         StringBuffer lower;
-        lower.append(prop).toLowerCase();
+        lower.append(prop);
+        if (!strchr(lower, ':'))
+            lower.toLowerCase();
         path.append(lower);
     }
     else
@@ -8497,7 +8503,9 @@ IStringIterator& CLocalWorkUnit::getDebugValues(const char *prop) const
 int CLocalWorkUnit::getDebugValueInt(const char *propname, int defVal) const
 {
     StringBuffer lower;
-    lower.append(propname).toLowerCase();
+    lower.append(propname);
+    if (!strchr(lower, ':'))
+        lower.toLowerCase();
     CriticalBlock block(crit);
     StringBuffer prop("Debug/");
     prop.append(lower);
@@ -8507,7 +8515,9 @@ int CLocalWorkUnit::getDebugValueInt(const char *propname, int defVal) const
 __int64 CLocalWorkUnit::getDebugValueInt64(const char *propname, __int64 defVal) const
 {
     StringBuffer lower;
-    lower.append(propname).toLowerCase();
+    lower.append(propname);
+    if (!strchr(lower, ':'))
+        lower.toLowerCase();
     CriticalBlock block(crit);
     StringBuffer prop("Debug/");
     prop.append(lower);
@@ -8517,7 +8527,9 @@ __int64 CLocalWorkUnit::getDebugValueInt64(const char *propname, __int64 defVal)
 double CLocalWorkUnit::getDebugValueReal(const char *propname, double defVal) const
 {
     StringBuffer lower;
-    lower.append(propname).toLowerCase();
+    lower.append(propname);
+    if (!strchr(lower, ':'))
+        lower.toLowerCase();
     CriticalBlock block(crit);
     StringBuffer prop("Debug/");
     prop.append(lower);
@@ -8527,7 +8539,9 @@ double CLocalWorkUnit::getDebugValueReal(const char *propname, double defVal) co
 bool CLocalWorkUnit::getDebugValueBool(const char * propname, bool defVal) const
 {
     StringBuffer lower;
-    lower.append(propname).toLowerCase();
+    lower.append(propname);
+    if (!strchr(lower, ':'))
+        lower.toLowerCase();
     CriticalBlock block(crit);
     StringBuffer prop("Debug/");
     prop.append(lower);
@@ -8600,7 +8614,9 @@ void CLocalWorkUnit::addProcess(const char *type, const char *instance, unsigned
 void CLocalWorkUnit::setDebugValue(const char *propname, const char *value, bool overwrite)
 {
     StringBuffer lower;
-    lower.append(propname).toLowerCase();
+    lower.append(propname);
+    if (!strchr(lower, ':'))
+        lower.toLowerCase();
     CriticalBlock block(crit);
     StringBuffer prop("Debug/");
     prop.append(lower);
@@ -8615,7 +8631,9 @@ void CLocalWorkUnit::setDebugValue(const char *propname, const char *value, bool
 void CLocalWorkUnit::setDebugValueInt(const char *propname, int value, bool overwrite)
 {
     StringBuffer lower;
-    lower.append(propname).toLowerCase();
+    lower.append(propname);
+    if (!strchr(lower, ':'))
+        lower.toLowerCase();
     CriticalBlock block(crit);
     StringBuffer prop("Debug/");
     prop.append(lower);

diff --git a/roxie/ccd/ccd.hpp b/roxie/ccd/ccd.hpp
@@ -408,6 +408,7 @@ extern bool defaultExecuteDependenciesSequentially;
 extern bool defaultStartInputsSequentially;
 extern bool oneShotRoxie;
 extern bool traceStrands;
+extern unsigned minPayloadSize;
 
 extern int backgroundCopyClass;
 extern int backgroundCopyPrio;

diff --git a/roxie/ccd/ccdmain.cpp b/roxie/ccd/ccdmain.cpp
@@ -102,6 +102,7 @@ bool mergeAgentStatistics = true;
 PTreeReaderOptions defaultXmlReadFlags = ptr_ignoreWhiteSpace;
 bool runOnce = false;
 bool oneShotRoxie = false;
+unsigned minPayloadSize = 800;
 
 unsigned udpMulticastBufferSize = 262142;
 #if !defined(_CONTAINERIZED) && !defined(SUBCHANNELS_IN_HEADER)
@@ -122,7 +123,7 @@ bool lockSuperFiles;
 bool useRemoteResources;
 bool checkFileDate;
 bool lazyOpen;
-bool localAgent;
+bool localAgent = false;
 bool encryptInTransit;
 bool useAeron;
 bool ignoreOrphans;
@@ -690,13 +691,43 @@ int CCD_API roxie_main(int argc, const char *argv[], const char * defaultYaml)
         useOldTopology = checkFileExists(topologyFile.str());
         topology = loadConfiguration(useOldTopology ? nullptr : defaultYaml, argv, "roxie", "ROXIE", topologyFile, nullptr, "@netAddress");
         saveTopology();
+
+        // Any settings we read from topology that must NOT be overridden in workunit debug fields should be read at this point, before the following section
+        getAllowedPipePrograms(allowedPipePrograms, true);
+
+        // Allow workunit debug fields to override most roxie configuration values, for testing/debug purposes.
+
+        topology->getProp("@daliServers", fileNameServiceDali);
+        const char *wuid = topology->queryProp("@workunit");
+        if (wuid)
+        {
+            Owned<IRoxieDaliHelper> daliHelper;
+            Owned<IConstWorkUnit> wu;
+            daliHelper.setown(connectToDali(ROXIE_DALI_CONNECT_TIMEOUT));
+            wu.setown(daliHelper->attachWorkunit(wuid));
+            Owned<IStringIterator> debugValues = &wu->getDebugValues();
+            ForEach (*debugValues)
+            {
+                StringBuffer debugStr;
+                SCMStringBuffer valueStr;
+                StringBufferAdaptor aDebugStr(debugStr);
+                debugValues->str(aDebugStr);
+                if (startsWith(debugStr, "roxie:"))
+                {
+                    wu->getDebugValue(debugStr.str(), valueStr);
+                    debugStr.replaceString("roxie:", "@");
+                    topology->setProp(debugStr.str(), valueStr.str());
+                }
+            }
+        }
+
         if (topology->getPropBool("expert/@profileStartup", false))
         {
             double interval = topology->getPropReal("expert/@profileStartupInterval", 0.2);
             startupTracer.setInterval(interval);
             startupTracer.start();
         }
-        localAgent = topology->getPropBool("@localAgent", topology->getPropBool("@localSlave", false));  // legacy name
+        localAgent = topology->getPropBool("@localAgent", topology->getPropBool("@localSlave", localAgent));  // legacy name
         encryptInTransit = topology->getPropBool("@encryptInTransit", false) && !localAgent;
         if (encryptInTransit)
             initSecretUdpKey();
@@ -769,7 +800,6 @@ int CCD_API roxie_main(int argc, const char *argv[], const char * defaultYaml)
         installDefaultFileHooks(topology);
 
         Owned<const IQueryDll> standAloneDll;
-        const char *wuid = topology->queryProp("@workunit");
         if (wuid)
             setDefaultJobId(wuid);
         if (topology->hasProp("@loadWorkunit"))
@@ -933,6 +963,7 @@ int CCD_API roxie_main(int argc, const char *argv[], const char * defaultYaml)
             }
         }
 
+        minPayloadSize = topology->getPropInt("@minPayloadSize", minPayloadSize);
         acknowledgeAllRequests = topology->getPropBool("@acknowledgeAllRequests", acknowledgeAllRequests);
         headRegionSize = topology->getPropInt("@headRegionSize", 50);
         ccdMulticastPort = topology->getPropInt("@multicastPort", CCD_MULTICAST_PORT);
@@ -1181,8 +1212,6 @@ int CCD_API roxie_main(int argc, const char *argv[], const char * defaultYaml)
         maxFilesOpen[true] = topology->getPropInt("@maxRemoteFilesOpen", 1000);
         dafilesrvLookupTimeout = topology->getPropInt("@dafilesrvLookupTimeout", 10000);
         setRemoteFileTimeouts(dafilesrvLookupTimeout, 0);
-        topology->getProp("@daliServers", fileNameServiceDali);
-        getAllowedPipePrograms(allowedPipePrograms, true);
         trapTooManyActiveQueries = topology->getPropBool("@trapTooManyActiveQueries", true);
         maxEmptyLoopIterations = topology->getPropInt("@maxEmptyLoopIterations", 1000);
         maxGraphLoopIterations = topology->getPropInt("@maxGraphLoopIterations", 1000);

diff --git a/roxie/ccd/ccdserver.cpp b/roxie/ccd/ccdserver.cpp
@@ -77,7 +77,6 @@
 #include "rtldynfield.hpp"
 
 #define MAX_HTTP_HEADERSIZE 8000
-#define MIN_PAYLOAD_SIZE 800
 
 #ifdef _WIN32
 #pragma warning(disable : 4355)
@@ -4151,8 +4150,8 @@ class CRemoteResultAdaptor : implements IEngineRowStream, implements IFinalRoxie
         void init(unsigned minSize)
         {
             assertex(!buffer.length());
-            if (minSize < MIN_PAYLOAD_SIZE)
-                minSize = MIN_PAYLOAD_SIZE;
+            if (minSize < minPayloadSize)
+                minSize = minPayloadSize;
             unsigned headerSize = sizeof(RoxiePacketHeader)+owner.headerLength();
             unsigned bufferSize = headerSize+minSize;
             if (bufferSize < mtu_size)

diff --git a/testing/regress/ecl/remote.ecl b/testing/regress/ecl/remote.ecl
@@ -17,12 +17,18 @@
 
 //nohthor
 //nothor
+//version localAgent=true
+//version localAgent=false
+
 
 IMPORT Std.system.thorlib as thorlib;
 import ^ as root;
 multiPart := #IFDEFINED(root.multiPart, true);
 useLocal := #IFDEFINED(root.useLocal, false);
 useTranslation := #IFDEFINED(root.useTranslation, false);
+useLocalAgent := #IFDEFINED(root.localAgent, false);
+
+#option('roxie:localAgent', useLocalAgent);
 
 import $.setup;
 Files := setup.Files(multiPart, useLocal, useTranslation);

diff --git a/testing/regress/ecl/stresstext.ecl b/testing/regress/ecl/stresstext.ecl
@@ -23,6 +23,13 @@
 //version multiPart=false,variant='inplace_lzw'
 //version multiPart=false,variant='inplace_lz4hc'
 
+// The settings below may be useful when trying to analyse Roxie keyed join behaviour, as they will
+// eliminate some wait time for an agent queue to become available
+
+//#option('roxie:minPayloadSize', 10000)
+//#option('roxie:agentThreads', 400)
+//#option('roxie:prestartAgentThreads', true)
+
 import ^ as root;
 multiPart := #IFDEFINED(root.multiPart, true);
 variant := #IFDEFINED(root.variant, '');