Runtime.hpp

Go to the documentation of this file.

 
#pragma once
 
#include <atomic>
#include <chrono>
#include <cstddef>
#include <limits>
#include <utility>
 
#include "corium/ApplicationContext.hpp"
#include "corium/EventBus.hpp"
#include "corium/internal/VariantIndex.hpp"
#include "corium/policies/OverflowPolicies.hpp"
#include "corium/policies/QueuePolicies.hpp"
#include "corium/policies/SignalPolicies.hpp"
#include "corium/policies/StoragePolicies.hpp"
#include "corium/policies/TimerPolicies.hpp"
#include "corium/timers/TimerScheduler.hpp"
 
namespace corium {
 
template <typename Derived, typename EventVariant, std::size_t MaxServices>
class Application;
 
template <
    typename EventVariant = DefaultEvents,
    typename QueuePolicy = BoundedMpscQueuePolicy<EventVariant, 1024>,
    typename SignalPolicy = NoSignalPolicy,
    typename StoragePolicy = DefaultStoragePolicy,
    typename OverflowPolicy = DropNewestOverflowPolicy,
    typename TimerStoragePolicy = DefaultTimerStoragePolicy,
    typename ProfilerPolicy = profiler::NullProfiler
>
class BasicRuntime {
public:
    using EventVariantType = EventVariant;
    using EventType = EventVariant;
    using EventBusType = BasicEventBus<EventVariant, QueuePolicy, SignalPolicy, StoragePolicy, OverflowPolicy, ProfilerPolicy>;
    using ClockPolicyType = typename internal::get_timer_clock_policy<TimerStoragePolicy>::type;
    using TimerSchedulerType = TimerScheduler<EventVariant, TimerStoragePolicy::max_timers, ClockPolicyType>;
    using ProfilerPolicyType = ProfilerPolicy;
 
    enum class State : uint8_t {
        Created,
        Initializing,
        Running,
        Stopping,
        Terminated
    };
 
    BasicRuntime()
        : _eventBus(),
          _state(State::Created),
          _quitRequested(false)
    {
    }
 
    ~BasicRuntime()
    {
        shutdown();
    }
 
    BasicRuntime(const BasicRuntime&) = delete;
    BasicRuntime& operator=(const BasicRuntime&) = delete;
 
    [[nodiscard]] State state() const noexcept
    {
        return _state.load(std::memory_order_acquire);
    }
 
    void detachApplication() noexcept
    {
        _appShutdownCb = StaticCallback{};
    }
 
    template <typename Derived, typename AppEvents = EventVariant, std::size_t MaxServices = 8>
    void initialize(corium::Application<Derived, AppEvents, MaxServices>& application)
    {
        using AppEventVariant = typename corium::Application<Derived, AppEvents, MaxServices>::EventVariant;
        static_assert(std::is_same_v<AppEventVariant, EventVariant>,
            "Application EventVariant list must match Runtime EventVariant list!");
 
        _state.store(State::Initializing, std::memory_order_release);
        _appShutdownCb = StaticCallback{
            [](void* appPtr) {
                auto* app = static_cast<Derived*>(static_cast<corium::Application<Derived, AppEvents, MaxServices>*>(appPtr));
                app->shutdownServices();
                app->shutdown();
                app->resetContext();
            },
            &application
        };
 
        auto ctx = applicationContext();
        ctx.setTimerScheduler(_timerScheduler);
        ctx.setRuntimeDetach(this, [](void* rt) noexcept {
            static_cast<BasicRuntime*>(rt)->detachApplication();
        });
        application.setContext(ctx);
 
        registerCoreHandlers();
        application.registerHandlers();
        _eventBus.seal();
 
        application.initializeServices(applicationContext().eventSink());
        application.initialize();
 
        _state.store(State::Running, std::memory_order_release);
    }
 
    void pump()
    {
        pump((std::numeric_limits<std::size_t>::max)());
    }
 
    void pump(std::size_t maxEvents)
    {
        _timerScheduler.processDueTimers(_eventBus);
 
        std::size_t processed = 0;
        while (_state.load(std::memory_order_relaxed) == State::Running && !_quitRequested && processed < maxEvents) {
            if (!_eventBus.processOne()) {
                break;
            }
            processed++;
        }
    }
 
    template <typename Rep, typename Period>
    std::size_t waitAndPump(const std::chrono::duration<Rep, Period>& timeout)
    {
        _timerScheduler.processDueTimers(_eventBus);
 
        if (_eventBus.empty() && !_quitRequested) {
            _eventBus.signalPolicy().wait_for(timeout);
        }
 
        std::size_t processed = 0;
        while (_state.load(std::memory_order_relaxed) == State::Running && !_quitRequested) {
            if (!_eventBus.processOne()) {
                break;
            }
            processed++;
        }
        return processed;
    }
 
    std::size_t pumpBatch(std::size_t batchSize = 16, std::size_t maxTotal = (std::numeric_limits<std::size_t>::max)())
    {
        _timerScheduler.processDueTimers(_eventBus);
 
        std::size_t total = 0;
        while (_state.load(std::memory_order_relaxed) == State::Running && !_quitRequested && total < maxTotal) {
            std::size_t toProcess = (std::min)(batchSize, maxTotal - total);
            std::size_t processed = _eventBus.processBatch(toProcess);
            total += processed;
            if (processed < toProcess) {
                break;
            }
        }
        return total;
    }
 
    std::size_t drain()
    {
        _timerScheduler.processDueTimers(_eventBus);
        if (_state.load(std::memory_order_relaxed) != State::Running || _quitRequested) {
            return 0;
        }
        return _eventBus.drain();
    }
 
