Merge branch 'main' into destructor-race

Author: 2dm

benchmark/nixlbench/README.md

@@ -160,7 +160,7 @@ meson configure build
 #### Device and Network Configuration
 ```
 --device_list LIST         # Comma-separated device names (default: all)
---etcd_endpoints URL       # ETCD server URL for coordination (default: http://localhost:2379)
+--etcd_endpoints URL       # ETCD server URL for coordination (optional for storage backends)
 ```
 
 #### Storage Backend Options (GDS, GDS_MT, POSIX, HF3FS, OBJ)
@@ -208,11 +208,25 @@ meson configure build
 
 NIXL Benchmark uses an ETCD key-value store for coordination between benchmark workers. This is useful in containerized or cloud-native environments.
 
-To run the benchmark:
+**ETCD Requirements:**
+- **Required**: Network backends (UCX, UCX_MO, GPUNETIO, Mooncake) and multi-node setups
+- **Optional**: Storage backends (GDS, GDS_MT, POSIX, HF3FS, OBJ) running as single instances
+- **Required**: Storage backends when `--etcd_endpoints` is explicitly specified
+
+**For multi-node benchmarks:**
 
 1. Ensure ETCD server is running (e.g., `docker run -p 2379:2379 quay.io/coreos/etcd`
 2. Launch multiple nixlbench instances pointing to the same ETCD server
 
+**For single-instance storage benchmarks:**
+```bash
+# No ETCD needed - just run directly
+./nixlbench --backend GDS --filepath /mnt/storage/testfile
+
+# Or with explicit ETCD if coordination is needed
+./nixlbench --etcd_endpoints http://etcd-server:2379 --backend GDS --filepath /mnt/storage/testfile
+```
+
 Note: etcd can be installed directly on host as well:
 ```bash
 apt install etcd-server
@@ -255,26 +269,29 @@ The workers automatically coordinate ranks through ETCD as they connect.
 
 **GDS (GPU Direct Storage)**
 ```bash
-# Basic GDS benchmark
+# Basic GDS benchmark (no ETCD needed for single instance)
+./nixlbench --backend GDS --filepath /mnt/storage/testfile --storage_enable_direct
+
+# GDS with ETCD coordination
 ./nixlbench --etcd_endpoints http://etcd-server:2379 --backend GDS --filepath /mnt/storage/testfile --storage_enable_direct
 
 # GDS with custom batch settings
-./nixlbench --etcd_endpoints http://etcd-server:2379 --backend GDS --filepath /mnt/storage/testfile --gds_batch_pool_size 64 --gds_batch_limit 256
+./nixlbench --backend GDS --filepath /mnt/storage/testfile --gds_batch_pool_size 64 --gds_batch_limit 256
 ```
 
 **GDS_MT (Multi-threaded GDS)**
 ```bash
-# Multi-threaded GDS
-./nixlbench --etcd_endpoints http://etcd-server:2379 --backend GDS_MT --filepath /mnt/storage/testfile --gds_mt_num_threads 8
+# Multi-threaded GDS (no ETCD needed for single instance)
+./nixlbench --backend GDS_MT --filepath /mnt/storage/testfile --gds_mt_num_threads 8
 ```
 
 **POSIX Backend**
 ```bash
-# POSIX with AIO
-./nixlbench --etcd_endpoints http://etcd-server:2379 --backend POSIX --filepath /mnt/storage/testfile --posix_api_type AIO
+# POSIX with AIO (no ETCD needed for single instance)
+./nixlbench --backend POSIX --filepath /mnt/storage/testfile --posix_api_type AIO
 
 # POSIX with io_uring
-./nixlbench --etcd_endpoints http://etcd-server:2379 --backend POSIX --filepath /mnt/storage/testfile --posix_api_type URING --storage_enable_direct
+./nixlbench --backend POSIX --filepath /mnt/storage/testfile --posix_api_type URING --storage_enable_direct
 ```
 
 #### Worker Types
@@ -287,27 +304,30 @@ The workers automatically coordinate ranks through ETCD as they connect.
 
 ### Benchmarking the OBJ (S3) Backend
 
-For OBJ plugin benchmarking run etcd-server and a single nixlbench instance.
+For OBJ plugin benchmarking, ETCD is optional for single instances.
 
 Example:
 ```bash
-# Basic S3 benchmark using environment variables
+# Basic S3 benchmark using environment variables (no ETCD needed)
 AWS_ACCESS_KEY_ID=<access_key> AWS_SECRET_ACCESS_KEY=<secret_key> AWS_DEFAULT_REGION=<region> \
-./nixlbench --etcd_endpoints http://etcd-server:2379 --backend OBJ --obj_bucket_name <bucket_name>
+./nixlbench --backend OBJ --obj_bucket_name <bucket_name>
 
-# S3 benchmark using command line flags
-./nixlbench --etcd_endpoints http://etcd-server:2379 --backend OBJ \
+# S3 benchmark using command line flags (no ETCD needed)
+./nixlbench --backend OBJ \
   --obj_access_key <access_key> \
   --obj_secret_key <secret_key> \
   --obj_region <region> \
   --obj_bucket_name <bucket_name>
+
+# S3 benchmark with ETCD coordination (if needed)
+./nixlbench --etcd_endpoints http://etcd-server:2379 --backend OBJ --obj_bucket_name <bucket_name>
 ```
 
 **Performance Considerations:**
 Transfer times are higher than local storage, so consider reducing iterations:
 
 ```bash
-./nixlbench --etcd_endpoints http://etcd-server:2379 --backend OBJ \
+./nixlbench --backend OBJ \
   --obj_bucket_name test-bucket \
   --warmup_iter 32 --num_iter 32 --large_blk_iter_ftr 2
 ```

benchmark/nixlbench/src/utils/utils.cpp

@@ -97,7 +97,9 @@ DEFINE_int32(gds_mt_num_threads, 1, "Number of threads used by GDS MT plugin (De
 // For example- 0:mlx5_0,mlx5_1,mlx5_2,1:mlx5_3,mlx5_4, ...
 DEFINE_string(device_list, "all", "Comma-separated device name to use for \
 		      communication (only used with nixl worker)");
-DEFINE_string(etcd_endpoints, "http://localhost:2379", "ETCD server endpoints for communication");
+DEFINE_string(etcd_endpoints,
+              "",
+              "ETCD server endpoints for communication (optional for storage backends)");
 
 // POSIX options - only used when backend is POSIX
 DEFINE_string (posix_api_type,
@@ -279,6 +281,14 @@ xferBenchConfig::loadFromFlags() {
     posix_api_type = FLAGS_posix_api_type;
     storage_enable_direct = FLAGS_storage_enable_direct;
 
+    // Validate ETCD configuration
+    if (!isStorageBackend() && etcd_endpoints.empty()) {
+        // For non-storage backends, set default ETCD endpoint
+        etcd_endpoints = "http://localhost:2379";
+        std::cout << "Using default ETCD endpoint for non-storage backend: " << etcd_endpoints
+                  << std::endl;
+    }
+
     if (worker_type == XFERBENCH_WORKER_NVSHMEM) {
         if (!((XFERBENCH_SEG_TYPE_VRAM == initiator_seg_type) &&
               (XFERBENCH_SEG_TYPE_VRAM == target_seg_type) &&
@@ -373,7 +383,11 @@ xferBenchConfig::printConfig() {
     printSeparator('*');
     printOption("Runtime (--runtime_type=[etcd])", runtime_type);
     if (runtime_type == XFERBENCH_RT_ETCD) {
-        printOption("ETCD Endpoint ", etcd_endpoints);
+        if (etcd_endpoints.empty()) {
+            printOption("ETCD Endpoint ", "disabled (storage backend)");
+        } else {
+            printOption("ETCD Endpoint ", etcd_endpoints);
+        }
     }
     printOption("Worker type (--worker_type=[nixl,nvshmem])", worker_type);
     if (worker_type == XFERBENCH_WORKER_NIXL) {

benchmark/nixlbench/src/worker/nixl/nixl_worker.cpp

@@ -748,6 +748,7 @@ int
 xferBenchNixlWorker::exchangeMetadata() {
     int meta_sz, ret = 0;
 
+    // Skip metadata exchange for storage backends or when ETCD is not available
     if (xferBenchConfig::isStorageBackend()) {
         return 0;
     }
@@ -1054,6 +1055,12 @@ xferBenchNixlWorker::poll(size_t block_size) {
 
 int
 xferBenchNixlWorker::synchronizeStart() {
+    // For storage backends without ETCD, no synchronization needed
+    if (xferBenchConfig::isStorageBackend() && xferBenchConfig::etcd_endpoints.empty()) {
+        std::cout << "Single instance storage backend - no synchronization needed" << std::endl;
+        return 0;
+    }
+
     if (IS_PAIRWISE_AND_SG()) {
         std::cout << "Waiting for all processes to start... (expecting " << rt->getSize()
                   << " total: " << xferBenchConfig::num_initiator_dev << " initiators and "

benchmark/nixlbench/src/worker/worker.cpp

@@ -21,7 +21,60 @@
 
 #include <unistd.h>
 
+// Null runtime for storage backends that don't need ETCD
+class xferBenchNullRT : public xferBenchRT {
+public:
+    xferBenchNullRT() {
+        setSize(1);
+        setRank(0);
+    }
+
+    virtual ~xferBenchNullRT() {}
+
+    virtual int
+    sendInt(int *buffer, int dest_rank) override {
+        return 0;
+    }
+
+    virtual int
+    recvInt(int *buffer, int src_rank) override {
+        return 0;
+    }
+
+    virtual int
+    broadcastInt(int *buffer, size_t count, int root_rank) override {
+        return 0;
+    }
+
+    virtual int
+    sendChar(char *buffer, size_t count, int dest_rank) override {
+        return 0;
+    }
+
+    virtual int
+    recvChar(char *buffer, size_t count, int src_rank) override {
+        return 0;
+    }
+
+    virtual int
+    reduceSumDouble(double *local_value, double *global_value, int dest_rank) override {
+        *global_value = *local_value;
+        return 0;
+    }
+
+    virtual int
+    barrier(const std::string &barrier_id) override {
+        return 0;
+    }
+};
+
 static xferBenchRT *createRT(int *terminate) {
+    // For storage backends without ETCD endpoints, use null runtime
+    if (xferBenchConfig::isStorageBackend() && xferBenchConfig::etcd_endpoints.empty()) {
+        std::cout << "Using null runtime for storage backend without ETCD" << std::endl;
+        return new xferBenchNullRT();
+    }
+
     if (XFERBENCH_RT_ETCD == xferBenchConfig::runtime_type) {
         int total = 2;
         if (XFERBENCH_MODE_SG == xferBenchConfig::mode) {
@@ -46,6 +99,11 @@ static xferBenchRT *createRT(int *terminate) {
 }
 
 int xferBenchWorker::synchronize() {
+    // For storage backends without ETCD, no synchronization needed
+    if (xferBenchConfig::isStorageBackend() && xferBenchConfig::etcd_endpoints.empty()) {
+        return 0;
+    }
+
     if (rt->barrier("sync") != 0) {
         std::cerr << "Failed to synchronize" << std::endl;
         // assuming this is a fatal error, continue benchmarking after synchronization failure does
@@ -67,7 +125,10 @@ xferBenchWorker::xferBenchWorker(int *argc, char ***argv) {
 
     int rank = rt->getRank();
 
-    if (XFERBENCH_MODE_SG == xferBenchConfig::mode) {
+    // For storage backends without ETCD, always act as initiator
+    if (xferBenchConfig::isStorageBackend() && xferBenchConfig::etcd_endpoints.empty()) {
+        name = "initiator";
+    } else if (XFERBENCH_MODE_SG == xferBenchConfig::mode) {
         if (rank >= 0 && rank < xferBenchConfig::num_initiator_dev) {
             name = "initiator";
         } else {