Main.h

Go to the documentation of this file.

 
#ifndef QB_MAIN_H
#define QB_MAIN_H
#include <condition_variable>
#include <qb/system/container/unordered_map.h>
#include <thread>
#include <vector>
// include from qb
#include <qb/system/lockfree/mpsc.h>
#include "CoreSet.h"
#include "Event.h"
 
namespace qb {
 
class Main;
class VirtualCore;
class IActorFactory;

constexpr const CoreId NoAffinity = std::numeric_limits<CoreId>::max();

class CoreInitializer : nocopy {
    friend class Main;
    friend class VirtualCore;
 
public:
    class ActorBuilder {
    public:
        using ActorIdList = std::vector<ActorId>;
 
    private:
        friend class CoreInitializer;
 
        CoreInitializer &_initializer;
        ActorIdList      _ret_ids;
        bool             _valid;

        explicit ActorBuilder(CoreInitializer &initializer) noexcept;
 
    public:
        ActorBuilder()                        = delete;
        ActorBuilder(ActorBuilder const &rhs) = default;

        template <typename _Actor, typename... _Args>
        ActorBuilder &addActor(_Args &&...args) noexcept;

        [[nodiscard]] bool valid() const noexcept;
        explicit           operator bool() const noexcept;

        [[nodiscard]] ActorIdList idList() const noexcept;
    };
 
private:
    const CoreId _index;
    ServiceId    _next_id;
    CoreIdSet    _affinity;
    uint64_t     _latency;
 
    qb::unordered_set<ServiceId> _registered_services;
    std::vector<IActorFactory *> _actor_factories;
    //        CoreSet _restricted_communication; future use
 
public:
    CoreInitializer() = delete;
    explicit CoreInitializer(CoreId index);
    ~CoreInitializer() noexcept;

    void clear() noexcept;

    template <typename _Actor, typename... _Args>
    ActorId addActor(_Args &&...args) noexcept;

    [[nodiscard]] ActorBuilder builder() noexcept;

    CoreInitializer &setAffinity(CoreIdSet const &cores = {}) noexcept;

    CoreInitializer &setLatency(uint64_t latency = 0) noexcept;

    [[nodiscard]] CoreId           getIndex() const noexcept;
    [[nodiscard]] CoreIdSet const &getAffinity() const noexcept;
    [[nodiscard]] uint64_t         getLatency() const noexcept;
};

using CoreInitializerMap = qb::unordered_map<CoreId, CoreInitializer>;

class SharedCoreCommunication : nocopy {
    friend class VirtualCore;
    friend class Main;
    constexpr static const uint64_t MaxRingEvents =
        (((std::numeric_limits<uint16_t>::max)()) / QB_LOCKFREE_EVENT_BUCKET_BYTES);
    class Mailbox : public lockfree::mpsc::ringbuffer<EventBucket, MaxRingEvents, 0> {
        const uint64_t          _latency;
        std::mutex              _mtx;
        std::condition_variable _cv;
 
    public:
        explicit Mailbox(std::size_t const nb_producer, uint64_t const latency)
            : lockfree::mpsc::ringbuffer<EventBucket, MaxRingEvents, 0>(nb_producer)
            , _latency(latency) {}

        void
        wait() noexcept {
            if (_latency) {
                std::unique_lock lk(_mtx);
                _cv.wait_for(lk, std::chrono::nanoseconds(_latency));
            }
        }

        void
        notify() noexcept {
            if (_latency)
                _cv.notify_all();
        }

        [[nodiscard]] uint64_t
        getLatency() const noexcept {
            return _latency;
        }
    };
 
    const CoreSet                  _core_set;
    std::vector<std::atomic<bool>> _event_safe_deadlock;
    std::vector<Mailbox *>         _mail_boxes;
 
public:
    SharedCoreCommunication() = delete;
    explicit SharedCoreCommunication(
        CoreInitializerMap const &core_initializers) noexcept;
 
    ~SharedCoreCommunication() noexcept;

    [[nodiscard]] bool send(Event const &event) const noexcept;

    [[nodiscard]] Mailbox &getMailBox(CoreId id) const noexcept;

    [[nodiscard]] CoreId getNbCore() const noexcept;
};

struct CoreSpawnerParameter {
    const CoreId             id;
    
    CoreInitializer         &initializer;
    
    SharedCoreCommunication &shared_com;
    
    std::atomic<uint64_t>   &sync_start;
};

class Main {
    friend class VirtualCore;
    constexpr static const uint64_t MaxRingEvents =
        (((std::numeric_limits<uint16_t>::max)()) / QB_LOCKFREE_EVENT_BUCKET_BYTES);
    using Mailbox = lockfree::mpsc::ringbuffer<EventBucket, MaxRingEvents, 0>;
 
    static std::vector<Main *> _instances;
    static std::mutex          _instances_lock;
 
    std::atomic<uint64_t> _sync_start;
    static void           onSignal(int signal) noexcept;
    static void           start_thread(CoreSpawnerParameter const &params) noexcept;
    static bool           __wait__all__cores__ready(std::size_t            nb_core,
                                                    std::atomic<uint64_t> &sync_start) noexcept;
 
private:
    std::vector<std::thread> _cores;
    // Core Factory
    CoreInitializerMap                       _core_initializers;
    std::unique_ptr<SharedCoreCommunication> _shared_com;
    bool                                     _is_running;
 
public:
    using ActorIdList = CoreInitializer::ActorBuilder::ActorIdList;

    Main() noexcept;
    ~Main() noexcept;

    void start(bool async = true) noexcept;

    [[nodiscard]] bool hasError() const noexcept;

    static void stop() noexcept;

    void join();
 
public:
    template <typename _Actor, typename... _Args>
    ActorId addActor(CoreId index, _Args &&...args);

    [[nodiscard]] CoreInitializer &core(CoreId index);

    void setLatency(uint64_t latency = 0);

    [[nodiscard]] qb::CoreIdSet usedCoreSet() const;

    static void registerSignal(int signum) noexcept;
    static void unregisterSignal(int signum) noexcept;
    static void ignoreSignal(int signum) noexcept;
};

using engine = Main;
 
} // namespace qb
#endif // QB_MAIN_H