From 858db2645d05ce0e91db91ff4899161bc3a08464 Mon Sep 17 00:00:00 2001
From: Malavan Sotheeswaran <105669860+msotheeswaran@users.noreply.github.com>
Date: Thu, 30 Mar 2023 20:23:56 -0700
Subject: [PATCH] add g_expirelock to previously async accesses to expireset
 (#618)

* add expire_lock to previously async access to expireset

* add flash option to create-cluster
---
 src/db.cpp                          | 10 +++++++++-
 src/expire.cpp                      |  1 +
 src/object.cpp                      |  1 +
 utils/create-cluster/create-cluster |  8 +++++++-
 4 files changed, 18 insertions(+), 2 deletions(-)

diff --git a/src/db.cpp b/src/db.cpp
index 21d00ad10..87ed33616 100644
--- a/src/db.cpp
+++ b/src/db.cpp
@@ -1931,6 +1931,7 @@ void setExpire(client *c, redisDb *db, robj *key, expireEntry &&e)
  * is associated with this key (i.e. the key is non volatile) */
 expireEntry *redisDbPersistentDataSnapshot::getExpire(const char *key) {
     /* No expire? return ASAP */
+    std::unique_lock<fastlock> ul(g_expireLock);
     if (expireSize() == 0)
         return nullptr;
 
@@ -2727,8 +2728,11 @@ void redisDbPersistentData::clear(void(callback)(void*))
         m_cnewKeysPending = 0;
         m_fAllChanged++;
     }
+    {
+    std::unique_lock<fastlock> ul(g_expireLock);
     delete m_setexpire;
     m_setexpire = new (MALLOC_LOCAL) expireset();
+    }
     if (m_spstorage != nullptr)
         m_spstorage->clear(callback);
     dictEmpty(m_pdictTombstone,callback);
@@ -2861,7 +2865,8 @@ LNotFound:
             {
                 dictAdd(m_pdict, sdsNewKey, o);
                 o->SetFExpires(spexpire != nullptr);
-
+                
+                std::unique_lock<fastlock> ul(g_expireLock);
                 if (spexpire != nullptr)
                 {
                     auto itr = m_setexpire->find(sdsKey);
@@ -2884,6 +2889,7 @@ LNotFound:
     if (*pde != nullptr && dictGetVal(*pde) != nullptr)
     {
         robj *o = (robj*)dictGetVal(*pde);
+        std::unique_lock<fastlock> ul(g_expireLock);
         serverAssert(o->FExpires() == (m_setexpire->find(sdsKey) != m_setexpire->end()));
     }
 }
@@ -3256,6 +3262,7 @@ std::unique_ptr<expireEntry> deserializeExpire(sds key, const char *str, size_t
 
 sds serializeStoredObjectAndExpire(redisDbPersistentData *db, const char *key, robj_roptr o)
 {
+    std::unique_lock<fastlock> ul(g_expireLock);
     auto itrExpire = db->setexpire()->find(key);
     const expireEntry *pexpire = nullptr;
     if (itrExpire != db->setexpire()->end())
@@ -3380,6 +3387,7 @@ void redisDbPersistentData::prefetchKeysAsync(client *c, parsed_command &command
                     dictAdd(m_pdict, sharedKey, o);
                     o->SetFExpires(spexpire != nullptr);
 
+                    std::unique_lock<fastlock> ul(g_expireLock);
                     if (spexpire != nullptr)
                     {
                         auto itr = m_setexpire->find(sharedKey);
diff --git a/src/expire.cpp b/src/expire.cpp
index 2f284d513..b727183ad 100644
--- a/src/expire.cpp
+++ b/src/expire.cpp
@@ -481,6 +481,7 @@ void expireSlaveKeys(void) {
     if (slaveKeysWithExpire == NULL ||
         dictSize(slaveKeysWithExpire) == 0) return;
 
+    std::unique_lock<fastlock> ul(g_expireLock);
     int cycles = 0, noexpire = 0;
     mstime_t start = mstime();
     while(1) {
diff --git a/src/object.cpp b/src/object.cpp
index 9b5ada3af..d3122dbbd 100644
--- a/src/object.cpp
+++ b/src/object.cpp
@@ -1668,6 +1668,7 @@ robj *deserializeStoredObjectCore(const void *data, size_t cb)
 robj *deserializeStoredObject(const redisDbPersistentData *db, const char *key, const void *data, size_t cb)
 {
     robj *o = deserializeStoredObjectCore(data, cb);
+    std::unique_lock<fastlock> ul(g_expireLock);
     o->SetFExpires(db->setexpire()->exists(key));
     return o;
 }
diff --git a/utils/create-cluster/create-cluster b/utils/create-cluster/create-cluster
index 4733f0ccb..c8df34044 100755
--- a/utils/create-cluster/create-cluster
+++ b/utils/create-cluster/create-cluster
@@ -26,7 +26,11 @@ then
     while [ $((PORT < ENDPORT)) != "0" ]; do
         PORT=$((PORT+1))
         echo "Starting $PORT"
-        $BIN_PATH/keydb-server --port $PORT  --protected-mode $PROTECTED_MODE --cluster-enabled yes --cluster-config-file nodes-${PORT}.conf --cluster-node-timeout $TIMEOUT --appendonly yes --appendfilename appendonly-${PORT}.aof --dbfilename dump-${PORT}.rdb --logfile ${PORT}.log --daemonize yes ${ADDITIONAL_OPTIONS}
+        if [ "$2" == "flash" ]
+        then
+            ADDITIONAL_OPTIONS="--save \"\" \"\" \"\" --semi-ordered-set-bucket-size 8 --client-output-buffer-limit replica 1 1 0 --maxmemory 100000000 --storage-provider flash ./$PORT.flash"
+        fi
+        $BIN_PATH/keydb-server --server-threads 4 --port $PORT  --protected-mode $PROTECTED_MODE --cluster-enabled yes --cluster-config-file nodes-${PORT}.conf --cluster-node-timeout $TIMEOUT --dbfilename dump-${PORT}.rdb --logfile ${PORT}.log --daemonize yes ${ADDITIONAL_OPTIONS}
     done
     exit 0
 fi
@@ -97,6 +101,8 @@ then
     rm -rf appendonly*.aof
     rm -rf dump*.rdb
     rm -rf nodes*.conf
+    rm -rf *.flash
+    rm -rf temp*.rdb
     exit 0
 fi