Merge pull request #14049 from jakesmith/hpcc-24536-excess-thor-logging

HPCC-24536 Suppress some verbose Thor logging by default

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

thorlcr/activities/choosesets/thchoosesetsslave.cpp

@@ -71,7 +71,6 @@ public:
     {
         ActivityTimer s(slaveTimerStats, timeActivities);
         PARENT::start();
-        ActPrintLog("CHOOSESETS: Is Local");
     }
     CATCH_NEXTROW()
     {
@@ -117,23 +116,25 @@ class ChooseSetsActivity : public BaseChooseSetsActivity
         if (!receiveMsg(msg, queryJobChannel().queryMyRank()-1, mpTag))
             return;
         memcpy(tallies, msg.readDirect(numSets*sizeof(unsigned)), numSets*sizeof(unsigned));
-#if THOR_TRACE_LEVEL >= 5
-        StringBuffer s;
-        unsigned idx=0;
-        for (; idx<numSets; idx++)
-            s.append("[").append(tallies[idx]).append("]");
-        ActPrintLog("CHOOSESETS: Incoming count = %s", s.str());
-#endif
+        if (!REJECTLOG(MCthorDetailedDebugInfo))
+        {
+            StringBuffer s;
+            unsigned idx=0;
+            for (; idx<numSets; idx++)
+                s.append("[").append(tallies[idx]).append("]");
+            ::ActPrintLog(this, thorDetailedLogLevel, "CHOOSESETS: Incoming count = %s", s.str());
+        }
     }
     void sendTallies() // NB: not called on last node.
     {
-#if THOR_TRACE_LEVEL >= 5
-        StringBuffer s;
-        unsigned idx=0;
-        for (; idx<numSets; idx++)
-            s.append("[").append(tallies[idx]).append("]");
-        ActPrintLog("CHOOSESETS: Outgoing count = %s", s.str());
-#endif
+        if (!REJECTLOG(MCthorDetailedDebugInfo))
+        {
+            StringBuffer s;
+            unsigned idx=0;
+            for (; idx<numSets; idx++)
+                s.append("[").append(tallies[idx]).append("]");
+            ::ActPrintLog(this, thorDetailedLogLevel, "CHOOSESETS: Outgoing count = %s", s.str());
+        }
         CMessageBuffer msg;
         msg.append(numSets * sizeof(unsigned), tallies); 
         queryJobChannel().queryJobComm().send(msg, queryJobChannel().queryMyRank()+1, mpTag);
@@ -152,7 +153,6 @@ public:
     virtual void start() override
     {
         ActivityTimer s(slaveTimerStats, timeActivities);
-        ActPrintLog("CHOOSESETS: Is Global");
         PARENT::start();
 
         if (ensureStartFTLookAhead(0))
@@ -163,9 +163,6 @@ public:
     }
     virtual void abort() override
     {
-#if THOR_TRACE_LEVEL >= 5
-        ActPrintLog("CHOOSESETS: abort()");
-#endif
         CSlaveActivity::abort();
         if (!container.queryLocalOrGrouped() && !firstNode())
             cancelReceiveMsg(RANK_ALL, mpTag);
@@ -284,7 +281,6 @@ public:
     virtual void start() override
     {
         ActivityTimer s(slaveTimerStats, timeActivities);
-        ActPrintLog("CHOOSESETS: Is Global");
         if (counts)
         {
             free(counts);
@@ -371,7 +367,7 @@ class ChooseSetsLastActivity : public ChooseSetsPlusActivity
         for (unsigned idx=0; idx < numSets; idx++)
         {
             rowcount_t firstToCopy = 0;
-            ActPrintLog("CHOOSESETSLAST: %d P(%" RCPF "d) C(%" RCPF "d) L(%" I64F "d) T(%" RCPF "d)", idx, priorCounts[idx], counts[idx], limits[idx], totalCounts[idx]);
+            ::ActPrintLog(this, thorDetailedLogLevel, "CHOOSESETSLAST: %d P(%" RCPF "d) C(%" RCPF "d) L(%" I64F "d) T(%" RCPF "d)", idx, priorCounts[idx], counts[idx], limits[idx], totalCounts[idx]);
             if (((rowcount_t)limits[idx]) < totalCounts[idx])
                 firstToCopy = totalCounts[idx] - (rowcount_t)limits[idx];
             if (priorCounts[idx] + counts[idx] > firstToCopy)
@@ -382,9 +378,7 @@ class ChooseSetsLastActivity : public ChooseSetsPlusActivity
                     numToSkip[idx] = (unsigned)(firstToCopy - priorCounts[idx]);
                 numToReturn[idx] = (unsigned)(priorCounts[idx] + counts[idx] - firstToCopy);
                 skipAll = false;
-#if THOR_TRACE_LEVEL >= 5
-                ActPrintLog("CHOOSESETSLAST: Selection %d.  Range(%d,%" RCPF "d)", idx, numToSkip[idx], counts[idx]);
-#endif
+                ::ActPrintLog(this, thorDetailedLogLevel, "CHOOSESETSLAST: Selection %d.  Range(%d,%" RCPF "d)", idx, numToSkip[idx], counts[idx]);
             }
         }
         if (skipAll)

thorlcr/activities/countproject/thcountprojectslave.cpp

@@ -54,7 +54,6 @@ public:
     virtual void start()
     {
         ActivityTimer s(slaveTimerStats, timeActivities);
-        ActPrintLog("COUNTPROJECT: Is Local");
         anyThisGroup = false;
         PARENT::start();
     }
@@ -162,7 +161,6 @@ public:
         localRecCount = RCUNSET;
         onInputFinishSends = true;
         PARENT::start();
-        ActPrintLog( "COUNTPROJECT: Is Global");
         first = true;
         prevRecCount = 0;
         ThorDataLinkMetaInfo info;

thorlcr/activities/distribution/thdistribution.cpp

@@ -48,9 +48,7 @@ public:
             CMessageBuffer msg;
             if (!receiveMsg(msg, RANK_ALL, mpTag, &sender))
                 return;
-#if THOR_TRACE_LEVEL >= 5
-            ActPrintLog("Received distribution result from node %d", (unsigned)sender);
-#endif
+            ::ActPrintLog(this, thorDetailedLogLevel, "Received distribution result from node %d", (unsigned)sender);
             if (msg.length())
                 helper->merge(result, msg);
         }
@@ -59,11 +57,7 @@ public:
         tmp.append("<XML>");
         helper->gatherResult(result, tmp);
         tmp.append("</XML>");
-
-#if THOR_TRACE_LEVEL >= 5
-        ActPrintLog("Distribution result: %s", tmp.str());
-#endif
-
+        ::ActPrintLog(this, thorDetailedLogLevel, "Distribution result: %s", tmp.str());
         helper->sendResult(tmp.length(), tmp.str());
 
         destroyThorRow(result);

thorlcr/activities/distribution/thdistributionslave.cpp

@@ -57,7 +57,7 @@ public:
                 helper->process(aggy, row);     
                 processed++;
             }
-            ActPrintLog("DISTRIBUTION: processed %" RCPF "d records", processed & THORDATALINK_COUNT_MASK);
+            ::ActPrintLog(this, thorDetailedLogLevel, "DISTRIBUTION: processed %" RCPF "d records", processed & THORDATALINK_COUNT_MASK);
         }
         catch(CATCHALL)
         {

thorlcr/activities/enth/thenthslave.cpp

@@ -61,10 +61,8 @@ protected:
         }
         else
             abortSoon = true;
-#if THOR_TRACE_LEVEL >= 5
-        ActPrintLog("ENTH: init - Numerator = %" RCPF "d, Denominator = %" RCPF "d", numerator, denominator);   
-        ActPrintLog("%s: Initial value of counter %" RCPF "d", actStr.str(), counter);
-#endif
+        ::ActPrintLog(this, thorDetailedLogLevel, "ENTH: init - Numerator = %" RCPF "d, Denominator = %" RCPF "d", numerator, denominator);   
+        ::ActPrintLog(this, thorDetailedLogLevel, "%s: Initial value of counter %" RCPF "d", actStr.str(), counter);
     }
     void setLocalCountReq()
     {

thorlcr/activities/fetch/thfetchslave.cpp

@@ -139,11 +139,13 @@ public:
         assertex(offsetMapSz == sizeof(FPosTableEntry) * offsetCount);
         offsetTable = new FPosTableEntry[offsetCount];
         memcpy_iflen(offsetTable, offsetMap, offsetMapSz);
-        unsigned c;
-        for (c=0; c<offsetCount; c++)
+        if (!REJECTLOG(MCthorDetailedDebugInfo))
         {
-            FPosTableEntry &e = offsetTable[c];
-            ActPrintLog(&owner, "Table[%d] : base=%" I64F "d, top=%" I64F "d, slave=%d", c, e.base, e.top, e.index);
+            for (unsigned c=0; c<offsetCount; c++)
+            {
+                FPosTableEntry &e = offsetTable[c];
+                ActPrintLog(&owner, thorDetailedLogLevel, "Table[%d] : base=%" I64F "d, top=%" I64F "d, slave=%d", c, e.base, e.top, e.index);
+            }
         }
         files = parts.ordinality();
         if (files)

thorlcr/activities/group/thgroupslave.cpp

@@ -80,12 +80,7 @@ public:
         PARENT::start();
         eogNext = prevEog = eof = false;
         if (rolloverEnabled)
-        {
             useRollover = !lastNode();
-#ifdef _TESTING
-            ActPrintLog("Node number = %d, Total Nodes = %d", queryJobChannel().queryMyRank(), container.queryJob().querySlaves());
-#endif
-        }
 
         stream.set(inputStream);
         startLastGroup = getDataLinkGlobalCount();

thorlcr/activities/hashdistrib/thhashdistrib.cpp

@@ -155,7 +155,7 @@ public:
     }
     void process()
     {
-        ActPrintLog("ReDistributeActivityMaster::process");
+        ::ActPrintLog(this, thorDetailedLogLevel, "ReDistributeActivityMaster::process");
         HashDistributeMasterBase::process();        
         IHThorHashDistributeArg *helper = (IHThorHashDistributeArg *)queryHelper(); 
         unsigned n = container.queryJob().querySlaves();
@@ -167,26 +167,20 @@ public:
                 if (abortSoon)
                     return;
                 CMessageBuffer mb;
-#ifdef _TRACE
-                ActPrintLog("ReDistribute process, Receiving on tag %d",statstag);
-#endif
+                ::ActPrintLog(this, thorDetailedLogLevel, "ReDistribute process, Receiving on tag %d",statstag);
                 rank_t sender;
                 if (!receiveMsg(mb, RANK_ALL, statstag, &sender)||abortSoon) 
                     return;
-#ifdef _TRACE
-                ActPrintLog("ReDistribute process, Received size from %d",sender);
-#endif
+                ::ActPrintLog(this, thorDetailedLogLevel, "ReDistribute process, Received size from %d",sender);
                 sender--;
                 assertex((unsigned)sender<n);
                 mb.read(sizes[sender]);
             }
-            ActPrintLog("ReDistributeActivityMaster::process sizes got");
+            ::ActPrintLog(this, thorDetailedLogLevel, "ReDistributeActivityMaster::process sizes got");
             for (i=0;i<n;i++) {
                 CMessageBuffer mb;
                 mb.append(n*sizeof(offset_t),sizes);
-#ifdef _TRACE
-                ActPrintLog("ReDistribute process, Replying to node %d tag %d",i+1,statstag);
-#endif
+                ::ActPrintLog(this, thorDetailedLogLevel, "ReDistribute process, Replying to node %d tag %d",i+1,statstag);
                 if (!queryJobChannel().queryJobComm().send(mb, (rank_t)i+1, statstag))
                     return;
             }
@@ -210,7 +204,7 @@ public:
             ActPrintLog(e,"ReDistribute");
             throw;
         }
-        ActPrintLog("ReDistributeActivityMaster::process exit");
+        ::ActPrintLog(this, thorDetailedLogLevel, "ReDistributeActivityMaster::process exit");
     }
     void abort()
     {

thorlcr/activities/hashdistrib/thhashdistribslave.cpp

@@ -599,7 +599,7 @@ protected:
             if (owner.sendBlock(target, msg))
                 return;
             markStopped(target); // Probably a bit pointless if target is 'self' - process loop will have done already
-            owner.ActPrintLog("CSender::sendBlock stopped slave %d (finished=%d)", target+1, atomic_read(&numFinished));
+            ::ActPrintLog(owner.activity, thorDetailedLogLevel, "CSender::sendBlock stopped slave %d (finished=%d)", target+1, atomic_read(&numFinished));
         }
         void closeWrite()
         {
@@ -1076,7 +1076,7 @@ public:
         }
         else
             compressHandler = queryDefaultCompressHandler();
-        ActPrintLog("Using compressor: %s", compressHandler ? compressHandler->queryType() : "NONE");
+        ::ActPrintLog(activity, thorDetailedLogLevel, "Using compressor: %s", compressHandler ? compressHandler->queryType() : "NONE");
 
         allowSpill = activity->getOptBool(THOROPT_HDIST_SPILL, true);
         if (allowSpill)
@@ -1084,12 +1084,12 @@ public:
         writerPoolSize = activity->getOptUInt(THOROPT_HDIST_WRITE_POOL_SIZE, DEFAULT_WRITEPOOLSIZE);
         if (writerPoolSize>(numnodes*2))
             writerPoolSize = numnodes*2; // limit to 2 per target
-        ActPrintLog("Writer thread pool size : %d", writerPoolSize);
+        ::ActPrintLog(activity, thorDetailedLogLevel, "Writer thread pool size : %d", writerPoolSize);
         candidateLimit = activity->getOptUInt(THOROPT_HDIST_CANDIDATELIMIT);
-        ActPrintLog("candidateLimit : %d", candidateLimit);
-        ActPrintLog("inputBufferSize : %d, bucketSendSize = %d, pullBufferSize=%d", inputBufferSize, bucketSendSize, pullBufferSize);
+        ::ActPrintLog(activity, thorDetailedLogLevel, "candidateLimit : %d", candidateLimit);
+        ::ActPrintLog(activity, thorDetailedLogLevel, "inputBufferSize : %d, bucketSendSize = %d, pullBufferSize=%d", inputBufferSize, bucketSendSize, pullBufferSize);
         targetWriterLimit = activity->getOptUInt(THOROPT_HDIST_TARGETWRITELIMIT);
-        ActPrintLog("targetWriterLimit : %d", targetWriterLimit);
+        ::ActPrintLog(activity, thorDetailedLogLevel, "targetWriterLimit : %d", targetWriterLimit);
     }
 
     virtual void beforeDispose()
@@ -1133,7 +1133,7 @@ public:
 
     virtual IRowStream *connect(IThorRowInterfaces *_rowIf, IRowStream *_input, IHash *_ihash, ICompare *_iCompare, ICompare *_keepBestCompare)
     {
-        ActPrintLog("HASHDISTRIB: connect");
+        ::ActPrintLog(activity, thorDetailedLogLevel, "HASHDISTRIB: connect");
 
         rowIf.set(_rowIf);
         allocator = _rowIf->queryRowAllocator();
@@ -1164,7 +1164,7 @@ public:
         sendException.clear();
         recvException.clear();
         start();
-        ActPrintLog("HASHDISTRIB: connected");
+        ::ActPrintLog(activity, thorDetailedLogLevel, "HASHDISTRIB: connected");
         return piperd.getLink();
     }
 
