Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
b0728af6aa |
No files matched your search
+2
-3
@@ -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();
|
||||
}
|
||||
|
||||
@@ -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
@@ -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();
|
||||
|
||||
+9
-47
@@ -2908,13 +2908,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 +2959,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 +2996,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 +3015,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 +3024,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;
|
||||
|
||||
@@ -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
@@ -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
@@ -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)
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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}
|
||||
|
||||
Reference in new issue
Block a user