From a553e1a1dc087e1d5b91f88f55cfdd2d98d7b9d7 Mon Sep 17 00:00:00 2001
From: Vlad Buslov <vladbu@nvidia.com>
Date: Fri, 25 Apr 2025 12:58:14 +0300
Subject: [PATCH 1/2] Cleanup nixl_test

Nixl_test shares the same code for both target and initiator which caused
to acquire conditional checks for the process type all around the code
decreasing its readability and maintainability. Split the code into two
process type-specific functions.

Refactor code where it makes sense to simplify: change loop that verifies
memory contents and which can never terminate unless the contents match to
a bounded loop with timeouts, simplify unnecessary complicated
implementations of some loops, user constructors instead of manually
setting structure fields where possible, extract commonly used literals
into defines or static const vars.

Update the code to use C++ algorithms, data structures and memory
management primitives from their C counterparts.

Signed-off-by: Vlad Buslov <vladbu@nvidia.com>
---
 test/nixl/nixl_test.cpp | 328 +++++++++++++++++++---------------------
 1 file changed, 159 insertions(+), 169 deletions(-)

diff --git a/test/nixl/nixl_test.cpp b/test/nixl/nixl_test.cpp
index fb0828959c..1939a8ba2b 100644
--- a/test/nixl/nixl_test.cpp
+++ b/test/nixl/nixl_test.cpp
@@ -17,14 +17,18 @@
 #include <iostream>
 #include <string>
 #include <algorithm>
+#include <thread>
+#include <chrono>
 #include <nixl_descriptors.h>
 #include <nixl_params.h>
 #include <nixl.h>
 #include <cassert>
 #include "stream/metadata_stream.h"
 #include "serdes/serdes.h"
+
 #define NUM_TRANSFERS 1
 #define SIZE 1024
+#define MEM_VAL 0xBB
 
 /**
  * This test does p2p from using PUT.
@@ -33,208 +37,194 @@
  * target to initiator
  */
 