@@ -1215,7 +1215,7 @@ public:
         MemoryBuffer tempMb;
         try
         {
-            ActPrintLog("Read loop start");
+            ::ActPrintLog(activity, thorDetailedLogLevel, "Read loop start");
             CMessageBuffer recvMb;
             Owned<ISerialStream> stream = createMemoryBufferSerialStream(tempMb);
             CThorStreamDeserializerSource rowSource;
@@ -1224,14 +1224,12 @@ public:
             Owned<IExpander> expander = getExpander();
             while (left && !aborted)
             {
-#ifdef _FULL_TRACE
-                ActPrintLog("HDIST: Receiving block");
-#endif
+                ::ActPrintLog(activity, thorDetailedLogLevel, "HDIST: Receiving block");
                 unsigned n = recvBlock(recvMb);
                 if (n==(unsigned)-1)
                     break;
 #ifdef _FULL_TRACE
-                ActPrintLog("HDIST: Received block %d from slave %d",recvMb.length(),n+1);
+                ::ActPrintLog(activity, thorDetailedLogLevel, "HDIST: Received block %d from slave %d",recvMb.length(),n+1);
 #endif
                 if (recvMb.length())
                 {
@@ -1276,20 +1274,16 @@ public:
                 else
                 {
                     left--;
-                    ActPrintLog("HDIST: finished slave %d, %d left",n+1,left);
+                    ::ActPrintLog(activity, thorDetailedLogLevel, "HDIST: finished slave %d, %d left",n+1,left);
                 }
-#ifdef _FULL_TRACE
-                ActPrintLog("HDIST: Put block %d from slave %d",recvMb.length(),n+1);
-#endif
+                ::ActPrintLog(activity, thorDetailedLogLevel, "HDIST: Put block %d from slave %d",recvMb.length(),n+1);
             }
         }
         catch (IException *e)
         {
             setRecvExc(e);
         }
-#ifdef _FULL_TRACE
-        ActPrintLog("HDIST: waiting localFinishedSem");
-#endif
+        ::ActPrintLog(activity, thorDetailedLogLevel, "HDIST: waiting localFinishedSem");
     }
 
     void recvloopdone()
@@ -2081,7 +2075,7 @@ public:
     virtual void init(MemoryBuffer &data, MemoryBuffer &slaveData) override
     {
         mptag = container.queryJobChannel().deserializeMPTag(data);
-        ActPrintLog("HASHDISTRIB: %sinit tag %d",mergecmp?"merge, ":"",(int)mptag);
+        ::ActPrintLog(this, thorDetailedLogLevel, "HASHDISTRIB: %sinit tag %d",mergecmp?"merge, ":"",(int)mptag);
 
         if (mergecmp)
             distributor = createPullHashDistributor(this, queryJobChannel().queryJobComm(), mptag, false, this);
@@ -2109,7 +2103,7 @@ public:
     }
     virtual void stop() override
     {
-        ActPrintLog("HASHDISTRIB: stopping");
+        ::ActPrintLog(this, thorDetailedLogLevel, "HASHDISTRIB: stopping");
         if (out)
         {
             out->stop();
@@ -2123,11 +2117,6 @@ public:
         instrm.clear();
         PARENT::stop();
     }
-    virtual void kill() override
-    {
-        ActPrintLog("HASHDISTRIB: kill");
-        CSlaveActivity::kill();
-    }
     virtual void abort() override
     {
         CSlaveActivity::abort();
@@ -2373,18 +2362,19 @@ public:
             if (i==self)
                 sizes[i] = s;
         }
-        for (i=0;i<n;i++) {
 #ifdef _DEBUG
-            ActPrintLog(activity, "after Node %d has %" I64F "d",i, insz[i]);
+        for (i=0;i<n;i++)
+            ::ActPrintLog(activity, thorDetailedLogLevel, "after Node %d has %" I64F "d",i, insz[i]);
 #endif
-        }
         tot = 0;
-        for (i=0;i<n;i++) {
-            if (sizes[i]) {
+        for (i=0;i<n;i++)
+        {
+            if (sizes[i])
+            {
                 if (i==self)
-                    ActPrintLog(activity, "Keep %" I64F "d local",sizes[i]);
+                    ::ActPrintLog(activity, thorDetailedLogLevel, "Keep %" I64F "d local",sizes[i]);
                 else
-                    ActPrintLog(activity, "Redistribute %" I64F "d to %d",sizes[i],i);
+                    ::ActPrintLog(activity, thorDetailedLogLevel, "Redistribute %" I64F "d to %d",sizes[i],i);
             }
             tot += sizes[i];
         }
@@ -3142,7 +3132,6 @@ public:
     }
     void kill()
     {
-        ActPrintLog("kill");
         currentInput.clear();
         bucketHandler.clear();
         bucketHandlerStack.kill();
@@ -3333,7 +3322,7 @@ void CHashTableRowTable::rehash(const void **newRows)
     }
 
     if (maxRows)
-        ActPrintLog(&activity, "Rehashed bucket %d - old size = %d, new size = %d, elements = %d", owner->queryBucketNumber(), maxRows, newMaxRows, htElements);
+        ::ActPrintLog(&activity, thorDetailedLogLevel, "Rehashed bucket %d - old size = %d, new size = %d, elements = %d", owner->queryBucketNumber(), maxRows, newMaxRows, htElements);
 
     const void **oldRows = rows;
     rows = (const void **)_newRows.getClear();
@@ -3420,7 +3409,7 @@ bool CBucket::flush(bool critical)
         {
             if (clearHashTable(critical))
             {
-                PROGLOG("Flushed%s bucket %d - %d elements", critical?"(critical)":"", queryBucketNumber(), count);
+                LOG(MCthorDetailedDebugInfo, thorJob, "Flushed%s bucket %d - %d elements", critical?"(critical)":"", queryBucketNumber(), count);
                 return true;
             }
         }
@@ -3686,7 +3675,7 @@ void CBucketHandler::init(unsigned _numBuckets, IRowStream *keyStream)
         buckets[i] = new CBucket(owner, rowIf, keyIf, extractKey, i, &htRows);
         htRows.setOwner(buckets[i]);
     }
-    ActPrintLog(&owner, "Max %d buckets, current depth = %d", numBuckets, depth+1);
+    ::ActPrintLog(&owner, thorDetailedLogLevel, "Max %d buckets, current depth = %d", numBuckets, depth+1);
     initCallbacks();
     if (keyStream)
     {
@@ -3719,7 +3708,7 @@ CBucketHandler *CBucketHandler::getNextBucketHandler(Owned<IRowStream> &nextInpu
             Owned<IRowStream> keyStream = bucket->getSpillKeyStream(&keyCount);
             dbgassertex(keyStream);
             Owned<CBucketHandler> newBucketHandler = new CBucketHandler(owner, rowIf, keyIf, iRowHash, iKeyHash, iCompare, extractKey, depth+1, div*numBuckets);
-            ActPrintLog(&owner, "Created bucket handler %d, depth %d", currentBucket, depth+1);
+            ::ActPrintLog(&owner, thorDetailedLogLevel, "Created bucket handler %d, depth %d", currentBucket, depth+1);
             nextInput.setown(bucket->getSpillRowStream(&count));
             dbgassertex(nextInput);
             // Use peak in mem keys as estimate for next round of buckets.
@@ -3796,7 +3785,6 @@ public:
     }
     virtual void stop() override
     {
-        ActPrintLog("stopping");
         if (instrm)
         {
             instrm->stop();
@@ -3879,7 +3867,7 @@ public:
         joinargs = (IHThorHashJoinArg *)queryHelper();
         mptag = container.queryJobChannel().deserializeMPTag(data);
         mptag2 = container.queryJobChannel().deserializeMPTag(data);
-        ActPrintLog("HASHJOIN: init tags %d,%d",(int)mptag,(int)mptag2);
+        ::ActPrintLog(this, thorDetailedLogLevel, "HASHJOIN: init tags %d,%d",(int)mptag,(int)mptag2);
     }
     virtual void start()
     {
@@ -3887,7 +3875,6 @@ public:
         startAllInputs();
         leftdone = false;
         eof = false;
-        ActPrintLog("HASHJOIN: starting");
         inL = queryInput(0);
         inR = queryInput(1);
         IHash *ihashL = joinargs->queryHashLeft();
@@ -3957,7 +3944,6 @@ public:
     }
     virtual void stop()
     {
-        ActPrintLog("HASHJOIN: stopping");
         stopInputL();
         stopInputR();
         if (joinhelper)
@@ -3973,11 +3959,6 @@ public:
         }
         PARENT::stop();
     }
-    void kill()
-    {
-        ActPrintLog("HASHJOIN: kill");
-        CSlaveActivity::kill();
-    }
     void abort()
     {
         CSlaveActivity::abort();
@@ -4436,7 +4417,7 @@ public:
         if (!container.queryLocalOrGrouped())
         {
             mptag = container.queryJobChannel().deserializeMPTag(data);
-            ActPrintLog("HASHAGGREGATE: init tags %d",(int)mptag);
+            ::ActPrintLog(this, thorDetailedLogLevel, "HASHAGGREGATE: init tags %d",(int)mptag);
         }
         localAggTable.setown(createRowAggregator(*this, *helper, *helper));
         localAggTable->init(queryRowAllocator());
@@ -4447,7 +4428,7 @@ public:
         PARENT::start();
         doNextGroup(); // or local set if !grouped
         if (!container.queryGrouped())
-            ActPrintLog("Table before distribution contains %d entries", localAggTable->elementCount());
+            ::ActPrintLog(this, thorDetailedLogLevel, "Table before distribution contains %d entries", localAggTable->elementCount());
         if (!container.queryLocalOrGrouped() && container.queryJob().querySlaves()>1)
         {
             Owned<IRowStream> localAggStream = localAggTable->getRowStream(true);
@@ -4459,7 +4440,6 @@ public:
     }
     virtual void stop() override
     {
-        ActPrintLog("HASHAGGREGATE: stopping");
         if (localAggTable)
             localAggTable->reset();
         if (aggregateStream)
@@ -4573,55 +4553,46 @@ CActivityBase *createHashDistributeSlave(CGraphElementBase *container)
 {
     if (container&&(((IHThorHashDistributeArg *)container->queryHelper())->queryHash()==NULL))
         return createReDistributeSlave(container);
-    ActPrintLog(container, "HASHDISTRIB: createHashDistributeSlave");
     return new HashDistributeSlaveActivity(container);
 }
 
 CActivityBase *createNWayDistributeSlave(CGraphElementBase *container)
 {
-    ActPrintLog(container, "NWAYDISTRIB: createNWayDistributeSlave");
     return new NWayDistributeSlaveActivity(container);
 }
 
 CActivityBase *createHashDistributeMergeSlave(CGraphElementBase *container)
 {
-    ActPrintLog(container, "HASHDISTRIB: createHashDistributeMergeSlave");
     return new HashDistributeMergeSlaveActivity(container);
 }
 
 CActivityBase *createHashDedupSlave(CGraphElementBase *container)
 {
-    ActPrintLog(container, "HASHDEDUP: createHashDedupSlave");
     return new GlobalHashDedupSlaveActivity(container);
 }
 
 CActivityBase *createHashLocalDedupSlave(CGraphElementBase *container)
 {
-    ActPrintLog(container, "LOCALHASHDEDUP: createHashLocalDedupSlave");
     return new LocalHashDedupSlaveActivity(container);
 }
 
 CActivityBase *createHashJoinSlave(CGraphElementBase *container)
 {
-    ActPrintLog(container, "HASHJOIN: createHashJoinSlave");
     return new HashJoinSlaveActivity(container);
 }
 
 CActivityBase *createHashAggregateSlave(CGraphElementBase *container)
 {
-    ActPrintLog(container, "HASHAGGREGATE: createHashAggregateSlave");
     return new CHashAggregateSlave(container);
 }
 
 CActivityBase *createIndexDistributeSlave(CGraphElementBase *container)
 {
-    ActPrintLog(container, "DISTRIBUTEINDEX: createIndexDistributeSlave");
     return new IndexDistributeSlaveActivity(container);
 }
 
 CActivityBase *createReDistributeSlave(CGraphElementBase *container)
 {
-    ActPrintLog(container, "REDISTRIBUTE: createReDistributeSlave");
     return new ReDistributeSlaveActivity(container);
 }
 

thorlcr/activities/indexread/thindexreadslave.cpp

@@ -1108,7 +1108,7 @@ public:
                 }
                 if (_currentManager)
                     prepareManager(_currentManager);
-                ActPrintLog("INDEXGROUPAGGREGATE: Local aggregate table contains %d entries", localAggTable->elementCount());
+                ::ActPrintLog(this, thorDetailedLogLevel, "INDEXGROUPAGGREGATE: Local aggregate table contains %d entries", localAggTable->elementCount());
                 if (!container.queryLocal() && container.queryJob().querySlaves()>1)
                 {
                     Owned<IRowStream> localAggStream = localAggTable->getRowStream(true);

thorlcr/activities/indexwrite/thindexwriteslave.cpp

@@ -266,7 +266,7 @@ public:
     }
     virtual void process() override
     {
-        ActPrintLog("INDEXWRITE: Start");
+        ::ActPrintLog(this, thorDetailedLogLevel, "INDEXWRITE: Start");
         init();
 
         IRowStream *stream = inputStream;
@@ -404,9 +404,8 @@ public:
                 {
                     StringBuffer partFname;
                     getPartFilename(*partDesc, 0, partFname);
-                    ActPrintLog("INDEXWRITE: process: handling fname : %s", partFname.str());
+                    ::ActPrintLog(this, thorDetailedLogLevel, "INDEXWRITE: process: handling fname : %s", partFname.str());
                     open(*partDesc, false, helper->queryDiskRecordSize()->isVariableSize(), false);
-                    ActPrintLog("INDEXWRITE: write");
 
                     BooleanOnOff tf(receiving);
                     if (!refactor || !active)
@@ -418,7 +417,6 @@ public:
                             break;
                         processRow(row);
                     } while (!abortSoon);
-                    ActPrintLog("INDEXWRITE: write level 0 complete");
                 }
                 catch (CATCHALL)
                 {
@@ -433,7 +431,7 @@ public:
 
                 if (buildTlk)
                 {
-                    ActPrintLog("INDEXWRITE: sending rows");
+                    ::ActPrintLog(this, thorDetailedLogLevel, "INDEXWRITE: sending rows");
                     NodeInfoArray tlkRows;
 
                     CMessageBuffer msg;
@@ -476,7 +474,7 @@ public:
 
                     if (firstNode())
                     {
-                        ActPrintLog("INDEXWRITE: Waiting on tlk to complete");
+                        ::ActPrintLog(this, thorDetailedLogLevel, "INDEXWRITE: Waiting on tlk to complete");
 
                         // JCSMORE if refactor==true, is rowsToReceive here right??
                         unsigned rowsToReceive = (refactor ? (tlkDesc->queryOwner().numParts()-1) : container.queryJob().querySlaves()) -1; // -1 'cos got my own in array already
@@ -506,7 +504,7 @@ public:
 
                         StringBuffer path;
                         getPartFilename(*tlkDesc, 0, path);
-                        ActPrintLog("INDEXWRITE: creating toplevel key file : %s", path.str());
+                        ::ActPrintLog(this, thorDetailedLogLevel, "INDEXWRITE: creating toplevel key file : %s", path.str());
                         try
                         {
                             open(*tlkDesc, true, helper->queryDiskRecordSize()->isVariableSize(), true);
@@ -549,7 +547,7 @@ public:
                     }
                 }
             }
-            ActPrintLog("INDEXWRITE: All done");
+            ::ActPrintLog(this, thorDetailedLogLevel, "INDEXWRITE: All done");
         }
     }
     virtual void endProcess() override

thorlcr/activities/keyedjoin/thkeyedjoinslave-legacy.cpp

@@ -2027,11 +2027,13 @@ public:
                 fetchHandler = new CKeyedFetchHandler(*this);
 
                 FPosTableEntry *fPosToNodeMap = globalFPosToNodeMap ? globalFPosToNodeMap : localFPosToNodeMap;
-                unsigned c;
-                for (c=0; c<filePartTotal; c++)
+                if (!REJECTLOG(MCthorDetailedDebugInfo))
                 {
-                    FPosTableEntry &e = fPosToNodeMap[c];
-                    ActPrintLog("Table[%d] : base=%" I64F "d, top=%" I64F "d, slave=%d", c, e.base, e.top, e.index);
+                    for (unsigned c=0; c<filePartTotal; c++)
+                    {
+                        FPosTableEntry &e = fPosToNodeMap[c];
+                        ::ActPrintLog(this, thorDetailedLogLevel, "Table[%d] : base=%" I64F "d, top=%" I64F "d, slave=%d", c, e.base, e.top, e.index);
+                    }
                 }
                 unsigned i=0;
                 for(; i<dataParts.ordinality(); i++)

thorlcr/activities/lookupjoin/thlookupjoinslave.cpp

@@ -2156,9 +2156,12 @@ protected:
                 }
             }
             InterChannelBarrier();
