Make snapshot completion faster and add latency monitor

Former-commit-id: 8063be6ee70a652c22c3263dccf318366e208891

src/dict.cpp

@@ -541,7 +541,7 @@ void dictRelease(dict *d)
     zfree(d);
 }
 
-dictEntry *dictFind(dict *d, const void *key)
+dictEntry *dictFindWithPrev(dict *d, const void *key, dictEntry ***dePrevPtr, dictht **pht)
 {
     dictEntry *he;
     uint64_t h, idx, table;
@@ -550,11 +550,15 @@ dictEntry *dictFind(dict *d, const void *key)
     if (dictIsRehashing(d)) _dictRehashStep(d);
     h = dictHashKey(d, key);
     for (table = 0; table <= 1; table++) {
+        *pht = d->ht + table;
         idx = h & d->ht[table].sizemask;
         he = d->ht[table].table[idx];
+        *dePrevPtr = &d->ht[table].table[idx];
         while(he) {
-            if (key==he->key || dictCompareKeys(d, key, he->key))
+            if (key==he->key || dictCompareKeys(d, key, he->key)) {       
                 return he;
+            }
+            *dePrevPtr = &he->next;
             he = he->next;
         }
         if (!dictIsRehashing(d)) return NULL;
@@ -562,6 +566,13 @@ dictEntry *dictFind(dict *d, const void *key)
     return NULL;
 }
 
+dictEntry *dictFind(dict *d, const void *key)
+{
+    dictEntry **deT;
+    dictht *ht;
+    return dictFindWithPrev(d, key, &deT, &ht);
+}
+
 void *dictFetchValue(dict *d, const void *key) {
     dictEntry *he;
 

src/dict.h

@@ -167,6 +167,7 @@ dictEntry *dictUnlink(dict *ht, const void *key);
 void dictFreeUnlinkedEntry(dict *d, dictEntry *he);
 void dictRelease(dict *d);
 dictEntry * dictFind(dict *d, const void *key);
+dictEntry * dictFindWithPrev(dict *d, const void *key, dictEntry ***dePrevPtr, dictht **ht);
 void *dictFetchValue(dict *d, const void *key);
 int dictResize(dict *d);
 dictIterator *dictGetIterator(dict *d);

src/snapshot.cpp

@@ -200,6 +200,7 @@ void redisDbPersistentData::endSnapshotAsync(const redisDbPersistentDataSnapshot
 
     // do the expensive work of merging snapshots outside the ref
     const_cast<redisDbPersistentDataSnapshot*>(psnapshotT)->freeTombstoneObjects(1);    // depth is one because we just creted it
+    const_cast<redisDbPersistentDataSnapshot*>(psnapshotT)->consolidate_children(this, true);
     
     // Final Cleanup
     aeAcquireLock();
@@ -277,12 +278,17 @@ void redisDbPersistentData::endSnapshot(const redisDbPersistentDataSnapshot *psn
         return;
     }
 
+    mstime_t latency_endsnapshot;
+    latencyStartMonitor(latency_endsnapshot);
+
     // Stage 1 Loop through all the tracked deletes and remove them from the snapshot DB
     dictIterator *di = dictGetIterator(m_pdictTombstone);
     dictEntry *de;
     while ((de = dictNext(di)) != NULL)
     {
-        dictEntry *deSnapshot = dictFind(m_spdbSnapshotHOLDER->m_pdict, dictGetKey(de));
+        dictEntry **dePrev;
+        dictht *ht;
+        dictEntry *deSnapshot = dictFindWithPrev(m_spdbSnapshotHOLDER->m_pdict, dictGetKey(de), &dePrev, &ht);
         if (deSnapshot == nullptr && m_spdbSnapshotHOLDER->m_pdbSnapshot)
         {
             // The tombstone is for a grand child, propogate it (or possibly in the storage provider - but an extra tombstone won't hurt)
@@ -296,8 +302,13 @@ void redisDbPersistentData::endSnapshot(const redisDbPersistentDataSnapshot *psn
             continue;
         }
         
-        const char *key = (const char*)dictGetKey(deSnapshot);
+        // Delete the object from the source dict, we don't use dictDelete to avoid a second search
-        dictDelete(m_spdbSnapshotHOLDER->m_pdict, key);
+        dictFreeKey(m_spdbSnapshotHOLDER->m_pdict, deSnapshot);
+        dictFreeVal(m_spdbSnapshotHOLDER->m_pdict, deSnapshot);
+        serverAssert(*dePrev == deSnapshot);
+        *dePrev = deSnapshot->next;
+        zfree(deSnapshot);
+        ht->used--;
     }
     dictReleaseIterator(di);
     dictEmpty(m_pdictTombstone, nullptr);
@@ -339,6 +350,9 @@ void redisDbPersistentData::endSnapshot(const redisDbPersistentDataSnapshot *psn
     serverAssert(m_spdbSnapshotHOLDER != nullptr || dictSize(m_pdictTombstone) == 0);
     serverAssert(sizeStart == size());
 
+    latencyEndMonitor(latency_endsnapshot);
+    latencyAddSampleIfNeeded("end-mvcc-snapshot", latency_endsnapshot);
+
     freeMemoryIfNeededAndSafe(false);
 }