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

Use intset for non-empty slots

parent ef1ca229
...@@ -50,38 +50,28 @@ int keyIsExpired(redisDb *db, robj *key); ...@@ -50,38 +50,28 @@ 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 *dbNextDict(dbIterator *dbit) { dict *dbNextDict(dbIterator *dbit) {
while (dbit->index < dbit->db->dict_count - 1) { int64_t slot;
dbit->index++; if (intsetGet(dbit->db->owned_slots, dbit->index++, &slot)){
dict *d = dbit->db->dict[dbit->index]; return dbit->db->dict[slot];
/* There is a single dictionary in non cluster mode,
* in cluster mode return first non-empty sub-dictionary. */
if (!server.cluster_enabled || dictSize(d) > 0) return d;
} }
return NULL; return NULL;
} }
void dbInitIteratorAt(dbIterator *dbit, redisDb *db, int slot) {
serverAssert(slot == 0 || server.cluster_enabled);
dbit->db = db;
dbit->index = slot - 1; /* Start one slot ahead, as dbNextDict increments index right away. */
}
/* Returns DB iterator that can be used to iterate through sub-dictionaries. /* Returns DB iterator that can be used to iterate through sub-dictionaries.
* Primary database contains only one dictionary when node runs without cluster mode, * Primary database contains only one dictionary when node runs without cluster mode,
* or 16k dictionaries (one per slot) when node runs with cluster mode enabled. */ * or 16k dictionaries (one per slot) when node runs with cluster mode enabled. */
void dbInitIterator(dbIterator *dbit, redisDb *db) { void dbInitIterator(dbIterator *dbit, redisDb *db) {
dbInitIteratorAt(dbit, db, 0); dbit->db = db;
dbit->index = 0;
} }
/* Returns next 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 *dbGetNextUnvisitedSlot(redisDb *db, int *slot) { dict *dbGetNextUnvisitedSlot(redisDb *db, int *slot) {
if (*slot < db->dict_count - 1) { for (int i = *slot + 1; i < db->dict_count; i++) {
dbIterator dbit; if (!dictIsEmpty(db->dict[i])) {
dbInitIteratorAt(&dbit, db, *slot + 1); /* Scan on the current slot has already returned 0, find next non-empty dict. */ *slot = i;
dict *dict = dbNextDict(&dbit); return db->dict[i];
if (dict != NULL) {
*slot = dbit.index;
return dict;
} }
} }
*slot = -1; *slot = -1;
...@@ -125,7 +115,7 @@ void updateLFU(robj *val) { ...@@ -125,7 +115,7 @@ void updateLFU(robj *val) {
* expired on replicas even if the master is lagging expiring our key via DELs * expired on replicas even if the master is lagging expiring our key via DELs
* in the replication link. */ * in the replication link. */
robj *lookupKey(redisDb *db, robj *key, int flags) { robj *lookupKey(redisDb *db, robj *key, int flags) {
dictEntry *de = dictFind(getDict(db, key->ptr),key->ptr); dictEntry *de = dictFind(db->dict[getKeySlot(key->ptr)], key->ptr);
robj *val = NULL; robj *val = NULL;
if (de) { if (de) {
val = dictGetVal(de); val = dictGetVal(de);
...@@ -230,21 +220,32 @@ robj *lookupKeyWriteOrReply(client *c, robj *key, robj *reply) { ...@@ -230,21 +220,32 @@ robj *lookupKeyWriteOrReply(client *c, robj *key, robj *reply) {
* The program is aborted if the key already exists. */ * The program is aborted if the key already exists. */
void dbAdd(redisDb *db, robj *key, robj *val) { void dbAdd(redisDb *db, robj *key, robj *val) {
sds copy = sdsdup(key->ptr); sds copy = sdsdup(key->ptr);
dict *d = getDict(db, key->ptr); int slot = getKeySlot(key->ptr);
dict *d = db->dict[slot];
int was_empty = 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 (was_empty) {
uint8_t success = 0;
db->owned_slots = intsetAdd(db->owned_slots, slot, &success);
serverAssert(success);
}
signalKeyAsReady(db, key, val->type); signalKeyAsReady(db, key, val->type);
notifyKeyspaceEvent(NOTIFY_NEW,"new",key,db->id); notifyKeyspaceEvent(NOTIFY_NEW,"new",key,db->id);
} }
/* Return slot-specific dictionary for key based on key's hash slot in CME, or 0 in CMD.*/ /* Return slot-specific dictionary for key based on key's hash slot in CME, or 0 in CMD.*/
dict *getDict(redisDb *db, sds key) { int getKeySlot(sds key) {
if (server.current_client && server.current_client->slot >= 0) { /* This is performance optimization, that uses pre-set slot id, in order to avoid calculating key hash. */ /* This is performance optimization, that uses pre-set slot id from the current command,
return db->dict[server.current_client->slot]; * in order to avoid calculation of the key hash. Code paths that are using keys, that can be from different slots,
* MUST unset current client's slot value before calling any db functions, otherwise wrong dictionary can be used. */
if (server.current_client && server.current_client->slot >= 0) {
return server.current_client->slot;
} }
return db->dict[(server.cluster_enabled ? keyHashSlot(key, (int) sdslen(key)) : 0)]; return server.cluster_enabled ? keyHashSlot(key, (int) sdslen(key)) : 0;
} }
/* This is a special version of dbAdd() that is used only when loading /* This is a special version of dbAdd() that is used only when loading
...@@ -259,11 +260,19 @@ dict *getDict(redisDb *db, sds key) { ...@@ -259,11 +260,19 @@ dict *getDict(redisDb *db, sds key) {
* ownership of the SDS string, otherwise 0 is returned, and is up to the * ownership of the SDS string, otherwise 0 is returned, and is up to the
* caller to free the SDS string. */ * caller to free the SDS string. */
int dbAddRDBLoad(redisDb *db, sds key, robj *val) { int dbAddRDBLoad(redisDb *db, sds key, robj *val) {
dict *d = getDict(db, key); int slot = getKeySlot(key);
dict *d = db->dict[slot];
int was_empty = 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 (was_empty) {
uint8_t success = 0;
db->owned_slots = intsetAdd(db->owned_slots, slot, &success);
serverAssert(success);
}
return 1; return 1;
} }
...@@ -278,7 +287,7 @@ int dbAddRDBLoad(redisDb *db, sds key, robj *val) { ...@@ -278,7 +287,7 @@ int dbAddRDBLoad(redisDb *db, sds key, robj *val) {
* *
* The program is aborted if the key was not already present. */ * The program is aborted if the key was not already present. */
static void dbSetValue(redisDb *db, robj *key, robj *val, int overwrite) { static void dbSetValue(redisDb *db, robj *key, robj *val, int overwrite) {
dict *d = getDict(db, key->ptr); dict *d = db->dict[getKeySlot(key->ptr)];
dictEntry *de = dictFind(d, key->ptr); dictEntry *de = dictFind(d, key->ptr);
serverAssertWithInfo(NULL,key,de != NULL); serverAssertWithInfo(NULL,key,de != NULL);
...@@ -354,7 +363,7 @@ void setKey(client *c, redisDb *db, robj *key, robj *val, int flags) { ...@@ -354,7 +363,7 @@ void setKey(client *c, redisDb *db, robj *key, robj *val, int flags) {
robj *dbRandomKey(redisDb *db) { robj *dbRandomKey(redisDb *db) {
dictEntry *de; dictEntry *de;
int maxtries = 100; int maxtries = 100;
dict *randomDict = getFairRandomDict(db); dict *randomDict = getRandomDict(db);
while(1) { while(1) {
sds key; sds key;
...@@ -387,50 +396,18 @@ robj *dbRandomKey(redisDb *db) { ...@@ -387,50 +396,18 @@ robj *dbRandomKey(redisDb *db) {
} }
/* Return random non-empty dictionary from this DB. */ /* Return random non-empty dictionary from this DB. */
dict *getFairRandomDict(redisDb *db) {
if (db->dict_count == 1) return db->dict[0];
unsigned long target = randomULong();
unsigned long long int key_count = dbSize(db);
if (!key_count) return db->dict[0];
target %= key_count; /* Random key index in range [0..KEY_COUNT). */
/* In linearly enumerated key space, find a slot that contains target key. */
dbIterator dbit;
dbInitIterator(&dbit, db);
dict *d = NULL;
while ((d = dbNextDict(&dbit))) {
unsigned long long int ds = dictSize(d);
if (target < ds) { /* Found dict that contains target key. */
return d;
}
target -= ds;
}
serverPanic("Bug in random dict selection, iteration completed without finding a slot for the target element.");
}
/* Return random non-empty dictionary from this DB by probing random slots.
* This function can be worse than getFairRandomDict in two cases:
* - Dictionary is almost empty, which could result in too much probing.
* - Slot sizes are significantly imbalanced, which could result in unfairness.
* First case is resolved by falling back to getFairRandomDict after certain number of probing attempts.
* Second issue is ignored, meaning that all slots have same probability of being selected. */
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);
for (int i = 0; i < MAX_RANDOM_DICT_PROBE_ATTEMTPS; i++) { return db->dict[slot];
int candidate = rand() % db->dict_count;
if (dictSize(db->dict[candidate]) > 0) {
return db->dict[candidate];
}
}
/* If we can't find non-empty dict by probing a few random slots, then we'll fall back to slower iterative approach. */
return getFairRandomDict(db);
} }
/* Helper for sync and async delete. */ /* Helper for sync and async delete. */
int dbGenericDelete(redisDb *db, robj *key, int async, int flags) { int dbGenericDelete(redisDb *db, robj *key, int async, int flags) {
dictEntry **plink; dictEntry **plink;
int table; int table;
dict *d = getDict(db, key->ptr); int slot = getKeySlot(key->ptr);
dict *d = db->dict[slot];
dictEntry *de = dictTwoPhaseUnlinkFind(d,key->ptr,&plink,&table); dictEntry *de = dictTwoPhaseUnlinkFind(d,key->ptr,&plink,&table);
if (de) { if (de) {
robj *val = dictGetVal(de); robj *val = dictGetVal(de);
...@@ -453,6 +430,12 @@ int dbGenericDelete(redisDb *db, robj *key, int async, int flags) { ...@@ -453,6 +430,12 @@ int dbGenericDelete(redisDb *db, robj *key, int async, int flags) {
* the key, because it is shared with the main dictionary. */ * the key, because it is shared with the main dictionary. */
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 (dictIsEmpty(d)) {
int success = 0;
db->owned_slots = intsetRemove(db->owned_slots, slot, &success);
serverAssert(success);
}
db->key_count--; db->key_count--;
return 1; return 1;
} else { } else {
...@@ -548,6 +531,8 @@ long long emptyDbStructure(redisDb *dbarray, int dbnum, int async, ...@@ -548,6 +531,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);
dbarray[j].owned_slots = intsetNew();
} }
return removed; return removed;
...@@ -615,6 +600,7 @@ redisDb *initTempDb(void) { ...@@ -615,6 +600,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();
} }
return tempDb; return tempDb;
...@@ -632,6 +618,7 @@ void discardTempDb(redisDb *tempDb, void(callback)(dict*)) { ...@@ -632,6 +618,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); zfree(tempDb);
...@@ -1148,10 +1135,9 @@ cleanup: ...@@ -1148,10 +1135,9 @@ cleanup:
} }
void addSlotIdToCursor(int slot, unsigned long long int *cursor) { void addSlotIdToCursor(int slot, unsigned long long int *cursor) {
/* Slot id can be -1 if there are no more slots to visit. */ /* Slot id can be -1 when iteration is over and there are no more slots to visit. */
if (slot >= 0) { if (slot < 0) return;
*cursor = (*cursor << CLUSTER_SLOT_MASK_BITS) | slot; *cursor = (*cursor << CLUSTER_SLOT_MASK_BITS) | slot;
}
} }
int getAndClearSlotIdFromCursor(unsigned long long int *cursor) { int getAndClearSlotIdFromCursor(unsigned long long int *cursor) {
...@@ -1504,7 +1490,7 @@ void scanDatabaseForReadyKeys(redisDb *db) { ...@@ -1504,7 +1490,7 @@ void scanDatabaseForReadyKeys(redisDb *db) {
dictIterator *di = dictGetSafeIterator(db->blocking_keys); dictIterator *di = dictGetSafeIterator(db->blocking_keys);
while((de = dictNext(di)) != NULL) { while((de = dictNext(di)) != NULL) {
robj *key = dictGetKey(de); robj *key = dictGetKey(de);
dictEntry *kde = dictFind(getDict(db, key->ptr), key->ptr); dictEntry *kde = dictFind(db->dict[getKeySlot(key->ptr)], key->ptr);
if (kde) { if (kde) {
robj *value = dictGetVal(kde); robj *value = dictGetVal(kde);
signalKeyAsReady(db, key, value->type); signalKeyAsReady(db, key, value->type);
...@@ -1524,7 +1510,7 @@ void scanDatabaseForDeletedKeys(redisDb *emptied, redisDb *replaced_with) { ...@@ -1524,7 +1510,7 @@ void scanDatabaseForDeletedKeys(redisDb *emptied, redisDb *replaced_with) {
int existed = 0, exists = 0; int existed = 0, exists = 0;
int original_type = -1, curr_type = -1; int original_type = -1, curr_type = -1;
dictEntry *kde = dictFind(getDict(emptied, key->ptr), key->ptr); dictEntry *kde = dictFind(emptied->dict[getKeySlot(key->ptr)], key->ptr);
if (kde) { if (kde) {
robj *value = dictGetVal(kde); robj *value = dictGetVal(kde);
original_type = value->type; original_type = value->type;
...@@ -1532,7 +1518,7 @@ void scanDatabaseForDeletedKeys(redisDb *emptied, redisDb *replaced_with) { ...@@ -1532,7 +1518,7 @@ void scanDatabaseForDeletedKeys(redisDb *emptied, redisDb *replaced_with) {
} }
if (replaced_with) { if (replaced_with) {
kde = dictFind(getDict(replaced_with, key->ptr), key->ptr); kde = dictFind(replaced_with->dict[getKeySlot(key->ptr)], key->ptr);
if (kde) { if (kde) {
robj *value = dictGetVal(kde); robj *value = dictGetVal(kde);
curr_type = value->type; curr_type = value->type;
...@@ -1579,6 +1565,7 @@ int dbSwapDatabases(int id1, int id2) { ...@@ -1579,6 +1565,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;
db2->dict = aux.dict; db2->dict = aux.dict;
db2->expires = aux.expires; db2->expires = aux.expires;
...@@ -1586,6 +1573,7 @@ int dbSwapDatabases(int id1, int id2) { ...@@ -1586,6 +1573,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;
/* 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
...@@ -1625,6 +1613,7 @@ void swapMainDbWithTempDb(redisDb *tempDb) { ...@@ -1625,6 +1613,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;
newdb->dict = aux.dict; newdb->dict = aux.dict;
newdb->expires = aux.expires; newdb->expires = aux.expires;
...@@ -1632,6 +1621,7 @@ void swapMainDbWithTempDb(redisDb *tempDb) { ...@@ -1632,6 +1621,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;
/* 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
...@@ -1687,7 +1677,7 @@ void swapdbCommand(client *c) { ...@@ -1687,7 +1677,7 @@ void swapdbCommand(client *c) {
int removeExpire(redisDb *db, robj *key) { int removeExpire(redisDb *db, robj *key) {
/* An expire may only be removed if there is a corresponding entry in the /* An expire may only be removed if there is a corresponding entry in the
* main dict. Otherwise, the key will never be freed. */ * main dict. Otherwise, the key will never be freed. */
serverAssertWithInfo(NULL,key,dictFind(getDict(db, key->ptr),key->ptr) != NULL); serverAssertWithInfo(NULL,key, dictFind(db->dict[getKeySlot(key->ptr)], key->ptr) != NULL);
return dictDelete(db->expires,key->ptr) == DICT_OK; return dictDelete(db->expires,key->ptr) == DICT_OK;
} }
...@@ -1699,7 +1689,7 @@ void setExpire(client *c, redisDb *db, robj *key, long long when) { ...@@ -1699,7 +1689,7 @@ void setExpire(client *c, redisDb *db, robj *key, long long when) {
dictEntry *kde, *de; dictEntry *kde, *de;
/* Reuse the sds from the main dict in the expire dict */ /* Reuse the sds from the main dict in the expire dict */
kde = dictFind(getDict(db, key->ptr),key->ptr); kde = dictFind(db->dict[getKeySlot(key->ptr)], key->ptr);
serverAssertWithInfo(NULL,key,kde != NULL); serverAssertWithInfo(NULL,key,kde != NULL);
de = dictAddOrFind(db->expires,dictGetKey(kde)); de = dictAddOrFind(db->expires,dictGetKey(kde));
dictSetSignedIntegerVal(de,when); dictSetSignedIntegerVal(de,when);
...@@ -1720,7 +1710,7 @@ long long getExpire(redisDb *db, robj *key) { ...@@ -1720,7 +1710,7 @@ long long getExpire(redisDb *db, robj *key) {
/* The entry was found in the expire dict, this means it should also /* The entry was found in the expire dict, this means it should also
* be present in the main dict (safety check). */ * be present in the main dict (safety check). */
serverAssertWithInfo(NULL,key,dictFind(getDict(db, key->ptr),key->ptr) != NULL); serverAssertWithInfo(NULL,key, dictFind(db->dict[getKeySlot(key->ptr)], key->ptr) != NULL);
return dictGetSignedIntegerVal(de); return dictGetSignedIntegerVal(de);
} }
......
...@@ -611,7 +611,7 @@ NULL ...@@ -611,7 +611,7 @@ NULL
robj *val; robj *val;
char *strenc; char *strenc;
if ((de = dictFind(getDict(c->db, c->argv[2]->ptr),c->argv[2]->ptr)) == NULL) { if ((de = dictFind(c->db->dict[getKeySlot(c->argv[2]->ptr)], c->argv[2]->ptr)) == NULL) {
addReplyErrorObject(c,shared.nokeyerr); addReplyErrorObject(c,shared.nokeyerr);
return; return;
} }
...@@ -663,7 +663,7 @@ NULL ...@@ -663,7 +663,7 @@ NULL
robj *val; robj *val;
sds key; sds key;
if ((de = dictFind(getDict(c->db, c->argv[2]->ptr), c->argv[2]->ptr)) == NULL) { if ((de = dictFind(c->db->dict[getKeySlot(c->argv[2]->ptr)], c->argv[2]->ptr)) == NULL) {
addReplyErrorObject(c,shared.nokeyerr); addReplyErrorObject(c,shared.nokeyerr);
return; return;
} }
...@@ -764,7 +764,7 @@ NULL ...@@ -764,7 +764,7 @@ NULL
/* We don't use lookupKey because a debug command should /* We don't use lookupKey because a debug command should
* work on logically expired keys */ * work on logically expired keys */
dictEntry *de; dictEntry *de;
robj *o = ((de = dictFind(getDict(c->db, c->argv[j]->ptr),c->argv[j]->ptr)) == NULL) ? NULL : dictGetVal(de); robj *o = ((de = dictFind(c->db->dict[getKeySlot(c->argv[j]->ptr)], c->argv[j]->ptr)) == NULL) ? NULL : dictGetVal(de);
if (o) xorObjectDigest(c->db,c->argv[j],digest,o); if (o) xorObjectDigest(c->db,c->argv[j],digest,o);
sds d = sdsempty(); sds d = sdsempty();
...@@ -1878,7 +1878,7 @@ void logCurrentClient(client *cc, const char *title) { ...@@ -1878,7 +1878,7 @@ void logCurrentClient(client *cc, const char *title) {
dictEntry *de; dictEntry *de;
key = getDecodedObject(cc->argv[1]); key = getDecodedObject(cc->argv[1]);
de = dictFind(getDict(cc->db, key->ptr), key->ptr); de = dictFind(cc->db->dict[getKeySlot(key->ptr)], key->ptr);
if (de) { if (de) {
val = dictGetVal(de); val = dictGetVal(de);
serverLog(LL_WARNING,"key '%s' found in DB containing the following object:", (char*)key->ptr); serverLog(LL_WARNING,"key '%s' found in DB containing the following object:", (char*)key->ptr);
......
...@@ -679,12 +679,12 @@ void defragKey(redisDb *db, dictEntry *de) { ...@@ -679,12 +679,12 @@ void defragKey(redisDb *db, dictEntry *de) {
/* Try to defrag the key name. */ /* Try to defrag the key name. */
newsds = activeDefragSds(keysds); newsds = activeDefragSds(keysds);
if (newsds) { if (newsds) {
dictSetKey(getDict(db, newsds), de, newsds); dictSetKey(db->dict[getKeySlot(newsds)], de, newsds);
if (dictSize(db->expires)) { if (dictSize(db->expires)) {
/* We can't search in db->expires for that key after we've released /* 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 * 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. */ * compare, but we can find the entry using key hash and pointer. */
uint64_t hash = dictGetHash(getDict(db, newsds), newsds); uint64_t hash = dictGetHash(db->dict[getKeySlot(newsds)], newsds);
dictEntry *expire_de = dictFindEntryByPtrAndHash(db->expires, keysds, hash); dictEntry *expire_de = dictFindEntryByPtrAndHash(db->expires, keysds, hash);
if (expire_de) dictSetKey(db->expires, expire_de, newsds); if (expire_de) dictSetKey(db->expires, expire_de, newsds);
} }
...@@ -693,7 +693,7 @@ void defragKey(redisDb *db, dictEntry *de) { ...@@ -693,7 +693,7 @@ void defragKey(redisDb *db, dictEntry *de) {
/* Try to defrag robj and / or string value. */ /* Try to defrag robj and / or string value. */
ob = dictGetVal(de); ob = dictGetVal(de);
if ((newob = activeDefragStringOb(ob))) { if ((newob = activeDefragStringOb(ob))) {
dictSetVal(getDict(db, newsds), de, newob); dictSetVal(db->dict[getKeySlot(newsds)], de, newob);
ob = newob; ob = newob;
} }
...@@ -851,7 +851,7 @@ int defragLaterStep(redisDb *db, long long endtime) { ...@@ -851,7 +851,7 @@ int defragLaterStep(redisDb *db, long long endtime) {
} }
/* each time we enter this function we need to fetch the key from the dict again (if it still exists) */ /* each time we enter this function we need to fetch the key from the dict again (if it still exists) */
dictEntry *de = dictFind(getDict(db, defrag_later_current_key), defrag_later_current_key); dictEntry *de = dictFind(db->dict[getKeySlot(defrag_later_current_key)], defrag_later_current_key);
key_defragged = server.stat_active_defrag_hits; key_defragged = server.stat_active_defrag_hits;
do { do {
int quit = 0; int quit = 0;
......
...@@ -139,6 +139,7 @@ typedef struct { ...@@ -139,6 +139,7 @@ typedef struct {
#define dictHashKey(d, key) ((d)->type->hashFunction(key)) #define dictHashKey(d, key) ((d)->type->hashFunction(key))
#define dictSlots(d) (DICTHT_SIZE((d)->ht_size_exp[0])+DICTHT_SIZE((d)->ht_size_exp[1])) #define dictSlots(d) (DICTHT_SIZE((d)->ht_size_exp[0])+DICTHT_SIZE((d)->ht_size_exp[1]))
#define dictSize(d) ((d)->ht_used[0]+(d)->ht_used[1]) #define dictSize(d) ((d)->ht_used[0]+(d)->ht_used[1])
#define dictIsEmpty(d) ((d)->ht_used[0] == 0 && (d)->ht_used[1] == 0)
#define dictIsRehashing(d) ((d)->rehashidx != -1) #define dictIsRehashing(d) ((d)->rehashidx != -1)
#define dictPauseRehashing(d) ((d)->pauserehash++) #define dictPauseRehashing(d) ((d)->pauserehash++)
#define dictResumeRehashing(d) ((d)->pauserehash--) #define dictResumeRehashing(d) ((d)->pauserehash--)
......
...@@ -161,7 +161,7 @@ void evictionPoolPopulate(int dbid, dict *sampledict, redisDb *db, struct evicti ...@@ -161,7 +161,7 @@ void evictionPoolPopulate(int dbid, dict *sampledict, redisDb *db, struct evicti
* dictionary (but the expires one) we need to lookup the key * dictionary (but the expires one) we need to lookup the key
* again in the key dictionary to obtain the value object. */ * again in the key dictionary to obtain the value object. */
if (server.maxmemory_policy != MAXMEMORY_VOLATILE_TTL) { if (server.maxmemory_policy != MAXMEMORY_VOLATILE_TTL) {
if (!(server.maxmemory_policy & MAXMEMORY_FLAG_ALLKEYS)) de = dictFind(getDict(db, key), key); if (!(server.maxmemory_policy & MAXMEMORY_FLAG_ALLKEYS)) de = dictFind(db->dict[getKeySlot(key)], key);
o = dictGetVal(de); o = dictGetVal(de);
} }
...@@ -619,8 +619,8 @@ int performEvictions(void) { ...@@ -619,8 +619,8 @@ int performEvictions(void) {
bestdbid = pool[k].dbid; bestdbid = pool[k].dbid;
if (server.maxmemory_policy & MAXMEMORY_FLAG_ALLKEYS) { if (server.maxmemory_policy & MAXMEMORY_FLAG_ALLKEYS) {
de = dictFind(getDict(&server.db[bestdbid], pool[k].key), de = dictFind(server.db[bestdbid].dict[getKeySlot(pool[k].key)],
pool[k].key); pool[k].key);
} else { } else {
de = dictFind(server.db[bestdbid].expires, de = dictFind(server.db[bestdbid].expires,
pool[k].key); pool[k].key);
......
...@@ -394,7 +394,7 @@ void touchWatchedKey(redisDb *db, robj *key) { ...@@ -394,7 +394,7 @@ void touchWatchedKey(redisDb *db, robj *key) {
/* The key was already expired when WATCH was called. */ /* The key was already expired when WATCH was called. */
if (db == wk->db && if (db == wk->db &&
equalStringObjects(key, wk->key) && equalStringObjects(key, wk->key) &&
dictFind(getDict(db, key->ptr), key->ptr) == NULL) dictFind(db->dict[getKeySlot(key->ptr)], key->ptr) == NULL)
{ {
/* Already expired key is deleted, so logically no change. Clear /* Already expired key is deleted, so logically no change. Clear
* the flag. Deleted keys are not flagged as expired. */ * the flag. Deleted keys are not flagged as expired. */
...@@ -432,9 +432,9 @@ void touchAllWatchedKeysInDb(redisDb *emptied, redisDb *replaced_with) { ...@@ -432,9 +432,9 @@ void touchAllWatchedKeysInDb(redisDb *emptied, redisDb *replaced_with) {
dictIterator *di = dictGetSafeIterator(emptied->watched_keys); dictIterator *di = dictGetSafeIterator(emptied->watched_keys);
while((de = dictNext(di)) != NULL) { while((de = dictNext(di)) != NULL) {
robj *key = dictGetKey(de); robj *key = dictGetKey(de);
int exists_in_emptied = dictFind(getDict(emptied, key->ptr), key->ptr) != NULL; int exists_in_emptied = dictFind(emptied->dict[getKeySlot(key->ptr)], key->ptr) != NULL;
if (exists_in_emptied || if (exists_in_emptied ||
(replaced_with && dictFind(getDict(replaced_with, key->ptr), key->ptr))) (replaced_with && dictFind(replaced_with->dict[getKeySlot(key->ptr)], key->ptr)))
{ {
list *clients = dictGetVal(de); list *clients = dictGetVal(de);
if (!clients) continue; if (!clients) continue;
...@@ -442,7 +442,7 @@ void touchAllWatchedKeysInDb(redisDb *emptied, redisDb *replaced_with) { ...@@ -442,7 +442,7 @@ void touchAllWatchedKeysInDb(redisDb *emptied, redisDb *replaced_with) {
while((ln = listNext(&li))) { while((ln = listNext(&li))) {
watchedKey *wk = redis_member2struct(watchedKey, node, ln); watchedKey *wk = redis_member2struct(watchedKey, node, ln);
if (wk->expired) { if (wk->expired) {
if (!replaced_with || !dictFind(getDict(replaced_with, key->ptr), key->ptr)) { if (!replaced_with || !dictFind(replaced_with->dict[getKeySlot(key->ptr)], key->ptr)) {
/* Expired key now deleted. No logical change. Clear the /* Expired key now deleted. No logical change. Clear the
* flag. Deleted keys are not flagged as expired. */ * flag. Deleted keys are not flagged as expired. */
wk->expired = 0; wk->expired = 0;
......
...@@ -1536,7 +1536,7 @@ NULL ...@@ -1536,7 +1536,7 @@ NULL
return; return;
} }
} }
if ((de = dictFind(getDict(c->db, c->argv[2]->ptr),c->argv[2]->ptr)) == NULL) { if ((de = dictFind(c->db->dict[getKeySlot(c->argv[2]->ptr)], c->argv[2]->ptr)) == NULL) {
addReplyNull(c); addReplyNull(c);
return; return;
} }
......
...@@ -2594,6 +2594,7 @@ void initServer(void) { ...@@ -2594,6 +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();
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. */
......
...@@ -90,6 +90,7 @@ typedef struct redisObject robj; ...@@ -90,6 +90,7 @@ typedef struct redisObject robj;
#include "sha1.h" #include "sha1.h"
#include "endianconv.h" #include "endianconv.h"
#include "crc64.h" #include "crc64.h"
#include "intset.h"
/* helpers */ /* helpers */
#define numElements(x) (sizeof(x)/sizeof((x)[0])) #define numElements(x) (sizeof(x)/sizeof((x)[0]))
...@@ -138,7 +139,6 @@ typedef struct redisObject robj; ...@@ -138,7 +139,6 @@ typedef struct redisObject robj;
#define CONFIG_DEFAULT_PROC_TITLE_TEMPLATE "{title} {listen-addr} {server-mode}" #define CONFIG_DEFAULT_PROC_TITLE_TEMPLATE "{title} {listen-addr} {server-mode}"
#define INCREMENTAL_REHASHING_MAX_QUEUE_SIZE (1024*16) #define INCREMENTAL_REHASHING_MAX_QUEUE_SIZE (1024*16)
#define INCREMENTAL_REHASHING_THRESHOLD_MS 1 #define INCREMENTAL_REHASHING_THRESHOLD_MS 1
#define MAX_RANDOM_DICT_PROBE_ATTEMTPS 10
/* Bucket sizes for client eviction pools. Each bucket stores clients with /* Bucket sizes for client eviction pools. Each bucket stores clients with
* memory usage of up to twice the size of the bucket below it. */ * memory usage of up to twice the size of the bucket below it. */
...@@ -961,6 +961,7 @@ typedef struct redisDb { ...@@ -961,6 +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. */
} redisDb; } redisDb;
/* forward declaration for functions ctx */ /* forward declaration for functions ctx */
...@@ -3024,8 +3025,7 @@ void dismissMemoryInChild(void); ...@@ -3024,8 +3025,7 @@ void dismissMemoryInChild(void);
#define RESTART_SERVER_CONFIG_REWRITE (1<<1) /* CONFIG REWRITE before restart.*/ #define RESTART_SERVER_CONFIG_REWRITE (1<<1) /* CONFIG REWRITE before restart.*/
int restartServer(int flags, mstime_t delay); int restartServer(int flags, mstime_t delay);
unsigned long long int dbSize(redisDb *db); unsigned long long int dbSize(redisDb *db);
dict *getDict(redisDb *db, sds key); int getKeySlot(sds key);
dict *getFairRandomDict(redisDb *db);
dict *getRandomDict(redisDb *db); dict *getRandomDict(redisDb *db);
unsigned long dbSlots(redisDb *db); unsigned long dbSlots(redisDb *db);
void expandDb(const redisDb *db, uint64_t db_size); void expandDb(const redisDb *db, uint64_t db_size);
......
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