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

Add DB iterator to go over owned dictionaries

parent 90efd492
...@@ -2238,8 +2238,9 @@ int rewriteAppendOnlyFileRio(rio *aof) { ...@@ -2238,8 +2238,9 @@ int rewriteAppendOnlyFileRio(rio *aof) {
for (j = 0; j < server.dbnum; j++) { for (j = 0; j < server.dbnum; j++) {
char selectcmd[] = "*2\r\n$6\r\nSELECT\r\n"; char selectcmd[] = "*2\r\n$6\r\nSELECT\r\n";
redisDb *db = server.db + j; redisDb *db = server.db + j;
for (int k = 0; k < db->dict_count; k++) { dict *d;
dict *d = db->dict[k]; dbIterator *dbit = dbGetIterator(db);
while ((d = dbNextDict(dbit))) {
if (dictSize(d) == 0) continue; if (dictSize(d) == 0) continue;
di = dictGetSafeIterator(d); di = dictGetSafeIterator(d);
...@@ -2316,6 +2317,7 @@ int rewriteAppendOnlyFileRio(rio *aof) { ...@@ -2316,6 +2317,7 @@ int rewriteAppendOnlyFileRio(rio *aof) {
dictReleaseIterator(di); dictReleaseIterator(di);
di = NULL; di = NULL;
} }
dbReleaseIterator(dbit);
} }
return C_OK; return C_OK;
......
...@@ -48,6 +48,32 @@ ...@@ -48,6 +48,32 @@
int expireIfNeeded(redisDb *db, robj *key, int flags); int expireIfNeeded(redisDb *db, robj *key, int flags);
int keyIsExpired(redisDb *db, robj *key); int keyIsExpired(redisDb *db, robj *key);
dict *dbNextDict(dbIterator *iter) {
while (iter->index < iter->db->dict_count - 1) {
iter->index++;
dict *d = iter->db->dict[iter->index];
if (!server.cluster_enabled) return d; /* There is a single dictionary in non cluster mode. */
clusterNode *myself = server.cluster->myself;
/* We need a non-empty dictionary that's owned by this node. */
if (clusterNodeGetSlotBit(myself, iter->index) && dictSize(d) > 0) {
return d;
}
}
return NULL;
}
dbIterator *dbGetIterator(redisDb *db) {
dbIterator *iter = zmalloc(sizeof(*iter));
iter->db = db;
iter->index = -1;
return iter;
}
void dbReleaseIterator(dbIterator *iter) {
zfree(iter);
}
/* Update LFU when an object is accessed. /* Update LFU when an object is accessed.
* Firstly, decrement the counter if the decrement time is reached. * Firstly, decrement the counter if the decrement time is reached.
* Then logarithmically increment the counter, and update the access time. */ * Then logarithmically increment the counter, and update the access time. */
...@@ -782,8 +808,9 @@ void keysCommand(client *c) { ...@@ -782,8 +808,9 @@ void keysCommand(client *c) {
int plen = sdslen(pattern), allkeys; int plen = sdslen(pattern), allkeys;
long numkeys = 0; long numkeys = 0;
void *replylen = addReplyDeferredLen(c); void *replylen = addReplyDeferredLen(c);
for (int i = 0; i < c->db->dict_count; i++) { dict *d;
dict *d = c->db->dict[i]; dbIterator *dbit = dbGetIterator(c->db);
while ((d = dbNextDict(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);
...@@ -803,6 +830,7 @@ void keysCommand(client *c) { ...@@ -803,6 +830,7 @@ void keysCommand(client *c) {
break;} break;}
dictReleaseIterator(di); dictReleaseIterator(di);
} }
dbReleaseIterator(dbit);
setDeferredArrayLen(c,replylen,numkeys); setDeferredArrayLen(c,replylen,numkeys);
} }
...@@ -969,6 +997,7 @@ void scanGenericCommand(client *c, robj *o, unsigned long long cursor) { ...@@ -969,6 +997,7 @@ void scanGenericCommand(client *c, robj *o, unsigned long long cursor) {
/* In cluster mode there is a separate dictionary for each slot. /* In cluster mode there is a separate dictionary for each slot.
* If cursor is empty, we should try exploring next non-empty slot. */ * If cursor is empty, we should try exploring next non-empty slot. */
while (o == NULL && !cursor && slot < c->db->dict_count - 1) { while (o == NULL && !cursor && slot < c->db->dict_count - 1) {
// TODO (vitarb@) Consider using dbIterator, or add slot ownership check.
slot++; ht = c->db->dict[slot]; slot++; ht = c->db->dict[slot];
if (dictSize(ht) > 0) { if (dictSize(ht) > 0) {
hasUnvisitedSlot = 1; hasUnvisitedSlot = 1;
...@@ -1097,17 +1126,23 @@ void dbsizeCommand(client *c) { ...@@ -1097,17 +1126,23 @@ void dbsizeCommand(client *c) {
unsigned long long int dbSize(const redisDb *db) { unsigned long long int dbSize(const redisDb *db) {
unsigned long long size = 0; unsigned long long size = 0;
for (int i = 0; i < db->dict_count; i++) { dict *d;
size += dictSize(db->dict[i]); dbIterator *dbit = dbGetIterator(db);
while ((d = dbNextDict(dbit))) {
size += dictSize(d);
} }
dbReleaseIterator(dbit);
return size; return size;
} }
unsigned long dbSlots(const redisDb *db) { unsigned long dbSlots(const redisDb *db) {
unsigned long slots = 0; unsigned long slots = 0;
for (int i = 0; i < db->dict_count; i++) { dict *d;
slots += dictSlots(db->dict[i]); dbIterator *dbit = dbGetIterator(db);
while ((d = dbNextDict(dbit))) {
slots += dictSlots(d);
} }
dbReleaseIterator(dbit);
return slots; return slots;
} }
...@@ -1780,11 +1815,13 @@ int expireIfNeeded(redisDb *db, robj *key, int flags) { ...@@ -1780,11 +1815,13 @@ int expireIfNeeded(redisDb *db, robj *key, int flags) {
* this node owns. */ * this node owns. */
void expandDb(const redisDb *db, uint64_t db_size) { void expandDb(const redisDb *db, uint64_t db_size) {
if (server.cluster_enabled) { if (server.cluster_enabled) {
clusterNode *myself = server.cluster->myself; dict *d;
for (int i = 0; i < db->dict_count; i++) { dbIterator *dbit = dbGetIterator(db);
while ((d = dbNextDict(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. */
if (clusterNodeGetSlotBit(myself, i)) dictExpand(db->dict[i], (db_size / myself->numslots)); dictExpand(d, (db_size / server.cluster->myself->numslots));
} }
dbReleaseIterator(dbit);
} else { } else {
dictExpand(db->dict[0], db_size); dictExpand(db->dict[0], db_size);
} }
...@@ -2534,16 +2571,17 @@ void dbGetStats(char *buf, size_t bufsize, redisDb *db) { ...@@ -2534,16 +2571,17 @@ void dbGetStats(char *buf, size_t bufsize, redisDb *db) {
size_t orig_bufsize = bufsize; size_t orig_bufsize = bufsize;
dictStats *mainHtStats = NULL; dictStats *mainHtStats = NULL;
dictStats *rehashHtStats = NULL; dictStats *rehashHtStats = NULL;
dict *d;
for (int i = 0; i < db->dict_count; i++) { dbIterator *dbit = dbGetIterator(db);
dictStats *stats = dictGetStatsHt(db->dict[i], 0); while ((d = dbNextDict(dbit))) {
dictStats *stats = dictGetStatsHt(d, 0);
if (!mainHtStats) mainHtStats = stats; if (!mainHtStats) mainHtStats = stats;
else { else {
dictCombineStats(stats, mainHtStats); dictCombineStats(stats, mainHtStats);
dictFreeStats(stats); dictFreeStats(stats);
} }
if (dictIsRehashing(db->dict[i])) { if (dictIsRehashing(d)) {
stats = dictGetStatsHt(db->dict[i], 1); stats = dictGetStatsHt(d, 1);
if (!rehashHtStats) rehashHtStats = stats; if (!rehashHtStats) rehashHtStats = stats;
else { else {
dictCombineStats(stats, rehashHtStats); dictCombineStats(stats, rehashHtStats);
...@@ -2551,6 +2589,7 @@ void dbGetStats(char *buf, size_t bufsize, redisDb *db) { ...@@ -2551,6 +2589,7 @@ void dbGetStats(char *buf, size_t bufsize, redisDb *db) {
} }
} }
} }
dbReleaseIterator(dbit);
l = dictGetStatsMsg(buf, bufsize, mainHtStats); l = dictGetStatsMsg(buf, bufsize, mainHtStats);
dictFreeStats(mainHtStats); dictFreeStats(mainHtStats);
buf += l; buf += l;
......
...@@ -287,10 +287,12 @@ void computeDatasetDigest(unsigned char *final) { ...@@ -287,10 +287,12 @@ void computeDatasetDigest(unsigned char *final) {
for (j = 0; j < server.dbnum; j++) { for (j = 0; j < server.dbnum; j++) {
redisDb *db = server.db+j; redisDb *db = server.db+j;
int hasEntries = 0; int hasEntries = 0;
for (int k = 0; k < db->dict_count; k++) { dict *d;
if (dictSize(db->dict[k]) == 0) continue; dbIterator *dbit = dbGetIterator(db);
while ((d = dbNextDict(dbit))) {
if (dictSize(d) == 0) continue;
hasEntries = 1; hasEntries = 1;
di = dictGetSafeIterator(db->dict[k]); 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) {
...@@ -312,6 +314,7 @@ void computeDatasetDigest(unsigned char *final) { ...@@ -312,6 +314,7 @@ void computeDatasetDigest(unsigned char *final) {
} }
dictReleaseIterator(di); dictReleaseIterator(di);
} }
dbReleaseIterator(dbit);
if (hasEntries) { if (hasEntries) {
/* hash the DB id, so the same dataset moved in a different DB will lead to a different digest */ /* hash the DB id, so the same dataset moved in a different DB will lead to a different digest */
aux = htonl(j); aux = htonl(j);
......
...@@ -1323,8 +1323,9 @@ ssize_t rdbSaveDb(rio *rdb, int dbid, int rdbflags, long *key_counter) { ...@@ -1323,8 +1323,9 @@ ssize_t rdbSaveDb(rio *rdb, int dbid, int rdbflags, long *key_counter) {
if ((res = rdbSaveLen(rdb,expires_size)) < 0) goto werr; if ((res = rdbSaveLen(rdb,expires_size)) < 0) goto werr;
written += res; written += res;
for (int i = 0; i < db->dict_count; i++) { dict *d;
dict *d = db->dict[i]; dbIterator *dbit = dbGetIterator(db);
while ((d = dbNextDict(dbit))) {
if (!dictSize(d)) continue; if (!dictSize(d)) continue;
di = dictGetSafeIterator(d); di = dictGetSafeIterator(d);
/* Iterate this DB writing every entry */ /* Iterate this DB writing every entry */
...@@ -1359,6 +1360,7 @@ ssize_t rdbSaveDb(rio *rdb, int dbid, int rdbflags, long *key_counter) { ...@@ -1359,6 +1360,7 @@ ssize_t rdbSaveDb(rio *rdb, int dbid, int rdbflags, long *key_counter) {
dictReleaseIterator(di); dictReleaseIterator(di);
} }
dbReleaseIterator(dbit);
return written; return written;
werr: werr:
......
...@@ -588,11 +588,13 @@ int htNeedsResize(dict *dict) { ...@@ -588,11 +588,13 @@ int htNeedsResize(dict *dict) {
/* If the percentage of used slots in the HT reaches HASHTABLE_MIN_FILL /* If the percentage of used slots in the HT reaches HASHTABLE_MIN_FILL
* we resize the hash table to save memory */ * we resize the hash table to save memory */
void tryResizeHashTables(int dbid) { void tryResizeHashTables(int dbid) {
for (int i = 0; i < server.db[dbid].dict_count; i++) { dict *d;
dict *d = server.db[dbid].dict[i]; dbIterator *dbit = dbGetIterator(&server.db[dbid]);
while ((d = dbNextDict(dbit))) {
if (htNeedsResize(d)) if (htNeedsResize(d))
dictResize(d); dictResize(d);
} }
dbReleaseIterator(dbit);
if (htNeedsResize(server.db[dbid].expires)) if (htNeedsResize(server.db[dbid].expires))
dictResize(server.db[dbid].expires); dictResize(server.db[dbid].expires);
} }
......
...@@ -2354,6 +2354,17 @@ typedef struct { ...@@ -2354,6 +2354,17 @@ typedef struct {
unsigned char *lpi; /* listpack iterator */ unsigned char *lpi; /* listpack iterator */
} setTypeIterator; } setTypeIterator;
/* Struture for DB iterator that allows iterating across multiple slot specific dictionaries in cluster mode. */
typedef struct dbIterator {
redisDb *db;
int index;
} dbIterator;
/* dbIterator specific functions */
dict *dbNextDict(dbIterator *iter);
dbIterator *dbGetIterator(redisDb *db);
void dbReleaseIterator(dbIterator *iter);
/* Structure to hold hash iteration abstraction. Note that iteration over /* Structure to hold hash iteration abstraction. Note that iteration over
* hashes involves both fields and values. Because it is possible that * hashes involves both fields and values. Because it is possible that
* not both are required, store pointers in the iterator to avoid * not both are required, store pointers in the iterator to avoid
......
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