Compare commits

..
25 changed files with 149 additions and 254 deletions

No files matched your search

+3 -2
View File
@@ -33,10 +33,11 @@ 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, size_t celem) {
virtual void bulkInsert(char **rgkeys, size_t *rgcbkeys, char **rgvals, size_t *rgcbvals, char *rgfOverwrite, size_t celem) {
beginWriteBatch();
for (size_t ielem = 0; ielem < celem; ++ielem) {
insert(rgkeys[ielem], rgcbkeys[ielem], rgvals[ielem], rgcbvals[ielem], false);
bool fOverwrite = (rgfOverwrite != nullptr) ? rgfOverwrite[ielem] : false;
insert(rgkeys[ielem], rgcbkeys[ielem], rgvals[ielem], rgcbvals[ielem], fOverwrite);
}
endWriteBatch();
}
+1 -9
View File
@@ -145,9 +145,7 @@ FINAL_LDFLAGS=$(LDFLAGS) $(KEYDB_LDFLAGS) $(DEBUG)
FINAL_LIBS+=-lm -lz -lcrypto -lbz2 -lzstd -llz4 -lsnappy
ifneq ($(uname_S),Darwin)
ifneq ($(uname_S),FreeBSD)
FINAL_LIBS+=-latomic
endif
endif
# Linux ARM32 needs -latomic at linking time
ifneq (,$(findstring armv,$(uname_M)))
@@ -213,13 +211,7 @@ ifeq ($(uname_S),NetBSD)
else
ifeq ($(uname_S),FreeBSD)
# FreeBSD
FINAL_LIBS+= -lpthread -luuid -lexecinfo
FINAL_CFLAGS+= -I/usr/local/include
FINAL_CXXFLAGS+= -I/usr/local/include
FINAL_LDFLAGS+= -L/usr/local/lib
ifeq ($(USE_BACKTRACE),yes)
FINAL_CFLAGS+= -DUSE_BACKTRACE
endif
FINAL_LIBS+= -lpthread -lexecinfo
else
ifeq ($(uname_S),DragonFly)
# DragonFly
+4 -3
View File
@@ -21,7 +21,7 @@ dictType dbStorageCacheType = {
StorageCache::StorageCache(IStorage *storage, bool fCache)
: m_spstorage(storage)
{
if (!g_pserver->flash_disable_key_cache && fCache)
if (fCache)
m_pdict = dictCreate(&dbStorageCacheType, nullptr);
}
@@ -114,13 +114,14 @@ 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, size_t celem)
void StorageCache::bulkInsert(char **rgkeys, size_t *rgcbkeys, char **rgvals, size_t *rgcbvals, char *rgfOverwrite, 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;
@@ -152,7 +153,7 @@ void StorageCache::bulkInsert(char **rgkeys, size_t *rgcbkeys, char **rgvals, si
}
ul.unlock();
m_spstorage->bulkInsert(rgkeys, rgcbkeys, rgvals, rgcbvals, celem);
m_spstorage->bulkInsert(rgkeys, rgcbkeys, rgvals, rgcbvals, rgfOverwrite, 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, size_t celem);
void bulkInsert(char **rgkeys, size_t *rgcbkeys, char **rgvals, size_t *rgcbvals, char *rgfOverwrite, size_t celem);
void retrieve(sds key, IStorage::callbackSingle fn) const;
bool erase(sds key);
void emergencyFreeCache();
+1 -3
View File
@@ -2806,7 +2806,7 @@ standardConfig configs[] = {
createBoolConfig("appendonly", NULL, MODIFIABLE_CONFIG, g_pserver->aof_enabled, 0, NULL, updateAppendonly),
createBoolConfig("cluster-allow-reads-when-down", NULL, MODIFIABLE_CONFIG, g_pserver->cluster_allow_reads_when_down, 0, NULL, NULL),
createBoolConfig("delete-on-evict", NULL, MODIFIABLE_CONFIG, cserver.delete_on_evict, 0, NULL, NULL),
createBoolConfig("use-fork", NULL, IMMUTABLE_CONFIG, cserver.fForkBgSave, 1, NULL, NULL),
createBoolConfig("use-fork", NULL, IMMUTABLE_CONFIG, cserver.fForkBgSave, 0, NULL, NULL),
createBoolConfig("io-threads-do-reads", NULL, IMMUTABLE_CONFIG, fDummy, 0, NULL, NULL),
createBoolConfig("time-thread-priority", NULL, IMMUTABLE_CONFIG, cserver.time_thread_priority, 0, NULL, NULL),
createBoolConfig("prefetch-enabled", NULL, MODIFIABLE_CONFIG, g_pserver->prefetch_enabled, 1, NULL, NULL),
@@ -2912,7 +2912,6 @@ standardConfig configs[] = {
createLongLongConfig("repl-backlog-size", NULL, MODIFIABLE_CONFIG, 1, LLONG_MAX, g_pserver->repl_backlog_size, 1024*1024, MEMORY_CONFIG, NULL, updateReplBacklogSize), /* Default: 1mb */
createLongLongConfig("repl-backlog-disk-reserve", NULL, IMMUTABLE_CONFIG, 0, LLONG_MAX, cserver.repl_backlog_disk_size, 0, MEMORY_CONFIG, NULL, NULL),
createLongLongConfig("max-snapshot-slip", NULL, MODIFIABLE_CONFIG, 0, 5000, g_pserver->snapshot_slip, 400, 0, NULL, NULL),
createLongLongConfig("max-rand-count", NULL, MODIFIABLE_CONFIG, 0, LONG_MAX/2, g_pserver->rand_total_threshold, LONG_MAX/2, 0, NULL, NULL),
/* Unsigned Long Long configs */
createULongLongConfig("maxmemory", NULL, MODIFIABLE_CONFIG, 0, LLONG_MAX, g_pserver->maxmemory, 0, MEMORY_CONFIG, NULL, updateMaxmemory),
@@ -2940,7 +2939,6 @@ standardConfig configs[] = {
createBoolConfig("allow-write-during-load", NULL, MODIFIABLE_CONFIG, g_pserver->fWriteDuringActiveLoad, 0, NULL, NULL),
createBoolConfig("force-backlog-disk-reserve", NULL, MODIFIABLE_CONFIG, cserver.force_backlog_disk, 0, NULL, NULL),
createBoolConfig("soft-shutdown", NULL, MODIFIABLE_CONFIG, g_pserver->config_soft_shutdown, 0, NULL, NULL),
createBoolConfig("flash-disable-key-cache", NULL, MODIFIABLE_CONFIG, g_pserver->flash_disable_key_cache, 0, NULL, NULL),
createSizeTConfig("semi-ordered-set-bucket-size", NULL, MODIFIABLE_CONFIG, 0, 1024, g_semiOrderedSetTargetBucketSize, 0, INTEGER_CONFIG, NULL, NULL),
#ifdef USE_OPENSSL
+51 -31
View File
@@ -1059,7 +1059,7 @@ void keysCommand(client *c) {
sds pattern = szFromObj(c->argv[1]);
const redisDbPersistentDataSnapshot *snapshot = nullptr;
if (!(c->flags & (CLIENT_MULTI | CLIENT_BLOCKED | CLIENT_DENY_BLOCKING)) && !(serverTL->in_eval || serverTL->in_exec))
if (!(c->flags & (CLIENT_MULTI | CLIENT_BLOCKED | CLIENT_DENY_BLOCKING)))
snapshot = c->db->createSnapshot(c->mvccCheckpoint, true /* fOptional */);
if (snapshot != nullptr)
{
@@ -1224,7 +1224,7 @@ void scanGenericCommand(client *c, robj_roptr o, unsigned long cursor) {
}
}
if (o == nullptr && count >= 100 && !(serverTL->in_eval || serverTL->in_exec))
if (o == nullptr && count >= 100)
{
// Do an async version
if (c->asyncCommand(
@@ -1737,9 +1737,6 @@ int dbSwapDatabases(int id1, int id2) {
id2 < 0 || id2 >= cserver.dbnum) return C_ERR;
if (id1 == id2) return C_OK;
std::swap(g_pserver->db[id1], g_pserver->db[id2]);
//swap db's id too, otherwise db does not match its id
std::swap(g_pserver->db[id1]->id, g_pserver->db[id2]->id);
/* Note that we don't swap blocking_keys,
* ready_keys and watched_keys, since we want clients to
@@ -1769,7 +1766,7 @@ int dbSwapDatabases(int id1, int id2) {
/* SWAPDB db1 db2 */
void swapdbCommand(client *c) {
int id1, id2, oriId;
int id1, id2;
/* Not allowed in cluster mode: we have just DB 0 there. */
if (g_pserver->cluster_enabled) {
@@ -1786,9 +1783,6 @@ void swapdbCommand(client *c) {
"invalid second DB index") != C_OK)
return;
//get client's original db's id
oriId=c->db->id;
/* Swap... */
if (dbSwapDatabases(id1,id2) == C_ERR) {
addReplyError(c,"DB index is out of range");
@@ -1797,10 +1791,6 @@ void swapdbCommand(client *c) {
RedisModuleSwapDbInfo si = {REDISMODULE_SWAPDBINFO_VERSION,(int32_t)id1,(int32_t)id2};
moduleFireServerEvent(REDISMODULE_EVENT_SWAPDB,0,&si);
g_pserver->dirty++;
//set client's db to original db
c->db=g_pserver->db[oriId];
addReply(c,shared.ok);
}
}
@@ -1941,7 +1931,6 @@ 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;
@@ -2738,11 +2727,8 @@ 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);
@@ -2875,8 +2861,7 @@ 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);
@@ -2899,7 +2884,6 @@ 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()));
}
}
@@ -2924,15 +2908,13 @@ void redisDbPersistentData::storeDatabase()
dictReleaseIterator(di);
}
/* static */ void redisDbPersistentData::serializeAndStoreChange(StorageCache *storage, redisDbPersistentData *db, const char *key, bool fUpdate)
/* static */ sds redisDbPersistentData::serializeChange(redisDbPersistentData *db, const char *key)
{
auto itr = db->find_cached_threadsafe(key);
if (itr == nullptr)
return;
return nullptr;
robj *o = itr.val();
sds temp = serializeStoredObjectAndExpire(db, (const char*) itr.key(), o);
storage->insert((sds)key, temp, sdslen(temp), fUpdate);
sdsfree(temp);
return serializeStoredObjectAndExpire(db, (const char*) itr.key(), o);
}
bool redisDbPersistentData::processChanges(bool fSnapshot)
@@ -2975,10 +2957,30 @@ 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)
{
serializeAndStoreChange(m_spstorage.get(), this, (const char*)dictGetKey(de), (bool)dictGetVal(de));
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));
}
}
m_spstorage->bulkInsert(veckeys.data(), veccbkeys.data(), vecvals.data(), veccbvals.data(), vecoverwrite.data(), veckeys.size());
for (auto val : vecvals)
sdsfree(val);
dictReleaseIterator(di);
}
}
@@ -3012,7 +3014,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(), veckeys.size());
m_spstorage->bulkInsert(veckeys.data(), veccbkeys.data(), vecvals.data(), veccbvals.data(), nullptr, veckeys.size());
for (auto val : vecvals)
sdsfree(val);
dictReleaseIterator(di);
@@ -3031,7 +3033,7 @@ void redisDbPersistentData::bulkDirectStorageInsert(char **rgKeys, size_t *rgcbK
}
aeReleaseLock();
}
m_spstorage->bulkInsert(rgKeys, rgcbKeys, rgVals, rgcbVals, celem);
m_spstorage->bulkInsert(rgKeys, rgcbKeys, rgVals, rgcbVals, nullptr, celem);
}
void redisDbPersistentData::commitChanges(const redisDbPersistentDataSnapshot **psnapshotFree)
@@ -3040,10 +3042,30 @@ 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)
{
serializeAndStoreChange(m_spstorage.get(), (redisDbPersistentData*)m_pdbSnapshotStorageFlush, (const char*)dictGetKey(de), (bool)dictGetVal(de));
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));
}
}
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;
@@ -3272,7 +3294,6 @@ 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())
@@ -3397,7 +3418,6 @@ 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);
+3 -4
View File
@@ -380,7 +380,7 @@ size_t freeMemoryGetNotCountedMemory(void) {
/* also don't count the replication backlog memory
* that's where the replication clients get their memory from */
overhead += g_pserver->repl_backlog_size - g_pserver->repl_backlog_config_size;
overhead += g_pserver->repl_backlog_size;
if (g_pserver->aof_state != AOF_OFF) {
overhead += sdsalloc(g_pserver->aof_buf)+aofRewriteBufferSize();
@@ -552,7 +552,7 @@ static int evictionTimeProc(
UNUSED(clientData);
serverAssert(GlobalLocksAcquired());
if (performEvictions((bool)clientData) == EVICT_RUNNING) return 0; /* keep evicting */
if (performEvictions(false) == EVICT_RUNNING) return 0; /* keep evicting */
/* For EVICT_OK - things are good, no need to keep evicting.
* For EVICT_FAIL - there is nothing left to evict. */
@@ -771,7 +771,6 @@ int performEvictions(bool fPreSnapshot) {
if (db->removeCachedValue(bestkey, &deT)) {
mem_freed += splazy->addEntry(db->dictUnsafeKeyOnly(), deT);
ckeysFailed = 0;
g_pserver->stat_evictedkeys++;
}
else {
delta = 0;
@@ -838,7 +837,7 @@ int performEvictions(bool fPreSnapshot) {
if (!isEvictionProcRunning && serverTL->el != nullptr) {
isEvictionProcRunning = 1;
aeCreateTimeEvent(serverTL->el, 0,
evictionTimeProc, (void*)fPreSnapshot, NULL);
evictionTimeProc, NULL, NULL);
}
break;
}
-1
View File
@@ -481,7 +481,6 @@ 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) {
+2 -2
View File
@@ -656,7 +656,7 @@ void moduleHandlePropagationAfterCommandCallback(RedisModuleCtx *ctx) {
/* We don't need to do anything here if the server isn't inside
* a transaction. */
if (!serverTL->propagate_in_transaction) return;
if (!g_pserver->propagate_in_transaction) return;
/* If this command is executed from with Lua or MULTI/EXEC we do not
* need to propagate EXEC */
@@ -1814,7 +1814,7 @@ void moduleReplicateMultiIfNeeded(RedisModuleCtx *ctx) {
* the module command was called by a script. */
if (serverTL->in_eval || serverTL->in_exec) return;
/* If we already emitted MULTI return ASAP. */
if (serverTL->propagate_in_transaction) return;
if (g_pserver->propagate_in_transaction) return;
/* If this is a thread safe context, we do not want to wrap commands
* executed into MULTI/EXEC, they are executed as single commands
* from an external client in essence. */
+5 -5
View File
@@ -118,14 +118,14 @@ void discardCommand(client *c) {
void beforePropagateMulti() {
/* Propagating MULTI */
serverAssert(!serverTL->propagate_in_transaction);
serverTL->propagate_in_transaction = 1;
serverAssert(!g_pserver->propagate_in_transaction);
g_pserver->propagate_in_transaction = 1;
}
void afterPropagateExec() {
/* Propagating EXEC */
serverAssert(serverTL->propagate_in_transaction == 1);
serverTL->propagate_in_transaction = 0;
serverAssert(g_pserver->propagate_in_transaction == 1);
g_pserver->propagate_in_transaction = 0;
}
/* Send a MULTI command to all the slaves and AOF file. Check the execCommand
@@ -264,7 +264,7 @@ void execCommand(client *c) {
/* Make sure the EXEC command will be propagated as well if MULTI
* was already propagated. */
if (serverTL->propagate_in_transaction) {
if (g_pserver->propagate_in_transaction) {
int is_master = listLength(g_pserver->masters) == 0;
g_pserver->dirty++;
/* If inside the MULTI/EXEC block this instance was suddenly
+2 -2
View File
@@ -2753,7 +2753,7 @@ void readQueryFromClient(connection *conn) {
if (cserver.cthreads > 1 || g_pserver->m_pstorageFactory) {
parseClientCommandBuffer(c);
if (g_pserver->enable_async_commands && !serverTL->disable_async_commands && listLength(g_pserver->monitors) == 0 && (aeLockContention() || serverTL->rgdbSnapshot[c->db->id] || g_fTestMode) && !serverTL->in_eval && !serverTL->in_exec) {
if (g_pserver->enable_async_commands && !serverTL->disable_async_commands && listLength(g_pserver->monitors) == 0 && (aeLockContention() || serverTL->rgdbSnapshot[c->db->id] || g_fTestMode)) {
// Frequent writers aren't good candidates for this optimization, they cause us to renew the snapshot too often
// so we exclude them unless the snapshot we need already exists.
// Note: In test mode we want to create snapshots as often as possibl to excercise them - we don't care about perf
@@ -3995,7 +3995,7 @@ void pauseClients(mstime_t end, pause_type type) {
* to track this state so that we don't assert
* in propagate(). */
if (serverTL->in_exec) {
serverTL->client_pause_in_transaction = 1;
g_pserver->client_pause_in_transaction = 1;
}
}
-1
View File
@@ -1668,7 +1668,6 @@ 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;
}
+47 -120
View File
@@ -1334,7 +1334,7 @@ int rdbSaveRio(rio *rdb, const redisDbPersistentDataSnapshot **rgpdb, int *error
if (rdbSaveModulesAux(rdb, REDISMODULE_AUX_BEFORE_RDB) == -1) goto werr;
for (j = 0; j < cserver.dbnum; j++) {
const redisDbPersistentDataSnapshot *db = rgpdb != nullptr ? rgpdb[j] : g_pserver->db[j];
const redisDbPersistentDataSnapshot *db = rgpdb[j];
if (db->size() == 0) continue;
/* Write the SELECT DB opcode */
@@ -1715,7 +1715,7 @@ void getTempFileName(char tmpfile[], int tmpfileNum) {
char tmpfileNumString[214];
/* Generate temp rdb file name using aync-signal safe functions. */
int pid_len = ll2string(pid, sizeof(pid), g_pserver->in_fork_child ? getppid() : getpid());
int pid_len = ll2string(pid, sizeof(pid), getpid());
int tmpfileNum_len = ll2string(tmpfileNumString, sizeof(tmpfileNumString), tmpfileNum);
strcpy(tmpfile, "temp-");
strncpy(tmpfile+5, pid, pid_len);
@@ -3832,130 +3832,57 @@ int rdbSaveToSlavesSockets(rdbSaveInfo *rsi) {
}
/* Create the child process. */
if (cserver.fForkBgSave) {
pid_t childpid;
if ((childpid = redisFork(CHILD_TYPE_RDB)) == 0) {
/* Child */
int retval, dummy;
rio rdb;
openChildInfoPipe();
rioInitWithFd(&rdb,args->rdb_pipe_write);
for (int idb = 0; idb < cserver.dbnum; ++idb)
args->rgpdb[idb] = g_pserver->db[idb]->createSnapshot(getMvccTstamp(), false /*fOptional*/);
redisSetProcTitle("keydb-rdb-to-slaves");
redisSetCpuAffinity(g_pserver->bgsave_cpulist);
retval = rdbSaveRioWithEOFMark(&rdb,nullptr,nullptr,rsi);
if (retval == C_OK && rioFlush(&rdb) == 0)
retval = C_ERR;
if (retval == C_OK) {
sendChildCowInfo(CHILD_INFO_TYPE_RDB_COW_SIZE, "RDB");
}
rioFreeFd(&rdb);
/* wake up the reader, tell it we're done. */
close(args->rdb_pipe_write);
close(g_pserver->rdb_child_exit_pipe); /* close write end so that we can detect the close on the parent. */
/* hold exit until the parent tells us it's safe. we're not expecting
* to read anything, just get the error when the pipe is closed. */
dummy = read(args->safe_to_exit_pipe, pipefds, 1);
UNUSED(dummy);
exitFromChild((retval == C_OK) ? 0 : 1);
} else {
/* Parent */
close(args->safe_to_exit_pipe);
if (childpid == -1) {
serverLog(LL_WARNING,"Can't save in background: fork: %s",
strerror(errno));
/* Undo the state change. The caller will perform cleanup on
* all the slaves in BGSAVE_START state, but an early call to
* replicationSetupSlaveForFullResync() turned it into BGSAVE_END */
listRewind(g_pserver->slaves,&li);
while((ln = listNext(&li))) {
client *replica = (client*)ln->value;
if (replica->replstate == SLAVE_STATE_WAIT_BGSAVE_END) {
replica->replstate = SLAVE_STATE_WAIT_BGSAVE_START;
}
}
close(args->rdb_pipe_write);
close(g_pserver->rdb_pipe_read);
zfree(g_pserver->rdb_pipe_conns);
g_pserver->rdb_pipe_conns = NULL;
g_pserver->rdb_pipe_numconns = 0;
g_pserver->rdb_pipe_numconns_writing = 0;
args->rsi.~rdbSaveInfo();
zfree(args);
} else {
serverLog(LL_NOTICE,"Background RDB transfer started by pid %ld",
(long)childpid);
g_pserver->rdb_save_time_start = time(NULL);
g_pserver->rdb_child_type = RDB_CHILD_TYPE_SOCKET;
g_pserver->rdbThreadVars.fRdbThreadActive = true;
updateDictResizePolicy();
close(args->rdb_pipe_write); /* close write in parent so that it can detect the close on the child. */
aePostFunction(g_pserver->rgthreadvar[IDX_EVENT_LOOP_MAIN].el, []{
if (aeCreateFileEvent(serverTL->el, g_pserver->rdb_pipe_read, AE_READABLE, rdbPipeReadHandler, nullptr) == AE_ERR) {
serverPanic("Unrecoverable error creating g_pserver->rdb_pipe_read file event.");
}
});
}
return (childpid == -1) ? C_ERR : C_OK;
}
}
else {
openChildInfoPipe();
for (int idb = 0; idb < cserver.dbnum; ++idb)
args->rgpdb[idb] = g_pserver->db[idb]->createSnapshot(getMvccTstamp(), false /*fOptional*/);
g_pserver->rdbThreadVars.tmpfileNum++;
g_pserver->rdbThreadVars.fRdbThreadCancel = false;
pthread_attr_t tattr;
pthread_attr_init(&tattr);
pthread_attr_setstacksize(&tattr, 1 << 23); // 8 MB
if (pthread_create(&child, &tattr, rdbSaveToSlavesSocketsThread, args)) {
pthread_attr_destroy(&tattr);
serverLog(LL_WARNING,"Can't save in background: fork: %s",
strerror(errno));
/* Undo the state change. The caller will perform cleanup on
* all the slaves in BGSAVE_START state, but an early call to
* replicationSetupSlaveForFullResync() turned it into BGSAVE_END */
listRewind(g_pserver->slaves,&li);
while((ln = listNext(&li))) {
client *replica = (client*)ln->value;
if (replica->replstate == SLAVE_STATE_WAIT_BGSAVE_END) {
replica->replstate = SLAVE_STATE_WAIT_BGSAVE_START;
}
}
close(args->rdb_pipe_write);
close(g_pserver->rdb_pipe_read);
zfree(g_pserver->rdb_pipe_conns);
close(args->safe_to_exit_pipe);
g_pserver->rdb_pipe_conns = NULL;
g_pserver->rdb_pipe_numconns = 0;
g_pserver->rdb_pipe_numconns_writing = 0;
args->rsi.~rdbSaveInfo();
zfree(args);
closeChildInfoPipe();
return C_ERR;
}
g_pserver->rdbThreadVars.tmpfileNum++;
g_pserver->rdbThreadVars.fRdbThreadCancel = false;
pthread_attr_t tattr;
pthread_attr_init(&tattr);
pthread_attr_setstacksize(&tattr, 1 << 23); // 8 MB
if (pthread_create(&child, &tattr, rdbSaveToSlavesSocketsThread, args)) {
pthread_attr_destroy(&tattr);
g_pserver->child_type = CHILD_TYPE_RDB;
serverLog(LL_WARNING,"Can't save in background: fork: %s",
strerror(errno));
serverLog(LL_NOTICE,"Background RDB transfer started");
g_pserver->rdb_save_time_start = time(NULL);
serverAssert(!g_pserver->rdbThreadVars.fRdbThreadActive);
g_pserver->rdbThreadVars.rdb_child_thread = child;
g_pserver->rdbThreadVars.fRdbThreadActive = true;
g_pserver->rdb_child_type = RDB_CHILD_TYPE_SOCKET;
aePostFunction(g_pserver->rgthreadvar[IDX_EVENT_LOOP_MAIN].el, []{
if (aeCreateFileEvent(serverTL->el, g_pserver->rdb_pipe_read, AE_READABLE, rdbPipeReadHandler, nullptr) == AE_ERR) {
serverPanic("Unrecoverable error creating server.rdb_pipe_read file event.");
/* Undo the state change. The caller will perform cleanup on
* all the slaves in BGSAVE_START state, but an early call to
* replicationSetupSlaveForFullResync() turned it into BGSAVE_END */
listRewind(g_pserver->slaves,&li);
while((ln = listNext(&li))) {
client *replica = (client*)ln->value;
if (replica->replstate == SLAVE_STATE_WAIT_BGSAVE_END) {
replica->replstate = SLAVE_STATE_WAIT_BGSAVE_START;
}
});
}
close(args->rdb_pipe_write);
close(g_pserver->rdb_pipe_read);
zfree(g_pserver->rdb_pipe_conns);
close(args->safe_to_exit_pipe);
g_pserver->rdb_pipe_conns = NULL;
g_pserver->rdb_pipe_numconns = 0;
g_pserver->rdb_pipe_numconns_writing = 0;
args->rsi.~rdbSaveInfo();
zfree(args);
closeChildInfoPipe();
return C_ERR;
}
pthread_attr_destroy(&tattr);
g_pserver->child_type = CHILD_TYPE_RDB;
serverLog(LL_NOTICE,"Background RDB transfer started");
g_pserver->rdb_save_time_start = time(NULL);
serverAssert(!g_pserver->rdbThreadVars.fRdbThreadActive);
g_pserver->rdbThreadVars.rdb_child_thread = child;
g_pserver->rdbThreadVars.fRdbThreadActive = true;
g_pserver->rdb_child_type = RDB_CHILD_TYPE_SOCKET;
aePostFunction(g_pserver->rgthreadvar[IDX_EVENT_LOOP_MAIN].el, []{
if (aeCreateFileEvent(serverTL->el, g_pserver->rdb_pipe_read, AE_READABLE, rdbPipeReadHandler,NULL) == AE_ERR) {
serverPanic("Unrecoverable error creating server.rdb_pipe_read file event.");
}
});
return C_OK; /* Unreached. */
}
+9 -8
View File
@@ -3945,6 +3945,8 @@ void initServer(void) {
g_pserver->pubsub_channels = dictCreate(&keylistDictType,NULL);
g_pserver->pubsub_patterns = dictCreate(&keylistDictType,NULL);
g_pserver->cronloops = 0;
g_pserver->propagate_in_transaction = 0;
g_pserver->client_pause_in_transaction = 0;
g_pserver->child_pid = -1;
g_pserver->child_type = CHILD_TYPE_NONE;
g_pserver->rdbThreadVars.fRdbThreadCancel = false;
@@ -4318,12 +4320,12 @@ void propagate(struct redisCommand *cmd, int dbid, robj **argv, int argc,
* This way we'll deliver the MULTI/..../EXEC block as a whole and
* both the AOF and the replication link will have the same consistency
* and atomicity guarantees. */
if (serverTL->in_exec && !serverTL->propagate_in_transaction)
if (serverTL->in_exec && !g_pserver->propagate_in_transaction)
execCommandPropagateMulti(dbid);
/* This needs to be unreachable since the dataset should be fixed during
* client pause, otherwise data may be lossed during a failover. */
serverAssert(!(areClientsPaused() && !serverTL->client_pause_in_transaction));
serverAssert(!(areClientsPaused() && !g_pserver->client_pause_in_transaction));
if (g_pserver->aof_state != AOF_OFF && flags & PROPAGATE_AOF)
feedAppendOnlyFile(cmd,dbid,argv,argc);
@@ -4645,8 +4647,8 @@ void call(client *c, int flags) {
/* Client pause takes effect after a transaction has finished. This needs
* to be located after everything is propagated. */
if (!serverTL->in_exec && serverTL->client_pause_in_transaction) {
serverTL->client_pause_in_transaction = 0;
if (!serverTL->in_exec && g_pserver->client_pause_in_transaction) {
g_pserver->client_pause_in_transaction = 0;
}
/* If the client has keys tracking enabled for client side caching,
@@ -4737,9 +4739,8 @@ int processCommand(client *c, int callFlags) {
/* Both EXEC and EVAL call call() directly so there should be
* no way in_exec or in_eval or propagate_in_transaction is 1.
* That is unless lua_timedout, in which case client may run
* some commands. Also possible that some other thread set
* propagate_in_transaction if this is an async command. */
serverAssert(!serverTL->propagate_in_transaction);
* some commands. */
serverAssert(!g_pserver->propagate_in_transaction);
serverAssert(!serverTL->in_exec);
serverAssert(!serverTL->in_eval);
}
@@ -5171,7 +5172,7 @@ int prepareForShutdown(int flags) {
* to unlink file actully) in background thread.
* The temp rdb file fd may won't be closed when redis exits quickly,
* but OS will close this fd when process exits. */
rdbRemoveTempFile(g_pserver->rdbThreadVars.tmpfileNum, 0);
rdbRemoveTempFile(g_pserver->child_pid, 0);
}
/* Kill module child if there is one. */
+3 -8
View File
@@ -1217,7 +1217,7 @@ protected:
uint64_t m_mvccCheckpoint = 0;
private:
static void serializeAndStoreChange(StorageCache *storage, redisDbPersistentData *db, const char *key, bool fUpdate);
static sds serializeChange(redisDbPersistentData *db, const char *key);
void ensure(const char *key);
void ensure(const char *key, dictEntry **de);
@@ -2197,9 +2197,6 @@ struct redisServerThreadVars {
bool modulesEnabledThisAeLoop = false; /* In this loop of aeMain, were modules enabled before
the thread went to sleep? */
bool disable_async_commands = false; /* this is only valid for one cycle of the AE loop and is reset in afterSleep */
int propagate_in_transaction = 0; /* Make sure we don't propagate nested MULTI/EXEC */
int client_pause_in_transaction = 0; /* Was a client pause executed during this Exec? */
std::vector<client*> vecclientsProcess;
dictAsyncRehashCtl *rehashCtl = nullptr;
@@ -2305,7 +2302,9 @@ struct redisServer {
int sentinel_mode; /* True if this instance is a Sentinel. */
size_t initial_memory_usage; /* Bytes used after initialization. */
int always_show_logo; /* Show logo even for non-stdout logging. */
int propagate_in_transaction; /* Make sure we don't propagate nested MULTI/EXEC */
char *ignore_warnings; /* Config: warnings that should be ignored. */
int client_pause_in_transaction; /* Was a client pause executed during this Exec? */
pause_type client_pause_type; /* True if clients are currently paused */
/* Modules */
::dict *moduleapi; /* Exported core APIs dictionary for modules. */
@@ -2730,13 +2729,9 @@ struct redisServer {
long long repl_batch_offStart = -1;
long long repl_batch_idxStart = -1;
long long rand_total_threshold;
int config_soft_shutdown = false;
bool soft_shutdown = false;
int flash_disable_key_cache = false;
/* Lock Contention Ring Buffer */
static const size_t s_lockContentionSamples = 64;
uint16_t rglockSamples[s_lockContentionSamples];
+10 -2
View File
@@ -43,8 +43,9 @@ 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, size_t celem)
void RocksDBStorageProvider::bulkInsert(char **rgkeys, size_t *rgcbkeys, char **rgvals, size_t *rgcbvals, char *rgfOverwrite, size_t celem)
{
size_t coverwrites = 0;
if (celem >= 16384) {
rocksdb::Options options = DefaultRocksDBOptions();
rocksdb::SstFileWriter sst_file_writer(rocksdb::EnvOptions(), options, options.comparator);
@@ -92,8 +93,15 @@ 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;
m_count += celem - coverwrites;
}
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, size_t celem) override;
virtual void bulkInsert(char **rgkeys, size_t *rgcbkeys, char **rgvals, size_t *rgcbvals, char *rgfOverwrite, size_t celem) override;
virtual void batch_lock() override;
virtual void batch_unlock() override;
+2 -10
View File
@@ -729,10 +729,6 @@ void hincrbyfloatCommand(client *c) {
unsigned int vlen;
if (getLongDoubleFromObjectOrReply(c,c->argv[3],&incr,NULL) != C_OK) return;
if (isnan(incr) || isinf(incr)) {
addReplyError(c,"value is NaN or Infinity");
return;
}
if ((o = hashTypeLookupWriteOrCreate(c,c->argv[1])) == NULL) return;
if (hashTypeGetValue(o,szFromObj(c->argv[2]),&vstr,&vlen,&ll) == C_OK) {
if (vstr) {
@@ -1066,8 +1062,6 @@ void hrandfieldWithCountCommand(client *c, long l, int withvalues) {
addReplyBulkCBuffer(c, key, sdslen(key));
if (withvalues)
addReplyBulkCBuffer(c, value, sdslen(value));
if (c->flags & CLIENT_CLOSE_ASAP)
break;
}
} else if (hash->encoding == OBJ_ENCODING_ZIPLIST) {
ziplistEntry *keys, *vals = NULL;
@@ -1081,8 +1075,6 @@ void hrandfieldWithCountCommand(client *c, long l, int withvalues) {
count -= sample_count;
ziplistRandomPairs((unsigned char*)ptrFromObj(hash), sample_count, keys, vals);
harndfieldReplyWithZiplist(c, sample_count, keys, vals);
if (c->flags & CLIENT_CLOSE_ASAP)
break;
}
zfree(keys);
zfree(vals);
@@ -1229,13 +1221,13 @@ void hrandfieldCommand(client *c) {
ziplistEntry ele;
if (c->argc >= 3) {
if (getRangeLongFromObjectOrReply(c,c->argv[2],-LONG_MAX,LONG_MAX,&l,NULL) != C_OK) return;
if (getLongFromObjectOrReply(c,c->argv[2],&l,NULL) != C_OK) return;
if (c->argc > 4 || (c->argc == 4 && strcasecmp(szFromObj(c->argv[3]),"withvalues"))) {
addReplyErrorObject(c,shared.syntaxerr);
return;
} else if (c->argc == 4) {
withvalues = 1;
if (l < -g_pserver->rand_total_threshold || l > g_pserver->rand_total_threshold) {
if (l < LONG_MIN/2 || l > LONG_MAX/2) {
addReplyError(c,"value is out of range");
return;
}
+1 -7
View File
@@ -672,11 +672,7 @@ void srandmemberWithCountCommand(client *c) {
dict *d;
if (getRangeLongFromObjectOrReply(c,c->argv[2],-LONG_MAX,LONG_MAX,&l,NULL) != C_OK) return;
if (l < -g_pserver->rand_total_threshold || l > g_pserver->rand_total_threshold) {
addReplyError(c,"value is out of range");
return;
}
if (getLongFromObjectOrReply(c,c->argv[2],&l,NULL) != C_OK) return;
if (l >= 0) {
count = (unsigned long) l;
} else {
@@ -710,8 +706,6 @@ void srandmemberWithCountCommand(client *c) {
} else {
addReplyBulkCBuffer(c,ele,sdslen(ele));
}
if (c->flags & CLIENT_CLOSE_ASAP)
break;
}
return;
}
+2 -6
View File
@@ -4054,8 +4054,6 @@ void zrandmemberWithCountCommand(client *c, long l, int withscores) {
addReplyBulkCBuffer(c, key, sdslen(key));
if (withscores)
addReplyDouble(c, *(double*)dictGetVal(de));
if (c->flags & CLIENT_CLOSE_ASAP)
break;
}
} else if (zsetobj->encoding == OBJ_ENCODING_ZIPLIST) {
ziplistEntry *keys, *vals = NULL;
@@ -4069,8 +4067,6 @@ void zrandmemberWithCountCommand(client *c, long l, int withscores) {
count -= sample_count;
ziplistRandomPairs((unsigned char*)ptrFromObj(zsetobj), sample_count, keys, vals);
zarndmemberReplyWithZiplist(c, sample_count, keys, vals);
if (c->flags & CLIENT_CLOSE_ASAP)
break;
}
zfree(keys);
zfree(vals);
@@ -4214,13 +4210,13 @@ void zrandmemberCommand(client *c) {
ziplistEntry ele;
if (c->argc >= 3) {
if (getRangeLongFromObjectOrReply(c,c->argv[2],-LONG_MAX,LONG_MAX,&l,NULL) != C_OK) return;
if (getLongFromObjectOrReply(c,c->argv[2],&l,NULL) != C_OK) return;
if (c->argc > 4 || (c->argc == 4 && strcasecmp(szFromObj(c->argv[3]),"withscores"))) {
addReplyErrorObject(c,shared.syntaxerr);
return;
} else if (c->argc == 4) {
withscores = 1;
if (l < -g_pserver->rand_total_threshold || l > g_pserver->rand_total_threshold) {
if (l < LONG_MIN/2 || l > LONG_MAX/2) {
addReplyError(c,"value is out of range");
return;
}
-8
View File
@@ -182,12 +182,4 @@ start_server {tags {"obuf-limits"} overrides { server-threads 1 }} {
assert_equal "v2" [r get k2]
assert_equal "v3" [r get k3]
}
test "Obuf limit, HRANDFIELD with huge count stopped mid-run" {
r config set client-output-buffer-limit {normal 1000000 0 0}
r hset myhash a b
catch {r hrandfield myhash -999999999} e
assert_match "*I/O error*" $e
reconnect
}
}
-6
View File
@@ -71,8 +71,6 @@ start_server {tags {"hash"}} {
test "HRANDFIELD count overflow" {
r hmset myhash a 1
assert_error {*value is out of range*} {r hrandfield myhash -9223372036854770000 withvalues}
assert_error {*value is out of range*} {r hrandfield myhash -9223372036854775808 withvalues}
assert_error {*value is out of range*} {r hrandfield myhash -9223372036854775808}
} {}
test "HRANDFIELD with <count> against non existing key" {
@@ -833,8 +831,4 @@ start_server {tags {"hash"}} {
set _ $k
} {ZIP_INT_8B 127 ZIP_INT_16B 32767 ZIP_INT_32B 2147483647 ZIP_INT_64B 9223372036854775808 ZIP_INT_IMM_MIN 0 ZIP_INT_IMM_MAX 12}
test {HINCRBYFLOAT does not allow NaN or Infinity} {
assert_error "*value is NaN or Infinity*" {r hincrbyfloat hfoo field +inf}
assert_equal 0 [r exists hfoo]
}
}
-5
View File
@@ -588,11 +588,6 @@ start_server {
r srandmember nonexisting_key 100
} {}
test "SRANDMEMBER count overflow" {
r sadd myset a
assert_error {*value is out of range*} {r srandmember myset -9223372036854775808}
} {}
# Make sure we can distinguish between an empty array and a null response
r readraw 1
-2
View File
@@ -1717,8 +1717,6 @@ start_server {tags {"zset"}} {
test "ZRANDMEMBER count overflow" {
r zadd myzset 0 a
assert_error {*value is out of range*} {r zrandmember myzset -9223372036854770000 withscores}
assert_error {*value is out of range*} {r zrandmember myzset -9223372036854775808 withscores}
assert_error {*value is out of range*} {r zrandmember myzset -9223372036854775808}
} {}
# Make sure we can distinguish between an empty array and a null response
+1 -7
View File
@@ -26,11 +26,7 @@ then
while [ $((PORT < ENDPORT)) != "0" ]; do
PORT=$((PORT+1))
echo "Starting $PORT"
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}
$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}
done
exit 0
fi
@@ -101,8 +97,6 @@ then
rm -rf appendonly*.aof
rm -rf dump*.rdb
rm -rf nodes*.conf
rm -rf *.flash
rm -rf temp*.rdb
exit 0
fi