proactor.hpp

Go to the documentation of this file.

 
#ifndef JOIN_CORE_PROACTOR_HPP
#define JOIN_CORE_PROACTOR_HPP
 
// libjoin.
#include <join/slab_allocator.hpp>
#include <join/io_operation.hpp>
#include <join/function.hpp>
#ifdef JOIN_HAS_IO_URING
#include <join/io_policy.hpp>
#include <join/wheel.hpp>
#else
#include <join/reactor.hpp>
#endif
#include <join/backoff.hpp>
#include <join/thread.hpp>
#include <join/memory.hpp>
#include <join/queue.hpp>
 
// C++.
#include <unordered_map>
#include <utility>
#include <vector>
 
// C.
#include <sys/eventfd.h>
#include <sys/prctl.h>
#include <cerrno>
 
namespace join
{
    class CompletionHandler;
#ifdef JOIN_HAS_IO_URING
    template <typename IoPolicy>
    class BasicProactor;
    template <typename IoPolicy>
    class BasicProactorThread;
 
    using Proactor = BasicProactor<IoDefaultPolicy>;
    using HybridProactor = BasicProactor<IoHybridPolicy>;
    using SqpollProactor = BasicProactor<IoSqpollPolicy>;
    using ProactorThread = BasicProactorThread<IoDefaultPolicy>;
    using HybridProactorThread = BasicProactorThread<IoHybridPolicy>;
    using SqpollProactorThread = BasicProactorThread<IoSqpollPolicy>;
#else
    class BasicProactor;
    class BasicProactorThread;
 
    using Proactor = BasicProactor;
    using ProactorThread = BasicProactorThread;
#endif
}
 
class join::CompletionHandler
{
#ifdef JOIN_HAS_IO_URING
    template <typename IoPolicy>
    friend class join::BasicProactor;
#else
    friend class join::BasicProactor;
#endif
 
public:
    CompletionHandler () = default;
 
    CompletionHandler (const CompletionHandler& other) = default;
 
    CompletionHandler& operator= (const CompletionHandler& other) = default;
 
    CompletionHandler (CompletionHandler&& other) = default;
 
    CompletionHandler& operator= (CompletionHandler&& other) = default;
 
    virtual ~CompletionHandler () = default;
 
protected:
    virtual void onComplete ([[maybe_unused]] IoOperation& op, [[maybe_unused]] int result)
    {
        // do nothing.
    }
 
    virtual void onCancel ([[maybe_unused]] IoOperation& op, [[maybe_unused]] int result)
    {
        // do nothing.
    }
};
 
