Added posix plugin

Author: dfarge

meson.build

@@ -1,3 +1,5 @@
+#!/usr/bin/env meson
+
 # SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
 # SPDX-License-Identifier: Apache-2.0
 #

src/core/nixl_plugin_manager.cpp

@@ -371,5 +371,10 @@ void nixlPluginManager::registerBuiltinPlugins() {
         extern nixlBackendPlugin* createStaticGdsPlugin();
         registerStaticPlugin("GDS", createStaticGdsPlugin);
     #endif
+
+    #ifdef STATIC_PLUGIN_POSIX
+        extern nixlBackendPlugin* createStaticPosixPlugin();
+        registerStaticPlugin("POSIX", createStaticPosixPlugin);
+    #endif
     #endif
 }

src/plugins/meson.build

@@ -17,6 +17,7 @@ ucx_backend_inc_dirs = include_directories('./ucx')
 
 subdir('ucx')
 subdir('ucx_mo')
+subdir('posix')
 
 disable_gds_backend = get_option('disable_gds_backend')
 if not disable_gds_backend and cuda_dep.found()

src/plugins/posix/meson.build

@@ -0,0 +1,50 @@
+#!/usr/bin/env meson
+
+# SPDX-FileCopyrightText: Copyright (c) 2025 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.
+
+# Check for liburing dependency
+liburing_dep = dependency('liburing', required: true)
+
+if 'POSIX' in static_plugins
+    posix_backend_lib = static_library('POSIX',
+        'posix_backend.cpp', 'posix_backend.h',
+        'posix_plugin.cpp',
+        dependencies: [nixl_infra, liburing_dep],
+        link_args: ['-luring'],
+        include_directories: [nixl_inc_dirs, utils_inc_dirs],
+        install: false,
+        cpp_args: compile_flags,
+        name_prefix: 'libplugin_')  # Custom prefix for plugin libraries
+else
+    posix_backend_lib = shared_library('POSIX',
+        'posix_backend.cpp', 'posix_backend.h',
+        'posix_plugin.cpp',
+        dependencies: [nixl_infra, liburing_dep],
+        include_directories: [nixl_inc_dirs, utils_inc_dirs],
+        link_args: ['-luring'],
+        install: true,
+        cpp_args: ['-fPIC'],
+        name_prefix: 'libplugin_',  # Custom prefix for plugin libraries
+        install_dir: plugin_install_dir)
+    if get_option('buildtype') == 'debug'
+        run_command('sh', '-c',
+            'echo "POSIX=' + posix_backend_lib.full_path() + '" >> ' + plugin_build_dir + '/pluginlist',
+            check: true
+        )
+    endif
+endif
+
+posix_backend_interface = declare_dependency(link_with: posix_backend_lib)

src/plugins/posix/posix_backend.cpp