-            ActPrintLog("Shared memory manager memory report");
-            rightRowManager->reportMemoryUsage(false);
-            ActPrintLog("End of shared manager memory report");
+            if (!REJECTLOG(MCthorDetailedDebugInfo))
+            {
+                ::ActPrintLog(this, thorDetailedLogLevel, "Shared memory manager memory report");
+                rightRowManager->reportMemoryUsage(false);
+                ::ActPrintLog(this, thorDetailedLogLevel, "End of shared manager memory report");
+            }
         }
         else
         {
@@ -2178,9 +2181,12 @@ protected:
                 setFailoverToLocal();
             rhsCollated = lkJoinCh0->isRhsCollated();
         }
-        ActPrintLog("Channel memory manager report");
-        queryRowManager()->reportMemoryUsage(false);
-        ActPrintLog("End of channel memory manager report");
+        if (!REJECTLOG(MCthorDetailedDebugInfo))
+        {
+            ::ActPrintLog(this, thorDetailedLogLevel, "Channel memory manager report");
+            queryRowManager()->reportMemoryUsage(false);
+            ::ActPrintLog(this, thorDetailedLogLevel, "End of channel memory manager report");
+        }
         return !hasFailedOverToLocal();
     }
     /*

thorlcr/activities/merge/thmerge.cpp

@@ -18,10 +18,6 @@
 #include "thmerge.ipp"
 #include "tsorta.hpp"
 
-#ifdef _DEBUG
-#define _TRACE
-#endif
-
 //
 // LocalMergeActivityMaster
 
@@ -55,7 +51,6 @@ public:
     }
     virtual void process()
     {
-        ActPrintLog("GlobalMergeActivityMaster::process");
         CMasterActivity::process();     
         IHThorMergeArg *helper = (IHThorMergeArg *)queryHelper();   
         Owned<IThorRowInterfaces> rowif = createRowInterfaces(helper->queryOutputMeta());
@@ -74,15 +69,11 @@ public:
                 if (abortSoon)
                     return;
                 CMessageBuffer mb;
-#ifdef _TRACE
-                ActPrintLog("Merge process, Receiving on tag %d",replyTag);
-#endif
+                ::ActPrintLog(this, thorDetailedLogLevel, "Merge process, Receiving on tag %d",replyTag);
                 rank_t sender;
                 if (!receiveMsg(mb, RANK_ALL, replyTag, &sender)||abortSoon) 
                     return;
-#ifdef _TRACE
-                ActPrintLog("Merge process, Received sample from %d",sender);
-#endif
+                ::ActPrintLog(this, thorDetailedLogLevel, "Merge process, Received sample from %d",sender);
                 sender--;
                 assertex((unsigned)sender<n);
                 assertex(replytags[(unsigned)sender]==TAG_NULL);
@@ -101,9 +92,7 @@ public:
                 for (unsigned j = 0;j<n;j++)
                     serializeMPtag(mb,intertags[j]);
                 sample.serialize(mb);
-#ifdef _TRACE
-                ActPrintLog("Merge process, Replying to node %d tag %d",i+1,replytags[i]);
-#endif
+                ::ActPrintLog(this, thorDetailedLogLevel, "Merge process, Replying to node %d tag %d",i+1,replytags[i]);
                 if (!queryJobChannel().queryJobComm().send(mb, (rank_t)i+1, replytags[i]))
                     break;
             }

thorlcr/activities/merge/thmergeslave.cpp

@@ -232,7 +232,7 @@ public:
             }
 #endif
 
-            ActPrintLog("Merge: partitionpos[%d] = %" I64F "d",i,partitionpos[i]);
+            ::ActPrintLog(this, thorDetailedLogLevel, "Merge: partitionpos[%d] = %" I64F "d",i,partitionpos[i]);
         }
         delete [] intertags;
         provider.init(this,queryRowSerializer(),intertag);
@@ -274,7 +274,6 @@ public:
 
     void abort()
     {
-        ActPrintLog("abort");
         CSlaveActivity::abort();
         provider.stop();
     }
@@ -424,11 +423,6 @@ public:
         setRequireInitData(false);
         appendOutputLinked(this);
     }
-    void abort()
-    {
-        ActPrintLog("abort");
-        CSlaveActivity::abort();
-    }
 
 // IThorDataLink
     virtual void start() override

thorlcr/activities/msort/thmsort.cpp

@@ -116,7 +116,6 @@ protected:
     virtual void preStart(size32_t parentExtractSz, const byte *parentExtract)
     {
         CSortBaseActivityMaster::preStart(parentExtractSz, parentExtract);
-        ActPrintLog("preStart");
         imaster = CreateThorSorterMaster(this);
         unsigned s=0;
         for (; s<container.queryJob().querySlaves(); s++)
@@ -128,8 +127,6 @@ protected:
     }
     virtual void process()
     {
-        ActPrintLog("process");
-
         CSortBaseActivityMaster::process();
 
         IHThorSortArg *helper = (IHThorSortArg *)queryHelper();
@@ -195,7 +192,6 @@ protected:
             throw;
         }
         ::Release(imaster);
-        ActPrintLog("process exit");
     }
 };
 

thorlcr/activities/msort/thmsortslave.cpp

@@ -173,8 +173,6 @@ public:
     }
     virtual void kill() override
     {
-        ActPrintLog("MSortSlaveActivity::kill");
-
         {
             CriticalBlock block(statsCs);
             mergeStats(stats, sorter, spillStatistics);
@@ -222,7 +220,6 @@ public:
 
 CActivityBase *createMSortSlave(CGraphElementBase *container)
 {
-    ActPrintLog(container, "MSortSlaveActivity::createMSortSlave");
     return new MSortSlaveActivity(container);
 }
 

thorlcr/activities/msort/thsortu.cpp

@@ -1637,7 +1637,7 @@ class CMultiCoreJoinHelper: public CMultiCoreJoinHelperBase
         }
         int run()
         {
-            PROGLOG("CMultiCoreJoinHelper::cReader started");
+            LOG(MCthorDetailedDebugInfo, thorJob, "CMultiCoreJoinHelper::cReader started");
             try {
                 const void * row = parent->jhelper->nextRow();
                 assertex(!row);
@@ -1647,7 +1647,7 @@ class CMultiCoreJoinHelper: public CMultiCoreJoinHelperBase
             }
             for (unsigned i=0;i<parent->numworkers;i++) 
                 parent->addWork(NULL,NULL);
-            PROGLOG("CMultiCoreJoinHelper::cReader exit");
+            LOG(MCthorDetailedDebugInfo, thorJob, "CMultiCoreJoinHelper::cReader exit");
             return 0;
         }
     } reader;
@@ -1671,7 +1671,7 @@ class CMultiCoreJoinHelper: public CMultiCoreJoinHelperBase
         }
         int run()
         {
-            PROGLOG("CMultiCoreJoinHelper::cWorker started");
+            LOG(MCthorDetailedDebugInfo, thorJob, "CMultiCoreJoinHelper::cWorker started");
 
             Owned<IThorRowInterfaces> rowIf = parent->activity.getRowInterfaces();
             Owned<IEngineRowAllocator> allocator = parent->activity.getRowAllocator(rowIf->queryRowMetaData(), (roxiemem::RHFpacked|roxiemem::RHFunique));
@@ -1704,7 +1704,7 @@ class CMultiCoreJoinHelper: public CMultiCoreJoinHelperBase
             rowWriter->putRow(NULL); // end-of-stream
             rowWriter->flush();
             stopped = true; // NB: will not get past flush(), until all read
-            PROGLOG("CMultiCoreJoinHelper::cWorker exit");
+            LOG(MCthorDetailedDebugInfo, thorJob, "CMultiCoreJoinHelper::cWorker exit");
             return 0;
         }
         bool isStopped() const { return stopped; }
@@ -1848,7 +1848,7 @@ class CMultiCoreUnorderedJoinHelper: public CMultiCoreJoinHelperBase
         }
         int run()
         {
-            PROGLOG("CMulticoreUnorderedJoinHelper::cReader started");
+            LOG(MCthorDetailedDebugInfo, thorJob, "CMulticoreUnorderedJoinHelper::cReader started");
             try {
                 const void * row = parent->jhelper->nextRow();
                 assertex(!row);
@@ -1857,7 +1857,7 @@ class CMultiCoreUnorderedJoinHelper: public CMultiCoreJoinHelperBase
                 parent->setException(e,"CMulticoreUnorderedJoinHelper::cReader");
             }
             parent->stopWorkers();
-            PROGLOG("CMulticoreUnorderedJoinHelper::cReader exit");
+            LOG(MCthorDetailedDebugInfo, thorJob, "CMulticoreUnorderedJoinHelper::cReader exit");
             return 0;
         }
     } reader;
@@ -1876,7 +1876,7 @@ class CMultiCoreUnorderedJoinHelper: public CMultiCoreJoinHelperBase
             Owned<IEngineRowAllocator> allocator = parent->activity.getRowAllocator(rowIf->queryRowMetaData(), (roxiemem::RHFpacked|roxiemem::RHFunique));
 
             Owned<IRowWriter> rowWriter = parent->multiWriter->getWriter();
-            PROGLOG("CMulticoreUnorderedJoinHelper::cWorker started");
+            LOG(MCthorDetailedDebugInfo, thorJob, "CMulticoreUnorderedJoinHelper::cWorker started");
             for (;;)
             {
                 cWorkItem *work = parent->workqueue.dequeue();
@@ -1897,7 +1897,7 @@ class CMultiCoreUnorderedJoinHelper: public CMultiCoreJoinHelperBase
                     break;
                 }
             }
-            PROGLOG("CMulticoreUnorderedJoinHelper::cWorker exit");
+            LOG(MCthorDetailedDebugInfo, thorJob, "CMulticoreUnorderedJoinHelper::cWorker exit");
             return 0;
         }
     } **workers;
@@ -1911,7 +1911,7 @@ public:
         unsigned limit = activity.queryContainer().queryXGMML().getPropInt("hint[@name=\"limit\"]/@value", numworkers*1000);
         unsigned readGranularity = activity.queryContainer().queryXGMML().getPropInt("hint[@name=\"readgranularity\"]/@value", DEFAULT_WR_READ_GRANULARITY);
         unsigned writeGranularity = activity.queryContainer().queryXGMML().getPropInt("hint[@name=\"writegranularity\"]/@value", DEFAULT_WR_WRITE_GRANULARITY);
-        ActPrintLog(&activity, "CMultiCoreUnorderedJoinHelper: limit=%d, readGranularity=%d, writeGranularity=%d", limit, readGranularity, writeGranularity);
+        ActPrintLog(&activity, thorDetailedLogLevel, "CMultiCoreUnorderedJoinHelper: limit=%d, readGranularity=%d, writeGranularity=%d", limit, readGranularity, writeGranularity);
         multiWriter.setown(createSharedWriteBuffer(&activity, rowIf, limit, readGranularity, writeGranularity));
         workers = new cWorker *[numthreads];
         for (unsigned i=0;i<numthreads;i++) 
@@ -2013,7 +2013,7 @@ IJoinHelper *createJoinHelper(CActivityBase &activity, IHThorJoinArg *helper, IT
     unsigned numthreads = activity.getOptInt(THOROPT_JOINHELPER_THREADS, 0);
     if (0 == numthreads)
         numthreads = activity.queryMaxCores();
-    ActPrintLog(&activity, "Join helper using %d threads", numthreads);
+    ActPrintLog(&activity, thorDetailedLogLevel, "Join helper using %d threads", numthreads);
     if (unsortedoutput)
         return new CMultiCoreUnorderedJoinHelper(activity, numthreads, false, jhelper, helper, rowIf);
     return new CMultiCoreJoinHelper(activity, numthreads, false, jhelper, helper, rowIf);
@@ -2034,7 +2034,7 @@ IJoinHelper *createSelfJoinHelper(CActivityBase &activity, IHThorJoinArg *helper
     unsigned numthreads = activity.getOptInt(THOROPT_JOINHELPER_THREADS, 0);
     if (0 == numthreads)
         numthreads = activity.queryMaxCores();
-    ActPrintLog(&activity, "Self join helper using %d threads", numthreads);
+    ActPrintLog(&activity, thorDetailedLogLevel, "Self join helper using %d threads", numthreads);
     if (unsortedoutput)
         return new CMultiCoreUnorderedJoinHelper(activity, numthreads, true, jhelper, helper, rowIf);
     return new CMultiCoreJoinHelper(activity, numthreads, true, jhelper, helper, rowIf);

thorlcr/activities/nsplitter/thnsplitterslave.cpp

@@ -175,7 +175,7 @@ public:
                  break;
              cur = next+1;
         }
-        ActPrintLog("Number of connected outputs: %u", connectedOutputCount);
+        ::ActPrintLog(this, thorDetailedLogLevel, "Number of connected outputs: %u", connectedOutputCount);
     }
     virtual void reset() override
     {

thorlcr/activities/pipewrite/thpwslave.cpp

@@ -124,7 +124,6 @@ public:
         start();
         Owned<IPipeProcessException> pipeException;
 
-        ActPrintLog("process");
         try
         {
             if (!writeTransformer)
@@ -145,7 +144,7 @@ public:
         verifyPipe();
         if (pipeException) // NB: verifyPipe may throw error based on pipe prog. output 1st.
             throw pipeException.getClear();
-        ActPrintLog("Wrote %" RCPF "d records", processed & THORDATALINK_COUNT_MASK);
+        ::ActPrintLog(this, thorDetailedLogLevel, "Wrote %" RCPF "d records", processed & THORDATALINK_COUNT_MASK);
     }
     void endProcess()
     {
@@ -163,7 +162,6 @@ public:
     }
     void write()
     {
-        ActPrintLog("write");
         while (!abortSoon)
         {
             OwnedConstThorRow row = inputStream->ungroupedNextRow();

thorlcr/activities/rollup/throllupslave.cpp

@@ -104,9 +104,7 @@ public:
     }
     bool calcNextDedupAll(bool groupOp)
     {
-#if THOR_TRACE_LEVEL >=5
-        ActPrintLog(activity, "DedupAllHelper::calcNextDedupAll");
-#endif
+        ActPrintLog(activity, thorDetailedLogLevel, "DedupAllHelper::calcNextDedupAll");
         dedupIdx = 0;
         rows.kill();
 
@@ -123,7 +121,7 @@ public:
             throw checkAndCreateOOMContextException(activity, e, "loading group for dedup all", rowLoader->numRows(), inputOutputMeta, rowLoader->probeRow(0));
         }
         dedupCount = rows.ordinality();
-        ActPrintLog(activity, "DEDUP: rows loaded = %d",dedupCount);
+        ActPrintLog(activity, thorDetailedLogLevel, "DEDUP: rows loaded = %d",dedupCount);
 
         if (iStopInput)
             iStopInput->stopInput();

thorlcr/activities/selectnth/thselectnthslave.cpp

@@ -50,7 +50,7 @@ class CSelectNthSlaveActivity : public CSlaveActivity, implements ILookAheadStop
                 createDefaultIfFail = true;
         }
         startN = N;
-        ActPrintLog("SELECTNTH: Selecting row %" I64F "d", N);
+        ::ActPrintLog(this, thorDetailedLogLevel, "SELECTNTH: Selecting row %" I64F "d", N);
     }
     void sendN()
     {
@@ -105,19 +105,17 @@ public:
         }
 
         seenNth = false;
-        if (0 == rowN)
+        if (!REJECTLOG(MCthorDetailedDebugInfo))
         {
-            ThorDataLinkMetaInfo info;
-            queryInput(0)->getMetaInfo(info);
-            StringBuffer meta;
-            meta.appendf("META(totalRowsMin=%" I64F "d,totalRowsMax=%" I64F "d, spilled=%" I64F "d,byteTotal=%" I64F "d)",
-                info.totalRowsMin,info.totalRowsMax,info.spilled,info.byteTotal);
-#if 0                 
-            Owned<IThorException> e = MakeActivityWarning(this, -1, "%s", meta.str());
-            fireException(e);
-#else
-            ActPrintLog("%s",meta.str());
-#endif
+            if (0 == rowN)
+            {
+                ThorDataLinkMetaInfo info;
+                queryInput(0)->getMetaInfo(info);
+                StringBuffer meta;
+                meta.appendf("META(totalRowsMin=%" I64F "d,totalRowsMax=%" I64F "d, spilled=%" I64F "d,byteTotal=%" I64F "d)",
+                    info.totalRowsMin,info.totalRowsMax,info.spilled,info.byteTotal);
+                ::ActPrintLog(this, thorDetailedLogLevel, "%s", meta.str());
+            }
         }
         first = true;
     }

thorlcr/activities/selfjoin/thselfjoinslave.cpp

@@ -58,9 +58,7 @@ private:
 
     IRowStream * doLocalSelfJoin()
     {
-#if THOR_TRACE_LEVEL > 5
-        ActPrintLog("SELFJOIN: Performing local self-join");
-#endif
+        ::ActPrintLog(this, thorDetailedLogLevel, "SELFJOIN: Performing local self-join");
         Owned<IThorRowLoader> iLoader = createThorRowLoader(*this, ::queryRowInterfaces(input), compare, isUnstable() ? stableSort_none : stableSort_earlyAlloc, rc_mixed, SPILL_PRIORITY_SELFJOIN);
         Owned<IRowStream> rs = iLoader->load(inputStream, abortSoon);
         mergeStats(stats, iLoader, spillStatistics);  // Not sure of the best policy if rs spills later on.
@@ -70,9 +68,7 @@ private:
 
     IRowStream * doGlobalSelfJoin()
     {
-#if THOR_TRACE_LEVEL > 5
-        ActPrintLog("SELFJOIN: Performing global self-join");
-#endif
+        ::ActPrintLog(this, thorDetailedLogLevel, "SELFJOIN: Performing global self-join");
         sorter->Gather(::queryRowInterfaces(input), inputStream, compare, NULL, NULL, keyserializer, NULL, NULL, false, isUnstable(), abortSoon, NULL);
         PARENT::stopInput(0);
         if (abortSoon)
@@ -126,12 +122,12 @@ public:
         }
         compare = helper->queryCompareLeft();                   // NB not CompareLeftRight
         keyserializer = helper->querySerializeLeft();           // hopefully never need right
-        if(isLightweight) 
-            ActPrintLog("SELFJOIN: LIGHTWEIGHT");
+        if (isLightweight) 
+            ::ActPrintLog(this, thorDetailedLogLevel, "SELFJOIN: LIGHTWEIGHT");
         else if(isLocal) 
-            ActPrintLog("SELFJOIN: LOCAL");
+            ::ActPrintLog(this, thorDetailedLogLevel, "SELFJOIN: LOCAL");
         else
-            ActPrintLog("SELFJOIN: GLOBAL");
+            ::ActPrintLog(this, thorDetailedLogLevel, "SELFJOIN: GLOBAL");
     }
     virtual void reset() override
     {

thorlcr/activities/temptable/thtmptableslave.cpp

@@ -63,9 +63,9 @@ public:
             __uint64 nodeid = queryCodeContext()->getNodeNum();
             startRow = (nodeid * numRows) / nodes;
             maxRow = ((nodeid + 1) * numRows) / nodes;
-            ActPrintLog("InlineSLAVE: numRows = %" I64F "d, nodes = %" I64F
-                        "d, nodeid = %" I64F "d, start = %" I64F "d, max = %" I64F "d",
-                        numRows, nodes, nodeid, startRow, maxRow);
+            ::ActPrintLog(this, thorDetailedLogLevel, "InlineSLAVE: numRows = %" I64F "d, nodes = %" I64F
+                         "d, nodeid = %" I64F "d, start = %" I64F "d, max = %" I64F "d",
+                         numRows, nodes, nodeid, startRow, maxRow);
         }
         else
         {

thorlcr/activities/thdiskbase.cpp

@@ -250,7 +250,7 @@ void CWriteMasterBase::publish()
             {
                 StringBuffer clusterName;
                 fileDesc->getClusterGroupName(clusterIdx, clusterName, &queryNamedGroupStore());
-                PROGLOG("Creating blank parts for file '%s', cluster '%s'", fileName.get(), clusterName.str());
+                LOG(MCthorDetailedDebugInfo, thorJob, "Creating blank parts for file '%s', cluster '%s'", fileName.get(), clusterName.str());
                 unsigned p=0;
                 while (p<fileDesc->numParts())
                 {

thorlcr/activities/thdiskbaseslave.cpp

@@ -139,7 +139,7 @@ void CDiskPartHandlerBase::open()
         }
     }
 
-    ActPrintLog(&activity, "%s[part=%d]: Base offset to %" I64F "d", kindStr, which, fileBaseOffset);
+    ActPrintLog(&activity, thorDetailedLogLevel, "%s[part=%d]: Base offset to %" I64F "d", kindStr, which, fileBaseOffset);
 
     if (compressed)
     {
@@ -149,10 +149,10 @@ void CDiskPartHandlerBase::open()
             checkFileCrc = false;
             if (activity.crcCheckCompressed) // applies to encrypted too, (optional, default off)
             {
-                ActPrintLog(&activity, "Calculating crc for file: %s", filename.get());
+                ActPrintLog(&activity, thorDetailedLogLevel, "Calculating crc for file: %s", filename.get());
                 unsigned calcCrc = iFile->getCRC();
                 // NB: for compressed files should always be ~0
-                ActPrintLog(&activity, "Calculated crc = %x, storedCrc = %x", calcCrc, storedCrc);
+                ActPrintLog(&activity, thorDetailedLogLevel, "Calculated crc = %x, storedCrc = %x", calcCrc, storedCrc);
                 if (calcCrc != storedCrc)
                 {
                     IThorException *e = MakeActivityException(&activity, TE_FileCrc, "CRC Failure validating compressed file: %s", iFile->queryFilename());

thorlcr/activities/topn/thtopnslave.cpp

@@ -157,7 +157,7 @@ public:
                             break; // never reading from node 0 (0 == terminator)
                         StringBuffer s;
                         s.appendN(indent, ' ').append("Merging from node: ").append(node);
-                        ActPrintLog("%s", s.str());
+                        ::ActPrintLog(this, thorDetailedLogLevel, "%s", s.str());
                         streams.append(*createRowStreamFromNode(*this, node+1, queryJobChannel().queryJobComm(), mpTag, abortSoon));
                     }
                     Owned<IRowLinkCounter> linkcounter = new CThorRowLinkCounter;

thorlcr/activities/wuidwrite/thwuidwrite.cpp

@@ -57,10 +57,10 @@ protected:
     {
         if (resultData.length() || complete)
         {
-            ActPrintLog("flushing result");
+            ::ActPrintLog(this, thorDetailedLogLevel, "flushing result");
             addResult(numResults, resultData, complete);
             resultData.clear();
-            ActPrintLog("result flushed");
+            ::ActPrintLog(this, thorDetailedLogLevel, "result flushed");
         }
     }
 public:
@@ -308,7 +308,7 @@ public:
     virtual void flushResults(bool complete=false) override
     {
         assertex(complete);
-        ActPrintLog("dictionary result");
+        ::ActPrintLog(this, thorDetailedLogLevel, "dictionary result");
         Owned<IThorRowInterfaces> rowIf = createRowInterfaces(container.queryInput(0)->queryHelper()->queryOutputMeta());
         IOutputRowDeserializer *deserializer = rowIf->queryRowDeserializer();
         CMessageBuffer mb;
@@ -331,7 +331,7 @@ public:
         rtlSerializeDictionary(out, rowIf->queryRowSerializer(), builder.getcount(), builder.queryrows());
         addResult(usedCount, rowData, complete);
         resultData.clear();
-        ActPrintLog("dictionary flushed");
+        ::ActPrintLog(this, thorDetailedLogLevel, "dictionary flushed");
     }
 };
 

thorlcr/activities/wuidwrite/thwuidwriteslave.cpp

@@ -95,7 +95,7 @@ public:
         if (ensureStartFTLookAhead(0))
             setLookAhead(0, createRowStreamLookAhead(this, inputStream, queryRowInterfaces(input), WORKUNITWRITE_SMART_BUFFER_SIZE, ::canStall(input), grouped, RCUNBOUND, NULL, &container.queryJob().queryIDiskUsage()), false);
 
-        ActPrintLog("WORKUNITWRITE: processing first block");
+        ::ActPrintLog(this, thorDetailedLogLevel, "WORKUNITWRITE: processing first block");
 
         CMessageBuffer replyMb;
         replyMb.append((unsigned)0);
@@ -147,7 +147,7 @@ public:
         start();
         processed = THORDATALINK_STARTED;
 
-        ActPrintLog("WORKUNITWRITELOCAL: processing first block");
+        ::ActPrintLog(this, thorDetailedLogLevel, "WORKUNITWRITELOCAL: processing first block");
 
         CMessageBuffer reqMsg;
         reqMsg.setReplyTag(replyTag);

thorlcr/graph/thgraph.cpp

@@ -544,7 +544,7 @@ void CGraphElementBase::addInput(unsigned input, CGraphElementBase *inputAct, un
 
 void CGraphElementBase::connectInput(unsigned input, CGraphElementBase *inputAct, unsigned inputOutIdx)
 {
-    ActPrintLog("CONNECTING (id=%" ACTPF "d, idx=%d) to (id=%" ACTPF "d, idx=%d)", inputAct->queryId(), inputOutIdx, queryId(), input);
+    ::ActPrintLog(this, thorDetailedLogLevel, "CONNECTING (id=%" ACTPF "d, idx=%d) to (id=%" ACTPF "d, idx=%d)", inputAct->queryId(), inputOutIdx, queryId(), input);
     while (connectedInputs.ordinality()<=input) connectedInputs.append(NULL);
     connectedInputs.replace(new COwningSimpleIOConnection(LINK(inputAct), inputOutIdx), input);
     while (inputAct->connectedOutputs.ordinality()<=inputOutIdx) inputAct->connectedOutputs.append(NULL);
@@ -1183,12 +1183,12 @@ void traceMemUsage()
 {
     StringBuffer memStatsStr;
     roxiemem::memstats(memStatsStr);
-    PROGLOG("Roxiemem stats: %s", memStatsStr.str());
+    LOG(MCthorDetailedDebugInfo, thorJob, "Roxiemem stats: %s", memStatsStr.str());
     memsize_t heapUsage = getMapInfo("heap");
     if (heapUsage) // if 0, assumed to be unavailable
     {
         memsize_t rmtotal = roxiemem::getTotalMemoryLimit();
-        PROGLOG("Heap usage (excluding Roxiemem) : %" I64F "d bytes", (unsigned __int64)(heapUsage-rmtotal));
+        LOG(MCthorDetailedDebugInfo, thorJob, "Heap usage (excluding Roxiemem) : %" I64F "d bytes", (unsigned __int64)(heapUsage-rmtotal));
     }
 }
 
@@ -1372,9 +1372,12 @@ void CGraphBase::executeSubGraph(size32_t parentExtractSz, const byte *parentExt
     {
         if (!queryOwner())
         {
-            StringBuffer s;
-            toXML(&queryXGMML(), s, 2);
-            GraphPrintLog("Running graph [%s] : %s", isGlobal()?"global":"local", s.str());
+            if (!REJECTLOG(MCthorDetailedDebugInfo))
+            {
+                StringBuffer s;
+                toXML(&queryXGMML(), s, 2);
+                LOG(MCthorDetailedDebugInfo, thorJob, "Running graph [%s] : %s", isGlobal()?"global":"local", s.str());
+            }
         }
         if (localResults)
             localResults->clear();
@@ -1388,9 +1391,12 @@ void CGraphBase::executeSubGraph(size32_t parentExtractSz, const byte *parentExt
     if (!queryOwner())
     {
         GraphPrintLog("Graph Done");
-        StringBuffer memStr;
-        getSystemTraceInfo(memStr, PerfMonStandard | PerfMonExtended);
-        GraphPrintLog("%s", memStr.str());
+        if (!REJECTLOG(MCthorDetailedDebugInfo))
+        {
+            StringBuffer memStr;
+            getSystemTraceInfo(memStr, PerfMonStandard | PerfMonExtended);
+            ::GraphPrintLog(this, thorDetailedLogLevel, "%s", memStr.str());
+        }
     }
     if (exception)
         throw exception.getClear();
@@ -1776,6 +1782,8 @@ void CGraphBase::abort(IException *e)
     if (aborted)
         return;
 
+    GraphPrintLog("Aborted");
+
     {
         CriticalBlock cb(crit);
 
@@ -2297,7 +2305,7 @@ void CGraphTempHandler::deregisterFile(const char *name, bool kept)
         try
         {
             if (!removeTemp(name))
-                LOG(MCwarning, unknownJob, "Failed to delete tmp file : %s (not found)", name);
+                LOG(MCwarning, thorJob, "Failed to delete tmp file : %s (not found)", name);
         }
         catch (IException *e) { StringBuffer s("Failed to delete tmp file : "); FLLOG(MCwarning, thorJob, e, s.append(name).str()); }
     }
@@ -2847,7 +2855,8 @@ void CJobBase::endJob()
         callThreadTerminationHooks(true); // must call any installed thread termination functions, before unloading plugins
         ::Release(pluginMap);
 
-        traceMemUsage();
+        if (!REJECTLOG(MCthorDetailedDebugInfo))
+            traceMemUsage();
 
         if (numChannels > 1) // if only 1 - then channel allocator is same as sharedAllocator, leaks will be reported by the single channel
             checkAndReportLeaks(sharedAllocator->queryRowManager());
@@ -3016,7 +3025,7 @@ mptag_t CJobChannel::deserializeMPTag(MemoryBuffer &mb)
     deserializeMPtag(mb, tag);
     if (TAG_NULL != tag)
     {
-        PROGLOG("deserializeMPTag: tag = %d", (int)tag);
+        LOG(MCthorDetailedDebugInfo, thorJob, "deserializeMPTag: tag = %d", (int)tag);
         jobComm->flush(tag);
     }
     return tag;
@@ -3052,8 +3061,11 @@ void CJobChannel::clean()
     cleaned = true;
     wait();
 
-    queryRowManager()->reportMemoryUsage(false);
-    PROGLOG("CJobBase resetting memory manager");
+    if (!REJECTLOG(MCthorDetailedDebugInfo))
+    {
+        queryRowManager()->reportMemoryUsage(false);
+        LOG(MCthorDetailedDebugInfo, thorJob, "CJobBase resetting memory manager");
+    }
 
     if (graphExecutor)
     {
@@ -3199,8 +3211,11 @@ CActivityBase::~CActivityBase()
 
 void CActivityBase::abort()
 {
-    if (!abortSoon) ActPrintLog("Abort condition set");
-    abortSoon = true;
+    if (!abortSoon)
+    {
+        ::ActPrintLog(this, thorDetailedLogLevel, "Abort condition set");
+        abortSoon = true;
+    }
 }
 
 void CActivityBase::kill()

thorlcr/graph/thgraphmaster.cpp

@@ -286,7 +286,7 @@ void CSlaveMessageHandler::threadmain()
                 }
                 case smt_getPhysicalName:
                 {
-                    LOG(MCdebugProgress, unknownJob, "getPhysicalName called from node %d", sender-1);
+                    LOG(MCdebugProgress, thorJob, "getPhysicalName called from node %d", sender-1);
                     StringAttr logicalName;
                     unsigned partNo;
                     bool create;
@@ -305,7 +305,7 @@ void CSlaveMessageHandler::threadmain()
                 }
                 case smt_getFileOffset:
                 {
-                    LOG(MCdebugProgress, unknownJob, "getFileOffset called from node %d", sender-1);
+                    LOG(MCdebugProgress, thorJob, "getFileOffset called from node %d", sender-1);
                     StringAttr logicalName;
                     unsigned partNo;
                     msg.read(logicalName);
@@ -318,7 +318,7 @@ void CSlaveMessageHandler::threadmain()
                 }
                 case smt_actMsg:
                 {
-                    LOG(MCdebugProgress, unknownJob, "smt_actMsg called from node %d", sender-1);
+                    LOG(MCdebugProgress, thorJob, "smt_actMsg called from node %d", sender-1);
                     graph_id gid;
                     msg.read(gid);
                     activity_id id;
@@ -336,7 +336,7 @@ void CSlaveMessageHandler::threadmain()
                 {
                     unsigned slave;
                     msg.read(slave);
-                    LOG(MCdebugProgress, unknownJob, "smt_getresult called from slave %d", slave);
+                    LOG(MCdebugProgress, thorJob, "smt_getresult called from slave %d", slave);
                     graph_id gid;
                     msg.read(gid);
                     activity_id ownerId;
@@ -819,7 +819,7 @@ class CThorCodeContextMaster : public CThorCodeContextBase
         return getWorkUnitResult(workunit, name, sequence);
     }
     #define PROTECTED_GETRESULT(STEPNAME, SEQUENCE, KIND, KINDTEXT, ACTION) \
-        LOG(MCdebugProgress, unknownJob, "getResult%s(%s,%d)", KIND, STEPNAME?STEPNAME:"", SEQUENCE); \
+        LOG(MCdebugProgress, thorJob, "getResult%s(%s,%d)", KIND, STEPNAME?STEPNAME:"", SEQUENCE); \
         Owned<IConstWUResult> r = getResultForGet(STEPNAME, SEQUENCE); \
         try \
         { \
@@ -1084,7 +1084,7 @@ public:
     {
         try
         {
-            LOG(MCdebugProgress, unknownJob, "getExternalResultRaw %s", stepname);
+            LOG(MCdebugProgress, thorJob, "getExternalResultRaw %s", stepname);
 
             Owned<IConstWUResult> r = getExternalResult(wuid, stepname, sequence);
             return r->getResultHash();
@@ -1126,7 +1126,7 @@ public:
         tgt = NULL;
         try
         {
-            LOG(MCdebugProgress, unknownJob, "getExternalResultRaw %s", stepname);
+            LOG(MCdebugProgress, thorJob, "getExternalResultRaw %s", stepname);
 
             Variable2IDataVal result(&tlen, &tgt);
             Owned<IConstWUResult> r = getExternalResult(wuid, stepname, sequence);
@@ -1395,7 +1395,7 @@ mptag_t CJobMaster::allocateMPTag()
 {
     mptag_t tag = allocateClusterMPTag();
     queryJobChannel(0).queryJobComm().flush(tag);
-    PROGLOG("allocateMPTag: tag = %d", (int)tag);
+    LOG(MCthorDetailedDebugInfo, thorJob, "allocateMPTag: tag = %d", (int)tag);
     return tag;
 }
 
@@ -1404,7 +1404,7 @@ void CJobMaster::freeMPTag(mptag_t tag)
     if (TAG_NULL != tag)
     {
         freeClusterMPTag(tag);
-        PROGLOG("freeMPTag: tag = %d", (int)tag);
+        LOG(MCthorDetailedDebugInfo, thorJob, "freeMPTag: tag = %d", (int)tag);
         queryJobChannel(0).queryJobComm().flush(tag);
     }
 }

thorlcr/graph/thgraphslave.cpp

@@ -137,7 +137,7 @@ CSlaveActivity::~CSlaveActivity()
     inputs.kill();
     outputs.kill();
     if (data) delete [] data;
-    ActPrintLog("DESTROYED");
+    ::ActPrintLog(this, thorDetailedLogLevel, "DESTROYED");
 }
 
 bool CSlaveActivity::hasLookAhead(unsigned index) const
@@ -1308,7 +1308,7 @@ void CSlaveGraph::serializeDone(MemoryBuffer &mb)
 void CSlaveGraph::getDone(MemoryBuffer &doneInfoMb)
 {
     if (!started) return;
-    GraphPrintLog("Entering getDone");
+    ::GraphPrintLog(this, thorDetailedLogLevel, "Entering getDone");
     if (!queryOwner() || isGlobal())
     {
         try
@@ -1323,12 +1323,12 @@ void CSlaveGraph::getDone(MemoryBuffer &doneInfoMb)
         }
         catch (IException *)
         {
-            GraphPrintLog("Leaving getDone");
+            ::GraphPrintLog(this, thorDetailedLogLevel, "Leaving getDone");
             getDoneSem.signal();
             throw;
         }
     }
-    GraphPrintLog("Leaving getDone");
+    ::GraphPrintLog(this, thorDetailedLogLevel, "Leaving getDone");
     getDoneSem.signal();
 }
 
@@ -1744,11 +1744,14 @@ void CJobSlave::reportGraphEnd(graph_id gid)
 {
     if (nodesLoaded) // wouldn't mean much if parallel jobs running
         PROGLOG("Graph[%" GIDPF "u] - JHTree node stats:\ncacheAdds=%d\ncacheHits=%d\nnodesLoaded=%d\nblobCacheHits=%d\nblobCacheAdds=%d\nleafCacheHits=%d\nleafCacheAdds=%d\nnodeCacheHits=%d\nnodeCacheAdds=%d\n", gid, cacheAdds.load(), cacheHits.load(), nodesLoaded.load(), blobCacheHits.load(), blobCacheAdds.load(), leafCacheHits.load(), leafCacheAdds.load(), nodeCacheHits.load(), nodeCacheAdds.load());
-    JSocketStatistics stats;
-    getSocketStatistics(stats);
-    StringBuffer s;
-    getSocketStatisticsString(stats,s);
-    PROGLOG("Graph[%" GIDPF "u] - Socket statistics : %s\n", gid, s.str());
+    if (!REJECTLOG(MCthorDetailedDebugInfo))
+    {        
+        JSocketStatistics stats;
+        getSocketStatistics(stats);
+        StringBuffer s;
+        getSocketStatisticsString(stats,s);
+        LOG(MCthorDetailedDebugInfo, thorJob, "Graph[%" GIDPF "u] - Socket statistics : %s\n", gid, s.str());
+    }
     resetSocketStatistics();
 }
 
@@ -1787,7 +1790,7 @@ mptag_t CJobSlave::deserializeMPTag(MemoryBuffer &mb)
     deserializeMPtag(mb, tag);
     if (TAG_NULL != tag)
     {
-        PROGLOG("CJobSlave::deserializeMPTag: tag = %d", (int)tag);
+        LOG(MCthorDetailedDebugInfo, thorJob, "CJobSlave::deserializeMPTag: tag = %d", (int)tag);
         for (unsigned c=0; c<queryJobChannels(); c++)
             queryJobChannel(c).queryJobComm().flush(tag);
     }

thorlcr/graph/thgraphslave.hpp

@@ -63,9 +63,7 @@ public:
 
     inline void dataLinkStart()
     {
-#ifdef _TESTING
         owner.ActPrintLog("ITDL starting for output %d", outputId);
-#endif
         if (hasStarted())
         {
             if (!hasStopped())
@@ -81,9 +79,7 @@ public:
     {
         if (hasStarted())
             count = (count & THORDATALINK_COUNT_MASK) | THORDATALINK_STOPPED;
-#ifdef _TESTING
         owner.ActPrintLog("ITDL output %d stopped, count was %" RCPF "d", outputId, getDataLinkCount());
-#endif
     }
     inline void dataLinkIncrement() { dataLinkIncrement(1); }
     inline void dataLinkIncrement(rowcount_t v)

thorlcr/master/mawatchdog.cpp

@@ -55,7 +55,7 @@ public:
             markdead = false;
             StringBuffer epstr;
             ep.getUrlStr(epstr);
-            LOG(MCdebugProgress, unknownJob, "Watchdog : Marking Machine as Up! [%s]", epstr.str());
+            LOG(MCdebugProgress, thorJob, "Watchdog : Marking Machine as Up! [%s]", epstr.str());
         }
     }   
 };
@@ -129,12 +129,12 @@ void CMasterWatchdogBase::stop()
         synchronized block(mutex);
         if (stopped)
             return;
-        LOG(MCdebugProgress, unknownJob, "Stopping watchdog");
+        LOG(MCdebugProgress, thorJob, "Stopping watchdog");
         stopped = true;
     }
     stopReading();
     threaded.join();
-    LOG(MCdebugProgress, unknownJob, "Stopped watchdog");
+    LOG(MCdebugProgress, thorJob, "Stopped watchdog");
 }
 
 void CMasterWatchdogBase::checkMachineStatus()
@@ -152,7 +152,7 @@ void CMasterWatchdogBase::checkMachineStatus()
             else
             {
                 mstate->markdead = true;
-                LOG(MCdebugProgress, unknownJob, "Watchdog : Marking Machine as Down! [%s]", epstr.str());
+                LOG(MCdebugProgress, thorJob, "Watchdog : Marking Machine as Down! [%s]", epstr.str());
                 //removeSlave(mstate->ep); // more TBD
             }
         }
@@ -190,7 +190,7 @@ unsigned CMasterWatchdogBase::readPacket(HeartBeatPacketHeader &hb, MemoryBuffer
 
 void CMasterWatchdogBase::threadmain()
 {
-    LOG(MCdebugProgress, unknownJob, "Started watchdog");
+    LOG(MCdebugProgress, thorJob, "Started watchdog");
     unsigned lastbeat=msTick();
     unsigned lastcheck=lastbeat;
 
@@ -235,7 +235,7 @@ void CMasterWatchdogBase::threadmain()
                 {
                     StringBuffer epstr;
                     hb.sender.getUrlStr(epstr);
-                    LOG(MCdebugProgress, unknownJob, "Watchdog : Unknown Machine! [%s]", epstr.str()); //TBD
+                    LOG(MCdebugProgress, thorJob, "Watchdog : Unknown Machine! [%s]", epstr.str()); //TBD
                 }
             }
             unsigned now=msTick();

thorlcr/master/thdemonserver.cpp

@@ -75,7 +75,7 @@ private:
         catch (IException *e)
         {
             StringBuffer s;
-            LOG(MCwarning, unknownJob, "Failed to update progress information: %s", e->errorMessage(s).str());
+            LOG(MCwarning, thorJob, "Failed to update progress information: %s", e->errorMessage(s).str());
             e->Release();
         }
     }
@@ -155,7 +155,7 @@ private:
             catch (IException *E)
             {
                 StringBuffer s;
-                LOG(MCwarning, unknownJob, "Failed to update progress information: %s", E->errorMessage(s).str());
+                LOG(MCwarning, thorJob, "Failed to update progress information: %s", E->errorMessage(s).str());
                 E->Release();
             }
         }
@@ -187,7 +187,7 @@ private:
         catch (IException *e)
         {
             StringBuffer s;
-            LOG(MCwarning, unknownJob, "Failed to update progress information: %s", e->errorMessage(s).str());
+            LOG(MCwarning, thorJob, "Failed to update progress information: %s", e->errorMessage(s).str());
             e->Release();
         }
     }
@@ -212,7 +212,7 @@ public:
         if (0 == activeGraphs.ordinality())
         {
             StringBuffer urlStr;
-            LOG(MCdebugProgress, unknownJob, "heartbeat packet received with no active graphs");
+            LOG(MCdebugProgress, thorJob, "heartbeat packet received with no active graphs");
             return;
         }
         size32_t compressedProgressSz = progressMb.remaining();
@@ -230,12 +230,12 @@ public:
                 ForEachItemIn(g, activeGraphs) if (activeGraphs.item(g).queryGraphId() == graphId) graph = (CMasterGraph *)&activeGraphs.item(g);
                 if (!graph)
                 {
-                    LOG(MCdebugProgress, unknownJob, "heartbeat received from unknown graph %" GIDPF "d", graphId);
+                    LOG(MCdebugProgress, thorJob, "heartbeat received from unknown graph %" GIDPF "d", graphId);
                     break;
                 }
                 if (!graph->deserializeStats(slave, uncompressedMb))
                 {
-                    LOG(MCdebugProgress, unknownJob, "heartbeat error in graph %" GIDPF "d", graphId);
+                    LOG(MCdebugProgress, thorJob, "heartbeat error in graph %" GIDPF "d", graphId);
                     break;
                 }
             }

thorlcr/master/thgraphmanager.cpp

@@ -711,6 +711,12 @@ void CJobManager::run()
         {
             factory.setown(getWorkUnitFactory());
             workunit.setown(factory->openWorkUnit(wuid));
+
+            unsigned maxLogDetail = workunit->getDebugValueInt("maxlogdetail", DefaultDetail); 
+            ILogMsgFilter *existingLogHandler = queryLogMsgManager()->queryMonitorFilter(logHandler);
+            dbgassertex(existingLogHandler);
+            verifyex(queryLogMsgManager()->changeMonitorFilterOwn(logHandler, getCategoryLogMsgFilter(existingLogHandler->queryAudienceMask(), existingLogHandler->queryClassMask(), maxLogDetail)));
+
             allDone = execute(workunit, wuid, graphName, agentep);
             daliLock.clear();
             reply(workunit, wuid, NULL, agentep, allDone);

thorlcr/master/thmastermain.cpp

@@ -653,6 +653,10 @@ int main( int argc, const char *argv[]  )
             lf->setCreateAliasFile(false);
             logHandler = lf->beginLogging();
             createUNCFilename(lf->queryLogFileSpec(), logUrl, false);
+#ifndef _DEBUG
+            // keep duplicate logging output to stderr to aide debugging
+            queryLogMsgManager()->removeMonitor(queryStderrLogMsgHandler());
+#endif
 
             LOG(MCdebugProgress, thorJob, "Opened log file %s", logUrl.str());
         }

thorlcr/msort/tsortl.cpp

@@ -122,11 +122,11 @@ public:
         if (dsz.eos()) {
             inbuf.clear();
 #ifdef _FULL_TRACE
-            PROGLOG("CSocketRowStream.nextRow recv (%d,%x)",id,(unsigned)(memsize_t)socket.get());
+            LOG(MCthorDetailedDebugInfo, thorJob, "CSocketRowStream.nextRow recv (%d,%x)",id,(unsigned)(memsize_t)socket.get());
 #endif
             size32_t sz = socket->receive_block_size();
 #ifdef _FULL_TRACE
-            PROGLOG("CSocketRowStream.nextRow(%d,%x,%d)",id,(unsigned)(memsize_t)socket.get(),sz);
+            LOG(MCthorDetailedDebugInfo, thorJob, "CSocketRowStream.nextRow(%d,%x,%d)",id,(unsigned)(memsize_t)socket.get(),sz);
 #endif
             if (sz==0) {
                 // eof so terminate (no need to confirm)
@@ -138,7 +138,7 @@ public:
             socket->receive_block(buf,sz);
             assertex(!dsz.eos());
 #ifdef _FULL_TRACE
-            PROGLOG("CSocketRowStream.nextRow got (%d,%x,%d)",id,(unsigned)(memsize_t)socket.get(),sz);
+            LOG(MCthorDetailedDebugInfo, thorJob, "CSocketRowStream.nextRow got (%d,%x,%d)",id,(unsigned)(memsize_t)socket.get(),sz);
 #endif
         }
         RtlDynamicRowBuilder rowBuilder(allocator);
@@ -154,12 +154,12 @@ public:
             stopped = true;
             try {
 #ifdef _FULL_TRACE
-                PROGLOG("CSocketRowStream.stop(%x)",(unsigned)(memsize_t)socket.get());
+                LOG(MCthorDetailedDebugInfo, thorJob, "CSocketRowStream.stop(%x)",(unsigned)(memsize_t)socket.get());
 #endif
                 bool eof = true;
                 socket->write(&eof,sizeof(eof)); // confirm stop
 #ifdef _FULL_TRACE
-                PROGLOG("CSocketRowStream.stopped(%x)",(unsigned)(memsize_t)socket.get());
+                LOG(MCthorDetailedDebugInfo, thorJob, "CSocketRowStream.stopped(%x)",(unsigned)(memsize_t)socket.get());
 #endif
             }
             catch (IException *e) {
@@ -206,7 +206,7 @@ public:
             preallocated = bufsize+initSize;
 
 #ifdef _FULL_TRACE
-        PROGLOG("CSocketRowWriter(%d,%x) preallocated = %d",id,(unsigned)(memsize_t)socket.get(),preallocated);
+        LOG(MCthorDetailedDebugInfo, thorJob, "CSocketRowWriter(%d,%x) preallocated = %d",id,(unsigned)(memsize_t)socket.get(),preallocated);
 #endif
     }
 
@@ -227,11 +227,11 @@ public:
             flush();
         try {
 #ifdef _FULL_TRACE
-            PROGLOG("CSocketRowWriter.stop(%x)",(unsigned)(memsize_t)socket.get());
+            LOG(MCthorDetailedDebugInfo, thorJob, "CSocketRowWriter.stop(%x)",(unsigned)(memsize_t)socket.get());
 #endif
             socket->send_block(NULL,0);
 #ifdef _FULL_TRACE
-            PROGLOG("CSocketRowWriter.stopped(%x)",(unsigned)(memsize_t)socket.get());
+            LOG(MCthorDetailedDebugInfo, thorJob, "CSocketRowWriter.stopped(%x)",(unsigned)(memsize_t)socket.get());
 #endif
         }
         catch (IJSOCK_Exception *e) { // already gone!
@@ -260,11 +260,11 @@ public:
     {
         size32_t l = outbuf.length();
 #ifdef _FULL_TRACE
-        PROGLOG("CSocketRowWriter.flush(%d,%x,%d)",id,(unsigned)(memsize_t)socket.get(),l);
+        LOG(MCthorDetailedDebugInfo, thorJob, "CSocketRowWriter.flush(%d,%x,%d)",id,(unsigned)(memsize_t)socket.get(),l);
 #endif
         if (l) {
             if (!socket->send_block(outbuf.bufferBase(),l)) {
-                PROGLOG("CSocketRowWriter remote stop");
+                LOG(MCthorDetailedDebugInfo, thorJob, "CSocketRowWriter remote stop");
                 stopped = true;
             }
             pos += l;
@@ -293,7 +293,7 @@ IRowStream *ConnectMergeRead(unsigned id, IThorRowInterfaces *rowif,SocketEndpoi
 #ifdef _FULL_TRACE
     StringBuffer s;
     nodeaddr.getUrlStr(s);
-    PROGLOG("ConnectMergeRead(%d,%s,%x,%" RCPF "d,%" RCPF "u)",id,s.str(),(unsigned)(memsize_t)socket.get(),startrec,numrecs);
+    LOG(MCthorDetailedDebugInfo, thorJob, "ConnectMergeRead(%d,%s,%x,%" RCPF "d,%" RCPF "u)",id,s.str(),(unsigned)(memsize_t)socket.get(),startrec,numrecs);
 #endif
     hdr.winrev();
     socket->write(&hdr,sizeof(hdr));
@@ -333,7 +333,7 @@ ISocketRowWriter *ConnectMergeWrite(IThorRowInterfaces *rowif,ISocket *socket,si
 #ifdef _FULL_TRACE
     char name[100];
     int port = socket->peer_name(name,sizeof(name));
-    PROGLOG("ConnectMergeWrite(%d,%s:%d,%x,%" RCPF "d,%" RCPF "u)",hdr.id,name,port,(unsigned)(memsize_t)socket,startrec,numrecs);
+    LOG(MCthorDetailedDebugInfo, thorJob, "ConnectMergeWrite(%d,%s:%d,%x,%" RCPF "d,%" RCPF "u)",hdr.id,name,port,(unsigned)(memsize_t)socket,startrec,numrecs);
 #endif
     return new CSocketRowWriter(hdr.id,rowif,socket,bufsize);
 }

thorlcr/msort/tsortm.cpp

@@ -47,10 +47,6 @@
 #include "tsortm.hpp"
 
 
-#ifdef _TESTING
-#define _TRACE
-#endif
-
 #ifdef _DEBUG
 //#define _MEMTRACE
 #endif
@@ -124,7 +120,7 @@ public:
         try {
 
             StringBuffer epstr;
-            ActPrintLog(activity, "Connect to %s:%d",endpoint.getIpText(epstr).str(),(unsigned)mpport);
+            ActPrintLog(activity, thorDetailedLogLevel, "Connect to %s:%d",endpoint.getIpText(epstr).str(),(unsigned)mpport);
             SocketEndpoint ep = endpoint;
             ep.port = mpport;
             Owned<INode> node = createINode(ep);
@@ -262,7 +258,7 @@ public:
 
     void ConnectSlaves()
     {
-        ActPrintLog(activity, "CSortMaster::ConnectSlaves");
+        ActPrintLog(activity, thorDetailedLogLevel, "CSortMaster::ConnectSlaves");
         class casyncfor: public CAsyncFor
         {
         public:
@@ -317,30 +313,6 @@ public:
         partitioninfo = NULL;
     }
 
-
-    void TraceLeft(NODESTATE state)
-    {
-        unsigned left=slaves.ordinality();
-        ForEachItemIn(i,slaves) {
-            CSortNode &slave = slaves.item(i);          
-            if (slave.state !=state)
-                left--;
-        }
-        ActPrintLog(activity, " Left=%d",left);
-        if (left<5) {
-            StringBuffer s;
-            ForEachItemIn(j,slaves) {
-                CSortNode &slave = slaves.item(j);          
-                if (slave.state==state) {
-                    s.append(' ');
-                    slave.endpoint.getIpText(s);
-                }
-            }
-            ActPrintLog(activity, "%s",s.str());
-        }
-    }
-
-
     void SortSetup(IThorRowInterfaces *_rowif,ICompare *_icompare,ISortKeySerializer *_keyserializer,bool cosort,bool needconnect,const char *_cosortfilenames,IThorRowInterfaces *_auxrowif)
     {
         ActPrintLog(activity, "Sort setup cosort=%s, needconnect=%s %s",cosort?"true":"false",needconnect?"true":"false",_keyserializer?"has key serializer":"");
@@ -456,25 +428,6 @@ public:
         ActPrintLog(activity, "Sort Done");
     }
 
-
-    bool GetNode(unsigned num,SocketEndpoint &endpoint)
-    {
-        ActPrintLog(activity, "GetNode %u",num);
-        if (num<slaves.ordinality()) {
-            CSortNode &slave = slaves.item(num);            
-            endpoint = slave.endpoint;
-            endpoint.port++;    // for socket server
-            return true;
-        }
-        return false;
-    }
-
-    unsigned NumActiveNodes()
-    {
-        return slaves.ordinality();
-    }
-
-
     unsigned __int64 CalcMinMax(OwnedConstThorRow &min, OwnedConstThorRow &max)
     {
         // initialize min/max keys
@@ -517,21 +470,22 @@ public:
             estrecsize = ers/ersn;
         else
             estrecsize = 100;
-#ifdef _TRACE
-        if (min)
-            traceKey(keyIf->queryRowSerializer(),"Min",min);
-        if (max)
-            traceKey(keyIf->queryRowSerializer(),"Max",max);
-        if (min&&max)
+        if (!REJECTLOG(MCthorDetailedDebugInfo))
         {
-            int cmp=icompare->docompare(min,max);
-            if (cmp==0) 
-                ActPrintLog(activity, "Min == Max : All keys equal!");
-            else if (cmp>0)
-                ActPrintLog(activity, "ERROR: Min > Max!");
+            if (min)
+                traceKey(keyIf->queryRowSerializer(),"Min",min);
+            if (max)
+                traceKey(keyIf->queryRowSerializer(),"Max",max);
+            if (min&&max)
+            {
+                int cmp=icompare->docompare(min,max);
+                if (cmp==0) 
+                    ActPrintLog(activity, "Min == Max : All keys equal!");
+                else if (cmp>0)
+                    ActPrintLog(activity, "ERROR: Min > Max!");
+            }
         }
         ActPrintLog(activity, "Tot = %" I64F "d", tot);
-#endif
         return tot;
     }
 
@@ -688,7 +642,6 @@ public:
             }
         } afor3(slaves, splitMap, numnodes, numsplits);
         afor3.For(numnodes, 20, true);
-#ifdef _TRACE
 #ifdef TRACE_PARTITION
         for (i=0;i<numnodes;i++) {
             StringBuffer str;
@@ -699,7 +652,6 @@ public:
             DBGLOG("%s",str.str());
         }
 #endif
-#endif
         return splitMap.getClear();
     }
 
@@ -742,10 +694,8 @@ public:
             MemoryBuffer mbmd;
             for (;;)
             {
-#ifdef _TRACE
                 iter++;
-                ActPrintLog(activity, "Split: %d",iter);
-#endif
+                ActPrintLog(activity, thorDetailedLogLevel, "Split: %d",iter);
                 emin.serializeCompress(mbmn.clear());
                 emax.serializeCompress(mbmx.clear());
                 class casyncfor: public CAsyncFor
@@ -976,7 +926,6 @@ public:
     rowcount_t *UsePartitionInfo(PartitionInfo &pi, bool uppercmp)
     {
         unsigned i;
-#ifdef _TRACE
 #ifdef TRACE_PARTITION
         ActPrintLog(activity, "UsePartitionInfo %s",uppercmp?"upper":"");
         for (i=0;i<pi.splitkeys.ordinality();i++)
@@ -986,7 +935,6 @@ public:
             traceKey(pi.prowif->queryRowSerializer(), s.str(), pi.splitkeys.query(i));
         }
 #endif
-#endif
         // first find split points
         unsigned numnodes = pi.numnodes;
         unsigned numsplits = numnodes-1;
@@ -1025,7 +973,6 @@ public:
             *mapp = slave.numrecs; // final entry is number in node
             mapp++;
         }
-#ifdef _TRACE
 #ifdef TRACE_PARTITION
         ActPrintLog(activity, "UsePartitionInfo result");
         rowcount_t *p = splitMap;
@@ -1041,7 +988,6 @@ public:
             ActPrintLog(activity, "%s",s.str());
         }
 #endif
-#endif
         return splitMap.getClear();
     }
 
@@ -1119,7 +1065,7 @@ public:
         if (skewError<0.000000001)
             skewError = 1.0/(double) n;
         double cSkew = ((double)n*(double)max/(double)total - 1.0) / ((double)n-1.0);
-        ActPrintLog(activity, "Skew check: Threshold %" I64F "d/%" I64F "d  Skew: %f/[warning=%f, error=%f]",
+        ActPrintLog(activity, thorDetailedLogLevel, "Skew check: Threshold %" I64F "d/%" I64F "d  Skew: %f/[warning=%f, error=%f]",
                   (unsigned __int64)max*(unsigned __int64)estrecsize,threshold,cSkew,skewWarning,skewError);
         if ((unsigned __int64)max*(unsigned __int64)estrecsize>threshold)
         {
@@ -1351,12 +1297,15 @@ public:
                         i--;
                     }
                 }
-                for (i=0;i<numnodes;i++)
+                if (!REJECTLOG(MCthorDetailedDebugInfo))
                 {
-                    CSortNode &slave = slaves.item(i);
-                    char url[100];
-                    slave.endpoint.getUrlStr(url,sizeof(url));
-                    ActPrintLog(activity, "Split point %d: %" RCPF "d rows on %s", i, tot[i], url);
+                    for (i=0;i<numnodes;i++)
+                    {
+                        CSortNode &slave = slaves.item(i);
+                        char url[100];
+                        slave.endpoint.getUrlStr(url,sizeof(url));
+                        ActPrintLog(activity, thorDetailedLogLevel, "Split point %d: %" RCPF "d rows on %s", i, tot[i], url);
+                    }
                 }
                 Owned<IThorException> e = CheckSkewed(threshold,skewWarning,skewError,numnodes,total,max);
                 if (e)

thorlcr/msort/tsortmp.cpp

@@ -290,7 +290,7 @@ bool SortSlaveMP::marshall(ISortSlaveMP &slave, ICommunicator* comm, mptag_t tag
     rank_t sender;
     comm->recv(mb,0,tag,&sender);       // NB only recv from master
     if (mb.length()==0) {
-        PROGLOG("Stopping SortSlaveMP::marshall");
+        LOG(MCthorDetailedDebugInfo, thorJob, "Stopping SortSlaveMP::marshall");
         return false;
     }
     byte fn;
@@ -301,7 +301,7 @@ bool SortSlaveMP::marshall(ISortSlaveMP &slave, ICommunicator* comm, mptag_t tag
     mbout.append(okout);
 #ifdef FULLTRACE
     StringBuffer tmp1;
-    PROGLOG(">SortSlaveMP::marshall(%d) got %d from %s tag %d replytag %d",(int)fn, mb.length(), mb.getSender().getUrlStr(tmp1).str(),tag,mb.getReplyTag());
+    LOG(MCthorDetailedDebugInfo, thorJob, ">SortSlaveMP::marshall(%d) got %d from %s tag %d replytag %d",(int)fn, mb.length(), mb.getSender().getUrlStr(tmp1).str(),tag,mb.getReplyTag());
 #endif
     bool replydone = false;
     Owned<IException> err;
@@ -531,7 +531,7 @@ bool SortSlaveMP::marshall(ISortSlaveMP &slave, ICommunicator* comm, mptag_t tag
     if (!replydone) {
 #ifdef FULLTRACE
         StringBuffer tmp1;
-        PROGLOG("<SortSlaveMP::marshall(%d) send %d to %s tag %d",(int)fn, mbout.length(), mbout.getSender().getUrlStr(tmp1).str(),mbout.getReplyTag());
+        LOG(MCthorDetailedDebugInfo, thorJob, "<SortSlaveMP::marshall(%d) send %d to %s tag %d",(int)fn, mbout.length(), mbout.getSender().getUrlStr(tmp1).str(),mbout.getReplyTag());
 #endif
         comm->reply(mbout);
     }

thorlcr/msort/tsorts.cpp

@@ -34,8 +34,6 @@
 #include "thbuf.hpp"
 #include "thgraph.hpp"
 
-#define _TRACE
-
 #ifdef _DEBUG
 //#define TRACE_UNIQUE
 //#define _FULL_TRACE
@@ -628,11 +626,11 @@ class CThorSorter : public CSimpleInterface, implements IThorSorter, implements
     {
         try
         {
-            ActPrintLog(activity, "Creating SortSlaveServer on tag %d MP",mpTagRPC);    
+            ActPrintLog(activity, thorDetailedLogLevel, "Creating SortSlaveServer on tag %d MP",mpTagRPC);    
             while(SortSlaveMP::marshall(*this,clusterComm,mpTagRPC)&&!stopping)
                ;
             stopping = true;
-            ActPrintLog(activity, "Exiting SortSlaveServer on tag %d",mpTagRPC);    
+            ActPrintLog(activity, thorDetailedLogLevel, "Exiting SortSlaveServer on tag %d",mpTagRPC);    
         }
         catch (IJSOCK_Exception *e)
         {
@@ -656,27 +654,24 @@ class CThorSorter : public CSimpleInterface, implements IThorSorter, implements
         }
         stopping = true;
     }
-#ifdef _TRACE
     void TraceRow(const char *s, const void *k)
     {
-        traceKey(rowif->queryRowSerializer(), s, k);
+        if (!REJECTLOG(MCthorDetailedDebugInfo))
+            traceKey(rowif->queryRowSerializer(), s, k);
     }
     void TraceKey(const char *s, const void *k)
     {
-        traceKey(keyIf->queryRowSerializer(), s, k);
+        if (!REJECTLOG(MCthorDetailedDebugInfo))
+            traceKey(keyIf->queryRowSerializer(), s, k);
     }
-#else
-#define TraceRow(msg, row)
-#define TraceKey(msg, key)
-#endif
 
     void stop()
     {
         if (!stopping) {
             stopping = true;
-            ActPrintLog(activity,"stopmarshall");
+            ActPrintLog(activity, thorDetailedLogLevel, "stopmarshall");
             SortSlaveMP::stopmarshall(clusterComm,mpTagRPC);
-            ActPrintLog(activity,"stopmarshalldone");
+            ActPrintLog(activity, thorDetailedLogLevel, "stopmarshalldone");
         }
         if (exc)
             throw exc.getClear();
@@ -799,11 +794,11 @@ public:
         try
         {
             stop();
-            ActPrintLog(activity, "Joining Sort Slave Server");
+            ActPrintLog(activity, thorDetailedLogLevel, "Joining Sort Slave Server");
             verifyex(threaded.join(10*60*1000));
             myendpoint.set(NULL,0);
             rowArray.kill();
-            ActPrintLog(activity, "~CThorSorter");
+            ActPrintLog(activity, thorDetailedLogLevel, "~CThorSorter");
         }
         catch (IException *e)
         {
@@ -815,7 +810,7 @@ public:
 // ISortSlaveMP
     virtual bool Connect(unsigned _partno, unsigned _numnodes)
     {
-        ActPrintLog(activity, "Connected to slave %d of %d",_partno,_numnodes);
+        ActPrintLog(activity, thorDetailedLogLevel, "Connected to slave %d of %d",_partno,_numnodes);
         numnodes = _numnodes;
         partno = _partno;
         transferserver.clear();
@@ -861,9 +856,7 @@ public:
         rowToKeySerializer->serialize(msz, (const byte *)kp);
 
         avrecsize = (size32_t)(grandtotalsize/grandtotal);
-#ifdef _TRACE
-        ActPrintLog(activity, "Ave Rec Size = %u", avrecsize);
-#endif
+        ActPrintLog(activity, thorDetailedLogLevel, "Ave Rec Size = %u", avrecsize);
         keybufsize = mb.length();
         keybuf = mb.detach();
         return 2;
@@ -1101,14 +1094,14 @@ public:
         startmergesem.signal();
         ActPrintLog(activity, "StartMiniSort output started");
         traceWait("finishedmergesem(2)",finishedmergesem);
-        DBGLOG("StartMiniSort output done");
+        LOG(MCthorDetailedDebugInfo, thorJob, "StartMiniSort output done");
         merger.clear();
         intercept.clear();
-        DBGLOG("StartMiniSort exit");
+        LOG(MCthorDetailedDebugInfo, thorJob, "StartMiniSort exit");
     }
     virtual void Close()
     {
-        ActPrintLog(activity, "Close");
+        ActPrintLog(activity, thorDetailedLogLevel, "Close");
         try {
             if (transferserver)
                 transferserver->subjoin(); // need to have finished merge threads 
@@ -1124,19 +1117,19 @@ public:
     }
     virtual void CloseWait()
     {
-        ActPrintLog(activity, "Close finished");
+        ActPrintLog(activity, thorDetailedLogLevel, "Close finished");
         if (closeexc.get())
             throw closeexc.getClear();
         rowArray.kill();
     }
     virtual void Disconnect()
     {
-        ActPrintLog(activity, "Disconnecting from slave %d of %d",partno,numnodes);
+        ActPrintLog(activity, thorDetailedLogLevel, "Disconnecting from slave %d of %d",partno,numnodes);
         if (transferserver) {
             transferserver->stop();
             transferserver.clear();
         }
-        ActPrintLog(activity, "Disconnected from slave %d of %d",partno,numnodes);
+        ActPrintLog(activity, thorDetailedLogLevel, "Disconnected from slave %d of %d",partno,numnodes);
     }
 
 // ISortSlaveBase
@@ -1177,14 +1170,14 @@ public:
             Owned<IRowLinkCounter> linkcounter = new CThorRowLinkCounter;
             merger.setown(createRowStreamMerger(readers.ordinality(), readers.getArray(), rowCompare, false, linkcounter));
         }
-        ActPrintLog(activity, "Global Merger Created: %d streams", readers.ordinality());
+        ActPrintLog(activity, thorDetailedLogLevel, "Global Merger Created: %d streams", readers.ordinality());
         startmergesem.signal();
         traceWait("finishedmergesem",finishedmergesem);
-        ActPrintLog(activity, "Global Merged completed");
+        ActPrintLog(activity, thorDetailedLogLevel, "Global Merged completed");
         merger.clear();
         readers.kill(); // NB: need to be cleared before intercept, which clears up files
         intercept.clear();
-        ActPrintLog(activity, "Global Merge exit");
+        ActPrintLog(activity, thorDetailedLogLevel, "Global Merge exit");
     }
 
 // IThorSorter

thorlcr/msort/tsorts1.cpp

@@ -27,8 +27,6 @@
 #include "thmem.hpp"
 
 
-#define _TRACE
-
 #ifdef _DEBUG
 //#define TRACE_UNIQUE
 #define _FULL_TRACE
@@ -60,17 +58,13 @@ public:
     CMergeReadStream(IThorRowInterfaces *rowif, unsigned streamno,SocketEndpoint &targetep, rowcount_t startrec, rowcount_t numrecs)
     {
         endpoint = targetep;
-#ifdef _TRACE
         char url[100];
         targetep.getUrlStr(url,sizeof(url));
-        DBGLOG("SORT Merge READ: Stream(%u) %s, pos=%" RCPF "d len=%" RCPF "u",streamno,url,startrec,numrecs);
-#endif
+        LOG(MCthorDetailedDebugInfo, thorJob, "SORT Merge READ: Stream(%u) %s, pos=%" RCPF "d len=%" RCPF "u",streamno,url,startrec,numrecs);
         SocketEndpoint mergeep = targetep;
         mergeep.port+=SOCKETSERVERINC; 
         stream = ConnectMergeRead(streamno,rowif,mergeep,startrec,numrecs);
-#ifdef _TRACE
-        DBGLOG("SORT Merge READ: Stream(%u) connected to %s",streamno,url);
-#endif
+        LOG(MCthorDetailedDebugInfo, thorJob, "SORT Merge READ: Stream(%u) connected to %s",streamno,url);
     }
     virtual ~CMergeReadStream()
     {
@@ -92,7 +86,7 @@ public:
 #ifdef _FULL_TRACE
             char url[100];
             endpoint.getUrlStr(url,sizeof(url));
-            DBGLOG("SORT Merge READ: EOS for %s",url);
+            LOG(MCthorDetailedDebugInfo, thorJob, "SORT Merge READ: EOS for %s",url);
 #endif
             eos();
         }
@@ -105,7 +99,7 @@ public:
 #ifdef _FULL_TRACE
             char url[100];
             endpoint.getUrlStr(url,sizeof(url));
-            DBGLOG("SORT Merge READ: stop for %s",url);
+            LOG(MCthorDetailedDebugInfo, thorJob, "SORT Merge READ: stop for %s",url);
 #endif
             stream->stop();
             eos();
@@ -147,24 +141,22 @@ public:
     ~CSortMerge()
     {
 #ifdef _FULL_TRACE
-        PROGLOG("~CSortMerge in");
+        LOG(MCthorDetailedDebugInfo, thorJob, "~CSortMerge in");
 #endif
         if (started)
             closedown();
 #ifdef _FULL_TRACE
-        PROGLOG("~CSortMerge out");
+        LOG(MCthorDetailedDebugInfo, thorJob, "~CSortMerge out");
 #endif
     }
     void init()
     {
         CriticalBlock block(crit);
         started = true;
-#ifdef _TRACE
         char name[64];
         int port = socket->peer_name(name,sizeof(name));
         url.append(name).append(':').append(port);
-        DBGLOG("SORT Merge WRITE: start %s, pos=%" RCPF "d, len=%" RCPF "d",url.str(),poscount,numrecs);
-#endif
+        LOG(MCthorDetailedDebugInfo, thorJob, "SORT Merge WRITE: start %s, pos=%" RCPF "d, len=%" RCPF "d",url.str(),poscount,numrecs);
         rowcount_t pos=poscount;
         try
         {
@@ -214,7 +206,7 @@ public:
         char peer[16];
         if (socket) {
             socket->peer_name(peer,sizeof(peer)-1);
-            PROGLOG("waitdone %s",peer);
+            LOG(MCthorDetailedDebugInfo, thorJob, "waitdone %s",peer);
         }
         else
             peer[0] = 0;
@@ -223,7 +215,7 @@ public:
         if (exception)
             throw exception.getClear();
         if (peer[0])
-            PROGLOG("waitdone exit");
+            LOG(MCthorDetailedDebugInfo, thorJob, "waitdone exit");
     }
     bool notifySelected(ISocket *sock,unsigned selected)
     {
@@ -232,11 +224,11 @@ public:
                 if (closing) {
                     closing = false;
 #ifdef _FULL_TRACE
-                    PROGLOG("notifySelected calling closedown");
+                    LOG(MCthorDetailedDebugInfo, thorJob, "notifySelected calling closedown");
 #endif
                     closedown();
 #ifdef _FULL_TRACE
-                    PROGLOG("notifySelected called closedown");
+                    LOG(MCthorDetailedDebugInfo, thorJob, "notifySelected called closedown");
 #endif
                     done = true;
                     donesem.signal();
@@ -478,7 +470,7 @@ public:
                 respos += vMAPL(j,i)-vMAPL(j,i-1);      // note we are adding up all of the lower as we want start
 
         rowcount_t totalrows = resnum;
-        DBGLOG("Output start = %" RCPF "d, num = %" RCPF "u",respos,resnum);
+        LOG(MCthorDetailedDebugInfo, thorJob, "Output start = %" RCPF "d, num = %" RCPF "u",respos,resnum);
 
         IArrayOf<IRowStream> readers;
         IException *exc = NULL;
@@ -493,7 +485,7 @@ public:
                 {
                     if (i==partno)
                     {
-                        DBGLOG("SORT Merge READ: Stream(%u) local, pos=%" RCPF "u len=%" RCPF "u",i,sstart,snum);
+                        LOG(MCthorDetailedDebugInfo, thorJob, "SORT Merge READ: Stream(%u) local, pos=%" RCPF "u len=%" RCPF "u",i,sstart,snum);
                         readers.append(*slave.createMergeInputStream(sstart,snum));
                     }
                     else
@@ -539,7 +531,7 @@ void CSortMerge::closedown()
 {
     CriticalBlock block(crit);
 #ifdef _FULL_TRACE
-    DBGLOG("SORT Merge: closing %s",url.str());
+    LOG(MCthorDetailedDebugInfo, thorJob, "SORT Merge: closing %s",url.str());
 #endif
     if (!socket)
         return;
@@ -574,9 +566,7 @@ void CSortMerge::closedown()
         throw;
     }
     started = false;
-#ifdef _TRACE
-    DBGLOG("SORT Merge: finished %s, %d rows merged",url.str(),ndone);
-#endif
+    LOG(MCthorDetailedDebugInfo, thorJob, "SORT Merge: finished %s, %d rows merged",url.str(),ndone);
 }
 
 IMergeTransferServer *createMergeTransferServer(ISortSlaveBase *parent)

thorlcr/slave/slave.cpp

@@ -58,13 +58,13 @@ void ProcessSlaveActivity::beforeDispose()
 {
     // Note - we can't throw from the destructor, so do this in beforeDispose instead
     // If the exception is thrown then we are liable to leak the object, but we are dying anyway...
-    ActPrintLog("destroying ProcessSlaveActivity");
-    ActPrintLog("ProcessSlaveActivity : joining process thread");
+    ::ActPrintLog(this, thorDetailedLogLevel, "destroying ProcessSlaveActivity");
+    ::ActPrintLog(this, thorDetailedLogLevel, "ProcessSlaveActivity : joining process thread");
     // NB: The activity thread should have already stopped,
     //     if it is still alive at job shutdown and cannot be joined then the thread is in an unknown state.
     if (!threaded.join(FATAL_ACTJOIN_TIMEOUT))
         throw MakeThorFatal(NULL, TE_FailedToAbortSlaves, "Activity %" ACTPF "d failed to stop", container.queryId());
-    ActPrintLog("AFTER ProcessSlaveActivity : joining process thread");
+    ::ActPrintLog(this, thorDetailedLogLevel, "AFTER ProcessSlaveActivity : joining process thread");
 }
 
 void ProcessSlaveActivity::startProcess(bool async)

thorlcr/slave/slavmain.cpp

@@ -1554,11 +1554,11 @@ public:
     }
     virtual void reset() override
     {
-        DBGLOG("KJService reset()");
+        LOG(MCthorDetailedDebugInfo, thorJob, "KJService reset()");
         processorPool->stopAll(true);
         processorPool->joinAll(false);
         clearAll();
-        DBGLOG("KJService reset() done");
+        LOG(MCthorDetailedDebugInfo, thorJob, "KJService reset() done");
     }
     virtual void start() override
     {
@@ -1569,7 +1569,7 @@ public:
     {
         if (aborted)
             return;
-        PROGLOG("KJService stop()");
+        LOG(MCthorDetailedDebugInfo, thorJob, "KJService stop()");
         queryNodeComm().cancel(RANK_ALL, keyLookupMpTag);
         processorPool->stopAll(true);
         processorPool->joinAll(true);
@@ -1693,7 +1693,7 @@ public:
     {
         queryNodeComm().cancel(0, masterSlaveMpTag);
     }
-    virtual void slaveMain()
+    void slaveMain(ILogMsgHandler *logHandler)
     {
         rank_t slaveProc = queryNodeGroup().rank()-1;
         unsigned totSlaveProcs = queryNodeClusterWidth();
@@ -1859,6 +1859,11 @@ public:
                         StringBuffer user;
                         workUnitInfo->getProp("user", user);
 
+                        unsigned maxLogDetail = workUnitInfo->getPropInt("Debug/maxLogDetail", DefaultDetail);
+                        ILogMsgFilter *existingLogHandler = queryLogMsgManager()->queryMonitorFilter(logHandler);
+                        dbgassertex(existingLogHandler);
+                        verifyex(queryLogMsgManager()->changeMonitorFilterOwn(logHandler, getCategoryLogMsgFilter(existingLogHandler->queryAudienceMask(), existingLogHandler->queryClassMask(), maxLogDetail)));
+
                         PROGLOG("Started wuid=%s, user=%s, graph=%s\n", wuid.get(), user.str(), graphName.get());
                         PROGLOG("Using query: %s", soPath.str());
 
@@ -2324,7 +2329,7 @@ public:
     virtual IKJService &queryKeyedJoinService() override { return *kjService.get(); }
 };
 
-void slaveMain(bool &jobListenerStopped)
+void slaveMain(bool &jobListenerStopped, ILogMsgHandler *logHandler)
 {
     unsigned masterMemMB = globals->getPropInt("@masterTotalMem");
     HardwareInfo hdwInfo;
@@ -2363,7 +2368,7 @@ void slaveMain(bool &jobListenerStopped)
 
 #endif
 
-    jobListener.slaveMain();
+    jobListener.slaveMain(logHandler);
 }
 
 void abortSlave()

thorlcr/slave/slavmain.hpp

@@ -18,8 +18,9 @@
 #ifndef SLAVMAIN_HPP
 #define SLAVMAIN_HPP
 
+interface ILogMsgHandler;
 void abortSlave();
-void slaveMain(bool &jobListenerStopped);
+void slaveMain(bool &jobListenerStopped, ILogMsgHandler *logHandler);
 void enableThorSlaveAsDaliClient();
 void disableThorSlaveAsDaliClient();
 

thorlcr/slave/slwatchdog.cpp

@@ -117,7 +117,7 @@ public:
         CriticalBlock b(crit);
         activeGraphs.append(*LINK(&graph));
         StringBuffer str("Watchdog: Start Job ");
-        LOG(MCdebugProgress, thorJob, "%s", str.append(graph.queryGraphId()).str());
+        LOG(MCthorDetailedDebugInfo, thorJob, "%s", str.append(graph.queryGraphId()).str());
     }
     void stopGraph(CGraphBase &graph, MemoryBuffer *mb)
     {
@@ -125,7 +125,7 @@ public:
         if (NotFound != activeGraphs.find(graph))
         {
             StringBuffer str("Watchdog: Stop Job ");
-            LOG(MCdebugProgress, thorJob, "%s", str.append(graph.queryGraphId()).str());
+            LOG(MCthorDetailedDebugInfo, thorJob, "%s", str.append(graph.queryGraphId()).str());
             if (mb)
             {
                 DelayedSizeMarker sizeMark(*mb);
@@ -167,7 +167,7 @@ public:
 // IThreaded
     virtual void threadmain() override
     {
-        LOG(MCdebugProgress, thorJob, "Watchdog: thread running");
+        LOG(MCthorDetailedDebugInfo, thorJob, "Watchdog: thread running");
         gatherAndSend(); // send initial data
         assertex(HEARTBEAT_INTERVAL>=8);
         unsigned count = HEARTBEAT_INTERVAL+getRandom()%8-4;

thorlcr/slave/thslavemain.cpp

@@ -274,7 +274,7 @@ public:
 #endif
 
 
-void startSlaveLog()
+ILogMsgHandler *startSlaveLog()
 {
 #ifndef _CONTAINERIZED
     StringBuffer fileName("thorslave");
@@ -283,7 +283,11 @@ void startSlaveLog()
     lf->setPostfix(slaveNumStr.append(mySlaveNum).str());
     lf->setCreateAliasFile(false);
     lf->setName(fileName.str());//override default filename
-    lf->beginLogging();
+    ILogMsgHandler *logHandler = lf->beginLogging();
+#ifndef _DEBUG 
+    // keep duplicate logging output to stderr to aide debugging
+    queryLogMsgManager()->removeMonitor(queryStderrLogMsgHandler());
+#endif
 
     LOG(MCdebugProgress, thorJob, "Opened log file %s", lf->queryLogFileSpec());
 #else
@@ -291,6 +295,7 @@ void startSlaveLog()
 #endif
     setupContainerizedStorageLocations();
     LOG(MCdebugProgress, thorJob, "Build %s", BUILD_TAG);
+    return logHandler;
 }
 
 void setSlaveAffinity(unsigned processOnNode)
@@ -314,7 +319,6 @@ void setSlaveAffinity(unsigned processOnNode)
         bindMemoryToLocalNodes();
 }
 
-
 int main( int argc, const char *argv[]  )
 {
     // If using systemd, we will be using daemon code, writing own pid
@@ -374,12 +378,9 @@ int main( int argc, const char *argv[]  )
         /* NB: in cloud/non-local storage mode, slave number is not known until after registration with the master
         * For the time being log file names are based on their slave number, so can only start when known.
         */