#ifdef JOIN_HAS_IO_URING
template <typename IoPolicy = join::IoDefaultPolicy>
class join::BasicProactor
#else
class join::BasicProactor : public join::EventHandler
#endif
{
private:
    static constexpr size_t _queueSize = 1024;
 
#ifdef JOIN_HAS_IO_URING
    class WheelPolicy
    {
    public:
        explicit WheelPolicy (BasicProactor& proactor) noexcept
        : _proactor (proactor)
        {
        }
 
        template <class Wheel>
        int start ([[maybe_unused]] Wheel& wheel, [[maybe_unused]] std::chrono::nanoseconds period) noexcept
        {
            // do nothing, the event loop advances the wheel.
            return 0;
        }
 
        void stop () noexcept
        {
            // do nothing, the event loop advances the wheel.
        }
 
        bool isRunnerThread () const noexcept
        {
            return _proactor.isProactorThread ();
        }
 
        template <class Wheel>
        int flush ([[maybe_unused]] Wheel& wheel) noexcept
        {
            if (JOIN_UNLIKELY (!_proactor.isRunning ()))
            {
                return -1;
            }
 
            return _proactor.wakeup ();
        }
 
    private:
        template <class Wheel>
        static uint64_t advance (Wheel& wheel) noexcept
        {
            return wheel.advance ();
        }
 
        template <class Wheel>
        static __kernel_timespec* timeout (const Wheel& wheel, __kernel_timespec& ts) noexcept
        {
            const std::chrono::nanoseconds next = wheel.next ();
 
            if (next == std::chrono::nanoseconds::max ())
            {
                return nullptr;
            }
 
            ts.tv_sec = next.count () / 1'000'000'000;
            ts.tv_nsec = next.count () % 1'000'000'000;
 
            return &ts;
        }
 
        BasicProactor& _proactor;
 
        friend class BasicProactor;
    };
#endif
 
public:
    using InvokeHandler = Function<void (), 64>;
 
#ifdef JOIN_HAS_IO_URING
    static_assert (sleep_ns_v<IoPolicy> == 0 || tick_ns_v<IoPolicy> >= sleep_ns_v<IoPolicy>,
                   "tick must not be finer than the backoff sleep");
 
    using Wheel = BasicWheel<Monotonic, WheelPolicy, _queueSize, tick_ns_v<IoPolicy>>;
#else
    using Wheel = Reactor::Wheel;
#endif
 
    explicit BasicProactor ();
 
    BasicProactor (const BasicProactor& other) = delete;
 
    BasicProactor& operator= (const BasicProactor& other) = delete;
 
    BasicProactor (BasicProactor&& other) = delete;
 
    BasicProactor& operator= (BasicProactor&& other) = delete;
 
    ~BasicProactor () noexcept;
 
    int submit (IoOperation& op, bool flush = false, bool sync = false) noexcept;
 
    int cancel (IoOperation& op, bool flush = false, bool sync = false) noexcept;
 
    int invoke (InvokeHandler* fn, bool sync = true) noexcept;
 
#ifdef JOIN_HAS_IO_URING
    int flush (bool sync = false) noexcept;
#endif
 
    void suspend (IoOperation& op) noexcept;
 
    void resume (IoOperation& op, CompletionHandler& handler) noexcept;
 
    void run ();
 
    void stop (bool sync = true) noexcept;
 
    void waitStopped () const noexcept;
 
#ifdef JOIN_HAS_IO_URING
    template <size_t Count, size_t... Sizes>
    int registerFixedBuffers (LocalMem::SlabAllocator<Count, Sizes...>& arena) noexcept;
 
    int unregisterFixedBuffers () noexcept;
#endif
 
    template <size_t Count, size_t Size>
    int registerBufferRing (uint16_t group, LocalMem::SlabAllocator<Count, Size>& arena);
 
    template <size_t Count, size_t Size>
    int registerBufferRing (LocalMem::SlabAllocator<Count, Size>& arena);
 
    int unregisterBufferRing (uint16_t group);
 
#ifdef JOIN_HAS_NUMA
    int mbind (int numa) const noexcept;
#endif
 
    int mlock () const noexcept;
 
    bool isRunning () const noexcept;
 
    bool isProactorThread () const noexcept;
 
    Wheel& wheel () noexcept;
 
private:
    enum class WakeupState
    {
        Sleeping, 
        Waking,   
        Spinning, 
    };
 
    enum class CommandType
    {
        Submit, 
        Cancel, 
        Invoke, 
        Stop,   
#ifdef JOIN_HAS_IO_URING
        Flush, 
#endif
    };
 
    struct alignas (64) Command
    {
        CommandType type;        
        IoOperation* op;         
        bool flush;              
        std::atomic<bool>* done; 
        std::error_code* errc;   
        InvokeHandler* fn;       
    };
 
    int writeCommand (const Command& cmd) noexcept;
 
#ifdef JOIN_HAS_IO_URING
    int wakeup () noexcept;
#endif
 
    void readCommands () noexcept;
 
    void processCommand (const Command& cmd) noexcept;
 
    int submitOperation (IoOperation& op, bool flush) noexcept;
 
    int cancelOperation (IoOperation& op, bool flush) noexcept;
 
    void cancelAllOperations () noexcept;
 
    int invokeFunction (InvokeHandler* fn) noexcept;
 
    void notifyOperation (IoOperation& op, int result, bool cancelled) noexcept;
 
    void dispatchOperation (IoOperation& op, int result, bool cancelled) noexcept;
 
    void endOperation (IoOperation& op, int result, bool cancelled = false) noexcept;
 
    void resetOperation (IoOperation& op) noexcept;
 
#ifdef JOIN_HAS_IO_URING
    template <size_t Count, size_t... Sizes, size_t... Is>
    int registerFixedBuffers (LocalMem::SlabAllocator<Count, Sizes...>& arena, std::index_sequence<Is...>) noexcept;
 
    io_uring_sqe* getSqe () noexcept;
 
    void prepareSqe (io_uring_sqe* sqe, IoOperation& op) noexcept;
 
    void dispatchCqe (io_uring_cqe* cqe) noexcept;
 
    void eventLoop () noexcept;
 
    void eventLoop (std::false_type, std::false_type) noexcept;
 
    void eventLoop (std::true_type, std::false_type) noexcept;
 
    void eventLoop (std::true_type, std::true_type) noexcept;
#else
    void onEvent (int fd, uint32_t revents) noexcept override;
 
    void drainQueue (IoOperation& tail, int result, uint32_t revents, bool cancelled) noexcept;
 
    void processQueue (int fd, bool write, uint32_t revents) noexcept;
 
    void processOperation (IoOperation& op, uint32_t revents) noexcept;
 
    static int executeOperation (IoOperation& op, uint32_t revents) noexcept;
 
    bool unqueueOperation (IoOperation& op) noexcept;
 
    static bool isWriteOperation (const IoOperation& op) noexcept;
#endif
 
#ifdef JOIN_HAS_IO_URING
    Wheel _wheel{*this};
#endif
 
    LocalMem::Mpsc::Queue<Command> _commands;
 
#ifdef JOIN_HAS_IO_URING
    alignas (64) std::atomic<WakeupState> _wakeupState{WakeupState::Spinning};
#else
    alignas (64) std::atomic<WakeupState> _wakeupState{WakeupState::Sleeping};
#endif
 
    int _wakeup = -1;
 
#ifdef JOIN_HAS_IO_URING
    alignas (64) uint64_t _wakeupBuf = 0;
 
    IoOperation _wakeupOp = {};
 
    io_uring _ring = {};
 
    std::vector<IoOperation*> _pendingOps;
 
    std::atomic<pid_t> _threadId{Thread::invalidId};
 
    std::atomic<bool> _running{false};
#else
    std::vector<IoOperation*> _readOps;
 
    std::vector<IoOperation*> _writeOps;
 
    Reactor _reactor;
#endif
 
    std::unordered_map<uint16_t, IoRingBuffer> _bufferRings;
};
 
// =========================================================================
//   CLASS     : BasicProactor
//   METHOD    : submit
// =========================================================================
#ifdef JOIN_HAS_IO_URING
template <typename IoPolicy>
int join::BasicProactor<IoPolicy>::submit (IoOperation& op, bool flush, bool sync) noexcept
#else
inline int join::BasicProactor::submit (IoOperation& op, bool flush, bool sync) noexcept
#endif
{
    if (isProactorThread ())
    {
        return submitOperation (op, flush);
    }
 
    std::atomic<bool> done{false}, *pdone = nullptr;
    std::error_code errc, *perrc = nullptr;
 
    if (JOIN_UNLIKELY (sync))
    {
        pdone = &done;
        perrc = &errc;
    }
 
    if (JOIN_UNLIKELY (writeCommand ({CommandType::Submit, &op, flush, pdone, perrc, nullptr}) == -1))
    {
        return -1;  // LCOV_EXCL_LINE
    }
 
    if (JOIN_UNLIKELY (sync))
    {
        Backoff backoff;
        while (!done.load (std::memory_order_acquire))
        {
            backoff ();
        }
 
        if (JOIN_UNLIKELY (errc))
        {
            lastError = errc;
            return -1;
        }
    }
 
    return 0;
}
 
// =========================================================================
//   CLASS     : BasicProactor
//   METHOD    : cancel
// =========================================================================
#ifdef JOIN_HAS_IO_URING
template <typename IoPolicy>
int join::BasicProactor<IoPolicy>::cancel (IoOperation& op, bool flush, bool sync) noexcept
#else
inline int join::BasicProactor::cancel (IoOperation& op, bool flush, bool sync) noexcept
#endif
{
    if (isProactorThread ())
    {
        return cancelOperation (op, flush);
    }
 
    std::atomic<bool> done{false}, *pdone = nullptr;
    std::error_code errc, *perrc = nullptr;
 
    if (JOIN_UNLIKELY (sync))
    {
        pdone = &done;
        perrc = &errc;
    }
 
    if (JOIN_UNLIKELY (writeCommand ({CommandType::Cancel, &op, flush, pdone, perrc, nullptr}) == -1))
    {
        return -1;  // LCOV_EXCL_LINE
    }
 
    if (JOIN_UNLIKELY (sync))
    {
        Backoff backoff;
        while (!done.load (std::memory_order_acquire))
        {
            backoff ();
        }
 
        if (JOIN_UNLIKELY (errc))
        {
            lastError = errc;
            return -1;
        }
    }
 
    return 0;
}
 
