don't delete db or snapshots on shutdown, still delete storage provider

Former-commit-id: edb840ce10ea77ce654ba27c9eadbf98bbc13403

src/db.cpp

@@ -2571,6 +2571,12 @@ void redisDbPersistentData::setStorageProvider(StorageCache *pstorage)
     m_spstorage = std::unique_ptr<StorageCache>(pstorage);
 }
 
+void redisDbPersistentData::endStorageProvider()
+{
+    serverAssert(m_spstorage != nullptr);
+    m_spstorage.reset();
+}
+
 void clusterStorageLoadCallback(const char *rgchkey, size_t cch, void *)
 {
     slotToKeyUpdateKeyCore(rgchkey, cch, true /*add*/);
@@ -2599,6 +2605,14 @@ void redisDb::storageProviderInitialize()
     }
 }
 
+void redisDb::storageProviderDelete()
+{
+    if (g_pserver->m_pstorageFactory != nullptr)
+    {
+        this->endStorageProvider();
+    }
+}
+
 bool redisDbPersistentData::insert(char *key, robj *o, bool fAssumeNew, dict_iter *piterExisting)
 {
     if (!fAssumeNew && (g_pserver->m_pstorageFactory != nullptr || m_pdbSnapshot != nullptr))

src/server.cpp

@@ -7079,13 +7079,6 @@ void *workerThreadMain(void *parg)
     serverAssert(!GlobalLocksAcquired());
     aeDeleteEventLoop(el);
 
-    aeAcquireLock();
-    for (int idb = 0; idb < cserver.dbnum; ++idb) {
-        if (g_pserver->rgthreadvar[iel].rgdbSnapshot[idb] != nullptr)
-            g_pserver->db[idb]->endSnapshot(g_pserver->rgthreadvar[iel].rgdbSnapshot[idb]);
-    }
-    aeReleaseLock();
-
     return NULL;
 }
 
@@ -7507,8 +7500,7 @@ int main(int argc, char **argv) {
     if (!fLockAcquired)
         g_fInCrash = true;  // We don't actually crash right away, because we want to sync any storage providers
     for (int idb = 0; idb < cserver.dbnum; ++idb) {
-        delete g_pserver->db[idb];
-        g_pserver->db[idb] = nullptr;
+        g_pserver->db[idb]->storageProviderDelete();
     }
     // If we couldn't acquire the global lock it means something wasn't shutdown and we'll probably deadlock
     serverAssert(fLockAcquired);

src/server.h

@@ -1158,6 +1158,7 @@ public:
     bool FRehashing() const { return dictIsRehashing(m_pdict) || dictIsRehashing(m_pdictTombstone); }
 
     void setStorageProvider(StorageCache *pstorage);
+    void endStorageProvider();
 
     void trackChanges(bool fBulk, size_t sizeHint = 0);
     bool FTrackingChanges() const { return !!m_fTrackingChanges; }
@@ -1301,6 +1302,7 @@ struct redisDb : public redisDbPersistentDataSnapshot
 
     void initialize(int id);
     void storageProviderInitialize();
+    void storageProviderDelete();
     virtual ~redisDb();
 
     void dbOverwriteCore(redisDb::iter itr, sds keySds, robj *val, bool fUpdateMvcc, bool fRemoveExpire);