Compare commits

..
12 changed files with 29 additions and 88 deletions

No files matched your search

@@ -88,7 +88,7 @@ After=network.target
Documentation=https://docs.keydb.dev, man:${BINARY}(1)
[Service]
Type=forking
Type=notify
ExecStart=/usr/bin/${BINARY} /etc/keydb/${NAMESPACED}.conf
ExecStop=/bin/kill -s TERM \$MAINPID
PIDFile=/var/run/${NAMESPACED}/${BINARY}.pid
@@ -90,7 +90,7 @@ After=network.target
Documentation=https://docs.keydb.dev, man:${BINARY}(1)
[Service]
Type=forking
Type=notify
ExecStart=/usr/bin/${BINARY} /etc/keydb/${NAMESPACED}.conf
ExecStop=/bin/kill -s TERM \$MAINPID
PIDFile=/var/run/${NAMESPACED}/${BINARY}.pid
@@ -4,7 +4,7 @@ After=network.target
Documentation=https://docs.keydb.dev, man:keydb-server(1)
[Service]
Type=forking
Type=notify
ExecStart=/usr/bin/keydb-server /etc/keydb/keydb.conf
ExecStop=/bin/kill -s TERM $MAINPID
PIDFile=/var/run/keydb/keydb-server.pid
+2 -3
View File
@@ -33,11 +33,10 @@ public:
virtual bool enumerate(callback fn) const = 0;
virtual size_t count() const = 0;
virtual void bulkInsert(char **rgkeys, size_t *rgcbkeys, char **rgvals, size_t *rgcbvals, char *rgfOverwrite, size_t celem) {
virtual void bulkInsert(char **rgkeys, size_t *rgcbkeys, char **rgvals, size_t *rgcbvals, size_t celem) {
beginWriteBatch();
for (size_t ielem = 0; ielem < celem; ++ielem) {
bool fOverwrite = (rgfOverwrite != nullptr) ? rgfOverwrite[ielem] : false;
insert(rgkeys[ielem], rgcbkeys[ielem], rgvals[ielem], rgcbvals[ielem], fOverwrite);
insert(rgkeys[ielem], rgcbkeys[ielem], rgvals[ielem], rgcbvals[ielem], false);
}
endWriteBatch();
}
+2 -3
View File
@@ -114,14 +114,13 @@ void StorageCache::insert(sds key, const void *data, size_t cbdata, bool fOverwr
}
long _dictKeyIndex(dict *d, const void *key, uint64_t hash, dictEntry **existing);
void StorageCache::bulkInsert(char **rgkeys, size_t *rgcbkeys, char **rgvals, size_t *rgcbvals, char *rgfOverwrite, size_t celem)
void StorageCache::bulkInsert(char **rgkeys, size_t *rgcbkeys, char **rgvals, size_t *rgcbvals, size_t celem)
{
std::vector<dictEntry*> vechashes;
if (m_pdict != nullptr) {
vechashes.reserve(celem);
for (size_t ielem = 0; ielem < celem; ++ielem) {
if (rgfOverwrite != nullptr && rgfOverwrite[ielem]) continue;
dictEntry *de = (dictEntry*)zmalloc(sizeof(dictEntry));
de->key = (void*)dictGenHashFunction(rgkeys[ielem], (int)rgcbkeys[ielem]);
de->v.u64 = 1;
@@ -153,7 +152,7 @@ void StorageCache::bulkInsert(char **rgkeys, size_t *rgcbkeys, char **rgvals, si
}
ul.unlock();
m_spstorage->bulkInsert(rgkeys, rgcbkeys, rgvals, rgcbvals, rgfOverwrite, celem);
m_spstorage->bulkInsert(rgkeys, rgcbkeys, rgvals, rgcbvals, celem);
bulkInsertsInProgress--;
}
+1 -1
View File
@@ -41,7 +41,7 @@ public:
void clear(void(callback)(void*));
void clearAsync();
void insert(sds key, const void *data, size_t cbdata, bool fOverwrite);
void bulkInsert(char **rgkeys, size_t *rgcbkeys, char **rgvals, size_t *rgcbvals, char *rgfOverwrite, size_t celem);
void bulkInsert(char **rgkeys, size_t *rgcbkeys, char **rgvals, size_t *rgcbvals, size_t celem);
void retrieve(sds key, IStorage::callbackSingle fn) const;
bool erase(sds key);
void emergencyFreeCache();
+16 -57
View File
@@ -2607,15 +2607,12 @@ void clusterStorageLoadCallback(const char *rgchkey, size_t cch, void *)
slotToKeyUpdateKeyCore(rgchkey, cch, true /*add*/);
}
void moduleLoadCallback(const char * rgchKey, size_t, void *data) {
redisObjectStack keyobj;
initStaticStringObject(keyobj, const_cast<char *>(rgchKey));
moduleNotifyKeyspaceEvent(NOTIFY_LOADED, "loaded", &keyobj, *(int *)data);
}
void moduleClusterLoadCallback(const char * rgchKey, size_t cchKey, void *data) {
clusterStorageLoadCallback(rgchKey, cchKey, data);
moduleLoadCallback(rgchKey, cchKey, data);
void moduleLoadCallback(const char * rgchKey, size_t cchKey, void *data) {
if (g_pserver->cluster_enabled) {
clusterStorageLoadCallback(rgchKey, cchKey, data);
}
robj *keyobj = createEmbeddedStringObject(rgchKey, cchKey);
moduleNotifyKeyspaceEvent(NOTIFY_LOADED, "loaded", keyobj, *(int *)data);
}
void redisDb::initialize(int id)
@@ -2636,7 +2633,7 @@ void redisDb::storageProviderInitialize()
{
if (g_pserver->m_pstorageFactory != nullptr)
{
IStorageFactory::key_load_iterator itr = g_pserver->cluster_enabled ? moduleClusterLoadCallback : moduleLoadCallback;
IStorageFactory::key_load_iterator itr = moduleLoadCallback;
this->setStorageProvider(StorageCache::create(g_pserver->m_pstorageFactory, id, itr, &id));
}
}
@@ -2908,13 +2905,15 @@ void redisDbPersistentData::storeDatabase()
dictReleaseIterator(di);
}
/* static */ sds redisDbPersistentData::serializeChange(redisDbPersistentData *db, const char *key)
/* static */ void redisDbPersistentData::serializeAndStoreChange(StorageCache *storage, redisDbPersistentData *db, const char *key, bool fUpdate)
{
auto itr = db->find_cached_threadsafe(key);
if (itr == nullptr)
return nullptr;
return;
robj *o = itr.val();
return serializeStoredObjectAndExpire(db, (const char*) itr.key(), o);
sds temp = serializeStoredObjectAndExpire(db, (const char*) itr.key(), o);
storage->insert((sds)key, temp, sdslen(temp), fUpdate);
sdsfree(temp);
}
bool redisDbPersistentData::processChanges(bool fSnapshot)
@@ -2957,30 +2956,10 @@ bool redisDbPersistentData::processChanges(bool fSnapshot)
{
dictIterator *di = dictGetIterator(m_dictChanged);
dictEntry *de;
std::vector<char*> veckeys;
std::vector<size_t> veccbkeys;
std::vector<char*> vecvals;
std::vector<size_t> veccbvals;
std::vector<char> vecoverwrite;
veckeys.reserve(dictSize(m_dictChanged));
veccbkeys.reserve(dictSize(m_dictChanged));
vecvals.reserve(dictSize(m_dictChanged));
veccbvals.reserve(dictSize(m_dictChanged));
vecoverwrite.reserve(dictSize(m_dictChanged));
while ((de = dictNext(di)) != nullptr)
{
sds val = serializeChange(this, (const char*)dictGetKey(de));
if (val != nullptr) {
veckeys.push_back((char*)dictGetKey(de));
veccbkeys.push_back(sdslen((sds)dictGetKey(de)));
vecvals.push_back(val);
veccbvals.push_back(sdslen(val));
vecoverwrite.push_back((bool)dictGetVal(de));
}
serializeAndStoreChange(m_spstorage.get(), this, (const char*)dictGetKey(de), (bool)dictGetVal(de));
}
m_spstorage->bulkInsert(veckeys.data(), veccbkeys.data(), vecvals.data(), veccbvals.data(), vecoverwrite.data(), veckeys.size());
for (auto val : vecvals)
sdsfree(val);
dictReleaseIterator(di);
}
}
@@ -3014,7 +2993,7 @@ void redisDbPersistentData::processChangesAsync(std::atomic<int> &pendingJobs)
vecvals.push_back(temp);
veccbvals.push_back(sdslen(temp));
}
m_spstorage->bulkInsert(veckeys.data(), veccbkeys.data(), vecvals.data(), veccbvals.data(), nullptr, veckeys.size());
m_spstorage->bulkInsert(veckeys.data(), veccbkeys.data(), vecvals.data(), veccbvals.data(), veckeys.size());
for (auto val : vecvals)
sdsfree(val);
dictReleaseIterator(di);
@@ -3033,7 +3012,7 @@ void redisDbPersistentData::bulkDirectStorageInsert(char **rgKeys, size_t *rgcbK
}
aeReleaseLock();
}
m_spstorage->bulkInsert(rgKeys, rgcbKeys, rgVals, rgcbVals, nullptr, celem);
m_spstorage->bulkInsert(rgKeys, rgcbKeys, rgVals, rgcbVals, celem);
}
void redisDbPersistentData::commitChanges(const redisDbPersistentDataSnapshot **psnapshotFree)
@@ -3042,30 +3021,10 @@ void redisDbPersistentData::commitChanges(const redisDbPersistentDataSnapshot **
{
dictIterator *di = dictGetIterator(m_dictChangedStorageFlush);
dictEntry *de;
std::vector<char*> veckeys;
std::vector<size_t> veccbkeys;
std::vector<char*> vecvals;
std::vector<size_t> veccbvals;
std::vector<char> vecoverwrite;
veckeys.reserve(dictSize(m_dictChanged));
veccbkeys.reserve(dictSize(m_dictChanged));
vecvals.reserve(dictSize(m_dictChanged));
veccbvals.reserve(dictSize(m_dictChanged));
vecoverwrite.resize(dictSize(m_dictChanged));
while ((de = dictNext(di)) != nullptr)
{
sds val = serializeChange((redisDbPersistentData*)m_pdbSnapshotStorageFlush, (const char*)dictGetKey(de));
if (val != nullptr) {
veckeys.push_back((char*)dictGetKey(de));
veccbkeys.push_back(sdslen((sds)dictGetKey(de)));
vecvals.push_back(val);
veccbvals.push_back(sdslen(val));
vecoverwrite.push_back((bool)dictGetVal(de));
}
serializeAndStoreChange(m_spstorage.get(), (redisDbPersistentData*)m_pdbSnapshotStorageFlush, (const char*)dictGetKey(de), (bool)dictGetVal(de));
}
m_spstorage->bulkInsert(veckeys.data(), veccbkeys.data(), vecvals.data(), veccbvals.data(), vecoverwrite.data(), veckeys.size());
for (auto val : vecvals)
sdsfree(val);
dictReleaseIterator(di);
dictRelease(m_dictChangedStorageFlush);
m_dictChangedStorageFlush = nullptr;
-3
View File
@@ -1264,9 +1264,6 @@ void scriptingInit(int setup) {
/* Finally set the table as 'redis' global var. */
lua_setglobal(lua,"redis");
/* Set table as 'keydb' global var as well */
lua_getglobal(lua,"redis");
lua_setglobal(lua,"keydb");
/* Replace math.random and math.randomseed with our implementations. */
lua_getglobal(lua,"math");
+1 -1
View File
@@ -1217,7 +1217,7 @@ protected:
uint64_t m_mvccCheckpoint = 0;
private:
static sds serializeChange(redisDbPersistentData *db, const char *key);
static void serializeAndStoreChange(StorageCache *storage, redisDbPersistentData *db, const char *key, bool fUpdate);
void ensure(const char *key);
void ensure(const char *key, dictEntry **de);
+2 -10
View File
@@ -43,9 +43,8 @@ void RocksDBStorageProvider::insert(const char *key, size_t cchKey, void *data,
++m_count;
}
void RocksDBStorageProvider::bulkInsert(char **rgkeys, size_t *rgcbkeys, char **rgvals, size_t *rgcbvals, char *rgfOverwrite, size_t celem)
void RocksDBStorageProvider::bulkInsert(char **rgkeys, size_t *rgcbkeys, char **rgvals, size_t *rgcbvals, size_t celem)
{
size_t coverwrites = 0;
if (celem >= 16384) {
rocksdb::Options options = DefaultRocksDBOptions();
rocksdb::SstFileWriter sst_file_writer(rocksdb::EnvOptions(), options, options.comparator);
@@ -93,15 +92,8 @@ void RocksDBStorageProvider::bulkInsert(char **rgkeys, size_t *rgcbkeys, char **
m_spdb->Write(WriteOptions(), spbatch.get());
}
if (rgfOverwrite != nullptr) {
for (size_t ielem = 0; ielem < celem; ++ielem) {
if (rgfOverwrite[ielem])
++coverwrites;
}
}
std::unique_lock<fastlock> l(m_lock);
m_count += celem - coverwrites;
m_count += celem;
}
bool RocksDBStorageProvider::erase(const char *key, size_t cchKey)
+1 -1
View File
@@ -38,7 +38,7 @@ public:
virtual void beginWriteBatch() override;
virtual void endWriteBatch() override;
virtual void bulkInsert(char **rgkeys, size_t *rgcbkeys, char **rgvals, size_t *rgcbvals, char *rgfOverwrite, size_t celem) override;
virtual void bulkInsert(char **rgkeys, size_t *rgcbkeys, char **rgvals, size_t *rgcbvals, size_t celem) override;
virtual void batch_lock() override;
virtual void batch_unlock() override;
+1 -6
View File
@@ -43,15 +43,10 @@ start_server {tags {"scripting"}} {
r eval {return redis.call('get',KEYS[1])} 1 mykey
} {myval}
test {EVAL - keys command works? } {
test { EVAL - keys command works? } {
r eval {return redis.call('keys', 'test')} 0
}
test {EVAL - KeyDB global works } {
r eval {return keydb.call('get', KEYS[1])} 1 mykey
assert_equal [r eval {return redis.call('get',KEYS[1])} 1 mykey] [r eval {return keydb.call('get', KEYS[1])} 1 mykey]
}
test {EVALSHA - Can we call a SHA1 if already defined?} {
r evalsha fd758d1589d044dd850a6f05d52f2eefd27f033f 1 mykey
} {myval}