From 2758bb9538cd1b9489444af3ef5ef3ca8c20e3fc Mon Sep 17 00:00:00 2001 From: mrabine Date: Thu, 17 Sep 2026 23:10:26 +0200 Subject: [PATCH] Harmonise async acceptor api --- core/include/join/async_acceptor.hpp | 127 ++++++++++++++++++------ core/include/join/async_operation.hpp | 17 +++- core/include/join/async_socket.hpp | 31 ++++++ core/include/join/proactor.hpp | 36 ------- core/tests/tcp_async_acceptor_test.cpp | 39 +++++++- core/tests/tcp_async_socket_test.cpp | 23 +++++ core/tests/unix_async_acceptor_test.cpp | 39 +++++++- 7 files changed, 235 insertions(+), 77 deletions(-) diff --git a/core/include/join/async_acceptor.hpp b/core/include/join/async_acceptor.hpp index 1f1866a1..df623ca2 100644 --- a/core/include/join/async_acceptor.hpp +++ b/core/include/join/async_acceptor.hpp @@ -36,9 +36,6 @@ #include #include -// C. -#include - namespace join { /** @@ -117,7 +114,7 @@ namespace join do { - cancelAccept (); + cancel (); backoff (); } @@ -130,9 +127,11 @@ namespace join * @brief start an asynchronous acceptation. * @param handler handler invoked on completion. * @param flags accepted socket creation flags. + * @param flush flush the submission queue. * @return 0 on success, -1 on failure. */ - int asyncAccept (AcceptHandler handler, int flags = SOCK_NONBLOCK | SOCK_CLOEXEC) noexcept + ssize_t asyncAccept (AcceptHandler handler, int flags = SOCK_NONBLOCK | SOCK_CLOEXEC, + bool flush = true) noexcept { if (JOIN_UNLIKELY (!_acceptor.opened ())) { @@ -140,7 +139,7 @@ namespace join return -1; } - if (JOIN_UNLIKELY (!arm (_acceptOp.op))) + if (JOIN_UNLIKELY (!armOp ())) { lastError = make_error_code (Errc::InUse); return -1; @@ -151,10 +150,11 @@ namespace join _acceptOp.op = IoOperation::makeAccept (_acceptor.handle (), _acceptOp.remote.addr (), &_acceptOp.remoteLen, flags | SOCK_NONBLOCK, this); - if (_proactor->submit (_acceptOp.op, true, false) == -1) + if (_proactor->submit (_acceptOp.op, flush, false) == -1) { // LCOV_EXCL_START _acceptOp.acceptHandler.reset (); + disarmOp (IoOperation::State::Submitted); return -1; // LCOV_EXCL_STOP } @@ -166,9 +166,11 @@ namespace join * @brief start an asynchronous multishot acceptation, staying armed until cancelled or failed. * @param handler handler invoked on each acceptation, the last call reporting more as false. * @param flags accepted socket creation flags. + * @param flush flush the submission queue. * @return 0 on success, -1 on failure. */ - int asyncAcceptMulti (AcceptHandler handler, int flags = SOCK_NONBLOCK | SOCK_CLOEXEC) noexcept + ssize_t asyncAcceptMulti (AcceptHandler handler, int flags = SOCK_NONBLOCK | SOCK_CLOEXEC, + bool flush = true) noexcept { if (JOIN_UNLIKELY (!_acceptor.opened ())) { @@ -176,7 +178,7 @@ namespace join return -1; } - if (JOIN_UNLIKELY (!arm (_acceptOp.op))) + if (JOIN_UNLIKELY (!armOp ())) { lastError = make_error_code (Errc::InUse); return -1; @@ -186,10 +188,11 @@ namespace join _acceptOp.acceptHandler = std::move (handler); _acceptOp.op = IoOperation::makeAcceptMulti (_acceptor.handle (), flags | SOCK_NONBLOCK, this); - if (_proactor->submit (_acceptOp.op, true, false) == -1) + if (_proactor->submit (_acceptOp.op, flush, false) == -1) { // LCOV_EXCL_START _acceptOp.acceptHandler.reset (); + disarmOp (IoOperation::State::Submitted); return -1; // LCOV_EXCL_STOP } @@ -201,7 +204,7 @@ namespace join * @brief cancel the acceptation in flight, if any. * @return 0 on success, -1 on failure. */ - int cancelAccept () noexcept + int cancel () noexcept { if (!inFlight (_acceptOp.op)) { @@ -216,6 +219,17 @@ namespace join return 0; } +#ifdef JOIN_HAS_IO_URING + /** + * @brief flush the pending submissions of the proactor driving this acceptor. + * @return 0 on success, -1 on failure. + */ + int flush () noexcept + { + return _proactor->flush (false); + } +#endif + /** * @brief determine the local endpoint associated with this acceptor. * @return local endpoint. @@ -272,57 +286,114 @@ namespace join protected: /** - * @brief method called when the acceptation completes. + * @brief method called when an operation completes. * @param op completed operation. * @param result accepted file descriptor, or negative errno. */ - void onComplete ([[maybe_unused]] IoOperation& op, int result) override + void onComplete (IoOperation& op, int result) override { - completeAccept (result); + std::error_code code = + (result < 0) ? std::error_code (-result, std::generic_category ()) : std::error_code (); + + dispatch (op, code, (result > 0) ? static_cast (result) : 0); } /** - * @brief method called when the acceptation is cancelled. + * @brief method called when an operation is cancelled. * @param op cancelled operation. * @param result negative errno. */ - void onCancel ([[maybe_unused]] IoOperation& op, [[maybe_unused]] int result) override + void onCancel (IoOperation& op, [[maybe_unused]] int result) override { - completeAccept (-ECANCELED); + dispatch (op, make_error_code (std::errc::operation_canceled), 0); } /** - * @brief invoke the completion handler. - * @param result accepted file descriptor, or negative errno. + * @brief invoke the completion handler of the given operation. + * @param op completed operation. + * @param code error code reported by the kernel. + * @param size accepted file descriptor, 0 when the operation failed or was cancelled. */ - void completeAccept (int result) noexcept + virtual void dispatch (IoOperation& op, const std::error_code& code, size_t size) noexcept { - Socket sock = (result < 0) ? Socket () : Socket (result, _acceptOp.remote); - std::error_code code = - (result < 0) ? std::error_code (-result, std::generic_category ()) : std::error_code (); + completeAccept (reinterpret_cast (&op), code, size); + } - if (_acceptOp.op.more) + /** + * @brief invoke the accept completion handler. + * @param accept completed accept operation. + * @param code error code reported by the kernel. + * @param handle accepted file descriptor. + */ + void completeAccept (AsyncAccept* accept, const std::error_code& code, size_t handle) noexcept + { + Socket sock = code ? Socket () : Socket (static_cast (handle), accept->remote); + + if (accept->op.more) { - if (JOIN_LIKELY (_acceptOp.acceptHandler)) + if (JOIN_LIKELY (accept->acceptHandler)) { - _acceptOp.acceptHandler (std::move (sock), code, true); + accept->acceptHandler (std::move (sock), code, true); } return; } - AcceptHandler handler = std::move (_acceptOp.acceptHandler); + AcceptHandler handler = std::move (accept->acceptHandler); if (JOIN_LIKELY (handler)) { handler (std::move (sock), code, false); } - IoOperation::State expected = IoOperation::State::Busy; + disarmOp (IoOperation::State::Busy); + } + + /** + * @brief reserve the accept operation for submission. + * @return true if the operation was reserved, false if already in flight. + */ + bool armOp () noexcept + { + IoOperation::State expected = IoOperation::State::Idle; + + return _acceptOp.op.state.compare_exchange_strong (expected, IoOperation::State::Submitted, + std::memory_order_acquire, std::memory_order_relaxed) || + (expected == IoOperation::State::Busy); + } + + /** + * @brief release the accept operation reservation. + * @param expected state the operation is expected to be in. + */ + void disarmOp (IoOperation::State expected) noexcept + { _acceptOp.op.state.compare_exchange_strong (expected, IoOperation::State::Idle, std::memory_order_release, std::memory_order_relaxed); } + /** + * @brief check if an operation is in flight. + * @param op operation to check. + * @return true if the operation is in flight, false otherwise. + */ + bool inFlight (const IoOperation& op) const noexcept + { + return op.state.load (std::memory_order_acquire) == IoOperation::State::Submitted; + } + + /** + * @brief check if an operation is in flight or completing. + * @param op operation to check. + * @return true if the operation is in flight or completing, false otherwise. + */ + bool pending (const IoOperation& op) const noexcept + { + IoOperation::State state = op.state.load (std::memory_order_acquire); + + return (state == IoOperation::State::Submitted) || (state == IoOperation::State::Busy); + } + private: /// proactor driving the operations. Proactor* _proactor; diff --git a/core/include/join/async_operation.hpp b/core/include/join/async_operation.hpp index b25e02ff..ac1a07be 100644 --- a/core/include/join/async_operation.hpp +++ b/core/include/join/async_operation.hpp @@ -142,14 +142,21 @@ namespace join /** * @brief asynchronous operation traits. */ - template - struct AsyncOp + template + struct BasicAsyncOpTraits { + static_assert (sizeof...(Ops) > 0, "traits must describe at least one operation"); + /// size of the largest asynchronous operation. - static constexpr size_t maxSize = - std::max ({sizeof (BasicAsyncWait), sizeof (BasicAsyncRead), - sizeof (BasicAsyncWrite)}); + static constexpr size_t maxSize = std::max ({sizeof (Ops)...}); }; + + /** + * @brief asynchronous socket operation traits. + */ + template + using AsyncOp = BasicAsyncOpTraits, BasicAsyncRead, + BasicAsyncWrite>; } #endif diff --git a/core/include/join/async_socket.hpp b/core/include/join/async_socket.hpp index 8fd60515..20fdd876 100644 --- a/core/include/join/async_socket.hpp +++ b/core/include/join/async_socket.hpp @@ -315,6 +315,15 @@ namespace join } #ifdef JOIN_HAS_IO_URING + /** + * @brief flush the pending submissions of the proactor driving this socket. + * @return 0 on success, -1 on failure. + */ + int flush () noexcept + { + return _proactor->flush (false); + } + /** * @brief register the arena chunks as fixed buffers on the proactor driving this socket. * @param arena arena to register. @@ -591,6 +600,28 @@ namespace join return 0; } + /** + * @brief check if an operation is in flight. + * @param op operation to check. + * @return true if the operation is in flight, false otherwise. + */ + bool inFlight (const IoOperation& op) const noexcept + { + return op.state.load (std::memory_order_acquire) == IoOperation::State::Submitted; + } + + /** + * @brief check if an operation is in flight or completing. + * @param op operation to check. + * @return true if the operation is in flight or completing, false otherwise. + */ + bool pending (const IoOperation& op) const noexcept + { + IoOperation::State state = op.state.load (std::memory_order_acquire); + + return (state == IoOperation::State::Submitted) || (state == IoOperation::State::Busy); + } + /// proactor driving the operations. Proactor* _proactor; diff --git a/core/include/join/proactor.hpp b/core/include/join/proactor.hpp index 2c75c5bc..83e24955 100644 --- a/core/include/join/proactor.hpp +++ b/core/include/join/proactor.hpp @@ -142,42 +142,6 @@ class join::CompletionHandler { // do nothing. } - - /** - * @brief arm an operation for submission. - * @param op operation to arm. - * @return true if the operation was armed, false if already in flight. - */ - bool arm (IoOperation& op) noexcept - { - IoOperation::State expected = IoOperation::State::Idle; - - return op.state.compare_exchange_strong (expected, IoOperation::State::Submitted, std::memory_order_acquire, - std::memory_order_relaxed) || - (expected == IoOperation::State::Busy); - } - - /** - * @brief check if an operation is in flight. - * @param op operation to check. - * @return true if the operation is in flight, false otherwise. - */ - bool inFlight (const IoOperation& op) const noexcept - { - return op.state.load (std::memory_order_acquire) == IoOperation::State::Submitted; - } - - /** - * @brief check if an operation is in flight or completing. - * @param op operation to check. - * @return true if the operation is in flight or completing, false otherwise. - */ - bool pending (const IoOperation& op) const noexcept - { - IoOperation::State state = op.state.load (std::memory_order_acquire); - - return (state == IoOperation::State::Submitted) || (state == IoOperation::State::Busy); - } }; /** diff --git a/core/tests/tcp_async_acceptor_test.cpp b/core/tests/tcp_async_acceptor_test.cpp index 8f27a192..e9697783 100644 --- a/core/tests/tcp_async_acceptor_test.cpp +++ b/core/tests/tcp_async_acceptor_test.cpp @@ -385,18 +385,18 @@ TEST_F (TcpAsyncAcceptor, closeFromHandler) } /** - * @brief Test cancelAccept method. + * @brief Test cancel method. */ -TEST_F (TcpAsyncAcceptor, cancelAccept) +TEST_F (TcpAsyncAcceptor, cancel) { Tcp::AsyncAcceptor server; - ASSERT_EQ (server.cancelAccept (), 0) << join::lastError.message (); + ASSERT_EQ (server.cancel (), 0) << join::lastError.message (); ASSERT_EQ (server.create ({_address, _port}), 0) << join::lastError.message (); ASSERT_EQ (server.asyncAccept (onReport), 0) << join::lastError.message (); - ASSERT_EQ (server.cancelAccept (), 0) << join::lastError.message (); + ASSERT_EQ (server.cancel (), 0) << join::lastError.message (); { ScopedLock lock (_mut); @@ -489,6 +489,37 @@ TEST_F (TcpAsyncAcceptor, handle) ASSERT_EQ (server.handle (), -1); } +#ifdef JOIN_HAS_IO_URING +/** + * @brief Test flush method. + */ +TEST_F (TcpAsyncAcceptor, flush) +{ + Tcp::AsyncAcceptor server; + Tcp::Socket client (Tcp::Socket::Blocking); + + ASSERT_EQ (server.create ({_address, _port}), 0) << join::lastError.message (); + + ASSERT_EQ (server.asyncAccept (onReport, SOCK_NONBLOCK | SOCK_CLOEXEC, false), 0) << join::lastError.message (); + ASSERT_EQ (server.flush (), 0) << join::lastError.message (); + + ASSERT_EQ (client.connect ({_address, _port}), 0) << join::lastError.message (); + + { + ScopedLock lock (_mut); + ASSERT_TRUE (_cond.timedWait (lock, std::chrono::milliseconds (_timeout), [] () { + return _completions >= 1; + })); + ASSERT_FALSE (_code) << _code.message (); + } + + ASSERT_TRUE (peer ().connected ()); + + client.close (); + server.close (); +} +#endif + /** * @brief main function. */ diff --git a/core/tests/tcp_async_socket_test.cpp b/core/tests/tcp_async_socket_test.cpp index 8d9d5e3e..25bab0d4 100644 --- a/core/tests/tcp_async_socket_test.cpp +++ b/core/tests/tcp_async_socket_test.cpp @@ -1628,6 +1628,29 @@ TEST_F (TcpAsyncSocket, handle) } #ifdef JOIN_HAS_IO_URING +/** + * @brief Test flush method. + */ +TEST_F (TcpAsyncSocket, flush) +{ + Tcp::AsyncSocket client; + + ASSERT_EQ (client.open (), 0) << join::lastError.message (); + + ASSERT_NE (client.asyncWait (true, false, onReportWait, false), -1) << join::lastError.message (); + ASSERT_EQ (client.flush (), 0) << join::lastError.message (); + + { + ScopedLock lock (_mut); + ASSERT_TRUE (_cond.timedWait (lock, std::chrono::milliseconds (_timeout), [] () { + return _completions >= 1; + })); + ASSERT_EQ (_code, Errc::ConnectionClosed); + } + + client.close (); +} + /** * @brief Test registerFixedBuffers method. */ diff --git a/core/tests/unix_async_acceptor_test.cpp b/core/tests/unix_async_acceptor_test.cpp index 6182caab..e187a8bd 100644 --- a/core/tests/unix_async_acceptor_test.cpp +++ b/core/tests/unix_async_acceptor_test.cpp @@ -388,18 +388,18 @@ TEST_F (UnixAsyncAcceptor, closeFromHandler) } /** - * @brief Test cancelAccept method. + * @brief Test cancel method. */ -TEST_F (UnixAsyncAcceptor, cancelAccept) +TEST_F (UnixAsyncAcceptor, cancel) { UnixStream::AsyncAcceptor server; - ASSERT_EQ (server.cancelAccept (), 0) << join::lastError.message (); + ASSERT_EQ (server.cancel (), 0) << join::lastError.message (); ASSERT_EQ (server.create (_path), 0) << join::lastError.message (); ASSERT_EQ (server.asyncAccept (onReport), 0) << join::lastError.message (); - ASSERT_EQ (server.cancelAccept (), 0) << join::lastError.message (); + ASSERT_EQ (server.cancel (), 0) << join::lastError.message (); { ScopedLock lock (_mut); @@ -490,6 +490,37 @@ TEST_F (UnixAsyncAcceptor, handle) ASSERT_EQ (server.handle (), -1); } +#ifdef JOIN_HAS_IO_URING +/** + * @brief Test flush method. + */ +TEST_F (UnixAsyncAcceptor, flush) +{ + UnixStream::AsyncAcceptor server; + UnixStream::Socket client (UnixStream::Socket::Blocking); + + ASSERT_EQ (server.create (_path), 0) << join::lastError.message (); + + ASSERT_EQ (server.asyncAccept (onReport, SOCK_NONBLOCK | SOCK_CLOEXEC, false), 0) << join::lastError.message (); + ASSERT_EQ (server.flush (), 0) << join::lastError.message (); + + ASSERT_EQ (client.connect (_path), 0) << join::lastError.message (); + + { + ScopedLock lock (_mut); + ASSERT_TRUE (_cond.timedWait (lock, std::chrono::milliseconds (_timeout), [] () { + return _completions >= 1; + })); + ASSERT_FALSE (_code) << _code.message (); + } + + ASSERT_TRUE (peer ().connected ()); + + client.close (); + server.close (); +} +#endif + /** * @brief main function. */