current_thread.hpp

//                  ReactivePlusPlus library
//
//          Copyright Aleksey Loginov 2023 - present.
// Distributed under the Boost Software License, Version 1.0.
//    (See accompanying file LICENSE_1_0.txt or copy at
//          https://www.boost.org/LICENSE_1_0.txt)
//
// Project home: https://github.com/AlexInLog/ReactivePlusPlus
//
 
#pragma once
 
#include <rpp/schedulers/fwd.hpp>
 
#include <rpp/schedulers/details/queue.hpp>
#include <rpp/schedulers/details/utils.hpp>
#include <rpp/schedulers/details/worker.hpp>
#include <rpp/utils/functors.hpp>
 
namespace rpp::schedulers
{
    class current_thread
    {
    public:
        friend class new_thread;
        class worker_strategy;
 
        static details::schedulables_queue<worker_strategy>*& get_queue()
        {
            thread_local details::schedulables_queue<worker_strategy>* s_queue{};
            return s_queue;
        }
 
        struct is_queue_is_empty
        {
            const details::schedulables_queue<worker_strategy>& queue;
 
            bool operator()() const { return queue.is_empty(); }
        };
 
 
        static void drain_queue() noexcept
        {
            while (get_queue() && !get_queue()->is_empty())
            {
                auto top = get_queue()->pop();
                if (top->is_disposed())
                    continue;
 
                details::sleep_until(top->get_timepoint());
 
                while (true)
                {
                    if (const auto res = top->make_advanced_call())
                    {
                        if (!top->is_disposed())
                        {
                            if (get_queue()->is_empty())
                            {
                                if (const auto d = std::get_if<delay_from_now>(&res->get()))
                                {
                                    std::this_thread::sleep_for(d->value);
                                }
                                else
                                {
                                    details::sleep_until(top->handle_advanced_call(res.value()));
                                }
                                continue;
                            }
                            const auto tp = top->handle_advanced_call(res.value());
                            get_queue()->emplace(tp, std::move(top));
                        }
                    }
                    break;
                }
            }
 
            get_queue() = nullptr;
        }
 
        class worker_strategy
        {
        public:
            template<rpp::schedulers::constraint::schedulable_handler Handler, typename... Args, constraint::schedulable_fn<Handler, Args...> Fn>
            static void defer_for(duration duration, Fn&& fn, Handler&& handler, Args&&... args)
            {
                if (handler.is_disposed())
                    return;
 
                if (!get_queue())
                {
                    details::schedulables_queue<worker_strategy> queue{};
                    get_queue() = &queue;
 
                    const auto timepoint = details::immediate_scheduling_while_condition<worker_strategy>(duration, is_queue_is_empty{queue}, fn, handler, args...);
                    if (!timepoint || handler.is_disposed())
                        return drain_queue();
 
                    get_queue()->emplace(timepoint.value(), std::forward<Fn>(fn), std::forward<Handler>(handler), std::forward<Args>(args)...);
                    return drain_queue();
                }
 
                get_queue()->emplace(now() + duration, std::forward<Fn>(fn), std::forward<Handler>(handler), std::forward<Args>(args)...);
            }
 
            template<rpp::schedulers::constraint::schedulable_handler Handler, typename... Args, constraint::schedulable_fn<Handler, Args...> Fn>
            static void defer_to(time_point tp, Fn&& fn, Handler&& handler, Args&&... args)
            {
                if (handler.is_disposed())
                    return;
 
                if (get_queue())
                {
                    get_queue()->emplace(tp, std::forward<Fn>(fn), std::forward<Handler>(handler), std::forward<Args>(args)...);
                }
                else
                {
                    defer_for(tp - now(), std::forward<Fn>(fn), std::forward<Handler>(handler), std::forward<Args>(args)...);
                }
            }
 
            static rpp::schedulers::time_point now() { return details::now(); }
        };
 
    private:
        class own_queue_guard
        {
        public:
            own_queue_guard()
                : m_clear_on_destruction{!get_queue()}
            {
                if (m_clear_on_destruction)
                    get_queue() = &m_queue;
            }
            ~own_queue_guard()
            {
                if (m_clear_on_destruction)
                    drain_queue();
            }
            own_queue_guard(const own_queue_guard&) = delete;
            own_queue_guard(own_queue_guard&&)      = delete;
 
        private:
            details::schedulables_queue<worker_strategy> m_queue{};
            bool                                         m_clear_on_destruction{};
        };
 
    public:
        static own_queue_guard own_queue_and_drain_finally_if_not_owned()
        {
            return own_queue_guard{};
        }
 
        static rpp::schedulers::worker<worker_strategy> create_worker()
        {
            return rpp::schedulers::worker<worker_strategy>{};
        }
    };
} // namespace rpp::schedulers