Unverified Commit 7759ec7c authored by guybe7's avatar guybe7 Committed by GitHub
Browse files

Cleanup: propagate and alsoPropagate do not need redisCommand (#9502)

The `cmd` argument was completely unused, and all the code that bothered to pass it was unnecessary.
This is a prepartion for a future commit that treats subcommands as commands
parent 03fcc211
...@@ -802,7 +802,7 @@ int loadAppendOnlyFile(char *filename) { ...@@ -802,7 +802,7 @@ int loadAppendOnlyFile(char *filename) {
goto cleanup; goto cleanup;
} }
if (cmd == server.multiCommand) valid_before_multi = valid_up_to; if (cmd->proc == multiCommand) valid_before_multi = valid_up_to;
/* Run the command in the context of a fake client */ /* Run the command in the context of a fake client */
fakeClient->cmd = fakeClient->lastcmd = cmd; fakeClient->cmd = fakeClient->lastcmd = cmd;
......
...@@ -345,15 +345,10 @@ void serveClientsBlockedOnSortedSetKey(robj *o, readyList *rl) { ...@@ -345,15 +345,10 @@ void serveClientsBlockedOnSortedSetKey(robj *o, readyList *rl) {
/* Replicate the command. */ /* Replicate the command. */
robj *argv[2]; robj *argv[2];
struct redisCommand *cmd = where == ZSET_MIN ? argv[0] = where == ZSET_MIN ? shared.zpopmin : shared.zpopmax;
server.zpopminCommand :
server.zpopmaxCommand;
argv[0] = createStringObject(cmd->name,strlen(cmd->name));
argv[1] = rl->key; argv[1] = rl->key;
incrRefCount(rl->key); incrRefCount(rl->key);
propagate(cmd,receiver->db->id, propagate(receiver->db->id,argv,2,PROPAGATE_AOF|PROPAGATE_REPL);
argv,2,PROPAGATE_AOF|PROPAGATE_REPL);
decrRefCount(argv[0]);
decrRefCount(argv[1]); decrRefCount(argv[1]);
} }
} }
......
...@@ -1451,7 +1451,7 @@ void propagateExpire(redisDb *db, robj *key, int lazy) { ...@@ -1451,7 +1451,7 @@ void propagateExpire(redisDb *db, robj *key, int lazy) {
* Even if module executed a command without asking for propagation. */ * Even if module executed a command without asking for propagation. */
int prev_replication_allowed = server.replication_allowed; int prev_replication_allowed = server.replication_allowed;
server.replication_allowed = 1; server.replication_allowed = 1;
propagate(server.delCommand,db->id,argv,2,PROPAGATE_AOF|PROPAGATE_REPL); propagate(db->id,argv,2,PROPAGATE_AOF|PROPAGATE_REPL);
server.replication_allowed = prev_replication_allowed; server.replication_allowed = prev_replication_allowed;
decrRefCount(argv[0]); decrRefCount(argv[0]);
......
...@@ -630,8 +630,7 @@ void moduleHandlePropagationAfterCommandCallback(RedisModuleCtx *ctx) { ...@@ -630,8 +630,7 @@ void moduleHandlePropagationAfterCommandCallback(RedisModuleCtx *ctx) {
/* Handle the replication of the final EXEC, since whatever a command /* Handle the replication of the final EXEC, since whatever a command
* emits is always wrapped around MULTI/EXEC. */ * emits is always wrapped around MULTI/EXEC. */
alsoPropagate(server.execCommand,c->db->id,&shared.exec,1, alsoPropagate(c->db->id,&shared.exec,1,PROPAGATE_AOF|PROPAGATE_REPL);
PROPAGATE_AOF|PROPAGATE_REPL);
afterPropagateExec(); afterPropagateExec();
/* If this is not a module command context (but is instead a simple /* If this is not a module command context (but is instead a simple
...@@ -645,7 +644,7 @@ void moduleHandlePropagationAfterCommandCallback(RedisModuleCtx *ctx) { ...@@ -645,7 +644,7 @@ void moduleHandlePropagationAfterCommandCallback(RedisModuleCtx *ctx) {
redisOp *rop = &server.also_propagate.ops[j]; redisOp *rop = &server.also_propagate.ops[j];
int target = rop->target; int target = rop->target;
if (target) if (target)
propagate(rop->cmd,rop->dbid,rop->argv,rop->argc,target); propagate(rop->dbid,rop->argv,rop->argc,target);
} }
redisOpArrayFree(&server.also_propagate); redisOpArrayFree(&server.also_propagate);
/* Restore the previous oparray in case of nexted use of the API. */ /* Restore the previous oparray in case of nexted use of the API. */
...@@ -2260,10 +2259,10 @@ int RM_Replicate(RedisModuleCtx *ctx, const char *cmdname, const char *fmt, ...) ...@@ -2260,10 +2259,10 @@ int RM_Replicate(RedisModuleCtx *ctx, const char *cmdname, const char *fmt, ...)
* will stop being used, so accumulating stuff does not make much sense, * will stop being used, so accumulating stuff does not make much sense,
* nor we could easily use the alsoPropagate() API from threads. */ * nor we could easily use the alsoPropagate() API from threads. */
if (ctx->flags & REDISMODULE_CTX_THREAD_SAFE) { if (ctx->flags & REDISMODULE_CTX_THREAD_SAFE) {
propagate(cmd,ctx->client->db->id,argv,argc,target); propagate(ctx->client->db->id,argv,argc,target);
} else { } else {
moduleReplicateMultiIfNeeded(ctx); moduleReplicateMultiIfNeeded(ctx);
alsoPropagate(cmd,ctx->client->db->id,argv,argc,target); alsoPropagate(ctx->client->db->id,argv,argc,target);
} }
/* Release the argv. */ /* Release the argv. */
...@@ -2285,7 +2284,7 @@ int RM_Replicate(RedisModuleCtx *ctx, const char *cmdname, const char *fmt, ...) ...@@ -2285,7 +2284,7 @@ int RM_Replicate(RedisModuleCtx *ctx, const char *cmdname, const char *fmt, ...)
* *
* The function always returns REDISMODULE_OK. */ * The function always returns REDISMODULE_OK. */
int RM_ReplicateVerbatim(RedisModuleCtx *ctx) { int RM_ReplicateVerbatim(RedisModuleCtx *ctx) {
alsoPropagate(ctx->client->cmd,ctx->client->db->id, alsoPropagate(ctx->client->db->id,
ctx->client->argv,ctx->client->argc, ctx->client->argv,ctx->client->argc,
PROPAGATE_AOF|PROPAGATE_REPL); PROPAGATE_AOF|PROPAGATE_REPL);
server.dirty++; server.dirty++;
......
...@@ -129,13 +129,11 @@ void afterPropagateExec() { ...@@ -129,13 +129,11 @@ void afterPropagateExec() {
* implementation for more information. */ * implementation for more information. */
void execCommandPropagateMulti(int dbid) { void execCommandPropagateMulti(int dbid) {
beforePropagateMulti(); beforePropagateMulti();
propagate(server.multiCommand,dbid,&shared.multi,1, propagate(dbid,&shared.multi,1,PROPAGATE_AOF|PROPAGATE_REPL);
PROPAGATE_AOF|PROPAGATE_REPL);
} }
void execCommandPropagateExec(int dbid) { void execCommandPropagateExec(int dbid) {
propagate(server.execCommand,dbid,&shared.exec,1, propagate(dbid,&shared.exec,1,PROPAGATE_AOF|PROPAGATE_REPL);
PROPAGATE_AOF|PROPAGATE_REPL);
afterPropagateExec(); afterPropagateExec();
} }
......
...@@ -3001,6 +3001,8 @@ void createSharedObjects(void) { ...@@ -3001,6 +3001,8 @@ void createSharedObjects(void) {
shared.persist = createStringObject("PERSIST",7); shared.persist = createStringObject("PERSIST",7);
shared.set = createStringObject("SET",3); shared.set = createStringObject("SET",3);
shared.eval = createStringObject("EVAL",4); shared.eval = createStringObject("EVAL",4);
shared.zpopmin = createStringObject("ZPOPMIN",7);
shared.zpopmax = createStringObject("ZPOPMAX",7);
/* Shared command argument */ /* Shared command argument */
shared.left = createStringObject("left",4); shared.left = createStringObject("left",4);
...@@ -3157,21 +3159,6 @@ void initServerConfig(void) { ...@@ -3157,21 +3159,6 @@ void initServerConfig(void) {
server.commands = dictCreate(&commandTableDictType); server.commands = dictCreate(&commandTableDictType);
server.orig_commands = dictCreate(&commandTableDictType); server.orig_commands = dictCreate(&commandTableDictType);
populateCommandTable(); populateCommandTable();
server.delCommand = lookupCommandByCString("del");
server.multiCommand = lookupCommandByCString("multi");
server.lpushCommand = lookupCommandByCString("lpush");
server.lpopCommand = lookupCommandByCString("lpop");
server.rpopCommand = lookupCommandByCString("rpop");
server.zpopminCommand = lookupCommandByCString("zpopmin");
server.zpopmaxCommand = lookupCommandByCString("zpopmax");
server.sremCommand = lookupCommandByCString("srem");
server.execCommand = lookupCommandByCString("exec");
server.expireCommand = lookupCommandByCString("expire");
server.pexpireCommand = lookupCommandByCString("pexpire");
server.xclaimCommand = lookupCommandByCString("xclaim");
server.xgroupCommand = lookupCommandByCString("xgroup");
server.rpoplpushCommand = lookupCommandByCString("rpoplpush");
server.lmoveCommand = lookupCommandByCString("lmove");
/* Debugging */ /* Debugging */
server.watchdog_period = 0; server.watchdog_period = 0;
...@@ -4013,14 +4000,11 @@ void redisOpArrayInit(redisOpArray *oa) { ...@@ -4013,14 +4000,11 @@ void redisOpArrayInit(redisOpArray *oa) {
oa->numops = 0; oa->numops = 0;
} }
int redisOpArrayAppend(redisOpArray *oa, struct redisCommand *cmd, int dbid, int redisOpArrayAppend(redisOpArray *oa, int dbid, robj **argv, int argc, int target) {
robj **argv, int argc, int target)
{
redisOp *op; redisOp *op;
oa->ops = zrealloc(oa->ops,sizeof(redisOp)*(oa->numops+1)); oa->ops = zrealloc(oa->ops,sizeof(redisOp)*(oa->numops+1));
op = oa->ops+oa->numops; op = oa->ops+oa->numops;
op->cmd = cmd;
op->dbid = dbid; op->dbid = dbid;
op->argv = argv; op->argv = argv;
op->argc = argc; op->argc = argc;
...@@ -4089,11 +4073,7 @@ struct redisCommand *lookupCommandOrOriginal(sds name) { ...@@ -4089,11 +4073,7 @@ struct redisCommand *lookupCommandOrOriginal(sds name) {
* command execution, for example when serving a blocked client, you * command execution, for example when serving a blocked client, you
* want to use propagate(). * want to use propagate().
*/ */
void propagate(struct redisCommand *cmd, int dbid, robj **argv, int argc, void propagate(int dbid, robj **argv, int argc, int flags) {
int flags)
{
UNUSED(cmd);
if (!server.replication_allowed) if (!server.replication_allowed)
return; return;
...@@ -4118,8 +4098,7 @@ void propagate(struct redisCommand *cmd, int dbid, robj **argv, int argc, ...@@ -4118,8 +4098,7 @@ void propagate(struct redisCommand *cmd, int dbid, robj **argv, int argc,
/* Used inside commands to schedule the propagation of additional commands /* Used inside commands to schedule the propagation of additional commands
* after the current command is propagated to AOF / Replication. * after the current command is propagated to AOF / Replication.
* *
* 'cmd' must be a pointer to the Redis command to replicate, dbid is the * dbid is the database ID the command should be propagated into.
* database ID the command should be propagated into.
* Arguments of the command to propagate are passed as an array of redis * Arguments of the command to propagate are passed as an array of redis
* objects pointers of len 'argc', using the 'argv' vector. * objects pointers of len 'argc', using the 'argv' vector.
* *
...@@ -4127,9 +4106,7 @@ void propagate(struct redisCommand *cmd, int dbid, robj **argv, int argc, ...@@ -4127,9 +4106,7 @@ void propagate(struct redisCommand *cmd, int dbid, robj **argv, int argc,
* so it is up to the caller to release the passed argv (but it is usually * so it is up to the caller to release the passed argv (but it is usually
* stack allocated). The function automatically increments ref count of * stack allocated). The function automatically increments ref count of
* passed objects, so the caller does not need to. */ * passed objects, so the caller does not need to. */
void alsoPropagate(struct redisCommand *cmd, int dbid, robj **argv, int argc, void alsoPropagate(int dbid, robj **argv, int argc, int target) {
int target)
{
robj **argvcopy; robj **argvcopy;
int j; int j;
...@@ -4140,7 +4117,7 @@ void alsoPropagate(struct redisCommand *cmd, int dbid, robj **argv, int argc, ...@@ -4140,7 +4117,7 @@ void alsoPropagate(struct redisCommand *cmd, int dbid, robj **argv, int argc,
argvcopy[j] = argv[j]; argvcopy[j] = argv[j];
incrRefCount(argv[j]); incrRefCount(argv[j]);
} }
redisOpArrayAppend(&server.also_propagate,cmd,dbid,argvcopy,argc,target); redisOpArrayAppend(&server.also_propagate,dbid,argvcopy,argc,target);
} }
/* It is possible to call the function forceCommandPropagation() inside a /* It is possible to call the function forceCommandPropagation() inside a
...@@ -4345,7 +4322,7 @@ void call(client *c, int flags) { ...@@ -4345,7 +4322,7 @@ void call(client *c, int flags) {
* propagation is needed. Note that modules commands handle replication * propagation is needed. Note that modules commands handle replication
* in an explicit way, so we never replicate them automatically. */ * in an explicit way, so we never replicate them automatically. */
if (propagate_flags != PROPAGATE_NONE && !(c->cmd->flags & CMD_MODULE)) if (propagate_flags != PROPAGATE_NONE && !(c->cmd->flags & CMD_MODULE))
propagate(c->cmd,c->db->id,c->argv,c->argc,propagate_flags); propagate(c->db->id,c->argv,c->argc,propagate_flags);
} }
/* Restore the old replication flags, since call() can be executed /* Restore the old replication flags, since call() can be executed
...@@ -4385,7 +4362,7 @@ void call(client *c, int flags) { ...@@ -4385,7 +4362,7 @@ void call(client *c, int flags) {
if (!(flags&CMD_CALL_PROPAGATE_AOF)) target &= ~PROPAGATE_AOF; if (!(flags&CMD_CALL_PROPAGATE_AOF)) target &= ~PROPAGATE_AOF;
if (!(flags&CMD_CALL_PROPAGATE_REPL)) target &= ~PROPAGATE_REPL; if (!(flags&CMD_CALL_PROPAGATE_REPL)) target &= ~PROPAGATE_REPL;
if (target) if (target)
propagate(rop->cmd,rop->dbid,rop->argv,rop->argc,target); propagate(rop->dbid,rop->argv,rop->argc,target);
} }
if (multi_emitted) { if (multi_emitted) {
......
...@@ -1095,7 +1095,6 @@ extern clientBufferLimitsConfig clientBufferLimitsDefaults[CLIENT_TYPE_OBUF_COUN ...@@ -1095,7 +1095,6 @@ extern clientBufferLimitsConfig clientBufferLimitsDefaults[CLIENT_TYPE_OBUF_COUN
typedef struct redisOp { typedef struct redisOp {
robj **argv; robj **argv;
int argc, dbid, target; int argc, dbid, target;
struct redisCommand *cmd;
} redisOp; } redisOp;
/* Defines an array of Redis operations. There is an API to add to this /* Defines an array of Redis operations. There is an API to add to this
...@@ -1311,12 +1310,6 @@ struct redisServer { ...@@ -1311,12 +1310,6 @@ struct redisServer {
off_t loading_loaded_bytes; off_t loading_loaded_bytes;
time_t loading_start_time; time_t loading_start_time;
off_t loading_process_events_interval_bytes; off_t loading_process_events_interval_bytes;
/* Fast pointers to often looked up command */
struct redisCommand *delCommand, *multiCommand, *lpushCommand,
*lpopCommand, *rpopCommand, *zpopminCommand,
*zpopmaxCommand, *sremCommand, *execCommand,
*expireCommand, *pexpireCommand, *xclaimCommand,
*xgroupCommand, *rpoplpushCommand, *lmoveCommand;
/* Fields used only for stats */ /* Fields used only for stats */
time_t stat_starttime; /* Server start time */ time_t stat_starttime; /* Server start time */
long long stat_numcommands; /* Number of processed commands */ long long stat_numcommands; /* Number of processed commands */
...@@ -2380,8 +2373,8 @@ struct redisCommand *lookupCommand(sds name); ...@@ -2380,8 +2373,8 @@ struct redisCommand *lookupCommand(sds name);
struct redisCommand *lookupCommandByCString(const char *s); struct redisCommand *lookupCommandByCString(const char *s);
struct redisCommand *lookupCommandOrOriginal(sds name); struct redisCommand *lookupCommandOrOriginal(sds name);
void call(client *c, int flags); void call(client *c, int flags);
void propagate(struct redisCommand *cmd, int dbid, robj **argv, int argc, int flags); void propagate(int dbid, robj **argv, int argc, int flags);
void alsoPropagate(struct redisCommand *cmd, int dbid, robj **argv, int argc, int target); void alsoPropagate(int dbid, robj **argv, int argc, int target);
void redisOpArrayInit(redisOpArray *oa); void redisOpArrayInit(redisOpArray *oa);
void redisOpArrayFree(redisOpArray *oa); void redisOpArrayFree(redisOpArray *oa);
void forceCommandPropagation(client *c, int flags); void forceCommandPropagation(client *c, int flags);
......
...@@ -962,8 +962,6 @@ void serveClientBlockedOnList(client *receiver, robj *o, robj *key, robj *dstkey ...@@ -962,8 +962,6 @@ void serveClientBlockedOnList(client *receiver, robj *o, robj *key, robj *dstkey
if (dstkey == NULL) { if (dstkey == NULL) {
/* Propagate the [LR]POP operation. */ /* Propagate the [LR]POP operation. */
struct redisCommand *cmd = (wherefrom == LIST_HEAD) ?
server.lpopCommand : server.rpopCommand;
argv[0] = (wherefrom == LIST_HEAD) ? shared.lpop : argv[0] = (wherefrom == LIST_HEAD) ? shared.lpop :
shared.rpop; shared.rpop;
argv[1] = key; argv[1] = key;
...@@ -976,7 +974,7 @@ void serveClientBlockedOnList(client *receiver, robj *o, robj *key, robj *dstkey ...@@ -976,7 +974,7 @@ void serveClientBlockedOnList(client *receiver, robj *o, robj *key, robj *dstkey
serverAssert(llen > 0); serverAssert(llen > 0);
argv[2] = createStringObjectFromLongLong((count > llen) ? llen : count); argv[2] = createStringObjectFromLongLong((count > llen) ? llen : count);
propagate(cmd, db->id, argv, 3, PROPAGATE_AOF|PROPAGATE_REPL); propagate(db->id, argv, 3, PROPAGATE_AOF|PROPAGATE_REPL);
decrRefCount(argv[2]); decrRefCount(argv[2]);
/* Pop a range of elements in a nested arrays way. */ /* Pop a range of elements in a nested arrays way. */
...@@ -984,7 +982,7 @@ void serveClientBlockedOnList(client *receiver, robj *o, robj *key, robj *dstkey ...@@ -984,7 +982,7 @@ void serveClientBlockedOnList(client *receiver, robj *o, robj *key, robj *dstkey
return; return;
} }
propagate(cmd, db->id, argv, 2, PROPAGATE_AOF|PROPAGATE_REPL); propagate(db->id, argv, 2, PROPAGATE_AOF|PROPAGATE_REPL);
/* BRPOP/BLPOP */ /* BRPOP/BLPOP */
value = listTypePop(o, wherefrom); value = listTypePop(o, wherefrom);
...@@ -1015,10 +1013,7 @@ void serveClientBlockedOnList(client *receiver, robj *o, robj *key, robj *dstkey ...@@ -1015,10 +1013,7 @@ void serveClientBlockedOnList(client *receiver, robj *o, robj *key, robj *dstkey
argv[2] = dstkey; argv[2] = dstkey;
argv[3] = getStringObjectFromListPosition(wherefrom); argv[3] = getStringObjectFromListPosition(wherefrom);
argv[4] = getStringObjectFromListPosition(whereto); argv[4] = getStringObjectFromListPosition(whereto);
propagate(isbrpoplpush ? server.rpoplpushCommand : server.lmoveCommand, propagate(db->id,argv,(isbrpoplpush ? 3 : 5),PROPAGATE_AOF|PROPAGATE_REPL);
db->id,argv,(isbrpoplpush ? 3 : 5),
PROPAGATE_AOF|
PROPAGATE_REPL);
/* Notify event ("lpush" or "rpush" was notified by lmoveHandlePush). */ /* Notify event ("lpush" or "rpush" was notified by lmoveHandlePush). */
notifyKeyspaceEvent(NOTIFY_LIST,wherefrom == LIST_TAIL ? "rpop" : "lpop", notifyKeyspaceEvent(NOTIFY_LIST,wherefrom == LIST_TAIL ? "rpop" : "lpop",
......
...@@ -533,8 +533,7 @@ void spopWithCountCommand(client *c) { ...@@ -533,8 +533,7 @@ void spopWithCountCommand(client *c) {
/* Replicate/AOF this command as an SREM operation */ /* Replicate/AOF this command as an SREM operation */
propargv[2] = objele; propargv[2] = objele;
alsoPropagate(server.sremCommand,c->db->id,propargv,3, alsoPropagate(c->db->id,propargv,3,PROPAGATE_AOF|PROPAGATE_REPL);
PROPAGATE_AOF|PROPAGATE_REPL);
decrRefCount(objele); decrRefCount(objele);
} }
} else { } else {
...@@ -576,8 +575,7 @@ void spopWithCountCommand(client *c) { ...@@ -576,8 +575,7 @@ void spopWithCountCommand(client *c) {
/* Replicate/AOF this command as an SREM operation */ /* Replicate/AOF this command as an SREM operation */
propargv[2] = objele; propargv[2] = objele;
alsoPropagate(server.sremCommand,c->db->id,propargv,3, alsoPropagate(c->db->id,propargv,3,PROPAGATE_AOF|PROPAGATE_REPL);
PROPAGATE_AOF|PROPAGATE_REPL);
decrRefCount(objele); decrRefCount(objele);
} }
setTypeReleaseIterator(si); setTypeReleaseIterator(si);
......
...@@ -1355,7 +1355,7 @@ void streamPropagateXCLAIM(client *c, robj *key, streamCG *group, robj *groupnam ...@@ -1355,7 +1355,7 @@ void streamPropagateXCLAIM(client *c, robj *key, streamCG *group, robj *groupnam
* the command execution context. Moreover this will just alter the * the command execution context. Moreover this will just alter the
* consumer group state, and we don't need MULTI/EXEC wrapping because * consumer group state, and we don't need MULTI/EXEC wrapping because
* there is no message state cross-message atomicity required. */ * there is no message state cross-message atomicity required. */
propagate(server.xclaimCommand,c->db->id,argv,14,PROPAGATE_AOF|PROPAGATE_REPL); propagate(c->db->id,argv,14,PROPAGATE_AOF|PROPAGATE_REPL);
decrRefCount(argv[3]); decrRefCount(argv[3]);
decrRefCount(argv[7]); decrRefCount(argv[7]);
decrRefCount(argv[9]); decrRefCount(argv[9]);
...@@ -1380,7 +1380,7 @@ void streamPropagateGroupID(client *c, robj *key, streamCG *group, robj *groupna ...@@ -1380,7 +1380,7 @@ void streamPropagateGroupID(client *c, robj *key, streamCG *group, robj *groupna
* the command execution context. Moreover this will just alter the * the command execution context. Moreover this will just alter the
* consumer group state, and we don't need MULTI/EXEC wrapping because * consumer group state, and we don't need MULTI/EXEC wrapping because
* there is no message state cross-message atomicity required. */ * there is no message state cross-message atomicity required. */
propagate(server.xgroupCommand,c->db->id,argv,5,PROPAGATE_AOF|PROPAGATE_REPL); propagate(c->db->id,argv,5,PROPAGATE_AOF|PROPAGATE_REPL);
decrRefCount(argv[4]); decrRefCount(argv[4]);
} }
...@@ -1402,7 +1402,7 @@ void streamPropagateConsumerCreation(client *c, robj *key, robj *groupname, sds ...@@ -1402,7 +1402,7 @@ void streamPropagateConsumerCreation(client *c, robj *key, robj *groupname, sds
* the command execution context. Moreover this will just alter the * the command execution context. Moreover this will just alter the
* consumer group state, and we don't need MULTI/EXEC wrapping because * consumer group state, and we don't need MULTI/EXEC wrapping because
* there is no message state cross-message atomicity required. */ * there is no message state cross-message atomicity required. */
propagate(server.xgroupCommand,c->db->id,argv,5,PROPAGATE_AOF|PROPAGATE_REPL); propagate(c->db->id,argv,5,PROPAGATE_AOF|PROPAGATE_REPL);
decrRefCount(argv[4]); decrRefCount(argv[4]);
} }
......
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