From f48039e990381f9984ab6ce074be02aa0cd66852 Mon Sep 17 00:00:00 2001 From: mrabine Date: Sun, 27 Sep 2026 01:24:01 +0200 Subject: [PATCH] Add epoll operation queue --- core/include/join/async_datagram_socket.hpp | 6 +- core/include/join/async_raw_socket.hpp | 22 +- core/include/join/async_socket.hpp | 22 +- core/include/join/async_stream_socket.hpp | 2 +- core/include/join/io_operation.hpp | 7 +- core/include/join/proactor.hpp | 49 +- core/include/join/proactor_epoll_impl.hpp | 451 +++++++++++------- core/src/io_operation.cpp | 8 + core/tests/icmp_async_socket_test.cpp | 4 - core/tests/io_operation_test.cpp | 12 + core/tests/proactor_test.cpp | 106 +++- core/tests/raw_async_socket_test.cpp | 4 - core/tests/tcp_async_socket_test.cpp | 8 +- core/tests/udp_async_socket_test.cpp | 12 +- .../tests/unix_async_datagram_socket_test.cpp | 4 - core/tests/unix_async_stream_socket_test.cpp | 4 - 16 files changed, 465 insertions(+), 256 deletions(-) diff --git a/core/include/join/async_datagram_socket.hpp b/core/include/join/async_datagram_socket.hpp index 6851d91e..9e44e9cd 100644 --- a/core/include/join/async_datagram_socket.hpp +++ b/core/include/join/async_datagram_socket.hpp @@ -170,7 +170,7 @@ namespace join if (this->_proactor->submit (read->op, flush, false) == -1) { // LCOV_EXCL_START - this->releaseOp (read); + this->releaseOperation (read); return -1; // LCOV_EXCL_STOP } @@ -218,7 +218,7 @@ namespace join if (this->_proactor->submit (read->op, flush, false) == -1) { // LCOV_EXCL_START - this->releaseOp (read); + this->releaseOperation (read); return -1; // LCOV_EXCL_STOP } @@ -278,7 +278,7 @@ namespace join if (this->_proactor->submit (write->op, flush, false) == -1) { // LCOV_EXCL_START - this->releaseOp (write); + this->releaseOperation (write); return -1; // LCOV_EXCL_STOP } diff --git a/core/include/join/async_raw_socket.hpp b/core/include/join/async_raw_socket.hpp index 0302620f..47817475 100644 --- a/core/include/join/async_raw_socket.hpp +++ b/core/include/join/async_raw_socket.hpp @@ -196,7 +196,7 @@ namespace join if (this->_proactor->submit (read->op, flush, false) == -1) { // LCOV_EXCL_START - this->releaseOp (read); + this->releaseOperation (read); return -1; // LCOV_EXCL_STOP } @@ -240,7 +240,7 @@ namespace join if (this->_proactor->submit (read->op, flush, false) == -1) { // LCOV_EXCL_START - this->releaseOp (read); + this->releaseOperation (read); return -1; // LCOV_EXCL_STOP } @@ -280,7 +280,7 @@ namespace join if (this->_proactor->submit (read->op, flush, false) == -1) { // LCOV_EXCL_START - this->releaseOp (read); + this->releaseOperation (read); return -1; // LCOV_EXCL_STOP } @@ -331,7 +331,7 @@ namespace join if (this->_proactor->submit (write->op, flush, false) == -1) { // LCOV_EXCL_START - this->releaseOp (write); + this->releaseOperation (write); return -1; // LCOV_EXCL_STOP } @@ -375,7 +375,7 @@ namespace join if (this->_proactor->submit (write->op, flush, false) == -1) { // LCOV_EXCL_START - this->releaseOp (write); + this->releaseOperation (write); return -1; // LCOV_EXCL_STOP } @@ -396,7 +396,7 @@ namespace join if ((op != nullptr) && (static_cast (op->code) == IoOperation::Opcode::Connect)) { - return this->cancelOp (*op); + return this->cancelOperation (*op); } } @@ -461,7 +461,7 @@ namespace join this->_socket._state = Socket::Connected; } - this->releaseOp (connect); + this->releaseOperation (connect); if (JOIN_LIKELY (handler)) { @@ -540,7 +540,7 @@ namespace join ReadHandler handler = std::move (read->readHandler); ReadFromHandler fromHandler = std::move (read->readFromHandler); - this->releaseOp (read); + this->releaseOperation (read); if (fromHandler) { @@ -563,7 +563,7 @@ namespace join if (JOIN_LIKELY (!write->op.multishot)) { - this->releaseOp (write); + this->releaseOperation (write); } if (JOIN_LIKELY (handler)) @@ -578,7 +578,7 @@ namespace join */ AsyncRead* allocateRead () noexcept { - return this->template allocateOp (); + return this->template allocateOperation (); } /** @@ -587,7 +587,7 @@ namespace join */ AsyncWrite* allocateWrite () noexcept { - return this->template allocateOp (); + return this->template allocateOperation (); } }; } diff --git a/core/include/join/async_socket.hpp b/core/include/join/async_socket.hpp index 09a3eedc..09a8848c 100644 --- a/core/include/join/async_socket.hpp +++ b/core/include/join/async_socket.hpp @@ -228,7 +228,7 @@ namespace join return -1; } - AsyncWait* wait = allocateOp (); + AsyncWait* wait = allocateOperation (); if (JOIN_UNLIKELY (wait == nullptr)) { lastError = make_error_code (Errc::OutOfMemory); @@ -244,7 +244,7 @@ namespace join if (_proactor->submit (wait->op, flush, false) == -1) { // LCOV_EXCL_START - releaseOp (wait); + releaseOperation (wait); return -1; // LCOV_EXCL_STOP } @@ -274,7 +274,7 @@ namespace join return -1; } - AsyncWait* wait = allocateOp (); + AsyncWait* wait = allocateOperation (); if (JOIN_UNLIKELY (wait == nullptr)) { lastError = make_error_code (Errc::OutOfMemory); @@ -290,7 +290,7 @@ namespace join if (_proactor->submit (wait->op, flush, false) == -1) { // LCOV_EXCL_START - releaseOp (wait); + releaseOperation (wait); return -1; // LCOV_EXCL_STOP } @@ -313,7 +313,7 @@ namespace join IoOperation* op = _ops[index].load (std::memory_order_acquire); - return (op != nullptr) ? cancelOp (*op) : 0; + return (op != nullptr) ? cancelOperation (*op) : 0; } #ifdef JOIN_HAS_IO_URING @@ -498,11 +498,11 @@ namespace join if (wait->op.more) { - cancelOp (wait->op); + cancelOperation (wait->op); } else { - releaseOp (wait); + releaseOperation (wait); } if (handler) @@ -516,7 +516,7 @@ namespace join * @return allocated operation, or nullptr if the arena is exhausted. */ template - Op* allocateOp () noexcept + Op* allocateOperation () noexcept { static_assert (sizeof (Op) <= _opSize, "operation larger than an arena slot"); @@ -537,7 +537,7 @@ namespace join * @param operation operation to release. */ template - void releaseOp (Op* operation) noexcept + void releaseOperation (Op* operation) noexcept { const uint32_t index = _arena.getIndex (operation); @@ -586,7 +586,7 @@ namespace join IoOperation* op = slot.load (std::memory_order_acquire); if (op != nullptr) { - cancelOp (*op); + cancelOperation (*op); } } } @@ -623,7 +623,7 @@ namespace join * @param op operation to cancel. * @return 0 on success, -1 on failure. */ - int cancelOp (IoOperation& op) noexcept + int cancelOperation (IoOperation& op) noexcept { if (!inFlight (op)) { diff --git a/core/include/join/async_stream_socket.hpp b/core/include/join/async_stream_socket.hpp index 1f3454f8..4d56dc30 100644 --- a/core/include/join/async_stream_socket.hpp +++ b/core/include/join/async_stream_socket.hpp @@ -136,7 +136,7 @@ namespace join if (this->_proactor->submit (connect->op, true, false) == -1) { // LCOV_EXCL_START - this->releaseOp (connect); + this->releaseOperation (connect); this->_socket.close (); return -1; // LCOV_EXCL_STOP diff --git a/core/include/join/io_operation.hpp b/core/include/join/io_operation.hpp index 5c5667ce..9851111c 100644 --- a/core/include/join/io_operation.hpp +++ b/core/include/join/io_operation.hpp @@ -389,8 +389,13 @@ namespace join */ int fd () const noexcept; - /// index of this operation in the proactor pending ops (io_uring only). +#ifdef JOIN_HAS_IO_URING + /// index of this operation in the proactor pending ops. uint32_t index = 0; +#else + /// next operation queued on the same descriptor and direction. + IoOperation* next = nullptr; +#endif /// operation code. uint8_t code = 0; diff --git a/core/include/join/proactor.hpp b/core/include/join/proactor.hpp index d7d7f854..9a15b817 100644 --- a/core/include/join/proactor.hpp +++ b/core/include/join/proactor.hpp @@ -579,11 +579,35 @@ class join::BasicProactor : public join::EventHandler void eventLoop (std::true_type, std::true_type) noexcept; #else /** - * @brief return true if operation requires EPOLLOUT. - * @param op operation. - * @return true for Connect, Write, WriteFixed, SendMsg, Send and Poll without POLLIN. + * @brief method called when events are reported on handle. + * @param fd file descriptor. + * @param revents events reported by the reactor. + */ + void onEvent (int fd, uint32_t revents) noexcept override; + + /** + * @brief complete every operation of a queue detached from its descriptor, in submission order. + * @param tail last operation of the queue. + * @param result result reported to the operations. + * @param revents events reported by the reactor, handed to poll operations. + * @param cancelled true if the operations are cancelled. + */ + void drainQueue (IoOperation& tail, int result, uint32_t revents, bool cancelled) noexcept; + + /** + * @brief notify every poll operation of a queue whose descriptor is ready and execute its first other operation. + * @param fd file descriptor. + * @param write true to serve the write queue, false to serve the read queue. + * @param revents events reported by the reactor. */ - static bool isWriteOp (const IoOperation& op) noexcept; + void processQueue (int fd, bool write, uint32_t revents) noexcept; + + /** + * @brief execute an operation whose descriptor is ready. + * @param op operation to execute. + * @param revents events reported by the reactor. + */ + void processOperation (IoOperation& op, uint32_t revents) noexcept; /** * @brief execute the syscall described by op. @@ -591,14 +615,21 @@ class join::BasicProactor : public join::EventHandler * @param revents events reported by the reactor. * @return bytes transferred or ready events (>= 0), or -errno (< 0). */ - static int executeOp (IoOperation& op, uint32_t revents) noexcept; + static int executeOperation (IoOperation& op, uint32_t revents) noexcept; /** - * @brief method called when events are reported on handle. - * @param fd file descriptor. - * @param revents events reported by the reactor. + * @brief remove an operation from the queue of its descriptor and direction. + * @param op operation to remove. + * @return true if the operation was queued, false otherwise. */ - void onEvent (int fd, uint32_t revents) noexcept override; + bool unqueueOperation (IoOperation& op) noexcept; + + /** + * @brief return true if operation requires EPOLLOUT. + * @param op operation. + * @return true for Connect, Write, WriteFixed, SendMsg, Send and Poll without POLLIN. + */ + static bool isWriteOperation (const IoOperation& op) noexcept; #endif /// command queue size. diff --git a/core/include/join/proactor_epoll_impl.hpp b/core/include/join/proactor_epoll_impl.hpp index 1de7070d..4411cc18 100644 --- a/core/include/join/proactor_epoll_impl.hpp +++ b/core/include/join/proactor_epoll_impl.hpp @@ -469,29 +469,13 @@ inline int join::BasicProactor::submitOperation (IoOperation& op, [[maybe_unused _writeOps.resize (newSize, nullptr); } - bool isWrite = isWriteOp (op); - - if (JOIN_UNLIKELY ((isWrite && (_writeOps[op.fd ()] == &op)) || (!isWrite && (_readOps[op.fd ()] == &op)))) + if (JOIN_UNLIKELY (op.next != nullptr)) { lastError = make_error_code (std::errc::device_or_resource_busy); return -1; } - if (JOIN_UNLIKELY ((isWrite && (_writeOps[op.fd ()] != nullptr)) || (!isWrite && (_readOps[op.fd ()] != nullptr)))) - { - resetOperation (op); - lastError = make_error_code (Errc::InvalidParam); - return -1; - } - - if (isWrite) - { - _writeOps[op.fd ()] = &op; - } - else - { - _readOps[op.fd ()] = &op; - } + IoOperation*& last = isWriteOperation (op) ? _writeOps[op.fd ()] : _readOps[op.fd ()]; if (JOIN_UNLIKELY (ring != nullptr)) { @@ -499,17 +483,22 @@ inline int join::BasicProactor::submitOperation (IoOperation& op, [[maybe_unused ring->bind (); } + if (last != nullptr) + { + op.next = last->next; + last->next = &op; + last = &op; + return 0; + } + + op.next = &op; + last = &op; + int err = _reactor.addHandler (op.fd (), this, _readOps[op.fd ()] != nullptr, _writeOps[op.fd ()] != nullptr); if (JOIN_UNLIKELY (err == -1)) { - if (isWrite) - { - _writeOps[op.fd ()] = nullptr; - } - else - { - _readOps[op.fd ()] = nullptr; - } + op.next = nullptr; + last = nullptr; if (JOIN_UNLIKELY (ring != nullptr)) { @@ -547,30 +536,19 @@ inline int join::BasicProactor::cancelOperation (IoOperation& op, [[maybe_unused return -1; } - bool isWrite = isWriteOp (op); - - if (JOIN_UNLIKELY ((isWrite && (_writeOps[op.fd ()] != &op)) || (!isWrite && (_readOps[op.fd ()] != &op)))) + if (JOIN_UNLIKELY (!unqueueOperation (op))) { lastError = make_error_code (Errc::InvalidParam); return -1; } - if (isWrite) - { - _writeOps[op.fd ()] = nullptr; - } - else - { - _readOps[op.fd ()] = nullptr; - } - int ret = 0; if (_readOps[op.fd ()] == nullptr && _writeOps[op.fd ()] == nullptr) { ret = _reactor.delHandler (op.fd ()); } - else + else if ((isWriteOperation (op) ? _writeOps[op.fd ()] : _readOps[op.fd ()]) == nullptr) { ret = _reactor.addHandler (op.fd (), this, _readOps[op.fd ()] != nullptr, _writeOps[op.fd ()] != nullptr); } @@ -596,11 +574,11 @@ inline void join::BasicProactor::cancelAllOperations () noexcept } if (rOp != nullptr) { - dispatchOperation (*rOp, -ECANCELED, true); + drainQueue (*rOp, -ECANCELED, 0, true); } if (wOp != nullptr) { - dispatchOperation (*wOp, -ECANCELED, true); + drainQueue (*wOp, -ECANCELED, 0, true); } } } @@ -618,20 +596,13 @@ inline void join::BasicProactor::endOperation (IoOperation& op, int result, bool return; // LCOV_EXCL_LINE } - if (isWriteOp (op)) - { - _writeOps[fd] = nullptr; - } - else - { - _readOps[fd] = nullptr; - } + unqueueOperation (op); if (_readOps[fd] == nullptr && _writeOps[fd] == nullptr) { _reactor.delHandler (fd); } - else + else if ((isWriteOperation (op) ? _writeOps[fd] : _readOps[fd]) == nullptr) { _reactor.addHandler (fd, this, _readOps[fd] != nullptr, _writeOps[fd] != nullptr); } @@ -641,30 +612,230 @@ inline void join::BasicProactor::endOperation (IoOperation& op, int result, bool // ========================================================================= // CLASS : BasicProactor -// METHOD : isWriteOp +// METHOD : onEvent // ========================================================================= -inline bool join::BasicProactor::isWriteOp (const IoOperation& op) noexcept +inline void join::BasicProactor::onEvent (int fd, uint32_t revents) noexcept { - switch (static_cast (op.code)) + if (JOIN_UNLIKELY (fd == _wakeup)) { - case IoOperation::Opcode::Poll: - return (op.data.poll.events & POLLIN) == 0; - case IoOperation::Opcode::Connect: - case IoOperation::Opcode::Write: - case IoOperation::Opcode::WriteFixed: - case IoOperation::Opcode::SendMsg: - case IoOperation::Opcode::Send: - return true; - default: - return false; + readCommands (); + return; + } + + if (JOIN_UNLIKELY (revents & (EPOLLERR | EPOLLRDHUP | EPOLLHUP))) + { + IoOperation* rOp = std::exchange (_readOps[fd], nullptr); + IoOperation* wOp = std::exchange (_writeOps[fd], nullptr); + if (JOIN_LIKELY (rOp || wOp)) + { + _reactor.delHandler (fd); + } + + int result = (revents & EPOLLERR) ? -ECONNRESET : 0; + + if (rOp != nullptr) + { + drainQueue (*rOp, result, revents, false); + } + if (wOp != nullptr) + { + drainQueue (*wOp, result, revents, false); + } + + return; + } + + if (revents & EPOLLIN) + { + processQueue (fd, false, revents); + } + + if (revents & EPOLLOUT) + { + processQueue (fd, true, revents); + } +} + +// ========================================================================= +// CLASS : BasicProactor +// METHOD : drainQueue +// ========================================================================= +inline void join::BasicProactor::drainQueue (IoOperation& tail, int result, uint32_t revents, bool cancelled) noexcept +{ + IoOperation* op = tail.next; + tail.next = nullptr; + + while (op != nullptr) + { + IoOperation* next = op->next; + op->next = nullptr; + + int res = result; + + if ((revents != 0) && (op->code == static_cast (IoOperation::Opcode::Poll))) + { + res = executeOperation (*op, revents); + } + + dispatchOperation (*op, res, cancelled); + + op = next; } } // ========================================================================= // CLASS : BasicProactor -// METHOD : executeOp +// METHOD : processQueue // ========================================================================= -inline int join::BasicProactor::executeOp (IoOperation& op, uint32_t revents) noexcept +inline void join::BasicProactor::processQueue (int fd, bool write, uint32_t revents) noexcept +{ + IoOperation* last = write ? _writeOps[fd] : _readOps[fd]; + if (last == nullptr) + { + return; + } + + size_t count = 1; + for (IoOperation* it = last->next; it != last; it = it->next) + { + ++count; + } + + IoOperation* op = last->next; + bool consumed = false; + + for (size_t i = 0; (op != nullptr) && (i < count); ++i) + { + IoOperation* following = (op == last) ? nullptr : op->next; + bool poll = (op->code == static_cast (IoOperation::Opcode::Poll)); + + if (poll || !consumed) + { + consumed = consumed || !poll; + processOperation (*op, revents); + } + + last = write ? _writeOps[fd] : _readOps[fd]; + op = nullptr; + + if ((following != nullptr) && (last != nullptr)) + { + for (IoOperation* it = last->next;; it = it->next) + { + if (it == following) + { + op = following; + break; + } + + if (it == last) + { + break; + } + } + } + } +} + +// ========================================================================= +// CLASS : BasicProactor +// METHOD : processOperation +// ========================================================================= +inline void join::BasicProactor::processOperation (IoOperation& op, uint32_t revents) noexcept +{ + IoOperation::State current = op.state.load (std::memory_order_acquire); + Backoff backoff; + + while (JOIN_UNLIKELY (current == IoOperation::State::Suspended)) + { + backoff (); + current = op.state.load (std::memory_order_acquire); + } + + if (JOIN_UNLIKELY (current == IoOperation::State::Idle)) + { + return; // LCOV_EXCL_LINE + } + + IoRingBuffer* br = nullptr; + uint16_t bid = 0; + + if (op.ring != nullptr) + { + br = op.ring; + bid = static_cast (br->select ()); + + if (op.code == static_cast (IoOperation::Opcode::RecvMsg)) + { + uint32_t reserved = op.data.msg.namelen + op.data.msg.controllen; + if (JOIN_UNLIKELY (reserved >= br->size ())) + { + endOperation (op, -EFAULT, false); + br->recycle (bid); + return; + } + + char* base = static_cast (br->get (bid)); + + op.data.msg.msg->msg_name = base; + op.data.msg.msg->msg_namelen = op.data.msg.namelen; + op.data.msg.msg->msg_control = base + op.data.msg.namelen; + op.data.msg.msg->msg_controllen = op.data.msg.controllen; + op.data.msg.msg->msg_iov->iov_base = base + reserved; + op.data.msg.msg->msg_iov->iov_len = br->size () - reserved; + op.data.msg.msg->msg_iovlen = 1; + op.data.msg.msg->msg_flags = 0; + } + else + { + op.data.stream.buf = br->get (bid); + op.data.stream.len = br->size (); + } + } + + int result = executeOperation (op, revents); + + if (JOIN_UNLIKELY ((result == -EAGAIN) && (op.code != static_cast (IoOperation::Opcode::Connect)))) + { + if (br != nullptr) + { + br->recycle (bid); + } + return; + } + + if ((br != nullptr) && (result >= 0) && (op.code == static_cast (IoOperation::Opcode::RecvMsg))) + { + op.data.msg.msg->msg_iov->iov_len = static_cast (result); + + if (op.data.msg.msg->msg_controllen < sizeof (cmsghdr)) + { + op.data.msg.msg->msg_control = nullptr; + } + } + + if (op.multishot && + ((result > 0) || ((result == 0) && (op.code == static_cast (IoOperation::Opcode::Accept))))) + { + op.more = true; + notifyOperation (op, result, false); + } + else + { + endOperation (op, result, false); + } + + if (br != nullptr) + { + br->recycle (bid); + } +} + +// ========================================================================= +// CLASS : BasicProactor +// METHOD : executeOperation +// ========================================================================= +inline int join::BasicProactor::executeOperation (IoOperation& op, uint32_t revents) noexcept { for (;;) { @@ -767,141 +938,59 @@ inline int join::BasicProactor::executeOp (IoOperation& op, uint32_t revents) no // ========================================================================= // CLASS : BasicProactor -// METHOD : onEvent +// METHOD : unqueueOperation // ========================================================================= -inline void join::BasicProactor::onEvent (int fd, uint32_t revents) noexcept +inline bool join::BasicProactor::unqueueOperation (IoOperation& op) noexcept { - if (JOIN_UNLIKELY (fd == _wakeup)) + if (op.next == nullptr) { - readCommands (); - return; + return false; } - if (JOIN_UNLIKELY (revents & (EPOLLERR | EPOLLRDHUP | EPOLLHUP))) - { - IoOperation* rOp = std::exchange (_readOps[fd], nullptr); - IoOperation* wOp = std::exchange (_writeOps[fd], nullptr); - if (JOIN_LIKELY (rOp || wOp)) - { - _reactor.delHandler (fd); - } + IoOperation*& last = isWriteOperation (op) ? _writeOps[op.fd ()] : _readOps[op.fd ()]; + IoOperation* prev = last; - int result = (revents & EPOLLERR) ? -ECONNRESET : 0; - int rResult = result; - int wResult = result; - - if ((rOp != nullptr) && (rOp->code == static_cast (IoOperation::Opcode::Poll))) - { - rResult = executeOp (*rOp, revents); - } - - if ((wOp != nullptr) && (wOp->code == static_cast (IoOperation::Opcode::Poll))) - { - wResult = executeOp (*wOp, revents); - } - - if (rOp != nullptr) - { - dispatchOperation (*rOp, rResult, false); - } - if (wOp != nullptr) - { - dispatchOperation (*wOp, wResult, false); - } - - return; - } - - IoOperation* op = (revents & EPOLLIN) ? _readOps[fd] : _writeOps[fd]; - if (JOIN_UNLIKELY (op == nullptr)) - { - return; - } - - IoOperation::State current = op->state.load (std::memory_order_acquire); - Backoff backoff; - - while (JOIN_UNLIKELY (current == IoOperation::State::Suspended)) + while (prev->next != &op) { - backoff (); - current = op->state.load (std::memory_order_acquire); + prev = prev->next; } - if (JOIN_UNLIKELY (current == IoOperation::State::Idle)) + if (prev == &op) { - return; // LCOV_EXCL_LINE + last = nullptr; } - - IoRingBuffer* br = nullptr; - uint16_t bid = 0; - - if (op->ring != nullptr) + else { - br = op->ring; - bid = static_cast (br->select ()); - - if (op->code == static_cast (IoOperation::Opcode::RecvMsg)) - { - uint32_t reserved = op->data.msg.namelen + op->data.msg.controllen; - if (JOIN_UNLIKELY (reserved >= br->size ())) - { - endOperation (*op, -EFAULT, false); - br->recycle (bid); - return; - } - - char* base = static_cast (br->get (bid)); - - op->data.msg.msg->msg_name = base; - op->data.msg.msg->msg_namelen = op->data.msg.namelen; - op->data.msg.msg->msg_control = base + op->data.msg.namelen; - op->data.msg.msg->msg_controllen = op->data.msg.controllen; - op->data.msg.msg->msg_iov->iov_base = base + reserved; - op->data.msg.msg->msg_iov->iov_len = br->size () - reserved; - op->data.msg.msg->msg_iovlen = 1; - op->data.msg.msg->msg_flags = 0; - } - else - { - op->data.stream.buf = br->get (bid); - op->data.stream.len = br->size (); - } - } - - int result = executeOp (*op, revents); + prev->next = op.next; - if (JOIN_UNLIKELY ((result == -EAGAIN) && (op->code != static_cast (IoOperation::Opcode::Connect)))) - { - if (br != nullptr) + if (last == &op) { - br->recycle (bid); + last = prev; } - return; } - if ((br != nullptr) && (result >= 0) && (op->code == static_cast (IoOperation::Opcode::RecvMsg))) - { - op->data.msg.msg->msg_iov->iov_len = static_cast (result); - - if (op->data.msg.msg->msg_controllen < sizeof (cmsghdr)) - { - op->data.msg.msg->msg_control = nullptr; - } - } + op.next = nullptr; - if (op->multishot && - ((result > 0) || ((result == 0) && (op->code == static_cast (IoOperation::Opcode::Accept))))) - { - op->more = true; - notifyOperation (*op, result, false); - } - else - { - endOperation (*op, result, false); - } + return true; +} - if (br != nullptr) +// ========================================================================= +// CLASS : BasicProactor +// METHOD : isWriteOperation +// ========================================================================= +inline bool join::BasicProactor::isWriteOperation (const IoOperation& op) noexcept +{ + switch (static_cast (op.code)) { - br->recycle (bid); + case IoOperation::Opcode::Poll: + return (op.data.poll.events & POLLIN) == 0; + case IoOperation::Opcode::Connect: + case IoOperation::Opcode::Write: + case IoOperation::Opcode::WriteFixed: + case IoOperation::Opcode::SendMsg: + case IoOperation::Opcode::Send: + return true; + default: + return false; } } diff --git a/core/src/io_operation.cpp b/core/src/io_operation.cpp index 12b435a8..094fab1e 100644 --- a/core/src/io_operation.cpp +++ b/core/src/io_operation.cpp @@ -38,7 +38,11 @@ IoOperation::IoOperation (const IoOperation& other) noexcept code = other.code; state.store (other.state.load (std::memory_order_relaxed), std::memory_order_relaxed); resume = other.resume; +#ifdef JOIN_HAS_IO_URING index = other.index; +#else + next = other.next; +#endif linked = other.linked; multishot = other.multishot; more = other.more; @@ -55,7 +59,11 @@ IoOperation::IoOperation (const IoOperation& other) noexcept IoOperation& IoOperation::operator= (const IoOperation& other) noexcept { code = other.code; +#ifdef JOIN_HAS_IO_URING index = other.index; +#else + next = other.next; +#endif linked = other.linked; multishot = other.multishot; more = other.more; diff --git a/core/tests/icmp_async_socket_test.cpp b/core/tests/icmp_async_socket_test.cpp index 8329ff31..27d5c306 100644 --- a/core/tests/icmp_async_socket_test.cpp +++ b/core/tests/icmp_async_socket_test.cpp @@ -429,7 +429,6 @@ TEST_F (IcmpAsyncSocket, asyncWait) { } -#ifdef JOIN_HAS_IO_URING for (size_t i = 0; i < Icmp::AsyncSocket::_opCount; ++i) { ASSERT_NE (client.asyncWait (nullptr, true, false), -1) << join::lastError.message (); @@ -437,7 +436,6 @@ TEST_F (IcmpAsyncSocket, asyncWait) ASSERT_EQ (client.asyncWait (nullptr, true, false), -1); ASSERT_EQ (join::lastError, Errc::OutOfMemory); -#endif client.close (); } @@ -633,7 +631,6 @@ TEST_F (IcmpAsyncSocket, asyncReadFromMulti) ASSERT_FALSE (_more); } -#ifdef JOIN_HAS_IO_URING for (size_t i = 0; i < Icmp::AsyncSocket::_opCount; ++i) { ASSERT_NE (server.asyncReadFromMulti (nullptr, 0), -1) << join::lastError.message (); @@ -641,7 +638,6 @@ TEST_F (IcmpAsyncSocket, asyncReadFromMulti) ASSERT_EQ (server.asyncReadFromMulti (nullptr, 0), -1); ASSERT_EQ (join::lastError, Errc::OutOfMemory); -#endif client.close (); server.close (); diff --git a/core/tests/io_operation_test.cpp b/core/tests/io_operation_test.cpp index ef605e75..b56c6b00 100644 --- a/core/tests/io_operation_test.cpp +++ b/core/tests/io_operation_test.cpp @@ -49,7 +49,11 @@ TEST (IoOperation, copy) auto op = IoOperation::makeRecvmsgMulti (8, 3, &msg, MSG_DONTWAIT, nullptr); op.state.store (IoOperation::State::Submitted); op.resume = IoOperation::State::Submitted; +#ifdef JOIN_HAS_IO_URING op.index = 4; +#else + op.next = &op; +#endif op.linked = true; IoOperation copy (op); @@ -57,7 +61,11 @@ TEST (IoOperation, copy) ASSERT_EQ (copy.code, op.code); ASSERT_EQ (copy.state.load (), op.state.load ()); ASSERT_EQ (copy.resume, op.resume); +#ifdef JOIN_HAS_IO_URING ASSERT_EQ (copy.index, op.index); +#else + ASSERT_EQ (copy.next, op.next); +#endif ASSERT_EQ (copy.linked, op.linked); ASSERT_EQ (copy.multishot, op.multishot); ASSERT_EQ (copy.group, op.group); @@ -72,7 +80,11 @@ TEST (IoOperation, copy) ASSERT_EQ (assigned.code, op.code); ASSERT_EQ (assigned.state.load (), IoOperation::State::Idle); ASSERT_EQ (assigned.resume, IoOperation::State::Idle); +#ifdef JOIN_HAS_IO_URING ASSERT_EQ (assigned.index, op.index); +#else + ASSERT_EQ (assigned.next, op.next); +#endif ASSERT_EQ (assigned.linked, op.linked); ASSERT_EQ (assigned.multishot, op.multishot); ASSERT_EQ (assigned.group, op.group); diff --git a/core/tests/proactor_test.cpp b/core/tests/proactor_test.cpp index 1d51909a..0dc51ef4 100644 --- a/core/tests/proactor_test.cpp +++ b/core/tests/proactor_test.cpp @@ -57,6 +57,7 @@ class ProactorTest : public CompletionHandler, public ::testing::Test { _op = nullptr; _result = 0; + _spareResult = 0; _cancellations = 0; _completions = 0; _iovlen = 0; @@ -140,6 +141,10 @@ class ProactorTest : public CompletionHandler, public ::testing::Test _control = op.data.msg.msg->msg_control; _flags = op.data.msg.msg->msg_flags; } + if (&op == &_spareOp) + { + _spareResult = result; + } _result = result; _op = &op; ++_completions; @@ -250,6 +255,9 @@ class ProactorTest : public CompletionHandler, public ::testing::Test /// last operation result. static int _result; + /// result of the last completion of the spare operation. + static int _spareResult; + /// number of cancellations received. static int _cancellations; @@ -296,6 +304,7 @@ IoOperation* ProactorTest::_cancelled = nullptr; bool ProactorTest::_stopFromHandler = false; bool ProactorTest::_suspendFromHandler = false; int ProactorTest::_result = 0; +int ProactorTest::_spareResult = 0; int ProactorTest::_cancellations = 0; int ProactorTest::_completions = 0; size_t ProactorTest::_iovlen = 0; @@ -472,22 +481,50 @@ TEST_F (ProactorTest, submit) _result = 0; } -#ifndef JOIN_HAS_IO_URING - _spareOp = IoOperation::makeRead (_server.handle (), _buf, sizeof (_buf), this); + char spare[16] = {}; + _spareOp = IoOperation::makeRead (_server.handle (), spare, sizeof (spare), this); + ASSERT_EQ (proactor.submit (_spareOp, true, true), 0) << join::lastError.message (); ASSERT_EQ (proactor.submit (_spareOp, true, true), -1); - ASSERT_EQ (join::lastError, Errc::InvalidParam); + ASSERT_EQ (join::lastError, std::errc::device_or_resource_busy); + + int base = 0; + IoOperation* first = nullptr; - ASSERT_EQ (proactor.submit (_spareOp, true, false), 0) << join::lastError.message (); + { + ScopedLock lock (_mut); + base = _completions; + } + + ASSERT_EQ (_client.writeExactly ("first", 5), 0) << join::lastError.message (); { ScopedLock lock (_mut); ASSERT_TRUE (_cond.timedWait (lock, std::chrono::milliseconds (_timeout), [&] () { - return _op == &_spareOp && _result == -EINVAL; + return _completions == base + 1 && _result == 5; })); + first = _op; _op = nullptr; _result = 0; } -#endif + + ASSERT_EQ (_client.writeExactly ("second", 6), 0) << join::lastError.message (); + + { + ScopedLock lock (_mut); + ASSERT_TRUE (_cond.timedWait (lock, std::chrono::milliseconds (_timeout), [&] () { + return _completions == base + 2 && _result == 6; + })); + ASSERT_NE (_op, first); + _op = nullptr; + _result = 0; + } + + ASSERT_TRUE ((first == &_readOp) || (first == &_spareOp)); + ASSERT_EQ (std::string ((first == &_readOp) ? _buf : spare, 5), "first"); + ASSERT_EQ (std::string ((first == &_readOp) ? spare : _buf, 6), "second"); + + _readOp = IoOperation::makeRead (_server.handle (), _buf, sizeof (_buf), this); + ASSERT_EQ (proactor.submit (_readOp, true, true), 0) << join::lastError.message (); ASSERT_EQ (proactor.cancel (_readOp, true, true), 0) << join::lastError.message (); { @@ -570,6 +607,42 @@ TEST_F (ProactorTest, cancel) ASSERT_EQ (_cancelResult, 0); _handlerProactor = nullptr; + // cancel operations queued behind another one. + char middle[16] = {}, tail[16] = {}; + IoOperation tailOp = IoOperation::makeRead (_server.handle (), tail, sizeof (tail), this); + + _readOp = IoOperation::makeRead (_server.handle (), _buf, sizeof (_buf), this); + _spareOp = IoOperation::makeRead (_server.handle (), middle, sizeof (middle), this); + ASSERT_EQ (proactor.submit (_readOp, true, true), 0) << join::lastError.message (); + ASSERT_EQ (proactor.submit (_spareOp, true, true), 0) << join::lastError.message (); + ASSERT_EQ (proactor.submit (tailOp, true, true), 0) << join::lastError.message (); + + ASSERT_EQ (proactor.cancel (_spareOp, true, true), 0) << join::lastError.message (); + { + ScopedLock lock (_mut); + ASSERT_TRUE (_cond.timedWait (lock, std::chrono::milliseconds (_timeout), [&] () { + return _cancelled == &_spareOp; + })); + } + + ASSERT_EQ (proactor.cancel (tailOp, true, true), 0) << join::lastError.message (); + { + ScopedLock lock (_mut); + ASSERT_TRUE (_cond.timedWait (lock, std::chrono::milliseconds (_timeout), [&] () { + return _cancelled == &tailOp; + })); + } + + ASSERT_EQ (_client.writeExactly ("queued", 6), 0) << join::lastError.message (); + { + ScopedLock lock (_mut); + ASSERT_TRUE (_cond.timedWait (lock, std::chrono::milliseconds (_timeout), [&] () { + return _op == &_readOp && _result == 6; + })); + _op = nullptr; + _result = 0; + } + proactor.stop (); th.join (); } @@ -980,6 +1053,22 @@ TEST_F (ProactorTest, asyncPollMulti) ASSERT_EQ (_server.readExactly (_buf, 4, _timeout), 0) << join::lastError.message (); } + // a read queued behind the armed poll. + char spare[16] = {}; + _spareOp = IoOperation::makeRead (_server.handle (), spare, sizeof (spare), this); + ASSERT_EQ (ProactorThread::proactor ().submit (_spareOp, true, true), 0) << join::lastError.message (); + ASSERT_EQ (_client.writeExactly ("read", 4, _timeout), 0) << join::lastError.message (); + + { + ScopedLock lock (_mut); + ASSERT_TRUE (_cond.timedWait (lock, std::chrono::milliseconds (_timeout), [&] () { + return _spareResult == 4; + })); + ASSERT_EQ (std::string (spare, 4), "read"); + _op = nullptr; + _result = 0; + } + ASSERT_EQ (ProactorThread::proactor ().cancel (_readOp, true, true), 0) << join::lastError.message (); { @@ -1604,15 +1693,18 @@ TEST_F (ProactorTest, onClose) ASSERT_EQ (ProactorThread::mlock (), 0) << join::lastError.message (); ASSERT_GT (ProactorThread::handle (), 0); + char spare[16] = {}; _readOp = IoOperation::makeRead (_server.handle (), _buf, sizeof (_buf), this); + _spareOp = IoOperation::makeRead (_server.handle (), spare, sizeof (spare), this); ASSERT_EQ (ProactorThread::proactor ().submit (_readOp, true, true), 0) << join::lastError.message (); + ASSERT_EQ (ProactorThread::proactor ().submit (_spareOp, true, true), 0) << join::lastError.message (); _client.close (); { ScopedLock lock (_mut); ASSERT_TRUE (_cond.timedWait (lock, std::chrono::milliseconds (_timeout), [&] () { - return _op == &_readOp && _result == 0; + return _completions >= 2 && _result == 0; })); _op = nullptr; _result = 0; diff --git a/core/tests/raw_async_socket_test.cpp b/core/tests/raw_async_socket_test.cpp index 5c8a46d4..86961fba 100644 --- a/core/tests/raw_async_socket_test.cpp +++ b/core/tests/raw_async_socket_test.cpp @@ -425,7 +425,6 @@ TEST_F (RawAsyncSocket, asyncWait) { } -#ifdef JOIN_HAS_IO_URING for (size_t i = 0; i < Raw::AsyncSocket::_opCount; ++i) { ASSERT_NE (rawSocket.asyncWait (nullptr, true, false), -1) << join::lastError.message (); @@ -433,7 +432,6 @@ TEST_F (RawAsyncSocket, asyncWait) ASSERT_EQ (rawSocket.asyncWait (nullptr, true, false), -1); ASSERT_EQ (join::lastError, Errc::OutOfMemory); -#endif rawSocket.close (); } @@ -525,7 +523,6 @@ TEST_F (RawAsyncSocket, asyncReadMulti) ASSERT_EQ (code (), std::errc::operation_canceled); ASSERT_FALSE (_more); -#ifdef JOIN_HAS_IO_URING for (size_t i = 0; i < Raw::AsyncSocket::_opCount; ++i) { ASSERT_NE (rawSocket.asyncReadMulti (nullptr, 0), -1) << join::lastError.message (); @@ -533,7 +530,6 @@ TEST_F (RawAsyncSocket, asyncReadMulti) ASSERT_EQ (rawSocket.asyncReadMulti (nullptr, 0), -1); ASSERT_EQ (join::lastError, Errc::OutOfMemory); -#endif rawSocket.close (); diff --git a/core/tests/tcp_async_socket_test.cpp b/core/tests/tcp_async_socket_test.cpp index df591c2d..a4f3db74 100644 --- a/core/tests/tcp_async_socket_test.cpp +++ b/core/tests/tcp_async_socket_test.cpp @@ -504,7 +504,6 @@ TEST_F (TcpAsyncSocket, asyncWait) ASSERT_EQ (::recv (client.handle (), _buf, sizeof (_buf), 0), 5) << strerror (errno); ASSERT_EQ (std::string (_buf, 5), "hello"); -#ifdef JOIN_HAS_IO_URING for (size_t i = 0; i < Tcp::AsyncSocket::_opCount; ++i) { ASSERT_NE (client.asyncWait (nullptr, true, false), -1) << join::lastError.message (); @@ -517,7 +516,6 @@ TEST_F (TcpAsyncSocket, asyncWait) { ASSERT_EQ (client.cancel (i), 0) << join::lastError.message (); } -#endif ASSERT_NE (client.asyncWait (onReportWait, true, false), -1) << join::lastError.message (); @@ -763,7 +761,6 @@ TEST_F (TcpAsyncSocket, asyncReadMulti) ASSERT_FALSE (_more); } -#ifdef JOIN_HAS_IO_URING for (size_t i = 0; i < Tcp::AsyncSocket::_opCount; ++i) { ASSERT_NE (client.asyncReadMulti (nullptr, 0), -1) << join::lastError.message (); @@ -771,7 +768,6 @@ TEST_F (TcpAsyncSocket, asyncReadMulti) ASSERT_EQ (client.asyncReadMulti (nullptr, 0), -1); ASSERT_EQ (join::lastError, Errc::OutOfMemory); -#endif client.close (); @@ -1245,7 +1241,6 @@ TEST_F (TcpAsyncSocket, cancel) -1) << join::lastError.message (); -#ifdef JOIN_HAS_IO_URING for (size_t i = 1; i < Tcp::AsyncSocket::_opCount; ++i) { ASSERT_NE (client.asyncRead (nullptr, _buf, sizeof (_buf)), -1) << join::lastError.message (); @@ -1254,6 +1249,7 @@ TEST_F (TcpAsyncSocket, cancel) ASSERT_EQ (client.asyncRead (nullptr, _buf, sizeof (_buf)), -1); ASSERT_EQ (join::lastError, Errc::OutOfMemory); +#ifdef JOIN_HAS_IO_URING ASSERT_EQ (client.asyncReadFixed (nullptr, _buf, sizeof (_buf), 0), -1); ASSERT_EQ (join::lastError, Errc::OutOfMemory); #endif @@ -1329,7 +1325,6 @@ TEST_F (TcpAsyncSocket, cancel) ASSERT_NE (sender.asyncWrite (onWrite, _buf, sizeof (_buf)), -1) << join::lastError.message (); -#ifdef JOIN_HAS_IO_URING for (size_t i = 1; i < Tcp::AsyncSocket::_opCount; ++i) { ASSERT_NE (sender.asyncWrite (nullptr, _buf, sizeof (_buf)), -1) << join::lastError.message (); @@ -1338,6 +1333,7 @@ TEST_F (TcpAsyncSocket, cancel) ASSERT_EQ (sender.asyncWrite (nullptr, _buf, sizeof (_buf)), -1); ASSERT_EQ (join::lastError, Errc::OutOfMemory); +#ifdef JOIN_HAS_IO_URING ASSERT_EQ (sender.asyncWriteFixed (nullptr, _buf, sizeof (_buf), 0), -1); ASSERT_EQ (join::lastError, Errc::OutOfMemory); #endif diff --git a/core/tests/udp_async_socket_test.cpp b/core/tests/udp_async_socket_test.cpp index d8cc9f10..394bdcea 100644 --- a/core/tests/udp_async_socket_test.cpp +++ b/core/tests/udp_async_socket_test.cpp @@ -595,7 +595,6 @@ TEST_F (UdpAsyncSocket, asyncWait) ASSERT_EQ (::recv (client.handle (), _buf, sizeof (_buf), 0), 5) << strerror (errno); ASSERT_EQ (std::string (_buf, 5), "hello"); -#ifdef JOIN_HAS_IO_URING for (size_t i = 0; i < Udp::AsyncSocket::_opCount; ++i) { ASSERT_NE (client.asyncWait (nullptr, true, false), -1) << join::lastError.message (); @@ -608,7 +607,6 @@ TEST_F (UdpAsyncSocket, asyncWait) { ASSERT_EQ (client.cancel (i), 0) << join::lastError.message (); } -#endif client.close (); ASSERT_EQ (client.connect (closed), 0) << join::lastError.message (); @@ -746,15 +744,13 @@ TEST_F (UdpAsyncSocket, asyncWriteTo) ASSERT_NE (client.asyncWriteTo (onReport, "hello", 5, dest), -1) << join::lastError.message (); ASSERT_TRUE (client.opened ()); -#ifdef JOIN_HAS_IO_URING for (size_t i = 1; i < Udp::AsyncSocket::_opCount; ++i) { - ASSERT_NE (client.asyncWriteTo (nullptr, "hello", 5, dest), -1) << join::lastError.message (); + ASSERT_NE (client.asyncWriteTo (onReport, "hello", 5, dest), -1) << join::lastError.message (); } ASSERT_EQ (client.asyncWriteTo (nullptr, "hello", 5, dest), -1); ASSERT_EQ (join::lastError, Errc::OutOfMemory); -#endif Thread th ([&proactor] () { proactor.run (); @@ -763,7 +759,7 @@ TEST_F (UdpAsyncSocket, asyncWriteTo) { ScopedLock lock (_mut); ASSERT_TRUE (_cond.timedWait (lock, std::chrono::milliseconds (_timeout), [] () { - return _completions >= 1; + return _completions >= static_cast (Udp::AsyncSocket::_opCount); })); ASSERT_FALSE (_code) << _code.message (); ASSERT_EQ (_transferred, 5u); @@ -800,7 +796,6 @@ TEST_F (UdpAsyncSocket, asyncReadFrom) ASSERT_EQ (std::string (_buf, 5), "hello"); } -#ifdef JOIN_HAS_IO_URING for (size_t i = 0; i < Udp::AsyncSocket::_opCount; ++i) { ASSERT_NE (client.asyncReadFrom (nullptr, _buf, sizeof (_buf), _from), -1) << join::lastError.message (); @@ -808,7 +803,6 @@ TEST_F (UdpAsyncSocket, asyncReadFrom) ASSERT_EQ (client.asyncReadFrom (nullptr, _buf, sizeof (_buf), _from), -1); ASSERT_EQ (join::lastError, Errc::OutOfMemory); -#endif client.close (); @@ -861,7 +855,6 @@ TEST_F (UdpAsyncSocket, asyncReadFromMulti) ASSERT_FALSE (_more); } -#ifdef JOIN_HAS_IO_URING for (size_t i = 0; i < Udp::AsyncSocket::_opCount; ++i) { ASSERT_NE (client.asyncReadFromMulti (nullptr, 0), -1) << join::lastError.message (); @@ -869,7 +862,6 @@ TEST_F (UdpAsyncSocket, asyncReadFromMulti) ASSERT_EQ (client.asyncReadFromMulti (nullptr, 0), -1); ASSERT_EQ (join::lastError, Errc::OutOfMemory); -#endif client.close (); diff --git a/core/tests/unix_async_datagram_socket_test.cpp b/core/tests/unix_async_datagram_socket_test.cpp index 7561affc..74e0700d 100644 --- a/core/tests/unix_async_datagram_socket_test.cpp +++ b/core/tests/unix_async_datagram_socket_test.cpp @@ -460,7 +460,6 @@ TEST_F (UnixAsyncDatagramSocket, asyncWait) ASSERT_EQ (::recv (client.handle (), _buf, sizeof (_buf), 0), 5) << strerror (errno); -#ifdef JOIN_HAS_IO_URING for (size_t i = 0; i < UnixDgram::AsyncSocket::_opCount; ++i) { ASSERT_NE (client.asyncWait (nullptr, true, false), -1) << join::lastError.message (); @@ -468,7 +467,6 @@ TEST_F (UnixAsyncDatagramSocket, asyncWait) ASSERT_EQ (client.asyncWait (nullptr, true, false), -1); ASSERT_EQ (join::lastError, Errc::OutOfMemory); -#endif client.close (); } @@ -773,7 +771,6 @@ TEST_F (UnixAsyncDatagramSocket, asyncReadFromMulti) ASSERT_FALSE (_more); } -#ifdef JOIN_HAS_IO_URING for (size_t i = 0; i < UnixDgram::AsyncSocket::_opCount; ++i) { ASSERT_NE (client.asyncReadFromMulti (nullptr, 0), -1) << join::lastError.message (); @@ -781,7 +778,6 @@ TEST_F (UnixAsyncDatagramSocket, asyncReadFromMulti) ASSERT_EQ (client.asyncReadFromMulti (nullptr, 0), -1); ASSERT_EQ (join::lastError, Errc::OutOfMemory); -#endif client.close (); diff --git a/core/tests/unix_async_stream_socket_test.cpp b/core/tests/unix_async_stream_socket_test.cpp index bb65d0a4..72f44fb3 100644 --- a/core/tests/unix_async_stream_socket_test.cpp +++ b/core/tests/unix_async_stream_socket_test.cpp @@ -420,7 +420,6 @@ TEST_F (UnixAsyncStreamSocket, asyncWait) ASSERT_EQ (::recv (client.handle (), _buf, sizeof (_buf), 0), 5) << strerror (errno); ASSERT_EQ (std::string (_buf, 5), "hello"); -#ifdef JOIN_HAS_IO_URING for (size_t i = 0; i < UnixStream::AsyncSocket::_opCount; ++i) { ASSERT_NE (client.asyncWait (nullptr, true, false), -1) << join::lastError.message (); @@ -433,7 +432,6 @@ TEST_F (UnixAsyncStreamSocket, asyncWait) { ASSERT_EQ (client.cancel (i), 0) << join::lastError.message (); } -#endif ASSERT_NE (client.asyncWait (onReportWait, true, false), -1) << join::lastError.message (); @@ -803,7 +801,6 @@ TEST_F (UnixAsyncStreamSocket, asyncReadMulti) ASSERT_FALSE (_more); } -#ifdef JOIN_HAS_IO_URING for (size_t i = 0; i < UnixStream::AsyncSocket::_opCount; ++i) { ASSERT_NE (client.asyncReadMulti (nullptr, 0), -1) << join::lastError.message (); @@ -811,7 +808,6 @@ TEST_F (UnixAsyncStreamSocket, asyncReadMulti) ASSERT_EQ (client.asyncReadMulti (nullptr, 0), -1); ASSERT_EQ (join::lastError, Errc::OutOfMemory); -#endif client.close ();