Ensure size() count is accurate, and fix find_threadsafe which didn't check children

Former-commit-id: ffa44b636afd4fe907c79fe7deebac3d1e091ee9

src/db.cpp

@@ -372,7 +372,13 @@ bool redisDbPersistentData::syncDelete(robj *key)
         removeExpire(key, itr);
     if (dictDelete(m_pdict,ptrFromObj(key)) == DICT_OK) {
         if (m_pdbSnapshot != nullptr)
-            dictAdd(m_pdictTombstone, sdsdup(szFromObj(key)), nullptr);
+        {
+            auto itr = m_pdbSnapshot->find_threadsafe(szFromObj(key));
+            if (itr != nullptr)
+            {
+                dictAdd(m_pdictTombstone, sdsdup(szFromObj(key)), nullptr);
+            }
+        }
         if (g_pserver->cluster_enabled) slotToKeyDel(key);
         return 1;
     } else {
@@ -1899,9 +1905,16 @@ void redisDb::initialize(int id)
 
 bool redisDbPersistentData::insert(char *key, robj *o)
 {
+    ensure(key);
     int res = dictAdd(m_pdict, key, o);
     if (res == DICT_OK)
+    {
+        if (m_pdbSnapshot != nullptr && m_pdbSnapshot->find_threadsafe(key) != nullptr)
+        {
+            serverAssert(dictFind(m_pdictTombstone, key) != nullptr);
+        }
         trackkey(key);
+    }
     return (res == DICT_OK);
 }
 
@@ -2036,6 +2049,7 @@ void redisDbPersistentData::ensure(const char *sdsKey, dictEntry **pde)
                 serverAssert(objNew->mvcc_tstamp == itr.val()->mvcc_tstamp);
             }
             *pde = dictFind(m_pdict, sdsKey);
+            dictAdd(m_pdictTombstone, sdsdup(sdsKey), nullptr);
         }
     }
     else if (*pde != nullptr && dictGetVal(*pde) == nullptr)

src/server.h

@@ -1315,11 +1315,7 @@ public:
     using redisDbPersistentData::end;
 
     dict_iter random_threadsafe() const;
-    dict_iter find_threadsafe(const char *key) const
-    {
-        dictEntry *de = dictFind(m_pdict, key);
-        return dict_iter(de);
-    }
+    dict_iter find_threadsafe(const char *key) const;
 
     // These need to be fixed
     using redisDbPersistentData::size;

src/snapshot.cpp

@@ -120,7 +120,12 @@ void redisDbPersistentData::endSnapshot(const redisDbPersistentDataSnapshot *psn
     {
         dictEntry *deSnapshot = dictFind(m_spdbSnapshotHOLDER->m_pdict, dictGetKey(de));
         if (deSnapshot == nullptr)
-            continue;   // sometimes we delete things that were never in the snapshot
+        {
+            // The tombstone is for a grand child, propogate it
+            serverAssert(m_spdbSnapshotHOLDER->m_pdbSnapshot->find_threadsafe((const char*)dictGetKey(de)) != nullptr);
+            dictAdd(m_spdbSnapshotHOLDER->m_pdictTombstone, sdsdup((sds)dictGetKey(de)), nullptr);
+            continue;
+        }
         
         robj *obj = (robj*)dictGetVal(deSnapshot);
         const char *key = (const char*)dictGetKey(deSnapshot);
@@ -157,6 +162,7 @@ void redisDbPersistentData::endSnapshot(const redisDbPersistentDataSnapshot *psn
     
     // Stage 3 swap the databases with the snapshot
     std::swap(m_pdict, m_spdbSnapshotHOLDER->m_pdict);
+    std::swap(m_pdictTombstone, m_spdbSnapshotHOLDER->m_pdictTombstone);
 
     // Stage 4 merge all expires
     // TODO
@@ -183,6 +189,7 @@ void redisDbPersistentData::endSnapshot(const redisDbPersistentDataSnapshot *psn
     serverAssert(m_spdbSnapshotHOLDER != nullptr || m_pdbSnapshot == nullptr);
     serverAssert(m_pdbSnapshot == m_spdbSnapshotHOLDER.get() || m_pdbSnapshot == nullptr);
     serverAssert((m_refCount == 0 && m_pdict->iterators == 0) || (m_refCount != 0 && m_pdict->iterators == 1));
+    serverAssert(m_spdbSnapshotHOLDER != nullptr || dictSize(m_pdictTombstone) == 0);
 }
 
 dict_iter redisDbPersistentDataSnapshot::random_threadsafe() const
@@ -204,14 +211,32 @@ dict_iter redisDbPersistentDataSnapshot::random_threadsafe() const
     return dict_iter(de);
 }
 
+dict_iter redisDbPersistentDataSnapshot::find_threadsafe(const char *key) const
+{
+    dictEntry *de = dictFind(m_pdict, key);
+    if (de == nullptr && m_pdbSnapshot != nullptr)
+    {
+        auto itr = m_pdbSnapshot->find_threadsafe(key);
+        if (itr != nullptr && dictFind(m_pdictTombstone, itr.key()) == nullptr)
+            return itr;
+    }
+    return dict_iter(de);
+}
+
 bool redisDbPersistentDataSnapshot::iterate_threadsafe(std::function<bool(const char*, robj_roptr o)> fn) const
 {
     dictEntry *de = nullptr;
     bool fResult = true;
 
+    // Take the size so we can ensure we visited every element exactly once
+    //  use volatile to ensure it's not checked too late.  This makes it more
+    //  likely we'll detect races (but it won't gurantee it)
+    volatile size_t celem = size();
+
     dictIterator *di = dictGetIterator(m_pdict);
     while((de = dictNext(di)) != nullptr)
     {
+        --celem;
         if (!fn((const char*)dictGetKey(de), (robj*)dictGetVal(de)))
         {
             fResult = false;
@@ -228,14 +253,17 @@ bool redisDbPersistentDataSnapshot::iterate_threadsafe(std::function<bool(const
             dictEntry *deCurrent = dictFind(m_pdict, key);
             if (deCurrent != nullptr)
                 return true;
+            
             dictEntry *deTombstone = dictFind(m_pdictTombstone, key);
             if (deTombstone != nullptr)
                 return true;
 
             // Alright it's a key in the use keyspace, lets ensure it and then pass it off
+            --celem;
             return fn(key, o);
         });
     }
 
+    serverAssert(!fResult || celem == 0);
     return fResult;
 }