proactor.hpp

Go to the documentation of this file.

 
#ifndef JOIN_CORE_PROACTOR_HPP
#define JOIN_CORE_PROACTOR_HPP
 
// libjoin.
#include <join/io_operation.hpp>
#include <join/allocator.hpp>
#include <join/function.hpp>
#ifdef JOIN_HAS_IO_URING
#include <join/io_policy.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 <cerrno>
 
namespace join
{
    class CompletionHandler;
#ifdef JOIN_HAS_IO_URING
    template <typename Policy>
    class BasicProactor;
    template <typename Policy>
    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 Policy>
    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 Policy = join::IoDefaultPolicy>
class join::BasicProactor
#else
class join::BasicProactor : public join::EventHandler
#endif
{
public:
    using InvokeHandler = Function<void (), 64>;
 
    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::Allocator<Count, Sizes...>& arena) noexcept;
 
    int unregisterFixedBuffers () noexcept;
#endif
 
    template <size_t Count, size_t Size>
    int registerBufferRing (uint16_t group, LocalMem::Allocator<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;
 
private:
#ifdef JOIN_HAS_IO_URING
    int initWakeup (std::true_type) noexcept;
 
    int initWakeup (std::false_type) noexcept;
 
    void initWakeupOp (std::true_type) noexcept;
 
    void initWakeupOp (std::false_type) noexcept;
 
    void initCqEntries (io_uring_params&, std::false_type) noexcept;
 
    void initCqEntries (io_uring_params& params, std::true_type) noexcept;
 
    void initSqThreadIdle (io_uring_params&, std::false_type) noexcept;
 
    void initSqThreadIdle (io_uring_params& params, std::true_type) noexcept;
 
    void initSqThreadCpu (io_uring_params&, std::false_type) noexcept;
 
    void initSqThreadCpu (io_uring_params& params, std::true_type) noexcept;
#endif
 
    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 writeCommand (const Command& cmd, std::true_type) noexcept;
 
    int writeCommand (const Command& cmd, std::false_type) 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::Allocator<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 dispatchCqe (io_uring_cqe* cqe, std::true_type) noexcept;
 
    void dispatchCqe (io_uring_cqe* cqe, std::false_type) 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
    static bool isWriteOp (const IoOperation& op) noexcept;
 
    static int executeOp (IoOperation& op, uint32_t revents) noexcept;
 
    void onEvent (int fd, uint32_t revents) noexcept override;
#endif
 
    static constexpr size_t _queueSize = 1024;
 
    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;
 
    static constexpr pthread_t _invalidThreadId = static_cast<pthread_t> (-1);
 
    std::atomic<pthread_t> _threadId{_invalidThreadId};
 
    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 Policy>
int join::BasicProactor<Policy>::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 Policy>
int join::BasicProactor<Policy>::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 Policy>
int join::BasicProactor<Policy>::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 Policy>
void join::BasicProactor<Policy>::suspend (IoOperation* op) noexcept
#else
inline void join::BasicProactor::suspend (IoOperation* op) noexcept
#endif
{
    if (JOIN_UNLIKELY (op == nullptr))
    {
        return;
    }
 
    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 Policy>
void join::BasicProactor<Policy>::resume (IoOperation* op, CompletionHandler* handler) noexcept
#else
inline void join::BasicProactor::resume (IoOperation* op, CompletionHandler* handler) noexcept
#endif
{
    if (JOIN_UNLIKELY (op == nullptr))
    {
        return;
    }
 
    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 Policy>
int join::BasicProactor<Policy>::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 Policy>
void join::BasicProactor<Policy>::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 Policy>
void join::BasicProactor<Policy>::dispatchOperation (IoOperation* op, int result, bool cancelled) noexcept
#else
inline void join::BasicProactor::dispatchOperation (IoOperation* op, int result, bool cancelled) noexcept
#endif
{
    if (JOIN_UNLIKELY (op == nullptr))
    {
        return;  // LCOV_EXCL_LINE
    }
 
    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 Policy>
void join::BasicProactor<Policy>::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 Policy = join::IoDefaultPolicy>
class join::BasicProactorThread
#else
class join::BasicProactorThread
#endif
{
public:
#ifdef JOIN_HAS_IO_URING
    static BasicProactor<Policy>& 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<Policy> _proactor;
#else
    BasicProactor _proactor;
#endif
 
    Thread _dispatcher;
};
 
#endif