diff --git a/src/StorageCache.cpp b/src/StorageCache.cpp index 23b9af90a..64ae0051e 100644 --- a/src/StorageCache.cpp +++ b/src/StorageCache.cpp @@ -1,36 +1,80 @@ #include "server.h" +uint64_t hashPassthrough(const void *hash) { + return static_cast(reinterpret_cast(hash)); +} + +int hashCompare(void *, const void *key1, const void *key2) { + auto diff = (reinterpret_cast(key1) - reinterpret_cast(key2)); + return !diff; +} + +dictType dbStorageCacheType = { + hashPassthrough, /* hash function */ + NULL, /* key dup */ + NULL, /* val dup */ + hashCompare, /* key compare */ + NULL, /* key destructor */ + NULL /* val destructor */ +}; + +StorageCache::StorageCache(IStorage *storage) + : m_spstorage(storage) +{ + m_pdict = dictCreate(&dbStorageCacheType, nullptr); +} + void StorageCache::clear() { std::unique_lock ul(m_lock); - if (m_setkeys != nullptr) - m_setkeys->clear(); + if (m_pdict != nullptr) + dictEmpty(m_pdict, nullptr); m_spstorage->clear(); + m_collisionCount = 0; } void StorageCache::cacheKey(sds key) { - if (m_setkeys == nullptr) + if (m_pdict == nullptr) return; - m_setkeys->insert(sdsimmutablestring(sdsdupshared(key))); + uintptr_t hash = dictSdsHash(key); + if (dictAdd(m_pdict, reinterpret_cast(hash), (void*)1) != DICT_OK) { + dictEntry *de = dictFind(m_pdict, reinterpret_cast(hash)); + serverAssert(de != nullptr); + de->v.s64++; + m_collisionCount++; + } } void StorageCache::cacheKey(const char *rgch, size_t cch) { - if (m_setkeys == nullptr) + if (m_pdict == nullptr) return; - m_setkeys->insert(sdsimmutablestring(sdsnewlen(rgch, cch))); + uintptr_t hash = dictGenHashFunction(rgch, (int)cch); + if (dictAdd(m_pdict, reinterpret_cast(hash), (void*)1) != DICT_OK) { + dictEntry *de = dictFind(m_pdict, reinterpret_cast(hash)); + serverAssert(de != nullptr); + de->v.s64++; + m_collisionCount++; + } } bool StorageCache::erase(sds key) { bool result = m_spstorage->erase(key, sdslen(key)); std::unique_lock ul(m_lock); - if (result && m_setkeys != nullptr) + if (result && m_pdict != nullptr) { - auto itr = m_setkeys->find(sdsview(key)); - serverAssert(itr != m_setkeys->end()); - m_setkeys->erase(itr); + uint64_t hash = dictSdsHash(key); + dictEntry *de = dictFind(m_pdict, reinterpret_cast(hash)); + serverAssert(de != nullptr); + de->v.s64--; + serverAssert(de->v.s64 >= 0); + if (de->v.s64 == 0) { + dictDelete(m_pdict, reinterpret_cast(hash)); + } else { + m_collisionCount--; + } } return result; } @@ -38,7 +82,7 @@ bool StorageCache::erase(sds key) void StorageCache::insert(sds key, const void *data, size_t cbdata, bool fOverwrite) { std::unique_lock ul(m_lock); - if (!fOverwrite && m_setkeys != nullptr) + if (!fOverwrite && m_pdict != nullptr) { cacheKey(key); } @@ -54,16 +98,16 @@ const StorageCache *StorageCache::clone() return cacheNew; } -void StorageCache::retrieve(sds key, IStorage::callbackSingle fn, sds *cachedKey) const +void StorageCache::retrieve(sds key, IStorage::callbackSingle fn) const { std::unique_lock ul(m_lock); - if (m_setkeys != nullptr) + if (m_pdict != nullptr) { - auto itr = m_setkeys->find(sdsview(key)); - if (itr == m_setkeys->end()) + uint64_t hash = dictSdsHash(key); + dictEntry *de = dictFind(m_pdict, reinterpret_cast(hash)); + + if (de == nullptr) return; // Not found - if (cachedKey != nullptr) - *cachedKey = sdsdupshared(itr->get()); } ul.unlock(); m_spstorage->retrieve(key, sdslen(key), fn); @@ -73,8 +117,9 @@ size_t StorageCache::count() const { std::unique_lock ul(m_lock); size_t count = m_spstorage->count(); - if (m_setkeys != nullptr) - serverAssert(count == m_setkeys->size()); + if (m_pdict != nullptr) { + serverAssert(count == (dictSize(m_pdict) + m_collisionCount)); + } return count; } diff --git a/src/StorageCache.h b/src/StorageCache.h index 4fa3c3a08..33492cab1 100644 --- a/src/StorageCache.h +++ b/src/StorageCache.h @@ -4,12 +4,11 @@ class StorageCache { std::shared_ptr m_spstorage; - std::unique_ptr> m_setkeys; + dict *m_pdict = nullptr; + int m_collisionCount = 0; mutable fastlock m_lock {"StorageCache"}; - StorageCache(IStorage *storage) - : m_spstorage(storage) - {} + StorageCache(IStorage *storage); void cacheKey(sds key); void cacheKey(const char *rgchKey, size_t cchKey); @@ -31,9 +30,10 @@ class StorageCache public: static StorageCache *create(IStorageFactory *pfactory, int db, IStorageFactory::key_load_iterator fn, void *privdata) { StorageCache *cache = new StorageCache(nullptr); - if (pfactory->FSlow()) + if (!pfactory->FSlow()) { - cache->m_setkeys = std::make_unique>(20); + dictRelease(cache->m_pdict); + cache->m_pdict = nullptr; } load_iter_data data = {cache, fn, privdata}; cache->m_spstorage = std::shared_ptr(pfactory->create(db, key_load_itr, (void*)&data)); @@ -42,7 +42,7 @@ public: void clear(); void insert(sds key, const void *data, size_t cbdata, bool fOverwrite); - void retrieve(sds key, IStorage::callbackSingle fn, sds *sharedKeyOut) const; + void retrieve(sds key, IStorage::callbackSingle fn) const; bool erase(sds key); bool enumerate(IStorage::callback fn) const { return m_spstorage->enumerate(fn); } diff --git a/src/db.cpp b/src/db.cpp index 8b6b24d31..b60462c09 100644 --- a/src/db.cpp +++ b/src/db.cpp @@ -49,7 +49,6 @@ struct dbBackup { int keyIsExpired(const redisDbPersistentDataSnapshot *db, robj *key); int expireIfNeeded(redisDb *db, robj *key, robj *o); void slotToKeyUpdateKeyCore(const char *key, size_t keylen, int add); -void changedescDtor(void *privdata, void *obj); std::unique_ptr deserializeExpire(sds key, const char *str, size_t cch, size_t *poffset); sds serializeStoredObjectAndExpire(redisDbPersistentData *db, const char *key, robj_roptr o); @@ -59,15 +58,10 @@ dictType dictChangeDescType { NULL, /* key dup */ NULL, /* val dup */ dictSdsKeyCompare, /* key compare */ - nullptr, /* key destructor */ - changedescDtor /* val destructor */ + dictSdsDestructor, /* key destructor */ + nullptr /* val destructor */ }; -void changedescDtor(void *, void *obj) { - redisDbPersistentData::changedesc *desc = (redisDbPersistentData::changedesc*)obj; - delete desc; -} - /* Update LFU when an object is accessed. * Firstly, decrement the counter if the decrement time is reached. * Then logarithmically increment the counter, and update the access time. */ @@ -437,8 +431,8 @@ bool redisDbPersistentData::syncDelete(robj *key) dictEntry *de = dictUnlink(m_dictChanged, szFromObj(key)); if (de != nullptr) { - changedesc *desc = (changedesc*)dictGetVal(de); - if (!desc->fUpdate) + bool fUpdate = (bool)dictGetVal(de); + if (!fUpdate) --m_cnewKeysPending; dictFreeUnlinkedEntry(m_dictChanged, de); } @@ -2641,20 +2635,18 @@ LNotFound: { if (dictSize(m_pdict) != size()) // if all keys are cached then no point in looking up the database { - sds sdsNewKey = nullptr; // the storage cache will give us its cached key if available robj *o = nullptr; + sds sdsNewKey = sdsdupshared(sdsKey); std::unique_ptr spexpire; m_spstorage->retrieve((sds)sdsKey, [&](const char *, size_t, const void *data, size_t cb){ size_t offset = 0; - spexpire = deserializeExpire((sds)sdsNewKey, (const char*)data, cb, &offset); - o = deserializeStoredObject(this, sdsKey, reinterpret_cast(data) + offset, cb - offset); + spexpire = deserializeExpire(sdsNewKey, (const char*)data, cb, &offset); + o = deserializeStoredObject(this, sdsNewKey, reinterpret_cast(data) + offset, cb - offset); serverAssert(o != nullptr); - }, &sdsNewKey); + }); if (o != nullptr) { - if (sdsNewKey == nullptr) - sdsNewKey = sdsdupshared(sdsKey); dictAdd(m_pdict, sdsNewKey, o); o->SetFExpires(spexpire != nullptr); @@ -2667,11 +2659,8 @@ LNotFound: serverAssert(m_setexpire->find(sdsKey) != m_setexpire->end()); } serverAssert(o->FExpires() == (m_setexpire->find(sdsKey) != m_setexpire->end())); - } - else - { - if (sdsNewKey != nullptr) - sdsfree(sdsNewKey); // BUG but don't bother crashing + } else { + sdsfree(sdsNewKey); } *pde = dictFind(m_pdict, sdsKey); @@ -2705,14 +2694,14 @@ void redisDbPersistentData::storeDatabase() dictReleaseIterator(di); } -/* static */ void redisDbPersistentData::serializeAndStoreChange(StorageCache *storage, redisDbPersistentData *db, const redisDbPersistentData::changedesc &change) +/* static */ void redisDbPersistentData::serializeAndStoreChange(StorageCache *storage, redisDbPersistentData *db, const char *key, bool fUpdate) { - auto itr = db->find_cached_threadsafe(change.strkey.get()); + auto itr = db->find_cached_threadsafe(key); if (itr == nullptr) return; robj *o = itr.val(); sds temp = serializeStoredObjectAndExpire(db, (const char*) itr.key(), o); - storage->insert((sds)change.strkey.get(), temp, sdslen(temp), change.fUpdate); + storage->insert((sds)key, temp, sdslen(temp), fUpdate); sdsfree(temp); } @@ -2756,8 +2745,7 @@ bool redisDbPersistentData::processChanges(bool fSnapshot) dictEntry *de; while ((de = dictNext(di)) != nullptr) { - changedesc *change = (changedesc*)dictGetVal(de); - serializeAndStoreChange(m_spstorage.get(), this, *change); + serializeAndStoreChange(m_spstorage.get(), this, (const char*)dictGetKey(de), (bool)dictGetVal(de)); } dictReleaseIterator(di); } @@ -2776,8 +2764,7 @@ void redisDbPersistentData::commitChanges(const redisDbPersistentDataSnapshot ** dictEntry *de; while ((de = dictNext(di)) != nullptr) { - changedesc *change = (changedesc*)dictGetVal(de); - serializeAndStoreChange(m_spstorage.get(), (redisDbPersistentData*)m_pdbSnapshotStorageFlush, *change); + serializeAndStoreChange(m_spstorage.get(), (redisDbPersistentData*)m_pdbSnapshotStorageFlush, (const char*)dictGetKey(de), (bool)dictGetVal(de)); } dictReleaseIterator(di); dictRelease(m_dictChangedStorageFlush); @@ -2866,14 +2853,18 @@ bool redisDbPersistentData::removeCachedValue(const char *key) return true; } -void redisDbPersistentData::trackChanges(bool fBulk) +void redisDbPersistentData::trackChanges(bool fBulk, size_t sizeHint) { m_fTrackingChanges.fetch_add(1, std::memory_order_relaxed); if (fBulk) m_fAllChanged.fetch_add(1, std::memory_order_acq_rel); - if (m_dictChanged == nullptr) + if (m_dictChanged == nullptr) { m_dictChanged = dictCreate(&dictChangeDescType, nullptr); + } + + if (sizeHint > 0) + dictExpand(m_dictChanged, sizeHint, false); } void redisDbPersistentData::removeAllCachedValues() @@ -2886,7 +2877,16 @@ void redisDbPersistentData::removeAllCachedValues() trackChanges(false); } - dictEmpty(m_pdict, nullptr); + if (m_pdict->iterators == 0) { + dict *dT = m_pdict; + m_pdict = dictCreate(&dbDictType, this); + dictExpand(m_pdict, dictSize(dT)/2, false); // Make room for about half so we don't excessively rehash + g_pserver->asyncworkqueue->AddWorkFunction([dT]{ + dictRelease(dT); + }, true); + } else { + dictEmpty(m_pdict, nullptr); + } } void redisDbPersistentData::trackkey(const char *key, bool fUpdate) @@ -2894,8 +2894,7 @@ void redisDbPersistentData::trackkey(const char *key, bool fUpdate) if (m_fTrackingChanges && !m_fAllChanged && m_spstorage) { dictEntry *de = dictFind(m_dictChanged, key); if (de == nullptr) { - changedesc *desc = new changedesc(sdsdupshared(key), fUpdate); - dictAdd(m_dictChanged, (void*)desc->strkey.get(), desc); + dictAdd(m_dictChanged, (void*)sdsdupshared(key), (void*)fUpdate); if (!fUpdate) ++m_cnewKeysPending; } @@ -3024,20 +3023,17 @@ void redisDbPersistentData::prefetchKeysAsync(client *c, parsed_command &command std::vector>> vecInserts; for (robj *objKey : veckeys) { - sds sharedKey = nullptr; + sds sharedKey = sdsdupshared((sds)szFromObj(objKey)); std::unique_ptr spexpire; robj *o = nullptr; m_spstorage->retrieve((sds)szFromObj(objKey), [&](const char *, size_t, const void *data, size_t cb){ size_t offset = 0; - spexpire = deserializeExpire((sds)szFromObj(objKey), (const char*)data, cb, &offset); - o = deserializeStoredObject(this, szFromObj(objKey), reinterpret_cast(data) + offset, cb - offset); + spexpire = deserializeExpire(sharedKey, (const char*)data, cb, &offset); + o = deserializeStoredObject(this, sharedKey, reinterpret_cast(data) + offset, cb - offset); serverAssert(o != nullptr); - }, &sharedKey); + }); if (o != nullptr) { - if (sharedKey == nullptr) - sharedKey = sdsdupshared(szFromObj(objKey)); - vecInserts.emplace_back(sharedKey, o, std::move(spexpire)); } else if (sharedKey != nullptr) { sdsfree(sharedKey); diff --git a/src/rdb.cpp b/src/rdb.cpp index aefc730c0..3d4b9f345 100644 --- a/src/rdb.cpp +++ b/src/rdb.cpp @@ -2364,6 +2364,21 @@ void rdbLoadProgressCallback(rio *r, const void *buf, size_t len) { } } +class EvictionPolicyCleanup +{ + int oldpolicy; + +public: + EvictionPolicyCleanup() { + oldpolicy = g_pserver->maxmemory_policy; + g_pserver->maxmemory_policy = MAXMEMORY_ALLKEYS_RANDOM; + } + + ~EvictionPolicyCleanup() { + g_pserver->maxmemory_policy = oldpolicy; + } +}; + /* Load an RDB file from the rio stream 'rdb'. On success C_OK is returned, * otherwise C_ERR is returned and 'errno' is set accordingly. */ int rdbLoadRio(rio *rdb, int rdbflags, rdbSaveInfo *rsi) { @@ -2380,9 +2395,13 @@ int rdbLoadRio(rio *rdb, int rdbflags, rdbSaveInfo *rsi) { sds key = nullptr; bool fLastKeyExpired = false; + // If we're running flash we may evict during load. We want a fast eviction function + // because there isn't any difference in use times between keys anyways + EvictionPolicyCleanup ecleanup; + for (int idb = 0; idb < cserver.dbnum; ++idb) { - g_pserver->db[idb]->trackChanges(true); + g_pserver->db[idb]->trackChanges(true, 1024); } rdb->update_cksum = rdbLoadProgressCallback; @@ -2403,7 +2422,7 @@ int rdbLoadRio(rio *rdb, int rdbflags, rdbSaveInfo *rsi) { now = mstime(); lru_clock = LRU_CLOCK(); - + while(1) { robj *val; @@ -2645,17 +2664,23 @@ int rdbLoadRio(rio *rdb, int rdbflags, rdbSaveInfo *rsi) { } else { /* If we have a storage provider check if we need to evict some keys to stay under our memory limit, do this every 16 keys to limit the perf impact */ - if (g_pserver->m_pstorageFactory && (ckeysLoaded % 16) == 0) + if (g_pserver->m_pstorageFactory && (ckeysLoaded % 128) == 0) { - if (getMaxmemoryState(NULL,NULL,NULL,NULL) != C_OK || (ckeysLoaded % (1024)) == 0) + bool fHighMemory = (getMaxmemoryState(NULL,NULL,NULL,NULL) != C_OK); + if (fHighMemory || (ckeysLoaded % (1024)) == 0) { for (int idb = 0; idb < cserver.dbnum; ++idb) { if (g_pserver->db[idb]->processChanges(false)) g_pserver->db[idb]->commitChanges(); - g_pserver->db[idb]->trackChanges(false); + if (fHighMemory && !(rsi && rsi->fForceSetKey)) { + g_pserver->db[idb]->removeAllCachedValues(); // During load we don't go through the normal eviction unless we're merging (i.e. an active replica) + fHighMemory = false; // we took care of it + } + g_pserver->db[idb]->trackChanges(false, 1024); } - freeMemoryIfNeeded(false /*fQuickCycle*/, false /* fPreSnapshot*/); + if (fHighMemory) + freeMemoryIfNeeded(false /*fQuickCycle*/, false /* fPreSnapshot*/); } } diff --git a/src/server.h b/src/server.h index 2c962cb60..42a32ecde 100644 --- a/src/server.h +++ b/src/server.h @@ -1042,7 +1042,6 @@ class redisDbPersistentDataSnapshot; class redisDbPersistentData { friend void dictDbKeyDestructor(void *privdata, void *key); - friend void changedescDtor(void*, void*); friend class redisDbPersistentDataSnapshot; public: @@ -1110,7 +1109,7 @@ public: void setStorageProvider(StorageCache *pstorage); - void trackChanges(bool fBulk); + void trackChanges(bool fBulk, size_t sizeHint = 0); // Process and commit changes for secondary storage. Note that process and commit are seperated // to allow you to release the global lock before commiting. To prevent deadlocks you *must* @@ -1147,15 +1146,7 @@ protected: uint64_t m_mvccCheckpoint = 0; private: - struct changedesc - { - sdsimmutablestring strkey; - bool fUpdate; - - changedesc(const char *strkey, bool fUpdate) : strkey(strkey), fUpdate(fUpdate) {} - }; - - static void serializeAndStoreChange(StorageCache *storage, redisDbPersistentData *db, const changedesc &change); + static void serializeAndStoreChange(StorageCache *storage, redisDbPersistentData *db, const char *key, bool fUpdate); void ensure(const char *key); void ensure(const char *key, dictEntry **de);