-        bool loggingStarted = false;
+        ILogMsgHandler *slaveLogHandler = nullptr;
         if (NotFound != mySlaveNum)
-        {
-            startSlaveLog();
-            loggingStarted = true;
-        }
+            slaveLogHandler = startSlaveLog();
 
         // In container world, SLAVE= will not be used
         const char *slave = globals->queryProp("@slave");
@@ -416,8 +417,8 @@ int main( int argc, const char *argv[]  )
 
         if (RegisterSelf(masterEp))
         {
-            if (!loggingStarted)
-                startSlaveLog();
+            if (!slaveLogHandler)
+                slaveLogHandler = startSlaveLog();
 
             if (globals->getPropBool("Debug/@slaveDaliClient"))
                 enableThorSlaveAsDaliClient();
@@ -526,7 +527,7 @@ int main( int argc, const char *argv[]  )
                 startPerformanceMonitor(pinterval, PerfMonStandard, nullptr);
 
             installDefaultFileHooks(globals);
-            slaveMain(jobListenerStopped);
+            slaveMain(jobListenerStopped, slaveLogHandler);
         }
 
         LOG(MCdebugProgress, thorJob, "ThorSlave terminated OK");

thorlcr/thorutil/thbuf.cpp

@@ -801,7 +801,7 @@ class CSharedWriteAheadBase : public CSimpleInterface, implements ISharedSmartBu
         {
             unsigned whichChunk = queryCOutput(output).currentChunkNum - lowestChunk;
 #ifdef TRACE_WRITEAHEAD
-            PROGLOG("output=%d, chunk=%d (whichChunk=%d)", output, currentChunkNum, whichChunk);
+            LOG(MCthorDetailedDebugInfo, thorJob, "output=%d, chunk=%d (whichChunk=%d)", output, currentChunkNum, whichChunk);
 #endif
             rowSet.setown(readRows(output, whichChunk));
             assertex(rowSet);
@@ -1519,7 +1519,7 @@ public:
             Owned<Chunk> chunk = savedChunks.dequeue();
             if (!chunk) break;
         }
