feat(posix): Asynchronous Open in POSIX-uring

src/plugins/posix/README.md

@@ -26,6 +26,22 @@ Optionally POSIX plugin can also use liburing.
 `"<modes>:<path>"` string in `metaInfo` (path-mode, backend owns the
 open/close); see [`src/utils/file/README.md`](../../utils/file/README.md#path-mode-file-registration).
 
+## io_uring path opens
+
+When `use_uring=true`, path-mode files are opened asynchronously by default.
+File registration can therefore return before the open completes and will not
+reliably report open errors. Such errors are reported when a transfer uses the
+file.
+
+Set `uring_open_synchronous=true` when file registration must wait for the open
+and report any open error directly. This option defaults to `false` and is
+ignored unless io_uring is selected.
+
+```cpp
+params["use_uring"] = "true";
+params["uring_open_synchronous"] = "true";
+```
+
 ## Dependencies
 To enable Linux AIO support, you need to install the libaio package:
 

src/plugins/posix/io_queue.cpp

@@ -19,17 +19,25 @@
 #include "common/nixl_log.h"
 #include <absl/strings/str_format.h>
 
+#include <limits>
+
 #ifdef HAVE_POSIXAIO
 std::unique_ptr<nixlPosixIOQueue>
-nixlPosixIOQueueAIOCreate(uint32_t ios_pool_size, uint32_t kernel_queue_size);
+nixlPosixIOQueueAIOCreate(uint32_t ios_pool_size,
+                          uint32_t kernel_queue_size,
+                          bool open_synchronous);
 #endif
 #ifdef HAVE_LIBURING
 std::unique_ptr<nixlPosixIOQueue>
-nixlPosixIOQueueUringCreate(uint32_t ios_pool_size, uint32_t kernel_queue_size);
+nixlPosixIOQueueUringCreate(uint32_t ios_pool_size,
+                            uint32_t kernel_queue_size,
+                            bool open_synchronous);
 #endif
 #ifdef HAVE_LINUXAIO
 std::unique_ptr<nixlPosixIOQueue>
-nixlPosixIOQueueLinuxAIOCreate(uint32_t ios_pool_size, uint32_t kernel_queue_size);
+nixlPosixIOQueueLinuxAIOCreate(uint32_t ios_pool_size,
+                               uint32_t kernel_queue_size,
+                               bool open_synchronous);
 #endif
 
 static const struct {
@@ -47,10 +55,57 @@ static const struct {
 #endif
 };
 
+nixl_status_t
+nixlPosixIOQueue::registerFile(uint64_t dev_id, const std::string &meta_info) {
+    const bool path_mode = nixl::parsePathMeta(meta_info).has_value();
+    if (!path_mode && dev_id > static_cast<uint64_t>(std::numeric_limits<int>::max())) {
+        return NIXL_ERR_INVALID_PARAM;
+    }
+    auto file = files_.find(dev_id);
+    if (file != files_.end()) {
+        if (path_mode || file->second.pathMode) {
+            return NIXL_ERR_INVALID_PARAM;
+        }
+        file->second.registrations++;
+        return NIXL_SUCCESS;
+    }
+
+    files_.try_emplace(dev_id, dev_id, meta_info, path_mode);
+    return NIXL_SUCCESS;
+}
+
+nixl_status_t
+nixlPosixIOQueue::deregisterFile(uint64_t dev_id) {
+    auto file = files_.find(dev_id);
+    if (file == files_.end()) {
+        return NIXL_SUCCESS;
+    }
+    if (--file->second.registrations == 0) {
+        files_.erase(file);
+    }
+    return NIXL_SUCCESS;
+}
+
+nixl_status_t
+nixlPosixIOQueue::enqueue(uint64_t dev_id,
+                          void *buf,
+                          size_t len,
+                          off_t offset,
+                          bool read,
+                          nixlPosixIOQueueDoneCb clb,
+                          void *ctx) {
+    auto file = files_.find(dev_id);
+    if (file == files_.end()) {
+        return NIXL_ERR_INVALID_PARAM;
+    }
+    return enqueueFd(file->second.fileFd.fd(), buf, len, offset, read, std::move(clb), ctx);
+}
+
 std::unique_ptr<nixlPosixIOQueue>
 nixlPosixIOQueue::instantiate(std::string_view io_queue_type,
                               uint32_t ios_pool_size,
-                              uint32_t kernel_queue_size) {
+                              uint32_t kernel_queue_size,
+                              bool open_synchronous) {
     for (const auto &factory : factories) {
         if (io_queue_type == factory.name) {
             if (ios_pool_size == 0) {
@@ -61,7 +116,7 @@ nixlPosixIOQueue::instantiate(std::string_view io_queue_type,
                 kernel_queue_size = DEF_KERNEL_QUEUE_SIZE;
                 NIXL_INFO << "Using default kernel queue size: " << kernel_queue_size;
             }
-            return factory.createFn(ios_pool_size, kernel_queue_size);
+            return factory.createFn(ios_pool_size, kernel_queue_size, open_synchronous);
         }
     }
     return nullptr;

src/plugins/posix/io_queue.h

@@ -19,11 +19,15 @@
 #define POSIX_IO_QUEUE_H
 
 #include <stdint.h>
+#include <sys/types.h>
 #include <list>
 #include <memory>
+#include <string>
+#include <unordered_map>
 #include <vector>
 #include <functional>
 #include "backend_aux.h"
+#include "file/file_path_mode.h"
 
 using nixlPosixIOQueueDoneCb = std::function<void(void *ctx, uint32_t data_size, int error)>;
 using nixlPosixIOQueueCancelDoneCb = std::function<void(void *ctx)>;
@@ -32,22 +36,40 @@ class nixlPosixIOQueue {
 public:
     using nixlPosixIOQueueCreateFn =
         std::function<std::unique_ptr<nixlPosixIOQueue>(uint32_t ios_pool_size,
-                                                        uint32_t kernel_queue_size)>;
+                                                        uint32_t kernel_queue_size,
+                                                        bool open_synchronous)>;
 
     nixlPosixIOQueue(uint32_t ios_pool_size, uint32_t kernel_queue_size)
         : ios_pool_size_(normalizedIOSPoolSize(ios_pool_size)),
           kernel_queue_size_(normalizedKernelQueueSize(kernel_queue_size)) {}
 
     virtual ~nixlPosixIOQueue() {}
 
+    /**
+     * @brief Register a file identifier for subsequent I/O.
+     * @param dev_id Identifier used by enqueue().
+     * @param meta_info File registration metadata.
+     * @return NIXL_SUCCESS on success, or an error status otherwise.
+     */
     virtual nixl_status_t
-    enqueue(int fd,
+    registerFile(uint64_t dev_id, const std::string &meta_info);
+
+    /**
+     * @brief Deregister a file identifier.
+     * @param dev_id Identifier previously passed to registerFile().
+     * @return NIXL_SUCCESS on success, or an error status otherwise.
+     */
+    virtual nixl_status_t
+    deregisterFile(uint64_t dev_id);
+
+    virtual nixl_status_t
+    enqueue(uint64_t dev_id,
             void *buf,
             size_t len,
             off_t offset,
             bool read,
             nixlPosixIOQueueDoneCb clb,
-            void *ctx) = 0;
+            void *ctx);
     virtual nixl_status_t
     post(void) = 0;
     virtual nixl_status_t
@@ -61,7 +83,10 @@ class nixlPosixIOQueue {
     }
 
     static std::unique_ptr<nixlPosixIOQueue>
-    instantiate(std::string_view io_queue_type, uint32_t ios_pool_size, uint32_t kernel_queue_size);
+    instantiate(std::string_view io_queue_type,
+                uint32_t ios_pool_size,
+                uint32_t kernel_queue_size,
+                bool open_synchronous = false);
     static std::string_view
     getDefaultIoQueueType(void);
 
@@ -73,6 +98,25 @@ class nixlPosixIOQueue {
     static constexpr uint32_t DEF_KERNEL_QUEUE_SIZE = 256;
 
 protected:
+    struct registeredFile {
+        registeredFile(uint64_t dev_id, const std::string &meta_info, bool is_path_mode)
+            : fileFd(is_path_mode ? -1 : static_cast<int>(dev_id), meta_info),
+              pathMode(is_path_mode) {}
+
+        nixl::FileFd fileFd;
+        size_t registrations = 1;
+        bool pathMode;
+    };
+
+    virtual nixl_status_t
+    enqueueFd(int fd,
+              void *buf,
+              size_t len,
+              off_t offset,
+              bool read,
+              nixlPosixIOQueueDoneCb clb,
+              void *ctx) = 0;
+
     static uint32_t
     normalizedIOSPoolSize(uint32_t ios_pool_size) {
         return std::clamp(ios_pool_size, MIN_IOS_POOL_SIZE, MAX_IOS_POOL_SIZE);
@@ -85,6 +129,7 @@ class nixlPosixIOQueue {
 
     uint32_t ios_pool_size_;
     uint32_t kernel_queue_size_;
+    std::unordered_map<uint64_t, registeredFile> files_;
 };
 
 template<typename Entry> class nixlPosixIOQueueImpl : public nixlPosixIOQueue {