Compare commits

..
18 Commits
Author SHA1 Message Date
John Sully c50a3d0bbd Remove performance zapping test code 2019-10-21 01:48:58 -04:00
John Sully 5eb84db03a reenable optimized string dup code 2019-10-21 00:08:38 -04:00
John Sully bd9d8e05b0 Forkless background save. NOT OPTIMIZED 2019-10-20 23:54:05 -04:00
John Sully 2ab4c6b99e Factor out dependencies on rdb_child_pid which will be going away 2019-10-17 23:10:45 -04:00
John Sully c12be59475 Dependencies for forkless RDB saving 2019-10-17 22:53:50 -04:00
John Sully fcdda67d87 Ensure RDB saving only uses data from the persistent data struct, and cleanup inheritance 2019-10-17 17:20:47 -04:00
John Sully dfdd41f15c Support serializing all types 2019-10-16 14:23:28 -04:00
John Sully e8debc83e8 Cleanup inheritance 2019-10-16 13:45:36 -04:00
John Sully 88152a8f4f Merge branch 'unstable' into advanced_db 2019-10-16 13:23:19 -04:00
John Sully c4f086ae93 Fix clang warnings 2019-10-13 15:48:55 -04:00
John Sully 3f0c1f94eb Fix failing travis CI build 2019-10-13 15:41:53 -04:00
John Sully f6855bc34d Merge branch 'advanced_db' of https://github.com/JohnSully/redis2 into advanced_db 2019-10-13 15:35:30 -04:00
John Sully 898efbfc0c Add the IStorage interface and wire it up 2019-10-13 15:24:08 -04:00
John Sully 8d63f5b560 Isolate the persistent parts of the database from the runtime parts 2019-10-13 15:24:08 -04:00
John Sully 024365521c Hide the database dict 2019-10-13 15:13:01 -04:00
John Sully cbd4b4df19 Add the IStorage interface and wire it up 2019-09-20 14:52:22 -04:00
John Sully 08995b065c Isolate the persistent parts of the database from the runtime parts 2019-09-19 22:36:29 -04:00
John Sully 065846a9c3 Hide the database dict 2019-09-19 17:44:42 -04:00
27 changed files with 1434 additions and 600 deletions

No files matched your search

