Commit 8bf660af authored by antirez's avatar antirez
Browse files

Some rework of #7234.

parent 9da134cd
...@@ -370,6 +370,7 @@ static int processTimeEvents(aeEventLoop *eventLoop) { ...@@ -370,6 +370,7 @@ static int processTimeEvents(aeEventLoop *eventLoop) {
* if flags has AE_DONT_WAIT set the function returns ASAP until all * if flags has AE_DONT_WAIT set the function returns ASAP until all
* the events that's possible to process without to wait are processed. * the events that's possible to process without to wait are processed.
* if flags has AE_CALL_AFTER_SLEEP set, the aftersleep callback is called. * if flags has AE_CALL_AFTER_SLEEP set, the aftersleep callback is called.
* if flags has AE_CALL_BEFORE_SLEEP set, the beforesleep callback is called.
* *
* The function returns the number of events processed. */ * The function returns the number of events processed. */
int aeProcessEvents(aeEventLoop *eventLoop, int flags) int aeProcessEvents(aeEventLoop *eventLoop, int flags)
...@@ -428,7 +429,7 @@ int aeProcessEvents(aeEventLoop *eventLoop, int flags) ...@@ -428,7 +429,7 @@ int aeProcessEvents(aeEventLoop *eventLoop, int flags)
tvp = &tv; tvp = &tv;
} }
if (eventLoop->beforesleep != NULL) if (eventLoop->beforesleep != NULL && flags & AE_CALL_BEFORE_SLEEP)
eventLoop->beforesleep(eventLoop); eventLoop->beforesleep(eventLoop);
/* Call the multiplexing API, will return only on timeout or when /* Call the multiplexing API, will return only on timeout or when
...@@ -525,7 +526,9 @@ int aeWait(int fd, int mask, long long milliseconds) { ...@@ -525,7 +526,9 @@ int aeWait(int fd, int mask, long long milliseconds) {
void aeMain(aeEventLoop *eventLoop) { void aeMain(aeEventLoop *eventLoop) {
eventLoop->stop = 0; eventLoop->stop = 0;
while (!eventLoop->stop) { while (!eventLoop->stop) {
aeProcessEvents(eventLoop, AE_ALL_EVENTS|AE_CALL_AFTER_SLEEP); aeProcessEvents(eventLoop, AE_ALL_EVENTS|
AE_CALL_BEFORE_SLEEP|
AE_CALL_AFTER_SLEEP);
} }
} }
......
...@@ -47,11 +47,12 @@ ...@@ -47,11 +47,12 @@
things to disk before sending replies, and want things to disk before sending replies, and want
to do that in a group fashion. */ to do that in a group fashion. */
#define AE_FILE_EVENTS 1 #define AE_FILE_EVENTS (1<<0)
#define AE_TIME_EVENTS 2 #define AE_TIME_EVENTS (1<<1)
#define AE_ALL_EVENTS (AE_FILE_EVENTS|AE_TIME_EVENTS) #define AE_ALL_EVENTS (AE_FILE_EVENTS|AE_TIME_EVENTS)
#define AE_DONT_WAIT 4 #define AE_DONT_WAIT (1<<2)
#define AE_CALL_AFTER_SLEEP 8 #define AE_CALL_BEFORE_SLEEP (1<<3)
#define AE_CALL_AFTER_SLEEP (1<<4)
#define AE_NOMORE -1 #define AE_NOMORE -1
#define AE_DELETED_EVENT_ID -1 #define AE_DELETED_EVENT_ID -1
......
...@@ -2863,8 +2863,9 @@ int processEventsWhileBlocked(void) { ...@@ -2863,8 +2863,9 @@ int processEventsWhileBlocked(void) {
ProcessingEventsWhileBlocked = 1; ProcessingEventsWhileBlocked = 1;
while (iterations--) { while (iterations--) {
int events = 0; int events = 0;
events += aeProcessEvents(server.el, AE_FILE_EVENTS|AE_DONT_WAIT); events += aeProcessEvents(server.el,
events += handleClientsWithPendingWrites(); AE_FILE_EVENTS|AE_DONT_WAIT|
AE_CALL_BEFORE_SLEEP|AE_CALL_AFTER_SLEEP);
if (!events) break; if (!events) break;
count += events; count += events;
} }
......
...@@ -2092,22 +2092,33 @@ extern int ProcessingEventsWhileBlocked; ...@@ -2092,22 +2092,33 @@ extern int ProcessingEventsWhileBlocked;
/* This function gets called every time Redis is entering the /* This function gets called every time Redis is entering the
* main loop of the event driven library, that is, before to sleep * main loop of the event driven library, that is, before to sleep
* for ready file descriptors. * for ready file descriptors.
*
* Note: This function is (currently) called from two functions: * Note: This function is (currently) called from two functions:
* 1. aeMain - The main server loop * 1. aeMain - The main server loop
* 2. processEventsWhileBlocked - Process clients during RDB/AOF load * 2. processEventsWhileBlocked - Process clients during RDB/AOF load
*
* If it was called from processEventsWhileBlocked we don't want * If it was called from processEventsWhileBlocked we don't want
* to perform all actions (For example, we don't want to expire * to perform all actions (For example, we don't want to expire
* keys), but we do need to perform some actions. * keys), but we do need to perform some actions.
*
* The most important is freeClientsInAsyncFreeQueue but we also * The most important is freeClientsInAsyncFreeQueue but we also
* call some other low-risk functions. */ * call some other low-risk functions. */
void beforeSleep(struct aeEventLoop *eventLoop) { void beforeSleep(struct aeEventLoop *eventLoop) {
UNUSED(eventLoop); UNUSED(eventLoop);
if (!ProcessingEventsWhileBlocked) { /* Just call a subset of vital functions in case we are re-entering
/* Handle precise timeouts of blocked clients. */ * the event loop from processEventsWhileBlocked(). */
handleBlockedClientsTimeout(); if (ProcessingEventsWhileBlocked) {
handleClientsWithPendingReadsUsingThreads();
tlsProcessPendingData();
handleClientsWithPendingWrites();
freeClientsInAsyncFreeQueue();
return;
} }
/* Handle precise timeouts of blocked clients. */
handleBlockedClientsTimeout();
/* We should handle pending reads clients ASAP after event loop. */ /* We should handle pending reads clients ASAP after event loop. */
handleClientsWithPendingReadsUsingThreads(); handleClientsWithPendingReadsUsingThreads();
...@@ -2117,56 +2128,54 @@ void beforeSleep(struct aeEventLoop *eventLoop) { ...@@ -2117,56 +2128,54 @@ void beforeSleep(struct aeEventLoop *eventLoop) {
/* If tls still has pending unread data don't sleep at all. */ /* If tls still has pending unread data don't sleep at all. */
aeSetDontWait(server.el, tlsHasPendingData()); aeSetDontWait(server.el, tlsHasPendingData());
if (!ProcessingEventsWhileBlocked) { /* Call the Redis Cluster before sleep function. Note that this function
/* Call the Redis Cluster before sleep function. Note that this function * may change the state of Redis Cluster (from ok to fail or vice versa),
* may change the state of Redis Cluster (from ok to fail or vice versa), * so it's a good idea to call it before serving the unblocked clients
* so it's a good idea to call it before serving the unblocked clients * later in this function. */
* later in this function. */ if (server.cluster_enabled) clusterBeforeSleep();
if (server.cluster_enabled) clusterBeforeSleep();
/* Run a fast expire cycle (the called function will return
/* Run a fast expire cycle (the called function will return * ASAP if a fast cycle is not needed). */
* ASAP if a fast cycle is not needed). */ 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
/* Unblock all the clients blocked for synchronous replication * in WAIT. */
* in WAIT. */ if (listLength(server.clients_waiting_acks))
if (listLength(server.clients_waiting_acks)) processClientsWaitingReplicas();
processClientsWaitingReplicas();
/* Check if there are clients unblocked by modules that implement
/* Check if there are clients unblocked by modules that implement * blocking commands. */
* blocking commands. */ if (moduleCount()) moduleHandleBlockedClients();
if (moduleCount()) moduleHandleBlockedClients();
/* Try to process pending commands for clients that were just unblocked. */
/* Try to process pending commands for clients that were just unblocked. */ if (listLength(server.unblocked_clients))
if (listLength(server.unblocked_clients)) processUnblockedClients();
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. Note that we do this after
* during the previous event loop iteration. Note that we do this after * processUnblockedClients(), so if there are multiple pipelined WAITs
* processUnblockedClients(), so if there are multiple pipelined WAITs * and the just unblocked WAIT gets blocked again, we don't have to wait
* and the just unblocked WAIT gets blocked again, we don't have to wait * a server cron cycle in absence of other event loop events. See #6623. */
* a server cron cycle in absence of other event loop events. See #6623. */ if (server.get_ack_from_slaves) {
if (server.get_ack_from_slaves) { robj *argv[3];
robj *argv[3];
argv[0] = createStringObject("REPLCONF",8);
argv[0] = createStringObject("REPLCONF",8); argv[1] = createStringObject("GETACK",6);
argv[1] = createStringObject("GETACK",6); argv[2] = createStringObject("*",1); /* Not used argument. */
argv[2] = createStringObject("*",1); /* Not used argument. */ replicationFeedSlaves(server.slaves, server.slaveseldb, argv, 3);
replicationFeedSlaves(server.slaves, server.slaveseldb, argv, 3); decrRefCount(argv[0]);
decrRefCount(argv[0]); decrRefCount(argv[1]);
decrRefCount(argv[1]); decrRefCount(argv[2]);
decrRefCount(argv[2]); server.get_ack_from_slaves = 0;
server.get_ack_from_slaves = 0; }
}
/* 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();
/* Write the AOF buffer on disk */ /* Write the AOF buffer on disk */
flushAppendOnlyFile(0); flushAppendOnlyFile(0);
}
/* Handle writes with pending output buffers. */ /* Handle writes with pending output buffers. */
handleClientsWithPendingWritesUsingThreads(); handleClientsWithPendingWritesUsingThreads();
...@@ -2174,12 +2183,10 @@ void beforeSleep(struct aeEventLoop *eventLoop) { ...@@ -2174,12 +2183,10 @@ void beforeSleep(struct aeEventLoop *eventLoop) {
/* Close clients that need to be closed asynchronous */ /* Close clients that need to be closed asynchronous */
freeClientsInAsyncFreeQueue(); freeClientsInAsyncFreeQueue();
if (!ProcessingEventsWhileBlocked) { /* Before we are going to sleep, let the threads access the dataset by
/* Before we are going to sleep, let the threads access the dataset by * releasing the GIL. Redis main thread will not touch anything at this
* releasing the GIL. Redis main thread will not touch anything at this * time. */
* time. */ if (moduleCount()) moduleReleaseGIL();
if (moduleCount()) moduleReleaseGIL();
}
} }
/* This function is called immadiately after the event loop multiplexing /* This function is called immadiately after the event loop multiplexing
......
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