ublk-cpp v0.0
Loading...
Searching...
No Matches
queue.hpp
Go to the documentation of this file.
1
5
6#pragma once
7
10#include "ublk/handler.hpp"
11#include "ublk/raw.hpp"
12#include "ublk/ublk_cmd.h"
13#include <atomic>
14#include <cerrno>
15#include <condy.hpp>
16#include <cstdint>
17#include <exception>
18#include <stdexcept>
19#include <sys/mman.h>
20#include <system_error>
21#include <unistd.h>
22#include <utility>
23
24namespace ublk {
25namespace detail {
26
27namespace ex = condy::detail::ex;
28
29inline bool need_alloc_buf(uint64_t flags) noexcept {
30 return !(flags & (UBLK_F_SUPPORT_ZERO_COPY | UBLK_F_USER_COPY));
31}
32
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);
38}
39
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);
43}
44
45template <typename T>
46concept has_res = requires(T t) {
47 { t.res } -> std::convertible_to<int32_t>;
48};
49
50template <typename T>
51concept has_zone_lba = requires(T t) {
52 { t.zone_lba } -> std::convertible_to<uint64_t>;
53};
54
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>) {
59 res = r;
60 zone_lba = 0;
61 } else {
62 static_assert(has_res<R>,
63 "returned result is neither int32_t nor has a "
64 "res field");
65 res = r.res;
66 zone_lba = 0;
67 if constexpr (has_zone_lba<R>) {
68 zone_lba = r.zone_lba;
69 }
70 }
71}
72
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>) {
76 return err;
77 } else if constexpr (std::same_as<E, std::exception_ptr>) {
78 try {
79 std::rethrow_exception(err);
80 } catch (const std::system_error &se) {
81 return se.code();
82 } catch (...) {
83 return std::error_code(v, std::generic_category());
84 }
85 } else {
86 return std::error_code(v, std::generic_category());
87 }
88}
89
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) |
94 ex::stopped_as_error(
95 std::error_code(ECANCELED, std::generic_category())) |
96 ex::upon_error([](const auto &err) noexcept {
97 return normalize_error(err, EIO);
98 }) |
99 ex::then([&](auto r) noexcept {
100 using R = decltype(r);
101 if constexpr (std::same_as<R, std::error_code>) {
102 res = -r.value();
103 zone_lba = 0;
104 } else {
105 extract_io_result(r, res, zone_lba);
106 }
107 return std::make_tuple(res, zone_lba);
108 });
109}
110
111template <typename Fd, IoHandler Handler, typename Alloc> class IoQueue {
112public:
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,
115 const Alloc &alloc)
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) {
119 bool ok = false;
120 auto d = defer([&] noexcept {
121 if (!ok) {
122 cleanup_();
123 }
124 });
125
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,
129 ublkc_fd_raw, off);
130 if (addr == MAP_FAILED) {
131 throw std::system_error(errno, std::generic_category(),
132 "mmap io_desc");
133 }
134 iod_base_ = reinterpret_cast<ublksrv_io_desc *>(addr);
135
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);
142 }
143 }
144
145 ok = true;
146 }
147
148 IoQueue(const IoQueue &) = delete;
149 IoQueue &operator=(const IoQueue &) = delete;
150 IoQueue(IoQueue &&) = delete;
151 IoQueue &operator=(IoQueue &&) = delete;
152
153 ~IoQueue() { cleanup_(); }
154
155public:
156 template <typename Sched> ex::task<void, TaskEnv<Sched, Alloc>> run() {
157 if constexpr (QueueHandler<Handler>) {
158 co_await handler_.init_queue(q_id_);
159 }
160
161 auto d = defer([&]() noexcept {
162 if constexpr (QueueHandler<Handler>) {
163 handler_.destroy_queue(q_id_);
164 }
165 });
166
167 ex::simple_counting_scope scope;
168 AllocVector<std::exception_ptr, decltype(alloc_)> errs(queue_depth_,
169 alloc_);
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));
173 auto s =
174 ex::starts_on(sched, std::move(task)) |
175 ex::upon_error([&, tag](const std::exception_ptr &ep) noexcept {
176 errs[tag] = ep;
177 });
178 ex::spawn(std::move(s), scope.get_token());
179 }
180
181 co_await scope.join();
182
183 for (auto &ep : errs) {
184 if (ep) {
185 std::rethrow_exception(ep);
186 }
187 }
188 }
189
190private:
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);
196 }
197 buf_bases_.clear();
198 }
199 if (iod_base_) {
200 size_t size = queue_depth_ * sizeof(ublksrv_io_desc);
201 munmap(iod_base_, size);
202 iod_base_ = nullptr;
203 }
204 }
205
206 void *get_buf_(uint16_t tag) const noexcept {
207 if (buf_bases_.empty()) {
208 return nullptr;
209 }
210 return buf_bases_[tag];
211 }
212
213 template <typename Sched>
214 ex::task<void, TaskEnv<Sched, Alloc>>
215 worker_(uint16_t tag, const ublksrv_io_desc *iod, void *buf) {
216 int32_t r;
217 bool zero_copy = flags_ & UBLK_F_SUPPORT_ZERO_COPY;
218 bool auto_reg = flags_ & UBLK_F_AUTO_BUF_REG;
219
220 uint64_t buf_addr = reinterpret_cast<uint64_t>(buf);
221 uint64_t sqe_addr = 0;
222 if (auto_reg) {
223 ublk_auto_buf_reg reg = {};
224 reg.index = tag;
225 reg.flags = UBLK_AUTO_BUF_REG_FALLBACK;
226 sqe_addr = ublk_auto_buf_reg_to_sqe_addr(&reg);
227 }
228
229 r = co_await (
230 raw::fetch_req(ublkc_fd_, q_id_, tag, buf_addr, sqe_addr) |
231 ex::upon_error(
232 [&](std::error_code ec) noexcept { return -ec.value(); }));
233 if (r == UBLK_IO_RES_NEED_GET_DATA) {
234 r = co_await (raw::need_get_data(ublkc_fd_, q_id_, tag, buf_addr) |
235 ex::upon_error([&](std::error_code ec) noexcept {
236 return -ec.value();
237 }));
238 }
239 if (r == UBLK_IO_RES_ABORT) {
240 co_return;
241 } else if (r < 0) {
242 throw std::system_error(-r, std::generic_category(), "fetch_req");
243 }
244
245 while (true) {
246 bool io_buf = need_io_buf(flags_, iod->op_flags);
247 if (io_buf) {
248 co_await raw::register_io_buf(ublkc_fd_, q_id_, tag, tag);
249 }
250
251 int32_t res;
252 uint64_t zone_lba;
253 const IoData io_data{q_id_, tag, iod, buf};
254 co_await wrap_handle_io(handler_.handle_io(io_data), res, zone_lba);
255
256 if (io_buf) {
257 co_await raw::unregister_io_buf(ublkc_fd_, q_id_, tag, tag);
258 }
259
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;
264 }
265 r = co_await (raw::commit_and_fetch_req(ublkc_fd_, q_id_, tag, res,
266 curr_buf_addr, sqe_addr) |
267 ex::upon_error([&](std::error_code ec) noexcept {
268 return -ec.value();
269 }));
270 if (r == UBLK_IO_RES_NEED_GET_DATA) {
271 r = co_await (
272 raw::need_get_data(ublkc_fd_, q_id_, tag, buf_addr) |
273 ex::upon_error([&](std::error_code ec) noexcept {
274 return -ec.value();
275 }));
276 }
277 if (r == UBLK_IO_RES_ABORT) {
278 co_return;
279 } else if (r < 0) {
280 throw std::system_error(-r, std::generic_category(),
281 "commit_and_fetch_req");
282 }
283 }
284 }
285
286private:
287 Fd ublkc_fd_;
288 uint16_t q_id_;
289 uint64_t flags_;
290 uint16_t queue_depth_;
291 uint32_t max_io_buf_bytes_;
292 Handler &handler_;
293 Alloc alloc_;
294 ublksrv_io_desc *iod_base_ = nullptr;
295 AllocVector<void *, Alloc> buf_bases_;
296};
297
298template <typename Fd, IoHandler Handler, typename Alloc> class BatchIoQueue {
299public:
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) {
306 bool ok = false;
307 auto d = defer([&] noexcept {
308 if (!ok) {
309 cleanup_();
310 }
311 });
312
313 {
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,
317 ublkc_fd_raw, off);
318 if (addr == MAP_FAILED) {
319 throw std::system_error(errno, std::generic_category(),
320 "mmap io_desc");
321 }
322 iod_base_ = reinterpret_cast<ublksrv_io_desc *>(addr);
323 }
324
325 const size_t page = sysconf(_SC_PAGESIZE);
326 {
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(),
333 "mlock commit_buf");
334 }
335 }
336
337 {
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(),
346 "mlock fetch_buf");
347 }
348 }
349
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);
355 }
356 }
357
358 ok = true;
359 }
360
361 BatchIoQueue(const BatchIoQueue &) = delete;
362 BatchIoQueue &operator=(const BatchIoQueue &) = delete;
363 BatchIoQueue(BatchIoQueue &&) = delete;
364 BatchIoQueue &operator=(BatchIoQueue &&) = delete;
365
366 ~BatchIoQueue() { cleanup_(); }
367
368public:
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);
372
373 if constexpr (QueueHandler<Handler>) {
374 co_await handler_.init_queue(q_id_);
375 }
376
377 auto d = defer([&]() noexcept {
378 if constexpr (QueueHandler<Handler>) {
379 handler_.destroy_queue(q_id_);
380 }
381 });
382
383 for (uint16_t tag = 0; tag < queue_depth_; tag++) {
384 set_result_(0, tag, tag, 0);
385 }
386 co_await raw::prep_io_cmds(ublkc_fd_, q_id_, get_batch_flags_(),
387 queue_depth_, commit_element_size_(),
388 commit_buf_[0],
389 queue_depth_ * commit_element_size_());
390
391 FlusherState flusher;
392 AllocVector<WorkerState, decltype(alloc)> workers(queue_depth_, alloc);
393
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_;
400
401 {
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);
406 });
407 ex::spawn(std::move(s), scope.get_token());
408 }
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));
412 auto s =
413 ex::starts_on(sched, std::move(task)) |
414 ex::upon_error([&, tag](const std::exception_ptr &ep) noexcept {
415 worker_errs[tag] = ep;
416 }) |
417 ex::then([&] noexcept {
418 if (--running_workers == 0) {
419 flusher.futex.notify_one();
420 }
421 });
422 ex::spawn(std::move(s), scope.get_token());
423 }
424
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_));
428
429 size_t off = 0;
430 // NOLINTNEXTLINE(bugprone-exception-escape)
431 auto fetch_cb = [&](std::pair<int32_t, condy::BufferInfo> r) noexcept {
432 auto &[res, info] = r;
433 auto bid = info.bid;
434 auto *fetch_buf = fetch_buf_[bid];
435 assert(res >= 0);
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();
443 }
444 off = end;
445 bool consumed = info.num_buffers;
446 if (consumed) {
447 auto r = queue.push(condy::buffer(fetch_buf, fetch_buf_size_));
448 assert(r == bid);
449 off = 0;
450 }
451 };
452
453 int32_t res = 0;
454
455 auto s =
456 raw::fetch_io_cmds(ublkc_fd_, queue, q_id_, fetch_cb) |
457 ex::then([](int32_t r, condy::BufferInfo) noexcept { return r; }) |
458 ex::upon_error(
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();
465 }
466 });
467
468 co_await ex::when_all(std::move(s), scope.join());
469
470 for (auto &ep : worker_errs) {
471 if (ep) {
472 std::rethrow_exception(ep);
473 }
474 }
475 if (flusher_err) {
476 std::rethrow_exception(flusher_err);
477 }
478 assert(res < 0);
479 if (res != UBLK_IO_RES_ABORT) {
480 throw std::system_error(-res, std::generic_category(),
481 "fetch_io_cmds");
482 }
483 }
484
485private:
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);
491 }
492 buf_bases_.clear();
493 }
494 if (fetch_buf_[0]) {
495 munlock(fetch_buf_[0], fetch_buf_size_ * 2);
496 free_aligned(alloc_, fetch_buf_[0], fetch_buf_size_ * 2, page);
497 }
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);
501 }
502 if (iod_base_) {
503 size_t size = queue_depth_ * sizeof(ublksrv_io_desc);
504 munmap(iod_base_, size);
505 iod_base_ = nullptr;
506 }
507 }
508
509 template <typename T>
510 static auto read_once_(const std::atomic<T> &a) noexcept {
511 return a.load(std::memory_order_relaxed);
512 }
513
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);
517 }
518
519 bool need_f_buf_addr_() noexcept {
520 return !(flags_ & UBLK_F_AUTO_BUF_REG) && need_alloc_buf(flags_);
521 }
522
523 bool need_f_zone_lba_() noexcept { return flags_ & UBLK_F_ZONED; }
524
525 uint16_t get_batch_flags_() noexcept {
526 uint16_t f = 0;
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;
531 }
532 if (need_f_zone_lba_()) {
533 f |= UBLK_BATCH_F_HAS_ZONE_LBA;
534 }
535 return f;
536 }
537
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);
542 }
543 if (need_f_zone_lba_()) {
544 size += sizeof(uint64_t);
545 }
546 return size;
547 }
548
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_());
553 }
554
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));
560 }
561
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);
567 }
568 return *reinterpret_cast<uint64_t *>(
569 reinterpret_cast<char *>(&commit_f_hdr(index, pos)) + off);
570 }
571
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);
575 hdr.tag = tag;
576 hdr.result = res;
577 if (flags_ & UBLK_F_AUTO_BUF_REG) {
578 hdr.buf_index = tag;
579 } else if (need_f_buf_addr_()) {
580 commit_f_buf_addr(index, pos) =
581 reinterpret_cast<uint64_t>(buf_bases_[tag]);
582 }
583 }
584
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;
588 }
589
590 void *get_buf_(uint16_t tag) const noexcept {
591 if (buf_bases_.empty()) {
592 return nullptr;
593 }
594 return buf_bases_[tag];
595 }
596
597private:
598 struct FlusherState {
599 size_t index = 0;
600 std::atomic<uint16_t> pending = 0;
601 condy::Futex<uint16_t> futex{pending};
602 };
603
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);
608 while (true) {
609 auto nr = read_once_(state.pending);
610 if (nr > 0) {
611 auto cur_index = state.index;
612 state.index = 1 - state.index;
613 write_once_(state.pending, 0);
614 auto s = raw::commit_io_cmds(
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 "
622 "count");
623 }
624 });
625 co_await std::move(s);
626 co_await ex::schedule(sched);
627 } else if (running_workers == 0) {
628 co_return;
629 } else {
630 co_await state.futex.wait(0);
631 }
632 }
633 }
634
635 struct WorkerState {
636 std::atomic<bool> flag = false;
637 condy::Futex<bool> futex{flag};
638 };
639
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) {
644 while (true) {
645 if (!stopped && !read_once_(state.flag)) {
646 co_await state.futex.wait(false);
647 }
648 if (stopped) {
649 co_return;
650 }
651 assert(read_once_(state.flag));
652 write_once_(state.flag, false);
653
654 bool io_buf = need_io_buf(flags_, iod->op_flags);
655 if (io_buf) {
656 co_await raw::register_io_buf(ublkc_fd_, q_id_, tag, tag);
657 }
658
659 int32_t res;
660 uint64_t zone_lba;
661 const IoData io_data{q_id_, tag, iod, buf};
662 co_await wrap_handle_io(handler_.handle_io(io_data), res, zone_lba);
663
664 if (io_buf) {
665 co_await raw::unregister_io_buf(ublkc_fd_, q_id_, tag, tag);
666 }
667
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);
673 }
674 write_once_(flusher.pending, slot + 1);
675 if (slot == 0) {
676 flusher.futex.notify_one();
677 }
678 }
679 }
680
681private:
682 Fd ublkc_fd_;
683 uint16_t q_id_;
684 uint64_t flags_;
685 uint16_t queue_depth_;
686 uint32_t max_io_buf_bytes_;
687 Handler &handler_;
688 Alloc alloc_;
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_;
695};
696
697} // namespace detail
698} // namespace ublk
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...
Definition raw.hpp:481
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...
Definition raw.hpp:447
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).
Definition raw.hpp:575
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).
Definition raw.hpp:555
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...
Definition raw.hpp:515
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_...
Definition raw.hpp:498
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).
Definition raw.hpp:533
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...
Definition raw.hpp:422
The main namespace of the ublk-cpp library.
Definition ublk.hpp:21
Low-level io_uring command senders for ublk control and I/O commands.
Task environment utils.
Small utility helpers.