    template <typename Rep, typename Period>
    TimerId scheduleDelayed(EventVariant event, const std::chrono::duration<Rep, Period>& delay, EventPriority priority = EventPriority::Normal)
    {
        return _timerScheduler.scheduleDelayed(std::move(event), delay, priority);
    }
 
    template <typename DurationType>
    TimerId scheduleDelayed(EventVariant event, DurationType delay, EventPriority priority = EventPriority::Normal)
        requires (!std::is_same_v<DurationType, std::chrono::microseconds> && !std::is_same_v<DurationType, std::chrono::milliseconds>)
    {
        return _timerScheduler.scheduleDelayed(std::move(event), delay, priority);
    }
 
    template <typename Rep, typename Period>
    TimerId schedulePeriodic(EventVariant event, const std::chrono::duration<Rep, Period>& interval, EventPriority priority = EventPriority::Normal)
    {
        return _timerScheduler.schedulePeriodic(std::move(event), interval, priority);
    }
 
    template <typename DurationType>
    TimerId schedulePeriodic(EventVariant event, DurationType interval, EventPriority priority = EventPriority::Normal)
        requires (!std::is_same_v<DurationType, std::chrono::microseconds> && !std::is_same_v<DurationType, std::chrono::milliseconds>)
    {
        return _timerScheduler.schedulePeriodic(std::move(event), interval, priority);
    }
 
    bool cancelTimer(TimerId id) noexcept
    {
        return _timerScheduler.cancelTimer(id);
    }
 
    void shutdown() noexcept
    {
        auto st = _state.load(std::memory_order_acquire);
        if (st == State::Stopping || st == State::Terminated) {
            return;
        }
 
        _state.store(State::Stopping, std::memory_order_release);
        if (_appShutdownCb) {
            auto cb = _appShutdownCb;
            _appShutdownCb = StaticCallback{};
            cb();
        }
        _state.store(State::Terminated, std::memory_order_release);
    }
 
    void requestQuit() noexcept
    {
        _quitRequested.store(true, std::memory_order_release);
    }
 
    [[nodiscard]] bool quitRequested() const noexcept
    {
        auto st = _state.load(std::memory_order_acquire);
        return _quitRequested.load(std::memory_order_acquire) || st == State::Stopping || st == State::Terminated;
    }
 
    void setOnQueueNonEmpty(StaticCallback callback)
    {
        _eventBus.setOnQueueNonEmpty(callback);
    }
 
    [[nodiscard]] SignalPolicy& signalPolicy() noexcept
    {
        return _eventBus.signalPolicy();
    }
 
    [[nodiscard]] const SignalPolicy& signalPolicy() const noexcept
    {
        return _eventBus.signalPolicy();
    }
 
    [[nodiscard]] OverflowPolicy& overflowPolicy() noexcept
    {
        return _eventBus.overflowPolicy();
    }
 
    [[nodiscard]] const OverflowPolicy& overflowPolicy() const noexcept
    {
        return _eventBus.overflowPolicy();
    }
 
    [[nodiscard]] TimerSchedulerType& timerScheduler() noexcept
    {
        return _timerScheduler;
    }
 
    [[nodiscard]] const TimerSchedulerType& timerScheduler() const noexcept
    {
        return _timerScheduler;
    }
 
    [[nodiscard]] EventSinkT<EventVariant> eventSink() noexcept
    {
        return _eventBus.sink();
    }
 
    [[nodiscard]] ProfilerPolicyType& profiler() noexcept
    {
        return _eventBus.profiler();
    }
 
    [[nodiscard]] const ProfilerPolicyType& profiler() const noexcept
    {
        return _eventBus.profiler();
    }
 
private:
    ApplicationContext<EventVariant> applicationContext()
    {
        auto ctx = ApplicationContext<EventVariant>{
            _eventBus,
            StaticCallback{
                [](void* c) { static_cast<BasicRuntime*>(c)->requestQuit(); },
                this
            }
        };
        ctx.setTimerScheduler(_timerScheduler);
        return ctx;
    }
 
    void registerCoreHandlers()
    {
        if constexpr (has_variant_type_v<QuitEvent, EventVariant>) {
            _eventBus.template registerHandler<QuitEvent>([this](const QuitEvent&) {
                _quitRequested.store(true, std::memory_order_release);
            });
        }
    }
 
    EventBusType _eventBus;
    TimerSchedulerType _timerScheduler{};
    StaticCallback _appShutdownCb;
    std::atomic<State> _state{State::Created};
    std::atomic<bool> _quitRequested{false};
};
 
using Runtime = BasicRuntime<DefaultEvents, BoundedMpscQueuePolicy<DefaultEvents, 1024>, NoSignalPolicy, DefaultStoragePolicy, DropNewestOverflowPolicy, DefaultTimerStoragePolicy, profiler::NullProfiler>;
 
template <
    typename EventVariant = DefaultEvents,
    typename QueuePolicy = BoundedMpscQueuePolicy<EventVariant, 1024>,
    typename SignalPolicy = NoSignalPolicy,
    typename StoragePolicy = DefaultStoragePolicy,
    typename OverflowPolicy = DropNewestOverflowPolicy,
    typename TimerStoragePolicy = DefaultTimerStoragePolicy,
    typename ProfilerPolicy = profiler::NullProfiler
>
using RuntimeT = BasicRuntime<EventVariant, QueuePolicy, SignalPolicy, StoragePolicy, OverflowPolicy, TimerStoragePolicy, ProfilerPolicy>;
 
} // namespace corium
 
#include "corium/RuntimeBuilder.hpp"