-        PROGLOG("CSharedWriteAheadDisk: highOffset=%" I64F "d", highOffset);
+        LOG(MCthorDetailedDebugInfo, thorJob, "CSharedWriteAheadDisk: highOffset=%" I64F "d", highOffset);
     }
     virtual void reset()
     {

thorlcr/thorutil/thmem.cpp

@@ -1646,11 +1646,10 @@ protected:
         StringBuffer tempPrefix, tempName;
         if (iCompare)
         {
-            ActPrintLog(&activity, "%sSorting %" RIPF "d rows", tracingPrefix.str(), spillableRows.numCommitted());
             CCycleTimer timer;
             spillableRows.sort(*iCompare, maxCores); // sorts committed rows
             sortCycles += timer.elapsedCycles();
-            ActPrintLog(&activity, "%sSort took: %f", tracingPrefix.str(), ((float)timer.elapsedMs())/1000);
+            ActPrintLog(&activity, "%sSorting %" RIPF "u rows took: %f", tracingPrefix.str(), spillableRows.numCommitted(), ((float)timer.elapsedMs())/1000);
             tempPrefix.append("srt");
         }
         tempPrefix.appendf("spill_%d", activity.queryId());
@@ -1864,7 +1863,7 @@ public:
              * memory usage.
              */
             size32_t compBlkSz = activity.getOptUInt(THOROPT_SORT_COMPBLKSZ, DEFAULT_SORT_COMPBLKSZ);
-            activity.ActPrintLog("%sSpilling will use compressed block size = %u", tracingPrefix.str(), compBlkSz);
+            ActPrintLog(&activity, thorDetailedLogLevel, "%sSpilling will use compressed block size = %u", tracingPrefix.str(), compBlkSz);
             spillableRows.setCompBlockSize(compBlkSz);
         }
     }

