async_acceptor.hpp

Go to the documentation of this file.

 
#ifndef JOIN_CORE_ASYNC_ACCEPTOR_HPP
#define JOIN_CORE_ASYNC_ACCEPTOR_HPP
 
// libjoin.
#include <join/async_stream_socket.hpp>
#include <join/acceptor.hpp>
#include <join/function.hpp>
#include <join/proactor.hpp>
#include <join/backoff.hpp>
 
// C++.
#include <system_error>
#include <utility>
 
namespace join
{
    template <class Protocol, class ProactorType>
    class BasicAsyncStreamAcceptor : public CompletionHandler
    {
    public:
        using Acceptor = BasicStreamAcceptor<Protocol>;
        using Endpoint = typename Protocol::Endpoint;
        using Socket = typename Protocol::Socket;
        using AsyncAccept = BasicAsyncAccept<Protocol, ProactorType>;
        using AcceptHandler = typename AsyncAccept::Accept;
 
        explicit BasicAsyncStreamAcceptor (ProactorType& proactor = ProactorThread::proactor ())
        : _proactor (&proactor)
        {
        }
 
        BasicAsyncStreamAcceptor (const BasicAsyncStreamAcceptor& other) = delete;
 
        BasicAsyncStreamAcceptor& operator= (const BasicAsyncStreamAcceptor& other) = delete;
 
        BasicAsyncStreamAcceptor (BasicAsyncStreamAcceptor&& other) = delete;
 
        BasicAsyncStreamAcceptor& operator= (BasicAsyncStreamAcceptor&& other) = delete;
 
        ~BasicAsyncStreamAcceptor ()
        {
            close ();
        }
 
        int create (const Endpoint& endpoint, int flags = SOCK_CLOEXEC | SOCK_NONBLOCK) noexcept
        {
            return _acceptor.create (endpoint, flags | SOCK_NONBLOCK);
        }
 
        void close () noexcept
        {
            Backoff backoff;
 
            do
            {
                cancel ();
 
                backoff ();
            }
            while (!_proactor->isProactorThread () && pending (_acceptOp.op));
 
            _acceptor.close ();
        }
 
        ssize_t asyncAccept (AcceptHandler handler, int flags = SOCK_NONBLOCK | SOCK_CLOEXEC,
                             bool flush = true) noexcept
        {
            if (JOIN_UNLIKELY (!_acceptor.opened ()))
            {
                lastError = make_error_code (Errc::OperationFailed);
                return -1;
            }
 
            if (JOIN_UNLIKELY (!armOp ()))
            {
                lastError = make_error_code (Errc::InUse);
                return -1;
            }
 
            _acceptOp.remoteLen = sizeof (struct sockaddr_storage);
            _acceptOp.acceptHandler = std::move (handler);
            _acceptOp.op = IoOperation::makeAccept (_acceptor.handle (), _acceptOp.remote.addr (), &_acceptOp.remoteLen,
                                                    flags | SOCK_NONBLOCK, this);
 
            if (_proactor->submit (_acceptOp.op, flush, false) == -1)
            {
                // LCOV_EXCL_START
                _acceptOp.acceptHandler.reset ();
                disarmOp (IoOperation::State::Submitted);
                return -1;
                // LCOV_EXCL_STOP
            }
 
            return 0;
        }
 
        ssize_t asyncAcceptMulti (AcceptHandler handler, int flags = SOCK_NONBLOCK | SOCK_CLOEXEC,
                                  bool flush = true) noexcept
        {
            if (JOIN_UNLIKELY (!_acceptor.opened ()))
            {
                lastError = make_error_code (Errc::OperationFailed);
                return -1;
            }
 
            if (JOIN_UNLIKELY (!armOp ()))
            {
                lastError = make_error_code (Errc::InUse);
                return -1;
            }
 
            _acceptOp.remote = Endpoint ();
            _acceptOp.acceptHandler = std::move (handler);
            _acceptOp.op = IoOperation::makeAcceptMulti (_acceptor.handle (), flags | SOCK_NONBLOCK, this);
 
            if (_proactor->submit (_acceptOp.op, flush, false) == -1)
            {
                // LCOV_EXCL_START
                _acceptOp.acceptHandler.reset ();
                disarmOp (IoOperation::State::Submitted);
                return -1;
                // LCOV_EXCL_STOP
            }
 
            return 0;
        }
 
        int cancel () noexcept
        {
            if (!inFlight (_acceptOp.op))
            {
                return 0;
            }
 
            if (_proactor->cancel (_acceptOp.op, true, true) == -1)
            {
                return (lastError == Errc::OperationFailed) ? 0 : -1;
            }
 
            return 0;
        }
 
#ifdef JOIN_HAS_IO_URING
        int flush () noexcept
        {
            return _proactor->flush (false);
        }
#endif
 
        Endpoint localEndpoint () const
        {
            return _acceptor.localEndpoint ();
        }
 
        bool opened () const noexcept
        {
            return _acceptor.opened ();
        }
 
        int family () const noexcept
        {
            return _acceptor.family ();
        }
 
        int type () const noexcept
        {
            return _acceptor.type ();
        }
 
        int protocol () const noexcept
        {
            return _acceptor.protocol ();
        }
 
        int handle () const noexcept
        {
            return _acceptor.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
        {
            completeAccept (reinterpret_cast<AsyncAccept*> (&op), code, size);
        }
 
        void completeAccept (AsyncAccept* accept, const std::error_code& code, size_t handle) noexcept
        {
            Socket sock = code ? Socket () : Socket (static_cast<int> (handle), accept->remote);
 
            if (accept->op.more)
            {
                if (JOIN_LIKELY (accept->acceptHandler))
                {
                    accept->acceptHandler (std::move (sock), code, true);
                }
 
                return;
            }
 
            AcceptHandler handler = std::move (accept->acceptHandler);
 
            if (JOIN_LIKELY (handler))
            {
                handler (std::move (sock), code, false);
            }
 
            disarmOp (IoOperation::State::Busy);
        }
 
        bool armOp () noexcept
        {
            IoOperation::State expected = IoOperation::State::Idle;
 
            return _acceptOp.op.state.compare_exchange_strong (expected, IoOperation::State::Submitted,
                                                               std::memory_order_acquire, std::memory_order_relaxed) ||
                   (expected == IoOperation::State::Busy);
        }
 
        void disarmOp (IoOperation::State expected) noexcept
        {
            _acceptOp.op.state.compare_exchange_strong (expected, IoOperation::State::Idle, std::memory_order_release,
                                                        std::memory_order_relaxed);
        }
 
        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);
        }
 
    private:
        ProactorType* _proactor;
 
        Acceptor _acceptor;
 
        AsyncAccept _acceptOp;
    };
}
 
#endif