eventloop.h

Go to the documentation of this file.

 
#pragma once
 
 
#include <promeki/config.h>
#if PROMEKI_ENABLE_CORE
#include <variant>
#include <functional>
#include <promeki/function.h>
#include <promeki/namespace.h>
#include <promeki/atomic.h>
#include <promeki/duration.h>
#include <promeki/hashmap.h>
#include <promeki/mutex.h>
#include <promeki/queue.h>
#include <promeki/list.h>
#include <promeki/string.h>
#include <promeki/stringregistry.h>
#include <promeki/timestamp.h>
#include <promeki/event.h>
#include <promeki/uniqueptr.h>
 
PROMEKI_NAMESPACE_BEGIN
 
class ObjectBase;
 
// Defined in eventloop.cpp; owns the platform-specific wake file
// descriptor (eventfd on Linux, self-pipe elsewhere) used to
// interrupt poll() when new work is posted.
class EventLoopWakeFd;
 
// Defined in eventloop.cpp; caches the poll() descriptor set so a
// loop servicing many I/O sources doesn't rebuild an N-entry array
// on every wake.  Holds POSIX pollfd state we don't want to leak
// into this header; rebuilt only when a source is added or removed.
class EventLoopPollCache;
 
class EventLoop {
        public:
                enum ProcessEventsFlag : uint32_t {
                        ExcludeTimers = 0x01, 
                        ExcludePosted = 0x02, 
                        WaitForMore = 0x04    
                };
 
                using Label = StringRegistry<"EventLoopLabel">::Item;
 
                static EventLoop *current();
 
                EventLoop();
 
                ~EventLoop();
 
                EventLoop(const EventLoop &) = delete;
                EventLoop &operator=(const EventLoop &) = delete;
 
                int exec();
 
                void processEvents(uint32_t flags = 0, unsigned int timeoutMs = 0);
 
                void quit(int returnCode = 0);
 
                void postCallable(Function<void()> func);
 
                void postCallable(Label label, Function<void()> func);
 
                void postCallable(ObjectBase *owner, Label label, Function<void()> func);
 
                void postEvent(ObjectBase *receiver, Event *event);
 
                int startTimer(ObjectBase *receiver, unsigned int intervalMs, bool singleShot = false);
 
                int startTimer(unsigned int intervalMs, Function<void()> func, bool singleShot = false);
 
                void stopTimer(int timerId);
 
                bool isRunning() const { return _running.value(); }
 
                int exitCode() const { return _exitCode.value(); }
 
                unsigned int nextTimerTimeout() const;
 
                static constexpr uint32_t IoRead = 0x01;  
                static constexpr uint32_t IoWrite = 0x02; 
                static constexpr uint32_t IoError = 0x04; 
                using IoCallback = Function<void(int fd, uint32_t events)>;
 
                int addIoSource(int fd, uint32_t events, IoCallback cb);
 
                void removeIoSource(int handle);
 
                struct Report {
                        struct EventStat {
                                Duration elapsed;
                                int64_t  count = 0;
                        };
 
                        String   loopName;          
                        Duration wallElapsed;       
                        Duration sleep;             
                        Duration queueWait;         
                        Duration timers;            
                        Duration events;            
                        Duration callables;         
                        Duration io;                
                        Duration overhead;          
 
                        int64_t  timersCount    = 0; 
                        int64_t  eventsCount    = 0; 
                        int64_t  callablesCount = 0; 
                        int64_t  ioCount        = 0; 
 
                        HashMap<int, EventStat> eventsByType;
 
                        HashMap<uint64_t, EventStat> callablesByLabel;
                };
 
                using ReportFunction = Function<void(const Report &)>;
 
                void setName(const String &name);
 
                String name() const;
 
                void installMonitor(const Duration &interval, ReportFunction fn = {});
 
                void removeMonitor();
 
                bool hasMonitor() const;
 
                Report peekStats() const;
 
                Report consumeStats();
 
        private:
                struct CallableItem {
                                Function<void()> func;
                                uint64_t              labelId = Label::InvalidID;
                };
                struct EventItem {
                                ObjectBase *receiver;
                                Event      *event;
                };
                struct QuitItem {
                                int code;
                };
                using Item = std::variant<CallableItem, EventItem, QuitItem>;
 
                struct TimerInfo {
                                int                   id;
                                ObjectBase           *receiver; 
                                Function<void()> func;     
                                unsigned int          intervalMs;
                                bool                  singleShot;
                                TimeStamp             nextFire;
                };
 
                static thread_local EventLoop *_current;
 
                Queue<Item>  _queue;
                Atomic<bool> _running;
                Atomic<int>  _exitCode;
 
                // Timer list access is guarded by _timersMutex.  Any
                // thread may install or stop timers via startTimer /
                // stopTimer, so the mutex is acquired on every touch.
                // processTimers() takes a snapshot of the ready-to-fire
                // entries under the lock, releases the lock, and then
                // invokes callbacks — this avoids deadlocks if a timer
                // callback calls startTimer() or stopTimer() on the
                // same event loop, and keeps the lock hold time bounded.
                mutable Mutex   _timersMutex;
                List<TimerInfo> _timers;
                Atomic<int>     _nextTimerId{1};
 
