diff --git a/benchmark/nixlbench/contrib/Dockerfile b/benchmark/nixlbench/contrib/Dockerfile index d826b7f04d..e7cc9062c4 100644 --- a/benchmark/nixlbench/contrib/Dockerfile +++ b/benchmark/nixlbench/contrib/Dockerfile @@ -27,7 +27,7 @@ FROM ${BASE_IMAGE}:${BASE_IMAGE_TAG} AS os_setup_stage ARG ARCH="x86_64" ARG DEFAULT_PYTHON_VERSION RUN apt-get update -y && \ - DEBIAN_FRONTEND=noninteractive apt-get -y install \ + DEBIAN_FRONTEND=noninteractive apt-get -y install --no-install-recommends \ ninja-build \ pybind11-dev \ libclang-dev \ @@ -47,7 +47,11 @@ RUN apt-get update -y && \ libgtest-dev \ hwloc \ libhwloc-dev \ - build-essential + build-essential \ + libhiredis-dev \ + libevent-dev \ + redis-tools && \ + rm -rf /var/lib/apt/lists/* # Add DOCA repo + upgrade always so the RDMA reinstall below resolves MLNX/DOCA versions. # Skip only the DOCA SDK install when the base image already ships it. @@ -123,6 +127,8 @@ ARG LIBFABRIC_VERSION="v1.21.0" ARG NPROC ARG ABSL_TAG="lts_2025_08_14" ARG GRPC_TAG="v1.73.0" +ARG NIXL_ENABLE_PLUGINS="" +ARG NIXL_STATIC_PLUGINS="" WORKDIR /workspace @@ -258,7 +264,9 @@ WORKDIR /workspace/nixl RUN rm -rf build && \ mkdir build && \ - meson setup build --prefix=/usr/local/nixl --buildtype=$BUILD_TYPE && \ + meson setup build --prefix=/usr/local/nixl --buildtype=$BUILD_TYPE \ + -Denable_plugins=${NIXL_ENABLE_PLUGINS} \ + -Dstatic_plugins=${NIXL_STATIC_PLUGINS} && \ cd build && \ ninja -j${NPROC:-$(nproc)} && \ ninja install diff --git a/benchmark/nixlbench/contrib/build.sh b/benchmark/nixlbench/contrib/build.sh index dbd2eea339..bedc2ece94 100755 --- a/benchmark/nixlbench/contrib/build.sh +++ b/benchmark/nixlbench/contrib/build.sh @@ -1,6 +1,6 @@ #!/bin/bash -# SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. # SPDX-License-Identifier: Apache-2.0 # # Licensed under the Apache License, Version 2.0 (the "License"); @@ -129,6 +129,9 @@ get_options() { missing_requirement $1 fi ;; + --redis-only) + BUILD_ARGS+=" --build-arg NIXL_ENABLE_PLUGINS=REDIS --build-arg NIXL_STATIC_PLUGINS=REDIS" + ;; --) shift break @@ -185,6 +188,7 @@ show_help() { echo " [--python-versions python versions to build for, comma separated]" echo " [--tag tag for image]" echo " [--arch [x86_64|aarch64] to select target architecture]" + echo " [--redis-only build NIXL with REDIS plugin only (static)]" exit 0 } diff --git a/benchmark/nixlbench/src/utils/meson.build b/benchmark/nixlbench/src/utils/meson.build index 3bb017ba0b..c8157e6a8b 100644 --- a/benchmark/nixlbench/src/utils/meson.build +++ b/benchmark/nixlbench/src/utils/meson.build @@ -25,7 +25,8 @@ utils_deps = [ openmp_dep, gflags_dep, tomlplusplus_dep, - dependency('dl', required: true) + dependency('dl', required: true), + dependency('hiredis', required: true), ] if cuda_available diff --git a/benchmark/nixlbench/src/utils/utils.cpp b/benchmark/nixlbench/src/utils/utils.cpp index 32942ea0b8..da01ee0aff 100644 --- a/benchmark/nixlbench/src/utils/utils.cpp +++ b/benchmark/nixlbench/src/utils/utils.cpp @@ -35,11 +35,130 @@ #include #include #include +#include #include "runtime/etcd/etcd_rt.h" #include "utils/neuron.h" #include "utils/utils.h" +namespace { + +bool +parseRedisBenchPort(int &port) { + port = 6379; + const char *port_env = getenv("REDIS_PORT"); + if (!port_env) { + return true; + } + + try { + size_t parsed = 0; + int parsed_port = std::stoi(port_env, &parsed); + if (parsed != std::string(port_env).size() || parsed_port <= 0 || parsed_port > 65535) { + std::cerr << "Invalid REDIS_PORT value: " << port_env << std::endl; + return false; + } + port = parsed_port; + return true; + } + catch (const std::exception &) { + std::cerr << "Invalid REDIS_PORT value: " << port_env << std::endl; + return false; + } +} + +redisContext * +connectRedisBench(const char *operation) { + const char *host = getenv("REDIS_HOST"); + if (!host) { + host = "127.0.0.1"; + } + + int port = 6379; + if (!parseRedisBenchPort(port)) { + return nullptr; + } + + struct timeval timeout = {5, 0}; + redisContext *ctx = redisConnectWithTimeout(host, port, timeout); + if (!ctx || ctx->err) { + std::cerr << "Redis connect failed for " << operation << ": " + << (ctx ? ctx->errstr : "null context") << std::endl; + if (ctx) { + redisFree(ctx); + } + return nullptr; + } + + return ctx; +} + +bool +authRedisBench(redisContext *ctx, const char *operation) { + const char *password = getenv("REDIS_PASSWORD"); + if (!password || password[0] == '\0') { + return true; + } + + redisReply *auth_reply = (redisReply *)redisCommand(ctx, "AUTH %s", password); + if (!auth_reply || auth_reply->type == REDIS_REPLY_ERROR) { + std::cerr << "Redis AUTH failed during " << operation << std::endl; + if (auth_reply) { + freeReplyObject(auth_reply); + } + return false; + } + freeReplyObject(auth_reply); + return true; +} + +bool +getRedisBench(size_t buffer_size, const std::string &key, void *buffer) { + if (buffer_size > 0 && !buffer) { + std::cerr << "Invalid Redis GET output buffer for key: " << key << std::endl; + return false; + } + + redisContext *ctx = connectRedisBench("consistency GET"); + if (!ctx) { + return false; + } + + if (!authRedisBench(ctx, "consistency GET")) { + redisFree(ctx); + return false; + } + + redisReply *reply = (redisReply *)redisCommand(ctx, "GET %b", key.data(), key.size()); + bool success = false; + if (reply && reply->type == REDIS_REPLY_STRING) { + const size_t reply_len = static_cast(reply->len); + if (reply_len == buffer_size) { + if (buffer_size > 0) { + memcpy(buffer, reply->str, buffer_size); + } + success = true; + } else { + std::cerr << "Redis GET size mismatch for key " << key << ": expected " << buffer_size + << " bytes, got " << reply_len << " bytes" << std::endl; + } + } else if (reply && reply->type == REDIS_REPLY_NIL) { + std::cerr << "Redis GET failed, key not found: " << key << std::endl; + } else if (reply && reply->type == REDIS_REPLY_ERROR) { + std::cerr << "Redis GET error for key " << key << ": " << reply->str << std::endl; + } else { + std::cerr << "Redis GET failed for key " << key << std::endl; + } + + if (reply) { + freeReplyObject(reply); + } + redisFree(ctx); + return success; +} + +} // namespace + // Define command line parameters #define NB_ARG_STRING(param_name, def_val, help_text) DEFINE_string(param_name, def_val, help_text) #define NB_ARG_BOOL(param_name, def_val, help_text) DEFINE_bool(param_name, def_val, help_text) @@ -63,7 +182,7 @@ NB_ARG_STRING(worker_type, XFERBENCH_WORKER_NIXL, "Type of worker [nixl, nvshmem NB_ARG_STRING(backend, XFERBENCH_BACKEND_UCX, "Name of NIXL backend [UCX, GDS, GDS_MT, POSIX, GPUNETIO, Mooncake, HF3FS, OBJ, " - "GUSLI, AZURE_BLOB] (only used with nixl worker)"); + "REDIS, GUSLI, AZURE_BLOB] (only used with nixl worker)"); NB_ARG_STRING(initiator_seg_type, XFERBENCH_SEG_TYPE_DRAM, "Type of memory segment for initiator [DRAM, VRAM]. Note: Storage backends always " @@ -679,8 +798,9 @@ xferBenchConfig::printConfig() { } printOption("Worker type (--worker_type=[nixl,nvshmem])", worker_type); if (worker_type == XFERBENCH_WORKER_NIXL) { - printOption("Backend (--backend=[UCX,GDS,GDS_MT,POSIX,Mooncake,HF3FS,OBJ,AZURE_BLOB])", - backend); + printOption( + "Backend (--backend=[UCX,GDS,GDS_MT,POSIX,Mooncake,HF3FS,OBJ,REDIS,GUSLI,AZURE_BLOB])", + backend); printOption("Enable pt (--enable_pt=[0,1])", std::to_string(enable_pt)); printOption("Progress threads (--progress_threads=N)", std::to_string(progress_threads)); printOption("Device list (--device_list=dev1,dev2,...)", device_list); @@ -823,6 +943,7 @@ xferBenchConfig::isStorageBackend() { XFERBENCH_BACKEND_HF3FS == xferBenchConfig::backend || XFERBENCH_BACKEND_POSIX == xferBenchConfig::backend || XFERBENCH_BACKEND_OBJ == xferBenchConfig::backend || + XFERBENCH_BACKEND_REDIS == xferBenchConfig::backend || XFERBENCH_BACKEND_GUSLI == xferBenchConfig::backend || XFERBENCH_BACKEND_AZURE_BLOB == xferBenchConfig::backend || XFERBENCH_BACKEND_INFINIA == xferBenchConfig::backend); @@ -1078,6 +1199,11 @@ xferBenchUtils::checkConsistency(std::vector> &iov_lis exit(EXIT_FAILURE); } close(fd); + } else if (xferBenchConfig::backend == XFERBENCH_BACKEND_REDIS) { + if (!getRedisBench(len, iov.metaInfo, addr)) { + std::cerr << "Failed to get Redis key: " << iov.metaInfo << std::endl; + exit(EXIT_FAILURE); + } } else { ssize_t rc = pread(iov.devId, addr, len, iov.addr); if (rc < 0) { @@ -1303,6 +1429,54 @@ xferBenchUtils::buildAwsCredentials() { return env_setup; } +bool +xferBenchUtils::putRedis(size_t buffer_size, const std::string &key) { + if (buffer_size == 0) { + std::cerr << "Invalid Redis seed size: 0" << std::endl; + return false; + } + + void *buf = nullptr; + const size_t align = xferBenchConfig::page_size > 0 ? xferBenchConfig::page_size : 4096; + if (posix_memalign(&buf, align, buffer_size) != 0) { + std::cerr << "Failed to allocate buffer for Redis seed data" << std::endl; + return false; + } + memset(buf, XFERBENCH_TARGET_BUFFER_ELEMENT, buffer_size); + + redisContext *ctx = connectRedisBench("seed SET"); + if (!ctx) { + free(buf); + return false; + } + + if (!authRedisBench(ctx, "seed SET")) { + redisFree(ctx); + free(buf); + return false; + } + + redisReply *reply = (redisReply *)redisCommand( + ctx, "SET %b %b", key.data(), (size_t)key.size(), buf, (size_t)buffer_size); + free(buf); + + bool ok = false; + if (reply && (reply->type == REDIS_REPLY_STATUS || reply->type == REDIS_REPLY_STRING)) { + ok = true; + std::cout << "Seeded Redis key for READ: " << key << std::endl; + } else if (reply && reply->type == REDIS_REPLY_ERROR) { + std::cerr << "Redis SET failed for key " << key << ": " << reply->str << std::endl; + } else { + std::cerr << "Redis SET failed for key " << key << std::endl; + } + + if (reply) { + freeReplyObject(reply); + } + redisFree(ctx); + return ok; +} + bool xferBenchUtils::putObj(size_t buffer_size, const std::string &name) { if (xferBenchConfig::backend == XFERBENCH_BACKEND_INFINIA) { diff --git a/benchmark/nixlbench/src/utils/utils.h b/benchmark/nixlbench/src/utils/utils.h index a0ed72ece8..4dca773b35 100644 --- a/benchmark/nixlbench/src/utils/utils.h +++ b/benchmark/nixlbench/src/utils/utils.h @@ -94,6 +94,7 @@ #define XFERBENCH_BACKEND_MOONCAKE "Mooncake" #define XFERBENCH_BACKEND_HF3FS "HF3FS" #define XFERBENCH_BACKEND_OBJ "OBJ" +#define XFERBENCH_BACKEND_REDIS "REDIS" #define XFERBENCH_BACKEND_GUSLI "GUSLI" #define XFERBENCH_BACKEND_UCCL "UCCL" #define XFERBENCH_BACKEND_AZURE_BLOB "AZURE_BLOB" @@ -376,6 +377,8 @@ class xferBenchUtils { static bool putObj(size_t buffer_size, const std::string &name); static bool + putRedis(size_t buffer_size, const std::string &key); + static bool getObj(const std::string &name); static bool rmObj(const std::string &name); diff --git a/benchmark/nixlbench/src/worker/nixl/nixl_worker.cpp b/benchmark/nixlbench/src/worker/nixl/nixl_worker.cpp index 8b88a320db..d72d1a1867 100644 --- a/benchmark/nixlbench/src/worker/nixl/nixl_worker.cpp +++ b/benchmark/nixlbench/src/worker/nixl/nixl_worker.cpp @@ -263,6 +263,10 @@ xferBenchNixlWorker::xferBenchNixlWorker(const std::vector &devices } else { std::cout << "OBJ backend with standard S3 enabled" << std::endl; } + } else if (0 == xferBenchConfig::backend.compare(XFERBENCH_BACKEND_REDIS)) { + // REDIS backend: host/port/password via REDIS_HOST, REDIS_PORT, REDIS_PASSWORD + std::cout << "REDIS backend configured (using defaults or environment variables)" + << std::endl; } else if (0 == xferBenchConfig::backend.compare(XFERBENCH_BACKEND_GUSLI)) { // GUSLI backend requires direct I/O - enable it automatically if (!xferBenchConfig::storage_enable_direct) { @@ -1021,6 +1025,33 @@ xferBenchNixlWorker::allocateMemory(int num_threads) { CHECK_NIXL_ERROR(agent->registerMem(desc_list, &opt_args), "registerMem failed"); remote_regs_.emplace_back(*agent, backend_engine, BLK_SEG, std::move(iov_list)); } + } else if (xferBenchConfig::backend == XFERBENCH_BACKEND_REDIS) { + struct timeval tv; + gettimeofday(&tv, nullptr); + uint64_t timestamp = tv.tv_sec * 1000000ULL + tv.tv_usec; + + for (int list_idx = 0; list_idx < num_threads; list_idx++) { + std::vector iov_list; + for (i = 0; i < num_devices; i++) { + std::string unique_name = "nixlbench_redis" + std::to_string(list_idx) + "_" + + std::to_string(i) + "_" + std::to_string(timestamp); + + if (xferBenchConfig::op_type == XFERBENCH_OP_READ) { + const size_t seed_size = xferBenchConfig::max_block_size; + if (!xferBenchUtils::putRedis(seed_size, unique_name)) { + std::cerr << "Failed to seed Redis key: " << unique_name << std::endl; + exit(EXIT_FAILURE); + } + } + + xferBenchIOV redis_desc(0, buffer_size, i, unique_name); + std::cout << "Creating Redis key: " << unique_name << std::endl; + iov_list.push_back(redis_desc); + } + nixl_reg_dlist_t desc_list = iovListToNixlRegDlist(iov_list, DRAM_SEG); + CHECK_NIXL_ERROR(agent->registerMem(desc_list, &opt_args), "registerMem failed"); + remote_regs_.emplace_back(*agent, backend_engine, DRAM_SEG, std::move(iov_list)); + } } else if (xferBenchConfig::isStorageBackend()) { int num_buffers = num_threads * num_devices; int num_files = xferBenchConfig::num_files; @@ -1227,6 +1258,11 @@ xferBenchNixlWorker::exchangeIOV(const std::vector> &l if (basic_desc) { remote_iov_list.push_back(basic_desc.value()); } + } else if (XFERBENCH_BACKEND_REDIS == xferBenchConfig::backend) { + xferBenchIOV redis_remote(iov); + redis_remote.addr = 0; + redis_remote.len = block_size; + remote_iov_list.push_back(redis_remote); } else if (XFERBENCH_BACKEND_GUSLI == xferBenchConfig::backend) { xferBenchIOV iov_remote(iov); iov_remote.addr = gusli_devices[devidx].dev_offset + file_offset; @@ -1316,6 +1352,8 @@ prepareTransferDescriptors(nixl_xfer_dlist_t &local_desc, // Set remote descriptor type based on backend if (xferBenchConfig::isObjStorageBackend()) { remote_desc = nixl_xfer_dlist_t(OBJ_SEG); + } else if (XFERBENCH_BACKEND_REDIS == xferBenchConfig::backend) { + remote_desc = nixl_xfer_dlist_t(DRAM_SEG); } else if (XFERBENCH_BACKEND_GUSLI == xferBenchConfig::backend) { remote_desc = nixl_xfer_dlist_t(BLK_SEG); } else if (xferBenchConfig::isStorageBackend()) { @@ -1330,6 +1368,8 @@ static nixl_mem_t getRemoteSegType() { if (xferBenchConfig::isObjStorageBackend()) { return OBJ_SEG; + } else if (XFERBENCH_BACKEND_REDIS == xferBenchConfig::backend) { + return DRAM_SEG; } else if (XFERBENCH_BACKEND_GUSLI == xferBenchConfig::backend) { return BLK_SEG; } else if (xferBenchConfig::isStorageBackend()) { diff --git a/meson.build b/meson.build index 6e1f7ad7ca..64dc3bd3ad 100644 --- a/meson.build +++ b/meson.build @@ -29,7 +29,7 @@ if enable_plugins_opt != '' and disable_plugins_opt != '' error('Cannot specify both enable_plugins and disable_plugins options') endif -all_plugins = ['UCX', 'LIBFABRIC', 'POSIX', 'OBJ', 'GDS', 'GDS_MT', 'MOONCAKE', 'HF3FS', 'GUSLI', 'GPUNETIO', 'UCCL', 'AZURE_BLOB', 'INFINIA', 'TELEMETRY_DOCA'] +all_plugins = ['UCX', 'LIBFABRIC', 'POSIX', 'OBJ', 'REDIS', 'GDS', 'GDS_MT', 'MOONCAKE', 'HF3FS', 'GUSLI', 'GPUNETIO', 'UCCL', 'AZURE_BLOB', 'INFINIA', 'TELEMETRY_DOCA'] enabled_plugins = {} diff --git a/src/core/meson.build b/src/core/meson.build index ee542bf94d..ec305afdc4 100644 --- a/src/core/meson.build +++ b/src/core/meson.build @@ -42,6 +42,10 @@ if 'OBJ' in static_plugins nixl_lib_deps += [ obj_backend_interface ] endif +if 'REDIS' in static_plugins + nixl_lib_deps += [ redis_backend_interface ] +endif + disable_gds_backend = get_option('disable_gds_backend') if not disable_gds_backend and 'GDS' in static_plugins nixl_lib_deps += [ gds_backend_interface, cuda_dep ] diff --git a/src/core/nixl_plugin_manager.cpp b/src/core/nixl_plugin_manager.cpp index 2a12915b33..3ce62748c3 100644 --- a/src/core/nixl_plugin_manager.cpp +++ b/src/core/nixl_plugin_manager.cpp @@ -847,6 +847,10 @@ void nixlPluginManager::registerBuiltinPlugins() { NIXL_REGISTER_STATIC_PLUGIN(Backend, OBJ) #endif +#ifdef STATIC_PLUGIN_REDIS + NIXL_REGISTER_STATIC_PLUGIN(Backend, REDIS) +#endif + #ifdef STATIC_PLUGIN_MOONCAKE NIXL_REGISTER_STATIC_PLUGIN(Backend, MOONCAKE) #endif diff --git a/src/plugins/meson.build b/src/plugins/meson.build index ce7103cb63..a7b2eb15dd 100644 --- a/src/plugins/meson.build +++ b/src/plugins/meson.build @@ -29,6 +29,10 @@ if enabled_plugins.get('OBJ') subdir('obj') endif +if enabled_plugins.get('REDIS') + subdir('redis') +endif + if enabled_plugins.get('AZURE_BLOB') subdir('azure_blob') endif diff --git a/src/plugins/redis/README.md b/src/plugins/redis/README.md new file mode 100644 index 0000000000..6117e01eb2 --- /dev/null +++ b/src/plugins/redis/README.md @@ -0,0 +1,259 @@ + + +# NIXL Redis Plugin + +The `REDIS` plugin exposes Redis as a local NIXL storage backend. It supports + +- asynchronous `NIXL_WRITE`/`NIXL_READ` transfers and +- synchronous `queryMem()` checks. + +## Architecture + +The plugin follows the direct-backend layout similar to existing backends: + +```text +src/plugins/redis/ + redis_backend.h/.cpp nixlRedisKVEngine and NIXL transfer logic + redis_client.h/.cpp hiredis client and its unit-test interface + redis_plugin.cpp NIXL plugin registration + meson.build REDIS plugin target + README.md +``` + +`nixlRedisKVEngine` derives directly from `nixlBackendEngine`. The internal `iRedisClient` +interface isolates hiredis/libevent and permits Redis-free unit tests; it is not a generic KV +extension API. + +The production client uses a `RedisConnectionPool` (default: 8 slots). Each slot holds: + +- One hiredis/libevent async connection for `SET` and `GET`. +- One blocking hiredis connection for `EXISTS`. +- One libevent thread. Hiredis callbacks complete NIXL request promises directly. + +Operations are dispatched round-robin across pool slots to saturate multiple Redis server +threads and reduce per-slot head-of-line blocking. + +## Dependencies + +- `hiredis` with async API support +- `libevent` +- `libevent_pthreads` + +If REDIS is explicitly enabled or selected as a static plugin, missing hiredis/libevent is a +configuration error. During a default all-plugin build, missing dependencies cause REDIS to be +skipped with a warning. + +## Configuration + +`RedisConfig` is the Redis client's resolved, internal configuration value. The backend calls +`RedisConfig::fromBackendParams()` once during construction and passes the result to +`RedisConnectionPool`; the client does not repeatedly read backend parameters or environment +variables. + +```cpp +struct RedisConfig { + std::string host = "localhost"; + int port = 6379; + std::string username; + std::string password; + int db = 0; + int pool_size = 8; +}; +``` + +Each setting is resolved in this precedence order: + +1. A valid value in the NIXL backend parameter map. +2. The corresponding `REDIS_*` environment variable, when one exists. +3. The built-in default. + +An explicitly provided string value, including an empty username or password, takes precedence +over its environment fallback. Port values that fail validation fall through to `REDIS_PORT` and +then to `6379`; database parse failures fall back to `0`. The logical database currently has no +environment fallback. Invalid `pool_size` values fall back to `8`. + +| Parameter | Environment fallback | Default | Description | +|-----------|----------------------|---------|-------------| +| `host` | `REDIS_HOST` | `localhost` | Redis hostname or IP address | +| `port` | `REDIS_PORT` | `6379` | Redis TCP port | +| `username` | `REDIS_USERNAME` | empty | Redis ACL username | +| `password` | `REDIS_PASSWORD` | empty | Redis AUTH password | +| `db` | none | `0` | Redis logical database | +| `pool_size` | `REDIS_POOL_SIZE` | `8` | Number of concurrent Redis connections | + +Authentication behavior is determined by the resolved credentials: + +- When `password` is empty, the client does not send `AUTH`. This is correct for a Redis server + that allows unauthenticated access. A password-protected server will reject subsequent commands. +- When only `password` is set, the client sends legacy/default-user `AUTH password`. +- When both `username` and `password` are set, the client sends ACL-style + `AUTH username password`. A username without a password is rejected during backend creation. + +```cpp +nixl_b_params_t params = { + {"host", "127.0.0.1"}, + {"port", "6379"}, + {"username", "nixl"}, + {"password", "example-password"}, + {"db", "2"}, + {"pool_size", "8"}, +}; +agent.createBackend("REDIS", params); +``` + +## Transfer behavior + +| NIXL operation | Redis operation | Result | +|----------------|-----------------|--------| +| `NIXL_WRITE` | `SET key bytes` | Stores local DRAM bytes | +| `NIXL_READ` | `GET key` | Copies the exact Redis value into local DRAM | +| `queryMem` | `EXISTS key` | Reports whether the key exists | + +Local descriptors must be `DRAM_SEG`; remote descriptors may be `DRAM_SEG` or `OBJ_SEG`. A Redis +key comes from descriptor `metaInfo` when present, otherwise from the decimal `devId`. `postXfer()` +resolves every remote key before dispatching commands, so an invalid descriptor cannot produce a +partially submitted transfer. + +`postXfer()` returns `NIXL_IN_PROG`; `checkXfer()` polls the request futures and returns success or +the first backend error once all completed work has been observed. + +## Build and test + +Build the plugin: + +```bash +meson setup build -Denable_plugins=REDIS +meson compile -C build REDIS +``` + +NIXL does not add tests to a `release` build, even when `build_tests` is enabled. To build and run +the Redis unit tests, install GoogleTest and GoogleMock and configure a separate non-release build: + +```bash +# Debian/Ubuntu +sudo apt-get install libgtest-dev libgmock-dev + +meson setup build-redis-tests \ + --buildtype=debug \ + -Dbuild_tests=true \ + -Denable_plugins=REDIS +meson compile -C build-redis-tests unit +meson devenv -C build-redis-tests \ + ./test/gtest/unit/unit \ + '--gtest_filter=redisConfigTest.*:redisEngineTest.*' +``` + +The filter runs only the Redis configuration and backend suites. A successful run currently +reports 11 tests from 2 test suites. These tests inject `mockRedisClient` and do not require a +running Redis server; use the live smoke test below to exercise hiredis and a real server. + +The dynamic plugin is produced at: + +```text +build/src/plugins/redis/libplugin_REDIS.so +``` + +For a static build: + +```bash +meson setup build -Denable_plugins=REDIS -Dstatic_plugins=REDIS +meson compile -C build +``` + +## Live nixlbench smoke test + +The unit tests use an injected Redis client and do not connect to a server. Use this smoke test to +exercise plugin creation, hiredis/libevent, and an actual Redis write/read data path. The example +uses a static REDIS plugin so the standalone nixlbench binary does not depend on dynamic plugin +discovery. + +Start an isolated Redis server on the loopback interface and verify that it is ready: + +```bash +docker run --detach --rm --name nixl-redis-smoke \ + --publish 127.0.0.1:6379:6379 redis:7-alpine +docker exec nixl-redis-smoke redis-cli PING +``` + +The readiness command must print `PONG`. + +Install a static REDIS-enabled NIXL build into a temporary prefix, then build nixlbench against +that installation: + +```bash +export NIXL_PREFIX=/tmp/nixl-redis-smoke-install + +meson setup build-redis \ + --prefix="$NIXL_PREFIX" \ + -Denable_plugins=REDIS \ + -Dstatic_plugins=REDIS +meson compile -C build-redis +meson install -C build-redis + +meson setup build-nixlbench benchmark/nixlbench \ + -Dnixl_path="$NIXL_PREFIX" +meson compile -C build-nixlbench +build-nixlbench/nixlbench --help | grep REDIS +``` + +The NIXL build requires the plugin dependencies listed above. The standalone nixlbench build also +requires the hiredis development package because it seeds Redis before a READ and independently +checks transferred data when consistency checking is enabled. + +Run fixed-size 4 KiB WRITE and READ benchmarks. These commands use one thread, one descriptor per +batch, one in-flight request, 32 warm-up iterations, and 208 measured iterations: + +```bash +export REDIS_HOST=127.0.0.1 +export REDIS_PORT=6379 +export LD_LIBRARY_PATH="$NIXL_PREFIX/lib/$(uname -m)-linux-gnu${LD_LIBRARY_PATH:+:$LD_LIBRARY_PATH}" + +build-nixlbench/nixlbench \ + --backend=REDIS \ + --runtime_type=ASIO \ + --op_type=WRITE \ + --start_block_size=4096 \ + --max_block_size=4096 \ + --start_batch_size=1 \ + --max_batch_size=1 \ + --pipeline_depth=1 \ + --num_threads=1 \ + --total_buffer_size=67108864 \ + --warmup_iter=32 \ + --num_iter=208 \ + --check_consistency=true + +build-nixlbench/nixlbench \ + --backend=REDIS \ + --runtime_type=ASIO \ + --op_type=READ \ + --start_block_size=4096 \ + --max_block_size=4096 \ + --start_batch_size=1 \ + --max_batch_size=1 \ + --pipeline_depth=1 \ + --num_threads=1 \ + --total_buffer_size=67108864 \ + --warmup_iter=32 \ + --num_iter=208 \ + --check_consistency=true +``` + +Each command must exit successfully and print a result row for block size `4096`. The READ run +must also print `Seeded Redis key for READ`; with `--check_consistency=true`, nixlbench reports an +error and exits unsuccessfully if the transferred data differs. The reported bandwidth and latency +are loopback smoke-test measurements, not production performance results. + +The example uses Redis database 0 without authentication. `REDIS_PASSWORD` may be set for a +password-protected default user. nixlbench's direct seed/consistency helper does not currently +support `REDIS_USERNAME` or selecting a nonzero `db`, even though the plugin itself supports both. + +Inspect the created keys if desired, then stop the server: + +```bash +docker exec nixl-redis-smoke redis-cli DBSIZE +docker stop nixl-redis-smoke +``` diff --git a/src/plugins/redis/meson.build b/src/plugins/redis/meson.build new file mode 100644 index 0000000000..3b01d8304a --- /dev/null +++ b/src/plugins/redis/meson.build @@ -0,0 +1,105 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +if not enabled_plugins.get('REDIS') + subdir_done() +endif + +redis_sources = [ + 'redis_backend.cpp', + 'redis_backend.h', + 'redis_client.cpp', + 'redis_client.h', + 'redis_plugin.cpp', +] + +hiredis_async_dep = dependency('hiredis_async', required: false) +libevent_dep = dependency('libevent', required: false) +libevent_pthreads_dep = dependency('libevent_pthreads', required: false) + +if not hiredis_async_dep.found() + hiredis_async_dep = dependency('hiredis', required: false) +endif + +if not hiredis_async_dep.found() or not libevent_dep.found() + cc = meson.get_compiler('cpp') + hiredis_lib = cc.find_library('hiredis', required: false) + libevent_lib = cc.find_library('event', required: false) + libevent_pthreads_lib = cc.find_library('event_pthreads', required: false) + + if hiredis_lib.found() and libevent_lib.found() + if not hiredis_async_dep.found() + hiredis_async_dep = hiredis_lib + endif + if not libevent_dep.found() + libevent_dep = libevent_lib + endif + if libevent_pthreads_lib.found() and not libevent_pthreads_dep.found() + libevent_pthreads_dep = libevent_pthreads_lib + endif + endif +endif + +if not hiredis_async_dep.found() or not libevent_dep.found() + if is_explicit_enable or 'REDIS' in static_plugins + error('REDIS plugin requested but hiredis and libevent are required. Install libhiredis-dev and libevent-dev (or equivalent), then reconfigure.') + endif + warning('hiredis-async or libevent not available, skipping REDIS plugin build') + subdir_done() +endif + +compile_defs_redis = [] +if is_variable('compile_defs') + compile_defs_redis = compile_defs +endif +compile_defs_redis += ['-DHAVE_HIREDIS_ASYNC'] + +redis_plugin_deps = [nixl_infra, nixl_common_dep, hiredis_async_dep, libevent_dep] +if libevent_pthreads_dep.found() + redis_plugin_deps += [libevent_pthreads_dep] +else + cc = meson.get_compiler('cpp') + event_pthreads = cc.find_library('event_pthreads', required: false) + if event_pthreads.found() + redis_plugin_deps += [event_pthreads] + else + warning('libevent_pthreads not found; REDIS plugin may fail to link (evthread_use_pthreads)') + endif +endif + +if not is_variable('compile_flags') + compile_flags = [] +endif + +if 'REDIS' in static_plugins + redis_backend_lib = static_library( + 'REDIS', + redis_sources, + dependencies: redis_plugin_deps, + cpp_args: compile_defs_redis + compile_flags, + include_directories: [nixl_inc_dirs, utils_inc_dirs], + install: false, + name_prefix: 'libplugin_') +else + redis_backend_lib = shared_library( + 'REDIS', + redis_sources, + dependencies: redis_plugin_deps, + cpp_args: compile_defs_redis + compile_flags + ['-fPIC'], + include_directories: [nixl_inc_dirs, utils_inc_dirs], + install: true, + name_prefix: 'libplugin_', + install_dir: plugin_install_dir, + install_rpath: '$ORIGIN/..') + if get_option('buildtype') == 'debug' + run_command('sh', '-c', + 'echo "REDIS=' + redis_backend_lib.full_path() + '" >> ' + plugin_build_dir + '/pluginlist', + check: true + ) + endif +endif + +redis_backend_interface = declare_dependency( + link_with: redis_backend_lib, + dependencies: redis_plugin_deps, +) diff --git a/src/plugins/redis/redis_backend.cpp b/src/plugins/redis/redis_backend.cpp new file mode 100644 index 0000000000..9955a0b218 --- /dev/null +++ b/src/plugins/redis/redis_backend.cpp @@ -0,0 +1,285 @@ +/* + * SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. + * SPDX-License-Identifier: Apache-2.0 + */ + +#include "redis_backend.h" + +#include "common/nixl_log.h" + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace { + +nixl_b_params_t * +getInitCustomParams(const nixlBackendInitParams *init_params) { + return init_params ? init_params->customParams : nullptr; +} + +bool +isValidPrepXferParams(const nixl_xfer_op_t &operation, + const nixl_meta_dlist_t &local, + const nixl_meta_dlist_t &remote, + const std::string &remote_agent, + const std::string &local_agent) { + if (operation != NIXL_WRITE && operation != NIXL_READ) { + NIXL_ERROR << absl::StrFormat("Error: Invalid operation type: %d", operation); + return false; + } + + if (local.descCount() == 0) { + NIXL_ERROR << "Error: Transfer descriptor lists must not be empty"; + return false; + } + + if (local.descCount() != remote.descCount()) { + NIXL_ERROR << absl::StrFormat( + "Error: Local and remote descriptor counts must match (%d != %d)", + local.descCount(), + remote.descCount()); + return false; + } + + if (remote_agent != local_agent) { + NIXL_WARN << absl::StrFormat( + "Warning: Remote agent doesn't match the requesting agent (%s). Got %s", + local_agent, + remote_agent); + } + + if (local.getType() != DRAM_SEG) { + NIXL_ERROR << absl::StrFormat("Error: Local memory type must be DRAM_SEG, got %d", + local.getType()); + return false; + } + + if (remote.getType() != DRAM_SEG && remote.getType() != OBJ_SEG) { + NIXL_ERROR << absl::StrFormat( + "Error: Remote memory type must be DRAM_SEG or OBJ_SEG, got %d", remote.getType()); + return false; + } + + return true; +} + +class nixlRedisBackendReqH : public nixlBackendReqH { +public: + std::vector>> statusPromises_; + std::vector> statusFutures_; + + nixl_status_t + getOverallStatus() { + while (!statusFutures_.empty()) { + if (statusFutures_.back().wait_for(std::chrono::seconds(0)) != + std::future_status::ready) { + return NIXL_IN_PROG; + } + + auto current_status = statusFutures_.back().get(); + if (current_status != NIXL_SUCCESS) { + statusFutures_.clear(); + statusPromises_.clear(); + return current_status; + } + statusFutures_.pop_back(); + statusPromises_.pop_back(); + } + return NIXL_SUCCESS; + } +}; + +class nixlRedisMetadata : public nixlBackendMD { +public: + nixlRedisMetadata(nixl_mem_t nixl_mem, uint64_t dev_id, std::string redis_key) + : nixlBackendMD(true), + nixlMem(nixl_mem), + devId(dev_id), + redisKey(std::move(redis_key)) {} + + nixl_mem_t nixlMem; + uint64_t devId; + std::string redisKey; +}; + +} // namespace + +nixlRedisKVEngine::nixlRedisKVEngine(const nixlBackendInitParams *init_params) + : nixlBackendEngine(init_params) { + auto config = RedisConfig::fromBackendParams(getInitCustomParams(init_params)); + redisClient_ = std::make_shared(std::move(config)); + NIXL_INFO << "Redis backend initialized"; +} + +nixlRedisKVEngine::nixlRedisKVEngine(const nixlBackendInitParams *init_params, + std::shared_ptr redis_client) + : nixlBackendEngine(init_params), + redisClient_(std::move(redis_client)) {} + +nixl_status_t +nixlRedisKVEngine::registerMem(const nixlBlobDesc &mem, + const nixl_mem_t &nixl_mem, + nixlBackendMD *&out) { + const auto supported_mems = getSupportedMems(); + if (std::find(supported_mems.begin(), supported_mems.end(), nixl_mem) == supported_mems.end()) { + out = nullptr; + return NIXL_ERR_NOT_SUPPORTED; + } + + std::string redis_key = mem.metaInfo.empty() ? std::to_string(mem.devId) : mem.metaInfo; + auto redis_md = std::make_unique(nixl_mem, mem.devId, redis_key); + devIdToRedisKey_[mem.devId] = redis_key; + out = redis_md.release(); + return NIXL_SUCCESS; +} + +nixl_status_t +nixlRedisKVEngine::deregisterMem(nixlBackendMD *meta) { + auto *redis_md = static_cast(meta); + if (redis_md) { + std::unique_ptr redis_md_ptr(redis_md); + devIdToRedisKey_.erase(redis_md->devId); + } + return NIXL_SUCCESS; +} + +nixl_status_t +nixlRedisKVEngine::queryMem(const nixl_reg_dlist_t &descs, + std::vector &resp) const { + if (!redisClient_) { + NIXL_ERROR << "Failed to query memory: no Redis client available"; + return NIXL_ERR_BACKEND; + } + + resp.clear(); + resp.reserve(descs.descCount()); + bool has_error = false; + + try { + for (int i = 0; i < descs.descCount(); ++i) { + const auto &desc = descs[i]; + const std::string key = + desc.metaInfo.empty() ? std::to_string(desc.devId) : desc.metaInfo; + const auto exists = redisClient_->checkKeyExistsSync(key); + if (!exists.has_value()) { + resp.emplace_back(std::nullopt); + has_error = true; + } else { + resp.emplace_back(*exists ? nixl_query_resp_t{nixl_b_params_t{}} : std::nullopt); + } + } + } + catch (const std::exception &e) { + NIXL_ERROR << "Failed to query memory: " << e.what(); + return NIXL_ERR_BACKEND; + } + + return has_error ? NIXL_ERR_BACKEND : NIXL_SUCCESS; +} + +nixl_status_t +nixlRedisKVEngine::prepXfer(const nixl_xfer_op_t &operation, + const nixl_meta_dlist_t &local, + const nixl_meta_dlist_t &remote, + const std::string &remote_agent, + nixlBackendReqH *&handle, + const nixl_opt_b_args_t *) const { + if (!isValidPrepXferParams(operation, local, remote, remote_agent, localAgent)) { + return NIXL_ERR_INVALID_PARAM; + } + + handle = new nixlRedisBackendReqH(); + return NIXL_SUCCESS; +} + +nixl_status_t +nixlRedisKVEngine::postXfer(const nixl_xfer_op_t &operation, + const nixl_meta_dlist_t &local, + const nixl_meta_dlist_t &remote, + const std::string &, + nixlBackendReqH *&handle, + const nixl_opt_b_args_t *) const { + if (!handle || (operation != NIXL_WRITE && operation != NIXL_READ)) { + return NIXL_ERR_INVALID_PARAM; + } + + if (local.descCount() == 0 || local.descCount() != remote.descCount()) { + NIXL_ERROR << absl::StrFormat( + "Invalid transfer descriptor counts for Redis postXfer (%d local, %d remote)", + local.descCount(), + remote.descCount()); + return NIXL_ERR_INVALID_PARAM; + } + + if (!redisClient_) { + return NIXL_ERR_BACKEND; + } + + auto *req_h = static_cast(handle); + req_h->statusFutures_.clear(); + req_h->statusPromises_.clear(); + + // Resolve every key before dispatching any command so invalid descriptors cannot + // produce a partially submitted Redis transfer. + std::vector redis_keys; + redis_keys.reserve(remote.descCount()); + for (int i = 0; i < remote.descCount(); ++i) { + const auto &remote_desc = remote[i]; + std::string redis_key; + + if (remote_desc.metadataP) { + auto *redis_md = dynamic_cast(remote_desc.metadataP); + if (redis_md) { + redis_key = redis_md->redisKey; + } + } + + if (redis_key.empty()) { + auto it = devIdToRedisKey_.find(remote_desc.devId); + if (it == devIdToRedisKey_.end()) { + return NIXL_ERR_INVALID_PARAM; + } + redis_key = it->second; + } + redis_keys.push_back(std::move(redis_key)); + } + + for (int i = 0; i < local.descCount(); ++i) { + const auto &local_desc = local[i]; + auto status_promise = std::make_shared>(); + req_h->statusPromises_.push_back(status_promise); + req_h->statusFutures_.push_back(status_promise->get_future()); + + if (operation == NIXL_WRITE) { + redisClient_->putKeyAsync( + redis_keys[i], local_desc.addr, local_desc.len, status_promise); + } else { + redisClient_->getKeyAsync( + redis_keys[i], local_desc.addr, local_desc.len, status_promise); + } + } + + return NIXL_IN_PROG; +} + +nixl_status_t +nixlRedisKVEngine::checkXfer(nixlBackendReqH *handle) const { + if (!handle) { + return NIXL_ERR_INVALID_PARAM; + } + return static_cast(handle)->getOverallStatus(); +} + +nixl_status_t +nixlRedisKVEngine::releaseReqH(nixlBackendReqH *handle) const { + delete static_cast(handle); + return NIXL_SUCCESS; +} diff --git a/src/plugins/redis/redis_backend.h b/src/plugins/redis/redis_backend.h new file mode 100644 index 0000000000..5d879449e7 --- /dev/null +++ b/src/plugins/redis/redis_backend.h @@ -0,0 +1,109 @@ +/* + * SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. + * SPDX-License-Identifier: Apache-2.0 + */ + +#ifndef NIXL_SRC_PLUGINS_REDIS_REDIS_BACKEND_H +#define NIXL_SRC_PLUGINS_REDIS_REDIS_BACKEND_H + +#include "backend/backend_engine.h" +#include "redis_client.h" + +#include +#include +#include +#include + +inline constexpr const char *REDIS_PLUGIN_NAME = "REDIS"; +inline constexpr const char *REDIS_PLUGIN_VERSION = "0.1.0"; + +/** NIXL backend engine for Redis key-value storage. */ +class nixlRedisKVEngine : public nixlBackendEngine { +public: + explicit nixlRedisKVEngine(const nixlBackendInitParams *init_params); + nixlRedisKVEngine(const nixlBackendInitParams *init_params, + std::shared_ptr redis_client); + ~nixlRedisKVEngine() override = default; + + bool + supportsRemote() const override { + return false; + } + + bool + supportsLocal() const override { + return true; + } + + bool + supportsNotif() const override { + return false; + } + + nixl_mem_list_t + getSupportedMems() const override { + return {OBJ_SEG, DRAM_SEG}; + } + + nixl_status_t + registerMem(const nixlBlobDesc &mem, const nixl_mem_t &nixl_mem, nixlBackendMD *&out) override; + + nixl_status_t + deregisterMem(nixlBackendMD *meta) override; + + nixl_status_t + queryMem(const nixl_reg_dlist_t &descs, std::vector &resp) const override; + + nixl_status_t + connect(const std::string &) override { + return NIXL_SUCCESS; + } + + nixl_status_t + disconnect(const std::string &) override { + return NIXL_SUCCESS; + } + + nixl_status_t + unloadMD(nixlBackendMD *) override { + return NIXL_SUCCESS; + } + + nixl_status_t + loadLocalMD(nixlBackendMD *input, nixlBackendMD *&output) override { + if (!input) { + output = nullptr; + return NIXL_ERR_INVALID_PARAM; + } + output = input; + return NIXL_SUCCESS; + } + + nixl_status_t + prepXfer(const nixl_xfer_op_t &operation, + const nixl_meta_dlist_t &local, + const nixl_meta_dlist_t &remote, + const std::string &remote_agent, + nixlBackendReqH *&handle, + const nixl_opt_b_args_t *opt_args = nullptr) const override; + + nixl_status_t + postXfer(const nixl_xfer_op_t &operation, + const nixl_meta_dlist_t &local, + const nixl_meta_dlist_t &remote, + const std::string &remote_agent, + nixlBackendReqH *&handle, + const nixl_opt_b_args_t *opt_args = nullptr) const override; + + nixl_status_t + checkXfer(nixlBackendReqH *handle) const override; + + nixl_status_t + releaseReqH(nixlBackendReqH *handle) const override; + +private: + std::shared_ptr redisClient_; + std::unordered_map devIdToRedisKey_; +}; + +#endif // NIXL_SRC_PLUGINS_REDIS_REDIS_BACKEND_H diff --git a/src/plugins/redis/redis_client.cpp b/src/plugins/redis/redis_client.cpp new file mode 100644 index 0000000000..fb88e0d31a --- /dev/null +++ b/src/plugins/redis/redis_client.cpp @@ -0,0 +1,684 @@ +/* + * SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. + * SPDX-License-Identifier: Apache-2.0 + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include "redis_client.h" +#include "common/nixl_log.h" +#include +#include +#include +#include +#include +#include +#include +#include + +namespace { + +std::string +getStringSetting(const nixl_b_params_t *custom_params, + const char *param_name, + const char *env_name, + const char *default_value = "") { + if (custom_params) { + const auto it = custom_params->find(param_name); + if (it != custom_params->end()) { + return it->second; + } + } + const char *env_value = std::getenv(env_name); + return env_value ? std::string(env_value) : default_value; +} + +std::optional +parseRedisPort(const std::string &value, const char *source) { + try { + size_t parsed = 0; + int port = std::stoi(value, &parsed); + if (parsed != value.size() || port <= 0 || port > 65535) { + NIXL_WARN << absl::StrFormat( + "Invalid %s value '%s', using default 6379", source, value); + return std::nullopt; + } + return port; + } + catch (const std::exception &) { + NIXL_WARN << absl::StrFormat("Invalid %s value '%s', using default 6379", source, value); + return std::nullopt; + } +} + +int +getRedisPort(const nixl_b_params_t *custom_params) { + if (custom_params && custom_params->count("port") > 0) { + auto port = parseRedisPort(custom_params->at("port"), "Redis port"); + if (port) { + return *port; + } + } + const char *env_port = std::getenv("REDIS_PORT"); + if (env_port) { + auto port = parseRedisPort(env_port, "REDIS_PORT"); + if (port) { + return *port; + } + } + return 6379; +} + +int +getRedisDB(const nixl_b_params_t *custom_params) { + if (custom_params && custom_params->count("db") > 0) { + try { + return std::stoi(custom_params->at("db")); + } + catch (const std::exception &) { + NIXL_WARN << "Invalid db value, using default 0"; + } + } + return 0; +} + +std::optional +parseRedisPoolSize(const std::string &value, const char *source) { + try { + size_t parsed = 0; + int val = std::stoi(value, &parsed); + if (parsed != value.size() || val <= 0) { + NIXL_WARN << absl::StrFormat("Invalid %s value '%s', using default 8", source, value); + return std::nullopt; + } + return val; + } + catch (const std::exception &) { + NIXL_WARN << absl::StrFormat("Invalid %s value '%s', using default 8", source, value); + return std::nullopt; + } +} + +int +getRedisPoolSize(const nixl_b_params_t *custom_params) { + if (custom_params && custom_params->count("pool_size") > 0) { + auto val = parseRedisPoolSize(custom_params->at("pool_size"), "pool_size"); + if (val) { + return *val; + } + } + const char *env_val = std::getenv("REDIS_POOL_SIZE"); + if (env_val) { + auto val = parseRedisPoolSize(env_val, "REDIS_POOL_SIZE"); + if (val) { + return *val; + } + } + return 8; +} + +} // namespace + +RedisConfig +RedisConfig::fromBackendParams(const nixl_b_params_t *custom_params) { + RedisConfig config; + config.host = getStringSetting(custom_params, "host", "REDIS_HOST", "localhost"); + config.port = getRedisPort(custom_params); + config.username = getStringSetting(custom_params, "username", "REDIS_USERNAME"); + config.password = getStringSetting(custom_params, "password", "REDIS_PASSWORD"); + config.db = getRedisDB(custom_params); + config.pool_size = getRedisPoolSize(custom_params); + if (!config.username.empty() && config.password.empty()) { + throw std::invalid_argument("Redis username requires a password"); + } + return config; +} + +#ifdef HAVE_HIREDIS_ASYNC + +namespace { + +using redis_event_task_t = std::function; + +void +runEventTask(evutil_socket_t, short, void *arg) { + std::unique_ptr task(static_cast(arg)); + (*task)(); +} + +bool +checkRedisReplyOk(redisReply *reply, const char *command) { + if (!reply) { + NIXL_ERROR << absl::StrFormat("Redis %s: no reply", command); + return false; + } + if (reply->type == REDIS_REPLY_ERROR) { + NIXL_ERROR << absl::StrFormat("Redis %s error: %s", command, reply->str); + return false; + } + if (reply->type == REDIS_REPLY_STATUS && strcmp(reply->str, "OK") != 0) { + NIXL_ERROR << absl::StrFormat("Redis %s unexpected status: %s", command, reply->str); + return false; + } + return true; +} + +void +setPromiseStatus(const std::shared_ptr> &promise, bool success) { + if (!promise) { + return; + } + promise->set_value(success ? NIXL_SUCCESS : NIXL_ERR_BACKEND); +} + +} // namespace + +hiredisAsyncClient::hiredisAsyncClient(RedisConfig config) + : config_(std::move(config)), + asyncContext_(nullptr), + syncContext_(nullptr), + eventBase_(nullptr), + connected_(false), + initDone_(false), + initSucceeded_(false) { + evthread_use_pthreads(); + + eventBase_ = event_base_new(); + if (!eventBase_) { + throw std::runtime_error("Failed to create event base"); + } + + asyncContext_ = redisAsyncConnect(config_.host.c_str(), config_.port); + if (!asyncContext_ || asyncContext_->err) { + if (asyncContext_) { + std::string err_msg = + absl::StrFormat("Failed to connect to Redis: %s", asyncContext_->errstr); + redisAsyncFree(asyncContext_); + event_base_free(eventBase_); + throw std::runtime_error(err_msg); + } + event_base_free(eventBase_); + throw std::runtime_error("Failed to allocate Redis async context"); + } + + asyncContext_->data = this; + + if (redisLibeventAttach(asyncContext_, eventBase_) != REDIS_OK) { + std::string err_msg = + absl::StrFormat("Failed to attach Redis to event base: %s", asyncContext_->errstr); + redisAsyncFree(asyncContext_); + event_base_free(eventBase_); + throw std::runtime_error(err_msg); + } + + redisAsyncSetConnectCallback(asyncContext_, connectCallback); + redisAsyncSetDisconnectCallback(asyncContext_, disconnectCallback); + + eventLoopThread_ = std::thread(&hiredisAsyncClient::processEventLoop, this); + + int retries = 100; + while (!initDone_.load() && retries-- > 0) { + std::this_thread::sleep_for(std::chrono::milliseconds(10)); + } + + if (!initSucceeded_.load()) { + stopEventLoop(); + throw std::runtime_error("Failed to connect to Redis within timeout"); + } + + connectSyncContext(); + + NIXL_INFO << absl::StrFormat( + "Connected to Redis at %s:%d (db=%d)", config_.host, config_.port, config_.db); +} + +void +hiredisAsyncClient::connectSyncContext() { + struct timeval timeout = {5, 0}; + syncContext_ = redisConnectWithTimeout(config_.host.c_str(), config_.port, timeout); + if (!syncContext_ || syncContext_->err) { + std::string err_msg = syncContext_ ? syncContext_->errstr : "allocation failed"; + if (syncContext_) { + redisFree(syncContext_); + syncContext_ = nullptr; + } + NIXL_WARN << absl::StrFormat( + "Sync Redis connection for EXISTS failed (%s:%d): %s; queryMem will return errors", + config_.host, + config_.port, + err_msg); + return; + } + + if (!config_.password.empty()) { + redisReply *reply = config_.username.empty() ? + static_cast( + redisCommand(syncContext_, "AUTH %s", config_.password.c_str())) : + static_cast(redisCommand( + syncContext_, "AUTH %s %s", config_.username.c_str(), config_.password.c_str())); + if (!checkRedisReplyOk(reply, "AUTH")) { + freeReplyObject(reply); + redisFree(syncContext_); + syncContext_ = nullptr; + NIXL_WARN << "Sync Redis AUTH failed; queryMem will return errors"; + return; + } + freeReplyObject(reply); + } + + if (config_.db != 0) { + redisReply *reply = + static_cast(redisCommand(syncContext_, "SELECT %d", config_.db)); + if (!checkRedisReplyOk(reply, "SELECT")) { + freeReplyObject(reply); + redisFree(syncContext_); + syncContext_ = nullptr; + NIXL_WARN << "Sync Redis SELECT failed; queryMem will return errors"; + return; + } + freeReplyObject(reply); + } + + NIXL_INFO << absl::StrFormat("Sync Redis connection ready for EXISTS at %s:%d (db=%d)", + config_.host, + config_.port, + config_.db); +} + +hiredisAsyncClient::~hiredisAsyncClient() { + stopEventLoop(); + if (syncContext_) { + redisFree(syncContext_); + syncContext_ = nullptr; + } +} + +void +hiredisAsyncClient::processEventLoop() { + event_base_dispatch(eventBase_); +} + +bool +hiredisAsyncClient::scheduleOnEventLoop(std::function task) { + if (!eventBase_) { + return false; + } + + auto *owned_task = new redis_event_task_t(std::move(task)); + timeval immediate = {0, 0}; + if (event_base_once(eventBase_, -1, EV_TIMEOUT, runEventTask, owned_task, &immediate) != 0) { + delete owned_task; + return false; + } + return true; +} + +void +hiredisAsyncClient::freeAsyncContextInEventLoop() { + redisAsyncContext *ctx = asyncContext_; + asyncContext_ = nullptr; + if (ctx) { + redisAsyncFree(ctx); + } +} + +void +hiredisAsyncClient::stopEventLoop() { + connected_.store(false); + + if (eventBase_ && eventLoopThread_.joinable()) { + bool scheduled = scheduleOnEventLoop([this]() { + freeAsyncContextInEventLoop(); + event_base_loopbreak(eventBase_); + }); + if (!scheduled) { + event_base_loopbreak(eventBase_); + } + eventLoopThread_.join(); + } + + if (eventBase_) { + event_base_free(eventBase_); + eventBase_ = nullptr; + } + asyncContext_ = nullptr; +} + +void +hiredisAsyncClient::completeAsyncInitialization(bool success) { + connected_.store(success); + initSucceeded_.store(success); + initDone_.store(true); +} + +void +hiredisAsyncClient::startAsyncSelect() { + if (config_.db == 0) { + completeAsyncInitialization(true); + return; + } + + int ret = redisAsyncCommand(asyncContext_, selectCallback, this, "SELECT %d", config_.db); + if (ret != REDIS_OK) { + NIXL_ERROR << "Failed to queue Redis SELECT command"; + completeAsyncInitialization(false); + freeAsyncContextInEventLoop(); + event_base_loopbreak(eventBase_); + } +} + +void +hiredisAsyncClient::startAsyncInitialization() { + if (!config_.password.empty()) { + const int ret = config_.username.empty() ? + redisAsyncCommand( + asyncContext_, authCallback, this, "AUTH %s", config_.password.c_str()) : + redisAsyncCommand(asyncContext_, + authCallback, + this, + "AUTH %s %s", + config_.username.c_str(), + config_.password.c_str()); + if (ret != REDIS_OK) { + NIXL_ERROR << "Failed to queue Redis AUTH command"; + completeAsyncInitialization(false); + freeAsyncContextInEventLoop(); + event_base_loopbreak(eventBase_); + } + return; + } + + // An empty password means the Redis deployment permits unauthenticated access. + startAsyncSelect(); +} + +void +hiredisAsyncClient::connectCallback(const redisAsyncContext *c, int status) { + auto *client = static_cast(c->data); + if (status != REDIS_OK) { + NIXL_ERROR << absl::StrFormat("Redis connection error: %s", c->errstr); + client->completeAsyncInitialization(false); + client->freeAsyncContextInEventLoop(); + event_base_loopbreak(client->eventBase_); + } else { + client->startAsyncInitialization(); + } +} + +void +hiredisAsyncClient::disconnectCallback(const redisAsyncContext *c, int status) { + auto *client = static_cast(c->data); + if (status != REDIS_OK) { + NIXL_WARN << absl::StrFormat("Redis disconnection error: %s", c->errstr); + } + client->connected_.store(false); +} + +void +hiredisAsyncClient::authCallback(redisAsyncContext *c, void *reply, void *privdata) { + auto *client = static_cast(privdata); + auto *r = static_cast(reply); + + if (!checkRedisReplyOk(r, "AUTH")) { + client->completeAsyncInitialization(false); + client->freeAsyncContextInEventLoop(); + event_base_loopbreak(client->eventBase_); + return; + } + + client->startAsyncSelect(); +} + +void +hiredisAsyncClient::selectCallback(redisAsyncContext *c, void *reply, void *privdata) { + auto *client = static_cast(privdata); + auto *r = static_cast(reply); + + if (!checkRedisReplyOk(r, "SELECT")) { + client->completeAsyncInitialization(false); + client->freeAsyncContextInEventLoop(); + event_base_loopbreak(client->eventBase_); + return; + } + + client->completeAsyncInitialization(true); +} + +void +hiredisAsyncClient::setCallback(redisAsyncContext *c, void *reply, void *privdata) { + auto *ctx = static_cast(privdata); + auto *r = static_cast(reply); + + bool success = false; + if (r && r->type == REDIS_REPLY_STATUS) { + success = (strcmp(r->str, "OK") == 0); + } else if (r && r->type == REDIS_REPLY_ERROR) { + NIXL_ERROR << absl::StrFormat("Redis SET error: %s", r->str); + } + + auto promise_ptr = ctx->promise_ptr; + delete ctx; + setPromiseStatus(promise_ptr, success); +} + +void +hiredisAsyncClient::getCallback(redisAsyncContext *c, void *reply, void *privdata) { + auto *ctx = static_cast(privdata); + auto *r = static_cast(reply); + + bool success = false; + if (r && r->type == REDIS_REPLY_STRING) { + const size_t reply_len = static_cast(r->len); + if (reply_len != ctx->data_len) { + NIXL_ERROR << absl::StrFormat( + "Redis GET size mismatch: expected %zu bytes, got %zu bytes", + ctx->data_len, + reply_len); + } else if (ctx->data_len == 0) { + success = true; + } else if (ctx->data_ptr) { + std::memcpy(reinterpret_cast(ctx->data_ptr), r->str, ctx->data_len); + success = true; + } + } else if (r && r->type == REDIS_REPLY_NIL) { + NIXL_WARN << "Redis GET: key not found"; + } else if (r && r->type == REDIS_REPLY_ERROR) { + NIXL_ERROR << absl::StrFormat("Redis GET error: %s", r->str); + } + + auto promise_ptr = ctx->promise_ptr; + delete ctx; + setPromiseStatus(promise_ptr, success); +} + +void +hiredisAsyncClient::putKeyAsync(std::string_view key, + uintptr_t data_ptr, + size_t data_len, + std::shared_ptr> promise) { + if (!connected_.load()) { + setPromiseStatus(promise, false); + return; + } + + std::string key_copy(key); + bool scheduled = scheduleOnEventLoop( + [this, key = std::move(key_copy), data_ptr, data_len, promise]() mutable { + if (!connected_.load() || !asyncContext_) { + setPromiseStatus(promise, false); + return; + } + + auto *ctx = new CallbackContext; + ctx->promise_ptr = promise; + + int ret = redisAsyncCommand(asyncContext_, + setCallback, + ctx, + "SET %b %b", + key.data(), + key.size(), + reinterpret_cast(data_ptr), + data_len); + + if (ret != REDIS_OK) { + auto promise_ptr = ctx->promise_ptr; + delete ctx; + setPromiseStatus(promise_ptr, false); + } + }); + + if (!scheduled) { + setPromiseStatus(promise, false); + } +} + +void +hiredisAsyncClient::getKeyAsync(std::string_view key, + uintptr_t data_ptr, + size_t data_len, + std::shared_ptr> promise) { + if (!connected_.load()) { + setPromiseStatus(promise, false); + return; + } + + std::string key_copy(key); + bool scheduled = scheduleOnEventLoop( + [this, key = std::move(key_copy), data_ptr, data_len, promise]() mutable { + if (!connected_.load() || !asyncContext_) { + setPromiseStatus(promise, false); + return; + } + + auto *ctx = new CallbackContext; + ctx->data_ptr = data_ptr; + ctx->data_len = data_len; + ctx->promise_ptr = promise; + + int ret = redisAsyncCommand( + asyncContext_, getCallback, ctx, "GET %b", key.data(), key.size()); + + if (ret != REDIS_OK) { + auto promise_ptr = ctx->promise_ptr; + delete ctx; + setPromiseStatus(promise_ptr, false); + } + }); + + if (!scheduled) { + setPromiseStatus(promise, false); + } +} + +std::optional +hiredisAsyncClient::checkKeyExistsSync(std::string_view key) { + std::lock_guard lock(syncMutex_); + + if (!syncContext_ || syncContext_->err) { + NIXL_ERROR << "Sync Redis connection unavailable for EXISTS"; + return std::nullopt; + } + + redisReply *reply = + static_cast(redisCommand(syncContext_, "EXISTS %b", key.data(), key.size())); + + if (!reply) { + NIXL_ERROR << "Redis EXISTS: no reply"; + return std::nullopt; + } + + if (reply->type == REDIS_REPLY_ERROR) { + NIXL_ERROR << absl::StrFormat("Redis EXISTS error: %s", reply->str); + freeReplyObject(reply); + return std::nullopt; + } + + if (reply->type != REDIS_REPLY_INTEGER) { + NIXL_ERROR << absl::StrFormat("Redis EXISTS unexpected reply type: %d", reply->type); + freeReplyObject(reply); + return std::nullopt; + } + + bool exists = (reply->integer == 1); + freeReplyObject(reply); + return exists; +} + +#else // HAVE_HIREDIS_ASYNC + +hiredisAsyncClient::hiredisAsyncClient(RedisConfig config) { + throw std::runtime_error("hiredis-async not available"); +} + +hiredisAsyncClient::~hiredisAsyncClient() {} + +void +hiredisAsyncClient::putKeyAsync(std::string_view key, + uintptr_t data_ptr, + size_t data_len, + std::shared_ptr> promise) { + if (promise) { + promise->set_value(NIXL_ERR_BACKEND); + } +} + +void +hiredisAsyncClient::getKeyAsync(std::string_view key, + uintptr_t data_ptr, + size_t data_len, + std::shared_ptr> promise) { + if (promise) { + promise->set_value(NIXL_ERR_BACKEND); + } +} + +std::optional +hiredisAsyncClient::checkKeyExistsSync(std::string_view key) { + return std::nullopt; +} + +#endif // HAVE_HIREDIS_ASYNC + +RedisConnectionPool::RedisConnectionPool(RedisConfig config) { + clients_.reserve(static_cast(config.pool_size)); + for (int i = 0; i < config.pool_size; ++i) { + clients_.push_back(std::make_unique(config)); + } +} + +iRedisClient & +RedisConnectionPool::nextSlot() { + return *clients_[nextSlot_.fetch_add(1, std::memory_order_relaxed) % clients_.size()]; +} + +void +RedisConnectionPool::putKeyAsync(std::string_view key, + uintptr_t data_ptr, + size_t data_len, + std::shared_ptr> promise) { + nextSlot().putKeyAsync(key, data_ptr, data_len, std::move(promise)); +} + +void +RedisConnectionPool::getKeyAsync(std::string_view key, + uintptr_t data_ptr, + size_t data_len, + std::shared_ptr> promise) { + nextSlot().getKeyAsync(key, data_ptr, data_len, std::move(promise)); +} + +std::optional +RedisConnectionPool::checkKeyExistsSync(std::string_view key) { + return nextSlot().checkKeyExistsSync(key); +} diff --git a/src/plugins/redis/redis_client.h b/src/plugins/redis/redis_client.h new file mode 100644 index 0000000000..c78f31c5cb --- /dev/null +++ b/src/plugins/redis/redis_client.h @@ -0,0 +1,169 @@ +/* + * SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. + * SPDX-License-Identifier: Apache-2.0 + */ + +#ifndef NIXL_SRC_PLUGINS_REDIS_REDIS_CLIENT_H +#define NIXL_SRC_PLUGINS_REDIS_REDIS_CLIENT_H + +#include "nixl_types.h" + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#ifdef HAVE_HIREDIS_ASYNC +#include +#include +#include +#include +#include +#else +struct event_base; +struct redisAsyncContext; +struct redisContext; +#endif + +/** Resolved Redis connection settings. Explicit backend parameters override environment values. */ +struct RedisConfig { + std::string host = "localhost"; + int port = 6379; + std::string username; + std::string password; + int db = 0; + int pool_size = 8; + + static RedisConfig + fromBackendParams(const nixl_b_params_t *custom_params); +}; + +/** Redis operations used by the NIXL REDIS backend. */ +class iRedisClient { +public: + virtual ~iRedisClient() = default; + + virtual void + putKeyAsync(std::string_view key, + uintptr_t data_ptr, + size_t data_len, + std::shared_ptr> promise) = 0; + + virtual void + getKeyAsync(std::string_view key, + uintptr_t data_ptr, + size_t data_len, + std::shared_ptr> promise) = 0; + + virtual std::optional + checkKeyExistsSync(std::string_view key) = 0; +}; + +/** Hiredis implementation with async SET/GET and a separate synchronous EXISTS connection. */ +class hiredisAsyncClient : public iRedisClient { +public: + explicit hiredisAsyncClient(RedisConfig config); + ~hiredisAsyncClient() override; + + void + putKeyAsync(std::string_view key, + uintptr_t data_ptr, + size_t data_len, + std::shared_ptr> promise) override; + + void + getKeyAsync(std::string_view key, + uintptr_t data_ptr, + size_t data_len, + std::shared_ptr> promise) override; + + std::optional + checkKeyExistsSync(std::string_view key) override; + +private: + struct CallbackContext { + uintptr_t data_ptr; + size_t data_len; + std::shared_ptr> promise_ptr; + }; + + static void + connectCallback(const redisAsyncContext *c, int status); + static void + disconnectCallback(const redisAsyncContext *c, int status); + static void + authCallback(redisAsyncContext *c, void *reply, void *privdata); + static void + selectCallback(redisAsyncContext *c, void *reply, void *privdata); + static void + setCallback(redisAsyncContext *c, void *reply, void *privdata); + static void + getCallback(redisAsyncContext *c, void *reply, void *privdata); + + void + processEventLoop(); + void + connectSyncContext(); + void + startAsyncInitialization(); + void + startAsyncSelect(); + void + completeAsyncInitialization(bool success); + void + freeAsyncContextInEventLoop(); + bool + scheduleOnEventLoop(std::function task); + void + stopEventLoop(); + + RedisConfig config_; + redisAsyncContext *asyncContext_; + redisContext *syncContext_; + struct event_base *eventBase_; + std::thread eventLoopThread_; + std::atomic connected_; + std::atomic initDone_; + std::atomic initSucceeded_; + mutable std::mutex syncMutex_; +}; + +/** + * Round-robin connection pool over multiple hiredisAsyncClient instances. + * Each slot has its own event loop thread and TCP connections. + */ +class RedisConnectionPool : public iRedisClient { +public: + explicit RedisConnectionPool(RedisConfig config); + ~RedisConnectionPool() override = default; + + void + putKeyAsync(std::string_view key, + uintptr_t data_ptr, + size_t data_len, + std::shared_ptr> promise) override; + + void + getKeyAsync(std::string_view key, + uintptr_t data_ptr, + size_t data_len, + std::shared_ptr> promise) override; + + std::optional + checkKeyExistsSync(std::string_view key) override; + +private: + iRedisClient & + nextSlot(); + + std::vector> clients_; + std::atomic nextSlot_{0}; +}; + +#endif // NIXL_SRC_PLUGINS_REDIS_REDIS_CLIENT_H diff --git a/src/plugins/redis/redis_plugin.cpp b/src/plugins/redis/redis_plugin.cpp new file mode 100644 index 0000000000..d4090abfdf --- /dev/null +++ b/src/plugins/redis/redis_plugin.cpp @@ -0,0 +1,40 @@ +/* + * SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. + * SPDX-License-Identifier: Apache-2.0 + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include "backend/backend_plugin.h" +#include "redis_backend.h" + +using redis_plugin_t = nixlBackendPluginCreator; + +static const nixl_mem_list_t supported_segments = {DRAM_SEG, OBJ_SEG}; + +#ifdef STATIC_PLUGIN_REDIS +nixlBackendPlugin * +createStaticREDISPlugin() { + return redis_plugin_t::create( + NIXL_PLUGIN_API_VERSION, REDIS_PLUGIN_NAME, REDIS_PLUGIN_VERSION, {}, supported_segments); +} +#else +extern "C" NIXL_PLUGIN_EXPORT nixlBackendPlugin * +nixl_plugin_init() { + return redis_plugin_t::create( + NIXL_PLUGIN_API_VERSION, REDIS_PLUGIN_NAME, REDIS_PLUGIN_VERSION, {}, supported_segments); +} + +extern "C" NIXL_PLUGIN_EXPORT void +nixl_plugin_fini() {} +#endif diff --git a/test/gtest/unit/descriptors/sec_desc_list.cpp b/test/gtest/unit/descriptors/sec_desc_list.cpp index 3ce1029f5e..c1acd8c5f2 100644 --- a/test/gtest/unit/descriptors/sec_desc_list.cpp +++ b/test/gtest/unit/descriptors/sec_desc_list.cpp @@ -207,7 +207,7 @@ TEST_F(secDescListTest, AddRandomBatches) { list.addDescs(std::move(batch)); num_added += batch_size; - ASSERT_EQ(list.descCount(), num_added); + ASSERT_EQ(static_cast(list.descCount()), num_added); assertSorted(list); } } diff --git a/test/gtest/unit/meson.build b/test/gtest/unit/meson.build index b68a5b96ef..032fd42aa1 100644 --- a/test/gtest/unit/meson.build +++ b/test/gtest/unit/meson.build @@ -23,6 +23,11 @@ if is_variable('obj_backend_interface') unit_test_deps += [obj_unit_test_dep] endif +if is_variable('redis_backend_interface') + subdir('redis') + unit_test_deps += [redis_unit_test_dep] +endif + subdir('agent') unit_test_deps += [agent_unit_test_dep] diff --git a/test/gtest/unit/redis/meson.build b/test/gtest/unit/redis/meson.build new file mode 100644 index 0000000000..23c636d458 --- /dev/null +++ b/test/gtest/unit/redis/meson.build @@ -0,0 +1,27 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +redis_unit_test_dep = declare_dependency( + sources: [ + 'redis.cpp', + ], + include_directories: [ + nixl_inc_dirs, utils_inc_dirs, + '../../../../src/plugins/redis', + ], + dependencies: [ + redis_backend_interface, + ], +) diff --git a/test/gtest/unit/redis/redis.cpp b/test/gtest/unit/redis/redis.cpp new file mode 100644 index 0000000000..29bd6b1de2 --- /dev/null +++ b/test/gtest/unit/redis/redis.cpp @@ -0,0 +1,473 @@ +/* + * SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. + * SPDX-License-Identifier: Apache-2.0 + */ + +#include + +#include "redis_backend.h" + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace gtest::redis { + +class scopedRedisEnvironment { +public: + scopedRedisEnvironment() { + save("REDIS_HOST", host_); + save("REDIS_PORT", port_); + save("REDIS_USERNAME", username_); + save("REDIS_PASSWORD", password_); + save("REDIS_POOL_SIZE", pool_size_); + } + + ~scopedRedisEnvironment() { + restore("REDIS_HOST", host_); + restore("REDIS_PORT", port_); + restore("REDIS_USERNAME", username_); + restore("REDIS_PASSWORD", password_); + restore("REDIS_POOL_SIZE", pool_size_); + } + + void + clear() { + unsetenv("REDIS_HOST"); + unsetenv("REDIS_PORT"); + unsetenv("REDIS_USERNAME"); + unsetenv("REDIS_PASSWORD"); + unsetenv("REDIS_POOL_SIZE"); + } + +private: + static void + save(const char *name, std::optional &value) { + if (const char *current = std::getenv(name)) { + value = current; + } + } + + static void + restore(const char *name, const std::optional &value) { + if (value) { + setenv(name, value->c_str(), 1); + } else { + unsetenv(name); + } + } + + std::optional host_; + std::optional port_; + std::optional username_; + std::optional password_; + std::optional pool_size_; +}; + +TEST(redisConfigTest, UsesUnauthenticatedDefaultsWhenCredentialsAreAbsent) { + scopedRedisEnvironment environment; + environment.clear(); + nixl_b_params_t params; + + const auto config = RedisConfig::fromBackendParams(¶ms); + + EXPECT_EQ(config.host, "localhost"); + EXPECT_EQ(config.port, 6379); + EXPECT_TRUE(config.username.empty()); + EXPECT_TRUE(config.password.empty()); + EXPECT_EQ(config.db, 0); +} + +TEST(redisConfigTest, BackendParametersOverrideEnvironmentFallbacks) { + scopedRedisEnvironment environment; + environment.clear(); + setenv("REDIS_HOST", "environment-host", 1); + setenv("REDIS_PORT", "6380", 1); + setenv("REDIS_USERNAME", "environment-user", 1); + setenv("REDIS_PASSWORD", "environment-password", 1); + nixl_b_params_t params = { + {"host", "parameter-host"}, + {"port", "6381"}, + {"username", "parameter-user"}, + {"password", "parameter-password"}, + {"db", "2"}, + }; + + const auto config = RedisConfig::fromBackendParams(¶ms); + + EXPECT_EQ(config.host, "parameter-host"); + EXPECT_EQ(config.port, 6381); + EXPECT_EQ(config.username, "parameter-user"); + EXPECT_EQ(config.password, "parameter-password"); + EXPECT_EQ(config.db, 2); +} + +TEST(redisConfigTest, RejectsAclUsernameWithoutPassword) { + scopedRedisEnvironment environment; + environment.clear(); + nixl_b_params_t params = {{"username", "acl-user"}}; + + EXPECT_THROW(RedisConfig::fromBackendParams(¶ms), std::invalid_argument); +} + +TEST(redisConfigTest, UsesDefaultPoolSizeWhenAbsent) { + scopedRedisEnvironment environment; + environment.clear(); + nixl_b_params_t params; + const auto config = RedisConfig::fromBackendParams(¶ms); + EXPECT_EQ(config.pool_size, 8); +} + +TEST(redisConfigTest, BackendParamOverridesPoolSize) { + scopedRedisEnvironment environment; + environment.clear(); + nixl_b_params_t params = {{"pool_size", "4"}}; + const auto config = RedisConfig::fromBackendParams(¶ms); + EXPECT_EQ(config.pool_size, 4); +} + +TEST(redisConfigTest, EnvVarSetsPoolSize) { + scopedRedisEnvironment environment; + environment.clear(); + setenv("REDIS_POOL_SIZE", "16", 1); + nixl_b_params_t params; + const auto config = RedisConfig::fromBackendParams(¶ms); + EXPECT_EQ(config.pool_size, 16); +} + +TEST(redisConfigTest, BackendParamOverridesPoolSizeEnvVar) { + scopedRedisEnvironment environment; + environment.clear(); + setenv("REDIS_POOL_SIZE", "16", 1); + nixl_b_params_t params = {{"pool_size", "2"}}; + const auto config = RedisConfig::fromBackendParams(¶ms); + EXPECT_EQ(config.pool_size, 2); +} + +TEST(redisConfigTest, InvalidPoolSizeFallsBackToDefault) { + scopedRedisEnvironment environment; + environment.clear(); + nixl_b_params_t params = {{"pool_size", "bad"}}; + const auto config = RedisConfig::fromBackendParams(¶ms); + EXPECT_EQ(config.pool_size, 8); +} + +TEST(redisConfigTest, ZeroPoolSizeFallsBackToDefault) { + scopedRedisEnvironment environment; + environment.clear(); + nixl_b_params_t params = {{"pool_size", "0"}}; + const auto config = RedisConfig::fromBackendParams(¶ms); + EXPECT_EQ(config.pool_size, 8); +} + +TEST(redisConfigTest, NegativePoolSizeFallsBackToDefault) { + scopedRedisEnvironment environment; + environment.clear(); + nixl_b_params_t params = {{"pool_size", "-1"}}; + const auto config = RedisConfig::fromBackendParams(¶ms); + EXPECT_EQ(config.pool_size, 8); +} + +class mockRedisClient : public iRedisClient { +public: + void + putKeyAsync(std::string_view key, + uintptr_t, + size_t, + std::shared_ptr> promise) override { + putKeys_.emplace_back(key); + completeOrQueue(std::move(promise)); + } + + void + getKeyAsync(std::string_view key, + uintptr_t, + size_t, + std::shared_ptr> promise) override { + getKeys_.emplace_back(key); + completeOrQueue(std::move(promise)); + } + + std::optional + checkKeyExistsSync(std::string_view key) override { + checkedKeys_.emplace_back(key); + if (existsResults_.empty()) { + return true; + } + auto result = existsResults_.front(); + existsResults_.pop_front(); + return result; + } + + void + setCompletionStatus(nixl_status_t status) { + completionStatus_ = status; + } + + void + setCompleteImmediately(bool value) { + completeImmediately_ = value; + } + + void + completePending(nixl_status_t status) { + for (auto &promise : pendingPromises_) { + promise->set_value(status); + } + pendingPromises_.clear(); + } + + void + setExistsResults(std::deque> results) { + existsResults_ = std::move(results); + } + + const std::vector & + putKeys() const { + return putKeys_; + } + + const std::vector & + getKeys() const { + return getKeys_; + } + + const std::vector & + checkedKeys() const { + return checkedKeys_; + } + +private: + void + completeOrQueue(std::shared_ptr> promise) { + if (completeImmediately_) { + promise->set_value(completionStatus_); + } else { + pendingPromises_.push_back(std::move(promise)); + } + } + + nixl_status_t completionStatus_ = NIXL_SUCCESS; + bool completeImmediately_ = true; + std::deque> existsResults_; + std::vector>> pendingPromises_; + std::vector putKeys_; + std::vector getKeys_; + std::vector checkedKeys_; +}; + +class redisEngineTest : public ::testing::Test { +protected: + void + SetUp() override { + initParams_.localAgent = "redis-test-agent"; + initParams_.type = "REDIS"; + initParams_.customParams = &customParams_; + initParams_.enableProgTh = false; + initParams_.pthrDelay = 0; + initParams_.syncMode = nixl_thread_sync_t::NIXL_THREAD_SYNC_RW; + + mockClient_ = std::make_shared(); + engine_ = std::make_unique(&initParams_, mockClient_); + } + + nixlBackendMD * + registerRemote(uint64_t dev_id, std::string key, nixl_mem_t type = OBJ_SEG) { + nixlBackendMD *metadata = nullptr; + EXPECT_EQ(engine_->registerMem(nixlBlobDesc(0, 16, dev_id, key), type, metadata), + NIXL_SUCCESS); + EXPECT_NE(metadata, nullptr); + return metadata; + } + + nixlBackendReqH * + prepareTransfer(nixl_xfer_op_t operation, nixl_meta_dlist_t &local, nixl_meta_dlist_t &remote) { + nixlBackendReqH *handle = nullptr; + EXPECT_EQ( + engine_->prepXfer(operation, local, remote, initParams_.localAgent, handle, nullptr), + NIXL_SUCCESS); + EXPECT_NE(handle, nullptr); + return handle; + } + + nixlBackendInitParams initParams_; + nixl_b_params_t customParams_; + std::shared_ptr mockClient_; + std::unique_ptr engine_; +}; + +TEST_F(redisEngineTest, ReportsRedisBackendCapabilities) { + EXPECT_FALSE(engine_->supportsRemote()); + EXPECT_TRUE(engine_->supportsLocal()); + EXPECT_FALSE(engine_->supportsNotif()); + EXPECT_EQ(engine_->getSupportedMems(), (nixl_mem_list_t{OBJ_SEG, DRAM_SEG})); + + nixlBackendMD *output = reinterpret_cast(1); + EXPECT_EQ(engine_->loadLocalMD(nullptr, output), NIXL_ERR_INVALID_PARAM); + EXPECT_EQ(output, nullptr); + EXPECT_EQ(engine_->connect(initParams_.localAgent), NIXL_SUCCESS); + EXPECT_EQ(engine_->disconnect(initParams_.localAgent), NIXL_SUCCESS); +} + +TEST_F(redisEngineTest, RegistersMetadataKeyAndDevIdFallback) { + auto *namedMetadata = registerRemote(11, "registered-key"); + auto *fallbackMetadata = registerRemote(22, ""); + + std::vector firstBuffer(16, 'a'); + std::vector secondBuffer(16, 'b'); + nixl_meta_dlist_t localDescs(DRAM_SEG); + localDescs.addDesc(nixlMetaDesc( + reinterpret_cast(firstBuffer.data()), firstBuffer.size(), 1, nullptr)); + localDescs.addDesc(nixlMetaDesc( + reinterpret_cast(secondBuffer.data()), secondBuffer.size(), 2, nullptr)); + nixl_meta_dlist_t remoteDescs(OBJ_SEG); + remoteDescs.addDesc(nixlMetaDesc(0, firstBuffer.size(), 11, nullptr)); + remoteDescs.addDesc(nixlMetaDesc(0, secondBuffer.size(), 22, nullptr)); + + auto *handle = prepareTransfer(NIXL_WRITE, localDescs, remoteDescs); + EXPECT_EQ(engine_->postXfer( + NIXL_WRITE, localDescs, remoteDescs, initParams_.localAgent, handle, nullptr), + NIXL_IN_PROG); + EXPECT_EQ(mockClient_->putKeys(), (std::vector{"registered-key", "22"})); + EXPECT_EQ(engine_->checkXfer(handle), NIXL_SUCCESS); + + EXPECT_EQ(engine_->releaseReqH(handle), NIXL_SUCCESS); + EXPECT_EQ(engine_->deregisterMem(namedMetadata), NIXL_SUCCESS); + EXPECT_EQ(engine_->deregisterMem(fallbackMetadata), NIXL_SUCCESS); + + nixlBackendMD *unsupported = reinterpret_cast(1); + EXPECT_EQ(engine_->registerMem(nixlBlobDesc(), VRAM_SEG, unsupported), NIXL_ERR_NOT_SUPPORTED); + EXPECT_EQ(unsupported, nullptr); +} + +TEST_F(redisEngineTest, QueryMemPreservesFoundMissingAndErrorResponses) { + mockClient_->setExistsResults({true, false, std::nullopt}); + nixl_reg_dlist_t descs(OBJ_SEG); + descs.addDesc(nixlBlobDesc(nixlBasicDesc(0, 0, 1), "found")); + descs.addDesc(nixlBlobDesc(nixlBasicDesc(0, 0, 2), "missing")); + descs.addDesc(nixlBlobDesc(nixlBasicDesc(0, 0, 3), "")); + + std::vector responses; + EXPECT_EQ(engine_->queryMem(descs, responses), NIXL_ERR_BACKEND); + ASSERT_EQ(responses.size(), 3U); + EXPECT_TRUE(responses[0].has_value()); + EXPECT_FALSE(responses[1].has_value()); + EXPECT_FALSE(responses[2].has_value()); + EXPECT_EQ(mockClient_->checkedKeys(), (std::vector{"found", "missing", "3"})); +} + +TEST_F(redisEngineTest, PrepXferRejectsInvalidRequests) { + nixl_meta_dlist_t emptyLocal(DRAM_SEG); + nixl_meta_dlist_t emptyRemote(OBJ_SEG); + nixlBackendReqH *handle = reinterpret_cast(1); + EXPECT_EQ(engine_->prepXfer( + NIXL_WRITE, emptyLocal, emptyRemote, initParams_.localAgent, handle, nullptr), + NIXL_ERR_INVALID_PARAM); + EXPECT_EQ(handle, reinterpret_cast(1)); + + std::vector buffer(16); + nixl_meta_dlist_t local(DRAM_SEG); + local.addDesc( + nixlMetaDesc(reinterpret_cast(buffer.data()), buffer.size(), 1, nullptr)); + nixl_meta_dlist_t remote(OBJ_SEG); + remote.addDesc(nixlMetaDesc(0, buffer.size(), 2, nullptr)); + + handle = nullptr; + EXPECT_EQ(engine_->prepXfer(static_cast(99), + local, + remote, + initParams_.localAgent, + handle, + nullptr), + NIXL_ERR_INVALID_PARAM); + EXPECT_EQ(handle, nullptr); + + nixl_meta_dlist_t wrongLocal(OBJ_SEG); + wrongLocal.addDesc(nixlMetaDesc(0, buffer.size(), 1, nullptr)); + EXPECT_EQ( + engine_->prepXfer(NIXL_READ, wrongLocal, remote, initParams_.localAgent, handle, nullptr), + NIXL_ERR_INVALID_PARAM); +} + +TEST_F(redisEngineTest, PollsReadUntilClientCompletes) { + auto *metadata = registerRemote(22, "read-key"); + mockClient_->setCompleteImmediately(false); + std::vector buffer(16); + nixl_meta_dlist_t local(DRAM_SEG); + local.addDesc( + nixlMetaDesc(reinterpret_cast(buffer.data()), buffer.size(), 1, nullptr)); + nixl_meta_dlist_t remote(OBJ_SEG); + remote.addDesc(nixlMetaDesc(0, buffer.size(), 22, nullptr)); + + auto *handle = prepareTransfer(NIXL_READ, local, remote); + EXPECT_EQ(engine_->postXfer(NIXL_READ, local, remote, initParams_.localAgent, handle, nullptr), + NIXL_IN_PROG); + EXPECT_EQ(mockClient_->getKeys(), (std::vector{"read-key"})); + EXPECT_EQ(engine_->checkXfer(handle), NIXL_IN_PROG); + mockClient_->completePending(NIXL_SUCCESS); + EXPECT_EQ(engine_->checkXfer(handle), NIXL_SUCCESS); + + EXPECT_EQ(engine_->releaseReqH(handle), NIXL_SUCCESS); + EXPECT_EQ(engine_->deregisterMem(metadata), NIXL_SUCCESS); +} + +TEST_F(redisEngineTest, PropagatesAsyncClientFailure) { + auto *metadata = registerRemote(22, "failed-key"); + mockClient_->setCompletionStatus(NIXL_ERR_BACKEND); + std::vector buffer(16); + nixl_meta_dlist_t local(DRAM_SEG); + local.addDesc( + nixlMetaDesc(reinterpret_cast(buffer.data()), buffer.size(), 1, nullptr)); + nixl_meta_dlist_t remote(OBJ_SEG); + remote.addDesc(nixlMetaDesc(0, buffer.size(), 22, nullptr)); + + auto *handle = prepareTransfer(NIXL_WRITE, local, remote); + EXPECT_EQ(engine_->postXfer(NIXL_WRITE, local, remote, initParams_.localAgent, handle, nullptr), + NIXL_IN_PROG); + EXPECT_EQ(engine_->checkXfer(handle), NIXL_ERR_BACKEND); + EXPECT_EQ(engine_->releaseReqH(handle), NIXL_SUCCESS); + EXPECT_EQ(engine_->deregisterMem(metadata), NIXL_SUCCESS); +} + +TEST_F(redisEngineTest, RejectsNullTransferHandles) { + nixl_meta_dlist_t local(DRAM_SEG); + nixl_meta_dlist_t remote(OBJ_SEG); + nixlBackendReqH *handle = nullptr; + EXPECT_EQ(engine_->postXfer(NIXL_WRITE, local, remote, initParams_.localAgent, handle, nullptr), + NIXL_ERR_INVALID_PARAM); + EXPECT_EQ(engine_->checkXfer(nullptr), NIXL_ERR_INVALID_PARAM); +} + +TEST_F(redisEngineTest, PostXferDoesNotDispatchPartialCommandsWhenLaterKeyIsMissing) { + std::vector firstBuffer(16, 'a'); + std::vector secondBuffer(16, 'b'); + + auto *registeredMetadata = registerRemote(11, "registered-key", DRAM_SEG); + + nixl_meta_dlist_t localDescs(DRAM_SEG); + localDescs.addDesc(nixlMetaDesc( + reinterpret_cast(firstBuffer.data()), firstBuffer.size(), 1, nullptr)); + localDescs.addDesc(nixlMetaDesc( + reinterpret_cast(secondBuffer.data()), secondBuffer.size(), 2, nullptr)); + + nixl_meta_dlist_t remoteDescs(DRAM_SEG); + remoteDescs.addDesc(nixlMetaDesc(0, firstBuffer.size(), 11, nullptr)); + remoteDescs.addDesc(nixlMetaDesc(0, secondBuffer.size(), 22, nullptr)); + + auto *handle = prepareTransfer(NIXL_WRITE, localDescs, remoteDescs); + EXPECT_EQ(engine_->postXfer( + NIXL_WRITE, localDescs, remoteDescs, initParams_.localAgent, handle, nullptr), + NIXL_ERR_INVALID_PARAM); + EXPECT_TRUE(mockClient_->putKeys().empty()); + + EXPECT_EQ(engine_->releaseReqH(handle), NIXL_SUCCESS); + EXPECT_EQ(engine_->deregisterMem(registeredMetadata), NIXL_SUCCESS); +} + +} // namespace gtest::redis