src/test/csf/Scheduler.h

#pragma once
 
#include <xrpl/basics/ByteUtilities.h>
#include <xrpl/beast/clock/manual_clock.h>
 
#include <boost/container/pmr/monotonic_buffer_resource.hpp>
#include <boost/intrusive/set.hpp>
 
#include <type_traits>
#include <utility>
 
namespace xrpl::test::csf {

class Scheduler
{
public:
    using clock_type = beast::ManualClock<std::chrono::steady_clock>;
 
    using duration = clock_type::duration;
 
    using time_point = clock_type::time_point;
 
private:
    using by_when_hook =
        boost::intrusive::set_base_hook<boost::intrusive::link_mode<boost::intrusive::normal_link>>;
 
    struct Event : by_when_hook
    {
        time_point when;
 
        Event(Event const&) = delete;
        Event&
        operator=(Event const&) = delete;
 
        virtual ~Event() = default;
 
        // Called to perform the event
        virtual void
        operator()() const = 0;
 
        Event(time_point when) : when(when)
        {
        }
 
        bool
        operator<(Event const& other) const
        {
            return when < other.when;
        }
    };
 
    template <class Handler>
    class EventImpl : public Event
    {
        Handler const h_;
 
    public:
        EventImpl(EventImpl const&) = delete;
 
        EventImpl&
        operator=(EventImpl const&) = delete;
 
        template <class DeducedHandler>
        EventImpl(time_point when, DeducedHandler&& h)
            : Event(when), h_(std::forward<DeducedHandler>(h))
        {
        }
 
        void
        operator()() const override
        {
            h_();
        }
    };
 
    class QueueType
    {
    private:
        using by_when_set = boost::intrusive::
            make_multiset<Event, boost::intrusive::constant_time_size<false>>::type;
        // alloc_ is owned by the scheduler
        boost::container::pmr::monotonic_buffer_resource* alloc_;
        by_when_set byWhen_;
 
    public:
        using iterator = by_when_set::iterator;
 
        QueueType(QueueType const&) = delete;
        QueueType&
        operator=(QueueType const&) = delete;
 
        explicit QueueType(boost::container::pmr::monotonic_buffer_resource* alloc);
 
        ~QueueType();
 
        [[nodiscard]] bool
        empty() const;
 
        iterator
        begin();
 
        iterator
        end();
 
        template <class Handler>
        by_when_set::iterator
        emplace(time_point when, Handler&& h);
 
        iterator
        erase(iterator iter);
    };
 
    boost::container::pmr::monotonic_buffer_resource alloc_{kilobytes(256)};
    QueueType queue_;
 
    // Aged containers that rely on this clock take a non-const reference =(
    mutable clock_type clock_;
 
public:
    Scheduler(Scheduler const&) = delete;
    Scheduler&
    operator=(Scheduler const&) = delete;
 
    Scheduler();

    clock_type&
    clock() const;

    time_point
    now() const;
 
    // Used to cancel timers
    struct CancelToken;

    template <class Function>
    CancelToken
    at(time_point const& when, Function&& f);

    template <class Function>
    CancelToken
    in(duration const& delay, Function&& f);

    void
    cancel(CancelToken const& token);

    bool
    stepOne();

    bool
    step();

    template <class Function>
    bool
    stepWhile(Function&& func);

    bool
    stepUntil(time_point const& until);

    template <class Period, class Rep>
    bool
    stepFor(std::chrono::duration<Period, Rep> const& amount);
};
 
//------------------------------------------------------------------------------
 
inline Scheduler::QueueType::QueueType(boost::container::pmr::monotonic_buffer_resource* alloc)
    : alloc_(alloc)
{
}
 
inline Scheduler::QueueType::~QueueType()
{
    for (auto iter = byWhen_.begin(); iter != byWhen_.end();)
    {
        auto e = &*iter;
        ++iter;
        e->~Event();
        alloc_->deallocate(e, sizeof(e));  // NOLINT(bugprone-sizeof-expression)
    }
}
 
inline bool
Scheduler::QueueType::empty() const
{
    return byWhen_.empty();
}
 
inline auto
Scheduler::QueueType::begin() -> iterator
{
    return byWhen_.begin();
}
 
inline auto
Scheduler::QueueType::end() -> iterator
{
    return byWhen_.end();
}
 
template <class Handler>
inline auto
Scheduler::QueueType::emplace(time_point when, Handler&& h) -> by_when_set::iterator
{
    using event_type = EventImpl<std::decay_t<Handler>>;
    auto const p = alloc_->allocate(sizeof(event_type));
    auto& e = *new (p) event_type(when, std::forward<Handler>(h));
    return byWhen_.insert(e);
}
 
inline auto
Scheduler::QueueType::erase(iterator iter) -> by_when_set::iterator
{
    auto& e = *iter;
    auto next = byWhen_.erase(iter);
    e.~Event();
    alloc_->deallocate(&e, sizeof(e));
    return next;
}
 
//-----------------------------------------------------------------------------
struct Scheduler::CancelToken
{
private:
    QueueType::iterator iter_;
 
public:
    CancelToken() = delete;
    CancelToken(CancelToken const&) = default;
    CancelToken&
    operator=(CancelToken const&) = default;
 
private:
    friend class Scheduler;
    CancelToken(QueueType::iterator iter) : iter_(iter)
    {
    }
};
 
//------------------------------------------------------------------------------
inline Scheduler::Scheduler() : queue_(&alloc_)
{
}
 
inline auto
Scheduler::clock() const -> clock_type&
{
    return clock_;
}
 
inline auto
Scheduler::now() const -> time_point
{
    return clock_.now();
}
 
template <class Function>
inline auto
Scheduler::at(time_point const& when, Function&& f) -> CancelToken
{
    return queue_.emplace(when, std::forward<Function>(f));
}
 
template <class Function>
inline auto
Scheduler::in(duration const& delay, Function&& f) -> CancelToken
{
    return at(clock_.now() + delay, std::forward<Function>(f));
}
 
inline void
Scheduler::cancel(CancelToken const& token)
{
    queue_.erase(token.iter_);
}
 
inline bool
Scheduler::stepOne()
{
    if (queue_.empty())
        return false;
    auto const iter = queue_.begin();
    clock_.set(iter->when);
    (*iter)();
    queue_.erase(iter);
    return true;
}
 
inline bool
Scheduler::step()
{
    if (!stepOne())
        return false;
    for (;;)
    {
        if (!stepOne())
            break;
    }
    return true;
}
 
template <class Function>
inline bool
Scheduler::stepWhile(Function&& f)
{
    bool ran = false;
    while (f() && stepOne())
        ran = true;
    return ran;
}
 
inline bool
Scheduler::stepUntil(time_point const& until)
{
    // VFALCO This routine needs optimizing
    if (queue_.empty())
    {
        clock_.set(until);
        return false;
    }
    auto iter = queue_.begin();
    if (iter->when > until)
    {
        clock_.set(until);
        return true;
    }
    do
    {
        stepOne();
        iter = queue_.begin();
    } while (iter != queue_.end() && iter->when <= until);
    clock_.set(until);
    return iter != queue_.end();
}
 
template <class Period, class Rep>
inline bool
Scheduler::stepFor(std::chrono::duration<Period, Rep> const& amount)
{
    return stepUntil(now() + amount);
}
 
}  // namespace xrpl::test::csf