HPCC-17563 Race condition updating factory stats

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

system/jlib/jmutex.hpp

@@ -892,7 +892,7 @@ public:
  * A template function for implementing a singleton object.  Using the same example as above would require:
 
     static std::atomic<void *> sobj;
-    static CCriticalSection slock;
+    static CriticalSection slock;
     void *get()
     {
         return querySingleton(sobj, slock, []{ return createSObj; });

system/jlib/jstats.cpp

@@ -1649,16 +1649,26 @@ void CRuntimeStatistic::merge(unsigned __int64 otherValue, StatsMergeAction merg
 CRuntimeStatisticCollection::~CRuntimeStatisticCollection()
 {
     delete [] values;
-    delete nested;
+    delete queryNested();
+}
+
+CNestedRuntimeStatisticMap * CRuntimeStatisticCollection::queryNested() const
+{
+    return nested.load(std::memory_order_relaxed);
+}
+
+CNestedRuntimeStatisticMap * CRuntimeStatisticCollection::createNested() const
+{
+    return new CNestedRuntimeStatisticMap;
 }
 
 CNestedRuntimeStatisticMap & CRuntimeStatisticCollection::ensureNested()
 {
-    if (!nested)
-        nested = new CNestedRuntimeStatisticMap;
-    return *nested;
+    return *querySingleton(nested, nestlock, [this]{ return this->createNested(); });
 }
 
+CriticalSection CRuntimeStatisticCollection::nestlock;
+
 void CRuntimeStatisticCollection::merge(const CRuntimeStatisticCollection & other)
 {
     ForEachItemIn(i, other)
@@ -1757,8 +1767,9 @@ void CRuntimeStatisticCollection::recordStatistics(IStatisticGatherer & target)
         }
     }
     reportIgnoredStats();
-    if (nested)
-        nested->recordStatistics(target);
+    CNestedRuntimeStatisticMap *qn = queryNested();
+    if (qn)
+        qn->recordStatistics(target);
 }
 
 void CRuntimeStatisticCollection::reportIgnoredStats() const
@@ -1779,8 +1790,9 @@ StringBuffer & CRuntimeStatisticCollection::toXML(StringBuffer &str) const
             str.appendf("<%s>%" I64F "d</%s>", name, value, name);
         }
     }
-    if (nested)
-        nested->toXML(str);
+    CNestedRuntimeStatisticMap *qn = queryNested();
+    if (qn)
+        qn->toXML(str);
     return str;
 }
 
@@ -1797,8 +1809,9 @@ StringBuffer & CRuntimeStatisticCollection::toStr(StringBuffer &str) const
             formatStatistic(str, value, kind);
         }
     }
-    if (nested)
-        nested->toStr(str);
+    CNestedRuntimeStatisticMap *qn = queryNested();
+    if (qn)
+        qn->toStr(str);
     return str;
 }
 
@@ -1875,10 +1888,11 @@ bool CRuntimeStatisticCollection::serialize(MemoryBuffer& out) const
     }
 
     bool nonEmpty = (numValid != 0);
-    out.append(nested != nullptr);
-    if (nested)
+    CNestedRuntimeStatisticMap *qn = queryNested();
+    out.append(qn != nullptr);
+    if (qn)
     {
-        if (nested->serialize(out))
+        if (qn->serialize(out))
             nonEmpty = true;
     }
     return nonEmpty;
@@ -1924,11 +1938,9 @@ CRuntimeSummaryStatisticCollection::~CRuntimeSummaryStatisticCollection()
     delete[] derived;
 }
 
-CNestedRuntimeStatisticMap & CRuntimeSummaryStatisticCollection::ensureNested()
+CNestedRuntimeStatisticMap * CRuntimeSummaryStatisticCollection::createNested() const
 {
-    if (!nested)
-        nested = new CNestedSummaryRuntimeStatisticMap;
-    return *nested;
+    return new CNestedSummaryRuntimeStatisticMap;
 }
 
 void CRuntimeSummaryStatisticCollection::mergeStatistic(StatisticKind kind, unsigned __int64 value)
@@ -2067,18 +2079,34 @@ void CNestedRuntimeStatisticCollection::updateDelta(CNestedRuntimeStatisticColle
 
 CNestedRuntimeStatisticCollection & CNestedRuntimeStatisticMap::addNested(const StatsScopeId & scope, const StatisticsMapping & mapping)
 {
-    ForEachItemIn(i, map)
+    unsigned mapSize;
+    unsigned entry;
     {
-        CNestedRuntimeStatisticCollection & cur = map.item(i);
-        if (cur.matches(scope))
-            return cur;
+        ReadLockBlock b(lock);
+        mapSize = map.length();
+        for (entry = 0; entry < mapSize; entry++)
+        {
+            CNestedRuntimeStatisticCollection & cur = map.item(entry);
+            if (cur.matches(scope))
+                return cur;
+        }
+    }
+    {
+        WriteLockBlock b(lock);
+        // Check no-one added anything between the read and write locks
+        mapSize = map.length();
+        for (; entry < mapSize; entry++)
+        {
+            CNestedRuntimeStatisticCollection & cur = map.item(entry);
+            if (cur.matches(scope))
+                return cur;
+        }
+        CNestedRuntimeStatisticCollection * stats = new CNestedRuntimeStatisticCollection(scope, createStats(mapping));
+        map.append(*stats);
+        return *stats;
     }
-    CNestedRuntimeStatisticCollection * stats = new CNestedRuntimeStatisticCollection(scope, createStats(mapping));
-    map.append(*stats);
-    return *stats;
 }
 
-
 void CNestedRuntimeStatisticMap::deserialize(MemoryBuffer& in)
 {
     unsigned numItems;
@@ -2111,6 +2139,7 @@ void CNestedRuntimeStatisticMap::deserializeMerge(MemoryBuffer& in)
 
 void CNestedRuntimeStatisticMap::merge(const CNestedRuntimeStatisticMap & other)
 {
+    ReadLockBlock b(other.lock);
     ForEachItemIn(i, other.map)
     {
         CNestedRuntimeStatisticCollection & cur = other.map.item(i);
@@ -2121,6 +2150,7 @@ void CNestedRuntimeStatisticMap::merge(const CNestedRuntimeStatisticMap & other)
 
 void CNestedRuntimeStatisticMap::updateDelta(CNestedRuntimeStatisticMap & target, const CNestedRuntimeStatisticMap & source)
 {
+    ReadLockBlock b(source.lock);
     ForEachItemIn(i, source.map)
     {
         CNestedRuntimeStatisticCollection & curSource = source.map.item(i);
@@ -2132,6 +2162,7 @@ void CNestedRuntimeStatisticMap::updateDelta(CNestedRuntimeStatisticMap & target
 
 bool CNestedRuntimeStatisticMap::serialize(MemoryBuffer& out) const
 {
+    ReadLockBlock b(lock);
     out.appendPacked(map.ordinality());
     bool nonEmpty = false;
     ForEachItemIn(i, map)
@@ -2144,12 +2175,14 @@ bool CNestedRuntimeStatisticMap::serialize(MemoryBuffer& out) const
 
 void CNestedRuntimeStatisticMap::recordStatistics(IStatisticGatherer & target) const
 {
+    ReadLockBlock b(lock);
     ForEachItemIn(i, map)
         map.item(i).recordStatistics(target);
 }
 
 StringBuffer & CNestedRuntimeStatisticMap::toStr(StringBuffer &str) const
 {
+    ReadLockBlock b(lock);
     ForEachItemIn(i, map)
         map.item(i).toStr(str);
     return str;
@@ -2157,6 +2190,7 @@ StringBuffer & CNestedRuntimeStatisticMap::toStr(StringBuffer &str) const
 
 StringBuffer & CNestedRuntimeStatisticMap::toXML(StringBuffer &str) const
 {
+    ReadLockBlock b(lock);
     ForEachItemIn(i, map)
         map.item(i).toXML(str);
     return str;

system/jlib/jstats.h

@@ -20,6 +20,7 @@
 #define JSTATS_H
 
 #include "jlib.hpp"
+#include "jmutex.hpp"
 
 #include "jstatcodes.h"
 
@@ -402,7 +403,9 @@ public:
 
 
 protected:
-    virtual CNestedRuntimeStatisticMap & ensureNested();
+    virtual CNestedRuntimeStatisticMap *createNested() const;
+    CNestedRuntimeStatisticMap & ensureNested();
+    CNestedRuntimeStatisticMap *queryNested() const;
     void reportIgnoredStats() const;
     void mergeStatistic(StatisticKind kind, unsigned __int64 value, StatsMergeAction mergeAction)
     {
@@ -413,7 +416,8 @@ protected:
 protected:
     const StatisticsMapping & mapping;
     CRuntimeStatistic * values;
-    CNestedRuntimeStatisticMap * nested = nullptr;
+    std::atomic<CNestedRuntimeStatisticMap *> nested {nullptr};
+    static CriticalSection nestlock;
 };
 
 //NB: Serialize and deserialize are not currently implemented.
@@ -444,7 +448,7 @@ protected:
     };
 
 protected:
-    virtual CNestedRuntimeStatisticMap & ensureNested() override;
+    virtual CNestedRuntimeStatisticMap *createNested() const override;
 
 protected:
     DerivedStats * derived;
@@ -501,6 +505,7 @@ protected:
 
 protected:
     CIArrayOf<CNestedRuntimeStatisticCollection> map;
+    mutable ReadWriteLock lock;
 };
 
 class CNestedSummaryRuntimeStatisticMap : public CNestedRuntimeStatisticMap