Unverified Commit acf3495e authored by Viktor Söderqvist's avatar Viktor Söderqvist Committed by GitHub
Browse files

Sort out the mess around writable replicas and lookupKeyRead/Write (#9572)

Writable replicas now no longer use the values of expired keys. Expired keys are
deleted when lookupKeyWrite() is used, even on a writable replica. Previously,
writable replicas could use the value of an expired key in write commands such
as INCR, SUNIONSTORE, etc..

This commit also sorts out the mess around the functions lookupKeyRead() and
lookupKeyWrite() so they now indicate what we intend to do with the key and
are not affected by the command calling them.

Multi-key commands like SUNIONSTORE, ZUNIONSTORE, COPY and SORT with the
store option now use lookupKeyRead() for the keys they're reading from (which will
not allow reading from logically expired keys).

This commit also fixes a bug where PFCOUNT could return a value of an
expired key.

Test modules commands have their readonly and write flags updated to correctly
reflect their lookups for reading or writing. Modules are not required to
correctly reflect this in their command flags, but this change is made for
consistency since the tests serve as usage examples.

Fixes #6842. Fixes #7475.
parent 4d870078
...@@ -737,7 +737,8 @@ int loadAppendOnlyFile(char *filename) { ...@@ -737,7 +737,8 @@ int loadAppendOnlyFile(char *filename) {
* to the same file we're about to read. */ * to the same file we're about to read. */
server.aof_state = AOF_OFF; server.aof_state = AOF_OFF;
fakeClient = createAOFClient(); client *old_client = server.current_client;
fakeClient = server.current_client = createAOFClient();
startLoadingFile(fp, filename, RDBFLAGS_AOF_PREAMBLE); startLoadingFile(fp, filename, RDBFLAGS_AOF_PREAMBLE);
/* Check if this AOF file has an RDB preamble. In that case we need to /* Check if this AOF file has an RDB preamble. In that case we need to
...@@ -928,6 +929,7 @@ fmterr: /* Format error. */ ...@@ -928,6 +929,7 @@ fmterr: /* Format error. */
cleanup: cleanup:
if (fakeClient) freeClient(fakeClient); if (fakeClient) freeClient(fakeClient);
server.current_client = old_client;
fclose(fp); fclose(fp);
stopLoading(ret == AOF_OK); stopLoading(ret == AOF_OK);
return ret; return ret;
......
...@@ -591,8 +591,7 @@ void handleClientsBlockedOnKeys(void) { ...@@ -591,8 +591,7 @@ void handleClientsBlockedOnKeys(void) {
updateCachedTime(0); updateCachedTime(0);
/* Serve clients blocked on the key. */ /* Serve clients blocked on the key. */
robj *o = lookupKeyWrite(rl->db,rl->key); robj *o = lookupKeyReadWithFlags(rl->db, rl->key, LOOKUP_NONOTIFY | LOOKUP_NOSTATS);
if (o != NULL) { if (o != NULL) {
if (o->type == OBJ_LIST) if (o->type == OBJ_LIST)
serveClientsBlockedOnListKey(o,rl); serveClientsBlockedOnListKey(o,rl);
......
...@@ -6012,8 +6012,9 @@ clusterNode *getNodeByQuery(client *c, struct redisCommand *cmd, robj **argv, in ...@@ -6012,8 +6012,9 @@ clusterNode *getNodeByQuery(client *c, struct redisCommand *cmd, robj **argv, in
} }
/* Migrating / Importing slot? Count keys we don't have. */ /* Migrating / Importing slot? Count keys we don't have. */
int flags = LOOKUP_NOTOUCH | LOOKUP_NOSTATS | LOOKUP_NONOTIFY;
if ((migrating_slot || importing_slot) && if ((migrating_slot || importing_slot) &&
lookupKeyRead(&server.db[0],thiskey) == NULL) lookupKeyReadWithFlags(&server.db[0], thiskey, flags) == NULL)
{ {
missing_keys++; missing_keys++;
} }
......
...@@ -39,6 +39,7 @@ ...@@ -39,6 +39,7 @@
* C-level DB API * C-level DB API
*----------------------------------------------------------------------------*/ *----------------------------------------------------------------------------*/
int expireIfNeeded(redisDb *db, robj *key, int force_delete_expired);
int keyIsExpired(redisDb *db, robj *key); int keyIsExpired(redisDb *db, robj *key);
/* Update LFU when an object is accessed. /* Update LFU when an object is accessed.
...@@ -50,14 +51,53 @@ void updateLFU(robj *val) { ...@@ -50,14 +51,53 @@ void updateLFU(robj *val) {
val->lru = (LFUGetTimeInMinutes()<<8) | counter; val->lru = (LFUGetTimeInMinutes()<<8) | counter;
} }
/* Low level key lookup API, not actually called directly from commands /* Lookup a key for read or write operations, or return NULL if the key is not
* implementations that should instead rely on lookupKeyRead(), * found in the specified DB. This function implements the functionality of
* lookupKeyWrite() and lookupKeyReadWithFlags(). */ * 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 couters.
* 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) { robj *lookupKey(redisDb *db, robj *key, int flags) {
dictEntry *de = dictFind(db->dict,key->ptr); dictEntry *de = dictFind(db->dict,key->ptr);
robj *val = NULL;
if (de) { if (de) {
robj *val = dictGetVal(de); val = dictGetVal(de);
int force_delete_expired = flags & LOOKUP_WRITE;
if (force_delete_expired) {
/* Forcing deletion of expired keys on a replica makes the replica
* inconsistent with the master. The reason it's allowed for write
* commands is to make writable replicas behave consistently. It
* shall not be used in readonly commands. Modules are accepted so
* that we don't break old modules. */
client *c = server.in_eval ? server.lua_client : server.current_client;
serverAssert(!c || !c->cmd || (c->cmd->flags & (CMD_WRITE|CMD_MODULE)));
}
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. /* Update the access time for the ageing algorithm.
* Don't do it if we have a saving child, as this will trigger * Don't do it if we have a saving child, as this will trigger
* a copy on write madness. */ * a copy on write madness. */
...@@ -68,75 +108,33 @@ robj *lookupKey(redisDb *db, robj *key, int flags) { ...@@ -68,75 +108,33 @@ robj *lookupKey(redisDb *db, robj *key, int flags) {
val->lru = LRU_CLOCK(); val->lru = LRU_CLOCK();
} }
} }
return val;
if (!(flags & (LOOKUP_NOSTATS | LOOKUP_WRITE)))
server.stat_keyspace_hits++;
/* TODO: Use separate hits stats for WRITE */
} else { } else {
return NULL; 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 /* Lookup a key for read operations, or return NULL if the key is not found
* in the specified DB. * in the specified DB.
* *
* As a side effect of calling this function:
* 1. A key gets expired if it reached it's TTL.
* 2. The key 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.
*
* This API should not be used when we write to the key after obtaining * 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. * the object linked to the key, but only for read only operations.
* *
* Flags change the behavior of this command: * 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
* LOOKUP_NONE (or zero): no special flags are passed. * the key. */
* LOOKUP_NOTOUCH: don't alter the last access time of the key.
*
* Note: this function also returns NULL if the key is logically expired
* but still existing, in case this is a slave, since this API is called only
* for read operations. Even if the key expiry is master-driven, we can
* correctly report a key is expired on slaves even if the master is lagging
* expiring our key via DELs in the replication link. */
robj *lookupKeyReadWithFlags(redisDb *db, robj *key, int flags) { robj *lookupKeyReadWithFlags(redisDb *db, robj *key, int flags) {
robj *val; serverAssert(!(flags & LOOKUP_WRITE));
return lookupKey(db, key, flags);
if (expireIfNeeded(db,key) == 1) {
/* If we are in the context of a master, expireIfNeeded() returns 1
* when the key is no longer valid, so we can return NULL ASAP. */
if (server.masterhost == NULL)
goto keymiss;
/* However if we are in the context of a slave, expireIfNeeded() will
* not really try to expire the key, it only returns information
* about the "logical" status of the key: key expiring is up to the
* master in order to have a consistent view of master's data set.
*
* However, if the command caller is not the master, and as additional
* safety measure, the command invoked is a read-only command, we can
* safely return NULL here, and provide a more consistent behavior
* to clients accessing expired values in a read-only fashion, that
* will say the key as non existing.
*
* Notably this covers GETs when slaves are used to scale reads. */
if (server.current_client &&
server.current_client != server.master &&
server.current_client->cmd &&
server.current_client->cmd->flags & CMD_READONLY)
{
goto keymiss;
}
}
val = lookupKey(db,key,flags);
if (val == NULL)
goto keymiss;
server.stat_keyspace_hits++;
return val;
keymiss:
if (!(flags & LOOKUP_NONOTIFY)) {
notifyKeyspaceEvent(NOTIFY_KEY_MISS, "keymiss", key, db->id);
}
server.stat_keyspace_misses++;
return NULL;
} }
/* Like lookupKeyReadWithFlags(), but does not use any flag, which is the /* Like lookupKeyReadWithFlags(), but does not use any flag, which is the
...@@ -146,13 +144,13 @@ robj *lookupKeyRead(redisDb *db, robj *key) { ...@@ -146,13 +144,13 @@ robj *lookupKeyRead(redisDb *db, robj *key) {
} }
/* Lookup a key for write operations, and as a side effect, if needed, expires /* Lookup a key for write operations, and as a side effect, if needed, expires
* the key if its TTL is reached. * 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 * Returns the linked value object if the key exists or NULL if the key
* does not exist in the specified DB. */ * does not exist in the specified DB. */
robj *lookupKeyWriteWithFlags(redisDb *db, robj *key, int flags) { robj *lookupKeyWriteWithFlags(redisDb *db, robj *key, int flags) {
expireIfNeeded(db,key); return lookupKey(db, key, flags | LOOKUP_WRITE);
return lookupKey(db,key,flags);
} }
robj *lookupKeyWrite(redisDb *db, robj *key) { robj *lookupKeyWrite(redisDb *db, robj *key) {
...@@ -292,7 +290,7 @@ robj *dbRandomKey(redisDb *db) { ...@@ -292,7 +290,7 @@ robj *dbRandomKey(redisDb *db) {
* return a key name that may be already expired. */ * return a key name that may be already expired. */
return keyobj; return keyobj;
} }
if (expireIfNeeded(db,keyobj)) { if (expireIfNeeded(db,keyobj,0)) {
decrRefCount(keyobj); decrRefCount(keyobj);
continue; /* search for another key. This expired. */ continue; /* search for another key. This expired. */
} }
...@@ -641,7 +639,7 @@ void delGenericCommand(client *c, int lazy) { ...@@ -641,7 +639,7 @@ void delGenericCommand(client *c, int lazy) {
int numdel = 0, j; int numdel = 0, j;
for (j = 1; j < c->argc; j++) { for (j = 1; j < c->argc; j++) {
expireIfNeeded(c->db,c->argv[j]); expireIfNeeded(c->db,c->argv[j],0);
int deleted = lazy ? dbAsyncDelete(c->db,c->argv[j]) : int deleted = lazy ? dbAsyncDelete(c->db,c->argv[j]) :
dbSyncDelete(c->db,c->argv[j]); dbSyncDelete(c->db,c->argv[j]);
if (deleted) { if (deleted) {
...@@ -939,7 +937,7 @@ void scanGenericCommand(client *c, robj *o, unsigned long cursor) { ...@@ -939,7 +937,7 @@ void scanGenericCommand(client *c, robj *o, unsigned long cursor) {
} }
/* Filter element if it is an expired key. */ /* Filter element if it is an expired key. */
if (!filter && o == NULL && expireIfNeeded(c->db, kobj)) filter = 1; if (!filter && o == NULL && expireIfNeeded(c->db, kobj, 0)) filter = 1;
/* Remove the element and its associated value if needed. */ /* Remove the element and its associated value if needed. */
if (filter) { if (filter) {
...@@ -1206,7 +1204,7 @@ void copyCommand(client *c) { ...@@ -1206,7 +1204,7 @@ void copyCommand(client *c) {
} }
/* Check if the element exists and get a reference */ /* Check if the element exists and get a reference */
o = lookupKeyWrite(c->db, key); o = lookupKeyRead(c->db, key);
if (!o) { if (!o) {
addReply(c,shared.czero); addReply(c,shared.czero);
return; return;
...@@ -1266,8 +1264,11 @@ void scanDatabaseForReadyLists(redisDb *db) { ...@@ -1266,8 +1264,11 @@ void scanDatabaseForReadyLists(redisDb *db) {
dictIterator *di = dictGetSafeIterator(db->blocking_keys); dictIterator *di = dictGetSafeIterator(db->blocking_keys);
while((de = dictNext(di)) != NULL) { while((de = dictNext(di)) != NULL) {
robj *key = dictGetKey(de); robj *key = dictGetKey(de);
robj *value = lookupKey(db,key,LOOKUP_NOTOUCH); dictEntry *kde = dictFind(db->dict,key->ptr);
if (value) signalKeyAsReady(db, key, value->type); if (kde) {
robj *value = dictGetVal(kde);
signalKeyAsReady(db, key, value->type);
}
} }
dictReleaseIterator(di); dictReleaseIterator(di);
} }
...@@ -1529,10 +1530,10 @@ int keyIsExpired(redisDb *db, robj *key) { ...@@ -1529,10 +1530,10 @@ int keyIsExpired(redisDb *db, robj *key) {
* is via lookupKey*() family of functions. * is via lookupKey*() family of functions.
* *
* The behavior of the function depends on the replication role of the * The behavior of the function depends on the replication role of the
* instance, because slave instances do not expire keys, they wait * instance, because by default replicas do not delete expired keys. They
* for DELs from the master for consistency matters. However even * wait for DELs from the master for consistency matters. However even
* slaves will try to have a coherent return value for the function, * replicas will try to have a coherent return value for the function,
* so that read commands executed in the slave side will be able to * 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 * behave like if the key is expired even if still present (because the
* master has yet to propagate the DEL). * master has yet to propagate the DEL).
* *
...@@ -1540,20 +1541,34 @@ int keyIsExpired(redisDb *db, robj *key) { ...@@ -1540,20 +1541,34 @@ int keyIsExpired(redisDb *db, robj *key) {
* key will be evicted from the database. Also this may trigger the * key will be evicted from the database. Also this may trigger the
* propagation of a DEL/UNLINK command in AOF / replication stream. * 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, * The return value of the function is 0 if the key is still valid,
* otherwise the function returns 1 if the key is expired. */ * otherwise the function returns 1 if the key is expired. */
int expireIfNeeded(redisDb *db, robj *key) { int expireIfNeeded(redisDb *db, robj *key, int force_delete_expired) {
if (!keyIsExpired(db,key)) return 0; if (!keyIsExpired(db,key)) return 0;
/* If we are running in the context of a slave, instead of /* If we are running in the context of a replica, instead of
* evicting the expired key from the database, we return ASAP: * evicting the expired key from the database, we return ASAP:
* the slave key expiration is controlled by the master that will * the replica key expiration is controlled by the master that will
* send us synthesized DEL operations for expired keys. * 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, * 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 * that is, 0 if we think the key should be still valid, 1 if
* we think the key is expired at this time. */ * we think the key is expired at this time.
if (server.masterhost != NULL) return 1; *
* 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, /* If clients are paused, we keep the current dataset constant,
* but return to the client what we believe is the right state. Typically, * but return to the client what we believe is the right state. Typically,
......
...@@ -1258,8 +1258,15 @@ void pfcountCommand(client *c) { ...@@ -1258,8 +1258,15 @@ void pfcountCommand(client *c) {
/* Case 2: cardinality of the single HLL. /* Case 2: cardinality of the single HLL.
* *
* The user specified a single key. Either return the cached value * The user specified a single key. Either return the cached value
* or compute one and update the cache. */ * or compute one and update the cache.
o = lookupKeyWrite(c->db,c->argv[1]); *
* Since a HLL is a regular Redis string type value, updating the cache does
* modify the value. We do a lookupKeyRead anyway since this is flagged as a
* read-only command. The difference is that with lookupKeyWrite, a
* logically expired key on a replica is deleted, while with lookupKeyRead
* it isn't, but the lookup returns NULL either way if the key is logically
* expired, which is what matters here. */
o = lookupKeyRead(c->db,c->argv[1]);
if (o == NULL) { if (o == NULL) {
/* No key? Cardinality is zero since no element was added, otherwise /* No key? Cardinality is zero since no element was added, otherwise
* we would have a key as HLLADD creates it as a side effect. */ * we would have a key as HLLADD creates it as a side effect. */
...@@ -1296,8 +1303,7 @@ void pfcountCommand(client *c) { ...@@ -1296,8 +1303,7 @@ void pfcountCommand(client *c) {
hdr->card[5] = (card >> 40) & 0xff; hdr->card[5] = (card >> 40) & 0xff;
hdr->card[6] = (card >> 48) & 0xff; hdr->card[6] = (card >> 48) & 0xff;
hdr->card[7] = (card >> 56) & 0xff; hdr->card[7] = (card >> 56) & 0xff;
/* This is not considered a read-only command even if the /* This is considered a read-only command even if the cached value
* data structure is not modified, since the cached value
* may be modified and given that the HLL is a Redis string * may be modified and given that the HLL is a Redis string
* we need to propagate the change. */ * we need to propagate the change. */
signalModifiedKey(c,c->db,c->argv[1]); signalModifiedKey(c,c->db,c->argv[1]);
......
...@@ -2623,11 +2623,9 @@ int removeExpire(redisDb *db, robj *key); ...@@ -2623,11 +2623,9 @@ int removeExpire(redisDb *db, robj *key);
void deleteExpiredKeyAndPropagate(redisDb *db, robj *keyobj); void deleteExpiredKeyAndPropagate(redisDb *db, robj *keyobj);
void propagateExpire(redisDb *db, robj *key, int lazy); void propagateExpire(redisDb *db, robj *key, int lazy);
int keyIsExpired(redisDb *db, robj *key); int keyIsExpired(redisDb *db, robj *key);
int expireIfNeeded(redisDb *db, robj *key);
long long getExpire(redisDb *db, robj *key); long long getExpire(redisDb *db, robj *key);
void setExpire(client *c, redisDb *db, robj *key, long long when); void setExpire(client *c, redisDb *db, robj *key, long long when);
int checkAlreadyExpired(long long when); int checkAlreadyExpired(long long when);
robj *lookupKey(redisDb *db, robj *key, int flags);
robj *lookupKeyRead(redisDb *db, robj *key); robj *lookupKeyRead(redisDb *db, robj *key);
robj *lookupKeyWrite(redisDb *db, robj *key); robj *lookupKeyWrite(redisDb *db, robj *key);
robj *lookupKeyReadOrReply(client *c, robj *key, robj *reply); robj *lookupKeyReadOrReply(client *c, robj *key, robj *reply);
...@@ -2639,8 +2637,11 @@ robj *objectCommandLookupOrReply(client *c, robj *key, robj *reply); ...@@ -2639,8 +2637,11 @@ robj *objectCommandLookupOrReply(client *c, robj *key, robj *reply);
int objectSetLRUOrLFU(robj *val, long long lfu_freq, long long lru_idle, int objectSetLRUOrLFU(robj *val, long long lfu_freq, long long lru_idle,
long long lru_clock, int lru_multiplier); long long lru_clock, int lru_multiplier);
#define LOOKUP_NONE 0 #define LOOKUP_NONE 0
#define LOOKUP_NOTOUCH (1<<0) #define LOOKUP_NOTOUCH (1<<0) /* Don't update LRU. */
#define LOOKUP_NONOTIFY (1<<1) #define LOOKUP_NONOTIFY (1<<1) /* Don't trigger keyspace event on key misses. */
#define LOOKUP_NOSTATS (1<<2) /* Don't update keyspace hits/misses couters. */
#define LOOKUP_WRITE (1<<3) /* Delete expired keys even in replicas. */
void dbAdd(redisDb *db, robj *key, robj *val); void dbAdd(redisDb *db, robj *key, robj *val);
int dbAddRDBLoad(redisDb *db, sds key, robj *val); int dbAddRDBLoad(redisDb *db, sds key, robj *val);
void dbOverwrite(redisDb *db, robj *key, robj *val); void dbOverwrite(redisDb *db, robj *key, robj *val);
......
...@@ -58,7 +58,7 @@ redisSortOperation *createSortOperation(int type, robj *pattern) { ...@@ -58,7 +58,7 @@ redisSortOperation *createSortOperation(int type, robj *pattern) {
* *
* The returned object will always have its refcount increased by 1 * The returned object will always have its refcount increased by 1
* when it is non-NULL. */ * when it is non-NULL. */
robj *lookupKeyByPattern(redisDb *db, robj *pattern, robj *subst, int writeflag) { robj *lookupKeyByPattern(redisDb *db, robj *pattern, robj *subst) {
char *p, *f, *k; char *p, *f, *k;
sds spat, ssub; sds spat, ssub;
robj *keyobj, *fieldobj = NULL, *o; robj *keyobj, *fieldobj = NULL, *o;
...@@ -106,10 +106,7 @@ robj *lookupKeyByPattern(redisDb *db, robj *pattern, robj *subst, int writeflag) ...@@ -106,10 +106,7 @@ robj *lookupKeyByPattern(redisDb *db, robj *pattern, robj *subst, int writeflag)
decrRefCount(subst); /* Incremented by decodeObject() */ decrRefCount(subst); /* Incremented by decodeObject() */
/* Lookup substituted key */ /* Lookup substituted key */
if (!writeflag) o = lookupKeyRead(db, keyobj);
o = lookupKeyRead(db,keyobj);
else
o = lookupKeyWrite(db,keyobj);
if (o == NULL) goto noobj; if (o == NULL) goto noobj;
if (fieldobj) { if (fieldobj) {
...@@ -270,10 +267,7 @@ void sortCommandGeneric(client *c, int readonly) { ...@@ -270,10 +267,7 @@ void sortCommandGeneric(client *c, int readonly) {
} }
/* Lookup the key to sort. It must be of the right types */ /* Lookup the key to sort. It must be of the right types */
if (!storekey) sortval = lookupKeyRead(c->db, c->argv[1]);
sortval = lookupKeyRead(c->db,c->argv[1]);
else
sortval = lookupKeyWrite(c->db,c->argv[1]);
if (sortval && sortval->type != OBJ_SET && if (sortval && sortval->type != OBJ_SET &&
sortval->type != OBJ_LIST && sortval->type != OBJ_LIST &&
sortval->type != OBJ_ZSET) sortval->type != OBJ_ZSET)
...@@ -459,7 +453,7 @@ void sortCommandGeneric(client *c, int readonly) { ...@@ -459,7 +453,7 @@ void sortCommandGeneric(client *c, int readonly) {
robj *byval; robj *byval;
if (sortby) { if (sortby) {
/* lookup value to sort by */ /* lookup value to sort by */
byval = lookupKeyByPattern(c->db,sortby,vector[j].obj,storekey!=NULL); byval = lookupKeyByPattern(c->db,sortby,vector[j].obj);
if (!byval) continue; if (!byval) continue;
} else { } else {
/* use object itself to sort by */ /* use object itself to sort by */
...@@ -522,7 +516,7 @@ void sortCommandGeneric(client *c, int readonly) { ...@@ -522,7 +516,7 @@ void sortCommandGeneric(client *c, int readonly) {
while((ln = listNext(&li))) { while((ln = listNext(&li))) {
redisSortOperation *sop = ln->value; redisSortOperation *sop = ln->value;
robj *val = lookupKeyByPattern(c->db,sop->pattern, robj *val = lookupKeyByPattern(c->db,sop->pattern,
vector[j].obj,storekey!=NULL); vector[j].obj);
if (sop->type == SORT_OP_GET) { if (sop->type == SORT_OP_GET) {
if (!val) { if (!val) {
...@@ -552,7 +546,7 @@ void sortCommandGeneric(client *c, int readonly) { ...@@ -552,7 +546,7 @@ void sortCommandGeneric(client *c, int readonly) {
while((ln = listNext(&li))) { while((ln = listNext(&li))) {
redisSortOperation *sop = ln->value; redisSortOperation *sop = ln->value;
robj *val = lookupKeyByPattern(c->db,sop->pattern, robj *val = lookupKeyByPattern(c->db,sop->pattern,
vector[j].obj,storekey!=NULL); vector[j].obj);
if (sop->type == SORT_OP_GET) { if (sop->type == SORT_OP_GET) {
if (!val) val = createStringObject("",0); if (!val) val = createStringObject("",0);
......
...@@ -871,9 +871,7 @@ void sinterGenericCommand(client *c, robj **setkeys, ...@@ -871,9 +871,7 @@ void sinterGenericCommand(client *c, robj **setkeys,
int encoding, empty = 0; int encoding, empty = 0;
for (j = 0; j < setnum; j++) { for (j = 0; j < setnum; j++) {
robj *setobj = dstkey ? robj *setobj = lookupKeyRead(c->db, setkeys[j]);
lookupKeyWrite(c->db,setkeys[j]) :
lookupKeyRead(c->db,setkeys[j]);
if (!setobj) { if (!setobj) {
/* A NULL is considered an empty set */ /* A NULL is considered an empty set */
empty += 1; empty += 1;
...@@ -1059,9 +1057,7 @@ void sunionDiffGenericCommand(client *c, robj **setkeys, int setnum, ...@@ -1059,9 +1057,7 @@ void sunionDiffGenericCommand(client *c, robj **setkeys, int setnum,
int diff_algo = 1; int diff_algo = 1;
for (j = 0; j < setnum; j++) { for (j = 0; j < setnum; j++) {
robj *setobj = dstkey ? robj *setobj = lookupKeyRead(c->db, setkeys[j]);
lookupKeyWrite(c->db,setkeys[j]) :
lookupKeyRead(c->db,setkeys[j]);
if (!setobj) { if (!setobj) {
sets[j] = NULL; sets[j] = NULL;
continue; continue;
......
...@@ -2551,9 +2551,7 @@ void zunionInterDiffGenericCommand(client *c, robj *dstkey, int numkeysIndex, in ...@@ -2551,9 +2551,7 @@ void zunionInterDiffGenericCommand(client *c, robj *dstkey, int numkeysIndex, in
/* read keys to be used for input */ /* read keys to be used for input */
src = zcalloc(sizeof(zsetopsrc) * setnum); src = zcalloc(sizeof(zsetopsrc) * setnum);
for (i = 0, j = numkeysIndex+1; i < setnum; i++, j++) { for (i = 0, j = numkeysIndex+1; i < setnum; i++, j++) {
robj *obj = dstkey ? robj *obj = lookupKeyRead(c->db, c->argv[j]);
lookupKeyWrite(c->db,c->argv[j]) :
lookupKeyRead(c->db,c->argv[j]);
if (obj != NULL) { if (obj != NULL) {
if (obj->type != OBJ_ZSET && obj->type != OBJ_SET) { if (obj->type != OBJ_ZSET && obj->type != OBJ_SET) {
zfree(src); zfree(src);
...@@ -3646,9 +3644,7 @@ void zrangeGenericCommand(zrange_result_handler *handler, int argc_start, int st ...@@ -3646,9 +3644,7 @@ void zrangeGenericCommand(zrange_result_handler *handler, int argc_start, int st
} }
/* Step 3: Lookup the key and get the range. */ /* Step 3: Lookup the key and get the range. */
zobj = handler->dstkey ? zobj = lookupKeyRead(c->db, key);
lookupKeyWrite(c->db,key) :
lookupKeyRead(c->db,key);
if (zobj == NULL) { if (zobj == NULL) {
if (store) { if (store) {
handler->beginResultEmission(handler); handler->beginResultEmission(handler);
......
...@@ -64,6 +64,68 @@ start_server {tags {"repl external:skip"}} { ...@@ -64,6 +64,68 @@ start_server {tags {"repl external:skip"}} {
after 6000 after 6000
r -1 dbsize r -1 dbsize
} {0} } {0}
test {Writable replica doesn't return expired keys} {
r select 5
assert {[r dbsize] == 0}
r debug set-active-expire 0
r set key1 5 px 10
r set key2 5 px 10
r -1 select 5
wait_for_condition 50 100 {
[r -1 dbsize] == 2 && [r -1 exists key1 key2] == 0
} else {
fail "Keys didn't replicate or didn't expire."
}
r -1 config set slave-read-only no
assert_equal 2 [r -1 dbsize] ; # active expire is off
assert_equal 1 [r -1 incr key1] ; # incr expires and re-creates key1
assert_equal -1 [r -1 ttl key1] ; # incr created key1 without TTL
assert_equal {} [r -1 get key2] ; # key2 expired but not deleted
assert_equal 2 [r -1 dbsize]
# cleanup
r debug set-active-expire 1
r -1 del key1 key2
r -1 config set slave-read-only yes
r del key1 key2
}
test {PFCOUNT updates cache on readonly replica} {
r select 5
assert {[r dbsize] == 0}
r pfadd key a b c d e f g h i j k l m n o p q
set strval [r get key]
r -1 select 5
wait_for_condition 50 100 {
[r -1 dbsize] == 1
} else {
fail "Replication timeout."
}
assert {$strval == [r -1 get key]}
assert_equal 17 [r -1 pfcount key]
assert {$strval != [r -1 get key]}; # cache updated
# cleanup
r del key
}
test {PFCOUNT doesn't use expired key on readonly replica} {
r select 5
assert {[r dbsize] == 0}
r debug set-active-expire 0
r pfadd key a b c d e f g h i j k l m n o p q
r pexpire key 10
r -1 select 5
wait_for_condition 50 100 {
[r -1 dbsize] == 1 && [r -1 exists key] == 0
} else {
fail "Key didn't replicate or didn't expire."
}
assert_equal [r -1 pfcount key] 0 ; # expired key not used
assert_equal [r -1 dbsize] 1 ; # but it's also not deleted
# cleanup
r debug set-active-expire 1
r del key
}
} }
} }
......
...@@ -85,6 +85,40 @@ start_server {tags {"repl external:skip"}} { ...@@ -85,6 +85,40 @@ start_server {tags {"repl external:skip"}} {
} }
} }
start_server {tags {"repl external:skip"}} {
start_server {} {
set master [srv -1 client]
set master_host [srv -1 host]
set master_port [srv -1 port]
set slave [srv 0 client]
test {First server should have role slave after SLAVEOF} {
$slave slaveof $master_host $master_port
wait_for_condition 50 100 {
[s 0 master_link_status] eq {up}
} else {
fail "Replication not started."
}
}
test {Replication of an expired key does not delete the expired key} {
$master debug set-active-expire 0
$master set k 1 ex 1
wait_for_ofs_sync $master $slave
exec kill -SIGSTOP [srv 0 pid]
$master incr k
after 1000
# Stopping the replica for one second to makes sure the INCR arrives
# to the replica after the key is logically expired.
exec kill -SIGCONT [srv 0 pid]
wait_for_ofs_sync $master $slave
# Check that k is locigally expired but is present in the replica.
assert_equal 0 [$slave exists k]
$slave debug object k ; # Raises exception if k is gone.
}
}
}
start_server {tags {"repl external:skip"}} { start_server {tags {"repl external:skip"}} {
start_server {} { start_server {} {
set master [srv -1 client] set master [srv -1 client]
......
...@@ -157,7 +157,7 @@ int RedisModule_OnLoad(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) ...@@ -157,7 +157,7 @@ int RedisModule_OnLoad(RedisModuleCtx *ctx, RedisModuleString **argv, int argc)
if (RedisModule_Init(ctx,"aclcheck",1,REDISMODULE_APIVER_1)== REDISMODULE_ERR) if (RedisModule_Init(ctx,"aclcheck",1,REDISMODULE_APIVER_1)== REDISMODULE_ERR)
return REDISMODULE_ERR; return REDISMODULE_ERR;
if (RedisModule_CreateCommand(ctx,"aclcheck.set.check.key", set_aclcheck_key,"",0,0,0) == REDISMODULE_ERR) if (RedisModule_CreateCommand(ctx,"aclcheck.set.check.key", set_aclcheck_key,"write",0,0,0) == REDISMODULE_ERR)
return REDISMODULE_ERR; return REDISMODULE_ERR;
if (RedisModule_CreateCommand(ctx,"aclcheck.publish.check.channel", publish_aclcheck_channel,"",0,0,0) == REDISMODULE_ERR) if (RedisModule_CreateCommand(ctx,"aclcheck.publish.check.channel", publish_aclcheck_channel,"",0,0,0) == REDISMODULE_ERR)
...@@ -169,7 +169,8 @@ int RedisModule_OnLoad(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) ...@@ -169,7 +169,8 @@ int RedisModule_OnLoad(RedisModuleCtx *ctx, RedisModuleString **argv, int argc)
if (RedisModule_CreateCommand(ctx,"aclcheck.rm_call.check.cmd.module.user", rm_call_aclcheck_cmd_module_user,"",0,0,0) == REDISMODULE_ERR) if (RedisModule_CreateCommand(ctx,"aclcheck.rm_call.check.cmd.module.user", rm_call_aclcheck_cmd_module_user,"",0,0,0) == REDISMODULE_ERR)
return REDISMODULE_ERR; return REDISMODULE_ERR;
if (RedisModule_CreateCommand(ctx,"aclcheck.rm_call", rm_call_aclcheck,"",0,0,0) == REDISMODULE_ERR) if (RedisModule_CreateCommand(ctx,"aclcheck.rm_call", rm_call_aclcheck,
"write",0,0,0) == REDISMODULE_ERR)
return REDISMODULE_ERR; return REDISMODULE_ERR;
return REDISMODULE_OK; return REDISMODULE_OK;
......
...@@ -936,12 +936,12 @@ int RedisModule_OnLoad(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) ...@@ -936,12 +936,12 @@ int RedisModule_OnLoad(RedisModuleCtx *ctx, RedisModuleString **argv, int argc)
return REDISMODULE_ERR; return REDISMODULE_ERR;
if (RedisModule_CreateCommand(ctx,"test.basics", if (RedisModule_CreateCommand(ctx,"test.basics",
TestBasics,"readonly",1,1,1) == REDISMODULE_ERR) TestBasics,"write",1,1,1) == REDISMODULE_ERR)
return REDISMODULE_ERR; return REDISMODULE_ERR;
/* the following commands are used by an external test and should not be added to TestBasics */ /* the following commands are used by an external test and should not be added to TestBasics */
if (RedisModule_CreateCommand(ctx,"test.rmcallautomode", if (RedisModule_CreateCommand(ctx,"test.rmcallautomode",
TestCallRespAutoMode,"readonly",1,1,1) == REDISMODULE_ERR) TestCallRespAutoMode,"write",1,1,1) == REDISMODULE_ERR)
return REDISMODULE_ERR; return REDISMODULE_ERR;
if (RedisModule_CreateCommand(ctx,"test.getresp", if (RedisModule_CreateCommand(ctx,"test.getresp",
......
...@@ -214,7 +214,8 @@ int RedisModule_OnLoad(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) ...@@ -214,7 +214,8 @@ int RedisModule_OnLoad(RedisModuleCtx *ctx, RedisModuleString **argv, int argc)
if (RedisModule_CreateCommand(ctx, "acquire_gil", acquire_gil, "", 0, 0, 0) == REDISMODULE_ERR) if (RedisModule_CreateCommand(ctx, "acquire_gil", acquire_gil, "", 0, 0, 0) == REDISMODULE_ERR)
return REDISMODULE_ERR; return REDISMODULE_ERR;
if (RedisModule_CreateCommand(ctx, "do_rm_call", do_rm_call, "", 0, 0, 0) == REDISMODULE_ERR) if (RedisModule_CreateCommand(ctx, "do_rm_call", do_rm_call,
"write", 0, 0, 0) == REDISMODULE_ERR)
return REDISMODULE_ERR; return REDISMODULE_ERR;
if (RedisModule_CreateCommand(ctx, "do_bg_rm_call", do_bg_rm_call, "", 0, 0, 0) == REDISMODULE_ERR) if (RedisModule_CreateCommand(ctx, "do_bg_rm_call", do_bg_rm_call, "", 0, 0, 0) == REDISMODULE_ERR)
......
...@@ -470,35 +470,35 @@ int RedisModule_OnLoad(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) ...@@ -470,35 +470,35 @@ int RedisModule_OnLoad(RedisModuleCtx *ctx, RedisModuleString **argv, int argc)
if (fsltype == NULL) if (fsltype == NULL)
return REDISMODULE_ERR; return REDISMODULE_ERR;
if (RedisModule_CreateCommand(ctx,"fsl.push",fsl_push,"",1,1,1) == REDISMODULE_ERR) if (RedisModule_CreateCommand(ctx,"fsl.push",fsl_push,"write",1,1,1) == REDISMODULE_ERR)
return REDISMODULE_ERR; return REDISMODULE_ERR;
if (RedisModule_CreateCommand(ctx,"fsl.bpop",fsl_bpop,"",1,1,1) == REDISMODULE_ERR) if (RedisModule_CreateCommand(ctx,"fsl.bpop",fsl_bpop,"write",1,1,1) == REDISMODULE_ERR)
return REDISMODULE_ERR; return REDISMODULE_ERR;
if (RedisModule_CreateCommand(ctx,"fsl.bpopgt",fsl_bpopgt,"",1,1,1) == REDISMODULE_ERR) if (RedisModule_CreateCommand(ctx,"fsl.bpopgt",fsl_bpopgt,"write",1,1,1) == REDISMODULE_ERR)
return REDISMODULE_ERR; return REDISMODULE_ERR;
if (RedisModule_CreateCommand(ctx,"fsl.bpoppush",fsl_bpoppush,"",1,2,1) == REDISMODULE_ERR) if (RedisModule_CreateCommand(ctx,"fsl.bpoppush",fsl_bpoppush,"write",1,2,1) == REDISMODULE_ERR)
return REDISMODULE_ERR; return REDISMODULE_ERR;
if (RedisModule_CreateCommand(ctx,"fsl.getall",fsl_getall,"",1,1,1) == REDISMODULE_ERR) if (RedisModule_CreateCommand(ctx,"fsl.getall",fsl_getall,"",1,1,1) == REDISMODULE_ERR)
return REDISMODULE_ERR; return REDISMODULE_ERR;
if (RedisModule_CreateCommand(ctx, "blockonkeys.popall", blockonkeys_popall, if (RedisModule_CreateCommand(ctx, "blockonkeys.popall", blockonkeys_popall,
"", 1, 1, 1) == REDISMODULE_ERR) "write", 1, 1, 1) == REDISMODULE_ERR)
return REDISMODULE_ERR; return REDISMODULE_ERR;
if (RedisModule_CreateCommand(ctx, "blockonkeys.lpush", blockonkeys_lpush, if (RedisModule_CreateCommand(ctx, "blockonkeys.lpush", blockonkeys_lpush,
"", 1, 1, 1) == REDISMODULE_ERR) "write", 1, 1, 1) == REDISMODULE_ERR)
return REDISMODULE_ERR; return REDISMODULE_ERR;
if (RedisModule_CreateCommand(ctx, "blockonkeys.lpush_unblock", blockonkeys_lpush, if (RedisModule_CreateCommand(ctx, "blockonkeys.lpush_unblock", blockonkeys_lpush,
"", 1, 1, 1) == REDISMODULE_ERR) "write", 1, 1, 1) == REDISMODULE_ERR)
return REDISMODULE_ERR; return REDISMODULE_ERR;
if (RedisModule_CreateCommand(ctx, "blockonkeys.blpopn", blockonkeys_blpopn, if (RedisModule_CreateCommand(ctx, "blockonkeys.blpopn", blockonkeys_blpopn,
"", 1, 1, 1) == REDISMODULE_ERR) "write", 1, 1, 1) == REDISMODULE_ERR)
return REDISMODULE_ERR; return REDISMODULE_ERR;
return REDISMODULE_OK; return REDISMODULE_OK;
......
...@@ -206,19 +206,22 @@ int RedisModule_OnLoad(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) ...@@ -206,19 +206,22 @@ int RedisModule_OnLoad(RedisModuleCtx *ctx, RedisModuleString **argv, int argc)
if (datatype == NULL) if (datatype == NULL)
return REDISMODULE_ERR; return REDISMODULE_ERR;
if (RedisModule_CreateCommand(ctx,"datatype.set", datatype_set,"deny-oom",1,1,1) == REDISMODULE_ERR) if (RedisModule_CreateCommand(ctx,"datatype.set", datatype_set,
"write deny-oom", 1, 1, 1) == REDISMODULE_ERR)
return REDISMODULE_ERR; return REDISMODULE_ERR;
if (RedisModule_CreateCommand(ctx,"datatype.get", datatype_get,"",1,1,1) == REDISMODULE_ERR) if (RedisModule_CreateCommand(ctx,"datatype.get", datatype_get,"",1,1,1) == REDISMODULE_ERR)
return REDISMODULE_ERR; return REDISMODULE_ERR;
if (RedisModule_CreateCommand(ctx,"datatype.restore", datatype_restore,"deny-oom",1,1,1) == REDISMODULE_ERR) if (RedisModule_CreateCommand(ctx,"datatype.restore", datatype_restore,
"write deny-oom", 1, 1, 1) == REDISMODULE_ERR)
return REDISMODULE_ERR; return REDISMODULE_ERR;
if (RedisModule_CreateCommand(ctx,"datatype.dump", datatype_dump,"",1,1,1) == REDISMODULE_ERR) if (RedisModule_CreateCommand(ctx,"datatype.dump", datatype_dump,"",1,1,1) == REDISMODULE_ERR)
return REDISMODULE_ERR; return REDISMODULE_ERR;
if (RedisModule_CreateCommand(ctx,"datatype.swap", datatype_swap,"",1,1,1) == REDISMODULE_ERR) if (RedisModule_CreateCommand(ctx, "datatype.swap", datatype_swap,
"write", 1, 1, 1) == REDISMODULE_ERR)
return REDISMODULE_ERR; return REDISMODULE_ERR;
return REDISMODULE_OK; return REDISMODULE_OK;
......
...@@ -81,7 +81,7 @@ int RedisModule_OnLoad(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) ...@@ -81,7 +81,7 @@ int RedisModule_OnLoad(RedisModuleCtx *ctx, RedisModuleString **argv, int argc)
REDISMODULE_NOT_USED(argc); REDISMODULE_NOT_USED(argc);
if (RedisModule_Init(ctx, "hash", 1, REDISMODULE_APIVER_1) == if (RedisModule_Init(ctx, "hash", 1, REDISMODULE_APIVER_1) ==
REDISMODULE_OK && REDISMODULE_OK &&
RedisModule_CreateCommand(ctx, "hash.set", hash_set, "", RedisModule_CreateCommand(ctx, "hash.set", hash_set, "write",
1, 1, 1) == REDISMODULE_OK) { 1, 1, 1) == REDISMODULE_OK) {
return REDISMODULE_OK; return REDISMODULE_OK;
} else { } else {
......
...@@ -250,19 +250,23 @@ int RedisModule_OnLoad(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) ...@@ -250,19 +250,23 @@ int RedisModule_OnLoad(RedisModuleCtx *ctx, RedisModuleString **argv, int argc)
return REDISMODULE_ERR; return REDISMODULE_ERR;
} }
if (RedisModule_CreateCommand(ctx,"keyspace.del_key_copy", cmdDelKeyCopy,"",0,0,0) == REDISMODULE_ERR){ if (RedisModule_CreateCommand(ctx, "keyspace.del_key_copy", cmdDelKeyCopy,
"write", 0, 0, 0) == REDISMODULE_ERR){
return REDISMODULE_ERR; return REDISMODULE_ERR;
} }
if (RedisModule_CreateCommand(ctx,"keyspace.incr_case1", cmdIncrCase1,"",0,0,0) == REDISMODULE_ERR){ if (RedisModule_CreateCommand(ctx, "keyspace.incr_case1", cmdIncrCase1,
"write", 0, 0, 0) == REDISMODULE_ERR){
return REDISMODULE_ERR; return REDISMODULE_ERR;
} }
if (RedisModule_CreateCommand(ctx,"keyspace.incr_case2", cmdIncrCase2,"",0,0,0) == REDISMODULE_ERR){ if (RedisModule_CreateCommand(ctx, "keyspace.incr_case2", cmdIncrCase2,
"write", 0, 0, 0) == REDISMODULE_ERR){
return REDISMODULE_ERR; return REDISMODULE_ERR;
} }
if (RedisModule_CreateCommand(ctx,"keyspace.incr_case3", cmdIncrCase3,"",0,0,0) == REDISMODULE_ERR){ if (RedisModule_CreateCommand(ctx, "keyspace.incr_case3", cmdIncrCase3,
"write", 0, 0, 0) == REDISMODULE_ERR){
return REDISMODULE_ERR; return REDISMODULE_ERR;
} }
......
...@@ -219,15 +219,15 @@ int RedisModule_OnLoad(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) ...@@ -219,15 +219,15 @@ int RedisModule_OnLoad(RedisModuleCtx *ctx, RedisModuleString **argv, int argc)
if (RedisModule_Init(ctx, "list", 1, REDISMODULE_APIVER_1) == REDISMODULE_OK && if (RedisModule_Init(ctx, "list", 1, REDISMODULE_APIVER_1) == REDISMODULE_OK &&
RedisModule_CreateCommand(ctx, "list.getall", list_getall, "", RedisModule_CreateCommand(ctx, "list.getall", list_getall, "",
1, 1, 1) == REDISMODULE_OK && 1, 1, 1) == REDISMODULE_OK &&
RedisModule_CreateCommand(ctx, "list.edit", list_edit, "", RedisModule_CreateCommand(ctx, "list.edit", list_edit, "write",
1, 1, 1) == REDISMODULE_OK && 1, 1, 1) == REDISMODULE_OK &&
RedisModule_CreateCommand(ctx, "list.get", list_get, "", RedisModule_CreateCommand(ctx, "list.get", list_get, "write",
1, 1, 1) == REDISMODULE_OK && 1, 1, 1) == REDISMODULE_OK &&
RedisModule_CreateCommand(ctx, "list.set", list_set, "", RedisModule_CreateCommand(ctx, "list.set", list_set, "write",
1, 1, 1) == REDISMODULE_OK && 1, 1, 1) == REDISMODULE_OK &&
RedisModule_CreateCommand(ctx, "list.insert", list_insert, "", RedisModule_CreateCommand(ctx, "list.insert", list_insert, "write",
1, 1, 1) == REDISMODULE_OK && 1, 1, 1) == REDISMODULE_OK &&
RedisModule_CreateCommand(ctx, "list.delete", list_delete, "", RedisModule_CreateCommand(ctx, "list.delete", list_delete, "write",
1, 1, 1) == REDISMODULE_OK) { 1, 1, 1) == REDISMODULE_OK) {
return REDISMODULE_OK; return REDISMODULE_OK;
} else { } else {
......
...@@ -239,17 +239,17 @@ int RedisModule_OnLoad(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) ...@@ -239,17 +239,17 @@ int RedisModule_OnLoad(RedisModuleCtx *ctx, RedisModuleString **argv, int argc)
if (RedisModule_CreateCommand(ctx,"propagate-test.mixed", if (RedisModule_CreateCommand(ctx,"propagate-test.mixed",
propagateTestMixedCommand, propagateTestMixedCommand,
"",1,1,1) == REDISMODULE_ERR) "write",1,1,1) == REDISMODULE_ERR)
return REDISMODULE_ERR; return REDISMODULE_ERR;
if (RedisModule_CreateCommand(ctx,"propagate-test.nested", if (RedisModule_CreateCommand(ctx,"propagate-test.nested",
propagateTestNestedCommand, propagateTestNestedCommand,
"",1,1,1) == REDISMODULE_ERR) "write",1,1,1) == REDISMODULE_ERR)
return REDISMODULE_ERR; return REDISMODULE_ERR;
if (RedisModule_CreateCommand(ctx,"propagate-test.incr", if (RedisModule_CreateCommand(ctx,"propagate-test.incr",
propagateTestIncr, propagateTestIncr,
"",1,1,1) == REDISMODULE_ERR) "write",1,1,1) == REDISMODULE_ERR)
return REDISMODULE_ERR; return REDISMODULE_ERR;
return REDISMODULE_OK; return REDISMODULE_OK;
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment