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

Make dbGetNextNonEmptySlot use intset and address some other feedback

parent 3d1a1781
...@@ -398,5 +398,6 @@ void clusterUpdateMyselfAnnouncedPorts(void); ...@@ -398,5 +398,6 @@ void clusterUpdateMyselfAnnouncedPorts(void);
sds clusterGenNodesDescription(int filter, int use_pport); sds clusterGenNodesDescription(int filter, int use_pport);
sds genClusterInfoString(); sds genClusterInfoString();
void freeClusterLink(clusterLink *link); void freeClusterLink(clusterLink *link);
int clusterNodeGetSlotBit(clusterNode *n, int slot);
#endif /* __CLUSTER_H */ #endif /* __CLUSTER_H */
...@@ -70,19 +70,21 @@ void dbIteratorInit(dbIterator *dbit, redisDb *db) { ...@@ -70,19 +70,21 @@ void dbIteratorInit(dbIterator *dbit, redisDb *db) {
dbit->cur_slot = -1; dbit->cur_slot = -1;
} }
/* Returns next non-empty dictionary strictly after provided slot and updates slot id in the supplied reference. /* Returns next non-empty dictionary strictly after provided slot and updates slot id in the supplied reference. */
* This function doesn't use dbIterator in order to provide deterministic iteration across all owned slots. */ dict *dbGetNextNonEmptySlot(redisDb *db, int *slot) {
dict *dbGetNextUnvisitedSlot(redisDb *db, int *slot) { uint32_t pos;
for (int i = *slot + 1; i < db->dict_count; i++) { int found = intsetSearch(db->non_empty_dicts, *slot, &pos);
if (!dictIsEmpty(db->dict[i])) { if (found) pos++; /* If current slot exists in the non-empty list, then we want next one, otherwise pos already points to it. */
*slot = i; int64_t next_slot;
return db->dict[i]; if (intsetGet(db->non_empty_dicts, pos, &next_slot)) {
} *slot = (int) next_slot;
return db->dict[*slot];
} }
*slot = -1; *slot = -1;
return NULL; return NULL;
} }
/* Update LFU when an object is accessed. /* Update LFU when an object is accessed.
* Firstly, decrement the counter if the decrement time is reached. * Firstly, decrement the counter if the decrement time is reached.
* Then logarithmically increment the counter, and update the access time. */ * Then logarithmically increment the counter, and update the access time. */
...@@ -1031,7 +1033,7 @@ void scanGenericCommand(client *c, robj *o, unsigned long long cursor) { ...@@ -1031,7 +1033,7 @@ void scanGenericCommand(client *c, robj *o, unsigned long long cursor) {
/* In cluster mode there is a separate dictionary for each slot. /* In cluster mode there is a separate dictionary for each slot.
* If cursor is empty, we should try exploring next non-empty slot. */ * If cursor is empty, we should try exploring next non-empty slot. */
if (o == NULL && !cursor) { if (o == NULL && !cursor) {
ht = dbGetNextUnvisitedSlot(c->db, &slot); ht = dbGetNextNonEmptySlot(c->db, &slot);
} }
} while ((cursor || slot > 0) && /* Continue iteration if there are more slots to visit, or cursor hasn't reached the end of dict yet. */ } while ((cursor || slot > 0) && /* Continue iteration if there are more slots to visit, or cursor hasn't reached the end of dict yet. */
maxiterations-- && maxiterations-- &&
......
...@@ -1020,7 +1020,7 @@ void activeDefragCycle(void) { ...@@ -1020,7 +1020,7 @@ void activeDefragCycle(void) {
if (!expires_cursor) if (!expires_cursor)
cursor = dictScanDefrag(d, cursor, defragScanCallback, cursor = dictScanDefrag(d, cursor, defragScanCallback,
&defragfns, db); &defragfns, db);
if (!cursor) d = dbGetNextUnvisitedSlot(db, &slot); if (!cursor) d = dbGetNextNonEmptySlot(db, &slot);
/* When done scanning the keyspace dict, we scan the expire dict. */ /* When done scanning the keyspace dict, we scan the expire dict. */
if (!cursor && slot > -1) if (!cursor && slot > -1)
expires_cursor = dictScanDefrag(db->expires, expires_cursor, expires_cursor = dictScanDefrag(db->expires, expires_cursor,
......
...@@ -114,7 +114,7 @@ static intset *intsetResize(intset *is, uint32_t len) { ...@@ -114,7 +114,7 @@ static intset *intsetResize(intset *is, uint32_t len) {
* sets "pos" to the position of the value within the intset. Return 0 when * sets "pos" to the position of the value within the intset. Return 0 when
* the value is not present in the intset and sets "pos" to the position * the value is not present in the intset and sets "pos" to the position
* where "value" can be inserted. */ * where "value" can be inserted. */
static uint8_t intsetSearch(intset *is, int64_t value, uint32_t *pos) { uint8_t intsetSearch(intset *is, int64_t value, uint32_t *pos) {
int min = 0, max = intrev32ifbe(is->length)-1, mid = -1; int min = 0, max = intrev32ifbe(is->length)-1, mid = -1;
int64_t cur = -1; int64_t cur = -1;
......
...@@ -42,6 +42,7 @@ intset *intsetNew(void); ...@@ -42,6 +42,7 @@ intset *intsetNew(void);
intset *intsetAdd(intset *is, int64_t value, uint8_t *success); intset *intsetAdd(intset *is, int64_t value, uint8_t *success);
intset *intsetRemove(intset *is, int64_t value, int *success); intset *intsetRemove(intset *is, int64_t value, int *success);
uint8_t intsetFind(intset *is, int64_t value); uint8_t intsetFind(intset *is, int64_t value);
uint8_t intsetSearch(intset *is, int64_t value, uint32_t *pos);
int64_t intsetRandom(intset *is); int64_t intsetRandom(intset *is);
int64_t intsetMax(intset *is); int64_t intsetMax(intset *is);
int64_t intsetMin(intset *is); int64_t intsetMin(intset *is);
......
...@@ -10369,7 +10369,7 @@ int RM_Scan(RedisModuleCtx *ctx, RedisModuleScanCursor *cursor, RedisModuleScanC ...@@ -10369,7 +10369,7 @@ int RM_Scan(RedisModuleCtx *ctx, RedisModuleScanCursor *cursor, RedisModuleScanC
dict *dict = ctx->client->db->dict[slot]; dict *dict = ctx->client->db->dict[slot];
cursor->cursor = dictScan(dict, cursor->cursor, moduleScanCallback, &data); cursor->cursor = dictScan(dict, cursor->cursor, moduleScanCallback, &data);
if (cursor->cursor == 0) { if (cursor->cursor == 0) {
dict = dbGetNextUnvisitedSlot(ctx->client->db, &slot); dict = dbGetNextNonEmptySlot(ctx->client->db, &slot);
if (dict == NULL) { if (dict == NULL) {
cursor->done = 1; cursor->done = 1;
ret = 0; ret = 0;
......
...@@ -181,6 +181,5 @@ int rdbFunctionLoad(rio *rdb, int ver, functionsLibCtx* lib_ctx, int rdbflags, s ...@@ -181,6 +181,5 @@ int rdbFunctionLoad(rio *rdb, int ver, functionsLibCtx* lib_ctx, int rdbflags, s
int rdbSaveRio(int req, rio *rdb, int *error, int rdbflags, rdbSaveInfo *rsi); int rdbSaveRio(int req, rio *rdb, int *error, int rdbflags, rdbSaveInfo *rsi);
ssize_t rdbSaveFunctions(rio *rdb); ssize_t rdbSaveFunctions(rio *rdb);
rdbSaveInfo *rdbPopulateSaveInfo(rdbSaveInfo *rsi); rdbSaveInfo *rdbPopulateSaveInfo(rdbSaveInfo *rsi);
int clusterNodeGetSlotBit(clusterNode *n, int slot);
#endif #endif
...@@ -2370,7 +2370,7 @@ typedef struct dbIterator { ...@@ -2370,7 +2370,7 @@ typedef struct dbIterator {
void dbIteratorInit(dbIterator *dbit, redisDb *db); void dbIteratorInit(dbIterator *dbit, redisDb *db);
dict *dbIteratorNextDict(dbIterator *dbit); dict *dbIteratorNextDict(dbIterator *dbit);
int dbIterCurSlot(dbIterator *dbit); int dbIterCurSlot(dbIterator *dbit);
dict *dbGetNextUnvisitedSlot(redisDb *db, int *slot); dict *dbGetNextNonEmptySlot(redisDb *db, int *slot);
/* SCAN specific commands for easy cursor manipulation, shared between main code and modules. */ /* SCAN specific commands for easy cursor manipulation, shared between main code and modules. */
int getAndClearSlotIdFromCursor(unsigned long long int *cursor); 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