Commit dd53453e authored by antirez's avatar antirez
Browse files

Threaded core commands: clients wait queue WIP.

parent 52db745c
...@@ -144,13 +144,15 @@ void queueClientForReprocessing(client *c) { ...@@ -144,13 +144,15 @@ void queueClientForReprocessing(client *c) {
/* Unblock a client calling the right function depending on the kind /* Unblock a client calling the right function depending on the kind
* of operation the client is blocking for. */ * of operation the client is blocking for. */
void unblockClient(client *c) { void unblockClient(client *c) {
if (c->btype == BLOCKED_LIST || int btype = c->btype;
c->btype == BLOCKED_ZSET ||
c->btype == BLOCKED_STREAM) { if (btype == BLOCKED_LIST ||
btype == BLOCKED_ZSET ||
btype == BLOCKED_STREAM) {
unblockClientWaitingData(c); unblockClientWaitingData(c);
} else if (c->btype == BLOCKED_WAIT) { } else if (btype == BLOCKED_WAIT) {
unblockClientWaitingReplicas(c); unblockClientWaitingReplicas(c);
} else if (c->btype == BLOCKED_MODULE) { } else if (btype == BLOCKED_MODULE) {
if (moduleClientIsBlockedOnKeys(c)) unblockClientWaitingData(c); if (moduleClientIsBlockedOnKeys(c)) unblockClientWaitingData(c);
unblockClientFromModule(c); unblockClientFromModule(c);
} else { } else {
...@@ -159,11 +161,15 @@ void unblockClient(client *c) { ...@@ -159,11 +161,15 @@ void unblockClient(client *c) {
/* Clear the flags, and put the client in the unblocked list so that /* Clear the flags, and put the client in the unblocked list so that
* we'll process new commands in its query buffer ASAP. */ * we'll process new commands in its query buffer ASAP. */
server.blocked_clients--; server.blocked_clients--;
server.blocked_clients_by_type[c->btype]--; server.blocked_clients_by_type[btype]--;
c->flags &= ~CLIENT_BLOCKED; c->flags &= ~CLIENT_BLOCKED;
c->btype = BLOCKED_NONE; c->btype = BLOCKED_NONE;
removeClientFromTimeoutTable(c); removeClientFromTimeoutTable(c);
queueClientForReprocessing(c); queueClientForReprocessing(c);
/* Re-execute the command if it was not executed at all since the
* client was blocked because of key locks. */
if (btype == BLOCKED_LOCK) processCommand(c);
} }
/* This function gets called when a blocked client timed out in order to /* This function gets called when a blocked client timed out in order to
...@@ -172,7 +178,8 @@ void unblockClient(client *c) { ...@@ -172,7 +178,8 @@ void unblockClient(client *c) {
void replyToBlockedClientTimedOut(client *c) { void replyToBlockedClientTimedOut(client *c) {
if (c->btype == BLOCKED_LIST || if (c->btype == BLOCKED_LIST ||
c->btype == BLOCKED_ZSET || c->btype == BLOCKED_ZSET ||
c->btype == BLOCKED_STREAM) { c->btype == BLOCKED_STREAM ||
c->btype == BLOCKED_LOCK) {
addReplyNullArray(c); addReplyNullArray(c);
} else if (c->btype == BLOCKED_WAIT) { } else if (c->btype == BLOCKED_WAIT) {
addReplyLongLong(c,replicationCountAcksByOffset(c->bpop.reploffset)); addReplyLongLong(c,replicationCountAcksByOffset(c->bpop.reploffset));
......
...@@ -3500,6 +3500,11 @@ int processCommand(client *c) { ...@@ -3500,6 +3500,11 @@ int processCommand(client *c) {
} }
} }
/* We want to block this client if the keys it is going to access
* are locked. */
if (dictSize(c->db->locked_keys)) {
}
/* Handle the maxmemory directive. /* Handle the maxmemory directive.
* *
* Note that we do not want to reclaim memory if we are here re-entering * Note that we do not want to reclaim memory if we are here re-entering
......
...@@ -265,7 +265,8 @@ typedef long long ustime_t; /* microsecond time type. */ ...@@ -265,7 +265,8 @@ typedef long long ustime_t; /* microsecond time type. */
#define BLOCKED_MODULE 3 /* Blocked by a loadable module. */ #define BLOCKED_MODULE 3 /* Blocked by a loadable module. */
#define BLOCKED_STREAM 4 /* XREAD. */ #define BLOCKED_STREAM 4 /* XREAD. */
#define BLOCKED_ZSET 5 /* BZPOP et al. */ #define BLOCKED_ZSET 5 /* BZPOP et al. */
#define BLOCKED_NUM 6 /* Number of blocked states. */ #define BLOCKED_LOCK 6 /* Waiting for locked keys. */
#define BLOCKED_NUM 7 /* Number of blocked states. */
/* Client request types */ /* Client request types */
#define PROTO_REQ_INLINE 1 #define PROTO_REQ_INLINE 1
......
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