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>
#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/queue.hpp>
 
// C++.
#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:
    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;
 
#ifdef JOIN_HAS_IO_URING
    int flush (bool sync = false) noexcept;
#endif
 
    void run ();
 
    void stop (bool sync = true) noexcept;
 
#ifdef JOIN_HAS_IO_URING
    int registerBuffers (const std::vector<iovec>& iovecs) noexcept;
 
    int unregisterBuffers () noexcept;
#endif
 
#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 CommandType
    {
        Submit, 
        Cancel, 
        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;   
    };
 
    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;
 
    void dispatchOperation (IoOperation* op, int result, bool cancelled) noexcept;
 
    void endOperation (IoOperation* op, int result, bool cancelled = false) noexcept;
 
#ifdef JOIN_HAS_IO_URING
    io_uring_sqe* getSqe () noexcept;
 
    void prepareSqe (io_uring_sqe* sqe, IoOperation* op) noexcept;
 
    void rearmWakeup () 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 (uint8_t code) noexcept;
 
    static int executeOp (IoOperation* op) noexcept;
 
    void onReadable (int fd) noexcept override;
 
    void onWriteable (int fd) noexcept override;
 
    void onClose (int fd) noexcept override;
 
    void onError (int fd) noexcept override;
#endif
 
    static constexpr size_t _queueSize = 1024;
 
    LocalMem::Mpsc::Queue<Command> _commands;
 
    std::atomic<bool> _stopping{false};
 
    int _wakeup = -1;
 
#ifdef JOIN_HAS_IO_URING
    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
};
 
// =========================================================================
//   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}) == -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}) == -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    : 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
    }
 
    if (JOIN_LIKELY (op->handler))
    {
        if (cancelled)
        {
            op->handler->onCancel (op, result);
        }
        else
        {
            op->handler->onComplete (op, result);
        }
    }
 
    op->state = IoOperation::State::Idle;
}
 
#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 ();
        });
    }
 
    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