/* * Copyright (c) 2009-2012, Salvatore Sanfilippo * All rights reserved. * * Redistribution and use in source and binary forms, with or without * modification, are permitted provided that the following conditions are met: * * * Redistributions of source code must retain the above copyright notice, * this list of conditions and the following disclaimer. * * Redistributions in binary form must reproduce the above copyright * notice, this list of conditions and the following disclaimer in the * documentation and/or other materials provided with the distribution. * * Neither the name of Redis nor the names of its contributors may be used * to endorse or promote products derived from this software without * specific prior written permission. * * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE * POSSIBILITY OF SUCH DAMAGE. */ #include "server.h" #include "cluster.h" #include "atomicvar.h" #include "latency.h" #include "script.h" #include "functions.h" #include #include /*----------------------------------------------------------------------------- * C-level DB API *----------------------------------------------------------------------------*/ int expireIfNeeded(redisDb *db, robj *key, int force_delete_expired); int keyIsExpired(redisDb *db, robj *key); /* Update LFU when an object is accessed. * Firstly, decrement the counter if the decrement time is reached. * Then logarithmically increment the counter, and update the access time. */ void updateLFU(robj *val) { unsigned long counter = LFUDecrAndReturn(val); counter = LFULogIncr(counter); val->lru = (LFUGetTimeInMinutes()<<8) | counter; } /* Lookup a key for read or write operations, or return NULL if the key is not * found in the specified DB. This function implements the functionality of * lookupKeyRead(), lookupKeyWrite() and their ...WithFlags() variants. * * Side-effects of calling this function: * * 1. A key gets expired if it reached it's TTL. * 2. The key's last access time is updated. * 3. The global keys hits/misses stats are updated (reported in INFO). * 4. If keyspace notifications are enabled, a "keymiss" notification is fired. * * Flags change the behavior of this command: * * LOOKUP_NONE (or zero): No special flags are passed. * LOOKUP_NOTOUCH: Don't alter the last access time of the key. * LOOKUP_NONOTIFY: Don't trigger keyspace event on key miss. * LOOKUP_NOSTATS: Don't increment key hits/misses counters. * LOOKUP_WRITE: Prepare the key for writing (delete expired keys even on * replicas, use separate keyspace stats and events (TODO)). * * Note: this function also returns NULL if the key is logically expired but * still existing, in case this is a replica and the LOOKUP_WRITE is not set. * Even if the key expiry is master-driven, we can correctly report a key is * expired on replicas even if the master is lagging expiring our key via DELs * in the replication link. */ robj *lookupKey(redisDb *db, robj *key, int flags) { dictEntry *de = dictFind(db->dict,key->ptr); robj *val = NULL; if (de) { val = dictGetVal(de); /* Forcing deletion of expired keys on a replica makes the replica * inconsistent with the master. We forbid it on readonly replicas, but * we have to allow it on writable replicas to make write commands * behave consistently. * * It's possible that the WRITE flag is set even during a readonly * command, since the command may trigger events that cause modules to * perform additional writes. */ int is_ro_replica = server.masterhost && server.repl_slave_ro; int force_delete_expired = flags & LOOKUP_WRITE && !is_ro_replica; if (expireIfNeeded(db, key, force_delete_expired)) { /* The key is no longer valid. */ val = NULL; } } if (val) { /* Update the access time for the ageing algorithm. * Don't do it if we have a saving child, as this will trigger * a copy on write madness. */ if (!hasActiveChildProcess() && !(flags & LOOKUP_NOTOUCH)){ if (server.maxmemory_policy & MAXMEMORY_FLAG_LFU) { updateLFU(val); } else { val->lru = LRU_CLOCK(); } } if (!(flags & (LOOKUP_NOSTATS | LOOKUP_WRITE))) server.stat_keyspace_hits++; /* TODO: Use separate hits stats for WRITE */ } else { if (!(flags & (LOOKUP_NONOTIFY | LOOKUP_WRITE))) notifyKeyspaceEvent(NOTIFY_KEY_MISS, "keymiss", key, db->id); if (!(flags & (LOOKUP_NOSTATS | LOOKUP_WRITE))) server.stat_keyspace_misses++; /* TODO: Use separate misses stats and notify event for WRITE */ } return val; } /* Lookup a key for read operations, or return NULL if the key is not found * in the specified DB. * * This API should not be used when we write to the key after obtaining * the object linked to the key, but only for read only operations. * * This function is equivalent to lookupKey(). The point of using this function * rather than lookupKey() directly is to indicate that the purpose is to read * the key. */ robj *lookupKeyReadWithFlags(redisDb *db, robj *key, int flags) { serverAssert(!(flags & LOOKUP_WRITE)); return lookupKey(db, key, flags); } /* Like lookupKeyReadWithFlags(), but does not use any flag, which is the * common case. */ robj *lookupKeyRead(redisDb *db, robj *key) { return lookupKeyReadWithFlags(db,key,LOOKUP_NONE); } /* Lookup a key for write operations, and as a side effect, if needed, expires * the key if its TTL is reached. It's equivalent to lookupKey() with the * LOOKUP_WRITE flag added. * * Returns the linked value object if the key exists or NULL if the key * does not exist in the specified DB. */ robj *lookupKeyWriteWithFlags(redisDb *db, robj *key, int flags) { return lookupKey(db, key, flags | LOOKUP_WRITE); } robj *lookupKeyWrite(redisDb *db, robj *key) { return lookupKeyWriteWithFlags(db, key, LOOKUP_NONE); } robj *lookupKeyReadOrReply(client *c, robj *key, robj *reply) { robj *o = lookupKeyRead(c->db, key); if (!o) addReplyOrErrorObject(c, reply); return o; } robj *lookupKeyWriteOrReply(client *c, robj *key, robj *reply) { robj *o = lookupKeyWrite(c->db, key); if (!o) addReplyOrErrorObject(c, reply); return o; } /* Add the key to the DB. It's up to the caller to increment the reference * counter 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); dictEntry *de = dictAddRaw(db->dict, copy, NULL); serverAssertWithInfo(NULL, key, de != NULL); dictSetVal(db->dict, de, val); signalKeyAsReady(db, key, val->type); if (server.cluster_enabled) slotToKeyAddEntry(de, db); notifyKeyspaceEvent(NOTIFY_NEW,"new",key,db->id); } /* This is a special version of dbAdd() that is used only when loading * keys from the RDB file: the key is passed as an SDS string that is * retained by the function (and not freed by the caller). * * Moreover this function will not abort if the key is already busy, to * give more control to the caller, nor will signal the key as ready * since it is not useful in this context. * * The function returns 1 if the key was added to the database, taking * ownership of the SDS string, otherwise 0 is returned, and is up to the * caller to free the SDS string. */ int dbAddRDBLoad(redisDb *db, sds key, robj *val) { dictEntry *de = dictAddRaw(db->dict, key, NULL); if (de == NULL) return 0; dictSetVal(db->dict, de, val); if (server.cluster_enabled) slotToKeyAddEntry(de, db); return 1; } /* 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) { dictEntry *de = dictFind(db->dict,key->ptr); serverAssertWithInfo(NULL,key,de != NULL); dictEntry auxentry = *de; robj *old = dictGetVal(de); if (server.maxmemory_policy & MAXMEMORY_FLAG_LFU) { val->lru = old->lru; } /* Although the key is not really deleted from the database, we regard * overwrite as two steps of unlink+add, so we still need to call the unlink * callback of the module. */ moduleNotifyKeyUnlink(key,old,db->id); /* We want to try to unblock any client using a blocking XREADGROUP */ if (old->type == OBJ_STREAM) signalKeyAsReady(db,key,old->type); dictSetVal(db->dict, de, val); if (server.lazyfree_lazy_server_del) { freeObjAsync(key,old,db->id); dictSetVal(db->dict, &auxentry, NULL); } dictFreeVal(db->dict, &auxentry); } /* 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. * * 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), * unless 'SETKEY_KEEPTTL' is enabled in flags. * 4) The key lookup can take place outside this interface outcome will be * delivered with 'SETKEY_ALREADY_EXIST' or 'SETKEY_DOESNT_EXIST' * * All the new keys in the database should be created via this interface. * The client 'c' argument may be set to NULL if the operation is performed * in a context where there is no clear client performing the operation. */ void setKey(client *c, redisDb *db, robj *key, robj *val, int flags) { int keyfound = 0; if (flags & SETKEY_ALREADY_EXIST) keyfound = 1; else if (!(flags & SETKEY_DOESNT_EXIST)) keyfound = (lookupKeyWrite(db,key) != NULL); if (!keyfound) { dbAdd(db,key,val); } else { dbOverwrite(db,key,val); } incrRefCount(val); if (!(flags & SETKEY_KEEPTTL)) removeExpire(db,key); if (!(flags & SETKEY_NO_SIGNAL)) signalModifiedKey(c,db,key); } /* Return a random key, in form of a Redis object. * If there are no keys, NULL is returned. * * The function makes sure to return keys not already expired. */ robj *dbRandomKey(redisDb *db) { dictEntry *de; int maxtries = 100; int allvolatile = dictSize(db->dict) == dictSize(db->expires); while(1) { sds key; robj *keyobj; de = dictGetFairRandomKey(db->dict); if (de == NULL) return NULL; key = dictGetKey(de); keyobj = createStringObject(key,sdslen(key)); if (dictFind(db->expires,key)) { if (allvolatile && server.masterhost && --maxtries == 0) { /* If the DB is composed only of keys with an expire set, * it could happen that all the keys are already logically * expired in the slave, so the function cannot stop because * expireIfNeeded() is false, nor it can stop because * dictGetFairRandomKey() returns NULL (there are keys to return). * To prevent the infinite loop we do some tries, but if there * are the conditions for an infinite loop, eventually we * return a key name that may be already expired. */ return keyobj; } if (expireIfNeeded(db,keyobj,0)) { decrRefCount(keyobj); continue; /* search for another key. This expired. */ } } return keyobj; } } /* Helper for sync and async delete. */ static int dbGenericDelete(redisDb *db, robj *key, int async) { /* 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); dictEntry *de = dictUnlink(db->dict,key->ptr); if (de) { robj *val = dictGetVal(de); /* Tells the module that the key has been unlinked from the database. */ moduleNotifyKeyUnlink(key,val,db->id); /* We want to try to unblock any client using a blocking XREADGROUP */ if (val->type == OBJ_STREAM) signalKeyAsReady(db,key,val->type); if (async) { freeObjAsync(key, val, db->id); dictSetVal(db->dict, de, NULL); } if (server.cluster_enabled) slotToKeyDelEntry(de, db); dictFreeUnlinkedEntry(db->dict,de); return 1; } else { return 0; } } /* Delete a key, value, and associated expiration entry if any, from the DB */ int dbSyncDelete(redisDb *db, robj *key) { return dbGenericDelete(db, key, 0); } /* Delete a key, value, and associated expiration entry if any, from the DB. If * the value consists of many allocations, it may be freed asynchronously. */ int dbAsyncDelete(redisDb *db, robj *key) { return dbGenericDelete(db, key, 1); } /* This is a wrapper whose behavior depends on the Redis lazy free * configuration. Deletes the key synchronously or asynchronously. */ int dbDelete(redisDb *db, robj *key) { return dbGenericDelete(db, key, server.lazyfree_lazy_server_del); } /* Prepare the string object stored at 'key' to be modified destructively * to implement commands like SETBIT or APPEND. * * An object is usually ready to be modified unless one of the two conditions * are true: * * 1) The object 'o' is shared (refcount > 1), we don't want to affect * other users. * 2) The object encoding is not "RAW". * * If the object is found in one of the above conditions (or both) by the * function, an unshared / not-encoded copy of the string object is stored * at 'key' in the specified 'db'. Otherwise the object 'o' itself is * returned. * * USAGE: * * The object 'o' is what the caller already obtained by looking up 'key' * in 'db', the usage pattern looks like this: * * o = lookupKeyWrite(db,key); * if (checkType(c,o,OBJ_STRING)) return; * o = dbUnshareStringValue(db,key,o); * * At this point the caller is ready to modify the object, for example * using an sdscat() call to append some data, or anything else. */ robj *dbUnshareStringValue(redisDb *db, robj *key, robj *o) { serverAssert(o->type == OBJ_STRING); if (o->refcount != 1 || o->encoding != OBJ_ENCODING_RAW) { robj *decoded = getDecodedObject(o); o = createRawStringObject(decoded->ptr, sdslen(decoded->ptr)); decrRefCount(decoded); dbOverwrite(db,key,o); } return o; } /* Remove all keys from the database(s) structure. The dbarray argument * may not be the server main DBs (could be a temporary DB). * * The dbnum can be -1 if all the DBs should be emptied, or the specified * DB index if we want to empty only a single database. * The function returns the number of keys removed from the database(s). */ long long emptyDbStructure(redisDb *dbarray, int dbnum, int async, void(callback)(dict*)) { long long removed = 0; int startdb, enddb; if (dbnum == -1) { startdb = 0; enddb = server.dbnum-1; } else { startdb = enddb = dbnum; } for (int j = startdb; j <= enddb; j++) { removed += dictSize(dbarray[j].dict); if (async) { emptyDbAsync(&dbarray[j]); } else { dictEmpty(dbarray[j].dict,callback); dictEmpty(dbarray[j].expires,callback); } /* Because all keys of database are removed, reset average ttl. */ dbarray[j].avg_ttl = 0; dbarray[j].expires_cursor = 0; } return removed; } /* Remove all data (keys and functions) from all the databases in a * Redis server. If callback is given the function is called from * time to time to signal that work is in progress. * * The dbnum can be -1 if all the DBs should be flushed, or the specified * DB number if we want to flush only a single Redis database number. * * Flags are be EMPTYDB_NO_FLAGS if no special flags are specified or * EMPTYDB_ASYNC if we want the memory to be freed in a different thread * and the function to return ASAP. EMPTYDB_NOFUNCTIONS can also be set * to specify that we do not want to delete the functions. * * On success the function returns the number of keys removed from the * database(s). Otherwise -1 is returned in the specific case the * DB number is out of range, and errno is set to EINVAL. */ long long emptyData(int dbnum, int flags, void(callback)(dict*)) { int async = (flags & EMPTYDB_ASYNC); int with_functions = !(flags & EMPTYDB_NOFUNCTIONS); RedisModuleFlushInfoV1 fi = {REDISMODULE_FLUSHINFO_VERSION,!async,dbnum}; long long removed = 0; if (dbnum < -1 || dbnum >= server.dbnum) { errno = EINVAL; return -1; } /* Fire the flushdb modules event. */ moduleFireServerEvent(REDISMODULE_EVENT_FLUSHDB, REDISMODULE_SUBEVENT_FLUSHDB_START, &fi); /* Make sure the WATCHed keys are affected by the FLUSH* commands. * Note that we need to call the function while the keys are still * there. */ signalFlushedDb(dbnum, async); /* Empty redis database structure. */ removed = emptyDbStructure(server.db, dbnum, async, callback); /* Flush slots to keys map if enable cluster, we can flush entire * slots to keys map whatever dbnum because only support one DB * in cluster mode. */ if (server.cluster_enabled) slotToKeyFlush(server.db); if (dbnum == -1) flushSlaveKeysWithExpireList(); if (with_functions) { serverAssert(dbnum == -1); functionsLibCtxClearCurrent(async); } /* Also fire the end event. Note that this event will fire almost * immediately after the start event if the flush is asynchronous. */ moduleFireServerEvent(REDISMODULE_EVENT_FLUSHDB, REDISMODULE_SUBEVENT_FLUSHDB_END, &fi); return removed; } /* Initialize temporary db on replica for use during diskless replication. */ redisDb *initTempDb(void) { redisDb *tempDb = zcalloc(sizeof(redisDb)*server.dbnum); for (int i=0; i= server.dbnum) return C_ERR; c->db = &server.db[id]; return C_OK; } long long dbTotalServerKeyCount() { long long total = 0; int j; for (j = 0; j < server.dbnum; j++) { total += dictSize(server.db[j].dict); } return total; } /*----------------------------------------------------------------------------- * 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. *----------------------------------------------------------------------------*/ /* Note that the 'c' argument may be NULL if the key was modified out of * a context of a client. */ void signalModifiedKey(client *c, redisDb *db, robj *key) { touchWatchedKey(db,key); trackingInvalidateKey(c,key,1); } void signalFlushedDb(int dbid, int async) { int startdb, enddb; if (dbid == -1) { startdb = 0; enddb = server.dbnum-1; } else { startdb = enddb = dbid; } for (int j = startdb; j <= enddb; j++) { scanDatabaseForDeletedStreams(&server.db[j], NULL); touchAllWatchedKeysInDb(&server.db[j], NULL); } trackingInvalidateKeysOnFlush(async); /* Changes in this method may take place in swapMainDbWithTempDb as well, * where we execute similar calls, but with subtle differences as it's * not simply flushing db. */ } /*----------------------------------------------------------------------------- * Type agnostic commands operating on the key space *----------------------------------------------------------------------------*/ /* Return the set of flags to use for the emptyDb() call for FLUSHALL * and FLUSHDB commands. * * sync: flushes the database in an sync manner. * async: flushes the database in an async manner. * no option: determine sync or async according to the value of lazyfree-lazy-user-flush. * * On success C_OK is returned and the flags are stored in *flags, otherwise * C_ERR is returned and the function sends an error to the client. */ int getFlushCommandFlags(client *c, int *flags) { /* Parse the optional ASYNC option. */ if (c->argc == 2 && !strcasecmp(c->argv[1]->ptr,"sync")) { *flags = EMPTYDB_NO_FLAGS; } else if (c->argc == 2 && !strcasecmp(c->argv[1]->ptr,"async")) { *flags = EMPTYDB_ASYNC; } else if (c->argc == 1) { *flags = server.lazyfree_lazy_user_flush ? EMPTYDB_ASYNC : EMPTYDB_NO_FLAGS; } else { addReplyErrorObject(c,shared.syntaxerr); return C_ERR; } return C_OK; } /* Flushes the whole server data set. */ void flushAllDataAndResetRDB(int flags) { server.dirty += emptyData(-1,flags,NULL); if (server.child_type == CHILD_TYPE_RDB) killRDBChild(); 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; rdbSaveInfo rsi, *rsiptr; rsiptr = rdbPopulateSaveInfo(&rsi); rdbSave(SLAVE_REQ_NONE,server.rdb_filename,rsiptr); server.dirty = saved_dirty; } /* Without that extra dirty++, when db was already empty, FLUSHALL will * not be replicated nor put into the AOF. */ server.dirty++; #if defined(USE_JEMALLOC) /* jemalloc 5 doesn't release pages back to the OS when there's no traffic. * for large databases, flushdb blocks for long anyway, so a bit more won't * harm and this way the flush and purge will be synchronous. */ if (!(flags & EMPTYDB_ASYNC)) jemalloc_purge(); #endif } /* FLUSHDB [ASYNC] * * Flushes the currently SELECTed Redis DB. */ void flushdbCommand(client *c) { int flags; if (getFlushCommandFlags(c,&flags) == C_ERR) return; /* flushdb should not flush the functions */ server.dirty += emptyData(c->db->id,flags | EMPTYDB_NOFUNCTIONS,NULL); addReply(c,shared.ok); #if defined(USE_JEMALLOC) /* jemalloc 5 doesn't release pages back to the OS when there's no traffic. * for large databases, flushdb blocks for long anyway, so a bit more won't * harm and this way the flush and purge will be synchronous. */ if (!(flags & EMPTYDB_ASYNC)) jemalloc_purge(); #endif } /* FLUSHALL [ASYNC] * * Flushes the whole server data set. */ void flushallCommand(client *c) { int flags; if (getFlushCommandFlags(c,&flags) == C_ERR) return; /* flushall should not flush the functions */ flushAllDataAndResetRDB(flags | EMPTYDB_NOFUNCTIONS); addReply(c,shared.ok); } /* This command implements DEL and LAZYDEL. */ void delGenericCommand(client *c, int lazy) { int numdel = 0, j; for (j = 1; j < c->argc; j++) { expireIfNeeded(c->db,c->argv[j],0); int deleted = lazy ? dbAsyncDelete(c->db,c->argv[j]) : dbSyncDelete(c->db,c->argv[j]); if (deleted) { signalModifiedKey(c,c->db,c->argv[j]); notifyKeyspaceEvent(NOTIFY_GENERIC, "del",c->argv[j],c->db->id); server.dirty++; numdel++; } } addReplyLongLong(c,numdel); } void delCommand(client *c) { delGenericCommand(c,server.lazyfree_lazy_user_del); } void unlinkCommand(client *c) { delGenericCommand(c,1); } /* EXISTS key1 key2 ... key_N. * Return value is the number of keys existing. */ void existsCommand(client *c) { long long count = 0; int j; for (j = 1; j < c->argc; j++) { if (lookupKeyReadWithFlags(c->db,c->argv[j],LOOKUP_NOTOUCH)) count++; } addReplyLongLong(c,count); } void selectCommand(client *c) { int id; if (getIntFromObjectOrReply(c, c->argv[1], &id, NULL) != C_OK) return; if (server.cluster_enabled && id != 0) { addReplyError(c,"SELECT is not allowed in cluster mode"); return; } if (selectDb(c,id) == C_ERR) { addReplyError(c,"DB index is out of range"); } else { addReply(c,shared.ok); } } void randomkeyCommand(client *c) { robj *key; if ((key = dbRandomKey(c->db)) == NULL) { addReplyNull(c); return; } addReplyBulk(c,key); decrRefCount(key); } void keysCommand(client *c) { dictIterator *di; dictEntry *de; sds pattern = c->argv[1]->ptr; int plen = sdslen(pattern), allkeys; unsigned long numkeys = 0; void *replylen = addReplyDeferredLen(c); di = dictGetSafeIterator(c->db->dict); allkeys = (pattern[0] == '*' && plen == 1); while((de = dictNext(di)) != NULL) { sds key = dictGetKey(de); robj *keyobj; if (allkeys || stringmatchlen(pattern,plen,key,sdslen(key),0)) { keyobj = createStringObject(key,sdslen(key)); if (!keyIsExpired(c->db,keyobj)) { addReplyBulk(c,keyobj); numkeys++; } decrRefCount(keyobj); } } dictReleaseIterator(di); setDeferredArrayLen(c,replylen,numkeys); } /* This callback is used by scanGenericCommand in order to collect elements * returned by the dictionary iterator into a list. */ void scanCallback(void *privdata, const dictEntry *de) { void **pd = (void**) privdata; list *keys = pd[0]; robj *o = pd[1]; robj *key, *val = NULL; if (o == NULL) { sds sdskey = dictGetKey(de); key = createStringObject(sdskey, sdslen(sdskey)); } else if (o->type == OBJ_SET) { sds keysds = dictGetKey(de); key = createStringObject(keysds,sdslen(keysds)); } else if (o->type == OBJ_HASH) { sds sdskey = dictGetKey(de); sds sdsval = dictGetVal(de); key = createStringObject(sdskey,sdslen(sdskey)); val = createStringObject(sdsval,sdslen(sdsval)); } else if (o->type == OBJ_ZSET) { sds sdskey = dictGetKey(de); key = createStringObject(sdskey,sdslen(sdskey)); val = createStringObjectFromLongDouble(*(double*)dictGetVal(de),0); } else { serverPanic("Type not handled in SCAN callback."); } listAddNodeTail(keys, key); if (val) listAddNodeTail(keys, val); } /* Try to parse a SCAN cursor stored at object 'o': * if the cursor is valid, store it as unsigned integer into *cursor and * returns C_OK. Otherwise return C_ERR and send an error to the * client. */ int parseScanCursorOrReply(client *c, robj *o, unsigned long *cursor) { char *eptr; /* Use strtoul() because we need an *unsigned* long, so * getLongLongFromObject() does not cover the whole cursor space. */ errno = 0; *cursor = strtoul(o->ptr, &eptr, 10); if (isspace(((char*)o->ptr)[0]) || eptr[0] != '\0' || errno == ERANGE) { addReplyError(c, "invalid cursor"); return C_ERR; } return C_OK; } /* This command implements SCAN, HSCAN and SSCAN commands. * If object 'o' is passed, then it must be a Hash, Set or Zset object, otherwise * if 'o' is NULL the command will operate on the dictionary associated with * the current database. * * When 'o' is not NULL the function assumes that the first argument in * the client arguments vector is a key so it skips it before iterating * in order to parse options. * * In the case of a Hash object the function returns both the field and value * of every element on the Hash. */ void scanGenericCommand(client *c, robj *o, unsigned long cursor) { int i, j; list *keys = listCreate(); listNode *node, *nextnode; long count = 10; sds pat = NULL; sds typename = NULL; int patlen = 0, use_pattern = 0; dict *ht; /* Object must be NULL (to iterate keys names), or the type of the object * must be Set, Sorted Set, or Hash. */ serverAssert(o == NULL || o->type == OBJ_SET || o->type == OBJ_HASH || o->type == OBJ_ZSET); /* Set i to the first option argument. The previous one is the cursor. */ i = (o == NULL) ? 2 : 3; /* Skip the key argument if needed. */ /* Step 1: Parse options. */ while (i < c->argc) { j = c->argc - i; if (!strcasecmp(c->argv[i]->ptr, "count") && j >= 2) { if (getLongFromObjectOrReply(c, c->argv[i+1], &count, NULL) != C_OK) { goto cleanup; } if (count < 1) { addReplyErrorObject(c,shared.syntaxerr); goto cleanup; } i += 2; } else if (!strcasecmp(c->argv[i]->ptr, "match") && j >= 2) { pat = c->argv[i+1]->ptr; patlen = sdslen(pat); /* The pattern always matches if it is exactly "*", so it is * equivalent to disabling it. */ use_pattern = !(patlen == 1 && pat[0] == '*'); i += 2; } else if (!strcasecmp(c->argv[i]->ptr, "type") && o == NULL && j >= 2) { /* SCAN for a particular type only applies to the db dict */ typename = c->argv[i+1]->ptr; i+= 2; } else { addReplyErrorObject(c,shared.syntaxerr); goto cleanup; } } /* Step 2: Iterate the collection. * * Note that if the object is encoded with a listpack, intset, or any other * representation that is not a hash table, we are sure that it is also * composed of a small number of elements. So to avoid taking state we * just return everything inside the object in a single call, setting the * cursor to zero to signal the end of the iteration. */ /* Handle the case of a hash table. */ ht = NULL; if (o == NULL) { ht = c->db->dict; } else if (o->type == OBJ_SET && o->encoding == OBJ_ENCODING_HT) { ht = o->ptr; } else if (o->type == OBJ_HASH && o->encoding == OBJ_ENCODING_HT) { ht = o->ptr; count *= 2; /* We return key / value for this type. */ } else if (o->type == OBJ_ZSET && o->encoding == OBJ_ENCODING_SKIPLIST) { zset *zs = o->ptr; ht = zs->dict; count *= 2; /* We return key / value for this type. */ } if (ht) { void *privdata[2]; /* We set the max number of iterations to ten times the specified * COUNT, so if the hash table is in a pathological state (very * sparsely populated) we avoid to block too much time at the cost * of returning no or very few elements. */ long maxiterations = count*10; /* We pass two pointers to the callback: the list to which it will * add new elements, and the object containing the dictionary so that * it is possible to fetch more data in a type-dependent way. */ privdata[0] = keys; privdata[1] = o; do { cursor = dictScan(ht, cursor, scanCallback, NULL, privdata); } while (cursor && maxiterations-- && listLength(keys) < (unsigned long)count); } else if (o->type == OBJ_SET) { int pos = 0; int64_t ll; while(intsetGet(o->ptr,pos++,&ll)) listAddNodeTail(keys,createStringObjectFromLongLong(ll)); cursor = 0; } else if (o->type == OBJ_HASH || o->type == OBJ_ZSET) { unsigned char *p = lpFirst(o->ptr); unsigned char *vstr; int64_t vlen; unsigned char intbuf[LP_INTBUF_SIZE]; while(p) { vstr = lpGet(p,&vlen,intbuf); listAddNodeTail(keys, createStringObject((char*)vstr,vlen)); p = lpNext(o->ptr,p); } cursor = 0; } else { serverPanic("Not handled encoding in SCAN."); } /* Step 3: Filter elements. */ node = listFirst(keys); while (node) { robj *kobj = listNodeValue(node); nextnode = listNextNode(node); int filter = 0; /* Filter element if it does not match the pattern. */ if (use_pattern) { if (sdsEncodedObject(kobj)) { if (!stringmatchlen(pat, patlen, kobj->ptr, sdslen(kobj->ptr), 0)) filter = 1; } else { char buf[LONG_STR_SIZE]; int len; serverAssert(kobj->encoding == OBJ_ENCODING_INT); len = ll2string(buf,sizeof(buf),(long)kobj->ptr); if (!stringmatchlen(pat, patlen, buf, len, 0)) filter = 1; } } /* Filter an element if it isn't the type we want. */ if (!filter && o == NULL && typename){ robj* typecheck = lookupKeyReadWithFlags(c->db, kobj, LOOKUP_NOTOUCH); char* type = getObjectTypeName(typecheck); if (strcasecmp((char*) typename, type)) filter = 1; } /* Filter element if it is an expired key. */ if (!filter && o == NULL && expireIfNeeded(c->db, kobj, 0)) filter = 1; /* Remove the element and its associated value if needed. */ if (filter) { decrRefCount(kobj); listDelNode(keys, node); } /* If this is a hash or a sorted set, we have a flat list of * key-value elements, so if this element was filtered, remove the * value, or skip it if it was not filtered: we only match keys. */ if (o && (o->type == OBJ_ZSET || o->type == OBJ_HASH)) { node = nextnode; serverAssert(node); /* assertion for valgrind (avoid NPD) */ nextnode = listNextNode(node); if (filter) { kobj = listNodeValue(node); decrRefCount(kobj); listDelNode(keys, node); } } node = nextnode; } /* Step 4: Reply to the client. */ addReplyArrayLen(c, 2); addReplyBulkLongLong(c,cursor); addReplyArrayLen(c, listLength(keys)); while ((node = listFirst(keys)) != NULL) { robj *kobj = listNodeValue(node); addReplyBulk(c, kobj); decrRefCount(kobj); listDelNode(keys, node); } cleanup: listSetFreeMethod(keys,decrRefCountVoid); listRelease(keys); } /* The SCAN command completely relies on scanGenericCommand. */ void scanCommand(client *c) { unsigned long cursor; if (parseScanCursorOrReply(c,c->argv[1],&cursor) == C_ERR) return; scanGenericCommand(c,NULL,cursor); } void dbsizeCommand(client *c) { addReplyLongLong(c,dictSize(c->db->dict)); } void lastsaveCommand(client *c) { addReplyLongLong(c,server.lastsave); } char* getObjectTypeName(robj *o) { char* type; if (o == NULL) { type = "none"; } else { switch(o->type) { case OBJ_STRING: type = "string"; break; case OBJ_LIST: type = "list"; break; case OBJ_SET: type = "set"; break; case OBJ_ZSET: type = "zset"; break; case OBJ_HASH: type = "hash"; break; case OBJ_STREAM: type = "stream"; break; case OBJ_MODULE: { moduleValue *mv = o->ptr; type = mv->type->name; }; break; default: type = "unknown"; break; } } return type; } void typeCommand(client *c) { robj *o; o = lookupKeyReadWithFlags(c->db,c->argv[1],LOOKUP_NOTOUCH); addReplyStatus(c, getObjectTypeName(o)); } void shutdownCommand(client *c) { int flags = SHUTDOWN_NOFLAGS; int abort = 0; for (int i = 1; i < c->argc; i++) { if (!strcasecmp(c->argv[i]->ptr,"nosave")) { flags |= SHUTDOWN_NOSAVE; } else if (!strcasecmp(c->argv[i]->ptr,"save")) { flags |= SHUTDOWN_SAVE; } else if (!strcasecmp(c->argv[i]->ptr, "now")) { flags |= SHUTDOWN_NOW; } else if (!strcasecmp(c->argv[i]->ptr, "force")) { flags |= SHUTDOWN_FORCE; } else if (!strcasecmp(c->argv[i]->ptr, "abort")) { abort = 1; } else { addReplyErrorObject(c,shared.syntaxerr); return; } } if ((abort && flags != SHUTDOWN_NOFLAGS) || (flags & SHUTDOWN_NOSAVE && flags & SHUTDOWN_SAVE)) { /* Illegal combo. */ addReplyErrorObject(c,shared.syntaxerr); return; } if (abort) { if (abortShutdown() == C_OK) addReply(c, shared.ok); else addReplyError(c, "No shutdown in progress."); return; } if (!(flags & SHUTDOWN_NOW) && c->flags & CLIENT_DENY_BLOCKING) { addReplyError(c, "SHUTDOWN without NOW or ABORT isn't allowed for DENY BLOCKING client"); return; } if (!(flags & SHUTDOWN_NOSAVE) && scriptIsTimedout()) { /* Script timed out. Shutdown allowed only with the NOSAVE flag. See * also processCommand where these errors are returned. */ if (server.busy_module_yield_flags && server.busy_module_yield_reply) { addReplyErrorFormat(c, "-BUSY %s", server.busy_module_yield_reply); } else if (server.busy_module_yield_flags) { addReplyErrorObject(c, shared.slowmoduleerr); } else if (scriptIsEval()) { addReplyErrorObject(c, shared.slowevalerr); } else { addReplyErrorObject(c, shared.slowscripterr); } return; } blockClient(c, BLOCKED_SHUTDOWN); if (prepareForShutdown(flags) == C_OK) exit(0); /* If we're here, then shutdown is ongoing (the client is still blocked) or * failed (the client has received an error). */ } void renameGenericCommand(client *c, int nx) { robj *o; long long expire; int samekey = 0; /* When source and dest key is the same, no operation is performed, * if the key exists, however we still return an error on unexisting key. */ if (sdscmp(c->argv[1]->ptr,c->argv[2]->ptr) == 0) samekey = 1; if ((o = lookupKeyWriteOrReply(c,c->argv[1],shared.nokeyerr)) == NULL) return; if (samekey) { addReply(c,nx ? shared.czero : shared.ok); return; } incrRefCount(o); expire = getExpire(c->db,c->argv[1]); if (lookupKeyWrite(c->db,c->argv[2]) != NULL) { if (nx) { decrRefCount(o); addReply(c,shared.czero); return; } /* 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,c->db,c->argv[2],expire); dbDelete(c->db,c->argv[1]); signalModifiedKey(c,c->db,c->argv[1]); signalModifiedKey(c,c->db,c->argv[2]); notifyKeyspaceEvent(NOTIFY_GENERIC,"rename_from", c->argv[1],c->db->id); notifyKeyspaceEvent(NOTIFY_GENERIC,"rename_to", c->argv[2],c->db->id); server.dirty++; addReply(c,nx ? shared.cone : shared.ok); } void renameCommand(client *c) { renameGenericCommand(c,0); } void renamenxCommand(client *c) { renameGenericCommand(c,1); } void moveCommand(client *c) { robj *o; redisDb *src, *dst; int srcid, dbid; long long expire; 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; if (getIntFromObjectOrReply(c, c->argv[2], &dbid, NULL) != C_OK) return; if (selectDb(c,dbid) == C_ERR) { addReplyError(c,"DB index is out of range"); return; } dst = c->db; selectDb(c,srcid); /* Back to the source DB */ /* If the user is moving using as target the same * DB as the source DB it is probably an error. */ if (src == dst) { addReplyErrorObject(c,shared.sameobjecterr); return; } /* Check if the element exists and get a reference */ o = lookupKeyWrite(c->db,c->argv[1]); if (!o) { addReply(c,shared.czero); return; } expire = getExpire(c->db,c->argv[1]); /* 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); if (expire != -1) setExpire(c,dst,c->argv[1],expire); incrRefCount(o); /* OK! key moved, free the entry in the source DB */ dbDelete(src,c->argv[1]); signalModifiedKey(c,src,c->argv[1]); signalModifiedKey(c,dst,c->argv[1]); notifyKeyspaceEvent(NOTIFY_GENERIC, "move_from",c->argv[1],src->id); notifyKeyspaceEvent(NOTIFY_GENERIC, "move_to",c->argv[1],dst->id); server.dirty++; addReply(c,shared.cone); } void copyCommand(client *c) { robj *o; redisDb *src, *dst; int srcid, dbid; long long expire; int j, replace = 0, delete = 0; /* Obtain source and target DB pointers * Default target DB is the same as the source DB * Parse the REPLACE option and targetDB option. */ src = c->db; dst = c->db; srcid = c->db->id; dbid = c->db->id; for (j = 3; j < c->argc; j++) { int additional = c->argc - j - 1; if (!strcasecmp(c->argv[j]->ptr,"replace")) { replace = 1; } else if (!strcasecmp(c->argv[j]->ptr, "db") && additional >= 1) { if (getIntFromObjectOrReply(c, c->argv[j+1], &dbid, NULL) != C_OK) return; if (selectDb(c, dbid) == C_ERR) { addReplyError(c,"DB index is out of range"); return; } dst = c->db; selectDb(c,srcid); /* Back to the source DB */ j++; /* Consume additional arg. */ } else { addReplyErrorObject(c,shared.syntaxerr); return; } } if ((server.cluster_enabled == 1) && (srcid != 0 || dbid != 0)) { addReplyError(c,"Copying to another database is not allowed in cluster mode"); return; } /* If the user select the same DB as * the source DB and using newkey as the same key * it is probably an error. */ robj *key = c->argv[1]; robj *newkey = c->argv[2]; if (src == dst && (sdscmp(key->ptr, newkey->ptr) == 0)) { addReplyErrorObject(c,shared.sameobjecterr); return; } /* Check if the element exists and get a reference */ o = lookupKeyRead(c->db, key); if (!o) { addReply(c,shared.czero); return; } expire = getExpire(c->db,key); /* Return zero if the key already exists in the target DB. * If REPLACE option is selected, delete newkey from targetDB. */ if (lookupKeyWrite(dst,newkey) != NULL) { if (replace) { delete = 1; } else { addReply(c,shared.czero); return; } } /* Duplicate object according to object's type. */ robj *newobj; switch(o->type) { case OBJ_STRING: newobj = dupStringObject(o); break; case OBJ_LIST: newobj = listTypeDup(o); break; case OBJ_SET: newobj = setTypeDup(o); break; case OBJ_ZSET: newobj = zsetDup(o); break; case OBJ_HASH: newobj = hashTypeDup(o); break; case OBJ_STREAM: newobj = streamDup(o); break; case OBJ_MODULE: newobj = moduleTypeDupOrReply(c, key, newkey, dst->id, o); if (!newobj) return; break; default: addReplyError(c, "unknown type object"); return; } if (delete) { dbDelete(dst,newkey); } dbAdd(dst,newkey,newobj); if (expire != -1) setExpire(c, dst, newkey, expire); /* OK! key copied */ signalModifiedKey(c,dst,c->argv[2]); notifyKeyspaceEvent(NOTIFY_GENERIC,"copy_to",c->argv[2],dst->id); server.dirty++; addReply(c,shared.cone); } /* Helper function for dbSwapDatabases(): scans the list of keys that have * one or more blocked clients for B[LR]POP or other blocking commands * and signal the keys as ready if they are of the right type. See the comment * where the function is used for more info. */ void scanDatabaseForReadyKeys(redisDb *db) { dictEntry *de; dictIterator *di = dictGetSafeIterator(db->blocking_keys); while((de = dictNext(di)) != NULL) { robj *key = dictGetKey(de); dictEntry *kde = dictFind(db->dict,key->ptr); if (kde) { robj *value = dictGetVal(kde); signalKeyAsReady(db, key, value->type); } } dictReleaseIterator(di); } /* Since we are unblocking XREADGROUP clients in the event the * key was deleted/overwritten we must do the same in case the * database was flushed/swapped. */ void scanDatabaseForDeletedStreams(redisDb *emptied, redisDb *replaced_with) { /* Optimization: If no clients are in type BLOCKED_STREAM, * we can skip this loop. */ if (!server.blocked_clients_by_type[BLOCKED_STREAM]) return; dictEntry *de; dictIterator *di = dictGetSafeIterator(emptied->blocking_keys); while((de = dictNext(di)) != NULL) { robj *key = dictGetKey(de); int was_stream = 0, is_stream = 0; dictEntry *kde = dictFind(emptied->dict, key->ptr); if (kde) { robj *value = dictGetVal(kde); was_stream = value->type == OBJ_STREAM; } if (replaced_with) { dictEntry *kde = dictFind(replaced_with->dict, key->ptr); if (kde) { robj *value = dictGetVal(kde); is_stream = value->type == OBJ_STREAM; } } /* We want to try to unblock any client using a blocking XREADGROUP */ if (was_stream && !is_stream) signalKeyAsReady(emptied, key, OBJ_STREAM); } dictReleaseIterator(di); } /* Swap two databases at runtime so that all clients will magically see * the new database even if already connected. Note that the client * structure c->db points to a given DB, so we need to be smarter and * swap the underlying referenced structures, otherwise we would need * to fix all the references to the Redis DB structure. * * Returns C_ERR if at least one of the DB ids are out of range, otherwise * C_OK is returned. */ int dbSwapDatabases(int id1, int id2) { if (id1 < 0 || id1 >= server.dbnum || id2 < 0 || id2 >= server.dbnum) return C_ERR; if (id1 == id2) return C_OK; redisDb aux = server.db[id1]; redisDb *db1 = &server.db[id1], *db2 = &server.db[id2]; /* Swapdb should make transaction fail if there is any * client watching keys */ touchAllWatchedKeysInDb(db1, db2); touchAllWatchedKeysInDb(db2, db1); /* Try to unblock any XREADGROUP clients if the key no longer exists. */ scanDatabaseForDeletedStreams(db1, db2); scanDatabaseForDeletedStreams(db2, db1); /* Swap hash tables. Note that we don't swap blocking_keys, * ready_keys and watched_keys, since we want clients to * remain in the same DB they were. */ db1->dict = db2->dict; db1->expires = db2->expires; db1->avg_ttl = db2->avg_ttl; db1->expires_cursor = db2->expires_cursor; db2->dict = aux.dict; db2->expires = aux.expires; db2->avg_ttl = aux.avg_ttl; db2->expires_cursor = aux.expires_cursor; /* Now we need to handle clients blocked on lists: as an effect * of swapping the two DBs, a client that was waiting for list * X in a given DB, may now actually be unblocked if X happens * to exist in the new version of the DB, after the swap. * * However normally we only do this check for efficiency reasons * in dbAdd() when a list is created. So here we need to rescan * the list of clients blocked on lists and signal lists as ready * if needed. */ scanDatabaseForReadyKeys(db1); scanDatabaseForReadyKeys(db2); return C_OK; } /* Logically, this discards (flushes) the old main database, and apply the newly loaded * database (temp) as the main (active) database, the actual freeing of old database * (which will now be placed in the temp one) is done later. */ void swapMainDbWithTempDb(redisDb *tempDb) { if (server.cluster_enabled) { /* Swap slots_to_keys from tempdb just loaded with main db slots_to_keys. */ clusterSlotToKeyMapping *aux = server.db->slots_to_keys; server.db->slots_to_keys = tempDb->slots_to_keys; tempDb->slots_to_keys = aux; } for (int i=0; idict = newdb->dict; activedb->expires = newdb->expires; activedb->avg_ttl = newdb->avg_ttl; activedb->expires_cursor = newdb->expires_cursor; newdb->dict = aux.dict; newdb->expires = aux.expires; newdb->avg_ttl = aux.avg_ttl; newdb->expires_cursor = aux.expires_cursor; /* Now we need to handle clients blocked on lists: as an effect * of swapping the two DBs, a client that was waiting for list * X in a given DB, may now actually be unblocked if X happens * to exist in the new version of the DB, after the swap. * * However normally we only do this check for efficiency reasons * in dbAdd() when a list is created. So here we need to rescan * the list of clients blocked on lists and signal lists as ready * if needed. */ scanDatabaseForReadyKeys(activedb); } trackingInvalidateKeysOnFlush(1); flushSlaveKeysWithExpireList(); } /* SWAPDB db1 db2 */ void swapdbCommand(client *c) { int id1, id2; /* Not allowed in cluster mode: we have just DB 0 there. */ if (server.cluster_enabled) { addReplyError(c,"SWAPDB is not allowed in cluster mode"); return; } /* Get the two DBs indexes. */ if (getIntFromObjectOrReply(c, c->argv[1], &id1, "invalid first DB index") != C_OK) return; if (getIntFromObjectOrReply(c, c->argv[2], &id2, "invalid second DB index") != C_OK) return; /* Swap... */ if (dbSwapDatabases(id1,id2) == C_ERR) { addReplyError(c,"DB index is out of range"); return; } else { RedisModuleSwapDbInfo si = {REDISMODULE_SWAPDBINFO_VERSION,id1,id2}; moduleFireServerEvent(REDISMODULE_EVENT_SWAPDB,0,&si); server.dirty++; addReply(c,shared.ok); } } /*----------------------------------------------------------------------------- * Expires API *----------------------------------------------------------------------------*/ 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. */ serverAssertWithInfo(NULL,key,dictFind(db->dict,key->ptr) != NULL); return dictDelete(db->expires,key->ptr) == DICT_OK; } /* Set an expire to the specified key. If the expire is set in the context * of an user calling a command 'c' is the client, otherwise 'c' is set * to NULL. The 'when' parameter is the absolute unix time in milliseconds * after which the key will no longer be considered valid. */ void setExpire(client *c, redisDb *db, robj *key, long long when) { dictEntry *kde, *de; /* Reuse the sds from the main dict in the expire dict */ kde = dictFind(db->dict,key->ptr); serverAssertWithInfo(NULL,key,kde != NULL); de = dictAddOrFind(db->expires,dictGetKey(kde)); dictSetSignedIntegerVal(de,when); int writable_slave = server.masterhost && server.repl_slave_ro == 0; if (c && writable_slave && !(c->flags & CLIENT_MASTER)) rememberSlaveKeyWithExpire(db,key); } /* Return the expire time of the specified key, or -1 if no expire * is associated with this key (i.e. the key is non volatile) */ long long getExpire(redisDb *db, robj *key) { dictEntry *de; /* No expire? return ASAP */ if (dictSize(db->expires) == 0 || (de = dictFind(db->expires,key->ptr)) == NULL) return -1; /* The entry was found in the expire dict, this means it should also * be present in the main dict (safety check). */ serverAssertWithInfo(NULL,key,dictFind(db->dict,key->ptr) != NULL); return dictGetSignedIntegerVal(de); } /* Delete the specified expired key and propagate expire. */ void deleteExpiredKeyAndPropagate(redisDb *db, robj *keyobj) { mstime_t expire_latency; latencyStartMonitor(expire_latency); if (server.lazyfree_lazy_expire) dbAsyncDelete(db,keyobj); else dbSyncDelete(db,keyobj); latencyEndMonitor(expire_latency); latencyAddSampleIfNeeded("expire-del",expire_latency); notifyKeyspaceEvent(NOTIFY_EXPIRED,"expired",keyobj,db->id); signalModifiedKey(NULL, db, keyobj); propagateDeletion(db,keyobj,server.lazyfree_lazy_expire); server.stat_expiredkeys++; } /* Propagate expires into slaves and the AOF file. * When a key expires in the master, a DEL operation for this key is sent * to all the slaves and the AOF file if enabled. * * This way the key expiry is centralized in one place, and since both * AOF and the master->slave link guarantee operation ordering, everything * will be consistent even if we allow write operations against expiring * keys. * * This function may be called from: * 1. Within call(): Example: Lazy-expire on key access. * In this case the caller doesn't have to do anything * because call() handles server.also_propagate(); or * 2. Outside of call(): Example: Active-expire, eviction. * In this the caller must remember to call * propagatePendingCommands, preferably at the end of * the deletion batch, so that DELs will be wrapped * in MULTI/EXEC */ void propagateDeletion(redisDb *db, robj *key, int lazy) { robj *argv[2]; argv[0] = lazy ? shared.unlink : shared.del; argv[1] = key; incrRefCount(argv[0]); incrRefCount(argv[1]); /* If the master decided to expire a key we must propagate it to replicas no matter what.. * Even if module executed a command without asking for propagation. */ int prev_replication_allowed = server.replication_allowed; server.replication_allowed = 1; alsoPropagate(db->id,argv,2,PROPAGATE_AOF|PROPAGATE_REPL); server.replication_allowed = prev_replication_allowed; decrRefCount(argv[0]); decrRefCount(argv[1]); } /* Check if the key is expired. */ int keyIsExpired(redisDb *db, robj *key) { mstime_t when = getExpire(db,key); mstime_t now; 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 in the context of a Lua script, we pretend that time is * blocked to when the Lua script started. This way a key can expire * only the first time it is accessed and not in the middle of the * script execution, making propagation to slaves / AOF consistent. * See issue #1525 on Github for more information. */ if (server.script_caller) { now = scriptTimeSnapshot(); } /* If we are in the middle of a command execution, we still want to use * a reference time that does not change: in that case we just use the * cached time, that we update before each call in the call() function. * This way we avoid that commands such as RPOPLPUSH or similar, that * may re-open the same key multiple times, can invalidate an already * open object in a next call, if the next call will see the key expired, * while the first did not. */ else if (server.fixed_time_expire > 0) { now = server.mstime; } /* For the other cases, we want to use the most fresh time we have. */ else { now = mstime(); } /* The key expired if the current (virtual or real) time is greater * than the expire time of the key. */ return now > when; } /* This function is called when we are going to perform some operation * in a given key, but such key may be already logically expired even if * it still exists in the database. The main way this function is called * is via lookupKey*() family of functions. * * The behavior of the function depends on the replication role of the * instance, because by default replicas do not delete expired keys. They * wait for DELs from the master for consistency matters. However even * replicas will try to have a coherent return value for the function, * so that read commands executed in the replica side will be able to * behave like if the key is expired even if still present (because the * master has yet to propagate the DEL). * * In masters as a side effect of finding a key which is expired, such * key will be evicted from the database. Also this may trigger the * propagation of a DEL/UNLINK command in AOF / replication stream. * * On replicas, this function does not delete expired keys by default, but * it still returns 1 if the key is logically expired. To force deletion * of logically expired keys even on replicas, set force_delete_expired to * a non-zero value. Note though that if the current client is executing * replicated commands from the master, keys are never considered expired. * * The return value of the function is 0 if the key is still valid, * otherwise the function returns 1 if the key is expired. */ int expireIfNeeded(redisDb *db, robj *key, int force_delete_expired) { if (!keyIsExpired(db,key)) return 0; /* If we are running in the context of a replica, instead of * evicting the expired key from the database, we return ASAP: * the replica key expiration is controlled by the master that will * send us synthesized DEL operations for expired keys. The * exception is when write operations are performed on writable * replicas. * * Still we try to return the right information to the caller, * that is, 0 if we think the key should be still valid, 1 if * we think the key is expired at this time. * * When replicating commands from the master, keys are never considered * expired. */ if (server.masterhost != NULL) { if (server.current_client == server.master) return 0; if (!force_delete_expired) return 1; } /* If clients are paused, we keep the current dataset constant, * but return to the client what we believe is the right state. Typically, * at the end of the pause we will properly expire the key OR we will * have failed over and the new primary will send us the expire. */ if (checkClientPauseTimeoutAndReturnIfPaused()) return 1; /* Delete the key */ deleteExpiredKeyAndPropagate(db,key); return 1; } /* ----------------------------------------------------------------------------- * API to get key arguments from commands * ---------------------------------------------------------------------------*/ /* Prepare the getKeysResult struct to hold numkeys, either by using the * pre-allocated keysbuf or by allocating a new array on the heap. * * This function must be called at least once before starting to populate * the result, and can be called repeatedly to enlarge the result array. */ keyReference *getKeysPrepareResult(getKeysResult *result, int numkeys) { /* GETKEYS_RESULT_INIT initializes keys to NULL, point it to the pre-allocated stack * buffer here. */ if (!result->keys) { serverAssert(!result->numkeys); result->keys = result->keysbuf; } /* Resize if necessary */ if (numkeys > result->size) { if (result->keys != result->keysbuf) { /* We're not using a static buffer, just (re)alloc */ result->keys = zrealloc(result->keys, numkeys * sizeof(keyReference)); } else { /* We are using a static buffer, copy its contents */ result->keys = zmalloc(numkeys * sizeof(keyReference)); if (result->numkeys) memcpy(result->keys, result->keysbuf, result->numkeys * sizeof(keyReference)); } result->size = numkeys; } return result->keys; } /* Returns a bitmask with all the flags found in any of the key specs of the command. * The 'inv' argument means we'll return a mask with all flags that are missing in at least one spec. */ int64_t getAllKeySpecsFlags(struct redisCommand *cmd, int inv) { int64_t flags = 0; for (int j = 0; j < cmd->key_specs_num; j++) { keySpec *spec = cmd->key_specs + j; flags |= inv? ~spec->flags : spec->flags; } return flags; } /* Fetch the keys based of the provided key specs. Returns the number of keys found, or -1 on error. * There are several flags that can be used to modify how this function finds keys in a command. * * GET_KEYSPEC_INCLUDE_NOT_KEYS: Return 'fake' keys as if they were keys. * GET_KEYSPEC_RETURN_PARTIAL: Skips invalid and incomplete keyspecs but returns the keys * found in other valid keyspecs. */ int getKeysUsingKeySpecs(struct redisCommand *cmd, robj **argv, int argc, int search_flags, getKeysResult *result) { int j, i, k = 0, last, first, step; keyReference *keys; for (j = 0; j < cmd->key_specs_num; j++) { keySpec *spec = cmd->key_specs + j; serverAssert(spec->begin_search_type != KSPEC_BS_INVALID); /* Skip specs that represent 'fake' keys */ if ((spec->flags & CMD_KEY_NOT_KEY) && !(search_flags & GET_KEYSPEC_INCLUDE_NOT_KEYS)) { continue; } first = 0; if (spec->begin_search_type == KSPEC_BS_INDEX) { first = spec->bs.index.pos; } else if (spec->begin_search_type == KSPEC_BS_KEYWORD) { int start_index = spec->bs.keyword.startfrom > 0 ? spec->bs.keyword.startfrom : argc+spec->bs.keyword.startfrom; int end_index = spec->bs.keyword.startfrom > 0 ? argc-1: 1; for (i = start_index; i != end_index; i = start_index <= end_index ? i + 1 : i - 1) { if (i >= argc || i < 1) break; if (!strcasecmp((char*)argv[i]->ptr,spec->bs.keyword.keyword)) { first = i+1; break; } } /* keyword not found */ if (!first) { continue; } } else { /* unknown spec */ goto invalid_spec; } if (spec->find_keys_type == KSPEC_FK_RANGE) { step = spec->fk.range.keystep; if (spec->fk.range.lastkey >= 0) { last = first + spec->fk.range.lastkey; } else { if (!spec->fk.range.limit) { last = argc + spec->fk.range.lastkey; } else { serverAssert(spec->fk.range.lastkey == -1); last = first + ((argc-first)/spec->fk.range.limit + spec->fk.range.lastkey); } } } else if (spec->find_keys_type == KSPEC_FK_KEYNUM) { step = spec->fk.keynum.keystep; long long numkeys; if (spec->fk.keynum.keynumidx >= argc) goto invalid_spec; sds keynum_str = argv[first + spec->fk.keynum.keynumidx]->ptr; if (!string2ll(keynum_str,sdslen(keynum_str),&numkeys) || numkeys < 0) { /* Unable to parse the numkeys argument or it was invalid */ goto invalid_spec; } first += spec->fk.keynum.firstkey; last = first + (int)numkeys-1; } else { /* unknown spec */ goto invalid_spec; } int count = ((last - first)+1); keys = getKeysPrepareResult(result, count); /* First or last is out of bounds, which indicates a syntax error */ if (last >= argc || last < first || first >= argc) { goto invalid_spec; } for (i = first; i <= last; i += step) { if (i >= argc || i < first) { /* Modules commands, and standard commands with a not fixed number * of arguments (negative arity parameter) do not have dispatch * time arity checks, so we need to handle the case where the user * passed an invalid number of arguments here. In this case we * return no keys and expect the command implementation to report * an arity or syntax error. */ if (cmd->flags & CMD_MODULE || cmd->arity < 0) { continue; } else { serverPanic("Redis built-in command declared keys positions not matching the arity requirements."); } } keys[k].pos = i; keys[k++].flags = spec->flags; } /* Handle incomplete specs (only after we added the current spec * to `keys`, just in case GET_KEYSPEC_RETURN_PARTIAL was given) */ if (spec->flags & CMD_KEY_INCOMPLETE) { goto invalid_spec; } /* Done with this spec */ continue; invalid_spec: if (search_flags & GET_KEYSPEC_RETURN_PARTIAL) { continue; } else { result->numkeys = 0; return -1; } } result->numkeys = k; return k; } /* Return all the arguments that are keys in the command passed via argc / argv. * This function will eventually replace getKeysFromCommand. * * The command returns the positions of all the key arguments inside the array, * so the actual return value is a heap allocated array of integers. The * length of the array is returned by reference into *numkeys. * * Along with the position, this command also returns the flags that are * associated with how Redis will access the key. * * 'cmd' must be point to the corresponding entry into the redisCommand * table, according to the command name in argv[0]. */ int getKeysFromCommandWithSpecs(struct redisCommand *cmd, robj **argv, int argc, int search_flags, getKeysResult *result) { /* The command has at least one key-spec not marked as NOT_KEY */ int has_keyspec = (getAllKeySpecsFlags(cmd, 1) & CMD_KEY_NOT_KEY); /* The command has at least one key-spec marked as VARIABLE_FLAGS */ int has_varflags = (getAllKeySpecsFlags(cmd, 0) & CMD_KEY_VARIABLE_FLAGS); /* Flags indicating that we have a getkeys callback */ int has_module_getkeys = cmd->flags & CMD_MODULE_GETKEYS; int has_native_getkeys = !(cmd->flags & CMD_MODULE) && cmd->getkeys_proc; /* The key-spec that's auto generated by RM_CreateCommand sets VARIABLE_FLAGS since no flags are given. * If the module provides getkeys callback, we'll prefer it, but if it didn't, we'll use key-spec anyway. */ if ((cmd->flags & CMD_MODULE) && has_varflags && !has_module_getkeys) has_varflags = 0; /* We prefer key-specs if there are any, and their flags are reliable. */ if (has_keyspec && !has_varflags) { int ret = getKeysUsingKeySpecs(cmd,argv,argc,search_flags,result); if (ret >= 0) return ret; /* If the specs returned with an error (probably an INVALID or INCOMPLETE spec), * fallback to the callback method. */ } /* Resort to getkeys callback methods. */ if (has_module_getkeys) return moduleGetCommandKeysViaAPI(cmd,argv,argc,result); /* We use native getkeys as a last resort, since not all these native getkeys provide * flags properly (only the ones that correspond to INVALID, INCOMPLETE or VARIABLE_FLAGS do.*/ if (has_native_getkeys) return cmd->getkeys_proc(cmd,argv,argc,result); return 0; } /* This function returns a sanity check if the command may have keys. */ int doesCommandHaveKeys(struct redisCommand *cmd) { return (!(cmd->flags & CMD_MODULE) && cmd->getkeys_proc) || /* has getkeys_proc (non modules) */ (cmd->flags & CMD_MODULE_GETKEYS) || /* module with GETKEYS */ (getAllKeySpecsFlags(cmd, 1) & CMD_KEY_NOT_KEY); /* has at least one key-spec not marked as NOT_KEY */ } /* A simplified channel spec table that contains all of the redis commands * and which channels they have and how they are accessed. */ typedef struct ChannelSpecs { redisCommandProc *proc; /* Command procedure to match against */ uint64_t flags; /* CMD_CHANNEL_* flags for this command */ int start; /* The initial position of the first channel */ int count; /* The number of channels, or -1 if all remaining * arguments are channels. */ } ChannelSpecs; ChannelSpecs commands_with_channels[] = { {subscribeCommand, CMD_CHANNEL_SUBSCRIBE, 1, -1}, {ssubscribeCommand, CMD_CHANNEL_SUBSCRIBE, 1, -1}, {unsubscribeCommand, CMD_CHANNEL_UNSUBSCRIBE, 1, -1}, {sunsubscribeCommand, CMD_CHANNEL_UNSUBSCRIBE, 1, -1}, {psubscribeCommand, CMD_CHANNEL_PATTERN | CMD_CHANNEL_SUBSCRIBE, 1, -1}, {punsubscribeCommand, CMD_CHANNEL_PATTERN | CMD_CHANNEL_UNSUBSCRIBE, 1, -1}, {publishCommand, CMD_CHANNEL_PUBLISH, 1, 1}, {spublishCommand, CMD_CHANNEL_PUBLISH, 1, 1}, {NULL,0} /* Terminator. */ }; /* Returns 1 if the command may access any channels matched by the flags * argument. */ int doesCommandHaveChannelsWithFlags(struct redisCommand *cmd, int flags) { /* If a module declares get channels, we are just going to assume * has channels. This API is allowed to return false positives. */ if (cmd->flags & CMD_MODULE_GETCHANNELS) { return 1; } for (ChannelSpecs *spec = commands_with_channels; spec->proc != NULL; spec += 1) { if (cmd->proc == spec->proc) { return !!(spec->flags & flags); } } return 0; } /* Return all the arguments that are channels in the command passed via argc / argv. * This function behaves similar to getKeysFromCommandWithSpecs, but with channels * instead of keys. * * The command returns the positions of all the channel arguments inside the array, * so the actual return value is a heap allocated array of integers. The * length of the array is returned by reference into *numkeys. * * Along with the position, this command also returns the flags that are * associated with how Redis will access the channel. * * 'cmd' must be point to the corresponding entry into the redisCommand * table, according to the command name in argv[0]. */ int getChannelsFromCommand(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result) { keyReference *keys; /* If a module declares get channels, use that. */ if (cmd->flags & CMD_MODULE_GETCHANNELS) { return moduleGetCommandChannelsViaAPI(cmd, argv, argc, result); } /* Otherwise check the channel spec table */ for (ChannelSpecs *spec = commands_with_channels; spec != NULL; spec += 1) { if (cmd->proc == spec->proc) { int start = spec->start; int stop = (spec->count == -1) ? argc : start + spec->count; if (stop > argc) stop = argc; int count = 0; keys = getKeysPrepareResult(result, stop - start); for (int i = start; i < stop; i++ ) { keys[count].pos = i; keys[count++].flags = spec->flags; } result->numkeys = count; return count; } } return 0; } /* The base case is to use the keys position as given in the command table * (firstkey, lastkey, step). * This function works only on command with the legacy_range_key_spec, * all other commands should be handled by getkeys_proc. * * If the commands keyspec is incomplete, no keys will be returned, and the provided * keys function should be called instead. * * NOTE: This function does not guarantee populating the flags for * the keys, in order to get flags you should use getKeysUsingKeySpecs. */ int getKeysUsingLegacyRangeSpec(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result) { int j, i = 0, last, first, step; keyReference *keys; UNUSED(argv); if (cmd->legacy_range_key_spec.begin_search_type == KSPEC_BS_INVALID) { result->numkeys = 0; return 0; } first = cmd->legacy_range_key_spec.bs.index.pos; last = cmd->legacy_range_key_spec.fk.range.lastkey; if (last >= 0) last += first; step = cmd->legacy_range_key_spec.fk.range.keystep; if (last < 0) last = argc+last; int count = ((last - first)+1); keys = getKeysPrepareResult(result, count); for (j = first; j <= last; j += step) { if (j >= argc || j < first) { /* Modules commands, and standard commands with a not fixed number * of arguments (negative arity parameter) do not have dispatch * time arity checks, so we need to handle the case where the user * passed an invalid number of arguments here. In this case we * return no keys and expect the command implementation to report * an arity or syntax error. */ if (cmd->flags & CMD_MODULE || cmd->arity < 0) { result->numkeys = 0; return 0; } else { serverPanic("Redis built-in command declared keys positions not matching the arity requirements."); } } keys[i].pos = j; /* Flags are omitted from legacy key specs */ keys[i++].flags = 0; } result->numkeys = i; return i; } /* Return all the arguments that are keys in the command passed via argc / argv. * * The command returns the positions of all the key arguments inside the array, * so the actual return value is a heap allocated array of integers. The * length of the array is returned by reference into *numkeys. * * 'cmd' must be point to the corresponding entry into the redisCommand * table, according to the command name in argv[0]. * * This function uses the command table if a command-specific helper function * is not required, otherwise it calls the command-specific function. */ int getKeysFromCommand(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result) { if (cmd->flags & CMD_MODULE_GETKEYS) { return moduleGetCommandKeysViaAPI(cmd,argv,argc,result); } else if (!(cmd->flags & CMD_MODULE) && cmd->getkeys_proc) { return cmd->getkeys_proc(cmd,argv,argc,result); } else { return getKeysUsingLegacyRangeSpec(cmd,argv,argc,result); } } /* Free the result of getKeysFromCommand. */ void getKeysFreeResult(getKeysResult *result) { if (result && result->keys != result->keysbuf) zfree(result->keys); } /* Helper function to extract keys from following commands: * COMMAND [destkey] [...] [...] ... * * eg: * ZUNION ... * ZUNIONSTORE ... * * 'storeKeyOfs': destkey index, 0 means destkey not exists. * 'keyCountOfs': num-keys index. * 'firstKeyOfs': firstkey index. * 'keyStep': the interval of each key, usually this value is 1. * * The commands using this functoin have a fully defined keyspec, so returning flags isn't needed. */ int genericGetKeys(int storeKeyOfs, int keyCountOfs, int firstKeyOfs, int keyStep, robj **argv, int argc, getKeysResult *result) { int i, num; keyReference *keys; num = atoi(argv[keyCountOfs]->ptr); /* Sanity check. Don't return any key if the command is going to * reply with syntax error. (no input keys). */ if (num < 1 || num > (argc - firstKeyOfs)/keyStep) { result->numkeys = 0; return 0; } int numkeys = storeKeyOfs ? num + 1 : num; keys = getKeysPrepareResult(result, numkeys); result->numkeys = numkeys; /* Add all key positions for argv[firstKeyOfs...n] to keys[] */ for (i = 0; i < num; i++) { keys[i].pos = firstKeyOfs+(i*keyStep); keys[i].flags = 0; } if (storeKeyOfs) { keys[num].pos = storeKeyOfs; keys[num].flags = 0; } return result->numkeys; } int sintercardGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result) { UNUSED(cmd); return genericGetKeys(0, 1, 2, 1, argv, argc, result); } int zunionInterDiffStoreGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result) { UNUSED(cmd); return genericGetKeys(1, 2, 3, 1, argv, argc, result); } int zunionInterDiffGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result) { UNUSED(cmd); return genericGetKeys(0, 1, 2, 1, argv, argc, result); } int evalGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result) { UNUSED(cmd); return genericGetKeys(0, 2, 3, 1, argv, argc, result); } int functionGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result) { UNUSED(cmd); return genericGetKeys(0, 2, 3, 1, argv, argc, result); } int lmpopGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result) { UNUSED(cmd); return genericGetKeys(0, 1, 2, 1, argv, argc, result); } int blmpopGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result) { UNUSED(cmd); return genericGetKeys(0, 2, 3, 1, argv, argc, result); } int zmpopGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result) { UNUSED(cmd); return genericGetKeys(0, 1, 2, 1, argv, argc, result); } int bzmpopGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result) { UNUSED(cmd); return genericGetKeys(0, 2, 3, 1, argv, argc, result); } /* Helper function to extract keys from the SORT RO command. * * SORT * * The second argument of SORT is always a key, however an arbitrary number of * keys may be accessed while doing the sort (the BY and GET args), so the * key-spec declares incomplete keys which is why we have to provide a concrete * implementation to fetch the keys. * * This command declares incomplete keys, so the flags are correctly set for this function */ int sortROGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result) { keyReference *keys; UNUSED(cmd); UNUSED(argv); UNUSED(argc); keys = getKeysPrepareResult(result, 1); keys[0].pos = 1; /* is always present. */ keys[0].flags = CMD_KEY_RO | CMD_KEY_ACCESS; return 1; } /* Helper function to extract keys from the SORT command. * * SORT ... STORE ... * * The first argument of SORT is always a key, however a list of options * follow in SQL-alike style. Here we parse just the minimum in order to * correctly identify keys in the "STORE" option. * * This command declares incomplete keys, so the flags are correctly set for this function */ int sortGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result) { int i, j, num, found_store = 0; keyReference *keys; UNUSED(cmd); num = 0; keys = getKeysPrepareResult(result, 2); /* Alloc 2 places for the worst case. */ keys[num].pos = 1; /* is always present. */ keys[num++].flags = CMD_KEY_RO | CMD_KEY_ACCESS; /* Search for STORE option. By default we consider options to don't * have arguments, so if we find an unknown option name we scan the * next. However there are options with 1 or 2 arguments, so we * provide a list here in order to skip the right number of args. */ struct { char *name; int skip; } skiplist[] = { {"limit", 2}, {"get", 1}, {"by", 1}, {NULL, 0} /* End of elements. */ }; for (i = 2; i < argc; i++) { for (j = 0; skiplist[j].name != NULL; j++) { if (!strcasecmp(argv[i]->ptr,skiplist[j].name)) { i += skiplist[j].skip; break; } else if (!strcasecmp(argv[i]->ptr,"store") && i+1 < argc) { /* Note: we don't increment "num" here and continue the loop * to be sure to process the *last* "STORE" option if multiple * ones are provided. This is same behavior as SORT. */ found_store = 1; keys[num].pos = i+1; /* */ keys[num].flags = CMD_KEY_OW | CMD_KEY_UPDATE; break; } } } result->numkeys = num + found_store; return result->numkeys; } /* This command declares incomplete keys, so the flags are correctly set for this function */ int migrateGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result) { int i, num, first; keyReference *keys; UNUSED(cmd); /* Assume the obvious form. */ first = 3; num = 1; /* But check for the extended one with the KEYS option. */ if (argc > 6) { for (i = 6; i < argc; i++) { if (!strcasecmp(argv[i]->ptr,"keys") && sdslen(argv[3]->ptr) == 0) { first = i+1; num = argc-first; break; } } } keys = getKeysPrepareResult(result, num); for (i = 0; i < num; i++) { keys[i].pos = first+i; keys[i].flags = CMD_KEY_RW | CMD_KEY_ACCESS | CMD_KEY_DELETE; } result->numkeys = num; return num; } /* Helper function to extract keys from following commands: * GEORADIUS key x y radius unit [WITHDIST] [WITHHASH] [WITHCOORD] [ASC|DESC] * [COUNT count] [STORE key] [STOREDIST key] * GEORADIUSBYMEMBER key member radius unit ... options ... * * This command has a fully defined keyspec, so returning flags isn't needed. */ int georadiusGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result) { int i, num; keyReference *keys; UNUSED(cmd); /* Check for the presence of the stored key in the command */ int stored_key = -1; for (i = 5; i < argc; i++) { char *arg = argv[i]->ptr; /* For the case when user specifies both "store" and "storedist" options, the * second key specified would override the first key. This behavior is kept * the same as in georadiusCommand method. */ if ((!strcasecmp(arg, "store") || !strcasecmp(arg, "storedist")) && ((i+1) < argc)) { stored_key = i+1; i++; } } num = 1 + (stored_key == -1 ? 0 : 1); /* Keys in the command come from two places: * argv[1] = key, * argv[5...n] = stored key if present */ keys = getKeysPrepareResult(result, num); /* Add all key positions to keys[] */ keys[0].pos = 1; keys[0].flags = 0; if(num > 1) { keys[1].pos = stored_key; keys[1].flags = 0; } result->numkeys = num; return num; } /* XREAD [BLOCK ] [COUNT ] [GROUP ] * STREAMS key_1 key_2 ... key_N ID_1 ID_2 ... ID_N * * This command has a fully defined keyspec, so returning flags isn't needed. */ int xreadGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result) { int i, num = 0; keyReference *keys; UNUSED(cmd); /* We need to parse the options of the command in order to seek the first * "STREAMS" string which is actually the option. This is needed because * "STREAMS" could also be the name of the consumer group and even the * name of the stream key. */ int streams_pos = -1; for (i = 1; i < argc; i++) { char *arg = argv[i]->ptr; if (!strcasecmp(arg, "block")) { i++; /* Skip option argument. */ } else if (!strcasecmp(arg, "count")) { i++; /* Skip option argument. */ } else if (!strcasecmp(arg, "group")) { i += 2; /* Skip option argument. */ } else if (!strcasecmp(arg, "noack")) { /* Nothing to do. */ } else if (!strcasecmp(arg, "streams")) { streams_pos = i; break; } else { break; /* Syntax error. */ } } if (streams_pos != -1) num = argc - streams_pos - 1; /* Syntax error. */ if (streams_pos == -1 || num == 0 || num % 2 != 0) { result->numkeys = 0; return 0; } num /= 2; /* We have half the keys as there are arguments because there are also the IDs, one per key. */ keys = getKeysPrepareResult(result, num); for (i = streams_pos+1; i < argc-num; i++) { keys[i-streams_pos-1].pos = i; keys[i-streams_pos-1].flags = 0; } result->numkeys = num; return num; } /* Helper function to extract keys from the SET command, which may have * a read flag if the GET argument is passed in. */ int setGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result) { keyReference *keys; UNUSED(cmd); keys = getKeysPrepareResult(result, 1); keys[0].pos = 1; /* We always know the position */ result->numkeys = 1; for (int i = 3; i < argc; i++) { char *arg = argv[i]->ptr; if ((arg[0] == 'g' || arg[0] == 'G') && (arg[1] == 'e' || arg[1] == 'E') && (arg[2] == 't' || arg[2] == 'T') && arg[3] == '\0') { keys[0].flags = CMD_KEY_RW | CMD_KEY_ACCESS | CMD_KEY_UPDATE; return 1; } } keys[0].flags = CMD_KEY_OW | CMD_KEY_UPDATE; return 1; } /* Helper function to extract keys from the BITFIELD command, which may be * read-only if the BITFIELD GET subcommand is used. */ int bitfieldGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result) { keyReference *keys; UNUSED(cmd); keys = getKeysPrepareResult(result, 1); keys[0].pos = 1; /* We always know the position */ result->numkeys = 1; for (int i = 2; i < argc; i++) { int remargs = argc - i - 1; /* Remaining args other than current. */ char *arg = argv[i]->ptr; if (!strcasecmp(arg, "get") && remargs >= 2) { keys[0].flags = CMD_KEY_RO | CMD_KEY_ACCESS; return 1; } } keys[0].flags = CMD_KEY_RW | CMD_KEY_ACCESS | CMD_KEY_UPDATE; return 1; }