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

Limit number of dictionaries that can be resized in one cron iteration

parent 7313244d
...@@ -535,6 +535,7 @@ long long emptyDbStructure(redisDb *dbarray, int dbnum, int async, ...@@ -535,6 +535,7 @@ long long emptyDbStructure(redisDb *dbarray, int dbnum, int async,
/* Because all keys of database are removed, reset average ttl. */ /* Because all keys of database are removed, reset average ttl. */
dbarray[j].avg_ttl = 0; dbarray[j].avg_ttl = 0;
dbarray[j].expires_cursor = 0; dbarray[j].expires_cursor = 0;
dbarray[j].resize_cursor = 0;
dbarray[j].key_count = 0; dbarray[j].key_count = 0;
zfree(dbarray[j].non_empty_dicts); zfree(dbarray[j].non_empty_dicts);
dbarray[j].non_empty_dicts = intsetNew(); dbarray[j].non_empty_dicts = intsetNew();
...@@ -1568,6 +1569,7 @@ int dbSwapDatabases(int id1, int id2) { ...@@ -1568,6 +1569,7 @@ int dbSwapDatabases(int id1, int id2) {
db1->expires = db2->expires; db1->expires = db2->expires;
db1->avg_ttl = db2->avg_ttl; db1->avg_ttl = db2->avg_ttl;
db1->expires_cursor = db2->expires_cursor; db1->expires_cursor = db2->expires_cursor;
db1->resize_cursor = db2->resize_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->non_empty_dicts = db2->non_empty_dicts; db1->non_empty_dicts = db2->non_empty_dicts;
...@@ -1576,6 +1578,7 @@ int dbSwapDatabases(int id1, int id2) { ...@@ -1576,6 +1578,7 @@ int dbSwapDatabases(int id1, int id2) {
db2->expires = aux.expires; db2->expires = aux.expires;
db2->avg_ttl = aux.avg_ttl; db2->avg_ttl = aux.avg_ttl;
db2->expires_cursor = aux.expires_cursor; db2->expires_cursor = aux.expires_cursor;
db2->resize_cursor = aux.resize_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->non_empty_dicts = aux.non_empty_dicts; db2->non_empty_dicts = aux.non_empty_dicts;
...@@ -1616,6 +1619,7 @@ void swapMainDbWithTempDb(redisDb *tempDb) { ...@@ -1616,6 +1619,7 @@ void swapMainDbWithTempDb(redisDb *tempDb) {
activedb->expires = newdb->expires; activedb->expires = newdb->expires;
activedb->avg_ttl = newdb->avg_ttl; activedb->avg_ttl = newdb->avg_ttl;
activedb->expires_cursor = newdb->expires_cursor; activedb->expires_cursor = newdb->expires_cursor;
activedb->resize_cursor = newdb->resize_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->non_empty_dicts = newdb->non_empty_dicts; activedb->non_empty_dicts = newdb->non_empty_dicts;
...@@ -1624,6 +1628,7 @@ void swapMainDbWithTempDb(redisDb *tempDb) { ...@@ -1624,6 +1628,7 @@ void swapMainDbWithTempDb(redisDb *tempDb) {
newdb->expires = aux.expires; newdb->expires = aux.expires;
newdb->avg_ttl = aux.avg_ttl; newdb->avg_ttl = aux.avg_ttl;
newdb->expires_cursor = aux.expires_cursor; newdb->expires_cursor = aux.expires_cursor;
newdb->resize_cursor = aux.resize_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->non_empty_dicts = aux.non_empty_dicts; newdb->non_empty_dicts = aux.non_empty_dicts;
......
...@@ -590,13 +590,20 @@ int htNeedsResize(dict *dict) { ...@@ -590,13 +590,20 @@ int htNeedsResize(dict *dict) {
void tryResizeHashTables(int dbid) { void tryResizeHashTables(int dbid) {
dict *d; dict *d;
dbIterator dbit; dbIterator dbit;
dbIteratorInit(&dbit, &server.db[dbid]); redisDb *db = &server.db[dbid];
while ((d = dbIteratorNextDict(&dbit))) { dbIteratorInit(&dbit, db);
dbit.index = db->resize_cursor;
for (int i = 0; i < CRON_DICTS_PER_CALL; i++) {
d = dbIteratorNextDict(&dbit);
if (d == NULL) break;
if (htNeedsResize(d)) if (htNeedsResize(d))
dictResize(d); dictResize(d);
} }
if (htNeedsResize(server.db[dbid].expires)) /* Save current index in the resize cursor, or start over if we've reached the end.*/
dictResize(server.db[dbid].expires); db->resize_cursor = dbit.cur_slot == -1 ? 0 : dbit.index;
if (htNeedsResize(db->expires))
dictResize(db->expires);
} }
/* Our hash table implementation performs rehashing incrementally while /* Our hash table implementation performs rehashing incrementally while
...@@ -2584,6 +2591,7 @@ void initServer(void) { ...@@ -2584,6 +2591,7 @@ void initServer(void) {
server.db[j].dict = dictCreateMultiple(&dbDictType, slotCount); server.db[j].dict = dictCreateMultiple(&dbDictType, slotCount);
server.db[j].expires = dictCreate(&dbExpiresDictType); server.db[j].expires = dictCreate(&dbExpiresDictType);
server.db[j].expires_cursor = 0; server.db[j].expires_cursor = 0;
server.db[j].resize_cursor = 0;
server.db[j].blocking_keys = dictCreate(&keylistDictType); server.db[j].blocking_keys = dictCreate(&keylistDictType);
server.db[j].blocking_keys_unblock_on_nokey = dictCreate(&objectKeyPointerValueDictType); server.db[j].blocking_keys_unblock_on_nokey = dictCreate(&objectKeyPointerValueDictType);
server.db[j].ready_keys = dictCreate(&objectKeyPointerValueDictType); server.db[j].ready_keys = dictCreate(&objectKeyPointerValueDictType);
......
...@@ -115,6 +115,7 @@ typedef struct redisObject robj; ...@@ -115,6 +115,7 @@ typedef struct redisObject robj;
#define CONFIG_MAX_HZ 500 #define CONFIG_MAX_HZ 500
#define MAX_CLIENTS_PER_CLOCK_TICK 200 /* HZ is adapted based on that. */ #define MAX_CLIENTS_PER_CLOCK_TICK 200 /* HZ is adapted based on that. */
#define CRON_DBS_PER_CALL 16 #define CRON_DBS_PER_CALL 16
#define CRON_DICTS_PER_CALL 1024 /* Number of dictionaries that can be resized by cron job during one call. */
#define NET_MAX_WRITES_PER_EVENT (1024*64) #define NET_MAX_WRITES_PER_EVENT (1024*64)
#define PROTO_SHARED_SELECT_CMDS 10 #define PROTO_SHARED_SELECT_CMDS 10
#define OBJ_SHARED_INTEGERS 10000 #define OBJ_SHARED_INTEGERS 10000
...@@ -957,6 +958,7 @@ typedef struct redisDb { ...@@ -957,6 +958,7 @@ typedef struct redisDb {
int id; /* Database ID */ int id; /* Database ID */
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. */
int resize_cursor; /* Cron job uses this cursor to gradually resize dictionaries. */
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. */ 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. */
......
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