Latency normalization with SCAN

Former-commit-id: 237f25d854e70d4d7a3095fdf56aaa80770e492e

src/db.cpp

@@ -896,7 +896,10 @@ void keysCommand(client *c) {
                 locker.arm(c);
 
                 unblockClient(c);
-                db->endSnapshot(snapshot);
+
+                locker.disarm();
+                lock.unlock();
+                db->endSnapshotAsync(snapshot);
                 aeAcquireLock();
             });
         });
@@ -1044,7 +1047,7 @@ void scanGenericCommand(client *c, robj_roptr o, unsigned long cursor) {
         // Do an async version
         const redisDbPersistentDataSnapshot *snapshot = nullptr;
         if (!(c->flags & (CLIENT_MULTI | CLIENT_BLOCKED)))
-            snapshot = c->db->createSnapshot(c->mvccCheckpoint, true /* fOptional */);
+            snapshot = c->db->createSnapshot(c->mvccCheckpoint, false /* fOptional */);
         if (snapshot != nullptr)
         {
             aeEventLoop *el = serverTL->el;
@@ -1082,9 +1085,16 @@ void scanGenericCommand(client *c, robj_roptr o, unsigned long cursor) {
                     locker.arm(c);
 
                     unblockClient(c);
+                    mstime_t timeScanFilter;
+                    latencyStartMonitor(timeScanFilter);
                     scanFilterAndReply(c, keys, nullptr, nullptr, false, nullptr, cursorResult);
+                    latencyEndMonitor(timeScanFilter);
+                    latencyAddSampleIfNeeded("scan-async-filter", timeScanFilter);
 
-                    db->endSnapshot(snapshot);
+                    locker.disarm();
+                    lock.unlock();
+
+                    db->endSnapshotAsync(snapshot);
                     listSetFreeMethod(keys,decrRefCountVoid);
                     listRelease(keys);
                     aeAcquireLock();

src/snapshot.cpp

@@ -182,7 +182,8 @@ void redisDbPersistentData::restoreSnapshot(const redisDbPersistentDataSnapshot
 //  have some internal heuristics to do a synchronous endSnapshot if it makes sense
 void redisDbPersistentData::endSnapshotAsync(const redisDbPersistentDataSnapshot *psnapshot)
 {
-    aeAcquireLock();
+    mstime_t latency;
+    aeAcquireLock(); latencyStartMonitor(latency);
         if (m_pdbSnapshotASYNC && m_pdbSnapshotASYNC->m_mvccCheckpoint <= psnapshot->m_mvccCheckpoint)
         {
             // Free a stale async snapshot so consolidate_children can clean it up later
@@ -198,6 +199,8 @@ void redisDbPersistentData::endSnapshotAsync(const redisDbPersistentDataSnapshot
         {
             // For small snapshots it makes more sense just to merge it directly
             endSnapshot(psnapshot);
+            latencyEndMonitor(latency);
+            latencyAddSampleIfNeeded("end-snapshot-async-synchronous-path", latency);
             aeReleaseLock();
             return;
         }
@@ -206,19 +209,22 @@ void redisDbPersistentData::endSnapshotAsync(const redisDbPersistentDataSnapshot
         auto psnapshotT = createSnapshot(LLONG_MAX, false);
         endSnapshot(psnapshot); // this will just dec the ref count since our new snapshot has a ref 
         psnapshot = nullptr;
-    aeReleaseLock();
+    aeReleaseLock(); latencyEndMonitor(latency);
+    latencyAddSampleIfNeeded("end-snapshot-async-phase-1", latency);
 
     // 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();
+    aeAcquireLock(); latencyStartMonitor(latency);
         if (m_pdbSnapshotASYNC == nullptr)
             m_pdbSnapshotASYNC = psnapshotT;
         else
             endSnapshot(psnapshotT);    // finally clean up our temp snapshot
-    aeReleaseLock();
+    aeReleaseLock(); latencyEndMonitor(latency);
+    
+    latencyAddSampleIfNeeded("end-snapshot-async-phase-2", latency);
 }
 
 void redisDbPersistentDataSnapshot::freeTombstoneObjects(int depth)
@@ -262,6 +268,9 @@ void redisDbPersistentData::endSnapshot(const redisDbPersistentDataSnapshot *psn
         return;
     }
 
+    mstime_t latency_endsnapshot;
+    latencyStartMonitor(latency_endsnapshot);
+
     // Alright we're ready to be free'd, but first dump all the refs on our child snapshots
     if (m_spdbSnapshotHOLDER->m_refCount == 1)
         recursiveFreeSnapshots(m_spdbSnapshotHOLDER.get());
@@ -288,9 +297,6 @@ 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;