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

Populate non empty dictionaries only in cluster mode and rename owned_slots field

parent 402a277b
...@@ -51,12 +51,14 @@ int keyIsExpired(redisDb *db, robj *key); ...@@ -51,12 +51,14 @@ int keyIsExpired(redisDb *db, robj *key);
/* Returns next dictionary from the iterator, or NULL if iteration is complete. */ /* Returns next dictionary from the iterator, or NULL if iteration is complete. */
dict *dbIteratorNextDict(dbIterator *dbit) { dict *dbIteratorNextDict(dbIterator *dbit) {
int64_t slot; int64_t slot;
if (intsetGet(dbit->db->owned_slots, dbit->index++, &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; dbit->cur_slot = (int) slot;
return dbit->db->dict[slot]; return dbit->db->dict[slot];
} }
dbit->cur_slot = -1; return dbit->cur_slot >= 0 ? dbit->db->dict[dbit->cur_slot] : NULL;
return NULL;
} }
/* Returns DB iterator that can be used to iterate through sub-dictionaries. /* Returns DB iterator that can be used to iterate through sub-dictionaries.
...@@ -225,15 +227,15 @@ void dbAdd(redisDb *db, robj *key, robj *val) { ...@@ -225,15 +227,15 @@ void dbAdd(redisDb *db, robj *key, robj *val) {
sds copy = sdsdup(key->ptr); sds copy = sdsdup(key->ptr);
int slot = getKeySlot(key->ptr); int slot = getKeySlot(key->ptr);
dict *d = db->dict[slot]; dict *d = db->dict[slot];
int was_empty = dictIsEmpty(d); int new_dict = server.cluster_enabled && dictIsEmpty(d);
dictEntry *de = dictAddRaw(d, copy, NULL); dictEntry *de = dictAddRaw(d, copy, NULL);
serverAssertWithInfo(NULL, key, de != NULL); serverAssertWithInfo(NULL, key, de != NULL);
dictSetVal(d, de, val); dictSetVal(d, de, val);
db->key_count++; db->key_count++;
/* If dict transitioned from empty to non-empty, we should add it to the list of owned slots. */ /* If dict transitioned from empty to non-empty, we should add it to the list of owned slots. Cluster mode only. */
if (was_empty) { if (new_dict) {
uint8_t success = 0; uint8_t success = 0;
db->owned_slots = intsetAdd(db->owned_slots, slot, &success); db->non_empty_dicts = intsetAdd(db->non_empty_dicts, slot, &success);
serverAssert(success); serverAssert(success);
} }
signalKeyAsReady(db, key, val->type); signalKeyAsReady(db, key, val->type);
...@@ -265,15 +267,15 @@ int getKeySlot(sds key) { ...@@ -265,15 +267,15 @@ int getKeySlot(sds key) {
int dbAddRDBLoad(redisDb *db, sds key, robj *val) { int dbAddRDBLoad(redisDb *db, sds key, robj *val) {
int slot = getKeySlot(key); int slot = getKeySlot(key);
dict *d = db->dict[slot]; dict *d = db->dict[slot];
int was_empty = dictIsEmpty(d); int new_dict = server.cluster_enabled && dictIsEmpty(d);
dictEntry *de = dictAddRaw(d, key, NULL); dictEntry *de = dictAddRaw(d, key, NULL);
if (de == NULL) return 0; if (de == NULL) return 0;
dictSetVal(d, de, val); dictSetVal(d, de, val);
db->key_count++; db->key_count++;
/* If dict transitioned from empty to non-empty, we should add it to the list of owned slots. */ /* If dict transitioned from empty to non-empty, we should add it to the list of owned slots. */
if (was_empty) { if (new_dict) {
uint8_t success = 0; uint8_t success = 0;
db->owned_slots = intsetAdd(db->owned_slots, slot, &success); db->non_empty_dicts = intsetAdd(db->non_empty_dicts, slot, &success);
serverAssert(success); serverAssert(success);
} }
return 1; return 1;
...@@ -401,7 +403,7 @@ robj *dbRandomKey(redisDb *db) { ...@@ -401,7 +403,7 @@ robj *dbRandomKey(redisDb *db) {
/* Return random non-empty dictionary from this DB. */ /* Return random non-empty dictionary from this DB. */
dict *getRandomDict(redisDb *db) { dict *getRandomDict(redisDb *db) {
if (db->dict_count == 1 || dbSize(db) == 0) return db->dict[0]; if (db->dict_count == 1 || dbSize(db) == 0) return db->dict[0];
int64_t slot = intsetRandom(db->owned_slots); int64_t slot = intsetRandom(db->non_empty_dicts);
return db->dict[slot]; return db->dict[slot];
} }
...@@ -434,9 +436,9 @@ int dbGenericDelete(redisDb *db, robj *key, int async, int flags) { ...@@ -434,9 +436,9 @@ int dbGenericDelete(redisDb *db, robj *key, int async, int flags) {
if (dictSize(db->expires) > 0) dictDelete(db->expires,key->ptr); if (dictSize(db->expires) > 0) dictDelete(db->expires,key->ptr);
dictTwoPhaseUnlinkFree(d,de,plink,table); 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 we've removed last entry from the dict, then we need to also remove it from the list of owned non-empty slots. */
if (dictIsEmpty(d)) { if (server.cluster_enabled && dictIsEmpty(d)) {
int success = 0; int success = 0;
db->owned_slots = intsetRemove(db->owned_slots, slot, &success); db->non_empty_dicts = intsetRemove(db->non_empty_dicts, slot, &success);
serverAssert(success); serverAssert(success);
} }
db->key_count--; db->key_count--;
...@@ -534,8 +536,8 @@ long long emptyDbStructure(redisDb *dbarray, int dbnum, int async, ...@@ -534,8 +536,8 @@ long long emptyDbStructure(redisDb *dbarray, int dbnum, int async,
dbarray[j].avg_ttl = 0; dbarray[j].avg_ttl = 0;
dbarray[j].expires_cursor = 0; dbarray[j].expires_cursor = 0;
dbarray[j].key_count = 0; dbarray[j].key_count = 0;
zfree(dbarray[j].owned_slots); zfree(dbarray[j].non_empty_dicts);
dbarray[j].owned_slots = intsetNew(); dbarray[j].non_empty_dicts = intsetNew();
} }
return removed; return removed;
...@@ -603,7 +605,7 @@ redisDb *initTempDb(void) { ...@@ -603,7 +605,7 @@ redisDb *initTempDb(void) {
tempDb[i].dict_count = (server.cluster_enabled) ? CLUSTER_SLOTS : 1; tempDb[i].dict_count = (server.cluster_enabled) ? CLUSTER_SLOTS : 1;
tempDb[i].dict = dictCreateMultiple(&dbDictType, tempDb[i].dict_count); tempDb[i].dict = dictCreateMultiple(&dbDictType, tempDb[i].dict_count);
tempDb[i].expires = dictCreate(&dbExpiresDictType); tempDb[i].expires = dictCreate(&dbExpiresDictType);
tempDb[i].owned_slots = intsetNew(); tempDb[i].non_empty_dicts = intsetNew();
} }
return tempDb; return tempDb;
...@@ -621,7 +623,7 @@ void discardTempDb(redisDb *tempDb, void(callback)(dict*)) { ...@@ -621,7 +623,7 @@ void discardTempDb(redisDb *tempDb, void(callback)(dict*)) {
} }
zfree(tempDb[i].dict); zfree(tempDb[i].dict);
dictRelease(tempDb[i].expires); dictRelease(tempDb[i].expires);
zfree(tempDb[i].owned_slots); zfree(tempDb[i].non_empty_dicts);
} }
zfree(tempDb); zfree(tempDb);
...@@ -1568,7 +1570,7 @@ int dbSwapDatabases(int id1, int id2) { ...@@ -1568,7 +1570,7 @@ int dbSwapDatabases(int id1, int id2) {
db1->expires_cursor = db2->expires_cursor; db1->expires_cursor = db2->expires_cursor;
db1->dict_count = db2->dict_count; db1->dict_count = db2->dict_count;
db1->key_count = db2->key_count; db1->key_count = db2->key_count;
db1->owned_slots = db2->owned_slots; db1->non_empty_dicts = db2->non_empty_dicts;
db2->dict = aux.dict; db2->dict = aux.dict;
db2->expires = aux.expires; db2->expires = aux.expires;
...@@ -1576,7 +1578,7 @@ int dbSwapDatabases(int id1, int id2) { ...@@ -1576,7 +1578,7 @@ int dbSwapDatabases(int id1, int id2) {
db2->expires_cursor = aux.expires_cursor; db2->expires_cursor = aux.expires_cursor;
db2->dict_count = aux.dict_count; db2->dict_count = aux.dict_count;
db2->key_count = aux.key_count; db2->key_count = aux.key_count;
db2->owned_slots = aux.owned_slots; db2->non_empty_dicts = aux.non_empty_dicts;
/* Now we need to handle clients blocked on lists: as an effect /* Now we need to handle clients blocked on lists: as an effect
* of swapping the two DBs, a client that was waiting for list * of swapping the two DBs, a client that was waiting for list
...@@ -1616,7 +1618,7 @@ void swapMainDbWithTempDb(redisDb *tempDb) { ...@@ -1616,7 +1618,7 @@ void swapMainDbWithTempDb(redisDb *tempDb) {
activedb->expires_cursor = newdb->expires_cursor; activedb->expires_cursor = newdb->expires_cursor;
activedb->dict_count = newdb->dict_count; activedb->dict_count = newdb->dict_count;
activedb->key_count = newdb->key_count; activedb->key_count = newdb->key_count;
activedb->owned_slots = newdb->owned_slots; activedb->non_empty_dicts = newdb->non_empty_dicts;
newdb->dict = aux.dict; newdb->dict = aux.dict;
newdb->expires = aux.expires; newdb->expires = aux.expires;
...@@ -1624,7 +1626,7 @@ void swapMainDbWithTempDb(redisDb *tempDb) { ...@@ -1624,7 +1626,7 @@ void swapMainDbWithTempDb(redisDb *tempDb) {
newdb->expires_cursor = aux.expires_cursor; newdb->expires_cursor = aux.expires_cursor;
newdb->dict_count = aux.dict_count; newdb->dict_count = aux.dict_count;
newdb->key_count = aux.key_count; newdb->key_count = aux.key_count;
newdb->owned_slots = aux.owned_slots; newdb->non_empty_dicts = aux.non_empty_dicts;
/* Now we need to handle clients blocked on lists: as an effect /* Now we need to handle clients blocked on lists: as an effect
* of swapping the two DBs, a client that was waiting for list * of swapping the two DBs, a client that was waiting for list
......
...@@ -2594,7 +2594,7 @@ void initServer(void) { ...@@ -2594,7 +2594,7 @@ void initServer(void) {
server.db[j].rehashing = listCreate(); server.db[j].rehashing = listCreate();
server.db[j].dict_count = slotCount; server.db[j].dict_count = slotCount;
server.db[j].key_count = 0; server.db[j].key_count = 0;
server.db[j].owned_slots = intsetNew(); server.db[j].non_empty_dicts = intsetNew();
listSetFreeMethod(server.db[j].defrag_later,(void (*)(void*))sdsfree); listSetFreeMethod(server.db[j].defrag_later,(void (*)(void*))sdsfree);
} }
evictionPoolAlloc(); /* Initialize the LRU keys pool. */ evictionPoolAlloc(); /* Initialize the LRU keys pool. */
......
...@@ -961,7 +961,7 @@ typedef struct redisDb { ...@@ -961,7 +961,7 @@ typedef struct redisDb {
list *rehashing; /* List of dictionaries in this DB that are currently rehashing. */ 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. */ 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. */ unsigned long long key_count; /* Total number of keys in this DB. */
intset *owned_slots; /* Set of owned non-empty slots. */ intset *non_empty_dicts; /* Set of non-empty dictionaries. */
} redisDb; } redisDb;
/* forward declaration for functions ctx */ /* forward declaration for functions ctx */
......
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