Implement new synchronization read-write locking mode

src/api/cpp/backend/backend_aux.h

@@ -51,6 +51,7 @@ class nixlBackendInitParams {
 
         bool              enableProgTh;
         nixlTime::us_t    pthrDelay;
+        nixl_thread_sync_t syncMode;
 };
 
 // Pure virtual class to have a common pointer type

src/api/cpp/nixl_types.h

@@ -70,6 +70,7 @@ enum nixl_status_t {
 enum class nixl_thread_sync_t {
     NIXL_THREAD_SYNC_NONE,
     NIXL_THREAD_SYNC_STRICT,
+    NIXL_THREAD_SYNC_RW,
     NIXL_THREAD_SYNC_DEFAULT = NIXL_THREAD_SYNC_NONE,
 };
 

src/core/nixl_agent.cpp

@@ -213,6 +213,7 @@ nixlAgent::createBackend(const nixl_backend_t &type,
     init_params.customParams = const_cast<nixl_b_params_t*>(&params);
     init_params.enableProgTh = data->config.useProgThread;
     init_params.pthrDelay    = data->config.pthrDelay;
+    init_params.syncMode     = data->config.syncMode;
 
     // First, try to load the backend as a plugin
     auto& plugin_manager = nixlPluginManager::getInstance();
@@ -654,7 +655,7 @@ nixlAgent::createXferReq(const nixl_xfer_op_t &operation,
 
     req_hndl = nullptr;
 
-    NIXL_LOCK_GUARD(data->lock);
+    NIXL_SHARED_LOCK_GUARD(data->lock);
     if (data->remoteSections.count(remote_agent) == 0)
     {
         return NIXL_ERR_NOT_FOUND;
@@ -772,7 +773,7 @@ nixlAgent::postXferReq(nixlXferReqH *req_hndl,
     if (!req_hndl)
         return NIXL_ERR_INVALID_PARAM;
 
-    NIXL_LOCK_GUARD(data->lock);
+    NIXL_SHARED_LOCK_GUARD(data->lock);
     // Check if the remote was invalidated before post/repost
     if (data->remoteSections.count(req_hndl->remoteAgent) == 0) {
         delete req_hndl;
@@ -827,7 +828,7 @@ nixlAgent::postXferReq(nixlXferReqH *req_hndl,
 nixl_status_t
 nixlAgent::getXferStatus (nixlXferReqH *req_hndl) const {
 
-    NIXL_LOCK_GUARD(data->lock);
+    NIXL_SHARED_LOCK_GUARD(data->lock);
     // If the status is done, no need to recheck.
     if (req_hndl->status == NIXL_IN_PROG) {
         // Check if the remote was invalidated before completion
@@ -854,7 +855,7 @@ nixlAgent::queryXferBackend(const nixlXferReqH* req_hndl,
 nixl_status_t
 nixlAgent::releaseXferReq(nixlXferReqH *req_hndl) const {
 
-    NIXL_LOCK_GUARD(data->lock);
+    NIXL_SHARED_LOCK_GUARD(data->lock);
     //attempt to cancel request
     if(req_hndl->status == NIXL_IN_PROG) {
         req_hndl->status = req_hndl->engine->checkXfer(
@@ -942,7 +943,7 @@ nixlAgent::genNotif(const std::string &remote_agent,
     nixlBackendEngine* backend = nullptr;
     backend_list_t*    backend_list;
 
-    NIXL_LOCK_GUARD(data->lock);
+    NIXL_SHARED_LOCK_GUARD(data->lock);
     if (!extra_params || extra_params->backends.size() == 0) {
         backend_list = &data->notifEngines;
         if (backend_list->empty())

src/core/sync.h

@@ -18,30 +18,67 @@
 #define SYNC_H
 #include "common/util.h"
 #include "nixl_params.h"
-#include <mutex>
+#include "absl/synchronization/mutex.h"
+#include <shared_mutex>
 
 class nixlLock {
     public:
-        nixlLock(const nixl_thread_sync_t sync_mode): syncMode(sync_mode)
-        {}
+        nixlLock(const nixl_thread_sync_t sync_mode) {
+            switch (sync_mode) {
+            case nixl_thread_sync_t::NIXL_THREAD_SYNC_NONE:
+                lock_cb = unlock_cb = lock_shared_cb = unlock_shared_cb = []() {};
+                break;
+            case nixl_thread_sync_t::NIXL_THREAD_SYNC_STRICT:
+                lock_cb = lock_shared_cb = [this]() {
+                    m.Lock();
+                };
+                unlock_cb = unlock_shared_cb = [this]() {
+                    m.Unlock();
+                };
+                break;
+            case nixl_thread_sync_t::NIXL_THREAD_SYNC_RW:
+                lock_cb = [this]() {
+                    m.Lock();
+                };
+                unlock_cb = [this]() {
+                    m.Unlock();
+                };
+                lock_shared_cb = [this]() {
+                    m.ReaderLock();
+                };
+                unlock_shared_cb = [this]() {
+                    m.ReaderUnlock();
+                };
+                break;
+            }
+        }
 
         void lock() {
-            if (syncMode == nixl_thread_sync_t::NIXL_THREAD_SYNC_STRICT) {
-                m.lock();
-            }
+            lock_cb();
+        }
+
+        void lock_shared() {
+            lock_shared_cb();
         }
 
         void unlock() {
-            if (syncMode == nixl_thread_sync_t::NIXL_THREAD_SYNC_STRICT) {
-                m.unlock();
-            }
+            unlock_cb();
+        }
+
+        void unlock_shared() {
+            unlock_shared_cb();
         }
 
     private:
-        nixl_thread_sync_t syncMode;
-        std::mutex m;
+        std::function<void()> lock_cb;
+        std::function<void()> unlock_cb;
+        std::function<void()> lock_shared_cb;
+        std::function<void()> unlock_shared_cb;
+
+        absl::Mutex m;
 };
 
 #define NIXL_LOCK_GUARD(lock) const std::lock_guard<nixlLock> UNIQUE_NAME(lock_guard) (lock)
+#define NIXL_SHARED_LOCK_GUARD(lock) const std::shared_lock<nixlLock> UNIQUE_NAME(lock_guard) (lock)
 
 #endif /* SYNC_H */

src/plugins/ucx/ucx_backend.cpp

@@ -485,8 +485,8 @@ nixlUcxEngine::nixlUcxEngine (const nixlBackendInitParams* init_params)
     uc = std::make_shared<nixlUcxContext>(devs, sizeof(nixlUcxIntReq),
                                           _internalRequestInit,
                                           _internalRequestFini,
-                                          NIXL_UCX_MT_WORKER, pthrOn,
-                                          err_handling_mode);
+                                          pthrOn,
+                                          err_handling_mode, numWorkers, init_params->syncMode);
 
     for (unsigned int i = 0; i < numWorkers; i++)
         uws.emplace_back(std::make_unique<nixlUcxWorker>(uc));

src/utils/common/meson.build

@@ -18,6 +18,7 @@ absl_log_dep = abseil_proj.get_variable('absl_log_dep')
 absl_strings_dep = abseil_proj.get_variable('absl_strings_dep')
 absl_status_dep = abseil_proj.get_variable('absl_status_dep')
 absl_strings_dep = abseil_proj.get_variable('absl_strings_dep')
+absl_synchronization_dep = abseil_proj.get_variable('absl_synchronization_dep')
 
 nixl_common_inc = include_directories('.')
 
@@ -26,6 +27,7 @@ nixl_common_deps = [
     absl_strings_dep,
     absl_status_dep,
     absl_strings_dep,
+    absl_synchronization_dep,
 ]
 
 # Define a shared library for common utilities

src/utils/ucx/ucx_utils.cpp

@@ -306,34 +306,28 @@ nixlUcxContext::nixlUcxContext(std::vector<std::string> devs,
                                size_t req_size,
                                nixlUcxContext::req_cb_t init_cb,
                                nixlUcxContext::req_cb_t fini_cb,
-                               nixl_ucx_mt_t __mt_type,
                                bool prog_thread,
-                               ucp_err_handling_mode_t __err_handling_mode)
+                               ucp_err_handling_mode_t __err_handling_mode,
+                               unsigned long num_workers,
+                               nixl_thread_sync_t sync_mode)
 {
     ucp_params_t ucp_params;
     ucp_config_t *ucp_config;
     ucs_status_t status = UCS_OK;
 
-    mt_type           = __mt_type;
+    // With strict synchronization model nixlAgent serializes access to backends, with more
+    // permissive models backends need to account for concurrent access and ensure their internal
+    // state is properly protected. Progress thread creates internal concurrency in UCX backend
+    // irrespective of nixlAgent synchronization model.
+    mt_type = (sync_mode == nixl_thread_sync_t::NIXL_THREAD_SYNC_RW || prog_thread) ?
+        NIXL_UCX_MT_WORKER : NIXL_UCX_MT_SINGLE;
     err_handling_mode = __err_handling_mode;
 
     ucp_params.field_mask = UCP_PARAM_FIELD_FEATURES | UCP_PARAM_FIELD_MT_WORKERS_SHARED;
     ucp_params.features = UCP_FEATURE_RMA | UCP_FEATURE_AMO32 | UCP_FEATURE_AMO64 | UCP_FEATURE_AM;
     if (prog_thread)
         ucp_params.features |= UCP_FEATURE_WAKEUP;
-
-    switch(mt_type) {
-    case NIXL_UCX_MT_SINGLE:
-        ucp_params.mt_workers_shared = 0;
-        break;
-    case NIXL_UCX_MT_WORKER:
-    case NIXL_UCX_MT_CTX:
-        ucp_params.mt_workers_shared = 1;
-        break;
-    default:
-        assert(mt_type < NIXL_UCX_MT_MAX);
-        abort();
-    }
+    ucp_params.mt_workers_shared = num_workers > 1 ? 1 : 0;
 
     if (req_size) {
         ucp_params.request_size = req_size;

src/utils/ucx/ucx_utils.h

@@ -129,8 +129,8 @@ class nixlUcxContext {
     using req_cb_t = void(void *request);
     nixlUcxContext(std::vector<std::string> devices,
                    size_t req_size, req_cb_t init_cb, req_cb_t fini_cb,
-                   nixl_ucx_mt_t mt_type, bool prog_thread,
-                   ucp_err_handling_mode_t err_handling_mode);
+                   bool prog_thread, ucp_err_handling_mode_t err_handling_mode,
+                   unsigned long num_workers, nixl_thread_sync_t sync_mode);
     ~nixlUcxContext();
 
     static bool mtLevelIsSupproted(nixl_ucx_mt_t mt_type);

test/gtest/mocks/mock_dram_engine.h

@@ -74,6 +74,7 @@ class MockDramBackendEngine : public nixlBackendEngine {
   }
   nixl_status_t getConnInfo(std::string &str) const override {
     assert(sharedState > 0);
+    str = "MockDramBackendEngineConnInfo";
     return NIXL_SUCCESS;
   }
   nixl_status_t loadRemoteConnInfo(const std::string &remote_agent,

test/gtest/multi_threading.cpp

@@ -28,10 +28,12 @@ class MultiThreadingTestFixture : public testing::Test {
     uintptr_t addr = 0;
     size_t len = 1024;
     uint64_t dev_id = 0;
+    std::string local_agent_name = "test_agent";
+    std::string remote_agent_name = "remote_agent";
 
-    nixlAgent createAgent() {
-        nixlAgentConfig cfg(false, false, 0, 0, 100000, nixl_thread_sync_t::NIXL_THREAD_SYNC_STRICT);
-        return nixlAgent("test_agent", cfg);
+    nixlAgent createAgent(const std::string &name) {
+        nixlAgentConfig cfg(false, false, 0, nixl_thread_sync_t::NIXL_THREAD_SYNC_RW);
+        return nixlAgent(name, cfg);
     }
 
     nixl_opt_args_t createExtraParams(nixlBackendH* backend) {
@@ -43,7 +45,7 @@ class MultiThreadingTestFixture : public testing::Test {
     nixlBackendH* verifyMockDramBackendCreation(nixlAgent& agent) {
         nixlBackendH* backend_handle = nullptr;
         nixl_b_params_t params;
-        nixl_status_t status = agent.createBackend("MOCK_DRAM", params, backend_handle);
+        auto status = agent.createBackend("MOCK_DRAM", params, backend_handle);
         EXPECT_EQ(status, NIXL_SUCCESS);
         EXPECT_NE(backend_handle, nullptr);
         return backend_handle;
@@ -54,7 +56,7 @@ class MultiThreadingTestFixture : public testing::Test {
         nixlDescList<nixlBlobDesc> desc_list(DRAM_SEG);
         desc_list.addDesc(blob);
 
-        nixl_status_t status = agent.registerMem(desc_list, &extra_params);
+        auto status = agent.registerMem(desc_list, &extra_params);
         EXPECT_EQ(status, NIXL_SUCCESS);
     }
 
@@ -67,7 +69,7 @@ class MultiThreadingTestFixture : public testing::Test {
         src_list.addDesc(basic_desc);
         dst_list.addDesc(basic_desc);
 
-        nixl_status_t status = agent.createXferReq(NIXL_WRITE, src_list, dst_list, "test_agent", xfer_req, &extra_params);
+        auto status = agent.createXferReq(NIXL_WRITE, src_list, dst_list, "test_agent", xfer_req, &extra_params);
         EXPECT_EQ(status, NIXL_SUCCESS);
         EXPECT_NE(xfer_req, nullptr);
 
@@ -84,7 +86,7 @@ class MultiThreadingTestFixture : public testing::Test {
 
 TEST_F(MultiThreadingTestFixture, ConcurrentTransfersWithPerThreadAgent) {
     auto transfer_sequence = [&]() {
-        nixlAgent agent = createAgent();
+        nixlAgent agent = createAgent(local_agent_name);
         nixlBackendH* backend = verifyMockDramBackendCreation(agent);
         nixl_opt_args_t extra_params = createExtraParams(backend);
 
@@ -101,7 +103,7 @@ TEST_F(MultiThreadingTestFixture, ConcurrentTransfersWithPerThreadAgent) {
 
 TEST_F(MultiThreadingTestFixture, ConcurrentAddPlugingDirWithPerThreadAgent) {
     auto transfer_sequence = [&]() {
-        nixlAgent agent = createAgent();
+        nixlAgent agent = createAgent(local_agent_name);
         nixlBackendH* backend = verifyMockDramBackendCreation(agent);
         nixl_opt_args_t extra_params = createExtraParams(backend);
 
@@ -122,7 +124,7 @@ TEST_F(MultiThreadingTestFixture, ConcurrentAddPlugingDirWithPerThreadAgent) {
 }
 
 TEST_F(MultiThreadingTestFixture, ConcurrentTransfersWithPerThreadMemory) {
-    nixlAgent agent = createAgent();
+    nixlAgent agent = createAgent(local_agent_name);
     nixlBackendH* backend = verifyMockDramBackendCreation(agent);
     nixl_opt_args_t extra_params = createExtraParams(backend);
 
@@ -139,7 +141,7 @@ TEST_F(MultiThreadingTestFixture, ConcurrentTransfersWithPerThreadMemory) {
 }
 
 TEST_F(MultiThreadingTestFixture, ConcurrentTransfers) {
-    nixlAgent agent = createAgent();
+    nixlAgent agent = createAgent(local_agent_name);
     nixlBackendH* backend = verifyMockDramBackendCreation(agent);
     nixl_opt_args_t extra_params = createExtraParams(backend);
 
@@ -156,8 +158,35 @@ TEST_F(MultiThreadingTestFixture, ConcurrentTransfers) {
     t2.join();
 }
 
+TEST_F(MultiThreadingTestFixture, GenerateNotification) {
+    nixlAgent local_agent = createAgent(local_agent_name);
+    nixlBackendH* backend = verifyMockDramBackendCreation(local_agent);
+    nixlAgent remote_agent = createAgent(remote_agent_name);
+    verifyMockDramBackendCreation(remote_agent);
+    nixl_opt_args_t extra_params = createExtraParams(backend);
+
+    verifyMemoryRegistration(local_agent, extra_params);
+
+    std::string md;
+    auto status = remote_agent.getLocalMD(md);
+    EXPECT_EQ(status, NIXL_SUCCESS);
+    status = local_agent.loadRemoteMD(md, remote_agent_name);
+    EXPECT_EQ(status, NIXL_SUCCESS);
+
+    auto transfer_sequence = [&]() {
+        auto status = local_agent.genNotif(remote_agent_name, "hello!");
+        EXPECT_EQ(status, NIXL_SUCCESS);
+    };
+
+    std::thread t1(transfer_sequence);
+    std::thread t2(transfer_sequence);
+
+    t1.join();
+    t2.join();
+}
+
 TEST_F(MultiThreadingTestFixture, RegisterMemWithMockDram) {
-    nixlAgent agent = createAgent();
+    nixlAgent agent = createAgent(local_agent_name);
     nixlBackendH* backend = verifyMockDramBackendCreation(agent);
     nixl_opt_args_t extra_params = createExtraParams(backend);