Commit 6283de4c authored by Vitaly Arbuzov's avatar Vitaly Arbuzov
Browse files

Fixed module Scan API and cleaned up regular Scan implementation

parent a68a9731
......@@ -48,7 +48,6 @@
int expireIfNeeded(redisDb *db, robj *key, int flags);
int keyIsExpired(redisDb *db, robj *key);
dict *dbNextDict(dbIterator *iter) {
while (iter->index < iter->db->dict_count - 1) {
iter->index++;
......@@ -64,12 +63,31 @@ dict *dbNextDict(dbIterator *iter) {
}
dbIterator *dbGetIterator(redisDb *db) {
return dbGetIteratorAt(db, 0);
}
dbIterator *dbGetIteratorAt(redisDb *db, int slot) {
serverAssert(slot == 0 || server.cluster_enabled);
dbIterator *iter = zmalloc(sizeof(*iter));
iter->db = db;
iter->index = -1;
iter->index = slot - 1; /* Start one slot ahead, as dbNextDict increments index right away. */
return iter;
}
dict *dbGetNextUnvisitedSlot(redisDb *db, int *slot) {
if (*slot < db->dict_count - 1) {
dbIterator *dbit = dbGetIteratorAt(db, *slot + 1); /* Scan on the current slot has already returned 0, find next non-empty dict. */
dict *dict = dbNextDict(dbit);
if (dict != NULL) {
*slot = dbit->index;
return dict;
}
dbReleaseIterator(dbit);
}
*slot = -1;
return NULL;
}
void dbReleaseIterator(dbIterator *iter) {
zfree(iter);
}
......@@ -958,12 +976,10 @@ void scanGenericCommand(client *c, robj *o, unsigned long long cursor) {
/* Handle the case of a hash table. */
ht = NULL;
unsigned long long slot_mask = (unsigned long long)(CLUSTER_SLOTS - 1) << SLOT_MASK_SHIFT;
/* During main dictionary traversal in cluster mode, 48 lower bits in the cursor are used for positioning in the HT.
* Following 14 bits are used for the slot number, ranging from 0 to 2^14-1.
* Slot is always 0 at the start of iteration and can be incremented only in cluster mode. */
int slot = (int) ((cursor & slot_mask) >> SLOT_MASK_SHIFT);
cursor = cursor & ~slot_mask; // clear slot mask from cursor for the time of iteration.
int slot = getAndClearSlotIdFromCursor(&cursor);
if (o == NULL) {
ht = c->db->dict[slot];
} else if (o->type == OBJ_SET && o->encoding == OBJ_ENCODING_HT) {
......@@ -990,24 +1006,17 @@ void scanGenericCommand(client *c, robj *o, unsigned long long cursor) {
* it is possible to fetch more data in a type-dependent way. */
privdata[0] = keys;
privdata[1] = o;
int hasUnvisitedSlot;
do {
hasUnvisitedSlot = 0;
cursor = dictScan(ht, cursor, scanCallback, privdata);
/* In cluster mode there is a separate dictionary for each slot.
* If cursor is empty, we should try exploring next non-empty slot. */
while (o == NULL && !cursor && slot < c->db->dict_count - 1) {
// TODO (vitarb@) Consider using dbIterator, or add slot ownership check.
slot++; ht = c->db->dict[slot];
if (dictSize(ht) > 0) {
hasUnvisitedSlot = 1;
break; // Found next non-empty slot.
}
if (o == NULL && !cursor) {
ht = dbGetNextUnvisitedSlot(c->db, &slot);
}
} while ((cursor || hasUnvisitedSlot) &&
} while ((cursor || slot > 0) &&
maxiterations-- &&
listLength(keys) < (unsigned long)count);
if (cursor) cursor |= ((unsigned long long) slot) << SLOT_MASK_SHIFT;
listLength(keys) < (unsigned long) count);
addSlotIdToCursor(slot, &cursor);
} else if (o->type == OBJ_SET) {
char *str;
size_t len;
......@@ -1111,6 +1120,18 @@ cleanup:
listRelease(keys);
}
void addSlotIdToCursor(int slot, unsigned long long int *cursor) {
if (slot > 0) {
(*cursor) |= ((unsigned long long) slot) << SLOT_MASK_SHIFT;
}
}
int getAndClearSlotIdFromCursor(unsigned long long int *cursor) {
int slot = (int) ((*cursor & SLOT_MASK) >> SLOT_MASK_SHIFT);
(*cursor) = (*cursor) & ~SLOT_MASK;
return slot;
}
/* The SCAN command completely relies on scanGenericCommand. */
void scanCommand(client *c) {
unsigned long long cursor;
......
......@@ -10263,7 +10263,7 @@ typedef struct {
} ScanCBData;
 
typedef struct RedisModuleScanCursor{
unsigned long cursor;
unsigned long long cursor;
int done;
}RedisModuleScanCursor;
 
......@@ -10365,11 +10365,17 @@ int RM_Scan(RedisModuleCtx *ctx, RedisModuleScanCursor *cursor, RedisModuleScanC
}
int ret = 1;
ScanCBData data = { ctx, privdata, fn };
// cursor->cursor = dictScan(ctx->client->db->dict, cursor->cursor, moduleScanCallback, &data); FIXME (vitarb) https://sim.amazon.com/issues/ELMO-63799
int slot = getAndClearSlotIdFromCursor(&cursor->cursor);
dict *dict = ctx->client->db->dict[slot];
cursor->cursor = dictScan(dict, cursor->cursor, moduleScanCallback, &data);
if (cursor->cursor == 0) {
cursor->done = 1;
ret = 0;
dict = dbGetNextUnvisitedSlot(ctx->client->db, &slot);
if (dict == NULL) {
cursor->done = 1;
ret = 0;
}
}
addSlotIdToCursor(slot, &cursor->cursor);
errno = 0;
return ret;
}
......
......@@ -2363,8 +2363,14 @@ typedef struct dbIterator {
/* dbIterator specific functions */
dict *dbNextDict(dbIterator *iter);
dbIterator *dbGetIterator(redisDb *db);
dbIterator *dbGetIteratorAt(redisDb *db, int slot);
dict *dbGetNextUnvisitedSlot(redisDb *db, int *slot);
void dbReleaseIterator(dbIterator *iter);
/* SCAN specific commands for easy cursor manipulation, shared between main code and modules. */
int getAndClearSlotIdFromCursor(unsigned long long int *cursor);
void addSlotIdToCursor(int slot, unsigned long long int *cursor);
/* Structure to hold hash iteration abstraction. Note that iteration over
* hashes involves both fields and values. Because it is possible that
* not both are required, store pointers in the iterator to avoid
......@@ -3107,6 +3113,7 @@ sds keyspaceEventsFlagsToString(int flags);
#define PERCENT_CONFIG (1<<1) /* Indicates if this value can be loaded as a percent (and stored as a negative int) */
#define OCTAL_CONFIG (1<<2) /* This value uses octal representation */
#define SLOT_MASK_SHIFT 48 /* Slot id is stored in high bits of the Scan API cursor, this number defines by how many bits it's shifted. */
#define SLOT_MASK ((unsigned long long)(CLUSTER_SLOTS - 1) << SLOT_MASK_SHIFT)
/* Enum Configs contain an array of configEnum objects that match a string with an integer. */
typedef struct configEnum {
......
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