12#include "ublk/ublk_cmd.h"
20#include <system_error>
27namespace ex = condy::detail::ex;
29inline bool need_alloc_buf(uint64_t flags)
noexcept {
30 return !(flags & (UBLK_F_SUPPORT_ZERO_COPY | UBLK_F_USER_COPY));
33inline bool need_io_buf(uint64_t flags, uint32_t op_flags)
noexcept {
34 bool zero_copy = flags & UBLK_F_SUPPORT_ZERO_COPY;
35 bool auto_reg = flags & UBLK_F_AUTO_BUF_REG;
36 bool need_reg_buf = op_flags & UBLK_IO_F_NEED_REG_BUF;
37 return need_reg_buf || (zero_copy && !auto_reg);
40inline off_t io_desc_offset(
size_t q_id)
noexcept {
41 return UBLKSRV_CMD_BUF_OFFSET +
42 q_id * UBLK_MAX_QUEUE_DEPTH *
sizeof(ublksrv_io_desc);
46concept has_res =
requires(T t) {
47 { t.res } -> std::convertible_to<int32_t>;
51concept has_zone_lba =
requires(T t) {
52 { t.zone_lba } -> std::convertible_to<uint64_t>;
55inline void extract_io_result(
auto &&r, int32_t &res,
56 uint64_t &zone_lba)
noexcept {
57 using R = std::decay_t<
decltype(r)>;
58 if constexpr (std::convertible_to<R, int32_t>) {
62 static_assert(has_res<R>,
63 "returned result is neither int32_t nor has a "
67 if constexpr (has_zone_lba<R>) {
68 zone_lba = r.zone_lba;
73inline std::error_code normalize_error(
const auto &err,
int v)
noexcept {
74 using E = std::decay_t<
decltype(err)>;
75 if constexpr (std::same_as<E, std::error_code>) {
77 }
else if constexpr (std::same_as<E, std::exception_ptr>) {
79 std::rethrow_exception(err);
80 }
catch (
const std::system_error &se) {
83 return std::error_code(v, std::generic_category());
86 return std::error_code(v, std::generic_category());
90template <
typename Sender>
91inline auto wrap_handle_io(Sender &&sender, int32_t &res,
92 uint64_t &zone_lba)
noexcept {
93 return std::forward<Sender>(sender) |
95 std::error_code(ECANCELED, std::generic_category())) |
96 ex::upon_error([](
const auto &err)
noexcept {
97 return normalize_error(err, EIO);
99 ex::then([&](
auto r)
noexcept {
100 using R =
decltype(r);
101 if constexpr (std::same_as<R, std::error_code>) {
105 extract_io_result(r, res, zone_lba);
107 return std::make_tuple(res, zone_lba);
111template <
typename Fd, IoHandler Handler,
typename Alloc>
class IoQueue {
113 IoQueue(Fd ublkc_fd,
int ublkc_fd_raw, uint16_t q_id, uint64_t flags,
114 uint16_t queue_depth, uint32_t max_io_buf_bytes, Handler &handler,
116 : ublkc_fd_(ublkc_fd), q_id_(q_id), flags_(flags),
117 queue_depth_(queue_depth), max_io_buf_bytes_(max_io_buf_bytes),
118 handler_(handler), alloc_(alloc) {
120 auto d = defer([&]
noexcept {
126 size_t size = queue_depth_ *
sizeof(ublksrv_io_desc);
127 auto off = io_desc_offset(q_id_);
128 void *addr = mmap(0, size, PROT_READ, MAP_SHARED | MAP_POPULATE,
130 if (addr == MAP_FAILED) {
131 throw std::system_error(errno, std::generic_category(),
134 iod_base_ =
reinterpret_cast<ublksrv_io_desc *
>(addr);
136 const size_t page = sysconf(_SC_PAGESIZE);
137 if (need_alloc_buf(flags_)) {
138 buf_bases_.reserve(queue_depth_);
139 for (uint16_t tag = 0; tag < queue_depth_; tag++) {
140 void *buf = alloc_aligned(alloc_, max_io_buf_bytes_, page);
141 buf_bases_.push_back(buf);
148 IoQueue(
const IoQueue &) =
delete;
149 IoQueue &operator=(
const IoQueue &) =
delete;
150 IoQueue(IoQueue &&) =
delete;
151 IoQueue &operator=(IoQueue &&) =
delete;
153 ~IoQueue() { cleanup_(); }
156 template <
typename Sched> ex::task<void, TaskEnv<Sched, Alloc>> run() {
157 if constexpr (QueueHandler<Handler>) {
158 co_await handler_.init_queue(q_id_);
161 auto d = defer([&]()
noexcept {
162 if constexpr (QueueHandler<Handler>) {
163 handler_.destroy_queue(q_id_);
167 ex::simple_counting_scope scope;
168 AllocVector<std::exception_ptr,
decltype(alloc_)> errs(queue_depth_,
170 auto sched =
co_await ex::read_env(ex::get_start_scheduler);
171 for (uint16_t tag = 0; tag < queue_depth_; tag++) {
172 auto task = worker_<Sched>(tag, iod_base_ + tag, get_buf_(tag));
174 ex::starts_on(sched, std::move(task)) |
175 ex::upon_error([&, tag](
const std::exception_ptr &ep)
noexcept {
178 ex::spawn(std::move(s), scope.get_token());
181 co_await scope.join();
183 for (
auto &ep : errs) {
185 std::rethrow_exception(ep);
191 void cleanup_() noexcept {
192 if (!buf_bases_.empty()) {
193 const size_t page = sysconf(_SC_PAGESIZE);
194 for (
auto *buf : buf_bases_) {
195 free_aligned(alloc_, buf, max_io_buf_bytes_, page);
200 size_t size = queue_depth_ *
sizeof(ublksrv_io_desc);
201 munmap(iod_base_, size);
206 void *get_buf_(uint16_t tag)
const noexcept {
207 if (buf_bases_.empty()) {
210 return buf_bases_[tag];
213 template <
typename Sched>
214 ex::task<void, TaskEnv<Sched, Alloc>>
215 worker_(uint16_t tag,
const ublksrv_io_desc *iod,
void *buf) {
217 bool zero_copy = flags_ & UBLK_F_SUPPORT_ZERO_COPY;
218 bool auto_reg = flags_ & UBLK_F_AUTO_BUF_REG;
220 uint64_t buf_addr =
reinterpret_cast<uint64_t
>(buf);
221 uint64_t sqe_addr = 0;
223 ublk_auto_buf_reg reg = {};
225 reg.flags = UBLK_AUTO_BUF_REG_FALLBACK;
226 sqe_addr = ublk_auto_buf_reg_to_sqe_addr(®);
232 [&](std::error_code ec)
noexcept {
return -ec.value(); }));
233 if (r == UBLK_IO_RES_NEED_GET_DATA) {
235 ex::upon_error([&](std::error_code ec)
noexcept {
239 if (r == UBLK_IO_RES_ABORT) {
242 throw std::system_error(-r, std::generic_category(),
"fetch_req");
246 bool io_buf = need_io_buf(flags_, iod->op_flags);
253 const IoData io_data{q_id_, tag, iod, buf};
254 co_await wrap_handle_io(handler_.handle_io(io_data), res, zone_lba);
260 uint64_t curr_buf_addr = buf_addr;
261 if (ublksrv_get_op(iod) == UBLK_IO_OP_ZONE_APPEND) {
262 assert(buf_addr == 0);
263 curr_buf_addr = zone_lba;
266 curr_buf_addr, sqe_addr) |
267 ex::upon_error([&](std::error_code ec)
noexcept {
270 if (r == UBLK_IO_RES_NEED_GET_DATA) {
273 ex::upon_error([&](std::error_code ec)
noexcept {
277 if (r == UBLK_IO_RES_ABORT) {
280 throw std::system_error(-r, std::generic_category(),
281 "commit_and_fetch_req");
290 uint16_t queue_depth_;
291 uint32_t max_io_buf_bytes_;
294 ublksrv_io_desc *iod_base_ =
nullptr;
295 AllocVector<void *, Alloc> buf_bases_;
298template <
typename Fd, IoHandler Handler,
typename Alloc>
class BatchIoQueue {
300 BatchIoQueue(Fd ublkc_fd,
int ublkc_fd_raw, uint16_t q_id, uint64_t flags,
301 uint16_t queue_depth, uint32_t max_io_buf_bytes,
302 Handler &handler,
const Alloc &alloc)
303 : ublkc_fd_(ublkc_fd), q_id_(q_id), flags_(flags),
304 queue_depth_(queue_depth), max_io_buf_bytes_(max_io_buf_bytes),
305 handler_(handler), alloc_(alloc) {
307 auto d = defer([&]
noexcept {
314 size_t size = queue_depth_ *
sizeof(ublksrv_io_desc);
315 auto off = io_desc_offset(q_id_);
316 void *addr = mmap(0, size, PROT_READ, MAP_SHARED | MAP_POPULATE,
318 if (addr == MAP_FAILED) {
319 throw std::system_error(errno, std::generic_category(),
322 iod_base_ =
reinterpret_cast<ublksrv_io_desc *
>(addr);
325 const size_t page = sysconf(_SC_PAGESIZE);
327 size_t size = align_up(queue_depth_ * commit_element_size_(), page);
328 commit_buf_size_ = size;
329 commit_buf_[0] = alloc_aligned(alloc_, size * 2, page);
330 commit_buf_[1] =
static_cast<char *
>(commit_buf_[0]) + size;
331 if (mlock(commit_buf_[0], size * 2) < 0) {
332 throw std::system_error(errno, std::generic_category(),
338 size_t size = align_up(queue_depth_ *
sizeof(uint16_t), page);
339 fetch_buf_size_ = size;
340 fetch_buf_[0] =
reinterpret_cast<uint16_t *
>(
341 alloc_aligned(alloc_, size * 2, page));
342 fetch_buf_[1] =
reinterpret_cast<uint16_t *
>(
343 reinterpret_cast<char *
>(fetch_buf_[0]) + size);
344 if (mlock(fetch_buf_[0], size * 2) < 0) {
345 throw std::system_error(errno, std::generic_category(),
350 if (need_alloc_buf(flags_)) {
351 buf_bases_.reserve(queue_depth_);
352 for (uint16_t tag = 0; tag < queue_depth_; tag++) {
353 void *buf = alloc_aligned(alloc_, max_io_buf_bytes_, page);
354 buf_bases_.push_back(buf);
361 BatchIoQueue(
const BatchIoQueue &) =
delete;
362 BatchIoQueue &operator=(
const BatchIoQueue &) =
delete;
363 BatchIoQueue(BatchIoQueue &&) =
delete;
364 BatchIoQueue &operator=(BatchIoQueue &&) =
delete;
366 ~BatchIoQueue() { cleanup_(); }
369 template <
typename Sched> ex::task<void, TaskEnv<Sched, Alloc>> run() {
370 auto sched =
co_await ex::read_env(ex::get_scheduler);
371 auto alloc =
co_await ex::read_env(ex::get_allocator);
373 if constexpr (QueueHandler<Handler>) {
374 co_await handler_.init_queue(q_id_);
377 auto d = defer([&]()
noexcept {
378 if constexpr (QueueHandler<Handler>) {
379 handler_.destroy_queue(q_id_);
383 for (uint16_t tag = 0; tag < queue_depth_; tag++) {
384 set_result_(0, tag, tag, 0);
387 queue_depth_, commit_element_size_(),
389 queue_depth_ * commit_element_size_());
391 FlusherState flusher;
392 AllocVector<WorkerState,
decltype(alloc)> workers(queue_depth_, alloc);
394 ex::simple_counting_scope scope;
395 std::exception_ptr flusher_err;
396 AllocVector<std::exception_ptr,
decltype(alloc)> worker_errs(
397 queue_depth_, alloc);
398 bool worker_stopped =
false;
399 size_t running_workers = queue_depth_;
402 auto task = flusher_<Sched>(flusher, running_workers);
403 auto s = ex::starts_on(sched, std::move(task)) |
404 ex::upon_error([&](std::exception_ptr ep)
noexcept {
405 flusher_err = std::move(ep);
407 ex::spawn(std::move(s), scope.get_token());
409 for (uint16_t tag = 0; tag < queue_depth_; tag++) {
410 auto task = worker_<Sched>(workers[tag], flusher, worker_stopped,
411 tag, iod_base_ + tag, get_buf_(tag));
413 ex::starts_on(sched, std::move(task)) |
414 ex::upon_error([&, tag](
const std::exception_ptr &ep)
noexcept {
415 worker_errs[tag] = ep;
417 ex::then([&]
noexcept {
418 if (--running_workers == 0) {
419 flusher.futex.notify_one();
422 ex::spawn(std::move(s), scope.get_token());
425 condy::ProvidedBufferQueue queue(2, IOU_PBUF_RING_INC);
426 queue.push(condy::buffer(fetch_buf_[0], fetch_buf_size_));
427 queue.push(condy::buffer(fetch_buf_[1], fetch_buf_size_));
431 auto fetch_cb = [&](std::pair<int32_t, condy::BufferInfo> r)
noexcept {
432 auto &[res, info] = r;
434 auto *fetch_buf = fetch_buf_[bid];
436 size_t nr_tags =
static_cast<size_t>(res) /
sizeof(uint16_t);
437 size_t end = off + nr_tags;
438 for (
size_t j = off; j < end; j++) {
439 uint16_t tag = fetch_buf[j];
440 auto &worker = workers[tag];
441 write_once_(worker.flag,
true);
442 worker.futex.notify_one();
445 bool consumed = info.num_buffers;
447 auto r = queue.push(condy::buffer(fetch_buf, fetch_buf_size_));
457 ex::then([](int32_t r, condy::BufferInfo)
noexcept {
return r; }) |
459 [&](std::error_code ec)
noexcept {
return -ec.value(); }) |
460 ex::then([&](int32_t r)
noexcept { res = r; }) |
461 ex::then([&]()
noexcept {
462 worker_stopped =
true;
463 for (
auto &worker : workers) {
464 worker.futex.notify_one();
468 co_await ex::when_all(std::move(s), scope.join());
470 for (
auto &ep : worker_errs) {
472 std::rethrow_exception(ep);
476 std::rethrow_exception(flusher_err);
479 if (res != UBLK_IO_RES_ABORT) {
480 throw std::system_error(-res, std::generic_category(),
486 void cleanup_() noexcept {
487 const size_t page = sysconf(_SC_PAGESIZE);
488 if (!buf_bases_.empty()) {
489 for (
auto *buf : buf_bases_) {
490 free_aligned(alloc_, buf, max_io_buf_bytes_, page);
495 munlock(fetch_buf_[0], fetch_buf_size_ * 2);
496 free_aligned(alloc_, fetch_buf_[0], fetch_buf_size_ * 2, page);
498 if (commit_buf_[0]) {
499 munlock(commit_buf_[0], commit_buf_size_ * 2);
500 free_aligned(alloc_, commit_buf_[0], commit_buf_size_ * 2, page);
503 size_t size = queue_depth_ *
sizeof(ublksrv_io_desc);
504 munmap(iod_base_, size);
509 template <
typename T>
510 static auto read_once_(
const std::atomic<T> &a)
noexcept {
511 return a.load(std::memory_order_relaxed);
514 template <
typename T,
typename V>
515 static void write_once_(std::atomic<T> &a, V v)
noexcept {
516 a.store(v, std::memory_order_relaxed);
519 bool need_f_buf_addr_() noexcept {
520 return !(flags_ & UBLK_F_AUTO_BUF_REG) && need_alloc_buf(flags_);
523 bool need_f_zone_lba_() noexcept {
return flags_ & UBLK_F_ZONED; }
525 uint16_t get_batch_flags_() noexcept {
527 if (flags_ & UBLK_F_AUTO_BUF_REG) {
528 f |= UBLK_BATCH_F_AUTO_BUF_REG_FALLBACK;
529 }
else if (need_f_buf_addr_()) {
530 f |= UBLK_BATCH_F_HAS_BUF_ADDR;
532 if (need_f_zone_lba_()) {
533 f |= UBLK_BATCH_F_HAS_ZONE_LBA;
538 size_t commit_element_size_() noexcept {
539 size_t size =
sizeof(ublk_elem_header);
540 if (need_f_buf_addr_()) {
541 size +=
sizeof(uint64_t);
543 if (need_f_zone_lba_()) {
544 size +=
sizeof(uint64_t);
549 ublk_elem_header &commit_f_hdr(
size_t index,
size_t pos)
noexcept {
550 return *
reinterpret_cast<ublk_elem_header *
>(
551 static_cast<char *
>(commit_buf_[index]) +
552 pos * commit_element_size_());
555 uint64_t &commit_f_buf_addr(
size_t index,
size_t pos)
noexcept {
556 assert(need_f_buf_addr_());
557 return *
reinterpret_cast<uint64_t *
>(
558 reinterpret_cast<char *
>(&commit_f_hdr(index, pos)) +
559 sizeof(ublk_elem_header));
562 uint64_t &commit_f_zone_lba(
size_t index,
size_t pos)
noexcept {
563 assert(need_f_zone_lba_());
564 size_t off =
sizeof(ublk_elem_header);
565 if (need_f_buf_addr_()) {
566 off +=
sizeof(uint64_t);
568 return *
reinterpret_cast<uint64_t *
>(
569 reinterpret_cast<char *
>(&commit_f_hdr(index, pos)) + off);
572 void set_result_(
size_t index,
size_t pos, uint16_t tag,
573 int32_t res)
noexcept {
574 auto &hdr = commit_f_hdr(index, pos);
577 if (flags_ & UBLK_F_AUTO_BUF_REG) {
579 }
else if (need_f_buf_addr_()) {
580 commit_f_buf_addr(index, pos) =
581 reinterpret_cast<uint64_t
>(buf_bases_[tag]);
585 void set_zone_lba_(
size_t index,
size_t pos, uint64_t lba)
noexcept {
586 assert(need_f_zone_lba_());
587 commit_f_zone_lba(index, pos) = lba;
590 void *get_buf_(uint16_t tag)
const noexcept {
591 if (buf_bases_.empty()) {
594 return buf_bases_[tag];
598 struct FlusherState {
600 std::atomic<uint16_t> pending = 0;
601 condy::Futex<uint16_t> futex{pending};
604 template <
typename Sched>
605 ex::task<void, TaskEnv<Sched, Alloc>> flusher_(FlusherState &state,
606 size_t &running_workers) {
607 auto sched =
co_await ex::read_env(ex::get_start_scheduler);
609 auto nr = read_once_(state.pending);
611 auto cur_index = state.index;
612 state.index = 1 - state.index;
613 write_once_(state.pending, 0);
615 ublkc_fd_, q_id_, get_batch_flags_(), nr,
616 commit_element_size_(), commit_buf_[cur_index],
617 nr * commit_element_size_()) |
618 ex::then([&](int32_t r) {
619 if (r != nr * commit_element_size_()) {
620 throw std::runtime_error(
621 "commit_io_cmds returned unexpected byte "
625 co_await std::move(s);
626 co_await ex::schedule(sched);
627 }
else if (running_workers == 0) {
630 co_await state.futex.wait(0);
636 std::atomic<bool> flag =
false;
637 condy::Futex<bool> futex{flag};
640 template <
typename Sched>
641 ex::task<void, TaskEnv<Sched, Alloc>>
642 worker_(WorkerState &state, FlusherState &flusher,
bool &stopped,
643 uint16_t tag,
const ublksrv_io_desc *iod,
void *buf) {
645 if (!stopped && !read_once_(state.flag)) {
646 co_await state.futex.wait(
false);
651 assert(read_once_(state.flag));
652 write_once_(state.flag,
false);
654 bool io_buf = need_io_buf(flags_, iod->op_flags);
661 const IoData io_data{q_id_, tag, iod, buf};
662 co_await wrap_handle_io(handler_.handle_io(io_data), res, zone_lba);
668 auto cur_index = flusher.index;
669 auto slot = read_once_(flusher.pending);
670 set_result_(cur_index, slot, tag, res);
671 if (ublksrv_get_op(iod) == UBLK_IO_OP_ZONE_APPEND) {
672 set_zone_lba_(cur_index, slot, zone_lba);
674 write_once_(flusher.pending, slot + 1);
676 flusher.futex.notify_one();
685 uint16_t queue_depth_;
686 uint32_t max_io_buf_bytes_;
689 ublksrv_io_desc *iod_base_ =
nullptr;
690 size_t commit_buf_size_ = 0;
691 void *commit_buf_[2] = {};
692 size_t fetch_buf_size_ = 0;
693 uint16_t *fetch_buf_[2] = {};
694 AllocVector<void *, Alloc> buf_bases_;
Handler concepts for ublk server.
auto need_get_data(Fd fd, uint16_t q_id, uint16_t tag, uint64_t buf_addr) noexcept
Copy the data of a write request into the user buffer after the driver returns UBLK_IO_RES_NEED_GET_D...
auto commit_and_fetch_req(Fd fd, uint16_t q_id, uint16_t tag, int32_t result, uint64_t buf_addr, uint64_t sqe_addr) noexcept
Overload of commit_and_fetch_req() with an explicit sqe->addr, used for automatic request buffer regi...
auto fetch_io_cmds(Fd fd, condy::ProvidedBufferQueue &buffers, uint16_t q_id, MultiShotFunc &&func) noexcept
Fetch I/O commands in multishot style into a provided buffer queue (UBLK_F_BATCH_IO).
auto commit_io_cmds(Fd fd, uint16_t q_id, uint16_t flags, uint16_t nr_elem, uint8_t elem_bytes, const void *elems, uint32_t elems_len) noexcept
Commit a batch of completed I/O commands for a queue (UBLK_F_BATCH_IO).
auto unregister_io_buf(Fd fd, uint16_t q_id, uint16_t tag, uint64_t buf_index) noexcept
Unregister the buffer of an in-flight request from the io_uring buffer table after zero-copy I/O comp...
auto register_io_buf(Fd fd, uint16_t q_id, uint16_t tag, uint64_t buf_index) noexcept
Register the buffer of an in-flight request into the io_uring buffer table for zero-copy I/O (UBLK_F_...
auto prep_io_cmds(Fd fd, uint16_t q_id, uint16_t flags, uint16_t nr_elem, uint8_t elem_bytes, const void *elems, uint32_t elems_len) noexcept
Prepare a batch of I/O commands for a queue (UBLK_F_BATCH_IO).
auto fetch_req(Fd fd, uint16_t q_id, uint16_t tag, uint64_t buf_addr, uint64_t sqe_addr) noexcept
Overload of fetch_req() with an explicit sqe->addr, used for automatic request buffer registration (U...
The main namespace of the ublk-cpp library.
Low-level io_uring command senders for ublk control and I/O commands.