Skip to content

Commit

Permalink
dict.c: added optional callback to dictEmpty().
Browse files Browse the repository at this point in the history
Redis hash table implementation has many non-blocking features like
incremental rehashing, however while deleting a large hash table there
was no way to have a callback called to do some incremental work.

This commit adds this support, as an optiona callback argument to
dictEmpty() that is currently called at a fixed interval (one time every
65k deletions).
  • Loading branch information
antirez committed Dec 10, 2013
1 parent 26cf5c8 commit b6610a5
Show file tree
Hide file tree
Showing 8 changed files with 22 additions and 21 deletions.
12 changes: 6 additions & 6 deletions src/db.c
Expand Up @@ -166,14 +166,14 @@ int dbDelete(redisDb *db, robj *key) {
}
}

long long emptyDb() {
long long emptyDb(void(callback)(void*)) {
int j;
long long removed = 0;

for (j = 0; j < server.dbnum; j++) {
removed += dictSize(server.db[j].dict);
dictEmpty(server.db[j].dict);
dictEmpty(server.db[j].expires);
dictEmpty(server.db[j].dict,callback);
dictEmpty(server.db[j].expires,callback);
}
return removed;
}
Expand Down Expand Up @@ -209,14 +209,14 @@ void signalFlushedDb(int dbid) {
void flushdbCommand(redisClient *c) {
server.dirty += dictSize(c->db->dict);
signalFlushedDb(c->db->id);
dictEmpty(c->db->dict);
dictEmpty(c->db->expires);
dictEmpty(c->db->dict,NULL);
dictEmpty(c->db->expires,NULL);
addReply(c,shared.ok);
}

void flushallCommand(redisClient *c) {
signalFlushedDb(-1);
server.dirty += emptyDb();
server.dirty += emptyDb(NULL);
addReply(c,shared.ok);
if (server.rdb_child_pid != -1) {
kill(server.rdb_child_pid,SIGUSR1);
Expand Down
4 changes: 2 additions & 2 deletions src/debug.c
Expand Up @@ -261,15 +261,15 @@ void debugCommand(redisClient *c) {
addReply(c,shared.err);
return;
}
emptyDb();
emptyDb(NULL);
if (rdbLoad(server.rdb_filename) != REDIS_OK) {
addReplyError(c,"Error trying to load the RDB dump");
return;
}
redisLog(REDIS_WARNING,"DB reloaded by DEBUG RELOAD");
addReply(c,shared.ok);
} else if (!strcasecmp(c->argv[1]->ptr,"loadaof")) {
emptyDb();
emptyDb(NULL);
if (loadAppendOnlyFile(server.aof_filename) != REDIS_OK) {
addReply(c,shared.err);
return;
Expand Down
15 changes: 8 additions & 7 deletions src/dict.c
Expand Up @@ -444,14 +444,15 @@ int dictDeleteNoFree(dict *ht, const void *key) {
}

/* Destroy an entire dictionary */
int _dictClear(dict *d, dictht *ht)
{
int _dictClear(dict *d, dictht *ht, void(callback)(void *)) {
unsigned long i;

/* Free all the elements */
for (i = 0; i < ht->size && ht->used > 0; i++) {
dictEntry *he, *nextHe;

if (callback && (i & 65535) == 0) callback(d->privdata);

if ((he = ht->table[i]) == NULL) continue;
while(he) {
nextHe = he->next;
Expand All @@ -472,8 +473,8 @@ int _dictClear(dict *d, dictht *ht)
/* Clear & Release the hash table */
void dictRelease(dict *d)
{
_dictClear(d,&d->ht[0]);
_dictClear(d,&d->ht[1]);
_dictClear(d,&d->ht[0],NULL);
_dictClear(d,&d->ht[1],NULL);
zfree(d);
}

Expand Down Expand Up @@ -882,9 +883,9 @@ static int _dictKeyIndex(dict *d, const void *key)
return idx;
}

void dictEmpty(dict *d) {
_dictClear(d,&d->ht[0]);
_dictClear(d,&d->ht[1]);
void dictEmpty(dict *d, void(callback)(void*)) {
_dictClear(d,&d->ht[0],callback);
_dictClear(d,&d->ht[1],callback);
d->rehashidx = -1;
d->iterators = 0;
}
Expand Down
2 changes: 1 addition & 1 deletion src/dict.h
Expand Up @@ -160,7 +160,7 @@ dictEntry *dictGetRandomKey(dict *d);
void dictPrintStats(dict *d);
unsigned int dictGenHashFunction(const void *key, int len);
unsigned int dictGenCaseHashFunction(const unsigned char *buf, int len);
void dictEmpty(dict *d);
void dictEmpty(dict *d, void(callback)(void*));
void dictEnableResize(void);
void dictDisableResize(void);
int dictRehash(dict *d, int n);
Expand Down
2 changes: 1 addition & 1 deletion src/redis.h
Expand Up @@ -1165,7 +1165,7 @@ void setKey(redisDb *db, robj *key, robj *val);
int dbExists(redisDb *db, robj *key);
robj *dbRandomKey(redisDb *db);
int dbDelete(redisDb *db, robj *key);
long long emptyDb();
long long emptyDb(void(callback)(void*));
int selectDb(redisClient *c, int id);
void signalModifiedKey(redisDb *db, robj *key);
void signalFlushedDb(int dbid);
Expand Down
4 changes: 2 additions & 2 deletions src/replication.c
Expand Up @@ -782,7 +782,7 @@ void readSyncBulkPayload(aeEventLoop *el, int fd, void *privdata, int mask) {
}
redisLog(REDIS_NOTICE, "MASTER <-> SLAVE sync: Flushing old data");
signalFlushedDb(-1);
emptyDb();
emptyDb(NULL);
/* Before loading the DB into memory we need to delete the readable
* handler, otherwise it will get called recursively since
* rdbLoad() will call the event loop to process events from time to
Expand Down Expand Up @@ -1445,7 +1445,7 @@ void replicationScriptCacheInit(void) {
* to reclaim otherwise unused memory.
*/
void replicationScriptCacheFlush(void) {
dictEmpty(server.repl_scriptcache_dict);
dictEmpty(server.repl_scriptcache_dict,NULL);
listRelease(server.repl_scriptcache_fifo);
server.repl_scriptcache_fifo = listCreate();
}
Expand Down
2 changes: 1 addition & 1 deletion src/sentinel.c
Expand Up @@ -393,7 +393,7 @@ void initSentinel(void) {

/* Remove usual Redis commands from the command table, then just add
* the SENTINEL command. */
dictEmpty(server.commands);
dictEmpty(server.commands,NULL);
for (j = 0; j < sizeof(sentinelcmds)/sizeof(sentinelcmds[0]); j++) {
int retval;
struct redisCommand *cmd = sentinelcmds+j;
Expand Down
2 changes: 1 addition & 1 deletion src/t_list.c
Expand Up @@ -837,7 +837,7 @@ void unblockClientWaitingData(redisClient *c) {
dictReleaseIterator(di);

/* Cleanup the client structure */
dictEmpty(c->bpop.keys);
dictEmpty(c->bpop.keys,NULL);
if (c->bpop.target) {
decrRefCount(c->bpop.target);
c->bpop.target = NULL;
Expand Down

0 comments on commit b6610a5

Please sign in to comment.