fix indent in evict.cpp

src/IStorage.h

@@ -31,6 +31,7 @@ public:
     virtual void retrieve(const char *key, size_t cchKey, callbackSingle fn) const = 0;
     virtual size_t clear() = 0;
     virtual bool enumerate(callback fn) const = 0;
+    virtual size_t stateful_enumerate(callback fn) = 0;
     virtual size_t count() const = 0;
 
     virtual void bulkInsert(char **rgkeys, size_t *rgcbkeys, char **rgvals, size_t *rgcbvals, size_t celem) {

src/StorageCache.h

@@ -49,6 +49,7 @@ public:
     void expand(uint64_t slots);
 
     bool enumerate(IStorage::callback fn) const { return m_spstorage->enumerate(fn); }
+    size_t stateful_enumerate(IStorage::callback fn) { return m_spstorage->stateful_enumerate(fn); }
 
     void beginWriteBatch();
     void endWriteBatch() { m_spstorage->endWriteBatch(); }

src/evict.cpp

@@ -771,7 +771,7 @@ int performEvictions(bool fPreSnapshot) {
                 if (db->removeCachedValue(bestkey, &deT)) {
                     mem_freed += splazy->addEntry(db->dictUnsafeKeyOnly(), deT);
                     ckeysFailed = 0;
-		    g_pserver->stat_evictedkeys++;
+                    g_pserver->stat_evictedkeys++;
                 }
                 else {
                     delta = 0;

src/expire.cpp

@@ -416,6 +416,33 @@ void activeExpireCycleCore(int type) {
             return true;
         }, &check);
 
+        if (!timelimit_exit && db->getStorageCache() != nullptr)
+            db->getStorageCache()->stateful_enumerate([&] (const char *key, size_t keylen, const void *val, size_t vallen) {
+                sds sdsKey = sdsnewlen(key, keylen);
+                size_t offset = 0;
+                auto e = deserializeExpire(sdsKey, (const char*)val, vallen, &offset);
+                if (e->when() < now)
+                {
+                    expired += activeExpireCycleExpire(db, *e, now, tried);
+                }
+                sdsfree(sdsKey);
+
+                if ((tried % ACTIVE_EXPIRE_CYCLE_LOOKUPS_PER_LOOP) == 0)
+                {
+                    /* We can't block forever here even if there are many keys to
+                    * expire. So after a given amount of milliseconds return to the
+                    * caller waiting for the other active expire cycle. */
+                    elapsed = ustime()-start;
+                    if (elapsed > timelimit) {
+                        timelimit_exit = 1;
+                        g_pserver->stat_expired_time_cap_reached_count++;
+                        return false;
+                    }
+                    check = ACTIVE_EXPIRE_CYCLE_FAST_DURATION;
+                }
+                return true;
+            });
+
         total_expired += expired;
     }
 

src/server.h

@@ -1209,6 +1209,7 @@ public:
     bool FSnapshot() const { return m_spdbSnapshotHOLDER != nullptr; }
 
     std::unique_ptr<const StorageCache> CloneStorageCache() { return std::unique_ptr<const StorageCache>(m_spstorage->clone()); }
+    std::shared_ptr<StorageCache> getStorageCache() { return m_spstorage; }
     void bulkDirectStorageInsert(char **rgKeys, size_t *rgcbKeys, char **rgVals, size_t *rgcbVals, size_t celem);
 
     dict_iter find_cached_threadsafe(const char *key) const;
@@ -1371,6 +1372,7 @@ struct redisDb : public redisDbPersistentDataSnapshot
     using redisDbPersistentData::FTrackingChanges;
     using redisDbPersistentData::CloneStorageCache;
     using redisDbPersistentData::bulkDirectStorageInsert;
+    using redisDbPersistentData::getStorageCache;
 
 public:
     const redisDbPersistentDataSnapshot *createSnapshot(uint64_t mvccCheckpoint, bool fOptional) {

src/storage/rocksdb.cpp

@@ -168,6 +168,28 @@ bool RocksDBStorageProvider::enumerate(callback fn) const
     return !it->Valid();
 }
 
+size_t RocksDBStorageProvider::stateful_enumerate(callback fn)
+{
+    if (m_iter == nullptr || !m_iter->Valid()) {
+        m_iter = std::unique_ptr<rocksdb::Iterator>(m_spdb->NewIterator(ReadOptions(), m_spcolfamily.get()));
+        m_iter->SeekToFirst();
+    }
+    size_t count = 0;
+    while (m_iter->Valid()) {
+        if (FInternalKey(m_iter->key().data(), m_iter->key().size()))
+            continue;
+
+        if (fn(m_iter->key().data(), m_iter->key().size(), m_iter->value().data(), m_iter->value().size())) {
+            m_iter->Next();
+            ++count;
+        } else {
+            break;
+        }
+    }
+    assert(m_iter->status().ok()); // Check for any errors found during the scan
+    return count;
+}
+
 const IStorage *RocksDBStorageProvider::clone() const
 {
     std::unique_lock<fastlock> l(m_lock);

src/storage/rocksdb.h

@@ -22,6 +22,7 @@ class RocksDBStorageProvider : public IStorage
     rocksdb::ReadOptions m_readOptionsTemplate;
     size_t m_count = 0;
     mutable fastlock m_lock {"RocksDBStorageProvider"};
+    std::unique_ptr<rocksdb::Iterator> m_iter;
 
 public:
     RocksDBStorageProvider(RocksDBStorageFactory *pfactory, std::shared_ptr<rocksdb::DB> &spdb, std::shared_ptr<rocksdb::ColumnFamilyHandle> &spcolfam, const rocksdb::Snapshot *psnapshot, size_t count);
@@ -32,6 +33,7 @@ public:
     virtual void retrieve(const char *key, size_t cchKey, callbackSingle fn) const override;
     virtual size_t clear() override;
     virtual bool enumerate(callback fn) const override;
+    virtual size_t stateful_enumerate(callback fn) override;
 
     virtual const IStorage *clone() const override;
 

src/storage/teststorageprovider.cpp

@@ -73,6 +73,26 @@ bool TestStorageProvider::enumerate(callback fn) const
     }
     return fAll;
 }