// =========================================================================
//   CLASS     : BasicProactor
//   METHOD    : invoke
// =========================================================================
#ifdef JOIN_HAS_IO_URING
template <typename IoPolicy>
int join::BasicProactor<IoPolicy>::invoke (InvokeHandler* fn, bool sync) noexcept
#else
inline int join::BasicProactor::invoke (InvokeHandler* fn, bool sync) noexcept
#endif
{
    if (isProactorThread ())
    {
        return invokeFunction (fn);
    }
 
    std::atomic<bool> done{false}, *pdone = nullptr;
    std::error_code errc, *perrc = nullptr;
 
    if (JOIN_LIKELY (sync))
    {
        pdone = &done;
        perrc = &errc;
    }
 
    if (JOIN_UNLIKELY (writeCommand ({CommandType::Invoke, nullptr, false, pdone, perrc, fn}) == -1))
    {
        return -1;  // LCOV_EXCL_LINE
    }
 
    if (JOIN_LIKELY (sync))
    {
        Backoff backoff;
        while (!done.load (std::memory_order_acquire))
        {
            backoff ();
        }
 
        if (JOIN_UNLIKELY (errc))
        {
            lastError = errc;
            return -1;
        }
    }
 
    return 0;
}
 
// =========================================================================
//   CLASS     : BasicProactor
//   METHOD    : suspend
// =========================================================================
#ifdef JOIN_HAS_IO_URING
template <typename IoPolicy>
void join::BasicProactor<IoPolicy>::suspend (IoOperation& op) noexcept
#else
inline void join::BasicProactor::suspend (IoOperation& op) noexcept
#endif
{
    Backoff backoff;
    for (;;)
    {
        IoOperation::State expected = op.state.load (std::memory_order_acquire);
        if ((expected == IoOperation::State::Idle) || (expected == IoOperation::State::Submitted))
        {
            if (op.state.compare_exchange_strong (expected, IoOperation::State::Suspended, std::memory_order_acquire,
                                                  std::memory_order_relaxed))
            {
                op.resume = expected;
                return;
            }
        }
        else if ((expected == IoOperation::State::Busy) && isProactorThread ())
        {
            return;
        }
 
        backoff ();
    }
}
 
// =========================================================================
//   CLASS     : BasicProactor
//   METHOD    : resume
// =========================================================================
#ifdef JOIN_HAS_IO_URING
template <typename IoPolicy>
void join::BasicProactor<IoPolicy>::resume (IoOperation& op, CompletionHandler& handler) noexcept
#else
inline void join::BasicProactor::resume (IoOperation& op, CompletionHandler& handler) noexcept
#endif
{
    op.handler = &handler;
    IoOperation::State expected = IoOperation::State::Suspended;
    op.state.compare_exchange_strong (expected, op.resume, std::memory_order_release, std::memory_order_relaxed);
}
 
// =========================================================================
//   CLASS     : BasicProactor
//   METHOD    : invokeFunction
// =========================================================================
#ifdef JOIN_HAS_IO_URING
template <typename IoPolicy>
int join::BasicProactor<IoPolicy>::invokeFunction (InvokeHandler* fn) noexcept
#else
inline int join::BasicProactor::invokeFunction (InvokeHandler* fn) noexcept
#endif
{
    if (JOIN_UNLIKELY ((fn == nullptr) || !*fn))
    {
        lastError = make_error_code (Errc::InvalidParam);
        return -1;
    }
 
    (*fn) ();
 
    return 0;
}
 
// =========================================================================
//   CLASS     : BasicProactor
//   METHOD    : notifyOperation
// =========================================================================
#ifdef JOIN_HAS_IO_URING
template <typename IoPolicy>
void join::BasicProactor<IoPolicy>::notifyOperation (IoOperation& op, int result, bool cancelled) noexcept
#else
inline void join::BasicProactor::notifyOperation (IoOperation& op, int result, bool cancelled) noexcept
#endif
{
    if (JOIN_LIKELY (op.handler))
    {
        if (cancelled)
        {
            op.handler->onCancel (op, result);
        }
        else
        {
            op.handler->onComplete (op, result);
        }
    }
}
 