@@ -0,0 +1,319 @@
+/*
+ * SPDX-FileCopyrightText: Copyright (c) 2025 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 <iostream>
+#include <liburing.h>
+#include <algorithm>
+#include <cmath>
+#include "posix_backend.h"
+#include <unordered_map>
+
+namespace {
+    static constexpr int max_posix_ring_log = 6; //posix_ring_log
+    static constexpr int max_posix_ring_size = 1 << max_posix_ring_log;
+}
+
+void nixlPosixBackendReqH::createUringParams() {
+    // We only need 2 params, as only the last one will be the remainder, and others will be max_posix_ring_size
+    num_uring_params = std::min(num_urings, 2);
+    uring_params.resize(num_uring_params);
+
+    // Calculate the remainder, but if it's 0, set to max_posix_ring_size instead
+    unsigned int remainder = (this->local_desc_count - 1) % max_posix_ring_size + 1;
+
+    switch (num_uring_params) {
+        case 2:
+            uring_params[1] = {
+                .sq_entries = max_posix_ring_size,
+                .cq_entries = max_posix_ring_size,
+            };
+        case 1:
+            uring_params[0] = {
+                .sq_entries = remainder,
+                .cq_entries = remainder,
+            };
+            break;
+        default:
+            throw UringInitError();
+    }
+}
+
+void nixlPosixBackendReqH::exitUrings(int num_urings) {
+    for (int i = 0; i < num_urings; ++i)
+        io_uring_queue_exit(&uring[i]);
+}
+
+int nixlPosixBackendReqH::createUringsAndNumEntries() {
+    num_entries.resize(num_urings);
+    uring.resize(num_urings);
+    // Calculate the remainder, but if it's 0, set to max_posix_ring_size instead
+    unsigned int remainder = (this->local_desc_count - 1) % max_posix_ring_size + 1;
+
+    for (int i = 0; i < num_urings - 1; ++i) {
+        num_entries[i] = max_posix_ring_size;
+        int uring_init_status = io_uring_queue_init_params(max_posix_ring_size, &uring[i], &uring_params[1]);
+        if (uring_init_status != 0) {
+            std::cerr << "Error in initializing io_uring " << i << "\n";
+            this->exitUrings(i);
+            return uring_init_status;
+        }
+    }
+
+    num_entries[num_urings - 1] = remainder;
+    int uring_init_status = io_uring_queue_init_params(remainder, &uring[num_urings - 1], &uring_params[0]);
+    if (uring_init_status != 0) {
+        std::cerr << "Error in initializing io_uring " << num_urings - 1 << "\n";
+        this->exitUrings(num_urings);
+    }
+
+    return uring_init_status;
+}
+
+const std::unordered_map<nixl_xfer_op_t, nixlPosixBackendReqH::io_uring_prep_func_t> nixlPosixBackendReqH::op_to_func = {
+    {NIXL_READ, (io_uring_prep_func_t)io_uring_prep_read},
+    {NIXL_WRITE, (io_uring_prep_func_t)io_uring_prep_write}
+};
+
+nixlPosixBackendReqH::nixlPosixBackendReqH(const nixl_xfer_op_t &operation,
+                                           const nixl_meta_dlist_t &local,
+                                           const nixl_meta_dlist_t &remote,
+                                           const nixl_opt_b_args_t* opt_args)
+    : local_desc_count(local.descCount()), operation(operation), local(local),
+      remote(remote), opt_args(opt_args), is_prepped(false), erred(false)
+{
+    // Initialize io_uring_prep_func based on the operation
+    auto it = this->op_to_func.find(operation);
+    if (it == this->op_to_func.end()) {
+        throw OperationError();
+    }
+    this->io_uring_prep_func = it->second;
+
+    // Calculate the number of io_urings needed based on the number of descriptors
+    this->num_urings = static_cast<int>(std::ceil(static_cast<double>(local_desc_count) / max_posix_ring_size));
+
+    this->num_completed.resize(this->num_urings, 0);
+
+    this->createUringParams();
+
+    if (this->createUringsAndNumEntries() != 0)
+        throw UringInitError();
+}
+
+nixlPosixBackendReqH::~nixlPosixBackendReqH() {
+    this->exitUrings(this->num_urings);
+}
+
+bool nixlPosixBackendReqH::isErred() const {
+    return erred;
+}
+
+nixl_status_t nixlPosixBackendReqH::updateNumCompleted(unsigned int uring_idx) {
+    if (this->isErred()) // TODO: Talk with team if this is the correct error code
+        return NIXL_ERR_INVALID_PARAM;
+
+    if (this->num_completed[uring_idx] == this->num_entries[uring_idx])
+        return NIXL_SUCCESS;
+
+    struct io_uring_cqe *cqes[max_posix_ring_size];
+
+    int num_ret_cqes = io_uring_peek_batch_cqe(&this->uring[uring_idx], cqes, max_posix_ring_size);
+    for (int i = 0; i < num_ret_cqes; ++i) {
+        int res = cqes[i]->res;
+        if (res < 0) {
+            io_uring_cqe_seen(&this->uring[uring_idx], cqes[i]);
+            this->erred = true;
+            return NIXL_ERR_BACKEND;
+        }
+        io_uring_cqe_seen(&this->uring[uring_idx], cqes[i]);
+    }
+
+    this->num_completed[uring_idx] += num_ret_cqes;
+    return NIXL_SUCCESS;
+}
+
+bool nixlPosixBackendReqH::isPrepped() const {
+    return this->is_prepped;
+}
+
+nixl_status_t nixlPosixBackendReqH::prepXfer() {
+    if (this->isErred()) // TODO: Talk with team if this is the correct error code
+        return NIXL_ERR_INVALID_PARAM;
+
+    if (this->is_prepped)
+        return NIXL_SUCCESS;
+
+    struct io_uring_sqe *entry;
+
+    for (int global_entry_index = 0; global_entry_index < this->local_desc_count; ++global_entry_index) {
+        // Calculate the index of the io_uring to use for this entry
+        int uring_index = global_entry_index / max_posix_ring_size;
+        entry = io_uring_get_sqe(&this->uring[uring_index]);
+        if (!entry) {
+            std::cerr << "Error in getting sqe\n";
+            return NIXL_ERR_BACKEND;
+        }
+        io_uring_prep_func(entry, this->remote[global_entry_index].devId,
+                           (void *)this->local[global_entry_index].addr,
+                           this->remote[global_entry_index].len,
+                           this->remote[global_entry_index].addr);
+    }
+
+    this->is_prepped = true;
+    return NIXL_SUCCESS;
+}
+
+nixl_status_t nixlPosixBackendReqH::checkXfer() {
+    if (this->isErred()) // TODO: Talk with team if this is the correct error code
+        return NIXL_ERR_INVALID_PARAM;
+
+    nixl_status_t status;
+    for (int i = 0; i < this->num_urings; ++i) {
+        status = this->updateNumCompleted(i);
+        if (status != NIXL_SUCCESS)
+            return status;
+        if (this->num_completed[i] != this->num_entries[i])
+            return NIXL_IN_PROG;
+    }
+
+    return NIXL_SUCCESS;
+}
+
+nixl_status_t nixlPosixBackendReqH::postXfer() {
+    for (int i = 0; i < this->num_urings; ++i) {
+        int ret = io_uring_submit(&this->uring[i]);
+        if (ret !=  this->num_entries[i]) {
+            std::cerr << "Error in submitting io_uring " << i << " - ";
+            std::cerr << "ret: " << ret << " - num_entries: " << this->num_entries[i] << std::endl;
+            return NIXL_ERR_BACKEND;
+        }
+    }
+    return NIXL_IN_PROG;
+}
+
+const nixl_mem_list_t nixlPosixEngine::supported_mems = {
+    FILE_SEG,
+    DRAM_SEG
+};
+
+nixl_status_t nixlPosixEngine::registerMem(const nixlBlobDesc &mem,
+                                           const nixl_mem_t &nixl_mem,
+                                           nixlBackendMD* &out) {
+    if (std::find(supported_mems.begin(), supported_mems.end(), nixl_mem) != supported_mems.end())
+        return NIXL_SUCCESS;
+
+    return NIXL_ERR_NOT_SUPPORTED;
+}
+
+nixl_status_t nixlPosixEngine::deregisterMem(nixlBackendMD *) {
+    return NIXL_SUCCESS;
+}
+
+bool isPrepXferParamsValid(const nixl_xfer_op_t &operation,
+                          const nixl_meta_dlist_t &local,
+                          const nixl_meta_dlist_t &remote) {
+    bool is_valid = true;
+    if (local.getType() != DRAM_SEG) {
+        std::cerr << "Error: local memory type must be DRAM_SEG\n";
+        is_valid = false;
+    }
+
+    // TODO: Consult with team if to change these to else ifs (right now it tells all the errors)
+    if (remote.getType() != FILE_SEG) {
+        std::cerr << "Error: remote memory type must be FILE_SEG\n";
+        is_valid = false;
+    }
+
+    if (local.descCount() != remote.descCount()) {
+        std::cerr <<"Error in count of local and remote descriptors\n";
+        is_valid = false;
+    }
+
+    if (local.getType() != DRAM_SEG) {
+        std::cerr << "Error: local memory type must be DRAM_SEG\n";
+        is_valid = false;
+    }
+
+    if (remote.getType() != FILE_SEG) {
+        std::cerr << "Error: remote memory type must be FILE_SEG\n";
+        is_valid = false;
+    }
+
+    return is_valid;
+}
+
+nixl_status_t nixlPosixEngine::prepXfer(const nixl_xfer_op_t &operation,
+                                        const nixl_meta_dlist_t &local,
+                                        const nixl_meta_dlist_t &remote,
+                                        const std::string &remote_agent __attribute__((unused)),
+                                        nixlBackendReqH* &handle,
+                                        const nixl_opt_b_args_t* opt_args) {
+    if (!isPrepXferParamsValid(operation, local, remote))
+        return NIXL_ERR_INVALID_PARAM;
+
+    try {
+        auto posix_handle = std::make_unique<nixlPosixBackendReqH>(operation, local, remote, opt_args);
+
+        nixl_status_t status = posix_handle->prepXfer();
+        if (status != NIXL_SUCCESS)
+            return status;
+
+        handle = posix_handle.release();
+        return NIXL_SUCCESS;
+    } catch (const nixlPosixBackendReqH::OperationError& e) {
+        std::cerr << "Operation error: " << e.what() << std::endl;
+        return NIXL_ERR_INVALID_PARAM;
+    } catch (const nixlPosixBackendReqH::UringInitError& e) {
+        std::cerr << "Uring initialization error: " << e.what() << std::endl;
+        return NIXL_ERR_BACKEND;
+    } catch (const std::exception& e) {
+        std::cerr << "Unexpected error: " << e.what() << std::endl;
+        return NIXL_ERR_BACKEND;
+    }
+}
+
+nixl_status_t nixlPosixEngine::postXfer(const nixl_xfer_op_t &operation,
+                                       const nixl_meta_dlist_t &local,
+                                       const nixl_meta_dlist_t &remote,
+                                       const std::string &remote_agent __attribute__((unused)),
+                                       nixlBackendReqH* &handle,
+                                       const nixl_opt_b_args_t* opt_args) {
+    nixl_status_t status = NIXL_SUCCESS;
+
+    nixlPosixBackendReqH  *posix_handle = (nixlPosixBackendReqH *)handle;
+    if (!posix_handle->isPrepped()) {
+        status = this->prepXfer(operation, local, remote, remote_agent, handle, opt_args);
+        if (status != NIXL_SUCCESS)
+            return status;
+    }
+
+    status = posix_handle->postXfer();
+    if (status != NIXL_SUCCESS && status != NIXL_IN_PROG)
+        std::cerr << "Error in submitting io_uring\n";
+
+    return status;
+}
+
+nixl_status_t nixlPosixEngine::checkXfer(nixlBackendReqH* handle) {
+    nixlPosixBackendReqH *posix_handle = (nixlPosixBackendReqH *)handle;
+    return posix_handle->checkXfer();
+}
+
+nixl_status_t nixlPosixEngine::releaseReqH(nixlBackendReqH* handle) {
+    nixlPosixBackendReqH *posix_handle = (nixlPosixBackendReqH *)handle;
+    delete posix_handle;
+    return NIXL_SUCCESS;
+}