thorlcr/thorutil/thormisc.cpp

@@ -172,6 +172,14 @@ void ActPrintLog(const CActivityBase *activity, const char *format, ...)
     va_end(args);
 }
 
+void ActPrintLog(const CActivityBase *activity, unsigned traceLevel, const char *format, ...)
+{
+    va_list args;
+    va_start(args, format);
+    ActPrintLogArgs(&activity->queryContainer(), thorlog_null, MCdebugInfo(traceLevel), format, args);
+    va_end(args);
+}
+
 void ActPrintLog(const CActivityBase *activity, IException *e, const char *format, ...)
 {
     va_list args;
@@ -223,6 +231,7 @@ void GraphPrintLog(CGraphBase *graph, IException *e, const char *format, ...)
     va_end(args);
 }
 
+
 class DECL_EXCEPTION CThorException : public CSimpleInterface, implements IThorException
 {
 protected:
@@ -731,7 +740,7 @@ void ClearDir(const char *dir)
 void ClearTempDirs()
 {
     TempNameHandler.clearDirs(true);
-    PROGLOG("temp directory cleared");
+    LOG(MCthorDetailedDebugInfo, thorJob, "temp directory cleared");
 }
 
 

thorlcr/thorutil/thormisc.hpp

@@ -107,7 +107,9 @@
 #define INITIAL_SELFJOIN_MATCH_WARNING_LEVEL 20000  // max of row matches before selfjoin emits warning
 
 #define THOR_SEM_RETRY_TIMEOUT 2
-#define THOR_TRACE_LEVEL 5
+
+// Logging
+extern graph_decl const LogMsgJobInfo thorJob;
 
 enum ThorExceptionAction { tea_null, tea_warning, tea_abort, tea_shutdown };
 
@@ -261,7 +263,7 @@ public:
     void stop() { running = false; todo.signal(); }
     void inform(IException *e)
     {
-        LOG(MCdebugProgress, unknownJob, "INFORM [%s]", description.get());
+        LOG(MCdebugProgress, thorJob, "INFORM [%s]", description.get());
         CriticalBlock block(crit);
         if (exception.get())
             e->Release();
@@ -276,7 +278,7 @@ public:
         CriticalBlock block(crit);
         IException *e = exception.getClear();
         if (e)
-            LOG(MCdebugProgress, unknownJob, "CLEARING TIMEOUT [%s]", description.get());
+            LOG(MCdebugProgress, thorJob, "CLEARING TIMEOUT [%s]", description.get());
         todo.signal();
         return e;
     }
@@ -398,6 +400,7 @@ extern graph_decl void ActPrintLogEx(const CGraphElementBase *container, const A
 extern graph_decl void ActPrintLogArgs(const CGraphElementBase *container, const ActLogEnum flags, const LogMsgCategory &logCat, const char *format, va_list args) __attribute__((format(printf,4,0)));
 extern graph_decl void ActPrintLogArgs(const CGraphElementBase *container, IException *e, const ActLogEnum flags, const LogMsgCategory &logCat, const char *format, va_list args) __attribute__((format(printf,5,0)));
 extern graph_decl void ActPrintLog(const CActivityBase *activity, const char *format, ...) __attribute__((format(printf, 2, 3)));
+extern graph_decl void ActPrintLog(const CActivityBase *activity, unsigned traceLevel, const char *format, ...) __attribute__((format(printf, 3, 4)));
 extern graph_decl void ActPrintLog(const CActivityBase *activity, IException *e, const char *format, ...) __attribute__((format(printf, 3, 4)));
 extern graph_decl void ActPrintLog(const CActivityBase *activity, IException *e);
 
@@ -409,6 +412,14 @@ inline void ActPrintLog(const CGraphElementBase *container, const char *format,
     ActPrintLogArgs(container, thorlog_ecl, MCdebugProgress, format, args);
     va_end(args);
 }
+inline void ActPrintLog(const CGraphElementBase *container, unsigned traceLevel, const char *format, ...) __attribute__((format(printf, 3, 4)));
+inline void ActPrintLog(const CGraphElementBase *container, unsigned traceLevel, const char *format, ...)
+{
+    va_list args;
+    va_start(args, format);
+    ActPrintLogArgs(container, thorlog_ecl, MCdebugProgress(traceLevel), format, args);
+    va_end(args);
+}
 inline void ActPrintLogEx(const CGraphElementBase *container, IException *e, const ActLogEnum flags, const LogMsgCategory &logCat, const char *format, ...) __attribute__((format(printf, 5, 6)));
 inline void ActPrintLogEx(const CGraphElementBase *container, IException *e, const ActLogEnum flags, const LogMsgCategory &logCat, const char *format, ...)
 {
@@ -458,6 +469,16 @@ inline void GraphPrintLog(CGraphBase *graph, const char *format, ...)
     GraphPrintLogArgs(graph, thorlog_null, MCdebugProgress, format, args);
     va_end(args);
 }
+
+inline void GraphPrintLog(CGraphBase *graph, unsigned traceLevel, const char *format, ...) __attribute__((format(printf, 3, 4)));
+inline void GraphPrintLog(CGraphBase *graph, unsigned traceLevel, const char *format, ...)
+{
+    va_list args;
+    va_start(args, format);
+    GraphPrintLogArgs(graph, thorlog_null, MCdebugInfo(traceLevel), format, args);
+    va_end(args);
+}
+
 extern graph_decl IThorException *MakeActivityException(CActivityBase *activity, int code, const char *_format, ...) __attribute__((format(printf, 3, 4)));
 extern graph_decl IThorException *MakeActivityException(CActivityBase *activity, IException *e, const char *xtra, ...) __attribute__((format(printf, 3, 4)));
 extern graph_decl IThorException *MakeActivityException(CActivityBase *activity, IException *e);
@@ -494,8 +515,6 @@ extern graph_decl Owned<IPropertyTree> globals;
 extern graph_decl mptag_t masterSlaveMpTag;
 extern graph_decl mptag_t kjServiceMpTag;
 enum SlaveMsgTypes:unsigned { smt_errorMsg=1, smt_initGraphReq, smt_initActDataReq, smt_dataReq, smt_getPhysicalName, smt_getFileOffset, smt_actMsg, smt_getresult };
-// Logging
-extern graph_decl const LogMsgJobInfo thorJob;
 
 extern graph_decl StringBuffer &getCompoundQueryName(StringBuffer &compoundName, const char *queryName, unsigned version);
 
@@ -563,5 +582,9 @@ inline void readUnderlyingType(MemoryBuffer &mb, T &v)
     mb.read(reinterpret_cast<typename std::underlying_type<T>::type &> (v));
 }
 
+constexpr unsigned thorDetailedLogLevel = 200;
+constexpr LogMsgCategory MCthorDetailedDebugInfo(MCdebugInfo(thorDetailedLogLevel));
+
+
 #endif