Commit ed138c0a authored by Vitaly Arbuzov's avatar Vitaly Arbuzov
Browse files

Use binary index in slot iteration instead of maintaining an intset

parent af1bb938
......@@ -51,15 +51,10 @@ void cumulativeKeyCountAdd(redisDb *db, int idx, long delta);
/* Returns next dictionary from the iterator, or NULL if iteration is complete. */
dict *dbIteratorNextDict(dbIterator *dbit) {
int64_t slot;
dbit->cur_slot = -1;
if (!server.cluster_enabled) {
dbit->cur_slot = dbit->index++ ? -1 : 0;
} else if (intsetGet(dbit->db->non_empty_dicts, dbit->index++, &slot)){
dbit->cur_slot = (int) slot;
return dbit->db->dict[slot];
}
return dbit->cur_slot >= 0 ? dbit->db->dict[dbit->cur_slot] : NULL;
if (dbit->next_slot == -1) return NULL;
dbit->slot = dbit->next_slot;
dbit->next_slot = dbGetNextNonEmptySlot(dbit->db, dbit->slot);
return dbit->db->dict[dbit->slot];
}
/* Returns next entry from the multi slot db. */
......@@ -69,7 +64,7 @@ dictEntry *dbIteratorNext(dbIterator *dbit) {
dict *d = dbIteratorNextDict(dbit);
if (!d) return NULL;
dictInitSafeIterator(&dbit->di, d);
return dictNext(&dbit->di);
de = dictNext(&dbit->di);
}
return de;
}
......@@ -79,25 +74,15 @@ dictEntry *dbIteratorNext(dbIterator *dbit) {
* or 16k dictionaries (one per slot) when node runs with cluster mode enabled. */
void dbIteratorInit(dbIterator *dbit, redisDb *db) {
dbit->db = db;
dbit->index = 0;
dbit->cur_slot = -1;
dbit->slot = -1;
dbit->next_slot = findSlotByKeyIndex(dbit->db, 1); /* Finds first non-empty slot. */
dictInitSafeIterator(&dbit->di, NULL);
}
/* Returns next non-empty dictionary strictly after provided slot and updates slot id in the supplied reference. */
dict *dbGetNextNonEmptySlot(redisDb *db, int *slot) {
if (server.cluster_enabled) {
uint32_t pos;
int found = intsetSearch(db->non_empty_dicts, *slot, &pos);
if (found) pos++; /* If current slot exists in the non-empty list, then we want next one, otherwise pos already points to it. */
int64_t next_slot;
if (intsetGet(db->non_empty_dicts, pos, &next_slot)) {
*slot = (int) next_slot;
return db->dict[*slot];
}
}
*slot = -1;
return NULL;
/* Returns next non-empty slot strictly after given one, or -1 if provided slot is the last one. */
int dbGetNextNonEmptySlot(redisDb *db, int slot) {
unsigned long long next_key = cumulativeKeyCountRead(db, slot) + 1;
return next_key <= dbSize(db) ? findSlotByKeyIndex(db, next_key) : -1;
}
......@@ -245,17 +230,10 @@ void dbAdd(redisDb *db, robj *key, robj *val) {
sds copy = sdsdup(key->ptr);
int slot = getKeySlot(key->ptr);
dict *d = db->dict[slot];
int new_dict = server.cluster_enabled && dictIsEmpty(d);
dictEntry *de = dictAddRaw(d, copy, NULL);
serverAssertWithInfo(NULL, key, de != NULL);
dictSetVal(d, de, val);
db->key_count++;
/* If dict transitioned from empty to non-empty, we should add it to the list of owned slots. Cluster mode only. */
if (new_dict) {
uint8_t success = 0;
db->non_empty_dicts = intsetAdd(db->non_empty_dicts, slot, &success);
serverAssert(success);
}
if (server.cluster_enabled) {
cumulativeKeyCountAdd(db, slot, 1);
}
......@@ -288,17 +266,10 @@ int getKeySlot(sds key) {
int dbAddRDBLoad(redisDb *db, sds key, robj *val) {
int slot = getKeySlot(key);
dict *d = db->dict[slot];
int new_dict = server.cluster_enabled && dictIsEmpty(d);
dictEntry *de = dictAddRaw(d, key, NULL);
if (de == NULL) return 0;
dictSetVal(d, de, val);
db->key_count++;
/* If dict transitioned from empty to non-empty, we should add it to the list of owned slots. */
if (new_dict) {
uint8_t success = 0;
db->non_empty_dicts = intsetAdd(db->non_empty_dicts, slot, &success);
serverAssert(success);
}
if (server.cluster_enabled) {
cumulativeKeyCountAdd(db, slot, 1);
}
......@@ -456,8 +427,12 @@ unsigned long long cumulativeKeyCountRead(redisDb *db, int slot) {
* Implementation uses binary search on top of binary index tree.
* Time complexity of this function is O(log^2(CLUSTER_SLOTS)). */
dict *getFairRandomDict(redisDb *db) {
if (!server.cluster_enabled || dbSize(db) == 0) return db->dict[0];
/* We want to find a slot containing target-th element in a key space ordered by slot id.
unsigned long target = dbSize(db) ? (randomULong() % dbSize(db)) + 1 : 0;
int slot = findSlotByKeyIndex(db, target);
return db->dict[slot];
}
/* Finds a slot containing target element in a key space ordered by slot id.
* Consider this example. Slots are represented by brackets and keys by dots:
* #0 #1 #2 #3 #4
* [..][....][...][.......][.]
......@@ -466,11 +441,13 @@ dict *getFairRandomDict(redisDb *db) {
*
* In this case slot #3 contains key that we are trying to find.
* */
unsigned long target = (randomULong() % dbSize(db)) + 1;
int findSlotByKeyIndex(redisDb *db, unsigned long target) {
if (!server.cluster_enabled || dbSize(db) == 0) return 0;
serverAssert(target <= dbSize(db));
int lo = 0, hi = CLUSTER_SLOTS - 1;
/* We use binary search to find a slot, we are allowed to do this, because we have a quick way to find a total number of keys
* up until certain slot, using binary index tree. */
while (lo < hi) {
while (lo <= hi) {
int mid = lo + (hi - lo) / 2;
unsigned long keys_up_to_mid = cumulativeKeyCountRead(db, mid); /* Total number of keys up until a given slot (inclusive). */
unsigned long keys_in_mid = dictSize(db->dict[mid]); /* Number of keys in the slot. */
......@@ -479,13 +456,10 @@ dict *getFairRandomDict(redisDb *db) {
} else if (target <= keys_up_to_mid - keys_in_mid) { /* Target is to the left from mid. */
hi = mid - 1;
} else { /* Located target. */
break;
return mid;
}
}
int slot = lo + (hi - lo) / 2;
dict *result = db->dict[slot];
serverAssert(!dictIsEmpty(result));
return result;
serverPanic("Unable to find a slot that contains target key.");
}
/* Helper for sync and async delete. */
......@@ -516,12 +490,6 @@ int dbGenericDelete(redisDb *db, robj *key, int async, int flags) {
* the key, because it is shared with the main dictionary. */
if (dictSize(db->expires) > 0) dictDelete(db->expires,key->ptr);
dictTwoPhaseUnlinkFree(d,de,plink,table);
/* If we've removed last entry from the dict, then we need to also remove it from the list of owned non-empty slots. */
if (server.cluster_enabled && dictIsEmpty(d)) {
int success = 0;
db->non_empty_dicts = intsetRemove(db->non_empty_dicts, slot, &success);
serverAssert(success);
}
if (server.cluster_enabled) {
cumulativeKeyCountAdd(db, slot, -1);
}
......@@ -624,8 +592,6 @@ long long emptyDbStructure(redisDb *dbarray, int dbnum, int async,
if (server.cluster_enabled) {
zfree(dbarray[j].binary_index);
dbarray[j].binary_index = zcalloc(sizeof(unsigned long long) * (CLUSTER_SLOTS + 1));
zfree(dbarray[j].non_empty_dicts);
dbarray[j].non_empty_dicts = intsetNew();
}
}
......@@ -694,7 +660,6 @@ redisDb *initTempDb(void) {
tempDb[i].dict_count = (server.cluster_enabled) ? CLUSTER_SLOTS : 1;
tempDb[i].dict = dictCreateMultiple(&dbDictType, tempDb[i].dict_count);
tempDb[i].expires = dictCreate(&dbExpiresDictType);
tempDb[i].non_empty_dicts = server.cluster_enabled ? intsetNew() : NULL;
tempDb[i].binary_index = server.cluster_enabled ? zcalloc(sizeof(unsigned long long) * (CLUSTER_SLOTS + 1)) : NULL;
}
......@@ -715,7 +680,6 @@ void discardTempDb(redisDb *tempDb, void(callback)(dict*)) {
dictRelease(tempDb[i].expires);
if (server.cluster_enabled) {
zfree(tempDb[i].binary_index);
zfree(tempDb[i].non_empty_dicts);
}
}
......@@ -1117,7 +1081,8 @@ void scanGenericCommand(client *c, robj *o, unsigned long long cursor) {
/* In cluster mode there is a separate dictionary for each slot.
* If cursor is empty, we should try exploring next non-empty slot. */
if (o == NULL && !cursor) {
ht = dbGetNextNonEmptySlot(c->db, &slot);
slot = dbGetNextNonEmptySlot(c->db, slot);
ht = slot != -1 ? c->db->dict[slot] : NULL;
}
} while ((cursor || slot > 0) && /* Continue iteration if there are more slots to visit, or cursor hasn't reached the end of dict yet. */
maxiterations-- &&
......@@ -1660,7 +1625,6 @@ int dbSwapDatabases(int id1, int id2) {
db1->resize_cursor = db2->resize_cursor;
db1->dict_count = db2->dict_count;
db1->key_count = db2->key_count;
db1->non_empty_dicts = db2->non_empty_dicts;
db1->binary_index = db2->binary_index;
db2->dict = aux.dict;
......@@ -1670,7 +1634,6 @@ int dbSwapDatabases(int id1, int id2) {
db2->resize_cursor = aux.resize_cursor;
db2->dict_count = aux.dict_count;
db2->key_count = aux.key_count;
db2->non_empty_dicts = aux.non_empty_dicts;
db2->binary_index = aux.binary_index;
/* Now we need to handle clients blocked on lists: as an effect
......@@ -1712,7 +1675,6 @@ void swapMainDbWithTempDb(redisDb *tempDb) {
activedb->resize_cursor = newdb->resize_cursor;
activedb->dict_count = newdb->dict_count;
activedb->key_count = newdb->key_count;
activedb->non_empty_dicts = newdb->non_empty_dicts;
activedb->binary_index = newdb->binary_index;
newdb->dict = aux.dict;
......@@ -1722,7 +1684,6 @@ void swapMainDbWithTempDb(redisDb *tempDb) {
newdb->resize_cursor = aux.resize_cursor;
newdb->dict_count = aux.dict_count;
newdb->key_count = aux.key_count;
newdb->non_empty_dicts = aux.non_empty_dicts;
newdb->binary_index = aux.binary_index;
/* Now we need to handle clients blocked on lists: as an effect
......
......@@ -1007,9 +1007,9 @@ void activeDefragCycle(void) {
db = &server.db[current_db];
cursor = 0;
}
int slot = 0;
dict *d = db->dict[slot];
int slot = findSlotByKeyIndex(db, 1);
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, endtime)) {
quit = 1; /* time is up, we didn't finish all the work */
......@@ -1020,7 +1020,7 @@ void activeDefragCycle(void) {
if (!expires_cursor)
cursor = dictScanDefrag(d, cursor, defragScanCallback,
&defragfns, db);
if (!cursor) d = dbGetNextNonEmptySlot(db, &slot);
if (!cursor) slot = dbGetNextNonEmptySlot(db, slot);
/* When done scanning the keyspace dict, we scan the expire dict. */
if (!cursor && slot > -1)
expires_cursor = dictScanDefrag(db->expires, expires_cursor,
......
......@@ -10847,8 +10847,8 @@ int RM_Scan(RedisModuleCtx *ctx, RedisModuleScanCursor *cursor, RedisModuleScanC
dict *dict = ctx->client->db->dict[slot];
cursor->cursor = dictScan(dict, cursor->cursor, moduleScanCallback, &data);
if (cursor->cursor == 0) {
dict = dbGetNextNonEmptySlot(ctx->client->db, &slot);
if (dict == NULL) {
slot = dbGetNextNonEmptySlot(ctx->client->db, slot);
if (slot == -1) {
cursor->done = 1;
ret = 0;
}
......
......@@ -1325,15 +1325,15 @@ ssize_t rdbSaveDb(rio *rdb, int dbid, int rdbflags, long *key_counter) {
/* Iterate this DB writing every entry */
while ((de = dbIteratorNext(&dbit)) != NULL) {
/* Save slot info. */
if (server.cluster_enabled && dbit.cur_slot != last_slot) {
serverAssert(dbit.cur_slot >= 0 && dbit.cur_slot < CLUSTER_SLOTS);
if (server.cluster_enabled && dbit.slot != last_slot) {
serverAssert(dbit.slot >= 0 && dbit.slot < CLUSTER_SLOTS);
if ((res = rdbSaveType(rdb, RDB_OPCODE_SLOT_INFO)) < 0) goto werr;
written += res;
if ((res = rdbSaveLen(rdb, dbit.cur_slot)) < 0) goto werr;
if ((res = rdbSaveLen(rdb, dbit.slot)) < 0) goto werr;
written += res;
if ((res = rdbSaveLen(rdb, dictSize(db->dict[dbit.cur_slot]))) < 0) goto werr;
if ((res = rdbSaveLen(rdb, dictSize(db->dict[dbit.slot]))) < 0) goto werr;
written += res;
last_slot = dbit.cur_slot;
last_slot = dbit.slot;
}
sds keystr = dictGetKey(de);
robj key, *o = dictGetVal(de);
......
......@@ -592,15 +592,17 @@ void tryResizeHashTables(int dbid) {
dbIterator dbit;
redisDb *db = &server.db[dbid];
dbIteratorInit(&dbit, db);
dbit.index = db->resize_cursor;
if (db->resize_cursor != -1) {
dbit.next_slot = db->resize_cursor;
}
for (int i = 0; i < CRON_DICTS_PER_CALL; i++) {
d = dbIteratorNextDict(&dbit);
if (d == NULL) break;
if (!d) break;
if (htNeedsResize(d))
dictResize(d);
}
/* Save current index in the resize cursor, or start over if we've reached the end.*/
db->resize_cursor = dbit.cur_slot == -1 ? 0 : dbit.index;
/* Save current iterator position in the resize_cursor. */
db->resize_cursor = dbit.next_slot;
if (htNeedsResize(db->expires))
dictResize(db->expires);
......@@ -2633,7 +2635,6 @@ void initServer(void) {
server.db[j].rehashing = listCreate();
server.db[j].dict_count = slotCount;
server.db[j].key_count = 0;
server.db[j].non_empty_dicts = server.cluster_enabled ? intsetNew() : NULL;
server.db[j].binary_index = server.cluster_enabled ? zcalloc(sizeof(unsigned long long) * (CLUSTER_SLOTS + 1)) : NULL;
listSetFreeMethod(server.db[j].defrag_later,(void (*)(void*))sdsfree);
}
......
......@@ -975,7 +975,6 @@ typedef struct redisDb {
list *rehashing; /* List of dictionaries in this DB that are currently rehashing. */
int dict_count; /* Indicates total number of dictionaires owned by this DB, 1 dict per slot in cluster mode. */
unsigned long long key_count; /* Total number of keys in this DB. */
intset *non_empty_dicts; /* Set of non-empty dictionaries. */
unsigned long long *binary_index; /* Binary indexed tree (BIT) that describes cumulative key frequencies up until given slot. */
} redisDb;
......@@ -2437,8 +2436,8 @@ typedef struct {
/* Structure for DB iterator that allows iterating across multiple slot specific dictionaries in cluster mode. */
typedef struct dbIterator {
redisDb *db;
int index;
int cur_slot;
int slot;
int next_slot;
dictIterator di;
} dbIterator;
......@@ -2446,7 +2445,8 @@ typedef struct dbIterator {
void dbIteratorInit(dbIterator *dbit, redisDb *db);
dict *dbIteratorNextDict(dbIterator *dbit);
dictEntry *dbIteratorNext(dbIterator *iter);
dict *dbGetNextNonEmptySlot(redisDb *db, int *slot);
int dbGetNextNonEmptySlot(redisDb *db, int slot);
int findSlotByKeyIndex(redisDb *db, unsigned long target);
/* SCAN specific commands for easy cursor manipulation, shared between main code and modules. */
int getAndClearSlotIdFromCursor(unsigned long long int *cursor);
......
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