HPCC-32788 Reduce the index cache contention to improve parallelism

- Ensure old cache entries are freed outside of a critical section

Signed-off-by: Gavin Halliday <[email protected]>

system/jhtree/jhtree.cpp

@@ -662,6 +662,38 @@ class CNodeMapping : public HTMapping<CNodeCacheEntry, CKeyIdAndPos>
     CNodeMapping * next = nullptr;
 };
 
+class DelayedCacheEntryReleaser : public IRemovedMappingCallback
+{
+    //This number should be high enough so that in all common cases it is not exceeded.
+    //There is little downside in making it too large - other than cache locality on the stack.
+    //It will depend on the number of worker threads and the variation in expanded size of the
+    //different leaf nodes.  Running stresstest with 15 workers and the default compression I
+    //have seen numbers as high as 30.
+    static constexpr unsigned maxFixed = 40;
+public:
+    ~DelayedCacheEntryReleaser()
+    {
+        for (unsigned i=0; i < numFixed; i++)
+            fixedPending[i]->Release();
+    }
+    virtual void noteRemoval(void * _mapping) override
+    {
+        CNodeMapping *mapping = reinterpret_cast<CNodeMapping *>(_mapping);
+        //Save the node onto a list, so it will be released when this object is released.
+        CJHTreeNode * node = const_cast<CJHTreeNode *>(mapping->query().getNode());
+        if (numFixed < maxFixed)
+            fixedPending[numFixed++] = node;
+        else
+            pending.append(*node);
+    }
+protected:
+    CIArray pending;
+    //Use a fixed array for a small number of allocations to avoid a heap allocation inside the critsec
+    CJHTreeNode * fixedPending[maxFixed]; // deliberately uninitialized
+    unsigned numFixed = 0;
+
+};
+
 typedef OwningSimpleHashTableOf<CNodeMapping, CKeyIdAndPos> CNodeTable;
 class CNodeMRUCache final : public CMRUCacheOf<CKeyIdAndPos, CNodeCacheEntry, CNodeMapping, CNodeTable>
 {
@@ -700,10 +732,10 @@ class CNodeMRUCache final : public CMRUCacheOf<CKeyIdAndPos, CNodeCacheEntry, CN
         size32_t oldMemLimit = memLimit;
         memLimit = _memLimit;
         if (full())
-            makeSpace();
+            makeSpace(nullptr);
         return oldMemLimit;
     }
-    virtual void makeSpace()
+    virtual void makeSpace(IRemovedMappingCallback * callback)
     {
         // remove LRU until !full
         // This code could walk the list, rather than restarting at the end each time - but there are unlikely to be
@@ -732,6 +764,8 @@ class CNodeMRUCache final : public CMRUCacheOf<CKeyIdAndPos, CNodeCacheEntry, CN
 
             mruList.remove(tail);
             numEvicts.fastAdd(1);
+            if (callback)
+                callback->noteRemoval(tail);
             table.removeExact(tail);
         }
         while (full());
@@ -2667,6 +2701,7 @@ const CJHTreeNode *CNodeCache::getCachedNode(const INodeLoader *keyIndex, unsign
     Owned<CNodeCacheEntry> ownedCacheEntry; // ensure node gets cleaned up if it fails to load
     bool alreadyExists = true;
     {
+        DelayedCacheEntryReleaser delayedReleaser;
         CNodeCacheEntry * cacheEntry;
         unsigned hashcode = curCache.getKeyHash(key);
 
@@ -2691,7 +2726,7 @@ const CJHTreeNode *CNodeCache::getCachedNode(const INodeLoader *keyIndex, unsign
         else
         {
             cacheEntry = new CNodeCacheEntry;
-            curCache.replace(key, *cacheEntry);
+            curCache.replace(key, *cacheEntry, &delayedReleaser);
             alreadyExists = false;
             curCache.numAdds.fastAdd(1);
         }

system/jhtree/jhutil.hpp

@@ -22,6 +22,13 @@
 #include "jqueue.hpp"
 #include "jhtree.hpp"
 
+//This interface can be used to record the mapping items that are being removed when making more space.
+//If the cache is checked within a critical section this allows the removed items to be released outside the critsec
+interface IRemovedMappingCallback
+{
+    virtual void noteRemoval(void * mapping) = 0;
+};
+
 // TABLE should be SuperHashTable derivative to contain MAPPING's
 // MAPPING should be something that constructs with (KEY, ENTRY) and impl. query returning ref. to ENTRY
 template <class KEY, class ENTRY, class MAPPING, class TABLE>
@@ -38,13 +45,15 @@ class CMRUCacheOf : public CInterface//, public IInterface
     } table;
     DListOf<MAPPING> mruList;
 
-    void clear(int count)
+    void clear(int count, IRemovedMappingCallback * callback)
     {
         for (;;)
         {
             MAPPING *tail = mruList.dequeueTail();
             if (!tail)
                 break;
+            if (callback)
+                callback->noteRemoval(tail);
             table.removeExact(tail);
             if ((-1 != count) && (0 == --count))
                 break;
@@ -57,7 +66,16 @@ class CMRUCacheOf : public CInterface//, public IInterface
     void replace(KEY key, ENTRY &entry)
     {
         if (full())
-            makeSpace();
+            makeSpace(nullptr);
+
+        MAPPING * mapping = new MAPPING(key, entry); // owns entry
+        table.replace(*mapping);
+        mruList.enqueueHead(mapping);
+    }
+    void replace(KEY key, ENTRY &entry, IRemovedMappingCallback * callback)
+    {
+        if (full())
+            makeSpace(callback);
 
         MAPPING * mapping = new MAPPING(key, entry); // owns entry
         table.replace(*mapping);
@@ -106,7 +124,7 @@ class CMRUCacheOf : public CInterface//, public IInterface
         mruList.dequeue(mapping);
         return true;
     }
-    void kill() { clear(-1); }
+    void kill() { clear(-1, nullptr); }
     void promote(MAPPING *mapping)
     {
         mruList.moveToHead(mapping);
@@ -115,7 +133,7 @@ class CMRUCacheOf : public CInterface//, public IInterface
     {
         return new SuperHashIteratorOf<MAPPING>(table);
     }
-    virtual void makeSpace() { }
+    virtual void makeSpace(IRemovedMappingCallback *) { }
     virtual bool full() { return false; }
     virtual void elementAdded(MAPPING *mapping) { }
     virtual void elementRemoved(MAPPING *mapping) { }
@@ -132,14 +150,14 @@ class CMRUCacheMaxCountOf : public CMRUCacheOf<KEY, ENTRY, MAPPING, TABLE>
     unsigned setCacheLimit(unsigned _cacheMax)
     {
         if (SELF::table.count() > _cacheMax)
-            this->clear(_cacheMax - SELF::table.count());
+            this->clear(_cacheMax - SELF::table.count(), nullptr);
         unsigned oldCacheMax = cacheMax;
         cacheMax = _cacheMax;
         return oldCacheMax;
     }
-    virtual void makeSpace()
+    virtual void makeSpace(IRemovedMappingCallback * callback)
     {
-        SELF::clear(cacheOverflow);
+        SELF::clear(cacheOverflow, callback);
     }
     virtual bool full()
     {