Unverified Commit 383d902c authored by ranshid's avatar ranshid Committed by GitHub
Browse files

reprocess command when client is unblocked on keys (#11012)

*TL;DR*
---------------------------------------
Following the discussion over the issue [#7551](https://github.com/redis/redis/issues/7551

)
We decided to refactor the client blocking code to eliminate some of the code duplications
and to rebuild the infrastructure better for future key blocking cases.


*In this PR*
---------------------------------------
1. reprocess the command once a client becomes unblocked on key (instead of running
   custom code for the unblocked path that's different than the one that would have run if
   blocking wasn't needed)
2. eliminate some (now) irrelevant code for handling unblocking lists/zsets/streams etc...
3. modify some tests to intercept the error in cases of error on reprocess after unblock (see
   details in the notes section below)
4. replace '$' on the client argv with current stream id. Since once we reprocess the stream
   XREAD we need to read from the last msg and not wait for new msg  in order to prevent
   endless block loop. 
5. Added statistics to the info "Clients" section to report the:
   * `total_blocking_keys` - number of blocking keys
   * `total_blocking_keys_on_nokey` - number of blocking keys which have at least 1 client
      which would like
   to be unblocked on when the key is deleted.
6. Avoid expiring unblocked key during unblock. Previously we used to lookup the unblocked key
   which might have been expired during the lookup. Now we lookup the key using NOTOUCH and
   NOEXPIRE to avoid deleting it at this point, so propagating commands in blocked.c is no longer needed.
7. deprecated command flags. We decided to remove the CMD_CALL_STATS and CMD_CALL_SLOWLOG
   and make an explicit verification in the call() function in order to decide if stats update should take place.
   This should simplify the logic and also mitigate existing issues: for example module calls which are
   triggered as part of AOF loading might still report stats even though they are called during AOF loading.

*Behavior changes*
---------------------------------------------------

1. As this implementation prevents writing dedicated code handling unblocked streams/lists/zsets,
since we now re-process the command once the client is unblocked some errors will be reported differently.
The old implementation used to issue
``UNBLOCKED the stream key no longer exists``
in the following cases:
   - The stream key has been deleted (ie. calling DEL)
   - The stream and group existed but the key type was changed by overriding it (ie. with set command)
   - The key not longer exists after we swapdb with a db which does not contains this key
   - After swapdb when the new db has this key but with different type.
   
In the new implementation the reported errors will be the same as if the command was processed after effect:
**NOGROUP** - in case key no longer exists, or **WRONGTYPE** in case the key was overridden with a different type.

2. Reprocessing the command means that some checks will be reevaluated once the
client is unblocked.
For example, ACL rules might change since the command originally was executed and
will fail once the client is unblocked.
Another example is OOM condition checks which might enable the command to run and
block but fail the command reprocess once the client is unblocked.

3. One of the changes in this PR is that no command stats are being updated once the
command is blocked (all stats will be updated once the client is unblocked). This implies
that when we have many clients blocked, users will no longer be able to get that information
from the command stats. However the information can still be gathered from the client list.

**Client blocking**
---------------------------------------------------

the blocking on key will still be triggered the same way as it is done today.
in order to block the current client on list of keys, the call to
blockForKeys will still need to be made which will perform the same as it is today:

*  add the client to the list of blocked clients on each key
*  keep the key with a matching list node (position in the global blocking clients list for that key)
   in the client private blocking key dict.
*  flag the client with CLIENT_BLOCKED
*  update blocking statistics
*  register the client on the timeout table

**Key Unblock**
---------------------------------------------------

Unblocking a specific key will be triggered (same as today) by calling signalKeyAsReady.
the implementation in that part will stay the same as today - adding the key to the global readyList.
The reason to maintain the readyList (as apposed to iterating over all clients blocked on the specific key)
is in order to keep the signal operation as short as possible, since it is called during the command processing.
The main change is that instead of going through a dedicated code path that operates the blocked command
we will just call processPendingCommandsAndResetClient.

**ClientUnblock (keys)**
---------------------------------------------------

1. Unblocking clients on keys will be triggered after command is
   processed and during the beforeSleep
8. the general schema is:
9. For each key *k* in the readyList:
```            
For each client *c* which is blocked on *k*:
            in case either:
	          1. *k* exists AND the *k* type matches the current client blocking type
	  	      OR
	          2. *k* exists and *c* is blocked on module command
	    	      OR
	          3. *k* does not exists and *c* was blocked with the flag
	             unblock_on_deleted_key
                 do:
                                  1. remove the client from the list of clients blocked on this key
                                  2. remove the blocking list node from the client blocking key dict
                                  3. remove the client from the timeout list
                                  10. queue the client on the unblocked_clients list
                                  11. *NEW*: call processCommandAndResetClient(c);
```
*NOTE:* for module blocked clients we will still call the moduleUnblockClientByHandle
              which will queue the client for processing in moduleUnblockedClients list.

**Process Unblocked clients**
---------------------------------------------------

The process of all unblocked clients is done in the beforeSleep and no change is planned
in that part.

The general schema will be:
For each client *c* in server.unblocked_clients:

        * remove client from the server.unblocked_clients
        * set back the client readHandler
        * continue processing the pending command and input buffer.

*Some notes regarding the new implementation*
---------------------------------------------------

1. Although it was proposed, it is currently difficult to remove the
   read handler from the client while it is blocked.
   The reason is that a blocked client should be unblocked when it is
   disconnected, or we might consume data into void.

2. While this PR mainly keep the current blocking logic as-is, there
   might be some future additions to the infrastructure that we would
   like to have:
   - allow non-preemptive blocking of client - sometimes we can think
     that a new kind of blocking can be expected to not be preempt. for
     example lets imagine we hold some keys on disk and when a command
     needs to process them it will block until the keys are uploaded.
     in this case we will want the client to not disconnect or be
     unblocked until the process is completed (remove the client read
     handler, prevent client timeout, disable unblock via debug command etc...).
   - allow generic blocking based on command declared keys - we might
     want to add a hook before command processing to check if any of the
     declared keys require the command to block. this way it would be
     easier to add new kinds of key-based blocking mechanisms.
Co-authored-by: default avatarOran Agra <oran@redislabs.com>
Signed-off-by: default avatarRan Shidlansik <ranshid@amazon.com>
parent af0a4fe2
This diff is collapsed.
......@@ -7215,10 +7215,10 @@ void clusterRedirectClient(client *c, clusterNode *n, int hashslot, int error_co
* returns 1. Otherwise 0 is returned and no operation is performed. */
int clusterRedirectBlockedClientIfNeeded(client *c) {
if (c->flags & CLIENT_BLOCKED &&
(c->btype == BLOCKED_LIST ||
c->btype == BLOCKED_ZSET ||
c->btype == BLOCKED_STREAM ||
c->btype == BLOCKED_MODULE))
(c->bstate.btype == BLOCKED_LIST ||
c->bstate.btype == BLOCKED_ZSET ||
c->bstate.btype == BLOCKED_STREAM ||
c->bstate.btype == BLOCKED_MODULE))
{
dictEntry *de;
dictIterator *di;
......@@ -7234,11 +7234,11 @@ int clusterRedirectBlockedClientIfNeeded(client *c) {
/* If the client is blocked on module, but not on a specific key,
* don't unblock it (except for the CLUSTER_FAIL case above). */
if (c->btype == BLOCKED_MODULE && !moduleClientIsBlockedOnKeys(c))
if (c->bstate.btype == BLOCKED_MODULE && !moduleClientIsBlockedOnKeys(c))
return 0;
/* All keys must belong to the same slot, so check first key only. */
di = dictGetIterator(c->bpop.keys);
di = dictGetIterator(c->bstate.keys);
if ((de = dictNext(di)) != NULL) {
robj *key = dictGetKey(de);
int slot = keyHashSlot((char*)key->ptr, sdslen(key->ptr));
......
......@@ -1147,7 +1147,7 @@ void shutdownCommand(client *c) {
return;
}
blockClient(c, BLOCKED_SHUTDOWN);
blockClientShutdown(c);
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). */
......
......@@ -131,6 +131,15 @@ typedef void (dictScanBucketFunction)(dict *d, dictEntry **bucketref);
#define dictSetDoubleVal(entry, _val_) \
do { (entry)->v.d = _val_; } while(0)
#define dictIncrSignedIntegerVal(entry, _val_) \
((entry)->v.s64 += _val_)
#define dictIncrUnsignedIntegerVal(entry, _val_) \
((entry)->v.u64 += _val_)
#define dictIncrDoubleVal(entry, _val_) \
((entry)->v.d += _val_)
#define dictFreeKey(d, entry) \
if ((d)->type->keyDestructor) \
(d)->type->keyDestructor((d), (entry)->key)
......
......@@ -6158,7 +6158,7 @@ RedisModuleCallReply *RM_Call(RedisModuleCtx *ctx, const char *cmdname, const ch
server.replication_allowed = replicate && server.replication_allowed;
 
/* Run the command */
int call_flags = CMD_CALL_SLOWLOG | CMD_CALL_STATS | CMD_CALL_FROM_MODULE;
int call_flags = CMD_CALL_FROM_MODULE;
if (replicate) {
if (!(flags & REDISMODULE_ARGV_NO_AOF))
call_flags |= CMD_CALL_PROPAGATE_AOF;
......@@ -7282,7 +7282,7 @@ void RM_LatencyAddSample(const char *event, mstime_t latency) {
* The structure RedisModuleBlockedClient will be always deallocated when
* running the list of clients blocked by a module that need to be unblocked. */
void unblockClientFromModule(client *c) {
RedisModuleBlockedClient *bc = c->bpop.module_blocked_handle;
RedisModuleBlockedClient *bc = c->bstate.module_blocked_handle;
 
/* Call the disconnection callback if any. Note that
* bc->disconnect_callback is set to NULL if the client gets disconnected
......@@ -7346,8 +7346,8 @@ RedisModuleBlockedClient *moduleBlockClient(RedisModuleCtx *ctx, RedisModuleCmdF
int islua = scriptIsRunning();
int ismulti = server.in_exec;
 
c->bpop.module_blocked_handle = zmalloc(sizeof(RedisModuleBlockedClient));
RedisModuleBlockedClient *bc = c->bpop.module_blocked_handle;
c->bstate.module_blocked_handle = zmalloc(sizeof(RedisModuleBlockedClient));
RedisModuleBlockedClient *bc = c->bstate.module_blocked_handle;
ctx->module->blocked_clients++;
 
/* We need to handle the invalid operation of calling modules blocking
......@@ -7371,16 +7371,16 @@ RedisModuleBlockedClient *moduleBlockClient(RedisModuleCtx *ctx, RedisModuleCmdF
bc->unblocked = 0;
bc->background_timer = 0;
bc->background_duration = 0;
c->bpop.timeout = timeout;
c->bstate.timeout = timeout;
 
if (islua || ismulti) {
c->bpop.module_blocked_handle = NULL;
c->bstate.module_blocked_handle = NULL;
addReplyError(c, islua ?
"Blocking module command called from Lua script" :
"Blocking module command called from transaction");
} else {
if (keys) {
blockForKeys(c,BLOCKED_MODULE,keys,numkeys,-1,timeout,NULL,NULL,NULL,flags&REDISMODULE_BLOCK_UNBLOCK_DELETED);
blockForKeys(c,BLOCKED_MODULE,keys,numkeys,timeout,flags&REDISMODULE_BLOCK_UNBLOCK_DELETED);
} else {
blockClient(c,BLOCKED_MODULE);
}
......@@ -7397,7 +7397,7 @@ RedisModuleBlockedClient *moduleBlockClient(RedisModuleCtx *ctx, RedisModuleCmdF
* This function returns 1 if client was served (and should be unblocked) */
int moduleTryServeClientBlockedOnKey(client *c, robj *key) {
int served = 0;
RedisModuleBlockedClient *bc = c->bpop.module_blocked_handle;
RedisModuleBlockedClient *bc = c->bstate.module_blocked_handle;
 
/* Protect against re-processing: don't serve clients that are already
* in the unblocking list for any reason (including RM_UnblockClient()
......@@ -7566,14 +7566,14 @@ int moduleUnblockClientByHandle(RedisModuleBlockedClient *bc, void *privdata) {
/* This API is used by the Redis core to unblock a client that was blocked
* by a module. */
void moduleUnblockClient(client *c) {
RedisModuleBlockedClient *bc = c->bpop.module_blocked_handle;
RedisModuleBlockedClient *bc = c->bstate.module_blocked_handle;
moduleUnblockClientByHandle(bc,NULL);
}
 
/* Return true if the client 'c' was blocked by a module using
* RM_BlockClientOnKeys(). */
int moduleClientIsBlockedOnKeys(client *c) {
RedisModuleBlockedClient *bc = c->bpop.module_blocked_handle;
RedisModuleBlockedClient *bc = c->bstate.module_blocked_handle;
return bc->blocked_on_keys;
}
 
......@@ -7740,10 +7740,10 @@ void moduleHandleBlockedClients(void) {
* moduleBlockedClientTimedOut().
*/
int moduleBlockedClientMayTimeout(client *c) {
if (c->btype != BLOCKED_MODULE)
if (c->bstate.btype != BLOCKED_MODULE)
return 1;
 
RedisModuleBlockedClient *bc = c->bpop.module_blocked_handle;
RedisModuleBlockedClient *bc = c->bstate.module_blocked_handle;
return (bc && bc->timeout_callback != NULL);
}
 
......@@ -7752,7 +7752,7 @@ int moduleBlockedClientMayTimeout(client *c) {
* does not need to do any cleanup. Eventually the module will call the
* API to unblock the client and the memory will be released. */
void moduleBlockedClientTimedOut(client *c) {
RedisModuleBlockedClient *bc = c->bpop.module_blocked_handle;
RedisModuleBlockedClient *bc = c->bstate.module_blocked_handle;
 
/* Protect against re-processing: don't serve clients that are already
* in the unblocking list for any reason (including RM_UnblockClient()
......@@ -7767,9 +7767,8 @@ void moduleBlockedClientTimedOut(client *c) {
long long prev_error_replies = server.stat_total_error_replies;
bc->timeout_callback(&ctx,(void**)c->argv,c->argc);
moduleFreeContext(&ctx);
if (!bc->blocked_on_keys) {
updateStatsOnUnblock(c, bc->background_duration, 0, server.stat_total_error_replies != prev_error_replies);
}
updateStatsOnUnblock(c, bc->background_duration, 0, server.stat_total_error_replies != prev_error_replies);
/* For timeout events, we do not want to call the disconnect callback,
* because the blocked client will be automatically disconnected in
* this case, and the user can still hook using the timeout callback. */
......
......@@ -165,6 +165,7 @@ client *createClient(connection *conn) {
c->flags = 0;
c->slot = -1;
c->ctime = c->lastinteraction = server.unixtime;
c->duration = 0;
clientSetDefaultAuth(c);
c->replstate = REPL_STATE_NONE;
c->repl_start_cmd_stream_on_ack = 0;
......@@ -184,15 +185,7 @@ client *createClient(connection *conn) {
c->obuf_soft_limit_reached_time = 0;
listSetFreeMethod(c->reply,freeClientReplyValue);
listSetDupMethod(c->reply,dupClientReplyValue);
c->btype = BLOCKED_NONE;
c->bpop.timeout = 0;
c->bpop.keys = dictCreate(&objectKeyHeapPointerValueDictType);
c->bpop.target = NULL;
c->bpop.xread_group = NULL;
c->bpop.xread_consumer = NULL;
c->bpop.xread_group_noack = 0;
c->bpop.numreplicas = 0;
c->bpop.reploffset = 0;
initClientBlockingState(c);
c->woff = 0;
c->watched_keys = listCreate();
c->pubsub_channels = dictCreate(&objectKeyPointerValueDictType);
......@@ -1588,7 +1581,7 @@ void freeClient(client *c) {
/* Deallocate structures used to block on blocking ops. */
if (c->flags & CLIENT_BLOCKED) unblockClient(c);
dictRelease(c->bpop.keys);
dictRelease(c->bstate.keys);
/* UNWATCH all the keys */
unwatchAllKeys(c);
......@@ -2033,6 +2026,8 @@ void resetClient(client *c) {
c->multibulklen = 0;
c->bulklen = -1;
c->slot = -1;
c->duration = 0;
c->flags &= ~CLIENT_EXECUTING_COMMAND;
if (c->deferred_reply_errors)
listRelease(c->deferred_reply_errors);
......@@ -3142,12 +3137,11 @@ NULL
* it also doesn't expect to be unblocked by CLIENT UNBLOCK */
if (target && target->flags & CLIENT_BLOCKED && moduleBlockedClientMayTimeout(target)) {
if (unblock_error)
addReplyError(target,
unblockClientOnError(target,
"-UNBLOCKED client unblocked via CLIENT UNBLOCK");
else
replyToBlockedClientTimedOut(target);
unblockClient(target);
updateStatsOnUnblock(target, 0, 0, 1);
unblockClientOnTimeout(target);
addReply(c,shared.cone);
} else {
addReply(c,shared.czero);
......
......@@ -3480,11 +3480,7 @@ void waitCommand(client *c) {
/* Otherwise block the client and put it into our list of clients
* waiting for ack from slaves. */
c->bpop.timeout = timeout;
c->bpop.reploffset = offset;
c->bpop.numreplicas = numreplicas;
listAddNodeHead(server.clients_waiting_acks,c);
blockClient(c,BLOCKED_WAIT);
blockForReplication(c,timeout,offset,numreplicas);
/* Make sure that the server will send an ACK request to all the slaves
* before returning to the event loop. */
......@@ -3518,16 +3514,16 @@ void processClientsWaitingReplicas(void) {
* offset and number of replicas, we remember it so the next client
* may be unblocked without calling replicationCountAcksByOffset()
* if the requested offset / replicas were equal or less. */
if (last_offset && last_offset >= c->bpop.reploffset &&
last_numreplicas >= c->bpop.numreplicas)
if (last_offset && last_offset >= c->bstate.reploffset &&
last_numreplicas >= c->bstate.numreplicas)
{
unblockClient(c);
addReplyLongLong(c,last_numreplicas);
} else {
int numreplicas = replicationCountAcksByOffset(c->bpop.reploffset);
int numreplicas = replicationCountAcksByOffset(c->bstate.reploffset);
if (numreplicas >= c->bpop.numreplicas) {
last_offset = c->bpop.reploffset;
if (numreplicas >= c->bstate.numreplicas) {
last_offset = c->bstate.reploffset;
last_numreplicas = numreplicas;
unblockClient(c);
addReplyLongLong(c,numreplicas);
......
......@@ -559,7 +559,7 @@ void scriptCall(scriptRunCtx *run_ctx, sds *err) {
goto error;
}
int call_flags = CMD_CALL_SLOWLOG | CMD_CALL_STATS;
int call_flags = CMD_CALL_NONE;
if (run_ctx->repl_flags & PROPAGATE_AOF) {
call_flags |= CMD_CALL_PROPAGATE_AOF;
}
......
......@@ -981,7 +981,7 @@ cleanup:
c->argc = c->argv_len = 0;
c->user = NULL;
c->argv = NULL;
freeClientArgv(c);
resetClient(c);
inuse--;
if (raise_error) {
......
......@@ -92,6 +92,10 @@ const char *replstateToString(int replstate);
/*============================ Utility functions ============================ */
/* This macro tells if we are in the context of loading an AOF. */
#define isAOFLoadingContext() \
((server.current_client && server.current_client->id == CLIENT_ID_AOF) ? 1 : 0)
/* We use a private localtime implementation which is fork-safe. The logging
* function of Redis may be called from other threads. */
void nolocks_localtime(struct tm *tmp, time_t t, time_t tz, int dst);
......@@ -3374,8 +3378,6 @@ int incrCommandStatsOnError(struct redisCommand *cmd, int flags) {
*
* The following flags can be passed:
* CMD_CALL_NONE No flags.
* CMD_CALL_SLOWLOG Check command speed and log in the slow log if needed.
* CMD_CALL_STATS Populate command stats.
* CMD_CALL_PROPAGATE_AOF Append command to AOF if it modified the dataset
* or if the client flags are forcing propagation.
* CMD_CALL_PROPAGATE_REPL Send command to slaves if it modified the dataset
......@@ -3411,6 +3413,15 @@ void call(client *c, int flags) {
long long dirty;
uint64_t client_old_flags = c->flags;
struct redisCommand *real_cmd = c->realcmd;
/* When call() is issued during loading the AOF we don't want commands called
* from module, exec or LUA to go into the slowlog or to populate statistics. */
int update_command_stats = !isAOFLoadingContext();
/* We want to be aware of a client which is making a first time attempt to execute this command
* and a client which is reprocessing command again (after being unblocked).
* Blocked clients can be blocked in different places and not always it means the call() function has been
* called. For example this is required for avoiding double logging to monitors.*/
int reprocessing_command = ((!server.execution_nesting) && (c->flags & CLIENT_EXECUTING_COMMAND)) ? 1 : 0;
/* Initialization: clear the flags that must be set by the command on
* demand, and initialize the array for additional commands propagation. */
......@@ -3429,10 +3440,11 @@ void call(client *c, int flags) {
const long long call_timer = ustime();
/* Update cache time, in case we have nested calls we want to
* update only on the first call */
/* Update cache time, and indicate we are starting command execution.
* in case we have nested calls we want to update only on the first call */
if (server.execution_nesting++ == 0) {
updateCachedTimeWithUs(0,call_timer);
c->flags |= CLIENT_EXECUTING_COMMAND;
}
monotime monotonic_start = 0;
......@@ -3440,7 +3452,13 @@ void call(client *c, int flags) {
monotonic_start = getMonotonicUs();
c->cmd->proc(c);
server.execution_nesting--;
if (--server.execution_nesting == 0) {
/* In case client is blocked after trying to execute the command,
* it means the execution is not yet completed and we MIGHT reprocess the command in the future. */
if (!(c->flags & CLIENT_BLOCKED))
c->flags &= ~(CLIENT_EXECUTING_COMMAND);
}
/* In order to avoid performance implication due to querying the clock using a system call 3 times,
* we use a monotonic clock, when we are sure its cost is very low, and fall back to non-monotonic call otherwise. */
......@@ -3450,7 +3468,7 @@ void call(client *c, int flags) {
else
duration = ustime() - call_timer;
c->duration = duration;
c->duration += duration;
dirty = server.dirty-dirty;
if (dirty < 0) dirty = 0;
......@@ -3471,11 +3489,6 @@ void call(client *c, int flags) {
c->flags |= CLIENT_CLOSE_AFTER_REPLY;
}
/* When EVAL is called loading the AOF we don't want commands called
* from Lua to go into the slowlog or to populate statistics. */
if (server.loading && c->flags & CLIENT_SCRIPT)
flags &= ~(CMD_CALL_SLOWLOG | CMD_CALL_STATS);
/* If the caller is Lua, we want to force the EVAL caller to propagate
* the script if the command flag or client flag are forcing the
* propagation. */
......@@ -3493,7 +3506,7 @@ void call(client *c, int flags) {
/* Record the latency this command induced on the main thread.
* unless instructed by the caller not to log. (happens when processing
* a MULTI-EXEC from inside an AOF). */
if (flags & CMD_CALL_SLOWLOG) {
if (update_command_stats) {
char *latency_event = (real_cmd->flags & CMD_FAST) ?
"fast-command" : "command";
latencyAddSampleIfNeeded(latency_event,duration/1000);
......@@ -3501,12 +3514,15 @@ void call(client *c, int flags) {
/* Log the command into the Slow log if needed.
* If the client is blocked we will handle slowlog when it is unblocked. */
if ((flags & CMD_CALL_SLOWLOG) && !(c->flags & CLIENT_BLOCKED))
slowlogPushCurrentCommand(c, real_cmd, duration);
/* Send the command to clients in MONITOR mode if applicable.
* Administrative commands are considered too dangerous to be shown. */
if (!(c->cmd->flags & (CMD_SKIP_MONITOR|CMD_ADMIN))) {
if (update_command_stats && !(c->flags & CLIENT_BLOCKED))
slowlogPushCurrentCommand(c, real_cmd, c->duration);
/* Send the command to clients in MONITOR mode if applicable,
* since some administrative commands are considered too dangerous to be shown.
* Other exceptions is a client which is unblocked and retring to process the command
* or we are currently in the process of loading AOF. */
if (update_command_stats && !reprocessing_command &&
!(c->cmd->flags & (CMD_SKIP_MONITOR|CMD_ADMIN))) {
robj **argv = c->original_argv ? c->original_argv : c->argv;
int argc = c->original_argv ? c->original_argc : c->argc;
replicationFeedMonitors(c,server.monitors,c->db->id,argv,argc);
......@@ -3517,13 +3533,13 @@ void call(client *c, int flags) {
if (!(c->flags & CLIENT_BLOCKED))
freeClientOriginalArgv(c);
/* populate the per-command statistics that we show in INFO commandstats. */
if (flags & CMD_CALL_STATS) {
real_cmd->microseconds += duration;
/* populate the per-command statistics that we show in INFO commandstats.
* If the client is blocked we will handle latency stats and duration when it is unblocked. */
if (update_command_stats && !(c->flags & CLIENT_BLOCKED)) {
real_cmd->calls++;
/* If the client is blocked we will handle latency stats when it is unblocked. */
real_cmd->microseconds += c->duration;
if (server.latency_tracking_enabled && !(c->flags & CLIENT_BLOCKED))
updateCommandLatencyHistogram(&(real_cmd->latency_histogram), duration*1000);
updateCommandLatencyHistogram(&(real_cmd->latency_histogram), c->duration*1000);
}
/* Propagate the command into the AOF and replication link.
......@@ -3584,7 +3600,8 @@ void call(client *c, int flags) {
}
}
server.stat_numcommands++;
if (!(c->flags & CLIENT_BLOCKED))
server.stat_numcommands++;
/* Record peak memory after each command and before the eviction that runs
* before the next command. */
......@@ -3735,6 +3752,10 @@ int processCommand(client *c) {
moduleCallCommandFilters(c);
/* in case we are starting to ProcessCommand and we already have a command we assume
* this is a reprocessing of this command, so we do not want to perform some of the actions again. */
int client_reprocessing_command = c->cmd ? 1 : 0;
/* Handle possible security attacks. */
if (!strcasecmp(c->argv[0]->ptr,"host:") || !strcasecmp(c->argv[0]->ptr,"post")) {
securityWarningCommand(c);
......@@ -3746,36 +3767,40 @@ int processCommand(client *c) {
if (server.busy_module_yield_flags != BUSY_MODULE_YIELD_NONE &&
!(server.busy_module_yield_flags & BUSY_MODULE_YIELD_CLIENTS))
{
c->bpop.timeout = 0;
blockClient(c,BLOCKED_POSTPONE);
blockPostponeClient(c);
return C_OK;
}
/* Now lookup the command and check ASAP about trivial error conditions
* such as wrong arity, bad command name and so forth. */
c->cmd = c->lastcmd = c->realcmd = lookupCommand(c->argv,c->argc);
sds err;
if (!commandCheckExistence(c, &err)) {
rejectCommandSds(c, err);
return C_OK;
}
if (!commandCheckArity(c, &err)) {
rejectCommandSds(c, err);
return C_OK;
}
/* Check if the command is marked as protected and the relevant configuration allows it */
if (c->cmd->flags & CMD_PROTECTED) {
if ((c->cmd->proc == debugCommand && !allowProtectedAction(server.enable_debug_cmd, c)) ||
(c->cmd->proc == moduleCommand && !allowProtectedAction(server.enable_module_cmd, c)))
{
rejectCommandFormat(c,"%s command not allowed. If the %s option is set to \"local\", "
"you can run it from a local connection, otherwise you need to set this option "
"in the configuration file, and then restart the server.",
c->cmd->proc == debugCommand ? "DEBUG" : "MODULE",
c->cmd->proc == debugCommand ? "enable-debug-command" : "enable-module-command");
* such as wrong arity, bad command name and so forth.
* In case we are reprocessing a command after it was blocked,
* we do not have to repeat the same checks */
if (!client_reprocessing_command) {
c->cmd = c->lastcmd = c->realcmd = lookupCommand(c->argv,c->argc);
sds err;
if (!commandCheckExistence(c, &err)) {
rejectCommandSds(c, err);
return C_OK;
}
if (!commandCheckArity(c, &err)) {
rejectCommandSds(c, err);
return C_OK;
}
/* Check if the command is marked as protected and the relevant configuration allows it */
if (c->cmd->flags & CMD_PROTECTED) {
if ((c->cmd->proc == debugCommand && !allowProtectedAction(server.enable_debug_cmd, c)) ||
(c->cmd->proc == moduleCommand && !allowProtectedAction(server.enable_module_cmd, c)))
{
rejectCommandFormat(c,"%s command not allowed. If the %s option is set to \"local\", "
"you can run it from a local connection, otherwise you need to set this option "
"in the configuration file, and then restart the server.",
c->cmd->proc == debugCommand ? "DEBUG" : "MODULE",
c->cmd->proc == debugCommand ? "enable-debug-command" : "enable-module-command");
return C_OK;
}
}
}
......@@ -4028,8 +4053,7 @@ int processCommand(client *c) {
((isPausedActions(PAUSE_ACTION_CLIENT_ALL)) ||
((isPausedActions(PAUSE_ACTION_CLIENT_WRITE)) && is_may_replicate_command)))
{
c->bpop.timeout = 0;
blockClient(c,BLOCKED_POSTPONE);
blockPostponeClient(c);
return C_OK;
}
......@@ -5444,7 +5468,9 @@ sds genRedisInfoString(dict *section_dict, int all_sections, int everything) {
/* Clients */
if (all_sections || (dictFind(section_dict,"clients") != NULL)) {
size_t maxin, maxout;
unsigned long blocking_keys, blocking_keys_on_nokey;
getExpansiveClientsInfo(&maxin,&maxout);
totalNumberOfBlockingKeys(&blocking_keys, &blocking_keys_on_nokey);
if (sections++) info = sdscat(info,"\r\n");
info = sdscatprintf(info,
"# Clients\r\n"
......@@ -5455,14 +5481,18 @@ sds genRedisInfoString(dict *section_dict, int all_sections, int everything) {
"client_recent_max_output_buffer:%zu\r\n"
"blocked_clients:%d\r\n"
"tracking_clients:%d\r\n"
"clients_in_timeout_table:%llu\r\n",
"clients_in_timeout_table:%llu\r\n"
"total_blocking_keys:%lu\r\n"
"total_blocking_keys_on_nokey:%lu\r\n",
listLength(server.clients)-listLength(server.slaves),
getClusterConnectionsCount(),
server.maxclients,
maxin, maxout,
server.blocked_clients,
server.tracking_clients,
(unsigned long long) raxSize(server.clients_timeout_table));
(unsigned long long) raxSize(server.clients_timeout_table),
blocking_keys,
blocking_keys_on_nokey);
}
/* Memory */
......
......@@ -359,7 +359,11 @@ extern int configOOMScoreAdjValuesDefaults[CONFIG_OOM_COUNT];
#define CLIENT_LUA_DEBUG_SYNC (1<<26) /* EVAL debugging without fork() */
#define CLIENT_MODULE (1<<27) /* Non connected client used by some module. */
#define CLIENT_PROTECTED (1<<28) /* Client should not be freed for now. */
/* #define CLIENT_... (1<<29) currently unused, feel free to use in the future */
#define CLIENT_EXECUTING_COMMAND (1<<29) /* Indicates that the client is currently in the process of handling
a command. usually this will be marked only during call()
however, blocked clients might have this flag kept until they
will try to reprocess the command. */
#define CLIENT_PENDING_COMMAND (1<<30) /* Indicates the client has a fully
* parsed command ready for execution. */
#define CLIENT_TRACKING (1ULL<<31) /* Client enabled keys tracking in order to
......@@ -388,15 +392,18 @@ extern int configOOMScoreAdjValuesDefaults[CONFIG_OOM_COUNT];
/* Client block type (btype field in client structure)
* if CLIENT_BLOCKED flag is set. */
#define BLOCKED_NONE 0 /* Not blocked, no CLIENT_BLOCKED flag set. */
#define BLOCKED_LIST 1 /* BLPOP & co. */
#define BLOCKED_WAIT 2 /* WAIT for synchronous replication. */
#define BLOCKED_MODULE 3 /* Blocked by a loadable module. */
#define BLOCKED_STREAM 4 /* XREAD. */
#define BLOCKED_ZSET 5 /* BZPOP et al. */
#define BLOCKED_POSTPONE 6 /* Blocked by processCommand, re-try processing later. */
#define BLOCKED_SHUTDOWN 7 /* SHUTDOWN. */
#define BLOCKED_NUM 8 /* Number of blocked states. */
typedef enum blocking_type {
BLOCKED_NONE, /* Not blocked, no CLIENT_BLOCKED flag set. */
BLOCKED_LIST, /* BLPOP & co. */
BLOCKED_WAIT, /* WAIT for synchronous replication. */
BLOCKED_MODULE, /* Blocked by a loadable module. */
BLOCKED_STREAM, /* XREAD. */
BLOCKED_ZSET, /* BZPOP et al. */
BLOCKED_POSTPONE, /* Blocked by processCommand, re-try processing later. */
BLOCKED_SHUTDOWN, /* SHUTDOWN. */
BLOCKED_NUM, /* Number of blocked states. */
BLOCKED_END /* End of enumeration */
} blocking_type;
/* Client request types */
#define PROTO_REQ_INLINE 1
......@@ -569,13 +576,11 @@ typedef enum {
/* Command call flags, see call() function */
#define CMD_CALL_NONE 0
#define CMD_CALL_SLOWLOG (1<<0)
#define CMD_CALL_STATS (1<<1)
#define CMD_CALL_PROPAGATE_AOF (1<<2)
#define CMD_CALL_PROPAGATE_REPL (1<<3)
#define CMD_CALL_PROPAGATE_AOF (1<<0)
#define CMD_CALL_PROPAGATE_REPL (1<<1)
#define CMD_CALL_PROPAGATE (CMD_CALL_PROPAGATE_AOF|CMD_CALL_PROPAGATE_REPL)
#define CMD_CALL_FULL (CMD_CALL_SLOWLOG | CMD_CALL_STATS | CMD_CALL_PROPAGATE)
#define CMD_CALL_FROM_MODULE (1<<4) /* From RM_Call */
#define CMD_CALL_FULL (CMD_CALL_PROPAGATE)
#define CMD_CALL_FROM_MODULE (1<<2) /* From RM_Call */
/* Command propagation flags, see propagateNow() function */
#define PROPAGATE_NONE 0
......@@ -992,27 +997,13 @@ typedef struct multiState {
* The fields used depend on client->btype. */
typedef struct blockingState {
/* Generic fields. */
long count; /* Elements to pop if count was specified (BLMPOP/BZMPOP), -1 otherwise. */
mstime_t timeout; /* Blocking operation timeout. If UNIX current time
* is > timeout then the operation timed out. */
/* BLOCKED_LIST, BLOCKED_ZSET and BLOCKED_STREAM */
dict *keys; /* The keys we are waiting to terminate a blocking
* operation such as BLPOP or XREAD. Or NULL. */
robj *target; /* The key that should receive the element,
* for BLMOVE. */
struct blockPos {
int wherefrom; /* Where to pop from */
int whereto; /* Where to push to */
} blockpos; /* The positions in the src/dst lists/zsets
* where we want to pop/push an element
* for BLPOP, BRPOP, BLMOVE and BZMPOP. */
/* BLOCK_STREAM */
size_t xread_count; /* XREAD COUNT option. */
robj *xread_group; /* XREADGROUP group name. */
robj *xread_consumer; /* XREADGROUP consumer name. */
int xread_group_noack;
blocking_type btype; /* Type of blocking op if CLIENT_BLOCKED. */
mstime_t timeout; /* Blocking operation timeout. If UNIX current time
* is > timeout then the operation timed out. */
int unblock_on_nokey; /* Whether to unblock the client when at least one of the keys
is deleted or does not exist anymore */
/* BLOCKED_LIST, BLOCKED_ZSET and BLOCKED_STREAM or any other Keys related blocking */
dict *keys; /* The keys we are blocked on */
/* BLOCKED_WAIT */
int numreplicas; /* Number of replicas we are waiting for ACK. */
......@@ -1029,7 +1020,7 @@ typedef struct blockingState {
* operation such as B[LR]POP, but received new data in the context of the
* last executed command.
*
* After the execution of every command or script, we run this list to check
* After the execution of every command or script, we iterate over this list to check
* if as a result we should serve data to clients blocked, unblocking them.
* Note that server.ready_keys will not have duplicates as there dictionary
* also called ready_keys in every structure representing a Redis database,
......@@ -1167,8 +1158,7 @@ typedef struct client {
int slave_capa; /* Slave capabilities: SLAVE_CAPA_* bitwise OR. */
int slave_req; /* Slave requirements: SLAVE_REQ_* */
multiState mstate; /* MULTI/EXEC state */
int btype; /* Type of blocking op if CLIENT_BLOCKED. */
blockingState bpop; /* blocking state */
blockingState bstate; /* blocking state */
long long woff; /* Last write global replication offset. */
list *watched_keys; /* Keys WATCHED for MULTI/EXEC CAS */
dict *pubsub_channels; /* channels a client is interested in (SUBSCRIBE) */
......@@ -2635,7 +2625,6 @@ int listTypeEqual(listTypeEntry *entry, robj *o);
void listTypeDelete(listTypeIterator *iter, listTypeEntry *entry);
robj *listTypeDup(robj *o);
void listTypeDelRange(robj *o, long start, long stop);
void unblockClientWaitingData(client *c);
void popGenericCommand(client *c, int where);
void listElementsRemoved(client *c, robj *key, int where, robj *o, long count, int signal, int *deleted);
typedef enum {
......@@ -2938,6 +2927,7 @@ int overMaxmemoryAfterAlloc(size_t moremem);
uint64_t getCommandFlags(client *c);
int processCommand(client *c);
int processPendingCommandAndInputBuffer(client *c);
int processCommandAndResetClient(client *c);
void setupSignalHandlers(void);
void removeSignalHandlers(void);
int createSocketAcceptHandler(connListener *sfd, aeFileProc *accept_handler);
......@@ -3166,6 +3156,7 @@ int objectSetLRUOrLFU(robj *val, long long lfu_freq, long long lru_idle,
#define LOOKUP_NOSTATS (1<<2) /* Don't update keyspace hits/misses counters. */
#define LOOKUP_WRITE (1<<3) /* Delete expired keys even in replicas. */
#define LOOKUP_NOEXPIRE (1<<4) /* Avoid deleting lazy expired keys. */
#define LOKKUP_NOEFFECTS (LOOKUP_NONOTIFY | LOOKUP_NOSTATS | LOOKUP_NOTOUCH | LOOKUP_NOEXPIRE) /* Avoid any effects from fetching the key */
void dbAdd(redisDb *db, robj *key, robj *val);
int dbAddRDBLoad(redisDb *db, sds key, robj *val);
......@@ -3281,20 +3272,28 @@ typedef struct luaScript {
robj *body;
} luaScript;
/* Blocked clients */
/* Blocked clients API */
void processUnblockedClients(void);
void initClientBlockingState(client *c);
void blockClient(client *c, int btype);
void unblockClient(client *c);
void unblockClientOnTimeout(client *c);
void unblockClientOnError(client *c, const char *err_str);
void queueClientForReprocessing(client *c);
void replyToBlockedClientTimedOut(client *c);
int getTimeoutFromObjectOrReply(client *c, robj *object, mstime_t *timeout, int unit);
void disconnectAllBlockedClients(void);
void handleClientsBlockedOnKeys(void);
void signalKeyAsReady(redisDb *db, robj *key, int type);
void blockForKeys(client *c, int btype, robj **keys, int numkeys, mstime_t timeout, int unblock_on_nokey);
void blockClientShutdown(client *c);
void blockPostponeClient(client *c);
void blockForReplication(client *c, mstime_t timeout, long long offset, long numreplicas);
void signalDeletedKeyAsReady(redisDb *db, robj *key, int type);
void blockForKeys(client *c, int btype, robj **keys, int numkeys, long count, mstime_t timeout, robj *target, struct blockPos *blockpos, streamID *ids, int unblock_on_nokey);
void updateStatsOnUnblock(client *c, long blocked_us, long reply_us, int had_errors);
void scanDatabaseForDeletedKeys(redisDb *emptied, redisDb *replaced_with);
void totalNumberOfBlockingKeys(unsigned long *blocking_keys, unsigned long *bloking_keys_on_nokey);
/* timeout.c -- Blocked clients timeout and connections timeout. */
void addClientToTimeoutTable(client *c);
......
......@@ -1200,103 +1200,6 @@ void rpoplpushCommand(client *c) {
lmoveGenericCommand(c, LIST_TAIL, LIST_HEAD);
}
/*-----------------------------------------------------------------------------
* Blocking POP operations
*----------------------------------------------------------------------------*/
/* This is a helper function for handleClientsBlockedOnKeys(). Its work
* is to serve a specific client (receiver) that is blocked on 'key'
* in the context of the specified 'db', doing the following:
*
* 1) Provide the client with the 'value' element or a range of elements.
* We will do the pop in here and caller does not need to bother the return.
* 2) If the dstkey is not NULL (we are serving a BLMOVE) also push the
* 'value' element on the destination list (the "push" side of the command).
* 3) Propagate the resulting BRPOP, BLPOP, BLMPOP and additional xPUSH if any into
* the AOF and replication channel.
*
* The argument 'wherefrom' is LIST_TAIL or LIST_HEAD, and indicates if the
* 'value' element was popped from the head (BLPOP) or tail (BRPOP) so that
* we can propagate the command properly.
*
* The argument 'whereto' is LIST_TAIL or LIST_HEAD, and indicates if the
* 'value' element is to be pushed to the head or tail so that we can
* propagate the command properly.
*
* 'deleted' is an optional output argument to get an indication
* if the key got deleted by this function. */
void serveClientBlockedOnList(client *receiver, robj *o, robj *key, robj *dstkey, redisDb *db, int wherefrom, int whereto, int *deleted)
{
robj *argv[5];
robj *value = NULL;
if (deleted) *deleted = 0;
if (dstkey == NULL) {
/* Propagate the [LR]POP operation. */
argv[0] = (wherefrom == LIST_HEAD) ? shared.lpop :
shared.rpop;
argv[1] = key;
if (receiver->lastcmd->proc == blmpopCommand) {
/* Propagate the [LR]POP COUNT operation. */
long count = receiver->bpop.count;
serverAssert(count > 0);
long llen = listTypeLength(o);
serverAssert(llen > 0);
argv[2] = createStringObjectFromLongLong((count > llen) ? llen : count);
alsoPropagate(db->id, argv, 3, PROPAGATE_AOF|PROPAGATE_REPL);
decrRefCount(argv[2]);
/* Pop a range of elements in a nested arrays way. */
listPopRangeAndReplyWithKey(receiver, o, key, wherefrom, count, 0, deleted);
return;
}
alsoPropagate(db->id, argv, 2, PROPAGATE_AOF|PROPAGATE_REPL);
/* BRPOP/BLPOP */
value = listTypePop(o, wherefrom);
serverAssert(value != NULL);
addReplyArrayLen(receiver,2);
addReplyBulk(receiver,key);
addReplyBulk(receiver,value);
/* We don't call signalModifiedKey() as it was already called
* when an element was pushed on the list. */
listElementsRemoved(receiver,key,wherefrom,o,1,0,deleted);
} else {
/* BLMOVE */
robj *dstobj =
lookupKeyWrite(receiver->db,dstkey);
if (!(dstobj &&
checkType(receiver,dstobj,OBJ_LIST)))
{
value = listTypePop(o, wherefrom);
serverAssert(value != NULL);
/* "lpush" or "rpush" notify event will be notified by lmoveHandlePush. */
lmoveHandlePush(receiver,dstkey,dstobj,value,whereto);
/* Propagate the LMOVE/RPOPLPUSH operation. */
int isbrpoplpush = (receiver->lastcmd->proc == brpoplpushCommand);
argv[0] = isbrpoplpush ? shared.rpoplpush : shared.lmove;
argv[1] = key;
argv[2] = dstkey;
argv[3] = getStringObjectFromListPosition(wherefrom);
argv[4] = getStringObjectFromListPosition(whereto);
alsoPropagate(db->id,argv,(isbrpoplpush ? 3 : 5),PROPAGATE_AOF|PROPAGATE_REPL);
/* We don't call signalModifiedKey() as it was already called
* when an element was pushed on the list. */
listElementsRemoved(receiver,key,wherefrom,o,1,0,deleted);
}
}
if (value) decrRefCount(value);
}
/* Blocking RPOP/LPOP/LMPOP
*
* 'numkeys' is the number of keys.
......@@ -1368,8 +1271,7 @@ void blockingPopGenericCommand(client *c, robj **keys, int numkeys, int where, i
}
/* If the keys do not exist we must block */
struct blockPos pos = {where};
blockForKeys(c,BLOCKED_LIST,keys,numkeys,count,timeout,NULL,&pos,NULL,0);
blockForKeys(c,BLOCKED_LIST,keys,numkeys,timeout,0);
}
/* BLPOP <key> [<key> ...] <timeout> */
......@@ -1393,8 +1295,7 @@ void blmoveGenericCommand(client *c, int wherefrom, int whereto, mstime_t timeou
addReplyNull(c);
} else {
/* The list is empty and the client blocks. */
struct blockPos pos = {wherefrom, whereto};
blockForKeys(c,BLOCKED_LIST,c->argv + 1,1,-1,timeout,c->argv[2],&pos,NULL,0);
blockForKeys(c,BLOCKED_LIST,c->argv + 1,1,timeout,0);
}
} else {
/* The list exists and has elements, so
......
......@@ -2409,27 +2409,19 @@ void xreadCommand(client *c) {
addReplyNullArray(c);
goto cleanup;
}
blockForKeys(c, BLOCKED_STREAM, c->argv+streams_arg, streams_count,
-1, timeout, NULL, NULL, ids, xreadgroup);
/* If no COUNT is given and we block, set a relatively small count:
* in case the ID provided is too low, we do not want the server to
* block just to serve this client a huge stream of messages. */
c->bpop.xread_count = count ? count : XREAD_BLOCKED_DEFAULT_COUNT;
/* If this is a XREADGROUP + GROUP we need to remember for which
* group and consumer name we are blocking, so later when one of the
* keys receive more data, we can call streamReplyWithRange() passing
* the right arguments. */
if (groupname) {
incrRefCount(groupname);
incrRefCount(consumername);
c->bpop.xread_group = groupname;
c->bpop.xread_consumer = consumername;
c->bpop.xread_group_noack = noack;
} else {
c->bpop.xread_group = NULL;
c->bpop.xread_consumer = NULL;
/* We change the '$' to the current last ID for this stream. this is
* Since later on when we unblock on arriving data - we would like to
* re-process the command and in case '$' stays we will spin-block forever.
*/
for (int id_idx = 0; id_idx < streams_count; id_idx++) {
int arg_idx = id_idx + streams_arg + streams_count;
if (strcmp(c->argv[arg_idx]->ptr,"$") == 0) {
robj *argv_streamid = createObjectFromStreamID(&ids[id_idx]);
rewriteClientCommandArgument(c, arg_idx, argv_streamid);
decrRefCount(argv_streamid);
}
}
blockForKeys(c, BLOCKED_STREAM, c->argv+streams_arg, streams_count, timeout, xreadgroup);
goto cleanup;
}
......
......@@ -4071,8 +4071,7 @@ void blockingGenericZpopCommand(client *c, robj **keys, int numkeys, int where,
}
/* If the keys do not exist we must block */
struct blockPos pos = {where};
blockForKeys(c,BLOCKED_ZSET,keys,numkeys,count,timeout,NULL,&pos,NULL,0);
blockForKeys(c,BLOCKED_ZSET,keys,numkeys,timeout,0);
}
// BZPOPMIN key [key ...] timeout
......
......@@ -36,12 +36,11 @@
* Otherwise 0 is returned and no operation is performed. */
int checkBlockedClientTimeout(client *c, mstime_t now) {
if (c->flags & CLIENT_BLOCKED &&
c->bpop.timeout != 0
&& c->bpop.timeout < now)
c->bstate.timeout != 0
&& c->bstate.timeout < now)
{
/* Handle blocking operation specific timeout. */
replyToBlockedClientTimedOut(c);
unblockClient(c);
unblockClientOnTimeout(c);
return 1;
} else {
return 0;
......@@ -71,7 +70,7 @@ int clientsCronHandleTimeout(client *c, mstime_t now_ms) {
* into keys no longer served by this server. */
if (server.cluster_enabled) {
if (clusterRedirectBlockedClientIfNeeded(c))
unblockClient(c);
unblockClientOnError(c, NULL);
}
}
return 0;
......@@ -112,8 +111,8 @@ void decodeTimeoutKey(unsigned char *buf, uint64_t *toptr, client **cptr) {
* to handle blocked clients timeouts. The client is not added to the list
* if its timeout is zero (block forever). */
void addClientToTimeoutTable(client *c) {
if (c->bpop.timeout == 0) return;
uint64_t timeout = c->bpop.timeout;
if (c->bstate.timeout == 0) return;
uint64_t timeout = c->bstate.timeout;
unsigned char buf[CLIENT_ST_KEYLEN];
encodeTimeoutKey(buf,timeout,c);
if (raxTryInsert(server.clients_timeout_table,buf,sizeof(buf),NULL,NULL))
......@@ -125,7 +124,7 @@ void addClientToTimeoutTable(client *c) {
void removeClientFromTimeoutTable(client *c) {
if (!(c->flags & CLIENT_IN_TO_TABLE)) return;
c->flags &= ~CLIENT_IN_TO_TABLE;
uint64_t timeout = c->bpop.timeout;
uint64_t timeout = c->bstate.timeout;
unsigned char buf[CLIENT_ST_KEYLEN];
encodeTimeoutKey(buf,timeout,c);
raxRemove(server.clients_timeout_table,buf,sizeof(buf),NULL);
......
......@@ -182,7 +182,8 @@ start_server {tags {"repl external:skip"}} {
[$B lrange foo 0 -1] eq {a b c}
} else {
fail "Master and replica have different digest: [$A debug digest] VS [$B debug digest]"
}
}
assert_match {*calls=1,*,rejected_calls=0,failed_calls=1*} [cmdrstat blpop $B]
}
}
}
......
......@@ -359,6 +359,14 @@ int test_rm_call(RedisModuleCtx *ctx, RedisModuleString **argv, int argc){
return REDISMODULE_OK;
}
/* wrapper for RM_Call which also replicates the module command */
int test_rm_call_replicate(RedisModuleCtx *ctx, RedisModuleString **argv, int argc){
test_rm_call(ctx, argv, argc);
RedisModule_ReplicateVerbatim(ctx);
return REDISMODULE_OK;
}
/* wrapper for RM_Call with flags */
int test_rm_call_flags(RedisModuleCtx *ctx, RedisModuleString **argv, int argc){
if(argc < 3){
......@@ -497,6 +505,8 @@ int RedisModule_OnLoad(RedisModuleCtx *ctx, RedisModuleString **argv, int argc)
return REDISMODULE_ERR;
if (RedisModule_CreateCommand(ctx, "test.rm_call_flags", test_rm_call_flags,"allow-stale", 0, 0, 0) == REDISMODULE_ERR)
return REDISMODULE_ERR;
if (RedisModule_CreateCommand(ctx, "test.rm_call_replicate", test_rm_call_replicate,"allow-stale", 0, 0, 0) == REDISMODULE_ERR)
return REDISMODULE_ERR;
return REDISMODULE_OK;
}
......@@ -173,6 +173,49 @@ start_server {tags {"introspection"}} {
$rd close
}
test {MONITOR log blocked command only once} {
# need to reconnect in order to reset the clients state
reconnect
set rd [redis_deferring_client]
set bc [redis_deferring_client]
r del mylist
$rd monitor
$rd read ; # Discard the OK
$bc blpop mylist 0
wait_for_blocked_clients_count 1
r lpush mylist 1
wait_for_blocked_clients_count 0
r lpush mylist 2
# we expect to see the blpop on the monitor first
assert_match {*"blpop"*"mylist"*"0"*} [$rd read]
# we scan out all the info commands on the monitor
set monitor_output [$rd read]
while { [string match {*"info"*} $monitor_output] } {
set monitor_output [$rd read]
}
# we expect to locate the lpush right when the client was unblocked
assert_match {*"lpush"*"mylist"*"1"*} $monitor_output
# we scan out all the info commands
set monitor_output [$rd read]
while { [string match {*"info"*} $monitor_output] } {
set monitor_output [$rd read]
}
# we expect to see the next lpush and not duplicate blpop command
assert_match {*"lpush"*"mylist"*"2"*} $monitor_output
$rd close
$bc close
}
test {CLIENT GETNAME should return NIL if name is not assigned} {
r client getname
......
set testmodule [file normalize tests/modules/propagate.so]
set miscmodule [file normalize tests/modules/misc.so]
set keyspace_events [file normalize tests/modules/keyspace_events.so]
tags "modules" {
......@@ -721,5 +722,42 @@ tags "modules aof" {
assert_equal [s 0 unexpected_error_replies] 0
}
}
test "Modules RM_Call does not update stats during aof load: AOF-load type $aofload_type" {
start_server [list overrides [list loadmodule "$miscmodule"]] {
# Enable the AOF
r config set appendonly yes
r config set auto-aof-rewrite-percentage 0 ; # Disable auto-rewrite.
waitForBgrewriteaof r
r config resetstat
r set foo bar
r EVAL {return redis.call('SET', KEYS[1], ARGV[1])} 1 foo bar2
r test.rm_call_replicate set foo bar3
r EVAL {return redis.call('test.rm_call_replicate',ARGV[1],KEYS[1],ARGV[2])} 1 foo set bar4
r multi
r set foo bar5
r EVAL {return redis.call('SET', KEYS[1], ARGV[1])} 1 foo bar6
r test.rm_call_replicate set foo bar7
r EVAL {return redis.call('test.rm_call_replicate',ARGV[1],KEYS[1],ARGV[2])} 1 foo set bar8
r exec
assert_match {*calls=8,*,rejected_calls=0,failed_calls=0} [cmdrstat set r]
# Load the AOF
if {$aofload_type == "debug_cmd"} {
r config resetstat
r debug loadaof
} else {
r config rewrite
restart_server 0 true false
wait_done_loading r
}
assert_no_match {*calls=*} [cmdrstat set r]
}
}
}
}
......@@ -200,4 +200,26 @@ start_server {tags {"slowlog"} overrides {slowlog-log-slower-than 1000000}} {
assert_equal 3 [llength [r slowlog get -1]]
assert_equal 3 [llength [r slowlog get 3]]
}
test {SLOWLOG - blocking command is reported only after unblocked} {
# Cleanup first
r del mylist
# create a test client
set rd [redis_deferring_client]
# config the slowlog and reset
r config set slowlog-log-slower-than 0
r config set slowlog-max-len 110
r slowlog reset
$rd BLPOP mylist 0
wait_for_blocked_clients_count 1 50 20
assert_equal 0 [llength [regexp -all -inline (?=BLPOP) [r slowlog get]]]
r LPUSH mylist 1
wait_for_blocked_clients_count 0 50 20
assert_equal 1 [llength [regexp -all -inline (?=BLPOP) [r slowlog get]]]
$rd close
}
}
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