wip

Author: jakesmith

initfiles/etc/DIR_NAME/environment.xml.in

@@ -922,6 +922,7 @@
                daliServers="mydali"
                description="Thor process"
                fileCacheLimit="1800"
+               globalMemorySize="200"
                heapRetainMemory="false"
                heapUseHugePages="false"
                heapUseTransparentHugePages="true"
@@ -935,7 +936,7 @@
                replicateAsync="false"
                replicateOutputs="false"
                slaveport="20100"
-               slavesPerNode="1"
+               slavesPerNode="2"
                watchdogEnabled="true"
                watchdogProgressEnabled="true">
    <SwapNode AutoSwapNode="false"/>

testing/regress/ecl/graphspill3.ecl

@@ -0,0 +1,20 @@
+#option('pickBestEngine', false);
+
+numRecs := 10;
+rec := RECORD
+ unsigned8 id;
+ unsigned8 other;
+ string984 rest := 'rest';
+ string24  to1k := '1k';
+END;
+
+ds := DATASET(numRecs, TRANSFORM(rec, SELF.id := HASH(COUNTER), SELF.other := COUNTER), DISTRIBUTED);
+
+sd1 := SORT(NOFOLD(ds), id);
+sd2 := SORT(NOFOLD(ds), other);
+sd2p := PROJECT(sd2, TRANSFORM({sd2.id}, SELF := LEFT));
+
+PARALLEL(
+ OUTPUT(DEDUP(sd1, other, HASH));
+ OUTPUT(DEDUP(sd2p, id, HASH));
+);

thorlcr/graph/thgraph.cpp

@@ -2375,33 +2375,19 @@ void CGraphTempHandler::clearTemps()
     tmpFiles.kill();
 }
 
-void CGraphTempHandler::getUsageStats(graph_id gid, offset_t & graphSpillSize)
+offset_t CGraphTempHandler::getUsageStats()
 {
     CriticalBlock b(crit);
     Owned<IFileUsageIterator> iter = getIterator();
-    graphSpillSize = 0;
+    offset_t activeSpillSize = 0;
     ForEach(*iter)
     {
         CFileUsageEntry &entry = iter->query();
-        if (entry.queryGraphId() == gid)
-            graphSpillSize += entry.getSize();
+        activeSpillSize += entry.getSize();
     }
+    return activeSpillSize;
 }
 
-void CGraphTempHandler::serializeUsageStats(MemoryBuffer &mb, graph_id gid)
-{
-    offset_t graphSpillSize;
-    getUsageStats(gid, graphSpillSize);
-    mb.append(graphSpillSize);
-}
-
-void CGraphTempHandler::setUsageStats(CRuntimeStatisticCollection &rsc, graph_id gid)
-{
-    offset_t graphSpillSize;
-    getUsageStats(gid, graphSpillSize);
-    rsc.setStatistic(StSizeActiveSpillFile, graphSpillSize);
-    rsc.setStatistic(StPeakSizeNodeSpillFile, graphSpillSize);  // StatsMergeMax should track the peak
-}
 /////
 
 class CGraphExecutor;

thorlcr/graph/thgraph.hpp

@@ -221,12 +221,7 @@ interface IGraphTempHandler : extends IInterface
     virtual void deregisterFile(const char *name, bool kept=false) = 0;
     virtual void clearTemps() = 0;
     virtual IFileUsageIterator *getIterator() = 0;
-    virtual void serializeUsageStats(MemoryBuffer &mb, graph_id gid) = 0;
-    static void serializeNullUsageStats(MemoryBuffer &mb)
-    {
-        mb.append((offset_t)0);
-    }
-    virtual void setUsageStats(CRuntimeStatisticCollection &rsc, graph_id gid) = 0;
+    virtual offset_t getUsageStats() = 0;
 };
 
 class CGraphDependency : public CInterface
@@ -560,7 +555,6 @@ class graph_decl CGraphTempHandler : implements IGraphTempHandler, public CInter
     mutable CriticalSection crit;
     bool errorOnMissing;
 
-    void getUsageStats(graph_id gid, offset_t & graphSpillSize);
 public:
     IMPLEMENT_IINTERFACE;
 
@@ -592,8 +586,7 @@ class graph_decl CGraphTempHandler : implements IGraphTempHandler, public CInter
         };
         return new CIterator(tmpFiles);
     }
-    virtual void serializeUsageStats(MemoryBuffer &mb, graph_id gid) override;
-    virtual void setUsageStats(CRuntimeStatisticCollection &rsc, graph_id gid) override;
+    virtual offset_t getUsageStats() override;
 };
 
 class graph_decl CGraphStub : public CInterface, implements IThorChildGraph

thorlcr/graph/thgraphmaster.cpp

@@ -2722,11 +2722,6 @@ void CMasterGraph::handleSlaveDone(unsigned node, MemoryBuffer &mb)
         sdMb.setBuffer(len, (void *)d);
         act->slaveDone(node, sdMb);
     }
-    offset_t nodeGraphSpill;
-    mb.read(nodeGraphSpill);
-    totalActiveSpillSize += nodeGraphSpill;
-    if (nodeGraphSpill>peakNodeSpillFile)
-        peakNodeSpillFile = nodeGraphSpill;
 }
 
 void CMasterGraph::getFinalProgress()

thorlcr/graph/thgraphslave.cpp

@@ -1259,14 +1259,17 @@ bool CSlaveGraph::serializeStats(MemoryBuffer &mb)
 
     CRuntimeStatisticCollection stats(graphStatistics);
     stats.setStatistic(StNumExecutions, numExecuted);
+    offset_t graphSpillSize = 0;
     if (!owner)
-        queryJob().queryTempHandler()->setUsageStats(stats, gid);
+        graphSpillSize = queryJob().queryTempHandler()->getUsageStats();
     else
     {
         IGraphTempHandler *tempHandler = queryTempHandler(false);
         if (tempHandler)
-            tempHandler->setUsageStats(stats, gid);
+            graphSpillSize = tempHandler->getUsageStats();
     }
+
+    stats.mergeStatistic(StPeakSizeNodeSpillFile, graphSpillSize);
     stats.serialize(mb);
 
     unsigned cPos = mb.length();
@@ -1340,17 +1343,6 @@ void CSlaveGraph::serializeDone(MemoryBuffer &mb)
         }
     }
     mb.writeDirect(cPos, sizeof(count), &count);
-
-    if (!owner)
-        queryJob().queryTempHandler()->serializeUsageStats(mb, gid);
-    else
-    {
-        IGraphTempHandler *tempHandler = queryTempHandler(false);
-        if (tempHandler)
-            tempHandler->serializeUsageStats(mb, gid);
-        else
-            IGraphTempHandler::serializeNullUsageStats(mb);
-    }
 }
 
 void CSlaveGraph::getDone(MemoryBuffer &doneInfoMb)