Commit 1f9c280a authored by Vitaly Arbuzov's avatar Vitaly Arbuzov
Browse files

Add slot size into RDB snapshot and use it to resize dictionaries on load

parent c91044b0
...@@ -2243,8 +2243,8 @@ int rewriteAppendOnlyFileRio(rio *aof) { ...@@ -2243,8 +2243,8 @@ int rewriteAppendOnlyFileRio(rio *aof) {
redisDb *db = server.db + j; redisDb *db = server.db + j;
dict *d; dict *d;
dbIterator dbit; dbIterator dbit;
dbInitIterator(&dbit, db); dbIteratorInit(&dbit, db);
while ((d = dbNextDict(&dbit))) { while ((d = dbIteratorNextDict(&dbit))) {
if (dictSize(d) == 0) continue; if (dictSize(d) == 0) continue;
di = dictGetSafeIterator(d); di = dictGetSafeIterator(d);
/* Iterate this DB writing every entry */ /* Iterate this DB writing every entry */
......
...@@ -49,20 +49,23 @@ int expireIfNeeded(redisDb *db, robj *key, int flags); ...@@ -49,20 +49,23 @@ int expireIfNeeded(redisDb *db, robj *key, int flags);
int keyIsExpired(redisDb *db, robj *key); 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 *dbIteratorNextDict(dbIterator *dbit) {
int64_t slot; int64_t slot;
if (intsetGet(dbit->db->owned_slots, dbit->index++, &slot)){ if (intsetGet(dbit->db->owned_slots, dbit->index++, &slot)){
dbit->cur_slot = (int) slot;
return dbit->db->dict[slot]; return dbit->db->dict[slot];
} }
dbit->cur_slot = -1;
return 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.
* 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 dbIteratorInit(dbIterator *dbit, redisDb *db) {
dbit->db = db; dbit->db = db;
dbit->index = 0; dbit->index = 0;
dbit->cur_slot = -1;
} }
/* Returns next non-empty 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.
...@@ -842,8 +845,8 @@ void keysCommand(client *c) { ...@@ -842,8 +845,8 @@ void keysCommand(client *c) {
void *replylen = addReplyDeferredLen(c); void *replylen = addReplyDeferredLen(c);
dict *d; dict *d;
dbIterator dbit; dbIterator dbit;
dbInitIterator(&dbit, c->db); dbIteratorInit(&dbit, c->db);
while ((d = dbNextDict(&dbit))) { while ((d = dbIteratorNextDict(&dbit))) {
if (dictSize(d) == 0) continue; if (dictSize(d) == 0) continue;
di = dictGetSafeIterator(d); di = dictGetSafeIterator(d);
allkeys = (pattern[0] == '*' && plen == 1); allkeys = (pattern[0] == '*' && plen == 1);
...@@ -1167,8 +1170,8 @@ unsigned long dbSlots(redisDb *db) { ...@@ -1167,8 +1170,8 @@ unsigned long dbSlots(redisDb *db) {
unsigned long slots = 0; unsigned long slots = 0;
dict *d; dict *d;
dbIterator dbit; dbIterator dbit;
dbInitIterator(&dbit, db); dbIteratorInit(&dbit, db);
while ((d = dbNextDict(&dbit))) { while ((d = dbIteratorNextDict(&dbit))) {
slots += dictSlots(d); slots += dictSlots(d);
} }
return slots; return slots;
...@@ -1853,8 +1856,8 @@ void expandDb(const redisDb *db, uint64_t db_size) { ...@@ -1853,8 +1856,8 @@ void expandDb(const redisDb *db, uint64_t db_size) {
if (server.cluster_enabled) { if (server.cluster_enabled) {
dict *d; dict *d;
dbIterator dbit; dbIterator dbit;
dbInitIterator(&dbit, (redisDb *) db); dbIteratorInit(&dbit, (redisDb *) db);
while ((d = dbNextDict(&dbit))) { while ((d = dbIteratorNextDict(&dbit))) {
/* We don't exact number of keys that would fall into each slot, but we can approximate it, assuming even distribution. */ /* We don't exact number of keys that would fall into each slot, but we can approximate it, assuming even distribution. */
dictExpand(d, (db_size / server.cluster->myself->numslots)); dictExpand(d, (db_size / server.cluster->myself->numslots));
} }
...@@ -2609,8 +2612,8 @@ void dbGetStats(char *buf, size_t bufsize, redisDb *db) { ...@@ -2609,8 +2612,8 @@ void dbGetStats(char *buf, size_t bufsize, redisDb *db) {
dictStats *rehashHtStats = NULL; dictStats *rehashHtStats = NULL;
dict *d; dict *d;
dbIterator dbit; dbIterator dbit;
dbInitIterator(&dbit, db); dbIteratorInit(&dbit, db);
while ((d = dbNextDict(&dbit))) { while ((d = dbIteratorNextDict(&dbit))) {
dictStats *stats = dictGetStatsHt(d, 0); dictStats *stats = dictGetStatsHt(d, 0);
if (!mainHtStats) mainHtStats = stats; if (!mainHtStats) mainHtStats = stats;
else { else {
......
...@@ -289,8 +289,8 @@ void computeDatasetDigest(unsigned char *final) { ...@@ -289,8 +289,8 @@ void computeDatasetDigest(unsigned char *final) {
int hasEntries = 0; int hasEntries = 0;
dict *d; dict *d;
dbIterator dbit; dbIterator dbit;
dbInitIterator(&dbit, db); dbIteratorInit(&dbit, db);
while ((d = dbNextDict(&dbit))) { while ((d = dbIteratorNextDict(&dbit))) {
if (dictSize(d) == 0) continue; if (dictSize(d) == 0) continue;
hasEntries = 1; hasEntries = 1;
di = dictGetSafeIterator(d); di = dictGetSafeIterator(d);
......
...@@ -1323,9 +1323,21 @@ ssize_t rdbSaveDb(rio *rdb, int dbid, int rdbflags, long *key_counter) { ...@@ -1323,9 +1323,21 @@ ssize_t rdbSaveDb(rio *rdb, int dbid, int rdbflags, long *key_counter) {
dict *d; dict *d;
dbIterator dbit; dbIterator dbit;
dbInitIterator(&dbit, db); dbIteratorInit(&dbit, db);
while ((d = dbNextDict(&dbit))) { while ((d = dbIteratorNextDict(&dbit))) {
if (!dictSize(d)) continue; if (!dictSize(d)) continue;
/* Save slot info. */
if (server.cluster_enabled) {
serverAssert(dbit.cur_slot >= 0 && dbit.cur_slot < CLUSTER_SLOTS);
if ((res = rdbSaveType(rdb, RDB_OPCODE_SLOT_INFO)) < 0) goto werr;
written += res;
if ((res = rdbSaveLen(rdb, dbit.cur_slot)) < 0) goto werr;
written += res;
if ((res = rdbSaveLen(rdb, dictSize(d))) < 0) goto werr;
written += res;
}
di = dictGetSafeIterator(d); di = dictGetSafeIterator(d);
/* Iterate this DB writing every entry */ /* Iterate this DB writing every entry */
while ((de = dictNext(di)) != NULL) { while ((de = dictNext(di)) != NULL) {
...@@ -3070,9 +3082,24 @@ int rdbLoadRioWithLoadingCtx(rio *rdb, int rdbflags, rdbSaveInfo *rsi, rdbLoadin ...@@ -3070,9 +3082,24 @@ int rdbLoadRioWithLoadingCtx(rio *rdb, int rdbflags, rdbSaveInfo *rsi, rdbLoadin
goto eoferr; goto eoferr;
if ((expires_size = rdbLoadLen(rdb,NULL)) == RDB_LENERR) if ((expires_size = rdbLoadLen(rdb,NULL)) == RDB_LENERR)
goto eoferr; goto eoferr;
/* In cluster mode we don't resize whole DB, instead we rely on RDB_OPCODE_SLOT_INFO for resizing slots. */
if (!server.cluster_enabled) {
expandDb(db, db_size); expandDb(db, db_size);
}
dictExpand(db->expires,expires_size); dictExpand(db->expires,expires_size);
continue; /* Read next opcode. */ continue; /* Read next opcode. */
} else if (type == RDB_OPCODE_SLOT_INFO) {
if (!server.cluster_enabled) {
continue; /* Ignore gracefully. */
}
uint64_t slot_id, slot_size;
if ((slot_id = rdbLoadLen(rdb,NULL)) == RDB_LENERR)
goto eoferr;
if ((slot_size = rdbLoadLen(rdb,NULL)) == RDB_LENERR)
goto eoferr;
/* In cluster mode we resize individual slot specific dictionaries based on the number of keys that slot holds. */
dictExpand(db->dict[slot_id], slot_size);
continue; /* Read next opcode. */
} else if (type == RDB_OPCODE_AUX) { } else if (type == RDB_OPCODE_AUX) {
/* AUX: generic string-string fields. Use to add state to RDB /* AUX: generic string-string fields. Use to add state to RDB
* which is backward compatible. Implementations of RDB loading * which is backward compatible. Implementations of RDB loading
......
...@@ -104,6 +104,7 @@ ...@@ -104,6 +104,7 @@
#define rdbIsObjectType(t) (((t) >= 0 && (t) <= 7) || ((t) >= 9 && (t) <= 21)) #define rdbIsObjectType(t) (((t) >= 0 && (t) <= 7) || ((t) >= 9 && (t) <= 21))
/* Special RDB opcodes (saved/loaded with rdbSaveType/rdbLoadType). */ /* Special RDB opcodes (saved/loaded with rdbSaveType/rdbLoadType). */
#define RDB_OPCODE_SLOT_INFO 244 /* Individual slot info, such as slot id and size (cluster mode only). */
#define RDB_OPCODE_FUNCTION2 245 /* function library data */ #define RDB_OPCODE_FUNCTION2 245 /* function library data */
#define RDB_OPCODE_FUNCTION_PRE_GA 246 /* old function library data for 7.0 rc1 and rc2 */ #define RDB_OPCODE_FUNCTION_PRE_GA 246 /* old function library data for 7.0 rc1 and rc2 */
#define RDB_OPCODE_MODULE_AUX 247 /* Module auxiliary data. */ #define RDB_OPCODE_MODULE_AUX 247 /* Module auxiliary data. */
......
...@@ -276,6 +276,16 @@ int redis_check_rdb(char *rdbfilename, FILE *fp) { ...@@ -276,6 +276,16 @@ int redis_check_rdb(char *rdbfilename, FILE *fp) {
if ((expires_size = rdbLoadLen(&rdb,NULL)) == RDB_LENERR) if ((expires_size = rdbLoadLen(&rdb,NULL)) == RDB_LENERR)
goto eoferr; goto eoferr;
continue; /* Read type again. */ continue; /* Read type again. */
} else if (type == RDB_OPCODE_SLOT_INFO) {
if (!server.cluster_enabled) {
continue; /* Ignore gracefully. */
}
uint64_t slot_id, slot_size;
if ((slot_id = rdbLoadLen(&rdb,NULL)) == RDB_LENERR)
goto eoferr;
if ((slot_size = rdbLoadLen(&rdb,NULL)) == RDB_LENERR)
goto eoferr;
continue; /* Read type again. */
} else if (type == RDB_OPCODE_AUX) { } else if (type == RDB_OPCODE_AUX) {
/* AUX: generic string-string fields. Use to add state to RDB /* AUX: generic string-string fields. Use to add state to RDB
* which is backward compatible. Implementations of RDB loading * which is backward compatible. Implementations of RDB loading
......
...@@ -590,8 +590,8 @@ int htNeedsResize(dict *dict) { ...@@ -590,8 +590,8 @@ int htNeedsResize(dict *dict) {
void tryResizeHashTables(int dbid) { void tryResizeHashTables(int dbid) {
dict *d; dict *d;
dbIterator dbit; dbIterator dbit;
dbInitIterator(&dbit, &server.db[dbid]); dbIteratorInit(&dbit, &server.db[dbid]);
while ((d = dbNextDict(&dbit))) { while ((d = dbIteratorNextDict(&dbit))) {
if (htNeedsResize(d)) if (htNeedsResize(d))
dictResize(d); dictResize(d);
} }
......
...@@ -2361,11 +2361,13 @@ typedef struct { ...@@ -2361,11 +2361,13 @@ typedef struct {
typedef struct dbIterator { typedef struct dbIterator {
redisDb *db; redisDb *db;
int index; int index;
int cur_slot;
} dbIterator; } dbIterator;
/* DB iterator specific functions */ /* DB iterator specific functions */
void dbInitIterator(dbIterator *dbit, redisDb *db); void dbIteratorInit(dbIterator *dbit, redisDb *db);
dict *dbNextDict(dbIterator *dbit); dict *dbIteratorNextDict(dbIterator *dbit);
int dbIterCurSlot(dbIterator *dbit);
dict *dbGetNextUnvisitedSlot(redisDb *db, int *slot); dict *dbGetNextUnvisitedSlot(redisDb *db, int *slot);
/* SCAN specific commands for easy cursor manipulation, shared between main code and modules. */ /* SCAN specific commands for easy cursor manipulation, shared between main code and modules. */
......
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