Commit 0b9004e3 authored by Vitaly Arbuzov's avatar Vitaly Arbuzov
Browse files

Improve incremental rehashing by using a linked list instead of linear scan...

Improve incremental rehashing by using a linked list instead of linear scan when finding rehash target
parent d6c61d71
...@@ -7321,9 +7321,6 @@ int clusterRedirectBlockedClientIfNeeded(client *c) { ...@@ -7321,9 +7321,6 @@ int clusterRedirectBlockedClientIfNeeded(client *c) {
unsigned int delKeysInSlot(unsigned int slot) { unsigned int delKeysInSlot(unsigned int slot) {
unsigned int j = 0; unsigned int j = 0;
server.core_propagates = 1;
server.in_nested_call++;
dictIterator *iter = NULL; dictIterator *iter = NULL;
dictEntry *de = NULL; dictEntry *de = NULL;
iter = dictGetSafeIterator(server.db->dict[slot]); iter = dictGetSafeIterator(server.db->dict[slot]);
...@@ -7340,7 +7337,6 @@ unsigned int delKeysInSlot(unsigned int slot) { ...@@ -7340,7 +7337,6 @@ unsigned int delKeysInSlot(unsigned int slot) {
server.dirty++; server.dirty++;
} }
dictReleaseIterator(iter); dictReleaseIterator(iter);
serverAssert(server.core_propagates); /* This function should not be re-entrant */
return j; return j;
} }
......
...@@ -309,7 +309,7 @@ void setKey(client *c, redisDb *db, robj *key, robj *val, int flags) { ...@@ -309,7 +309,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 = getRandomDict(db, 0); dict *randomDict = getRandomDict(db);
while(1) { while(1) {
sds key; sds key;
...@@ -342,13 +342,13 @@ robj *dbRandomKey(redisDb *db) { ...@@ -342,13 +342,13 @@ robj *dbRandomKey(redisDb *db) {
} }
/* Return random non-empty dictionary from this DB, if shouldBeRehashing is set, then it ignores dicts that aren't rehashing. */ /* Return random non-empty dictionary from this DB, if shouldBeRehashing is set, then it ignores dicts that aren't rehashing. */
dict *getRandomDict(redisDb *db, int shouldBeRehashing) { dict *getRandomDict(redisDb *db) {
if (db->dict_count == 1) return db->dict[0]; if (db->dict_count == 1) return db->dict[0];
int i = 0, r = 0; int i = 0, r = 0;
for (int j = 0; j < CLUSTER_SLOTS; j++) { for (int j = 0; j < CLUSTER_SLOTS; j++) {
// Skip empty dicts or if we want only rehashing dicts and the dict isn't rehashing. // Skip empty dicts or if we want only rehashing dicts and the dict isn't rehashing.
if (dictSize(db->dict[j]) == 0 || (shouldBeRehashing && !dictIsRehashing(db->dict[j]))) continue; if (dictSize(db->dict[j]) == 0) continue;
if (i == 0 || (rand() % (i + 1)) == 0) { if (i == 0 || (rand() % (i + 1)) == 0) {
r = j; // Select K-th non-empty bucket with 1/K probability, this keeps balanced probabilities for all non-empty buckets. r = j; // Select K-th non-empty bucket with 1/K probability, this keeps balanced probabilities for all non-empty buckets.
} }
......
...@@ -291,6 +291,7 @@ int _dictExpand(dict *d, unsigned long size, int* malloc_failed) ...@@ -291,6 +291,7 @@ int _dictExpand(dict *d, unsigned long size, int* malloc_failed)
d->ht_used[1] = new_ht_used; d->ht_used[1] = new_ht_used;
d->ht_table[1] = new_ht_table; d->ht_table[1] = new_ht_table;
d->rehashidx = 0; d->rehashidx = 0;
if (d->type->rehashingStarted) d->type->rehashingStarted(d);
return DICT_OK; return DICT_OK;
} }
...@@ -410,15 +411,16 @@ long long timeInMilliseconds(void) { ...@@ -410,15 +411,16 @@ long long timeInMilliseconds(void) {
/* Rehash in ms+"delta" milliseconds. The value of "delta" is larger /* Rehash in ms+"delta" milliseconds. The value of "delta" is larger
* than 0, and is smaller than 1 in most cases. The exact upper bound * than 0, and is smaller than 1 in most cases. The exact upper bound
* depends on the running time of dictRehash(d,100).*/ * depends on the running time of dictRehash(d,100).*/
int dictRehashMilliseconds(dict *d, int ms) { int dictRehashMilliseconds(dict *d, unsigned int ms) {
if (d->pauserehash > 0) return 0; if (d->pauserehash > 0) return 0;
long long start = timeInMilliseconds(); monotime timer;
elapsedStart(&timer);
int rehashes = 0; int rehashes = 0;
while(dictRehash(d,100)) { while(dictRehash(d,100)) {
rehashes += 100; rehashes += 100;
if (timeInMilliseconds()-start > ms) break; if (elapsedMs(timer) >= ms) break;
} }
return rehashes; return rehashes;
} }
......
...@@ -65,6 +65,7 @@ typedef struct dictType { ...@@ -65,6 +65,7 @@ typedef struct dictType {
void (*keyDestructor)(dict *d, void *key); void (*keyDestructor)(dict *d, void *key);
void (*valDestructor)(dict *d, void *obj); void (*valDestructor)(dict *d, void *obj);
int (*expandAllowed)(size_t moreMem, double usedRatio); int (*expandAllowed)(size_t moreMem, double usedRatio);
void (*rehashingStarted)(dict *d);
/* Flags */ /* Flags */
/* The 'no_value' flag, if set, indicates that values are not used, i.e. the /* The 'no_value' flag, if set, indicates that values are not used, i.e. the
* dict is a set. When this flag is set, it's not possible to access the * dict is a set. When this flag is set, it's not possible to access the
...@@ -213,7 +214,7 @@ uint64_t dictGenCaseHashFunction(const unsigned char *buf, size_t len); ...@@ -213,7 +214,7 @@ uint64_t dictGenCaseHashFunction(const unsigned char *buf, size_t len);
void dictEmpty(dict *d, void(callback)(dict*)); void dictEmpty(dict *d, void(callback)(dict*));
void dictSetResizeEnabled(dictResizeEnable enable); void dictSetResizeEnabled(dictResizeEnable enable);
int dictRehash(dict *d, int n); int dictRehash(dict *d, int n);
int dictRehashMilliseconds(dict *d, int ms); int dictRehashMilliseconds(dict *d, unsigned int ms);
void dictSetHashFunctionSeed(uint8_t *seed); void dictSetHashFunctionSeed(uint8_t *seed);
uint8_t *dictGetHashFunctionSeed(void); uint8_t *dictGetHashFunctionSeed(void);
unsigned long dictScan(dict *d, unsigned long v, dictScanFunction *fn, void *privdata); unsigned long dictScan(dict *d, unsigned long v, dictScanFunction *fn, void *privdata);
......
...@@ -593,7 +593,7 @@ int performEvictions(void) { ...@@ -593,7 +593,7 @@ int performEvictions(void) {
for (i = 0; i < server.dbnum; i++) { for (i = 0; i < server.dbnum; i++) {
db = server.db+i; db = server.db+i;
dict = (server.maxmemory_policy & MAXMEMORY_FLAG_ALLKEYS) ? dict = (server.maxmemory_policy & MAXMEMORY_FLAG_ALLKEYS) ?
getRandomDict(db, 0) : db->expires; getRandomDict(db) : db->expires;
if ((keys = dictSize(dict)) != 0) { if ((keys = dictSize(dict)) != 0) {
evictionPoolPopulate(i, dict, db, pool); evictionPoolPopulate(i, dict, db, pool);
total_keys += keys; total_keys += keys;
...@@ -643,7 +643,7 @@ int performEvictions(void) { ...@@ -643,7 +643,7 @@ int performEvictions(void) {
j = (++next_db) % server.dbnum; j = (++next_db) % server.dbnum;
db = server.db+j; db = server.db+j;
dict = (server.maxmemory_policy == MAXMEMORY_ALLKEYS_RANDOM) ? dict = (server.maxmemory_policy == MAXMEMORY_ALLKEYS_RANDOM) ?
getRandomDict(db, 0) : db->expires; getRandomDict(db) : db->expires;
if (dictSize(dict) != 0) { if (dictSize(dict) != 0) {
de = dictGetRandomKey(dict); de = dictGetRandomKey(dict);
bestkey = dictGetKey(de); bestkey = dictGetKey(de);
......
...@@ -392,6 +392,16 @@ int dictExpandAllowed(size_t moreMem, double usedRatio) { ...@@ -392,6 +392,16 @@ int dictExpandAllowed(size_t moreMem, double usedRatio) {
} }
} }
/* Adds dictionary to the rehashing list in cluster mode, which allows us
* to quickly find rehash targets during incremental rehashing.
* In non-cluster mode, we don't need this list as there is only one dictionary per DB. */
void dictRehashingStarted(dict *d) {
if (!server.cluster_enabled || !server.activerehashing) return;
/* Safety check against queue overflow. */
if (listLength(server.db[0].rehashing) > INCREMENTAL_REHASHING_MAX_QUEUE_SIZE) return;
listAddNodeTail(server.db[0].rehashing, d);
}
/* Generic hash table type where keys are Redis Objects, Values /* Generic hash table type where keys are Redis Objects, Values
* dummy pointers. */ * dummy pointers. */
dictType objectKeyPointerValueDictType = { dictType objectKeyPointerValueDictType = {
...@@ -447,6 +457,7 @@ dictType dbDictType = { ...@@ -447,6 +457,7 @@ dictType dbDictType = {
dictSdsDestructor, /* key destructor */ dictSdsDestructor, /* key destructor */
dictObjectDestructor, /* val destructor */ dictObjectDestructor, /* val destructor */
dictExpandAllowed, /* allow to expand */ dictExpandAllowed, /* allow to expand */
dictRehashingStarted,
}; };
/* Db->expires */ /* Db->expires */
...@@ -594,15 +605,36 @@ void tryResizeHashTables(int dbid) { ...@@ -594,15 +605,36 @@ void tryResizeHashTables(int dbid) {
* The function returns 1 if some rehashing was performed, otherwise 0 * The function returns 1 if some rehashing was performed, otherwise 0
* is returned. */ * is returned. */
int incrementallyRehash(int dbid) { int incrementallyRehash(int dbid) {
/* Keys dictionary */ /* Rehash main dictionary. */
dict *d = getRandomDict(&server.db[dbid], 1); if (server.cluster_enabled) {
listNode *node, *nextNode;
monotime timer;
elapsedStart(&timer);
/* Our goal is to rehash as many slot specific dictionaries as we can before reaching predefined threshold,
* while removing those that already finished rehashing from the queue. */
while ((node = listFirst(server.db[dbid].rehashing))) {
if (dictIsRehashing((dict *) listNodeValue(node))) {
dictRehashMilliseconds(listNodeValue(node), INCREMENTAL_REHASHING_THRESHOLD_MS);
if (elapsedMs(timer) >= INCREMENTAL_REHASHING_THRESHOLD_MS) {
return 1; /* Reached the time limit. */
}
} else { /* It is possible that rehashing has already completed for this dictionary, simply remove it from the queue. */
nextNode = listNextNode(node);
listDelNode(server.db[dbid].rehashing, node);
node = nextNode;
}
}
/* When cluster mode is disabled, only one dict is used for the entire DB and rehashing list isn't populated. */
} else {
dict *d = server.db[dbid].dict[0];
if (dictIsRehashing(d)) { if (dictIsRehashing(d)) {
dictRehashMilliseconds(d, 1); dictRehashMilliseconds(d, INCREMENTAL_REHASHING_THRESHOLD_MS);
return 1; /* already used our millisecond for this loop... */ return 1; /* already used our millisecond for this loop... */
} }
/* Expires */ }
/* Rehash expires. */
if (dictIsRehashing(server.db[dbid].expires)) { if (dictIsRehashing(server.db[dbid].expires)) {
dictRehashMilliseconds(server.db[dbid].expires,1); dictRehashMilliseconds(server.db[dbid].expires, INCREMENTAL_REHASHING_THRESHOLD_MS);
return 1; /* already used our millisecond for this loop... */ return 1; /* already used our millisecond for this loop... */
} }
return 0; return 0;
...@@ -2557,6 +2589,7 @@ void initServer(void) { ...@@ -2557,6 +2589,7 @@ void initServer(void) {
server.db[j].id = j; server.db[j].id = j;
server.db[j].avg_ttl = 0; server.db[j].avg_ttl = 0;
server.db[j].defrag_later = listCreate(); server.db[j].defrag_later = listCreate();
server.db[j].rehashing = listCreate();
server.db[j].dict_count = slotCount; server.db[j].dict_count = slotCount;
listSetFreeMethod(server.db[j].defrag_later,(void (*)(void*))sdsfree); listSetFreeMethod(server.db[j].defrag_later,(void (*)(void*))sdsfree);
} }
......
...@@ -136,6 +136,8 @@ typedef struct redisObject robj; ...@@ -136,6 +136,8 @@ typedef struct redisObject robj;
#define CONFIG_BINDADDR_MAX 16 #define CONFIG_BINDADDR_MAX 16
#define CONFIG_MIN_RESERVED_FDS 32 #define CONFIG_MIN_RESERVED_FDS 32
#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_THRESHOLD_MS 1
/* 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. */
...@@ -955,6 +957,7 @@ typedef struct redisDb { ...@@ -955,6 +957,7 @@ typedef struct redisDb {
long long avg_ttl; /* Average TTL, just for stats */ long long avg_ttl; /* Average TTL, just for stats */
unsigned long expires_cursor; /* Cursor of the active expire cycle. */ unsigned long expires_cursor; /* Cursor of the active expire cycle. */
list *defrag_later; /* List of key names to attempt to defrag one by one, gradually. */ list *defrag_later; /* List of key names to attempt to defrag one by one, gradually. */
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. */
} redisDb; } redisDb;
...@@ -3005,7 +3008,7 @@ void dismissMemoryInChild(void); ...@@ -3005,7 +3008,7 @@ void dismissMemoryInChild(void);
int restartServer(int flags, mstime_t delay); int restartServer(int flags, mstime_t delay);
unsigned long long int dbSize(const redisDb *db); unsigned long long int dbSize(const redisDb *db);
dict *getDict(redisDb *db, sds key); dict *getDict(redisDb *db, sds key);
dict *getRandomDict(redisDb *db, int shouldBeRehashing); dict *getRandomDict(redisDb *db);
unsigned long dbSlots(const redisDb *db); unsigned long dbSlots(const 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