complete ops inline on start failure (#190)

* remove internal check_cqe32()

* check if runtime exist

* check if already canceled

* add tests

Author: wokron

include/condy/cqe_handler.hpp

@@ -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]};
     }
 };
@@ -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]}};
     }
 };
@@ -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);

include/condy/detail/finish_handles.hpp

@@ -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()) {

include/condy/detail/op_states.hpp

@@ -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:

include/condy/detail/ring.hpp

@@ -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 {

include/condy/zcrx.hpp

@@ -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();
         }

tests/test_senders.cpp

@@ -1,3 +1,4 @@
+#include "condy/async_operations.hpp"
 #include "condy/awaiter_operations.hpp"
 #include "condy/channel.hpp"
 #include "condy/detail/async_operations.hpp"
@@ -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>
@@ -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());
+}