Scheduler.hpp

// SPDX-FileCopyrightText: 2025 Vincent Leroy
// SPDX-License-Identifier: MIT
//
// This file is part of dfx.
//
// Licensed under the MIT License. See the LICENSE file in the project root
// for full license information.
 
#pragma once
 
// Standard includes
#include <mutex>
 
// Project includes
#include "FdListener.hpp"
#include "ThreadPool.hpp"
#include <dfx-fdwatch/BorrowedFd.hpp>
#include <dfx-fdwatch/OwnedFd.hpp>
#include <dfx-hooks/Delivery.hpp>
#include <dfx-hooks/Graph.hpp>
#include <dfx-runtime-api/NodeReactor.hpp>
#include <dfx-utilities/CopyMoveControl.hpp>
#include <dfx-utilities/SystemConfig.hpp>
 
namespace dfx::Runtime
{
class Scheduler : public Hooks::Delivery, public Hooks::Graph, public Api::NodeReactor
{
private:
    struct FdData
    {
        FdWatch::BorrowedFd fd;
        FdWatch::FdCallback cb;
        Core::NodeWPtr node;
    };
 
    struct DeferedCbData
    {
        FdWatch::Callback callback;
        Core::NodeWPtr node;
    };
 
    using FdDatas = std::vector<FdData>;
 
public:
    Scheduler(Utils::SystemConfig & sysConfig);

    DISABLE_COPY_AND_MOVE(Scheduler);

    ~Scheduler() override;
 
public:
    template<DerivedFromTask T, typename ... Args>
    void emplaceTask(Args && ... args)
    { _threadPool.emplaceTask<T>(std::forward<Args>(args)...); }

    void pushTask(TaskPtr task)
    { _threadPool.pushTask(std::move(task)); }
 
public:
    void start();

    void stop();

    bool isRunning() const noexcept { return _isRunning; }
 
public:
    HookResult preDelivery(Core::NodePtr src, Core::OutputPort const & out,
                           Core::NodePtr dst, Core::InputPort & in,
                           Core::MessagePtr & message) override;

    HookResult postDelivery(Core::NodePtr src, Core::OutputPort const & out,
                            Core::NodePtr dst, Core::InputPort & in) override;

    void registerNodeFd(FdWatch::BorrowedFd fd, FdWatch::EventInterests events, FdWatch::FdCallback cb, Core::NodeWPtr node) override;

    void deregisterNodeFd(FdWatch::BorrowedFd fd) noexcept override;

    void deferNodeCall(FdWatch::Callback cb, Core::NodeWPtr node) override;

    void onNodeCreated(Core::NodePtr node) override;

    void onNodeRemoved(Core::NodePtr node) override;

    void onChannelCreated(Core::ChannelPtr channel) override;

    void onChannelRemoved(Core::ChannelPtr channel) override;
 
private:
    FdDatas::iterator _getFdData(FdWatch::BorrowedFd fd) noexcept;
    FdDatas::const_iterator _getFdData(FdWatch::BorrowedFd fd) const noexcept;
 
private:
    void _onFdTriggered(FdWatch::BorrowedFd fd, FdWatch::EventTriggers events);
 
private:
    std::atomic_bool _isRunning = false;
 
private:
    std::atomic_uint32_t _maxNesting = 0;
 
private:
    std::list<Core::NodeWPtr> _nodes;
    std::mutex                _nodeMutex;
 
private:
    FdDatas            _fdData;
    mutable std::mutex _fdMutex;
 
private:
    // Destroy the FdListener then the ThreadPool
    // in order to avoid issue with a pending task
    // that will execute code in the scheduler that
    // is actively being destroyed
    ThreadPool       _threadPool;
    FdListener       _fdListener;
};
} // !namespace dfx::Runtime