+
+size_t TestStorageProvider::stateful_enumerate(callback fn)
+{
+    size_t count = 0;
+    if (m_iter == m_map.end())
+        m_iter = m_map.begin();
+
+    while (m_iter != m_map.end())
+    {
+        if (fn(m_iter->first.data(), m_iter->first.size(), m_iter->second.data(), m_iter->second.size()))
+        {
+            m_iter++;
+            count++;
+        } else
+        {
+            break;
+        }
+    }
+    return count;
+}
     
 size_t TestStorageProvider::count() const
 {

src/storage/teststorageprovider.h

@@ -14,6 +14,7 @@ class TestStorageFactory : public IStorageFactory
 class TestStorageProvider final : public IStorage
 {
     std::unordered_map<std::string, std::string> m_map;
+    std::unordered_map<std::string, std::string>::iterator m_iter = m_map.end();
 
 public:
     TestStorageProvider();
@@ -24,6 +25,7 @@ public:
     virtual void retrieve(const char *key, size_t cchKey, callbackSingle fn) const override;
     virtual size_t clear() override;
     virtual bool enumerate(callback fn) const override;
+    virtual size_t stateful_enumerate(callback fn) override;
     virtual size_t count() const override;
 
     virtual void flush() override;

tests/unit/flash.tcl

@@ -120,6 +120,21 @@ if {$::flash_enabled} {
             assert_equal {2} [r scard set1]
         }
 
+        test { Keys in storage are expired } {
+            r flushdb
+            r psetex key1 10000 a
+            r psetex key2 10000 a
+            r psetex key3 10000 a
+            r flushall cache
+
+            after 1000
+            wait_for_condition 1000 50 {
+                [r dbsize] == 0
+            } else {
+                fail "keys in ssd only not expired"
+            }
+        }
+
         r flushall
         # If a weak storage memory model is set, wait for any pending snapshot writes to finish
         after 500