Unverified Commit 8cd62f82 authored by guybe7's avatar guybe7 Committed by GitHub
Browse files

Refactor the per-slot dict-array db.c into a new kvstore data structure (#12822)

# Description
Gather most of the scattered `redisDb`-related code from the per-slot
dict PR (#11695) and turn it to a new data structure, `kvstore`. i.e.
it's a class that represents an array of dictionaries.

# Motivation
The main motivation is code cleanliness, the idea of using an array of
dictionaries is very well-suited to becoming a self-contained data
structure.
This allowed cleaning some ugly code, among others: loops that run twice
on the main dict and expires dict, and duplicate code for allocating and
releasing this data structure.

# Notes
1. This PR reverts the part of https://github.com/redis/redis/pull/12848
where the `rehashing` list is global (handling rehashing `dict`s is
under the responsibility of `kvstore`, and should not be managed by the
server)
2. This PR also replaces the type of `server.pubsubshard_channels` from
`dict**` to `kvstore` (original PR:
https://github.com/redis/redis/pull/12804). After that was done,
server.pubsub_channels was also chosen to be a `kvstore` (with only one
`dict`, which seems odd) just to make the code cleaner by making it the
same type as `server.pubsubshard_channels`, see
`pubsubtype.serverPubSubChannels`
3. the keys and expires kvstores are currenlty configured to allocate
the individual dicts only when the first key is added (unlike before, in
which they allocated them in advance), but they won't release them when
the last key is deleted.

Worth mentioning that due to the recent change the reply of DEBUG
HTSTATS changed, in case no keys were ever added to the db.

before:
```
127.0.0.1:6379> DEBUG htstats 9
[Dictionary HT]
Hash table 0 stats (main hash table):
No stats available for empty dictionaries
[Expires HT]
Hash table 0 stats (main hash table):
No stats available for empty dictionaries
```

after:
```
127.0.0.1:6379> DEBUG htstats 9
[Dictionary HT]
[Expires HT]
```
parent f20774ec
......@@ -345,7 +345,7 @@ endif
REDIS_SERVER_NAME=redis-server$(PROG_SUFFIX)
REDIS_SENTINEL_NAME=redis-sentinel$(PROG_SUFFIX)
REDIS_SERVER_OBJ=threads_mngr.o adlist.o quicklist.o ae.o anet.o dict.o server.o sds.o zmalloc.o lzf_c.o lzf_d.o pqsort.o zipmap.o sha1.o ziplist.o release.o networking.o util.o object.o db.o replication.o rdb.o t_string.o t_list.o t_set.o t_zset.o t_hash.o config.o aof.o pubsub.o multi.o debug.o sort.o intset.o syncio.o cluster.o cluster_legacy.o crc16.o endianconv.o slowlog.o eval.o bio.o rio.o rand.o memtest.o syscheck.o crcspeed.o crc64.o bitops.o sentinel.o notify.o setproctitle.o blocked.o hyperloglog.o latency.o sparkline.o redis-check-rdb.o redis-check-aof.o geo.o lazyfree.o module.o evict.o expire.o geohash.o geohash_helper.o childinfo.o defrag.o siphash.o rax.o t_stream.o listpack.o localtime.o lolwut.o lolwut5.o lolwut6.o acl.o tracking.o socket.o tls.o sha256.o timeout.o setcpuaffinity.o monotonic.o mt19937-64.o resp_parser.o call_reply.o script_lua.o script.o functions.o function_lua.o commands.o strl.o connection.o unix.o logreqres.o
REDIS_SERVER_OBJ=threads_mngr.o adlist.o quicklist.o ae.o anet.o dict.o kvstore.o server.o sds.o zmalloc.o lzf_c.o lzf_d.o pqsort.o zipmap.o sha1.o ziplist.o release.o networking.o util.o object.o db.o replication.o rdb.o t_string.o t_list.o t_set.o t_zset.o t_hash.o config.o aof.o pubsub.o multi.o debug.o sort.o intset.o syncio.o cluster.o cluster_legacy.o crc16.o endianconv.o slowlog.o eval.o bio.o rio.o rand.o memtest.o syscheck.o crcspeed.o crc64.o bitops.o sentinel.o notify.o setproctitle.o blocked.o hyperloglog.o latency.o sparkline.o redis-check-rdb.o redis-check-aof.o geo.o lazyfree.o module.o evict.o expire.o geohash.o geohash_helper.o childinfo.o defrag.o siphash.o rax.o t_stream.o listpack.o localtime.o lolwut.o lolwut5.o lolwut6.o acl.o tracking.o socket.o tls.o sha256.o timeout.o setcpuaffinity.o monotonic.o mt19937-64.o resp_parser.o call_reply.o script_lua.o script.o functions.o function_lua.o commands.o strl.o connection.o unix.o logreqres.o
REDIS_CLI_NAME=redis-cli$(PROG_SUFFIX)
REDIS_CLI_OBJ=anet.o adlist.o dict.o redis-cli.o zmalloc.o release.o ae.o redisassert.o crcspeed.o crc64.o siphash.o crc16.o monotonic.o cli_common.o mt19937-64.o strl.o cli_commands.o
REDIS_BENCHMARK_NAME=redis-benchmark$(PROG_SUFFIX)
......
......@@ -1903,12 +1903,6 @@ int ACLCheckAllPerm(client *c, int *idxptr) {
return ACLCheckAllUserCommandPerm(c->user, c->cmd, c->argv, c->argc, idxptr);
}
int totalSubscriptions(void) {
return dictSize(server.pubsub_patterns) +
dictSize(server.pubsub_channels) +
server.shard_channel_count;
}
/* If 'new' can access all channels 'original' could then return NULL;
Otherwise return a list of channels that the new user can access */
list *getUpcomingChannelList(user *new, user *original) {
......@@ -2017,7 +2011,7 @@ int ACLShouldKillPubsubClient(client *c, list *upcoming) {
* permissions specified via the upcoming argument, and kill them if so. */
void ACLKillPubsubClientsIfNeeded(user *new, user *original) {
/* Do nothing if there are no subscribers. */
if (totalSubscriptions() == 0)
if (pubsubTotalSubscriptions() == 0)
return;
list *channels = getUpcomingChannelList(new, original);
......@@ -2450,7 +2444,7 @@ sds ACLLoadFromFile(const char *filename) {
/* If there are some subscribers, we need to check if we need to drop some clients. */
rax *user_channels = NULL;
if (totalSubscriptions() > 0) {
if (pubsubTotalSubscriptions() > 0) {
user_channels = raxNew();
}
......
......@@ -2244,7 +2244,7 @@ int rewriteAppendOnlyFileRio(rio *aof) {
int j;
long key_count = 0;
long long updated_time = 0;
dbIterator *dbit = NULL;
kvstoreIterator *kvs_it = NULL;
/* Record timestamp at the beginning of rewriting AOF. */
if (server.aof_timestamp_enabled) {
......@@ -2258,15 +2258,15 @@ int rewriteAppendOnlyFileRio(rio *aof) {
for (j = 0; j < server.dbnum; j++) {
char selectcmd[] = "*2\r\n$6\r\nSELECT\r\n";
redisDb *db = server.db + j;
if (dbSize(db, DB_MAIN) == 0) continue;
if (kvstoreSize(db->keys) == 0) continue;
/* SELECT the new DB */
if (rioWrite(aof,selectcmd,sizeof(selectcmd)-1) == 0) goto werr;
if (rioWriteBulkLongLong(aof,j) == 0) goto werr;
dbit = dbIteratorInit(db, DB_MAIN);
kvs_it = kvstoreIteratorInit(db->keys);
/* Iterate this DB writing every entry */
while((de = dbIteratorNext(dbit)) != NULL) {
while((de = kvstoreIteratorNext(kvs_it)) != NULL) {
sds keystr;
robj key, *o;
long long expiretime;
......@@ -2331,12 +2331,12 @@ int rewriteAppendOnlyFileRio(rio *aof) {
if (server.rdb_key_save_delay)
debugDelay(server.rdb_key_save_delay);
}
dbReleaseIterator(dbit);
kvstoreIteratorRelease(kvs_it);
}
return C_OK;
werr:
if (dbit) dbReleaseIterator(dbit);
if (kvs_it) kvstoreIteratorRelease(kvs_it);
return C_ERR;
}
......
......@@ -817,7 +817,7 @@ static int shouldReturnTlsInfo(void) {
}
unsigned int countKeysInSlot(unsigned int slot) {
return dictSize(server.db->dict[slot]);
return kvstoreDictSize(server.db->keys, slot);
}
void clusterCommandHelp(client *c) {
......@@ -919,7 +919,7 @@ void clusterCommand(client *c) {
addReplyArrayLen(c,numkeys);
dictIterator *iter = NULL;
dictEntry *de = NULL;
iter = dictGetIterator(server.db->dict[slot]);
iter = kvstoreDictGetIterator(server.db->keys, slot);
for (unsigned int i = 0; i < numkeys; i++) {
de = dictNext(iter);
serverAssert(de != NULL);
......
......@@ -5104,7 +5104,7 @@ int verifyClusterConfigWithData(void) {
/* Make sure we only have keys in DB0. */
for (j = 1; j < server.dbnum; j++) {
if (dbSize(&server.db[j], DB_MAIN)) return C_ERR;
if (kvstoreSize(server.db[j].keys)) return C_ERR;
}
/* Check that all the slots we see populated memory have a corresponding
......@@ -5140,7 +5140,7 @@ int verifyClusterConfigWithData(void) {
/* Remove all the shard channel related information not owned by the current shard. */
static inline void removeAllNotOwnedShardChannelSubscriptions(void) {
if (!server.shard_channel_count) return;
if (!kvstoreSize(server.pubsubshard_channels)) return;
clusterNode *currmaster = clusterNodeIsMaster(myself) ? myself : myself->slaveof;
for (int j = 0; j < CLUSTER_SLOTS; j++) {
if (server.cluster->slots[j] != currmaster) {
......@@ -5734,16 +5734,17 @@ void removeChannelsInSlot(unsigned int slot) {
pubsubShardUnsubscribeAllChannelsInSlot(slot);
}
/* Remove all the keys in the specified hash slot.
* The number of removed items is returned. */
unsigned int delKeysInSlot(unsigned int hashslot) {
if (!kvstoreDictSize(server.db->keys, hashslot))
return 0;
unsigned int j = 0;
dictIterator *iter = NULL;
dictEntry *de = NULL;
iter = dictGetSafeIterator(server.db->dict[hashslot]);
iter = kvstoreDictGetSafeIterator(server.db->keys, hashslot);
while((de = dictNext(iter)) != NULL) {
enterExecutionUnit(1, 0);
sds sdskey = dictGetKey(de);
......@@ -5768,8 +5769,7 @@ unsigned int delKeysInSlot(unsigned int hashslot) {
/* Get the count of the channels for a given slot. */
unsigned int countChannelsInSlot(unsigned int hashslot) {
dict *d = server.pubsubshard_channels[hashslot];
return d ? dictSize(d) : 0;
return kvstoreDictSize(server.pubsubshard_channels, hashslot);
}
int clusterNodeIsMyself(clusterNode *n) {
......@@ -5939,7 +5939,7 @@ int clusterCommandSpecial(client *c) {
}
} else if (!strcasecmp(c->argv[1]->ptr,"flushslots") && c->argc == 2) {
/* CLUSTER FLUSHSLOTS */
if (dbSize(&server.db[0], DB_MAIN) != 0) {
if (kvstoreSize(server.db[0].keys) != 0) {
addReplyError(c,"DB must be empty to perform CLUSTER FLUSHSLOTS.");
return 1;
}
......@@ -6205,7 +6205,7 @@ int clusterCommandSpecial(client *c) {
* slots nor keys to accept to replicate some other node.
* Slaves can switch to another master without issues. */
if (clusterNodeIsMaster(myself) &&
(myself->numslots != 0 || dbSize(&server.db[0], DB_MAIN) != 0)) {
(myself->numslots != 0 || kvstoreSize(server.db[0].keys) != 0)) {
addReplyError(c,
"To set a master the node must be empty and "
"without assigned slots.");
......@@ -6339,7 +6339,7 @@ int clusterCommandSpecial(client *c) {
/* Slaves can be reset while containing data, but not master nodes
* that must be empty. */
if (clusterNodeIsMaster(myself) && dbSize(c->db, DB_MAIN) != 0) {
if (clusterNodeIsMaster(myself) && kvstoreSize(c->db->keys) != 0) {
addReplyError(c,"CLUSTER RESET can't be called with "
"master nodes containing keys");
return 1;
......
This diff is collapsed.
......@@ -76,7 +76,6 @@ int bugReportStart(void);
void printCrashReport(void);
void bugReportEnd(int killViaSignal, int sig);
void logStackTrace(void *eip, int uplevel, int current_thread);
void dbGetStats(char *buf, size_t bufsize, redisDb *db, int full, dbKeyType keyType);
void sigalrmSignalHandler(int sig, siginfo_t *info, void *secret);
/* ================================= Debugging ============================== */
......@@ -290,15 +289,16 @@ void computeDatasetDigest(unsigned char *final) {
for (j = 0; j < server.dbnum; j++) {
redisDb *db = server.db+j;
if (dbSize(db, DB_MAIN) == 0) continue;
dbIterator *dbit = dbIteratorInit(db, DB_MAIN);
if (kvstoreSize(db->keys) == 0)
continue;
kvstoreIterator *kvs_it = kvstoreIteratorInit(db->keys);
/* hash the DB id, so the same dataset moved in a different DB will lead to a different digest */
aux = htonl(j);
mixDigest(final,&aux,sizeof(aux));
/* Iterate this DB writing every entry */
while((de = dbIteratorNext(dbit)) != NULL) {
while((de = kvstoreIteratorNext(kvs_it)) != NULL) {
sds key;
robj *keyobj, *o;
......@@ -315,7 +315,7 @@ void computeDatasetDigest(unsigned char *final) {
xorDigest(final,digest,20);
decrRefCount(keyobj);
}
dbReleaseIterator(dbit);
kvstoreIteratorRelease(kvs_it);
}
}
......@@ -606,7 +606,7 @@ NULL
robj *val;
char *strenc;
if ((de = dbFind(c->db, c->argv[2]->ptr, DB_MAIN)) == NULL) {
if ((de = dbFind(c->db, c->argv[2]->ptr)) == NULL) {
addReplyErrorObject(c,shared.nokeyerr);
return;
}
......@@ -658,7 +658,7 @@ NULL
robj *val;
sds key;
if ((de = dbFind(c->db, c->argv[2]->ptr, DB_MAIN)) == NULL) {
if ((de = dbFind(c->db, c->argv[2]->ptr)) == NULL) {
addReplyErrorObject(c,shared.nokeyerr);
return;
}
......@@ -719,7 +719,7 @@ NULL
return;
}
if (dbExpand(c->db, keys, DB_MAIN, 1) == C_ERR) {
if (dbExpand(c->db, keys, 1) == C_ERR) {
addReplyError(c, "OOM in dictTryExpand");
return;
}
......@@ -767,7 +767,7 @@ NULL
/* We don't use lookupKey because a debug command should
* work on logically expired keys */
dictEntry *de;
robj *o = ((de = dbFind(c->db, c->argv[j]->ptr, DB_MAIN)) == NULL) ? NULL : dictGetVal(de);
robj *o = ((de = dbFind(c->db, c->argv[j]->ptr)) == NULL) ? NULL : dictGetVal(de);
if (o) xorObjectDigest(c->db,c->argv[j],digest,o);
sds d = sdsempty();
......@@ -911,11 +911,11 @@ NULL
full = 1;
stats = sdscatprintf(stats,"[Dictionary HT]\n");
dbGetStats(buf, sizeof(buf), &server.db[dbid], full, DB_MAIN);
kvstoreGetStats(server.db[dbid].keys, buf, sizeof(buf), full);
stats = sdscat(stats,buf);
stats = sdscatprintf(stats,"[Expires HT]\n");
dbGetStats(buf, sizeof(buf), &server.db[dbid], full, DB_EXPIRES);
kvstoreGetStats(server.db[dbid].expires, buf, sizeof(buf), full);
stats = sdscat(stats,buf);
addReplyVerbatim(c,stats,sdslen(stats),"txt");
......@@ -2051,7 +2051,7 @@ void logCurrentClient(client *cc, const char *title) {
dictEntry *de;
key = getDecodedObject(cc->argv[1]);
de = dbFind(cc->db, key->ptr, DB_MAIN);
de = dbFind(cc->db, key->ptr);
if (de) {
val = dictGetVal(de);
serverLog(LL_WARNING,"key '%s' found in DB containing the following object:", (char*)key->ptr);
......
......@@ -684,21 +684,21 @@ void defragKey(defragCtx *ctx, dictEntry *de) {
/* Try to defrag the key name. */
newsds = activeDefragSds(keysds);
if (newsds) {
dictSetKey(db->dict[slot], de, newsds);
if (dbSize(db, DB_EXPIRES)) {
kvstoreDictSetKey(db->keys, slot, de, newsds);
if (kvstoreSize(db->expires)) {
/* We can't search in db->expires for that key after we've released
* the pointer it holds, since it won't be able to do the string
* compare, but we can find the entry using key hash and pointer. */
uint64_t hash = dictGetHash(db->dict[slot], newsds);
dictEntry *expire_de = dictFindEntryByPtrAndHash(db->expires[slot], keysds, hash);
if (expire_de) dictSetKey(db->expires[slot], expire_de, newsds);
uint64_t hash = kvstoreGetHash(db->keys, newsds);
dictEntry *expire_de = kvstoreDictFindEntryByPtrAndHash(db->expires, slot, keysds, hash);
if (expire_de) kvstoreDictSetKey(db->expires, slot, expire_de, newsds);
}
}
/* Try to defrag robj and / or string value. */
ob = dictGetVal(de);
if ((newob = activeDefragStringOb(ob))) {
dictSetVal(db->dict[slot], de, newob);
kvstoreDictSetVal(db->keys, slot, de, newob);
ob = newob;
}
......@@ -856,7 +856,7 @@ int defragLaterStep(redisDb *db, int slot, 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->dict[slot], defrag_later_current_key);
dictEntry *de = kvstoreDictFind(db->keys, slot, defrag_later_current_key);
key_defragged = server.stat_active_defrag_hits;
do {
int quit = 0;
......@@ -1022,13 +1022,12 @@ void activeDefragCycle(void) {
db = &server.db[current_db];
cursor = 0;
expires_cursor = 0;
slot = findSlotByKeyIndex(db, 1, DB_MAIN);
slot = kvstoreFindDictIndexByKeyIndex(db->keys, 1);
defrag_later_item_in_progress = 0;
ctx.db = db;
ctx.slot = slot;
}
do {
dict *d = db->dict[slot];
/* before scanning the next bucket, see if we have big keys left from the previous bucket to scan */
if (defragLaterStep(db, slot, endtime)) {
quit = 1; /* time is up, we didn't finish all the work */
......@@ -1038,13 +1037,14 @@ void activeDefragCycle(void) {
if (!defrag_later_item_in_progress) {
/* Scan the keyspace dict unless we're scanning the expire dict. */
if (!expires_cursor)
cursor = dictScanDefrag(d, cursor, defragScanCallback,
&defragfns, &ctx);
cursor = kvstoreDictScanDefrag(db->keys, slot, cursor,
defragScanCallback,
&defragfns, &ctx);
/* When done scanning the keyspace dict, we scan the expire dict. */
if (!cursor)
expires_cursor = dictScanDefrag(db->expires[slot], expires_cursor,
scanCallbackCountScanned,
&defragfns, NULL);
expires_cursor = kvstoreDictScanDefrag(db->expires, slot, expires_cursor,
scanCallbackCountScanned,
&defragfns, NULL);
}
if (!(cursor || expires_cursor)) {
/* Move to the next slot only if regular and large item scanning has been completed. */
......@@ -1052,7 +1052,7 @@ void activeDefragCycle(void) {
defrag_later_item_in_progress = 1;
continue;
}
slot = dbGetNextNonEmptySlot(db, slot, DB_MAIN);
slot = kvstoreGetNextNonEmptyDictIndex(db->keys, slot);
defrag_later_item_in_progress = 0;
ctx.slot = slot;
}
......
......@@ -194,16 +194,6 @@ dict *dictCreate(dictType *type)
return d;
}
/* Create an array of dictionaries */
dict **dictCreateMultiple(dictType *type, int count)
{
dict **d = zmalloc(sizeof(dict*) * count);
for (int i = 0; i < count; i++) {
d[i] = dictCreate(type);
}
return d;
}
/* Initialize the hash table */
int _dictInit(dict *d, dictType *type)
{
......
......@@ -51,6 +51,7 @@ typedef struct dictEntry dictEntry; /* opaque */
typedef struct dict dict;
typedef struct dictType {
/* Callbacks */
uint64_t (*hashFunction)(const void *key);
void *(*keyDup)(dict *d, const void *key);
void *(*valDup)(dict *d, const void *obj);
......@@ -66,6 +67,10 @@ typedef struct dictType {
/* Allow a dict to carry extra caller-defined metadata. The
* extra memory is initialized to 0 when a dict is allocated. */
size_t (*dictMetadataBytes)(dict *d);
/* Data */
void *userdata;
/* Flags */
/* The 'no_value' flag, if set, indicates that values are not used, i.e. the
* dict is a set. When this flag is set, it's not possible to access the
......@@ -177,7 +182,6 @@ typedef enum {
/* API */
dict *dictCreate(dictType *type);
dict **dictCreateMultiple(dictType *type, int count);
int dictExpand(dict *d, unsigned long size);
int dictTryExpand(dict *d, unsigned long size);
int dictShrink(dict *d, unsigned long size);
......
......@@ -143,11 +143,12 @@ void evictionPoolAlloc(void) {
* We insert keys on place in ascending order, so keys with the smaller
* idle time are on the left, and keys with the higher idle time on the
* right. */
int evictionPoolPopulate(int dbid, int slot, dict *sampledict, dict *keydict, struct evictionPoolEntry *pool) {
int evictionPoolPopulate(redisDb *db, kvstore *samplekvs, struct evictionPoolEntry *pool) {
int j, k, count;
dictEntry *samples[server.maxmemory_samples];
count = dictGetSomeKeys(sampledict,samples,server.maxmemory_samples);
int slot = kvstoreGetFairRandomDictIndex(samplekvs);
count = kvstoreDictGetSomeKeys(samplekvs,slot,samples,server.maxmemory_samples);
for (j = 0; j < count; j++) {
unsigned long long idle;
sds key;
......@@ -161,7 +162,8 @@ int evictionPoolPopulate(int dbid, int slot, dict *sampledict, dict *keydict, st
* dictionary (but the expires one) we need to lookup the key
* again in the key dictionary to obtain the value object. */
if (server.maxmemory_policy != MAXMEMORY_VOLATILE_TTL) {
if (sampledict != keydict) de = dictFind(keydict, key);
if (samplekvs != db->keys)
de = kvstoreDictFind(db->keys, slot, key);
o = dictGetVal(de);
}
......@@ -236,7 +238,7 @@ int evictionPoolPopulate(int dbid, int slot, dict *sampledict, dict *keydict, st
pool[k].key = pool[k].cached;
}
pool[k].idle = idle;
pool[k].dbid = dbid;
pool[k].dbid = db->id;
pool[k].slot = slot;
}
......@@ -578,16 +580,12 @@ int performEvictions(void) {
sds bestkey = NULL;
int bestdbid;
redisDb *db;
dict *dict;
dictEntry *de;
if (server.maxmemory_policy & (MAXMEMORY_FLAG_LRU|MAXMEMORY_FLAG_LFU) ||
server.maxmemory_policy == MAXMEMORY_VOLATILE_TTL)
{
struct evictionPoolEntry *pool = EvictionPoolLRU;
dbKeyType keyType = (server.maxmemory_policy & MAXMEMORY_FLAG_ALLKEYS ?
DB_MAIN : DB_EXPIRES);
while (bestkey == NULL) {
unsigned long total_keys = 0;
......@@ -596,17 +594,21 @@ int performEvictions(void) {
* every DB. */
for (i = 0; i < server.dbnum; i++) {
db = server.db+i;
kvstore *kvs;
if (server.maxmemory_policy & MAXMEMORY_FLAG_ALLKEYS) {
kvs = db->keys;
} else {
kvs = db->expires;
}
unsigned long sampled_keys = 0;
unsigned long current_db_keys = dbSize(db, keyType);
unsigned long current_db_keys = kvstoreSize(kvs);
if (current_db_keys == 0) continue;
total_keys += current_db_keys;
int l = dbNonEmptySlots(db, keyType);
int l = kvstoreNumNonEmptyDicts(kvs);
/* Do not exceed the number of non-empty slots when looping. */
while (l--) {
int slot = getFairRandomSlot(db, keyType);
dict = (keyType == DB_MAIN ? db->dict[slot] : db->expires[slot]);
sampled_keys += evictionPoolPopulate(i, slot, dict, db->dict[slot], pool);
sampled_keys += evictionPoolPopulate(db, kvs, pool);
/* We have sampled enough keys in the current db, exit the loop. */
if (sampled_keys >= (unsigned long) server.maxmemory_samples)
break;
......@@ -624,13 +626,13 @@ int performEvictions(void) {
if (pool[k].key == NULL) continue;
bestdbid = pool[k].dbid;
kvstore *kvs;
if (server.maxmemory_policy & MAXMEMORY_FLAG_ALLKEYS) {
de = dictFind(server.db[bestdbid].dict[pool[k].slot],
pool[k].key);
kvs = server.db[bestdbid].keys;
} else {
de = dictFind(server.db[bestdbid].expires[pool[k].slot],
pool[k].key);
kvs = server.db[bestdbid].expires;
}
de = kvstoreDictFind(kvs, pool[k].slot, pool[k].key);
/* Remove the entry from the pool. */
if (pool[k].key != pool[k].cached)
......@@ -660,10 +662,15 @@ int performEvictions(void) {
for (i = 0; i < server.dbnum; i++) {
j = (++next_db) % server.dbnum;
db = server.db+j;
dict = (server.maxmemory_policy == MAXMEMORY_ALLKEYS_RANDOM) ?
db->dict[getFairRandomSlot(db, DB_MAIN)] : db->expires[getFairRandomSlot(db, DB_EXPIRES)];
if (dictSize(dict) != 0) {
de = dictGetRandomKey(dict);
kvstore *kvs;
if (server.maxmemory_policy == MAXMEMORY_ALLKEYS_RANDOM) {
kvs = db->keys;
} else {
kvs = db->expires;
}
int slot = kvstoreGetFairRandomDictIndex(kvs);
de = kvstoreDictGetRandomKey(kvs, slot);
if (de) {
bestkey = dictGetKey(de);
bestdbid = j;
break;
......
......@@ -253,7 +253,8 @@ void activeExpireCycle(int type) {
* distribute the time evenly across DBs. */
current_db++;
if (dbSize(db, DB_EXPIRES)) dbs_performed++;
if (kvstoreSize(db->expires))
dbs_performed++;
/* Continue to expire if at the end of the cycle there are still
* a big percentage of keys to expire, compared to the number of keys
......@@ -264,7 +265,7 @@ void activeExpireCycle(int type) {
iteration++;
/* If there is nothing to expire try next DB ASAP. */
if ((num = dbSize(db, DB_EXPIRES)) == 0) {
if ((num = kvstoreSize(db->expires)) == 0) {
db->avg_ttl = 0;
break;
}
......@@ -294,7 +295,7 @@ void activeExpireCycle(int type) {
int origin_ttl_samples = data.ttl_samples;
while (data.sampled < num && checked_buckets < max_buckets) {
db->expires_cursor = dbScan(db, DB_EXPIRES, db->expires_cursor, -1, expireScanCallback, isExpiryDictValidForSamplingCb, &data);
db->expires_cursor = kvstoreScan(db->expires, db->expires_cursor, -1, expireScanCallback, isExpiryDictValidForSamplingCb, &data);
if (db->expires_cursor == 0) {
db_done = 1;
break;
......@@ -429,7 +430,7 @@ void expireSlaveKeys(void) {
while(dbids && dbid < server.dbnum) {
if ((dbids & 1) != 0) {
redisDb *db = server.db+dbid;
dictEntry *expire = dictFind(db->expires[getKeySlot(keyname)],keyname);
dictEntry *expire = dbFindExpires(db, keyname);
int expired = 0;
if (expire &&
......
This diff is collapsed.
#ifndef DICTARRAY_H_
#define DICTARRAY_H_
#include "dict.h"
#include "adlist.h"
typedef struct _kvstore kvstore;
typedef struct _kvstoreIterator kvstoreIterator;
typedef int (kvstoreScanShouldSkipDict)(dict *d);
typedef int (kvstoreExpandShouldSkipDictIndex)(int didx);
#define KVSTORE_ALLOCATE_DICTS_ON_DEMAND (1<<0)
#define KVSTORE_FREE_EMPTY_DICTS (1<<1)
kvstore *kvstoreCreate(dictType *type, int num_dicts_bits, int flags);
void kvstoreEmpty(kvstore *kvs, void(callback)(dict*));
void kvstoreRelease(kvstore *kvs);
unsigned long long kvstoreSize(kvstore *kvs);
unsigned long kvstoreBuckets(kvstore *kvs);
size_t kvstoreMemUsage(kvstore *kvs);
unsigned long long kvstoreScan(kvstore *kvs, unsigned long long cursor,
int onlydidx, dictScanFunction *scan_cb,
kvstoreScanShouldSkipDict *skip_cb,
void *privdata);
int kvstoreExpand(kvstore *kvs, uint64_t newsize, int try_expand, kvstoreExpandShouldSkipDictIndex *skip_cb);
int kvstoreGetFairRandomDictIndex(kvstore *kvs);
void kvstoreGetStats(kvstore *kvs, char *buf, size_t bufsize, int full);
int kvstoreFindDictIndexByKeyIndex(kvstore *kvs, unsigned long target);
int kvstoreGetNextNonEmptyDictIndex(kvstore *kvs, int didx);
int kvstoreNumNonEmptyDicts(kvstore *kvs);
int kvstoreNumDicts(kvstore *kvs);
uint64_t kvstoreGetHash(kvstore *kvs, const void *key);
/* kvstore iterator specific functions */
kvstoreIterator *kvstoreIteratorInit(kvstore *kvs);
void kvstoreIteratorRelease(kvstoreIterator *kvs_it);
dict *kvstoreIteratorNextDict(kvstoreIterator *kvs_it);
int kvstoreIteratorGetCurrentDictIndex(kvstoreIterator *kvs_it);
dictEntry *kvstoreIteratorNext(kvstoreIterator *kvs_it);
/* Rehashing */
void kvstoreTryResizeDicts(kvstore *kvs, int limit);
uint64_t kvstoreIncrementallyRehash(kvstore *kvs, uint64_t threshold_ms);
/* Specific dict access by dict-index */
unsigned long kvstoreDictSize(kvstore *kvs, int didx);
dictIterator *kvstoreDictGetIterator(kvstore *kvs, int didx);
dictIterator *kvstoreDictGetSafeIterator(kvstore *kvs, int didx);
dictEntry *kvstoreDictGetRandomKey(kvstore *kvs, int didx);
dictEntry *kvstoreDictGetFairRandomKey(kvstore *kvs, int didx);
dictEntry *kvstoreDictFindEntryByPtrAndHash(kvstore *kvs, int didx, const void *oldptr, uint64_t hash);
unsigned int kvstoreDictGetSomeKeys(kvstore *kvs, int didx, dictEntry **des, unsigned int count);
int kvstoreDictExpand(kvstore *kvs, int didx, unsigned long size);
unsigned long kvstoreDictScanDefrag(kvstore *kvs, int didx, unsigned long v, dictScanFunction *fn, dictDefragFunctions *defragfns, void *privdata);
void *kvstoreDictFetchValue(kvstore *kvs, int didx, const void *key);
dictEntry *kvstoreDictFind(kvstore *kvs, int didx, void *key);
dictEntry *kvstoreDictAddRaw(kvstore *kvs, int didx, void *key, dictEntry **existing);
void kvstoreDictSetKey(kvstore *kvs, int didx, dictEntry* de, void *key);
void kvstoreDictSetVal(kvstore *kvs, int didx, dictEntry *de, void *val);
dictEntry *kvstoreDictTwoPhaseUnlinkFind(kvstore *kvs, int didx, const void *key, dictEntry ***plink, int *table_index);
void kvstoreDictTwoPhaseUnlinkFree(kvstore *kvs, int didx, dictEntry *he, dictEntry **plink, int table_index);
int kvstoreDictDelete(kvstore *kvs, int didx, const void *key);
#endif /* DICTARRAY_H_ */
......@@ -2,6 +2,7 @@
#include "bio.h"
#include "atomicvar.h"
#include "functions.h"
#include "cluster.h"
static redisAtomic size_t lazyfree_objects = 0;
static redisAtomic size_t lazyfreed_objects = 0;
......@@ -19,19 +20,14 @@ void lazyfreeFreeObject(void *args[]) {
* database which was substituted with a fresh one in the main thread
* when the database was logically deleted. */
void lazyfreeFreeDatabase(void *args[]) {
dict **ht1 = (dict **) args[0];
dict **ht2 = (dict **) args[1];
int *dictCount = (int *) args[2];
for (int i=0; i<*dictCount; i++) {
size_t numkeys = dictSize(ht1[i]);
dictRelease(ht1[i]);
dictRelease(ht2[i]);
atomicDecr(lazyfree_objects,numkeys);
atomicIncr(lazyfreed_objects,numkeys);
}
zfree(ht1);
zfree(ht2);
zfree(dictCount);
kvstore *da1 = args[0];
kvstore *da2 = args[1];
size_t numkeys = kvstoreSize(da1);
kvstoreRelease(da1);
kvstoreRelease(da2);
atomicDecr(lazyfree_objects,numkeys);
atomicIncr(lazyfreed_objects,numkeys);
}
/* Release the key tracking table. */
......@@ -179,28 +175,12 @@ void freeObjAsync(robj *key, robj *obj, int dbid) {
* create a new empty set of hash tables and scheduling the old ones for
* lazy freeing. */
void emptyDbAsync(redisDb *db) {
dbDictMetadata *metadata;
for (int i = 0; i < db->dict_count; i++) {
metadata = (dbDictMetadata *)dictMetadata(db->dict[i]);
if (metadata->rehashing_node) {
listDelNode(server.rehashing, metadata->rehashing_node);
metadata->rehashing_node = NULL;
}
metadata = (dbDictMetadata *)dictMetadata(db->expires[i]);
if (metadata->rehashing_node) {
listDelNode(server.rehashing, metadata->rehashing_node);
metadata->rehashing_node = NULL;
}
}
dict **oldDict = db->dict;
dict **oldExpires = db->expires;
atomicIncr(lazyfree_objects,dbSize(db, DB_MAIN));
db->dict = dictCreateMultiple(&dbDictType, db->dict_count);
db->expires = dictCreateMultiple(&dbExpiresDictType, db->dict_count);
int *count = zmalloc(sizeof(int));
*count = db->dict_count;
bioCreateLazyFreeJob(lazyfreeFreeDatabase, 3, oldDict, oldExpires, count);
int slotCountBits = server.cluster_enabled? CLUSTER_SLOT_MASK_BITS : 0;
kvstore *oldkeys = db->keys, *oldexpires = db->expires;
db->keys = kvstoreCreate(&dbDictType, slotCountBits, KVSTORE_ALLOCATE_DICTS_ON_DEMAND);
db->expires = kvstoreCreate(&dbExpiresDictType, slotCountBits, KVSTORE_ALLOCATE_DICTS_ON_DEMAND);
atomicIncr(lazyfree_objects, kvstoreSize(oldkeys));
bioCreateLazyFreeJob(lazyfreeFreeDatabase, 2, oldkeys, oldexpires);
}
/* Free the key tracking table.
......
......@@ -4295,7 +4295,7 @@ void RM_ResetDataset(int restart_aof, int async) {
 
/* Returns the number of keys in the current db. */
unsigned long long RM_DbSize(RedisModuleCtx *ctx) {
return dbSize(ctx->client->db, DB_MAIN);
return dbSize(ctx->client->db);
}
 
/* Returns a name of a random key, or NULL if current db is empty. */
......@@ -11058,7 +11058,7 @@ int RM_Scan(RedisModuleCtx *ctx, RedisModuleScanCursor *cursor, RedisModuleScanC
}
int ret = 1;
ScanCBData data = { ctx, privdata, fn };
cursor->cursor = dbScan(ctx->client->db, DB_MAIN, cursor->cursor, -1, moduleScanCallback, NULL, &data);
cursor->cursor = dbScan(ctx->client->db, cursor->cursor, moduleScanCallback, &data);
if (cursor->cursor == 0) {
cursor->done = 1;
ret = 0;
......
......@@ -394,7 +394,7 @@ void touchWatchedKey(redisDb *db, robj *key) {
/* The key was already expired when WATCH was called. */
if (db == wk->db &&
equalStringObjects(key, wk->key) &&
dictFind(db->dict[calculateKeySlot(key->ptr)], key->ptr) == NULL)
dbFind(db, key->ptr) == NULL)
{
/* Already expired key is deleted, so logically no change. Clear
* the flag. Deleted keys are not flagged as expired. */
......@@ -432,9 +432,9 @@ void touchAllWatchedKeysInDb(redisDb *emptied, redisDb *replaced_with) {
dictIterator *di = dictGetSafeIterator(emptied->watched_keys);
while((de = dictNext(di)) != NULL) {
robj *key = dictGetKey(de);
int exists_in_emptied = dictFind(emptied->dict[calculateKeySlot(key->ptr)], key->ptr) != NULL;
int exists_in_emptied = dbFind(emptied, key->ptr) != NULL;
if (exists_in_emptied ||
(replaced_with && dictFind(replaced_with->dict[calculateKeySlot(key->ptr)], key->ptr)))
(replaced_with && dbFind(replaced_with, key->ptr) != NULL))
{
list *clients = dictGetVal(de);
if (!clients) continue;
......@@ -442,7 +442,7 @@ void touchAllWatchedKeysInDb(redisDb *emptied, redisDb *replaced_with) {
while((ln = listNext(&li))) {
watchedKey *wk = redis_member2struct(watchedKey, node, ln);
if (wk->expired) {
if (!replaced_with || !dictFind(replaced_with->dict[calculateKeySlot(key->ptr)], key->ptr)) {
if (!replaced_with || !dbFind(replaced_with, key->ptr)) {
/* Expired key now deleted. No logical change. Clear the
* flag. Deleted keys are not flagged as expired. */
wk->expired = 0;
......
......@@ -1246,18 +1246,19 @@ struct redisMemOverhead *getMemoryOverheadData(void) {
for (j = 0; j < server.dbnum; j++) {
redisDb *db = server.db+j;
unsigned long long keyscount = dbSize(db, DB_MAIN);
unsigned long long keyscount = kvstoreSize(db->keys);
if (keyscount == 0) continue;
mh->total_keys += keyscount;
mh->db = zrealloc(mh->db,sizeof(mh->db[0])*(mh->num_dbs+1));
mh->db[mh->num_dbs].dbid = j;
mem = dbMemUsage(db, DB_MAIN);
mem = kvstoreMemUsage(db->keys) +
keyscount * sizeof(robj);
mh->db[mh->num_dbs].overhead_ht_main = mem;
mem_total+=mem;
mem = dbMemUsage(db, DB_EXPIRES);
mem = kvstoreMemUsage(db->expires);
mh->db[mh->num_dbs].overhead_ht_expires = mem;
mem_total+=mem;
......@@ -1544,7 +1545,7 @@ NULL
return;
}
}
if ((de = dbFind(c->db, c->argv[2]->ptr, DB_MAIN)) == NULL) {
if ((de = dbFind(c->db, c->argv[2]->ptr)) == NULL) {
addReplyNull(c);
return;
}
......
......@@ -36,7 +36,7 @@ typedef struct pubsubtype {
int shard;
dict *(*clientPubSubChannels)(client*);
int (*subscriptionCount)(client*);
dict **(*serverPubSubChannels)(unsigned int);
kvstore **serverPubSubChannels;
robj **subscribeMsg;
robj **unsubscribeMsg;
robj **messageBulk;
......@@ -62,22 +62,12 @@ dict* getClientPubSubChannels(client *c);
*/
dict* getClientPubSubShardChannels(client *c);
/*
* Get server's global Pub/Sub channels dict.
*/
dict **getServerPubSubChannels(unsigned int slot);
/*
* Get server's shard level Pub/Sub channels dict.
*/
dict **getServerPubSubShardChannels(unsigned int slot);
/*
* Get list of channels client is subscribed to.
* If a pattern is provided, the subset of channels is returned
* matching the pattern.
*/
void channelList(client *c, sds pat, dict** pubsub_channels, int is_sharded);
void channelList(client *c, sds pat, kvstore *pubsub_channels);
/*
* Pub/Sub type for global channels.
......@@ -86,7 +76,7 @@ pubsubtype pubSubType = {
.shard = 0,
.clientPubSubChannels = getClientPubSubChannels,
.subscriptionCount = clientSubscriptionsCount,
.serverPubSubChannels = getServerPubSubChannels,
.serverPubSubChannels = &server.pubsub_channels,
.subscribeMsg = &shared.subscribebulk,
.unsubscribeMsg = &shared.unsubscribebulk,
.messageBulk = &shared.messagebulk,
......@@ -99,7 +89,7 @@ pubsubtype pubSubShardType = {
.shard = 1,
.clientPubSubChannels = getClientPubSubShardChannels,
.subscriptionCount = clientShardSubscriptionsCount,
.serverPubSubChannels = getServerPubSubShardChannels,
.serverPubSubChannels = &server.pubsubshard_channels,
.subscribeMsg = &shared.ssubscribebulk,
.unsubscribeMsg = &shared.sunsubscribebulk,
.messageBulk = &shared.smessagebulk,
......@@ -218,15 +208,14 @@ void addReplyPubsubPatUnsubscribed(client *c, robj *pattern) {
/* Return the number of pubsub channels + patterns is handled. */
int serverPubsubSubscriptionCount(void) {
return dictSize(server.pubsub_channels) + dictSize(server.pubsub_patterns);
return kvstoreSize(server.pubsub_channels) + dictSize(server.pubsub_patterns);
}
/* Return the number of pubsub shard level channels is handled. */
int serverPubsubShardSubscriptionCount(void) {
return server.shard_channel_count;
return kvstoreSize(server.pubsubshard_channels);
}
/* Return the number of channels + patterns a client is subscribed to. */
int clientSubscriptionsCount(client *c) {
return dictSize(c->pubsub_channels) + dictSize(c->pubsub_patterns);
......@@ -245,16 +234,6 @@ dict* getClientPubSubShardChannels(client *c) {
return c->pubsubshard_channels;
}
dict **getServerPubSubChannels(unsigned int slot) {
UNUSED(slot);
return &server.pubsub_channels;
}
dict **getServerPubSubShardChannels(unsigned int slot) {
serverAssert(server.cluster_enabled || slot == 0);
return &server.pubsubshard_channels[slot];
}
/* Return the number of pubsub + pubsub shard level channels
* a client is subscribed to. */
int clientTotalPubSubSubscriptionCount(client *c) {
......@@ -278,8 +257,7 @@ void unmarkClientAsPubSub(client *c) {
/* Subscribe a client to a channel. Returns 1 if the operation succeeded, or
* 0 if the client was already subscribed to that channel. */
int pubsubSubscribeChannel(client *c, robj *channel, pubsubtype type) {
dict **d_ptr;
dictEntry *de;
dictEntry *de, *existing;
dict *clients = NULL;
int retval = 0;
unsigned int slot = 0;
......@@ -292,23 +270,17 @@ int pubsubSubscribeChannel(client *c, robj *channel, pubsubtype type) {
if (server.cluster_enabled && type.shard) {
slot = getKeySlot(channel->ptr);
}
d_ptr = type.serverPubSubChannels(slot);
if (*d_ptr == NULL) {
*d_ptr = dictCreate(&objToDictDictType);
de = NULL;
de = kvstoreDictAddRaw(*type.serverPubSubChannels, slot, channel, &existing);
if (existing) {
clients = dictGetVal(existing);
} else {
de = dictFind(*d_ptr, channel);
}
if (de == NULL) {
clients = dictCreate(&clientDictType);
dictAdd(*d_ptr, channel, clients);
kvstoreDictSetVal(*type.serverPubSubChannels, slot, de, clients);
incrRefCount(channel);
if (type.shard) {
server.shard_channel_count++;
}
} else {
clients = dictGetVal(de);
}
serverAssert(dictAdd(clients, c, NULL) != DICT_ERR);
}
/* Notify the client */
......@@ -319,7 +291,6 @@ int pubsubSubscribeChannel(client *c, robj *channel, pubsubtype type) {
/* Unsubscribe a client from a channel. Returns 1 if the operation succeeded, or
* 0 if the client was not subscribed to the specified channel. */
int pubsubUnsubscribeChannel(client *c, robj *channel, int notify, pubsubtype type) {
dict *d;
dictEntry *de;
dict *clients;
int retval = 0;
......@@ -334,9 +305,7 @@ int pubsubUnsubscribeChannel(client *c, robj *channel, int notify, pubsubtype ty
if (server.cluster_enabled && type.shard) {
slot = getKeySlot(channel->ptr);
}
d = *type.serverPubSubChannels(slot);
serverAssertWithInfo(c,NULL,d != NULL);
de = dictFind(d, channel);
de = kvstoreDictFind(*type.serverPubSubChannels, slot, channel);
serverAssertWithInfo(c,NULL,de != NULL);
clients = dictGetVal(de);
serverAssertWithInfo(c, NULL, dictDelete(clients, c) == DICT_OK);
......@@ -344,15 +313,7 @@ int pubsubUnsubscribeChannel(client *c, robj *channel, int notify, pubsubtype ty
/* Free the dict and associated hash entry at all if this was
* the latest client, so that it will be possible to abuse
* Redis PUBSUB creating millions of channels. */
dictDelete(d, channel);
if (type.shard) {
if (dictSize(d) == 0) {
dictRelease(d);
dict **d_ptr = type.serverPubSubChannels(slot);
*d_ptr = NULL;
}
server.shard_channel_count--;
}
kvstoreDictDelete(*type.serverPubSubChannels, slot, channel);
}
}
/* Notify the client */
......@@ -365,11 +326,10 @@ int pubsubUnsubscribeChannel(client *c, robj *channel, int notify, pubsubtype ty
/* Unsubscribe all shard channels in a slot. */
void pubsubShardUnsubscribeAllChannelsInSlot(unsigned int slot) {
dict *d = server.pubsubshard_channels[slot];
if (!d) {
if (!kvstoreDictSize(server.pubsubshard_channels, slot))
return;
}
dictIterator *di = dictGetSafeIterator(d);
dictIterator *di = kvstoreDictGetSafeIterator(server.pubsubshard_channels, slot);
dictEntry *de;
while ((de = dictNext(di)) != NULL) {
robj *channel = dictGetKey(de);
......@@ -389,12 +349,9 @@ void pubsubShardUnsubscribeAllChannelsInSlot(unsigned int slot) {
}
}
dictReleaseIterator(iter);
server.shard_channel_count--;
dictDelete(d, channel);
kvstoreDictDelete(server.pubsubshard_channels, slot, channel);
}
dictReleaseIterator(di);
dictRelease(d);
server.pubsubshard_channels[slot] = NULL;
}
/* Subscribe a client to a pattern. Returns 1 if the operation succeeded, or 0 if the client was already subscribed to that pattern. */
......@@ -513,7 +470,6 @@ int pubsubUnsubscribeAllPatterns(client *c, int notify) {
*/
int pubsubPublishMessageInternal(robj *channel, robj *message, pubsubtype type) {
int receivers = 0;
dict *d;
dictEntry *de;
dictIterator *di;
unsigned int slot = 0;
......@@ -522,8 +478,7 @@ int pubsubPublishMessageInternal(robj *channel, robj *message, pubsubtype type)
if (server.cluster_enabled && type.shard) {
slot = keyHashSlot(channel->ptr, sdslen(channel->ptr));
}
d = *type.serverPubSubChannels(slot);
de = d ? dictFind(d, channel) : NULL;
de = kvstoreDictFind(*type.serverPubSubChannels, slot, channel);
if (de) {
dict *clients = dictGetVal(de);
dictEntry *entry;
......@@ -693,14 +648,14 @@ NULL
{
/* PUBSUB CHANNELS [<pattern>] */
sds pat = (c->argc == 2) ? NULL : c->argv[2]->ptr;
channelList(c, pat, &server.pubsub_channels, 0);
channelList(c, pat, server.pubsub_channels);
} else if (!strcasecmp(c->argv[1]->ptr,"numsub") && c->argc >= 2) {
/* PUBSUB NUMSUB [Channel_1 ... Channel_N] */
int j;
addReplyArrayLen(c,(c->argc-2)*2);
for (j = 2; j < c->argc; j++) {
dict *d = dictFetchValue(server.pubsub_channels, c->argv[j]);
dict *d = kvstoreDictFetchValue(server.pubsub_channels, 0, c->argv[j]);
addReplyBulk(c,c->argv[j]);
addReplyLongLong(c, d ? dictSize(d) : 0);
......@@ -713,35 +668,33 @@ NULL
{
/* PUBSUB SHARDCHANNELS */
sds pat = (c->argc == 2) ? NULL : c->argv[2]->ptr;
channelList(c,pat,server.pubsubshard_channels,server.cluster_enabled);
channelList(c,pat,server.pubsubshard_channels);
} else if (!strcasecmp(c->argv[1]->ptr,"shardnumsub") && c->argc >= 2) {
/* PUBSUB SHARDNUMSUB [ShardChannel_1 ... ShardChannel_N] */
int j;
addReplyArrayLen(c, (c->argc-2)*2);
for (j = 2; j < c->argc; j++) {
unsigned int slot = calculateKeySlot(c->argv[j]->ptr);
dict *d = server.pubsubshard_channels[slot];
dict *clients = d ? dictFetchValue(d, c->argv[j]) : NULL;
dict *clients = kvstoreDictFetchValue(server.pubsubshard_channels, slot, c->argv[j]);
addReplyBulk(c,c->argv[j]);
addReplyLongLong(c, d ? dictSize(clients) : 0);
addReplyLongLong(c, clients ? dictSize(clients) : 0);
}
} else {
addReplySubcommandSyntaxError(c);
}
}
void channelList(client *c, sds pat, dict **pubsub_channels, int is_sharded) {
void channelList(client *c, sds pat, kvstore *pubsub_channels) {
long mblen = 0;
void *replylen;
unsigned int slot_cnt = is_sharded ? CLUSTER_SLOTS : 1;
unsigned int slot_cnt = kvstoreNumDicts(pubsub_channels);
replylen = addReplyDeferredLen(c);
for (unsigned int i = 0; i < slot_cnt; i++) {
if (pubsub_channels[i] == NULL) {
if (!kvstoreDictSize(pubsub_channels, i))
continue;
}
dictIterator *di = dictGetIterator(pubsub_channels[i]);
dictIterator *di = kvstoreDictGetIterator(pubsub_channels, i);
dictEntry *de;
while((de = dictNext(di)) != NULL) {
robj *cobj = dictGetKey(de);
......@@ -805,3 +758,9 @@ size_t pubsubMemOverhead(client *c) {
mem += dictMemUsage(c->pubsubshard_channels);
return mem;
}
int pubsubTotalSubscriptions(void) {
return dictSize(server.pubsub_patterns) +
kvstoreSize(server.pubsub_channels) +
kvstoreSize(server.pubsubshard_channels);
}
......@@ -1301,12 +1301,12 @@ ssize_t rdbSaveDb(rio *rdb, int dbid, int rdbflags, long *key_counter) {
dictEntry *de;
ssize_t written = 0;
ssize_t res;
dbIterator *dbit = NULL;
kvstoreIterator *kvs_it = NULL;
static long long info_updated_time = 0;
char *pname = (rdbflags & RDBFLAGS_AOF_PREAMBLE) ? "AOF rewrite" : "RDB";
redisDb *db = server.db + dbid;
unsigned long long int db_size = dbSize(db, DB_MAIN);
unsigned long long int db_size = kvstoreSize(db->keys);
if (db_size == 0) return 0;
/* Write the SELECT DB opcode */
......@@ -1316,7 +1316,7 @@ ssize_t rdbSaveDb(rio *rdb, int dbid, int rdbflags, long *key_counter) {
written += res;
/* Write the RESIZE DB opcode. */
unsigned long long expires_size = dbSize(db, DB_EXPIRES);
unsigned long long expires_size = kvstoreSize(db->expires);
if ((res = rdbSaveType(rdb,RDB_OPCODE_RESIZEDB)) < 0) goto werr;
written += res;
if ((res = rdbSaveLen(rdb,db_size)) < 0) goto werr;
......@@ -1324,20 +1324,20 @@ ssize_t rdbSaveDb(rio *rdb, int dbid, int rdbflags, long *key_counter) {
if ((res = rdbSaveLen(rdb,expires_size)) < 0) goto werr;
written += res;
dbit = dbIteratorInit(db, DB_MAIN);
kvs_it = kvstoreIteratorInit(db->keys);
int last_slot = -1;
/* Iterate this DB writing every entry */
while ((de = dbIteratorNext(dbit)) != NULL) {
int curr_slot = dbIteratorGetCurrentSlot(dbit);
while ((de = kvstoreIteratorNext(kvs_it)) != NULL) {
int curr_slot = kvstoreIteratorGetCurrentDictIndex(kvs_it);
/* Save slot info. */
if (server.cluster_enabled && curr_slot != last_slot) {
if ((res = rdbSaveType(rdb, RDB_OPCODE_SLOT_INFO)) < 0) goto werr;
written += res;
if ((res = rdbSaveLen(rdb, curr_slot)) < 0) goto werr;
written += res;
if ((res = rdbSaveLen(rdb, dictSize(db->dict[curr_slot]))) < 0) goto werr;
if ((res = rdbSaveLen(rdb, kvstoreDictSize(db->keys, curr_slot))) < 0) goto werr;
written += res;
if ((res = rdbSaveLen(rdb, dictSize(db->expires[curr_slot]))) < 0) goto werr;
if ((res = rdbSaveLen(rdb, kvstoreDictSize(db->expires, curr_slot))) < 0) goto werr;
written += res;
last_slot = curr_slot;
}
......@@ -1368,11 +1368,11 @@ ssize_t rdbSaveDb(rio *rdb, int dbid, int rdbflags, long *key_counter) {
}
}
}
dbReleaseIterator(dbit);
kvstoreIteratorRelease(kvs_it);
return written;
werr:
if (dbit) dbReleaseIterator(dbit);
if (kvs_it) kvstoreIteratorRelease(kvs_it);
return -1;
}
......@@ -3027,7 +3027,6 @@ int rdbLoadRio(rio *rdb, int rdbflags, rdbSaveInfo *rsi) {
return retval;
}
/* Load an RDB file from the rio stream 'rdb'. On success C_OK is returned,
* otherwise C_ERR is returned.
* The rdb_loading_ctx argument holds objects to which the rdb will be loaded to,
......@@ -3131,8 +3130,8 @@ int rdbLoadRioWithLoadingCtx(rio *rdb, int rdbflags, rdbSaveInfo *rsi, rdbLoadin
continue; /* Ignore gracefully. */
}
/* In cluster mode we resize individual slot specific dictionaries based on the number of keys that slot holds. */
dictExpand(db->dict[slot_id], slot_size);
dictExpand(db->expires[slot_id], expires_slot_size);
kvstoreDictExpand(db->keys, slot_id, slot_size);
kvstoreDictExpand(db->expires, slot_id, slot_size);
should_expand_db = 0;
continue; /* Read next opcode. */
} else if (type == RDB_OPCODE_AUX) {
......@@ -3266,8 +3265,8 @@ int rdbLoadRioWithLoadingCtx(rio *rdb, int rdbflags, rdbSaveInfo *rsi, rdbLoadin
/* If there is no slot info, it means that it's either not cluster mode or we are trying to load legacy RDB file.
* In this case we want to estimate number of keys per slot and resize accordingly. */
if (should_expand_db) {
dbExpand(db, db_size, DB_MAIN, 0);
dbExpand(db, expires_size, DB_EXPIRES, 0);
dbExpand(db, db_size, 0);
dbExpandExpires(db, db_size, 0);
should_expand_db = 0;
}
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment