mirrors/valkey
1
0
Fork 0
mirror of http://github.com/valkey-io/valkey synced 2024-11-22 09:17:20 +00:00
Code Issues Packages Projects Releases Wiki Activity

Kill diskless fork child asap when the last replica drop (#1227)

Browse Source 
We originally checked the replica connection to whether to kill the
diskless child only when rdbPipeReadHandler is triggered. Actually
we can check it when the replica is disconnected, so that we don't
have to wait for rdbPipeReadHandler to be triggered and can kill
the forkless child as soon as possible.

In this way, when the child or rdbPipeReadHandler is stuck for some
reason, we can kill the child faster and release the fork resources.

Signed-off-by: Binbin <binloveplay1314@qq.com>
This commit is contained in:
Binbin 2024-11-15 16:34:32 +08:00 committed by GitHub
parent d3f3b9cc3a
commit 4e2493e5c9
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 11 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
src/networking.c
View File

@ -1555,12 +1555,17 @@ void unlinkClient(client *c) {
* in which case it needs to be cleaned from that list */ * in which case it needs to be cleaned from that list */
if (c->flag.replica && c->repl_state == REPLICA_STATE_WAIT_BGSAVE_END && server.rdb_pipe_conns) { if (c->flag.replica && c->repl_state == REPLICA_STATE_WAIT_BGSAVE_END && server.rdb_pipe_conns) {
int i; int i;
int still_alive = 0;
for (i = 0; i < server.rdb_pipe_numconns; i++) { for (i = 0; i < server.rdb_pipe_numconns; i++) {
if (server.rdb_pipe_conns[i] == c->conn) { if (server.rdb_pipe_conns[i] == c->conn) {
rdbPipeWriteHandlerConnRemoved(c->conn); rdbPipeWriteHandlerConnRemoved(c->conn);
server.rdb_pipe_conns[i] = NULL; server.rdb_pipe_conns[i] = NULL;
break;
} }
if (server.rdb_pipe_conns[i]) still_alive++;
}
if (still_alive == 0) {
serverLog(LL_NOTICE, "Diskless rdb transfer, last replica dropped, killing fork child.");
killRDBChild();
} }
} }
/* Only use shutdown when the fork is active and we are the parent. */ /* Only use shutdown when the fork is active and we are the parent. */
@ -1781,6 +1786,7 @@ void freeClient(client *c) {
if (server.saveparamslen == 0 && c->repl_state == REPLICA_STATE_WAIT_BGSAVE_END && if (server.saveparamslen == 0 && c->repl_state == REPLICA_STATE_WAIT_BGSAVE_END &&
server.child_type == CHILD_TYPE_RDB && server.rdb_child_type == RDB_CHILD_TYPE_DISK && server.child_type == CHILD_TYPE_RDB && server.rdb_child_type == RDB_CHILD_TYPE_DISK &&
anyOtherReplicaWaitRdb(c) == 0) { anyOtherReplicaWaitRdb(c) == 0) {
serverLog(LL_NOTICE, "Background saving, persistence disabled, last replica dropped, killing fork child.");
killRDBChild(); killRDBChild();
} }
if (c->repl_state == REPLICA_STATE_SEND_BULK) { if (c->repl_state == REPLICA_STATE_SEND_BULK) {

12
src/replication.c
View File

@ -1669,7 +1669,9 @@ void rdbPipeReadHandler(struct aeEventLoop *eventLoop, int fd, void *clientData,
if (!conn) continue; if (!conn) continue;
stillUp++; stillUp++;
} }
serverLog(LL_NOTICE, "Diskless rdb transfer, done reading from pipe, %d replicas still up.", stillUp); if (stillUp) {
serverLog(LL_NOTICE, "Diskless rdb transfer, done reading from pipe, %d replicas still up.", stillUp);
}
/* Now that the replicas have finished reading, notify the child that it's safe to exit. /* Now that the replicas have finished reading, notify the child that it's safe to exit.
* When the server detects the child has exited, it can mark the replica as online, and * When the server detects the child has exited, it can mark the replica as online, and
* start streaming the replication buffers. */ * start streaming the replication buffers. */
@ -1678,7 +1680,6 @@ void rdbPipeReadHandler(struct aeEventLoop *eventLoop, int fd, void *clientData,
return; return;
} }
int stillAlive = 0;
for (i = 0; i < server.rdb_pipe_numconns; i++) { for (i = 0; i < server.rdb_pipe_numconns; i++) {
ssize_t nwritten; ssize_t nwritten;
connection *conn = server.rdb_pipe_conns[i]; connection *conn = server.rdb_pipe_conns[i];
@ -1708,15 +1709,10 @@ void rdbPipeReadHandler(struct aeEventLoop *eventLoop, int fd, void *clientData,
server.rdb_pipe_numconns_writing++; server.rdb_pipe_numconns_writing++;
connSetWriteHandler(conn, rdbPipeWriteHandler); connSetWriteHandler(conn, rdbPipeWriteHandler);
} }
stillAlive++;
} }
if (stillAlive == 0) {
serverLog(LL_WARNING, "Diskless rdb transfer, last replica dropped, killing fork child.");
killRDBChild();
}
/* Remove the pipe read handler if at least one write handler was set. */ /* Remove the pipe read handler if at least one write handler was set. */
if (server.rdb_pipe_numconns_writing || stillAlive == 0) { if (server.rdb_pipe_numconns_writing) {
aeDeleteFileEvent(server.el, server.rdb_pipe_read, AE_READABLE); aeDeleteFileEvent(server.el, server.rdb_pipe_read, AE_READABLE);
break; break;
} }