Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 0 additions & 12 deletions include/condy/cqe_handler.hpp
Original file line number Diff line number Diff line change
Expand Up @@ -57,10 +57,6 @@ class SelectBufferCQEHandler {
*/
struct NVMePassthruCQEHandler {
std::pair<int32_t, uint64_t> operator()(io_uring_cqe *cqe) noexcept {
assert(
detail::Context::current().runtime()->ring_internal().check_cqe32(
cqe) &&
"Expected big CQE for NVMe passthrough");
return {cqe->res, cqe->big_cqe[0]};
}
};
Expand Down Expand Up @@ -98,10 +94,6 @@ struct SCSIBsgResult {
*/
struct SCSIBsgPassthruCQEHandler {
std::pair<int32_t, SCSIBsgResult> operator()(io_uring_cqe *cqe) noexcept {
assert(
detail::Context::current().runtime()->ring_internal().check_cqe32(
cqe) &&
"Expected big CQE for SCSI BSG passthrough");
return {cqe->res, SCSIBsgResult{cqe->big_cqe[0]}};
}
};
Expand Down Expand Up @@ -138,10 +130,6 @@ struct TxTimestampResult {
struct TxTimestampCQEHandler {
std::pair<int32_t, TxTimestampResult>
operator()(io_uring_cqe *cqe) noexcept {
assert(
detail::Context::current().runtime()->ring_internal().check_cqe32(
cqe) &&
"Expected big CQE for TX timestamp operations");
TxTimestampResult result;
result.tstype =
static_cast<int>(cqe->flags >> IORING_TIMESTAMP_TYPE_SHIFT);
Expand Down
5 changes: 5 additions & 0 deletions include/condy/detail/finish_handles.hpp
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,11 @@ class OpFinishHandle : public OpFinishHandleBase {
CONDY_DELETE_COPY_MOVE(OpFinishHandle);

public:
bool stop_requested() noexcept {
auto stop_token = receiver_.get_stop_token();
return stop_token.stop_requested();
}

void maybe_set_cancel(Runtime *runtime) noexcept {
auto stop_token = receiver_.get_stop_token();
if (stop_token.stop_possible()) {
Expand Down
28 changes: 21 additions & 7 deletions include/condy/detail/op_states.hpp
Original file line number Diff line number Diff line change
Expand Up @@ -31,22 +31,36 @@ template <typename Handle, PrepFuncLike Func> class OpSenderOperationState {

public:
void start(unsigned int flags) noexcept {
auto &context = Context::current();
auto &ring = context.runtime()->ring_internal();
auto *runtime = Context::current().runtime();
if (runtime == nullptr) {
fail_(-EINVAL);
return;
}
if (finish_handle_.get().stop_requested()) {
fail_(-ECANCELED);
return;
}
auto &ring = runtime->ring_internal();
io_uring_sqe *sqe = prep_func_(&ring);
if (sqe == nullptr) {
io_uring_cqe cqe = {};
cqe.res = -EINVAL;
finish_handle_.get().handle(&cqe);
fail_(-EINVAL);
return;
}
context.runtime()->pend_work_internal();
runtime->pend_work_internal();
io_uring_sqe_set_flags(sqe, sqe->flags | flags);
auto work = encode_work(&finish_handle_.get(), WorkType::Common);
io_uring_sqe_set_data64(sqe, work);
ring.maybe_submit();

finish_handle_.get().maybe_set_cancel(context.runtime());
finish_handle_.get().maybe_set_cancel(runtime);
}

private:
void fail_(int32_t res) noexcept {
assert(res < 0);
io_uring_cqe cqe[2] = {};
cqe[0].res = res;
finish_handle_.get().handle(cqe);
}

private:
Expand Down
13 changes: 0 additions & 13 deletions include/condy/detail/ring.hpp
Original file line number Diff line number Diff line change
Expand Up @@ -133,19 +133,6 @@ class Ring {
}
#endif

bool check_cqe32([[maybe_unused]] io_uring_cqe *cqe) const noexcept {
auto ring_flags = ring_.flags;
if (ring_flags & IORING_SETUP_CQE32) {
return true;
}
#if CONDY_URING_VERSION_GE(2, 13) // >= 2.13
if (ring_flags & IORING_SETUP_CQE_MIXED) {
return cqe->flags & IORING_CQE_F_32;
}
#endif
return false;
}

private:
template <io_uring_sqe *(*get_sqe)(struct io_uring *)>
io_uring_sqe *get_sqe_() noexcept {
Expand Down
2 changes: 0 additions & 2 deletions include/condy/zcrx.hpp
Original file line number Diff line number Diff line change
Expand Up @@ -203,8 +203,6 @@ class ZeroCopyRxBufferPool {
uint32_t zcrx_id() const noexcept { return zcrx_id_; }

ZeroCopyRxBuffer handle_finish(io_uring_cqe *cqe) noexcept {
assert(ring_->check_cqe32(cqe) && "Expected big CQE for ZeroCopyRx");

if (cqe->res < 0) {
return ZeroCopyRxBuffer();
}
Expand Down
47 changes: 46 additions & 1 deletion tests/test_senders.cpp
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
#include "condy/async_operations.hpp"
#include "condy/awaiter_operations.hpp"
#include "condy/channel.hpp"
#include "condy/detail/async_operations.hpp"
Expand All @@ -6,7 +7,9 @@
#include "condy/sync_wait.hpp"
#include <atomic>
#include <cassert>
#include <cerrno>
#include <doctest.h>
#include <optional>
#include <stdexcept>
#include <stop_token>
#include <thread>
Expand Down Expand Up @@ -557,4 +560,46 @@ TEST_CASE("test senders - cancel from other runtime thread") {
condy::sync_wait(cancel_task());

t1.join();
}
}

TEST_CASE("test senders - start without runtime") {
auto coro = []() -> condy::Coro<int> {
int r = co_await condy::async_nop();
REQUIRE(r == -EINVAL);
co_return r;
}();
auto handle = coro.release();

REQUIRE(!handle.done());
handle.resume();
REQUIRE(handle.done());
REQUIRE(handle.promise().value() == -EINVAL);

handle.destroy();
}

#if CONDY_URING_VERSION_GE(2, 13) // >= 2.13
TEST_CASE("test senders - start without sqe128 support") {
auto func = []() -> condy::Coro<void> {
int r = co_await condy::async_nop128();
REQUIRE(r == -EINVAL);
};
condy::sync_wait(func());
}
#endif

TEST_CASE("test senders - start with stopped token") {
using condy::operators::operator||;

auto func = []() -> condy::Coro<void> {
condy::Channel<int> ch(1);
REQUIRE(ch.try_push(42) == 0);

auto res = co_await (ch.pop() || condy::async_nop());
REQUIRE(res.index() == 0);
auto [r, item] = std::get<0>(res);
REQUIRE(r == 0);
REQUIRE(item == 42);
};
condy::sync_wait(func());
}
Loading