Files
redis/tests/integration/psync2-master-restart.tcl
7a58b432c7 Handle primary/replica clients in IO threads (#14335)
# Problem

While introducing Async IO
threads(https://github.com/redis/redis/pull/13695) primary and replica
clients were left to be handled inside main thread due to data race and
synchronization issues. This PR solves this issue with the additional
hope it increases performance of replication.

# Overview

## Moving the clients to IO threads

Since clients first participate in a handshake and an RDB replication
phases it was decided they are moved to IO-thread after RDB replication
is done. For primary client this was trivial as the master client is
created only after RDB sync (+ some additional checks one can see in
`isClientMustHandledByMainThread`). Replica clients though are moved to
IO threads immediately after connection (as are all clients) so
currently in `unstable` replication happens while this client is in
IO-thread. In this PR it was moved to main thread after receiving the
first `REPLCONF` message from the replica, but it is a bit hacky and we
can remove it. I didn't find issues between the two versions.

## Primary client (replica node)

We have few issues here:
- during `serverCron` a `replicationCron` is ran which periodically
sends `REPLCONF ACK` message to the master, also checks for timed-out
master. In order to prevent data races we utilize`IOThreadClientsCron`.
The client is periodically sent to main thread and during
`processClientsFromIOThread` it's checked if it needs to run the
replication cron behaviour.

- data races with main thread - specifically `lastinteraction` and
`read_reploff` members of the primary client that are written to in
`readQueryFromClient` could be accessed at the same time from main
thread during execution of `INFO REPLICATION`(`genRedisInfoString`). To
solve this the members were duplicated so if the client is in IO-thread
it writes to the duplicates and they are synced with the original
variables each time the client is send to main thread ( that means `INFO
REPLICATION` could potentially return stale values).

- During `freeClient` the primary client is fetched to main thread but
when caching it(`replicationCacheMaster`) the thread id will remain the
id of the IO thread it was from. This creates problems when resurrecting
the master client. Here the call to `unbindClientFromIOThreadEventLoop`
in `freeClient` was rewritten to call `keepClientInMainThread` which
automatically fixes the problem.

- During `exitScriptTimedoutMode` the master is queued for reprocessing
(specifically process any pending commands ASAP after it's unblocked).
We do that by putting it in the `server.unblocked_clients` list, which
are processed in the next `beforeSleep` cycle in main thread. Since this
will create a contention between main and IO thread, we just skip this
queueing in `unblocked_clients` and just queue the client to main thread
- the `processClientsFromIOThread` will process the pending commands
just as main would have.

## Replica clients (primary node)

We move the client after RDB replication is done and after replication
backlog is fed with its first message.
We do that so that the client's reference to the first replication
backlog node is initialized before it's read from IO-thread, hence no
contention with main thread on it.

### Shared replication buffer

Currently in unstable the replication buffer is shared amongst clients.
This is done via clients holding references to the nodes inside the
buffer. A node from the buffer can be trimmed once each replica client
has read it and send its contents. The reference is
`client->ref_repl_buf_node`. The replication buffer is written to by
main thread in `feedReplicationBuffer` and the refcounting is intrusive
- it's inside the replication-buffer nodes themselves.

Since the replica client changes the refcount (decreases the refcount of
the node it has just read, and increases the refcount of the next node
it starts to read) during `writeToClient` we have a data race with main
thread when it feeds the replication buffer. Moreover, main thread also
updates the `used` size of the node - how much it has written to it,
compared to its capacity which the replica client relies on to know how
much to read. Obviously replica being in IO-thread creates another data
race here. To mitigate these issues a few new variables were added to
the client's struct:

- `io_curr_repl_node` - starting node this replica is reading from
inside IO-thread
- `io_bound_repl_node` - the last node in the replication buffer the
replica sees before being send to IO-thread.

These values are only allowed to be updated in main thread. The client
keeps track of how much it has read into the buffer via the old
`ref_repl_buf_node`. Generally while in IO-thread the replica client
will now keep refcount of the `io_curr_repl_node` until it's processed
all the nodes up to `io_bound_repl_node` - at that point its returned to
main thread which can safely update the refcounts.
The `io_bound_repl_node` reference is there so the replica knows when to
stop reading from the repl buffer - imagine that replica reads from the
last node of the replication buffer while main thread feeds data to it -
we will create a data race on the `used` value
(`_writeToClientSlave`(IO-thread) vs `feedReplicationBuffer`(main)).
That's why this value is updated just before the replica is being send
to IO thread.
*NOTE*, this means that when replicas are handled by IO threads they
will hold more than one node at a time (i.e `io_curr_repl_node` up to
`io_bound_repl_node`) meaning trimming will happen a bit less
frequently. Tests show no significant problems with that.
(tnx to @ShooterIT for the `io_curr_repl_node` and `io_bound_repl_node`
mechanism as my initial implementation had similar semantics but was way
less clear)

Example of how this works:

* Replication buffer state at time N:
   | node 0| ... | node M, used_size K |
* replica caches `io_curr_repl_node`=0, `io_bound_repl_node`=M and
`io_bound_block_pos`=K
* replica moves to IO thread and processes all the data it sees
* Replication buffer state at time N + 1:
| node 0| ... | node M, used_size Full | |node M + 1| |node M + 2,
used_size L|, where Full > M
* replica moves to main thread at time N + 1, at this point following
happens
   - refcount to node 0 (io_curr_repl_node) is decreased
- `ref_repl_buf_node` becomes node M(io_bound_repl_node) (we still have
size-K bytes to process from there)
- refcount to node M is increased (now all nodes from 0 up to M-1
including can be trimmed unless some other replica holds reference to
them)
- And just before the replica is send back to IO thread the following
are updated:
   - `io_bound_repl_node` ref becomes node M+2
   - `io_bound_block_pos` becomes L

Note that replica client is only moved to main if it has processed all
the data it knows about (i.e up to `io_bound_repl_node` +
`io_bound_block_pos`)

### Replica clients kept in main as much as possible

During implementation an issue arose - how fast is the replica client
able to get knowledge about new data from the replication buffer and how
fast can it trim it. In order for that to happen ASAP whenever a replica
is moved to main it remains there until the replication buffer is fed
new data. At that point its put in the pending write queue and special
cased in handleClientsWithPendingWrites so that its send to IO thread
ASAP to write the new data to replica. Also since each time the replica
writes its whole repl data it knows about that means after it's send to
main thread `processClientsFromIOThread` is able to immediately update
the refcounts and trim whatever it can.

### ACK messages from primary

Slave clients need to periodically read `REPLCONF ACK` messages from
client. Since replica can remain in main thread indefinitely if no DB
change occurs, a new atomic `pending_read` was added during
`readQueryFromClient`. If a replica client has a pending read it's
returned back to IO-thread in order to process the read even if there is
no pending repl data to write.

### Replicas during shutdown

During shutdown the main thread pauses write actions and periodically
checks if all replicas have reached the same replication offset as the
primary node. During `finishShutdown` that may or may not be the case.
Either way a client data may be read from the replicas and even we may
try to write any pending data to them inside `flushSlavesOutputBuffers`.
In order to prevent races all the replicas from IO threads are moved to
main via `fetchClientFromIOThread`. A cancel of the shutdown should be
ok, since the mechanism employed by `handleClientsWithPendingWrites`
should return the client back to IO thread when needed.

## Notes

While adding new tests timing issues with Tsan tests were found and
fixed.

Also there is a data race issue caught by Tsan on the `last_error`
member of the `client` struct. It happens when both IO-thread and main
thread make a syscall using a `client` instance - this can happen only
for primary and replica clients since their data can be accessed by
commands send from other clients. Specific example is the `INFO
REPLICATION` command.
Although other such races were fixed, as described above, this once is
insignificant and it was decided to be ignored in `tsan.sup`.

---------

Co-authored-by: Yuan Wang <[email protected]>
Co-authored-by: Yuan Wang <[email protected]>
2026-01-21 16:19:12 +02:00

234 lines
9.0 KiB
Tcl

start_server {tags {"psync2 external:skip"}} {
start_server {} {
start_server {} {
set master [srv 0 client]
set master_host [srv 0 host]
set master_port [srv 0 port]
set replica [srv -1 client]
set replica_host [srv -1 host]
set replica_port [srv -1 port]
set sub_replica [srv -2 client]
# Make sure the server saves an RDB on shutdown
$master config set save "3600 1"
# Because we will test partial resync later, we don't want a timeout to cause
# the master-replica disconnect, then the extra reconnections will break the
# sync_partial_ok stat test
$master config set repl-timeout 3600
$replica config set repl-timeout 3600
$sub_replica config set repl-timeout 3600
# Avoid PINGs
$master config set repl-ping-replica-period 3600
$master config rewrite
# Build replication chain
$replica replicaof $master_host $master_port
$sub_replica replicaof $replica_host $replica_port
wait_for_condition 50 100 {
[status $replica master_link_status] eq {up} &&
[status $sub_replica master_link_status] eq {up}
} else {
fail "Replication not started."
}
test "PSYNC2: Partial resync after Master restart using RDB aux fields when offset is 0" {
assert {[status $master master_repl_offset] == 0}
set replid [status $master master_replid]
$replica config resetstat
catch {
restart_server 0 true false true now
set master [srv 0 client]
}
wait_for_condition 50 1000 {
[status $replica master_link_status] eq {up} &&
[status $sub_replica master_link_status] eq {up}
} else {
fail "Replicas didn't sync after master restart"
}
# Make sure master restore replication info correctly
assert {[status $master master_replid] != $replid}
assert {[status $master master_repl_offset] == 0}
assert {[status $master master_replid2] eq $replid}
assert {[status $master second_repl_offset] == 1}
# Make sure master set replication backlog correctly
assert {[status $master repl_backlog_active] == 1}
assert {[status $master repl_backlog_first_byte_offset] == 1}
assert {[status $master repl_backlog_histlen] == 0}
# Partial resync after Master restart
assert {[status $master sync_partial_ok] == 1}
assert {[status $replica sync_partial_ok] == 1}
}
# Generate some data
createComplexDataset $master 1000
test "PSYNC2: Partial resync after Master restart using RDB aux fields with data" {
wait_for_condition 500 100 {
[status $master master_repl_offset] == [status $replica master_repl_offset] &&
[status $master master_repl_offset] == [status $sub_replica master_repl_offset]
} else {
fail "Replicas and master offsets were unable to match *exactly*."
}
set replid [status $master master_replid]
set offset [status $master master_repl_offset]
$replica config resetstat
catch {
# SHUTDOWN NOW ensures master doesn't send GETACK to replicas before
# shutting down which would affect the replication offset.
restart_server 0 true false true now
set master [srv 0 client]
}
wait_for_condition 50 1000 {
[status $replica master_link_status] eq {up} &&
[status $sub_replica master_link_status] eq {up}
} else {
fail "Replicas didn't sync after master restart"
}
# Make sure master restore replication info correctly
assert {[status $master master_replid] != $replid}
assert {[status $master master_repl_offset] == $offset}
assert {[status $master master_replid2] eq $replid}
assert {[status $master second_repl_offset] == [expr $offset+1]}
# Make sure master set replication backlog correctly
assert {[status $master repl_backlog_active] == 1}
assert {[status $master repl_backlog_first_byte_offset] == [expr $offset+1]}
assert {[status $master repl_backlog_histlen] == 0}
# Partial resync after Master restart
assert {[status $master sync_partial_ok] == 1}
assert {[status $replica sync_partial_ok] == 1}
}
test "PSYNC2: Partial resync after Master restart using RDB aux fields with expire" {
$master debug set-active-expire 0
for {set j 0} {$j < 1024} {incr j} {
$master select [expr $j%16]
$master set $j somevalue px 10
}
after 20
# Wait until master has received ACK from replica. If the master thinks
# that any replica is lagging when it shuts down, master would send
# GETACK to the replicas, affecting the replication offset.
set offset [status $master master_repl_offset]
wait_for_condition 500 100 {
[string match "*slave0:*,offset=$offset,*" [$master info replication]] &&
$offset == [status $replica master_repl_offset] &&
$offset == [status $sub_replica master_repl_offset]
} else {
show_cluster_status
fail "Replicas and master offsets were unable to match *exactly*."
}
set offset [status $master master_repl_offset]
$replica config resetstat
catch {
# Unlike the test above, here we use SIGTERM, which behaves
# differently compared to SHUTDOWN NOW if there are lagging
# replicas. This is just to increase coverage and let each test use
# a different shutdown approach. In this case there are no lagging
# replicas though.
restart_server 0 true false
set master [srv 0 client]
}
wait_for_condition 50 1000 {
[status $replica master_link_status] eq {up} &&
[status $sub_replica master_link_status] eq {up}
} else {
fail "Replicas didn't sync after master restart"
}
set expired_offset [status $master repl_backlog_histlen]
# Stale keys expired and master_repl_offset grows correctly
assert {[status $master rdb_last_load_keys_expired] == 1024}
assert {[status $master master_repl_offset] == [expr $offset+$expired_offset]}
# Partial resync after Master restart
assert {[status $master sync_partial_ok] == 1}
assert {[status $replica sync_partial_ok] == 1}
set digest [$master debug digest]
wait_for_condition 10 100 {
$digest eq [$replica debug digest] &&
$digest eq [$sub_replica debug digest]
} else {
fail "Replica and sub-replica didn't sync after master restart in time..."
}
}
test "PSYNC2: Full resync after Master restart when too many key expired" {
$master config set repl-backlog-size 16384
$master config rewrite
$master debug set-active-expire 0
# Make sure replication backlog is full and will be trimmed.
for {set j 0} {$j < 2048} {incr j} {
$master select [expr $j%16]
$master set $j somevalue px 10
}
##### hash-field-expiration
# Hashes of type OBJ_ENCODING_LISTPACK_EX won't be discarded during
# RDB load, even if they are expired.
$master hset myhash1 f1 v1 f2 v2 f3 v3
$master hpexpire myhash1 10 FIELDS 3 f1 f2 f3
# Hashes of type RDB_TYPE_HASH_METADATA will be discarded during RDB load.
$master config set hash-max-listpack-entries 0
$master hset myhash2 f1 v1 f2 v2
$master hpexpire myhash2 10 FIELDS 2 f1 f2
$master config set hash-max-listpack-entries 1
after 20
wait_for_condition 500 100 {
[status $master master_repl_offset] == [status $replica master_repl_offset] &&
[status $master master_repl_offset] == [status $sub_replica master_repl_offset]
} else {
fail "Replicas and master offsets were unable to match *exactly*."
}
$replica config resetstat
catch {
# Unlike the test above, here we use SIGTERM. This is just to
# increase coverage and let each test use a different shutdown
# approach.
restart_server 0 true false
set master [srv 0 client]
}
wait_for_condition 50 1000 {
[status $replica master_link_status] eq {up} &&
[status $sub_replica master_link_status] eq {up}
} else {
fail "Replicas didn't sync after master restart"
}
# Replication backlog is full
assert {[status $master repl_backlog_first_byte_offset] > [status $master second_repl_offset]}
assert {[status $master sync_partial_ok] == 0}
assert {[status $master sync_full] == 1}
assert {[status $master rdb_last_load_keys_expired] == 2048}
assert {[status $replica sync_full] == 1}
set digest [$master debug digest]
assert {$digest eq [$replica debug digest]}
assert {$digest eq [$sub_replica debug digest]}
}
}}}