ai-dynamo · gaowayne · Jun 11, 2026 · Jun 12, 2026 · Jun 13, 2026 · Jun 13, 2026
@@ -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

@@ -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
 }
 

@@ -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

@@ -35,11 +35,130 @@
 #include <fcntl.h>
 #include <filesystem>
 #include <gflags/gflags.h>
+#include <hiredis/hiredis.h>
 
 #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<size_t>(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<std::vector<xferBenchIOV>> &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) {

@@ -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);

@@ -263,6 +263,10 @@ xferBenchNixlWorker::xferBenchNixlWorker(const std::vector<std::string> &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<xferBenchIOV> 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<std::vector<xferBenchIOV>> &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()) {