Commit 297922e6 authored by Vitaly Arbuzov's avatar Vitaly Arbuzov
Browse files

Add ability to use dbIterator as dictIterator

parent 72935b9d
......@@ -2235,7 +2235,6 @@ werr:
}
int rewriteAppendOnlyFileRio(rio *aof) {
dictIterator *di = NULL;
dictEntry *de;
int j;
long key_count = 0;
......@@ -2256,86 +2255,78 @@ int rewriteAppendOnlyFileRio(rio *aof) {
if (rioWrite(aof, selectcmd, sizeof(selectcmd) - 1) == 0) goto werr;
if (rioWriteBulkLongLong(aof, j) == 0) goto werr;
redisDb *db = server.db + j;
dict *d;
dbIterator dbit;
dbIteratorInit(&dbit, db);
while ((d = dbIteratorNextDict(&dbit))) {
if (dictSize(d) == 0) continue;
di = dictGetSafeIterator(d);
/* Iterate this DB writing every entry */
while ((de = dictNext(di)) != NULL) {
sds keystr;
robj key, *o;
long long expiretime;
size_t aof_bytes_before_key = aof->processed_bytes;
keystr = dictGetKey(de);
o = dictGetVal(de);
initStaticStringObject(key, keystr);
expiretime = getExpire(db, &key);
/* Save the key and associated value */
if (o->type == OBJ_STRING) {
/* Emit a SET command */
char cmd[] = "*3\r\n$3\r\nSET\r\n";
if (rioWrite(aof, cmd, sizeof(cmd) - 1) == 0) goto werr;
/* Key and value */
if (rioWriteBulkObject(aof, &key) == 0) goto werr;
if (rioWriteBulkObject(aof, o) == 0) goto werr;
} else if (o->type == OBJ_LIST) {
if (rewriteListObject(aof, &key, o) == 0) goto werr;
} else if (o->type == OBJ_SET) {
if (rewriteSetObject(aof, &key, o) == 0) goto werr;
} else if (o->type == OBJ_ZSET) {
if (rewriteSortedSetObject(aof, &key, o) == 0) goto werr;
} else if (o->type == OBJ_HASH) {
if (rewriteHashObject(aof, &key, o) == 0) goto werr;
} else if (o->type == OBJ_STREAM) {
if (rewriteStreamObject(aof, &key, o) == 0) goto werr;
} else if (o->type == OBJ_MODULE) {
if (rewriteModuleObject(aof, &key, o, j) == 0) goto werr;
} else {
serverPanic("Unknown object type");
}
/* Iterate this DB writing every entry */
while ((de = dbIteratorNext(&dbit)) != NULL) {
sds keystr;
robj key, *o;
long long expiretime;
size_t aof_bytes_before_key = aof->processed_bytes;
keystr = dictGetKey(de);
o = dictGetVal(de);
initStaticStringObject(key, keystr);
expiretime = getExpire(db, &key);
/* Save the key and associated value */
if (o->type == OBJ_STRING) {
/* Emit a SET command */
char cmd[] = "*3\r\n$3\r\nSET\r\n";
if (rioWrite(aof, cmd, sizeof(cmd) - 1) == 0) goto werr;
/* Key and value */
if (rioWriteBulkObject(aof, &key) == 0) goto werr;
if (rioWriteBulkObject(aof, o) == 0) goto werr;
} else if (o->type == OBJ_LIST) {
if (rewriteListObject(aof, &key, o) == 0) goto werr;
} else if (o->type == OBJ_SET) {
if (rewriteSetObject(aof, &key, o) == 0) goto werr;
} else if (o->type == OBJ_ZSET) {
if (rewriteSortedSetObject(aof, &key, o) == 0) goto werr;
} else if (o->type == OBJ_HASH) {
if (rewriteHashObject(aof, &key, o) == 0) goto werr;
} else if (o->type == OBJ_STREAM) {
if (rewriteStreamObject(aof, &key, o) == 0) goto werr;
} else if (o->type == OBJ_MODULE) {
if (rewriteModuleObject(aof, &key, o, j) == 0) goto werr;
} else {
serverPanic("Unknown object type");
}
/* In fork child process, we can try to release memory back to the
* OS and possibly avoid or decrease COW. We give the dismiss
* mechanism a hint about an estimated size of the object we stored. */
size_t dump_size = aof->processed_bytes - aof_bytes_before_key;
if (server.in_fork_child) dismissObject(o, dump_size);
/* Save the expire time */
if (expiretime != -1) {
char cmd[] = "*3\r\n$9\r\nPEXPIREAT\r\n";
if (rioWrite(aof, cmd, sizeof(cmd) - 1) == 0) goto werr;
if (rioWriteBulkObject(aof, &key) == 0) goto werr;
if (rioWriteBulkLongLong(aof, expiretime) == 0) goto werr;
}
/* In fork child process, we can try to release memory back to the
* OS and possibly avoid or decrease COW. We give the dismiss
* mechanism a hint about an estimated size of the object we stored. */
size_t dump_size = aof->processed_bytes - aof_bytes_before_key;
if (server.in_fork_child) dismissObject(o, dump_size);
/* Save the expire time */
if (expiretime != -1) {
char cmd[] = "*3\r\n$9\r\nPEXPIREAT\r\n";
if (rioWrite(aof, cmd, sizeof(cmd) - 1) == 0) goto werr;
if (rioWriteBulkObject(aof, &key) == 0) goto werr;
if (rioWriteBulkLongLong(aof, expiretime) == 0) goto werr;
}
/* Update info every 1 second (approximately).
* in order to avoid calling mstime() on each iteration, we will
* check the diff every 1024 keys */
if ((key_count++ & 1023) == 0) {
long long now = mstime();
if (now - updated_time >= 1000) {
sendChildInfo(CHILD_INFO_TYPE_CURRENT_INFO, key_count, "AOF rewrite");
updated_time = now;
}
/* Update info every 1 second (approximately).
* in order to avoid calling mstime() on each iteration, we will
* check the diff every 1024 keys */
if ((key_count++ & 1023) == 0) {
long long now = mstime();
if (now - updated_time >= 1000) {
sendChildInfo(CHILD_INFO_TYPE_CURRENT_INFO, key_count, "AOF rewrite");
updated_time = now;
}
/* Delay before next key if required (for testing) */
if (server.rdb_key_save_delay)
debugDelay(server.rdb_key_save_delay);
}
dictReleaseIterator(di);
di = NULL;
/* Delay before next key if required (for testing) */
if (server.rdb_key_save_delay)
debugDelay(server.rdb_key_save_delay);
}
}
return C_OK;
werr:
if (di) dictReleaseIterator(di);
return C_ERR;
}
......
......@@ -61,6 +61,23 @@ dict *dbIteratorNextDict(dbIterator *dbit) {
return dbit->cur_slot >= 0 ? dbit->db->dict[dbit->cur_slot] : NULL;
}
/* Returns next entry from the multi slot db. */
dictEntry *dbIteratorNext(dbIterator *dbit) {
if (!dbit->di.d) { /* No current dict, need to get a new one. */
dict *d = dbIteratorNextDict(dbit);
if (!d) return NULL;
dictInitSafeIterator(&dbit->di, d);
}
dictEntry *de = dictNext(&dbit->di);
if (!de) { /* Reached the end of the dictionary, check if there are more. */
dict *d = dbIteratorNextDict(dbit);
if (!d) return NULL;
dictInitSafeIterator(&dbit->di, d);
return dictNext(&dbit->di);
}
return de;
}
/* Returns DB iterator that can be used to iterate through sub-dictionaries.
* 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. */
......@@ -68,6 +85,7 @@ void dbIteratorInit(dbIterator *dbit, redisDb *db) {
dbit->db = db;
dbit->index = 0;
dbit->cur_slot = -1;
dictInitSafeIterator(&dbit->di, NULL);
}
/* Returns next non-empty dictionary strictly after provided slot and updates slot id in the supplied reference. */
......@@ -846,34 +864,28 @@ void randomkeyCommand(client *c) {
}
void keysCommand(client *c) {
dictIterator *di;
dictEntry *de;
sds pattern = c->argv[1]->ptr;
int plen = sdslen(pattern), allkeys;
long numkeys = 0;
void *replylen = addReplyDeferredLen(c);
dict *d;
dbIterator dbit;
dbIteratorInit(&dbit, c->db);
while ((d = dbIteratorNextDict(&dbit))) {
if (dictSize(d) == 0) continue;
di = dictGetSafeIterator(d);
allkeys = (pattern[0] == '*' && plen == 1);
while((de = dictNext(di)) != NULL) {
sds key = dictGetKey(de);
robj *keyobj;
if (allkeys || stringmatchlen(pattern,plen,key,sdslen(key),0)) {
keyobj = createStringObject(key,sdslen(key));
if (!keyIsExpired(c->db,keyobj)) {
addReplyBulk(c,keyobj);
numkeys++;
}
decrRefCount(keyobj);
allkeys = (pattern[0] == '*' && plen == 1);
while ((de = dbIteratorNext(&dbit)) != NULL) {
sds key = dictGetKey(de);
robj *keyobj;
if (allkeys || stringmatchlen(pattern, plen, key, sdslen(key), 0)) {
keyobj = createStringObject(key, sdslen(key));
if (!keyIsExpired(c->db, keyobj)) {
addReplyBulk(c, keyobj);
numkeys++;
}
decrRefCount(keyobj);
}
if (c->flags & CLIENT_CLOSE_ASAP)
break;}
dictReleaseIterator(di);
break;
}
setDeferredArrayLen(c,replylen,numkeys);
}
......
......@@ -277,7 +277,6 @@ void xorObjectDigest(redisDb *db, robj *keyobj, unsigned char *digest, robj *o)
* a different digest. */
void computeDatasetDigest(unsigned char *final) {
unsigned char digest[20];
dictIterator *di = NULL;
dictEntry *de;
int j;
uint32_t aux;
......@@ -286,39 +285,31 @@ void computeDatasetDigest(unsigned char *final) {
for (j = 0; j < server.dbnum; j++) {
redisDb *db = server.db+j;
int hasEntries = 0;
dict *d;
if (dbSize(db) == 0) continue;
dbIterator dbit;
dbIteratorInit(&dbit, db);
while ((d = dbIteratorNextDict(&dbit))) {
if (dictSize(d) == 0) continue;
hasEntries = 1;
di = dictGetSafeIterator(d);
/* Iterate this DB writing every entry */
while ((de = dictNext(di)) != NULL) {
sds key;
robj *keyobj, *o;
/* hash the DB id, so the same dataset moved in a different DB will lead to a different digest */
aux = htonl(j);
mixDigest(final, &aux, sizeof(aux));
memset(digest, 0, 20); /* This key-val digest */
key = dictGetKey(de);
keyobj = createStringObject(key, sdslen(key));
/* Iterate this DB writing every entry */
while ((de = dbIteratorNext(&dbit)) != NULL) {
sds key;
robj *keyobj, *o;
mixDigest(digest, key, sdslen(key));
memset(digest, 0, 20); /* This key-val digest */
key = dictGetKey(de);
keyobj = createStringObject(key, sdslen(key));
o = dictGetVal(de);
xorObjectDigest(db, keyobj, digest, o);
mixDigest(digest, key, sdslen(key));
/* We can finally xor the key-val digest to the final digest */
xorDigest(final, digest, 20);
decrRefCount(keyobj);
}
dictReleaseIterator(di);
}
if (hasEntries) {
/* hash the DB id, so the same dataset moved in a different DB will lead to a different digest */
aux = htonl(j);
mixDigest(final, &aux, sizeof(aux));
o = dictGetVal(de);
xorObjectDigest(db, keyobj, digest, o);
/* We can finally xor the key-val digest to the final digest */
xorDigest(final, digest, 20);
decrRefCount(keyobj);
}
}
}
......
......@@ -2439,12 +2439,13 @@ typedef struct dbIterator {
redisDb *db;
int index;
int cur_slot;
dictIterator di;
} dbIterator;
/* DB iterator specific functions */
void dbIteratorInit(dbIterator *dbit, redisDb *db);
dict *dbIteratorNextDict(dbIterator *dbit);
int dbIterCurSlot(dbIterator *dbit);
dictEntry *dbIteratorNext(dbIterator *iter);
dict *dbGetNextNonEmptySlot(redisDb *db, int *slot);
/* 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