Add Redis source code for testing

1 files changed, 331 insertions, 0 deletions

diff --git a/examples/redis-unstable/tests/integration/replication-iothreads.tcl b/examples/redis-unstable/tests/integration/replication-iothreads.tcl
new file mode 100644
index 0000000..12d0845
--- /dev/null
+++ b/examples/redis-unstable/tests/integration/replication-iothreads.tcl
@@ -0,0 +1,331 @@
+#
+# Copyright (c) 2025-Present, Redis Ltd.
+# All rights reserved.
+#
+# Licensed under your choice of (a) the Redis Source Available License 2.0
+# (RSALv2); or (b) the Server Side Public License v1 (SSPLv1); or (c) the
+# GNU Affero General Public License v3 (AGPLv3).
+#
+
+# Tests for master and slave clients in IO threads
+
+# Helper function to get master client IO thread from INFO replication
+proc get_master_client_io_thread {r} {
+    return [status $r master_client_io_thread]
+}
+
+# Helper function to get slave client IO thread from INFO replication
+proc get_slave_client_io_thread {r slave_idx} {
+    set info [$r info replication]
+    set lines [split $info "\r\n"]
+
+    foreach line $lines {
+        if {[string match "slave${slave_idx}:*" $line]} {
+            # Parse the slave line to extract io-thread value
+            set parts [split $line ","]
+            foreach part $parts {
+                if {[string match "*io-thread=*" $part]} {
+                    set kv [split $part "="]
+                    assert_equal [llength $kv] 2
+                    return [lindex $kv 1]
+                }
+            }
+        }
+    }
+    return -1
+}
+
+start_server {overrides {io-threads 4 save ""} tags {"iothreads repl network external:skip"}} {
+start_server {overrides {io-threads 4 save ""}} {
+    set master [srv 0 client]
+    set master_host [srv 0 host]
+    set master_port [srv 0 port]
+    set slave [srv -1 client]
+
+    test {Setup slave} {
+        $slave slaveof $master_host $master_port
+        wait_for_condition 1000 100 {
+            [s -1 master_link_status] eq {up}
+        } else {
+            fail "Replication not started."
+        }
+    }
+
+    test {Master client moves to IO thread after sync complete} {
+        # Check master client thread assignment (master client is on slave side)
+        wait_for_condition 100 100 {
+            [get_master_client_io_thread $slave] > 0
+        } else {
+            fail "Master client was not assigned to IO thread"
+        }
+    }
+
+    test {Slave client assignment to IO threads} {
+        # Verify slave is connected and online
+        wait_replica_online $master 0
+
+        # Slave client is connected - force a write so that it's assigned to an
+        # IO thread.
+        assert_equal "OK" [$master set x x]
+
+        # Check slave client thread assignment
+        wait_for_condition 50 100 {
+            [get_slave_client_io_thread $master 0] > 0
+        } else {
+            fail "Slave client was not assigned to IO thread"
+        }
+    }
+
+    test {WAIT command works with master/slave in IO threads} {
+        # Test basic WAIT functionality
+        $master set wait_test_key1 value1
+        $master set wait_test_key2 value2
+        $master incr wait_counter
+
+        assert {[$master wait 1 2000] == 1}
+
+        # Verify data reached slave
+        wait_for_condition 10 100 {
+            [$slave get wait_test_key1] eq "value1" &&
+            [$slave get wait_test_key2] eq "value2" &&
+            [$slave get wait_counter] eq "1"
+        } else {
+            fail "commands not propagated to IO thread slave in time"
+        }
+    }
+
+    test {Replication data integrity with IO threads} {
+        # Generate significant replication traffic
+        for {set i 0} {$i < 100} {incr i} {
+            $master set bulk_key_$i [string repeat "data" 10]
+            $master lpush bulk_list element_$i
+            $master zadd bulk_zset $i member_$i
+            if {$i % 20 == 0} {
+                # Periodically verify WAIT works
+                assert {[$master wait 1 2000] == 1}
+            }
+        }
+
+        # Final verification
+        wait_for_condition 50 100 {
+            [$slave get bulk_key_99] eq [string repeat "data" 10] &&
+            [$slave llen bulk_list] eq 100 &&
+            [$slave zcard bulk_zset] eq 100
+        } else {
+            fail "Replication data integrity failed"
+        }
+    }
+
+    test {WAIT timeout behavior with slave in IO thread} {
+        set slave_pid [srv -1 pid]
+
+        # Pause slave to test timeout
+        pause_process $slave_pid
+
+        # Should timeout and return 0 acks
+        $master set timeout_test_key timeout_value
+        set start_time [clock milliseconds]
+        assert {[$master wait 1 2000] == 0}
+        set elapsed [expr {[clock milliseconds] - $start_time}]
+        assert_range $elapsed 2000 2500
+
+        # Resume and verify recovery
+        resume_process $slave_pid
+
+        assert {[$master wait 1 2000] == 1}
+
+        # Verify data reached slave after resume
+        wait_for_condition 10 100 {
+            [$slave get timeout_test_key] eq "timeout_value"
+        } else {
+            fail "commands not propagated to IO thread slave in time"
+        }
+    }
+
+    test {Network interruption recovery with IO threads} {
+        # Generate traffic before interruption
+        for {set i 0} {$i < 50} {incr i} {
+            $master set pre_interrupt_$i value_$i
+        }
+
+        # Simulate network interruption
+        pause_process $slave_pid
+
+        # Continue writing during interruption
+        for {set i 0} {$i < 50} {incr i} {
+            $master set during_interrupt_$i value_$i
+        }
+
+        # WAIT should timeout
+        assert {[$master wait 1 2000] == 0}
+
+        # Resume slave and verify recovery
+        resume_process $slave_pid
+
+        # Verify WAIT works again
+        assert {[$master wait 1 2000] == 1}
+
+        # Wait for reconnection and catch up
+        wait_for_condition 100 100 {
+            [$slave get during_interrupt_49] eq "value_49"
+        } else {
+            fail "Slave didn't catch up after network recovery"
+        }
+
+        $master set post_recovery_test recovery_value
+        wait_for_condition 10 100 {
+          [$slave get post_recovery_test] eq "recovery_value"
+        } else {
+          fail "Slave didn't receive 'set post_recovery_test' command"
+        }
+
+        # Check thread assignments after recovery
+        wait_for_condition 100 100 {
+            [get_master_client_io_thread $slave] > 0
+        } else {
+            fail "Slave client not assigned to IO thread after recovery"
+        }
+    }
+
+    test {Replication reconnection cycles with IO threads} {
+        # Test multiple disconnect/reconnect cycles
+        for {set cycle 0} {$cycle < 3} {incr cycle} {
+            # Generate traffic
+            for {set i 0} {$i < 20} {incr i} {
+                $master set cycle_${cycle}_key_$i value_$i
+            }
+
+            assert {[$master wait 1 2000] == 1}
+
+            # Record thread assignments during cycle
+            set master_thread [get_master_client_io_thread $slave]
+            set slave_thread [get_slave_client_io_thread $master 0]
+            puts "Cycle $cycle - Master thread: $master_thread, Slave thread: $slave_thread"
+
+            # Disconnect and reconnect (except last cycle)
+            if {$cycle < 2} {
+                $slave replicaof no one
+                after 100
+                $slave replicaof $master_host $master_port
+                wait_for_sync $slave
+            }
+        }
+
+        # Verify final state
+        wait_for_condition 10 100 {
+            [$slave get cycle_2_key_19] eq "value_19"
+        } else {
+            fail "last command not propagated to IO thread slave in time"
+        }
+    }
+
+    test {INFO replication shows correct thread information} {
+        # Test INFO replication output format
+        set info [$master info replication]
+
+        # Should show master role
+        assert_match "*role:master*" $info
+
+        # Should have slave thread information
+        assert_match "*slave0:*io-thread=*" $info
+
+        # Test we can parse the thread ID
+        set slave_thread [get_slave_client_io_thread $master 0]
+        assert_morethan $slave_thread 0
+
+        # Test master client thread info
+        set slave_info [$slave info replication]
+        assert_match "*role:slave*" $slave_info
+        assert_match "*master_client_io_thread:*" $slave_info
+
+        set master_thread [get_master_client_io_thread $slave]
+        assert_morethan $master_thread 0
+    }
+}
+}
+
+start_server {overrides {io-threads 4 save ""} tags {"iothreads repl network external:skip"}} {
+start_server {overrides {io-threads 4 save ""}} {
+start_server {overrides {io-threads 4 save ""}} {
+start_server {overrides {io-threads 4 save ""}} {
+    set master [srv 0 client]
+    set master_host [srv 0 host]
+    set master_port [srv 0 port]
+    set slave1 [srv -1 client]
+    set slave2 [srv -2 client]
+    set slave3 [srv -3 client]
+
+    test {Multiple slaves across IO threads} {
+        # Setup replication for all slaves
+        $slave1 replicaof $master_host $master_port
+        $slave2 replicaof $master_host $master_port
+        $slave3 replicaof $master_host $master_port
+
+        # Wait for all slaves to be online
+        wait_replica_online $master 0
+        wait_replica_online $master 1
+        wait_replica_online $master 2
+
+        set iterations 5
+        while {[incr iterations -1] >= 0} {
+            # Slave clients are connected - force a write so that they are assigned
+            # to IO threads.
+            assert_equal "OK" [$master set x x]
+
+            wait_for_condition 10 100 {
+                ([get_slave_client_io_thread $master 0] > 0) &&
+                ([get_slave_client_io_thread $master 1] > 0) &&
+                ([get_slave_client_io_thread $master 2] > 0)
+            } else {
+                continue
+            }
+
+            break
+        }
+        if {$iterations < 0} {
+            fail "Replicas failed to be assigned to IO threads in time"
+        }
+
+        # Test concurrent replication to all slaves
+        for {set i 0} {$i < 200} {incr i} {
+            $master set multi_key_$i value_$i
+            if {$i % 50 == 0} {
+                assert {[$master wait 3 2000] == 3}
+            }
+        }
+
+        # Final verification all slaves got data
+        wait_for_condition 50 100 {
+            [$slave1 get multi_key_199] eq "value_199" &&
+            [$slave2 get multi_key_199] eq "value_199" &&
+            [$slave3 get multi_key_199] eq "value_199"
+        } else {
+            fail "Multi-slave replication failed"
+        }
+    }
+
+    test {WAIT with multiple slaves in IO threads} {
+        # Test various WAIT scenarios
+        $master set wait_multi_test1 value1
+        assert {[$master wait 3 2000] == 3}
+
+        $master set wait_multi_test2 value2
+        assert {[$master wait 2 2000] >= 2}
+
+        $master set wait_multi_test3 value3
+        assert {[$master wait 1 2000] >= 1}
+
+        # Verify all slaves have the data
+        wait_for_condition 10 100 {
+            [$slave1 get wait_multi_test3] eq "value3" &&
+            [$slave2 get wait_multi_test3] eq "value3" &&
+            [$slave3 get wait_multi_test3] eq "value3"
+        } else {
+            fail "commands not propagated to io thread slaves in time"
+        }
+    }
+}
+}
+}
+}
+