Merge branch 'candidate-7.12.x'

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

common/workunit/workunit.cpp

@@ -299,6 +299,7 @@ protected:
             case SSTworkflow:
             case SSTgraph:
                 // SSTworkflow and SSTgraph may be safely ignored.  They are not required to produce the statistics.
+                expandProcessTreeFromStats(rootTarget, target, &cur);
                 continue;
             case SSTfunction:
                 //MORE:Should function scopes be included in the graph scope somehow, and if so how?
@@ -3990,7 +3991,19 @@ public:
         if (workUnitTraceLevel > 1)
             DBGLOG("Releasing locked workunit %s", queryWuid());
         if (c)
-            c->unlockRemote();
+        {
+            try
+            {
+                c->unlockRemote();
+            }
+            catch (IException *E)
+            {
+                // Exceptions here should be very uncommon - but there's also not a lot we can do if we get one
+                // Allowing them to be thrown out of the destructor is going to terminate the program which is NOT what we want.
+                EXCLOG(E);
+                ::Release(E);
+            }
+        }
     }
 
     virtual IConstWorkUnit * unlock()
@@ -8456,6 +8469,9 @@ void CLocalWorkUnit::setStatistic(StatisticCreatorType creatorType, const char *
 
     if (!statTree)
     {
+        /* NB: Sasha archive uses this structure directly
+         * if it changes, the code in saarch.cpp needs updating
+         */
         statTree = stats->addPropTree("Statistic");
         statTree->setProp("@creator", creator);
         statTree->setProp("@scope", scope);

common/wuanalysis/anaerrorcodes.hpp

@@ -9,7 +9,7 @@ typedef enum
     ANA_DISTRIB_SKEW_INPUT_ID,
     ANA_DISTRIB_SKEW_OUTPUT_ID,
     ANA_IOSKEW_RECORDS_ID,
-    ANA_IOSKEW_CHILDRECORDS_ID,
+    ANA_UNUSED_ID,                                 /* May re-use but don't remove to avoid changing later id's */
     ANA_KJ_EXCESS_PREFILTER_ID
 } AnalyzerErrorCode;
 

common/wuanalysis/anarule.cpp

@@ -142,7 +142,7 @@ public:
             else
                 cost = (timeMaxLocalExecute - timeAvgLocalExecute);
 
-            result.set(ANA_IOSKEW_CHILDRECORDS_ID, cost, "Significant skew in records causes uneven %s time", category);
+            result.set(ANA_IOSKEW_RECORDS_ID, cost, "Significant skew in records causes uneven %s time", category);
             updateInformation(result, activity);
             return true;
         }

dali/base/dadfs.cpp

@@ -6990,7 +6990,7 @@ public:
 #define GROUP_CACHE_INTERVAL (1000*60)
 #define GROUP_EXCEPTION_CACHE_INTERVAL (1000*60*10)
 
-static GroupType translateGroupType(const char *groupType)
+GroupType translateGroupType(const char *groupType)
 {
     if (!groupType)
         return grp_unknown;

dali/base/dadfs.hpp

@@ -814,6 +814,8 @@ extern da_decl const char *normalizeLFN(const char *s, StringBuffer &normalized)
 
 extern da_decl IDFAttributesIterator *createSubFileFilter(IDFAttributesIterator *_iter,IUserDescriptor* _user, bool includesub, unsigned timems=INFINITE); // takes ownership of iter
 
+extern da_decl GroupType translateGroupType(const char *groupType);
+
 #define DFS_REPLICATE_QUEUE "dfs_replicate_queue"
 #define DRQ_STOP 0
 #define DRQ_REPLICATE 1

dali/base/dafdesc.cpp

@@ -3145,6 +3145,168 @@ static void setupContainerizedStorageLocations()
     }
 }
 
+struct GroupInformation : public CInterface
+{
+public:
+    GroupInformation(const char * _name) : name(_name) {}
+
+    unsigned ordinality() const { return hosts.ordinality(); }
+
+    bool checkIsSubset(const GroupInformation & other); // save information if it is a subset of other
+    void createStoragePlane(IPropertyTree * storage, unsigned copy) const;
+
+public:
+    StringBuffer name;
+    StringBuffer dir;
+    StringArray hosts;
+    const GroupInformation * container = nullptr;
+    unsigned containerOffset = 0;
+    GroupType groupType = grp_unknown;
+    bool dropZone = false;
+};
+
+using GroupInfoArray = CIArrayOf<GroupInformation>;
+
+bool GroupInformation::checkIsSubset(const GroupInformation & other)
+{
+    //Find the first ip that matches, and then check the next ips within the other list also match
+    unsigned thisSize = hosts.ordinality();
+    unsigned otherSize = other.hosts.ordinality();
+    assertex(thisSize <= otherSize);
+    for (unsigned i=0; i <= otherSize-thisSize; i++)
+    {
+        bool match = true;
+        for (unsigned j=0; j < thisSize; j++)
+        {
+            if (!strieq(hosts.item(j), other.hosts.item(i+j)))
+            {
+                match = false;
+                break;
+            }
+        }
+
+        if (match)
+        {
+            container = &other;
+            containerOffset = i;
+            return true;
+        }
+    }
+    return false;
+}
+
+void GroupInformation::createStoragePlane(IPropertyTree * storage, unsigned copy) const
+{
+    IPropertyTree * plane = storage->addPropTree("planes");
+    if (copy == 0)
+    {
+        plane->setProp("@name", name);
+    }
+    else
+    {
+        StringBuffer mirrorname;
+        mirrorname.append(name).append("_mirror").append(copy);
+        plane->setProp("@name", mirrorname);
+    }
+
+    //URL style drop zones don't generate a host entry, and will have a single device
+    if (ordinality() != 0)
+    {
+        if (container)
+        {
+            plane->setProp("@hosts", container->name);
+            if (containerOffset)
+                plane->setPropInt("@start", containerOffset);
+            if (ordinality() != container->ordinality())
+                plane->setPropInt("@size", ordinality());
+        }
+        else
+            plane->setProp("@hosts", name);
+
+        if (ordinality() > 1)
+            plane->setPropInt("@numDevices", ordinality());
+    }
+
+    if (copy)
+        plane->setPropInt("@offset", copy);
+
+    if (dir.length())
+        plane->setProp("@prefix", dir);
+    else
+        plane->setProp("@prefix", queryBaseDirectory(groupType, copy));
+
+    if (dropZone)
+        plane->setPropBool("@dropZone", true);
+
+    //MORE: If container is identical to this except for the name we could generate an information tag @alias
+}
+
+static void appendGroup(GroupInfoArray & groups, GroupInformation * group)
+{
+    if (!group->name)
+    {
+        group->Release();
+        return;
+    }
+
+    //Check for a duplicate group name.  This should never happen, but make sure it is caught if it did.
+    ForEachItemIn(i, groups)
+    {
+        if (strieq(groups.item(i).name, group->name))
+        {
+            DBGLOG("Unexpected duplicate group name %s", group->name.str());
+            group->Release();
+            return;
+        }
+    }
+
+    groups.append(*group);
+}
+
+
+static int compareGroupSize(CInterface * const * _left, CInterface * const * _right)
+{
+    const GroupInformation * left = static_cast<const GroupInformation *>(*_left);
+    const GroupInformation * right = static_cast<const GroupInformation *>(*_right);
+    int ret = (int) (right->hosts.ordinality() - left->hosts.ordinality());
+    if (ret)
+        return ret;
+    return stricmp(left->name, right->name);
+}
+
+
+static void generateHosts(IPropertyTree * storage, GroupInfoArray & groups)
+{
+    //This function is potentially O(N^2) in the number of groups, and O(n^2) in the number of nodes within those groups.
+    //In practice, it is unlikely to take very long unless there were vast numbers of groups.
+    ForEachItemIn(i, groups)
+    {
+        GroupInformation & cur = groups.item(i);
+        GroupInformation * container = nullptr;
+        for (unsigned j=0; j < i; j++)
+        {
+            GroupInformation & prev = groups.item(j);
+            if (cur.checkIsSubset(prev))
+                break;
+        }
+
+        // No containing hostGroup found, so generate a new entry for this set of hosts
+        if (!cur.container && cur.ordinality())
+        {
+            IPropertyTree * plane = storage->addPropTree("hostGroups");
+            plane->setProp("@name", cur.name);
+            StringBuffer host;
+            ForEachItemIn(i, cur.hosts)
+            {
+                IPropertyTree * entry2 = createPTree("host");
+                IPropertyTree * entry = plane->addPropTreeArrayItem("hosts", entry2);
+                entry->setProp("", cur.hosts.item(i));
+            }
+        }
+    }
+
+}
+
 static CriticalSection storageCS;
 void initializeStorageGroups(bool createPlanesFromGroups)
 {
@@ -3156,46 +3318,25 @@ void initializeStorageGroups(bool createPlanesFromGroups)
 #ifndef _CONTAINERIZED
     if (createPlanesFromGroups && !storage->hasProp("planes"))
     {
+        GroupInfoArray allGroups;
+
         //Create information about the storage planes from the groups published in dali
-        INamedGroupStore & groupStore = queryNamedGroupStore();
-        Owned<INamedGroupIterator> iter= groupStore.getIterator();
-        ForEach(*iter)
+        //Use the Groups section directly, rather than queryNamedGroupStore(), so that hostnames are preserved
         {
-            StringBuffer name, dir;
-            iter->get(name);
-            if (name.length())
+            Owned<IRemoteConnection> conn = querySDS().connect("/Groups", myProcessSession(), RTM_LOCK_READ, SDS_CONNECT_TIMEOUT);
+            Owned<IPropertyTreeIterator> groups = conn->queryRoot()->getElements("Group");
+            ForEach(*groups)
             {
-                IPropertyTree * plane = storage->addPropTree("planes");
-                plane->setProp("@name", name);
-                GroupType groupType;
-                Owned<IGroup> group = groupStore.lookup(name, dir, groupType);
-
-                if (dir.length())
-                    plane->setProp("@prefix", dir);
-                else
-                    plane->setProp("@prefix", queryBaseDirectory(groupType, 0));
-
-                if (!group)
-                    plane->setPropInt("@numDevices", 0);
-                else if (group->ordinality() != 1)
-                    plane->setPropInt("@numDevices", group->ordinality());
-
-                if (groupType == grp_thor)
+                IPropertyTree & group = groups->query();
+                Owned<GroupInformation> next = new GroupInformation(group.queryProp("@name"));
+                Owned<IPropertyTreeIterator> nodes = group.getElements("Node");
+                ForEach(*nodes)
                 {
-                    assertex(group);
-                    StringBuffer mirrorname;
-                    mirrorname.append(name).append("replica");
-
-                    IPropertyTree * mirror = storage->addPropTree("planes");
-                    mirror->setProp("@name", mirrorname);
-                    mirror->setProp("@prefix", queryBaseDirectory(groupType, 1));
-
-                    if (group->ordinality() != 1)
-                        mirror->setPropInt("@numDevices", group->ordinality());
-
-                    IPropertyTree * elem = plane->addPropTreeArrayItem("replication", createPTree());
-                    elem->setProp("", mirrorname);
+                    next->hosts.append(nodes->query().queryProp("@ip"));
                 }
+                next->groupType = translateGroupType(group.queryProp("@kind"));
+
+                appendGroup(allGroups, next.getClear());
             }
         }
 
@@ -3208,20 +3349,39 @@ void initializeStorageGroups(bool createPlanesFromGroups)
             {
                 IPropertyTree & cur = dropzones->query();
                 unsigned numServers = cur.getCount("ServerList");
+
+                //Allow url style drop zones, and drop zones with a single node.  Not sure what >1 would mean in legacy.
                 if (numServers <= 1)
                 {
-                    //If no instances then assume it is always local (e.g. azure)
-                    const char * ip = numServers ? cur.queryProp("ServerList[1]/@server") : nullptr;
-                    IPropertyTree * plane = storage->addPropTree("planes");
-                    plane->setProp("@name", cur.queryProp("@name"));
-                    plane->setProp("@prefix", cur.queryProp("@directory"));
-
-                    //Temporary.  This is likely to change to eventually create a hostgroup and set the @hosts property.
-                    if (ip && !streq(ip, "."))
-                        plane->setProp("@host", ip);
+                    Owned<GroupInformation> next = new GroupInformation(cur.queryProp("@name"));
+                    const char * ip = cur.queryProp("ServerList[1]/@server");
+
+                    next->dir.set(cur.queryProp("@directory"));
+                    next->dropZone= true;
+                    if (ip)
+                        next->hosts.append(ip);
+                    appendGroup(allGroups, next.getClear());
                 }
             }
         }
+
+        //Sort into size order, to help spot groups that are subsets of other groups
+        allGroups.sort(compareGroupSize);
+
+        //Now generate a list of hosts, if one group is a subset of another then do not generate the hosts
+        generateHosts(storage, allGroups);
+
+        //Finally generate the storage plane information
+        ForEachItemIn(i, allGroups)
+        {
+            const GroupInformation & cur = allGroups.item(i);
+            cur.createStoragePlane(storage, 0);
+            if (cur.groupType == grp_thor)
+                cur.createStoragePlane(storage, 1);
+        }
+
+        //Uncomment the following to trace the values that been generated
+        //printYAML(storage);
     }
 #endif
 

dali/dfu/dfuserver.cpp

@@ -185,6 +185,9 @@ int main(int argc, const char *argv[])
             engine.setown(createDFUengine());
             engine->setDFUServerName(name);
             addAbortHandler(exitDFUserver);
