diff --git a/src/plugins/posix/README.md b/src/plugins/posix/README.md index 9720ad4765..6140c6bf0c 100644 --- a/src/plugins/posix/README.md +++ b/src/plugins/posix/README.md @@ -26,6 +26,22 @@ Optionally POSIX plugin can also use liburing. `":"` 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: diff --git a/src/plugins/posix/io_queue.cpp b/src/plugins/posix/io_queue.cpp index 6b2d0931aa..e4c13f04b9 100644 --- a/src/plugins/posix/io_queue.cpp +++ b/src/plugins/posix/io_queue.cpp @@ -19,17 +19,25 @@ #include "common/nixl_log.h" #include +#include + #ifdef HAVE_POSIXAIO std::unique_ptr -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 -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 -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(std::numeric_limits::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::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; diff --git a/src/plugins/posix/io_queue.h b/src/plugins/posix/io_queue.h index f46dd9f7ee..deab170541 100644 --- a/src/plugins/posix/io_queue.h +++ b/src/plugins/posix/io_queue.h @@ -19,11 +19,15 @@ #define POSIX_IO_QUEUE_H #include +#include #include #include +#include +#include #include #include #include "backend_aux.h" +#include "file/file_path_mode.h" using nixlPosixIOQueueDoneCb = std::function; using nixlPosixIOQueueCancelDoneCb = std::function; @@ -32,7 +36,8 @@ class nixlPosixIOQueue { public: using nixlPosixIOQueueCreateFn = std::function(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)), @@ -40,14 +45,31 @@ class nixlPosixIOQueue { 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 - 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(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 files_; }; template class nixlPosixIOQueueImpl : public nixlPosixIOQueue { diff --git a/src/plugins/posix/io_uring_io_queue.cpp b/src/plugins/posix/io_uring_io_queue.cpp index b1e5770050..3f3a1275b5 100644 --- a/src/plugins/posix/io_uring_io_queue.cpp +++ b/src/plugins/posix/io_uring_io_queue.cpp @@ -17,273 +17,552 @@ #include "io_queue.h" #include "common/nixl_log.h" + #include #include -#include -#define MAX_IO_SUBMIT_BATCH_SIZE 64 -#define MAX_IO_CHECK_COMPLETED_BATCH_SIZE 64 - -enum class nixlPosixIoUringCQEKind { - IO, - CANCEL, +#include +#include +#include +#include +#include +#include +#include + +namespace { + +struct fileState; +struct ioSlot; + +ioSlot * +completeOpen(void *owner, int result); +ioSlot * +completeData(void *owner, int result); +ioSlot * +completeCancel(void *owner, int result); +ioSlot * +completeClose(void *owner, int result); + +struct completion { + using handler_t = ioSlot *(*)(void *, int); + + handler_t handler; + void *owner; }; -struct nixlPosixIoUringCQEData { - explicit nixlPosixIoUringCQEData(nixlPosixIoUringCQEKind kind) : kind_(kind) {} - - nixlPosixIoUringCQEKind kind_; - void *ctx_ = nullptr; +struct fileState { + enum class status_t { PENDING_OPEN, OPENING, OPEN, FAILED, PENDING_CLOSE, CLOSING, CLOSED }; + + explicit fileState(nixl::PathSpec path_spec) + : path(std::move(path_spec.path)), + flags(path_spec.flags), + mode(path_spec.mode), + completionData{completeOpen, this} {} + + std::string path; + int flags; + mode_t mode; + status_t status = status_t::PENDING_OPEN; + int fd = -1; + int openError = 0; + size_t activeIos = 0; + bool deregistered = false; + completion completionData; }; -struct nixlPosixIoUringIO : public nixlPosixIoUringCQEData { - nixlPosixIoUringIO() : nixlPosixIoUringCQEData(nixlPosixIoUringCQEKind::IO) {} +struct ioSlot { + enum class state_t { FREE, QUEUED, IN_FLIGHT }; - int fd; - void *buf_; - size_t len_; - off_t offset_; - bool read_; - nixlPosixIOQueueDoneCb clb_; - bool in_flight_ = false; // owned by the ring, not yet reaped - bool cancel_pending_ = false; // cancellation is queued or its CQE is pending -}; - -struct nixlPosixIoUringCancel : public nixlPosixIoUringCQEData { - nixlPosixIoUringCancel() : nixlPosixIoUringCQEData(nixlPosixIoUringCQEKind::CANCEL) {} + ioSlot() : data_completion_{completeData, this}, cancel_completion_{completeCancel, this} {} - nixlPosixIoUringIO *io_ = nullptr; - nixlPosixIOQueueCancelDoneCb clb_; + int fd = -1; + void *buf_ = nullptr; + size_t len_ = 0; + off_t offset_ = 0; + bool read_ = false; + nixlPosixIOQueueDoneCb clb_; + void *ctx_ = nullptr; + std::shared_ptr file_; + state_t state_ = state_t::FREE; + bool cancel_pending_ = false; + bool cancel_submitted_ = false; + nixlPosixIOQueueCancelDoneCb cancel_clb_; + completion data_completion_; + completion cancel_completion_; }; -class nixlPosixIOQueueUring : public nixlPosixIOQueueImpl { +class nixlPosixIOQueueUring : public nixlPosixIOQueueImpl { public: - nixlPosixIOQueueUring(uint32_t ios_pool_size, uint32_t kernel_queue_size); + nixlPosixIOQueueUring(uint32_t ios_pool_size, + uint32_t kernel_queue_size, + bool open_synchronous); - virtual nixl_status_t + nixl_status_t post(void) override; - virtual nixl_status_t - enqueue(int fd, + nixl_status_t + enqueue(uint64_t dev_id, void *buf, size_t len, off_t offset, bool read, nixlPosixIOQueueDoneCb clb, void *ctx) override; - virtual nixl_status_t - poll(void) override; - virtual unsigned - cancel(void *ctx, nixlPosixIOQueueCancelDoneCb clb) override; - virtual ~nixlPosixIOQueueUring() override; + nixl_status_t + registerFile(uint64_t dev_id, const std::string &meta_info) override; + nixl_status_t + deregisterFile(uint64_t dev_id) override; -protected: nixl_status_t - doCheckCompleted(void); + poll(void) override; + unsigned + cancel(void *ctx, nixlPosixIOQueueCancelDoneCb clb) override; + ~nixlPosixIOQueueUring() override; private: nixl_status_t - driveSubmissions(void); + enqueueFd(int fd, + void *buf, + size_t len, + off_t offset, + bool read, + nixlPosixIOQueueDoneCb clb, + void *ctx) override; + nixl_status_t + enqueueIO(int fd, + std::shared_ptr file, + void *buf, + size_t len, + off_t offset, + bool read, + nixlPosixIOQueueDoneCb clb, + void *ctx); void - failQueuedIOs(void *ctx); + doCheckCompleted(void); + nixl_status_t + submitPrepared(unsigned prepared); void - prepareSQEs(void); + retireFile(uint64_t dev_id); void - releaseIOIfIdle(nixlPosixIoUringIO *io); + completeQueuedIO(ioSlot *io, int error); - struct io_uring uring; // The io_uring instance for async I/O operations + struct io_uring uring_{}; + + uint32_t cq_capacity_ = 0; + size_t in_flight_cqes_ = 0; + unsigned pending_sqes_ = 0; bool terminal_error_ = false; - std::list cancels_to_submit_; - std::vector cancels_; + bool open_supported_ = false; + bool open_synchronous_; + std::unordered_map> path_files_; }; -nixlPosixIOQueueUring::nixlPosixIOQueueUring(uint32_t ios_pool_size, uint32_t kernel_queue_size) - : nixlPosixIOQueueImpl(ios_pool_size, kernel_queue_size), - cancels_(ios_.size()) { - for (size_t i = 0; i < ios_.size(); i++) { - cancels_[i].io_ = &ios_[i]; - } - +nixlPosixIOQueueUring::nixlPosixIOQueueUring(uint32_t ios_pool_size, + uint32_t kernel_queue_size, + bool open_synchronous) + : nixlPosixIOQueueImpl(ios_pool_size, kernel_queue_size), + open_synchronous_(open_synchronous) { io_uring_params params = {}; - int ret = io_uring_queue_init_params(kernel_queue_size_, &uring, ¶ms); + int ret = io_uring_queue_init_params(kernel_queue_size_, &uring_, ¶ms); if (ret < 0) { throw std::runtime_error( absl::StrFormat("Failed to initialize io_uring instance: %s", nixl_strerror(-ret))); } + cq_capacity_ = params.cq_entries; + + // Probe the running kernel because vendors routinely backport io_uring features. + io_uring_probe *probe = io_uring_get_probe_ring(&uring_); + open_supported_ = probe && io_uring_opcode_supported(probe, IORING_OP_OPENAT); + if (probe) { + io_uring_free_probe(probe); + } +} + +nixl_status_t +nixlPosixIOQueueUring::enqueueFd(int fd, + void *buf, + size_t len, + off_t offset, + bool read, + nixlPosixIOQueueDoneCb clb, + void *ctx) { + return enqueueIO(fd, {}, buf, len, offset, read, std::move(clb), ctx); +} + +nixl_status_t +nixlPosixIOQueueUring::enqueue(uint64_t dev_id, + void *buf, + size_t len, + off_t offset, + bool read, + nixlPosixIOQueueDoneCb clb, + void *ctx) { + auto file = path_files_.find(dev_id); + if (file == path_files_.end()) { + return nixlPosixIOQueue::enqueue(dev_id, buf, len, offset, read, std::move(clb), ctx); + } + return enqueueIO(-1, file->second, buf, len, offset, read, std::move(clb), ctx); +} + +nixl_status_t +nixlPosixIOQueueUring::enqueueIO(int fd, + std::shared_ptr file, + void *buf, + size_t len, + off_t offset, + bool read, + nixlPosixIOQueueDoneCb clb, + void *ctx) { + if (file && + (file->status == fileState::status_t::PENDING_CLOSE || + file->status == fileState::status_t::CLOSING || + file->status == fileState::status_t::CLOSED)) { + return NIXL_ERR_NOT_ALLOWED; + } + if (free_ios_.empty()) { + NIXL_ERROR << "No more free blocks available"; + return NIXL_ERR_NOT_ALLOWED; + } + ioSlot *io = free_ios_.front(); + free_ios_.pop_front(); + io->fd = fd; + io->buf_ = buf; + io->len_ = len; + io->offset_ = offset; + io->read_ = read; + io->clb_ = std::move(clb); + io->ctx_ = ctx; + io->file_ = std::move(file); + io->state_ = ioSlot::state_t::QUEUED; + io->cancel_pending_ = false; + io->cancel_submitted_ = false; + io->cancel_clb_ = {}; + if (io->file_) { + ++io->file_->activeIos; + } + return NIXL_SUCCESS; } -// Prepare pending cancellation SQEs before normal I/O SQEs, without submitting them. void -nixlPosixIOQueueUring::prepareSQEs(void) { - int num_sqes = 0; - while (num_sqes < MAX_IO_SUBMIT_BATCH_SIZE) { - if (cancels_to_submit_.empty() && ios_to_submit_.empty()) { - break; - } +queueClose(fileState *file) { + if (file->fd < 0 || file->status == fileState::status_t::PENDING_CLOSE || + file->status == fileState::status_t::CLOSING || + file->status == fileState::status_t::CLOSED) { + return; + } + file->status = fileState::status_t::PENDING_CLOSE; +} - struct io_uring_sqe *sqe = io_uring_get_sqe(&uring); - if (!sqe) { - break; - } +void +nixlPosixIOQueueUring::retireFile(uint64_t dev_id) { + auto file_it = path_files_.find(dev_id); + NIXL_ASSERT(file_it != path_files_.end()); + const auto &file = file_it->second; + file->deregistered = true; + if (file->status == fileState::status_t::PENDING_OPEN) { + file->status = fileState::status_t::CLOSED; + } else if (file->status == fileState::status_t::OPEN) { + queueClose(file.get()); + } else if (file->status == fileState::status_t::FAILED) { + file->status = fileState::status_t::CLOSED; + } + if (file->status == fileState::status_t::CLOSED) { + path_files_.erase(file_it); + } +} - if (!cancels_to_submit_.empty()) { - nixlPosixIoUringIO *io = cancels_to_submit_.front(); - cancels_to_submit_.pop_front(); - size_t index = static_cast(io - ios_.data()); - auto *io_data = static_cast(io); - io_uring_prep_cancel(sqe, io_data, 0); - io_uring_sqe_set_data(sqe, static_cast(&cancels_[index])); - NIXL_ASSERT(io->cancel_pending_); - } else { - nixlPosixIoUringIO *io = ios_to_submit_.front(); - ios_to_submit_.pop_front(); - if (io->read_) { - io_uring_prep_read(sqe, io->fd, io->buf_, io->len_, io->offset_); - } else { - io_uring_prep_write(sqe, io->fd, io->buf_, io->len_, io->offset_); - } - io_uring_sqe_set_data(sqe, static_cast(io)); - io->in_flight_ = true; +nixl_status_t +nixlPosixIOQueueUring::deregisterFile(uint64_t dev_id) { + doCheckCompleted(); + auto file_it = path_files_.find(dev_id); + if (file_it == path_files_.end()) { + return nixlPosixIOQueue::deregisterFile(dev_id); + } + const auto file = file_it->second; + if (file->activeIos != 0) { + return NIXL_ERR_NOT_ALLOWED; + } + retireFile(dev_id); + while (file->status != fileState::status_t::CLOSED) { + nixl_status_t status = poll(); + if (status < 0) { + return status; } - num_sqes++; } + return NIXL_SUCCESS; } -void -nixlPosixIOQueueUring::failQueuedIOs(void *ctx) { - for (auto it = ios_to_submit_.begin(); it != ios_to_submit_.end();) { - nixlPosixIoUringIO *io = *it; - if (io->ctx_ != ctx) { - ++it; - continue; +nixl_status_t +nixlPosixIOQueueUring::registerFile(uint64_t dev_id, const std::string &meta_info) { + auto path_spec = nixl::parsePathMeta(meta_info); + if (!path_spec) { + if (path_files_.contains(dev_id)) { + return NIXL_ERR_INVALID_PARAM; } - if (io->clb_) { - io->clb_(io->ctx_, 0, 1); + return nixlPosixIOQueue::registerFile(dev_id, meta_info); + } + + if (!open_supported_) { + return NIXL_ERR_NOT_SUPPORTED; + } + + if (path_files_.contains(dev_id) || files_.contains(dev_id)) { + return NIXL_ERR_NOT_ALLOWED; + } + + auto file = std::make_shared(std::move(*path_spec)); + path_files_.emplace(dev_id, file); + + nixl_status_t status = post(); + if (status < 0) { + retireFile(dev_id); + return status; + } + if (!open_synchronous_) { + return NIXL_SUCCESS; + } + + while (file->status == fileState::status_t::PENDING_OPEN || + file->status == fileState::status_t::OPENING) { + status = poll(); + if (status < 0) { + retireFile(dev_id); + return status; } - it = ios_to_submit_.erase(it); - free_ios_.push_back(io); } -} -void -nixlPosixIOQueueUring::releaseIOIfIdle(nixlPosixIoUringIO *io) { - if (!io->in_flight_ && !io->cancel_pending_) { - free_ios_.push_back(io); + if (file->status == fileState::status_t::FAILED) { + retireFile(dev_id); + return NIXL_ERR_BACKEND; } + return NIXL_SUCCESS; } nixl_status_t -nixlPosixIOQueueUring::post(void) { - return driveSubmissions(); +nixlPosixIOQueueUring::submitPrepared(unsigned prepared) { + pending_sqes_ += prepared; + while (pending_sqes_ > 0) { + int ret = io_uring_submit(&uring_); + if (ret == -EAGAIN || ret == -EBUSY || ret == -EINTR) { + return NIXL_IN_PROG; + } + if (ret < 0) { + NIXL_ERROR << "io_uring_submit failed: " << nixl_strerror(-ret); + terminal_error_ = true; + return NIXL_ERR_BACKEND; + } + if (ret == 0) { + return NIXL_IN_PROG; + } + const unsigned submitted = std::min(pending_sqes_, static_cast(ret)); + pending_sqes_ -= submitted; + in_flight_cqes_ += submitted; + } + return NIXL_IN_PROG; } -// Prepare I/O SQEs and submit every ring-ready SQE. nixl_status_t -nixlPosixIOQueueUring::driveSubmissions(void) { +nixlPosixIOQueueUring::post(void) { if (terminal_error_) { + return NIXL_ERR_BACKEND; + } + if (pending_sqes_ > 0) { + return submitPrepared(0); + } + std::erase_if(path_files_, [](const auto &entry) { + return entry.second->status == fileState::status_t::CLOSED; + }); + bool failed_io = false; + for (auto &io : ios_) { + if (io.state_ == ioSlot::state_t::QUEUED && io.file_ && + io.file_->status == fileState::status_t::FAILED) { + completeQueuedIO(&io, io.file_->openError); + failed_io = true; + } + } + if (failed_io) { return NIXL_IN_PROG; } - prepareSQEs(); - - int ret = io_uring_submit(&uring); - if (ret >= 0 || ret == -EAGAIN || ret == -EBUSY || ret == -EINTR) { + const size_t occupied_cqes = in_flight_cqes_ + pending_sqes_; + const size_t available_cqes = cq_capacity_ > occupied_cqes ? cq_capacity_ - occupied_cqes : 0; + if (available_cqes == 0) { return NIXL_IN_PROG; } - NIXL_ERROR << "io_uring_submit failed: " << nixl_strerror(-ret); - terminal_error_ = true; - return NIXL_IN_PROG; -} + unsigned prepared = 0; + for (auto &io : ios_) { + if (prepared >= available_cqes || io_uring_sq_space_left(&uring_) < 1) { + break; + } + if (!io.cancel_pending_ || io.cancel_submitted_) { + continue; + } + io_uring_sqe *sqe = io_uring_get_sqe(&uring_); + io_uring_prep_cancel(sqe, &io.data_completion_, 0); + io_uring_sqe_set_data(sqe, &io.cancel_completion_); + io.cancel_submitted_ = true; + ++prepared; + } -inline nixl_status_t -nixlPosixIOQueueUring::doCheckCompleted(void) { - struct io_uring_cqe *cqe; - unsigned head; - int count = 0; - io_uring_for_each_cqe(&uring, head, cqe) { - int res = cqe->res; - auto *data = static_cast(io_uring_cqe_get_data(cqe)); - NIXL_ASSERT(data); - nixlPosixIoUringIO *io; - if (data->kind_ == nixlPosixIoUringCQEKind::CANCEL) { - auto *cancel = static_cast(data); - io = cancel->io_; - NIXL_ASSERT(io && io->cancel_pending_); - io->cancel_pending_ = false; - if (cancel->clb_) { - cancel->clb_(cancel->ctx_); - } - cancel->clb_ = nullptr; - cancel->ctx_ = nullptr; + for (const auto &[dev_id, file] : path_files_) { + if (prepared >= available_cqes || io_uring_sq_space_left(&uring_) < 1) { + break; + } + if (file->status != fileState::status_t::PENDING_CLOSE) { + continue; + } + io_uring_sqe *sqe = io_uring_get_sqe(&uring_); + io_uring_prep_close(sqe, file->fd); + file->completionData.handler = completeClose; + io_uring_sqe_set_data(sqe, &file->completionData); + file->status = fileState::status_t::CLOSING; + ++prepared; + } + + // Prioritize I/O unblocked by completed opens over more registration opens. + for (auto &io : ios_) { + if (prepared >= available_cqes || io_uring_sq_space_left(&uring_) < 1) { + break; + } + if (io.state_ != ioSlot::state_t::QUEUED || + (io.file_ && io.file_->status != fileState::status_t::OPEN)) { + continue; + } + io_uring_sqe *data_sqe = io_uring_get_sqe(&uring_); + const int fd = io.file_ ? io.file_->fd : io.fd; + if (io.read_) { + io_uring_prep_read(data_sqe, fd, io.buf_, io.len_, io.offset_); } else { - io = static_cast(data); - int error = res < 0 || static_cast(res) != io->len_; - if (error) { - NIXL_DEBUG << absl::StrFormat( - "IO operation incomplete: result %d, expected %zu", res, io->len_); - } - if (io->clb_) { - io->clb_(io->ctx_, error ? 0 : static_cast(res), error); - } - io->in_flight_ = false; + io_uring_prep_write(data_sqe, fd, io.buf_, io.len_, io.offset_); } - releaseIOIfIdle(io); - if (++count == MAX_IO_CHECK_COMPLETED_BATCH_SIZE) { + io.state_ = ioSlot::state_t::IN_FLIGHT; + io_uring_sqe_set_data(data_sqe, &io.data_completion_); + ++prepared; + } + + for (const auto &[dev_id, file] : path_files_) { + if (prepared >= available_cqes || io_uring_sq_space_left(&uring_) < 1) { break; } + if (file->status != fileState::status_t::PENDING_OPEN) { + continue; + } + io_uring_sqe *open_sqe = io_uring_get_sqe(&uring_); + io_uring_prep_openat(open_sqe, AT_FDCWD, file->path.c_str(), file->flags, file->mode); + io_uring_sqe_set_data(open_sqe, &file->completionData); + file->status = fileState::status_t::OPENING; + ++prepared; } - // Mark all seen - io_uring_cq_advance(&uring, count); + return submitPrepared(prepared); +} - if (free_ios_.size() == ios_pool_size_) { - return NIXL_SUCCESS; // All ios and cancellation cleanup are done +void +nixlPosixIOQueueUring::completeQueuedIO(ioSlot *io, int error) { + if (io->clb_) { + io->clb_(io->ctx_, 0, error); } - - return NIXL_IN_PROG; // Some ios or cancellation SQEs still need to drain + if (io->file_ && io->file_->activeIos > 0) { + --io->file_->activeIos; + } + io->file_.reset(); + io->state_ = ioSlot::state_t::FREE; + free_ios_.push_back(io); } -nixl_status_t -nixlPosixIOQueueUring::enqueue(int fd, - void *buf, - size_t len, - off_t offset, - bool read, - nixlPosixIOQueueDoneCb clb, - void *ctx) { - if (free_ios_.empty()) { - NIXL_ERROR << "No more free blocks available"; - return NIXL_ERR_NOT_ALLOWED; +ioSlot * +completeOpen(void *owner, int result) { + auto *file = static_cast(owner); + if (result < 0) { + file->status = fileState::status_t::FAILED; + file->openError = -result; + if (file->deregistered) { + file->status = fileState::status_t::CLOSED; + } + NIXL_ERROR << absl::StrFormat( + "io_uring open failed for %s: %s", file->path, nixl_strerror(-result)); + return nullptr; } - nixlPosixIoUringIO *io = free_ios_.front(); - free_ios_.pop_front(); - io->fd = fd; - io->buf_ = buf; - io->len_ = len; - io->offset_ = offset; - io->read_ = read; - io->clb_ = clb; - io->ctx_ = ctx; - io->in_flight_ = false; - io->cancel_pending_ = false; + file->fd = result; + file->status = fileState::status_t::OPEN; + if (file->deregistered) { + queueClose(file); + } + return nullptr; +} - ios_to_submit_.push_back(io); +ioSlot * +completeData(void *owner, int result) { + auto *io = static_cast(owner); + const auto file = io->file_; + const int error = result < 0 ? -result : static_cast(result) != io->len_; + if (io->clb_) { + io->clb_(io->ctx_, error ? 0 : static_cast(result), error); + } + if (error) { + NIXL_DEBUG << absl::StrFormat( + "IO operation incomplete: result %d, expected %zu", result, io->len_); + } + if (file) { + --file->activeIos; + } + io->file_.reset(); + io->state_ = ioSlot::state_t::FREE; + if (io->cancel_pending_ && !io->cancel_submitted_) { + io->cancel_pending_ = false; + if (io->cancel_clb_) { + io->cancel_clb_(io->ctx_); + } + io->cancel_clb_ = {}; + } + return io->cancel_pending_ ? nullptr : io; +} - return NIXL_SUCCESS; +ioSlot * +completeCancel(void *owner, int) { + auto *io = static_cast(owner); + io->cancel_pending_ = false; + io->cancel_submitted_ = false; + if (io->cancel_clb_) { + io->cancel_clb_(io->ctx_); + } + io->cancel_clb_ = {}; + return io->state_ == ioSlot::state_t::FREE ? io : nullptr; } -nixl_status_t -nixlPosixIOQueueUring::poll(void) { - nixl_status_t completion_status = doCheckCompleted(); - if (completion_status == NIXL_SUCCESS) { - return NIXL_SUCCESS; +ioSlot * +completeClose(void *owner, int result) { + auto *file = static_cast(owner); + if (result < 0) { + NIXL_ERROR << absl::StrFormat( + "io_uring close failed for %s: %s", file->path, nixl_strerror(-result)); + ::close(file->fd); } - if (terminal_error_) { - return NIXL_ERR_BACKEND; + file->fd = -1; + file->status = fileState::status_t::CLOSED; + return nullptr; +} + +void +nixlPosixIOQueueUring::doCheckCompleted(void) { + io_uring_cqe *cqe; + unsigned head; + unsigned count = 0; + io_uring_for_each_cqe(&uring_, head, cqe) { + auto *completion_data = reinterpret_cast(io_uring_cqe_get_data(cqe)); + if (completion_data) { + if (auto *io = completion_data->handler(completion_data->owner, cqe->res)) { + free_ios_.push_back(io); + } + } + ++count; } - return driveSubmissions(); + if (count > 0) { + io_uring_cq_advance(&uring_, count); + in_flight_cqes_ -= std::min(in_flight_cqes_, static_cast(count)); + } } unsigned @@ -292,32 +571,74 @@ nixlPosixIOQueueUring::cancel(void *ctx, nixlPosixIOQueueCancelDoneCb clb) { return 0; } - failQueuedIOs(ctx); + for (auto &io : ios_) { + if (io.state_ == ioSlot::state_t::QUEUED && io.ctx_ == ctx) { + completeQueuedIO(&io, 1); + } + } - unsigned cancels_requested = 0; + unsigned requested = 0; for (auto &io : ios_) { - if (io.in_flight_ && io.ctx_ == ctx && !io.cancel_pending_) { - size_t index = static_cast(&io - ios_.data()); - io.cancel_pending_ = true; - cancels_[index].clb_ = clb; - cancels_[index].ctx_ = ctx; - cancels_to_submit_.push_back(&io); - cancels_requested++; + if (io.state_ != ioSlot::state_t::IN_FLIGHT || io.ctx_ != ctx || io.cancel_pending_) { + continue; } + io.cancel_pending_ = true; + io.cancel_clb_ = clb; + ++requested; } - if (cancels_requested != 0) { - // Best-effort cancellation blocks only its owning request until callbacks are invoked. - driveSubmissions(); + if (requested > 0) { + post(); } - return cancels_requested; + return requested; +} + +nixl_status_t +nixlPosixIOQueueUring::poll(void) { + doCheckCompleted(); + nixl_status_t post_status = post(); + if (post_status < 0) { + return post_status; + } + return free_ios_.size() == ios_pool_size_ ? NIXL_SUCCESS : NIXL_IN_PROG; } nixlPosixIOQueueUring::~nixlPosixIOQueueUring() { - io_uring_queue_exit(&uring); + // Drain outstanding operations before ring teardown. + while (!terminal_error_) { + doCheckCompleted(); + const nixl_status_t status = post(); + if (status < 0) { + break; + } + if (pending_sqes_ == 0 && in_flight_cqes_ == 0) { + break; + } + if (in_flight_cqes_ > 0) { + io_uring_cqe *cqe = nullptr; + const int ret = io_uring_wait_cqe(&uring_, &cqe); + if (ret < 0) { + NIXL_ERROR << "io_uring wait during shutdown failed: " << nixl_strerror(-ret); + break; + } + } + } + doCheckCompleted(); + + io_uring_queue_exit(&uring_); + for (const auto &[dev_id, file] : path_files_) { + if (file->fd >= 0) { + ::close(file->fd); + } + } } +} // namespace + std::unique_ptr -nixlPosixIOQueueUringCreate(uint32_t ios_pool_size, uint32_t kernel_queue_size) { - return std::make_unique(ios_pool_size, kernel_queue_size); +nixlPosixIOQueueUringCreate(uint32_t ios_pool_size, + uint32_t kernel_queue_size, + bool open_synchronous) { + return std::make_unique( + ios_pool_size, kernel_queue_size, open_synchronous); } diff --git a/src/plugins/posix/linux_aio_io_queue.cpp b/src/plugins/posix/linux_aio_io_queue.cpp index 35ca0a20d2..42d5d049e3 100644 --- a/src/plugins/posix/linux_aio_io_queue.cpp +++ b/src/plugins/posix/linux_aio_io_queue.cpp @@ -41,13 +41,13 @@ class nixlPosixIOQueueLinuxAIO : public nixlPosixIOQueueImpl -nixlPosixIOQueueLinuxAIOCreate(uint32_t ios_pool_size, uint32_t kernel_queue_size) { +nixlPosixIOQueueLinuxAIOCreate(uint32_t ios_pool_size, uint32_t kernel_queue_size, bool) { return std::make_unique(ios_pool_size, kernel_queue_size); } diff --git a/src/plugins/posix/posix_aio_io_queue.cpp b/src/plugins/posix/posix_aio_io_queue.cpp index ea551c8350..cd99c42f41 100644 --- a/src/plugins/posix/posix_aio_io_queue.cpp +++ b/src/plugins/posix/posix_aio_io_queue.cpp @@ -38,13 +38,13 @@ class nixlPosixIOQueueAIO : public nixlPosixIOQueueImpl { virtual nixl_status_t post(void) override; virtual nixl_status_t - enqueue(int fd, - void *buf, - size_t len, - off_t offset, - bool read, - nixlPosixIOQueueDoneCb clb, - void *ctx) override; + enqueueFd(int fd, + void *buf, + size_t len, + off_t offset, + bool read, + nixlPosixIOQueueDoneCb clb, + void *ctx) override; virtual nixl_status_t poll(void) override; virtual ~nixlPosixIOQueueAIO() override; @@ -65,13 +65,13 @@ nixlPosixIOQueueAIO::~nixlPosixIOQueueAIO() { } nixl_status_t -nixlPosixIOQueueAIO::enqueue(int fd, - void *buf, - size_t len, - off_t offset, - bool read, - nixlPosixIOQueueDoneCb clb, - void *ctx) { +nixlPosixIOQueueAIO::enqueueFd(int fd, + void *buf, + size_t len, + off_t offset, + bool read, + nixlPosixIOQueueDoneCb clb, + void *ctx) { if (free_ios_.empty()) { NIXL_ERROR << "No more free blocks available"; return NIXL_ERR_NOT_ALLOWED; @@ -176,6 +176,6 @@ nixlPosixIOQueueAIO::poll(void) { } std::unique_ptr -nixlPosixIOQueueAIOCreate(uint32_t ios_pool_size, uint32_t kernel_queue_size) { +nixlPosixIOQueueAIOCreate(uint32_t ios_pool_size, uint32_t kernel_queue_size, bool) { return std::make_unique(ios_pool_size, kernel_queue_size); } diff --git a/src/plugins/posix/posix_backend.cpp b/src/plugins/posix/posix_backend.cpp index 9b9b67abb1..aa698891b2 100644 --- a/src/plugins/posix/posix_backend.cpp +++ b/src/plugins/posix/posix_backend.cpp @@ -223,8 +223,8 @@ nixlPosixBackendReqH::postXfer() { for (auto [local_it, remote_it] = std::make_pair(local.begin(), remote.begin()); local_it != local.end() && remote_it != remote.end(); ++local_it, ++remote_it) { - int fd = static_cast(remote_it->metadataP)->file_fd.fd(); - nixl_status_t status = io_queue_->enqueue(fd, + uint64_t dev_id = static_cast(remote_it->metadataP)->devId; + nixl_status_t status = io_queue_->enqueue(dev_id, reinterpret_cast(local_it->addr), remote_it->len, remote_it->addr, @@ -233,9 +233,8 @@ nixlPosixBackendReqH::postXfer() { this); if (status != NIXL_SUCCESS) { - // Currently we do not support partial submissions, so it's all or nothing NIXL_ERROR << absl::StrFormat("Error preparing I/O operation: %d", status); - return status; + return queueResult(status); } } @@ -256,6 +255,7 @@ nixlPosixEngine::getPluginParams() { #endif #ifdef HAVE_LIBURING params["use_uring"] = "false"; + params["uring_open_synchronous"] = "false"; #endif #ifdef HAVE_POSIXAIO params["use_posix_aio"] = "false"; @@ -266,10 +266,14 @@ nixlPosixEngine::getPluginParams() { nixlPosixEngine::nixlPosixEngine(const nixlBackendInitParams *init_params) : nixlBackendEngine(init_params), io_queue_type_(getIoQueueType(init_params->customParams)), + uring_open_synchronous_(nixl::getBackendParamDefaulted(init_params->customParams, + "uring_open_synchronous", + false)), io_queue_(nixlPosixIOQueue::instantiate( io_queue_type_, nixl::getBackendParamDefaulted(init_params->customParams, "ios_pool_size", 0u), - nixl::getBackendParamDefaulted(init_params->customParams, "kernel_queue_size", 0u))), + nixl::getBackendParamDefaulted(init_params->customParams, "kernel_queue_size", 0u), + uring_open_synchronous_)), io_queue_lock_(init_params->syncMode) { if (io_queue_type_.empty()) { initErr = true; @@ -301,13 +305,22 @@ nixlPosixEngine::registerMem(const nixlBlobDesc &mem, return NIXL_ERR_INVALID_PARAM; } try { - out = new nixlPosixFileMD(mem.devId, mem.metaInfo); + auto file_md = std::make_unique(mem.devId); + nixl_status_t status; + { + NIXL_LOCK_GUARD(io_queue_lock_); + status = io_queue_->registerFile(mem.devId, mem.metaInfo); + } + if (status != NIXL_SUCCESS) { + return status; + } + resv.commit(); + out = file_md.release(); } catch (const std::system_error &e) { NIXL_ERROR << "POSIX path-mode open failed: " << e.what(); return NIXL_ERR_BACKEND; } - resv.commit(); } return NIXL_SUCCESS; } @@ -317,13 +330,17 @@ nixlPosixEngine::registerMem(const nixlBlobDesc &mem, nixl_status_t nixlPosixEngine::deregisterMem(nixlBackendMD *meta) { - // non-null meta is always a file MD. Release the path-mode reservation (path() empty in - // fd-mode) if (meta) { auto *file_md = static_cast(meta); - if (!file_md->file_fd.path().empty()) { - path_mode_devids_.release(file_md->devId); + nixl_status_t status; + { + NIXL_LOCK_GUARD(io_queue_lock_); + status = io_queue_->deregisterFile(file_md->devId); + } + if (status != NIXL_SUCCESS) { + return status; } + path_mode_devids_.release(file_md->devId); } delete meta; return NIXL_SUCCESS; diff --git a/src/plugins/posix/posix_backend.h b/src/plugins/posix/posix_backend.h index c8beb3ca1c..1fa571695b 100644 --- a/src/plugins/posix/posix_backend.h +++ b/src/plugins/posix/posix_backend.h @@ -29,8 +29,12 @@ #include "io_queue.h" #include "sync.h" -// POSIX reuses the shared owned-fd base (path-mode devId stored for dereg). -using nixlPosixFileMD = nixlFilePathMD; +class nixlPosixFileMD : public nixlBackendMD { +public: + explicit nixlPosixFileMD(uint64_t dev_id) : nixlBackendMD(true /*isPrivate*/), devId(dev_id) {} + + uint64_t devId; +}; class nixlPosixBackendReqH : public nixlBackendReqH { public: @@ -38,7 +42,7 @@ class nixlPosixBackendReqH : public nixlBackendReqH { const nixl_meta_dlist_t &local, const nixl_meta_dlist_t &remote, std::unique_ptr &io_queue); - ~nixlPosixBackendReqH() {}; + ~nixlPosixBackendReqH() override = default; nixl_status_t postXfer(); @@ -95,6 +99,7 @@ class nixlPosixBackendReqH : public nixlBackendReqH { class nixlPosixEngine : public nixlBackendEngine { private: std::string_view io_queue_type_; + bool uring_open_synchronous_; mutable std::unique_ptr io_queue_; mutable nixlLock io_queue_lock_; nixl::PathModeDevIdRegistry path_mode_devids_; diff --git a/src/utils/file/README.md b/src/utils/file/README.md index 2fb72a4d14..926236bb16 100644 --- a/src/utils/file/README.md +++ b/src/utils/file/README.md @@ -84,9 +84,16 @@ Test files are provided: ## Path-Mode File Registration Path-mode lets a caller declare a `FILE_SEG` descriptor by path in -`nixlBlobDesc::metaInfo` instead of pre-opening an fd; the backend -opens in `registerMem` and closes in `deregisterMem`. Motivation: -collapse N Python `os.open()` GIL crossings into one. +`nixlBlobDesc::metaInfo` instead of pre-opening an fd; the backend initiates +the open during `registerMem`, owns the resulting fd, and closes it during +`deregisterMem`. Motivation: collapse N Python `os.open()` GIL crossings into +one. + +With POSIX io_uring (`use_uring=true`), `registerMem` submits `OPENAT` +asynchronously by default and may return before the open completes. An invalid +or missing path can therefore register successfully, with the error reported +when a transfer first uses the descriptor. Set `uring_open_synchronous=true` +to make `registerMem` wait for the completion and report open errors directly. A `metaInfo` string is parsed as path-mode iff it matches: @@ -107,10 +114,12 @@ Examples: `ro:/var/cache/x.bin`, `rw,direct:/var/cache/x.bin`, (fail-loud); the design is strictly additive: any non-matching `metaInfo` falls through to caller-owned fd in `devId`. -Backends consume the shared helpers `nixl::parsePathMeta()` + -`nixlFilePathMD` from `file_path_mode.{h,cpp}`. POSIX uses -`nixlFilePathMD` directly; HF3FS / CUDA_GDS / GDS_MT extend their -existing per-descriptor MD struct with `owned` (and close the fd in -`deregisterMem` after the backend-specific teardown). The GDS per-fd -caches key on the *opened* fd, so two path-mode registrations of the -same path yield two cuFile handles (no path-level dedup). +Backends consume `nixl::parsePathMeta()` and the shared helpers from +`file_path_mode.{h,cpp}`. POSIX metadata (`nixlPosixFileMD`) stores only the +`uint64_t` device identifier; the selected I/O queue owns the registered file +state and lifetime. POSIX AIO queues keep a `FileFd` in their queue registry, +while the io_uring path owns asynchronous open and close state in `fileState`. +HF3FS / CUDA_GDS / GDS_MT extend their existing per-descriptor MD struct with +`owned` (and close the fd in `deregisterMem` after the backend-specific +teardown). The GDS per-fd caches key on the *opened* fd, so two path-mode +registrations of the same path yield two cuFile handles (no path-level dedup). diff --git a/test/unit/plugins/posix/meson.build b/test/unit/plugins/posix/meson.build index e63efe140c..1ecc619b30 100644 --- a/test/unit/plugins/posix/meson.build +++ b/test/unit/plugins/posix/meson.build @@ -50,7 +50,11 @@ if has_posix_plugin if has_io_uring nixl_posix_uring_app = executable( - 'nixl_posix_uring_test', 'nixl_posix_uring_test.cpp', + 'nixl_posix_uring_test', + [ + 'nixl_posix_uring_test.cpp', + 'nixl_posix_uring_path_agent_test.cpp', + ], dependencies: [nixl_dep, nixl_infra, file_utils_interface, absl_log_dep, posix_backend_interface, io_uring_dep], include_directories: [nixl_inc_dirs, utils_inc_dirs, @@ -58,6 +62,10 @@ if has_posix_plugin include_directories('../../../../src/plugins/posix')], link_args: ['-Wl,--export-dynamic'], install: true) - test('posix_uring_backend_test', nixl_posix_uring_app) + posix_uring_test_env = environment() + posix_uring_test_env.set( + 'NIXL_PLUGIN_DIR', join_paths(plugin_build_dir, 'src/plugins/posix')) + test('posix_uring_backend_test', nixl_posix_uring_app, + env: posix_uring_test_env) endif endif diff --git a/test/unit/plugins/posix/nixl_posix_aio_test.cpp b/test/unit/plugins/posix/nixl_posix_aio_test.cpp index 0918e84016..35aa22029c 100644 --- a/test/unit/plugins/posix/nixl_posix_aio_test.cpp +++ b/test/unit/plugins/posix/nixl_posix_aio_test.cpp @@ -82,12 +82,17 @@ struct aioTest { throw std::runtime_error("mkstemp failed"); } unlink(path); + if (queue->registerFile(fd, "") != NIXL_SUCCESS) { + close(fd); + throw std::runtime_error("file registration failed"); + } for (size_t i = 0; i < buffers.size(); i++) { std::memset(buffers[i].data(), static_cast(i + 1), buffers[i].size()); } } ~aioTest() { + queue->deregisterFile(fd); queue.reset(); close(fd); } @@ -288,7 +293,7 @@ main() { { setSubmitMode(submitMode::PARTIAL_THEN_ERROR); aioTest test(request_count + 1); - nixlPosixFileMD file_md(test.fd, ""); + nixlPosixFileMD file_md(test.fd); aioRequest failed(test, file_md, 0, request_count); aioRequest unrelated(test, file_md, request_count, 1); @@ -312,7 +317,7 @@ main() { constexpr int completion_error_request_count = 65; setSubmitMode(submitMode::COMPLETION_ERROR); aioTest test(completion_error_request_count); - nixlPosixFileMD file_md(test.fd, ""); + nixlPosixFileMD file_md(test.fd); aioRequest failed(test, file_md, 0, completion_error_request_count); nixl_status_t status = failed.request.postXfer(); diff --git a/test/unit/plugins/posix/nixl_posix_uring_path_agent_test.cpp b/test/unit/plugins/posix/nixl_posix_uring_path_agent_test.cpp new file mode 100644 index 0000000000..49827873da --- /dev/null +++ b/test/unit/plugins/posix/nixl_posix_uring_path_agent_test.cpp @@ -0,0 +1,262 @@ +/* + * SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. + * SPDX-License-Identifier: Apache-2.0 + */ + +#include "nixl.h" +#include "nixl_descriptors.h" +#include "nixl_params.h" + +#include +#include + +#include +#include +#include +#include +#include +#include + +namespace { + +constexpr char agent_name[] = "POSIXUringEagerOpenReuse"; + +bool +runXfer(nixlAgent &agent, + nixl_xfer_op_t operation, + const nixl_xfer_dlist_t &dram, + const nixl_xfer_dlist_t &file) { + nixlXferReqH *request = nullptr; + nixl_status_t status = agent.createXferReq(operation, dram, file, agent_name, request); + if (status == NIXL_SUCCESS) { + status = agent.postXferReq(request); + while (status == NIXL_IN_PROG) { + status = agent.getXferStatus(request); + } + agent.releaseXferReq(request); + } + return status == NIXL_SUCCESS; +} + +int +fail(const char *message, const char *path) { + unlink(path); + std::cerr << message << std::endl; + return 1; +} + +} // namespace + +int +runPosixUringPathAgentTest() { + nixlAgentConfig config; + nixlAgent agent(agent_name, config); + nixl_b_params_t params; + params["use_uring"] = "true"; + + nixlBackendH *backend = nullptr; + if (agent.createBackend("POSIX", params, backend) != NIXL_SUCCESS || !backend) { + std::cout << "SKIP: kernel lacks asynchronous io_uring open support" << std::endl; + return 77; + } + + char path[] = "/tmp/nixl-uring-eager-agent-XXXXXX"; + int fd = mkstemp(path); + if (fd < 0) { + return 1; + } + std::array expected; + expected.fill(0x6b); + if (write(fd, expected.data(), expected.size()) != static_cast(expected.size())) { + close(fd); + return fail("failed to initialize agent eager-open test", path); + } + close(fd); + + const uint64_t wide_dev_id = static_cast(std::numeric_limits::max()) + 1; + nixl_reg_dlist_t wide_path_reg(FILE_SEG); + nixlBlobDesc wide_path_desc; + wide_path_desc.addr = 0; + wide_path_desc.len = expected.size(); + wide_path_desc.devId = wide_dev_id; + wide_path_desc.metaInfo = std::string("ro:") + path; + wide_path_reg.addDesc(wide_path_desc); + if (agent.registerMem(wide_path_reg) != NIXL_SUCCESS) { + return fail("64-bit path-mode devId registration failed", path); + } + if (agent.registerMem(wide_path_reg) == NIXL_SUCCESS) { + return fail("duplicate path-mode devId registration was not rejected", path); + } + if (agent.deregisterMem(wide_path_reg) != NIXL_SUCCESS) { + return fail("64-bit path-mode devId deregistration failed", path); + } + if (agent.registerMem(wide_path_reg) != NIXL_SUCCESS) { + return fail("released path-mode devId could not be registered again", path); + } + if (agent.deregisterMem(wide_path_reg) != NIXL_SUCCESS) { + return fail("re-registered path-mode devId deregistration failed", path); + } + + nixl_reg_dlist_t invalid_fd_reg(FILE_SEG); + nixlBlobDesc invalid_fd_desc; + invalid_fd_desc.addr = 0; + invalid_fd_desc.len = expected.size(); + invalid_fd_desc.devId = wide_dev_id; + invalid_fd_reg.addDesc(invalid_fd_desc); + if (agent.registerMem(invalid_fd_reg) == NIXL_SUCCESS) { + return fail("out-of-range fd-mode devId was not rejected", path); + } + + nixl_reg_dlist_t file_reg(FILE_SEG); + nixlBlobDesc file_desc; + file_desc.addr = 0; + file_desc.len = expected.size(); + file_desc.devId = 1; + file_desc.metaInfo = std::string("ro:") + path; + file_reg.addDesc(file_desc); + + std::array buffer{}; + nixl_reg_dlist_t dram_reg(DRAM_SEG); + nixlBlobDesc dram_desc; + dram_desc.addr = reinterpret_cast(buffer.data()); + dram_desc.len = buffer.size(); + dram_desc.devId = 0; + dram_reg.addDesc(dram_desc); + + if (agent.registerMem(file_reg) != NIXL_SUCCESS || + agent.registerMem(dram_reg) != NIXL_SUCCESS) { + return fail("eager asynchronous path registration failed", path); + } + + const nixl_xfer_dlist_t file_xfer = file_reg.trim(); + const nixl_xfer_dlist_t dram_xfer = dram_reg.trim(); + if (!runXfer(agent, NIXL_READ, dram_xfer, file_xfer) || + !std::equal(buffer.begin(), buffer.end(), expected.begin())) { + return fail("first registered eager-open transfer failed", path); + } + + if (unlink(path) != 0) { + return fail("failed to unlink after first eager-open transfer", path); + } + buffer.fill(0); + if (!runXfer(agent, NIXL_READ, dram_xfer, file_xfer) || + !std::equal(buffer.begin(), buffer.end(), expected.begin())) { + return fail("second transfer did not reuse the registered open file", path); + } + + char create_path[] = "/tmp/nixl-uring-eager-create-XXXXXX"; + int create_fd = mkstemp(create_path); + if (create_fd < 0) { + return fail("failed to reserve eager-create path", path); + } + close(create_fd); + unlink(create_path); + + nixl_reg_dlist_t create_reg(FILE_SEG); + nixlBlobDesc create_desc; + create_desc.addr = 0; + create_desc.len = buffer.size(); + create_desc.devId = 2; + create_desc.metaInfo = std::string("rw,create:") + create_path; + create_reg.addDesc(create_desc); + buffer.fill(0x3c); + if (agent.registerMem(create_reg) != NIXL_SUCCESS) { + return fail("eager rw,create registration failed", create_path); + } + if (!runXfer(agent, NIXL_WRITE, dram_xfer, create_reg.trim()) || + access(create_path, F_OK) != 0) { + return fail("transfer waiting on eager create failed", create_path); + } + if (agent.deregisterMem(create_reg) != NIXL_SUCCESS) { + return fail("eager-created file deregistration failed", create_path); + } + unlink(create_path); + + char missing_path[] = "/tmp/nixl-uring-eager-missing-XXXXXX"; + int missing_fd = mkstemp(missing_path); + if (missing_fd < 0) { + return fail("failed to reserve missing-file path", path); + } + close(missing_fd); + unlink(missing_path); + + nixl_reg_dlist_t missing_reg(FILE_SEG); + nixlBlobDesc missing_desc; + missing_desc.addr = 0; + missing_desc.len = buffer.size(); + missing_desc.devId = 3; + missing_desc.metaInfo = std::string("ro:") + missing_path; + missing_reg.addDesc(missing_desc); + if (agent.registerMem(missing_reg) != NIXL_SUCCESS) { + return fail("missing path failed during eager asynchronous registration", missing_path); + } + if (runXfer(agent, NIXL_READ, dram_xfer, missing_reg.trim())) { + return fail("missing path unexpectedly transferred successfully", missing_path); + } + + std::array sibling_buffer{}; + nixl_reg_dlist_t sibling_dram_reg(DRAM_SEG); + nixlBlobDesc sibling_dram_desc; + sibling_dram_desc.addr = reinterpret_cast(sibling_buffer.data()); + sibling_dram_desc.len = sibling_buffer.size(); + sibling_dram_desc.devId = 0; + sibling_dram_reg.addDesc(sibling_dram_desc); + if (agent.registerMem(sibling_dram_reg) != NIXL_SUCCESS) { + return fail("failed to register mixed-transfer buffer", missing_path); + } + + nixl_xfer_dlist_t mixed_dram(DRAM_SEG); + mixed_dram.addDesc(*dram_xfer.begin()); + mixed_dram.addDesc(*sibling_dram_reg.trim().begin()); + nixl_xfer_dlist_t mixed_file(FILE_SEG); + mixed_file.addDesc(*file_xfer.begin()); + mixed_file.addDesc(*missing_reg.trim().begin()); + buffer.fill(0); + if (runXfer(agent, NIXL_READ, mixed_dram, mixed_file) || + std::any_of(buffer.begin(), buffer.end(), [](unsigned char value) { return value != 0; })) { + return fail("failed-open transfer did not cancel its valid sibling", missing_path); + } + if (!runXfer(agent, NIXL_READ, dram_xfer, file_xfer)) { + return fail("queue was unusable after failed-open cancellation", missing_path); + } + + if (agent.deregisterMem(missing_reg) != NIXL_SUCCESS || + agent.deregisterMem(file_reg) != NIXL_SUCCESS || + agent.deregisterMem(sibling_dram_reg) != NIXL_SUCCESS || + agent.deregisterMem(dram_reg) != NIXL_SUCCESS) { + return fail("failure-path deregistration failed", missing_path); + } + + nixlAgent synchronous_agent("POSIXUringOpenSynchronous", config); + nixl_b_params_t synchronous_params; + synchronous_params["use_uring"] = "true"; + synchronous_params["uring_open_synchronous"] = "true"; + nixlBackendH *synchronous_backend = nullptr; + if (synchronous_agent.createBackend("POSIX", synchronous_params, synchronous_backend) != + NIXL_SUCCESS || + !synchronous_backend) { + return fail("failed to create synchronous-open backend", missing_path); + } + + char synchronous_missing_path[] = "/tmp/nixl-uring-synchronous-missing-XXXXXX"; + int synchronous_missing_fd = mkstemp(synchronous_missing_path); + if (synchronous_missing_fd < 0) { + return fail("failed to reserve synchronous missing-file path", missing_path); + } + close(synchronous_missing_fd); + unlink(synchronous_missing_path); + + nixl_reg_dlist_t synchronous_missing_reg(FILE_SEG); + nixlBlobDesc synchronous_missing_desc; + synchronous_missing_desc.addr = 0; + synchronous_missing_desc.len = buffer.size(); + synchronous_missing_desc.devId = 4; + synchronous_missing_desc.metaInfo = std::string("ro:") + synchronous_missing_path; + synchronous_missing_reg.addDesc(synchronous_missing_desc); + if (synchronous_agent.registerMem(synchronous_missing_reg) == NIXL_SUCCESS) { + return fail("synchronous open did not report the failure", synchronous_missing_path); + } + + std::cout << "asynchronous open and synchronous open error reporting: OK" << std::endl; + return 0; +} diff --git a/test/unit/plugins/posix/nixl_posix_uring_test.cpp b/test/unit/plugins/posix/nixl_posix_uring_test.cpp index 95d97e90dd..7ed26201b0 100644 --- a/test/unit/plugins/posix/nixl_posix_uring_test.cpp +++ b/test/unit/plugins/posix/nixl_posix_uring_test.cpp @@ -7,6 +7,7 @@ #include #include #include +#include #include #include #include @@ -21,6 +22,9 @@ #define LIBURING_NOEXCEPT #endif +int +runPosixUringPathAgentTest(); + namespace { constexpr int request_count = 32, ring_entries = 16, max_poll_iterations = 2000; constexpr size_t block_size = 4096; @@ -63,12 +67,17 @@ struct uringTest { throw std::runtime_error("mkstemp failed"); } unlink(path); + if (queue->registerFile(fd, "") != NIXL_SUCCESS) { + close(fd); + throw std::runtime_error("file registration failed"); + } for (size_t i = 0; i < buffers.size(); i++) { std::memset(buffers[i].data(), static_cast(i + 1), buffers[i].size()); } } ~uringTest() { + queue->deregisterFile(fd); queue.reset(); close(fd); } @@ -121,6 +130,102 @@ struct uringRequest { request(operation, local, remote, test.queue) {} }; +int +countProcessFds(); + +bool +runEagerPathModeTest() { + char path[] = "/tmp/nixl_uring_eager_path_test_XXXXXX"; + int fd = mkstemp(path); + if (fd < 0) { + return false; + } + + std::array expected; + expected.fill(0x5a); + bool success = + write(fd, expected.data(), expected.size()) == static_cast(expected.size()); + close(fd); + if (!success) { + unlink(path); + return false; + } + + auto queue = nixlPosixIOQueue::instantiate("URING", 64, ring_entries); + if (!queue) { + unlink(path); + return false; + } + const int fds_before_open = countProcessFds(); + if (fds_before_open < 0) { + unlink(path); + return false; + } + + constexpr int dev_id = 1; + std::array buffer{}; + completionState state; + // A transfer posted before the open CQE is reaped must remain deferred until poll(). + success = queue->registerFile(dev_id, std::string("ro:") + path) == NIXL_SUCCESS && + queue->enqueue(dev_id, buffer.data(), buffer.size(), 0, true, completionCallback, &state) == + NIXL_SUCCESS && + queue->deregisterFile(dev_id) == NIXL_ERR_NOT_ALLOWED && queue->post() == NIXL_IN_PROG && + state.count == 0; + + for (int i = 0; success && state.count == 0 && i < max_poll_iterations; ++i) { + success = queue->poll() >= 0; + std::this_thread::sleep_for(poll_pause); + } + + success = success && state.count == 1 && state.errors == 0 && buffer == expected; + + if (queue->deregisterFile(dev_id) != NIXL_SUCCESS) { + success = false; + } + success = success && countProcessFds() == fds_before_open; + + unlink(path); + return success; +} + +int +countProcessFds() { + DIR *dir = opendir("/proc/self/fd"); + if (!dir) { + return -1; + } + int count = 0; + while (readdir(dir)) { + ++count; + } + closedir(dir); + return count; +} + +bool +runEagerOpenShutdownTest() { + const int before = countProcessFds(); + if (before < 0) { + return false; + } + + { + auto queue = nixlPosixIOQueue::instantiate("URING", 1, ring_entries); + if (!queue) { + return false; + } + + for (int i = 0; i < request_count; ++i) { + if (queue->registerFile(i, "ro:/dev/null") != NIXL_SUCCESS) { + return false; + } + } + // Queue destruction must reap pending OPENAT CQEs and close every resulting fd. + } + + return countProcessFds() == before; +} + #define URING_CHECK(condition) \ do { \ if (!(condition)) { \ @@ -161,8 +266,7 @@ io_uring_submit(struct io_uring *ring) LIBURING_NOEXCEPT { int main() { io_uring probe_ring{}; - io_uring_params probe_params{}; - int probe_ret = io_uring_queue_init_params(ring_entries, &probe_ring, &probe_params); + int probe_ret = io_uring_queue_init(ring_entries, &probe_ring, 0); if (probe_ret < 0) { std::cerr << "io_uring backend test requires a usable ring: " << std::strerror(-probe_ret) << " (" << probe_ret << ")" << std::endl; @@ -189,7 +293,7 @@ main() { } { uringTest test(submit_mode_t::PASS_THROUGH); - nixlPosixFileMD file_md(test.fd, ""); + nixlPosixFileMD file_md(test.fd); uringRequest cancelled(test, file_md, 0), unrelated(test, file_md, 1); nixl_status_t cancelled_status = cancelled.request.postXfer(); URING_CHECK(cancelled_status >= NIXL_IN_PROG); @@ -210,5 +314,11 @@ main() { URING_CHECK(cancelled_status == NIXL_SUCCESS || cancelled_status == NIXL_ERR_BACKEND); URING_CHECK(test.drain() == NIXL_SUCCESS && cancel_completions == 1); } + + URING_CHECK(runEagerPathModeTest()); + URING_CHECK(runEagerOpenShutdownTest()); + URING_CHECK(runPosixUringPathAgentTest() == 0); + std::cout << "io_uring eager path opens: regular-fd=tested" << std::endl; + return 0; }