diff --git a/sbin/unit-tests b/sbin/unit-tests index dc7cb524950..82cde41524c 100755 --- a/sbin/unit-tests +++ b/sbin/unit-tests @@ -209,6 +209,39 @@ run_coordinator_tests() { done } +#------------------------------------------------------------------------------ +# Run C++ coordinator unit tests +#------------------------------------------------------------------------------ +run_cpp_coord_tests() { + print_separator + echo "# Running C++ coordinator unit tests" + CPP_COORD_TESTS_DIR="$(cd $BINDIR/tests/cpptests/coord_tests; pwd)" + cd $ROOT_DIR/tests/cpptests/coord_tests + TEST_NAME=rstest_coord setup_sanitizer + + if [[ -z $TEST ]]; then + # Run all C++ coordinator tests + LOG_FILE="${LOGS_DIR}/rstest_coord.log" + echo "Running all C++ coordinator tests (log: $LOG_FILE)" + { $CPP_COORD_TESTS_DIR/rstest_coord > "$LOG_FILE" 2>&1; test_result=$?; (( EXIT_CODE |= $test_result )); } || true + + # Parse and display individual test results + parse_cpp_test_results "$LOG_FILE" + else + # Run single C++ coordinator test if requested + if [[ -f $CPP_COORD_TESTS_DIR/rstest_coord ]]; then + LOG_FILE="${LOGS_DIR}/rstest_coord_${TEST}.log" + echo "Running C++ coordinator test: $TEST (log: $LOG_FILE)" + { $CPP_COORD_TESTS_DIR/rstest_coord --gtest_filter=$TEST > "$LOG_FILE" 2>&1; test_result=$?; (( EXIT_CODE |= $test_result )); } || true + + # Parse and display results + parse_cpp_test_results "$LOG_FILE" + else + echo "C++ coordinator test binary not found: $CPP_COORD_TESTS_DIR/rstest_coord" + fi + fi +} + #------------------------------------------------------------------------------ # Run all unit tests #------------------------------------------------------------------------------ @@ -219,8 +252,11 @@ run_all_tests() { # Run C++ tests run_cpp_tests - # Run coordinator tests + # Run C coordinator tests run_coordinator_tests + + # Run C++ coordinator tests + run_cpp_coord_tests } #------------------------------------------------------------------------------ diff --git a/src/coord/config.c b/src/coord/config.c index 131a2c2063f..fef1c7b27fa 100644 --- a/src/coord/config.c +++ b/src/coord/config.c @@ -16,12 +16,13 @@ #include "hiredis/hiredis.h" #include "module.h" - #include #include extern RedisModuleCtx *RSDummyContext; +#define CEIL_DIV(a, b) ((a + b - 1) / b) +#define MAX_SEARCH_IO_THREADS (1 << 8) #define CONFIG_FROM_RSCONFIG(c) ((SearchClusterConfig *)(c)->chainedConfig) static SearchClusterConfig* getOrCreateRealConfig(RSConfig *config){ @@ -80,7 +81,10 @@ int triggerConnPerShard(RSConfig *config) { } else { connPerShard = config->numWorkerThreads + 1; } - MR_UpdateConnPerShard(connPerShard); + // The connPerShard will be applied to each of the ConnManager in each of the IO threads. + size_t conn_pool_size = CEIL_DIV(connPerShard, realConfig->coordinatorIOThreads); + + MR_UpdateConnPoolSize(conn_pool_size); return REDISMODULE_OK; } @@ -152,7 +156,7 @@ CONFIG_GETTER(getSearchThreads) { // search-threads int set_search_threads(const char *name, long long val, void *privdata, - RedisModuleString **err) { + RedisModuleString **err) { RSConfig *config = (RSConfig *)privdata; SearchClusterConfig *realConfig = getOrCreateRealConfig(config); realConfig->coordinatorPoolSize = (size_t)val; @@ -165,6 +169,36 @@ long long get_search_threads(const char *name, void *privdata) { return (long long)realConfig->coordinatorPoolSize; } +// SEARCH_IO_THREADS + +CONFIG_SETTER(setSearchIOThreads) { + SearchClusterConfig *realConfig = getOrCreateRealConfig((RSConfig *)config); + int acrc = AC_GetSize(ac, &realConfig->coordinatorIOThreads, AC_F_GE1); + CHECK_RETURN_PARSE_ERROR(acrc); + // Todo, the same as with the coord threads setting, this has no actual impact + return REDISMODULE_OK; +} + +CONFIG_GETTER(getSearchIOThreads) { + SearchClusterConfig *realConfig = getOrCreateRealConfig((RSConfig *)config); + return sdsfromlonglong(realConfig->coordinatorIOThreads); +} + +// search-io-threads +int set_search_io_threads(const char *name, long long val, void *privdata, + RedisModuleString **err) { + RSConfig *config = (RSConfig *)privdata; + SearchClusterConfig *realConfig = getOrCreateRealConfig(config); + realConfig->coordinatorIOThreads = (size_t)val; + return REDISMODULE_OK; +} + +long long get_search_io_threads(const char *name, void *privdata) { + RSConfig *config = (RSConfig *)privdata; + SearchClusterConfig *realConfig = getOrCreateRealConfig(config); + return (long long)realConfig->coordinatorIOThreads; +} + // TOPOLOGY_VALIDATION_TIMEOUT CONFIG_SETTER(setTopologyValidationTimeout) { SearchClusterConfig *realConfig = getOrCreateRealConfig((RSConfig *)config); @@ -223,6 +257,11 @@ static RSConfigOptions clusterOptions_g = { .setValue = setSearchThreads, .getValue = getSearchThreads, .flags = RSCONFIGVAR_F_IMMUTABLE,}, + {.name = "SEARCH_IO_THREADS", + .helpText = "Sets the number of I/O threads in the coordinator", + .setValue = setSearchIOThreads, + .getValue = getSearchIOThreads, + .flags = RSCONFIGVAR_F_IMMUTABLE}, {.name = "TOPOLOGY_VALIDATION_TIMEOUT", .helpText = "Sets the timeout for topology validation (in milliseconds). After this timeout, " "any pending requests will be processed, even if the topology is not fully connected. " @@ -280,6 +319,15 @@ int RegisterClusterModuleConfig(RedisModuleCtx *ctx) { ) ) + RM_TRY( + RedisModule_RegisterNumericConfig( + ctx, "search-io-threads", COORDINATOR_IO_THREADS_DEFAULT_SIZE, + REDISMODULE_CONFIG_IMMUTABLE | REDISMODULE_CONFIG_UNPREFIXED, 1, + MAX_SEARCH_IO_THREADS, get_search_io_threads, set_search_io_threads, NULL, + (void*)&RSGlobalConfig + ) + ) + RM_TRY( RedisModule_RegisterNumericConfig ( ctx, "search-topology-validation-timeout", DEFAULT_TOPOLOGY_VALIDATION_TIMEOUT, diff --git a/src/coord/config.h b/src/coord/config.h index aa34e1e0325..0fbe1f9f09d 100644 --- a/src/coord/config.h +++ b/src/coord/config.h @@ -23,6 +23,7 @@ typedef struct { size_t connPerShard; size_t cursorReplyThreshold; size_t coordinatorPoolSize; // number of threads in the coordinator thread pool + size_t coordinatorIOThreads; // number of I/O threads in the coordinator size_t topologyValidationTimeoutMS; } SearchClusterConfig; @@ -33,6 +34,7 @@ extern RedisModuleString *config_dummy_password; #define CLUSTER_TYPE_RLABS "redislabs" #define COORDINATOR_POOL_DEFAULT_SIZE 20 +#define COORDINATOR_IO_THREADS_DEFAULT_SIZE 1 #define DEFAULT_TOPOLOGY_VALIDATION_TIMEOUT 30000 #define DEFAULT_CURSOR_REPLY_THRESHOLD 1 #define DEFAULT_CONN_PER_SHARD 0 @@ -44,6 +46,7 @@ extern RedisModuleString *config_dummy_password; .timeoutMS = 0, \ .cursorReplyThreshold = DEFAULT_CURSOR_REPLY_THRESHOLD, \ .coordinatorPoolSize = COORDINATOR_POOL_DEFAULT_SIZE, \ + .coordinatorIOThreads = COORDINATOR_IO_THREADS_DEFAULT_SIZE, \ .topologyValidationTimeoutMS = DEFAULT_TOPOLOGY_VALIDATION_TIMEOUT, \ } diff --git a/src/coord/debug_commands.c b/src/coord/debug_commands.c index 96363462543..75cda1aa858 100644 --- a/src/coord/debug_commands.c +++ b/src/coord/debug_commands.c @@ -7,7 +7,7 @@ * GNU Affero General Public License v3 (AGPLv3). */ #include "coord/rmr/rmr.h" -#include "coord/rmr/rq.h" +#include "coord/rmr/io_runtime_ctx.h" #include "debug_commands.h" #include "debug_command_names.h" #include "coord/rmr/redis_cluster.h" @@ -53,7 +53,7 @@ DEBUG_COMMAND(clearTopology) { return RedisModule_ReplyWithError(ctx, NODEBUG_ERR); } if (argc != 2) return RedisModule_WrongArity(ctx); - RQ_Debug_ClearPendingTopo(); + MR_Debug_ClearPendingTopo(); return RedisModule_ReplyWithSimpleString(ctx, "OK"); } diff --git a/src/coord/rmr/cluster.c b/src/coord/rmr/cluster.c index f160bcaffd6..6b05163c68c 100644 --- a/src/coord/rmr/cluster.c +++ b/src/coord/rmr/cluster.c @@ -13,63 +13,28 @@ #include "rmalloc.h" #include +#include "rq.h" -void _MRCluster_UpdateNodes(MRCluster *cl) { - /* Get all the current node ids from the connection manager. We will remove all the nodes - * that are in the new topology, and after the update, delete all the nodes that are in this map - * and not in the new topology */ - dict *nodesToDisconnect = dictCreate(&dictTypeHeapStrings, NULL); - dictIterator *it = dictGetIterator(cl->mgr.map); - dictEntry *de; - while ((de = dictNext(it))) { - dictAdd(nodesToDisconnect, dictGetKey(de), NULL); - } - dictReleaseIterator(it); - - /* Walk the topology and add all nodes in it to the connection manager */ - for (int sh = 0; sh < cl->topo->numShards; sh++) { - for (int n = 0; n < cl->topo->shards[sh].numNodes; n++) { - MRClusterNode *node = &cl->topo->shards[sh].nodes[n]; - MRConnManager_Add(&cl->mgr, node->id, &node->endpoint, 0); - - /* This node is still valid, remove it from the nodes to delete list */ - dictDelete(nodesToDisconnect, node->id); - - /* See if this is us - if so we need to update the cluster's host and current id */ - // if (node->flags & MRNode_Self) { - // cl->myNode = node; - // cl->myShard = &cl->topo->shards[sh]; - // } - } - } - - // if we didn't remove the node from the original nodes map copy, it means it's not in the new topology, - // we need to disconnect the node's connections - it = dictGetIterator(nodesToDisconnect); - while ((de = dictNext(it))) { - MRConnManager_Disconnect(&cl->mgr, dictGetKey(de)); - } - dictReleaseIterator(it); - dictRelease(nodesToDisconnect); -} - -MRCluster *MR_NewCluster(MRClusterTopology *initialTopology, size_t conn_pool_size) { +/* Initialize the MapReduce engine with a node provider */ +MRCluster *MR_NewCluster(MRClusterTopology *initialTopology, size_t conn_pool_size, size_t num_io_threads) { MRCluster *cl = rm_new(MRCluster); - cl->topo = initialTopology; - MRConnManager_Init(&cl->mgr, conn_pool_size); - - if (cl->topo) { - _MRCluster_UpdateNodes(cl); + RS_ASSERT(num_io_threads > 0); + cl->num_io_threads = num_io_threads; + cl->current_round_robin = 0; // Initialize round-robin counter + cl->io_runtimes_pool = rm_malloc(cl->num_io_threads * sizeof(IORuntimeCtx*)); + for (size_t i = 0; i < cl->num_io_threads; i++) { + cl->io_runtimes_pool[i] = IORuntimeCtx_Create(conn_pool_size, initialTopology, i + 1, i == 0); } + return cl; } /* Find the shard responsible for a given slot */ -MRClusterShard *_MRCluster_FindShard(MRCluster *cl, unsigned slot) { +MRClusterShard *_MRCluster_FindShard(MRClusterTopology *topo, unsigned slot) { // TODO: Switch to binary search - for (int i = 0; i < cl->topo->numShards; i++) { - if (cl->topo->shards[i].startSlot <= slot && cl->topo->shards[i].endSlot >= slot) { - return &cl->topo->shards[i]; + for (int i = 0; i < topo->numShards; i++) { + if (topo->shards[i].startSlot <= slot && topo->shards[i].endSlot >= slot) { + return &topo->shards[i]; } } return NULL; @@ -135,7 +100,7 @@ static const char *MRGetShardKey(const MRCommand *cmd, size_t *len) { } -static mr_slot_t getSlotByCmd(const MRCommand *cmd, const MRCluster *cl) { +static mr_slot_t getSlotByCmd(const MRCommand *cmd, const MRClusterTopology *topo) { if(cmd->targetSlot >= 0){ return cmd->targetSlot; @@ -145,44 +110,49 @@ static mr_slot_t getSlotByCmd(const MRCommand *cmd, const MRCluster *cl) { const char *k = MRGetShardKey(cmd, &len); if (!k) return 0; // Default to crc16 - uint16_t crc = (cl->topo->hashFunc == MRHashFunc_CRC12) ? crc12(k, len) : crc16(k, len); - return crc % cl->topo->numSlots; + uint16_t crc = (topo->hashFunc == MRHashFunc_CRC12) ? crc12(k, len) : crc16(k, len); + return crc % topo->numSlots; } -MRConn* MRCluster_GetConn(MRCluster *cl, bool mastersOnly, MRCommand *cmd) { +MRConn* MRCluster_GetConn(IORuntimeCtx *ioRuntime, bool mastersOnly, MRCommand *cmd) { - if (!cl || !cl->topo) return NULL; + if (!ioRuntime->topo) return NULL; /* Get the cluster slot from the sharder */ - unsigned slot = getSlotByCmd(cmd, cl); + unsigned slot = getSlotByCmd(cmd, ioRuntime->topo); /* Get the shard from the slot map */ - MRClusterShard *sh = _MRCluster_FindShard(cl, slot); + MRClusterShard *sh = _MRCluster_FindShard(ioRuntime->topo, slot); if (!sh) return NULL; MRClusterNode *node = _MRClusterShard_SelectNode(sh, mastersOnly); if (!node) return NULL; - return MRConn_Get(&cl->mgr, node->id); + return MRConn_Get(&ioRuntime->conn_mgr, node->id); } /* Send a single command to the right shard in the cluster, with an optional control over node * selection */ -int MRCluster_SendCommand(MRCluster *cl, bool mastersOnly, MRCommand *cmd, - redisCallbackFn *fn, void *privdata) { - MRConn *conn = MRCluster_GetConn(cl, mastersOnly, cmd); +int MRCluster_SendCommand(IORuntimeCtx *ioRuntime, + bool mastersOnly, + MRCommand *cmd, + redisCallbackFn *fn, + void *privdata) { + MRConn *conn = MRCluster_GetConn(ioRuntime, mastersOnly, cmd); if (!conn) return REDIS_ERR; return MRConn_SendCommand(conn, cmd, fn, privdata); } -int MRCluster_CheckConnections(MRCluster *cl, bool mastersOnly) { - for (size_t i = 0; i < cl->topo->numShards; i++) { - MRClusterShard *sh = &cl->topo->shards[i]; +int MRCluster_CheckConnections(IORuntimeCtx *ioRuntime, + bool mastersOnly) { + struct MRClusterTopology *topo = ioRuntime->topo; + for (size_t i = 0; i < topo->numShards; i++) { + MRClusterShard *sh = &topo->shards[i]; for (size_t j = 0; j < sh->numNodes; j++) { if (mastersOnly && !(sh->nodes[j].flags & MRNode_Master)) { continue; } - if (!MRConn_Get(&cl->mgr, sh->nodes[j].id)) { + if (!MRConn_Get(&ioRuntime->conn_mgr, sh->nodes[j].id)) { return REDIS_ERR; } } @@ -192,16 +162,20 @@ int MRCluster_CheckConnections(MRCluster *cl, bool mastersOnly) { /* Multiplex a command to all coordinators, using a specific coordination strategy. Returns the * number of sent commands */ -int MRCluster_FanoutCommand(MRCluster *cl, bool mastersOnly, MRCommand *cmd, - redisCallbackFn *fn, void *privdata) { +int MRCluster_FanoutCommand(IORuntimeCtx *ioRuntime, + bool mastersOnly, + MRCommand *cmd, + redisCallbackFn *fn, + void *privdata) { + struct MRClusterTopology *topo = ioRuntime->topo; int ret = 0; - for (size_t i = 0; i < cl->topo->numShards; i++) { - MRClusterShard *sh = &cl->topo->shards[i]; + for (size_t i = 0; i < topo->numShards; i++) { + MRClusterShard *sh = &topo->shards[i]; for (size_t j = 0; j < sh->numNodes; j++) { if (mastersOnly && !(sh->nodes[j].flags & MRNode_Master)) { continue; } - MRConn *conn = MRConn_Get(&cl->mgr, sh->nodes[j].id); + MRConn *conn = MRConn_Get(&ioRuntime->conn_mgr, sh->nodes[j].id); if (conn) { if (MRConn_SendCommand(conn, cmd, fn, privdata) != REDIS_ERR) { ret++; @@ -212,100 +186,33 @@ int MRCluster_FanoutCommand(MRCluster *cl, bool mastersOnly, MRCommand *cmd, return ret; } -/* Initialize the connections to all shards */ -int MRCluster_ConnectAll(MRCluster *cl) { - return MRConnManager_ConnectAll(&cl->mgr); -} - -void MRClusterTopology_Free(MRClusterTopology *t) { - for (int s = 0; s < t->numShards; s++) { - for (int n = 0; n < t->shards[s].numNodes; n++) { - MRClusterNode_Free(&t->shards[s].nodes[n]); +void MRCluster_Free(MRCluster *cl) { + if (cl) { + // First, fire the shutdown event for all runtimes + if (cl->io_runtimes_pool) { + for (size_t i = 0; i < cl->num_io_threads; i++) { + IORuntimeCtx_FireShutdown(cl->io_runtimes_pool[i]); + } } - rm_free(t->shards[s].nodes); - } - rm_free(t->shards); - rm_free(t); -} - -void MRClusterNode_Free(MRClusterNode *n) { - MREndpoint_Free(&n->endpoint); - rm_free((char *)n->id); -} - -int MRCLuster_UpdateTopology(MRCluster *cl, MRClusterTopology *newTopo) { - - if (!newTopo) return REDIS_ERR; - // if the topology has updated, we update to the new one - if (newTopo->hashFunc == MRHashFunc_None && cl->topo) { - newTopo->hashFunc = cl->topo->hashFunc; - } - - MRClusterTopology *old = cl->topo; - cl->topo = newTopo; - _MRCluster_UpdateNodes(cl); - MRCluster_ConnectAll(cl); - if (old) { - MRClusterTopology_Free(old); - } - return REDIS_OK; -} - - -void MRCluster_UpdateConnPerShard(MRCluster *cl, size_t new_conn_pool_size) { - RS_ASSERT(new_conn_pool_size > 0); - size_t old_conn_pool_size = cl->mgr.nodeConns; - if (old_conn_pool_size > new_conn_pool_size) { - MRConnManager_Shrink(&cl->mgr, new_conn_pool_size); - } else if (old_conn_pool_size < new_conn_pool_size) { - MRConnManager_Expand(&cl->mgr, new_conn_pool_size); - } -} - -MRClusterShard MR_NewClusterShard(mr_slot_t startSlot, mr_slot_t endSlot, size_t capNodes) { - MRClusterShard ret = (MRClusterShard){ - .startSlot = startSlot, - .endSlot = endSlot, - .capNodes = capNodes, - .numNodes = 0, - .nodes = rm_calloc(capNodes, sizeof(MRClusterNode)), - }; - return ret; -} - -void MRClusterShard_AddNode(MRClusterShard *sh, MRClusterNode *n) { - if (sh->capNodes == sh->numNodes) { - sh->capNodes += 1; - sh->nodes = rm_realloc(sh->nodes, sh->capNodes * sizeof(MRClusterNode)); + // Then free the RuntimeCtx, it will join the threads + if (cl->io_runtimes_pool) { + for (size_t i = 0; i < cl->num_io_threads; i++) { + IORuntimeCtx_Free(cl->io_runtimes_pool[i]); + } + rm_free(cl->io_runtimes_pool); + } + rm_free(cl); } - sh->nodes[sh->numNodes++] = *n; } -MRClusterTopology *MR_NewTopology(size_t numShards, size_t numSlots, MRHashFunc hashFunc) { - MRClusterTopology *topo = rm_new(MRClusterTopology); - topo->numSlots = numSlots; - topo->hashFunc = hashFunc; - topo->numShards = 0; - topo->capShards = numShards; - topo->shards = rm_calloc(topo->capShards, sizeof(MRClusterShard)); - return topo; +size_t MRCluster_AssignRoundRobinIORuntimeIdx(MRCluster *cl) { + size_t idx = cl->current_round_robin; + cl->current_round_robin = (cl->current_round_robin + 1) % cl->num_io_threads; + return idx; } -void MRClusterTopology_AddShard(MRClusterTopology *topo, MRClusterShard *sh) { - if (topo->capShards == topo->numShards) { - topo->capShards++; - topo->shards = rm_realloc(topo->shards, topo->capShards * sizeof(MRClusterShard)); - } - topo->shards[topo->numShards++] = *sh; -} - -void MRClust_Free(MRCluster *cl) { - if (cl) { - if (cl->topo) - MRClusterTopology_Free(cl->topo); - if (cl->mgr.map) - MRConnManager_Free(&cl->mgr); - rm_free(cl); - } +IORuntimeCtx *MRCluster_GetIORuntimeCtx(const MRCluster *cl, size_t idx) { + RS_ASSERT(idx < cl->num_io_threads); + return cl->io_runtimes_pool[idx]; } diff --git a/src/coord/rmr/cluster.h b/src/coord/rmr/cluster.h index 8e0f65fb0ea..3526c941833 100644 --- a/src/coord/rmr/cluster.h +++ b/src/coord/rmr/cluster.h @@ -15,86 +15,50 @@ #include "endpoint.h" #include "command.h" #include "node.h" +#include "io_runtime_ctx.h" +#include "cluster_topology.h" -typedef uint16_t mr_slot_t; - -/* A "shard" represents a slot range of the cluster, with its associated nodes. For each sharding - * key, we select the slot based on the hash function, and then look for the shard in the cluster's - * shard array */ -typedef struct { - mr_slot_t startSlot; - mr_slot_t endSlot; - size_t numNodes; - size_t capNodes; - MRClusterNode *nodes; -} MRClusterShard; - -/* Create a new cluster shard to be added to a topology */ -MRClusterShard MR_NewClusterShard(mr_slot_t startSlot, mr_slot_t endSlots, size_t capNodes); -void MRClusterShard_AddNode(MRClusterShard *sh, MRClusterNode *n); - -#define MRHASHFUNC_CRC12_STR "CRC12" -#define MRHASHFUNC_CRC16_STR "CRC16" - -typedef enum { - MRHashFunc_None = 0, - MRHashFunc_CRC12, - MRHashFunc_CRC16, -} MRHashFunc; - -/* A topology is the mapping of slots to shards and nodes */ -typedef struct MRClusterTopology { - size_t numSlots; - MRHashFunc hashFunc; - size_t numShards; - size_t capShards; - MRClusterShard *shards; -} MRClusterTopology; - -MRClusterTopology *MR_NewTopology(size_t numShards, size_t numSlots, MRHashFunc hashFunc); -void MRClusterTopology_AddShard(MRClusterTopology *topo, MRClusterShard *sh); - -void MRClusterTopology_Free(MRClusterTopology *t); - -void MRClusterNode_Free(MRClusterNode *n); +#ifdef __cplusplus +extern "C" { +#endif /* A cluster has nodes and connections that can be used by the engine to send requests */ typedef struct { /* The connection manager holds a connection to each node, indexed by node id */ - MRConnManager mgr; - /* The latest topology of the cluster */ - MRClusterTopology *topo; + /* An MRCluster holds an array of Connection Managers (one per each I/O thread)*/ + IORuntimeCtx **io_runtimes_pool; + size_t num_io_threads; // Number of threads in the pool (including the control plane) + size_t current_round_robin; } MRCluster; -int MRCluster_CheckConnections(MRCluster *cl, bool mastersOnly); - /* Multiplex a non-sharding command to all coordinators, using a specific coordination strategy. The * return value is the number of nodes we managed to successfully send the command to */ -int MRCluster_FanoutCommand(MRCluster *cl, bool mastersOnly, MRCommand *cmd, redisCallbackFn *fn, +int MRCluster_FanoutCommand(IORuntimeCtx *ioRuntime, bool mastersOnly, MRCommand *cmd, redisCallbackFn *fn, void *privdata); /* Get a connected connection according to the cluster, strategy and command. * Returns NULL if no fitting connection exists at the moment */ -MRConn *MRCluster_GetConn(MRCluster *cl, bool mastersOnly, MRCommand *cmd); +MRConn *MRCluster_GetConn(IORuntimeCtx *ioRuntime, bool mastersOnly, MRCommand *cmd); /* Send a command to its appropriate shard, selecting a node based on the coordination strategy. * Returns REDIS_OK on success, REDIS_ERR on failure. Notice that that send is asynchronous so even * though we signal for success, the request may fail */ -int MRCluster_SendCommand(MRCluster *cl, bool mastersOnly, MRCommand *cmd, redisCallbackFn *fn, +int MRCluster_SendCommand(IORuntimeCtx *ioRuntime, bool mastersOnly, MRCommand *cmd, redisCallbackFn *fn, void *privdata); /* Asynchronously connect to all nodes in the cluster. This must be called before the io loop is * started */ -int MRCluster_ConnectAll(MRCluster *cl); +int MRCluster_ConnectAll(IORuntimeCtx *ioRuntime); /* Create a new cluster using a node provider */ -MRCluster *MR_NewCluster(MRClusterTopology *topology, size_t conn_pool_size); +MRCluster *MR_NewCluster(MRClusterTopology *topology, size_t conn_pool_size, size_t num_io_threads); + +void MRCluster_Free(MRCluster *cl); -/* Update the number of connections per shard */ -void MRCluster_UpdateConnPerShard(MRCluster *cl, size_t new_conn_pool_size); +size_t MRCluster_AssignRoundRobinIORuntimeIdx(MRCluster *cl); -/* Update the topology by calling the topology provider explicitly with ctx. If ctx is NULL, the - * provider's current context is used. Otherwise, we call its function with the given context */ -int MRCLuster_UpdateTopology(MRCluster *cl, MRClusterTopology *newTopology); +IORuntimeCtx *MRCluster_GetIORuntimeCtx(const MRCluster *cl, size_t idx); -void MRClust_Free(MRCluster *cl); +#ifdef __cplusplus +} +#endif diff --git a/src/coord/rmr/cluster_topology.c b/src/coord/rmr/cluster_topology.c new file mode 100644 index 00000000000..ea1ecb0775c --- /dev/null +++ b/src/coord/rmr/cluster_topology.c @@ -0,0 +1,89 @@ +/* + * Copyright (c) 2006-Present, Redis Ltd. + * All rights reserved. + * + * Licensed under your choice of 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). +*/ + +#include "cluster_topology.h" +#include "endpoint.h" +#include "rmalloc.h" + +void MRClusterShard_AddNode(MRClusterShard *sh, MRClusterNode *n) { + if (sh->capNodes == sh->numNodes) { + sh->capNodes += 1; + sh->nodes = rm_realloc(sh->nodes, sh->capNodes * sizeof(MRClusterNode)); + } + sh->nodes[sh->numNodes++] = *n; +} + +MRClusterShard MR_NewClusterShard(mr_slot_t startSlot, mr_slot_t endSlot, size_t capNodes) { + MRClusterShard ret = (MRClusterShard){ + .startSlot = startSlot, + .endSlot = endSlot, + .capNodes = capNodes, + .numNodes = 0, + .nodes = rm_calloc(capNodes, sizeof(MRClusterNode)), + }; + return ret; +} + +MRClusterTopology *MR_NewTopology(size_t numShards, size_t numSlots, MRHashFunc hashFunc) { + MRClusterTopology *topo = rm_new(MRClusterTopology); + topo->numSlots = numSlots; + topo->hashFunc = hashFunc; + topo->numShards = 0; + topo->capShards = numShards; + topo->shards = rm_calloc(topo->capShards, sizeof(MRClusterShard)); + return topo; +} + +void MRClusterTopology_AddShard(MRClusterTopology *topo, MRClusterShard *sh) { + if (topo->capShards == topo->numShards) { + topo->capShards++; + topo->shards = rm_realloc(topo->shards, topo->capShards * sizeof(MRClusterShard)); + } + topo->shards[topo->numShards++] = *sh; +} + +MRClusterTopology *MRClusterTopology_Clone(MRClusterTopology *t) { + if (!t) { + return NULL; + } + MRClusterTopology *topo = MR_NewTopology(t->numShards, t->numSlots, t->hashFunc); + for (int s = 0; s < t->numShards; s++) { + MRClusterShard *original_shard = &t->shards[s]; + MRClusterShard new_shard = MR_NewClusterShard(original_shard->startSlot, original_shard->endSlot, original_shard->numNodes); + for (int n = 0; n < original_shard->numNodes; n++) { + MRClusterNode *node = &original_shard->nodes[n]; + MRClusterShard_AddNode(&new_shard, node); + } + for (int n = 0; n < new_shard.numNodes; n++) { + new_shard.nodes[n].id = rm_strdup(original_shard->nodes[n].id); + MREndpoint_Copy(&new_shard.nodes[n].endpoint, &original_shard->nodes[n].endpoint); + new_shard.nodes[n].endpoint.port = original_shard->nodes[n].endpoint.port; + new_shard.nodes[n].flags = 0; + new_shard.nodes[n].flags = original_shard->nodes[n].flags; + } + MRClusterTopology_AddShard(topo, &new_shard); + } + return topo; +} + +void MRClusterNode_Free(MRClusterNode *n) { + MREndpoint_Free(&n->endpoint); + rm_free((char *)n->id); +} + +void MRClusterTopology_Free(MRClusterTopology *t) { + for (int s = 0; s < t->numShards; s++) { + for (int n = 0; n < t->shards[s].numNodes; n++) { + MRClusterNode_Free(&t->shards[s].nodes[n]); + } + rm_free(t->shards[s].nodes); + } + rm_free(t->shards); + rm_free(t); +} diff --git a/src/coord/rmr/cluster_topology.h b/src/coord/rmr/cluster_topology.h new file mode 100644 index 00000000000..b52921c4a2e --- /dev/null +++ b/src/coord/rmr/cluster_topology.h @@ -0,0 +1,66 @@ +/* + * Copyright (c) 2006-Present, Redis Ltd. + * All rights reserved. + * + * Licensed under your choice of 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). +*/ +#pragma once + +#include +#include +#include "endpoint.h" +#include "node.h" + +#ifdef __cplusplus +extern "C" { +#endif + +typedef uint16_t mr_slot_t; + +/* A "shard" represents a slot range of the cluster, with its associated nodes. For each sharding + * key, we select the slot based on the hash function, and then look for the shard in the cluster's + * shard array */ +typedef struct { + mr_slot_t startSlot; + mr_slot_t endSlot; + size_t numNodes; + size_t capNodes; + MRClusterNode *nodes; +} MRClusterShard; + +/* Create a new cluster shard to be added to a topology */ +MRClusterShard MR_NewClusterShard(mr_slot_t startSlot, mr_slot_t endSlots, size_t capNodes); +void MRClusterShard_AddNode(MRClusterShard *sh, MRClusterNode *n); + +#define MRHASHFUNC_CRC12_STR "CRC12" +#define MRHASHFUNC_CRC16_STR "CRC16" + +typedef enum { + MRHashFunc_None = 0, + MRHashFunc_CRC12, + MRHashFunc_CRC16, +} MRHashFunc; + +/* A topology is the mapping of slots to shards and nodes */ +typedef struct MRClusterTopology { + size_t numSlots; + MRHashFunc hashFunc; + size_t numShards; + size_t capShards; + MRClusterShard *shards; +} MRClusterTopology; + +MRClusterTopology *MR_NewTopology(size_t numShards, size_t numSlots, MRHashFunc hashFunc); +void MRClusterTopology_AddShard(MRClusterTopology *topo, MRClusterShard *sh); + +void MRClusterTopology_Free(MRClusterTopology *t); + +void MRClusterNode_Free(MRClusterNode *n); + +MRClusterTopology *MRClusterTopology_Clone(MRClusterTopology *t); + +#ifdef __cplusplus +} +#endif diff --git a/src/coord/rmr/conn.c b/src/coord/rmr/conn.c index b799cd79f03..a2c214fe2d2 100644 --- a/src/coord/rmr/conn.c +++ b/src/coord/rmr/conn.c @@ -21,13 +21,23 @@ #include #include +typedef struct MRConn{ + MREndpoint ep; + redisAsyncContext *conn; + MRConnState state; + void *timer; + int protocol; // 0 (undetermined), 2, or 3 + uv_loop_t *loop; +} MRConn; + + static void MRConn_ConnectCallback(const redisAsyncContext *c, int status); static void MRConn_DisconnectCallback(const redisAsyncContext *, int); static int MRConn_Connect(MRConn *conn); static void MRConn_SwitchState(MRConn *conn, MRConnState nextState); static void MRConn_Free(void *ptr); static void MRConn_Stop(MRConn *conn); -static MRConn *MR_NewConn(MREndpoint *ep); +static MRConn *MR_NewConn(MREndpoint *ep, uv_loop_t *loop); static int MRConn_StartNewConnection(MRConn *conn); static int MRConn_SendAuth(MRConn *conn); @@ -42,12 +52,13 @@ static int MRConn_SendAuth(MRConn *conn); ##__VA_ARGS__) /* detaches from our redis context */ -static redisAsyncContext *detachFromConn(MRConn *conn, int shouldFree) { +static redisAsyncContext *detachFromConn(MRConn *conn, bool shouldFree) { if (!conn->conn) { return NULL; } redisAsyncContext *ac = conn->conn; + //Freeing the cbData, not the connection or the uvloop ac->data = NULL; conn->conn = NULL; if (shouldFree) { @@ -64,7 +75,7 @@ typedef struct { MRConn **conns; } MRConnPool; -static MRConnPool *_MR_NewConnPool(MREndpoint *ep, size_t num) { +static MRConnPool *_MR_NewConnPool(MREndpoint *ep, size_t num, uv_loop_t *loop) { MRConnPool *pool = rm_malloc(sizeof(*pool)); *pool = (MRConnPool){ .num = num, @@ -74,13 +85,13 @@ static MRConnPool *_MR_NewConnPool(MREndpoint *ep, size_t num) { /* Create the connection */ for (size_t i = 0; i < num; i++) { - pool->conns[i] = MR_NewConn(ep); + pool->conns[i] = MR_NewConn(ep, loop); } return pool; } static void MRConnPool_Free(void *privdata, void *p) { - UNUSED(privdata); + uv_loop_t *loop = (uv_loop_t *)privdata; MRConnPool *pool = p; if (!pool) return; for (size_t i = 0; i < pool->num; i++) { @@ -94,6 +105,7 @@ static void MRConnPool_Free(void *privdata, void *p) { /* Get a connection from the connection pool. We select the next available connected connection with * a round robin selector */ static MRConn *MRConnPool_Get(MRConnPool *pool) { + for (size_t i = 0; i < pool->num; i++) { MRConn *conn = pool->conns[pool->rr]; @@ -138,6 +150,7 @@ void MRConnManager_ReplyState(MRConnManager *mgr, RedisModuleCtx *ctx) { RedisModule_ReplyWithString(ctx, key); RedisModule_FreeString(ctx, key); RedisModule_ReplyWithArray(ctx, pool->num); + for (size_t i = 0; i < pool->num; i++) { RedisModule_ReplyWithCString(ctx, MRConnState_Str(pool->conns[i]->state)); } @@ -177,7 +190,7 @@ int MRConn_SendCommand(MRConn *c, MRCommand *cmd, redisCallbackFn *fn, void *pri } /* Add a node to the connection manager. Return 1 if it's been added or 0 if it hasn't */ -int MRConnManager_Add(MRConnManager *m, const char *id, MREndpoint *ep, int connect) { +int MRConnManager_Add(MRConnManager *m, uv_loop_t *loop, const char *id, MREndpoint *ep, int connect) { /* First try to see if the connection is already in the manager */ dictEntry *ptr = dictFind(m->map, id); if (ptr) { @@ -192,7 +205,7 @@ int MRConnManager_Add(MRConnManager *m, const char *id, MREndpoint *ep, int conn // if the node has changed, we just replace the pool with a new one automatically } - MRConnPool *pool = _MR_NewConnPool(ep, m->nodeConns); + MRConnPool *pool = _MR_NewConnPool(ep, m->nodeConns, loop); if (connect) { for (size_t i = 0; i < pool->num; i++) { MRConn_Connect(pool->conns[i]); @@ -243,6 +256,7 @@ int MRConnManager_Disconnect(MRConnManager *m, const char *id) { return REDIS_ERR; } + // Shrink the connection pool to the given number of connections // Assumes that the number of connections is less than the current number of connections, // and that the new number of connections is greater than 0 @@ -255,6 +269,7 @@ void MRConnManager_Shrink(MRConnManager *m, size_t num) { for (size_t i = num; i < pool->num; i++) { MRConn_Stop(pool->conns[i]); } + pool->num = num; pool->rr %= num; // set the round robin counter to the new pool size bound pool->conns = rm_realloc(pool->conns, num * sizeof(MRConn *)); @@ -265,7 +280,7 @@ void MRConnManager_Shrink(MRConnManager *m, size_t num) { // Expand the connection pool to the given number of connections // Assumes that the number of connections is greater than the current number of connections -void MRConnManager_Expand(MRConnManager *m, size_t num) { +void MRConnManager_Expand(MRConnManager *m, size_t num, uv_loop_t *loop) { dictIterator *it = dictGetIterator(m->map); dictEntry *entry; while ((entry = dictNext(it))) { @@ -276,7 +291,7 @@ void MRConnManager_Expand(MRConnManager *m, size_t num) { // There should always be at least one connection in the pool MREndpoint *ep = &pool->conns[0]->ep; for (size_t i = pool->num; i < num; i++) { - pool->conns[i] = MR_NewConn(ep); + pool->conns[i] = MR_NewConn(ep, loop); MRConn_StartNewConnection(pool->conns[i]); } pool->num = num; @@ -301,16 +316,18 @@ static void freeConn(MRConn *conn) { rm_free(conn); } + static void signalCallback(uv_timer_t *tm) { MRConn *conn = tm->data; + if (conn->state == MRConn_Connected) { return; // Nothing to do here! } + redisAsyncContext *ac = conn->conn; + if (conn->state == MRConn_Freeing) { - if (conn->conn) { - redisAsyncContext *ac = conn->conn; - // detach the connection + if (ac) { ac->data = NULL; conn->conn = NULL; redisAsyncDisconnect(ac); @@ -321,12 +338,12 @@ static void signalCallback(uv_timer_t *tm) { if (conn->state == MRConn_ReAuth) { if (MRConn_SendAuth(conn) != REDIS_OK) { - detachFromConn(conn, 1); + detachFromConn(conn, true); MRConn_SwitchState(conn, MRConn_Connecting); } } else if (conn->state == MRConn_Connecting) { if (MRConn_Connect(conn) == REDIS_ERR) { - detachFromConn(conn, 1); + detachFromConn(conn, true); MRConn_SwitchState(conn, MRConn_Connecting); } } else { @@ -336,9 +353,10 @@ static void signalCallback(uv_timer_t *tm) { /* Safely transition to current state */ static void MRConn_SwitchState(MRConn *conn, MRConnState nextState) { + uv_loop_t *loop = conn->loop; if (!conn->timer) { conn->timer = rm_malloc(sizeof(uv_timer_t)); - uv_timer_init(uv_default_loop(), conn->timer); + uv_timer_init(loop, conn->timer); ((uv_timer_t *)conn->timer)->data = conn; } CONN_LOG(conn, "Switching state to %s", MRConnState_Str(nextState)); @@ -382,18 +400,22 @@ static void MRConn_SwitchState(MRConn *conn, MRConnState nextState) { activate_timer: if (!uv_is_active(conn->timer)) { + uv_timer_t *tm = conn->timer; uv_timer_start(conn->timer, signalCallback, nextTimeout, 0); } } static void MRConn_AuthCallback(redisAsyncContext *c, void *r, void *privdata) { MRConn *conn = c->data; + redisReply *rep = r; if (!conn || conn->state == MRConn_Freeing) { // Will be picked up by disconnect callback goto cleanup; } + uv_loop_t *loop = conn->loop; + if (c->err || !r) { detachFromConn(conn, !!r); MRConn_SwitchState(conn, MRConn_Connecting); @@ -420,6 +442,7 @@ static void MRConn_AuthCallback(redisAsyncContext *c, void *r, void *privdata) { static int MRConn_SendAuth(MRConn *conn) { CONN_LOG(conn, "Authenticating..."); + uv_loop_t *loop = conn->loop; // if we failed to send the auth command, start a reconnect loop size_t len = 0; @@ -429,7 +452,7 @@ static int MRConn_SendAuth(MRConn *conn) { RedisModule_ThreadSafeContextLock(RSDummyContext); const char *internal_secret = RedisModule_GetInternalSecret(RSDummyContext, &len); // Create a local copy of the secret so we can release the GIL. - int status = redisAsyncCommand(conn->conn, MRConn_AuthCallback, conn, + int status = redisAsyncCommand(conn->conn, MRConn_AuthCallback, loop, "AUTH %s %b", INTERNALAUTH_USERNAME, internal_secret, len); if (status == REDIS_ERR) { MRConn_SwitchState(conn, MRConn_ReAuth); @@ -439,7 +462,7 @@ static int MRConn_SendAuth(MRConn *conn) { } else { // On Enterprise, we use the password we got from `CLUSTERSET`. // If we got here, we know we have a password. - if (redisAsyncCommand(conn->conn, MRConn_AuthCallback, conn, "AUTH %s", + if (redisAsyncCommand(conn->conn, MRConn_AuthCallback, loop, "AUTH %s", conn->ep.password) == REDIS_ERR) { MRConn_SwitchState(conn, MRConn_ReAuth); return REDIS_ERR; @@ -462,10 +485,10 @@ static int MRConn_TlsPasswordCallback(char *buf, int size, int rwflag, void *u) } static SSL_CTX* MRConn_CreateSSLContext(const char *cacert_filename, - const char *cert_filename, - const char *private_key_filename, - const char *private_key_pass, - redisSSLContextError *error) + const char *cert_filename, + const char *private_key_filename, + const char *private_key_pass, + redisSSLContextError *error) { SSL_CTX *ssl_ctx = SSL_CTX_new(SSLv23_client_method()); if (!ssl_ctx) { @@ -566,20 +589,25 @@ static int checkTLS(char** client_key, char** client_cert, char** ca_cert, char* static void MRConn_ConnectCallback(const redisAsyncContext *c, int status) { MRConn *conn = c->data; if (!conn) { + // The connection was already freed, we need to clean up the redisAsyncContext if (status == REDIS_OK) { // We need to free it here because we will not be getting a disconnect // callback. redisAsyncFree((redisAsyncContext *)c); - } else { - // Will be freed anyway } return; } + if (conn->state == MRConn_Freeing) { + // The connection is being freed, we need to clean up the redisAsyncContext + // Before the connection was established, there was a request to Stop connection, so do not proceed further + return; + } + // if the connection is not stopped - try to reconnect if (status != REDIS_OK) { CONN_LOG(conn, "Error on connect: %s", c->errstr); - detachFromConn(conn, 0); // Free the connection as well - we have an error + detachFromConn(conn, false); // Free the connection as well - we have an error MRConn_SwitchState(conn, MRConn_Connecting); return; } @@ -598,7 +626,7 @@ static void MRConn_ConnectCallback(const redisAsyncContext *c, int status) { if (key_file_pass) rm_free(key_file_pass); if(ssl_context == NULL || ssl_error != 0) { CONN_LOG(conn, "Error on ssl context creation: %s", (ssl_error != 0) ? redisSSLContextGetError(ssl_error) : "Unknown error"); - detachFromConn(conn, 0); // Free the connection as well - we have an error + detachFromConn(conn, false); // Free the connection as well - we have an error MRConn_SwitchState(conn, MRConn_Connecting); if (ssl_context) SSL_CTX_free(ssl_context); return; @@ -614,7 +642,7 @@ static void MRConn_ConnectCallback(const redisAsyncContext *c, int status) { ((struct redisAsyncContext*)c)->c.funcs = old_callbacks; CONN_LOG(conn, "Error on tls auth, %s.", err); - detachFromConn(conn, 0); // Free the connection as well - we have an error + detachFromConn(conn, false); // Free the connection as well - we have an error MRConn_SwitchState(conn, MRConn_Connecting); if (ssl_context) SSL_CTX_free(ssl_context); return; @@ -626,7 +654,7 @@ static void MRConn_ConnectCallback(const redisAsyncContext *c, int status) { // a password is set to the `default` ACL user. if (!IsEnterprise() || conn->ep.password) { if (MRConn_SendAuth(conn) != REDIS_OK) { - detachFromConn(conn, 1); + detachFromConn(conn, true); MRConn_SwitchState(conn, MRConn_Connecting); } } else { @@ -640,26 +668,23 @@ static void MRConn_DisconnectCallback(const redisAsyncContext *c, int status) { /* Ignore */ return; } - if (conn->state != MRConn_Freeing) { - detachFromConn(conn, 0); + detachFromConn(conn, false); MRConn_SwitchState(conn, MRConn_Connecting); } else { freeConn(conn); } } -static MRConn *MR_NewConn(MREndpoint *ep) { +static MRConn *MR_NewConn(MREndpoint *ep, uv_loop_t *loop) { MRConn *conn = rm_malloc(sizeof(MRConn)); - *conn = (MRConn){.state = MRConn_Disconnected, .conn = NULL, .protocol = 0}; + *conn = (MRConn){.state = MRConn_Disconnected, .conn = NULL, .protocol = 0, .loop = loop}; MREndpoint_Copy(&conn->ep, ep); return conn; } /* Connect to a cluster node. Return REDIS_OK if either connected, or if */ static int MRConn_Connect(MRConn *conn) { - RS_ASSERT(!conn->conn); - redisOptions options = {.type = REDIS_CONN_TCP, .options = REDIS_OPT_NOAUTOFREEREPLIES, .endpoint.tcp = {.ip = conn->ep.host, .port = conn->ep.port}}; @@ -670,12 +695,11 @@ static int MRConn_Connect(MRConn *conn) { redisAsyncFree(c); return REDIS_ERR; } - conn->conn = c; conn->conn->data = conn; conn->state = MRConn_Connecting; - redisLibuvAttach(conn->conn, uv_default_loop()); + redisLibuvAttach(conn->conn, conn->loop); redisAsyncSetConnectCallback(conn->conn, MRConn_ConnectCallback); redisAsyncSetDisconnectCallback(conn->conn, MRConn_DisconnectCallback); diff --git a/src/coord/rmr/conn.h b/src/coord/rmr/conn.h index 21eeb4c4677..ffbf6bab8ad 100644 --- a/src/coord/rmr/conn.h +++ b/src/coord/rmr/conn.h @@ -15,6 +15,7 @@ #include "endpoint.h" #include "command.h" #include "util/dict.h" +#include /* * The state of the connection. @@ -52,14 +53,8 @@ static inline const char *MRConnState_Str(MRConnState state) { return " +#endif +#include "io_runtime_ctx.h" +#include "rmalloc.h" +#include "conn.h" +#include "cluster.h" +#include // Include the assertion header +#include "../config.h" + +// Atomically exchange the pending topology with a new topology. +// Returns the old pending topology (or NULL if there was no pending topology). +static inline queueItem *exchangePendingTopo(IORuntimeCtx *io_runtime_ctx, queueItem *newTopo) { + return __atomic_exchange_n(&io_runtime_ctx->pendingTopo, newTopo, __ATOMIC_SEQ_CST); +} + +static inline bool CheckAndSetIoRuntimeNotStarted(IORuntimeCtx *io_runtime_ctx) { + return __builtin_expect((__atomic_test_and_set(&io_runtime_ctx->uv_runtime.io_runtime_started_or_starting, __ATOMIC_ACQUIRE) == false), false); +} + +static inline bool CheckIoRuntimeStarted(IORuntimeCtx *io_runtime_ctx) { + return io_runtime_ctx->uv_runtime.io_runtime_started_or_starting; +} + +static void triggerPendingQueues(IORuntimeCtx *io_runtime_ctx) { + array_foreach(io_runtime_ctx->pendingQueues, async, uv_async_send(async)); + array_free(io_runtime_ctx->pendingQueues); + io_runtime_ctx->pendingQueues = NULL; +} + +static void rqAsyncCb(uv_async_t *async) { + IORuntimeCtx *io_runtime_ctx = async->data; + // EDGE CASE: If loop_th_ready is false when a shutdown is fired, it could happen that the shutdown comes before the pendingQueues that are here being + // "reescheduled". + if (!io_runtime_ctx->uv_runtime.loop_th_ready) { + // Topology is scheduled to change, add to the list of pending queues after topology is properly applied + array_ensure_append_1(io_runtime_ctx->pendingQueues, async); // try again later + return; + } + queueItem *req; + while (NULL != (req = RQ_Pop(io_runtime_ctx->queue, &io_runtime_ctx->uv_runtime.async))) { + req->cb(req->privdata); + rm_free(req); + RQ_Done(io_runtime_ctx->queue); + } +} + +extern RedisModuleCtx *RSDummyContext; + +static void topologyFailureCB(uv_timer_t *timer) { + IORuntimeCtx *io_runtime_ctx = (IORuntimeCtx *)timer->data; + RedisModule_Log(RSDummyContext, "warning", "IORuntime ID %zu: Topology validation failed: not all nodes connected", io_runtime_ctx->queue->id); + uv_timer_stop(&io_runtime_ctx->uv_runtime.topologyValidationTimer); // stop the validation timer + // Mark the event loop thread as ready. This will allow any pending requests to be processed + // (and fail, but it will unblock clients) + io_runtime_ctx->uv_runtime.loop_th_ready = true; + triggerPendingQueues(io_runtime_ctx); +} + +static int CheckTopologyConnections(const MRClusterTopology *topo, + IORuntimeCtx *ioRuntime, + bool mastersOnly) { + for (size_t i = 0; i < topo->numShards; i++) { + MRClusterShard *sh = &topo->shards[i]; + for (size_t j = 0; j < sh->numNodes; j++) { + if (mastersOnly && !(sh->nodes[j].flags & MRNode_Master)) { + continue; + } + if (!MRConn_Get(&ioRuntime->conn_mgr, sh->nodes[j].id)) { + return REDIS_ERR; + } + } + } + return REDIS_OK; +} + +static void topologyTimerCB(uv_timer_t *timer) { + IORuntimeCtx *io_runtime_ctx = (IORuntimeCtx *)timer->data; + const MRClusterTopology *topo = io_runtime_ctx->topo; + // Can we lock the topology? here? + if (CheckTopologyConnections(topo, io_runtime_ctx, true) == REDIS_OK) { + // We are connected to all master nodes. We can mark the event loop thread as ready + io_runtime_ctx->uv_runtime.loop_th_ready = true; + RedisModule_Log(RSDummyContext, "verbose", "IORuntime ID %zu: All nodes connected: IO thread is ready to handle requests", io_runtime_ctx->queue->id); + uv_timer_stop(&io_runtime_ctx->uv_runtime.topologyValidationTimer); // stop the timer repetition + uv_timer_stop(&io_runtime_ctx->uv_runtime.topologyFailureTimer); // stop failure timer (as we are connected) + triggerPendingQueues(io_runtime_ctx); + } else { + RedisModule_Log(RSDummyContext, "verbose", "IORuntime ID %zu: Waiting for all nodes to connect", io_runtime_ctx->queue->id); + } +} + +static void topologyAsyncCB(uv_async_t *async) { + IORuntimeCtx *io_runtime_ctx = (IORuntimeCtx *)async->data; + queueItem *task = exchangePendingTopo(io_runtime_ctx, NULL); // take the topology + if (task) { + // Apply new topology + RedisModule_Log(RSDummyContext, "verbose", "IORuntime ID %zu: Applying new topology", io_runtime_ctx->queue->id); + // Mark the event loop thread as not ready. This will ensure that the next event on the event loop + // will be the topology check. If the topology hasn't changed, the topology check will quickly + // mark the event loop thread as ready again. + io_runtime_ctx->uv_runtime.loop_th_ready = false; + task->cb(task->privdata); + rm_free(task); + // Finish this round of topology checks to give the topology connections a chance to connect. + // Schedule connectivity check immediately with a 1ms repeat interval + uv_timer_start(&io_runtime_ctx->uv_runtime.topologyValidationTimer, topologyTimerCB, 0, 1); + if (clusterConfig.topologyValidationTimeoutMS) { + // Schedule a timer to fail the topology validation if we don't connect to all nodes in time + uv_timer_start(&io_runtime_ctx->uv_runtime.topologyFailureTimer, topologyFailureCB, clusterConfig.topologyValidationTimeoutMS, 0); + } + } +} + +void shutdown_cb(uv_async_t* handle) { + IORuntimeCtx* io_runtime_ctx = (IORuntimeCtx*)handle->data; + // Stop the event loop first + RedisModule_Log(RSDummyContext, "verbose", "IORuntime ID %zu: Stopping event loop", io_runtime_ctx->queue->id); + uv_stop(&io_runtime_ctx->uv_runtime.loop); +} + +// Add this new function to walk and close all handles +static void close_walk_cb(uv_handle_t* handle, void* arg) { + if (!uv_is_closing(handle)) { + uv_close(handle, NULL); + } +} + +#define THREAD_NAME_MAX_LEN 32 +/* start the event loop side thread */ +static void sideThread(void *arg) { + IORuntimeCtx *io_runtime_ctx = arg; + /* Set thread name for profiling and debugging */ + char thread_name[THREAD_NAME_MAX_LEN]; // Increased buffer size to accommodate ID + snprintf(thread_name, sizeof(thread_name), "%s-uv-%zu", REDISEARCH_MODULE_NAME, io_runtime_ctx->queue->id); + +#if defined(__linux__) + /* Use prctl instead to prevent using _GNU_SOURCE flag and implicit + * declaration */ + prctl(PR_SET_NAME, thread_name); +#elif defined(__APPLE__) && defined(__MACH__) + pthread_setname_np(thread_name); +#else + RedisModule_Log(RSDummyContext, "verbose", + "sideThread(): pthread_setname_np is not supported on this system"); +#endif + // loop is initialized and handles are ready + //io_runtime_ctx->loop_th_ready = false; // Until topology is validated, no requests are allowed (will be accumulated in the pending queue) + uv_async_send(&io_runtime_ctx->uv_runtime.topologyAsync); // start the topology check + // Run the event loop + RedisModule_Log(RSDummyContext, "verbose", "IORuntime ID %zu: Running event loop", io_runtime_ctx->queue->id); + uv_run(&io_runtime_ctx->uv_runtime.loop, UV_RUN_DEFAULT); + RedisModule_Log(RSDummyContext, "verbose", "IORuntime ID %zu: Event loop stopped", io_runtime_ctx->queue->id); + // After the loop stops, close all handles https://github.com/libuv/libuv/issues/709 + uv_walk(&io_runtime_ctx->uv_runtime.loop, close_walk_cb, NULL); + // Run the loop one more time to process close callbacks + uv_run(&io_runtime_ctx->uv_runtime.loop, UV_RUN_ONCE); + uv_loop_close(&io_runtime_ctx->uv_runtime.loop); +} + +uv_loop_t* IORuntimeCtx_GetLoop(IORuntimeCtx *io_runtime_ctx) { + return &io_runtime_ctx->uv_runtime.loop; +} + +/* Initialize the connections to all shards */ +int IORuntimeCtx_ConnectAll(IORuntimeCtx *ioRuntime) { + return MRConnManager_ConnectAll(&ioRuntime->conn_mgr); +} + +void IORuntimeCtx_UpdateNodes(IORuntimeCtx *ioRuntime) { + /* Get all the current node ids from the connection manager. We will remove all the nodes + * that are in the new topology, and after the update, delete all the nodes that are in this map + * and not in the new topology */ + const struct MRClusterTopology *topo = ioRuntime->topo; + dict *nodesToDisconnect = dictCreate(&dictTypeHeapStrings, NULL); + + dictIterator *it = dictGetIterator(ioRuntime->conn_mgr.map); + dictEntry *de; + while ((de = dictNext(it))) { + dictAdd(nodesToDisconnect, dictGetKey(de), NULL); + } + dictReleaseIterator(it); + + /* Walk the topology and add all nodes in it to the connection manager */ + for (int sh = 0; sh < topo->numShards; sh++) { + for (int n = 0; n < topo->shards[sh].numNodes; n++) { + // Update all the conn Manager in each of the runtimes. + MRClusterNode *node = &topo->shards[sh].nodes[n]; + MRConnManager_Add(&ioRuntime->conn_mgr, &ioRuntime->uv_runtime.loop, node->id, &node->endpoint, 0); + /* This node is still valid, remove it from the nodes to delete list */ + dictDelete(nodesToDisconnect, node->id); + } + } + + // if we didn't remove the node from the original nodes map copy, it means it's not in the new topology, + // we need to disconnect the node's connections + it = dictGetIterator(nodesToDisconnect); + while ((de = dictNext(it))) { + MRConnManager_Disconnect(&ioRuntime->conn_mgr, dictGetKey(de)); + } + dictReleaseIterator(it); + dictRelease(nodesToDisconnect); +} + +int IORuntimeCtx_UpdateNodesAndConnectAll(IORuntimeCtx *ioRuntime) { + IORuntimeCtx_UpdateNodes(ioRuntime); + IORuntimeCtx_ConnectAll(ioRuntime); + return REDIS_OK; +} + +static void UV_Init(IORuntimeCtx *io_runtime_ctx) { + io_runtime_ctx->uv_runtime.loop_th_ready = false; + io_runtime_ctx->uv_runtime.io_runtime_started_or_starting = false; + io_runtime_ctx->uv_runtime.loop_th_created = false; + io_runtime_ctx->uv_runtime.loop_th_creation_failed = false; + uv_loop_init(&io_runtime_ctx->uv_runtime.loop); + uv_mutex_init(&io_runtime_ctx->uv_runtime.loop_th_created_mutex); + uv_cond_init(&io_runtime_ctx->uv_runtime.loop_th_created_cond); + io_runtime_ctx->uv_runtime.shutdownAsync.data = io_runtime_ctx; + io_runtime_ctx->uv_runtime.async.data = io_runtime_ctx; + io_runtime_ctx->uv_runtime.topologyAsync.data = io_runtime_ctx; + io_runtime_ctx->uv_runtime.topologyFailureTimer.data = io_runtime_ctx; + io_runtime_ctx->uv_runtime.topologyValidationTimer.data = io_runtime_ctx; + uv_timer_init(&io_runtime_ctx->uv_runtime.loop, &io_runtime_ctx->uv_runtime.topologyValidationTimer); + uv_timer_init(&io_runtime_ctx->uv_runtime.loop, &io_runtime_ctx->uv_runtime.topologyFailureTimer); + uv_async_init(&io_runtime_ctx->uv_runtime.loop, &io_runtime_ctx->uv_runtime.async, rqAsyncCb); + uv_async_init(&io_runtime_ctx->uv_runtime.loop, &io_runtime_ctx->uv_runtime.shutdownAsync, shutdown_cb); + uv_async_init(&io_runtime_ctx->uv_runtime.loop, &io_runtime_ctx->uv_runtime.topologyAsync, topologyAsyncCB); +} + +static void UV_Close(IORuntimeCtx *io_runtime_ctx) { + // Close all handles when thread wasn't initialized + uv_close((uv_handle_t*)&io_runtime_ctx->uv_runtime.topologyValidationTimer, NULL); + uv_close((uv_handle_t*)&io_runtime_ctx->uv_runtime.topologyFailureTimer, NULL); + uv_close((uv_handle_t*)&io_runtime_ctx->uv_runtime.async, NULL); + uv_close((uv_handle_t*)&io_runtime_ctx->uv_runtime.shutdownAsync, NULL); + uv_close((uv_handle_t*)&io_runtime_ctx->uv_runtime.topologyAsync, NULL); + + // Run the loop once to process the close callbacks + uv_run(&io_runtime_ctx->uv_runtime.loop, UV_RUN_ONCE); + + uv_loop_close(&io_runtime_ctx->uv_runtime.loop); +} + +IORuntimeCtx *IORuntimeCtx_Create(size_t conn_pool_size, struct MRClusterTopology *initialTopology, size_t id, bool take_topo_ownership) { + IORuntimeCtx *io_runtime_ctx = rm_malloc(sizeof(IORuntimeCtx)); + MRConnManager_Init(&io_runtime_ctx->conn_mgr, conn_pool_size); + io_runtime_ctx->queue = RQ_New(io_runtime_ctx->conn_mgr.nodeConns * PENDING_FACTOR, id); + io_runtime_ctx->pendingTopo = NULL; + io_runtime_ctx->pendingQueues = NULL; + + if (take_topo_ownership) { + io_runtime_ctx->topo = initialTopology; + } else { + io_runtime_ctx->topo = MRClusterTopology_Clone(initialTopology); + } + UV_Init(io_runtime_ctx); + + return io_runtime_ctx; +} + +void IORuntimeCtx_FireShutdown(IORuntimeCtx *io_runtime_ctx) { + if (CheckIoRuntimeStarted(io_runtime_ctx)) { + // There may be a delay between the thread starting and the loop running, we need to account for it + uv_async_send(&io_runtime_ctx->uv_runtime.shutdownAsync); + } +} + +void IORuntimeCtx_Free(IORuntimeCtx *io_runtime_ctx) { + if (CheckIoRuntimeStarted(io_runtime_ctx)) { + // Here we know that at least the thread will be created + uv_mutex_lock(&io_runtime_ctx->uv_runtime.loop_th_created_mutex); + while (!io_runtime_ctx->uv_runtime.loop_th_created && !io_runtime_ctx->uv_runtime.loop_th_creation_failed) { + uv_cond_wait(&io_runtime_ctx->uv_runtime.loop_th_created_cond, &io_runtime_ctx->uv_runtime.loop_th_created_mutex); + } + uv_mutex_unlock(&io_runtime_ctx->uv_runtime.loop_th_created_mutex); + if (!io_runtime_ctx->uv_runtime.loop_th_creation_failed) { + // Make sure IORuntimeCtx Free is not holding the GIL + uv_thread_join(&io_runtime_ctx->uv_runtime.loop_th); + } + } else { + UV_Close(io_runtime_ctx); + } + array_free(io_runtime_ctx->pendingQueues); + RQ_Free(io_runtime_ctx->queue); + MRConnManager_Free(&io_runtime_ctx->conn_mgr); + queueItem *task = exchangePendingTopo(io_runtime_ctx, NULL); + if (task) { + struct UpdateTopologyCtx *ctx = task->privdata; + if (ctx && ctx->new_topo) { + MRClusterTopology_Free(ctx->new_topo); + } + rm_free(ctx); + rm_free(task); + } + if (io_runtime_ctx->topo) { + MRClusterTopology_Free(io_runtime_ctx->topo); + } + + // Destroy synchronization primitives + uv_mutex_destroy(&io_runtime_ctx->uv_runtime.loop_th_created_mutex); + uv_cond_destroy(&io_runtime_ctx->uv_runtime.loop_th_created_cond); + + rm_free(io_runtime_ctx); +} + +//TODO(Joan): Handle potential error from uv_thread_create, what if thread is not properly created (Not sure other thdpools handle it) +void IORuntimeCtx_Start(IORuntimeCtx *io_runtime_ctx) { + // Initialize the loop and timers + // Verify that we are running on the event loop thread + uv_mutex_lock(&io_runtime_ctx->uv_runtime.loop_th_created_mutex); + int uv_thread_create_status = uv_thread_create(&io_runtime_ctx->uv_runtime.loop_th, sideThread, io_runtime_ctx); + io_runtime_ctx->uv_runtime.loop_th_created = true; + io_runtime_ctx->uv_runtime.loop_th_creation_failed = uv_thread_create_status != 0; + uv_cond_signal(&io_runtime_ctx->uv_runtime.loop_th_created_cond); + uv_mutex_unlock(&io_runtime_ctx->uv_runtime.loop_th_created_mutex); + RS_ASSERT(uv_thread_create_status == 0); + REDISMODULE_NOT_USED(uv_thread_create_status); + RedisModule_Log(RSDummyContext, "verbose", "Created event loop thread for IORuntime ID %zu", io_runtime_ctx->queue->id); +} + +void IORuntimeCtx_Schedule(IORuntimeCtx *io_runtime_ctx, MRQueueCallback cb, void *privdata) { + if (CheckAndSetIoRuntimeNotStarted(io_runtime_ctx)) { + //This guarantees only one worker thread will start the IORuntime because of the atomic check. If started but loop is not ready, still RQ will accumulate the request + // and would still be processed when the thread uvloop starts + IORuntimeCtx_Start(io_runtime_ctx); + } + RQ_Push(io_runtime_ctx->queue, cb, privdata); + uv_async_send(&io_runtime_ctx->uv_runtime.async); +} + +void IORuntimeCtx_RequestCompleted(IORuntimeCtx *io_runtime_ctx) { + RQ_Done(io_runtime_ctx->queue); +} + +void IORuntimeCtx_Schedule_Topology(IORuntimeCtx *io_runtime_ctx, MRQueueCallback cb, struct MRClusterTopology *topo, bool take_topo_ownership) { + struct queueItem *newTask = rm_new(struct queueItem); + struct queueItem *oldTask = NULL; + //Clone it so that this runtime can handle its own copy + struct MRClusterTopology *new_topo; + if (take_topo_ownership) { + new_topo = topo; + } else { + new_topo = MRClusterTopology_Clone(topo); + } + struct UpdateTopologyCtx *ctx = rm_new(struct UpdateTopologyCtx); + ctx->ioRuntime = io_runtime_ctx; + ctx->new_topo = new_topo; + newTask->cb = cb; + newTask->privdata = ctx; + oldTask = exchangePendingTopo(io_runtime_ctx, newTask); + // I need to trigger regardless of the thread running or not, it would be eventually picked, the same way a regular Request is scheduled without checking + // if the thread is running or not. Otherwise there may be a race condition where a topology is never scheduled. + uv_async_send(&io_runtime_ctx->uv_runtime.topologyAsync); // trigger the topology check + if (oldTask) { + // If there was an old task + struct UpdateTopologyCtx *oldCtx = oldTask->privdata; + if (oldCtx->new_topo) { + MRClusterTopology_Free(oldCtx->new_topo); + } + rm_free(oldCtx); + rm_free(oldTask); + } +} + +void IORuntimeCtx_Debug_ClearPendingTopo(IORuntimeCtx *io_runtime_ctx) { + queueItem *task = exchangePendingTopo(io_runtime_ctx, NULL); + if (task) { + struct UpdateTopologyCtx *ctx = task->privdata; + if (ctx && ctx->new_topo) { + MRClusterTopology_Free(ctx->new_topo); + } + rm_free(ctx); + rm_free(task); + } +} + +void IORuntimeCtx_UpdateConnPoolSize(IORuntimeCtx *ioRuntime, size_t new_conn_pool_size) { + RS_ASSERT(new_conn_pool_size > 0); + size_t old_conn_pool_size = ioRuntime->conn_mgr.nodeConns; + if (old_conn_pool_size > new_conn_pool_size) { + MRConnManager_Shrink(&ioRuntime->conn_mgr, new_conn_pool_size); + } else if (old_conn_pool_size < new_conn_pool_size) { + MRConnManager_Expand(&ioRuntime->conn_mgr, new_conn_pool_size, IORuntimeCtx_GetLoop(ioRuntime)); + } +} diff --git a/src/coord/rmr/io_runtime_ctx.h b/src/coord/rmr/io_runtime_ctx.h new file mode 100644 index 00000000000..d62dcc56826 --- /dev/null +++ b/src/coord/rmr/io_runtime_ctx.h @@ -0,0 +1,90 @@ +/* + * Copyright (c) 2006-Present, Redis Ltd. + * All rights reserved. + * + * Licensed under your choice of 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). +*/ + +#pragma once + +#include "rq.h" +#include "conn.h" +#include +#include "util/arr.h" +#include "cluster_topology.h" + +#ifdef __cplusplus +extern "C" { +#endif + +// `*50` for following the previous behavior +// #define MAX_CONCURRENT_REQUESTS (MR_CONN_POOL_SIZE * 50) +#define PENDING_FACTOR 50 + +typedef struct { + bool loop_th_ready; /* set to true when the event loop thread is ready to process requests. + * This is set to false when a new topology is applied, and set to true + * when the topology check is done. */ + bool io_runtime_started_or_starting; /* Set to true when the IO Runtime is starting or already started. We know that at least one thread (main or worker) is initializing the thread so we are sure (by having atomic access) + * that the thread will be started only once.*/ + uv_async_t async; + uv_loop_t loop; + uv_thread_t loop_th; + uv_timer_t topologyValidationTimer, topologyFailureTimer; + uv_async_t topologyAsync; + uv_async_t shutdownAsync; + + // Thread creation / joining synchronization. Avoid race condition of joining a thread that was not created. + bool loop_th_created; + bool loop_th_creation_failed; + uv_mutex_t loop_th_created_mutex; + uv_cond_t loop_th_created_cond; +} UVRuntime; + +//Structure to encapsulate the IO Runtime context for MR operations to take place +typedef struct { + // Connectivity / topology structures + MRConnManager conn_mgr; + struct MRClusterTopology *topo; + + // Request queue and topology requests + MRWorkQueue *queue; + struct queueItem *pendingTopo; // The pending topology to be applied + arrayof(uv_async_t *) pendingQueues; + + //UV runtime + UVRuntime uv_runtime; + +} IORuntimeCtx; + +struct UpdateTopologyCtx { + IORuntimeCtx *ioRuntime; + struct MRClusterTopology *new_topo; +}; + +IORuntimeCtx *IORuntimeCtx_Create(size_t conn_pool_size, struct MRClusterTopology *initialTopology, size_t id, bool take_topo_ownership); +void IORuntimeCtx_Start(IORuntimeCtx *io_runtime_ctx); +void IORuntimeCtx_Free(IORuntimeCtx *io_runtime_ctx); +void IORuntimeCtx_FireShutdown(IORuntimeCtx *io_runtime_ctx); + +//TODO: Have it return int status (return error if thread not created) +void IORuntimeCtx_Schedule(IORuntimeCtx *io_runtime_ctx, MRQueueCallback cb, void *privdata); + +void IORuntimeCtx_RequestCompleted(IORuntimeCtx *io_runtime_ctx); + +// Clears the pendingTopology request that may be queued to be updated, and return the topology that was pending. +void IORuntimeCtx_Debug_ClearPendingTopo(IORuntimeCtx *io_runtime_ctx); +uv_loop_t* IORuntimeCtx_GetLoop(IORuntimeCtx *io_runtime_ctx); +int IORuntimeCtx_ConnectAll(IORuntimeCtx *ioRuntime); +void IORuntimeCtx_UpdateNodes(IORuntimeCtx *ioRuntime); +/* Update the topology by calling the topology provider explicitly with ctx. If ctx is NULL, the + * provider's current context is used. Otherwise, we call its function with the given context */ +int IORuntimeCtx_UpdateNodesAndConnectAll(IORuntimeCtx *ioRuntime); +void IORuntimeCtx_Schedule_Topology(IORuntimeCtx *io_runtime_ctx, MRQueueCallback cb, struct MRClusterTopology *topo, bool take_topo_ownership); +void IORuntimeCtx_UpdateConnPoolSize(IORuntimeCtx *ioRuntime, size_t new_conn_pool_size); + +#ifdef __cplusplus +} +#endif diff --git a/src/coord/rmr/redis_cluster.c b/src/coord/rmr/redis_cluster.c index d38f3d1ca0b..09da335144a 100644 --- a/src/coord/rmr/redis_cluster.c +++ b/src/coord/rmr/redis_cluster.c @@ -123,7 +123,7 @@ extern size_t NumShards; void UpdateTopology(RedisModuleCtx *ctx) { MRClusterTopology *topo = RedisCluster_GetTopology(ctx); if (topo) { // if we didn't get a topology, do nothing. Log was already printed - RedisModule_Log(ctx, "debug", "Setting number of partitions to %ld", topo->numShards); + RedisModule_Log(ctx, "debug", "UpdateTopology: Setting number of partitions to %ld", topo->numShards); NumShards = topo->numShards; MR_UpdateTopology(topo); } diff --git a/src/coord/rmr/rmr.c b/src/coord/rmr/rmr.c index ee97a393264..fb17674997a 100644 --- a/src/coord/rmr/rmr.c +++ b/src/coord/rmr/rmr.c @@ -30,15 +30,17 @@ #include "hiredis/hiredis.h" #include "hiredis/async.h" +#include "io_runtime_ctx.h" #define REFCOUNT_INCR_MSG(caller, refcount) \ RS_DEBUG_LOG_FMT("%s: increased refCount to == %d", caller, refcount); #define REFCOUNT_DECR_MSG(caller, refcount) \ RS_DEBUG_LOG_FMT("%s: decreased refCount to == %d", caller, refcount); -/* Currently a single cluster is supported */ +#define CEIL_DIV(a, b) ((a + b - 1) / b) + +/* A cluster is a pool of IORuntimes. It is owned by the main thread and accessed in the coordinator threads */ static MRCluster *cluster_g = NULL; -static MRWorkQueue *rq_g = NULL; /* Coordination request timeout */ long long timeout_g = 5000; // unused value. will be set in MR_Init @@ -56,6 +58,7 @@ typedef struct MRCtx { RedisModuleBlockedClient *bc; bool mastersOnly; MRCommand cmd; + IORuntimeCtx *ioRuntime; /** * This is a reduce function inside the MRCtx. @@ -75,6 +78,7 @@ void MR_SetCoordinationStrategy(MRCtx *ctx, bool mastersOnly) { /* Create a new MapReduce context */ MRCtx *MR_CreateCtx(RedisModuleCtx *ctx, RedisModuleBlockedClient *bc, void *privdata, int replyCap) { + RS_ASSERT(cluster_g); MRCtx *ret = rm_malloc(sizeof(MRCtx)); ret->numReplied = 0; ret->numErrored = 0; @@ -88,7 +92,7 @@ MRCtx *MR_CreateCtx(RedisModuleCtx *ctx, RedisModuleBlockedClient *bc, void *pri ret->bc = bc; RS_ASSERT(ctx || bc); ret->fn = NULL; - + ret->ioRuntime = MRCluster_GetIORuntimeCtx(cluster_g, MRCluster_AssignRoundRobinIORuntimeIdx(cluster_g)); return ret; } @@ -117,6 +121,10 @@ int MRCtx_GetNumReplied(struct MRCtx *ctx) { return ctx->numReplied; } +void MRCtx_RequestCompleted(struct MRCtx *ctx) { + IORuntimeCtx_RequestCompleted(ctx->ioRuntime); +} + MRReply** MRCtx_GetReplies(struct MRCtx *ctx) { return ctx->replies; } @@ -134,9 +142,9 @@ void MRCtx_SetReduceFunction(struct MRCtx *ctx, MRReduceFunc fn) { } static void freePrivDataCB(RedisModuleCtx *ctx, void *p) { - MR_requestCompleted(); if (p) { MRCtx *mc = p; + IORuntimeCtx_RequestCompleted(mc->ioRuntime); MRCtx_Free(mc); } } @@ -185,31 +193,19 @@ static void fanoutCallback(redisAsyncContext *c, void *r, void *privdata) { } } -// `*50` for following the previous behavior -// #define MAX_CONCURRENT_REQUESTS (MR_CONN_POOL_SIZE * 50) -#define PENDING_FACTOR 50 /* Initialize the MapReduce engine with a node provider */ -void MR_Init(MRCluster *cl, long long timeoutMS) { - - cluster_g = cl; +void MR_Init(size_t num_io_threads, size_t conn_pool_size, long long timeoutMS) { + cluster_g = MR_NewCluster(NULL, conn_pool_size, num_io_threads); timeout_g = timeoutMS; - rq_g = RQ_New(cl->mgr.nodeConns * PENDING_FACTOR); -} - -int MR_CheckTopologyConnections(bool mastersOnly) { - return MRCluster_CheckConnections(cluster_g, mastersOnly); -} - -bool MR_CurrentTopologyExists() { - return cluster_g->topo != NULL; } /* The fanout request received in the event loop in a thread safe manner */ static void uvFanoutRequest(void *p) { MRCtx *mrctx = p; + IORuntimeCtx *ioRuntime = mrctx->ioRuntime; mrctx->numExpected = - MRCluster_FanoutCommand(cluster_g, mrctx->mastersOnly, &mrctx->cmd, fanoutCallback, mrctx); + MRCluster_FanoutCommand(ioRuntime, mrctx->mastersOnly, &mrctx->cmd, fanoutCallback, mrctx); if (mrctx->numExpected == 0) { RedisModuleBlockedClient *bc = mrctx->bc; @@ -219,10 +215,12 @@ static void uvFanoutRequest(void *p) { } } +// This function already runs in one of the IO threads. We need to make sure that the adequate RuntimeCtx is used. This info can be found in the MRCtx static void uvMapRequest(void *p) { MRCtx *mrctx = p; + IORuntimeCtx *ioRuntime = mrctx->ioRuntime; - int rc = MRCluster_SendCommand(cluster_g, mrctx->mastersOnly, &mrctx->cmd, fanoutCallback, mrctx); + int rc = MRCluster_SendCommand(ioRuntime, mrctx->mastersOnly, &mrctx->cmd, fanoutCallback, mrctx); mrctx->numExpected = (rc == REDIS_OK) ? 1 : 0; if (mrctx->numExpected == 0) { @@ -233,10 +231,6 @@ static void uvMapRequest(void *p) { } } -void MR_requestCompleted() { - RQ_Done(rq_g); -} - /* Fanout map - send the same command to all the shards, sending the collective * reply to the reducer callback */ int MR_Fanout(struct MRCtx *mrctx, MRReduceFunc reducer, MRCommand cmd, bool block) { @@ -246,9 +240,12 @@ int MR_Fanout(struct MRCtx *mrctx, MRReduceFunc reducer, MRCommand cmd, bool blo mrctx->redisCtx, unblockHandler, timeoutHandler, freePrivDataCB, 0); // timeout_g); RedisModule_BlockedClientMeasureTimeStart(mrctx->bc); } + //Is possible that mrctx->fn may already be there and reducer to be null mrctx->reducer = reducer; mrctx->cmd = cmd; - RQ_Push(rq_g, uvFanoutRequest, mrctx); + + + IORuntimeCtx_Schedule(mrctx->ioRuntime, uvFanoutRequest, mrctx); return REDIS_OK; } @@ -258,75 +255,118 @@ int MR_MapSingle(struct MRCtx *ctx, MRReduceFunc reducer, MRCommand cmd) { RS_ASSERT(!ctx->bc); ctx->bc = RedisModule_BlockClient(ctx->redisCtx, unblockHandler, timeoutHandler, freePrivDataCB, 0); // timeout_g); RedisModule_BlockedClientMeasureTimeStart(ctx->bc); - RQ_Push(rq_g, uvMapRequest, ctx); + IORuntimeCtx_Schedule(ctx->ioRuntime, uvMapRequest, ctx); return REDIS_OK; } /* on-loop update topology request. This can't be done from the main thread */ static void uvUpdateTopologyRequest(void *p) { - MRCLuster_UpdateTopology(cluster_g, p); + struct UpdateTopologyCtx *ctx = p; + IORuntimeCtx *ioRuntime = ctx->ioRuntime; + MRClusterTopology *old_topo = ioRuntime->topo; + ioRuntime->topo = ctx->new_topo; + IORuntimeCtx_UpdateNodesAndConnectAll(ioRuntime); + rm_free(ctx); + if (old_topo) { + MRClusterTopology_Free(old_topo); + } } -/* Set a new topology for the cluster */ +/* Set a new topology for the cluster.*/ void MR_UpdateTopology(MRClusterTopology *newTopo) { - // enqueue a request on the io thread, this can't be done from the main thread - RQ_Push_Topology(uvUpdateTopologyRequest, newTopo); + // TODO(Joan): Most likely we need to make sure we wait for the topology to properly be applied to every runtime context before returning. + for (size_t i = 0; i < cluster_g->num_io_threads; i++) { + IORuntimeCtx_Schedule_Topology(cluster_g->io_runtimes_pool[i], uvUpdateTopologyRequest, newTopo, i == 0); + } } +struct UpdateConnPoolSizeCtx { + IORuntimeCtx *ioRuntime; + size_t conn_pool_size; +}; + /* Modifying the connection pools cannot be done from the main thread */ -static void uvUpdateConnPerShard(void *p) { - size_t connPerShard = (uintptr_t)p; - MRCluster_UpdateConnPerShard(cluster_g, connPerShard); - RQ_UpdateMaxPending(rq_g, connPerShard * PENDING_FACTOR); - MR_requestCompleted(); +static void uvUpdateConnPoolSize(void *p) { + struct UpdateConnPoolSizeCtx *ctx = p; + IORuntimeCtx *ioRuntime = ctx->ioRuntime; + IORuntimeCtx_UpdateConnPoolSize(ioRuntime, ctx->conn_pool_size); + size_t max_pending = ioRuntime->conn_mgr.nodeConns * PENDING_FACTOR; + RQ_UpdateMaxPending(ioRuntime->queue, max_pending); + IORuntimeCtx_RequestCompleted(ioRuntime); + rm_free(ctx); } extern size_t NumShards; -void MR_UpdateConnPerShard(size_t connPerShard) { - if (!rq_g) return; // not initialized yet, we have nothing to update yet. +void MR_UpdateConnPoolSize(size_t conn_pool_size) { + if (!cluster_g) return; // not initialized yet, we have nothing to update yet. if (NumShards == 1) { // If we observe that there is only one shard from the main thread, // we know the uv thread is not initialized yet (and may never be). // We can update the connection pool size directly from the main thread. // This is mostly a no-op, as the connection pool is not in use (yet or at all). // This call should only update the connection pool `size` for when the connection pool is initialized. - MRCluster_UpdateConnPerShard(cluster_g, connPerShard); + for (size_t i = 0; i < cluster_g->num_io_threads; i++) { + IORuntimeCtx_UpdateConnPoolSize(cluster_g->io_runtimes_pool[i], conn_pool_size); + } } else { - void *p = (void *)(uintptr_t)connPerShard; - RQ_Push(rq_g, uvUpdateConnPerShard, p); + for (size_t i = 0; i < cluster_g->num_io_threads; i++) { + struct UpdateConnPoolSizeCtx *ctx = rm_malloc(sizeof(*ctx)); + ctx->ioRuntime = cluster_g->io_runtimes_pool[i]; + ctx->conn_pool_size = conn_pool_size; + IORuntimeCtx_Schedule(cluster_g->io_runtimes_pool[i], uvUpdateConnPoolSize, ctx); + } } } +struct ReplyClusterInfoCtx { + IORuntimeCtx *ioRuntime; + RedisModuleBlockedClient *bc; +}; + static void uvGetConnectionPoolState(void *p) { - RedisModuleBlockedClient *bc = p; + struct ReplyClusterInfoCtx *replyClusterInfoCtx = p; + IORuntimeCtx *ioRuntime = replyClusterInfoCtx->ioRuntime; + RedisModuleBlockedClient *bc = replyClusterInfoCtx->bc; RedisModuleCtx *ctx = RedisModule_GetThreadSafeContext(bc); - MRConnManager_ReplyState(&cluster_g->mgr, ctx); - MR_requestCompleted(); + MRConnManager_ReplyState(&ioRuntime->conn_mgr, ctx); + IORuntimeCtx_RequestCompleted(ioRuntime); RedisModule_FreeThreadSafeContext(ctx); RedisModule_BlockedClientMeasureTimeEnd(bc); RedisModule_UnblockClient(bc, NULL); + rm_free(replyClusterInfoCtx); } void MR_GetConnectionPoolState(RedisModuleCtx *ctx) { RedisModuleBlockedClient *bc = RedisModule_BlockClient(ctx, NULL, NULL, NULL, 0); RedisModule_BlockedClientMeasureTimeStart(bc); - RQ_Push(rq_g, uvGetConnectionPoolState, bc); + struct ReplyClusterInfoCtx *replyClusterInfoCtx = rm_new(struct ReplyClusterInfoCtx); + size_t idx = MRCluster_AssignRoundRobinIORuntimeIdx(cluster_g); + replyClusterInfoCtx->bc = bc; + replyClusterInfoCtx->ioRuntime = cluster_g->io_runtimes_pool[idx]; + IORuntimeCtx_Schedule(replyClusterInfoCtx->ioRuntime, uvGetConnectionPoolState, replyClusterInfoCtx); } static void uvReplyClusterInfo(void *p) { - RedisModuleBlockedClient *bc = p; + struct ReplyClusterInfoCtx *replyClusterInfoCtx = p; + IORuntimeCtx *ioRuntime = replyClusterInfoCtx->ioRuntime; + RedisModuleBlockedClient *bc = replyClusterInfoCtx->bc; RedisModuleCtx *ctx = RedisModule_GetThreadSafeContext(bc); - MR_ReplyClusterInfo(ctx, cluster_g->topo); - MR_requestCompleted(); + MR_ReplyClusterInfo(ctx, ioRuntime->topo); + IORuntimeCtx_RequestCompleted(ioRuntime); RedisModule_FreeThreadSafeContext(ctx); RedisModule_BlockedClientMeasureTimeEnd(bc); RedisModule_UnblockClient(bc, NULL); + rm_free(replyClusterInfoCtx); } void MR_uvReplyClusterInfo(RedisModuleCtx *ctx) { RedisModuleBlockedClient *bc = RedisModule_BlockClient(ctx, NULL, NULL, NULL, 0); RedisModule_BlockedClientMeasureTimeStart(bc); - RQ_Push(rq_g, uvReplyClusterInfo, bc); + struct ReplyClusterInfoCtx *replyClusterInfoCtx = rm_new(struct ReplyClusterInfoCtx); + size_t idx = MRCluster_AssignRoundRobinIORuntimeIdx(cluster_g); + replyClusterInfoCtx->bc = bc; + replyClusterInfoCtx->ioRuntime = cluster_g->io_runtimes_pool[idx]; + IORuntimeCtx_Schedule(replyClusterInfoCtx->ioRuntime, uvReplyClusterInfo, replyClusterInfoCtx); } void MR_ReplyClusterInfo(RedisModuleCtx *ctx, MRClusterTopology *topo) { @@ -449,6 +489,7 @@ struct MRIteratorCtx { // reference counter of the iterator. // When it reaches 0, both readers and the writer agree that the iterator can be released int8_t itRefCount; + IORuntimeCtx *ioRuntime; }; struct MRIteratorCallbackCtx { @@ -474,7 +515,8 @@ static void mrIteratorRedisCB(redisAsyncContext *c, void *r, void *privdata) { } int MRIteratorCallback_ResendCommand(MRIteratorCallbackCtx *ctx) { - return MRCluster_SendCommand(cluster_g, true, &ctx->cmd, mrIteratorRedisCB, ctx); + IORuntimeCtx *io_runtime_ctx = ctx->it->ctx.ioRuntime; + return MRCluster_SendCommand(io_runtime_ctx, true, &ctx->cmd, mrIteratorRedisCB, ctx); } // Use after modifying `pending` (or any other variable of the iterator) to make sure it's visible @@ -484,8 +526,9 @@ void MRIteratorCallback_ProcessDone(MRIteratorCallbackCtx *ctx) { if (!inProcess) { MRChannel_Unblock(ctx->it->ctx.chan); RS_DEBUG_LOG("MRIteratorCallback_ProcessDone: calling MRIterator_Release"); + IORuntimeCtx *ioRuntime = ctx->it->ctx.ioRuntime; // Save before potential free MRIterator_Release(ctx->it); - RQ_Done(rq_g); + IORuntimeCtx_RequestCompleted(ioRuntime); } } @@ -545,37 +588,41 @@ void MRIteratorCallback_AddReply(MRIteratorCallbackCtx *ctx, MRReply *rep) { MRChannel_Push(ctx->it->ctx.chan, rep); } +// This function already runs in one of the IO threads. We need to make sure that the adequate RuntimeCtx is used. This info can be found in the MRIterator ctx void iterStartCb(void *p) { MRIterator *it = p; + IORuntimeCtx *io_runtime_ctx = it->ctx.ioRuntime; + size_t numShards = io_runtime_ctx->topo->numShards; + it->len = numShards; + it->ctx.pending = numShards; + it->ctx.inProcess = numShards; // Initially all commands are in process - size_t len = cluster_g->topo->numShards; - it->len = len; - it->ctx.pending = len; - it->ctx.inProcess = len; // Initially all commands are in process - - it->cbxs = rm_realloc(it->cbxs, len * sizeof(*it->cbxs)); + it->cbxs = rm_realloc(it->cbxs, numShards * sizeof(*it->cbxs)); MRCommand *cmd = &it->cbxs->cmd; - cmd->targetSlot = cluster_g->topo->shards[0].startSlot; // Set the first command to target the first shard - for (size_t i = 1; i < len; i++) { + cmd->targetSlot = io_runtime_ctx->topo->shards[0].startSlot; // Set the first command to target the first shard + for (size_t i = 1; i < numShards; i++) { it->cbxs[i].it = it; it->cbxs[i].cmd = MRCommand_Copy(cmd); // Set each command to target a different shard - it->cbxs[i].cmd.targetSlot = cluster_g->topo->shards[i].startSlot; + it->cbxs[i].cmd.targetSlot = io_runtime_ctx->topo->shards[i].startSlot; } + // This implies that every connection to each shard will work inside a single IO thread for (size_t i = 0; i < it->len; i++) { - if (MRCluster_SendCommand(cluster_g, true, &it->cbxs[i].cmd, + if (MRCluster_SendCommand(io_runtime_ctx, true, &it->cbxs[i].cmd, mrIteratorRedisCB, &it->cbxs[i]) == REDIS_ERR) { MRIteratorCallback_Done(&it->cbxs[i], 1); } } } +// This function already runs in one of the IO threads. We need to make sure that the adequate RuntimeCtx is used. This info can be found in the MRIterator ctx void iterManualNextCb(void *p) { MRIterator *it = p; + IORuntimeCtx *io_runtime_ctx = it->ctx.ioRuntime; for (size_t i = 0; i < it->len; i++) { if (!it->cbxs[i].cmd.depleted) { - if (MRCluster_SendCommand(cluster_g, true, &it->cbxs[i].cmd, + if (MRCluster_SendCommand(io_runtime_ctx, true, &it->cbxs[i].cmd, mrIteratorRedisCB, &it->cbxs[i]) == REDIS_ERR) { MRIteratorCallback_Done(&it->cbxs[i], 1); } @@ -606,7 +653,7 @@ bool MR_ManuallyTriggerNextIfNeeded(MRIterator *it, size_t channelThreshold) { // We need to take a reference to the iterator for the next batch of commands. int8_t refCount = MRIterator_IncreaseRefCount(it); REFCOUNT_INCR_MSG("MR_ManuallyTriggerNextIfNeeded", refCount); - RQ_Push(rq_g, iterManualNextCb, it); + IORuntimeCtx_Schedule(it->ctx.ioRuntime, iterManualNextCb, it); return true; // We may have more replies (and we surely will) } // We have no pending commands and no more than channelThreshold replies to process. @@ -633,6 +680,7 @@ MRIterator *MR_Iterate(const MRCommand *cmd, MRIteratorCallback cb) { .inProcess = 1, .timedOut = false, .itRefCount = 2, + .ioRuntime = MRCluster_GetIORuntimeCtx(cluster_g, MRCluster_AssignRoundRobinIORuntimeIdx(cluster_g)), }, .cbxs = rm_new(MRIteratorCallbackCtx), }; @@ -641,8 +689,7 @@ MRIterator *MR_Iterate(const MRCommand *cmd, MRIteratorCallback cb) { .cmd = MRCommand_Copy(cmd), .it = ret, }; - - RQ_Push(rq_g, iterStartCb, ret); + IORuntimeCtx_Schedule(ret->ctx.ioRuntime, iterStartCb, ret); return ret; } @@ -693,7 +740,7 @@ void MRIterator_Release(MRIterator *it) { // The iterator will be released when DEL commands are done. refcount = MRIterator_IncreaseRefCount(it); REFCOUNT_INCR_MSG("MRIterator_Release: triggering DEL on the shards' cursors", refcount); - RQ_Push(rq_g, iterManualNextCb, it); + IORuntimeCtx_Schedule(it->ctx.ioRuntime, iterManualNextCb, it); } else { // No pending shards, so no remote resources to free. // Free the iterator and we are done. @@ -701,3 +748,17 @@ void MRIterator_Release(MRIterator *it) { MRIterator_Free(it); } } + +void MR_Debug_ClearPendingTopo() { + for (size_t i = 0; i < cluster_g->num_io_threads; i++) { + IORuntimeCtx_Debug_ClearPendingTopo(cluster_g->io_runtimes_pool[i]); + } +} + +void MR_FreeCluster() { + if (!cluster_g) return; + RedisModule_ThreadSafeContextUnlock(RSDummyContext); + MRCluster_Free(cluster_g); + cluster_g = NULL; + RedisModule_ThreadSafeContextLock(RSDummyContext); +} diff --git a/src/coord/rmr/rmr.h b/src/coord/rmr/rmr.h index 8eff697bbb6..bb7d457603e 100644 --- a/src/coord/rmr/rmr.h +++ b/src/coord/rmr/rmr.h @@ -29,35 +29,33 @@ int MR_MapSingle(struct MRCtx *ctx, MRReduceFunc reducer, MRCommand cmd); void MR_SetCoordinationStrategy(struct MRCtx *ctx, bool mastersOnly); -/* Initialize the MapReduce engine with a node provider */ -void MR_Init(MRCluster *cl, long long timeoutMS); +/* Initialize the MapReduce engine with a given number of I/O threads and connections per each node in the Cluster */ +void MR_Init(size_t num_io_threads, size_t conn_pool_size, long long timeoutMS); /* Set a new topology for the cluster */ void MR_UpdateTopology(MRClusterTopology *newTopology); -/* Get the current cluster topology */ -bool MR_CurrentTopologyExists(); - -/* Get the current cluster topology connectivity status */ -int MR_CheckTopologyConnections(bool mastersOnly); - void MR_ReplyClusterInfo(RedisModuleCtx *ctx, MRClusterTopology *topo); void MR_GetConnectionPoolState(RedisModuleCtx *ctx); void MR_uvReplyClusterInfo(RedisModuleCtx *ctx); -void MR_UpdateConnPerShard(size_t connPerShard); +void MR_UpdateConnPoolSize(size_t conn_pool_size); + +void MR_Debug_ClearPendingTopo(); + +void MR_FreeCluster(); /* Get the user stored private data from the context */ void *MRCtx_GetPrivData(struct MRCtx *ctx); struct RedisModuleCtx *MRCtx_GetRedisCtx(struct MRCtx *ctx); int MRCtx_GetNumReplied(struct MRCtx *ctx); +void MRCtx_RequestCompleted(struct MRCtx *ctx); MRReply** MRCtx_GetReplies(struct MRCtx *ctx); RedisModuleBlockedClient *MRCtx_GetBlockedClient(struct MRCtx *ctx); void MRCtx_SetReduceFunction(struct MRCtx *ctx, MRReduceFunc fn); -void MR_requestCompleted(); /* Free the MapReduce context */ diff --git a/src/coord/rmr/rq.c b/src/coord/rmr/rq.c index ea33977d139..f0c7e79851e 100644 --- a/src/coord/rmr/rq.c +++ b/src/coord/rmr/rq.c @@ -8,166 +8,15 @@ */ #define RQ_C__ -#if defined(__linux__) -#include -#endif -#include - #include #include #include "rq.h" #include "rmalloc.h" -#include "rmr.h" -#include "coord/config.h" #include "rmutil/rm_assert.h" - -struct queueItem { - void *privdata; - MRQueueCallback cb; - struct queueItem *next; -}; - -typedef struct MRWorkQueue { - struct queueItem *head; - struct queueItem *tail; - int pending; - int maxPending; - size_t sz; - struct { - struct queueItem *head; - size_t warnSize; - } pendingInfo; - uv_mutex_t lock; - uv_async_t async; -} MRWorkQueue; - -uv_thread_t loop_th; -static char loop_th_started = false; // set to true when the event loop thread is started -static char loop_th_running = false; // set to true when the event loop thread is initialized -static char loop_th_ready = false; /* set to true when the event loop thread is ready to process requests. - * This is set to false when a new topology is applied, and set to true - * when the topology check is done. */ -uv_timer_t topologyValidationTimer, topologyFailureTimer; -uv_async_t topologyAsync; -struct queueItem *pendingTopo = NULL; -arrayof(uv_async_t *) pendingQueues = NULL; - -// Atomically exchange the pending topology with a new topology. -// Returns the old pending topology (or NULL if there was no pending topology). -static inline struct queueItem *exchangePendingTopo(struct queueItem *newTopo) { - return __atomic_exchange_n(&pendingTopo, newTopo, __ATOMIC_SEQ_CST); -} - -// Atomically check if the event loop thread is uninitialized and mark it as initialized. -// Returns true if the event loop thread was uninitialized, and in this case the caller should -// start the event loop thread. Should normally return false. -static inline bool loopThreadUninitialized() { - return __builtin_expect((__atomic_test_and_set(&loop_th_started, __ATOMIC_ACQUIRE) == false), false); -} - -static void triggerPendingQueues() { - array_foreach(pendingQueues, async, uv_async_send(async)); - array_free(pendingQueues); - pendingQueues = NULL; -} - -extern RedisModuleCtx *RSDummyContext; - -static void topologyFailureCB(uv_timer_t *timer) { - RedisModule_Log(RSDummyContext, "warning", "Topology validation failed: not all nodes connected"); - uv_timer_stop(&topologyValidationTimer); // stop the validation timer - // Mark the event loop thread as ready. This will allow any pending requests to be processed - // (and fail, but it will unblock clients) - loop_th_ready = true; - triggerPendingQueues(); -} - -static void topologyTimerCB(uv_timer_t *timer) { - if (MR_CheckTopologyConnections(true) == REDIS_OK) { - // We are connected to all master nodes. We can mark the event loop thread as ready - loop_th_ready = true; - RedisModule_Log(RSDummyContext, "verbose", "All nodes connected"); - uv_timer_stop(&topologyValidationTimer); // stop the timer repetition - uv_timer_stop(&topologyFailureTimer); // stop failure timer (as we are connected) - triggerPendingQueues(); - } else { - RedisModule_Log(RSDummyContext, "verbose", "Waiting for all nodes to connect"); - } -} - -static void topologyAsyncCB(uv_async_t *async) { - struct queueItem *topo = exchangePendingTopo(NULL); // take the topology - if (topo) { - // Apply new topology - RedisModule_Log(RSDummyContext, "verbose", "Applying new topology"); - // Mark the event loop thread as not ready. This will ensure that the next event on the event loop - // will be the topology check. If the topology hasn't changed, the topology check will quickly - // mark the event loop thread as ready again. - loop_th_ready = false; - topo->cb(topo->privdata); - rm_free(topo); - // Finish this round of topology checks to give the topology connections a chance to connect. - // Schedule connectivity check immediately with a 1ms repeat interval - uv_timer_start(&topologyValidationTimer, topologyTimerCB, 0, 1); - if (clusterConfig.topologyValidationTimeoutMS) { - // Schedule a timer to fail the topology validation if we don't connect to all nodes in time - uv_timer_start(&topologyFailureTimer, topologyFailureCB, clusterConfig.topologyValidationTimeoutMS, 0); - } - } -} - -/* start the event loop side thread */ -static void sideThread(void *arg) { - REDISMODULE_NOT_USED(arg); - /* Set thread name for profiling and debugging */ - char *thread_name = REDISEARCH_MODULE_NAME "-uv"; - -#if defined(__linux__) - /* Use prctl instead to prevent using _GNU_SOURCE flag and implicit - * declaration */ - prctl(PR_SET_NAME, thread_name); -#elif defined(__APPLE__) && defined(__MACH__) - pthread_setname_np(thread_name); -#else - RedisModule_Log(RSDummyContext, "verbose", - "sideThread(): pthread_setname_np is not supported on this system"); -#endif - // Mark the event loop thread as running before triggering the topology check. - loop_th_running = true; - uv_async_send(&topologyAsync); // start the topology check - uv_run(uv_default_loop(), UV_RUN_DEFAULT); -} - -static void verify_uv_thread() { - if (loopThreadUninitialized()) { - uv_timer_init(uv_default_loop(), &topologyValidationTimer); - uv_timer_init(uv_default_loop(), &topologyFailureTimer); - uv_async_init(uv_default_loop(), &topologyAsync, topologyAsyncCB); - // Verify that we are running on the event loop thread - int uv_thread_create_status = uv_thread_create(&loop_th, sideThread, NULL); - RS_ASSERT(uv_thread_create_status == 0); - REDISMODULE_NOT_USED(uv_thread_create_status); - RedisModule_Log(RSDummyContext, "verbose", "Created event loop thread"); - } -} - -void RQ_Push_Topology(MRQueueCallback cb, MRClusterTopology *topo) { - struct queueItem *oldTask, *newTask = rm_new(struct queueItem); - newTask->cb = cb; - newTask->privdata = topo; - oldTask = exchangePendingTopo(newTask); - if (loop_th_running) { - uv_async_send(&topologyAsync); // trigger the topology check - } - if (oldTask) { - MRClusterTopology_Free(oldTask->privdata); - rm_free(oldTask); - } -} +#include "rq.h" void RQ_Push(MRWorkQueue *q, MRQueueCallback cb, void *privdata) { - verify_uv_thread(); - struct queueItem *item = rm_new(*item); + queueItem *item = rm_new(struct queueItem); item->cb = cb; item->privdata = privdata; item->next = NULL; @@ -184,10 +33,10 @@ void RQ_Push(MRWorkQueue *q, MRQueueCallback cb, void *privdata) { q->sz++; uv_mutex_unlock(&q->lock); - uv_async_send(&q->async); } -static struct queueItem *rqPop(MRWorkQueue *q) { +// To be called from the event loop thread, need to protect the link list +queueItem *RQ_Pop(MRWorkQueue *q, uv_async_t* async) { uv_mutex_lock(&q->lock); if (q->head == NULL) { @@ -197,13 +46,12 @@ static struct queueItem *rqPop(MRWorkQueue *q) { if (q->pending >= q->maxPending) { uv_mutex_unlock(&q->lock); // If the queue is full we need to wake up the drain callback - uv_async_send(&q->async); - + uv_async_send(async); // Handle pending info logging. Access only to a non-NULL head and pendingInfo, // So it's safe to do without the lock. if (q->head == q->pendingInfo.head && q->sz > q->pendingInfo.warnSize) { // If we hit the same head multiple times, we may have a problem. Log it once. - RedisModule_Log(RSDummyContext, "warning", "Work queue at max pending with the same head. Size: %zu", q->sz); + RedisModule_Log(RSDummyContext, "warning", "Queue ID %zu: Work queue at max pending with the same head. Size: %zu", q->id, q->sz); q->pendingInfo.warnSize = q->sz + (1 << 10); } else { q->pendingInfo.head = q->head; @@ -216,7 +64,7 @@ static struct queueItem *rqPop(MRWorkQueue *q) { q->pendingInfo.warnSize = 0; } - struct queueItem *r = q->head; + queueItem *r = q->head; q->head = r->next; if (!q->head) q->tail = NULL; q->sz--; @@ -226,27 +74,12 @@ static struct queueItem *rqPop(MRWorkQueue *q) { return r; } +// 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); } -static void rqAsyncCb(uv_async_t *async) { - if (!loop_th_ready) { - array_ensure_append_1(pendingQueues, async); // try again later - return; - } - MRWorkQueue *q = async->data; - struct queueItem *req; - while (NULL != (req = rqPop(q))) { - req->cb(req->privdata); - rm_free(req); - } -} - -MRWorkQueue *RQ_New(int maxPending) { - +MRWorkQueue *RQ_New(int maxPending, size_t id) { MRWorkQueue *q = rm_calloc(1, sizeof(*q)); q->sz = 0; q->head = NULL; @@ -256,21 +89,25 @@ MRWorkQueue *RQ_New(int maxPending) { q->pendingInfo.head = NULL; q->pendingInfo.warnSize = 0; uv_mutex_init(&q->lock); - uv_async_init(uv_default_loop(), &q->async, rqAsyncCb); - q->async.data = q; + q->id = id; return q; } -void RQ_UpdateMaxPending(MRWorkQueue *q, int maxPending) { +void RQ_Free(MRWorkQueue *q) { uv_mutex_lock(&q->lock); - q->maxPending = maxPending; + // clear the queue + queueItem *cur = q->head; + while (cur) { + queueItem *next = cur->next; + rm_free(cur); + cur = next; + } uv_mutex_unlock(&q->lock); + uv_mutex_destroy(&q->lock); + rm_free(q); } -void RQ_Debug_ClearPendingTopo() { - struct queueItem *topo = exchangePendingTopo(NULL); - if (topo) { - MRClusterTopology_Free(topo->privdata); - rm_free(topo); - } +// To be called from the event loop thread, no need to protect the maxPending +void RQ_UpdateMaxPending(MRWorkQueue *q, int maxPending) { + q->maxPending = maxPending; } diff --git a/src/coord/rmr/rq.h b/src/coord/rmr/rq.h index c4b1b5ae3f9..10a8deb385a 100644 --- a/src/coord/rmr/rq.h +++ b/src/coord/rmr/rq.h @@ -10,20 +10,39 @@ #pragma once #include +#include +#include typedef void (*MRQueueCallback)(void *); -#ifndef RQ_C__ -typedef struct MRWorkQueue MRWorkQueue; +typedef struct queueItem { + void *privdata; + MRQueueCallback cb; + struct queueItem *next; +} queueItem; + +typedef struct MRWorkQueue { + size_t id; + queueItem *head; + queueItem *tail; + int pending; + int maxPending; + size_t sz; + struct { + queueItem *head; + size_t warnSize; + } pendingInfo; + uv_mutex_t lock; +} MRWorkQueue; + +MRWorkQueue *RQ_New(int maxPending, size_t id); + +void RQ_Free(MRWorkQueue *q); -MRWorkQueue *RQ_New(int maxPending); void RQ_UpdateMaxPending(MRWorkQueue *q, int maxPending); void RQ_Done(MRWorkQueue *q); void RQ_Push(MRWorkQueue *q, MRQueueCallback cb, void *privdata); -struct MRClusterTopology; -void RQ_Push_Topology(MRQueueCallback cb, struct MRClusterTopology *topo); -void RQ_Debug_ClearPendingTopo(); -#endif // RQ_C__ +queueItem *RQ_Pop(MRWorkQueue *q, uv_async_t* async); diff --git a/src/module.c b/src/module.c index 20ed5e9bc90..001f9ce6b01 100644 --- a/src/module.c +++ b/src/module.c @@ -81,6 +81,7 @@ } \ } while(0); +#define CEIL_DIV(a, b) ((a + b - 1) / b) extern RSConfig RSGlobalConfig; @@ -1432,6 +1433,7 @@ void RediSearch_CleanupModule(void) { CleanPool_ThreadPoolDestroy(); ReindexPool_ThreadPoolDestroy(); ConcurrentSearch_ThreadPoolDestroy(); + MR_FreeCluster(); // free global structures Extensions_Free(); @@ -2902,7 +2904,7 @@ static int searchResultReducer(struct MRCtx *mc, int count, MRReply **replies) { // and since we already replied with error in this case (in the beginning of this function), // we can't pass `mc` to the unblock function. searchRequestCtx_Free(req); - MR_requestCompleted(); + MRCtx_RequestCompleted(mc); MRCtx_Free(mc); return res; } @@ -3410,7 +3412,7 @@ static int DistSearchUnblockClient(RedisModuleCtx *ctx, RedisModuleString **argv RedisModule_ReplyWithError(ctx, "Could not send query to cluster"); } searchRequestCtx_Free(MRCtx_GetPrivData(mrctx)); - MR_requestCompleted(); + MRCtx_RequestCompleted(mrctx); MRCtx_Free(mrctx); } return REDISMODULE_OK; @@ -3507,13 +3509,7 @@ int ProfileCommandHandler(RedisModuleCtx *ctx, RedisModuleString **argv, int arg } int ClusterInfoCommand(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) { - if (MR_CurrentTopologyExists()) { - // If we have a topology, we must read it from the uv thread - MR_uvReplyClusterInfo(ctx); - } else { - // If we don't have a topology, we can reply immediately - MR_ReplyClusterInfo(ctx, NULL); - } + MR_uvReplyClusterInfo(ctx); return REDISMODULE_OK; } @@ -3530,7 +3526,7 @@ int SetClusterCommand(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) { return REDISMODULE_ERR; } - RedisModule_Log(ctx, "debug", "Setting number of partitions to %ld", topo->numShards); + RedisModule_Log(ctx, "debug", "SetClusterCommand: Setting number of partitions to %ld", topo->numShards); NumShards = topo->numShards; // send the topology to the cluster @@ -3563,8 +3559,10 @@ static int initSearchCluster(RedisModuleCtx *ctx, RedisModuleString **argv, int num_connections_per_shard = RSGlobalConfig.numWorkerThreads + 1; } - MRCluster *cl = MR_NewCluster(NULL, num_connections_per_shard); - MR_Init(cl, clusterConfig.timeoutMS); + size_t num_io_threads = clusterConfig.coordinatorIOThreads; + size_t conn_pool_size = CEIL_DIV(num_connections_per_shard, num_io_threads); + + MR_Init(num_io_threads, conn_pool_size, clusterConfig.timeoutMS); return REDISMODULE_OK; } diff --git a/tests/cpptests/CMakeLists.txt b/tests/cpptests/CMakeLists.txt index fb0c5f94edf..518c75e11df 100644 --- a/tests/cpptests/CMakeLists.txt +++ b/tests/cpptests/CMakeLists.txt @@ -11,7 +11,7 @@ include_directories(${root}/src/redisearch_rs/headers) include_directories(.) if (NOT TEST_MODULE) - set(TEST_MODULE redisearch) + set(TEST_MODULE redisearch) endif() # redismock is a mock library for using redis module API in tests, defined in main CMakeLists.txt. @@ -28,6 +28,8 @@ target_link_libraries(test_distagg ${TEST_MODULE} redismock) set_target_properties(test_distagg PROPERTIES COMPILE_FLAGS "-fvisibility=default") add_test(name test_distagg COMMAND test_distagg) +# Add the coord_tests subdirectory +add_subdirectory(coord_tests) file(GLOB BENCHMARK_SOURCES "benchmark_*.cpp") add_executable(rsbench ${BENCHMARK_SOURCES} index_utils.cpp) diff --git a/tests/cpptests/coord_tests/CMakeLists.txt b/tests/cpptests/coord_tests/CMakeLists.txt new file mode 100644 index 00000000000..c51685f1403 --- /dev/null +++ b/tests/cpptests/coord_tests/CMakeLists.txt @@ -0,0 +1,26 @@ +set(CMAKE_CXX_STANDARD 17) +set(CMAKE_CXX_STANDARD_REQUIRED ON) + +get_filename_component(root ${CMAKE_CURRENT_LIST_DIR}/../../.. ABSOLUTE) + +include_directories("${gtest_SOURCE_DIR}/include") +include_directories(${root}/src) +include_directories(${root}/deps) +include_directories(${root}/src/coord/rmr) +include_directories(..) +include_directories(.) + +if (NOT TEST_MODULE) + set(TEST_MODULE redisearch) +endif() + +file(GLOB TEST_SOURCES "test_cpp_*.cpp") + +set(COMMON_FILES + ../common.cpp +) + +add_executable(rstest_coord ${TEST_SOURCES} ${COMMON_FILES}) +target_link_libraries(rstest_coord gtest ${TEST_MODULE} redismock ${CMAKE_LD_LIBS}) +set_target_properties(rstest_coord PROPERTIES LINKER_LANGUAGE CXX) +add_test(NAME rstest_coord COMMAND rstest_coord) diff --git a/tests/cpptests/coord_tests/test_cpp_cluster_io_threads.cpp b/tests/cpptests/coord_tests/test_cpp_cluster_io_threads.cpp new file mode 100644 index 00000000000..66048e9dedd --- /dev/null +++ b/tests/cpptests/coord_tests/test_cpp_cluster_io_threads.cpp @@ -0,0 +1,156 @@ +/* + * Copyright (c) 2006-Present, Redis Ltd. + * All rights reserved. + * + * Licensed under your choice of 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). +*/ + +#include "gtest/gtest.h" +#include "io_runtime_ctx.h" +#include "cluster.h" +#include "rmutil/alloc.h" +#include "rmutil/rm_assert.h" +#include "redismodule.h" +#include +#include "coord/config.h" + +// Helper function to create a test topology +// Callback for regular tasks +static void callback(void *privdata) { + usleep(100000); // 10ms delay + int *counter = static_cast(privdata); + (*counter)++; +} + +// Callback for topology updates +static void topoCallback(void *privdata) { + struct UpdateTopologyCtx *ctx = static_cast(privdata); + IORuntimeCtx *ioRuntime = ctx->ioRuntime; + // Update the topology + if (ioRuntime->topo) { + MRClusterTopology_Free(ioRuntime->topo); + } + ioRuntime->topo = ctx->new_topo; + + // Set loop_th_ready to true to allow processing requests + ioRuntime->uv_runtime.loop_th_ready = true; + rm_free(ctx); +} + +// Test fixture for cluster IO threads tests +class ClusterIOThreadsTest : public ::testing::Test { +protected: + static MRClusterTopology *getDummyTopology(size_t numSlots) { + MRClusterTopology *topo = static_cast(rm_malloc(sizeof(*topo))); + topo->hashFunc = MRHashFunc_CRC16; + topo->numShards = 0; + topo->numSlots = numSlots; + topo->shards = nullptr; + return topo; + } +}; + +static void UpdateNumIOThreads(MRCluster *cl, size_t num_io_threads) { + RS_ASSERT(num_io_threads > 0); + + if (num_io_threads == cl->num_io_threads) return; + + if (num_io_threads < cl->num_io_threads) { + // Then free the runtime contexts + for (size_t i = num_io_threads; i < cl->num_io_threads; i++) { + IORuntimeCtx_FireShutdown(cl->io_runtimes_pool[i]); + } + for (size_t i = num_io_threads; i < cl->num_io_threads; i++) { + IORuntimeCtx_Free(cl->io_runtimes_pool[i]); + } + // Resize the pool + cl->io_runtimes_pool = (IORuntimeCtx**)rm_realloc(cl->io_runtimes_pool, sizeof(IORuntimeCtx*) * num_io_threads); + } else { + // Need to increase the number of IO threads + // Resize the pool + cl->io_runtimes_pool = (IORuntimeCtx**)rm_realloc(cl->io_runtimes_pool, sizeof(IORuntimeCtx*) * num_io_threads); + + // Create new runtime contexts + for (size_t i = cl->num_io_threads; i < num_io_threads; i++) { + cl->io_runtimes_pool[i] = IORuntimeCtx_Create( + cl->io_runtimes_pool[0]->conn_mgr.nodeConns, + NULL, + i + 1, + false); + if (cl->io_runtimes_pool[0]->topo) { + //TODO(Joan): We should make sure this is the last topology from user, so the UpdateTopology request should wait to return + cl->io_runtimes_pool[i]->topo = MRClusterTopology_Clone(cl->io_runtimes_pool[0]->topo); + cl->io_runtimes_pool[i]->uv_runtime.loop_th_ready = true; + } + } + } + cl->num_io_threads = num_io_threads; +} + +TEST_F(ClusterIOThreadsTest, TestIOThreadsResize) { + // Create a cluster with 3 IO threads initially + MRCluster *cluster = MR_NewCluster(nullptr, 2, 3); + ASSERT_EQ(cluster->num_io_threads, 3); + + size_t first_num_io_threads = cluster->num_io_threads; + + // Create counters to track callback execution + int counters[5] = {0}; + MRClusterTopology *topo = getDummyTopology(4096); + + // Schedule callbacks on each IO runtime + for (int i = 0; i < cluster->num_io_threads; i++) { + IORuntimeCtx *ioRuntime = MRCluster_GetIORuntimeCtx(cluster, i); + IORuntimeCtx_Schedule_Topology(ioRuntime, topoCallback, topo, false); + // Schedule multiple callbacks on each runtime + for (int j = 0; j < 10; j++) { + IORuntimeCtx_Schedule(ioRuntime, callback, &counters[i]); + } + } + + // make sure topology is applied, it either is put before the async, or the Topology timer will triggerPendingQueues. + // Since the order of the callbacks is not guaranteed, we can't assert on the counters (even if 2 async_t are sent in an specific order, + // the order of processing is not guaranteed in the uvloop) + usleep(DEFAULT_TOPOLOGY_VALIDATION_TIMEOUT*1000); // 100ms + + // Change number of IO threads (increase) + UpdateNumIOThreads(cluster, 5); + ASSERT_EQ(cluster->num_io_threads, 5); + + // Schedule more callbacks on the new threads + for (int i = first_num_io_threads; i < cluster->num_io_threads; i++) { + IORuntimeCtx *ioRuntime = MRCluster_GetIORuntimeCtx(cluster, i); + for (int j = 0; j < 10; j++) { + IORuntimeCtx_Schedule(ioRuntime, callback, &counters[i]); + } + } + + // Change number of IO threads (decrease) + UpdateNumIOThreads(cluster, 1); + ASSERT_EQ(cluster->num_io_threads, 1); + // Schedule more callbacks on the new threads + for (int i = 0; i < cluster->num_io_threads; i++) { + IORuntimeCtx *ioRuntime = MRCluster_GetIORuntimeCtx(cluster, i); + for (int j = 0; j < 10; j++) { + IORuntimeCtx_Schedule(ioRuntime, callback, &counters[i]); + } + } + + for (int i = 0; i < cluster->num_io_threads; i++) { + IORuntimeCtx *ioRuntime = MRCluster_GetIORuntimeCtx(cluster, i); + IORuntimeCtx_FireShutdown(ioRuntime); + } + + // Free the topology before freeing the cluster + rm_free(topo); + MRCluster_Free(cluster); + ASSERT_EQ(counters[0], 20); + ASSERT_EQ(counters[1], 10); + // Thread that was removed should still have executed its callbacks + ASSERT_EQ(counters[2], 10); + // New threads that were added and then removed should have executed their callbacks + ASSERT_EQ(counters[3], 10); + ASSERT_EQ(counters[4], 10); +} diff --git a/tests/cpptests/coord_tests/test_cpp_io_runtime_ctx.cpp b/tests/cpptests/coord_tests/test_cpp_io_runtime_ctx.cpp new file mode 100644 index 00000000000..f111c780601 --- /dev/null +++ b/tests/cpptests/coord_tests/test_cpp_io_runtime_ctx.cpp @@ -0,0 +1,190 @@ +/* + * Copyright (c) 2006-Present, Redis Ltd. + * All rights reserved. + * + * Licensed under your choice of 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). +*/ + +#include "gtest/gtest.h" +#include "io_runtime_ctx.h" +#include "cluster.h" +#include "rmutil/alloc.h" +#include "rmutil/rm_assert.h" +#include "redismodule.h" +#include + +// Test callback for queue operations +static void testCallback(void *privdata) { + int *counter = (int *)privdata; + (*counter)++; +} + +// Test callback for topology updates +static void testTopoCallback(void *privdata) { + struct UpdateTopologyCtx *ctx = (struct UpdateTopologyCtx *)privdata; + IORuntimeCtx *ioRuntime = ctx->ioRuntime; + //Simulate what the TopologyValidationTimer should do + ioRuntime->uv_runtime.loop_th_ready = true; + MRClusterTopology *old_topo = ioRuntime->topo; + MRClusterTopology *new_topo = ctx->new_topo; + ioRuntime->topo = new_topo; + rm_free(ctx); + if (old_topo) { + MRClusterTopology_Free(old_topo); + } +} + +class IORuntimeCtxCommonTest : public ::testing::Test { +protected: + IORuntimeCtx *ctx; + static MRClusterTopology *getDummyTopology(size_t numSlots) { + MRClusterTopology *topo = static_cast(rm_malloc(sizeof(*topo))); + topo->hashFunc = MRHashFunc_CRC16; + topo->numShards = 0; + topo->numSlots = numSlots; + topo->shards = nullptr; + return topo; + } + + void SetUp() override { + struct MRClusterTopology *topo = getDummyTopology(4096); + ctx = IORuntimeCtx_Create(2, topo, 1, false); + MRClusterTopology_Free(topo); + } + + void TearDown() override { + // Clear any pending topology before shutdown + IORuntimeCtx_FireShutdown(ctx); + IORuntimeCtx_Free(ctx); + } +}; + +TEST_F(IORuntimeCtxCommonTest, InitialState) { + ASSERT_NE(ctx, nullptr); + ASSERT_NE(ctx->queue, nullptr); + ASSERT_EQ(ctx->pendingTopo, nullptr); + ASSERT_FALSE(ctx->uv_runtime.loop_th_ready); + ASSERT_FALSE(ctx->uv_runtime.io_runtime_started_or_starting); + ASSERT_EQ(ctx->pendingQueues, nullptr); + ASSERT_FALSE(ctx->uv_runtime.loop_th_created); + ASSERT_FALSE(ctx->uv_runtime.loop_th_creation_failed); +} + +TEST_F(IORuntimeCtxCommonTest, Schedule) { + int counter = 0; + IORuntimeCtx_Schedule(ctx, testCallback, &counter); + // Give some time for thread to start + usleep(100); + ASSERT_TRUE(ctx->uv_runtime.io_runtime_started_or_starting); + ASSERT_TRUE(ctx->uv_runtime.loop_th_created); + ASSERT_FALSE(ctx->uv_runtime.loop_th_creation_failed); + // Verify the callback has not been called yet, thread not ready because no Topology is called + ASSERT_EQ(counter, 0); + struct MRClusterTopology *topo = getDummyTopology(4091); + IORuntimeCtx_Schedule_Topology(ctx, testTopoCallback, topo, false); + MRClusterTopology_Free(topo); + + usleep(1000); + + for (int i = 0; i < 10; i++) { + IORuntimeCtx_Schedule(ctx, testCallback, &counter); + } + // Give some time for thread to start + usleep(1000); + // Now the Runtime processed the topology and the pending queue + ASSERT_EQ(counter, 11); +} + +TEST_F(IORuntimeCtxCommonTest, ScheduleTopology) { + // Create a new topology + MRClusterTopology *newTopo = getDummyTopology(4097); + + // Schedule the topology update + IORuntimeCtx_Schedule_Topology(ctx, testTopoCallback, newTopo, false); + MRClusterTopology_Free(newTopo); + + // Give some time to consider that if it had to be applied it would have time + usleep(2000); + + // Verify the topology was not yet updated (will be updated once a request is scheduled) + ASSERT_EQ(ctx->topo->numSlots, 4096); + + int counter = 0; + IORuntimeCtx_Schedule(ctx, testCallback, &counter); + + usleep(2000); + ASSERT_EQ(ctx->topo->numSlots, 4097); + + // We don't need to free newTopo here as it's handled by testTopoCallback +} + +TEST_F(IORuntimeCtxCommonTest, MultipleTopologyUpdates) { + // Schedule one dummy request to start the thread and still have the flag io_runtime_started_or_starting set to true + int counter = 0; + IORuntimeCtx_Schedule(ctx, testCallback, &counter); + // Schedule multiple topology updates in quick succession + for (int i = 3; i <= 5; i++) { + MRClusterTopology *newTopo = getDummyTopology(4096 + i); + IORuntimeCtx_Schedule_Topology(ctx, testTopoCallback, newTopo, false); + MRClusterTopology_Free(newTopo); + } + + // Give some time for the last topology to be applied + usleep(3000); + + // Only the last topology should be applied + ASSERT_EQ(ctx->topo->numSlots, 4101); +} + +TEST_F(IORuntimeCtxCommonTest, ClearPendingTopo) { + // Create a new topology but don't start the runtime + const char *new_hosts[] = {"localhost:6379", "localhost:6389", "localhost:6399", "localhost:6409", "localhost:6419"}; + MRClusterTopology *newTopo = getDummyTopology(2048); + + // Schedule the topology update + IORuntimeCtx_Schedule_Topology(ctx, testTopoCallback, newTopo, false); + + MRClusterTopology_Free(newTopo); + // Verify we have a pending topology + ASSERT_NE(ctx->pendingTopo, nullptr); + + // Clear the pending topology + IORuntimeCtx_Debug_ClearPendingTopo(ctx); +} + +TEST_F(IORuntimeCtxCommonTest, ShutdownWithPendingRequests) { + IORuntimeCtx *io_runtime_ctx = IORuntimeCtx_Create(2, NULL, 1, false); + int counter = 0; + + MRClusterTopology *newTopo = getDummyTopology(4097); + IORuntimeCtx_Schedule_Topology(io_runtime_ctx, testTopoCallback, newTopo, false); + MRClusterTopology_Free(newTopo); + + // Create a delayed callback that takes 100ms to complete + auto delayedCallback = [](void *privdata) { + int *counter = (int *)privdata; + usleep(100000); // 100ms delay + (*counter)++; + }; + + IORuntimeCtx_Schedule(io_runtime_ctx, testCallback, &counter); + // Send one request and make sure it runs to make the test better. Otherwise the async callback does not see the topology applied + // and delays the callback call (and shutdown call may be called before all the callbacks are called) + usleep(20000); + + // Schedule 10 delayed requests + for (int i = 0; i < 10; i++) { + IORuntimeCtx_Schedule(io_runtime_ctx, delayedCallback, &counter); + } + //usleep(100); // 100ms delay + ASSERT_LT(counter, 11); + + // Fire shutdown and wait for completion, the shutdown is scheduled to run at the end of the event loop (is just another event) + IORuntimeCtx_FireShutdown(io_runtime_ctx); + IORuntimeCtx_Free(io_runtime_ctx); + + // Verify all requests were processed despite shutdown + ASSERT_EQ(counter, 11); +} diff --git a/tests/ctests/coord_tests/CMakeLists.txt b/tests/ctests/coord_tests/CMakeLists.txt index 051b914eb40..0b4624e6217 100644 --- a/tests/ctests/coord_tests/CMakeLists.txt +++ b/tests/ctests/coord_tests/CMakeLists.txt @@ -1,21 +1,21 @@ if (NOT TEST_MODULE) - set(TEST_MODULE redisearch) + set(TEST_MODULE redisearch) endif() include_directories(${root}/src/coord/rmr) function(RMRTEST name) - add_executable(${name} ${name}.c) - add_dependencies(${name} ${TEST_MODULE} redismock) - target_link_libraries("${name}" redismock ${TEST_MODULE} ${CMAKE_LD_LIBS}) - add_test(NAME "${name}" COMMAND "${name}") - set_target_properties("${name}" PROPERTIES COMPILE_FLAGS "-fvisibility=default") + add_executable(${name} ${name}.c) + add_dependencies(${name} ${TEST_MODULE} redismock) + target_link_libraries("${name}" redismock ${TEST_MODULE} ${CMAKE_LD_LIBS}) + add_test(NAME "${name}" COMMAND "${name}") + set_target_properties("${name}" PROPERTIES COMPILE_FLAGS "-fvisibility=default") endfunction() file(GLOB TEST_SOURCES "test_*.c") foreach(n ${TEST_SOURCES}) - get_filename_component(test_name ${n} NAME_WE) - RMRTEST(${test_name}) + get_filename_component(test_name ${n} NAME_WE) + RMRTEST(${test_name}) endforeach() diff --git a/tests/ctests/coord_tests/test_cluster.c b/tests/ctests/coord_tests/test_cluster.c index 058dcd01ea4..c60899c3ebb 100644 --- a/tests/ctests/coord_tests/test_cluster.c +++ b/tests/ctests/coord_tests/test_cluster.c @@ -94,7 +94,7 @@ static const char *GetShardKey(const MRCommand *cmd, size_t *len) { *len = cmd->lens[1]; return cmd->strs[1]; } -static mr_slot_t CRCShardFunc(const MRCommand *cmd, const MRCluster *cl) { +static mr_slot_t CRCShardFunc(const MRCommand *cmd, const IORuntimeCtx *ioRuntime) { if(cmd->targetSlot >= 0){ return cmd->targetSlot; @@ -104,8 +104,8 @@ static mr_slot_t CRCShardFunc(const MRCommand *cmd, const MRCluster *cl) { const char *k = GetShardKey(cmd, &len); if (!k) return 0; // Default to crc16 - uint16_t crc = (cl->topo->hashFunc == MRHashFunc_CRC12) ? crc12(k, len) : crc16(k, len); - return crc % cl->topo->numSlots; + uint16_t crc = (ioRuntime->topo->hashFunc == MRHashFunc_CRC12) ? crc12(k, len) : crc16(k, len); + return crc % ioRuntime->topo->numSlots; } void testShardingFunc() { @@ -113,36 +113,99 @@ void testShardingFunc() { MRCommand cmd = MR_NewCommand(2, "foo", "baz"); const char *host = "localhost:6379"; MRClusterTopology *topo = getTopology(4096, 1, &host); - MRCluster *cl = MR_NewCluster(topo, 2); - mr_slot_t shard = CRCShardFunc(&cmd, cl); - mu_assert_int_eq(shard, 717); + MRCluster *cl = MR_NewCluster(topo, 2, 3); + for (int i = 0; i < cl->num_io_threads; i++) { + IORuntimeCtx *ioRuntime = MRCluster_GetIORuntimeCtx(cl, i); + mr_slot_t shard = CRCShardFunc(&cmd, ioRuntime); + mu_assert_int_eq(shard, 717); + } MRCommand_Free(&cmd); - MRClust_Free(cl); + MRCluster_Free(cl); } -MRClusterShard *_MRCluster_FindShard(MRCluster *cl, mr_slot_t slot); - -void testCluster() { - +void testClusterTopology_Clone() { int n = 4; const char *hosts[] = {"localhost:6379", "localhost:6389", "localhost:6399", "localhost:6409"}; MRClusterTopology *topo = getTopology(4096, n, hosts); - MRCluster *cl = MR_NewCluster(topo, 2); - mu_check(cl != NULL); - // mu_check(cl->tp == tp); - mu_check(cl->topo->numShards == n); - mu_check(cl->topo->numSlots == 4096); - - for (int i = 0; i < cl->topo->numShards; i++) { - MRClusterShard *sh = &cl->topo->shards[i]; - mu_check(sh->numNodes == 1); - mu_check(sh->startSlot == i * (4096 / n)); - mu_check(sh->endSlot == sh->startSlot + (4096 / n) - 1); - mu_check(!strcmp(sh->nodes[0].id, hosts[i])); + // Clone the topology + MRClusterTopology *cloned = MRClusterTopology_Clone(topo); + + // Verify the clone has the same basic properties + mu_check(cloned != NULL); + mu_check(cloned != topo); // Different memory address + mu_check(cloned->numShards == topo->numShards); + mu_check(cloned->numSlots == topo->numSlots); + mu_check(cloned->hashFunc == topo->hashFunc); + + // Verify each shard was properly cloned + for (int j = 0; j < topo->numShards; j++) { + MRClusterShard *original_sh = &topo->shards[j]; + MRClusterShard *cloned_sh = &cloned->shards[j]; + + mu_check(cloned_sh->startSlot == original_sh->startSlot); + mu_check(cloned_sh->endSlot == original_sh->endSlot); + mu_check(cloned_sh->numNodes == original_sh->numNodes); + + // Verify each node in the shard + for (int k = 0; k < original_sh->numNodes; k++) { + mu_check(strcmp(cloned_sh->nodes[k].id, original_sh->nodes[k].id) == 0); + mu_check(cloned_sh->nodes[k].id != original_sh->nodes[k].id); // Different memory address + mu_check(strcmp(cloned_sh->nodes[k].endpoint.host, original_sh->nodes[k].endpoint.host) == 0); + mu_check(cloned_sh->nodes[k].endpoint.port == original_sh->nodes[k].endpoint.port); + mu_check(cloned_sh->nodes[k].flags == original_sh->nodes[k].flags); + } } - MRClust_Free(cl); + // Modify the original to prove independence + topo->numSlots = 8192; + topo->shards[0].startSlot = 999; + + // Verify the clone remains unchanged + mu_check(cloned->numSlots == 4096); + mu_check(cloned->shards[0].startSlot != 999); + + // Clean up + MRClusterTopology_Free(topo); + MRClusterTopology_Free(cloned); +} + +MRClusterShard *_MRCluster_FindShard(MRClusterTopology *topo, mr_slot_t slot); + +void testCluster() { + for (int num_io_threads = 1; num_io_threads <= 4; num_io_threads++) { + int n = 4; + const char *hosts[] = {"localhost:6379", "localhost:6389", "localhost:6399", "localhost:6409"}; + MRClusterTopology *topo = getTopology(4096, n, hosts); + + mu_check(topo->numShards == n); + mu_check(topo->numSlots == 4096); + for (int j = 0; j < topo->numShards; j++) { + MRClusterShard *sh = &topo->shards[j]; + mu_check(sh->numNodes == 1); + mu_check(sh->startSlot == j * (4096 / n)); + mu_check(sh->endSlot == sh->startSlot + (4096 / n) - 1); + mu_check(!strcmp(sh->nodes[0].id, hosts[j])); + } + + MRCluster *cl = MR_NewCluster(topo, 2, num_io_threads); + mu_check(cl != NULL); + // mu_check(cl->tp == tp); + for (int i = 0; i < cl->num_io_threads; i++) { + IORuntimeCtx *ioRuntime = MRCluster_GetIORuntimeCtx(cl, i); + mu_check(ioRuntime->topo->numShards == n); + mu_check(ioRuntime->topo->numSlots == 4096); + for (int j = 0; j < ioRuntime->topo->numShards; j++) { + MRClusterShard *sh = &ioRuntime->topo->shards[j]; + mu_check(sh->numNodes == 1); + mu_check(sh->startSlot == j * (4096 / n)); + mu_check(sh->endSlot == sh->startSlot + (4096 / n) - 1); + mu_check(!strcmp(sh->nodes[0].id, hosts[j])); + } + } + + MRCluster_Free(cl); + } } void testClusterSharding() { @@ -150,17 +213,19 @@ void testClusterSharding() { const char *hosts[] = {"localhost:6379", "localhost:6389", "localhost:6399", "localhost:6409"}; MRClusterTopology *topo = getTopology(4096, n, hosts); - MRCluster *cl = MR_NewCluster(topo, 2); + MRCluster *cl = MR_NewCluster(topo, 2, 3); MRCommand cmd = MR_NewCommand(4, "_FT.SEARCH", "foob", "bar", "baz"); - mr_slot_t slot = CRCShardFunc(&cmd, cl); - mu_check(slot > 0); - MRClusterShard *sh = _MRCluster_FindShard(cl, slot); - mu_check(sh != NULL); - mu_check(sh->numNodes == 1); - mu_check(!strcmp(sh->nodes[0].id, hosts[3])); - + for (int i = 0; i < cl->num_io_threads; i++) { + IORuntimeCtx *ioRuntime = MRCluster_GetIORuntimeCtx(cl, i); + mr_slot_t slot = CRCShardFunc(&cmd, ioRuntime); + mu_check(slot > 0); + MRClusterShard *sh = _MRCluster_FindShard(ioRuntime->topo, slot); + mu_check(sh != NULL); + mu_check(sh->numNodes == 1); + mu_check(!strcmp(sh->nodes[0].id, hosts[3])); + } MRCommand_Free(&cmd); - MRClust_Free(cl); + MRCluster_Free(cl); } static void dummyLog(RedisModuleCtx *ctx, const char *level, const char *fmt, ...) {} @@ -172,6 +237,7 @@ int main(int argc, char **argv) { MU_RUN_TEST(testShardingFunc); MU_RUN_TEST(testCluster); MU_RUN_TEST(testClusterSharding); + MU_RUN_TEST(testClusterTopology_Clone); MU_REPORT(); return minunit_status; diff --git a/tests/pytests/test.py b/tests/pytests/test.py index 75f4038acb2..7f62e4624a6 100644 --- a/tests/pytests/test.py +++ b/tests/pytests/test.py @@ -4133,16 +4133,21 @@ def prepare_env(env): env.expect('SEARCH.CLUSTERSET', 'MYID', '0', 'RANGES', str(env.shardsCount), *shards).ok() @skip(cluster=False) -def test_rq_job_without_topology(env:Env): +def test_rq_job_without_topology(): + env = Env(moduleArgs="SEARCH_IO_THREADS 20") env.expect(debug_cmd(), 'PAUSE_TOPOLOGY_UPDATER').ok() env.expect(debug_cmd(), 'CLEAR_PENDING_TOPOLOGY').ok() workers = 5 env.expect(config_cmd(), 'SET', 'WORKERS', workers).ok() + num_io_threads = 20 + def compute_number_of_connections_in_each_ioruntime(num_connections): + return max(1, num_connections // num_io_threads) # Verify that the `SHARD_CONNECTION_STATES` debug command is blocked when the topology is not set. try: con = env.getConnection() with TimeLimit(2, 'Failed waiting (SUCCESS!)'): + print('Waiting for SHARD_CONNECTION_STATES to block...') con.execute_command(debug_cmd(), 'SHARD_CONNECTION_STATES') env.assertTrue(False, message='Expected to fail') except Exception as e: @@ -4151,7 +4156,7 @@ def test_rq_job_without_topology(env:Env): # Now re-set the topology and call the debug command again env.expect('SEARCH.CLUSTERREFRESH').ok() # We should also see the effect of setting the number of workers - env.expect(debug_cmd(), 'SHARD_CONNECTION_STATES').equal([ANY, [ANY] * (workers + 1)] * env.shardsCount) + env.expect(debug_cmd(), 'SHARD_CONNECTION_STATES').equal([ANY, [ANY] * (compute_number_of_connections_in_each_ioruntime(workers + 1))] * env.shardsCount) @skip(cluster=False) # this test is only relevant on cluster diff --git a/tests/pytests/test_config.py b/tests/pytests/test_config.py index a7f3c6caecf..f8fbdd965a6 100644 --- a/tests/pytests/test_config.py +++ b/tests/pytests/test_config.py @@ -139,7 +139,7 @@ def testSetConfigOptionsErrors(env): env.expect(config_cmd(), 'set', 'BM25STD_TANH_FACTOR', -1).contains('Value is outside acceptable bounds') env.expect(config_cmd(), 'set', 'BM25STD_TANH_FACTOR', 10001).contains('BM25STD_TANH_FACTOR must be between 1 and 10000') env.expect(config_cmd(), 'set', '_BG_INDEX_OOM_PAUSE_TIME', -1).contains('Value is outside acceptable bounds') - env.expect(config_cmd(), 'set', '_BG_INDEX_OOM_PAUSE_TIME', UINT32_MAX+1).contains('Value is outside acceptable bounds') + env.expect(config_cmd(), 'set', '_BG_INDEX_OOM_PAUSE_TIME', UINT32_MAX+1).contains('Value is outside acceptable bounds') env.expect(config_cmd(), 'set', 'INDEXER_YIELD_EVERY_OPS', -1).contains('Value is outside acceptable bounds') @skip(cluster=True) @@ -854,6 +854,7 @@ def testModuleLoadexNumericParamsLastWins(): def testNumericArgDeprecationMessage(): moduleArgs = '' for configName, argName, default, minValue, maxValue, immutable, clusterConfig in numericConfigs: + # Since the IO threads are not lazily started, we cannot set the max number of shards and all that to the max values moduleArgs += f'{argName} {maxValue} ' env = Env(noDefaultModuleArgs=True, moduleArgs=moduleArgs) diff --git a/tests/pytests/test_multithread.py b/tests/pytests/test_multithread.py index 45f5fe3ec10..da4944e7828 100644 --- a/tests/pytests/test_multithread.py +++ b/tests/pytests/test_multithread.py @@ -417,8 +417,9 @@ def test_switch_loader_modes(): env.expect('FT.CURSOR', 'DEL', 'idx', cursor3).noError().ok() @skip(cluster=False) -def test_change_num_connections(env: Env): - +def test_change_num_connections(): + # TODO(Joan): This test is changed because it seems hard to accumulate all the Status from all the I/O threads. To be revisited + env = initEnv('SEARCH_IO_THREADS 20') # Validate the default values env.expect(config_cmd(), 'GET', 'WORKERS').equal([['WORKERS', '0']]) env.expect(config_cmd(), 'GET', 'CONN_PER_SHARD').equal([['CONN_PER_SHARD', '0']]) @@ -433,6 +434,10 @@ def test_change_num_connections(env: Env): # ['127.0.0.1:6379', ['Connected', 'Connected'], # '127.0.0.1:6381', ['Connected', 'Connecting'], # '127.0.0.1:6383', ['Connected', 'Connected']] + num_io_threads = 20 + def compute_number_of_connections_in_each_ioruntime(num_connections): + return max(1, num_connections // num_io_threads) + def expected(conns): return [ ANY, # The shard id (host:port) @@ -440,32 +445,42 @@ def expected(conns): ] * env.shardsCount # By default, the number of connections is 1 - env.expect(debug_cmd(), 'SHARD_CONNECTION_STATES').equal(expected(1)) + env.expect(debug_cmd(), 'SHARD_CONNECTION_STATES').equal(expected(compute_number_of_connections_in_each_ioruntime(1))) # Increase the number of worker threads to 6 env.expect(config_cmd(), 'SET', 'WORKERS', '6').ok() # The number of connections should be 7 - env.expect(debug_cmd(), 'SHARD_CONNECTION_STATES').equal(expected(7)) + env.expect(debug_cmd(), 'SHARD_CONNECTION_STATES').equal(expected(compute_number_of_connections_in_each_ioruntime(7))) # Set the number of connections to 4 env.expect(config_cmd(), 'SET', 'CONN_PER_SHARD', '4').ok() # The number of connections should be 4 - env.expect(debug_cmd(), 'SHARD_CONNECTION_STATES').equal(expected(4)) + env.expect(debug_cmd(), 'SHARD_CONNECTION_STATES').equal(expected(compute_number_of_connections_in_each_ioruntime(4))) # Decrease the number of worker threads to 5 env.expect(config_cmd(), 'SET', 'WORKERS', '5').ok() # The number of connections should remain 4 - env.expect(debug_cmd(), 'SHARD_CONNECTION_STATES').equal(expected(4)) + env.expect(debug_cmd(), 'SHARD_CONNECTION_STATES').equal(expected(compute_number_of_connections_in_each_ioruntime(4))) # Set the number of connections to 0 env.expect(config_cmd(), 'SET', 'CONN_PER_SHARD', '0').ok() # The number of connections should be 6 (5 worker threads + 1) - env.expect(debug_cmd(), 'SHARD_CONNECTION_STATES').equal(expected(6)) + env.expect(debug_cmd(), 'SHARD_CONNECTION_STATES').equal(expected(compute_number_of_connections_in_each_ioruntime(6))) # Set back the number of worker threads to 0 env.expect(config_cmd(), 'SET', 'WORKERS', '0').ok() # The number of connections should be 1 again - env.expect(debug_cmd(), 'SHARD_CONNECTION_STATES').equal(expected(1)) + env.expect(debug_cmd(), 'SHARD_CONNECTION_STATES').equal(expected(compute_number_of_connections_in_each_ioruntime(1))) + + # Set back Connection per shard to 40 + env.expect(config_cmd(), 'SET', 'CONN_PER_SHARD', '40').ok() + # The number of connections should be 40 again, but only 2 should be seen + env.expect(debug_cmd(), 'SHARD_CONNECTION_STATES').equal(expected(compute_number_of_connections_in_each_ioruntime(40))) + + # Set back Connection per shard to 40 + env.expect(config_cmd(), 'SET', 'CONN_PER_SHARD', '100').ok() + # The number of connections should be 40 again, but only 2 should be seen + env.expect(debug_cmd(), 'SHARD_CONNECTION_STATES').equal(expected(compute_number_of_connections_in_each_ioruntime(100))) def test_change_workers_number(): @@ -580,3 +595,184 @@ def get_RP_name(profile_res): env.assertEqual(get_RP_name(res), 'Threadsafe-Loader', message="Expected not to be optimized") res = env.cmd('FT.PROFILE', 'idx', 'AGGREGATE', 'QUERY', '*', 'LOAD', 1, '@not-sortable') env.assertEqual(get_RP_name(res), 'Threadsafe-Loader', message="Expected not to be optimized") + +def _test_ft_search_with_io_threads(io_threads): + """Helper function to test queries with specific IO thread count""" + # Create environment with specific IO thread count + env = initEnv(moduleArgs=f'SEARCH_IO_THREADS {io_threads}') + + # Create index + env.expect('FT.CREATE', 'idx', 'SCHEMA', 'txt', 'TEXT', 'num', 'NUMERIC', 'SORTABLE').ok() + + # Add test documents + conn = getConnectionByEnv(env) + doc_count = 100 + for i in range(doc_count): + conn.execute_command('HSET', f'doc:{i}', + 'txt', f'hello world document {i}', + 'num', i) + + # Run different query types and verify results + + # 1. Simple search + res = env.cmd('FT.SEARCH', 'idx', 'hello', 'NOCONTENT') + env.assertEqual(res[0], doc_count, message=f"Simple search with {io_threads} IO threads") + + # 2. Numeric range query + res = env.cmd('FT.SEARCH', 'idx', '@num:[10 50]', 'NOCONTENT') + env.assertEqual(res[0], 41, message=f"Numeric range query with {io_threads} IO threads") + + # 3. Combined query with sorting + res = env.cmd('FT.SEARCH', 'idx', 'world @num:[20 40]', 'SORTBY', 'num', 'DESC', 'NOCONTENT') + env.assertEqual(res[0], 21, message=f"Combined query with {io_threads} IO threads") + # Check sort order (first result should be doc:40) + env.assertEqual(res[1], 'doc:40', message=f"Sort order with {io_threads} IO threads") + + # 4. Aggregate query + res = env.cmd('FT.AGGREGATE', 'idx', '*', + 'GROUPBY', '1', '@num', + 'REDUCE', 'count', '0', 'AS', 'count', + 'FILTER', '@count > 0') + env.assertEqual(len(res), doc_count + 1, message=f"Aggregate query with {io_threads} IO threads") + + # Clean up for next iteration + env.cmd('FLUSHALL') + env.flush() + env.stop() + + +@skip(cluster=False) +def test_ft_search_with_coord_1_io_thread(): + _test_ft_search_with_io_threads(1) + +@skip(cluster=False) +def test_ft_search_with_coord_5_io_threads(): + _test_ft_search_with_io_threads(5) + +@skip(cluster=False) +def test_ft_search_with_coord_10_io_threads(): + _test_ft_search_with_io_threads(10) + + +def _test_ft_aggregate_with_io_threads(io_threads): + """Helper function to test aggregate queries with specific IO thread count""" + # Create environment with specific IO thread count + env = initEnv(moduleArgs=f'SEARCH_IO_THREADS {io_threads}') + + # Create index + env.expect('FT.CREATE', 'idx', 'SCHEMA', 'txt', 'TEXT', 'num', 'NUMERIC', 'SORTABLE', 'tag', 'TAG').ok() + + # Add test documents + conn = getConnectionByEnv(env) + doc_count = 100 + for i in range(doc_count): + tag_value = f"tag{i % 10}" # Create 10 different tag values + conn.execute_command('HSET', f'doc:{i}', + 'txt', f'hello world document {i}', + 'num', i, + 'tag', tag_value) + + # Run different aggregate queries and verify results + + # 1. Simple aggregate + res = env.cmd('FT.AGGREGATE', 'idx', '*') + # Check exact structure - 100 empty arrays after the counter + env.assertEqual(len(res), 101, message=f"Simple aggregate with {io_threads} IO threads") + env.assertEqual(res[1:], [[] for _ in range(100)], message=f"Simple aggregate structure with {io_threads} IO threads") + + # 2. Aggregate with LOAD + res = env.cmd('FT.AGGREGATE', 'idx', '*', 'LOAD', 1, '@num', 'LIMIT', 0, 10) + # Check exact number of results and structure + env.assertEqual(len(res), 11, message=f"Aggregate with LOAD with {io_threads} IO threads") + # Verify each result has the correct format ['num', value] + for i in range(1, len(res)): + env.assertEqual(len(res[i]), 2, message=f"Result format with {io_threads} IO threads") + env.assertEqual(res[i][0], 'num', message=f"Field name with {io_threads} IO threads") + + # 3. Aggregate with GROUPBY + res = env.cmd('FT.AGGREGATE', 'idx', '*', + 'GROUPBY', 1, '@tag', + 'REDUCE', 'COUNT', 0, 'AS', 'count') + # Check exact number of groups and their structure + env.assertEqual(len(res), 11, message=f"Aggregate with GROUPBY with {io_threads} IO threads") + for i in range(1, len(res)): + env.assertEqual(len(res[i]), 4, message=f"Group format with {io_threads} IO threads") + env.assertEqual(res[i][0], 'tag', message=f"Group field name with {io_threads} IO threads") + env.assertEqual(res[i][2], 'count', message=f"Count field name with {io_threads} IO threads") + env.assertEqual(res[i][3], '10', message=f"Group count with {io_threads} IO threads") + + # 4. Aggregate with SORTBY + res = env.cmd('FT.AGGREGATE', 'idx', '*', + 'SORTBY', 2, '@num', 'DESC', + 'LIMIT', 0, 5) + # Check exact number of results and descending order + env.assertEqual(len(res), 6, message=f"Aggregate with SORTBY with {io_threads} IO threads") + # Verify descending order of results + expected_values = ['99', '98', '97', '96', '95'] + for i in range(5): + env.assertEqual(res[i+1][1], expected_values[i], + message=f"Sort order at position {i} with {io_threads} IO threads") + + # 5. Aggregate with APPLY + res = env.cmd('FT.AGGREGATE', 'idx', '*', + 'LOAD', 1, '@num', + 'APPLY', '@num * 2', 'AS', 'doubled', + 'LIMIT', 0, 5) + # Check exact structure and calculated values + env.assertEqual(len(res), 6, message=f"Aggregate with APPLY with {io_threads} IO threads") + for i in range(1, len(res)): + env.assertEqual(len(res[i]), 4, message=f"Result format with {io_threads} IO threads") + env.assertEqual(res[i][0], 'num', message=f"First field name with {io_threads} IO threads") + env.assertEqual(res[i][2], 'doubled', message=f"Second field name with {io_threads} IO threads") + # Verify doubled value is correct + num_val = int(res[i][1]) + doubled_val = int(res[i][3]) + env.assertEqual(doubled_val, num_val * 2, + message=f"APPLY calculation for {num_val} with {io_threads} IO threads") + + # 6. Aggregate with FILTER + res = env.cmd('FT.AGGREGATE', 'idx', '*', + 'LOAD', 1, '@num', + 'FILTER', '@num > 90') + # Check exact number of results (9 documents with num > 90) + env.assertEqual(len(res), 10, message=f"Aggregate with FILTER with {io_threads} IO threads") + # Verify all values are > 90 + for i in range(1, len(res)): + env.assertGreater(int(res[i][1]), 90, + message=f"Filter condition with {io_threads} IO threads") + + # 7. Complex aggregate with multiple operations + res = env.cmd('FT.AGGREGATE', 'idx', '*', + 'GROUPBY', 1, '@tag', + 'REDUCE', 'AVG', 1, '@num', 'AS', 'avg_num', + 'REDUCE', 'COUNT', 0, 'AS', 'count', + 'SORTBY', 2, '@avg_num', 'DESC') + # Check exact number of groups and descending order of avg_num + env.assertEqual(len(res), 11, message=f"Complex aggregate with {io_threads} IO threads") + # Verify descending order of avg_num + for i in range(1, len(res)-1): + current_avg = float(res[i][5]) # Fixed: value is at index 5, not 4 + next_avg = float(res[i+1][5]) # Fixed: value is at index 5, not 4 + env.assertGreaterEqual(current_avg, next_avg, + message=f"Sort order of avg_num with {io_threads} IO threads") + + # Clean up for next iteration + env.cmd('FLUSHALL') + env.flush() + env.stop() + +@skip(cluster=False) +def test_ft_aggregate_with_coord_1_io_thread(): + _test_ft_aggregate_with_io_threads(1) + +@skip(cluster=False) +def test_ft_aggregate_with_coord_5_io_threads(): + _test_ft_aggregate_with_io_threads(5) + +@skip(cluster=False) +def test_ft_aggregate_with_coord_10_io_threads(): + _test_ft_aggregate_with_io_threads(10) + +@skip(cluster=False) +def test_ft_aggregate_with_coord_20_io_threads(): + _test_ft_aggregate_with_io_threads(20)