Unverified Commit 0c653428 authored by Madelyn Olson's avatar Madelyn Olson Committed by GitHub
Browse files

Remove race condition and consistency issues with client pause and threaded IO (#8520)

clientsArePaused isn't thread safe because it has a side effect of attempting to unpause,
which may cause multiple threads concurrently updating the unblocked_clients global list.
This change resolves this issue by no longer postponing client for threaded reads when
clients are paused and then skipping the check for client paused for threaded reads,
in case one is postponed and then clients are paused. (I don't think this is strictly possible,
but being defensive seems better here)
parent 8d70d498
...@@ -98,6 +98,9 @@ void processUnblockedClients(void) { ...@@ -98,6 +98,9 @@ void processUnblockedClients(void) {
client *c; client *c;
while (listLength(server.unblocked_clients)) { while (listLength(server.unblocked_clients)) {
/* If clients are paused we yield for now, since
* we don't want to process any commands later. */
if (clientsArePaused()) return;
ln = listFirst(server.unblocked_clients); ln = listFirst(server.unblocked_clients);
serverAssert(ln != NULL); serverAssert(ln != NULL);
c = ln->value; c = ln->value;
...@@ -109,6 +112,11 @@ void processUnblockedClients(void) { ...@@ -109,6 +112,11 @@ void processUnblockedClients(void) {
* client is not blocked before to proceed, but things may change and * client is not blocked before to proceed, but things may change and
* the code is conceptually more correct this way. */ * the code is conceptually more correct this way. */
if (!(c->flags & CLIENT_BLOCKED)) { if (!(c->flags & CLIENT_BLOCKED)) {
/* If we have a queued command, execute it now. */
if (processPendingCommandsAndResetClient(c) == C_ERR) {
continue;
}
/* Then process client if it has more data in it's buffer. */
if (c->querybuf && sdslen(c->querybuf) > 0) { if (c->querybuf && sdslen(c->querybuf) > 0) {
processInputBuffer(c); processInputBuffer(c);
} }
......
...@@ -1903,6 +1903,19 @@ int processCommandAndResetClient(client *c) { ...@@ -1903,6 +1903,19 @@ int processCommandAndResetClient(client *c) {
return deadclient ? C_ERR : C_OK; return deadclient ? C_ERR : C_OK;
} }
/* This function will execute any fully parsed commands pending on
* the client. Returns C_ERR if the client is no longer valid after executing
* the command, and C_OK for all other cases. */
int processPendingCommandsAndResetClient(client *c) {
if (c->flags & CLIENT_PENDING_COMMAND) {
c->flags &= ~CLIENT_PENDING_COMMAND;
if (processCommandAndResetClient(c) == C_ERR) {
return C_ERR;
}
}
return C_OK;
}
/* This function is called every time, in the client structure 'c', there is /* This function is called every time, in the client structure 'c', there is
* more query buffer to process, because we read more data from the socket * more query buffer to process, because we read more data from the socket
* or because a client was blocked and later reactivated, so there could be * or because a client was blocked and later reactivated, so there could be
...@@ -1911,7 +1924,9 @@ void processInputBuffer(client *c) { ...@@ -1911,7 +1924,9 @@ void processInputBuffer(client *c) {
/* Keep processing while there is something in the input buffer */ /* Keep processing while there is something in the input buffer */
while(c->qb_pos < sdslen(c->querybuf)) { while(c->qb_pos < sdslen(c->querybuf)) {
/* Return if clients are paused. */ /* Return if clients are paused. */
if (!(c->flags & CLIENT_SLAVE) && clientsArePaused()) break; if (!(c->flags & CLIENT_SLAVE) &&
!(c->flags & CLIENT_PENDING_READ) &&
clientsArePaused()) break;
/* Immediately abort if the client is in the middle of something. */ /* Immediately abort if the client is in the middle of something. */
if (c->flags & CLIENT_BLOCKED) break; if (c->flags & CLIENT_BLOCKED) break;
...@@ -3269,6 +3284,7 @@ int handleClientsWithPendingWritesUsingThreads(void) { ...@@ -3269,6 +3284,7 @@ int handleClientsWithPendingWritesUsingThreads(void) {
int postponeClientRead(client *c) { int postponeClientRead(client *c) {
if (server.io_threads_active && if (server.io_threads_active &&
server.io_threads_do_reads && server.io_threads_do_reads &&
!clientsArePaused() &&
!ProcessingEventsWhileBlocked && !ProcessingEventsWhileBlocked &&
!(c->flags & (CLIENT_MASTER|CLIENT_SLAVE|CLIENT_PENDING_READ))) !(c->flags & (CLIENT_MASTER|CLIENT_SLAVE|CLIENT_PENDING_READ)))
{ {
...@@ -3336,15 +3352,17 @@ int handleClientsWithPendingReadsUsingThreads(void) { ...@@ -3336,15 +3352,17 @@ int handleClientsWithPendingReadsUsingThreads(void) {
client *c = listNodeValue(ln); client *c = listNodeValue(ln);
c->flags &= ~CLIENT_PENDING_READ; c->flags &= ~CLIENT_PENDING_READ;
listDelNode(server.clients_pending_read,ln); listDelNode(server.clients_pending_read,ln);
/* Clients can become paused while executing the queued commands,
if (c->flags & CLIENT_PENDING_COMMAND) { * so we need to check in between each command. If a pause was
c->flags &= ~CLIENT_PENDING_COMMAND; * executed, we still remove the command and it will get picked up
if (processCommandAndResetClient(c) == C_ERR) { * later when clients are unpaused and we re-queue all clients. */
/* If the client is no longer valid, we avoid if (clientsArePaused()) continue;
* processing the client later. So we just go
* to the next. */ if (processPendingCommandsAndResetClient(c) == C_ERR) {
continue; /* If the client is no longer valid, we avoid
} * processing the client later. So we just go
* to the next. */
continue;
} }
processInputBuffer(c); processInputBuffer(c);
......
...@@ -2020,6 +2020,7 @@ size_t freeMemoryGetNotCountedMemory(); ...@@ -2020,6 +2020,7 @@ size_t freeMemoryGetNotCountedMemory();
int freeMemoryIfNeeded(void); int freeMemoryIfNeeded(void);
int freeMemoryIfNeededAndSafe(void); int freeMemoryIfNeededAndSafe(void);
int processCommand(client *c); int processCommand(client *c);
int processPendingCommandsAndResetClient(client *c);
void setupSignalHandlers(void); void setupSignalHandlers(void);
struct redisCommand *lookupCommand(sds name); struct redisCommand *lookupCommand(sds name);
struct redisCommand *lookupCommandByCString(const char *s); struct redisCommand *lookupCommandByCString(const char *s);
......
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