async_socket.hpp

Go to the documentation of this file.

 
#ifndef JOIN_CORE_ASYNC_SOCKET_HPP
#define JOIN_CORE_ASYNC_SOCKET_HPP
 
// libjoin.
#include <join/async_operation.hpp>
#include <join/proactor.hpp>
#include <join/backoff.hpp>
#include <join/memory.hpp>
#include <join/socket.hpp>
#include <join/utils.hpp>
 
// C++.
#include <system_error>
#include <utility>
#include <atomic>
#include <array>
#include <new>
 
// C.
#include <sys/socket.h>
#include <poll.h>
 
namespace join
{
    template <class Protocol, class ProactorType, size_t OpCount>
    class BasicAsyncSocket : public CompletionHandler
    {
    public:
        using Socket = typename Protocol::Socket;
        using Endpoint = typename Protocol::Endpoint;
        using AsyncWait = BasicAsyncWait<Protocol, ProactorType>;
        using WaitHandler = typename AsyncWait::Wait;
 
        static constexpr size_t _opCount = OpCount;
 
        static_assert (OpCount > 0, "a socket needs at least one operation slot");
 
        static constexpr size_t _opSize = nextPow2 (AsyncOp<Protocol, ProactorType>::maxSize);
 
        using OpAllocator = LocalMem::SlabAllocator<_opCount, _opSize>;
 
    protected:
        explicit BasicAsyncSocket (ProactorType& proactor = ProactorThread::proactor ())
        : _proactor (&proactor)
        {
        }
 
        explicit BasicAsyncSocket (Socket&& sock, ProactorType& proactor = ProactorThread::proactor ())
        : _proactor (&proactor)
        , _socket (std::move (sock))
        {
        }
 
        BasicAsyncSocket (const BasicAsyncSocket& other) = delete;
 
        BasicAsyncSocket (BasicAsyncSocket&& other) noexcept
        : _proactor (other._proactor)
        {
            other.suspendAll ();
 
            _allocator = std::move (other._allocator);
 
            for (size_t i = 0; i < other._ops.size (); ++i)
            {
                _ops[i].store (other._ops[i].exchange (nullptr, std::memory_order_acq_rel), std::memory_order_release);
            }
 
            _socket = std::move (other._socket);
        }
 
    public:
        BasicAsyncSocket& operator= (const BasicAsyncSocket& other) = delete;
 
        BasicAsyncSocket& operator= (BasicAsyncSocket&& other) noexcept
        {
            close ();
 
            _proactor = other._proactor;
 
            other.suspendAll ();
 
            _allocator = std::move (other._allocator);
 
            for (size_t i = 0; i < other._ops.size (); ++i)
            {
                _ops[i].store (other._ops[i].exchange (nullptr, std::memory_order_acq_rel), std::memory_order_release);
            }
 
            _socket = std::move (other._socket);
 
            return *this;
        }
 
        ~BasicAsyncSocket ()
        {
            close ();
        }
 
        int open (const Protocol& protocol = Protocol ()) noexcept
        {
            return _socket.open (protocol);
        }
 
        void close () noexcept
        {
            bool pending = false;
            typename ProactorType::InvokeHandler fn = [this, &pending] () {
                cancelAll ();
                pending = pendingAny ();
            };
            Backoff backoff;
 
            do
            {
                if (_proactor->isRunning ())
                {
                    _proactor->invoke (&fn);
                }
                else
                {
                    fn ();
                }
 
                backoff ();
            }
            while (!_proactor->isProactorThread () && pending);
 
            _socket.close ();
        }
 
        template <size_t Count, size_t Size>
        int registerBufferRing (uint16_t group, LocalMem::SlabAllocator<Count, Size>& arena)
        {
            return _proactor->registerBufferRing (group, arena);
        }
 
        int unregisterBufferRing (uint16_t group)
        {
            return _proactor->unregisterBufferRing (group);
        }
 
