Merge pull request #15388 from richardkchapman/roxie-dali-attach

HPCC-26565 One-shot roxie may crash if agents connect to workunit

Reviewed-by: Gavin Halliday <ghalliday@hpccsystems.com>
Merged-by: Gavin Halliday <ghalliday@hpccsystems.com>

roxie/ccd/ccdcontext.cpp

@@ -2722,15 +2722,6 @@ public:
                 logctx.CTXLOG("Created wu %s for query statistics", statsWu->queryWuid());
             }
         }
-        else if (wuid)  // This is undocumented for developer debug use only
-        {
-            IRoxieDaliHelper *daliHelper = checkDaliConnection();
-            assertex(daliHelper );
-            workUnit.setown(daliHelper->attachWorkunit(wuid, _factory->queryDll()));
-            if (!workUnit)
-                throw MakeStringException(ROXIE_DALI_ERROR, "Failed to open workunit %s", wuid);
-            startWorkUnit(); // MORE - will go horribly wrong if specified wu doesn't match query
-        }
         else if (context->getPropBool("@debug", false))
         {
             bool breakAtStart = context->getPropBool("@break", true);
@@ -3569,7 +3560,7 @@ public:
         Owned <IRoxieDaliHelper> daliHelper = connectToDali();
         if (daliHelper && daliHelper->connected())
         {
-            Owned<IConstWorkUnit> externalWU = daliHelper->attachWorkunit(wuid, NULL);
+            Owned<IConstWorkUnit> externalWU = daliHelper->attachWorkunit(wuid);
             if (externalWU)
             {
                 externalWU->remoteCheckAccess(queryUserDescriptor(), false);

roxie/ccd/ccddali.cpp

@@ -167,7 +167,7 @@ public:
 class CRoxieDaliHelper : implements IRoxieDaliHelper, public CInterface
 {
 private:
-    static bool isConnected;
+    static std::atomic<bool> isConnected;
     static CRoxieDaliHelper *daliHelper;  // Note - this does not own the helper
     static CriticalSection daliHelperCrit;
     CriticalSection daliConnectionCrit;
@@ -436,21 +436,39 @@ public:
     IMPLEMENT_IINTERFACE;
     CRoxieDaliHelper() : serverStatus(NULL), connectWatcher(this)
     {
-        userdesc.setown(createUserDescriptor());
-        const char *roxieUser = NULL;
-        const char *roxiePassword = NULL;
-        if (topology)
-        {
-            roxieUser = topology->queryProp("@ldapUser");
-            roxiePassword = topology->queryProp("@ldapPassword");
-        }
-        if (!roxieUser)
-            roxieUser = "roxie";
-        if (!roxiePassword)
-            roxiePassword = "";
-        StringBuffer password;
-        decrypt(password, roxiePassword);
-        userdesc->set(roxieUser, password.str());
+        if (oneShotRoxie && topology && topology->hasProp("@workunit"))
+        {
+            if (!connect(ROXIE_DALI_CONNECT_TIMEOUT))
+                throw MakeStringException(ROXIE_DALI_ERROR, "Failed to connect to dali");
+            Owned<IWorkUnitFactory> wuFactory = getWorkUnitFactory();
+            Owned<IWorkUnit> w = wuFactory->updateWorkUnit(topology->queryProp("@workunit"));
+            if (!w)
+                throw MakeStringException(ROXIE_DALI_ERROR, "Failed to locate dll workunit info");
+            w->setAgentSession(myProcessSession());
+            if(topology->getPropBool("@resetWorkflow", false))
+                w->resetWorkflow();
+            userdesc.set(w->queryUserDescriptor());
+            w->commit();
+            w.clear();
+        }
+        else
+        {
+            userdesc.setown(createUserDescriptor());
+            const char *roxieUser = NULL;
+            const char *roxiePassword = NULL;
+            if (topology)
+            {
+                roxieUser = topology->queryProp("@ldapUser");
+                roxiePassword = topology->queryProp("@ldapPassword");
+            }
+            if (!roxieUser)
+                roxieUser = "roxie";
+            if (!roxiePassword)
+                roxiePassword = "";
+            StringBuffer password;
+            decrypt(password, roxiePassword);
+            userdesc->set(roxieUser, password.str());
+        }
         if (fileNameServiceDali.length())
             connectWatcher.start();
         else
@@ -649,31 +667,10 @@ public:
 #endif
     }
 
-    virtual IConstWorkUnit *attachWorkunit(const char *wuid, ILoadedDllEntry *source)
+    virtual IConstWorkUnit *attachWorkunit(const char *wuid)
     {
         assertex(isConnected);
         Owned<IWorkUnitFactory> wuFactory = getWorkUnitFactory();
-        Owned<IWorkUnit> w = wuFactory->updateWorkUnit(wuid);
-        if (!w)
-            return NULL;
-        w->setAgentSession(myProcessSession());
-        if(topology->getPropBool("@resetWorkflow", false))
-            w->resetWorkflow();
-        if (source)
-        {
-            StringBuffer wuXML;
-            if (getEmbeddedWorkUnitXML(source, wuXML))
-            {
-                Owned<ILocalWorkUnit> localWU = createLocalWorkUnit(wuXML);
-                queryExtendedWU(w)->copyWorkUnit(localWU, true, true);
-            }
-            else
-                throw MakeStringException(ROXIE_DALI_ERROR, "Failed to locate dll workunit info");
-        }
-        if (topology->hasProp("@workunit"))    // This really only works properly in one-shot mode
-            userdesc.set(w->queryUserDescriptor());
-        w->commit();
-        w.clear();
         return wuFactory->openWorkUnit(wuid);
     }
 
@@ -956,7 +953,7 @@ public:
     }
 } roxieDllServer;
 
-bool CRoxieDaliHelper::isConnected = false;
+std::atomic<bool> CRoxieDaliHelper::isConnected(false);
 CRoxieDaliHelper * CRoxieDaliHelper::daliHelper;
 CriticalSection CRoxieDaliHelper::daliHelperCrit;
 #ifdef ROXIE_DALI_CACHE

roxie/ccd/ccddali.hpp

@@ -47,7 +47,7 @@ interface IRoxieDaliHelper : extends IInterface
     virtual IFileDescriptor *checkClonedFromRemote(const char *id, IFileDescriptor *fdesc, bool cacheIt, bool isPrivilegedUser) = 0;
     virtual IDistributedFile *resolveLFN(const char *filename, bool cacheIt, bool writeAccess, bool isPrivilegedUser) = 0;
     virtual IFileDescriptor *resolveCachedLFN(const char *filename) = 0;
-    virtual IConstWorkUnit *attachWorkunit(const char *wuid, ILoadedDllEntry *source) = 0;
+    virtual IConstWorkUnit *attachWorkunit(const char *wuid) = 0;
     virtual IPropertyTree *getQuerySet(const char *id) = 0;
     virtual IDaliPackageWatcher *getQuerySetSubscription(const char *id, ISafeSDSSubscription *notifier) = 0;
     virtual IPropertyTree *getPackageSets() = 0;

roxie/ccd/ccdlistener.cpp

@@ -1169,7 +1169,7 @@ public:
         else
         {
             daliHelper.setown(connectToDali());
-            wu.setown(daliHelper->attachWorkunit(wuid.get(), NULL));
+            wu.setown(daliHelper->attachWorkunit(wuid.get()));
         }
         Owned<StringContextLogger> logctx = new StringContextLogger(wuid.get());
         if (wu->hasDebugValue("GlobalId"))

roxie/ccd/ccdqueue.cpp

@@ -1519,7 +1519,7 @@ public:
         if (!queryFactory && logctx.queryWuid())
         {
             Owned <IRoxieDaliHelper> daliHelper = connectToDali();
-            Owned<IConstWorkUnit> wu = daliHelper->attachWorkunit(logctx.queryWuid(), NULL);
+            Owned<IConstWorkUnit> wu = daliHelper->attachWorkunit(logctx.queryWuid());
             queryFactory.setown(createAgentQueryFactoryFromWu(wu, channel));
             if (queryFactory)
                 cacheOnDemandQuery(queryHash, channel, queryFactory);