Scheduler.hpp

// SPDX-FileCopyrightText: 2025-2026 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-utilities/BorrowedFd.hpp>
#include <dfx-utilities/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
    {
        Utils::BorrowedFd fd;
        FdWatch::FdCallback cb;
        Core::NodeWPtr node;
        FdWatch::EventInterests currentInterests;
        bool isArmed;
    };
 
    struct DeferedCbData
    {
        FdWatch::Callback callback;
        Core::NodeWPtr node;
    };
 
    using FdDatas = std::vector<FdData>;
 
public:
    explicit Scheduler(Utils::SystemConfig & sysConfig);

    DFX_DISABLE_COPY_AND_MOVE(Scheduler);

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

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

    void stop();

    bool isRunning() const noexcept { return _isRunning; }
 
public:
    HookResult preDelivery(Core::Endpoint const & src, Core::Endpoint const & dst,
                           Core::MessagePtr & message) override;

    HookResult postDelivery(Core::Endpoint const & src, Core::Endpoint const & dst) override;

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

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

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

    void updateNodeFdEvents(Utils::BorrowedFd fd, FdWatch::EventInterests events) 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:
    void _deliverTaskToNode(Api::TaskPtr task, Core::NodeWPtr node);
    void _deliverTasksToNode(std::span<Api::TaskPtr> tasks, Core::NodeWPtr node);
 
    FdDatas::iterator _getFdData(Utils::BorrowedFd fd) noexcept;
    FdDatas::const_iterator _getFdData(Utils::BorrowedFd fd) const noexcept;
 
private:
    void _onFdTriggered(std::vector<FdListener::FdInfo> fdInfos);
 
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