X-Git-Url: https://git.saurik.com/redis.git/blobdiff_plain/8d51fb6a80fb40abd0bb487d71435d3a30c2942e..c0ba9ebe13865189a3b21f7be8a910e349b59fda:/src/db.c diff --git a/src/db.c b/src/db.c index f3606077..2bc8d7c4 100644 --- a/src/db.c +++ b/src/db.c @@ -2,33 +2,48 @@ #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. + */ + robj *lookupKey(redisDb *db, robj *key) { dictEntry *de = dictFind(db->dict,key->ptr); if (de) { - robj *val = dictGetEntryVal(de); + robj *val = dictGetVal(de); /* 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 && 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); - } server.stat_keyspace_hits++; return val; } else { - /* FIXME: Check if the object is on disk, if it is, load it - * in a blocking way now. */ server.stat_keyspace_misses++; return NULL; } @@ -56,37 +71,45 @@ robj *lookupKeyWriteOrReply(redisClient *c, robj *key, robj *reply) { return o; } -/* Add the key to the DB. If the key already exists REDIS_ERR is returned, - * otherwise REDIS_OK is returned, and the caller should increment the - * refcount of 'val'. */ -int dbAdd(redisDb *db, robj *key, robj *val) { - /* Perform a lookup before adding the key, as we need to copy the - * key value. */ - if (dictFind(db->dict, key->ptr) != NULL) { - return REDIS_ERR; - } else { - sds copy = sdsdup(key->ptr); - dictAdd(db->dict, copy, val); - if (server.ds_enabled) { - /* FIXME: remove entry from negative cache */ - } - return REDIS_OK; - } +/* Add the key to the DB. It's up to the caller to increment the reference + * counte of the value if needed. + * + * The program is aborted if the key already exists. */ +void dbAdd(redisDb *db, robj *key, robj *val) { + sds copy = sdsdup(key->ptr); + int retval = dictAdd(db->dict, copy, val); + + redisAssertWithInfo(NULL,key,retval == REDIS_OK); + if (server.cluster_enabled) SlotToKeyAdd(key); + } + +/* Overwrite an existing key with a new value. Incrementing the reference + * count of the new value is up to the caller. + * This function does not modify the expire time of the existing key. + * + * The program is aborted if the key was not already present. */ +void dbOverwrite(redisDb *db, robj *key, robj *val) { + struct dictEntry *de = dictFind(db->dict,key->ptr); + + redisAssertWithInfo(NULL,key,de != NULL); + dictReplace(db->dict, key->ptr, val); } -/* If the key does not exist, this is just like dbAdd(). Otherwise - * the value associated to the key is replaced with the new one. +/* High level Set operation. This function can be used in order to set + * a key, whatever it was existing or not, to a new object. * - * 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) { - sds copy = sdsdup(key->ptr); - dictAdd(db->dict, copy, val); - return 1; + * 1) The ref count of the value object is incremented. + * 2) clients WATCHing for the destination key notified. + * 3) The expire time of the key is reset (the key is made persistent). */ +void setKey(redisDb *db, robj *key, robj *val) { + if (lookupKeyWrite(db,key) == NULL) { + dbAdd(db,key,val); } else { - dictReplace(db->dict, key->ptr, val); - return 0; + dbOverwrite(db,key,val); } + incrRefCount(val); + removeExpire(db,key); + touchWatchedKey(db,key); } int dbExists(redisDb *db, robj *key) { @@ -107,7 +130,7 @@ robj *dbRandomKey(redisDb *db) { de = dictGetRandomKey(db->dict); if (de == NULL) return NULL; - key = dictGetEntryKey(de); + key = dictGetKey(de); keyobj = createStringObject(key,sdslen(key)); if (dictFind(db->expires,key)) { if (expireIfNeeded(db,keyobj)) { @@ -121,22 +144,19 @@ 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.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. */ - /* 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; @@ -167,14 +187,10 @@ int selectDb(redisClient *c, int id) { void signalModifiedKey(redisDb *db, robj *key) { touchWatchedKey(db,key); - if (server.ds_enabled) - cacheScheduleForFlush(db,key); } void signalFlushedDb(int dbid) { touchWatchedKeysOnFlush(dbid); - if (server.ds_enabled) - dsFlushDb(dbid); } /*----------------------------------------------------------------------------- @@ -197,7 +213,13 @@ void flushallCommand(redisClient *c) { kill(server.bgsavechildpid,SIGKILL); rdbRemoveTempFile(server.bgsavechildpid); } - rdbSave(server.dbfilename); + if (server.saveparamslen > 0) { + /* Normally rdbSave() will reset dirty, but we don't want this here + * as otherwise FLUSHALL will not be replicated nor put into the AOF. */ + int saved_dirty = server.dirty; + rdbSave(server.dbfilename); + server.dirty = saved_dirty; + } server.dirty++; } @@ -226,6 +248,10 @@ 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) { addReplyError(c,"invalid DB index"); } else { @@ -256,7 +282,7 @@ void keysCommand(redisClient *c) { di = dictGetIterator(c->db->dict); allkeys = (pattern[0] == '*' && pattern[1] == '\0'); while((de = dictNext(di)) != NULL) { - sds key = dictGetEntryKey(de); + sds key = dictGetKey(de); robj *keyobj; if (allkeys || stringmatchlen(pattern,plen,key,sdslen(key),0)) { @@ -300,30 +326,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); @@ -332,6 +334,7 @@ void shutdownCommand(redisClient *c) { void renameGenericCommand(redisClient *c, int nx) { robj *o; + time_t expire; /* To use the same key as src and dst is probably an error */ if (sdscmp(c->argv[1]->ptr,c->argv[2]->ptr) == 0) { @@ -343,14 +346,18 @@ void renameGenericCommand(redisClient *c, int nx) { return; incrRefCount(o); - if (dbAdd(c->db,c->argv[2],o) == REDIS_ERR) { + expire = getExpire(c->db,c->argv[1]); + if (lookupKeyWrite(c->db,c->argv[2]) != NULL) { if (nx) { decrRefCount(o); addReply(c,shared.czero); return; } - dbReplace(c->db,c->argv[2],o); + /* Overwrite: delete the old key before creating the new one with the same name. */ + dbDelete(c->db,c->argv[2]); } + dbAdd(c->db,c->argv[2],o); + if (expire != -1) setExpire(c->db,c->argv[2],expire); dbDelete(c->db,c->argv[1]); signalModifiedKey(c->db,c->argv[1]); signalModifiedKey(c->db,c->argv[2]); @@ -371,6 +378,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; @@ -395,11 +407,12 @@ void moveCommand(redisClient *c) { return; } - /* Try to add the element to the target DB */ - if (dbAdd(dst,c->argv[1],o) == REDIS_ERR) { + /* Return zero if the key already exists in the target DB */ + if (lookupKeyWrite(dst,c->argv[1]) != NULL) { addReply(c,shared.czero); return; } + dbAdd(dst,c->argv[1],o); incrRefCount(o); /* OK! key moved, free the entry in the source DB */ @@ -415,7 +428,7 @@ void moveCommand(redisClient *c) { int removeExpire(redisDb *db, robj *key) { /* An expire may only be removed if there is a corresponding entry in the * main dict. Otherwise, the key will never be freed. */ - redisAssert(dictFind(db->dict,key->ptr) != NULL); + redisAssertWithInfo(NULL,key,dictFind(db->dict,key->ptr) != NULL); return dictDelete(db->expires,key->ptr) == DICT_OK; } @@ -424,8 +437,8 @@ void setExpire(redisDb *db, robj *key, time_t when) { /* Reuse the sds from the main dict in the expire dict */ de = dictFind(db->dict,key->ptr); - redisAssert(de != NULL); - dictReplace(db->expires,dictGetEntryKey(de),(void*)when); + redisAssertWithInfo(NULL,key,de != NULL); + dictReplace(db->expires,dictGetKey(de),(void*)when); } /* Return the expire time of the specified key, or -1 if no expire @@ -439,8 +452,8 @@ time_t getExpire(redisDb *db, robj *key) { /* The entry was found in the expire dict, this means it should also * be present in the main dict (safety check). */ - redisAssert(dictFind(db->dict,key->ptr) != NULL); - return (time_t) dictGetEntryVal(de); + redisAssertWithInfo(NULL,key,dictFind(db->dict,key->ptr) != NULL); + return (time_t) dictGetVal(de); } /* Propagate expires into slaves and the AOF file. @@ -470,6 +483,11 @@ 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 */ + + /* Don't expire anything while loading. It will be done later. */ + if (server.loading) return 0; + /* 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. @@ -481,8 +499,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; @@ -509,10 +525,24 @@ void expireGenericCommand(redisClient *c, robj *key, robj *param, long offset) { addReply(c,shared.czero); return; } - if (seconds <= 0) { - if (dbDelete(c->db,key)) server.dirty++; - addReply(c, shared.cone); + /* EXPIRE with negative TTL, or EXPIREAT with a timestamp into the past + * should never be executed as a DEL when load the AOF or in the context + * of a slave instance. + * + * Instead we take the other branch of the IF statement setting an expire + * (possibly in the past) and wait for an explicit DEL from the master. */ + if (seconds <= 0 && !server.loading && !server.masterhost) { + robj *aux; + + redisAssertWithInfo(c,key,dbDelete(c->db,key)); + server.dirty++; + + /* Replicate/AOF this as an explicit DEL. */ + aux = createStringObject("DEL",3); + rewriteClientCommandVector(c,2,aux,key); + decrRefCount(aux); signalModifiedKey(c->db,key); + addReply(c, shared.cone); return; } else { time_t when = time(NULL)+seconds; @@ -558,3 +588,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; +}