+58
View File
@@ -0,0 +1,58 @@
{
"files.associations": {
"zmalloc.h": "c",
"stat.h": "c",
"array": "cpp",
"atomic": "cpp",
"*.tcc": "cpp",
"cctype": "cpp",
"chrono": "cpp",
"clocale": "cpp",
"cmath": "cpp",
"condition_variable": "cpp",
"cstdarg": "cpp",
"cstddef": "cpp",
"cstdint": "cpp",
"cstdio": "cpp",
"cstdlib": "cpp",
"cstring": "cpp",
"ctime": "cpp",
"cwchar": "cpp",
"cwctype": "cpp",
"deque": "cpp",
"list": "cpp",
"unordered_map": "cpp",
"vector": "cpp",
"exception": "cpp",
"fstream": "cpp",
"functional": "cpp",
"future": "cpp",
"initializer_list": "cpp",
"iomanip": "cpp",
"iosfwd": "cpp",
"iostream": "cpp",
"istream": "cpp",
"limits": "cpp",
"memory": "cpp",
"mutex": "cpp",
"new": "cpp",
"numeric": "cpp",
"optional": "cpp",
"ostream": "cpp",
"ratio": "cpp",
"scoped_allocator": "cpp",
"sstream": "cpp",
"stdexcept": "cpp",
"streambuf": "cpp",
"string_view": "cpp",
"system_error": "cpp",
"thread": "cpp",
"cinttypes": "cpp",
"tuple": "cpp",
"type_traits": "cpp",
"typeinfo": "cpp",
"utility": "cpp",
"set": "cpp",
"algorithm": "cpp"
}
}
+14
View File
@@ -0,0 +1,14 @@
#pragma once
#include <functional>
class IStorage
{
public:
typedef std::function<void(const char *, size_t, const void *, size_t)> callback;
virtual void insert(const char *key, size_t cchKey, void *data, size_t cb) = 0;
virtual void erase(const char *key, size_t cchKey) = 0;
virtual void retrieve(const char *key, size_t cchKey, bool fDelete, callback fn) = 0;
virtual size_t clear() = 0;
virtual void enumerate(callback fn) = 0;
};
+35 -35
View File
@@ -264,7 +264,7 @@ int startAppendOnly(void) {
strerror(errno));
return C_ERR;
}
if (g_pserver->rdb_child_pid != -1) {
if (g_pserver->FRdbSaveInProgress()) {
g_pserver->aof_rewrite_scheduled = 1;
serverLog(LL_WARNING,"AOF was enabled but there is already a child process saving an RDB file on disk. An AOF background was scheduled to start when possible.");
} else {
@@ -397,7 +397,7 @@ void flushAppendOnlyFile(int force) {
* useful for graphing / monitoring purposes. */
if (sync_in_progress) {
latencyAddSampleIfNeeded("aof-write-pending-fsync",latency);
} else if (g_pserver->aof_child_pid != -1 || g_pserver->rdb_child_pid != -1) {
} else if (g_pserver->aof_child_pid != -1 || g_pserver->FRdbSaveInProgress()) {
latencyAddSampleIfNeeded("aof-write-active-child",latency);
} else {
latencyAddSampleIfNeeded("aof-write-alone",latency);
@@ -494,7 +494,7 @@ try_fsync:
/* Don't fsync if no-appendfsync-on-rewrite is set to yes and there are
* children doing I/O in the background. */
if (g_pserver->aof_no_fsync_on_rewrite &&
(g_pserver->aof_child_pid != -1 || g_pserver->rdb_child_pid != -1))
(g_pserver->aof_child_pid != -1 || g_pserver->FRdbSaveInProgress()))
return;
/* Perform the fsync if needed. */
@@ -1300,52 +1300,46 @@ ssize_t aofReadDiffFromParent(void) {
int rewriteAppendOnlyFileRio(rio *aof) {
dictIterator *di = NULL;
dictEntry *de;
size_t processed = 0;
int j;
for (j = 0; j < cserver.dbnum; j++) {
char selectcmd[] = "*2\r\n$6\r\nSELECT\r\n";
redisDb *db = g_pserver->db+j;
dict *d = db->pdict;
if (dictSize(d) == 0) continue;
di = dictGetSafeIterator(d);
if (db->size() == 0) continue;
/* SELECT the new DB */
if (rioWrite(aof,selectcmd,sizeof(selectcmd)-1) == 0) goto werr;
if (rioWriteBulkLongLong(aof,j) == 0) goto werr;
/* Iterate this DB writing every entry */
while((de = dictNext(di)) != NULL) {
sds keystr;
robj key, *o;
bool fComplete = db->iterate([&](const char *keystr, robj *o)->bool{
robj key;
keystr = (sds)dictGetKey(de);
o = (robj*)dictGetVal(de);
initStaticStringObject(key,keystr);
initStaticStringObject(key,(sds)keystr);
expireEntry *pexpire = getExpire(db,&key);
expireEntry *pexpire = db->getExpire(&key);
/* Save the key and associated value */
if (o->type == OBJ_STRING) {
/* Emit a SET command */
char cmd[]="*3\r\n$3\r\nSET\r\n";
if (rioWrite(aof,cmd,sizeof(cmd)-1) == 0) goto werr;
if (rioWrite(aof,cmd,sizeof(cmd)-1) == 0) return false;
/* Key and value */
if (rioWriteBulkObject(aof,&key) == 0) goto werr;
if (rioWriteBulkObject(aof,o) == 0) goto werr;
if (rioWriteBulkObject(aof,&key) == 0) return false;
if (rioWriteBulkObject(aof,o) == 0) return false;
} else if (o->type == OBJ_LIST) {
if (rewriteListObject(aof,&key,o) == 0) goto werr;
if (rewriteListObject(aof,&key,o) == 0) return false;
} else if (o->type == OBJ_SET) {
if (rewriteSetObject(aof,&key,o) == 0) goto werr;
if (rewriteSetObject(aof,&key,o) == 0) return false;
} else if (o->type == OBJ_ZSET) {
if (rewriteSortedSetObject(aof,&key,o) == 0) goto werr;
if (rewriteSortedSetObject(aof,&key,o) == 0) return false;
} else if (o->type == OBJ_HASH) {
if (rewriteHashObject(aof,&key,o) == 0) goto werr;
if (rewriteHashObject(aof,&key,o) == 0) return false;
} else if (o->type == OBJ_STREAM) {
if (rewriteStreamObject(aof,&key,o) == 0) goto werr;
if (rewriteStreamObject(aof,&key,o) == 0) return false;
} else if (o->type == OBJ_MODULE) {
if (rewriteModuleObject(aof,&key,o) == 0) goto werr;
if (rewriteModuleObject(aof,&key,o) == 0) return false;
} else {
serverPanic("Unknown object type");
}
@@ -1355,17 +1349,17 @@ int rewriteAppendOnlyFileRio(rio *aof) {
if (subExpire.subkey() == nullptr)
{
char cmd[]="*3\r\n$9\r\nPEXPIREAT\r\n";
if (rioWrite(aof,cmd,sizeof(cmd)-1) == 0) goto werr;
if (rioWriteBulkObject(aof,&key) == 0) goto werr;
if (rioWrite(aof,cmd,sizeof(cmd)-1) == 0) return false;
if (rioWriteBulkObject(aof,&key) == 0) return false;
}
else
{
char cmd[]="*4\r\n$12\r\nEXPIREMEMBER\r\n";
if (rioWrite(aof,cmd,sizeof(cmd)-1) == 0) goto werr;
if (rioWriteBulkObject(aof,&key) == 0) goto werr;
if (rioWrite(aof,subExpire.subkey(),sdslen(subExpire.subkey())) == 0) goto werr;
if (rioWrite(aof,cmd,sizeof(cmd)-1) == 0) return false;
if (rioWriteBulkObject(aof,&key) == 0) return false;
if (rioWrite(aof,subExpire.subkey(),sdslen(subExpire.subkey())) == 0) return false;
}
if (rioWriteBulkLongLong(aof,subExpire.when()) == 0) goto werr; // common
if (rioWriteBulkLongLong(aof,subExpire.when()) == 0) return false; // common
}
}
/* Read some diff from the parent process from time to time. */
@@ -1373,9 +1367,10 @@ int rewriteAppendOnlyFileRio(rio *aof) {
processed = aof->processed_bytes;
aofReadDiffFromParent();
}
}
dictReleaseIterator(di);
di = NULL;
return true;
});
if (!fComplete)
goto werr;
}
return C_OK;
@@ -1416,7 +1411,12 @@ int rewriteAppendOnlyFile(char *filename) {
if (g_pserver->aof_use_rdb_preamble) {
int error;
if (rdbSaveRio(&aof,&error,RDB_SAVE_AOF_PREAMBLE,NULL) == C_ERR) {
std::vector<redisDbPersistentData*> vecpdb;
for (int idb = 0; idb < cserver.dbnum; ++idb)
{
vecpdb.push_back(&g_pserver->db[idb]);
}
if (rdbSaveRio(&aof,vecpdb.data(),&error,RDB_SAVE_AOF_PREAMBLE,NULL) == C_ERR) {
errno = error;
goto werr;
}
@@ -1593,7 +1593,7 @@ int rewriteAppendOnlyFileBackground(void) {
pid_t childpid;
long long start;
if (g_pserver->aof_child_pid != -1 || g_pserver->rdb_child_pid != -1) return C_ERR;
if (g_pserver->aof_child_pid != -1 || g_pserver->FRdbSaveInProgress()) return C_ERR;
if (aofCreatePipes() != C_OK) return C_ERR;
openChildInfoPipe();
start = ustime();
@@ -1652,7 +1652,7 @@ int rewriteAppendOnlyFileBackground(void) {
void bgrewriteaofCommand(client *c) {
if (g_pserver->aof_child_pid != -1) {
addReplyError(c,"Background append only file rewriting already in progress");
} else if (g_pserver->rdb_child_pid != -1) {
} else if (g_pserver->FRdbSaveInProgress()) {
g_pserver->aof_rewrite_scheduled = 1;
addReplyStatus(c,"Background append only file rewriting scheduled");
} else if (rewriteAppendOnlyFileBackground() == C_OK) {
+24 -5
View File
@@ -3910,7 +3910,7 @@ int verifyClusterConfigWithData(void) {
/* Make sure we only have keys in DB0. */
for (j = 1; j < cserver.dbnum; j++) {
if (dictSize(g_pserver->db[j].pdict)) return C_ERR;
if (g_pserver->db[j].size()) return C_ERR;
}
/* Check that all the slots we see populated memory have a corresponding
@@ -4286,7 +4286,7 @@ NULL
clusterReplyMultiBulkSlots(c);
} else if (!strcasecmp(szFromObj(c->argv[1]),"flushslots") && c->argc == 2) {
/* CLUSTER FLUSHSLOTS */
if (dictSize(g_pserver->db[0].pdict) != 0) {
if (g_pserver->db[0].size() != 0) {
addReplyError(c,"DB must be empty to perform CLUSTER FLUSHSLOTS.");
return;
}
@@ -4621,7 +4621,7 @@ NULL
* slots nor keys to accept to replicate some other node.
* Slaves can switch to another master without issues. */
if (nodeIsMaster(myself) &&
(myself->numslots != 0 || dictSize(g_pserver->db[0].pdict) != 0)) {
(myself->numslots != 0 || g_pserver->db[0].size() != 0)) {
addReplyError(c,
"To set a master the node must be empty and "
"without assigned slots.");
@@ -4778,7 +4778,7 @@ NULL
/* Slaves can be reset while containing data, but not master nodes
* that must be empty. */
if (nodeIsMaster(myself) && dictSize(c->db->pdict) != 0) {
if (nodeIsMaster(myself) && c->db->size() != 0) {
addReplyError(c,"CLUSTER RESET can't be called with "
"master nodes containing keys");
return;
@@ -4794,6 +4794,7 @@ NULL
/* -----------------------------------------------------------------------------
* DUMP, RESTORE and MIGRATE commands
* -------------------------------------------------------------------------- */
ssize_t rdbSaveAuxFieldStrStr(rio *rdb, const char *key, const char *val);
/* Generates a DUMP-format representation of the object 'o', adding it to the
* io stream pointed by 'rio'. This function can't fail. */
@@ -4806,6 +4807,9 @@ void createDumpPayload(rio *payload, robj_roptr o, robj *key) {
rioInitWithBuffer(payload,sdsempty());
serverAssert(rdbSaveObjectType(payload,o));
serverAssert(rdbSaveObject(payload,o,key));
char szT[32];
snprintf(szT, 32, "%" PRIu64, o->mvcc_tstamp);
serverAssert(rdbSaveAuxFieldStrStr(payload,"mvcc-tstamp", szT) != -1);
/* Write the footer, this is how it looks like:
* ----------------+---------------------+---------------+
@@ -4941,6 +4945,21 @@ void restoreCommand(client *c) {
addReplyError(c,"Bad data format");
return;
}
if (rdbLoadType(&payload) == RDB_OPCODE_AUX)
{
robj *auxkey, *auxval;
if ((auxkey = rdbLoadStringObject(&payload)) == NULL) goto eoferr;
if ((auxval = rdbLoadStringObject(&payload)) == NULL) {
decrRefCount(auxkey);
goto eoferr;
}
if (strcasecmp(szFromObj(auxkey), "mvcc-tstamp") == 0) {
obj->mvcc_tstamp = strtoull(szFromObj(auxval), nullptr, 10);
}
decrRefCount(auxkey);
decrRefCount(auxval);
}
eoferr:
/* Remove the old key if needed. */
if (replace) dbDelete(c->db,c->argv[1]);
@@ -5194,7 +5213,7 @@ try_again:
/* Create RESTORE payload and generate the protocol to call the command. */
for (j = 0; j < num_keys; j++) {
long long ttl = 0;
expireEntry *pexpire = getExpire(c->db,kv[j]);
expireEntry *pexpire = c->db->getExpire(kv[j]);
long long expireat = -1;
if (pexpire != nullptr)
pexpire->FGetPrimaryExpire(&expireat);
+10 -2
View File
@@ -30,7 +30,16 @@ public:
zfree(m_data);
}
compactvector(compactvector &) noexcept = delete;
compactvector(const compactvector &src)
{
m_celem = src.m_celem;
m_max = src.m_max;
m_data = (T*)zmalloc(sizeof(T) * m_max, MALLOC_LOCAL);
for (size_t ielem = 0; ielem < m_celem; ++ielem)
{
new (m_data+ielem) T(src[ielem]);
}
}
compactvector(compactvector &&src) noexcept
{
@@ -42,7 +51,6 @@ public:
src.m_max = 0;
}
compactvector &operator=(const compactvector&) noexcept = delete;
compactvector &operator=(compactvector &&src) noexcept
{
zfree(m_data);
+425 -153
View File
@@ -41,7 +41,6 @@
int keyIsExpired(redisDb *db, robj *key);
int expireIfNeeded(redisDb *db, robj *key, robj *o);
void dbOverwriteCore(redisDb *db, dictEntry *de, robj *key, robj *val, bool fUpdateMvcc, bool fRemoveExpire);
/* Update LFU when an object is accessed.
* Firstly, decrement the counter if the decrement time is reached.
@@ -57,8 +56,7 @@ void updateExpire(redisDb *db, sds key, robj *valOld, robj *valNew)
serverAssert(valOld->FExpires());
serverAssert(!valNew->FExpires());
auto itr = db->setexpire->find(key);
serverAssert(itr != db->setexpire->end());
serverAssert(db->FKeyExpires((const char*)key));
valNew->SetFExpires(true);
valOld->SetFExpires(false);
@@ -70,14 +68,13 @@ void updateExpire(redisDb *db, sds key, robj *valOld, robj *valNew)
* implementations that should instead rely on lookupKeyRead(),
* lookupKeyWrite() and lookupKeyReadWithFlags(). */
static robj *lookupKey(redisDb *db, robj *key, int flags) {
dictEntry *de = dictFind(db->pdict,ptrFromObj(key));
if (de) {
robj *val = (robj*)dictGetVal(de);
auto itr = db->find(key);
if (itr) {
robj *val = itr.val();
/* Update the access time for the ageing algorithm.
* Don't do it if we have a saving child, as this will trigger
* a copy on write madness. */
if (g_pserver->rdb_child_pid == -1 &&
if (!g_pserver->FRdbSaveInProgress() &&
g_pserver->aof_child_pid == -1 &&
!(flags & LOOKUP_NOTOUCH))
{
@@ -90,6 +87,7 @@ static robj *lookupKey(redisDb *db, robj *key, int flags) {
if (flags & LOOKUP_UPDATEMVCC) {
val->mvcc_tstamp = getMvccTstamp();
db->trackkey(key);
}
return val;
} else {
@@ -195,13 +193,14 @@ robj *lookupKeyWriteOrReply(client *c, robj *key, robj *reply) {
return o;
}
int dbAddCore(redisDb *db, robj *key, robj *val) {
bool dbAddCore(redisDb *db, robj *key, robj *val) {
serverAssert(!val->FExpires());
sds copy = sdsdup(szFromObj(key));
int retval = dictAdd(db->pdict, copy, val);
val->mvcc_tstamp = key->mvcc_tstamp = getMvccTstamp();
bool fInserted = db->insert(copy, val);
if (g_pserver->fActiveReplica)
val->mvcc_tstamp = key->mvcc_tstamp = getMvccTstamp();
if (retval == DICT_OK)
if (fInserted)
{
if (val->type == OBJ_LIST ||
val->type == OBJ_ZSET)
@@ -213,7 +212,7 @@ int dbAddCore(redisDb *db, robj *key, robj *val) {
sdsfree(copy);
}
return retval;
return fInserted;
}
/* Add the key to the DB. It's up to the caller to increment the reference
@@ -222,23 +221,22 @@ int dbAddCore(redisDb *db, robj *key, robj *val) {
* The program is aborted if the key already exists. */
void dbAdd(redisDb *db, robj *key, robj *val)
{
int retval = dbAddCore(db, key, val);
serverAssertWithInfo(NULL,key,retval == DICT_OK);
bool fInserted = dbAddCore(db, key, val);
serverAssertWithInfo(NULL,key,fInserted);
}
void dbOverwriteCore(redisDb *db, dictEntry *de, robj *key, robj *val, bool fUpdateMvcc, bool fRemoveExpire)
void redisDb::dbOverwriteCore(redisDb::iter itr, robj *key, robj *val, bool fUpdateMvcc, bool fRemoveExpire)
{
dictEntry auxentry = *de;
robj *old = (robj*)dictGetVal(de);
robj *old = itr.val();
if (old->FExpires()) {
if (fRemoveExpire) {
removeExpire(db, key);
::removeExpire(this, key);
}
else {
if (val->getrefcount(std::memory_order_relaxed) == OBJ_SHARED_REFCOUNT)
val = dupStringObject(val);
updateExpire(db, (sds)dictGetKey(de), old, val);
::updateExpire(this, itr.key(), old, val);
}
}
@@ -251,14 +249,12 @@ void dbOverwriteCore(redisDb *db, dictEntry *de, robj *key, robj *val, bool fUpd
val->mvcc_tstamp = getMvccTstamp();
}
dictSetVal(db->pdict, de, val);
if (g_pserver->lazyfree_lazy_server_del)
freeObjAsync(itr.val());
else
decrRefCount(itr.val());
if (g_pserver->lazyfree_lazy_server_del) {
freeObjAsync(old);
dictSetVal(db->pdict, &auxentry, NULL);
}
dictFreeVal(db->pdict, &auxentry);
updateValue(itr, val);
}
/* Overwrite an existing key with a new value. Incrementing the reference
@@ -267,10 +263,10 @@ void dbOverwriteCore(redisDb *db, dictEntry *de, robj *key, robj *val, bool fUpd
*
* The program is aborted if the key was not already present. */
void dbOverwrite(redisDb *db, robj *key, robj *val) {
dictEntry *de = dictFind(db->pdict,ptrFromObj(key));
auto itr = db->find(key);
serverAssertWithInfo(NULL,key,de != NULL);
dbOverwriteCore(db, de, key, val, !!g_pserver->fActiveReplica, false);
serverAssertWithInfo(NULL,key,itr != nullptr);
db->dbOverwriteCore(itr, key, val, !!g_pserver->fActiveReplica, false);
}
/* Insert a key, handling duplicate keys according to fReplace */
@@ -278,14 +274,14 @@ int dbMerge(redisDb *db, robj *key, robj *val, int fReplace)
{
if (fReplace)
{
dictEntry *de = dictFind(db->pdict, ptrFromObj(key));
if (de == nullptr)
return (dbAddCore(db, key, val) == DICT_OK);
auto itr = db->find(key);
if (itr == nullptr)
return (dbAddCore(db, key, val) == true);
robj *old = (robj*)dictGetVal(de);
robj *old = itr.val();
if (old->mvcc_tstamp <= val->mvcc_tstamp)
{
dbOverwriteCore(db, de, key, val, false, true);
db->dbOverwriteCore(itr, key, val, false, true);
return true;
}
@@ -293,7 +289,7 @@ int dbMerge(redisDb *db, robj *key, robj *val, int fReplace)
}
else
{
return (dbAddCore(db, key, val) == DICT_OK);
return (dbAddCore(db, key, val) == true);
}
}
@@ -306,18 +302,18 @@ int dbMerge(redisDb *db, robj *key, robj *val, int fReplace)
*
* All the new keys in the database should be created via this interface. */
void setKey(redisDb *db, robj *key, robj *val) {
dictEntry *de = dictFind(db->pdict, ptrFromObj(key));
if (de == NULL) {
auto itr = db->find(key);
if (itr == NULL) {
dbAdd(db,key,val);
} else {
dbOverwriteCore(db,de,key,val,!!g_pserver->fActiveReplica,true);
db->dbOverwriteCore(itr,key,val,!!g_pserver->fActiveReplica,true);
}
incrRefCount(val);
signalModifiedKey(db,key);
}
int dbExists(redisDb *db, robj *key) {
return dictFind(db->pdict,ptrFromObj(key)) != NULL;
return (db->find(key) != nullptr);
}
/* Return a random key, in form of a Redis object.
@@ -325,21 +321,20 @@ int dbExists(redisDb *db, robj *key) {
*
* The function makes sure to return keys not already expired. */
robj *dbRandomKey(redisDb *db) {
dictEntry *de;
int maxtries = 100;
int allvolatile = dictSize(db->pdict) == db->setexpire->size();
bool allvolatile = db->expireSize() == db->size();
while(1) {
sds key;
robj *keyobj;
de = dictGetRandomKey(db->pdict);
if (de == NULL) return NULL;
auto itr = db->random();
if (itr == nullptr) return NULL;
key = (sds)dictGetKey(de);
key = itr.key();
keyobj = createStringObject(key,sdslen(key));
if (((robj*)dictGetVal(de))->FExpires())
if (itr.val()->FExpires())
{
if (allvolatile && listLength(g_pserver->masters) && --maxtries == 0) {
/* If the DB is composed only of keys with an expire set,
@@ -354,7 +349,7 @@ robj *dbRandomKey(redisDb *db) {
}
}
if (((robj*)dictGetVal(de))->FExpires())
if (itr.val()->FExpires())
{
if (expireIfNeeded(db,keyobj)) {
decrRefCount(keyobj);
@@ -366,15 +361,16 @@ robj *dbRandomKey(redisDb *db) {
}
}
/* Delete a key, value, and associated expiration entry if any, from the DB */
int dbSyncDelete(redisDb *db, robj *key) {
bool redisDbPersistentData::syncDelete(robj *key)
{
/* Deleting an entry from the expires dict will not free the sds of
* the key, because it is shared with the main dictionary. */
dictEntry *de = dictFind(db->pdict, szFromObj(key));
if (de != nullptr && ((robj*)dictGetVal(de))->FExpires())
removeExpireCore(db, key, de);
if (dictDelete(db->pdict,ptrFromObj(key)) == DICT_OK) {
auto itr = find(szFromObj(key));
trackkey(szFromObj(key));
if (itr != nullptr && itr.val()->FExpires())
removeExpire(key, itr);
if (dictDelete(m_pdict,ptrFromObj(key)) == DICT_OK) {
if (g_pserver->cluster_enabled) slotToKeyDel(key);
return 1;
} else {
@@ -382,6 +378,11 @@ int dbSyncDelete(redisDb *db, robj *key) {
}
}
/* Delete a key, value, and associated expiration entry if any, from the DB */
int dbSyncDelete(redisDb *db, robj *key) {
return db->syncDelete(key);
}
/* This is a wrapper whose behavior depends on the Redis lazy free
* configuration. Deletes the key synchronously or asynchronously. */
int dbDelete(redisDb *db, robj *key) {
@@ -427,7 +428,7 @@ robj *dbUnshareStringValue(redisDb *db, robj *key, robj *o) {
return o;
}
/* Remove all keys from all the databases in a Redis g_pserver->
/* Remove all keys from all the databases in a Redis DB
* If callback is given the function is called from time to time to
* signal that work is in progress.
*
@@ -459,15 +460,7 @@ long long emptyDb(int dbnum, int flags, void(callback)(void*)) {
}
for (int j = startdb; j <= enddb; j++) {
removed += dictSize(g_pserver->db[j].pdict);
if (async) {
emptyDbAsync(&g_pserver->db[j]);
} else {
dictEmpty(g_pserver->db[j].pdict,callback);
delete g_pserver->db[j].setexpire;
g_pserver->db[j].setexpire = new (MALLOC_LOCAL) expireset();
g_pserver->db[j].expireitr = g_pserver->db[j].setexpire->end();
}
removed += g_pserver->db[j].clear(!!async, callback);
}
if (g_pserver->cluster_enabled) {
if (async) {
@@ -553,14 +546,14 @@ void flushallCommand(client *c) {
signalFlushedDb(-1);
g_pserver->dirty += emptyDb(-1,flags,NULL);
addReply(c,shared.ok);
if (g_pserver->rdb_child_pid != -1) killRDBChild();
if (g_pserver->FRdbSaveInProgress()) killRDBChild();
if (g_pserver->saveparamslen > 0) {
/* Normally rdbSave() will reset dirty, but we don't want this here
* as otherwise FLUSHALL will not be replicated nor put into the AOF. */
int saved_dirty = g_pserver->dirty;
rdbSaveInfo rsi, *rsiptr;
rsiptr = rdbPopulateSaveInfo(&rsi);
rdbSave(rsiptr);
rdbSave(nullptr, rsiptr);
g_pserver->dirty = saved_dirty;
}
g_pserver->dirty++;
@@ -635,22 +628,35 @@ void randomkeyCommand(client *c) {
decrRefCount(key);
}
bool redisDbPersistentData::iterate(std::function<bool(const char*, robj*)> fn)
{
dictIterator *di = dictGetSafeIterator(m_pdict);
dictEntry *de = nullptr;
bool fResult = true;
while((de = dictNext(di)) != nullptr)
{
ensure(de);
if (!fn((const char*)dictGetKey(de), (robj*)dictGetVal(de)))
{
fResult = false;
break;
}
}
dictReleaseIterator(di);
return fResult;
}
void keysCommand(client *c) {
dictIterator *di;
dictEntry *de;
sds pattern = szFromObj(c->argv[1]);
int plen = sdslen(pattern), allkeys;
unsigned long numkeys = 0;
void *replylen = addReplyDeferredLen(c);
#ifdef MULTITHREADED_KEYS
aeReleaseLock();
#endif
di = dictGetSafeIterator(c->db->pdict);
allkeys = (pattern[0] == '*' && pattern[1] == '\0');
while((de = dictNext(di)) != NULL) {
sds key = (sds)dictGetKey(de);
c->db->iterate([&](const char *key, robj *)->bool {
robj *keyobj;
if (allkeys || stringmatchlen(pattern,plen,key,sdslen(key),0)) {
@@ -661,17 +667,15 @@ void keysCommand(client *c) {
}
decrRefCount(keyobj);
}
}
dictReleaseIterator(di);
return true;
});
setDeferredArrayLen(c,replylen,numkeys);
#ifdef MULTITHREADED_KEYS
fastlock_unlock(&c->db->lock); // we must release the DB lock before acquiring the AE lock to prevent deadlocks
AeLocker lock;
lock.arm(c);
fastlock_lock(&c->db->lock); // we still need the DB lock
lock.release();
#endif
}
/* This callback is used by scanGenericCommand in order to collect elements
@@ -799,7 +803,7 @@ void scanGenericCommand(client *c, robj_roptr o, unsigned long cursor) {
/* Handle the case of a hash table. */
ht = NULL;
if (o == nullptr) {
ht = c->db->pdict;
ht = c->db->dictUnsafeKeyOnly();
} else if (o->type == OBJ_SET && o->encoding == OBJ_ENCODING_HT) {
ht = (dict*)ptrFromObj(o);
} else if (o->type == OBJ_HASH && o->encoding == OBJ_ENCODING_HT) {
@@ -932,7 +936,7 @@ void scanCommand(client *c) {
}
void dbsizeCommand(client *c) {
addReplyLongLong(c,dictSize(c->db->pdict));
addReplyLongLong(c,c->db->size());
}
void lastsaveCommand(client *c) {
@@ -1015,7 +1019,7 @@ void renameGenericCommand(client *c, int nx) {
std::unique_ptr<expireEntry> spexpire;
{ // scope pexpireOld since it will be invalid soon
expireEntry *pexpireOld = getExpire(c->db,c->argv[1]);
expireEntry *pexpireOld = c->db->getExpire(c->argv[1]);
if (pexpireOld != nullptr)
spexpire = std::make_unique<expireEntry>(std::move(*pexpireOld));
}
@@ -1093,7 +1097,7 @@ void moveCommand(client *c) {
std::unique_ptr<expireEntry> spexpire;
{ // scope pexpireOld
expireEntry *pexpireOld = getExpire(c->db,c->argv[1]);
expireEntry *pexpireOld = c->db->getExpire(c->argv[1]);
if (pexpireOld != nullptr)
spexpire = std::make_unique<expireEntry>(std::move(*pexpireOld));
}
@@ -1152,17 +1156,14 @@ int dbSwapDatabases(int id1, int id2) {
/* Swap hash tables. Note that we don't swap blocking_keys,
* ready_keys and watched_keys, since we want clients to
* remain in the same DB they were. */
db1->pdict = db2->pdict;
db1->setexpire = db2->setexpire;
db1->expireitr = db2->expireitr;
redisDbPersistentData::swap(db1, db2);
db1->avg_ttl = db2->avg_ttl;
db1->last_expire_set = db2->last_expire_set;
db1->expireitr = db2->expireitr;
db2->pdict = aux.pdict;
db2->setexpire = aux.setexpire;
db2->expireitr = aux.expireitr;
db2->avg_ttl = aux.avg_ttl;
db2->last_expire_set = aux.last_expire_set;
db2->expireitr = aux.expireitr;
/* Now we need to handle clients blocked on lists: as an effect
* of swapping the two DBs, a client that was waiting for list
@@ -1211,37 +1212,38 @@ void swapdbCommand(client *c) {
* Expires API
*----------------------------------------------------------------------------*/
int removeExpire(redisDb *db, robj *key) {
dictEntry *de = dictFind(db->pdict,ptrFromObj(key));
return removeExpireCore(db, key, de);
auto itr = db->find(key);
return db->removeExpire(key, itr);
}
int removeExpireCore(redisDb *db, robj *key, dictEntry *de) {
int redisDbPersistentData::removeExpire(robj *key, dict_iter itr) {
/* An expire may only be removed if there is a corresponding entry in the
* main dict. Otherwise, the key will never be freed. */
serverAssertWithInfo(NULL,key,de != NULL);
serverAssertWithInfo(NULL,key,itr != nullptr);
robj *val = (robj*)dictGetVal(de);
robj *val = itr.val();
if (!val->FExpires())
return 0;
auto itr = db->setexpire->find((sds)dictGetKey(de));
serverAssert(itr != db->setexpire->end());
serverAssert(itr->key() == (sds)dictGetKey(de));
db->setexpire->erase(itr);
trackkey(key);
auto itrExpire = m_setexpire->find(itr.key());
serverAssert(itrExpire != m_setexpire->end());
serverAssert(itrExpire->key() == itr.key());
m_setexpire->erase(itrExpire);
val->SetFExpires(false);
return 1;
}
int removeSubkeyExpire(redisDb *db, robj *key, robj *subkey) {
dictEntry *de = dictFind(db->pdict,ptrFromObj(key));
serverAssertWithInfo(NULL,key,de != NULL);
int redisDbPersistentData::removeSubkeyExpire(robj *key, robj *subkey) {
auto de = find(szFromObj(key));
serverAssertWithInfo(NULL,key,de != nullptr);
robj *val = (robj*)dictGetVal(de);
robj *val = de.val();
if (!val->FExpires())
return 0;
auto itr = db->setexpire->find((sds)dictGetKey(de));
serverAssert(itr != db->setexpire->end());
serverAssert(itr->key() == (sds)dictGetKey(de));
auto itr = m_setexpire->find(de.key());
serverAssert(itr != m_setexpire->end());
serverAssert(itr->key() == de.key());
if (!itr->FFat())
return 0;
@@ -1259,7 +1261,7 @@ int removeSubkeyExpire(redisDb *db, robj *key, robj *subkey) {
}
if (itr->pfatentry()->size() == 0)
removeExpireCore(db, key, de);
this->removeExpire(key, de);
return found;
}
@@ -1269,49 +1271,23 @@ int removeSubkeyExpire(redisDb *db, robj *key, robj *subkey) {
* to NULL. The 'when' parameter is the absolute unix time in milliseconds
* after which the key will no longer be considered valid. */
void setExpire(client *c, redisDb *db, robj *key, robj *subkey, long long when) {
dictEntry *kde;
serverAssert(GlobalLocksAcquired());
/* Reuse the sds from the main dict in the expire dict */
kde = dictFind(db->pdict,ptrFromObj(key));
serverAssertWithInfo(NULL,key,kde != NULL);
if (((robj*)dictGetVal(kde))->getrefcount(std::memory_order_relaxed) == OBJ_SHARED_REFCOUNT)
{
// shared objects cannot have the expire bit set, create a real object
dictSetVal(db->pdict, kde, dupStringObject((robj*)dictGetVal(kde)));
}
/* Update TTL stats (exponential moving average) */
/* Note: We never have to update this on expiry since we reduce it by the current elapsed time here */
long long now = g_pserver->mstime;
db->avg_ttl -= (now - db->last_expire_set); // reduce the TTL by the time that has elapsed
if (db->setexpire->empty())
if (db->expireSize() == 0)
db->avg_ttl = 0;
else
db->avg_ttl -= db->avg_ttl / db->setexpire->size(); // slide one entry out the window
db->avg_ttl -= db->avg_ttl / db->expireSize(); // slide one entry out the window
if (db->avg_ttl < 0)
db->avg_ttl = 0; // TTLs are never negative
db->avg_ttl += (double)(when-now) / (db->setexpire->size()+1); // add the new entry
db->avg_ttl += (double)(when-now) / (db->expireSize()+1); // add the new entry
db->last_expire_set = now;
/* Update the expire set */
const char *szSubKey = (subkey != nullptr) ? szFromObj(subkey) : nullptr;
if (((robj*)dictGetVal(kde))->FExpires()) {
auto itr = db->setexpire->find((sds)dictGetKey(kde));
serverAssert(itr != db->setexpire->end());
expireEntry eNew(std::move(*itr));
eNew.update(szSubKey, when);
db->setexpire->erase(itr);
db->setexpire->insert(eNew);
}
else
{
expireEntry e((sds)dictGetKey(kde), szSubKey, when);
((robj*)dictGetVal(kde))->SetFExpires(true);
db->setexpire->insert(e);
}
db->setExpire(key, subkey, when);
int writable_slave = listLength(g_pserver->masters) && g_pserver->repl_slave_ro == 0;
if (c && writable_slave && !(c->flags & CLIENT_MASTER))
@@ -1320,26 +1296,24 @@ void setExpire(client *c, redisDb *db, robj *key, robj *subkey, long long when)
void setExpire(client *c, redisDb *db, robj *key, expireEntry &&e)
{
dictEntry *kde;
serverAssert(GlobalLocksAcquired());
/* Reuse the sds from the main dict in the expire dict */
kde = dictFind(db->pdict,ptrFromObj(key));
auto kde = db->find(key);
serverAssertWithInfo(NULL,key,kde != NULL);
if (((robj*)dictGetVal(kde))->getrefcount(std::memory_order_relaxed) == OBJ_SHARED_REFCOUNT)
if (kde.val()->getrefcount(std::memory_order_relaxed) == OBJ_SHARED_REFCOUNT)
{
// shared objects cannot have the expire bit set, create a real object
dictSetVal(db->pdict, kde, dupStringObject((robj*)dictGetVal(kde)));
db->updateValue(kde, dupStringObject(kde.val()));
}
if (((robj*)dictGetVal(kde))->FExpires())
if (kde.val()->FExpires())
removeExpire(db, key);
e.setKeyUnsafe((sds)dictGetKey(kde));
db->setexpire->insert(e);
((robj*)dictGetVal(kde))->SetFExpires(true);
e.setKeyUnsafe(kde.key());
db->setExpire(std::move(e));
kde.val()->SetFExpires(true);
int writable_slave = listLength(g_pserver->masters) && g_pserver->repl_slave_ro == 0;
@@ -1349,22 +1323,19 @@ void setExpire(client *c, redisDb *db, robj *key, expireEntry &&e)
/* Return the expire time of the specified key, or null if no expire
* is associated with this key (i.e. the key is non volatile) */
expireEntry *getExpire(redisDb *db, robj_roptr key) {
dictEntry *de;
expireEntry *redisDbPersistentData::getExpire(robj_roptr key) {
/* No expire? return ASAP */
if (db->setexpire->size() == 0)
if (expireSize() == 0)
return nullptr;
de = dictFind(db->pdict, ptrFromObj(key));
if (de == NULL)
auto itr = find(szFromObj(key));
if (itr == nullptr)
return nullptr;
robj *obj = (robj*)dictGetVal(de);
if (!obj->FExpires())
if (!itr.val()->FExpires())
return nullptr;
auto itr = db->setexpire->find((sds)dictGetKey(de));
return itr.operator->();
auto itrExpire = findExpire(itr.key());
return itrExpire.operator->();
}
/* Propagate expires into slaves and the AOF file.
@@ -1394,7 +1365,7 @@ void propagateExpire(redisDb *db, robj *key, int lazy) {
/* Check if the key is expired. Note, this does not check subexpires */
int keyIsExpired(redisDb *db, robj *key) {
expireEntry *pexpire = getExpire(db,key);
expireEntry *pexpire = db->getExpire(key);
if (pexpire == nullptr) return 0; /* No expire for this key */
@@ -1826,3 +1797,304 @@ unsigned int delKeysInSlot(unsigned int hashslot) {
unsigned int countKeysInSlot(unsigned int hashslot) {
return g_pserver->cluster->slots_keys_count[hashslot];
}
void redisDbPersistentData::initialize()
{
m_pdict = dictCreate(&dbDictType,this);
m_setexpire = new(MALLOC_LOCAL) expireset();
m_fAllChanged = false;
m_fTrackingChanges = 0;
}
void redisDb::initialize(int id)
{
redisDbPersistentData::initialize();
this->expireitr = setexpire()->end();
this->blocking_keys = dictCreate(&keylistDictType,NULL);
this->ready_keys = dictCreate(&objectKeyPointerValueDictType,NULL);
this->watched_keys = dictCreate(&keylistDictType,NULL);
this->id = id;
this->avg_ttl = 0;
this->last_expire_set = 0;
this->defrag_later = listCreate();
}
bool redisDbPersistentData::insert(char *key, robj *o)
{
int res = dictAdd(m_pdict, key, o);
if (res == DICT_OK)
trackkey(key);
return (res == DICT_OK);
}
void redisDbPersistentData::tryResize()
{
if (htNeedsResize(m_pdict))
dictResize(m_pdict);
}
size_t redisDb::clear(bool fAsync, void(callback)(void*))
{
size_t removed = size();
if (fAsync) {
redisDbPersistentData::emptyDbAsync();
} else {
redisDbPersistentData::clear(callback);
}
expireitr = setexpire()->end();
return removed;
}
void redisDbPersistentData::clear(void(callback)(void*))
{
dictEmpty(m_pdict,callback);
if (m_fTrackingChanges)
m_fAllChanged = true;
delete m_setexpire;
m_setexpire = new (MALLOC_LOCAL) expireset();
if (m_pstorage != nullptr)
m_pstorage->clear();
}
/* static */ void redisDbPersistentData::swap(redisDbPersistentData *db1, redisDbPersistentData *db2)
{
redisDbPersistentData aux = *db1;
db1->m_pdict = db2->m_pdict;
db1->m_fTrackingChanges = db2->m_fTrackingChanges;
db1->m_fAllChanged = db2->m_fAllChanged;
db1->m_setexpire = db2->m_setexpire;
db1->m_pstorage = db2->m_pstorage;
db1->m_spdbSnapshot = db2->m_spdbSnapshot;
db2->m_pdict = aux.m_pdict;
db2->m_fTrackingChanges = aux.m_fTrackingChanges;
db2->m_fAllChanged = aux.m_fAllChanged;
db2->m_setexpire = aux.m_setexpire;
db2->m_pstorage = aux.m_pstorage;
db2->m_spdbSnapshot = aux.m_spdbSnapshot;
db1->m_pdict->privdata = static_cast<redisDbPersistentData*>(db1);
db2->m_pdict->privdata = static_cast<redisDbPersistentData*>(db2);
}
void redisDbPersistentData::setExpire(robj *key, robj *subkey, long long when)
{
/* Reuse the sds from the main dict in the expire dict */
dictEntry *kde = dictFind(m_pdict,ptrFromObj(key));
serverAssertWithInfo(NULL,key,kde != NULL);
trackkey(key);
if (((robj*)dictGetVal(kde))->getrefcount(std::memory_order_relaxed) == OBJ_SHARED_REFCOUNT)
{
// shared objects cannot have the expire bit set, create a real object
dictSetVal(m_pdict, kde, dupStringObject((robj*)dictGetVal(kde)));
}
const char *szSubKey = (subkey != nullptr) ? szFromObj(subkey) : nullptr;
if (((robj*)dictGetVal(kde))->FExpires()) {
auto itr = m_setexpire->find((sds)dictGetKey(kde));
serverAssert(itr != m_setexpire->end());
expireEntry eNew(std::move(*itr));
eNew.update(szSubKey, when);
m_setexpire->erase(itr);
m_setexpire->insert(eNew);
}
else
{
expireEntry e((sds)dictGetKey(kde), szSubKey, when);
((robj*)dictGetVal(kde))->SetFExpires(true);
m_setexpire->insert(e);
}
}
void redisDbPersistentData::setExpire(expireEntry &&e)
{
trackkey(e.key());
m_setexpire->insert(e);
}
bool redisDb::FKeyExpires(const char *key)
{
return setexpireUnsafe()->find(key) != setexpire()->end();
}
void redisDbPersistentData::updateValue(dict_iter itr, robj *val)
{
trackkey(itr.key());
dictSetVal(m_pdict, itr.de, val);
}
void redisDbPersistentData::ensure(const char *key)
{
dictEntry *de = dictFind(m_pdict, key);
ensure(de);
}
void redisDbPersistentData::ensure(dictEntry *de)
{
if (de != nullptr && dictGetVal(de) == nullptr)
{
if (m_spdbSnapshot != nullptr)
{
auto itr = m_spdbSnapshot->find((const char*)dictGetKey(de));
serverAssert(itr != m_spdbSnapshot->end());
if (itr.val()->getrefcount(std::memory_order_relaxed) == OBJ_SHARED_REFCOUNT)
{
dictSetVal(m_pdict, de, itr.val());
}
else
{
sds strT = serializeStoredObject(itr.val());
robj *objNew = deserializeStoredObject(strT, sdslen(strT));
sdsfree(strT);
dictSetVal(m_pdict, de, objNew);
serverAssert(objNew->getrefcount(std::memory_order_relaxed) == 1);
serverAssert(objNew->mvcc_tstamp == itr.val()->mvcc_tstamp);
}
}
else
{
serverAssert(m_pstorage != nullptr);
sds key = (sds)dictGetKey(de);
m_pstorage->retrieve(key, sdslen(key), true, [&](const char *, size_t, const void *data, size_t cb){
robj *o = deserializeStoredObject(data, cb);
serverAssert(o != nullptr);
dictSetVal(m_pdict, de, o);
});
}
}
}
void redisDbPersistentData::storeKey(const char *szKey, size_t cchKey, robj *o)
{
sds temp = serializeStoredObject(o);
m_pstorage->insert(szKey, cchKey, temp, sdslen(temp));
sdsfree(temp);
}
void redisDbPersistentData::storeDatabase()
{
dictIterator *di = dictGetIterator(m_pdict);
dictEntry *de;
while ((de = dictNext(di)) != NULL) {
sds key = (sds)dictGetKey(de);
robj *o = (robj*)dictGetVal(de);
storeKey(key, sdslen(key), o);
}
dictReleaseIterator(di);
}
void redisDbPersistentData::processChanges()
{
--m_fTrackingChanges;
serverAssert(m_fTrackingChanges >= 0);
if (m_pstorage != nullptr)
{
if (m_fTrackingChanges == 0)
{
if (m_fAllChanged)
{
m_pstorage->clear();
storeDatabase();
}
else
{
for (auto &str : m_setchanged)
{
sds sdsKey = sdsnewlen(str.data(), str.size());
robj *o = find(sdsKey);
if (o != nullptr)
{
storeKey(str.data(), str.size(), o);
}
else
{
m_pstorage->erase(str.data(), str.size());
}
sdsfree(sdsKey);
}
}
}
}
m_setchanged.clear();
}
std::shared_ptr<redisDbPersistentData> redisDbPersistentData::createSnapshot()
{
serverAssert(GlobalLocksAcquired());
serverAssert(m_spdbSnapshot == nullptr);
auto spdb = std::make_shared<redisDbPersistentData>();
spdb->m_pdict = dictCreate(&dbDictType,spdb.get());
spdb->m_fAllChanged = false;
spdb->m_fTrackingChanges = 0;
spdb->m_pdict->rehashidx = m_pdict->rehashidx;
spdb->m_pdict->iterators++; // fake an iterator so it doesn't rehash
if (m_setexpire != nullptr)
spdb->m_setexpire = new (MALLOC_LOCAL) expireset(*m_setexpire);
for (unsigned iht = 0; iht < 2; ++iht)
{
spdb->m_pdict->ht[iht] = m_pdict->ht[iht];
if (m_pdict->ht[iht].size)
m_pdict->ht[iht].table = (dictEntry**)zcalloc(m_pdict->ht[iht].size*sizeof(dictEntry*), MALLOC_SHARED);
else
m_pdict->ht[iht].table = nullptr;
for (size_t idx = 0; idx < m_pdict->ht[iht].size; ++idx)
{
const dictEntry *deSrc = spdb->m_pdict->ht[iht].table[idx];
dictEntry **pdeDst = &m_pdict->ht[iht].table[idx];
while (deSrc != nullptr)
{
*pdeDst = (dictEntry*)zmalloc(sizeof(dictEntry), MALLOC_SHARED);
(*pdeDst)->key = deSrc->key;
(*pdeDst)->v.val = nullptr;
(*pdeDst)->next = nullptr;
pdeDst = &(*pdeDst)->next;
deSrc = deSrc->next;
}
}
}
m_spdbSnapshot = std::move(spdb);
return m_spdbSnapshot;
}
void redisDbPersistentData::endSnapshot(const redisDbPersistentData *psnapshot)
{
serverAssert(GlobalLocksAcquired());
serverAssert(m_spdbSnapshot.get() == psnapshot);
dictIterator *di = dictGetIterator(m_pdict);
dictEntry *de;
while ((de = dictNext(di)) != NULL)
{
dictEntry *deSnapshot = dictFind(m_spdbSnapshot->m_pdict, dictGetKey(de));
if (dictGetVal(de) == nullptr)
{
if (deSnapshot != nullptr)
{
de->v.val = deSnapshot->v.val;
deSnapshot->v.val = nullptr;
}
}
if (deSnapshot && (dictGetKey(deSnapshot) == dictGetKey(de)))
{
// The key is owned by the parent snapshot, so we modify the DB key dtor
// to ensure the key is not free'd during the delete
m_spdbSnapshot->m_pdict->type = &dbSnapshotDictType;
dictDelete(m_spdbSnapshot->m_pdict, dictGetKey(de));
m_spdbSnapshot->m_pdict->type = &dbDictType;
}
}
dictReleaseIterator(di);
m_spdbSnapshot->m_pdict->iterators--;
m_spdbSnapshot = nullptr;
}
redisDbPersistentData::~redisDbPersistentData()
{
dictRelease(m_pdict);
delete m_setexpire;
}
+23 -34
View File
@@ -126,7 +126,7 @@ void mixStringObjectDigest(unsigned char *digest, robj_roptr o) {
void xorObjectDigest(redisDb *db, robj_roptr keyobj, unsigned char *digest, robj_roptr o) {
uint32_t aux = htonl(o->type);
mixDigest(digest,&aux,sizeof(aux));
expireEntry *pexpire = getExpire(db,keyobj);
expireEntry *pexpire = db->getExpire(keyobj);
long long expiretime = -1;
char buf[128];
@@ -266,8 +266,6 @@ void xorObjectDigest(redisDb *db, robj_roptr keyobj, unsigned char *digest, robj
* a different digest. */
void computeDatasetDigest(unsigned char *final) {
unsigned char digest[20];
dictIterator *di = NULL;
dictEntry *de;
int j;
uint32_t aux;
@@ -276,8 +274,7 @@ void computeDatasetDigest(unsigned char *final) {
for (j = 0; j < cserver.dbnum; j++) {
redisDb *db = g_pserver->db+j;
if (dictSize(db->pdict) == 0) continue;
di = dictGetSafeIterator(db->pdict);
if (db->size() == 0) continue;
/* hash the DB id, so the same dataset moved in a different
* DB will lead to a different digest */
@@ -285,24 +282,21 @@ void computeDatasetDigest(unsigned char *final) {
mixDigest(final,&aux,sizeof(aux));
/* Iterate this DB writing every entry */
while((de = dictNext(di)) != NULL) {
sds key;
robj *keyobj, *o;
db->iterate([&](const char *key, robj *o)->bool {
robj *keyobj;
memset(digest,0,20); /* This key-val digest */
key = (sds)dictGetKey(de);
keyobj = createStringObject(key,sdslen(key));
mixDigest(digest,key,sdslen(key));
o = (robj*)dictGetVal(de);
xorObjectDigest(db,keyobj,digest,o);
/* We can finally xor the key-val digest to the final digest */
xorDigest(final,digest,20);
decrRefCount(keyobj);
}
dictReleaseIterator(di);
return true;
});
}
}
@@ -365,7 +359,7 @@ NULL
} else if (!strcasecmp(szFromObj(c->argv[1]),"reload")) {
rdbSaveInfo rsi, *rsiptr;
rsiptr = rdbPopulateSaveInfo(&rsi);
if (rdbSave(rsiptr) != C_OK) {
if (rdbSave(nullptr, rsiptr) != C_OK) {
addReply(c,shared.err);
return;
}
@@ -394,15 +388,14 @@ NULL
serverLog(LL_WARNING,"Append Only File loaded by DEBUG LOADAOF");
addReply(c,shared.ok);
} else if (!strcasecmp(szFromObj(c->argv[1]),"object") && c->argc == 3) {
dictEntry *de;
robj *val;
const char *strenc;
if ((de = dictFind(c->db->pdict,ptrFromObj(c->argv[2]))) == NULL) {
val = c->db->find(c->argv[2]);
if (val == NULL) {
addReply(c,shared.nokeyerr);
return;
}
val = (robj*)dictGetVal(de);
strenc = strEncoding(val->encoding);
char extra[138] = {0};
@@ -446,16 +439,14 @@ NULL
strenc, rdbSavedObjectLen(val),
val->lru, estimateObjectIdleTime(val)/1000, extra);
} else if (!strcasecmp(szFromObj(c->argv[1]),"sdslen") && c->argc == 3) {
dictEntry *de;
robj *val;
sds key;
auto itr = c->db->find(c->argv[2]);
robj *val = itr.val();
const char *key = itr.key();
if ((de = dictFind(c->db->pdict,ptrFromObj(c->argv[2]))) == NULL) {
if (val == NULL) {
addReply(c,shared.nokeyerr);
return;
}
val = (robj*)dictGetVal(de);
key = (sds)dictGetKey(de);
if (val->type != OBJ_STRING || !sdsEncodedObject(val)) {
addReplyError(c,"Not an sds encoded string.");
@@ -465,16 +456,16 @@ NULL
"val_sds_len:%lld, val_sds_avail:%lld, val_zmalloc: %lld",
(long long) sdslen(key),
(long long) sdsavail(key),
(long long) sdsZmallocSize(key),
(long long) sdsZmallocSize((sds)key),
(long long) sdslen(szFromObj(val)),
(long long) sdsavail(szFromObj(val)),
(long long) getStringObjectSdsUsedMemory(val));
}
} else if (!strcasecmp(szFromObj(c->argv[1]),"ziplist") && c->argc == 3) {
robj *o;
robj_roptr o;
if ((o = objectCommandLookupOrReply(c,c->argv[2],shared.nokeyerr))
== NULL) return;
== nullptr) return;
if (o->encoding != OBJ_ENCODING_ZIPLIST) {
addReplyError(c,"Not an sds encoded string.");
@@ -490,7 +481,7 @@ NULL
if (getLongFromObjectOrReply(c, c->argv[2], &keys, NULL) != C_OK)
return;
dictExpand(c->db->pdict,keys);
c->db->expand(keys);
for (j = 0; j < keys; j++) {
long valsize = 0;
snprintf(buf,sizeof(buf),"%s:%lu",
@@ -641,20 +632,20 @@ NULL
}
stats = sdscatprintf(stats,"[Dictionary HT]\n");
dictGetStats(buf,sizeof(buf),g_pserver->db[dbid].pdict);
g_pserver->db[dbid].getStats(buf,sizeof(buf));
stats = sdscat(stats,buf);
stats = sdscatprintf(stats,"[Expires set]\n");
g_pserver->db[dbid].setexpire->getstats(buf, sizeof(buf));
g_pserver->db[dbid].getExpireStats(buf, sizeof(buf));
stats = sdscat(stats, buf);
addReplyBulkSds(c,stats);
} else if (!strcasecmp(szFromObj(c->argv[1]),"htstats-key") && c->argc == 3) {
robj *o;
robj_roptr o;
dict *ht = NULL;
if ((o = objectCommandLookupOrReply(c,c->argv[2],shared.nokeyerr))
== NULL) return;
== nullptr) return;
/* Get the hash table reference from the object, if possible. */
switch (o->encoding) {
@@ -1256,12 +1247,10 @@ void logCurrentClient(void) {
* selected DB, and if so print info about the associated object. */
if (cc->argc >= 1) {
robj *val, *key;
dictEntry *de;
key = getDecodedObject(cc->argv[1]);
de = dictFind(cc->db->pdict, ptrFromObj(key));
if (de) {
val = (robj*)dictGetVal(de);
val = cc->db->find(key);
if (val) {
serverLog(LL_WARNING,"key '%s' found in DB containing the following object:", (char*)ptrFromObj(key));
serverLogObjectDebugInfo(val);
}
+12 -10
View File
@@ -781,12 +781,14 @@ long defragKey(redisDb *db, dictEntry *de) {
newsds = activeDefragSds(keysds);
if (newsds)
defragged++, de->key = newsds;
if (!db->setexpire->empty()) {
replaceSateliteOSetKeyPtr(*db->setexpire, keysds, newsds);
if (!db->setexpire()->empty()) {
replaceSateliteOSetKeyPtr(*const_cast<expireset*>(db->setexpire()), keysds, newsds);
}
/* Try to defrag robj and / or string value. */
ob = (robj*)dictGetVal(de);
if (ob == nullptr)
return defragged;
if ((newob = activeDefragStringOb(ob, &defragged))) {
de->v.val = newob;
ob = newob;
@@ -853,7 +855,7 @@ void defragScanCallback(void *privdata, const dictEntry *de) {
g_pserver->stat_active_defrag_scanned++;
}
/* Defrag scan callback for each hash table bicket,
/* Defrag scan callback for each hash table bucket,
* used in order to defrag the dictEntry allocations. */
void defragDictBucketCallback(void *privdata, dictEntry **bucketref) {
UNUSED(privdata); /* NOTE: this function is also used by both activeDefragCycle and scanLaterHash, etc. don't use privdata */
@@ -902,9 +904,8 @@ long defragOtherGlobals() {
/* returns 0 more work may or may not be needed (see non-zero cursor),
* and 1 if time is up and more work is needed. */
int defragLaterItem(dictEntry *de, unsigned long *cursor, long long endtime) {
if (de) {
robj *ob = (robj*)dictGetVal(de);
int defragLaterItem(robj *ob, unsigned long *cursor, long long endtime) {
if (ob) {
if (ob->type == OBJ_LIST) {
g_pserver->stat_active_defrag_hits += scanLaterList(ob);
*cursor = 0; /* list has no scan, we must finish it in one go */
@@ -959,11 +960,11 @@ int defragLaterStep(redisDb *db, long long endtime) {
}
/* each time we enter this function we need to fetch the key from the dict again (if it still exists) */
dictEntry *de = dictFind(db->pdict, current_key);
robj *o = db->find(current_key);
key_defragged = g_pserver->stat_active_defrag_hits;
do {
int quit = 0;
if (defragLaterItem(de, &cursor, endtime))
if (defragLaterItem(o, &cursor, endtime))
quit = 1; /* time is up, we didn't finish all the work */
/* Don't start a new BIG key in this loop, this is because the
@@ -1045,7 +1046,7 @@ void activeDefragCycle(void) {
mstime_t latency;
int quit = 0;
if (g_pserver->aof_child_pid!=-1 || g_pserver->rdb_child_pid!=-1)
if (g_pserver->aof_child_pid!=-1 || g_pserver->FRdbSaveInProgress())
return; /* Defragging memory while there's a fork will just do damage. */
/* Once a second, check if we the fragmentation justfies starting a scan
@@ -1112,7 +1113,8 @@ void activeDefragCycle(void) {
break; /* this will exit the function and we'll continue on the next cycle */
}
cursor = dictScan(db->pdict, cursor, defragScanCallback, defragDictBucketCallback, db);
// we actually look at the objects too but defragScanCallback can handle missing values
cursor = dictScan(db->dictUnsafeKeyOnly(), cursor, defragScanCallback, defragDictBucketCallback, db);
/* Once in 16 scan iterations, 512 pointer reallocations. or 64 keys
* (if we have a lot of pointers in one hash bucket or rehasing),
+34 -32
View File
@@ -87,7 +87,7 @@ unsigned int LRU_CLOCK(void) {
/* Given an object returns the min number of milliseconds the object was never
* requested, using an approximated LRU algorithm. */
unsigned long long estimateObjectIdleTime(robj *o) {
unsigned long long estimateObjectIdleTime(robj_roptr o) {
unsigned long long lruclock = LRU_CLOCK();
if (lruclock >= o->lru) {
return (lruclock - o->lru) * LRU_CLOCK_RESOLUTION;
@@ -252,19 +252,20 @@ struct visitFunctor
return count < g_pserver->maxmemory_samples;
}
};
void evictionPoolPopulate(int dbid, dict *dbdict, expireset *setexpire, struct evictionPoolEntry *pool)
void evictionPoolPopulate(int dbid, redisDb *db, expireset *setexpire, struct evictionPoolEntry *pool)
{
if (setexpire != nullptr)
{
visitFunctor visitor { dbid, dbdict, pool, 0 };
visitFunctor visitor { dbid, db->dictUnsafeKeyOnly(), pool, 0 };
setexpire->random_visit(visitor);
}
else
{
dictEntry **samples = (dictEntry**)alloca(g_pserver->maxmemory_samples * sizeof(dictEntry*));
int count = dictGetSomeKeys(dbdict,samples,g_pserver->maxmemory_samples);
int count = dictGetSomeKeys(db->dictUnsafeKeyOnly(),samples,g_pserver->maxmemory_samples);
for (int j = 0; j < count; j++) {
robj *o = (robj*)dictGetVal(samples[j]);
serverAssert(o != nullptr); // BUG!!! We have to get the info we need here without permanently rehydrating the obj
processEvictionCandidate(dbid, (sds)dictGetKey(samples[j]), o, nullptr, pool);
}
}
@@ -504,16 +505,16 @@ int freeMemoryIfNeeded(void) {
db = g_pserver->db+i;
if (g_pserver->maxmemory_policy & MAXMEMORY_FLAG_ALLKEYS)
{
if ((keys = dictSize(db->pdict)) != 0) {
evictionPoolPopulate(i, db->pdict, nullptr, pool);
if ((keys = db->size()) != 0) {
evictionPoolPopulate(i, db, nullptr, pool);
total_keys += keys;
}
}
else
{
keys = db->setexpire->size();
keys = db->expireSize();
if (keys != 0)
evictionPoolPopulate(i, db->pdict, db->setexpire, pool);
evictionPoolPopulate(i, db, db->setexpireUnsafe(), pool);
total_keys += keys;
}
}
@@ -525,9 +526,9 @@ int freeMemoryIfNeeded(void) {
bestdbid = pool[k].dbid;
sds key = nullptr;
dictEntry *de = dictFind(g_pserver->db[pool[k].dbid].pdict,pool[k].key);
if (de != nullptr && (g_pserver->maxmemory_policy & MAXMEMORY_FLAG_ALLKEYS || ((robj*)dictGetVal(de))->FExpires()))
key = (sds)dictGetKey(de);
auto itr = g_pserver->db[pool[k].dbid].find(pool[k].key);
if (itr != nullptr && (g_pserver->maxmemory_policy & MAXMEMORY_FLAG_ALLKEYS || itr.val()->FExpires()))
key = itr.key();
/* Remove the entry from the pool. */
if (pool[k].key != pool[k].cached)
@@ -559,18 +560,18 @@ int freeMemoryIfNeeded(void) {
db = g_pserver->db+j;
if (g_pserver->maxmemory_policy == MAXMEMORY_ALLKEYS_RANDOM)
{
if (dictSize(db->pdict) != 0) {
dictEntry *de = dictGetRandomKey(db->pdict);
bestkey = (sds)dictGetKey(de);
if (db->size() != 0) {
auto itr = db->random();
bestkey = itr.key();
bestdbid = j;
break;
}
}
else
{
if (!db->setexpire->empty())
if (db->expireSize())
{
bestkey = (sds)db->setexpire->random_value().key();
bestkey = (sds)db->random_expire().key();
bestdbid = j;
break;
}
@@ -581,16 +582,17 @@ int freeMemoryIfNeeded(void) {
/* Finally remove the selected key. */
if (bestkey) {
db = g_pserver->db+bestdbid;
robj *keyobj = createStringObject(bestkey,sdslen(bestkey));
propagateExpire(db,keyobj,g_pserver->lazyfree_lazy_eviction);
/* We compute the amount of memory freed by db*Delete() alone.
* It is possible that actually the memory needed to propagate
* the DEL in AOF and replication link is greater than the one
* we are freeing removing the key, but we can't account for
* that otherwise we would never exit the loop.
*
* AOF and Output buffer memory will be freed eventually so
* we only care about memory used by the key space. */
* It is possible that actually the memory needed to propagate
* the DEL in AOF and replication link is greater than the one
* we are freeing removing the key, but we can't account for
* that otherwise we would never exit the loop.
*
* AOF and Output buffer memory will be freed eventually so
* we only care about memory used by the key space. */
delta = (long long) zmalloc_used_memory();
latencyStartMonitor(eviction_latency);
if (g_pserver->lazyfree_lazy_eviction)
@@ -609,18 +611,18 @@ int freeMemoryIfNeeded(void) {
keys_freed++;
/* When the memory to free starts to be big enough, we may
* start spending so much time here that is impossible to
* deliver data to the slaves fast enough, so we force the
* transmission here inside the loop. */
* start spending so much time here that is impossible to
* deliver data to the slaves fast enough, so we force the
* transmission here inside the loop. */
if (slaves) flushSlavesOutputBuffers();
/* Normally our stop condition is the ability to release
* a fixed, pre-computed amount of memory. However when we
* are deleting objects in another thread, it's better to
* check, from time to time, if we already reached our target
* memory, since the "mem_freed" amount is computed only
* across the dbAsyncDelete() call, while the thread can
* release the memory all the time. */
* a fixed, pre-computed amount of memory. However when we
* are deleting objects in another thread, it's better to
* check, from time to time, if we already reached our target
* memory, since the "mem_freed" amount is computed only
* across the dbAsyncDelete() call, while the thread can
* release the memory all the time. */
if (g_pserver->lazyfree_lazy_eviction && !(keys_freed % 16)) {
if (getMaxmemoryState(NULL,NULL,NULL,NULL) == C_OK) {
/* Let's satisfy our stop condition. */
+14 -15
View File
@@ -74,8 +74,7 @@ void activeExpireCycleExpire(redisDb *db, expireEntry &e, long long now) {
}
expireEntryFat *pfat = e.pfatentry();
dictEntry *de = dictFind(db->pdict, e.key());
robj *val = (robj*)dictGetVal(de);
robj *val = db->find(e.key());
int deleted = 0;
while (!pfat->FEmpty())
{
@@ -157,7 +156,7 @@ void expireMemberCore(client *c, robj *key, robj *subkey, long long basetime, lo
/* No key, return zero. */
robj *val = lookupKeyWriteOrReply(c, key, shared.czero);
if (val == NULL) {
if (val == nullptr) {
return;
}
@@ -296,7 +295,7 @@ void activeExpireCycle(int type) {
now = mstime();
/* If there is nothing to expire try next DB ASAP. */
if (db->setexpire->empty())
if (db->setexpireUnsafe()->empty())
{
db->avg_ttl = 0;
db->last_expire_set = now;
@@ -306,7 +305,7 @@ void activeExpireCycle(int type) {
size_t expired = 0;
size_t tried = 0;
long long check = ACTIVE_EXPIRE_CYCLE_FAST_DURATION; // assume a check is roughly 1us. It isn't but good enough
db->expireitr = db->setexpire->enumerate(db->expireitr, now, [&](expireEntry &e) __attribute__((always_inline)) {
db->expireitr = db->setexpireUnsafe()->enumerate(db->expireitr, now, [&](expireEntry &e) __attribute__((always_inline)) {
if (e.when() < now)
{
activeExpireCycleExpire(db, e, now);
@@ -406,16 +405,16 @@ void expireSlaveKeys(void) {
redisDb *db = g_pserver->db+dbid;
// the expire is hashed based on the key pointer, so we need the point in the main db
dictEntry *deMain = dictFind(db->pdict, keyname);
auto itr = db->setexpire->end();
if (deMain != nullptr)
itr = db->setexpire->find((sds)dictGetKey(deMain));
auto itrDB = db->find(keyname);
auto itrExpire = db->setexpire()->end();
if (itrDB != nullptr)
itrExpire = db->setexpireUnsafe()->find(itrDB.key());
int expired = 0;
if (itr != db->setexpire->end())
if (itrExpire != db->setexpire()->end())
{
if (itr->when() < start) {
activeExpireCycleExpire(g_pserver->db+dbid,*itr,start);
if (itrExpire->when() < start) {
activeExpireCycleExpire(g_pserver->db+dbid,*itrExpire,start);
expired = 1;
}
}
@@ -424,7 +423,7 @@ void expireSlaveKeys(void) {
* corresponding bit in the new bitmap we set as value.
* At the end of the loop if the bitmap is zero, it means we
* no longer need to keep track of this key. */
if (itr != db->setexpire->end() && !expired) {
if (itrExpire != db->setexpire()->end() && !expired) {
noexpire++;
new_dbids |= (uint64_t)1 << dbid;
}
@@ -592,7 +591,7 @@ void ttlGenericCommand(client *c, int output_ms) {
/* The key exists. Return -1 if it has no expire, or the actual
* TTL value otherwise. */
expireEntry *pexpire = getExpire(c->db,c->argv[1]);
expireEntry *pexpire = c->db->getExpire(c->argv[1]);
if (c->argc == 2) {
// primary expire
@@ -648,7 +647,7 @@ void persistCommand(client *c) {
addReply(c,shared.czero);
}
} else if (c->argc == 3) {
if (removeSubkeyExpire(c->db, c->argv[1], c->argv[2])) {
if (c->db->removeSubkeyExpire(c->argv[1], c->argv[2])) {
addReply(c,shared.cone);
g_pserver->dirty++;
} else {
+2 -2
View File
@@ -46,13 +46,13 @@
#define ECCENT (sqrt(1.0 - (RATIO *RATIO)))
#define COM (0.5 * ECCENT)
/// @brief The usual PI/180 constant
const double DEG_TO_RAD = 0.017453292519943295769236907684886;
/// @brief Earth's quatratic mean radius for WGS-84
const double EARTH_RADIUS_IN_METERS = 6372797.560856;
const double MERCATOR_MAX = 20037726.37;
#if 0 //unused
const double MERCATOR_MIN = -20037726.37;
#endif
static inline double deg_rad(double ang) { return ang * D_R; }
static inline double rad_deg(double ang) { return ang / D_R; }
+20 -13
View File
@@ -51,18 +51,18 @@ size_t lazyfreeGetFreeEffort(robj *obj) {
* a lazy free list instead of being freed synchronously. The lazy free list
* will be reclaimed in a different bio.c thread. */
#define LAZYFREE_THRESHOLD 64
int dbAsyncDelete(redisDb *db, robj *key) {
bool redisDbPersistentData::asyncDelete(robj *key) {
/* If the value is composed of a few allocations, to free in a lazy way
* is actually just slower... So under a certain limit we just free
* the object synchronously. */
dictEntry *de = dictUnlink(db->pdict,ptrFromObj(key));
dictEntry *de = dictUnlink(m_pdict,ptrFromObj(key));
if (de) {
robj *val = (robj*)dictGetVal(de);
if (val->FExpires())
{
/* Deleting an entry from the expires dict will not free the sds of
* the key, because it is shared with the main dictionary. */
removeExpireCore(db,key,de);
removeExpire(key,dict_iter(de));
}
size_t free_effort = lazyfreeGetFreeEffort(val);
@@ -78,21 +78,25 @@ int dbAsyncDelete(redisDb *db, robj *key) {
if (free_effort > LAZYFREE_THRESHOLD && val->getrefcount(std::memory_order_relaxed) == 1) {
atomicIncr(lazyfree_objects,1);
bioCreateBackgroundJob(BIO_LAZY_FREE,val,NULL,NULL);
dictSetVal(db->pdict,de,NULL);
dictSetVal(m_pdict,de,NULL);
}
}
/* Release the key-val pair, or just the key if we set the val
* field to NULL in order to lazy free it later. */
if (de) {
dictFreeUnlinkedEntry(db->pdict,de);
dictFreeUnlinkedEntry(m_pdict,de);
if (g_pserver->cluster_enabled) slotToKeyDel(key);
return 1;
return true;
} else {
return 0;
return false;
}
}
int dbAsyncDelete(redisDb *db, robj *key) {
return db->asyncDelete(key);
}
/* Free an object, if the object is huge enough, free it in async way. */
void freeObjAsync(robj *o) {
size_t free_effort = lazyfreeGetFreeEffort(o);
@@ -107,12 +111,15 @@ void freeObjAsync(robj *o) {
/* Empty a Redis DB asynchronously. What the function does actually is to
* create a new empty set of hash tables and scheduling the old ones for
* lazy freeing. */
void emptyDbAsync(redisDb *db) {
dict *oldht1 = db->pdict;
auto *set = db->setexpire;
db->setexpire = new (MALLOC_LOCAL) expireset();
db->expireitr = db->setexpire->end();
db->pdict = dictCreate(&dbDictType,NULL);
void redisDbPersistentData::emptyDbAsync() {
dict *oldht1 = m_pdict;
auto *set = m_setexpire;
m_setexpire = new (MALLOC_LOCAL) expireset();
m_pdict = dictCreate(&dbDictType,this);
if (m_pstorage != nullptr)
m_pstorage->clear();
if (m_fTrackingChanges)
m_fAllChanged = true;
atomicIncr(lazyfree_objects,dictSize(oldht1));
bioCreateBackgroundJob(BIO_LAZY_FREE,NULL,oldht1,set);
}
+1 -1
View File
@@ -1644,7 +1644,7 @@ int RM_UnlinkKey(RedisModuleKey *key) {
* If no TTL is associated with the key or if the key is empty,
* REDISMODULE_NO_EXPIRE is returned. */
mstime_t RM_GetExpire(RedisModuleKey *key) {
expireEntry *pexpire = getExpire(key->db,key->key);
expireEntry *pexpire = key->db->getExpire(key->key);
mstime_t expire = -1;
if (pexpire != nullptr)
pexpire->FGetPrimaryExpire(&expire);
+1 -1
View File
@@ -333,7 +333,7 @@ void touchWatchedKeysOnFlush(int dbid) {
* key exists, mark the client as dirty, as the key will be
* removed. */
if (dbid == -1 || wk->db->id == dbid) {
if (dictFind(wk->db->pdict, ptrFromObj(wk->key)) != NULL)
if (wk->db->find(wk->key) != NULL)
c->flags |= CLIENT_DIRTY_CAS;
}
}
+130 -25
View File
@@ -789,7 +789,7 @@ size_t streamRadixTreeMemoryUsage(rax *rax) {
* case of aggregated data types where only "sample_size" elements
* are checked and averaged to estimate the total size. */
#define OBJ_COMPUTE_SIZE_DEF_SAMPLES 5 /* Default sample size. */
size_t objectComputeSize(robj *o, size_t sample_size) {
size_t objectComputeSize(robj_roptr o, size_t sample_size) {
sds ele, ele2;
dict *d;
dictIterator *di;
@@ -800,7 +800,7 @@ size_t objectComputeSize(robj *o, size_t sample_size) {
if(o->encoding == OBJ_ENCODING_INT) {
asize = sizeof(*o);
} else if(o->encoding == OBJ_ENCODING_RAW) {
asize = sdsAllocSize(szFromObj(o))+sizeof(*o);
asize = sdsAllocSize((sds)szFromObj(o))+sizeof(*o);
} else if(o->encoding == OBJ_ENCODING_EMBSTR) {
asize = sdslen(szFromObj(o))+2+sizeof(*o);
} else {
@@ -1054,20 +1054,20 @@ struct redisMemOverhead *getMemoryOverheadData(void) {
for (j = 0; j < cserver.dbnum; j++) {
redisDb *db = g_pserver->db+j;
long long keyscount = dictSize(db->pdict);
long long keyscount = db->size();
if (keyscount==0) continue;
mh->total_keys += keyscount;
mh->db = (decltype(mh->db))zrealloc(mh->db,sizeof(mh->db[0])*(mh->num_dbs+1), MALLOC_LOCAL);
mh->db[mh->num_dbs].dbid = j;
mem = dictSize(db->pdict) * sizeof(dictEntry) +
dictSlots(db->pdict) * sizeof(dictEntry*) +
dictSize(db->pdict) * sizeof(robj);
mem = db->size() * sizeof(dictEntry) +
db->slots() * sizeof(dictEntry*) +
db->size() * sizeof(robj);
mh->db[mh->num_dbs].overhead_ht_main = mem;
mem_total+=mem;
mem = db->setexpire->bytes_used();
mem = db->setexpire()->bytes_used();
mh->db[mh->num_dbs].overhead_ht_expires = mem;
mem_total+=mem;
@@ -1246,15 +1246,12 @@ void objectSetLRUOrLFU(robj *val, long long lfu_freq, long long lru_idle,
/* This is a helper function for the OBJECT command. We need to lookup keys
* without any modification of LRU or other parameters. */
robj *objectCommandLookup(client *c, robj *key) {
dictEntry *de;
if ((de = dictFind(c->db->pdict,ptrFromObj(key))) == NULL) return NULL;
return (robj*) dictGetVal(de);
robj_roptr objectCommandLookup(client *c, robj *key) {
return c->db->find(key);
}
robj *objectCommandLookupOrReply(client *c, robj *key, robj *reply) {
robj *o = objectCommandLookup(c,key);
robj_roptr objectCommandLookupOrReply(client *c, robj *key, robj *reply) {
robj_roptr o = objectCommandLookup(c,key);
if (!o) addReply(c, reply);
return o;
@@ -1263,7 +1260,7 @@ robj *objectCommandLookupOrReply(client *c, robj *key, robj *reply) {
/* Object command allows to inspect the internals of an Redis Object.
* Usage: OBJECT <refcount|encoding|idletime|freq> <key> */
void objectCommand(client *c) {
robj *o;
robj_roptr o;
if (c->argc == 2 && !strcasecmp(szFromObj(c->argv[1]),"help")) {
const char *help[] = {
@@ -1276,15 +1273,15 @@ NULL
addReplyHelp(c, help);
} else if (!strcasecmp(szFromObj(c->argv[1]),"refcount") && c->argc == 3) {
if ((o = objectCommandLookupOrReply(c,c->argv[2],shared.null[c->resp]))
== NULL) return;
== nullptr) return;
addReplyLongLong(c,o->getrefcount(std::memory_order_relaxed));
} else if (!strcasecmp(szFromObj(c->argv[1]),"encoding") && c->argc == 3) {
if ((o = objectCommandLookupOrReply(c,c->argv[2],shared.null[c->resp]))
== NULL) return;
== nullptr) return;
addReplyBulkCString(c,strEncoding(o->encoding));
} else if (!strcasecmp(szFromObj(c->argv[1]),"idletime") && c->argc == 3) {
if ((o = objectCommandLookupOrReply(c,c->argv[2],shared.null[c->resp]))
== NULL) return;
== nullptr) return;
if (g_pserver->maxmemory_policy & MAXMEMORY_FLAG_LFU) {
addReplyError(c,"An LFU maxmemory policy is selected, idle time not tracked. Please note that when switching between policies at runtime LRU and LFU data will take some time to adjust.");
return;
@@ -1292,7 +1289,7 @@ NULL
addReplyLongLong(c,estimateObjectIdleTime(o)/1000);
} else if (!strcasecmp(szFromObj(c->argv[1]),"freq") && c->argc == 3) {
if ((o = objectCommandLookupOrReply(c,c->argv[2],shared.null[c->resp]))
== NULL) return;
== nullptr) return;
if (!(g_pserver->maxmemory_policy & MAXMEMORY_FLAG_LFU)) {
addReplyError(c,"An LFU maxmemory policy is not selected, access frequency not tracked. Please note that when switching between policies at runtime LRU and LFU data will take some time to adjust.");
return;
@@ -1301,10 +1298,10 @@ NULL
* in case of the key has not been accessed for a long time,
* because we update the access time only
* when the key is read or overwritten. */
addReplyLongLong(c,LFUDecrAndReturn(o));
addReplyLongLong(c,LFUDecrAndReturn(o.unsafe_robjcast()));
} else if (!strcasecmp(szFromObj(c->argv[1]), "lastmodified") && c->argc == 3) {
if ((o = objectCommandLookupOrReply(c,c->argv[2],shared.null[c->resp]))
== NULL) return;
== nullptr) return;
addReplyLongLong(c, (g_pserver->mstime - (o->mvcc_tstamp >> MVCC_MS_SHIFT)) / 1000);
} else {
addReplySubcommandSyntaxError(c);
@@ -1327,7 +1324,6 @@ NULL
};
addReplyHelp(c, help);
} else if (!strcasecmp(szFromObj(c->argv[1]),"usage") && c->argc >= 3) {
dictEntry *de;
long long samples = OBJ_COMPUTE_SIZE_DEF_SAMPLES;
for (int j = 3; j < c->argc; j++) {
if (!strcasecmp(szFromObj(c->argv[j]),"samples") &&
@@ -1346,12 +1342,14 @@ NULL
return;
}
}
if ((de = dictFind(c->db->pdict,ptrFromObj(c->argv[2]))) == NULL) {
auto itr = c->db->find(c->argv[2]);
if (itr == nullptr) {
addReplyNull(c, shared.nullbulk);
return;
}
size_t usage = objectComputeSize((robj*)dictGetVal(de),samples);
usage += sdsAllocSize((sds)dictGetKey(de));
size_t usage = objectComputeSize(itr.val(),samples);
usage += sdsAllocSize(itr.key());
usage += sizeof(dictEntry);
addReplyLongLong(c,usage);
} else if (!strcasecmp(szFromObj(c->argv[1]),"stats") && c->argc == 2) {
@@ -1495,3 +1493,110 @@ void redisObject::setrefcount(unsigned ref)
serverAssert(!FExpires());
refcount.store(ref, std::memory_order_relaxed);
}
sds serializeStoredStringObject(sds str, robj_roptr o)
{
str = sdscatlen(str, &(*o), sizeof(robj));
switch (o->encoding)
{
case OBJ_ENCODING_RAW:
str = sdscatsds(str, (sds)szFromObj(o));
break;
case OBJ_ENCODING_INT:
break; //nop
case OBJ_ENCODING_EMBSTR:
size_t cb = zmalloc_size(o.unsafe_robjcast());
if (cb > sizeof(robj))
{
str = sdscatlen(str, o.unsafe_robjcast() + 1, cb - sizeof(robj));
}
break;
}
return str;
}
robj *deserializeStoredStringObject(const char *data, size_t cb)
{
const robj *oT = (const robj*)data;
robj *newObject = nullptr;
switch (oT->encoding)
{
case OBJ_ENCODING_INT:
serverAssert(cb == sizeof(robj));
[[fallthrough]];
case OBJ_ENCODING_EMBSTR:
newObject = (robj*)zmalloc(cb, MALLOC_LOCAL);
memcpy(newObject, data, cb);
newObject->setrefcount(1);
return newObject;
case OBJ_ENCODING_RAW:
newObject = (robj*)zmalloc(sizeof(robj), MALLOC_SHARED);
memcpy(newObject, data, sizeof(robj));
newObject->m_ptr = sdsnewlen(SDS_NOINIT,cb-sizeof(robj));
memcpy(newObject->m_ptr, data+sizeof(robj), cb-sizeof(robj));
newObject->setrefcount(1);
return newObject;
}
serverPanic("Unknown string object encoding from storage");
return nullptr;
}
robj *deserializeStoredObject(const void *data, size_t cb)
{
switch (((char*)data)[0])
{
case RDB_TYPE_STRING:
return deserializeStoredStringObject(((char*)data)+1, cb-1);
default:
rio payload;
int type;
robj *obj;
rioInitWithConstBuffer(&payload,data,cb);
if (((type = rdbLoadObjectType(&payload)) == -1) ||
((obj = rdbLoadObject(type,&payload,nullptr, OBJ_MVCC_INVALID)) == nullptr))
{
serverPanic("Bad data format");
}
if (rdbLoadType(&payload) == RDB_OPCODE_AUX)
{
robj *auxkey, *auxval;
if ((auxkey = rdbLoadStringObject(&payload)) == NULL) goto eoferr;
if ((auxval = rdbLoadStringObject(&payload)) == NULL) {
decrRefCount(auxkey);
goto eoferr;
}
if (strcasecmp(szFromObj(auxkey), "mvcc-tstamp") == 0) {
obj->mvcc_tstamp = strtoull(szFromObj(auxval), nullptr, 10);
}
decrRefCount(auxkey);
decrRefCount(auxval);
}
eoferr:
return obj;
}
serverPanic("Unknown object type loading from storage");
}
sds serializeStoredObject(robj_roptr o)
{
switch (o->type)
{
case OBJ_STRING:
{
sds sdsT = sdsnewlen(nullptr, 1);
sdsT[0] = RDB_TYPE_STRING;
return serializeStoredStringObject(sdsT, o);
}
default:
rio rdb;
createDumpPayload(&rdb,o,nullptr);
return (sds)rdb.io.buffer.ptr;
}
serverPanic("Attempting to store unknown object type");
}
+1 -1
View File
@@ -705,7 +705,7 @@ int raxGenericInsert(rax *rax, unsigned char *s, size_t len, void *data, void **
nodesize = sizeof(raxNode)+trimmedlen+raxPadding(trimmedlen)+
sizeof(raxNode*);
if (h->iskey && !h->isnull) nodesize += sizeof(void*);
trimmed = rax_malloc(nodesize);
trimmed = zcalloc(nodesize, MALLOC_LOCAL);//rax_malloc(nodesize);
}
if (postfixlen) {
+2 -2
View File
@@ -6,7 +6,7 @@ extern "C" {
#include <sys/wait.h>
/* Save the DB on disk. Return C_ERR on error, C_OK on success. */
int rdbSaveS3(char *s3bucket, rdbSaveInfo *rsi)
int rdbSaveS3(char *s3bucket, redisDbPersistentData **rgpdb, rdbSaveInfo *rsi)
{
int status = EXIT_FAILURE;
int fd[2];
@@ -40,7 +40,7 @@ int rdbSaveS3(char *s3bucket, rdbSaveInfo *rsi)
return C_ERR;
}
if (rdbSaveFp(fp, rsi) != C_OK)
if (rdbSaveFp(fp, rgpdb, rsi) != C_OK)
{
fclose(fp);
return C_ERR;
+284 -196
View File
@@ -42,6 +42,9 @@
#include <arpa/inet.h>
#include <sys/stat.h>
#include <sys/param.h>
#include <thread>
#include <future>
#include "aelocker.h"
#define rdbExitReportCorruptRDB(...) rdbCheckThenExit(__LINE__,__VA_ARGS__)
@@ -1103,21 +1106,21 @@ int rdbSaveInfoAuxFields(rio *rdb, int flags, rdbSaveInfo *rsi) {
if (rsi) {
if (rdbSaveAuxFieldStrInt(rdb,"repl-stream-db",rsi->repl_stream_db)
== -1) return -1;
if (rdbSaveAuxFieldStrStr(rdb,"repl-id",g_pserver->replid)
if (rdbSaveAuxFieldStrStr(rdb,"repl-id",rsi->repl_id)
== -1) return -1;
if (rdbSaveAuxFieldStrInt(rdb,"repl-offset",g_pserver->master_repl_offset)
if (rdbSaveAuxFieldStrInt(rdb,"repl-offset",rsi->master_repl_offset)
== -1) return -1;
}
if (rdbSaveAuxFieldStrInt(rdb,"aof-preamble",aof_preamble) == -1) return -1;
return 1;
}
int saveKey(rio *rdb, redisDb *db, int flags, size_t *processed, const char *keystr, robj *o)
int saveKey(rio *rdb, redisDbPersistentData *db, int flags, size_t *processed, const char *keystr, robj *o)
{
robj key;
initStaticStringObject(key,(char*)keystr);
expireEntry *pexpire = getExpire(db, &key);
expireEntry *pexpire = db->getExpire(&key);
if (rdbSaveKeyValuePair(rdb,&key,o,pexpire) == -1)
return 0;
@@ -1142,9 +1145,9 @@ int saveKey(rio *rdb, redisDb *db, int flags, size_t *processed, const char *key
* When the function returns C_ERR and if 'error' is not NULL, the
* integer pointed by 'error' is set to the value of errno just after the I/O
* error. */
int rdbSaveRio(rio *rdb, int *error, int flags, rdbSaveInfo *rsi) {
dictIterator *di = NULL;
int rdbSaveRio(rio *rdb, redisDbPersistentData **rgpdb, int *error, int flags, rdbSaveInfo *rsi) {
dictEntry *de;
dictIterator *di = NULL;
char magic[10];
int j;
uint64_t cksum;
@@ -1154,13 +1157,12 @@ int rdbSaveRio(rio *rdb, int *error, int flags, rdbSaveInfo *rsi) {
rdb->update_cksum = rioGenericUpdateChecksum;
snprintf(magic,sizeof(magic),"REDIS%04d",RDB_VERSION);
if (rdbWriteRaw(rdb,magic,9) == -1) goto werr;
if (rdbSaveInfoAuxFields(rdb,flags,rsi) == -1) goto werr;
for (j = 0; j < cserver.dbnum; j++) {
redisDb *db = g_pserver->db+j;
dict *d = db->pdict;
if (dictSize(d) == 0) continue;
di = dictGetSafeIterator(d);
redisDbPersistentData *db = rgpdb[j];
if (db->size() == 0) continue;
/* Write the SELECT DB opcode */
if (rdbSaveType(rdb,RDB_OPCODE_SELECTDB) == -1) goto werr;
@@ -1171,43 +1173,47 @@ int rdbSaveRio(rio *rdb, int *error, int flags, rdbSaveInfo *rsi) {
* However this does not limit the actual size of the DB to load since
* these sizes are just hints to resize the hash tables. */
uint64_t db_size, expires_size;
db_size = dictSize(db->pdict);
expires_size = db->setexpire->size();
db_size = db->size();
expires_size = db->expireSize();
if (rdbSaveType(rdb,RDB_OPCODE_RESIZEDB) == -1) goto werr;
if (rdbSaveLen(rdb,db_size) == -1) goto werr;
if (rdbSaveLen(rdb,expires_size) == -1) goto werr;
/* Iterate this DB writing every entry */
size_t ckeysExpired = 0;
while((de = dictNext(di)) != NULL) {
sds keystr = (sds)dictGetKey(de);
robj *o = (robj*)dictGetVal(de);
bool fSavedAll = db->iterate([&](const char *keystr, robj *o)->bool {
if (o->FExpires())
++ckeysExpired;
if (!saveKey(rdb, db, flags, &processed, keystr, o))
goto werr;
}
serverAssert(ckeysExpired == db->setexpire->size());
dictReleaseIterator(di);
di = NULL; /* So that we don't release it again on error. */
return false;
return !g_pserver->rdbThreadVars.fRdbThreadCancel;
});
if (!fSavedAll)
goto werr;
serverAssert(ckeysExpired == db->expireSize());
}
/* If we are storing the replication information on disk, persist
* the script cache as well: on successful PSYNC after a restart, we need
* to be able to process any EVALSHA inside the replication backlog the
* master will send us. */
{
AeLocker lock;
lock.arm(nullptr);
if (rsi && dictSize(g_pserver->lua_scripts)) {
di = dictGetIterator(g_pserver->lua_scripts);
while((de = dictNext(di)) != NULL) {
robj *body = (robj*)dictGetVal(de);
if (rdbSaveAuxField(rdb,"lua",3,szFromObj(body),sdslen(szFromObj(body))) == -1)
goto werr;
if (g_pserver->rdbThreadVars.fRdbThreadCancel)
goto werr;
}
dictReleaseIterator(di);
di = NULL; /* So that we don't release it again on error. */
}
} // AeLocker end scope
/* EOF opcode */
if (rdbSaveType(rdb,RDB_OPCODE_EOF) == -1) goto werr;
@@ -1233,7 +1239,7 @@ werr:
* While the suffix is the 40 bytes hex string we announced in the prefix.
* This way processes receiving the payload can understand when it ends
* without doing any processing of the content. */
int rdbSaveRioWithEOFMark(rio *rdb, int *error, rdbSaveInfo *rsi) {
int rdbSaveRioWithEOFMark(rio *rdb, redisDbPersistentData **rgpdb, int *error, rdbSaveInfo *rsi) {
char eofmark[RDB_EOF_MARK_SIZE];
getRandomHexChars(eofmark,RDB_EOF_MARK_SIZE);
@@ -1241,7 +1247,7 @@ int rdbSaveRioWithEOFMark(rio *rdb, int *error, rdbSaveInfo *rsi) {
if (rioWrite(rdb,"$EOF:",5) == 0) goto werr;
if (rioWrite(rdb,eofmark,RDB_EOF_MARK_SIZE) == 0) goto werr;
if (rioWrite(rdb,"\r\n",2) == 0) goto werr;
if (rdbSaveRio(rdb,error,RDB_SAVE_NONE,rsi) == C_ERR) goto werr;
if (rdbSaveRio(rdb,rgpdb,error,RDB_SAVE_NONE,rsi) == C_ERR) goto werr;
if (rioWrite(rdb,eofmark,RDB_EOF_MARK_SIZE) == 0) goto werr;
return C_OK;
@@ -1251,7 +1257,7 @@ werr: /* Write error. */
return C_ERR;
}
int rdbSaveFp(FILE *fp, rdbSaveInfo *rsi)
int rdbSaveFp(FILE *fp, redisDbPersistentData **rgpdb, rdbSaveInfo *rsi)
{
int error = 0;
rio rdb;
@@ -1261,31 +1267,41 @@ int rdbSaveFp(FILE *fp, rdbSaveInfo *rsi)
if (g_pserver->rdb_save_incremental_fsync)
rioSetAutoSync(&rdb,REDIS_AUTOSYNC_BYTES);
if (rdbSaveRio(&rdb,&error,RDB_SAVE_NONE,rsi) == C_ERR) {
if (rdbSaveRio(&rdb,rgpdb,&error,RDB_SAVE_NONE,rsi) == C_ERR) {
errno = error;
return C_ERR;
}
return C_OK;
}
int rdbSave(rdbSaveInfo *rsi)
int rdbSave(redisDbPersistentData **rgpdb, rdbSaveInfo *rsi)
{
std::vector<redisDbPersistentData*> vecdb;
if (rgpdb == nullptr)
{
for (int idb = 0; idb < cserver.dbnum; ++idb)
{
vecdb.push_back(&g_pserver->db[idb]);
}
rgpdb = vecdb.data();
}
int err = C_OK;
if (g_pserver->rdb_filename != NULL)
err = rdbSaveFile(g_pserver->rdb_filename, rsi);
err = rdbSaveFile(g_pserver->rdb_filename, rgpdb, rsi);
if (err == C_OK && g_pserver->rdb_s3bucketpath != NULL)
err = rdbSaveS3(g_pserver->rdb_s3bucketpath, rsi);
err = rdbSaveS3(g_pserver->rdb_s3bucketpath, rgpdb, rsi);
return err;
}
/* Save the DB on disk. Return C_ERR on error, C_OK on success. */
int rdbSaveFile(char *filename, rdbSaveInfo *rsi) {
int rdbSaveFile(char *filename, redisDbPersistentData **rgpdb, rdbSaveInfo *rsi) {
char tmpfile[256];
char cwd[MAXPATHLEN]; /* Current working dir path for error messages. */
FILE *fp;
snprintf(tmpfile,256,"temp-%d.rdb", (int) getpid());
snprintf(tmpfile,256,"temp-%d.rdb", g_pserver->rdbThreadVars.tmpfileNum);
fp = fopen(tmpfile,"w");
if (!fp) {
char *cwdp = getcwd(cwd,MAXPATHLEN);
@@ -1298,7 +1314,7 @@ int rdbSaveFile(char *filename, rdbSaveInfo *rsi) {
return C_ERR;
}
if (rdbSaveFp(fp, rsi) == C_ERR){
if (rdbSaveFp(fp, rgpdb, rsi) == C_ERR){
goto werr;
}
@@ -1323,9 +1339,12 @@ int rdbSaveFile(char *filename, rdbSaveInfo *rsi) {
}
serverLog(LL_NOTICE,"DB saved on disk");
g_pserver->dirty = 0;
g_pserver->lastsave = time(NULL);
g_pserver->lastbgsave_status = C_OK;
if (serverTL != nullptr)
{
g_pserver->dirty = 0;
g_pserver->lastsave = time(NULL);
g_pserver->lastbgsave_status = C_OK;
}
return C_OK;
werr:
@@ -1335,63 +1354,99 @@ werr:
return C_ERR;
}
struct rdbSaveThreadArgs
{
rdbSaveInfo rsi;
redisDbPersistentData *rgpdb[1]; // NOTE: Variable Length
};
void *rdbSaveThread(void *vargs)
{
rdbSaveThreadArgs *args = reinterpret_cast<rdbSaveThreadArgs*>(vargs);
serverAssert(serverTL == nullptr);
int retval = rdbSave(args->rgpdb, &args->rsi);
if (retval == C_OK) {
size_t private_dirty = zmalloc_get_private_dirty(-1);
if (private_dirty) {
serverLog(LL_NOTICE,
"RDB: %zu MB of memory used by copy-on-write",
private_dirty/(1024*1024));
}
g_pserver->child_info_data.cow_size = private_dirty;
sendChildInfo(CHILD_INFO_TYPE_RDB);
}
aeAcquireLock();
for (int idb = 0; idb < cserver.dbnum; ++idb)
g_pserver->db[idb].endSnapshot(args->rgpdb[idb]);
aeReleaseLock();
zfree(args);
return (retval == C_OK) ? (void*)0 : (void*)1;
}
int launchRdbSaveThread(pthread_t &child, rdbSaveInfo *rsi)
{
rdbSaveThreadArgs *args = (rdbSaveThreadArgs*)zmalloc(sizeof(rdbSaveThreadArgs) + ((cserver.dbnum-1)*sizeof(redisDbPersistentData*)), MALLOC_LOCAL);
rdbSaveInfo rsiT = RDB_SAVE_INFO_INIT;
if (rsi == nullptr)
rsi = &rsiT;
memcpy(&args->rsi, rsi, sizeof(rdbSaveInfo));
memcpy(&args->rsi.repl_id, g_pserver->replid, sizeof(g_pserver->replid));
args->rsi.master_repl_offset = g_pserver->master_repl_offset;
for (int idb = 0; idb < cserver.dbnum; ++idb)
args->rgpdb[idb] = g_pserver->db[idb].createSnapshot().get();
g_pserver->rdbThreadVars.tmpfileNum++;
g_pserver->rdbThreadVars.fRdbThreadCancel = false;
if (pthread_create(&child, NULL, rdbSaveThread, args)) {
for (int idb = 0; idb < cserver.dbnum; ++idb)
g_pserver->db[idb].endSnapshot(args->rgpdb[idb]);
zfree(args);
return C_ERR;
}
return C_OK;
}
int rdbSaveBackground(rdbSaveInfo *rsi) {
pid_t childpid;
pthread_t child;
long long start;
if (g_pserver->aof_child_pid != -1 || g_pserver->rdb_child_pid != -1) return C_ERR;
if (g_pserver->aof_child_pid != -1 || g_pserver->FRdbSaveInProgress()) return C_ERR;
g_pserver->dirty_before_bgsave = g_pserver->dirty;
g_pserver->lastbgsave_try = time(NULL);
openChildInfoPipe();
start = ustime();
if ((childpid = fork()) == 0) {
int retval;
/* Child */
closeListeningSockets(0);
redisSetProcTitle("keydb-rdb-bgsave");
retval = rdbSave(rsi);
if (retval == C_OK) {
size_t private_dirty = zmalloc_get_private_dirty(-1);
if (private_dirty) {
serverLog(LL_NOTICE,
"RDB: %zu MB of memory used by copy-on-write",
private_dirty/(1024*1024));
}
g_pserver->child_info_data.cow_size = private_dirty;
sendChildInfo(CHILD_INFO_TYPE_RDB);
}
exitFromChild((retval == C_OK) ? 0 : 1);
} else {
/* Parent */
g_pserver->stat_fork_time = ustime()-start;
g_pserver->stat_fork_rate = (double) zmalloc_used_memory() * 1000000 / g_pserver->stat_fork_time / (1024*1024*1024); /* GB per second. */
latencyAddSampleIfNeeded("fork",g_pserver->stat_fork_time/1000);
if (childpid == -1) {
closeChildInfoPipe();
g_pserver->lastbgsave_status = C_ERR;
serverLog(LL_WARNING,"Can't save in background: fork: %s",
strerror(errno));
return C_ERR;
}
serverLog(LL_NOTICE,"Background saving started by pid %d",childpid);
g_pserver->rdb_save_time_start = time(NULL);
g_pserver->rdb_child_pid = childpid;
g_pserver->rdb_child_type = RDB_CHILD_TYPE_DISK;
updateDictResizePolicy();
return C_OK;
g_pserver->stat_fork_time = ustime()-start;
g_pserver->stat_fork_rate = (double) zmalloc_used_memory() * 1000000 / g_pserver->stat_fork_time / (1024*1024*1024); /* GB per second. */
latencyAddSampleIfNeeded("fork",g_pserver->stat_fork_time/1000);
if (launchRdbSaveThread(child, rsi) != C_OK) {
closeChildInfoPipe();
g_pserver->lastbgsave_status = C_ERR;
serverLog(LL_WARNING,"Can't save in background: fork: %s",
strerror(errno));
return C_ERR;
}
return C_OK; /* unreached */
serverLog(LL_NOTICE,"Background saving started");
g_pserver->rdb_save_time_start = time(NULL);
g_pserver->rdbThreadVars.fRdbThreadActive = true;
g_pserver->rdbThreadVars.rdb_child_thread = child;
g_pserver->rdb_child_type = RDB_CHILD_TYPE_DISK;
updateDictResizePolicy();
return C_OK;
}
void rdbRemoveTempFile(pid_t childpid) {
void rdbRemoveTempFile(int tmpfileNum) {
char tmpfile[256];
snprintf(tmpfile,sizeof(tmpfile),"temp-%d.rdb", (int) childpid);
snprintf(tmpfile,sizeof(tmpfile),"temp-%d.rdb", tmpfileNum);
unlink(tmpfile);
}
@@ -1998,7 +2053,7 @@ int rdbLoadRio(rio *rdb, rdbSaveInfo *rsi, int loading_aof) {
goto eoferr;
if ((expires_size = rdbLoadLen(rdb,NULL)) == RDB_LENERR)
goto eoferr;
dictExpand(db->pdict,db_size);
db->expand(db_size);
continue; /* Read next opcode. */
} else if (type == RDB_OPCODE_AUX) {
/* AUX: generic string-string fields. Use to add state to RDB
@@ -2236,7 +2291,7 @@ void backgroundSaveDoneHandlerDisk(int exitcode, int bysignal) {
serverLog(LL_WARNING,
"Background saving terminated by signal %d", bysignal);
latencyStartMonitor(latency);
rdbRemoveTempFile(g_pserver->rdb_child_pid);
rdbRemoveTempFile(g_pserver->rdbThreadVars.tmpfileNum);
latencyEndMonitor(latency);
latencyAddSampleIfNeeded("rdb-unlink-temp-file",latency);
/* SIGUSR1 is whitelisted, so we have a way to kill a child without
@@ -2244,7 +2299,7 @@ void backgroundSaveDoneHandlerDisk(int exitcode, int bysignal) {
if (bysignal != SIGUSR1)
g_pserver->lastbgsave_status = C_ERR;
}
g_pserver->rdb_child_pid = -1;
g_pserver->rdbThreadVars.fRdbThreadActive = false;
g_pserver->rdb_child_type = RDB_CHILD_TYPE_NONE;
g_pserver->rdb_save_time_last = time(NULL)-g_pserver->rdb_save_time_start;
g_pserver->rdb_save_time_start = -1;
@@ -2269,7 +2324,7 @@ void backgroundSaveDoneHandlerSocket(int exitcode, int bysignal) {
serverLog(LL_WARNING,
"Background transfer terminated by signal %d", bysignal);
}
g_pserver->rdb_child_pid = -1;
g_pserver->rdbThreadVars.fRdbThreadActive = false;
g_pserver->rdb_child_type = RDB_CHILD_TYPE_NONE;
g_pserver->rdb_save_time_start = -1;
@@ -2368,26 +2423,114 @@ void backgroundSaveDoneHandler(int exitcode, int bysignal) {
* the child did not exit for an error, but because we wanted), and performs
* the cleanup needed. */
void killRDBChild(void) {
kill(g_pserver->rdb_child_pid,SIGUSR1);
rdbRemoveTempFile(g_pserver->rdb_child_pid);
g_pserver->rdbThreadVars.fRdbThreadCancel = true;
void *rval;
pthread_join(g_pserver->rdbThreadVars.rdb_child_thread,&rval);
g_pserver->rdbThreadVars.fRdbThreadActive = false;
g_pserver->rdbThreadVars.fRdbThreadCancel = false;
rdbRemoveTempFile(g_pserver->rdbThreadVars.tmpfileNum);
closeChildInfoPipe();
updateDictResizePolicy();
}
struct rdbSaveSocketThreadArgs
{
rdbSaveInfo rsi;
int *fds;
int numfds;
uint64_t *clientids;
redisDbPersistentData *rgpdb[1];
};
void *rdbSaveToSlavesSocketsThread(void *vargs)
{
/* Child */
serverAssert(serverTL == nullptr);
rdbSaveSocketThreadArgs *args = (rdbSaveSocketThreadArgs*)vargs;
int retval;
rio slave_sockets;
rioInitWithFdset(&slave_sockets,args->fds,args->numfds);
zfree(args->fds);
args->fds = nullptr;
retval = rdbSaveRioWithEOFMark(&slave_sockets,args->rgpdb,NULL,&args->rsi);
if (retval == C_OK && rioFlush(&slave_sockets) == 0)
retval = C_ERR;
if (retval == C_OK) {
size_t private_dirty = zmalloc_get_private_dirty(-1);
if (private_dirty) {
serverLog(LL_NOTICE,
"RDB: %zu MB of memory used by copy-on-write",
private_dirty/(1024*1024));
}
g_pserver->child_info_data.cow_size = private_dirty;
sendChildInfo(CHILD_INFO_TYPE_RDB);
/* If we are returning OK, at least one replica was served
* with the RDB file as expected, so we need to send a report
* to the parent via the pipe. The format of the message is:
*
* <len> <replica[0].id> <replica[0].error> ...
*
* len, replica IDs, and replica errors, are all uint64_t integers,
* so basically the reply is composed of 64 bits for the len field
* plus 2 additional 64 bit integers for each entry, for a total
* of 'len' entries.
*
* The 'id' represents the replica's client ID, so that the master
* can match the report with a specific replica, and 'error' is
* set to 0 if the replication process terminated with a success
* or the error code if an error occurred. */
void *msg = zmalloc(sizeof(uint64_t)*(1+2*args->numfds), MALLOC_LOCAL);
uint64_t *len = (uint64_t*)msg;
uint64_t *ids = len+1;
int j, msglen;
*len = args->numfds;
for (j = 0; j < args->numfds; j++) {
*ids++ = args->clientids[j];
*ids++ = slave_sockets.io.fdset.state[j];
}
/* Write the message to the parent. If we have no good slaves or
* we are unable to transfer the message to the parent, we exit
* with an error so that the parent will abort the replication
* process with all the childre that were waiting. */
msglen = sizeof(uint64_t)*(1+2*args->numfds);
if (*len == 0 ||
write(g_pserver->rdb_pipe_write_result_to_parent,msg,msglen)
!= msglen)
{
retval = C_ERR;
}
zfree(msg);
}
aeAcquireLock();
for (int idb = 0; idb < cserver.dbnum; ++idb)
g_pserver->db[idb].endSnapshot(args->rgpdb[idb]);
aeReleaseLock();
zfree(args->clientids);
zfree(args);
rioFreeFdset(&slave_sockets);
return (retval == C_OK) ? (void*)0 : (void*)1;
}
/* Spawn an RDB child that writes the RDB to the sockets of the slaves
* that are currently in SLAVE_STATE_WAIT_BGSAVE_START state. */
int rdbSaveToSlavesSockets(rdbSaveInfo *rsi) {
serverAssert(GlobalLocksAcquired());
int *fds;
uint64_t *clientids;
int numfds;
listNode *ln;
listIter li;
pid_t childpid;
pthread_t child;
long long start;
int pipefds[2];
rdbSaveSocketThreadArgs *args = nullptr;
if (g_pserver->aof_child_pid != -1 || g_pserver->rdb_child_pid != -1) return C_ERR;
if (g_pserver->aof_child_pid != -1 || g_pserver->FRdbSaveInProgress()) return C_ERR;
/* Before to fork, create a pipe that will be used in order to
* send back to the parent the IDs of the slaves that successfully
@@ -2396,22 +2539,27 @@ int rdbSaveToSlavesSockets(rdbSaveInfo *rsi) {
g_pserver->rdb_pipe_read_result_from_child = pipefds[0];
g_pserver->rdb_pipe_write_result_to_parent = pipefds[1];
args = (rdbSaveSocketThreadArgs*)zmalloc(sizeof(rdbSaveSocketThreadArgs) + sizeof(redisDbPersistentData*)*(cserver.dbnum-1), MALLOC_LOCAL);
/* Collect the file descriptors of the slaves we want to transfer
* the RDB to, which are i WAIT_BGSAVE_START state. */
fds = (int*)zmalloc(sizeof(int)*listLength(g_pserver->slaves), MALLOC_LOCAL);
args->fds = (int*)zmalloc(sizeof(int)*listLength(g_pserver->slaves), MALLOC_LOCAL);
/* We also allocate an array of corresponding client IDs. This will
* be useful for the child process in order to build the report
* (sent via unix pipe) that will be sent to the parent. */
clientids = (uint64_t*)zmalloc(sizeof(uint64_t)*listLength(g_pserver->slaves), MALLOC_LOCAL);
numfds = 0;
args->clientids = (uint64_t*)zmalloc(sizeof(uint64_t)*listLength(g_pserver->slaves), MALLOC_LOCAL);
args->numfds = 0;
memcpy(&args->rsi, rsi, sizeof(rdbSaveInfo));
memcpy(&args->rsi.repl_id, g_pserver->replid, sizeof(g_pserver->replid));
args->rsi.master_repl_offset = g_pserver->master_repl_offset;
listRewind(g_pserver->slaves,&li);
while((ln = listNext(&li))) {
client *replica = (client*)ln->value;
if (replica->replstate == SLAVE_STATE_WAIT_BGSAVE_START) {
clientids[numfds] = replica->id;
fds[numfds++] = replica->fd;
args->clientids[args->numfds] = replica->id;
args->fds[args->numfds++] = replica->fd;
replicationSetupSlaveForFullResync(replica,getPsyncInitialOffset());
/* Put the socket in blocking mode to simplify RDB transfer.
* We'll restore it when the children returns (since duped socket
@@ -2424,126 +2572,63 @@ int rdbSaveToSlavesSockets(rdbSaveInfo *rsi) {
/* Create the child process. */
openChildInfoPipe();
start = ustime();
if ((childpid = fork()) == 0) {
/* Child */
int retval;
rio slave_sockets;
rioInitWithFdset(&slave_sockets,fds,numfds);
zfree(fds);
for (int idb = 0; idb < cserver.dbnum; ++idb)
args->rgpdb[idb] = g_pserver->db[idb].createSnapshot().get();
closeListeningSockets(0);
redisSetProcTitle("keydb-rdb-to-slaves");
g_pserver->rdbThreadVars.tmpfileNum++;
g_pserver->rdbThreadVars.fRdbThreadCancel = false;
if (pthread_create(&child, nullptr, rdbSaveToSlavesSocketsThread, args)) {
serverLog(LL_WARNING,"Can't save in background: fork: %s",
strerror(errno));
retval = rdbSaveRioWithEOFMark(&slave_sockets,NULL,rsi);
if (retval == C_OK && rioFlush(&slave_sockets) == 0)
retval = C_ERR;
/* 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;
int j;
if (retval == C_OK) {
size_t private_dirty = zmalloc_get_private_dirty(-1);
if (private_dirty) {
serverLog(LL_NOTICE,
"RDB: %zu MB of memory used by copy-on-write",
private_dirty/(1024*1024));
}
g_pserver->child_info_data.cow_size = private_dirty;
sendChildInfo(CHILD_INFO_TYPE_RDB);
/* If we are returning OK, at least one replica was served
* with the RDB file as expected, so we need to send a report
* to the parent via the pipe. The format of the message is:
*
* <len> <replica[0].id> <replica[0].error> ...
*
* len, replica IDs, and replica errors, are all uint64_t integers,
* so basically the reply is composed of 64 bits for the len field
* plus 2 additional 64 bit integers for each entry, for a total
* of 'len' entries.
*
* The 'id' represents the replica's client ID, so that the master
* can match the report with a specific replica, and 'error' is
* set to 0 if the replication process terminated with a success
* or the error code if an error occurred. */
void *msg = zmalloc(sizeof(uint64_t)*(1+2*numfds), MALLOC_LOCAL);
uint64_t *len = (uint64_t*)msg;
uint64_t *ids = len+1;
int j, msglen;
*len = numfds;
for (j = 0; j < numfds; j++) {
*ids++ = clientids[j];
*ids++ = slave_sockets.io.fdset.state[j];
}
/* Write the message to the parent. If we have no good slaves or
* we are unable to transfer the message to the parent, we exit
* with an error so that the parent will abort the replication
* process with all the childre that were waiting. */
msglen = sizeof(uint64_t)*(1+2*numfds);
if (*len == 0 ||
write(g_pserver->rdb_pipe_write_result_to_parent,msg,msglen)
!= msglen)
{
retval = C_ERR;
}
zfree(msg);
}
zfree(clientids);
rioFreeFdset(&slave_sockets);
exitFromChild((retval == C_OK) ? 0 : 1);
} else {
/* Parent */
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;
int j;
for (j = 0; j < numfds; j++) {
if (replica->id == clientids[j]) {
replica->replstate = SLAVE_STATE_WAIT_BGSAVE_START;
break;
}
for (j = 0; j < args->numfds; j++) {
if (replica->id == args->clientids[j]) {
replica->replstate = SLAVE_STATE_WAIT_BGSAVE_START;
break;
}
}
close(pipefds[0]);
close(pipefds[1]);
closeChildInfoPipe();
} else {
g_pserver->stat_fork_time = ustime()-start;
g_pserver->stat_fork_rate = (double) zmalloc_used_memory() * 1000000 / g_pserver->stat_fork_time / (1024*1024*1024); /* GB per second. */
latencyAddSampleIfNeeded("fork",g_pserver->stat_fork_time/1000);
serverLog(LL_NOTICE,"Background RDB transfer started by pid %d",
childpid);
g_pserver->rdb_save_time_start = time(NULL);
g_pserver->rdb_child_pid = childpid;
g_pserver->rdb_child_type = RDB_CHILD_TYPE_SOCKET;
updateDictResizePolicy();
}
zfree(clientids);
zfree(fds);
return (childpid == -1) ? C_ERR : C_OK;
close(pipefds[0]);
close(pipefds[1]);
closeChildInfoPipe();
for (int idb = 0; idb < cserver.dbnum; ++idb)
g_pserver->db[idb].endSnapshot(args->rgpdb[idb]);
zfree(args->clientids);
zfree(args->fds);
zfree(args);
return C_ERR;
}
g_pserver->stat_fork_time = ustime()-start;
g_pserver->stat_fork_rate = (double) zmalloc_used_memory() * 1000000 / g_pserver->stat_fork_time / (1024*1024*1024); /* GB per second. */
latencyAddSampleIfNeeded("fork",g_pserver->stat_fork_time/1000);
serverLog(LL_NOTICE,"Background RDB transfer started");
g_pserver->rdb_save_time_start = time(NULL);
g_pserver->rdbThreadVars.rdb_child_thread = child;
g_pserver->rdbThreadVars.fRdbThreadActive = true;
g_pserver->rdb_child_type = RDB_CHILD_TYPE_SOCKET;
updateDictResizePolicy();
return C_OK; /* Unreached. */
}
void saveCommand(client *c) {
if (g_pserver->rdb_child_pid != -1) {
if (g_pserver->FRdbSaveInProgress()) {
addReplyError(c,"Background save already in progress");
return;
}
rdbSaveInfo rsi, *rsiptr;
rsiptr = rdbPopulateSaveInfo(&rsi);
if (rdbSave(rsiptr) == C_OK) {
if (rdbSave(nullptr, rsiptr) == C_OK) {
addReply(c,shared.ok);
} else {
addReply(c,shared.err);
@@ -2568,7 +2653,7 @@ void bgsaveCommand(client *c) {
rdbSaveInfo rsi, *rsiptr;
rsiptr = rdbPopulateSaveInfo(&rsi);
if (g_pserver->rdb_child_pid != -1) {
if (g_pserver->FRdbSaveInProgress()) {
addReplyError(c,"Background save already in progress");
} else if (g_pserver->aof_child_pid != -1) {
if (schedule) {
@@ -2600,6 +2685,9 @@ rdbSaveInfo *rdbPopulateSaveInfo(rdbSaveInfo *rsi) {
rdbSaveInfo rsi_init = RDB_SAVE_INFO_INIT;
*rsi = rsi_init;
memcpy(rsi->repl_id, g_pserver->replid, sizeof(g_pserver->replid));
rsi->master_repl_offset = g_pserver->master_repl_offset;
/* If the instance is a master, we can populate the replication info
* only when repl_backlog is not NULL. If the repl_backlog is NULL,
* it means that the instance isn't in any replication chains. In this
+5 -5
View File
@@ -139,11 +139,11 @@ int rdbLoad(rdbSaveInfo *rsi);
int rdbLoadFile(const char *filename, rdbSaveInfo *rsi);
int rdbSaveBackground(rdbSaveInfo *rsi);
int rdbSaveToSlavesSockets(rdbSaveInfo *rsi);
void rdbRemoveTempFile(pid_t childpid);
int rdbSave(rdbSaveInfo *rsi);
int rdbSaveFile(char *filename, rdbSaveInfo *rsi);
int rdbSaveFp(FILE *pf, rdbSaveInfo *rsi);
int rdbSaveS3(char *path, rdbSaveInfo *rsi);
void rdbRemoveTempFile(int childpid);
int rdbSave(redisDbPersistentData **rgpdb, rdbSaveInfo *rsi);
int rdbSaveFile(char *filename, redisDbPersistentData **rgpdb, rdbSaveInfo *rsi);
int rdbSaveFp(FILE *pf, redisDbPersistentData **rgpdb, rdbSaveInfo *rsi);
int rdbSaveS3(char *path, redisDbPersistentData **rgpdb, rdbSaveInfo *rsi);
int rdbLoadS3(char *path, rdbSaveInfo *rsi);
ssize_t rdbSaveObject(rio *rdb, robj_roptr o, robj *key);
size_t rdbSavedObjectLen(robj *o);
+6 -7
View File
@@ -858,7 +858,7 @@ void syncCommand(client *c) {
}
/* CASE 1: BGSAVE is in progress, with disk target. */
if (g_pserver->rdb_child_pid != -1 &&
if (g_pserver->FRdbSaveInProgress() &&
g_pserver->rdb_child_type == RDB_CHILD_TYPE_DISK)
{
/* Ok a background save is in progress. Let's check if it is a good
@@ -889,7 +889,7 @@ void syncCommand(client *c) {
}
/* CASE 2: BGSAVE is in progress, with socket target. */
} else if (g_pserver->rdb_child_pid != -1 &&
} else if (g_pserver->FRdbSaveInProgress() &&
g_pserver->rdb_child_type == RDB_CHILD_TYPE_SOCKET)
{
/* There is an RDB child process but it is writing directly to
@@ -1510,13 +1510,12 @@ void readSyncBulkPayload(aeEventLoop *el, int fd, void *privdata, int mask) {
int aof_is_enabled = g_pserver->aof_state != AOF_OFF;
/* Ensure background save doesn't overwrite synced data */
if (g_pserver->rdb_child_pid != -1) {
if (g_pserver->FRdbSaveInProgress()) {
serverLog(LL_NOTICE,
"Replica is about to load the RDB file received from the "
"master, but there is a pending RDB child running. "
"Killing process %ld and removing its temp file to avoid "
"any race",
(long) g_pserver->rdb_child_pid);
"Cancelling RDB the save and removing its temp file to avoid "
"any race");
killRDBChild();
}
@@ -3143,7 +3142,7 @@ void replicationCron(void) {
* In case of diskless replication, we make sure to wait the specified
* number of seconds (according to configuration) so that other slaves
* have the time to arrive before we start streaming. */
if (g_pserver->rdb_child_pid == -1 && g_pserver->aof_child_pid == -1) {
if (!g_pserver->FRdbSaveInProgress() && g_pserver->aof_child_pid == -1) {
time_t idle, max_idle = 0;
int slaves_waiting = 0;
int mincapa = -1;
+28
View File
@@ -73,6 +73,14 @@ static size_t rioBufferRead(rio *r, void *buf, size_t len) {
return 1;
}
static size_t rioConstBufferRead(rio *r, void *buf, size_t len) {
if (r->io.buffer.len-r->io.buffer.pos < (off_t)len)
return 0; /* not enough buffer to return len bytes. */
memcpy(buf,r->io.buffer.ptr+r->io.buffer.pos,len);
r->io.buffer.pos += len;
return 1;
}
/* Returns read/write position in buffer. */
static off_t rioBufferTell(rio *r) {
return r->io.buffer.pos;
@@ -97,12 +105,32 @@ static const rio rioBufferIO = {
{ { NULL, 0 } } /* union for io-specific vars */
};
static const rio rioConstBufferIO = {
rioConstBufferRead,
nullptr,
rioBufferTell,
rioBufferFlush,
NULL, /* update_checksum */
0, /* current checksum */
0, /* bytes read or written */
0, /* read/write chunk size */
{ { NULL, 0 } } /* union for io-specific vars */
};
void rioInitWithBuffer(rio *r, sds s) {
*r = rioBufferIO;
r->io.buffer.ptr = s;
r->io.buffer.pos = 0;
}
void rioInitWithConstBuffer(rio *r, const void *buf, size_t cb)
{
*r = rioConstBufferIO;
r->io.buffer.ptr = (sds)buf;
r->io.buffer.pos = 0;
r->io.buffer.len = cb;
}
/* --------------------- Stdio file pointer implementation ------------------- */
/* Returns 1 or 0 for success/failure. */
+2
View File
@@ -70,6 +70,7 @@ struct _rio {
struct {
sds ptr;
off_t pos;
off_t len; // For const buffers only
} buffer;
/* Stdio file pointer target. */
struct {
@@ -130,6 +131,7 @@ static inline int rioFlush(rio *r) {
void rioInitWithFile(rio *r, FILE *fp);
void rioInitWithBuffer(rio *r, sds s);
void rioInitWithConstBuffer(rio *r, const void *rgch, size_t cch);
void rioInitWithFdset(rio *r, int *fds, int numfds);
void rioFreeFdset(rio *r);
+2 -2
View File
@@ -93,9 +93,9 @@ public:
return end();
}
setiter end()
setiter end() const
{
setiter itr(this);
setiter itr(const_cast<semiorderedset<T,T_KEY,MEMMOVE_SAFE>*>(this));
itr.idxPrimary = m_data.size();
return itr;
}
+72 -41
View File
@@ -1179,6 +1179,21 @@ int dictSdsKeyCompare(void *privdata, const void *key1,
return memcmp(key1, key2, l1) == 0;
}
void dictSdsNOPDestructor(void *, void *) {}
void dictDbKeyDestructor(void *privdata, void *key)
{
redisDbPersistentData *owner = (redisDbPersistentData*)privdata;
serverAssert(owner != nullptr);
if (owner->m_spdbSnapshot != nullptr)
{
dictEntry *deSnapshot = dictFind(owner->m_spdbSnapshot->m_pdict, key);
if (deSnapshot && (key == dictGetKey(deSnapshot)))
return; // don't free, it's now owned by the snapshot
}
sdsfree((sds)key);
}
/* A case insensitive version used for the command lookup table and other
* places where case insensitive non binary-safe comparison is needed. */
int dictSdsKeyCaseCompare(void *privdata, const void *key1,
@@ -1314,10 +1329,19 @@ dictType dbDictType = {
NULL, /* key dup */
NULL, /* val dup */
dictSdsKeyCompare, /* key compare */
dictSdsDestructor, /* key destructor */
dictDbKeyDestructor, /* key destructor */
dictObjectDestructor /* val destructor */
};
dictType dbSnapshotDictType = {
dictSdsHash,
NULL,
NULL,
dictSdsKeyCompare,
dictSdsNOPDestructor,
dictObjectDestructor,
};
/* g_pserver->lua_scripts sha (as sds string) -> scripts (as robj) cache. */
dictType shaScriptObjectDictType = {
dictSdsCaseHash, /* hash function */
@@ -1439,8 +1463,7 @@ int htNeedsResize(dict *dict) {
/* If the percentage of used slots in the HT reaches HASHTABLE_MIN_FILL
* we resize the hash table to save memory */
void tryResizeHashTables(int dbid) {
if (htNeedsResize(g_pserver->db[dbid].pdict))
dictResize(g_pserver->db[dbid].pdict);
g_pserver->db[dbid].tryResize();
}
/* Our hash table implementation performs rehashing incrementally while
@@ -1450,10 +1473,10 @@ void tryResizeHashTables(int dbid) {
*
* The function returns 1 if some rehashing was performed, otherwise 0
* is returned. */
int incrementallyRehash(int dbid) {
int redisDbPersistentData::incrementallyRehash() {
/* Keys dictionary */
if (dictIsRehashing(g_pserver->db[dbid].pdict)) {
dictRehashMilliseconds(g_pserver->db[dbid].pdict,1);
if (dictIsRehashing(m_pdict)) {
dictRehashMilliseconds(m_pdict,1);
return 1; /* already used our millisecond for this loop... */
}
return 0;
@@ -1466,7 +1489,7 @@ int incrementallyRehash(int dbid) {
* for dict.c to resize the hash tables accordingly to the fact we have o not
* running childs. */
void updateDictResizePolicy(void) {
if (g_pserver->rdb_child_pid == -1 && g_pserver->aof_child_pid == -1)
if (!g_pserver->FRdbSaveInProgress() && g_pserver->aof_child_pid == -1)
dictEnableResize();
else
dictDisableResize();
@@ -1713,7 +1736,7 @@ void databasesCron(void) {
/* Perform hash tables rehashing if needed, but only if there are no
* other processes saving the DB on disk. Otherwise rehashing is bad
* as will cause a lot of copy-on-write of memory pages. */
if (g_pserver->rdb_child_pid == -1 && g_pserver->aof_child_pid == -1) {
if (!g_pserver->FRdbSaveInProgress() && g_pserver->aof_child_pid == -1) {
/* We use global counters so if we stop the computation at a given
* DB we'll be able to start from the successive in the next
* cron loop iteration. */
@@ -1734,7 +1757,7 @@ void databasesCron(void) {
/* Rehash */
if (g_pserver->activerehashing) {
for (j = 0; j < dbs_per_call; j++) {
int work_done = incrementallyRehash(rehash_db);
int work_done = g_pserver->db[rehash_db].incrementallyRehash();
if (work_done) {
/* If the function did some work, stop here, we'll do
* more at the next cron loop. */
@@ -1893,9 +1916,9 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
for (j = 0; j < cserver.dbnum; j++) {
long long size, used, vkeys;
size = dictSlots(g_pserver->db[j].pdict);
used = dictSize(g_pserver->db[j].pdict);
vkeys = g_pserver->db[j].setexpire->size();
size = g_pserver->db[j].slots();
used = g_pserver->db[j].size();
vkeys = g_pserver->db[j].expireSize();
if (used || vkeys) {
serverLog(LL_VERBOSE,"DB %d: %lld keys (%lld volatile) in %lld slots HT.",j,used,vkeys,size);
/* dictPrintStats(g_pserver->dict); */
@@ -1922,20 +1945,37 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
/* Start a scheduled AOF rewrite if this was requested by the user while
* a BGSAVE was in progress. */
if (g_pserver->rdb_child_pid == -1 && g_pserver->aof_child_pid == -1 &&
if (!g_pserver->FRdbSaveInProgress() && g_pserver->aof_child_pid == -1 &&
g_pserver->aof_rewrite_scheduled)
{
rewriteAppendOnlyFileBackground();
}
/* Check if a background saving or AOF rewrite in progress terminated. */
if (g_pserver->rdb_child_pid != -1 || g_pserver->aof_child_pid != -1 ||
if (g_pserver->FRdbSaveInProgress() || g_pserver->aof_child_pid != -1 ||
ldbPendingChildren())
{
int statloc;
pid_t pid;
if ((pid = wait3(&statloc,WNOHANG,NULL)) != 0) {
if (g_pserver->FRdbSaveInProgress())
{
void *rval = nullptr;
if (pthread_tryjoin_np(g_pserver->rdbThreadVars.rdb_child_thread, &rval))
{
if (errno != EBUSY && errno != EAGAIN)
serverLog(LL_WARNING, "Error joining the background RDB save thread: %s\n", strerror(errno));
}
else
{
int exitcode = (int)reinterpret_cast<ptrdiff_t>(rval);
backgroundSaveDoneHandler(exitcode,0);
if (exitcode == 0) receiveChildInfo();
updateDictResizePolicy();
closeChildInfoPipe();
}
}
else if ((pid = wait3(&statloc,WNOHANG,NULL)) != 0) {
int exitcode = WEXITSTATUS(statloc);
int bysignal = 0;
@@ -1943,13 +1983,9 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
if (pid == -1) {
serverLog(LL_WARNING,"wait3() returned an error: %s. "
"rdb_child_pid = %d, aof_child_pid = %d",
"aof_child_pid = %d",
strerror(errno),
(int) g_pserver->rdb_child_pid,
(int) g_pserver->aof_child_pid);
} else if (pid == g_pserver->rdb_child_pid) {
backgroundSaveDoneHandler(exitcode,bysignal);
if (!bysignal && exitcode == 0) receiveChildInfo();
} else if (pid == g_pserver->aof_child_pid) {
backgroundRewriteDoneHandler(exitcode,bysignal);
if (!bysignal && exitcode == 0) receiveChildInfo();
@@ -1990,7 +2026,7 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
/* Trigger an AOF rewrite if needed. */
if (g_pserver->aof_state == AOF_ON &&
g_pserver->rdb_child_pid == -1 &&
!g_pserver->FRdbSaveInProgress() &&
g_pserver->aof_child_pid == -1 &&
g_pserver->aof_rewrite_perc &&
g_pserver->aof_current_size > g_pserver->aof_rewrite_min_size)
@@ -2046,7 +2082,7 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
* Note: this code must be after the replicationCron() call above so
* make sure when refactoring this file to keep this order. This is useful
* because we want to give priority to RDB savings for replication. */
if (g_pserver->rdb_child_pid == -1 && g_pserver->aof_child_pid == -1 &&
if (!g_pserver->FRdbSaveInProgress() && g_pserver->aof_child_pid == -1 &&
g_pserver->rdb_bgsave_scheduled &&
(g_pserver->unixtime-g_pserver->lastbgsave_try > CONFIG_BGSAVE_RETRY_DELAY ||
g_pserver->lastbgsave_status == C_OK))
@@ -2932,16 +2968,7 @@ void initServer(void) {
/* Create the Redis databases, and initialize other internal state. */
for (int j = 0; j < cserver.dbnum; j++) {
new (&g_pserver->db[j]) redisDb;
g_pserver->db[j].pdict = dictCreate(&dbDictType,NULL);
g_pserver->db[j].setexpire = new(MALLOC_LOCAL) expireset();
g_pserver->db[j].expireitr = g_pserver->db[j].setexpire->end();
g_pserver->db[j].blocking_keys = dictCreate(&keylistDictType,NULL);
g_pserver->db[j].ready_keys = dictCreate(&objectKeyPointerValueDictType,NULL);
g_pserver->db[j].watched_keys = dictCreate(&keylistDictType,NULL);
g_pserver->db[j].id = j;
g_pserver->db[j].avg_ttl = 0;
g_pserver->db[j].last_expire_set = 0;
g_pserver->db[j].defrag_later = listCreate();
g_pserver->db[j].initialize(j);
}
/* Fixup Master Client Database */
@@ -2981,7 +3008,7 @@ void initServer(void) {
listSetFreeMethod(g_pserver->pubsub_patterns,freePubsubPattern);
listSetMatchMethod(g_pserver->pubsub_patterns,listMatchPubsubPattern);
g_pserver->cronloops = 0;
g_pserver->rdb_child_pid = -1;
g_pserver->rdbThreadVars.fRdbThreadActive = false;
g_pserver->aof_child_pid = -1;
g_pserver->rdb_child_type = RDB_CHILD_TYPE_NONE;
g_pserver->rdb_bgsave_scheduled = 0;
@@ -3707,7 +3734,11 @@ int processCommand(client *c, int callFlags) {
addReply(c,shared.queued);
} else {
std::unique_lock<decltype(c->db->lock)> ulock(c->db->lock);
for (int idb = 0; idb < cserver.dbnum; ++idb)
g_pserver->db[idb].trackChanges();
call(c,callFlags);
for (int idb = 0; idb < cserver.dbnum; ++idb)
g_pserver->db[idb].processChanges();
c->woff = g_pserver->master_repl_offset;
if (listLength(g_pserver->ready_keys))
handleClientsBlockedOnKeys();
@@ -3748,7 +3779,7 @@ int prepareForShutdown(int flags) {
/* Kill the saving child if there is a background saving in progress.
We want to avoid race conditions, for instance our saving child may
overwrite the synchronous saving did by SHUTDOWN. */
if (g_pserver->rdb_child_pid != -1) {
if (g_pserver->FRdbSaveInProgress()) {
serverLog(LL_WARNING,"There is a child saving an .rdb. Killing it!");
killRDBChild();
}
@@ -3779,7 +3810,7 @@ int prepareForShutdown(int flags) {
/* Snapshotting. Perform a SYNC SAVE and exit */
rdbSaveInfo rsi, *rsiptr;
rsiptr = rdbPopulateSaveInfo(&rsi);
if (rdbSave(rsiptr) != C_OK) {
if (rdbSave(nullptr, rsiptr) != C_OK) {
/* Ooops.. error saving! The best we can do is to continue
* operating. Note that if there was a background saving process,
* in the next cron() Redis will be notified that the background
@@ -4262,12 +4293,12 @@ sds genRedisInfoString(const char *section) {
"aof_last_cow_size:%zu\r\n",
g_pserver->loading,
g_pserver->dirty,
g_pserver->rdb_child_pid != -1,
g_pserver->FRdbSaveInProgress(),
(intmax_t)g_pserver->lastsave,
(g_pserver->lastbgsave_status == C_OK) ? "ok" : "err",
(intmax_t)g_pserver->rdb_save_time_last,
(intmax_t)((g_pserver->rdb_child_pid == -1) ?
-1 : time(NULL)-g_pserver->rdb_save_time_start),
(intmax_t)(g_pserver->FRdbSaveInProgress() ?
time(NULL)-g_pserver->rdb_save_time_start : -1),
g_pserver->stat_rdb_cow_bytes,
g_pserver->aof_state != AOF_OFF,
g_pserver->aof_child_pid != -1,
@@ -4583,8 +4614,8 @@ sds genRedisInfoString(const char *section) {
for (j = 0; j < cserver.dbnum; j++) {
long long keys, vkeys;
keys = dictSize(g_pserver->db[j].pdict);
vkeys = g_pserver->db[j].setexpire->size();
keys = g_pserver->db[j].size();
vkeys = g_pserver->db[j].expireSize();
// Adjust TTL by the current time
g_pserver->db[j].avg_ttl -= (g_pserver->mstime - g_pserver->db[j].last_expire_set);
@@ -4763,7 +4794,7 @@ static void sigShutdownHandler(int sig) {
* on disk. */
if (g_pserver->shutdown_asap && sig == SIGINT) {
serverLogFromHandler(LL_WARNING, "You insist... exiting now.");
rdbRemoveTempFile(getpid());
rdbRemoveTempFile(g_pserver->rdbThreadVars.tmpfileNum);
exit(1); /* Exit with an error since this was not a clean shutdown. */
} else if (g_pserver->loading) {
serverLogFromHandler(LL_WARNING, "Received shutdown signal during loading, exiting now.");
+226 -15
View File
@@ -42,6 +42,7 @@
#include <stdlib.h>
#include <cmath>
#include <string.h>
#include <string>
#include <time.h>
#include <limits.h>
#include <unistd.h>
@@ -54,6 +55,7 @@
#include <vector>
#include <algorithm>
#include <memory>
#include <set>
#ifdef __cplusplus
extern "C" {
#include <lua.h>
@@ -89,6 +91,7 @@ typedef long long mstime_t; /* millisecond time type. */
#include "sha1.h"
#include "endianconv.h"
#include "crc64.h"
#include "IStorage.h"
extern int g_fTestMode;
@@ -129,6 +132,11 @@ public:
return m_ptr;
}
const redisObject& operator*() const
{
return *m_ptr;
}
bool operator!() const
{
return !m_ptr;
@@ -783,6 +791,16 @@ public:
: when(when), spsubkey(subkey, sdsfree)
{}
subexpireEntry(const subexpireEntry &e)
: when(e.when), spsubkey(nullptr, sdsfree)
{
if (e.spsubkey)
spsubkey = std::unique_ptr<const char, void(*)(const char*)>((const char*)sdsdup((sds)e.spsubkey.get()), sdsfree);
}
subexpireEntry(subexpireEntry &&e) = default;
subexpireEntry& operator=(subexpireEntry &&e) = default;
bool operator<(long long when) const noexcept { return this->when < when; }
bool operator<(const subexpireEntry &se) { return this->when < se.when; }
};
@@ -795,6 +813,10 @@ public:
expireEntryFat(sds keyPrimary)
: m_keyPrimary(keyPrimary)
{}
expireEntryFat(const expireEntryFat &e) = default;
expireEntryFat(expireEntryFat &&e) = default;
long long when() const noexcept { return m_vecexpireEntries.front().when; }
const char *key() const noexcept { return m_keyPrimary; }
@@ -896,6 +918,14 @@ public:
m_when = LLONG_MIN;
}
expireEntry(const expireEntry &e)
{
u.m_key = e.u.m_key;
m_when = e.m_when;
if (e.FFat())
u.m_pfatentry = new (MALLOC_LOCAL) expireEntryFat(*e.u.m_pfatentry);
}
expireEntry(expireEntry &&e)
{
u.m_key = e.u.m_key;
@@ -1039,17 +1069,183 @@ typedef struct clientReplyBlock {
#endif
} clientReplyBlock;
struct dictEntry;
class dict_const_iter
{
friend struct redisDb;
friend class redisDbPersistentData;
protected:
dictEntry *de;
public:
explicit dict_const_iter(dictEntry *de)
: de(de)
{}
const char *key() const { return de ? (const char*)dictGetKey(de) : nullptr; }
robj_roptr val() const { return de ? (robj*)dictGetVal(de) : nullptr; }
const robj* operator->() const { return de ? (robj*)dictGetVal(de) : nullptr; }
operator robj_roptr() const { return de ? (robj*)dictGetVal(de) : nullptr; }
bool operator==(std::nullptr_t) const { return de == nullptr; }
bool operator!=(std::nullptr_t) const { return de != nullptr; }
bool operator==(const dict_const_iter &other) { return de == other.de; }
};
class dict_iter : public dict_const_iter
{
public:
explicit dict_iter(dictEntry *de)
: dict_const_iter(de)
{}
sds key() { return de ? (sds)dictGetKey(de) : nullptr; }
robj *val() { return de ? (robj*)dictGetVal(de) : nullptr; }
robj *operator->() { return de ? (robj*)dictGetVal(de) : nullptr; }
operator robj*() const { return de ? (robj*)dictGetVal(de) : nullptr; }
};
class redisDbPersistentData
{
friend void dictDbKeyDestructor(void *privdata, void *key);
public:
~redisDbPersistentData();
static void swap(redisDbPersistentData *db1, redisDbPersistentData *db2);
size_t slots() const { return dictSlots(m_pdict); }
size_t size() const { return dictSize(m_pdict); }
void expand(uint64_t slots) { dictExpand(m_pdict, slots); }
void trackkey(robj_roptr o)
{
trackkey(szFromObj(o));
}
void trackkey(const char *key)
{
if (m_fTrackingChanges && !m_fAllChanged)
m_setchanged.insert(std::string(key, sdslen(key)));
}
dict_iter find(const char *key)
{
dictEntry *de = dictFind(m_pdict, key);
ensure(de);
return dict_iter(de);
}
dict_iter find(robj_roptr key)
{
return find(szFromObj(key));
}
dict_iter random()
{
dictEntry *de = dictGetRandomKey(m_pdict);
ensure(de);
return dict_iter(de);
}
const expireEntry &random_expire()
{
return m_setexpire->random_value();
}
auto findExpire(const char *key)
{
return m_setexpire->find(key);
}
dict_iter end() { return dict_iter(nullptr); }
dict_const_iter end() const { return dict_const_iter(nullptr); }
void getStats(char *buf, size_t bufsize) { dictGetStats(buf, bufsize, m_pdict); }
void getExpireStats(char *buf, size_t bufsize) { m_setexpire->getstats(buf, bufsize); }
bool insert(char *k, robj *o);
void tryResize();
int incrementallyRehash();
void updateValue(dict_iter itr, robj *val);
bool syncDelete(robj *key);
bool asyncDelete(robj *key);
size_t expireSize() const { return m_setexpire->size(); }
int removeExpire(robj *key, dict_iter itr);
int removeSubkeyExpire(robj *key, robj *subkey);
void clear(void(callback)(void*));
void emptyDbAsync();
bool iterate(std::function<bool(const char*, robj*)> fn);
void setExpire(robj *key, robj *subkey, long long when);
void setExpire(expireEntry &&e);
expireEntry *getExpire(robj_roptr key);
void initialize();
void trackChanges() { m_fTrackingChanges++; }
void processChanges();
// This should only be used if you look at the key, we do not fixup
// objects stored elsewhere
dict *dictUnsafeKeyOnly() { return m_pdict; }
expireset *setexpireUnsafe() { return m_setexpire; }
const expireset *setexpire() { return m_setexpire; }
std::shared_ptr<redisDbPersistentData> createSnapshot();
void endSnapshot(const redisDbPersistentData *psnapshot);
private:
void ensure(const char *key);
void ensure(dictEntry *de);
void storeDatabase();
void storeKey(const char *key, size_t cchKey, robj *o);
// Keyspace
dict *m_pdict = nullptr; /* The keyspace for this DB */
int m_fTrackingChanges = 0; // Note: Stack based
bool m_fAllChanged = false;
std::set<std::string> m_setchanged;
IStorage *m_pstorage = nullptr;
// Expire
expireset *m_setexpire = nullptr;
std::shared_ptr<redisDbPersistentData> m_spdbSnapshot;
};
/* Redis database representation. There are multiple databases identified
* by integers from 0 (the default database) up to the max configured
* database. The database number is the 'id' field in the structure. */
typedef struct redisDb {
redisDb()
: expireitr(nullptr)
{};
dict *pdict; /* The keyspace for this DB */
expireset *setexpire;
expireset::setiter expireitr;
typedef struct redisDb : public redisDbPersistentData
{
// Legacy C API, Do not add more
friend void tryResizeHashTables(int);
friend int dbSyncDelete(redisDb *db, robj *key);
friend int dbAsyncDelete(redisDb *db, robj *key);
friend long long emptyDb(int dbnum, int flags, void(callback)(void*));
friend void scanGenericCommand(struct client *c, robj_roptr o, unsigned long cursor);
friend int dbSwapDatabases(int id1, int id2);
friend int removeExpire(redisDb *db, robj *key);
friend void setExpire(struct client *c, redisDb *db, robj *key, robj *subkey, long long when);
friend void setExpire(client *c, redisDb *db, robj *key, expireEntry &&e);
friend void evictionPoolPopulate(int dbid, redisDb *db, expireset *setexpire, struct evictionPoolEntry *pool);
friend void activeDefragCycle(void);
friend int freeMemoryIfNeeded(void);
friend void activeExpireCycle(int);
friend void expireSlaveKeys(void);
typedef ::dict_const_iter const_iter;
typedef ::dict_iter iter;
redisDb()
: expireitr(nullptr)
{}
void initialize(int id);
void dbOverwriteCore(redisDb::iter itr, robj *key, robj *val, bool fUpdateMvcc, bool fRemoveExpire);
bool FKeyExpires(const char *key);
size_t clear(bool fAsync, void(callback)(void*));
public:
expireset::setiter expireitr;
dict *blocking_keys; /* Keys with clients waiting for data (BLPOP)*/
dict *ready_keys; /* Blocked keys that received a PUSH */
dict *watched_keys; /* WATCHED keys for MULTI/EXEC CAS */
@@ -1392,6 +1588,9 @@ typedef struct rdbSaveInfo {
char repl_id[CONFIG_RUN_ID_SIZE+1]; /* Replication ID. */
long long repl_offset; /* Replication offset. */
int fForceSetKey;
/* Used In Save */
long long master_repl_offset;
} rdbSaveInfo;
#define RDB_SAVE_INFO_INIT {-1,0,"000000000000000000000000000000",-1, TRUE}
@@ -1649,7 +1848,13 @@ struct redisServer {
/* RDB persistence */
long long dirty; /* Changes to DB from the last save */
long long dirty_before_bgsave; /* Used to restore dirty on failed BGSAVE */
pid_t rdb_child_pid; /* PID of RDB saving child */
struct _rdbThreadVars
{
bool fRdbThreadActive = false;
volatile bool fRdbThreadCancel = false;
pthread_t rdb_child_thread; /* PID of RDB saving child */
int tmpfileNum = 0;
} rdbThreadVars;
struct saveparam *saveparams; /* Save points array for RDB */
int saveparamslen; /* Number of saving points */
char *rdb_filename; /* Name of RDB file */
@@ -1828,6 +2033,8 @@ struct redisServer {
/* System hardware info */
size_t system_memory_size; /* Total memory in system as reported by OS */
bool FRdbSaveInProgress() const { return rdbThreadVars.fRdbThreadActive; }
};
typedef struct pubsubPattern {
@@ -1933,9 +2140,11 @@ extern dictType zsetDictType;
extern dictType clusterNodesDictType;
extern dictType clusterNodesBlackListDictType;
extern dictType dbDictType;
extern dictType dbSnapshotDictType;
extern dictType shaScriptObjectDictType;
extern double R_Zero, R_PosInf, R_NegInf, R_Nan;
extern dictType hashDictType;
extern dictType keylistDictType;
extern dictType replScriptCacheDictType;
extern dictType keyptrDictType;
extern dictType modulesDictType;
@@ -2164,8 +2373,12 @@ const char *strEncoding(int encoding);
int compareStringObjects(robj *a, robj *b);
int collateStringObjects(robj *a, robj *b);
int equalStringObjects(robj *a, robj *b);
unsigned long long estimateObjectIdleTime(robj *o);
unsigned long long estimateObjectIdleTime(robj_roptr o);
void trimStringObjectIfNeeded(robj *o);
robj *deserializeStoredObject(const void *data, size_t cb);
sds serializeStoredObject(robj_roptr o);
#define sdsEncodedObject(objptr) (objptr->encoding == OBJ_ENCODING_RAW || objptr->encoding == OBJ_ENCODING_EMBSTR)
/* Synchronous I/O with timeout */
@@ -2218,7 +2431,7 @@ int writeCommandsDeniedByDiskError(void);
/* RDB persistence */
#include "rdb.h"
int rdbSaveRio(rio *rdb, int *error, int flags, rdbSaveInfo *rsi);
int rdbSaveRio(rio *rdb, redisDbPersistentData **rgpdb, int *error, int flags, rdbSaveInfo *rsi);
void killRDBChild(void);
/* AOF persistence */
@@ -2436,11 +2649,9 @@ int rewriteConfig(char *path);
/* db.c -- Keyspace access API */
int removeExpire(redisDb *db, robj *key);
int removeExpireCore(redisDb *db, robj *key, dictEntry *de);
int removeSubkeyExpire(redisDb *db, robj *key, robj *subkey);
void propagateExpire(redisDb *db, robj *key, int lazy);
int expireIfNeeded(redisDb *db, robj *key);
expireEntry *getExpire(redisDb *db, robj_roptr key);
void setExpire(client *c, redisDb *db, robj *key, robj *subkey, long long when);
void setExpire(client *c, redisDb *db, robj *key, expireEntry &&entry);
robj_roptr lookupKeyRead(redisDb *db, robj *key);
@@ -2448,8 +2659,8 @@ robj *lookupKeyWrite(redisDb *db, robj *key);
robj_roptr lookupKeyReadOrReply(client *c, robj *key, robj *reply);
robj *lookupKeyWriteOrReply(client *c, robj *key, robj *reply);
robj_roptr lookupKeyReadWithFlags(redisDb *db, robj *key, int flags);
robj *objectCommandLookup(client *c, robj *key);
robj *objectCommandLookupOrReply(client *c, robj *key, robj *reply);
robj_roptr objectCommandLookup(client *c, robj *key);
robj_roptr objectCommandLookupOrReply(client *c, robj *key, robj *reply);
void objectSetLRUOrLFU(robj *val, long long lfu_freq, long long lru_idle,
long long lru_clock);
#define LOOKUP_NONE 0
@@ -2482,7 +2693,6 @@ void slotToKeyAdd(robj *key);
void slotToKeyDel(robj *key);
void slotToKeyFlush(void);
int dbAsyncDelete(redisDb *db, robj *key);
void emptyDbAsync(redisDb *db);
void slotToKeyFlushAsync(void);
size_t lazyfreeGetPendingObjectsCount(void);
void freeObjAsync(robj *o);
@@ -2506,6 +2716,7 @@ void clusterPropagatePublish(robj *channel, robj *message);
void migrateCloseTimedoutSockets(void);
void clusterBeforeSleep(void);
int clusterSendModuleMessageToTarget(const char *target, uint64_t module_id, uint8_t type, unsigned char *payload, uint32_t len);
void createDumpPayload(rio *payload, robj_roptr o, robj *key);
/* Sentinel */
void initSentinelConfig(void);
+2 -2
View File
@@ -219,9 +219,9 @@ int setTypeRandomElement(robj *setobj, sds *sdsele, int64_t *llele) {
return setobj->encoding;
}
int setTypeRandomElement(robj_roptr setobj, const char **sdsele, const int64_t *llele)
int setTypeRandomElement(robj_roptr setobj, const char **sdsele, int64_t *llele)
{
return setTypeRandomElement(setobj.unsafe_robjcast(), (sds*)sdsele, const_cast<int64_t*>(llele));
return setTypeRandomElement(setobj.unsafe_robjcast(), (sds*)sdsele, llele);
}
unsigned long setTypeSize(robj_roptr subject) {
+1 -1
View File
@@ -1,3 +1,3 @@
#define KEYDB_REAL_VERSION "5.1.0"
#define KEYDB_REAL_VERSION "0.0.0"
extern const char *KEYDB_SET_VERSION; // Unlike real version, this can be overriden by the config