        ssize_t asyncWait (WaitHandler handler, bool wantRead, bool wantWrite, bool flush = true) noexcept
        {
            if (JOIN_UNLIKELY (!_socket.opened ()))
            {
                lastError = make_error_code (Errc::OperationFailed);
                return -1;
            }
 
            if (JOIN_UNLIKELY (wantRead == wantWrite))
            {
                lastError = make_error_code (Errc::InvalidParam);
                return -1;
            }
 
            AsyncWait* wait = allocateOperation<AsyncWait> ();
            if (JOIN_UNLIKELY (wait == nullptr))
            {
                lastError = make_error_code (Errc::OutOfMemory);
                return -1;
            }
 
            wait->waitHandler = std::move (handler);
            wait->op = IoOperation::makePoll (_socket.handle (), wantRead ? POLLIN : POLLOUT, this);
            wait->op.state.store (IoOperation::State::Submitted, std::memory_order_release);
 
            size_t index = _allocator.getIndex (wait);
 
            if (_proactor->submit (wait->op, flush, false) == -1)
            {
                // LCOV_EXCL_START
                releaseOperation (wait);
                return -1;
                // LCOV_EXCL_STOP
            }
 
            return static_cast<ssize_t> (index);
        }
 
        ssize_t asyncWaitMulti (WaitHandler handler, bool wantRead, bool wantWrite, bool flush = true) noexcept
        {
            if (JOIN_UNLIKELY (!_socket.opened ()))
            {
                lastError = make_error_code (Errc::OperationFailed);
                return -1;
            }
 
            if (JOIN_UNLIKELY (wantRead == wantWrite))
            {
                lastError = make_error_code (Errc::InvalidParam);
                return -1;
            }
 
            AsyncWait* wait = allocateOperation<AsyncWait> ();
            if (JOIN_UNLIKELY (wait == nullptr))
            {
                lastError = make_error_code (Errc::OutOfMemory);
                return -1;
            }
 
            wait->waitHandler = std::move (handler);
            wait->op = IoOperation::makePollMulti (_socket.handle (), wantRead ? POLLIN : POLLOUT, this);
            wait->op.state.store (IoOperation::State::Submitted, std::memory_order_release);
 
            size_t index = _allocator.getIndex (wait);
 
            if (_proactor->submit (wait->op, flush, false) == -1)
            {
                // LCOV_EXCL_START
                releaseOperation (wait);
                return -1;
                // LCOV_EXCL_STOP
            }
 
            return static_cast<ssize_t> (index);
        }
 
        int cancel (size_t index) noexcept
        {
            if (JOIN_UNLIKELY (index >= _ops.size ()))
            {
                lastError = make_error_code (Errc::InvalidParam);
                return -1;
            }
 
            IoOperation* op = _ops[index].load (std::memory_order_acquire);
 
            return (op != nullptr) ? cancelOperation (*op) : 0;
        }
 
#ifdef JOIN_HAS_IO_URING
        int flush () noexcept
        {
            return _proactor->flush (false);
        }
 
        template <size_t Count, size_t... Sizes>
        int registerFixedBuffers (LocalMem::SlabAllocator<Count, Sizes...>& arena) noexcept
        {
            return _proactor->registerFixedBuffers (arena);
        }
 
        int unregisterFixedBuffers () noexcept
        {
            return _proactor->unregisterFixedBuffers ();
        }
#endif
 
        Endpoint localEndpoint () const noexcept
        {
            return _socket.localEndpoint ();
        }
 
        bool opened () const noexcept
        {
            return _socket.opened ();
        }
 
        int family () const noexcept
        {
            return _socket.family ();
        }
 
        int type () const noexcept
        {
            return _socket.type ();
        }
 
        int protocol () const noexcept
        {
            return _socket.protocol ();
        }
 
        int handle () const noexcept
        {
            return _socket.handle ();
        }
 
    protected:
        void onComplete (IoOperation& op, int result) override
        {
            std::error_code code =
                (result < 0) ? std::error_code (-result, std::generic_category ()) : std::error_code ();
 
            dispatch (op, code, (result > 0) ? static_cast<size_t> (result) : 0);
        }
 
        void onCancel (IoOperation& op, [[maybe_unused]] int result) override
        {
            dispatch (op, make_error_code (std::errc::operation_canceled), 0);
        }
 
        virtual void dispatch (IoOperation& op, const std::error_code& code, size_t size) noexcept
        {
            completeWait (reinterpret_cast<AsyncWait*> (&op), code, size);
        }
 
