diff --git a/core/CMakeLists.txt b/core/CMakeLists.txt index 8956f2d6..d6f8ebae 100644 --- a/core/CMakeLists.txt +++ b/core/CMakeLists.txt @@ -38,6 +38,7 @@ set(PUBLIC_HEADERS include/join/stream_socket.hpp include/join/socket_stream.hpp include/join/acceptor.hpp + include/join/async_socket.hpp include/join/async_stream_socket.hpp include/join/async_acceptor.hpp include/join/cpu.hpp diff --git a/core/include/join/async_acceptor.hpp b/core/include/join/async_acceptor.hpp index de7880da..2fdcdeaa 100644 --- a/core/include/join/async_acceptor.hpp +++ b/core/include/join/async_acceptor.hpp @@ -92,7 +92,10 @@ namespace join , _ops (std::move (other._ops)) , _onAccept (std::move (other._onAccept)) { - _ops->acceptOp.handler = this; + if (_ops) + { + _ops->accept.op.handler = this; + } } /** @@ -102,17 +105,17 @@ namespace join */ BasicAsyncStreamAcceptor& operator= (BasicAsyncStreamAcceptor&& other) noexcept { - if (_ops) - { - close (); - } + close (); _acceptor = std::move (other._acceptor); _engine = other._engine; _ops = std::move (other._ops); _onAccept = std::move (other._onAccept); - _ops->acceptOp.handler = this; + if (_ops) + { + _ops->accept.op.handler = this; + } return *this; } @@ -122,10 +125,7 @@ namespace join */ ~BasicAsyncStreamAcceptor () { - if (_ops) - { - close (); - } + close (); } /** @@ -146,11 +146,11 @@ namespace join { cancelAccept (); - if (!_engine->isProactorThread ()) + if (_ops && !_engine->isProactorThread ()) { Backoff backoff; - while (_ops->acceptState.load (std::memory_order_acquire) != State::Idle) + while (_ops->accept.state.load (std::memory_order_acquire) != AsyncOperation::Idle) { backoff (); } @@ -167,35 +167,35 @@ namespace join */ int asyncAccept (AcceptHandler handler, int flags = SOCK_NONBLOCK | SOCK_CLOEXEC) noexcept { - if (!_acceptor.opened ()) + if (JOIN_UNLIKELY (!_acceptor.opened ())) { lastError = make_error_code (Errc::OperationFailed); return -1; } - State expected = State::Idle; + AsyncOperation::State expected = AsyncOperation::Idle; - if (!_ops->acceptState.compare_exchange_strong (expected, State::Pending, std::memory_order_acq_rel, - std::memory_order_acquire)) + if (!_ops->accept.state.compare_exchange_strong (expected, AsyncOperation::Pending, + std::memory_order_acquire, std::memory_order_acquire)) { - if ((expected != State::Dispatching) || !_engine->isProactorThread ()) + if ((expected != AsyncOperation::Dispatching) || !_engine->isProactorThread ()) { lastError = make_error_code (Errc::InUse); return -1; } - _ops->acceptState.store (State::Pending, std::memory_order_release); + _ops->accept.state.store (AsyncOperation::Pending, std::memory_order_release); } _ops->peerLen = sizeof (struct sockaddr_storage); _onAccept = std::move (handler); - _ops->acceptOp = IoOperation::makeAccept (_acceptor.handle (), _ops->peer.addr (), &_ops->peerLen, - flags | SOCK_NONBLOCK, this); + _ops->accept.op = IoOperation::makeAccept (_acceptor.handle (), _ops->peer.addr (), &_ops->peerLen, + flags | SOCK_NONBLOCK, this); - if (_engine->submit (&_ops->acceptOp, true, false) == -1) + if (_engine->submit (&_ops->accept.op, true, false) == -1) { // LCOV_EXCL_START - _ops->acceptState.store (State::Idle, std::memory_order_release); + _ops->accept.state.store (AsyncOperation::Idle, std::memory_order_release); _onAccept.reset (); return -1; // LCOV_EXCL_STOP @@ -210,12 +210,12 @@ namespace join */ int cancelAccept () noexcept { - if (_ops->acceptState.load (std::memory_order_acquire) == State::Idle) + if (!_ops || (_ops->accept.state.load (std::memory_order_acquire) == AsyncOperation::Idle)) { return 0; } - if (_engine->cancel (&_ops->acceptOp, true, true) == -1) + if (_engine->cancel (&_ops->accept.op, true, true) == -1) { return (lastError == Errc::OperationFailed) ? 0 : -1; } @@ -277,35 +277,7 @@ namespace join return _acceptor.handle (); } - /** - * @brief get the underlying synchronous acceptor. - * @return the underlying synchronous acceptor. - */ - Acceptor& acceptor () noexcept - { - return _acceptor; - } - - /** - * @brief get the underlying synchronous acceptor. - * @return the underlying synchronous acceptor. - */ - const Acceptor& acceptor () const noexcept - { - return _acceptor; - } - private: - /** - * @brief acceptation slot state. - */ - enum class State : uint8_t - { - Idle, /**< no acceptation in flight and no completion handler running. */ - Pending, /**< an acceptation is in flight. */ - Dispatching, /**< the completion handler is running. */ - }; - /** * @brief operation block, kept at a stable address across moves. */ @@ -338,10 +310,7 @@ namespace join } /// acceptation operation slot. - alignas (64) IoOperation acceptOp = {}; - - /// acceptation slot state. - alignas (64) std::atomic acceptState{State::Idle}; + AsyncOperation accept; /// peer endpoint, written by the kernel until the acceptation completes. Endpoint peer; @@ -368,11 +337,11 @@ namespace join { Ops* ops = _ops.get (); - ops->acceptState.store (State::Dispatching, std::memory_order_release); + ops->accept.state.store (AsyncOperation::Dispatching, std::memory_order_release); AcceptHandler handler = std::move (_onAccept); - if (!handler) + if (JOIN_UNLIKELY (!handler)) { if (result > -1) { @@ -388,9 +357,9 @@ namespace join handler (std::error_code (), AsyncSocket (Socket (result, ops->peer), *_engine)); } - State expected = State::Dispatching; - ops->acceptState.compare_exchange_strong (expected, State::Idle, std::memory_order_acq_rel, - std::memory_order_acquire); + AsyncOperation::State expected = AsyncOperation::Dispatching; + ops->accept.state.compare_exchange_strong (expected, AsyncOperation::Idle, std::memory_order_release, + std::memory_order_relaxed); } /** diff --git a/core/include/join/async_socket.hpp b/core/include/join/async_socket.hpp new file mode 100644 index 00000000..7bfb2d25 --- /dev/null +++ b/core/include/join/async_socket.hpp @@ -0,0 +1,590 @@ +/** + * MIT License + * + * Copyright (c) 2026 Mathieu Rabine + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +#ifndef JOIN_CORE_ASYNC_SOCKET_HPP +#define JOIN_CORE_ASYNC_SOCKET_HPP + +// libjoin. +#include +#include +#include +#include +#include +#include + +// C++. +#include +#include +#include +#include +#include +#include + +// C. +#include +#include +#include + +namespace join +{ + /** + * @brief asynchronous operation. + */ + struct AsyncOperation + { + /** + * @brief caller side operation state. + */ + enum State : uint8_t + { + Idle, /**< no operation in flight and no completion handler running. */ + Pending, /**< an operation is in flight. */ + Dispatching, /**< the completion handler is running. */ + }; + + /// operation. + IoOperation op = {}; + + /// caller side operation state. + alignas (64) std::atomic state{Idle}; + }; + + /** + * @brief basic asynchronous socket class. + */ + template + class BasicAsyncSocket : protected CompletionHandler + { + public: + using Socket = typename Protocol::Socket; + using Endpoint = typename Protocol::Endpoint; + using Option = typename Socket::Option; + + /// handler invoked on read completion. + using ReadHandler = Function; + + /// handler invoked on write completion. + using WriteHandler = Function; + + /** + * @brief create the socket instance. + * @param engine engine driving the operations. + */ + explicit BasicAsyncSocket (Engine& engine = ProactorThread::proactor ()) + : _engine (&engine) + { + } + + /** + * @brief create the socket instance adopting an already opened socket. + * @param sock socket to adopt. + * @param engine engine driving the operations. + */ + explicit BasicAsyncSocket (Socket&& sock, Engine& engine = ProactorThread::proactor ()) + : _socket (std::move (sock)) + , _engine (&engine) + { + } + + /** + * @brief copy constructor. + * @param other other object to copy. + */ + BasicAsyncSocket (const BasicAsyncSocket& other) = delete; + + /** + * @brief copy assignment operator. + * @param other other object to assign. + * @return assigned object. + */ + BasicAsyncSocket& operator= (const BasicAsyncSocket& other) = delete; + + /** + * @brief move constructor. + * @param other other object to move. + */ + BasicAsyncSocket (BasicAsyncSocket&& other) noexcept + : _socket (std::move (other._socket)) + , _engine (other._engine) + , _ops (std::move (other._ops)) + , _onRead (std::move (other._onRead)) + , _onWrite (std::move (other._onWrite)) + { + if (_ops) + { + _ops->read.op.handler = this; + _ops->write.op.handler = this; + } + } + + /** + * @brief move assignment operator. + * @param other other object to assign. + * @return assigned object. + */ + BasicAsyncSocket& operator= (BasicAsyncSocket&& other) noexcept + { + close (); + + _socket = std::move (other._socket); + _engine = other._engine; + _ops = std::move (other._ops); + _onRead = std::move (other._onRead); + _onWrite = std::move (other._onWrite); + + if (_ops) + { + _ops->read.op.handler = this; + _ops->write.op.handler = this; + } + + return *this; + } + + /** + * @brief destroy the socket instance. + */ + ~BasicAsyncSocket () + { + close (); + } + + /** + * @brief open socket using the given protocol. + * @param protocol protocol to use. + * @return 0 on success, -1 on failure. + */ + int open (const Protocol& protocol = Protocol ()) noexcept + { + return _socket.open (protocol); + } + + /** + * @brief close the socket, cancelling the operations in flight. + */ + void close () noexcept + { + cancelRead (); + cancelWrite (); + + if (_ops && !_engine->isProactorThread ()) + { + Backoff backoff; + + while ((_ops->read.state.load (std::memory_order_acquire) != AsyncOperation::Idle) || + (_ops->write.state.load (std::memory_order_acquire) != AsyncOperation::Idle)) + { + backoff (); + } + } + + _socket.close (); + } + + /** + * @brief start an asynchronous read. + * @param data buffer used to store the data received, valid until the handler is invoked. + * @param maxSize maximum number of bytes to read. + * @param handler handler invoked on completion. + * @return 0 on success, -1 on failure. + */ + int asyncRead (char* data, size_t maxSize, ReadHandler handler) noexcept + { + if (JOIN_UNLIKELY (!_ops || !_socket.opened ())) + { + lastError = make_error_code (Errc::OperationFailed); + return -1; + } + + AsyncOperation::State expected = AsyncOperation::Idle; + + if (!_ops->read.state.compare_exchange_strong (expected, AsyncOperation::Pending, std::memory_order_acquire, + std::memory_order_acquire)) + { + if ((expected != AsyncOperation::Dispatching) || !_engine->isProactorThread ()) + { + lastError = make_error_code (Errc::InUse); + return -1; + } + + _ops->read.state.store (AsyncOperation::Pending, std::memory_order_release); + } + + _onRead = std::move (handler); + + _ops->readIov.iov_base = data; + _ops->readIov.iov_len = maxSize; + + _ops->readMsg.msg_name = nullptr; + _ops->readMsg.msg_namelen = 0; + _ops->readMsg.msg_iov = &_ops->readIov; + _ops->readMsg.msg_iovlen = 1; + _ops->readMsg.msg_control = nullptr; + _ops->readMsg.msg_controllen = 0; + _ops->readMsg.msg_flags = 0; + + _ops->read.op = IoOperation::makeRecvmsg (_socket.handle (), &_ops->readMsg, 0, this); + + if (_engine->submit (&_ops->read.op, true, false) == -1) + { + // LCOV_EXCL_START + _ops->read.state.store (AsyncOperation::Idle, std::memory_order_release); + _onRead.reset (); + return -1; + // LCOV_EXCL_STOP + } + + return 0; + } + + /** + * @brief start an asynchronous write. + * @param data data buffer to send, valid until the handler is invoked. + * @param size number of bytes to write. + * @param handler handler invoked on completion. + * @return 0 on success, -1 on failure. + */ + int asyncWrite (const char* data, size_t size, WriteHandler handler) noexcept + { + if (JOIN_UNLIKELY (!_ops || !_socket.opened ())) + { + lastError = make_error_code (Errc::OperationFailed); + return -1; + } + + AsyncOperation::State expected = AsyncOperation::Idle; + + if (!_ops->write.state.compare_exchange_strong (expected, AsyncOperation::Pending, + std::memory_order_acquire, std::memory_order_acquire)) + { + if ((expected != AsyncOperation::Dispatching) || !_engine->isProactorThread ()) + { + lastError = make_error_code (Errc::InUse); + return -1; + } + + _ops->write.state.store (AsyncOperation::Pending, std::memory_order_release); + } + + _onWrite = std::move (handler); + + _ops->writeIov.iov_base = const_cast (data); + _ops->writeIov.iov_len = size; + + _ops->writeMsg.msg_name = nullptr; + _ops->writeMsg.msg_namelen = 0; + _ops->writeMsg.msg_iov = &_ops->writeIov; + _ops->writeMsg.msg_iovlen = 1; + _ops->writeMsg.msg_control = nullptr; + _ops->writeMsg.msg_controllen = 0; + _ops->writeMsg.msg_flags = 0; + + _ops->write.op = IoOperation::makeSendmsg (_socket.handle (), &_ops->writeMsg, MSG_NOSIGNAL, this); + + if (_engine->submit (&_ops->write.op, true, false) == -1) + { + // LCOV_EXCL_START + _ops->write.state.store (AsyncOperation::Idle, std::memory_order_release); + _onWrite.reset (); + return -1; + // LCOV_EXCL_STOP + } + + return 0; + } + + /** + * @brief cancel the read operation in flight, if any. + * @return 0 on success, -1 on failure. + */ + int cancelRead () noexcept + { + if (!_ops || (_ops->read.state.load (std::memory_order_acquire) == AsyncOperation::Idle)) + { + return 0; + } + + if (_engine->cancel (&_ops->read.op, true, true) == -1) + { + return (lastError == Errc::OperationFailed) ? 0 : -1; + } + + return 0; + } + + /** + * @brief cancel the connect or write operation in flight, if any. + * @return 0 on success, -1 on failure. + */ + int cancelWrite () noexcept + { + if (!_ops || (_ops->write.state.load (std::memory_order_acquire) == AsyncOperation::Idle)) + { + return 0; + } + + if (_engine->cancel (&_ops->write.op, true, true) == -1) + { + return (lastError == Errc::OperationFailed) ? 0 : -1; + } + + return 0; + } + + /** + * @brief assign the specified endpoint to the socket. + * @param endpoint endpoint to assign to the socket. + * @return 0 on success, -1 on failure. + */ + int bind (const Endpoint& endpoint) noexcept + { + return _socket.bind (endpoint); + } + + /** + * @brief assign the specified device to the socket. + * @param device device name. + * @return 0 on success, -1 on failure. + */ + int bindToDevice (const std::string& device) noexcept + { + return _socket.bindToDevice (device); + } + + /** + * @brief set the given option to the given value. + * @param option socket option. + * @param value option value. + * @return 0 on success, -1 on failure. + */ + int setOption (Option option, int value) noexcept + { + return _socket.setOption (option, value); + } + + /** + * @brief determine the local endpoint associated with this socket. + * @return local endpoint. + */ + Endpoint localEndpoint () const noexcept + { + return _socket.localEndpoint (); + } + + /** + * @brief check if the socket is opened. + * @return true if opened, false otherwise. + */ + bool opened () const noexcept + { + return _socket.opened (); + } + + /** + * @brief get the number of readable bytes. + * @return the number of readable bytes, -1 on failure. + */ + int canRead () const noexcept + { + return _socket.canRead (); + } + + /** + * @brief get address family. + * @return address family. + */ + int family () const noexcept + { + return _socket.family (); + } + + /** + * @brief get the protocol communication semantic. + * @return the protocol communication semantic. + */ + int type () const noexcept + { + return _socket.type (); + } + + /** + * @brief get socket protocol. + * @return socket protocol. + */ + int protocol () const noexcept + { + return _socket.protocol (); + } + + /** + * @brief get socket native handle. + * @return socket native handle. + */ + int handle () const noexcept + { + return _socket.handle (); + } + + protected: + /** + * @brief operation block, kept at a stable address across moves. + */ + struct Ops + { + /** + * @brief allocate a block honouring its extended alignment. + * @param size allocation size in bytes. + * @return pointer to the allocated storage. + */ + static void* operator new (size_t size) + { + void* mem = ::aligned_alloc (alignof (Ops), size); + + if (mem == nullptr) + { + throw std::bad_alloc (); // LCOV_EXCL_LINE + } + + return mem; + } + + /** + * @brief release storage allocated by operator new. + * @param mem storage to release. + */ + static void operator delete (void* mem) noexcept + { + ::free (mem); + } + + /// read operation slot. + AsyncOperation read; + + /// read message header, read by the kernel until the read completes. + msghdr readMsg = {}; + + /// read scatter gather entry, read by the kernel until the read completes. + iovec readIov = {}; + + /// connect or write operation slot. + AsyncOperation write; + + /// write message header, read by the kernel until the write completes. + msghdr writeMsg = {}; + + /// write scatter gather entry, read by the kernel until the write completes. + iovec writeIov = {}; + }; + + /// underlying synchronous socket. + Socket _socket; + + /// engine driving the operations. + Engine* _engine; + + /// operation block. + std::unique_ptr _ops{new Ops ()}; + + /// handler invoked on read completion. + ReadHandler _onRead; + + /// handler invoked on write completion. + WriteHandler _onWrite; + + /** + * @brief method called when an operation completes. + * @param op completed operation. + * @param result number of bytes transferred, or operation specific value. + */ + void onComplete (IoOperation* op, int result) override + { + dispatch (op, (result < 0) ? std::error_code (-result, std::generic_category ()) : std::error_code (), + (result > 0) ? static_cast (result) : 0); + } + + /** + * @brief method called when an operation is cancelled. + * @param op cancelled operation. + * @param result negative errno. + */ + void onCancel (IoOperation* op, [[maybe_unused]] int result) override + { + dispatch (op, make_error_code (std::errc::operation_canceled), 0); + } + + /** + * @brief invoke the handler owning the given operation slot. + * @param op completed or cancelled operation. + * @param code error code to report. + * @param size number of bytes transferred. + */ + void dispatch (IoOperation* op, const std::error_code& code, size_t size) noexcept + { + if (op == &_ops->read.op) + { + _ops->read.state.store (AsyncOperation::Dispatching, std::memory_order_release); + + ReadHandler handler = std::move (_onRead); + std::error_code result = code; + + if (JOIN_LIKELY (!result)) + { + if (JOIN_UNLIKELY (size == 0)) + { + result = make_error_code (Errc::ConnectionClosed); + } + else if (JOIN_UNLIKELY (_ops->readMsg.msg_flags & MSG_TRUNC)) + { + result = make_error_code (Errc::MessageTooLong); + } + } + + if (JOIN_LIKELY (handler)) + { + handler (result, size); + } + + AsyncOperation::State expected = AsyncOperation::Dispatching; + _ops->read.state.compare_exchange_strong (expected, AsyncOperation::Idle, std::memory_order_release, + std::memory_order_relaxed); + return; + } + + _ops->write.state.store (AsyncOperation::Dispatching, std::memory_order_release); + + WriteHandler handler = std::move (_onWrite); + + if (JOIN_LIKELY (handler)) + { + handler (code, size); + } + + AsyncOperation::State expected = AsyncOperation::Dispatching; + _ops->write.state.compare_exchange_strong (expected, AsyncOperation::Idle, std::memory_order_release, + std::memory_order_relaxed); + } + }; +} + +#endif diff --git a/core/include/join/async_stream_socket.hpp b/core/include/join/async_stream_socket.hpp index 8c260f61..4d99860f 100644 --- a/core/include/join/async_stream_socket.hpp +++ b/core/include/join/async_stream_socket.hpp @@ -27,21 +27,12 @@ // libjoin. #include -#include -#include -#include +#include // C++. #include #include -#include #include -#include - -// C. -#include -#include -#include namespace join { @@ -49,28 +40,27 @@ namespace join * @brief asynchronous stream socket class. */ template - class BasicAsyncStreamSocket : private CompletionHandler + class BasicAsyncStreamSocket : public BasicAsyncSocket { public: using Socket = BasicStreamSocket; using Endpoint = typename Protocol::Endpoint; - using Option = typename Socket::Option; /// handler invoked on connect completion. using ConnectHandler = Function; /// handler invoked on read completion. - using ReadHandler = Function; + using ReadHandler = typename BasicAsyncSocket::ReadHandler; /// handler invoked on write completion. - using WriteHandler = Function; + using WriteHandler = typename BasicAsyncSocket::WriteHandler; /** * @brief create the socket instance. * @param engine engine driving the operations. */ explicit BasicAsyncStreamSocket (Engine& engine = ProactorThread::proactor ()) - : _engine (&engine) + : BasicAsyncSocket (engine) { } @@ -80,8 +70,7 @@ namespace join * @param engine engine driving the operations. */ explicit BasicAsyncStreamSocket (Socket&& sock, Engine& engine = ProactorThread::proactor ()) - : _socket (std::move (sock)) - , _engine (&engine) + : BasicAsyncSocket (std::move (sock), engine) { } @@ -103,15 +92,10 @@ namespace join * @param other other object to move. */ BasicAsyncStreamSocket (BasicAsyncStreamSocket&& other) noexcept - : _socket (std::move (other._socket)) - , _engine (other._engine) - , _ops (std::move (other._ops)) + : BasicAsyncSocket (std::move (other)) + , _remote (std::move (other._remote)) , _onConnect (std::move (other._onConnect)) - , _onRead (std::move (other._onRead)) - , _onWrite (std::move (other._onWrite)) { - _ops->readOp.handler = this; - _ops->writeOp.handler = this; } /** @@ -121,20 +105,12 @@ namespace join */ BasicAsyncStreamSocket& operator= (BasicAsyncStreamSocket&& other) noexcept { - if (_ops) - { - close (); - } + this->close (); - _socket = std::move (other._socket); - _engine = other._engine; - _ops = std::move (other._ops); - _onConnect = std::move (other._onConnect); - _onRead = std::move (other._onRead); - _onWrite = std::move (other._onWrite); + BasicAsyncSocket::operator= (std::move (other)); - _ops->readOp.handler = this; - _ops->writeOp.handler = this; + _remote = std::move (other._remote); + _onConnect = std::move (other._onConnect); return *this; } @@ -144,62 +120,7 @@ namespace join */ ~BasicAsyncStreamSocket () { - if (_ops) - { - close (); - } - } - - /** - * @brief open socket using the given protocol. - * @param protocol protocol to use. - * @return 0 on success, -1 on failure. - */ - int open (const Protocol& protocol = Protocol ()) noexcept - { - return _socket.open (protocol); - } - - /** - * @brief close the socket, cancelling the operations in flight. - */ - void close () noexcept - { - cancelRead (); - cancelWrite (); - - if (!_engine->isProactorThread ()) - { - Backoff backoff; - - while ((_ops->readState.load (std::memory_order_acquire) != State::Idle) || - (_ops->writeState.load (std::memory_order_acquire) != State::Idle)) - { - backoff (); - } - } - - _socket.close (); - } - - /** - * @brief assign the specified endpoint to the socket. - * @param endpoint endpoint to assign to the socket. - * @return 0 on success, -1 on failure. - */ - int bind (const Endpoint& endpoint) noexcept - { - return _socket.bind (endpoint); - } - - /** - * @brief assign the specified device to the socket. - * @param device device name. - * @return 0 on success, -1 on failure. - */ - int bindToDevice (const std::string& device) noexcept - { - return _socket.bindToDevice (device); + this->close (); } /** @@ -210,127 +131,44 @@ namespace join */ int asyncConnect (const Endpoint& endpoint, ConnectHandler handler) noexcept { - if (!_socket.opened () && (_socket.open (endpoint.protocol ()) == -1)) - { - return -1; // LCOV_EXCL_LINE - } - - State expected = State::Idle; - - if (!_ops->writeState.compare_exchange_strong (expected, State::Pending, std::memory_order_acq_rel, - std::memory_order_acquire)) - { - if ((expected != State::Dispatching) || !_engine->isProactorThread ()) - { - lastError = make_error_code (Errc::InUse); - return -1; - } - - _ops->writeState.store (State::Pending, std::memory_order_release); - } - - _socket._state = Socket::Connecting; - _socket._remote = endpoint; - _onConnect = std::move (handler); - _ops->remote = endpoint; - _ops->writeOp = - IoOperation::makeConnect (_socket.handle (), _ops->remote.addr (), _ops->remote.length (), this); - - if (_engine->submit (&_ops->writeOp, true, false) == -1) - { - // LCOV_EXCL_START - _ops->writeState.store (State::Idle, std::memory_order_release); - _onConnect.reset (); - _socket.close (); - return -1; - // LCOV_EXCL_STOP - } - - return 0; - } - - /** - * @brief start an asynchronous read. - * @param data buffer used to store the data received, valid until the handler is invoked. - * @param maxSize maximum number of bytes to read. - * @param handler handler invoked on completion. - * @return 0 on success, -1 on failure. - */ - int asyncRead (char* data, size_t maxSize, ReadHandler handler) noexcept - { - if (!_socket.opened ()) + if (!this->_ops || !_remote) { lastError = make_error_code (Errc::OperationFailed); return -1; } - State expected = State::Idle; - - if (!_ops->readState.compare_exchange_strong (expected, State::Pending, std::memory_order_acq_rel, - std::memory_order_acquire)) - { - if ((expected != State::Dispatching) || !_engine->isProactorThread ()) - { - lastError = make_error_code (Errc::InUse); - return -1; - } - - _ops->readState.store (State::Pending, std::memory_order_release); - } - - _onRead = std::move (handler); - _ops->readOp = IoOperation::makeRecv (_socket.handle (), data, static_cast (maxSize), 0, this); - - if (_engine->submit (&_ops->readOp, true, false) == -1) - { - // LCOV_EXCL_START - _ops->readState.store (State::Idle, std::memory_order_release); - _onRead.reset (); - return -1; - // LCOV_EXCL_STOP - } - - return 0; - } - - /** - * @brief start an asynchronous write. - * @param data data buffer to send, valid until the handler is invoked. - * @param size number of bytes to write. - * @param handler handler invoked on completion. - * @return 0 on success, -1 on failure. - */ - int asyncWrite (const char* data, size_t size, WriteHandler handler) noexcept - { - if (!_socket.opened ()) + if (!this->_socket.opened () && (this->_socket.open (endpoint.protocol ()) == -1)) { - lastError = make_error_code (Errc::OperationFailed); - return -1; + return -1; // LCOV_EXCL_LINE } - State expected = State::Idle; + AsyncOperation::State expected = AsyncOperation::Idle; - if (!_ops->writeState.compare_exchange_strong (expected, State::Pending, std::memory_order_acq_rel, - std::memory_order_acquire)) + if (!this->_ops->write.state.compare_exchange_strong (expected, AsyncOperation::Pending, + std::memory_order_acquire, std::memory_order_acquire)) { - if ((expected != State::Dispatching) || !_engine->isProactorThread ()) + if ((expected != AsyncOperation::Dispatching) || !this->_engine->isProactorThread ()) { lastError = make_error_code (Errc::InUse); return -1; } - _ops->writeState.store (State::Pending, std::memory_order_release); + this->_ops->write.state.store (AsyncOperation::Pending, std::memory_order_release); } - _onWrite = std::move (handler); - _ops->writeOp = - IoOperation::makeSend (_socket.handle (), data, static_cast (size), MSG_NOSIGNAL, this); + this->_socket._state = Socket::Connecting; + this->_socket._remote = endpoint; + _onConnect = std::move (handler); + *_remote = endpoint; + this->_ops->write.op = + IoOperation::makeConnect (this->_socket.handle (), _remote->addr (), _remote->length (), this); - if (_engine->submit (&_ops->writeOp, true, false) == -1) + if (this->_engine->submit (&this->_ops->write.op, true, false) == -1) { // LCOV_EXCL_START - _ops->writeState.store (State::Idle, std::memory_order_release); - _onWrite.reset (); + this->_ops->write.state.store (AsyncOperation::Idle, std::memory_order_release); + _onConnect.reset (); + this->_socket.close (); return -1; // LCOV_EXCL_STOP } @@ -338,80 +176,13 @@ namespace join return 0; } - /** - * @brief cancel the read operation in flight, if any. - * @return 0 on success, -1 on failure. - */ - int cancelRead () noexcept - { - if (_ops->readState.load (std::memory_order_acquire) == State::Idle) - { - return 0; - } - - if (_engine->cancel (&_ops->readOp, true, true) == -1) - { - return (lastError == Errc::OperationFailed) ? 0 : -1; - } - - return 0; - } - - /** - * @brief cancel the connect or write operation in flight, if any. - * @return 0 on success, -1 on failure. - */ - int cancelWrite () noexcept - { - if (_ops->writeState.load (std::memory_order_acquire) == State::Idle) - { - return 0; - } - - if (_engine->cancel (&_ops->writeOp, true, true) == -1) - { - return (lastError == Errc::OperationFailed) ? 0 : -1; - } - - return 0; - } - - /** - * @brief set the given option to the given value. - * @param option socket option. - * @param value option value. - * @return 0 on success, -1 on failure. - */ - int setOption (Option option, int value) noexcept - { - return _socket.setOption (option, value); - } - - /** - * @brief determine the local endpoint associated with this socket. - * @return local endpoint. - */ - Endpoint localEndpoint () const noexcept - { - return _socket.localEndpoint (); - } - /** * @brief determine the remote endpoint associated with this socket. * @return remote endpoint. */ const Endpoint& remoteEndpoint () const noexcept { - return _socket.remoteEndpoint (); - } - - /** - * @brief check if the socket is opened. - * @return true if opened, false otherwise. - */ - bool opened () const noexcept - { - return _socket.opened (); + return this->_socket.remoteEndpoint (); } /** @@ -420,7 +191,7 @@ namespace join */ bool connected () noexcept { - return _socket.connected (); + return this->_socket.connected (); } /** @@ -429,16 +200,7 @@ namespace join */ bool connecting () const noexcept { - return _socket.connecting (); - } - - /** - * @brief get the number of readable bytes. - * @return the number of readable bytes, -1 on failure. - */ - int canRead () const noexcept - { - return _socket.canRead (); + return this->_socket.connecting (); } /** @@ -447,121 +209,10 @@ namespace join */ int mtu () const noexcept { - return _socket.mtu (); - } - - /** - * @brief get address family. - * @return address family. - */ - int family () const noexcept - { - return _socket.family (); - } - - /** - * @brief get the protocol communication semantic. - * @return the protocol communication semantic. - */ - int type () const noexcept - { - return _socket.type (); - } - - /** - * @brief get socket protocol. - * @return socket protocol. - */ - int protocol () const noexcept - { - return _socket.protocol (); - } - - /** - * @brief get socket native handle. - * @return socket native handle. - */ - int handle () const noexcept - { - return _socket.handle (); - } - - /** - * @brief get the underlying synchronous socket. - * @return the underlying synchronous socket. - */ - Socket& socket () noexcept - { - return _socket; - } - - /** - * @brief get the underlying synchronous socket. - * @return the underlying synchronous socket. - */ - const Socket& socket () const noexcept - { - return _socket; + return this->_socket.mtu (); } - private: - /** - * @brief operation slot state. - */ - enum class State : uint8_t - { - Idle, /**< no operation in flight and no completion handler running. */ - Pending, /**< an operation is in flight. */ - Dispatching, /**< the completion handler is running. */ - }; - - /** - * @brief operation block, kept at a stable address across moves. - */ - struct Ops - { - /** - * @brief allocate a block honouring its extended alignment. - * @param size allocation size in bytes. - * @return pointer to the allocated storage. - */ - static void* operator new (size_t size) - { - void* mem = ::aligned_alloc (alignof (Ops), size); - - if (mem == nullptr) - { - throw std::bad_alloc (); // LCOV_EXCL_LINE - } - - return mem; - } - - /** - * @brief release storage allocated by operator new. - * @param mem storage to release. - */ - static void operator delete (void* mem) noexcept - { - ::free (mem); - } - - /// read operation slot. - alignas (64) IoOperation readOp = {}; - - /// connect or write operation slot. - alignas (64) IoOperation writeOp = {}; - - /// read slot state. - alignas (64) std::atomic readState{State::Idle}; - - /// connect or write slot state. - alignas (64) std::atomic writeState{State::Idle}; - - /// remote endpoint, read by the kernel until the connect completes. - Endpoint remote; - }; - + protected: /** * @brief method called when an operation completes. * @param op completed operation. @@ -591,69 +242,69 @@ namespace join */ void dispatch (IoOperation* op, const std::error_code& code, size_t size) noexcept { - bool isRead = (op == &_ops->readOp); - std::atomic& state = isRead ? _ops->readState : _ops->writeState; + if (op == &this->_ops->read.op) + { + this->_ops->read.state.store (AsyncOperation::Dispatching, std::memory_order_release); - state.store (State::Dispatching, std::memory_order_release); + ReadHandler handler = std::move (this->_onRead); + std::error_code result = code; - if (isRead) - { - ReadHandler handler = std::move (_onRead); + if (JOIN_UNLIKELY (!result && (size == 0))) + { + result = make_error_code (Errc::ConnectionClosed); + } - if (handler) + if (JOIN_LIKELY (handler)) { - handler ((!code && (size == 0)) ? make_error_code (Errc::ConnectionClosed) : code, size); + handler (result, size); } + + AsyncOperation::State expected = AsyncOperation::Dispatching; + this->_ops->read.state.compare_exchange_strong (expected, AsyncOperation::Idle, + std::memory_order_release, std::memory_order_relaxed); + return; } - else if (op->code == static_cast (IoOperation::Opcode::Connect)) + + this->_ops->write.state.store (AsyncOperation::Dispatching, std::memory_order_release); + + if (JOIN_UNLIKELY (op->code == static_cast (IoOperation::Opcode::Connect))) { ConnectHandler handler = std::move (_onConnect); if (code) { - _socket.close (); + this->_socket.close (); } else { - _socket._state = Socket::Connected; + this->_socket._state = Socket::Connected; } - if (handler) + if (JOIN_LIKELY (handler)) { handler (code); } } else { - WriteHandler handler = std::move (_onWrite); + WriteHandler handler = std::move (this->_onWrite); - if (handler) + if (JOIN_LIKELY (handler)) { handler (code, size); } } - State expected = State::Dispatching; - state.compare_exchange_strong (expected, State::Idle, std::memory_order_acq_rel, std::memory_order_acquire); + AsyncOperation::State expected = AsyncOperation::Dispatching; + this->_ops->write.state.compare_exchange_strong (expected, AsyncOperation::Idle, std::memory_order_release, + std::memory_order_relaxed); } - /// underlying synchronous socket. - Socket _socket; - - /// engine driving the operations. - Engine* _engine; - - /// operation block. - std::unique_ptr _ops{new Ops ()}; + /// remote endpoint. + std::unique_ptr _remote{new Endpoint ()}; /// handler invoked on connect completion. ConnectHandler _onConnect; - - /// handler invoked on read completion. - ReadHandler _onRead; - - /// handler invoked on write completion. - WriteHandler _onWrite; }; } diff --git a/core/include/join/function.hpp b/core/include/join/function.hpp index dfd40150..37ce8724 100644 --- a/core/include/join/function.hpp +++ b/core/include/join/function.hpp @@ -33,6 +33,7 @@ // C. #include +#include namespace join { @@ -127,15 +128,28 @@ namespace join return static_cast ((*static_cast (storage)) (std::forward (args)...)); }; - _manager = [] (void* dst, void* src) noexcept { - DecayedFunc* source = static_cast (src); - new (dst) DecayedFunc (std::move (*source)); - source->~DecayedFunc (); - }; + _manager = nullptr; + _destructor = nullptr; - _destructor = [] (void* storage) noexcept { - static_cast (storage)->~DecayedFunc (); - }; + if (!std::is_trivially_copyable::value) + { + _manager = [] (void* dst, void* src) noexcept { + DecayedFunc* source = static_cast (src); + new (dst) DecayedFunc (std::move (*source)); + source->~DecayedFunc (); + }; + } + + if (!std::is_trivially_destructible::value) + { + _destructor = [] (void* storage) noexcept { + static_cast (storage)->~DecayedFunc (); + }; + } + + // an empty target holds no state, so relocating it copies nothing at all and never + // reads the storage, which is left uninitialized by the placement new above. + _size = std::is_empty::value ? 0 : sizeof (DecayedFunc); } /** @@ -203,13 +217,20 @@ namespace join */ void clear () noexcept { + if (_invoker == nullptr) + { + return; + } + if (_destructor) { _destructor (&_storage); - _invoker = nullptr; - _manager = nullptr; - _destructor = nullptr; } + + _invoker = nullptr; + _manager = nullptr; + _destructor = nullptr; + _size = 0; } /** @@ -220,14 +241,26 @@ namespace join _invoker = other._invoker; _manager = other._manager; _destructor = other._destructor; + _size = other._size; + + if (_invoker == nullptr) + { + return; + } if (_manager) { _manager (&_storage, &other._storage); - other._invoker = nullptr; - other._manager = nullptr; - other._destructor = nullptr; } + else + { + std::memcpy (&_storage, &other._storage, _size); + } + + other._invoker = nullptr; + other._manager = nullptr; + other._destructor = nullptr; + other._size = 0; } /// fixed-capacity aligned storage for the callable target. @@ -241,6 +274,9 @@ namespace join /// pointer to the static destructor wrapper. DestructorFunc _destructor = nullptr; + + /// number of bytes to relocate when the target is trivially copyable. + std::size_t _size = 0; }; } diff --git a/core/include/join/proactor_epoll_impl.hpp b/core/include/join/proactor_epoll_impl.hpp index 6a379a97..26bac1e5 100644 --- a/core/include/join/proactor_epoll_impl.hpp +++ b/core/include/join/proactor_epoll_impl.hpp @@ -168,7 +168,7 @@ inline int join::BasicProactor::writeCommand (const Command& cmd) noexcept return -1; // LCOV_EXCL_LINE } - if (_notified.exchange (true)) + if (_notified.exchange (true, std::memory_order_acq_rel)) { return 0; } diff --git a/core/include/join/proactor_uring_impl.hpp b/core/include/join/proactor_uring_impl.hpp index cfc1abd0..b5383b63 100644 --- a/core/include/join/proactor_uring_impl.hpp +++ b/core/include/join/proactor_uring_impl.hpp @@ -380,7 +380,7 @@ int join::BasicProactor::writeCommand (const Command& cmd, std::true_typ return -1; // LCOV_EXCL_LINE } - if (_notified.exchange (true)) + if (_notified.exchange (true, std::memory_order_acq_rel)) { return 0; } diff --git a/core/include/join/protocol.hpp b/core/include/join/protocol.hpp index a3f54207..24978cdb 100644 --- a/core/include/join/protocol.hpp +++ b/core/include/join/protocol.hpp @@ -54,6 +54,9 @@ namespace join template class BasicProactor; + template > + class BasicAsyncSocket; + template > class BasicAsyncStreamSocket; @@ -62,6 +65,9 @@ namespace join #else class BasicProactor; + template + class BasicAsyncSocket; + template class BasicAsyncStreamSocket; @@ -165,6 +171,7 @@ namespace join public: using Endpoint = BasicLinkLayerEndpoint; using Socket = BasicSocket; + using AsyncSocket = BasicAsyncSocket; /** * @brief default constructor. diff --git a/core/tests/CMakeLists.txt b/core/tests/CMakeLists.txt index 5da7931e..3ce26d2b 100644 --- a/core/tests/CMakeLists.txt +++ b/core/tests/CMakeLists.txt @@ -271,6 +271,11 @@ target_link_libraries(tcp_acceptor.gtest ${JOIN_CORE} GTest::gtest_main) add_test(NAME tcp_acceptor.gtest COMMAND tcp_acceptor.gtest) install(TARGETS tcp_acceptor.gtest RUNTIME DESTINATION ${CMAKE_INSTALL_DATADIR}/${PROJECT_NAME}/test) +add_executable(raw_async_socket.gtest raw_async_socket_test.cpp) +target_link_libraries(raw_async_socket.gtest ${JOIN_CORE} GTest::gtest_main) +add_test(NAME raw_async_socket.gtest COMMAND raw_async_socket.gtest) +install(TARGETS raw_async_socket.gtest RUNTIME DESTINATION ${CMAKE_INSTALL_DATADIR}/${PROJECT_NAME}/test) + add_executable(tcp_async_stream_socket.gtest tcp_async_stream_socket_test.cpp) target_link_libraries(tcp_async_stream_socket.gtest ${JOIN_CORE} GTest::gtest_main) add_test(NAME tcp_async_stream_socket.gtest COMMAND tcp_async_stream_socket.gtest) diff --git a/core/tests/raw_async_socket_test.cpp b/core/tests/raw_async_socket_test.cpp new file mode 100644 index 00000000..063d73a3 --- /dev/null +++ b/core/tests/raw_async_socket_test.cpp @@ -0,0 +1,480 @@ +/** + * MIT License + * + * Copyright (c) 2026 Mathieu Rabine + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +// libjoin. +#include +#include + +// Libraries. +#include + +// C. +#include +#include +#include + +using join::Errc; +using join::Mutex; +using join::Condition; +using join::ScopedLock; +using join::IpAddress; +using join::MacAddress; +using join::Raw; + +/** + * @brief Class used to test the raw asynchronous socket API. + */ +class RawAsyncSocket : public ::testing::Test +{ +public: + /** + * @brief set up test case. + */ + static void SetUpTestCase () + { + // fill in data. + memcpy (_packet.data, "this is a test", strlen ("this is a test")); + + // fill in UDP header. + _packet.ip.protocol = IPPROTO_UDP; + _packet.ip.saddr = *reinterpret_cast (IpAddress ("127.0.0.1").addr ()); + _packet.ip.daddr = *reinterpret_cast (IpAddress ("127.0.0.1").addr ()); + _packet.udp.source = htons (5000); + _packet.udp.dest = htons (5000); + _packet.udp.len = htons (sizeof (Packet) - sizeof (_packet.eth) - sizeof (_packet.ip)); + _packet.ip.tot_len = _packet.udp.len; + _packet.udp.check = + Raw::Socket::checksum (reinterpret_cast (&_packet.ip), sizeof (Packet) - sizeof (_packet.eth)); + + // fill in IP header. + _packet.ip.ihl = sizeof (_packet.ip) >> 2; + _packet.ip.version = IPVERSION; + _packet.ip.tos = IPTOS_CLASS_CS6 | IPTOS_ECN_NOT_ECT; + _packet.ip.tot_len = htons (sizeof (Packet) - sizeof (_packet.eth)); + _packet.ip.id = htons (join::randomize ()); + _packet.ip.frag_off = htons (IP_DF); + _packet.ip.ttl = IPDEFTTL; + _packet.ip.check = Raw::Socket::checksum (reinterpret_cast (&_packet.ip), sizeof (_packet.ip)); + + // fill in ETH header. + memcpy (_packet.eth.h_dest, MacAddress::wildcard.addr (), 6); + memcpy (_packet.eth.h_source, MacAddress::wildcard.addr (), 6); + _packet.eth.h_proto = htons (ETH_P_IP); + } + +protected: + /** + * @brief Sets up the test fixture. + */ + void SetUp () override + { + ScopedLock lock (_mut); + + _code = {}; + _completions = 0; + _transferred = 0; + } + + /** + * @brief report a completion to the waiting test. + * @param ec error reported by the socket. + * @param size number of bytes transferred. + */ + static void onCompletion (const std::error_code& ec, size_t size) + { + ScopedLock lock (_mut); + + _code = ec; + _transferred = size; + ++_completions; + _cond.signal (); + } + + /** + * @brief wait for the expected number of completions. + * @param expected number of completions to wait for. + * @return true on success, false on timeout. + */ + static bool wait (int expected) + { + ScopedLock lock (_mut); + + return _cond.timedWait (lock, std::chrono::milliseconds (_timeout), [expected] () { + return _completions >= expected; + }); + } + + /** + * @brief Raw packet. + */ + struct __attribute__ ((packed)) Packet + { + struct ethhdr eth = {}; + struct iphdr ip = {}; + struct udphdr udp = {}; + char data[16] = {}; + }; + + /// mutex. + static Mutex _mut; + + /// condition variable. + static Condition _cond; + + /// error reported by the last completion. + static std::error_code _code; + + /// number of completions reported. + static int _completions; + + /// number of bytes reported by the last completion. + static size_t _transferred; + + /// packet. + static Packet _packet; + + /// read buffer. + static char _buf[2048]; + + /// interface. + static const std::string _interface; + + /// timeout. + static const int _timeout; +}; + +Mutex RawAsyncSocket::_mut; +Condition RawAsyncSocket::_cond; +std::error_code RawAsyncSocket::_code; +int RawAsyncSocket::_completions = 0; +size_t RawAsyncSocket::_transferred = 0; +RawAsyncSocket::Packet RawAsyncSocket::_packet; +char RawAsyncSocket::_buf[2048] = {}; +const std::string RawAsyncSocket::_interface = "lo"; +const int RawAsyncSocket::_timeout = 1000; + +/** + * @brief Test move with an operation in flight. + */ +TEST_F (RawAsyncSocket, move) +{ + Raw::AsyncSocket rawSocket; + + ASSERT_EQ (rawSocket.bind (_interface), 0) << join::lastError.message (); + ASSERT_EQ (rawSocket.asyncRead (_buf, sizeof (_buf), onCompletion), 0) << join::lastError.message (); + + ASSERT_EQ (rawSocket.asyncRead (_buf, sizeof (_buf), nullptr), -1); + ASSERT_EQ (join::lastError, Errc::InUse); + + Raw::AsyncSocket moved (std::move (rawSocket)); + + ASSERT_EQ (moved.asyncRead (_buf, sizeof (_buf), nullptr), -1); + ASSERT_EQ (join::lastError, Errc::InUse); + + ASSERT_EQ (moved.asyncWrite (reinterpret_cast (&_packet), sizeof (_packet), nullptr), 0) + << join::lastError.message (); + + ASSERT_TRUE (wait (1)); + ASSERT_FALSE (_code) << _code.message (); + ASSERT_GT (_transferred, 0u); + + Raw::AsyncSocket assigned; + assigned = std::move (moved); + + ASSERT_TRUE (assigned.opened ()); + + // the moved from socket must stay safe to use. + ASSERT_EQ (moved.cancelRead (), 0) << join::lastError.message (); + ASSERT_EQ (moved.cancelWrite (), 0) << join::lastError.message (); + ASSERT_FALSE (moved.opened ()); + moved.close (); + + Raw::AsyncSocket chained (std::move (moved)); + ASSERT_FALSE (chained.opened ()); + + Raw::AsyncSocket reassigned; + reassigned = std::move (chained); + ASSERT_FALSE (reassigned.opened ()); + + assigned.close (); +} + +/** + * @brief Test open method. + */ +TEST_F (RawAsyncSocket, open) +{ + Raw::AsyncSocket rawSocket; + + ASSERT_EQ (rawSocket.open (), 0) << join::lastError.message (); + ASSERT_EQ (rawSocket.open (), -1); + ASSERT_EQ (join::lastError, Errc::InUse); + rawSocket.close (); +} + +/** + * @brief Test close method. + */ +TEST_F (RawAsyncSocket, close) +{ + Raw::AsyncSocket rawSocket; + + ASSERT_EQ (rawSocket.open (), 0) << join::lastError.message (); + ASSERT_TRUE (rawSocket.opened ()); + rawSocket.close (); + ASSERT_FALSE (rawSocket.opened ()); +} + +/** + * @brief Test bind method. + */ +TEST_F (RawAsyncSocket, bind) +{ + Raw::AsyncSocket rawSocket; + + ASSERT_EQ (rawSocket.bind (_interface), 0) << join::lastError.message (); + rawSocket.close (); +} + +/** + * @brief Test bindToDevice method. + */ +TEST_F (RawAsyncSocket, bindToDevice) +{ + Raw::AsyncSocket rawSocket; + + ASSERT_EQ (rawSocket.bindToDevice (_interface), -1); + ASSERT_EQ (rawSocket.open (), 0) << join::lastError.message (); + ASSERT_EQ (rawSocket.bindToDevice (_interface), 0) << join::lastError.message (); + ASSERT_EQ (rawSocket.bindToDevice ("foo"), -1); + rawSocket.close (); +} + +/** + * @brief Test asyncWrite method. + */ +TEST_F (RawAsyncSocket, asyncWrite) +{ + Raw::AsyncSocket rawSocket; + + ASSERT_EQ (rawSocket.asyncWrite (reinterpret_cast (&_packet), sizeof (_packet), nullptr), -1); + ASSERT_EQ (join::lastError, Errc::OperationFailed); + + ASSERT_EQ (rawSocket.bind (_interface), 0) << join::lastError.message (); + ASSERT_EQ (rawSocket.asyncWrite (reinterpret_cast (&_packet), sizeof (_packet), onCompletion), 0) + << join::lastError.message (); + + ASSERT_TRUE (wait (1)); + ASSERT_FALSE (_code) << _code.message (); + ASSERT_EQ (_transferred, sizeof (_packet)); + + rawSocket.close (); +} + +/** + * @brief Test asyncRead method. + */ +TEST_F (RawAsyncSocket, asyncRead) +{ + Raw::AsyncSocket rawSocket; + + ASSERT_EQ (rawSocket.asyncRead (_buf, sizeof (_buf), nullptr), -1); + ASSERT_EQ (join::lastError, Errc::OperationFailed); + + ASSERT_EQ (rawSocket.bind (_interface), 0) << join::lastError.message (); + ASSERT_EQ (rawSocket.asyncRead (_buf, sizeof (_buf), onCompletion), 0) << join::lastError.message (); + + ASSERT_EQ (rawSocket.asyncRead (_buf, sizeof (_buf), nullptr), -1); + ASSERT_EQ (join::lastError, Errc::InUse); + + ASSERT_EQ (rawSocket.asyncWrite (reinterpret_cast (&_packet), sizeof (_packet), nullptr), 0) + << join::lastError.message (); + + ASSERT_TRUE (wait (1)); + ASSERT_FALSE (_code) << _code.message (); + ASSERT_GT (_transferred, 0u); + + // a message larger than the buffer must be reported as truncated. + ASSERT_EQ (rawSocket.asyncRead (_buf, sizeof (_packet) / 2, onCompletion), 0) << join::lastError.message (); + ASSERT_EQ (rawSocket.asyncWrite (reinterpret_cast (&_packet), sizeof (_packet), nullptr), 0) + << join::lastError.message (); + + ASSERT_TRUE (wait (2)); + ASSERT_EQ (_code, Errc::MessageTooLong) << _code.message (); + + rawSocket.close (); +} + +/** + * @brief Test cancelRead method. + */ +TEST_F (RawAsyncSocket, cancelRead) +{ + Raw::AsyncSocket rawSocket; + + ASSERT_EQ (rawSocket.cancelRead (), 0) << join::lastError.message (); + + ASSERT_EQ (rawSocket.bind (_interface), 0) << join::lastError.message (); + ASSERT_EQ (rawSocket.asyncRead (_buf, sizeof (_buf), onCompletion), 0) << join::lastError.message (); + ASSERT_EQ (rawSocket.cancelRead (), 0) << join::lastError.message (); + + ASSERT_TRUE (wait (1)); + ASSERT_EQ (_code, std::errc::operation_canceled) << _code.message (); + + rawSocket.close (); +} + +/** + * @brief Test cancelWrite method. + */ +TEST_F (RawAsyncSocket, cancelWrite) +{ + Raw::AsyncSocket rawSocket; + + ASSERT_EQ (rawSocket.cancelWrite (), 0) << join::lastError.message (); + + ASSERT_EQ (rawSocket.bind (_interface), 0) << join::lastError.message (); + ASSERT_EQ (rawSocket.cancelWrite (), 0) << join::lastError.message (); + + rawSocket.close (); +} + +/** + * @brief Test setOption method. + */ +TEST_F (RawAsyncSocket, setOption) +{ + Raw::AsyncSocket rawSocket; + + ASSERT_EQ (rawSocket.setOption (Raw::Socket::RcvBuffer, 1500), -1); + ASSERT_EQ (rawSocket.open (), 0) << join::lastError.message (); + ASSERT_EQ (rawSocket.setOption (Raw::Socket::RcvBuffer, 1500), 0) << join::lastError.message (); + ASSERT_EQ (rawSocket.setOption (Raw::Socket::SndBuffer, 1500), 0) << join::lastError.message (); + rawSocket.close (); +} + +/** + * @brief Test localEndpoint method. + */ +TEST_F (RawAsyncSocket, localEndpoint) +{ + Raw::AsyncSocket rawSocket; + + ASSERT_EQ (rawSocket.bind (_interface), 0) << join::lastError.message (); + ASSERT_EQ (rawSocket.localEndpoint ().device (), _interface); + rawSocket.close (); +} + +/** + * @brief Test opened method. + */ +TEST_F (RawAsyncSocket, opened) +{ + Raw::AsyncSocket rawSocket; + + ASSERT_FALSE (rawSocket.opened ()); + ASSERT_EQ (rawSocket.open (), 0) << join::lastError.message (); + ASSERT_TRUE (rawSocket.opened ()); + rawSocket.close (); + ASSERT_FALSE (rawSocket.opened ()); +} + +/** + * @brief Test canRead method. + */ +TEST_F (RawAsyncSocket, canRead) +{ + Raw::AsyncSocket rawSocket; + + ASSERT_EQ (rawSocket.canRead (), -1); + ASSERT_EQ (rawSocket.bind (_interface), 0) << join::lastError.message (); + ASSERT_EQ (rawSocket.asyncWrite (reinterpret_cast (&_packet), sizeof (_packet), onCompletion), 0) + << join::lastError.message (); + + ASSERT_TRUE (wait (1)); + ASSERT_FALSE (_code) << _code.message (); + ASSERT_GT (rawSocket.canRead (), 0) << join::lastError.message (); + + rawSocket.close (); +} + +/** + * @brief Test family method. + */ +TEST_F (RawAsyncSocket, family) +{ + Raw::AsyncSocket rawSocket; + + ASSERT_EQ (rawSocket.open (), 0) << join::lastError.message (); + ASSERT_EQ (rawSocket.family (), AF_PACKET); + rawSocket.close (); +} + +/** + * @brief Test type method. + */ +TEST_F (RawAsyncSocket, type) +{ + Raw::AsyncSocket rawSocket; + + ASSERT_EQ (rawSocket.open (), 0) << join::lastError.message (); + ASSERT_EQ (rawSocket.type (), SOCK_RAW); + rawSocket.close (); +} + +/** + * @brief Test protocol method. + */ +TEST_F (RawAsyncSocket, protocol) +{ + Raw::AsyncSocket rawSocket; + + ASSERT_EQ (rawSocket.open (), 0) << join::lastError.message (); + ASSERT_EQ (rawSocket.protocol (), Raw ().protocol ()); + rawSocket.close (); +} + +/** + * @brief Test handle method. + */ +TEST_F (RawAsyncSocket, handle) +{ + Raw::AsyncSocket rawSocket; + + ASSERT_EQ (rawSocket.handle (), -1); + ASSERT_EQ (rawSocket.open (), 0) << join::lastError.message (); + ASSERT_GT (rawSocket.handle (), -1); + rawSocket.close (); + ASSERT_EQ (rawSocket.handle (), -1); +} + +/** + * @brief main function. + */ +int main (int argc, char** argv) +{ + testing::InitGoogleTest (&argc, argv); + + return RUN_ALL_TESTS (); +} diff --git a/core/tests/tcp_async_acceptor_test.cpp b/core/tests/tcp_async_acceptor_test.cpp index 527b0d41..b41fab40 100644 --- a/core/tests/tcp_async_acceptor_test.cpp +++ b/core/tests/tcp_async_acceptor_test.cpp @@ -183,6 +183,17 @@ TEST_F (TcpAsyncAcceptor, move) ASSERT_TRUE (assigned.opened ()); + ASSERT_EQ (moved.cancelAccept (), 0) << join::lastError.message (); + ASSERT_FALSE (moved.opened ()); + moved.close (); + + Tcp::AsyncAcceptor chained (std::move (moved)); + ASSERT_FALSE (chained.opened ()); + + Tcp::AsyncAcceptor reassigned; + reassigned = std::move (chained); + ASSERT_FALSE (reassigned.opened ()); + client.close (); assigned.close (); } @@ -454,18 +465,6 @@ TEST_F (TcpAsyncAcceptor, handle) ASSERT_EQ (server.handle (), -1); } -/** - * @brief Test acceptor method. - */ -TEST_F (TcpAsyncAcceptor, acceptor) -{ - Tcp::AsyncAcceptor server; - - ASSERT_EQ (server.create ({_address, _port}), 0) << join::lastError.message (); - ASSERT_TRUE (server.acceptor ().opened ()); - server.close (); -} - /** * @brief main function. */ diff --git a/core/tests/tcp_async_stream_socket_test.cpp b/core/tests/tcp_async_stream_socket_test.cpp index d93ef4ba..c2817d6a 100644 --- a/core/tests/tcp_async_stream_socket_test.cpp +++ b/core/tests/tcp_async_stream_socket_test.cpp @@ -29,6 +29,9 @@ // Libraries. #include +// C. +#include + using join::Errc; using join::Mutex; using join::Condition; @@ -293,6 +296,20 @@ TEST_F (TcpAsyncStreamSocket, move) ASSERT_TRUE (assigned.connected ()); + ASSERT_EQ (moved.cancelRead (), 0) << join::lastError.message (); + ASSERT_EQ (moved.cancelWrite (), 0) << join::lastError.message (); + ASSERT_EQ (moved.asyncConnect ({_host, _port}, nullptr), -1); + ASSERT_EQ (join::lastError, Errc::OperationFailed); + ASSERT_FALSE (moved.opened ()); + moved.close (); + + Tcp::AsyncSocket chained (std::move (moved)); + ASSERT_FALSE (chained.opened ()); + + Tcp::AsyncSocket reassigned; + reassigned = std::move (chained); + ASSERT_FALSE (reassigned.opened ()); + assigned.close (); } @@ -742,7 +759,7 @@ TEST_F (TcpAsyncStreamSocket, cancelWrite) ASSERT_EQ (peer.setOption (Tcp::Socket::RcvBuffer, 4096), 0) << join::lastError.message (); int filled = 0; - while ((filled < 4096) && (sender.socket ().write (_buf, sizeof (_buf)) != -1)) + while ((filled < 4096) && (::write (sender.handle (), _buf, sizeof (_buf)) != -1)) { ++filled; } @@ -991,18 +1008,6 @@ TEST_F (TcpAsyncStreamSocket, handle) ASSERT_EQ (client.handle (), -1); } -/** - * @brief Test socket method. - */ -TEST_F (TcpAsyncStreamSocket, socket) -{ - Tcp::AsyncSocket client; - - ASSERT_EQ (client.open (), 0) << join::lastError.message (); - ASSERT_TRUE (client.socket ().opened ()); - client.close (); -} - /** * @brief main function. */ diff --git a/core/tests/unix_async_acceptor_test.cpp b/core/tests/unix_async_acceptor_test.cpp index 01d24437..8c2480fe 100644 --- a/core/tests/unix_async_acceptor_test.cpp +++ b/core/tests/unix_async_acceptor_test.cpp @@ -186,6 +186,17 @@ TEST_F (UnixAsyncAcceptor, move) ASSERT_TRUE (assigned.opened ()); + ASSERT_EQ (moved.cancelAccept (), 0) << join::lastError.message (); + ASSERT_FALSE (moved.opened ()); + moved.close (); + + UnixStream::AsyncAcceptor chained (std::move (moved)); + ASSERT_FALSE (chained.opened ()); + + UnixStream::AsyncAcceptor reassigned; + reassigned = std::move (chained); + ASSERT_FALSE (reassigned.opened ()); + client.close (); assigned.close (); } @@ -455,18 +466,6 @@ TEST_F (UnixAsyncAcceptor, handle) ASSERT_EQ (server.handle (), -1); } -/** - * @brief Test acceptor method. - */ -TEST_F (UnixAsyncAcceptor, acceptor) -{ - UnixStream::AsyncAcceptor server; - - ASSERT_EQ (server.create (_path), 0) << join::lastError.message (); - ASSERT_TRUE (server.acceptor ().opened ()); - server.close (); -} - /** * @brief main function. */ diff --git a/core/tests/unix_async_stream_socket_test.cpp b/core/tests/unix_async_stream_socket_test.cpp index 8ebf3860..90271370 100644 --- a/core/tests/unix_async_stream_socket_test.cpp +++ b/core/tests/unix_async_stream_socket_test.cpp @@ -29,6 +29,9 @@ // Libraries. #include +// C. +#include + using join::Errc; using join::Mutex; using join::Condition; @@ -300,6 +303,20 @@ TEST_F (UnixAsyncStreamSocket, move) ASSERT_TRUE (assigned.connected ()); + ASSERT_EQ (moved.cancelRead (), 0) << join::lastError.message (); + ASSERT_EQ (moved.cancelWrite (), 0) << join::lastError.message (); + ASSERT_EQ (moved.asyncConnect ({_serverpath}, nullptr), -1); + ASSERT_EQ (join::lastError, Errc::OperationFailed); + ASSERT_FALSE (moved.opened ()); + moved.close (); + + UnixStream::AsyncSocket chained (std::move (moved)); + ASSERT_FALSE (chained.opened ()); + + UnixStream::AsyncSocket reassigned; + reassigned = std::move (chained); + ASSERT_FALSE (reassigned.opened ()); + assigned.close (); } @@ -747,7 +764,7 @@ TEST_F (UnixAsyncStreamSocket, cancelWrite) ASSERT_EQ (peer.setOption (UnixStream::Socket::RcvBuffer, 4096), 0) << join::lastError.message (); int filled = 0; - while ((filled < 4096) && (sender.socket ().write (_buf, sizeof (_buf)) != -1)) + while ((filled < 4096) && (::write (sender.handle (), _buf, sizeof (_buf)) != -1)) { ++filled; } @@ -978,18 +995,6 @@ TEST_F (UnixAsyncStreamSocket, handle) ASSERT_EQ (client.handle (), -1); } -/** - * @brief Test socket method. - */ -TEST_F (UnixAsyncStreamSocket, socket) -{ - UnixStream::AsyncSocket client; - - ASSERT_EQ (client.open (), 0) << join::lastError.message (); - ASSERT_TRUE (client.socket ().opened ()); - client.close (); -} - /** * @brief main function. */