X-Git-Url: https://git.saurik.com/redis.git/blobdiff_plain/aa81e4d5f4b4d942b9340fd2d401e7795fa9fd1e..3a73be752476f44dfb099863ac9424c2f27f2be0:/src/db.c diff --git a/src/db.c b/src/db.c index 1f185e29..669fba13 100644 --- a/src/db.c +++ b/src/db.c @@ -6,6 +6,35 @@ * C-level DB API *----------------------------------------------------------------------------*/ +/* Important notes on lookup and disk store. + * + * When disk store is enabled on lookup we can have different cases. + * + * a) The key is in memory: + * - If the key is not in IO_SAVEINPROG state we can access it. + * As if it's just IO_SAVE this means we have the key in the IO queue + * but can't be accessed by the IO thread (it requires to be + * translated into an IO Job by the cache cron function.) + * - If the key is in IO_SAVEINPROG we can't touch the key and have + * to blocking wait completion of operations. + * b) The key is not in memory: + * - If it's marked as non existing on disk as well (negative cache) + * we don't need to perform the disk access. + * - if the key MAY EXIST, but is not in memory, and it is marked as IO_SAVE + * then the key can only be a deleted one. As IO_SAVE keys are never + * evicted (dirty state), so the only possibility is that key was deleted. + * - if the key MAY EXIST we need to blocking load it. + * We check that the key is not in IO_SAVEINPROG state before accessing + * the disk object. If it is in this state, we wait. + */ + +void lookupWaitBusyKey(redisDb *db, robj *key) { + /* FIXME: wait just for this key, not everything */ + waitEmptyIOJobsQueue(); + processAllPendingIOJobs(); + redisAssert((cacheScheduleIOGetFlags(db,key) & REDIS_IO_SAVEINPROG) == 0); +} + robj *lookupKey(redisDb *db, robj *key) { dictEntry *de = dictFind(db->dict,key->ptr); if (de) { @@ -17,12 +46,12 @@ robj *lookupKey(redisDb *db, robj *key) { if (server.bgsavechildpid == -1 && server.bgrewritechildpid == -1) val->lru = server.lruclock; - if (server.ds_enabled && val->storage == REDIS_DS_SAVING) { - /* FIXME: change this code to just wait for our object to - * get out of the IO Job. As it is now it is correct but slow. */ - waitEmptyIOJobsQueue(); - processAllPendingIOJobs(); - redisAssert(val->storage != REDIS_DS_SAVING); + if (server.ds_enabled && + cacheScheduleIOGetFlags(db,key) & REDIS_IO_SAVEINPROG) + { + /* Need to wait for the key to get unbusy */ + redisLog(REDIS_DEBUG,"Lookup found a key in SAVEINPROG state. Waiting. (Key was in the cache)"); + lookupWaitBusyKey(db,key); } server.stat_keyspace_hits++; return val; @@ -32,14 +61,26 @@ robj *lookupKey(redisDb *db, robj *key) { /* Key not found in the in memory hash table, but if disk store is * enabled we may have this key on disk. If so load it in memory - * in a blocking way. - * - * FIXME: race condition here. If there was an already scheduled - * async loading of this key, what may happen is that the old - * key is loaded in memory if this gets deleted in the meantime. */ + * in a blocking way. */ if (server.ds_enabled && cacheKeyMayExist(db,key)) { - redisLog(REDIS_DEBUG,"Force loading key %s via lookup", - key->ptr); + long flags = cacheScheduleIOGetFlags(db,key); + + /* They key is not in cache, but it has a SAVE op in queue? + * The only possibility is that the key was deleted, since + * dirty keys are not evicted. */ + if (flags & REDIS_IO_SAVE) { + server.stat_keyspace_misses++; + return NULL; + } + + /* At this point we need to blocking load the key in memory. + * The first thing we do is waiting here if the key is busy. */ + if (flags & REDIS_IO_SAVEINPROG) { + redisLog(REDIS_DEBUG,"Lookup found a key in SAVEINPROG state. Waiting (while force loading)."); + lookupWaitBusyKey(db,key); + } + + redisLog(REDIS_DEBUG,"Force loading key %s via lookup", key->ptr); val = dsGet(db,key,&expire); if (val) { int retval = dbAdd(db,key,val); @@ -47,6 +88,8 @@ robj *lookupKey(redisDb *db, robj *key) { if (expire != -1) setExpire(db,key,expire); server.stat_keyspace_hits++; return val; + } else { + cacheSetKeyDoesNotExist(db,key); } } server.stat_keyspace_misses++; @@ -87,9 +130,7 @@ int dbAdd(redisDb *db, robj *key, robj *val) { } else { sds copy = sdsdup(key->ptr); dictAdd(db->dict, copy, val); - if (server.ds_enabled) { - /* FIXME: remove entry from negative cache */ - } + if (server.ds_enabled) cacheSetKeyMayExist(db,key); return REDIS_OK; } } @@ -100,16 +141,18 @@ int dbAdd(redisDb *db, robj *key, robj *val) { * On update (key already existed) 0 is returned. Otherwise 1. */ int dbReplace(redisDb *db, robj *key, robj *val) { robj *oldval; + int retval; if ((oldval = dictFetchValue(db->dict,key->ptr)) == NULL) { sds copy = sdsdup(key->ptr); dictAdd(db->dict, copy, val); - return 1; + retval = 1; } else { - val->storage = oldval->storage; dictReplace(db->dict, key->ptr, val); - return 0; + retval = 0; } + if (server.ds_enabled) cacheSetKeyMayExist(db,key); + return retval; } int dbExists(redisDb *db, robj *key) { @@ -147,10 +190,10 @@ int dbDelete(redisDb *db, robj *key) { /* If diskstore is enabled make sure to awake waiting clients for this key * as it is not really useful to wait for a key already deleted to be * loaded from disk. */ - if (server.ds_enabled) handleClientsBlockedOnSwappedKey(db,key); - - /* Mark this key as non existing on disk as well */ - cacheSetKeyDoesNotExistRemember(db,key); + if (server.ds_enabled) { + handleClientsBlockedOnSwappedKey(db,key); + cacheSetKeyDoesNotExist(db,key); + } /* Deleting an entry from the expires dict will not free the sds of * the key, because it is shared with the main dictionary. */ @@ -158,7 +201,8 @@ int dbDelete(redisDb *db, robj *key) { return dictDelete(db->dict,key->ptr) == DICT_OK; } -/* Empty the whole database */ +/* Empty the whole database. + * If diskstore is enabled this function will just flush the in-memory cache. */ long long emptyDb() { int j; long long removed = 0; @@ -167,6 +211,7 @@ long long emptyDb() { removed += dictSize(server.db[j].dict); dictEmpty(server.db[j].dict); dictEmpty(server.db[j].expires); + if (server.ds_enabled) dictEmpty(server.db[j].io_negcache); } return removed; } @@ -190,13 +235,11 @@ int selectDb(redisClient *c, int id) { void signalModifiedKey(redisDb *db, robj *key) { touchWatchedKey(db,key); if (server.ds_enabled) - cacheScheduleForFlush(db,key); + cacheScheduleIO(db,key,REDIS_IO_SAVE); } void signalFlushedDb(int dbid) { touchWatchedKeysOnFlush(dbid); - if (server.ds_enabled) - dsFlushDb(dbid); } /*----------------------------------------------------------------------------- @@ -208,6 +251,7 @@ void flushdbCommand(redisClient *c) { signalFlushedDb(c->db->id); dictEmpty(c->db->dict); dictEmpty(c->db->expires); + if (server.ds_enabled) dsFlushDb(c->db->id); addReply(c,shared.ok); } @@ -219,7 +263,10 @@ void flushallCommand(redisClient *c) { kill(server.bgsavechildpid,SIGKILL); rdbRemoveTempFile(server.bgsavechildpid); } - rdbSave(server.dbfilename); + if (server.ds_enabled) + dsFlushDb(-1); + else + rdbSave(server.dbfilename); server.dirty++; } @@ -240,7 +287,7 @@ void delCommand(redisClient *c) { if (cacheKeyMayExist(c->db,c->argv[j]) && dsExists(c->db,c->argv[j])) { - cacheScheduleForFlush(c->db,c->argv[j]); + cacheScheduleIO(c->db,c->argv[j],REDIS_IO_SAVE); deleted = 1; } } @@ -334,30 +381,6 @@ void typeCommand(redisClient *c) { addReplyStatus(c,type); } -void saveCommand(redisClient *c) { - if (server.bgsavechildpid != -1) { - addReplyError(c,"Background save already in progress"); - return; - } - if (rdbSave(server.dbfilename) == REDIS_OK) { - addReply(c,shared.ok); - } else { - addReply(c,shared.err); - } -} - -void bgsaveCommand(redisClient *c) { - if (server.bgsavechildpid != -1) { - addReplyError(c,"Background save already in progress"); - return; - } - if (rdbSaveBackground(server.dbfilename) == REDIS_OK) { - addReplyStatus(c,"Background saving started"); - } else { - addReply(c,shared.err); - } -} - void shutdownCommand(redisClient *c) { if (prepareForShutdown() == REDIS_OK) exit(0); @@ -504,6 +527,8 @@ void propagateExpire(redisDb *db, robj *key) { int expireIfNeeded(redisDb *db, robj *key) { time_t when = getExpire(db,key); + if (when < 0) return 0; /* No expire for this key */ + /* If we are running in the context of a slave, return ASAP: * the slave key expiration is controlled by the master that will * send us synthesized DEL operations for expired keys. @@ -515,8 +540,6 @@ int expireIfNeeded(redisDb *db, robj *key) { return time(NULL) > when; } - if (when < 0) return 0; - /* Return when this key has not expired */ if (time(NULL) <= when) return 0;