Fix race in writes

Author: JohnSully

src/AsyncWorkQueue.h

@@ -37,4 +37,13 @@ class AsyncWorkQueue
     void shutdown();
 
     void abandonThreads();
+
+    void waitForEmpty() {
+        for (;;) {
+            std::unique_lock<std::mutex> l(m_mutex);
+            if (m_workqueue.empty())
+                return;
+            sched_yield();
+        }
+    }
 };

src/debug.cpp

@@ -298,6 +298,11 @@ void computeDatasetDigest(unsigned char *final) {
 
     memset(final,0,20); /* Start with a clean result */
 
+    // For test reliabilty ensure all writes are applied
+    if (g_pserver->m_pstorageFactory) {
+        g_pserver->asyncwriteworkqueue->waitForEmpty();
+    }
+
     for (j = 0; j < cserver.dbnum; j++) {
         redisDb *db = g_pserver->db[j];
 

src/replication.cpp

@@ -1324,6 +1324,12 @@ int startBgsaveForReplication(int mincapa) {
     serverLog(LL_NOTICE,"Starting BGSAVE for SYNC with target: %s",
         socket_target ? "replicas sockets" : "disk");
 
+    // When FLASH is enabled we have a seperate responsibility of ensuring the write queue is flushed otherwise
+    //  the data won't match the repl buffer
+    if (g_pserver->m_pstorageFactory) {
+        g_pserver->asyncwriteworkqueue->waitForEmpty();
+    }
+
     rdbSaveInfo rsi, *rsiptr;
     rsiptr = rdbPopulateSaveInfo(&rsi);
     /* Only do rdbSave* when rsiptr is not NULL,

src/storage/rocksdb.cpp

@@ -344,11 +344,12 @@ StorageToken* RocksDBStorageProvider::begin_endWriteBatch(struct aeEventLoop *el
     tok->tspbatch = std::move(m_spbatch);
     tok->tspdb = m_spdb;
     m_spbatch = nullptr;
-    m_lock.unlock();
-    (*m_pfactory->m_wwqueue)->AddWorkFunction([this, el,callback,tok]{
+    // This has a race with future writes so async is disabled
+    //(*m_pfactory->m_wwqueue)->AddWorkFunction([this, el,callback,tok]{
         tok->tspdb->Write(WriteOptions(),tok->tspbatch.get()->GetWriteBatch());
         aePostFunction(el,callback,tok);
-    });
+    //});
+    m_lock.unlock();
 
     return tok;
 }