+
+            IPropertyTree * config = nullptr;
+            installDefaultFileHooks(config);
         }
         const char *q = queue.str();
         for (;;) {

dali/ft/ftslave.cpp

@@ -141,6 +141,7 @@ int main(int argc, char * argv[])
 {
     InitModuleObjects();
     setDaliServixSocketCaching(true);
+    installDefaultFileHooks(nullptr);
     FtSlave slave;
     slave.run(argc, argv);
     return 0;

dali/sasha/saarch.cpp

@@ -1007,11 +1007,12 @@ class CWorkUnitArchiver: public CBranchArchiver
             }
             getWorkUnitCreateTime(wuid,time);
             // get latest time stamp
-            Owned<IPropertyTreeIterator> iter = e.getElements("TimeStamps/TimeStamp/*");
+            Owned<IPropertyTreeIterator> iter = e.getElements("Statistics/Statistic");
             if (iter) {
                 ForEach(*iter) {
                     CDateTime cmp;
-                    cmp.setString(iter->query().queryProp(NULL));
+                    timestamp_type ts = iter->query().getPropInt64("@ts");
+                    cmp.setTimeStamp(ts);
                     if (time.isNull()||(!cmp.isNull()&&(cmp.compare(time)>0)))
                         time.set(cmp);
                 }

ecl/eclcmd/eclcmd_common.cpp

@@ -346,7 +346,10 @@ eclCmdOptionMatchIndicator EclCmdCommon::matchCommandLineOption(ArgvIterator &it
     if (iter.matchOption(optUsername, ECLOPT_USERNAME)||iter.matchOption(optUsername, ECLOPT_USERNAME_S))
         return EclCmdOptionMatch;
     if (iter.matchOption(optPassword, ECLOPT_PASSWORD)||iter.matchOption(optPassword, ECLOPT_PASSWORD_S))
+    {
+        optPasswordProvided = true;
         return EclCmdOptionMatch;
+    }
     if (iter.matchFlag(optVerbose, ECLOPT_VERBOSE) || iter.matchFlag(optVerbose, ECLOPT_VERBOSE_S))
         return EclCmdOptionMatch;
     if (iter.matchFlag(optSSL, ECLOPT_SSL) || iter.matchFlag(optSSL, ECLOPT_SSL_S))
@@ -376,7 +379,7 @@ bool EclCmdCommon::finalizeOptions(IProperties *globals)
     extractEclCmdOption(optUsername, globals, ECLOPT_USERNAME_ENV, ECLOPT_USERNAME_INI, NULL, NULL);
     extractEclCmdOption(optPassword, globals, ECLOPT_PASSWORD_ENV, ECLOPT_PASSWORD_INI, NULL, NULL);
 
-    if (!optUsername.isEmpty() && optPassword.isEmpty())
+    if (!optUsername.isEmpty() && !optPasswordProvided)
     {
         VStringBuffer prompt("%s's password: ", optUsername.get());
         StringBuffer pw;

ecl/eclcmd/eclcmd_common.hpp

@@ -276,6 +276,7 @@ public:
     StringAttr optPort;
     StringAttr optUsername;
     StringAttr optPassword;
+    bool optPasswordProvided = false;
     unsigned optWaitConnectMs = 0;
     unsigned optWaitReadSec = 0;
     bool optVerbose;

ecl/hqlcpp/hqlttcpp.cpp

@@ -1433,8 +1433,6 @@ IHqlExpression * SequenceNumberAllocator::doTransformRootExpr(IHqlExpression * e
     case no_keypatch:
     case no_outputscalar:
         return createTransformed(expr);         //NB: Do not common up!!!
-    case no_setmeta:
-        return LINK(expr);
     default:
         {
             OwnedHqlExpr transformed = transform(expr);

ecl/hthor/hthor.cpp

@@ -1708,7 +1708,7 @@ public:
 private:
     bool waitForPipe()
     {
-        Owned<IPipeProcessException> pipeException;
+        Owned<IException> pipeException;
         try
         {
             if (firstRead)
@@ -1721,8 +1721,9 @@ private:
             if (!readTransformer->eos())
                 return true;
         }
-        catch (IPipeProcessException *e)
+        catch (IException *e)
         {
+            // NB: the original exception is probably a IPipeProcessException, but because InterruptableSemaphore rethrows it, we must catch it as an IException
             pipeException.setown(e);
         }
         verifyPipe();
@@ -1811,7 +1812,7 @@ public:
 
     virtual void execute()
     {
-        Owned<IPipeProcessException> pipeException;
+        Owned<IException> pipeException;
         try
         {
             for (;;)
@@ -1836,8 +1837,9 @@ public:
             if (!recreate)
                 closePipe();
         }
-        catch (IPipeProcessException *e)
+        catch (IException *e)
         {
+            // NB: the original exception is probably a IPipeProcessException, but because InterruptableSemaphore rethrows it, we must catch it as an IException
             pipeException.setown(e);
         }
         verifyPipe();
@@ -11133,7 +11135,6 @@ void CHThorNewDiskReadActivity::ready()
     if (helper.getFlags() & TDRlimitskips)
         limit = (unsigned __int64) -1;
     stopAfter = helper.getChooseNLimit();
-    assertex(stopAfter != 0);
     if (!helper.transformMayFilter() && !helper.hasMatchFilter())
         remoteLimit = stopAfter;
     finishedParts = false;

esp/services/ws_dfu/ws_dfuXRefService.cpp

@@ -572,7 +572,11 @@ void addUsedFilesFromPackageMaps(MapStringTo<bool> &usedFileMap, const char *pro
     if (!packageSet)
         throw MakeStringException(ECLWATCH_PACKAGEMAP_NOTRESOLVED, "Unable to retrieve package information from dali /PackageMaps");
     StringArray pmids;
+#ifdef _CONTAINERIZED
+    Owned<IStringIterator> targets = getContainerTargetClusters("roxie", process);
+#else
     Owned<IStringIterator> targets = getTargetClusters("RoxieCluster", process);
+#endif
     ForEach(*targets)
     {
         SCMStringBuffer target;

esp/services/ws_fs/ws_fsService.cpp

@@ -878,7 +878,11 @@ bool CFileSprayEx::getOneDFUWorkunit(IEspContext& context, const char* wuid, IEs
     const char* clusterName = wu->getClusterName(cluster).str();
     if (clusterName && *clusterName)
     {
-        Owned<IStringIterator> targets = getTargetClusters(NULL, clusterName);
+#ifdef _CONTAINERIZED
+        Owned<IStringIterator> targets = getContainerTargetClusters(nullptr, clusterName);
+#else
+        Owned<IStringIterator> targets = getTargetClusters(nullptr, clusterName);
+#endif
         if (!targets->first())
             resultWU->setClusterName(clusterName);
         else

esp/services/ws_machine/ws_machineService.cpp

@@ -2918,7 +2918,11 @@ bool Cws_machineEx::onGetComponentUsage(IEspContext& context, IEspGetComponentUs
 
 StringArray& Cws_machineEx::listTargetClusterNames(IConstEnvironment* constEnv, StringArray& targetClusters)
 {
+#ifdef _CONTAINERIZED
+    Owned<IStringIterator> targets = getContainerTargetClusters(nullptr, nullptr);
+#else
     Owned<IStringIterator> targets = getTargetClusters(nullptr, nullptr);
+#endif
     ForEach(*targets)
     {
         SCMStringBuffer target;

esp/services/ws_sql/ws_sqlService.cpp

@@ -63,10 +63,12 @@ bool CwssqlEx::onGetDBMetaData(IEspContext &context, IEspGetDBMetaDataRequest &r
     if (includeStoredProcs)
     {
         const char * querysetfilter = req.getQuerySet();
-        Owned<IStringIterator> targets = getTargetClusters(NULL, NULL);
-
+#ifdef _CONTAINERIZED
+        Owned<IStringIterator> targets = getContainerTargetClusters(nullptr, nullptr);
+#else
+        Owned<IStringIterator> targets = getTargetClusters(nullptr, nullptr);
+#endif
         IArrayOf<IEspHPCCQuerySet> pquerysets;
-
         SCMStringBuffer target;
         ForEach(*targets)
         {
@@ -1968,7 +1970,11 @@ bool CwssqlEx::onGetResults(IEspContext &context, IEspGetResultsRequest &req, IE
 void CwssqlEx::refreshValidClusters()
 {
     validClusters.kill();
-    Owned<IStringIterator> it = getTargetClusters(NULL, NULL);
+#ifdef _CONTAINERIZED
+    Owned<IStringIterator> it = getContainerTargetClusters(nullptr, nullptr);
+#else
+    Owned<IStringIterator> it = getTargetClusters(nullptr, nullptr);
+#endif
     ForEach(*it)
     {
         SCMStringBuffer s;

esp/services/ws_workunits/ws_workunitsQuerySets.cpp

@@ -29,6 +29,7 @@
 #include "dautils.hpp"
 #include "httpclient.hpp"
 #include "portlist.h" //ROXIE_SERVER_PORT
+#include "TpWrapper.hpp"
 
 #define DALI_FILE_LOOKUP_TIMEOUT (1000*15*1)  // 15 seconds
 
@@ -37,8 +38,6 @@ const unsigned ROXIECONTROLQUERYTIMEOUT = 3000; //3 second
 const unsigned ROXIECONTROLQUERIESTIMEOUT = 30000; //30 second
 const unsigned ROXIELOCKCONNECTIONTIMEOUT = 60000; //60 second
 
-#define SDS_LOCK_TIMEOUT (5*60*1000) // 5mins, 30s a bit short
-
 //The CQuerySetQueryActionTypes[] has to match with the ESPenum QuerySetQueryActionTypes in the ecm file.
 static unsigned NumOfQuerySetQueryActionTypes = 7;
 static const char *QuerySetQueryActionTypes[] = { "Suspend", "Unsuspend", "ToggleSuspend", "Activate",
@@ -417,7 +416,11 @@ void QueryFilesInUse::loadTarget(IPropertyTree *t, const char *target, unsigned
 
 void QueryFilesInUse::loadTargets(IPropertyTree *t, unsigned flags)
 {
-    Owned<IStringIterator> targets = getTargetClusters("RoxieCluster", NULL);
+#ifdef _CONTAINERIZED
+    Owned<IStringIterator> targets = getContainerTargetClusters("roxie", nullptr);
+#else
+    Owned<IStringIterator> targets = getTargetClusters("RoxieCluster", nullptr);
+#endif
     SCMStringBuffer s;
     ForEach(*targets)
     {
@@ -935,7 +938,12 @@ bool CWsWorkunitsEx::onWUPublishWorkunit(IEspContext &context, IEspWUPublishWork
 bool CWsWorkunitsEx::onWUQuerysets(IEspContext &context, IEspWUQuerysetsRequest & req, IEspWUQuerysetsResponse & resp)
 {
     IArrayOf<IEspQuerySet> querySets;
-    Owned<IStringIterator> targets = getTargetClusters(NULL, NULL);
+#ifdef _CONTAINERIZED
+    Owned<IStringIterator> targets = getContainerTargetClusters(nullptr, nullptr);
+#else
+    Owned<IStringIterator> targets = getTargetClusters(nullptr, nullptr);
+#endif
+
     SCMStringBuffer target;
     ForEach(*targets)
     {
@@ -1612,7 +1620,11 @@ void CWsWorkunitsEx::getSuspendedQueriesByCluster(MapStringTo<bool> &suspendedQu
     }
     else
     {
+#ifdef _CONTAINERIZED
+        Owned<IStringIterator> targets = getContainerTargetClusters("roxie", nullptr);
+#else
         Owned<IStringIterator> targets = getTargetClusters("RoxieCluster", nullptr);
+#endif
         ForEach(*targets)
         {
             SCMStringBuffer target;
@@ -1666,7 +1678,11 @@ bool CWsWorkunitsEx::onWUListQueriesUsingFile(IEspContext &context, IEspWUListQu
     else // if (process && *process)
     {
         SCMStringBuffer targetStr;
+#ifdef _CONTAINERIZED
+        Owned<IStringIterator> targetClusters = getContainerTargetClusters("roxie", process);
+#else
         Owned<IStringIterator> targetClusters = getTargetClusters("RoxieCluster", process);
+#endif
         ForEach(*targetClusters)
             targets.append(targetClusters->str(targetStr).str());
         logMsg.append(", process ").append(process);

esp/smc/SMCLib/TpWrapper.cpp

@@ -2010,3 +2010,44 @@ extern TPWRAPPER_API ISashaCommand* archiveOrRestoreWorkunits(StringArray& wuids
             sashaAddress.str());
     return cmd.getClear();
 }
+
+extern TPWRAPPER_API IStringIterator* getContainerTargetClusters(const char* processType, const char* processName)
+{
+    Owned<CStringArrayIterator> ret = new CStringArrayIterator;
+    Owned<IPropertyTreeIterator> queues = queryComponentConfig().getElements("queues");
+    ForEach(*queues)
+    {
+        IPropertyTree& queue = queues->query();
+        if (!isEmptyString(processType))
+        {
+            const char* type = queue.queryProp("@type");
+            if (isEmptyString(type) || !strieq(type, processType))
+                continue;
+        }
+        const char* qName = queue.queryProp("@name");
+        if (isEmptyString(qName))
+            continue;
+
+        if (!isEmptyString(processName) && !strieq(qName, processName))
+            continue;
+
+        ret->append_unique(qName);
+    }
+    if (!isEmptyString(processType) && !strieq("roxie", processType))
+        return ret.getClear();
+
+    Owned<IPropertyTreeIterator> services = queryComponentConfig().getElements("services[@type='roxie']");
+    ForEach(*services)
+    {
+        IPropertyTree& service = services->query();
+        const char* targetName = service.queryProp("@target");
+        if (isEmptyString(targetName))
+            continue;
+
+        if (!isEmptyString(processName) && !strieq(targetName, processName))
+            continue;
+
+        ret->append_unique(targetName);
+    }
+    return ret.getClear();
+}

esp/smc/SMCLib/TpWrapper.hpp

@@ -211,6 +211,7 @@ private:
 };
 
 extern TPWRAPPER_API ISashaCommand* archiveOrRestoreWorkunits(StringArray& wuids, IProperties* params, bool archive, bool dfu);
+extern TPWRAPPER_API IStringIterator *getContainerTargetClusters(const char* processType, const char* processName);
 
 #endif //_ESPWIZ_TpWrapper_HPP__
 

esp/src/package-lock.json

@@ -88,35 +88,35 @@
       }
     },
     "@hpcc-js/api": {
-      "version": "2.8.35",
-      "resolved": "https://registry.npmjs.org/@hpcc-js/api/-/api-2.8.35.tgz",
-      "integrity": "sha512-PiZj5ZBTwb7ODpoCj8PazY1LlES24mlaYvI5cFE02jE5q2U3SMbLXI9maS3xj7k8H2WzjIfiHUDRZ1D/q85M8g==",
+      "version": "2.8.37",
+      "resolved": "https://registry.npmjs.org/@hpcc-js/api/-/api-2.8.37.tgz",
+      "integrity": "sha512-NSX1NZ1owZAffmtTh5iab1g//bJEzDTt3clCzb/kf0XRWcsdYvWc0YGrtfruF2kHpdueyQPkl02BQhhwV7Sk4w==",
       "requires": {
-        "@hpcc-js/common": "^2.40.0"
+        "@hpcc-js/common": "^2.42.0"
       }
     },
     "@hpcc-js/chart": {
-      "version": "2.50.0",
-      "resolved": "https://registry.npmjs.org/@hpcc-js/chart/-/chart-2.50.0.tgz",
-      "integrity": "sha512-Z3EFtZKfwg0G8LqQO8eyOVTTv1OTIoUvxDGbE06/k5u+jJW/BElh2w9VZl9PBowKVxP+Juiug7f71JVJkzQLZA==",
+      "version": "2.52.0",
+      "resolved": "https://registry.npmjs.org/@hpcc-js/chart/-/chart-2.52.0.tgz",
+      "integrity": "sha512-klIwVn5mFfXMmcN/06gr3gQYihYUsYqq/9is8v1OuxKobll01O61WUWuTv2VCVVWI1dc4AdJdO9IT8V3lMr2Cg==",
       "requires": {
-        "@hpcc-js/api": "^2.8.35",
-        "@hpcc-js/common": "^2.40.0",
+        "@hpcc-js/api": "^2.8.37",
+        "@hpcc-js/common": "^2.42.0",
         "@hpcc-js/util": "^2.26.0"
       }
     },
     "@hpcc-js/codemirror": {
-      "version": "2.31.0",
-      "resolved": "https://registry.npmjs.org/@hpcc-js/codemirror/-/codemirror-2.31.0.tgz",
-      "integrity": "sha512-z9MOrXHTCAADcluajxIr9diioMSwWyO9xkU2jdhe2zcVn6qtb7wMx+wqlBlyAvPGsFy6YVqn60pdf1x54xfEfg==",
+      "version": "2.33.0",
+      "resolved": "https://registry.npmjs.org/@hpcc-js/codemirror/-/codemirror-2.33.0.tgz",
+      "integrity": "sha512-1LGk+dKPe7LeAF5fVzZdGPLaPRipc9QvwiexvMpUhzaHl7dbWVl0aYhDUim9n4cwUkX4IoyE49tFf0doWju4mQ==",
       "requires": {
-        "@hpcc-js/common": "^2.40.0"
+        "@hpcc-js/common": "^2.42.0"
       }
     },
     "@hpcc-js/common": {
-      "version": "2.40.0",
-      "resolved": "https://registry.npmjs.org/@hpcc-js/common/-/common-2.40.0.tgz",
-      "integrity": "sha512-Q/+illzsBYnkamLoLIrPH4vHTXxkOdlWteqHaTvOB87W+EIE0bNM1ZXPm0Ufj0DItD8FxM9R4jCGkEsjM4v74A==",
+      "version": "2.42.0",
+      "resolved": "https://registry.npmjs.org/@hpcc-js/common/-/common-2.42.0.tgz",
+      "integrity": "sha512-FTyurUquhyeBzXfDzlakbypSUzK2TQm+xEkWnQs793p/7BY40JkDpSdH2wtslSu0rCs4VyXuNCfyszN4HK90nQ==",
       "requires": {
         "@hpcc-js/util": "^2.26.0",
         "@types/d3-array": "1.2.6",
@@ -137,9 +137,9 @@
       }
     },
     "@hpcc-js/comms": {
-      "version": "2.28.0",
-      "resolved": "https://registry.npmjs.org/@hpcc-js/comms/-/comms-2.28.0.tgz",
-      "integrity": "sha512-52gx+d1DFFEyBok2c21Iq5JbT5hQybhqIol5sgaPvKgV0sg2xgWc1O9xaFzwe6Aq8ujBF28swJv3hwjqOUIXwQ==",
+      "version": "2.29.0",
+      "resolved": "https://registry.npmjs.org/@hpcc-js/comms/-/comms-2.29.0.tgz",
+      "integrity": "sha512-7jbGFAjL3MnfG/9Cs8zMc3/gURUkm0b2mQHN2B2/oDjuBkmVFvJz7nu7i9GViQCOYOczLNf+LSoMGXoGjNSMqQ==",
       "requires": {
         "@hpcc-js/ddl-shim": "^2.17.15",
         "@hpcc-js/util": "^2.26.0",
@@ -163,11 +163,11 @@
       }
     },
     "@hpcc-js/dgrid": {
-      "version": "2.8.34",
-      "resolved": "https://registry.npmjs.org/@hpcc-js/dgrid/-/dgrid-2.8.34.tgz",
-      "integrity": "sha512-4k1LeGDQV5ozfmYmgy4k6z3KNcynM1x4iwr9AgcwIYOo25zt93ZdMxi5Qrn+/cunpXSeh2E1irhGwdx6xcjzXg==",
+      "version": "2.8.37",
+      "resolved": "https://registry.npmjs.org/@hpcc-js/dgrid/-/dgrid-2.8.37.tgz",
+      "integrity": "sha512-Z0Tj2wSsAFvNXB+blugzpMy/iWIwLcx7newNFnKlags94QMje4PiuzBkAWDCfHVNKjT1TA3VF6tLzIB/jIi/AQ==",
       "requires": {
-        "@hpcc-js/common": "^2.40.0",
+        "@hpcc-js/common": "^2.42.0",
         "@hpcc-js/ddl-shim": "^2.17.15",
         "@hpcc-js/dgrid-shim": "^2.11.22",
         "@hpcc-js/util": "^2.26.0"
@@ -179,52 +179,52 @@
       "integrity": "sha512-D2S7JBWJyTV819G00xis9yj4NA/tV+o79kiwBeqma5tAMF9UPBgOdtj24BsXVPbjM3uLiCD7irsfY0g+DuFD5Q=="
     },
     "@hpcc-js/eclwatch": {
-      "version": "2.13.0",
-      "resolved": "https://registry.npmjs.org/@hpcc-js/eclwatch/-/eclwatch-2.13.0.tgz",
-      "integrity": "sha512-C1oqlSGGXUB23O3cUQubU4+McUiudRB+9AqpwO52OD9KVM1Gdc3XQzFAE9oPGe+W4OQWOzuVZTSw1UYhy7O5hA==",
-      "requires": {
-        "@hpcc-js/codemirror": "^2.31.0",
-        "@hpcc-js/common": "^2.40.0",
-        "@hpcc-js/comms": "^2.25.0",
-        "@hpcc-js/dgrid": "^2.8.34",
-        "@hpcc-js/graph": "^2.41.0",
-        "@hpcc-js/layout": "^2.16.46",
-        "@hpcc-js/phosphor": "^2.14.29",
-        "@hpcc-js/timeline": "^2.18.0",
-        "@hpcc-js/tree": "^2.13.0",
+      "version": "2.18.0",
+      "resolved": "https://registry.npmjs.org/@hpcc-js/eclwatch/-/eclwatch-2.18.0.tgz",
+      "integrity": "sha512-4rROLoNmtxR2BRwKVAQCyXWVL6+MVJ+vg8U0dfu1UtRsXW1UFNta/RmwsSfsXglmwspQGtSRR819V2A6bjpxgg==",
+      "requires": {
+        "@hpcc-js/codemirror": "^2.33.0",
+        "@hpcc-js/common": "^2.42.0",
+        "@hpcc-js/comms": "^2.29.0",
+        "@hpcc-js/dgrid": "^2.8.36",
+        "@hpcc-js/graph": "^2.43.0",
+        "@hpcc-js/layout": "^2.18.0",
+        "@hpcc-js/phosphor": "^2.14.31",
+        "@hpcc-js/timeline": "^2.20.0",
+        "@hpcc-js/tree": "^2.15.0",
         "@hpcc-js/util": "^2.26.0"
       }
     },
     "@hpcc-js/graph": {
-      "version": "2.41.0",
-      "resolved": "https://registry.npmjs.org/@hpcc-js/graph/-/graph-2.41.0.tgz",
-      "integrity": "sha512-SNxct35AU88vfM5fmoGGraV0WpmUv/dTluTACwR+bH0ciWveS9RbrWCyX5OeRKny4lSDjEbY+R22lH/Uz/QzZg==",
-      "requires": {
-        "@hpcc-js/api": "^2.8.35",
-        "@hpcc-js/common": "^2.40.0",
-        "@hpcc-js/html": "^2.15.0",
-        "@hpcc-js/react": "^2.22.0",
+      "version": "2.43.0",
+      "resolved": "https://registry.npmjs.org/@hpcc-js/graph/-/graph-2.43.0.tgz",
+      "integrity": "sha512-vlh2/FT7IxVvwTpODIukbWU811CEW76Z5jyuUadX40zSUIeHTetlKei97SajwdJ75PJ7V5c6gw/8NjaGBkF/Dg==",
+      "requires": {
+        "@hpcc-js/api": "^2.8.37",
+        "@hpcc-js/common": "^2.42.0",
+        "@hpcc-js/html": "^2.17.0",
+        "@hpcc-js/react": "^2.24.0",
         "@hpcc-js/util": "^2.26.0"
       }
     },
     "@hpcc-js/html": {
-      "version": "2.15.0",
-      "resolved": "https://registry.npmjs.org/@hpcc-js/html/-/html-2.15.0.tgz",
-      "integrity": "sha512-KX4vdOxx7rP5QkpBh8cAhfOl8468UiONvVV58aUnexIMzWy84ndsiOIoqTWiIrL0dVpDWiOvMQqtEp1iRH9QOQ==",
+      "version": "2.17.0",
+      "resolved": "https://registry.npmjs.org/@hpcc-js/html/-/html-2.17.0.tgz",
+      "integrity": "sha512-8xJykiZgGLAwfssl3J+UjlIaeLd5ZuHvo1D+S6bsiP2/doC3PHP3gXIZ4rqqfg0eHQJwib5iBpjx327E71uMrA==",
       "requires": {
-        "@hpcc-js/common": "^2.40.0",
+        "@hpcc-js/common": "^2.42.0",
         "@hpcc-js/preact-shim": "^2.13.13",
         "@hpcc-js/util": "^2.26.0"
       }
     },
     "@hpcc-js/layout": {
-      "version": "2.16.46",
-      "resolved": "https://registry.npmjs.org/@hpcc-js/layout/-/layout-2.16.46.tgz",
-      "integrity": "sha512-JHAxspWkeSlXmE85dgOCIG14LCsZHCahuSJMd9vhMcxlua7rypG0p/k31PjKVY67vD5GThBQGLLe7ZWj8/9K9A==",
+      "version": "2.18.0",
+      "resolved": "https://registry.npmjs.org/@hpcc-js/layout/-/layout-2.18.0.tgz",
+      "integrity": "sha512-kFJuZ+fOy3NxNgGvHmjmHuswYITN9dRsKgytYXngWW3ZkTSjj6Ez6SN3Yo9SGs3D1fObwYWjm9DV7hbkCzswPA==",
       "requires": {
-        "@hpcc-js/api": "^2.8.35",
-        "@hpcc-js/common": "^2.40.0",
-        "@hpcc-js/dgrid": "^2.8.34"
+        "@hpcc-js/api": "^2.8.37",
+        "@hpcc-js/common": "^2.42.0",
+        "@hpcc-js/dgrid": "^2.8.36"
       }
     },
     "@hpcc-js/leaflet-shim": {
@@ -238,36 +238,36 @@
       }
     },
     "@hpcc-js/map": {
-      "version": "2.31.0",
-      "resolved": "https://registry.npmjs.org/@hpcc-js/map/-/map-2.31.0.tgz",
-      "integrity": "sha512-WupSo3vj8jd0wwX+gnoijj+EZK/80an/Qp0vCM0Klded63yb8zKqPIZ5qOFSd232ZsTrmiHMKUpgdGBFTnpWHg==",
-      "requires": {
-        "@hpcc-js/api": "^2.8.35",
-        "@hpcc-js/common": "^2.40.0",
-        "@hpcc-js/graph": "^2.41.0",
-        "@hpcc-js/layout": "^2.16.46",
+      "version": "2.34.0",
+      "resolved": "https://registry.npmjs.org/@hpcc-js/map/-/map-2.34.0.tgz",
+      "integrity": "sha512-Lri1zAXzdUoN6s8m3TH/8eg3XLz7/G9AArLtXrY3wzk0PFvYZfgMNIZVRiJ0WGT0YzEa07DpAge9nMC3wra0yg==",
+      "requires": {
+        "@hpcc-js/api": "^2.8.37",
+        "@hpcc-js/common": "^2.42.0",
+        "@hpcc-js/graph": "^2.43.0",
+        "@hpcc-js/layout": "^2.18.0",
         "@hpcc-js/leaflet-shim": "^2.1.14",
-        "@hpcc-js/other": "^2.13.49",
+        "@hpcc-js/other": "^2.13.51",
         "@hpcc-js/util": "^2.26.0"
       }
     },
     "@hpcc-js/other": {
-      "version": "2.13.49",
-      "resolved": "https://registry.npmjs.org/@hpcc-js/other/-/other-2.13.49.tgz",
-      "integrity": "sha512-5x413MNKEpiS3k5D/WjKxTcbVDlTU9s7Qls1QoEW7CExPXBC3DE4fViEAxRQZ0wciHGJxgpilpNDEKEhsNTK4w==",
+      "version": "2.13.51",
+      "resolved": "https://registry.npmjs.org/@hpcc-js/other/-/other-2.13.51.tgz",
+      "integrity": "sha512-a25A+0l+QXxcCe+ldHxxDXO3k8l9JklmL3B+YHHRfXrS4SNnp5cLwHex4E11eFcAiSK3/uP9fREkml5gFYZXiw==",
       "requires": {
-        "@hpcc-js/api": "^2.8.35",
-        "@hpcc-js/common": "^2.40.0",
-        "@hpcc-js/layout": "^2.16.46"
+        "@hpcc-js/api": "^2.8.37",
+        "@hpcc-js/common": "^2.42.0",
+        "@hpcc-js/layout": "^2.18.0"
       }
     },
     "@hpcc-js/phosphor": {
-      "version": "2.14.29",
-      "resolved": "https://registry.npmjs.org/@hpcc-js/phosphor/-/phosphor-2.14.29.tgz",
-      "integrity": "sha512-siwqxAhXIrglOrqkXhl9ULIWjkVK5GKsgJEFBSOmMMq1r8+J2aEVeUTRTB5w2evphZiij4Ha5AjHDk9qPWAdTw==",
+      "version": "2.14.31",
+      "resolved": "https://registry.npmjs.org/@hpcc-js/phosphor/-/phosphor-2.14.31.tgz",
+      "integrity": "sha512-n0YO/8UPzHiQ9B9IWmNmyZ0+wVquvqNBIpO1J8HVBafET9cgFNFlILGdrM8nsLbKei7dsWOP4gvgE0CbSo4ICA==",
       "requires": {
-        "@hpcc-js/common": "^2.40.0",
-        "@hpcc-js/other": "^2.13.49",
+        "@hpcc-js/common": "^2.42.0",
+        "@hpcc-js/other": "^2.13.51",
         "@hpcc-js/phosphor-shim": "^2.11.17",
         "@hpcc-js/util": "^2.26.0"
       }
@@ -292,33 +292,46 @@
       }
     },
     "@hpcc-js/react": {
-      "version": "2.22.0",
-      "resolved": "https://registry.npmjs.org/@hpcc-js/react/-/react-2.22.0.tgz",
-      "integrity": "sha512-zsfshyUeaZ94PKkKAzZctPC/ufPVwhKyjO8XUv50EbrEQU9qegNUlpppA+08Vxkh7m3insMyh64Tn6MPrJjBTg==",
+      "version": "2.24.0",
+      "resolved": "https://registry.npmjs.org/@hpcc-js/react/-/react-2.24.0.tgz",
+      "integrity": "sha512-qh91cR+B8NFNv0yQKsEc8eLJehJxmuQyfxnnJaBfUMYwe1I8vVwsohCznQIauY3k3DQmBmUkOmVwJV7KH6YNnw==",
       "requires": {
-        "@hpcc-js/common": "^2.40.0",
+        "@hpcc-js/common": "^2.42.0",
         "@hpcc-js/preact-shim": "^2.13.13"
       }
     },
     "@hpcc-js/timeline": {
-      "version": "2.18.0",
-      "resolved": "https://registry.npmjs.org/@hpcc-js/timeline/-/timeline-2.18.0.tgz",
-      "integrity": "sha512-entE3FHWiNrf6H3sNBa0KlOzSi1SluznR9hhoEz71xwWqC6v7g/crua67dLjBYXMpjpUOewJQlOTYumllcd9mw==",
-      "requires": {
-        "@hpcc-js/api": "^2.8.35",
-        "@hpcc-js/chart": "^2.50.0",
-        "@hpcc-js/common": "^2.40.0",
-        "@hpcc-js/html": "^2.15.0",
-        "@hpcc-js/react": "^2.22.0"
+      "version": "2.21.0",
+      "resolved": "https://registry.npmjs.org/@hpcc-js/timeline/-/timeline-2.21.0.tgz",
+      "integrity": "sha512-+BahIJZb2TXOAb7ykemReuZ6Td93tJBt0abdHEs7nc0rT9XOhB0gfsFQ/1EqWCZIbzHSSLAMAlyFw8PIwdeT6w==",
+      "requires": {
+        "@hpcc-js/api": "^2.8.37",
+        "@hpcc-js/chart": "^2.52.0",
+        "@hpcc-js/common": "^2.42.0",
+        "@hpcc-js/html": "^2.17.0",
+        "@hpcc-js/layout": "^2.19.0",
+        "@hpcc-js/react": "^2.24.0"
+      },
+      "dependencies": {
+        "@hpcc-js/layout": {
+          "version": "2.19.0",
+          "resolved": "https://registry.npmjs.org/@hpcc-js/layout/-/layout-2.19.0.tgz",
+          "integrity": "sha512-rVcFqBiGJ78MJZZZ35ENZg9KmUrgx0R7TGxqcKljI+UYKpzRU25pzEmin+sF1fBY7soiGH+lq7Fcdt5A1Qh8oQ==",
+          "requires": {
+            "@hpcc-js/api": "^2.8.37",
+            "@hpcc-js/common": "^2.42.0",
+            "@hpcc-js/dgrid": "^2.8.37"
+          }
+        }
       }
     },
     "@hpcc-js/tree": {
-      "version": "2.13.0",
-      "resolved": "https://registry.npmjs.org/@hpcc-js/tree/-/tree-2.13.0.tgz",
-      "integrity": "sha512-QAKe2fg4yjLEx/d8l4ERag9cKSuLWAqdIx9AV5D3VmcYq1c1yo34AxWtKG9XYasjrSmT314u9Hcpho2xiby+cA==",
+      "version": "2.15.0",
+      "resolved": "https://registry.npmjs.org/@hpcc-js/tree/-/tree-2.15.0.tgz",
+      "integrity": "sha512-T6LeliE6uSnjo81Gv4JNawO4hY7xKb9W1+L0iGLCZN9o7vxRpAU8BPgXO0omSyfdC+JUyB/vdxDpq4JfDddD4A==",
       "requires": {
-        "@hpcc-js/api": "^2.8.35",
-        "@hpcc-js/common": "^2.40.0"
+        "@hpcc-js/api": "^2.8.37",
+        "@hpcc-js/common": "^2.42.0"
       }
     },
     "@hpcc-js/util": {

esp/src/package.json

@@ -33,20 +33,20 @@
   },
   "main": "src/stub.js",
   "dependencies": {
-    "@hpcc-js/chart": "2.50.0",
-    "@hpcc-js/codemirror": "2.31.0",
-    "@hpcc-js/common": "2.40.0",
-    "@hpcc-js/comms": "^2.28.0",
+    "@hpcc-js/chart": "2.52.0",
+    "@hpcc-js/codemirror": "2.33.0",
+    "@hpcc-js/common": "2.42.0",
+    "@hpcc-js/comms": "2.29.0",
     "@hpcc-js/dataflow": "2.5.0",
-    "@hpcc-js/eclwatch": "2.13.0",
-    "@hpcc-js/graph": "2.41.0",
-    "@hpcc-js/html": "2.15.0",
-    "@hpcc-js/layout": "2.16.46",
-    "@hpcc-js/map": "2.31.0",
-    "@hpcc-js/other": "2.13.49",
-    "@hpcc-js/phosphor": "2.14.29",
-    "@hpcc-js/react": "2.22.0",
-    "@hpcc-js/tree": "2.13.0",
+    "@hpcc-js/eclwatch": "2.18.0",
+    "@hpcc-js/graph": "2.43.0",
+    "@hpcc-js/html": "2.17.0",
+    "@hpcc-js/layout": "2.18.0",
+    "@hpcc-js/map": "2.34.0",
+    "@hpcc-js/other": "2.13.51",
+    "@hpcc-js/phosphor": "2.14.31",
+    "@hpcc-js/react": "2.24.0",
+    "@hpcc-js/tree": "2.15.0",
     "@hpcc-js/util": "2.26.0",
     "@material-ui/core": "4.8.3",
     "@material-ui/icons": "4.9.1",
@@ -93,4 +93,4 @@
     "type": "git",
     "url": "https://github.com/hpcc-systems/HPCC-Platform"
   }
-}
+}

esp/src/src/Timings.ts

@@ -24,6 +24,11 @@ export class WUTimelinePatched extends WUTimeline {
 
     constructor() {
         super();
+        this._gantt.bucketHeight(16);
+        this.strokeWidth(0);
+        this.tooltipHTML(d => {
+            return d[d.length - 1].calcTooltip(); 
+        });
     }
 
     data(): any;

helm/hpcc/templates/roxie.yaml

@@ -2,6 +2,7 @@
 {{- if not $roxie.disabled -}}
 {{- include "hpcc.checkDefaultStoragePlane" (dict "root" $ "me" $roxie )}}
 {{- if not $roxie.localAgent -}}
+{{- $_ := set $roxie "localAgent" false -}}
 {{- $toponame := printf "%s-toposerver" $roxie.name -}}
 {{- $numChannels := $roxie.numChannels | int | default 1 -}}
 {{- $topoport := $roxie.topoport | int | default 9004 -}}

helm/hpcc/templates/thor.yaml

@@ -13,7 +13,7 @@
 {{- $eclAgentScope := dict "name" $eclAgentName "useChildProcesses" $eclAgentUseChildProcesses "replicas" $eclAgentReplicas | merge (pick . "keepJobs") }}
 {{- $thorAgentScope := dict "name" $thorAgentName "replicas" $thorAgentReplicas  | merge (pick . "keepJobs") }}
 {{- $hthorScope := dict "name" $hthorName | merge (pick . "multiJobLinger") }}
-{{- $thorScope := omit . "eclagent" "thoragent" "hthor" "logging" "eclAgentResources" "managerResources" "workerResources" "eclAgentUseChildProcesses" "eclAgentReplicas" "thorAgentReplicas" "eclAgentType" }}
+{{- $thorScope := omit . "eclagent" "thoragent" "hthor" "logging" "eclAgentResources" "eclAgentUseChildProcesses" "eclAgentReplicas" "thorAgentReplicas" "eclAgentType" }}
 apiVersion: apps/v1
 kind: Deployment
 metadata:

roxie/ccd/ccd.hpp

@@ -51,7 +51,15 @@
 
 #define ROXIE_STATEFILE_VERSION 2
 
+// Have not yet tested impact of new IBYTI handling in non-containerized systems
+
 #ifdef _CONTAINERIZED
+#define NEW_IBYTI
+#endif
+
+#if defined(_CONTAINERIZED) || defined (NEW_IBYTI)
+// Both containerized mode and new IBYTI mode assume subchannels are passed in header.
+// It SHOULD also work, and may be beneficial, in non-containerized systems but has not as yet been confirmed.
 #define SUBCHANNELS_IN_HEADER
 #endif
 
@@ -139,6 +147,8 @@ public:
     }
 };
 
+extern bool localAgent;
+
 class RoxiePacketHeader
 {
 private:
@@ -187,6 +197,8 @@ public:
 #ifdef SUBCHANNELS_IN_HEADER
     unsigned mySubChannel() const // NOTE - 0 based
     {
+        if (localAgent)
+            return 0;
         for (unsigned idx = 0; idx < MAX_SUBCHANNEL; idx++)
         {
             if (subChannels[idx].isMe())
@@ -197,6 +209,8 @@ public:
 
     bool hasBuddies() const
     {
+        if (localAgent)
+            return false;
         if (subChannels[1].isNull())
         {
             assert(subChannels[0].isMe());
@@ -267,7 +281,6 @@ extern bool debugPermitted;
 extern bool useRemoteResources;
 extern bool checkFileDate;
 extern bool lazyOpen;
-extern bool localAgent;
 extern bool useAeron;
 extern bool ignoreOrphans;
 extern bool doIbytiDelay;

roxie/ccd/ccddali.cpp

@@ -33,7 +33,13 @@
 #include "workflow.hpp"
 #include "mpcomm.hpp"
 
+#ifndef _CONTAINERIZED
+#define ROXIE_DALI_CACHE
+#endif
+
+#ifdef ROXIE_DALI_CACHE
 const char *roxieStateName = "RoxieLocalState.xml";
+#endif
 
 class CDaliPackageWatcher : public CInterface, implements ISafeSDSSubscription, implements ISDSNodeSubscription, implements IDaliPackageWatcher
 {
@@ -235,6 +241,7 @@ private:
         connectWatcher.join();
     }
 
+#ifdef ROXIE_DALI_CACHE
     // The cache is static since it outlives the dali connections
 
     static CriticalSection cacheCrit;
@@ -287,6 +294,12 @@ private:
                 cache->addPropTree(newLoc, LINK(val));
         }
     }
+#else
+    inline static void initCache() {}
+    inline static void loadCache() {}
+    inline IPropertyTree *readCache(const char *xpath) { return nullptr; }
+    inline static void writeCache(const char *foundLoc, const char *newLoc, IPropertyTree *val) {}
+#endif
 
     IPropertyTree *loadDaliTree(const char *path, const char *id)
     {
@@ -588,6 +601,7 @@ public:
 
     virtual void commitCache()
     {
+#ifdef ROXIE_DALI_CACHE
         if (isConnected && cache && !oneShotRoxie)
         {
             CriticalBlock b(cacheCrit);
@@ -606,6 +620,7 @@ public:
                 cacheFile->rename(oldCacheFileName);
             newFile->rename(cacheFileName);
         }
+#endif
     }
 
     virtual IConstWorkUnit *attachWorkunit(const char *wuid, ILoadedDllEntry *source)
@@ -702,8 +717,10 @@ public:
 
     static void releaseCache()
     {
+#ifdef ROXIE_DALI_CACHE
         CriticalBlock b(cacheCrit);
         cache.clear();
+#endif
     }
 
     virtual void releaseSubscription(IDaliPackageWatcher *subscription)
@@ -903,8 +920,10 @@ public:
 bool CRoxieDaliHelper::isConnected = false;
 CRoxieDaliHelper * CRoxieDaliHelper::daliHelper;
 CriticalSection CRoxieDaliHelper::daliHelperCrit;
+#ifdef ROXIE_DALI_CACHE
 CriticalSection CRoxieDaliHelper::cacheCrit;
 Owned<IPropertyTree> CRoxieDaliHelper::cache;
+#endif
 
 CriticalSection CRoxieDllServer::crit;
 

roxie/ccd/ccdmain.cpp

@@ -653,6 +653,11 @@ int CCD_API roxie_main(int argc, const char *argv[], const char * defaultYaml)
         topology = loadConfiguration(useOldTopology ? nullptr : defaultYaml, argv, "roxie", "ROXIE", topologyFile, nullptr, "@netAddress");
         saveTopology();
         localAgent = topology->getPropBool("@localAgent", topology->getPropBool("@localSlave", false));  // legacy name
+        numChannels = topology->getPropInt("@numChannels", 0);
+#ifdef _CONTAINERIZED
+        if (!numChannels)
+            throw makeStringException(MSGAUD_operator, ROXIE_INVALID_TOPOLOGY, "Invalid topology file - numChannels not set");
+#endif
         const char *channels = topology->queryProp("@channels");
         if (channels)
         {
@@ -675,7 +680,10 @@ int CCD_API roxie_main(int argc, const char *argv[], const char * defaultYaml)
         }
 #ifdef _CONTAINERIZED
         else if (localAgent)
-            agentChannels.push_back(std::pair<unsigned, unsigned>(1, 0));
+        {
+            for (unsigned channel = 1; channel <= numChannels; channel++)
+                agentChannels.push_back(std::pair<unsigned, unsigned>(channel, 0));
+        }
 #endif
         const char *topos = topology->queryProp("@topologyServers");
         StringArray topoValues;
@@ -877,7 +885,6 @@ int CCD_API roxie_main(int argc, const char *argv[], const char * defaultYaml)
             envInstallNASHooks(nas);
         }
         useAeron = topology->getPropBool("@useAeron", false);
-        numChannels = topology->getPropInt("@numChannels", 0);
         doIbytiDelay = topology->getPropBool("@doIbytiDelay", true);
         minIbytiDelay = topology->getPropInt("@minIbytiDelay", 2);
         initIbytiDelay = topology->getPropInt("@initIbytiDelay", 50);
@@ -1149,8 +1156,6 @@ int CCD_API roxie_main(int argc, const char *argv[], const char * defaultYaml)
 #endif
 
 #ifdef _CONTAINERIZED
-        if (!numChannels)
-            throw makeStringException(MSGAUD_operator, ROXIE_INVALID_TOPOLOGY, "Invalid topology file - numChannels not set");
         IpAddress myIP(".");
         myNode.setIp(myIP);
         if (topology->getPropBool("@server", true))

roxie/ccd/ccdqueue.cpp

@@ -1194,7 +1194,9 @@ class CRoxieWorker : public CInterface, implements IPooledThread
 {
     RoxieQueue *queue;
     CriticalSection actCrit;
+#ifndef NEW_IBYTI
     Semaphore ibytiSem;
+#endif
     bool stopped;
     bool abortJob;
     bool busy;
@@ -1247,8 +1249,10 @@ public:
         if (packet && packet->queryHeader().channel==channel)
         {
             abortJob = true;
+#ifndef NEW_IBYTI
             if (doIbytiDelay) 
                 ibytiSem.signal();
+#endif
             if (activity) 
                 activity->abort();
         }
@@ -1261,8 +1265,10 @@ public:
         {
             queryFound = true;
             abortJob = true;
+#ifndef NEW_IBYTI
             if (doIbytiDelay)
                 ibytiSem.signal();
+#endif
             if (activity) 
             {
                 // Try to stop/abort a job after it starts only if IBYTI comes from a higher priority agent 
@@ -1386,18 +1392,19 @@ public:
         try
         {   
             bool debugging = logctx.queryDebuggerActive();
-#ifdef SUBCHANNELS_IN_HEADER
             if (debugging)
             {
-                if (!mySubChannel)
-                    abortJob = true;  // when debugging, we always run on primary only... should really have sent to primary only too...
+                if (mySubChannel)
+                    abortJob = true;  // when debugging, we always run on primary only...
             }
+#ifndef NEW_IBYTI
+#ifdef SUBCHANNELS_IN_HEADER
             else if (doIbytiDelay && mySubChannel)
             {
                 unsigned delay = 0;
                 for (unsigned subChannel = 0; subChannel < mySubChannel; subChannel++)
                     delay += getIbytiDelay(header.subChannels[subChannel].getIpAddress());
-                unsigned start;
+                unsigned start = 0;
                 if (traceRoxiePackets)
                 {
                     StringBuffer x;
@@ -1418,11 +1425,6 @@ public:
                 }
             }
 #else
-            if (debugging)
-            {
-                if (mySubChannel)
-                    abortJob = true;  // when debugging, we always run on primary only...
-            }
             else if (doIbytiDelay && (numAgents > 1))
             {
                 unsigned hdrHashVal = header.priorityHash();
@@ -1468,6 +1470,7 @@ public:
                 }
             }
 #endif
+#endif
             if (abortJob)
             {
                 CriticalBlock b(actCrit);
@@ -1534,8 +1537,10 @@ public:
                     maxAgentsActive.store_max(agentsActive);
                     abortJob = false;
                     busy = true;
+#ifndef NEW_IBYTI
                     if (doIbytiDelay) 
                         ibytiSem.reinit(0U); // Make sure sem is is in no-signaled state
+#endif
                     packet.setown(queue->dequeue());
                     if (packet)
                     {
@@ -1930,6 +1935,293 @@ public:
     }
 };
 
+//------------------------------------------------------------------------------------------------------------
+#ifdef NEW_IBYTI
+
+class DelayedPacketQueue
+{
+    // Used to keep a list of all recently-received packets where we are not primary subchannel. There is one queue per subchannel level
+    // It is accessed ONLY from the main reader thread and does not need to be threadsafe (but does need to be fast)
+    // We use a doubly-linked list (not std::list as not quite flexible enough).
+
+    class DelayedPacketEntry
+    {
+        DelayedPacketEntry() = delete;
+        DelayedPacketEntry(const DelayedPacketEntry&) = delete;
+    public:
+        DelayedPacketEntry(IRoxieQueryPacket *_packet, unsigned _waitExpires)
+        : packet(_packet), waitExpires(_waitExpires)
+        {
+        }
+        ~DelayedPacketEntry()
+        {
+            if (prev)
+                prev->next = next;
+            if (next)
+                next->prev = prev;
+        }
+        bool matches(const RoxiePacketHeader &ibyti) const
+        {
+            return packet->queryHeader().matchPacket(ibyti);
+        }
+        IRoxieQueryPacket *getClear()
+        {
+            return packet.getClear();
+        }
+        StringBuffer & describe(StringBuffer &ret) const
+        {
+            return packet->queryHeader().toString(ret);
+        }
+
+        Owned<IRoxieQueryPacket> packet;
+        DelayedPacketEntry *next = nullptr;
+        DelayedPacketEntry *prev = nullptr;
+
+        unsigned waitExpires = 0;
+    };
+
+public:
+    DelayedPacketQueue() = default;
+    DelayedPacketQueue(const DelayedPacketQueue&) = delete;
+    ~DelayedPacketQueue()
+    {
+        while (head)
+            removeEntry(head);
+    }
+    bool doIBYTI(const RoxiePacketHeader &ibyti)
+    {
+        assert(GetCurrentThreadId()==roxiePacketReaderThread);
+        DelayedPacketEntry *finger = head;
+        while (finger)
+        {
+            if (finger->matches(ibyti))
+            {
+                if (traceRoxiePackets)
+                {
+                    StringBuffer s;
+                    DBGLOG("IBYTI removing delayed packet %s", finger->describe(s).str());
+                }
+                removeEntry(finger);
+                return true;
+            }
+            finger = finger->next;
+        }
+        return false;
+    }
+
+    void append(IRoxieQueryPacket *packet, unsigned expires)
+    {
+        // Goes on the end. But percolate the expiry time backwards
+        assert(GetCurrentThreadId()==roxiePacketReaderThread);
+        DelayedPacketEntry *newEntry = new DelayedPacketEntry(packet, expires);
+        if (traceRoxiePackets)
+        {
+            StringBuffer s;
+            DBGLOG("Adding delayed packet %s", packet->queryHeader().toString(s).str());
+        }
+        newEntry->prev = tail;
+        if (tail)
+        {
+            tail->next = newEntry;
+            for (DelayedPacketEntry *finger = tail; finger != nullptr; finger = finger->prev)
+            {
+                if ((int) (finger->waitExpires - expires) <= 0)
+                    break;
+                finger->waitExpires = expires;
+                finger = finger->prev;
+            }
+        }
+        else
+            head = newEntry;
+        tail = newEntry;
+    }
+
+    // Move any that we are done waiting for our buddy onto the active queue
+    void checkExpired(
+            unsigned now,
+#ifdef ROXIE_SLA_LOGIC
+            RoxieQueue &slaQueue,
+#endif
+            RoxieQueue &hiQueue, RoxieQueue &loQueue)
+    {
+        assert(GetCurrentThreadId()==roxiePacketReaderThread);
+        DelayedPacketEntry *finger = head;
+        while (finger)
+        {
+            if (((int) (finger->waitExpires - now)) <= 0)   // Oddly coded to handle wrapping
+            {
+                IRoxieQueryPacket *packet = finger->getClear();
+                const RoxiePacketHeader &header = packet->queryHeader();
+                if (traceRoxiePackets)
+                {
+                    StringBuffer s;
+                    DBGLOG("No IBYTI received yet for delayed packet %s", header.toString(s).str());
+                }
+#ifdef ROXIE_SLA_LOGIC
+                if (header.activityId & ROXIE_SLA_PRIORITY)
+                    slaQueue.enqueue(packet);
+                else
+#endif
+                if (header.activityId & ROXIE_HIGH_PRIORITY)
+                    hiQueue.enqueue(packet);
+                else
+                    loQueue.enqueue(packet);
+                for (unsigned subChannel = 0; subChannel < MAX_SUBCHANNEL; subChannel++)
+                {
+                    if (header.subChannels[subChannel].isMe() || header.subChannels[subChannel].isNull())
+                        break;
+                    noteNodeSick(header.subChannels[subChannel].getIpAddress());
+                }
+
+                DelayedPacketEntry *goer = finger;
+                finger = finger->next;
+                removeEntry(goer);
+            }
+            else
+                break;
+        }
+    }
+
+    // How long until the next time we want to call checkExpires() ?
+    unsigned timeout(unsigned now) const
+    {
+        assert(GetCurrentThreadId()==roxiePacketReaderThread);
+        if (head)
+        {
+            int delay = (int) (head->waitExpires - now);
+            if (delay <= 0)
+                return 0;
+            else
+                return (unsigned) delay;
+        }
+        else
+            return (unsigned) -1;
+    }
+
+private:
+    void removeEntry(DelayedPacketEntry *goer)
+    {
+        if (goer==head)
+            head = goer->next;
+        if (goer==tail)
+            tail = goer->prev;
+        delete goer;
+    }
+
+    DelayedPacketEntry *head = nullptr;
+    DelayedPacketEntry *tail = nullptr;
+
+};
+
+//------------------------------------------------------------------------------------------------------------
+
+class DelayedPacketQueueChannel : public CInterface
+{
+    // Manages a set of DelayedPacketQueues, one for each supported subchannel level.
+    DelayedPacketQueueChannel() = delete;
+    DelayedPacketQueueChannel(const DelayedPacketQueueChannel&) = delete;
+public:
+    DelayedPacketQueueChannel(unsigned _channel) : channel(_channel)
+    {
+    }
+    inline unsigned queryChannel() const { return channel; }
+    inline DelayedPacketQueue &queryQueue(unsigned subchannel)
+    {
+        assertex(subchannel);  // Subchannel 0 means primary and is never delayed
+        subchannel -= 1;
+        if (subchannel > maxSeen)
+            maxSeen = subchannel;
+        return queues[subchannel];
+    }
+    unsigned timeout(unsigned now) const
+    {
+        unsigned min = (unsigned) -1;
+        for (unsigned queue = 0; queue <= maxSeen; queue++)
+        {
+            unsigned t = queues[queue].timeout(now);
+            if (t < min)
+                min = t;
+        }
+        return min;
+    }
+    void checkExpired(
+            unsigned now,
+#ifdef ROXIE_SLA_LOGIC
+            RoxieQueue &slaQueue,
+#endif
+            RoxieQueue &hiQueue, RoxieQueue &loQueue)
+    {
+        for (unsigned queue = 0; queue <= maxSeen; queue++)
+        {
+            queues[queue].checkExpired(
+                    now,
+#ifdef ROXIE_SLA_LOGIC
+                    slaQueue,
+#endif
+                    hiQueue, loQueue);
+        }
+    }
+private:
+    DelayedPacketQueue queues[MAX_SUBCHANNEL-1];   // Note - primary subchannel is not included
+    unsigned channel = 0;
+    unsigned maxSeen = 0;
+};
+
+class DelayedPacketQueueManager
+{
+
+public:
+    DelayedPacketQueueManager() = default;
+    DelayedPacketQueueManager(const DelayedPacketQueueManager&) = delete;
+    inline DelayedPacketQueue &queryQueue(unsigned channel, unsigned subchannel)
+    {
+        // Note - there are normally no more than a couple of channels on a single agent.
+        // If that were to change we could make this a fixed size array
+        assert(GetCurrentThreadId()==roxiePacketReaderThread);
+        ForEachItemIn(idx, channels)
+        {
+            DelayedPacketQueueChannel &i = channels.item(idx);
+            if (i.queryChannel() == channel)
+                return i.queryQueue(subchannel);
+        }
+        channels.append(*new DelayedPacketQueueChannel(channel));
+        return channels.tos().queryQueue(subchannel);
+    }
+    unsigned timeout(unsigned now) const
+    {
+        unsigned ret = (unsigned) -1;
+        ForEachItemIn(idx, channels)
+        {
+            unsigned t = channels.item(idx).timeout(now);
+            if (t < ret)
+                ret = t;
+        }
+        return ret;
+    }
+    void checkExpired(
+            unsigned now,
+#ifdef ROXIE_SLA_LOGIC
+            RoxieQueue &slaQueue,
+#endif
+            RoxieQueue &hiQueue, RoxieQueue &loQueue)
+    {
+        ForEachItemIn(idx, channels)
+        {
+            channels.item(idx).checkExpired(
+                    now,
+#ifdef ROXIE_SLA_LOGIC
+                    slaQueue,
+#endif
+                    hiQueue, loQueue);
+        }
+    }
+private:
+    CIArrayOf<DelayedPacketQueueChannel> channels;
+};
+#endif
+
+//------------------------------------------------------------------------------------------------------------
+
 class RoxieSocketQueueManager : public RoxieReceiverBase
 {
 protected:
@@ -1939,6 +2231,9 @@ protected:
     Owned<TokenBucket> bucket;
     unsigned maxPacketSize = 0;
     std::atomic<bool> running = { false };
+#ifdef NEW_IBYTI
+    DelayedPacketQueueManager delayed;
+#endif
 
     class ReceiverThread : public Thread
     {
@@ -2098,25 +2393,28 @@ public:
         return ret;
     }
 
-    void doIbyti(RoxiePacketHeader &header, RoxieQueue &queue
-#ifndef SUBCHANNELS_IN_HEADER
-                 , const ITopologyServer* topology
-#endif
-                 )
+    void doIbyti(RoxiePacketHeader &header, RoxieQueue &queue)
     {
         assert(!localAgent);
         bool preActivity = false;
 #ifdef SUBCHANNELS_IN_HEADER
         unsigned mySubChannel = header.mySubChannel();
 #else
+        Owned<const ITopologyServer> topology = getTopology();
         const ChannelInfo &channelInfo = topology->queryChannelInfo(header.channel);
         unsigned mySubChannel = channelInfo.subChannel();
 #endif
 
         if (header.retries == QUERY_ABORTED)
         {
-            abortRunning(header, queue, false, preActivity);
-            queue.remove(header);
+            bool foundInQ = false;
+#ifdef NEW_IBYTI
+            foundInQ = mySubChannel != 0 && delayed.queryQueue(header.channel, mySubChannel).doIBYTI(header);
+#endif
+            if (!foundInQ)
+                foundInQ = queue.remove(header);
+            if (!foundInQ)
+                abortRunning(header, queue, false, preActivity);
 
             if (traceRoxiePackets || traceLevel > 10)
             {
@@ -2141,7 +2439,12 @@ public:
 #else
                 noteNodeHealthy(header.subChannels[subChannel].getIpAddress());
 #endif
-                bool foundInQ = queue.remove(header);
+                bool foundInQ = false;
+#ifdef NEW_IBYTI
+                foundInQ = mySubChannel != 0 && delayed.queryQueue(header.channel, mySubChannel).doIBYTI(header);
+#endif
+                if (!foundInQ)
+                    foundInQ = queue.remove(header);
                 if (foundInQ)
                 {
                     if (traceRoxiePackets || traceLevel > 10)
@@ -2181,84 +2484,100 @@ public:
     {
         // NOTE - this thread needs to do as little as possible - just read packets and queue them up - otherwise we can get packet loss due to buffer overflow
         // DO NOT put tracing on this thread except at very high tracelevels!
-#ifdef SUBCHANNELS_IN_HEADER
-        unsigned mySubchannel = header.mySubChannel();
-#else
-        Owned<const ITopologyServer> topology = getTopology();
-        if (!header.channel)
-        {
-            // Turn broadcast packet (channel 0), as early as possible, into non-0 channel packets.
-            // So retries and other communication with Roxie server (which uses non-0 channel numbers) will not cause double work or confusion.
-            // Unfortunately this is bad news for dropping packets
-            const std::vector<unsigned> channels = topology->queryChannels();
-            Owned<IRoxieQueryPacket> packet = createRoxiePacket(mb);
-            for (unsigned i = 1; i < channels.size(); i++)
-                queue.enqueue(packet->clonePacket(channels[i]));
-            header.channel = channels[0];
-            queue.enqueue(packet.getClear());
-            return;
-        }
-        unsigned mySubchannel = topology->queryChannelInfo(header.channel).subChannel();
-#endif
-        if (header.activityId == ROXIE_FILECALLBACK || header.activityId == ROXIE_DEBUGCALLBACK )
+        if ((header.activityId & ~ROXIE_PRIORITY_MASK) == 0)
+            doIbyti(header, queue);
+        else
         {
-            Owned<IRoxieQueryPacket> packet = createRoxiePacket(mb);
-            if (traceLevel > 10)
+            if (!header.channel)
             {
-                StringBuffer s; 
-                DBGLOG("ROXIE_CALLBACK %s", header.toString(s).str());
+                // Turn broadcast packet (channel 0), as early as possible, into non-0 channel packets.
+                // So retries and other communication with Roxie server (which uses non-0 channel numbers) will not cause double work or confusion.
+                // Unfortunately this is bad news for dropping packets
+            // In SUBCHANNELS_IN_HEADER mode this translation has been done on server before sending, except for some control messages like PING or UNLOAD
+
+                Owned<const ITopologyServer> topology = getTopology();
+                const std::vector<unsigned> channels = topology->queryChannels();
+                Owned<IRoxieQueryPacket> packet = createRoxiePacket(mb);
+                for (unsigned i = 1; i < channels.size(); i++)
+                    queue.enqueue(packet->clonePacket(channels[i]));
+                header.channel = channels[0];
+                queue.enqueue(packet.getClear());
+                return;
             }
-            doFileCallback(packet);
-        }
-        else if ((header.activityId & ~ROXIE_PRIORITY_MASK) == 0)
-            doIbyti(header, queue
-#ifndef SUBCHANNELS_IN_HEADER
-                    , topology
+#ifdef SUBCHANNELS_IN_HEADER
+            unsigned mySubchannel = header.mySubChannel();
+#else
+            Owned<const ITopologyServer> topology = getTopology();
+            unsigned mySubchannel = topology->queryChannelInfo(header.channel).subChannel();
 #endif
-                   ); // MORE - check how fast this is!
-        else if (IBYTIbufferSize && queue.lookupOrphanIBYTI(header))
-        {
-            if (traceRoxiePackets || traceLevel > 10)
+            if (header.activityId == ROXIE_FILECALLBACK || header.activityId == ROXIE_DEBUGCALLBACK )
             {
-                StringBuffer s;
-                DBGLOG("doIBYTI packet was too early : %s", header.toString(s).str());
+                Owned<IRoxieQueryPacket> packet = createRoxiePacket(mb);
+                if (traceLevel > 10)
+                {
+                    StringBuffer s;
+                    DBGLOG("ROXIE_CALLBACK %s", header.toString(s).str());
+                }
+                doFileCallback(packet);
             }
-            ibytiPacketsTooLate--;
-            ibytiPacketsTooEarly++;
-        }
-        else
-        {
-            Owned<IRoxieQueryPacket> packet = createRoxiePacket(mb);
-            AgentContextLogger logctx(packet);
-            unsigned retries = header.thisChannelRetries(mySubchannel);
-            if (retries)
+            else if (IBYTIbufferSize && queue.lookupOrphanIBYTI(header))
             {
-                // MORE - is this fast enough? By the time I am seeing retries I may already be under load. Could move onto a separate thread
-                assertex(header.channel); // should never see a retry on channel 0
-                if (retries >= SUBCHANNEL_MASK)
-                    return; // someone sent a failure or something - ignore it
-
-                // Send back an out-of-band immediately, to let Roxie server know that channel is still active
-                if (!(testAgentFailure & 0x800))
+                if (traceRoxiePackets || traceLevel > 10)
                 {
-                    RoxiePacketHeader newHeader(header, ROXIE_ALIVE, mySubchannel);
-                    Owned<IMessagePacker> output = ROQ->createOutputStream(newHeader, true, logctx);
-                    output->flush();
+                    StringBuffer s;
+                    DBGLOG("doIBYTI packet was too early : %s", header.toString(s).str());
                 }
-
-                // If it's a retry, look it up against already running, or output stream, or input queue
-                // if found, send an IBYTI and discard retry request
-                
-                if (!mySubchannel)
-                    retriesReceivedPrm++;
-                else  
-                    retriesReceivedSec++;
-                bool alreadyRunning = false;
-                Owned<IPooledThreadIterator> wi = queue.running();
-                ForEach(*wi)
+                ibytiPacketsTooLate--;
+                ibytiPacketsTooEarly++;
+            }
+            else
+            {
+                Owned<IRoxieQueryPacket> packet = createRoxiePacket(mb);
+                AgentContextLogger logctx(packet);
+                unsigned retries = header.thisChannelRetries(mySubchannel);
+                if (retries)
                 {
-                    CRoxieWorker &w = (CRoxieWorker &) wi->query();
-                    if (w.match(header))
+                    // MORE - is this fast enough? By the time I am seeing retries I may already be under load. Could move onto a separate thread
+                    assertex(header.channel); // should never see a retry on channel 0
+                    if (retries >= SUBCHANNEL_MASK)
+                        return; // someone sent a failure or something - ignore it
+
+                    // Send back an out-of-band immediately, to let Roxie server know that channel is still active
+                    if (!(testAgentFailure & 0x800))
+                    {
+                        RoxiePacketHeader newHeader(header, ROXIE_ALIVE, mySubchannel);
+                        Owned<IMessagePacker> output = ROQ->createOutputStream(newHeader, true, logctx);
+                        output->flush();
+                    }
+
+                    // If it's a retry, look it up against already running, or output stream, or input queue
+                    // if found, send an IBYTI and discard retry request
+
+                    if (!mySubchannel)
+                        retriesReceivedPrm++;
+                    else
+                        retriesReceivedSec++;
+                    bool alreadyRunning = false;
+                    Owned<IPooledThreadIterator> wi = queue.running();
+                    ForEach(*wi)
+                    {
+                        CRoxieWorker &w = (CRoxieWorker &) wi->query();
+                        if (w.match(header))
+                        {
+                            alreadyRunning = true;
+                            if (!mySubchannel)
+                                retriesIgnoredPrm++;
+                            else
+                                retriesIgnoredSec++;
+                            ROQ->sendIbyti(header, logctx, mySubchannel);
+                            if (logctx.queryTraceLevel() > 10)
+                            {
+                                StringBuffer xx; logctx.CTXLOG("Ignored retry on subchannel %u for running activity %s", mySubchannel, header.toString(xx).str());
+                            }
+                            break;
+                        }
+                    }
+                    if (!alreadyRunning && checkCompleted && ROQ->replyPending(header))
                     {
                         alreadyRunning = true;
                         if (!mySubchannel)
@@ -2268,35 +2587,33 @@ public:
                         ROQ->sendIbyti(header, logctx, mySubchannel);
                         if (logctx.queryTraceLevel() > 10)
                         {
-                            StringBuffer xx; logctx.CTXLOG("Ignored retry on subchannel %u for running activity %s", mySubchannel, header.toString(xx).str());
+                            StringBuffer xx; logctx.CTXLOG("Ignored retry on subchannel %u for completed activity %s", mySubchannel, header.toString(xx).str());
                         }
-                        break;
                     }
-                } 
-                if (!alreadyRunning && checkCompleted && ROQ->replyPending(header))
-                {
-                    alreadyRunning = true;
-                    if (!mySubchannel)
-                        retriesIgnoredPrm++;
-                    else 
-                        retriesIgnoredSec++;
-                    ROQ->sendIbyti(header, logctx, mySubchannel);
-                    if (logctx.queryTraceLevel() > 10)
+                    if (!alreadyRunning)
                     {
-                        StringBuffer xx; logctx.CTXLOG("Ignored retry on subchannel %u for completed activity %s", mySubchannel, header.toString(xx).str());
+                        if (logctx.queryTraceLevel() > 10)
+                        {
+                            StringBuffer xx; logctx.CTXLOG("Retry %d received on subchannel %u for %s", retries+1, mySubchannel, header.toString(xx).str());
+                        }
+                        queue.enqueueUnique(packet.getClear(), mySubchannel);
                     }
                 }
-                if (!alreadyRunning)
+                else // first time (not a retry).
                 {
-                    if (logctx.queryTraceLevel() > 10)
+#ifdef NEW_IBYTI
+                    if (mySubchannel != 0)  // i.e. I am not the primary here
                     {
-                        StringBuffer xx; logctx.CTXLOG("Retry %d received on subchannel %u for %s", retries+1, mySubchannel, header.toString(xx).str());
+                        unsigned delay = 0;
+                        for (unsigned subChannel = 0; subChannel < mySubchannel; subChannel++)
+                            delay += getIbytiDelay(header.subChannels[subChannel].getIpAddress());
+                        delayed.queryQueue(header.channel, mySubchannel).append(packet.getClear(), msTick()+delay);
                     }
-                    queue.enqueueUnique(packet.getClear(), mySubchannel);
+                    else
+#endif
+                        queue.enqueue(packet.getClear());
                 }
             }
-            else // first time (not a retry). 
-                queue.enqueue(packet.getClear());
         }
     }
 
@@ -2313,8 +2630,15 @@ public:
             {
                 // NOTE - this thread needs to do as little as possible - just read packets and queue them up - otherwise we can get packet loss due to buffer overflow
                 // DO NOT put tracing on this thread except at very high tracelevels!
+#ifdef NEW_IBYTI
+                unsigned timeout = delayed.timeout(msTick());
+                if (timeout>5000)
+                    timeout = 5000;
+#else
+                unsigned timeout = 5000;
+#endif
                 unsigned l;
-                multicastSocket->read(mb.reserve(maxPacketSize), sizeof(RoxiePacketHeader), maxPacketSize, l, 5);
+                multicastSocket->readtms(mb.reserve(maxPacketSize), sizeof(RoxiePacketHeader), maxPacketSize, l, timeout);
                 mb.setLength(l);
                 packetsReceived++;
                 RoxiePacketHeader &header = *(RoxiePacketHeader *) mb.toByteArray();
@@ -2371,6 +2695,14 @@ public:
                     break;
                 }
             }
+#ifdef NEW_IBYTI
+            delayed.checkExpired(
+                            msTick(),
+#ifdef ROXIE_SLA_LOGIC
+                            slaQueue,
+#endif
+                            hiQueue, loQueue);
+#endif
         }
         return 0;
     }
@@ -2763,17 +3095,13 @@ public:
             }
             else
             {
-#ifdef SUBCHANNELS_IN_HEADER
-                // In SUBCHANNELS_IN_HEADER mode this translation has been done on server before sending
-                throwUnexpected();
-#else
                 // Turn broadcast packet (channel 0), as early as possible, into non-0 channel packets.
                 // So retries and other communication with Roxie server (which uses non-0 channel numbers) will not cause double work or confusion.
+                // In SUBCHANNELS_IN_HEADER mode this translation has been done on server before sending, except for some control messages like PING or UNLOAD
                 for (unsigned i = 0; i < numChannels; i++)
                 {
                     targetQueue->enqueue(packet->clonePacket(i+1));
                 }
-#endif
             }
         }
     }

roxie/ccd/ccdserver.cpp

@@ -9728,10 +9728,7 @@ private:
         catch (IException *e)
         {
             // NB: the original exception is probably a IPipeProcessException, but because InterruptableSemaphore rethrows it, we must catch it as an IException
-            if (QUERYINTERFACE(e, IPipeProcessException))
-                pipeException.setown(e);
-            else
-                throw;
+            pipeException.setown(e);
         }
         verifyPipe();
         if (pipeException) // NB: verifyPipe may throw error based on pipe prog. output 1st.
@@ -9841,7 +9838,7 @@ public:
 
     virtual void onExecute()
     {
-        Owned<IPipeProcessException> pipeException;
+        Owned<IException> pipeException;
         try
         {
             for (;;)
@@ -9862,7 +9859,7 @@ public:
             if (!recreate)
                 closePipe();
         }
-        catch (IPipeProcessException *e)
+        catch (IException *e)
         {
             pipeException.setown(e);
         }
@@ -11760,7 +11757,7 @@ protected:
         }
         else
         {
-            if (isContainerized())
+            if (isContainerized() && fileNameServiceDali)
             {
                 StringBuffer nasGroupName;
                 queryNamedGroupStore().getNasGroupName(nasGroupName, 1);
@@ -11867,7 +11864,7 @@ public:
 
     virtual void reset()
     {
-        CRoxieServerActivity::reset();
+        CRoxieServerInternalSinkActivity::reset();
         diskout.clear();
         outSeq.clear();
         writer.clear();
@@ -12250,7 +12247,7 @@ class CRoxieServerIndexWriteActivity : public CRoxieServerInternalSinkActivity,
 
         if (!clusters.length())
         {
-            if (isContainerized())
+            if (isContainerized() && fileNameServiceDali)
             {
                 StringBuffer nasGroupName;
                 queryNamedGroupStore().getNasGroupName(nasGroupName, 1);
@@ -21010,7 +21007,7 @@ public:
     {
     }
 
-    virtual void doExecuteAction(unsigned parentExtractSize, const byte * parentExtract) 
+    virtual void doExecuteAction(unsigned parentExtractSize, const byte * parentExtract) override
     {
         bool cond;
         {
@@ -21021,6 +21018,19 @@ public:
         executeDependencies(parentExtractSize, parentExtract, cond ? 1 : 2);
     }
 
+    virtual void stop() override
+    {
+        if (state != STATEstopped)
+        {
+            ForEachItemIn(idx, dependencies)
+            {
+                if (dependencyControlIds.item(idx) != 0)
+                    dependencies.item(idx).stop();
+            }
+        }
+        CRoxieServerActionBaseActivity::stop();
+    }
+
 };
 
 class CRoxieServerIfActionActivityFactory : public CRoxieServerActivityFactory
@@ -21114,8 +21124,17 @@ public:
     virtual void doExecuteAction(unsigned parentExtractSize, const byte * parentExtract) 
     {
         unsigned numBranches = helper.numBranches();
-        for (unsigned branch=1; branch <= numBranches; branch++)
-            executeDependencies(parentExtractSize, parentExtract, branch);
+        try
+        {
+            for (unsigned branch=1; branch <= numBranches; branch++)
+                executeDependencies(parentExtractSize, parentExtract, branch);
+        }
+        catch (...)
+        {
+            for (unsigned branch=1; branch <= numBranches; branch++)
+                stopDependencies(parentExtractSize, parentExtract, branch);
+            throw;
+        }
     }
 
 };
@@ -21534,7 +21553,7 @@ public:
             }
         }
         size32_t outputLimitBytes = 0;
-        IConstWorkUnit *workunit = serverContext->queryWorkUnit();
+        IConstWorkUnit *workunit = sequence == ResultSequenceInternal ? nullptr : serverContext->queryWorkUnit();
         if (workunit)
         {
             size32_t outputLimit;
@@ -26990,9 +27009,58 @@ public:
     virtual void checkForAbort() { checkAbort(); }
 };
 
+class CRoxieServerSoapActionBase : public CRoxieServerSoapActivityBase
+{
+public:
+    CRoxieServerSoapActionBase(IRoxieAgentContext *_ctx, const IRoxieServerActivityFactory *_factory, IProbeManager *_probeManager)
+        : CRoxieServerSoapActivityBase(_ctx, _factory, _probeManager)
+    {
+    }
+
+    virtual void execute(unsigned parentExtractSize, const byte * parentExtract)
+    {
+        CriticalBlock b(ecrit);
+        if (exception)
+            throw(exception.getLink());
+        if (!executed)
+        {
+            try
+            {
+                executed = true;
+                start(parentExtractSize, parentExtract, false);
+                {
+                    ActivityTimer t(activityStats, timeActivities); // unfortunately this is not really best place for seeing in debugger.
+                    onExecute();
+                }
+                stop();
+            }
+            catch (IException *E)
+            {
+                ctx->notifyAbort(E);
+                exception.set(E);
+                abort();
+                throw E;
+            }
+        }
+    }
+
+    virtual void onExecute() = 0;
+
+    virtual void reset() override
+    {
+        executed = false;
+        exception.clear();
+        CRoxieServerSoapActivityBase::reset();
+    }
+
+protected:
+    bool executed = false;
+    Linked<IException> exception;
+    CriticalSection ecrit;
+};
 //---------------------------------------------------------------------------
 
-class CRoxieServerSoapRowCallActivity : public CRoxieServerSoapActivityBase 
+class CRoxieServerSoapRowCallActivity : public CRoxieServerSoapActivityBase
 {
     IHThorSoapCallArg & callHelper;
 
@@ -27060,16 +27128,15 @@ IRoxieServerActivityFactory *createRoxieServerSoapRowCallActivityFactory(unsigne
 
 //---------------------------------------------------------------------------
 
-class CRoxieServerSoapRowActionActivity : public CRoxieServerSoapActivityBase 
+class CRoxieServerSoapRowActionActivity : public CRoxieServerSoapActionBase
 {
 public:
     CRoxieServerSoapRowActionActivity(IRoxieAgentContext *_ctx, const IRoxieServerActivityFactory *_factory, IProbeManager *_probeManager)
-        : CRoxieServerSoapActivityBase(_ctx, _factory, _probeManager)
+        : CRoxieServerSoapActionBase(_ctx, _factory, _probeManager)
     {}
 
-    virtual void execute(unsigned parentExtractSize, const byte * parentExtract)
+    virtual void onExecute() override
     {
-        //MORE: parentExtract not passed to start - although shouldn't be a problem.
         soaphelper.setown(createSoapCallHelper(this, NULL, ctx->queryAuthToken(), SCrow, pClientCert, *this, this));
         soaphelper->start();
         soaphelper->waitUntilDone();
@@ -27193,11 +27260,11 @@ IRoxieServerActivityFactory *createRoxieServerSoapDatasetCallActivityFactory(uns
 
 //---------------------------------------------------------------------------
 
-class CRoxieServerSoapDatasetActionActivity : public CRoxieServerSoapActivityBase 
+class CRoxieServerSoapDatasetActionActivity : public CRoxieServerSoapActionBase
 {
 public:
     CRoxieServerSoapDatasetActionActivity(IRoxieAgentContext *_ctx, const IRoxieServerActivityFactory *_factory, IProbeManager *_probeManager)
-        : CRoxieServerSoapActivityBase(_ctx, _factory, _probeManager)
+        : CRoxieServerSoapActionBase(_ctx, _factory, _probeManager)
     {}
 
     virtual const void *getNextRow()
@@ -27209,34 +27276,15 @@ public:
         return nextrec;
     }
 
-    virtual void execute(unsigned parentExtractSize, const byte * parentExtract)
+    virtual void onExecute() override
     {
-        try
-        {
-            start(parentExtractSize, parentExtract, false);
-            soaphelper.setown(createSoapCallHelper(this, NULL, ctx->queryAuthToken(), SCdataset, pClientCert, *this, this));
-            soaphelper->start();
-            soaphelper->waitUntilDone();
-            IException *e = soaphelper->getError();
-            soaphelper.clear();
-            if (e)
-                throw e;
-            stop();
-        }
-        catch (IException *E)
-        {
-            ctx->notifyAbort(E);
-            abort();
-            throw;
-        }
-        catch(...)
-        {
-            Owned<IException> E = MakeStringException(ROXIE_INTERNAL_ERROR, "Unknown exception caught at %s:%d", sanitizeSourceFile(__FILE__), __LINE__);
-            ctx->notifyAbort(E);
-            abort();
-            throw;
-
-        }
+        soaphelper.setown(createSoapCallHelper(this, NULL, ctx->queryAuthToken(), SCdataset, pClientCert, *this, this));
+        soaphelper->start();
+        soaphelper->waitUntilDone();
+        IException *e = soaphelper->getError();
+        soaphelper.clear();
+        if (e)
+            throw e;
     }
 
     virtual IFinalRoxieInput *queryOutput(unsigned idx)

system/jlib/jsecrets.cpp

@@ -429,7 +429,16 @@ private:
 public:
     CVaultManager()
     {
-        IPropertyTree *config = queryComponentConfig().queryPropTree("vaults");
+        IPropertyTree *config = nullptr;
+        try
+        {
+            config = queryComponentConfig().queryPropTree("vaults");
+        }
+        catch (IException * e)
+        {
+            EXCLOG(e);
+            e->Release();
+        }
         if (!config)
             return;
         Owned<IPropertyTreeIterator> iter = config->getElements("*");

testing/regress/ecl/choosen0.ecl

@@ -0,0 +1,49 @@
+/*##############################################################################
+
+    HPCC SYSTEMS software Copyright (C) 2020 HPCC Systems®.
+
+    Licensed under the Apache License, Version 2.0 (the "License");
+    you may not use this file except in compliance with the License.
+    You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+    Unless required by applicable law or agreed to in writing, software
+    distributed under the License is distributed on an "AS IS" BASIS,
+    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+    See the License for the specific language governing permissions and
+    limitations under the License.
+############################################################################## */
+
+#onwarning (1036, ignore);
+//version useGeneric=false,multiPart=false
+//version useGeneric=true,multiPart=false
+//version useGeneric=false,multiPart=true
+//version useGeneric=true,multiPart=true
+
+import ^ as root;
+useGeneric := #IFDEFINED(root.useGeneric, true);
+multiPart := #IFDEFINED(root.multiPart, true);
+useLocal := #IFDEFINED(root.useLocal, false);
+useTranslation := #IFDEFINED(root.useTranslation, false);
+
+#option('fgenericDiskReads', useGeneric);
+
+//--- end of version configuration ---
+
+import $.setup;
+Files := setup.Files(multiPart, useLocal, useTranslation);
+
+d := nofold(dataset([{1},{2},{3}], { unsigned a} ));
+
+choosen(d, nofold(0));
+choosen(d, 1);
+choosen(d, 10);
+choosen(d, ALL);
+
+d1 := Files.DG_FlatFile;
+choosen(d1, nofold(0));
+choosen(d1, 1);
+choosen(d1, 10);
+choosen(d1, ALL);
+

testing/regress/ecl/ifaction2.ecl

@@ -0,0 +1,16 @@
+#option ('resourceConditionalActions', true);
+
+d := NOFOLD(DATASET([{1},{2},{3}], { unsigned v; }));
+
+spl1 := NOFOLD(d);
+spl2 := NOFOLD(d(v=1));
+spl3 := NOFOLD(d(v=2));
+
+s2 := NOFOLD(true) : STORED('s2');
+s3 := NOFOLD(false) : stored('s3');
+
+unstopped := IF(s2, OUTPUT(spl1), OUTPUT(spl2));
+
+stopped := IF(s3, unstopped, OUTPUT(spl3));
+
+stopped;

testing/regress/ecl/issue23168.ecl

@@ -15,7 +15,6 @@
     limitations under the License.
 ############################################################################## */
 
-//nothor
 
 import dbglog from Std.System.log;
 

testing/regress/ecl/key/choosen0.xml

@@ -0,0 +1,98 @@
+<Dataset name='Result 1'>
+</Dataset>
+<Dataset name='Result 2'>
+ <Row><a>1</a></Row>
+</Dataset>
+<Dataset name='Result 3'>
+ <Row><a>1</a></Row>
+ <Row><a>2</a></Row>
+ <Row><a>3</a></Row>
+</Dataset>
+<Dataset name='Result 4'>
+ <Row><a>1</a></Row>
+ <Row><a>2</a></Row>
+ <Row><a>3</a></Row>
+</Dataset>
+<Dataset name='Result 5'>
+</Dataset>
+<Dataset name='Result 6'>
+ <Row><dg_parentid>0</dg_parentid><dg_firstname>DAVID     </dg_firstname><dg_lastname>BAYLISS   </dg_lastname><dg_prange>1</dg_prange><filepos>0</filepos></Row>
+</Dataset>
+<Dataset name='Result 7'>
+ <Row><dg_parentid>0</dg_parentid><dg_firstname>DAVID     </dg_firstname><dg_lastname>BAYLISS   </dg_lastname><dg_prange>1</dg_prange><filepos>0</filepos></Row>
+ <Row><dg_parentid>1</dg_parentid><dg_firstname>DAVID     </dg_firstname><dg_lastname>BAYLISS   </dg_lastname><dg_prange>2</dg_prange><filepos>25</filepos></Row>
+ <Row><dg_parentid>2</dg_parentid><dg_firstname>DAVID     </dg_firstname><dg_lastname>BAYLISS   </dg_lastname><dg_prange>3</dg_prange><filepos>50</filepos></Row>
+ <Row><dg_parentid>3</dg_parentid><dg_firstname>DAVID     </dg_firstname><dg_lastname>BAYLISS   </dg_lastname><dg_prange>4</dg_prange><filepos>75</filepos></Row>
+ <Row><dg_parentid>4</dg_parentid><dg_firstname>DAVID     </dg_firstname><dg_lastname>DOLSON    </dg_lastname><dg_prange>1</dg_prange><filepos>100</filepos></Row>
+ <Row><dg_parentid>5</dg_parentid><dg_firstname>DAVID     </dg_firstname><dg_lastname>DOLSON    </dg_lastname><dg_prange>2</dg_prange><filepos>125</filepos></Row>
+ <Row><dg_parentid>6</dg_parentid><dg_firstname>DAVID     </dg_firstname><dg_lastname>DOLSON    </dg_lastname><dg_prange>3</dg_prange><filepos>150</filepos></Row>
+ <Row><dg_parentid>7</dg_parentid><dg_firstname>DAVID     </dg_firstname><dg_lastname>DOLSON    </dg_lastname><dg_prange>4</dg_prange><filepos>175</filepos></Row>
+ <Row><dg_parentid>8</dg_parentid><dg_firstname>DAVID     </dg_firstname><dg_lastname>BILLINGTON</dg_lastname><dg_prange>1</dg_prange><filepos>200</filepos></Row>
+ <Row><dg_parentid>9</dg_parentid><dg_firstname>DAVID     </dg_firstname><dg_lastname>BILLINGTON</dg_lastname><dg_prange>2</dg_prange><filepos>225</filepos></Row>
+</Dataset>
+<Dataset name='Result 8'>
+ <Row><dg_parentid>0</dg_parentid><dg_firstname>DAVID     </dg_firstname><dg_lastname>BAYLISS   </dg_lastname><dg_prange>1</dg_prange><filepos>0</filepos></Row>
+ <Row><dg_parentid>1</dg_parentid><dg_firstname>DAVID     </dg_firstname><dg_lastname>BAYLISS   </dg_lastname><dg_prange>2</dg_prange><filepos>25</filepos></Row>
+ <Row><dg_parentid>2</dg_parentid><dg_firstname>DAVID     </dg_firstname><dg_lastname>BAYLISS   </dg_lastname><dg_prange>3</dg_prange><filepos>50</filepos></Row>
+ <Row><dg_parentid>3</dg_parentid><dg_firstname>DAVID     </dg_firstname><dg_lastname>BAYLISS   </dg_lastname><dg_prange>4</dg_prange><filepos>75</filepos></Row>
+ <Row><dg_parentid>4</dg_parentid><dg_firstname>DAVID     </dg_firstname><dg_lastname>DOLSON    </dg_lastname><dg_prange>1</dg_prange><filepos>100</filepos></Row>
+ <Row><dg_parentid>5</dg_parentid><dg_firstname>DAVID     </dg_firstname><dg_lastname>DOLSON    </dg_lastname><dg_prange>2</dg_prange><filepos>125</filepos></Row>
+ <Row><dg_parentid>6</dg_parentid><dg_firstname>DAVID     </dg_firstname><dg_lastname>DOLSON    </dg_lastname><dg_prange>3</dg_prange><filepos>150</filepos></Row>
+ <Row><dg_parentid>7</dg_parentid><dg_firstname>DAVID     </dg_firstname><dg_lastname>DOLSON    </dg_lastname><dg_prange>4</dg_prange><filepos>175</filepos></Row>
+ <Row><dg_parentid>8</dg_parentid><dg_firstname>DAVID     </dg_firstname><dg_lastname>BILLINGTON</dg_lastname><dg_prange>1</dg_prange><filepos>200</filepos></Row>
+ <Row><dg_parentid>9</dg_parentid><dg_firstname>DAVID     </dg_firstname><dg_lastname>BILLINGTON</dg_lastname><dg_prange>2</dg_prange><filepos>225</filepos></Row>
+ <Row><dg_parentid>10</dg_parentid><dg_firstname>DAVID     </dg_firstname><dg_lastname>BILLINGTON</dg_lastname><dg_prange>3</dg_prange><filepos>250</filepos></Row>
+ <Row><dg_parentid>11</dg_parentid><dg_firstname>DAVID     </dg_firstname><dg_lastname>BILLINGTON</dg_lastname><dg_prange>4</dg_prange><filepos>275</filepos></Row>
+ <Row><dg_parentid>12</dg_parentid><dg_firstname>DAVID     </dg_firstname><dg_lastname>SMITH     </dg_lastname><dg_prange>1</dg_prange><filepos>300</filepos></Row>
+ <Row><dg_parentid>13</dg_parentid><dg_firstname>DAVID     </dg_firstname><dg_lastname>SMITH     </dg_lastname><dg_prange>2</dg_prange><filepos>325</filepos></Row>
+ <Row><dg_parentid>14</dg_parentid><dg_firstname>DAVID     </dg_firstname><dg_lastname>SMITH     </dg_lastname><dg_prange>3</dg_prange><filepos>350</filepos></Row>
+ <Row><dg_parentid>15</dg_parentid><dg_firstname>DAVID     </dg_firstname><dg_lastname>SMITH     </dg_lastname><dg_prange>4</dg_prange><filepos>375</filepos></Row>
+ <Row><dg_parentid>16</dg_parentid><dg_firstname>CLAIRE    </dg_firstname><dg_lastname>BAYLISS   </dg_lastname><dg_prange>1</dg_prange><filepos>400</filepos></Row>
+ <Row><dg_parentid>17</dg_parentid><dg_firstname>CLAIRE    </dg_firstname><dg_lastname>BAYLISS   </dg_lastname><dg_prange>2</dg_prange><filepos>425</filepos></Row>
+ <Row><dg_parentid>18</dg_parentid><dg_firstname>CLAIRE    </dg_firstname><dg_lastname>BAYLISS   </dg_lastname><dg_prange>3</dg_prange><filepos>450</filepos></Row>
+ <Row><dg_parentid>19</dg_parentid><dg_firstname>CLAIRE    </dg_firstname><dg_lastname>BAYLISS   </dg_lastname><dg_prange>4</dg_prange><filepos>475</filepos></Row>
+ <Row><dg_parentid>20</dg_parentid><dg_firstname>CLAIRE    </dg_firstname><dg_lastname>DOLSON    </dg_lastname><dg_prange>1</dg_prange><filepos>500</filepos></Row>
+ <Row><dg_parentid>21</dg_parentid><dg_firstname>CLAIRE    </dg_firstname><dg_lastname>DOLSON    </dg_lastname><dg_prange>2</dg_prange><filepos>525</filepos></Row>
+ <Row><dg_parentid>22</dg_parentid><dg_firstname>CLAIRE    </dg_firstname><dg_lastname>DOLSON    </dg_lastname><dg_prange>3</dg_prange><filepos>550</filepos></Row>
+ <Row><dg_parentid>23</dg_parentid><dg_firstname>CLAIRE    </dg_firstname><dg_lastname>DOLSON    </dg_lastname><dg_prange>4</dg_prange><filepos>575</filepos></Row>
+ <Row><dg_parentid>24</dg_parentid><dg_firstname>CLAIRE    </dg_firstname><dg_lastname>BILLINGTON</dg_lastname><dg_prange>1</dg_prange><filepos>600</filepos></Row>
+ <Row><dg_parentid>25</dg_parentid><dg_firstname>CLAIRE    </dg_firstname><dg_lastname>BILLINGTON</dg_lastname><dg_prange>2</dg_prange><filepos>625</filepos></Row>
+ <Row><dg_parentid>26</dg_parentid><dg_firstname>CLAIRE    </dg_firstname><dg_lastname>BILLINGTON</dg_lastname><dg_prange>3</dg_prange><filepos>650</filepos></Row>
+ <Row><dg_parentid>27</dg_parentid><dg_firstname>CLAIRE    </dg_firstname><dg_lastname>BILLINGTON</dg_lastname><dg_prange>4</dg_prange><filepos>675</filepos></Row>
+ <Row><dg_parentid>28</dg_parentid><dg_firstname>CLAIRE    </dg_firstname><dg_lastname>SMITH     </dg_lastname><dg_prange>1</dg_prange><filepos>700</filepos></Row>
+ <Row><dg_parentid>29</dg_parentid><dg_firstname>CLAIRE    </dg_firstname><dg_lastname>SMITH     </dg_lastname><dg_prange>2</dg_prange><filepos>725</filepos></Row>
+ <Row><dg_parentid>30</dg_parentid><dg_firstname>CLAIRE    </dg_firstname><dg_lastname>SMITH     </dg_lastname><dg_prange>3</dg_prange><filepos>750</filepos></Row>
+ <Row><dg_parentid>31</dg_parentid><dg_firstname>CLAIRE    </dg_firstname><dg_lastname>SMITH     </dg_lastname><dg_prange>4</dg_prange><filepos>775</filepos></Row>
+ <Row><dg_parentid>32</dg_parentid><dg_firstname>KELLY     </dg_firstname><dg_lastname>BAYLISS   </dg_lastname><dg_prange>1</dg_prange><filepos>800</filepos></Row>
+ <Row><dg_parentid>33</dg_parentid><dg_firstname>KELLY     </dg_firstname><dg_lastname>BAYLISS   </dg_lastname><dg_prange>2</dg_prange><filepos>825</filepos></Row>
+ <Row><dg_parentid>34</dg_parentid><dg_firstname>KELLY     </dg_firstname><dg_lastname>BAYLISS   </dg_lastname><dg_prange>3</dg_prange><filepos>850</filepos></Row>
+ <Row><dg_parentid>35</dg_parentid><dg_firstname>KELLY     </dg_firstname><dg_lastname>BAYLISS   </dg_lastname><dg_prange>4</dg_prange><filepos>875</filepos></Row>
+ <Row><dg_parentid>36</dg_parentid><dg_firstname>KELLY     </dg_firstname><dg_lastname>DOLSON    </dg_lastname><dg_prange>1</dg_prange><filepos>900</filepos></Row>
+ <Row><dg_parentid>37</dg_parentid><dg_firstname>KELLY     </dg_firstname><dg_lastname>DOLSON    </dg_lastname><dg_prange>2</dg_prange><filepos>925</filepos></Row>
+ <Row><dg_parentid>38</dg_parentid><dg_firstname>KELLY     </dg_firstname><dg_lastname>DOLSON    </dg_lastname><dg_prange>3</dg_prange><filepos>950</filepos></Row>
+ <Row><dg_parentid>39</dg_parentid><dg_firstname>KELLY     </dg_firstname><dg_lastname>DOLSON    </dg_lastname><dg_prange>4</dg_prange><filepos>975</filepos></Row>
+ <Row><dg_parentid>40</dg_parentid><dg_firstname>KELLY     </dg_firstname><dg_lastname>BILLINGTON</dg_lastname><dg_prange>1</dg_prange><filepos>1000</filepos></Row>
+ <Row><dg_parentid>41</dg_parentid><dg_firstname>KELLY     </dg_firstname><dg_lastname>BILLINGTON</dg_lastname><dg_prange>2</dg_prange><filepos>1025</filepos></Row>
+ <Row><dg_parentid>42</dg_parentid><dg_firstname>KELLY     </dg_firstname><dg_lastname>BILLINGTON</dg_lastname><dg_prange>3</dg_prange><filepos>1050</filepos></Row>
+ <Row><dg_parentid>43</dg_parentid><dg_firstname>KELLY     </dg_firstname><dg_lastname>BILLINGTON</dg_lastname><dg_prange>4</dg_prange><filepos>1075</filepos></Row>
+ <Row><dg_parentid>44</dg_parentid><dg_firstname>KELLY     </dg_firstname><dg_lastname>SMITH     </dg_lastname><dg_prange>1</dg_prange><filepos>1100</filepos></Row>
+ <Row><dg_parentid>45</dg_parentid><dg_firstname>KELLY     </dg_firstname><dg_lastname>SMITH     </dg_lastname><dg_prange>2</dg_prange><filepos>1125</filepos></Row>
+ <Row><dg_parentid>46</dg_parentid><dg_firstname>KELLY     </dg_firstname><dg_lastname>SMITH     </dg_lastname><dg_prange>3</dg_prange><filepos>1150</filepos></Row>
+ <Row><dg_parentid>47</dg_parentid><dg_firstname>KELLY     </dg_firstname><dg_lastname>SMITH     </dg_lastname><dg_prange>4</dg_prange><filepos>1175</filepos></Row>
+ <Row><dg_parentid>48</dg_parentid><dg_firstname>KIMBERLY  </dg_firstname><dg_lastname>BAYLISS   </dg_lastname><dg_prange>1</dg_prange><filepos>1200</filepos></Row>
+ <Row><dg_parentid>49</dg_parentid><dg_firstname>KIMBERLY  </dg_firstname><dg_lastname>BAYLISS   </dg_lastname><dg_prange>2</dg_prange><filepos>1225</filepos></Row>
+ <Row><dg_parentid>50</dg_parentid><dg_firstname>KIMBERLY  </dg_firstname><dg_lastname>BAYLISS   </dg_lastname><dg_prange>3</dg_prange><filepos>1250</filepos></Row>
+ <Row><dg_parentid>51</dg_parentid><dg_firstname>KIMBERLY  </dg_firstname><dg_lastname>BAYLISS   </dg_lastname><dg_prange>4</dg_prange><filepos>1275</filepos></Row>
+ <Row><dg_parentid>52</dg_parentid><dg_firstname>KIMBERLY  </dg_firstname><dg_lastname>DOLSON    </dg_lastname><dg_prange>1</dg_prange><filepos>1300</filepos></Row>
+ <Row><dg_parentid>53</dg_parentid><dg_firstname>KIMBERLY  </dg_firstname><dg_lastname>DOLSON    </dg_lastname><dg_prange>2</dg_prange><filepos>1325</filepos></Row>
+ <Row><dg_parentid>54</dg_parentid><dg_firstname>KIMBERLY  </dg_firstname><dg_lastname>DOLSON    </dg_lastname><dg_prange>3</dg_prange><filepos>1350</filepos></Row>
+ <Row><dg_parentid>55</dg_parentid><dg_firstname>KIMBERLY  </dg_firstname><dg_lastname>DOLSON    </dg_lastname><dg_prange>4</dg_prange><filepos>1375</filepos></Row>
+ <Row><dg_parentid>56</dg_parentid><dg_firstname>KIMBERLY  </dg_firstname><dg_lastname>BILLINGTON</dg_lastname><dg_prange>1</dg_prange><filepos>1400</filepos></Row>
+ <Row><dg_parentid>57</dg_parentid><dg_firstname>KIMBERLY  </dg_firstname><dg_lastname>BILLINGTON</dg_lastname><dg_prange>2</dg_prange><filepos>1425</filepos></Row>
+ <Row><dg_parentid>58</dg_parentid><dg_firstname>KIMBERLY  </dg_firstname><dg_lastname>BILLINGTON</dg_lastname><dg_prange>3</dg_prange><filepos>1450</filepos></Row>
+ <Row><dg_parentid>59</dg_parentid><dg_firstname>KIMBERLY  </dg_firstname><dg_lastname>BILLINGTON</dg_lastname><dg_prange>4</dg_prange><filepos>1475</filepos></Row>
+ <Row><dg_parentid>60</dg_parentid><dg_firstname>KIMBERLY  </dg_firstname><dg_lastname>SMITH     </dg_lastname><dg_prange>1</dg_prange><filepos>1500</filepos></Row>
+ <Row><dg_parentid>61</dg_parentid><dg_firstname>KIMBERLY  </dg_firstname><dg_lastname>SMITH     </dg_lastname><dg_prange>2</dg_prange><filepos>1525</filepos></Row>
+ <Row><dg_parentid>62</dg_parentid><dg_firstname>KIMBERLY  </dg_firstname><dg_lastname>SMITH     </dg_lastname><dg_prange>3</dg_prange><filepos>1550</filepos></Row>
+ <Row><dg_parentid>63</dg_parentid><dg_firstname>KIMBERLY  </dg_firstname><dg_lastname>SMITH     </dg_lastname><dg_prange>4</dg_prange><filepos>1575</filepos></Row>
+</Dataset>

testing/regress/ecl/key/ifaction2.xml

@@ -0,0 +1,3 @@
+<Dataset name='Result 3'>
+ <Row><v>2</v></Row>
+</Dataset>

testing/regress/ecl/keydiff1.ecl

@@ -18,6 +18,10 @@
 import $.setup;
 prefix := setup.Files(false, false).QueryFilePrefix;
 
+// This warning is suppressed because it depends on the size of the thor cluster
+// and consistant output is required by regression suite
+// (warning relates to skew in the child record causing uneven disk read time)
+#onwarning (30003,ignore);
 //noRoxie
 //noHthor
 import Std.File AS FileServices;

thorlcr/graph/thgraph.cpp

@@ -754,6 +754,12 @@ bool CGraphElementBase::prepareContext(size32_t parentExtractSz, const byte *par
                 default:
                     break;
             }
+            if (isActivitySink(getKind()))
+            {
+                // Suppress executing internal sinks with 0 generated dependencies
+                if (queryXGMML().getPropBool("att[@name='_internal']/@value", false) && (0 == owner->queryDependents(id)))
+                    return false;
+            }
             if (checkDependencies && ((unsigned)-1 != whichBranch))
             {
                 if (inputs.queryItem(whichBranch))
@@ -1067,18 +1073,6 @@ static void getGlobalDeps(CGraphBase &graph, CICopyArrayOf<CGraphDependency> &de
     }
 }
 
-static void noteDependency(CGraphElementBase *targetActivity, CGraphElementBase *sourceActivity, CGraphBase *targetGraph, CGraphBase *sourceGraph, unsigned controlId)
-{
-    targetActivity->addDependsOn(sourceGraph, controlId);
-    // NB: record dependency in source graph, serialized to slaves, used to decided if should run dependency sinks or not
-    Owned<IPropertyTree> dependencyFor = createPTree();
-    dependencyFor->setPropInt("@id", sourceActivity->queryId());
-    dependencyFor->setPropInt("@graphId", targetGraph->queryGraphId());
-    if (controlId)
-        dependencyFor->setPropInt("@conditionalId", controlId);
-    sourceGraph->queryXGMML().addPropTree("Dependency", dependencyFor.getClear());
-}
-
 static void addDependencies(IPropertyTree *xgmml, bool failIfMissing, CGraphTableCopy &graphs)
 {
     CGraphArrayCopy dependentchildGraphs;
@@ -1128,7 +1122,7 @@ static void addDependencies(IPropertyTree *xgmml, bool failIfMissing, CGraphTabl
                 targetActivity = targetGraph->queryElement(targetGraphContext);
             }
             assertex(targetActivity && sourceActivity);
-            noteDependency(targetActivity, sourceActivity, target, source, controlId);
+            source->noteDependency(targetActivity, sourceActivity, controlId, true);
         }
         else if (edge.getPropBool("att[@name=\"_conditionSource\"]/@value", false))
         { /* Ignore it */ }
@@ -1143,7 +1137,7 @@ static void addDependencies(IPropertyTree *xgmml, bool failIfMissing, CGraphTabl
         {
             if (!edge.getPropBool("att[@name=\"_childGraph\"]/@value", false)) // JCSMORE - not sure if necess. roxie seem to do.
                 controlId = edge.getPropInt("att[@name=\"_when\"]/@value", 0);
-            noteDependency(targetActivity, sourceActivity, target, source, controlId);
+            source->noteDependency(targetActivity, sourceActivity, controlId, true);
         }
     }
     ForEachItemIn(c, dependentchildGraphs)
@@ -1158,7 +1152,7 @@ static void addDependencies(IPropertyTree *xgmml, bool failIfMissing, CGraphTabl
             ForEachItemIn(gcd, globalChildGraphDeps)
             {
                 CGraphDependency &globalDep = globalChildGraphDeps.item(gcd);
-                noteDependency(&targetActivity, &sourceActivity, globalDep.graph, &childGraph, globalDep.controlId);
+                childGraph.noteDependency(&targetActivity, &sourceActivity, globalDep.controlId, true);
             }
         }
     }
@@ -1207,6 +1201,7 @@ CGraphBase::CGraphBase(CJobChannel &_jobChannel) : jobChannel(_jobChannel), job(
     parentExtractSz = 0;
     counter = 0; // loop/graph counter, will be set by loop/graph activity if needed
     loopBodySubgraph = false;
+    sourceActDependents.setown(createPTree());
 }
 
 CGraphBase::~CGraphBase()
@@ -1323,6 +1318,21 @@ IThorGraphStubIterator *CGraphBase::getChildStubIterator() const
     return new CIter(childGraphsTable);
 }
 
+void CGraphBase::noteDependency(CGraphElementBase *targetActivity, CGraphElementBase *sourceActivity, unsigned controlId, bool interGraph)
+{
+    if (interGraph)
+        targetActivity->addDependsOn(this, controlId);
+    // NB: record dependency in source graph, serialized to slaves, used to decided if should run dependency sinks or not
+    VStringBuffer srcActStr("act%u", sourceActivity->queryId());
+    sourceActDependents->setPropInt(srcActStr, sourceActDependents->getPropInt(srcActStr)+1);
+}
+
+unsigned CGraphBase::queryDependents(unsigned sourceActId)
+{
+    VStringBuffer srcActStr("act%u", sourceActId);
+    return sourceActDependents->getPropInt(srcActStr);
+}
+
 IThorGraphIterator *CGraphBase::getChildGraphIterator() const
 {
     CriticalBlock b(crit);
@@ -2019,6 +2029,10 @@ void CGraphBase::createFromXGMML(IPropertyTree *_node, CGraphBase *_owner, CGrap
         CGraphElementBase *source = queryElement(edge.getPropInt("@source"));
         CGraphElementBase *target = queryElement(edge.getPropInt("@target"));
         target->addInput(targetInput, source, sourceOutput);
+
+        int controlId = edge.getPropInt("att[@name=\"_when\"]/@value", 0);
+        if (controlId != 0)
+            noteDependency(target, source, controlId, false);
     }
     Owned<IThorActivityIterator> iter = getIterator();
     ForEach(*iter)

thorlcr/graph/thgraph.hpp

@@ -615,6 +615,7 @@ protected:
     unsigned counter;
     CReplyCancelHandler graphCancelHandler;
     bool loopBodySubgraph;
+    Owned<IPropertyTree> sourceActDependents;
 
 public:
     IMPLEMENT_IINTERFACE_USING(CGraphStub);
@@ -749,7 +750,8 @@ public:
     }
     IThorGraphIterator *getChildGraphIterator() const; // retrieves original child graphs
     IThorGraphStubIterator *getChildStubIterator() const; // retrieves child graph stubs, which redirect to parallel instances
-
+    void noteDependency(CGraphElementBase *targetActivity, CGraphElementBase *sourceActivity, unsigned controlId, bool interGraph);
+    unsigned queryDependents(unsigned sourceActId);
     void executeChildGraphs(size32_t parentExtractSz, const byte *parentExtract);
     void doExecute(size32_t parentExtractSz, const byte *parentExtract, bool checkDependencies);
     void doExecuteChild(size32_t parentExtractSz, const byte *parentExtract);

thorlcr/graph/thgraphmaster.cpp

@@ -2424,6 +2424,7 @@ void CMasterGraph::serializeGraphInit(MemoryBuffer &mb)
     mb.append((int)startBarrierTag);
     mb.append((int)waitBarrierTag);
     mb.append((int)doneBarrierTag);
+    sourceActDependents->serialize(mb);
     mb.append(queryChildGraphCount());
     Owned<IThorGraphIterator> childIter = getChildGraphIterator();
     ForEach (*childIter)

thorlcr/graph/thgraphslave.cpp

@@ -872,6 +872,7 @@ void CSlaveGraph::init(MemoryBuffer &mb)
     waitBarrier = queryJobChannel().createBarrier(waitBarrierTag);
     if (doneBarrierTag != TAG_NULL)
         doneBarrier = queryJobChannel().createBarrier(doneBarrierTag);
+    sourceActDependents.setown(createPTree(mb));
     unsigned subCount;
     mb.read(subCount);
     while (subCount--)