reactor.hpp

Go to the documentation of this file.

 
#ifndef JOIN_CORE_REACTOR_HPP
#define JOIN_CORE_REACTOR_HPP
 
// libjoin.
#include <join/function.hpp>
#include <join/thread.hpp>
#include <join/queue.hpp>
#include <join/wheel.hpp>
#include <join/utils.hpp>
 
// C++.
#include <unordered_map>
#include <unordered_set>
#include <atomic>
#include <chrono>
#include <limits>
 
// C.
#include <sys/epoll.h>
 
namespace join
{
    class EventHandler
    {
    public:
        EventHandler () = default;
 
        EventHandler (const EventHandler& other) = default;
 
        EventHandler& operator= (const EventHandler& other) = default;
 
        EventHandler (EventHandler&& other) = default;
 
        EventHandler& operator= (EventHandler&& other) = default;
 
        virtual ~EventHandler () = default;
 
    protected:
        virtual void onEvent (int fd, uint32_t revents)
        {
            if (JOIN_UNLIKELY (revents & EPOLLERR))
            {
                onError (fd);
            }
            else if (JOIN_UNLIKELY (revents & (EPOLLRDHUP | EPOLLHUP)))
            {
                onClose (fd);
            }
            else if (JOIN_LIKELY (revents & EPOLLIN))
            {
                onReadable (fd);
            }
            else if (revents & EPOLLOUT)
            {
                onWriteable (fd);
            }
        }
 
        virtual void onReadable ([[maybe_unused]] int fd)
        {
            // do nothing.
        }
 
        virtual void onWriteable ([[maybe_unused]] int fd)
        {
            // do nothing.
        }
 
        virtual void onClose ([[maybe_unused]] int fd)
        {
            // do nothing.
        }
 
        virtual void onError ([[maybe_unused]] int fd)
        {
            // do nothing.
        }
 
        friend class Reactor;
    };
 
    class Reactor
    {
    private:
        static constexpr size_t _deletedReserve = 64;
 
        static constexpr size_t _queueSize = 1024;
 
        static constexpr size_t _maxEvents = 1024;
 
        class WheelPolicy
        {
        public:
            explicit WheelPolicy (Reactor& reactor) noexcept
            : _reactor (reactor)
            {
            }
 
            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 _reactor.isReactorThread ();
            }
 
            template <class Wheel>
            int flush ([[maybe_unused]] Wheel& wheel) noexcept
            {
                if (JOIN_UNLIKELY (!_reactor.isRunning ()))
                {
                    return -1;
                }
 
                return _reactor.wakeup ();
            }
 
        private:
            template <class Wheel>
            static uint64_t advance (Wheel& wheel) noexcept
            {
                return wheel.advance ();
            }
 
            template <class Wheel>
            static timespec* timeout (const Wheel& wheel, timespec& ts) noexcept
            {
                const std::chrono::nanoseconds next = wheel.next ();
 
                if (next == std::chrono::nanoseconds::max ())
                {
                    return nullptr;
                }
 
                ts = toTimespec (next);
 
                return &ts;
            }
 
            Reactor& _reactor;
 
            friend class Reactor;
        };
 
    public:
        using InvokeHandler = Function<void (), 64>;
 
        using Wheel = BasicWheel<Monotonic, WheelPolicy, _queueSize, 1'000>;
 
        Reactor ();
 
        Reactor (const Reactor& other) = delete;
 
        Reactor& operator= (const Reactor& other) = delete;
 
        Reactor (Reactor&& other) = delete;
 
        Reactor& operator= (Reactor&& other) = delete;
 
        ~Reactor () noexcept;
 
        int addHandler (int fd, EventHandler* handler, bool wantRead = true, bool wantWrite = false,
                        bool sync = true) noexcept;
 
        int delHandler (int fd, bool sync = true) noexcept;
 
        int invoke (InvokeHandler* fn, bool sync = true) noexcept;
 
        void run ();
 
        void stop (bool sync = true) noexcept;
 
        void waitStopped () const noexcept;
 
#ifdef JOIN_HAS_NUMA
        int mbind (int numa) const noexcept;
#endif
 
        int mlock () const noexcept;
 
        bool isRunning () const noexcept;
 
        bool isReactorThread () const noexcept;
 
        Wheel& wheel () noexcept
        {
            return _wheel;
        }
 
    private:
        enum class CommandType
        {
            Add,
            Del,
            Invoke,
            Stop
        };
 
        struct alignas (64) Command
        {
            CommandType type;
            int fd;
            uint32_t events;
            EventHandler* handler;
            std::atomic<bool>* done;
            std::error_code* errc;
            InvokeHandler* fn;
        };
 
        int registerHandler (int fd, EventHandler* handler, uint32_t events) noexcept;
 
        int unregisterHandler (int fd) noexcept;
 
        int invokeFunction (InvokeHandler* fn) noexcept;
 
        int writeCommand (const Command& cmd) noexcept;
 
        int wakeup () noexcept;
 
        void processCommand (const Command& cmd) noexcept;
 
        void readCommands () noexcept;
 
        void dispatchEvent (const epoll_event& event);
 
        void eventLoop ();
 
        bool isActive (int fd) const noexcept;
 
        Wheel _wheel{*this};
 
        int _wakeup = -1;
 
        int _epoll = -1;
 
        alignas (64) std::atomic<bool> _notified{false};
 
        LocalMem::Mpsc::Queue<Command> _commands;
 
        std::unordered_map<int, EventHandler*> _handlers;
 
        std::unordered_set<int> _deleted;
 
        std::atomic<bool> _running{false};
 
        std::atomic<pid_t> _threadId{Thread::invalidId};
    };
 
    class ReactorThread
    {
    public:
        static Reactor& reactor ();
 
        static int affinity (int core);
 
        static int affinity ();
 
        static int priority (int prio);
 
        static int priority ();
 
        static pthread_t handle ();
 
#ifdef JOIN_HAS_NUMA
        static int mbind (int numa);
#endif
 
        static int mlock ();
 
    private:
        static ReactorThread& instance ();
 
        ReactorThread ();
 
        ReactorThread (const ReactorThread& other) = delete;
 
        ReactorThread& operator= (const ReactorThread& other) = delete;
 
        ReactorThread (ReactorThread&& other) noexcept = delete;
 
        ReactorThread& operator= (ReactorThread&& other) noexcept = delete;
 
        ~ReactorThread ();
 
        Reactor _reactor;
 
        Thread _dispatcher;
    };
}
 
#endif