X-Git-Url: https://git.saurik.com/redis.git/blobdiff_plain/7f00cd226438d3cd91238974346cee624d8920e1..a7b058dae612ad3ab31a2b2d6f07e5ff7fe06618:/src/db.c diff --git a/src/db.c b/src/db.c index ca520c82..354f9045 100644 --- a/src/db.c +++ b/src/db.c @@ -2,39 +2,100 @@ #include +void SlotToKeyAdd(robj *key); +void SlotToKeyDel(robj *key); + /*----------------------------------------------------------------------------- * 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) { robj *val = dictGetEntryVal(de); - 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); - /* Update the access time for the aging algorithm. */ - val->lru = server.lruclock; - } else { - int notify = (val->storage == REDIS_VM_LOADING); + /* Update the access time for the aging algorithm. + * Don't do it if we have a saving child, as this will trigger + * a copy on write madness. */ + if (server.bgsavechildpid == -1 && server.bgrewritechildpid == -1) + val->lru = server.lruclock; + + 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; + } 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)) { + 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; + } - /* Our value was swapped on disk. Bring it at home. */ - redisAssert(val->type == REDIS_VMPOINTER); - val = vmLoadObject(val); - dictGetEntryVal(de) = val; + /* 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); + } - /* Clients blocked by the VM subsystem may be waiting for - * this key... */ - if (notify) handleClientsBlockedOnSwappedKey(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); + redisAssert(retval == REDIS_OK); + if (expire != -1) setExpire(db,key,expire); + server.stat_keyspace_hits++; + return val; + } else { + cacheSetKeyDoesNotExist(db,key); } } - return val; - } else { + server.stat_keyspace_misses++; return NULL; } } @@ -72,6 +133,8 @@ 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); + if (server.cluster_enabled) SlotToKeyAdd(key); return REDIS_OK; } } @@ -81,14 +144,20 @@ 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; + if (server.cluster_enabled) SlotToKeyAdd(key); + 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) { @@ -123,18 +192,27 @@ 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. */ - 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); - return dictDelete(db->dict,key->ptr) == DICT_OK; + if (dictDelete(db->dict,key->ptr) == DICT_OK) { + if (server.cluster_enabled) SlotToKeyDel(key); + return 1; + } else { + return 0; + } } -/* 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; @@ -143,6 +221,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; } @@ -154,27 +233,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++; } @@ -182,10 +284,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); @@ -203,8 +317,12 @@ void existsCommand(redisClient *c) { void selectCommand(redisClient *c) { int id = atoi(c->argv[1]->ptr); + if (server.cluster_enabled && id != 0) { + addReplyError(c,"SELECT is not allowed in cluster mode"); + return; + } if (selectDb(c,id) == REDIS_ERR) { - addReplySds(c,sdsnew("-ERR invalid DB index\r\n")); + addReplyError(c,"invalid DB index"); } else { addReply(c,shared.ok); } @@ -228,11 +346,9 @@ void keysCommand(redisClient *c) { sds pattern = c->argv[1]->ptr; int plen = sdslen(pattern), allkeys; unsigned long numkeys = 0; - robj *lenobj = createObject(REDIS_STRING,NULL); + void *replylen = addDeferredMultiBulkLength(c); di = dictGetIterator(c->db->dict); - addReply(c,lenobj); - decrRefCount(lenobj); allkeys = (pattern[0] == '*' && pattern[1] == '\0'); while((de = dictNext(di)) != NULL) { sds key = dictGetEntryKey(de); @@ -248,17 +364,15 @@ void keysCommand(redisClient *c) { } } dictReleaseIterator(di); - lenobj->ptr = sdscatprintf(sdsempty(),"*%lu\r\n",numkeys); + setDeferredMultiBulkLength(c,replylen,numkeys); } void dbsizeCommand(redisClient *c) { - addReplySds(c, - sdscatprintf(sdsempty(),":%lu\r\n",dictSize(c->db->dict))); + addReplyLongLong(c,dictSize(c->db->dict)); } void lastsaveCommand(redisClient *c) { - addReplySds(c, - sdscatprintf(sdsempty(),":%lu\r\n",server.lastsave)); + addReplyLongLong(c,server.lastsave); } void typeCommand(redisClient *c) { @@ -267,50 +381,24 @@ void typeCommand(redisClient *c) { o = lookupKeyRead(c->db,c->argv[1]); if (o == NULL) { - type = "+none"; + type = "none"; } else { switch(o->type) { - case REDIS_STRING: type = "+string"; break; - case REDIS_LIST: type = "+list"; break; - case REDIS_SET: type = "+set"; break; - case REDIS_ZSET: type = "+zset"; break; - case REDIS_HASH: type = "+hash"; break; - default: type = "+unknown"; break; + case REDIS_STRING: type = "string"; break; + case REDIS_LIST: type = "list"; break; + case REDIS_SET: type = "set"; break; + case REDIS_ZSET: type = "zset"; break; + case REDIS_HASH: type = "hash"; break; + default: type = "unknown"; break; } } - addReplySds(c,sdsnew(type)); - addReply(c,shared.crlf); -} - -void saveCommand(redisClient *c) { - if (server.bgsavechildpid != -1) { - addReplySds(c,sdsnew("-ERR background save in progress\r\n")); - return; - } - if (rdbSave(server.dbfilename) == REDIS_OK) { - addReply(c,shared.ok); - } else { - addReply(c,shared.err); - } -} - -void bgsaveCommand(redisClient *c) { - if (server.bgsavechildpid != -1) { - addReplySds(c,sdsnew("-ERR background save already in progress\r\n")); - return; - } - if (rdbSaveBackground(server.dbfilename) == REDIS_OK) { - char *status = "+Background saving started\r\n"; - addReplySds(c,sdsnew(status)); - } else { - addReply(c,shared.err); - } + addReplyStatus(c,type); } void shutdownCommand(redisClient *c) { if (prepareForShutdown() == REDIS_OK) exit(0); - addReplySds(c, sdsnew("-ERR Errors trying to SHUTDOWN. Check logs.\r\n")); + addReplyError(c,"Errors trying to SHUTDOWN. Check logs."); } void renameGenericCommand(redisClient *c, int nx) { @@ -335,8 +423,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); } @@ -354,6 +442,11 @@ void moveCommand(redisClient *c) { redisDb *src, *dst; int srcid; + if (server.cluster_enabled) { + addReplyError(c,"MOVE is not allowed in cluster mode"); + return; + } + /* Obtain source and target DB pointers */ src = c->db; srcid = c->db->id; @@ -435,16 +528,14 @@ time_t getExpire(redisDb *db, robj *key) { * will be consistent even if we allow write operations against expiring * keys. */ void propagateExpire(redisDb *db, robj *key) { - struct redisCommand *cmd; robj *argv[2]; - cmd = lookupCommand("del"); argv[0] = createStringObject("DEL",3); argv[1] = key; incrRefCount(key); if (server.appendonly) - feedAppendOnlyFile(cmd,db->id,argv,2); + feedAppendOnlyFile(server.delCommand,db->id,argv,2); if (listLength(server.slaves)) replicationFeedSlaves(server.slaves,db->id,argv,2); @@ -455,6 +546,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. @@ -466,14 +559,11 @@ 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; /* Delete the key */ server.stat_expiredkeys++; - server.dirty++; propagateExpire(db,key); return dbDelete(db,key); } @@ -484,7 +574,7 @@ int expireIfNeeded(redisDb *db, robj *key) { void expireGenericCommand(redisClient *c, robj *key, robj *param, long offset) { dictEntry *de; - time_t seconds; + long seconds; if (getLongFromObjectOrReply(c, param, &seconds, NULL) != REDIS_OK) return; @@ -498,13 +588,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; } @@ -521,6 +611,7 @@ void expireatCommand(redisClient *c) { void ttlCommand(redisClient *c) { time_t expire, ttl = -1; + if (server.ds_enabled) lookupKeyRead(c->db,c->argv[1]); expire = getExpire(c->db,c->argv[1]); if (expire != -1) { ttl = (expire-time(NULL)); @@ -544,3 +635,108 @@ void persistCommand(redisClient *c) { } } } + +/* ----------------------------------------------------------------------------- + * API to get key arguments from commands + * ---------------------------------------------------------------------------*/ + +int *getKeysUsingCommandTable(struct redisCommand *cmd,robj **argv, int argc, int *numkeys) { + int j, i = 0, last, *keys; + REDIS_NOTUSED(argv); + + if (cmd->firstkey == 0) { + *numkeys = 0; + return NULL; + } + last = cmd->lastkey; + if (last < 0) last = argc+last; + keys = zmalloc(sizeof(int)*((last - cmd->firstkey)+1)); + for (j = cmd->firstkey; j <= last; j += cmd->keystep) { + redisAssert(j < argc); + keys[i++] = j; + } + *numkeys = i; + return keys; +} + +int *getKeysFromCommand(struct redisCommand *cmd,robj **argv, int argc, int *numkeys, int flags) { + if (cmd->getkeys_proc) { + return cmd->getkeys_proc(cmd,argv,argc,numkeys,flags); + } else { + return getKeysUsingCommandTable(cmd,argv,argc,numkeys); + } +} + +void getKeysFreeResult(int *result) { + zfree(result); +} + +int *noPreloadGetKeys(struct redisCommand *cmd,robj **argv, int argc, int *numkeys, int flags) { + if (flags & REDIS_GETKEYS_PRELOAD) { + *numkeys = 0; + return NULL; + } else { + return getKeysUsingCommandTable(cmd,argv,argc,numkeys); + } +} + +int *renameGetKeys(struct redisCommand *cmd,robj **argv, int argc, int *numkeys, int flags) { + if (flags & REDIS_GETKEYS_PRELOAD) { + int *keys = zmalloc(sizeof(int)); + *numkeys = 1; + keys[0] = 1; + return keys; + } else { + return getKeysUsingCommandTable(cmd,argv,argc,numkeys); + } +} + +int *zunionInterGetKeys(struct redisCommand *cmd,robj **argv, int argc, int *numkeys, int flags) { + int i, num, *keys; + REDIS_NOTUSED(cmd); + REDIS_NOTUSED(flags); + + num = atoi(argv[2]->ptr); + /* Sanity check. Don't return any key if the command is going to + * reply with syntax error. */ + if (num > (argc-3)) { + *numkeys = 0; + return NULL; + } + keys = zmalloc(sizeof(int)*num); + for (i = 0; i < num; i++) keys[i] = 3+i; + *numkeys = num; + return keys; +} + +/* Slot to Key API. This is used by Redis Cluster in order to obtain in + * a fast way a key that belongs to a specified hash slot. This is useful + * while rehashing the cluster. */ +void SlotToKeyAdd(robj *key) { + unsigned int hashslot = keyHashSlot(key->ptr,sdslen(key->ptr)); + + zslInsert(server.cluster.slots_to_keys,hashslot,key); + incrRefCount(key); +} + +void SlotToKeyDel(robj *key) { + unsigned int hashslot = keyHashSlot(key->ptr,sdslen(key->ptr)); + + zslDelete(server.cluster.slots_to_keys,hashslot,key); +} + +unsigned int GetKeysInSlot(unsigned int hashslot, robj **keys, unsigned int count) { + zskiplistNode *n; + zrangespec range; + int j = 0; + + range.min = range.max = hashslot; + range.minex = range.maxex = 0; + + n = zslFirstInRange(server.cluster.slots_to_keys, range); + while(n && n->score == hashslot && count--) { + keys[j++] = n->obj; + n = n->level[0].forward; + } + return j; +}