        void completeWait (AsyncWait* wait, const std::error_code& code, size_t revents) noexcept
        {
            std::error_code result = code;
 
            if (!result)
            {
                bool ready = (revents & wait->op.data.poll.events) != 0;
 
                if (JOIN_UNLIKELY (revents & POLLERR))
                {
                    int err = 0;
                    socklen_t len = sizeof (err);
                    ::getsockopt (_socket.handle (), SOL_SOCKET, SO_ERROR, &err, &len);
 
                    if (JOIN_LIKELY (err != 0))
                    {
                        result = std::error_code (err, std::generic_category ());
                    }
                    else
                    {
                        result = make_error_code (Errc::OperationFailed);  // LCOV_EXCL_LINE
                    }
                }
                else if (JOIN_UNLIKELY ((revents & POLLHUP) && !ready))
                {
                    result = make_error_code (Errc::ConnectionClosed);
                }
            }
 
            if (wait->op.more && !result && !(revents & POLLHUP))
            {
                if (JOIN_LIKELY (wait->waitHandler))
                {
                    wait->waitHandler (result, true);
                }
 
                return;
            }
 
            WaitHandler handler = std::move (wait->waitHandler);
 
            if (wait->op.more)
            {
                cancelOperation (wait->op);
            }
            else
            {
                releaseOperation (wait);
            }
 
            if (handler)
            {
                handler (result, false);
            }
        }
 
        template <class Op>
        Op* allocateOperation () noexcept
        {
            static_assert (sizeof (Op) <= _opSize, "operation larger than an allocator slot");
 
            void* chunk = _allocator.allocate (sizeof (Op));
            if (JOIN_UNLIKELY (chunk == nullptr))
            {
                return nullptr;
            }
 
            Op* operation = new (chunk) Op;
            _ops[_allocator.getIndex (chunk)].store (&operation->op, std::memory_order_release);
 
            return operation;
        }
 
        template <class Op>
        void releaseOperation (Op* operation) noexcept
        {
            const uint32_t index = _allocator.getIndex (operation);
 
            operation->~Op ();
            _allocator.deallocate (operation);
            _ops[index].store (nullptr, std::memory_order_release);
        }
 
        void suspendAll () noexcept
        {
            for (auto& slot : _ops)
            {
                IoOperation* op = slot.load (std::memory_order_acquire);
                if (op != nullptr)
                {
                    _proactor->suspend (*op);
                }
            }
        }
 
        void resumeAll () noexcept
        {
            for (auto& slot : _ops)
            {
                IoOperation* op = slot.load (std::memory_order_acquire);
                if (op != nullptr)
                {
                    _proactor->resume (*op, *this);
                }
            }
        }
 
        void cancelAll () noexcept
        {
            for (auto& slot : _ops)
            {
                IoOperation* op = slot.load (std::memory_order_acquire);
                if (op != nullptr)
                {
                    cancelOperation (*op);
                }
            }
        }
 
        bool pendingAny () const noexcept
        {
            for (auto& slot : _ops)
            {
                IoOperation* op = slot.load (std::memory_order_acquire);
                if ((op != nullptr) && pending (*op))
                {
                    return true;
                }
            }
 
            return false;
        }
 
        int cancelOperation (IoOperation& op) noexcept
        {
            if (!inFlight (op))
            {
                return 0;
            }
 
            if (_proactor->cancel (op, true, true) == -1)
            {
                return (lastError == Errc::OperationFailed) ? 0 : -1;
            }
 
            return 0;
        }
 
        bool inFlight (const IoOperation& op) const noexcept
        {
            return op.state.load (std::memory_order_acquire) == IoOperation::State::Submitted;
        }
 
        bool pending (const IoOperation& op) const noexcept
        {
            IoOperation::State state = op.state.load (std::memory_order_acquire);
 
            return (state == IoOperation::State::Submitted) || (state == IoOperation::State::Busy);
        }
 
        ProactorType* _proactor;
 
        Socket _socket;
 
        OpAllocator _allocator;
 
        std::array<std::atomic<IoOperation*>, _opCount> _ops{};
    };
}
 
#endif