Merge branch 'candidate-6.0.4'

Signed-off-by: Gavin Halliday <gavin.halliday@lexisnexis.com>

Conflicts:
	thorlcr/activities/selectnth/thselectnthslave.cpp
	version.cmake

common/thorhelper/thorcommon.cpp

@@ -1788,7 +1788,16 @@ void setAutoAffinity(unsigned curProcess, unsigned processPerMachine, const char
     if (optNodes)
         throw makeStringException(1, "Numa node list not yet supported");
 
-    unsigned numNumaNodes = numa_max_node()+1;
+    unsigned numaMap[NUMA_NUM_NODES];
+    unsigned numNumaNodes = 0;
+    for (unsigned i=0; i<=numa_max_node(); i++)
+    {
+        if (numa_bitmask_isbitset(numa_all_nodes_ptr, i))
+        {
+            numaMap[numNumaNodes] = i;
+            numNumaNodes++;
+        }
+    }
     if (numNumaNodes <= 1)
         return;
 
@@ -1797,20 +1806,20 @@ void setAutoAffinity(unsigned curProcess, unsigned processPerMachine, const char
 
 #if defined(LIBNUMA_API_VERSION) && (LIBNUMA_API_VERSION>=2)
     struct bitmask * cpus = numa_allocate_cpumask();
-    numa_node_to_cpus(curNode, cpus);
+    numa_node_to_cpus(numaMap[curNode], cpus);
     bool ok = (numa_sched_setaffinity(0, cpus) == 0);
     numa_bitmask_free(cpus);
 #else
     cpu_set_t cpus;
     CPU_ZERO(&cpus);
-    numa_node_to_cpus(curNode, (unsigned long *) &cpus, sizeof (cpus));
+    numa_node_to_cpus(numaMap[curNode], (unsigned long *) &cpus, sizeof (cpus));
     bool ok = sched_setaffinity (0, sizeof(cpus), &cpus) != 0;
 #endif
 
     if (!ok)
-        throw makeStringExceptionV(1, "Failed to set affinity for node %u", curNode);
+        throw makeStringExceptionV(1, "Failed to set affinity to numa node %u (id:%u)", curNode, numaMap[curNode]);
 
-    DBGLOG("Process bound to numa node %u of %u", curNode, numNumaNodes);
+    DBGLOG("Process bound to numa node %u (id:%u) of %u", curNode, numaMap[curNode], numNumaNodes);
 #endif
     clearAffinityCache();
 }
@@ -1820,7 +1829,12 @@ void bindMemoryToLocalNodes()
 #if defined(LIBNUMA_API_VERSION) && (LIBNUMA_API_VERSION>=2)
     numa_set_bind_policy(1);
 
-    unsigned numNumaNodes = numa_max_node() + 1;
+    unsigned numNumaNodes = 0;
+    for (unsigned i=0; i<=numa_max_node(); i++)
+    {
+        if (numa_bitmask_isbitset(numa_all_nodes_ptr, i))
+            numNumaNodes++;
+    }
     if (numNumaNodes <= 1)
         return;
     struct bitmask *nodes = numa_get_run_node_mask();

esp/services/ws_machine/metrics.xslt

@@ -66,13 +66,13 @@
             <script type="text/javascript">
               var autoRefreshVal=<xsl:value-of select="$autoRefresh"/>;
               var autoUpdateChecked=<xsl:value-of select="$autoupdatechecked"/>;
+              var viewColumnsStr = '<xsl:value-of select="$hpccStrings/st[@id='ViewColumns']"/>';
             <xsl:text disable-output-escaping="yes"><![CDATA[
          //############ begin auto reload script ######################
          var reloadTimer = null;
          var reloadTimeout = 0;
          var idCount = 0;
          var checkboxIDs = new Array();
-         var viewColumnsStr = '<xsl:value-of select="$hpccStrings/st[@id='ViewColumns']"/>';
 
          // This function gets called when the window has completely loaded.
          // It starts the reload timer with a default time value.

plugins/mysql/mysqlembed.cpp

@@ -1521,6 +1521,9 @@ public:
         if (nextParam != stmtInfo->queryInputBindings().numColumns())
             failx("Not enough parameters supplied (%d parameters supplied, but statement has %d bound columns)", nextParam, stmtInfo->queryInputBindings().numColumns());
         // We actually do the execute later, when the result is fetched
+        // Unless, there is no expected result, in that case execute query now
+        if (stmtInfo->queryResultBindings().numColumns() == 0)
+            lazyExecute();
     }
 protected:
     void lazyExecute()

system/security/LdapSecurity/ldapconnection.cpp

@@ -1348,11 +1348,11 @@ public:
     virtual void init(IPermissionProcessor* pp)
     {
         m_pp = pp;
-        if(m_ldapconfig->getServerType() == OPEN_LDAP)
+        static bool createdOU = false;
+        CriticalBlock block(lcCrit);
+        if (!createdOU)
         {
-            static bool createdOU = false;
-            CriticalBlock block(lcCrit);
-            if (!createdOU)
+            if(m_ldapconfig->getServerType() == OPEN_LDAP)
             {
                 try
                 {
@@ -1363,7 +1363,7 @@ public:
                 }
                 try
                 {
-                addGroup("Directory Administrators", NULL, NULL, m_ldapconfig->getBasedn());
+                    addGroup("Directory Administrators", NULL, NULL, m_ldapconfig->getBasedn());
                 }
                 catch(...)
                 {

system/security/LdapSecurity/ldapsecurity.cpp

@@ -528,8 +528,8 @@ void CLdapSecManager::init(const char *serviceName, IPropertyTree* cfg)
     else
         throwUnexpected();
 
-    ldap_client->init(pp);
     pp->setLdapClient(ldap_client);
+    ldap_client->init(pp);
 
     m_ldap_client.setown(ldap_client);
     m_pp.setown(pp);

thorlcr/activities/firstn/thfirstnslave.cpp

@@ -34,11 +34,6 @@ protected:
     bool stopped;
     IHThorFirstNArg *helper;
 
-    virtual void doStop()
-    {
-        PARENT::stop();
-    }
-
 public:
     CFirstNSlaveBase(CGraphElementBase *_container) : CSlaveActivity(_container)
     {
@@ -58,7 +53,7 @@ public:
         {
             abortSoon = true;
             stopped = true;
-            doStop();
+            PARENT::stop();
         }
     }
     virtual void getMetaInfo(ThorDataLinkMetaInfo &info)
@@ -215,14 +210,6 @@ class CFirstNSlaveGlobal : public CFirstNSlaveBase, implements ILookAheadStopNot
     ThorDataLinkMetaInfo inputMeta;
     Owned<IEngineRowStream> originalInputStream;
 
-protected:
-    virtual void doStop()
-    {
-        limitgot.signal(); // JIC not previously signalled by lookahead
-        onInputFinished(getDataLinkCount()+skipped);
-        PARENT::doStop();
-    }
-
 public:
     CFirstNSlaveGlobal(CGraphElementBase *container) : CFirstNSlaveBase(container)
     {
@@ -268,7 +255,7 @@ public:
     {
         CMessageBuffer msgMb;
         if (!receiveMsg(msgMb, 0, mpTag))
-            return false; // NB: can be triggered by onInputFinished with count==0, or abort()
+            return false; // NB: can be triggered by abort()
         msgMb.read(limit);
         msgMb.read(skipCount);
         skipped = 0;
@@ -295,7 +282,7 @@ public:
     void sendCount()
     {
         limitgot.wait();
-        
+
         rowcount_t read = 0;
         rowcount_t skip = skipCount;
         if (limit > 0)
@@ -329,7 +316,8 @@ public:
     CATCH_NEXTROW()
     {
         ActivityTimer t(totalCycles, timeActivities);
-        if (!abortSoon) {
+        if (!abortSoon)
+        {
             if (firstget)
             {
                 firstget = false;
@@ -377,12 +365,6 @@ public:
             sendCount();
         }
         ActPrintLog("FIRSTN: maximum row count %" RCPF "d", count);
-        if (0 == count)
-        {
-            limit = 0;
-            CriticalBlock b(crit);
-            cancelReceiveMsg(RANK_ALL, mpTag);
-        }
     }
 };
 

thorlcr/activities/lookupjoin/thlookupjoinslave.cpp

@@ -3219,16 +3219,19 @@ public:
     }
     virtual void stop()
     {
-        if (isGlobal())
+        if (hasStarted())
         {
-            if (gotRHS)
+            if (isGlobal())
             {
-                // Other channels sharing HT. So do not reset until all here
-                if (queryJob().queryJobChannels()>1)
-                    InterChannelBarrier();
+                if (gotRHS)
+                {
+                    // Other channels sharing HT. So do not reset until all here
+                    if (queryJob().queryJobChannels()>1)
+                        InterChannelBarrier();
+                }
+                else
+                    getRHS(true); // If global, need to handle RHS until all are slaves stop
             }
-            else
-                getRHS(true); // If global, need to handle RHS until all are slaves stop
         }
         PARENT::stop();
     }

thorlcr/activities/selectnth/thselectnthslave.cpp

@@ -28,6 +28,7 @@ class CSelectNthSlaveActivity : public CSlaveActivity, implements ILookAheadStop
     bool createDefaultIfFail;
     IHThorSelectNArg *helper;
     SpinLock spin; // MORE: Remove this and use an atomic variable for lookaheadN
+    Owned<IEngineRowStream> originalInputStream;
 
     void initN()
     {
@@ -72,19 +73,12 @@ public:
     }
 
 // IThorSlaveActivity overloaded methods
-    virtual void init(MemoryBuffer & data, MemoryBuffer &slaveData)
+    virtual void init(MemoryBuffer & data, MemoryBuffer &slaveData) override
     {
         if (!container.queryLocalOrGrouped())
             mpTag = container.queryJobChannel().deserializeMPTag(data);
     }
-    virtual void setInputStream(unsigned index, CThorInput &_input, bool consumerOrdered) override
-    {
-        PARENT::setInputStream(index, _input, consumerOrdered);
-        rowcount_t rowN = (rowcount_t)helper->getRowToSelect();
-        if (!isLocal && rowN)
-            setLookAhead(0, createRowStreamLookAhead(this, inputStream, queryRowInterfaces(input), SELECTN_SMART_BUFFER_SIZE, isSmartBufferSpillNeeded(this), false, rowN, this, &container.queryJob().queryIDiskUsage()));
-    }
-    virtual void start()
+    virtual void start() override
     {
         ActivityTimer s(totalCycles, timeActivities);
 
@@ -102,8 +96,17 @@ public:
             throw;
         }
 
+        rowcount_t rowN = (rowcount_t)helper->getRowToSelect();
+        IStartableEngineRowStream *lookAhead = nullptr;
+        if (!isLocal && rowN)
+        {
+            lookAhead = createRowStreamLookAhead(this, inputStream, queryRowInterfaces(input), SELECTN_SMART_BUFFER_SIZE, isSmartBufferSpillNeeded(this), false, rowN, this, &container.queryJob().queryIDiskUsage());
+            originalInputStream.setown(replaceInputStream(0, lookAhead));
+            lookAhead->start();
+        }
+
         seenNth = false;
-        if (0==helper->getRowToSelect())
+        if (0 == rowN)
         {
             ThorDataLinkMetaInfo info;
             queryInput(0)->getMetaInfo(info);
@@ -119,7 +122,15 @@ public:
         }
         first = true;
     }
-    virtual void abort()
+    virtual void stop() override
+    {
+        PARENT::stop();
+        if (originalInputStream)
+        {
+            Owned<IEngineRowStream> lookAhead = replaceInputStream(0, originalInputStream.getClear());
+        }
+    }
+    virtual void abort() override
     {
         CSlaveActivity::abort();
         if (!firstNode())
@@ -188,7 +199,7 @@ public:
         return ret.getClear();
     }
     virtual bool isGrouped() const override { return false; }
-    void getMetaInfo(ThorDataLinkMetaInfo &info)
+    virtual void getMetaInfo(ThorDataLinkMetaInfo &info) override
     {
         initMetaInfo(info);
         info.isSequential = true; 
@@ -196,7 +207,7 @@ public:
         calcMetaInfoSize(info, queryInput(0));
     }
 // IStartableEngineRowStream methods used for global selectn only
-    virtual void onInputFinished(rowcount_t count)
+    virtual void onInputFinished(rowcount_t count) override
     {
         SpinBlock b(spin);
         lookaheadN = count;

thorlcr/graph/thgraphmaster.cpp

@@ -233,6 +233,7 @@ void CSlaveMessageHandler::main()
                         assertex(element);
                         try
                         {
+                            element->reset();
                             element->doCreateActivity(parentExtractSz, parentExtract);
                         }
                         catch (IException *e)