* 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) {
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. */
- 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;
/* 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)) {
+ 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);
if (expire != -1) setExpire(db,key,expire);
server.stat_keyspace_hits++;
return val;
+ } else {
+ cacheSetKeyDoesNotExist(db,key);
}
}
server.stat_keyspace_misses++;
} 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;
}
}
* 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) {
/* 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.ds_enabled) handleClientsBlockedOnSwappedKey(db,key);
-
- /* FIXME: we need to delete the IO Job loading the key, or simply we can
- * wait for it to finish. */
+ /* 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. */
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;
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;
}
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);
}
/*-----------------------------------------------------------------------------
signalFlushedDb(c->db->id);
dictEmpty(c->db->dict);
dictEmpty(c->db->expires);
+ if (server.ds_enabled) dsFlushDb(c->db->id);
addReply(c,shared.ok);
}
kill(server.bgsavechildpid,SIGKILL);
rdbRemoveTempFile(server.bgsavechildpid);
}
- rdbSave(server.dbfilename);
+ if (server.ds_enabled)
+ dsFlushDb(-1);
+ else
+ rdbSave(server.dbfilename);
server.dirty++;
}
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;
}
}
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);
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.
return time(NULL) > when;
}
- if (when < 0) return 0;
-
/* Return when this key has not expired */
if (time(NULL) <= when) return 0;
projects / redis.git / blobdiff