                // Platform wake fd (eventfd on Linux, self-pipe
                // elsewhere).  Owned by the EventLoop; opened in the
                // constructor, closed in the destructor.  Written by
                // postCallable / postEvent / quit / startTimer /
                // stopTimer to unblock poll() in waitOnSources, and
                // included in every poll set as index 0.
                using WakeFdUPtr = UniquePtr<EventLoopWakeFd>;
                WakeFdUPtr _wake;
 
                // I/O source registration.  Mutation under _ioMutex,
                // poll set built under _ioMutex into a short-lived
                // stack copy, callbacks fired after the lock is
                // released so a callback may call addIoSource /
                // removeIoSource on the same EventLoop without
                // deadlocking.
                struct IoSource {
                                int        handle;
                                int        fd;
                                uint32_t   events;
                                IoCallback cb;
                                bool       pendingRemove = false;
                };
                mutable Mutex  _ioMutex;
                List<IoSource> _ioSources;
                Atomic<int>    _nextIoHandle{1};
 
                // Cached poll() descriptor set + dirty flag (see
                // EventLoopPollCache in eventloop.cpp).  Rebuilt under
                // _ioMutex only when an IoSource is added or removed,
                // then reused across waitOnSources iterations so the
                // steady-state wake path costs a lock + flag check
                // rather than an O(number-of-sources) array rebuild.
                using PollCacheUPtr = UniquePtr<EventLoopPollCache>;
                PollCacheUPtr _pollCache;
 
                void wakeSelf();
 
                bool dispatchItem(Item &item);
                void processTimers();
 
                // Waits on the wake fd + registered I/O sources via
                // poll(), for up to @p waitMs milliseconds (0 = wait
                // indefinitely — callers clamp by timers before
                // calling).  On return, drains the wake fd and fires
                // any ready I/O source callbacks.  On non-POSIX
                // platforms, falls back to the condvar-based
                // Queue::pop wait so Windows / Emscripten builds
                // still compile and behave as before (minus
                // IoSource support).
                void waitOnSources(unsigned int waitMs);
 
                // ----------------------------------------------------
                // Stats / monitor support.  The hot-path bracket reads
                // _monitorActive once at construction; when false it
                // skips the timestamp grab, the lock, and the
                // accumulator update.  When true it stamps once on
                // construction and once on destruction, takes the
                // mutex, adds the elapsed nanoseconds + 1 to the
                // attributed duration / count buckets, and updates
                // _eventsByType when an event-type is supplied.
                // ----------------------------------------------------
 
                struct EventStatNs {
                                int64_t elapsed = 0;
                                int64_t count   = 0;
                };
 
                Atomic<bool>                _monitorActive;
                mutable Mutex               _statsMutex;
                TimeStamp                   _statsLastSnapshot;
 
                int64_t                     _sleepNs        = 0;
                int64_t                     _queueWaitNs    = 0;
                int64_t                     _timersNs       = 0;
                int64_t                     _eventsNs       = 0;
                int64_t                     _callablesNs    = 0;
                int64_t                     _ioNs           = 0;
 
                int64_t                     _timersCount    = 0;
                int64_t                     _eventsCount    = 0;
                int64_t                     _callablesCount = 0;
                int64_t                     _ioCount        = 0;
 
                HashMap<int, EventStatNs>   _eventsByType;
 
                HashMap<uint64_t, EventStatNs> _callablesByLabel;
 
                int                         _monitorTimerId = 0;
                ReportFunction              _monitorFn;
                String                      _name;
 
                // RAII helper that brackets a single dispatch site.
                // Reads _monitorActive ONCE in the constructor and
                // caches the boolean; both the constructor's
                // TimeStamp::now() grab and the destructor's
                // accumulator update key off that cached value so
                // an enable-mid-bracket cannot leave the destructor
                // trying to attribute time without a start
                // timestamp.  When the cached gate is false, no
                // timestamp is stored and the destructor early-outs
                // without locking.
                class StatsBracket {
                                public:
                                        StatsBracket(EventLoop *loop, int64_t *durBucket,
                                                     int64_t *countBucket);
                                        ~StatsBracket();
                                        void attributeEventType(int type) { _eventType = type; }
                                        void attributeCallableLabel(uint64_t labelId) {
                                                _callableLabel = labelId;
                                        }
 
                                private:
                                        EventLoop  *_loop;
                                        int64_t    *_durBucket;
                                        int64_t    *_countBucket;
                                        TimeStamp   _start;
                                        int         _eventType = -1;
                                        uint64_t    _callableLabel = Label::InvalidID;
                                        bool        _active = false;
                };
 
                friend class StatsBracket;
 
                // Default formatter used when _monitorFn is empty.
                // Defined in eventloop.cpp and exposed via
                // installMonitor's "empty fn = default" semantics.
                static void defaultMonitorReporter(const Report &r);
};
 
PROMEKI_NAMESPACE_END
 
#endif // PROMEKI_ENABLE_CORE