Introduce Multiple I/O Threads for Coordinator

[tonyd3]

Multiple I/O uvloop threads for Coordinator

This pull request introduces a significant architectural change to the coordinator component by implementing a multi-threaded I/O model. Previously, the coordinator likely relied on a single I/O thread for all cluster communication. The core change involves abstracting the I/O runtime into IORuntimeCtx objects, each managing its own libuv event loop, request queue, and connection manager. The main MRCluster now maintains a pool of these IORuntimeCtx instances, distributing requests among them using a round-robin mechanism.

The primary outcome of this change is enhanced scalability and performance for cluster operations by enabling parallel I/O. It aims to improve responsiveness and throughput, especially in high-concurrency scenarios. The refactoring also leads to better isolation of I/O operations and a more robust framework for handling dynamic topology updates and connection management across multiple threads.

Key Changes

• Introduced IORuntimeCtx structure to encapsulate libuv event loop, request queue (MRWorkQueue), and connection manager (MRConnManager) for each I/O thread.
• The MRCluster now manages a pool of IORuntimeCtx instances, distributing incoming requests (e.g., MR_Fanout, MR_MapSingle) across them using a round-robin strategy.
• Refactored core cluster communication functions (MRCluster_SendCommand, MRCluster_FanoutCommand, MRCluster_GetConn) to operate on a specific IORuntimeCtx instance, rather than a single global connection manager.
• Implemented thread-safe mechanisms for IORuntimeCtx creation, starting, and freeing, including synchronization primitives (uv_mutex_t, uv_cond_t) to manage thread lifecycle.
• Modified topology update logic (MR_UpdateTopology) to propagate new cluster topologies to all IORuntimeCtx instances, with each instance maintaining its own copy of the topology.
• Updated connection pool resizing (MR_UpdateConnPoolSize) to dynamically adjust connections per IORuntimeCtx based on the new SEARCH_IO_THREADS configuration.
• Added a new configuration option SEARCH_IO_THREADS to control the number of I/O threads used by the coordinator.
• Enhanced shutdown procedures for IORuntimeCtx to ensure proper cleanup of libuv handles and graceful thread joining.
• Refactored RQ (Request Queue) to be an integral part of IORuntimeCtx, handling request pushing and popping in a thread-safe manner.
• Introduced new C++ unit tests (test_cpp_io_runtime_ctx`.cpp`, test_cpp_cluster_io_threads.cpp) to validate the behavior and thread safety of the new I/O runtime context and multi-threaded cluster operations.
• Updated Python integration tests (test_multithread`.py`, `test.py`, test_config.py) to account for the new multi-threaded I/O model and SEARCH_IO_THREADS configuration.

Affected Areas

• src/coord/rmr/io_runtime_ctx.c
• src/coord/rmr/io_runtime_ctx.h
• src/coord/rmr/cluster.c
• src/coord/rmr/cluster.h
• src/coord/rmr/rmr.c
• src/coord/rmr/rmr.h
• src/coord/rmr/rq.c
• src/coord/rmr/rq.h
• src/coord/rmr/conn.c
• src/coord/rmr/conn.h
• src/coord/rmr/cluster_topology.c
• src/coord/rmr/cluster_topology.h
• src/coord/config.c
• src/coord/config.h
• src/module.c
• src/coord/debug_commands.c
• tests/pytests/test_multithread.py
• tests/pytests/test.py
• tests/pytests/test_config.py
• tests/cpptests/coord_tests/test_cpp_io_runtime_ctx.cpp
• tests/cpptests/coord_tests/test_cpp_cluster_io_threads.cpp
• tests/cpptests/coord_tests/CMakeLists.txt
• tests/ctests/coord_tests/test_cluster.c
• tests/ctests/coord_tests/CMakeLists.txt
• tests/cpptests/CMakeLists.txt
• sbin/unit-tests

---

This summary was automatically generated by @propel-code-bot

[propel-code-bot]

[BestPractice]

The new function run_cpp_coord_tests is almost identical to the existing run_cpp_tests function. This duplication can be avoided by creating a generic function that takes parameters like the test binary name, directories, and a description. This would make the script more maintainable and easier to extend with new test suites in the future.

For example, you could create a generic function like this:

run_gtest_suite() {
    local test_binary_name=$1
    local test_binary_dir_rel=$2
    local test_source_dir_rel=$3
    local description=$4
    # ... implementation ...
}

And then call it for both test suites:

run_cpp_tests() {
    run_gtest_suite "rstest" "tests/cpptests" "tests/cpptests" "C++ unit tests"
}

run_cpp_coord_tests() {
    run_gtest_suite "rstest_coord" "tests/cpptests/coord_tests" "tests/cpptests/coord_tests" "C++ coordinator unit tests"
}

[propel-code-bot]

[CriticalError]

The pending counter in MRWorkQueue appears to be accessed from multiple threads without synchronization, which could lead to a race condition.

• RQ_Pop() increments pending and is called from the libuv event loop thread.
• RQ_Done() decrements pending and can be called from the main Redis thread (via unblockHandler) or a worker thread.

The decrement --q->pending is not atomic. This could lead to incorrect pending counts, potentially causing the queue to process requests incorrectly under load. The comment on line 77 also appears to be incorrect based on the call paths.

I see from the commit history that you've worked on this area. Could you clarify if my understanding of the threading model is incorrect, or if an atomic operation is needed here?

Suggested change

	// To be called from the event loop thread, after the request is done, no need to protect the pending
	void RQ_Done(MRWorkQueue *q) {
	uv_mutex_lock(&q->lock);
	--q->pending;
	uv_mutex_unlock(&q->lock);
	}
	// This can be called from multiple threads, must be atomic.
	void RQ_Done(MRWorkQueue *q) {
	__atomic_fetch_sub(&q->pending, 1, __ATOMIC_RELAXED);
	}

⚡ Committable suggestion

Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation.