HPCC-30057 Use DataBuffers to store returned rows in localAgent mode

Signed-off-by: Richard Chapman <[email protected]>

devdoc/roxie.md

@@ -281,3 +281,17 @@ Should the scope of the blacklist be different? Possible scopes are:
 
 Options 2 and 4 above would allow all aspects of the blacklisting behaviour to be specified by options on the SOAPCALL. We could control whether or not the
 blacklister is to be used at all via a SOAPCALL option with any of the above...
+
+Some notes on LocalAgent mode
+=============================
+
+In localAgent mode, the global queueManager object (normally a RoxieUdpSocketQueueManager) is replaced by a RoxieLocalQueueManager. Outbound packets are added directly to target queue, inbound are packed into DataBuffers.
+
+There is also "local optimizations" mode where any index operation reading a one-part file (does the same apply to one-part disk files?) just reads it directly on the server (regardless of localAgent setting). Typically still injected into receiver code though as otherwise handling exception cases, limits etc would all be duplicated/messy. Rows created in localOptimization mode are created directly in the caller's row manager, and are injected in serialized format.
+
+Why are inbound not created directly in the desired destination's allocator and then marked as serialized? Some lifespan issues... are they insurmountable?
+We do pack into dataBuffers rather than MemoryBuffers, which avoids a need to copy the data before the receiver can use it. Large rows get split and will require copying again, but we could set dataBufferSize to be bigger in localAgent mode to mitigate this somewhat.
+
+What is the lifespan issue? In-flight queries may be abandoned when a server-side query fails, times out, or no longer needs the data. Using DataBuffer does not have this issue as they are attached to the query's memory manager/allocation once read. Or we could bypass the agent queue altogether, but rather more refactoring needed for that (might almost be easier to extent the "local optimization" mode to use multiple threads at that point)
+
+abortPending, replyPending, and abortPendingData methods are unimplemented, which may lead to some inefficiencies?

helm/hpcc/values.schema.json

@@ -1585,6 +1585,11 @@
           "default": 0,
           "description": "Specify an IONICE value for the background copy thread, if backgroundCopyClass set to best-effort."
         },
+        "blockedLocalAgent": { 
+          "type": "boolean",
+          "default": true,
+          "description": "Used DataBuffer blocks to return agent data in localAgent mode."
+        },
         "callbackRetries": { 
           "type": "integer",
           "default": 3,

roxie/ccd/ccd.hpp

@@ -300,6 +300,7 @@ extern IPropertyTree *topology;
 extern MapStringTo<int> *preferredClusters;
 extern StringArray allQuerySetNames;
 
+extern bool blockedLocalAgent;
 extern bool acknowledgeAllRequests;
 extern bool alwaysTrustFormatCrcs;
 extern bool allFilesDynamic;

roxie/ccd/ccdmain.cpp

@@ -68,6 +68,7 @@ unsigned numServerThreads = 30;
 unsigned numAgentThreads = 30;
 bool prestartAgentThreads = false;
 unsigned numRequestArrayThreads = 5;
+bool blockedLocalAgent = true;
 bool acknowledgeAllRequests = true;
 unsigned headRegionSize;
 unsigned ccdMulticastPort;
@@ -756,8 +757,19 @@ int CCD_API roxie_main(int argc, const char *argv[], const char * defaultYaml)
         else
             setStatisticsComponentName(SCTroxie, "roxie", true);
 #ifdef _CONTAINERIZED
-        getDefaultStoragePlane(defaultPlane);
-        getDefaultIndexBuildStoragePlane(defaultIndexBuildPlane);
+        try
+        {
+            getDefaultStoragePlane(defaultPlane);
+            getDefaultIndexBuildStoragePlane(defaultIndexBuildPlane);
+        }
+        catch (IException *E)
+        {
+#ifdef _DEBUG            
+            E->Release();   // Useful for some local testing to be able to ignore these configuration errors
+#else
+            throw;
+#endif
+        }
 #endif
         installDefaultFileHooks(topology);
 
@@ -966,6 +978,7 @@ int CCD_API roxie_main(int argc, const char *argv[], const char * defaultYaml)
         }
 
         minPayloadSize = topology->getPropInt("@minPayloadSize", minPayloadSize);
+        blockedLocalAgent = topology->getPropBool("@blockedLocalAgent", blockedLocalAgent);
         acknowledgeAllRequests = topology->getPropBool("@acknowledgeAllRequests", acknowledgeAllRequests);
         headRegionSize = topology->getPropInt("@headRegionSize", 0);
         ccdMulticastPort = topology->getPropInt("@multicastPort", CCD_MULTICAST_PORT);