// =========================================================================
//   CLASS     : BasicProactor
//   METHOD    : dispatchOperation
// =========================================================================
#ifdef JOIN_HAS_IO_URING
template <typename IoPolicy>
void join::BasicProactor<IoPolicy>::dispatchOperation (IoOperation& op, int result, bool cancelled) noexcept
#else
inline void join::BasicProactor::dispatchOperation (IoOperation& op, int result, bool cancelled) noexcept
#endif
{
    Backoff backoff;
    for (;;)
    {
        IoOperation::State expected = op.state.load (std::memory_order_relaxed);
        if (expected != IoOperation::State::Suspended)
        {
            if (op.state.compare_exchange_strong (expected, IoOperation::State::Busy, std::memory_order_acquire,
                                                  std::memory_order_relaxed))
            {
                break;
            }
        }
        backoff ();  // LCOV_EXCL_LINE
    }
 
    if (op.ring != nullptr)
    {
        op.ring->unbind ();
        op.ring = nullptr;
    }
 
    op.more = false;
 
    notifyOperation (op, result, cancelled);
}
 
// =========================================================================
//   CLASS     : BasicProactor
//   METHOD    : resetOperation
// =========================================================================
#ifdef JOIN_HAS_IO_URING
template <typename IoPolicy>
void join::BasicProactor<IoPolicy>::resetOperation (IoOperation& op) noexcept
#else
inline void join::BasicProactor::resetOperation (IoOperation& op) noexcept
#endif
{
    IoOperation::State expected = IoOperation::State::Submitted;
    if (!op.state.compare_exchange_strong (expected, IoOperation::State::Idle, std::memory_order_release,
                                           std::memory_order_relaxed) &&
        (expected != IoOperation::State::Busy))  // LCOV_EXCL_LINE
    {
        op.resume = IoOperation::State::Idle;  // LCOV_EXCL_LINE
    }
}
 
#ifdef JOIN_HAS_IO_URING
#include "proactor_uring_impl.hpp"
#else
#include "proactor_epoll_impl.hpp"
#endif
 
#ifdef JOIN_HAS_IO_URING
template <typename IoPolicy = join::IoDefaultPolicy>
class join::BasicProactorThread
#else
class join::BasicProactorThread
#endif
{
public:
#ifdef JOIN_HAS_IO_URING
    static BasicProactor<IoPolicy>& proactor ()
#else
    static BasicProactor& proactor ()
#endif
    {
        return instance ()._proactor;
    }
 
    static int affinity (int core)
    {
        return instance ()._dispatcher.affinity (core);
    }
 
    static int affinity () noexcept
    {
        return instance ()._dispatcher.affinity ();
    }
 
    static int priority (int prio)
    {
        return instance ()._dispatcher.priority (prio);
    }
 
    static int priority () noexcept
    {
        return instance ()._dispatcher.priority ();
    }
 
    static pthread_t handle () noexcept
    {
        return instance ()._dispatcher.handle ();
    }
 
#ifdef JOIN_HAS_NUMA
    static int mbind (int numa) noexcept
    {
        return instance ()._proactor.mbind (numa);
    }
#endif
 
    static int mlock () noexcept
    {
        return instance ()._proactor.mlock ();
    }
 
private:
    static BasicProactorThread& instance ()
    {
        static BasicProactorThread proactorThread;
        return proactorThread;
    }
 
    BasicProactorThread ()
    {
        _dispatcher = Thread ([this] () {
            _proactor.run ();
        });
 
        Backoff backoff;
        while (!_proactor.isRunning ())
        {
            backoff ();
        }
    }
 
    BasicProactorThread (const BasicProactorThread&) = delete;
 
    BasicProactorThread& operator= (const BasicProactorThread&) = delete;
 
    BasicProactorThread (BasicProactorThread&&) = delete;
 
    BasicProactorThread& operator= (BasicProactorThread&&) = delete;
 
    ~BasicProactorThread ()
    {
        _proactor.stop ();
        _dispatcher.join ();
    }
 
#ifdef JOIN_HAS_IO_URING
    BasicProactor<IoPolicy> _proactor;
#else
    BasicProactor _proactor;
#endif
 
    Thread _dispatcher;
};
 
#endif