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/utils.hpp>
 
// C++.
#include <unordered_map>
#include <unordered_set>
#include <atomic>
 
// 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
    {
    public:
        using InvokeHandler = Function<void (), 64>;
 
        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;
 
    private:
        static constexpr size_t _deletedReserve = 64;
 
        static constexpr size_t _queueSize = 1024;
 
        static constexpr size_t _maxEvents = 1024;
 
        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;
 
        void processCommand (const Command& cmd) noexcept;
 
        void readCommands () noexcept;
 
        void dispatchEvent (const epoll_event& event);
 
        void eventLoop ();
 
        bool isActive (int fd) const noexcept;
 
        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};
 
        static constexpr pthread_t _invalidThreadId = static_cast<pthread_t> (-1);
 
        std::atomic<pthread_t> _threadId{_invalidThreadId};
    };
 
    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