-bool allBytesAre(void* buffer, size_t size, uint8_t value) {
-    uint8_t* byte_buffer = static_cast<uint8_t*>(buffer); // Cast void* to uint8_t*
-    // Iterate over each byte in the buffer
-    for (size_t i = 0; i < size; ++i) {
-        if (byte_buffer[i] != value) {
-            return false; // Return false if any byte doesn't match the value
-        }
+static const std::string target("target");
+static const std::string initiator("initiator");
+
+static std::vector<std::unique_ptr<uint8_t[]>> initMem(nixlAgent &agent,
+                                                       nixl_reg_dlist_t &dram,
+                                                       nixl_opt_args_t *extra_params,
+                                                       uint8_t val) {
+    std::vector<std::unique_ptr<uint8_t[]>> addrs;
+
+    for (int i = 0; i < NUM_TRANSFERS; i++) {
+        auto addr = std::make_unique<uint8_t[]>(SIZE);
+
+        std::fill_n(addr.get(), SIZE, val);
+        std::cout << "Allocating : " << (void *)addr.get() << ", "
+                  << "Setting to 0x" << std::hex << (unsigned)val << std::dec << std::endl;
+        dram.addDesc(nixlBlobDesc((uintptr_t)(addr.get()), SIZE, 0, ""));
+
+        addrs.push_back(std::move(addr));
     }
-    return true; // All bytes match the value
+    agent.registerMem(dram, extra_params);
+
+    return addrs;
 }
 
-int main(int argc, char *argv[]) {
-    int                     target_port;
-    nixl_status_t           ret = NIXL_SUCCESS;
-    void                    *addr[NUM_TRANSFERS];
-    std::string             role;
-    const char              *target_ip;
-    nixl_blob_t             remote_desc;
-    nixl_blob_t             tgt_metadata;
-    nixl_blob_t             tgt_md_init;
-    int                     status = 0;
-    bool                    rc = false;
-
-    /** NIXL declarations */
-    /** Agent and backend creation parameters */
+static void runTarget(const std::string &ip, int port) {
+    nixlAgentConfig cfg(true, true, port);
+
+    std::cout << "Starting Agent for target\n";
+    nixlAgent     agent(target, cfg);
+
     nixl_b_params_t params;
-    nixlBlobDesc    buf[NUM_TRANSFERS];
     nixlBackendH    *ucx;
+    agent.createBackend("UCX", params, ucx);
 
-    /** Serialization/Deserialization object to create a blob */
-    nixlSerDes *serdes        = new nixlSerDes();
-    nixlSerDes *remote_serdes = new nixlSerDes();
+    nixl_opt_args_t extra_params;
+    extra_params.backends.push_back(ucx);
 
-    /** Descriptors and Transfer Request */
-    nixl_reg_dlist_t  dram_for_ucx(DRAM_SEG);
-    nixlXferReqH      *treq;
+    nixl_reg_dlist_t dram_for_ucx(DRAM_SEG);
+    auto addrs = initMem(agent, dram_for_ucx, &extra_params, 0);
 
-    /** Argument Parsing */
-    if (argc < 4) {
-        std::cout <<"Enter the required arguments\n" << std::endl;
-        std::cout <<"<Role> " <<"Target IP> <Target Port>"
-                  << std::endl;
-        exit(-1);
-    }
+    nixl_blob_t tgt_metadata;
+    agent.getLocalMD(tgt_metadata);
 
-    role = std::string(argv[1]);
-    target_ip   = argv[2];
-    target_port = std::stoi(argv[3]);
-    std::transform(role.begin(), role.end(), role.begin(), ::tolower);
+    std::cout << " Start Control Path metadata exchanges \n";
 
-    if (!role.compare("initiator") && !role.compare("target")) {
-            std::cerr << "Invalid role. Use 'initiator' or 'target'."
-                      << "Currently "<< role <<std::endl;
-            return 1;
+    std::cout << " Desc List from Target to Initiator\n";
+    dram_for_ucx.print();
+
+    /** Only send desc list */
+    nixlSerDes serdes;
+    assert(dram_for_ucx.trim().serialize(&serdes) == NIXL_SUCCESS);
+
+    std::cout << " Wait for initiator and then send xfer descs\n";
+    std::string message = serdes.exportStr();
+    while (agent.genNotif(initiator, message, &extra_params) != NIXL_SUCCESS);
+    std::cout << " End Control Path metadata exchanges \n";
+
+    std::cout << " Start Data Path Exchanges \n";
+    std::cout << " Waiting to receive Data from Initiator\n";
+
+    bool rc = false;
+    for (int n_tries = 0; !rc && n_tries < 100; n_tries++) {
+        //Only works with progress thread now, as backend is protected
+        /** Sanity Check */
+        rc = std::all_of(addrs.begin(), addrs.end(), [](auto &addr) {
+            return std::all_of(addr.get(), addr.get() + SIZE, [](int x) {
+                return x == MEM_VAL;
+            });
+        });
+        if (!rc)
+            std::this_thread::sleep_for(std::chrono::milliseconds(10));
     }
+    if (!rc)
+        std::cerr << " UCX Transfer failed, buffers are different\n";
+    else
+        std::cout << " Transfer completed and Buffers match with Initiator\n"
+                  <<"  UCX Transfer Success!!!\n";
 
-    nixlAgentConfig cfg(true);
-    if ( role == "target" ) {
-        cfg = nixlAgentConfig(true, true, target_port);
-    } else {
-        //use default port
-        cfg = nixlAgentConfig(true, true);
-    }
+    std::cout <<"Cleanup.. \n";
+    agent.deregisterMem(dram_for_ucx, &extra_params);
+}
 
+static void runInitiator(const std::string &target_ip, int target_port) {
+    nixlAgentConfig cfg(true, true);
 
-    /*** End - Argument Parsing */
+    std::cout << "Starting Agent for initiator\n";
+    nixlAgent     agent(initiator, cfg);
 
-    /** Common to both Initiator and Target */
-    std::cout << "Starting Agent for "<< role << "\n";
-    nixlAgent     agent(role, cfg);
+    nixl_b_params_t params;
+    nixlBackendH    *ucx;
     agent.createBackend("UCX", params, ucx);
 
     nixl_opt_args_t extra_params;
     extra_params.backends.push_back(ucx);
 
-    for (int i = 0; i < NUM_TRANSFERS; i++) {
-        addr[i] = calloc(1, SIZE);
-        if (role != "target") {
-            memset(addr[i], 0xbb, SIZE);
-            std::cout << "Allocating for initiator : "
-                      << addr[i] << ", "
-                      << "Setting to 0xbb "
-                      << std::endl;
-        } else {
-            memset(addr[i], 0, SIZE);
-            std::cout << "Allocating for target : "
-                      << addr[i] << ", "
-                      << "Setting to 0 " << std::endl;
-        }
-        buf[i].addr  = (uintptr_t)(addr[i]);
-        buf[i].len   = SIZE;
-        buf[i].devId = 0;
-        dram_for_ucx.addDesc(buf[i]);
-    }
-
-    /** Register memory in both initiator and target */
-    agent.registerMem(dram_for_ucx, &extra_params);
-    if (role == "target") {
-        agent.getLocalMD(tgt_metadata);
-    }
+    nixl_reg_dlist_t dram_for_ucx(DRAM_SEG);
+    auto addrs = initMem(agent, dram_for_ucx, &extra_params, MEM_VAL);
 
     std::cout << " Start Control Path metadata exchanges \n";
-    if (role == "target") {
-        std::cout << " Desc List from Target to Initiator\n";
-        dram_for_ucx.print();
-
-        /** Only send desc list */
-        assert(dram_for_ucx.trim().serialize(serdes) == NIXL_SUCCESS);
+    std::cout << " Exchange metadata with Target \n";
 
-        std::cout << " Wait for initiator and then send xfer descs\n";
+    nixl_opt_args_t md_extra_params;
+    md_extra_params.ipAddr = target_ip;
+    md_extra_params.port = target_port;
 
-        std::string message = serdes->exportStr();
+    agent.fetchRemoteMD(target, &md_extra_params);
 
-        do{
-            ret = agent.genNotif("initiator", message, &extra_params);
-        } while(ret != NIXL_SUCCESS);
+    agent.sendLocalMD(&md_extra_params);
 
-        std::cout << " End Control Path metadata exchanges \n";
+    nixl_notifs_t notifs;
+    while(notifs.size() == 0) {
+        nixl_status_t ret = agent.getNotifs(notifs, &extra_params);
+        assert(ret >= 0);
+    }
+    std::string rrstr = notifs[target][0];
+    assert(rrstr.size() > 0);
+
+    nixlSerDes remote_serdes;
+    remote_serdes.importStr(rrstr);
+
+    std::cout << " Verify Deserialized Target's Desc List at Initiator\n";
+    nixl_xfer_dlist_t dram_target_ucx(&remote_serdes);
+    nixl_xfer_dlist_t dram_initiator_ucx = dram_for_ucx.trim();
+    dram_target_ucx.print();
+
+    std::cout << " End Control Path metadata exchanges \n";
+    std::cout << " Start Data Path Exchanges \n\n";
+    std::cout << " Create transfer request with UCX backend\n ";
+
+    // Need to do this in a loop with NIXL_ERR_NOT_FOUND
+    // UCX AM with desc list is faster than listener thread can recv/load MD with sockets
+    // Will be deprecated with ETCD or callbacks
+    nixlXferReqH *treq;
+    nixl_status_t ret = NIXL_SUCCESS;
+    do {
+        ret = agent.createXferReq(NIXL_WRITE, dram_initiator_ucx, dram_target_ucx,
+                                  target, treq, &extra_params);
+    } while (ret == NIXL_ERR_NOT_FOUND);
+
+    if (ret != NIXL_SUCCESS) {
+        std::cerr << "Error creating transfer request " << ret <<"\n";
+        exit(-1);
+    }
 
-        std::cout << " Start Data Path Exchanges \n";
-        std::cout << " Waiting to receive Data from Initiator\n";
+    std::cout << " Post the request with UCX backend\n ";
+    ret = agent.postXferReq(treq);
+    std::cout << " Initiator posted Data Path transfer\n";
+    std::cout << " Waiting for completion\n";
 
-        while (!rc) {
-            //Only works with progress thread now, as backend is protected
-            /** Sanity Check */
-            for (int i = 0; i < NUM_TRANSFERS; i++) {
-                rc = allBytesAre(addr[i], SIZE, 0xbb);
-                if (!rc)
-                    break;
-            }
-        }
-        if (!rc)
-            std::cerr << " UCX Transfer failed, buffers are different\n";
-        else
-            std::cout << " Transfer completed and Buffers match with Initiator\n"
-                      <<"  UCX Transfer Success!!!\n";
-
-    } else {
-
-        std::cout << " Exchange metadata with Target \n";
-        nixl_opt_args_t md_extra_params;
-        md_extra_params.ipAddr = target_ip;
-        md_extra_params.port = target_port;
-
-        agent.fetchRemoteMD("target", &md_extra_params);
-
-        agent.sendLocalMD(&md_extra_params);
-
-        nixl_notifs_t notifs;
-
-        while(notifs.size() == 0) {
-            ret = agent.getNotifs(notifs, &extra_params);
-            assert(ret >= 0);
-        }
-        std::string rrstr = notifs["target"][0];
-        assert(rrstr.size() > 0);
-
-        remote_serdes->importStr(rrstr);
-
-        std::cout << " Verify Deserialized Target's Desc List at Initiator\n";
-        nixl_xfer_dlist_t dram_target_ucx(remote_serdes);
-        nixl_xfer_dlist_t dram_initiator_ucx = dram_for_ucx.trim();
-        dram_target_ucx.print();
-
-        std::cout << " End Control Path metadata exchanges \n";
-        std::cout << " Start Data Path Exchanges \n\n";
-        std::cout << " Create transfer request with UCX backend\n ";
-
-        // Need to do this in a loop with NIXL_ERR_NOT_FOUND
-        // UCX AM with desc list is faster than listener thread can recv/load MD with sockets
-        // Will be deprecated with ETCD or callbacks
-        do {
-            ret = agent.createXferReq(NIXL_WRITE, dram_initiator_ucx, dram_target_ucx,
-                                      "target", treq, &extra_params);
-        } while (ret == NIXL_ERR_NOT_FOUND);
-
-        if (ret != NIXL_SUCCESS) {
-            std::cerr << "Error creating transfer request " << ret <<"\n";
-            exit(-1);
-        }
-
-        std::cout << " Post the request with UCX backend\n ";
-        status = agent.postXferReq(treq);
-        std::cout << " Initiator posted Data Path transfer\n";
-        std::cout << " Waiting for completion\n";
-
-        while (status != NIXL_SUCCESS) {
-            status = agent.getXferStatus(treq);
-            assert(status >= 0);
-        }
-        std::cout << " Completed Sending Data using UCX backend\n";
-        agent.releaseXferReq(treq);
-        agent.invalidateLocalMD(&md_extra_params);
+    while (ret != NIXL_SUCCESS) {
+        ret = agent.getXferStatus(treq);
+        assert(ret >= 0);
     }
+    std::cout << " Completed Sending Data using UCX backend\n";
+    agent.releaseXferReq(treq);
+    agent.invalidateLocalMD(&md_extra_params);
 
     std::cout <<"Cleanup.. \n";
     agent.deregisterMem(dram_for_ucx, &extra_params);
-    for (int i = 0; i < NUM_TRANSFERS; i++) {
-        free(addr[i]);
+}
+
+int main(int argc, char *argv[]) {
+    /** Argument Parsing */
+    if (argc < 4) {
+        std::cout <<"Enter the required arguments\n" << std::endl;
+        std::cout <<"<Role> " <<"Target IP> <Target Port>"
+                  << std::endl;
+        exit(-1);
     }
-    delete serdes;
-    delete remote_serdes;
+
+    std::string role = std::string(argv[1]);
+    const char  *target_ip   = argv[2];
+    int         target_port = std::stoi(argv[3]);
+
+    std::transform(role.begin(), role.end(), role.begin(), ::tolower);
+
+    if (!role.compare(initiator) && !role.compare(target)) {
+            std::cerr << "Invalid role. Use 'initiator' or 'target'."
+                      << "Currently "<< role <<std::endl;
+            return 1;
+    }
+
+    /*** End - Argument Parsing */
+
+    if (role == target)
+        runTarget(target_ip, target_port);
+    else
+        runInitiator(target_ip, target_port);
 
     return 0;
 }

From fc4df5558ce82fc674cad95c55c72c93a0e2a7a9 Mon Sep 17 00:00:00 2001
From: Vlad Buslov <vladbu@nvidia.com>
Date: Fri, 25 Apr 2025 14:37:22 +0300
Subject: [PATCH 2/2] Implement multi-threaded execution in nixl_test

With refactoring from previous commit it becomes trivial to extend the test
with multi-threading, so just take the parts that makes sense to execute
concurrently (everything past agent creation and backend initialization)
from both initiator and target functions and extract them into dedicated
helpers.

The only tricky part is notification management in initiator - since NIXL
API doesn't provide a way to specify how many maximum total notifications
the caller of getNotifs() is ready to receive so it is also impossible to
guarantee that every thread will receive exactly one. Extend the per-thread
code to store serdes objects instantiated from notifications in a shared
vector protected with mutex and indexed by the thread id to remove the
requirement for every thread to receive exactly one.

Signed-off-by: Vlad Buslov <vladbu@nvidia.com>
---
 test/nixl/nixl_test.cpp | 172 ++++++++++++++++++++++++++--------------
 1 file changed, 111 insertions(+), 61 deletions(-)

diff --git a/test/nixl/nixl_test.cpp b/test/nixl/nixl_test.cpp
index 1939a8ba2b..d618a6bcb0 100644
--- a/test/nixl/nixl_test.cpp
+++ b/test/nixl/nixl_test.cpp
@@ -25,8 +25,11 @@
 #include <cassert>
 #include "stream/metadata_stream.h"
 #include "serdes/serdes.h"
+#include <mutex>
+#include <vector>
 
-#define NUM_TRANSFERS 1
+#define NUM_TRANSFERS 2
+#define NUM_THREADS 4
 #define SIZE 1024
 #define MEM_VAL 0xBB
 
@@ -37,6 +40,11 @@
  * target to initiator
  */
 
+struct SharedNotificationState {
+    std::mutex mtx;
+    std::vector<nixlSerDes> remote_serdes;
+};
+
 static const std::string target("target");
 static const std::string initiator("initiator");
 
@@ -61,41 +69,29 @@ static std::vector<std::unique_ptr<uint8_t[]>> initMem(nixlAgent &agent,
     return addrs;
 }
 
-static void runTarget(const std::string &ip, int port) {
-    nixlAgentConfig cfg(true, true, port);
-
-    std::cout << "Starting Agent for target\n";
-    nixlAgent     agent(target, cfg);
-
-    nixl_b_params_t params;
-    nixlBackendH    *ucx;
-    agent.createBackend("UCX", params, ucx);
-
-    nixl_opt_args_t extra_params;
-    extra_params.backends.push_back(ucx);
-
+static void targetThread(nixlAgent &agent, nixl_opt_args_t *extra_params, int thread_id) {
     nixl_reg_dlist_t dram_for_ucx(DRAM_SEG);
-    auto addrs = initMem(agent, dram_for_ucx, &extra_params, 0);
+    auto addrs = initMem(agent, dram_for_ucx, extra_params, 0);
 
     nixl_blob_t tgt_metadata;
     agent.getLocalMD(tgt_metadata);
 
-    std::cout << " Start Control Path metadata exchanges \n";
+    std::cout << "Thread " << thread_id << " Start Control Path metadata exchanges\n";
 
-    std::cout << " Desc List from Target to Initiator\n";
+    std::cout << "Thread " << thread_id << " Desc List from Target to Initiator\n";
     dram_for_ucx.print();
 
     /** Only send desc list */
     nixlSerDes serdes;
     assert(dram_for_ucx.trim().serialize(&serdes) == NIXL_SUCCESS);
 
-    std::cout << " Wait for initiator and then send xfer descs\n";
+    std::cout << "Thread " << thread_id << " Wait for initiator and then send xfer descs\n";
     std::string message = serdes.exportStr();
-    while (agent.genNotif(initiator, message, &extra_params) != NIXL_SUCCESS);
-    std::cout << " End Control Path metadata exchanges \n";
+    while (agent.genNotif(initiator, message, extra_params) != NIXL_SUCCESS);
+    std::cout << "Thread " << thread_id << " End Control Path metadata exchanges\n";
 
-    std::cout << " Start Data Path Exchanges \n";
-    std::cout << " Waiting to receive Data from Initiator\n";
+    std::cout << "Thread " << thread_id << " Start Data Path Exchanges\n";
+    std::cout << "Thread " << thread_id << " Waiting to receive Data from Initiator\n";
 
     bool rc = false;
     for (int n_tries = 0; !rc && n_tries < 100; n_tries++) {
@@ -110,33 +106,23 @@ static void runTarget(const std::string &ip, int port) {
             std::this_thread::sleep_for(std::chrono::milliseconds(10));
     }
     if (!rc)
-        std::cerr << " UCX Transfer failed, buffers are different\n";
+        std::cerr << "Thread " << thread_id << " UCX Transfer failed, buffers are different\n";
     else
-        std::cout << " Transfer completed and Buffers match with Initiator\n"
-                  <<"  UCX Transfer Success!!!\n";
+        std::cout << "Thread " << thread_id << " Transfer completed and Buffers match with Initiator\n"
+                  << "Thread " << thread_id << " UCX Transfer Success!!!\n";
 
-    std::cout <<"Cleanup.. \n";
-    agent.deregisterMem(dram_for_ucx, &extra_params);
+    std::cout << "Thread " << thread_id << " Cleanup..\n";
+    agent.deregisterMem(dram_for_ucx, extra_params);
 }
 
-static void runInitiator(const std::string &target_ip, int target_port) {
-    nixlAgentConfig cfg(true, true);
-
-    std::cout << "Starting Agent for initiator\n";
-    nixlAgent     agent(initiator, cfg);
-
-    nixl_b_params_t params;
-    nixlBackendH    *ucx;
-    agent.createBackend("UCX", params, ucx);
-
-    nixl_opt_args_t extra_params;
-    extra_params.backends.push_back(ucx);
-
+static void initiatorThread(nixlAgent &agent, nixl_opt_args_t *extra_params,
+                          const std::string &target_ip, int target_port, int thread_id,
+                          SharedNotificationState &shared_state) {
     nixl_reg_dlist_t dram_for_ucx(DRAM_SEG);
-    auto addrs = initMem(agent, dram_for_ucx, &extra_params, MEM_VAL);
+    auto addrs = initMem(agent, dram_for_ucx, extra_params, MEM_VAL);
 
-    std::cout << " Start Control Path metadata exchanges \n";
-    std::cout << " Exchange metadata with Target \n";
+    std::cout << "Thread " << thread_id << " Start Control Path metadata exchanges\n";
+    std::cout << "Thread " << thread_id << " Exchange metadata with Target\n";
 
     nixl_opt_args_t md_extra_params;
     md_extra_params.ipAddr = target_ip;
@@ -146,25 +132,44 @@ static void runInitiator(const std::string &target_ip, int target_port) {
 
     agent.sendLocalMD(&md_extra_params);
 
-    nixl_notifs_t notifs;
-    while(notifs.size() == 0) {
-        nixl_status_t ret = agent.getNotifs(notifs, &extra_params);
+    // Wait for notifications and populate shared state
+    while (true) {
+        {
+            std::lock_guard<std::mutex> lock(shared_state.mtx);
+            if (shared_state.remote_serdes.size() >= NUM_THREADS) {
+                break;
+            }
+        }
+
+        nixl_notifs_t notifs;
+        nixl_status_t ret = agent.getNotifs(notifs, extra_params);
         assert(ret >= 0);
+
+        if (notifs.size() > 0) {
+            std::lock_guard<std::mutex> lock(shared_state.mtx);
+            for (const auto &notif : notifs[target]) {
+                nixlSerDes serdes;
+                serdes.importStr(notif);
+                shared_state.remote_serdes.push_back(serdes);
+            }
+        }
     }
-    std::string rrstr = notifs[target][0];
-    assert(rrstr.size() > 0);
 
+    // Get our thread's serdes instance
     nixlSerDes remote_serdes;
-    remote_serdes.importStr(rrstr);
+    {
+        std::lock_guard<std::mutex> lock(shared_state.mtx);
+        remote_serdes = shared_state.remote_serdes[thread_id];
+    }
 
-    std::cout << " Verify Deserialized Target's Desc List at Initiator\n";
+    std::cout << "Thread " << thread_id << " Verify Deserialized Target's Desc List at Initiator\n";
     nixl_xfer_dlist_t dram_target_ucx(&remote_serdes);
     nixl_xfer_dlist_t dram_initiator_ucx = dram_for_ucx.trim();
     dram_target_ucx.print();
 
-    std::cout << " End Control Path metadata exchanges \n";
-    std::cout << " Start Data Path Exchanges \n\n";
-    std::cout << " Create transfer request with UCX backend\n ";
+    std::cout << "Thread " << thread_id << " End Control Path metadata exchanges\n";
+    std::cout << "Thread " << thread_id << " Start Data Path Exchanges\n\n";
+    std::cout << "Thread " << thread_id << " Create transfer request with UCX backend\n";
 
     // Need to do this in a loop with NIXL_ERR_NOT_FOUND
     // UCX AM with desc list is faster than listener thread can recv/load MD with sockets
@@ -173,29 +178,74 @@ static void runInitiator(const std::string &target_ip, int target_port) {
     nixl_status_t ret = NIXL_SUCCESS;
     do {
         ret = agent.createXferReq(NIXL_WRITE, dram_initiator_ucx, dram_target_ucx,
-                                  target, treq, &extra_params);
+                                  target, treq, extra_params);
     } while (ret == NIXL_ERR_NOT_FOUND);
 
     if (ret != NIXL_SUCCESS) {
-        std::cerr << "Error creating transfer request " << ret <<"\n";
+        std::cerr << "Thread " << thread_id << " Error creating transfer request " << ret << "\n";
         exit(-1);
     }
 
-    std::cout << " Post the request with UCX backend\n ";
+    std::cout << "Thread " << thread_id << " Post the request with UCX backend\n";
     ret = agent.postXferReq(treq);
-    std::cout << " Initiator posted Data Path transfer\n";
-    std::cout << " Waiting for completion\n";
+    std::cout << "Thread " << thread_id << " Initiator posted Data Path transfer\n";
+    std::cout << "Thread " << thread_id << " Waiting for completion\n";
 
     while (ret != NIXL_SUCCESS) {
         ret = agent.getXferStatus(treq);
         assert(ret >= 0);
     }
-    std::cout << " Completed Sending Data using UCX backend\n";
+    std::cout << "Thread " << thread_id << " Completed Sending Data using UCX backend\n";
     agent.releaseXferReq(treq);
     agent.invalidateLocalMD(&md_extra_params);
 
-    std::cout <<"Cleanup.. \n";
-    agent.deregisterMem(dram_for_ucx, &extra_params);
+    std::cout << "Thread " << thread_id << " Cleanup..\n";
+    agent.deregisterMem(dram_for_ucx, extra_params);
+}
+
+static void runTarget(const std::string &ip, int port) {
+    nixlAgentConfig cfg(true, true, port, 0, 100000, nixl_thread_sync_t::NIXL_THREAD_SYNC_STRICT);
+
+    std::cout << "Starting Agent for target\n";
+    nixlAgent agent(target, cfg);
+
+    nixl_b_params_t params;
+    nixlBackendH *ucx;
+    agent.createBackend("UCX", params, ucx);
+
+    nixl_opt_args_t extra_params;
+    extra_params.backends.push_back(ucx);
+
+    std::vector<std::thread> threads;
+    for (int i = 0; i < NUM_THREADS; i++)
+        threads.emplace_back(targetThread, std::ref(agent), &extra_params, i);
+
+    for (auto &thread : threads)
+        thread.join();
+}
+
+static void runInitiator(const std::string &target_ip, int target_port) {
+    nixlAgentConfig cfg(true, true, 0, 0, 100000, nixl_thread_sync_t::NIXL_THREAD_SYNC_STRICT);
+
+    std::cout << "Starting Agent for initiator\n";
+    nixlAgent agent(initiator, cfg);
+
+    nixl_b_params_t params;
+    nixlBackendH *ucx;
+    agent.createBackend("UCX", params, ucx);
+
+    nixl_opt_args_t extra_params;
+    extra_params.backends.push_back(ucx);
+
+    SharedNotificationState shared_state;
+
+    std::vector<std::thread> threads;
+    for (int i = 0; i < NUM_THREADS; i++)
+        threads.emplace_back(initiatorThread, std::ref(agent), &extra_params,
+                             target_ip, target_port, i, std::ref(shared_state));
+
+    for (auto &thread : threads)
+        thread.join();
 }
 
 int main(int argc, char *argv[]) {