gvsafronov/futriix
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

fix pipelined WAIT performance issue.

Browse Source 
If client gets blocked again in `processUnblockedClients`, redis will not send
`REPLCONF GETACK *` to slaves untill next eventloop, so the client will be
blocked for 100ms by default(10hz) if no other file event fired.

move server.get_ack_from_slaves sinppet after `processUnblockedClients`, so
that both the first WAIT command that puts client in blocked context and the
following WAIT command processed in processUnblockedClients would trigger
redis-sever to send `REPLCONF GETACK *`, so that the eventloop would get
`REPLCONG ACK <reploffset>` from slaves and unblocked ASAP.
This commit is contained in:
srzhao 2019-11-26 10:43:57 +08:00 committed by antirez
parent f95a88d988
commit 4024bc7eee
1 changed files with 13 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

26
src/server.c
View File

@ -2110,6 +2110,19 @@ void beforeSleep(struct aeEventLoop *eventLoop) {
if (server.active_expire_enabled && server.masterhost == NULL) if (server.active_expire_enabled && server.masterhost == NULL)
activeExpireCycle(ACTIVE_EXPIRE_CYCLE_FAST); activeExpireCycle(ACTIVE_EXPIRE_CYCLE_FAST);
/* Unblock all the clients blocked for synchronous replication
* in WAIT. */
if (listLength(server.clients_waiting_acks))
processClientsWaitingReplicas();
/* Check if there are clients unblocked by modules that implement
* blocking commands. */
if (moduleCount()) moduleHandleBlockedClients();
/* Try to process pending commands for clients that were just unblocked. */
if (listLength(server.unblocked_clients))
processUnblockedClients();
/* Send all the slaves an ACK request if at least one client blocked /* Send all the slaves an ACK request if at least one client blocked
* during the previous event loop iteration. */ * during the previous event loop iteration. */
if (server.get_ack_from_slaves) { if (server.get_ack_from_slaves) {
@ -2125,19 +2138,6 @@ void beforeSleep(struct aeEventLoop *eventLoop) {
server.get_ack_from_slaves = 0; server.get_ack_from_slaves = 0;
} }
/* Unblock all the clients blocked for synchronous replication
* in WAIT. */
if (listLength(server.clients_waiting_acks))
processClientsWaitingReplicas();
/* Check if there are clients unblocked by modules that implement
* blocking commands. */
if (moduleCount()) moduleHandleBlockedClients();
/* Try to process pending commands for clients that were just unblocked. */
if (listLength(server.unblocked_clients))
processUnblockedClients();
/* Send the invalidation messages to clients participating to the /* Send the invalidation messages to clients participating to the
* client side caching protocol in broadcasting (BCAST) mode. */ * client side caching protocol in broadcasting (BCAST) mode. */
trackingBroadcastInvalidationMessages(); trackingBroadcastInvalidationMessages();