X-Git-Url: https://git.saurik.com/redis.git/blobdiff_plain/1b1f47c915c69eae40d99727267b147f7c5a44ac..c15a3887e08d468b96d4313cc19862b5e4b09977:/src/db.c diff --git a/src/db.c b/src/db.c index aa1c14ad..ae40d204 100644 --- a/src/db.c +++ b/src/db.c @@ -17,29 +17,44 @@ robj *lookupKey(redisDb *db, robj *key) { if (server.bgsavechildpid == -1 && server.bgrewritechildpid == -1) val->lru = server.lruclock; - if (server.vm_enabled) { - if (val->storage == REDIS_VM_MEMORY || - val->storage == REDIS_VM_SWAPPING) - { - /* If we were swapping the object out, cancel the operation */ - if (val->storage == REDIS_VM_SWAPPING) - vmCancelThreadedIOJob(val); - } else { - int notify = (val->storage == REDIS_VM_LOADING); - - /* Our value was swapped on disk. Bring it at home. */ - redisAssert(val->type == REDIS_VMPOINTER); - val = vmLoadObject(val); - dictGetEntryVal(de) = val; - - /* Clients blocked by the VM subsystem may be waiting for - * this key... */ - if (notify) handleClientsBlockedOnSwappedKey(db,key); - } + if (server.ds_enabled && + cacheScheduleIOGetFlags(db,key) & REDIS_IO_SAVEINPROG) + { + /* There is a save in progress for this object! + * Wait for it to get out. */ + waitEmptyIOJobsQueue(); + processAllPendingIOJobs(); + redisAssert(!(cacheScheduleIOGetFlags(db,key) & REDIS_IO_SAVEINPROG)); } server.stat_keyspace_hits++; return val; } else { + time_t expire; + robj *val; + + /* 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. */ + if (server.ds_enabled && cacheKeyMayExist(db,key)) { + if (cacheScheduleIOGetFlags(db,key) & REDIS_IO_SAVEINPROG) { + /* There is a save in progress for this object! + * Wait for it to get out. */ + waitEmptyIOJobsQueue(); + processAllPendingIOJobs(); + redisAssert((cacheScheduleIOGetFlags(db,key) & REDIS_IO_SAVEINPROG) == 0); + } + + redisLog(REDIS_DEBUG,"Force loading key %s via lookup", + key->ptr); + val = dsGet(db,key,&expire); + if (val) { + int retval = dbAdd(db,key,val); + redisAssert(retval == REDIS_OK); + if (expire != -1) setExpire(db,key,expire); + server.stat_keyspace_hits++; + return val; + } + } server.stat_keyspace_misses++; return NULL; } @@ -78,6 +93,7 @@ int dbAdd(redisDb *db, robj *key, robj *val) { } else { sds copy = sdsdup(key->ptr); dictAdd(db->dict, copy, val); + if (server.ds_enabled) cacheSetKeyMayExist(db,key); return REDIS_OK; } } @@ -87,14 +103,19 @@ 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) { - if (dictFind(db->dict,key->ptr) == NULL) { + 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 { 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) { @@ -129,11 +150,14 @@ robj *dbRandomKey(redisDb *db) { /* Delete a key, value, and associated expiration entry if any, from the DB */ int dbDelete(redisDb *db, robj *key) { - /* If VM is enabled make sure to awake waiting clients for this key: - * deleting the key will kill the I/O thread bringing the key from swap - * to memory, so the client will never be notified and unblocked if we - * don't do it now. */ - if (server.vm_enabled) handleClientsBlockedOnSwappedKey(db,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); + 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. */ if (dictSize(db->expires) > 0) dictDelete(db->expires,key->ptr); @@ -160,27 +184,50 @@ int selectDb(redisClient *c, int id) { return REDIS_OK; } +/*----------------------------------------------------------------------------- + * Hooks for key space changes. + * + * Every time a key in the database is modified the function + * signalModifiedKey() is called. + * + * Every time a DB is flushed the function signalFlushDb() is called. + *----------------------------------------------------------------------------*/ + +void signalModifiedKey(redisDb *db, robj *key) { + touchWatchedKey(db,key); + if (server.ds_enabled) + cacheScheduleIO(db,key,REDIS_IO_SAVE); +} + +void signalFlushedDb(int dbid) { + touchWatchedKeysOnFlush(dbid); +} + /*----------------------------------------------------------------------------- * Type agnostic commands operating on the key space *----------------------------------------------------------------------------*/ void flushdbCommand(redisClient *c) { server.dirty += dictSize(c->db->dict); - touchWatchedKeysOnFlush(c->db->id); + signalFlushedDb(c->db->id); dictEmpty(c->db->dict); dictEmpty(c->db->expires); + if (server.ds_enabled) dsFlushDb(c->db->id); addReply(c,shared.ok); } void flushallCommand(redisClient *c) { - touchWatchedKeysOnFlush(-1); + signalFlushedDb(-1); server.dirty += emptyDb(); addReply(c,shared.ok); if (server.bgsavechildpid != -1) { kill(server.bgsavechildpid,SIGKILL); rdbRemoveTempFile(server.bgsavechildpid); } - rdbSave(server.dbfilename); + if (server.ds_enabled) + dsFlushDb(-1); + else + rdbSave(server.dbfilename); server.dirty++; } @@ -188,10 +235,22 @@ void delCommand(redisClient *c) { int deleted = 0, j; for (j = 1; j < c->argc; j++) { + if (server.ds_enabled) { + lookupKeyRead(c->db,c->argv[j]); + /* FIXME: this can be optimized a lot, no real need to load + * a possibly huge value. */ + } if (dbDelete(c->db,c->argv[j])) { - touchWatchedKey(c->db,c->argv[j]); + signalModifiedKey(c->db,c->argv[j]); server.dirty++; deleted++; + } else if (server.ds_enabled) { + if (cacheKeyMayExist(c->db,c->argv[j]) && + dsExists(c->db,c->argv[j])) + { + cacheScheduleIO(c->db,c->argv[j],REDIS_IO_SAVE); + deleted = 1; + } } } addReplyLongLong(c,deleted); @@ -335,8 +394,8 @@ void renameGenericCommand(redisClient *c, int nx) { dbReplace(c->db,c->argv[2],o); } dbDelete(c->db,c->argv[1]); - touchWatchedKey(c->db,c->argv[1]); - touchWatchedKey(c->db,c->argv[2]); + signalModifiedKey(c->db,c->argv[1]); + signalModifiedKey(c->db,c->argv[2]); server.dirty++; addReply(c,nx ? shared.cone : shared.ok); } @@ -495,13 +554,13 @@ void expireGenericCommand(redisClient *c, robj *key, robj *param, long offset) { if (seconds <= 0) { if (dbDelete(c->db,key)) server.dirty++; addReply(c, shared.cone); - touchWatchedKey(c->db,key); + signalModifiedKey(c->db,key); return; } else { time_t when = time(NULL)+seconds; setExpire(c->db,key,when); addReply(c,shared.cone); - touchWatchedKey(c->db,key); + signalModifiedKey(c->db,key); server.dirty++; return; }