executor.hpp

// SPDX-License-Identifier: Apache-2.0
 
#pragma once
#include <chrono>
#include <condition_variable>
#include <core/async/ExecutorContext.hpp>
#include <core/async/IExecutor.hpp>
#include <core/async/ParkedWork.hpp>
#include <coroutine>
#include <exception>
#include <functional>
#include <mutex>
#include <queue>
#include <string>
#include <thread>
#include <vector>
 
#include "../attributes.hpp"
#include "logger.hpp"
#include "profiler.hpp"
 
namespace morph::exec {
 
struct IExecutor;
 
namespace detail {
 
/// @brief A `core::async::IExecutor` over a morph executor: how a core-cpp
///        strand's pump, and an `ExecutorScope`, reach a thread pool, a
///        main-thread pump or Qt.
///
/// Every `morph::exec::IExecutor` owns exactly one (`IExecutor::coreExecutor`),
/// so an executor has one core-cpp identity for its whole life.
///
/// A strand hands its base one bare coroutine handle per turn, which this
/// posts as a lambda holding nothing else: trivially copyable, so it fits a
/// `std::function`'s small buffer and a turn costs no allocation. The lambda
/// does not refer to this object, so it may be destroyed while a pump it
/// posted is still queued: the pump finds its strand closed and ends.
class CoreExecutorOver final : public ::core::async::IExecutor {
public:
    /// @param executor Where every resumption is posted. Borrowed: it must
    ///        outlive every strand over this object and run what it queued.
    explicit CoreExecutorOver(::morph::exec::IExecutor& executor MORPH_LIFETIMEBOUND) : _executor{&executor} {}
 
    using ::core::async::IExecutor::submit;
 
    /// @brief Posts @p handle's resumption.
    /// @param handle The coroutine to resume; borrowed.
    void submit(std::coroutine_handle<> handle) override;
 
    /// @brief Posts @p work's resumption. An executor that drops the task
    ///        unrun releases its claim, which frees a chain nobody owns.
    /// @param work The coroutine to resume, and its claim.
    void submit(::core::async::ParkedWork work) override;
 
private:
    ::morph::exec::IExecutor* _executor;
};
 
}  // namespace detail
 
/// @brief Abstract executor interface.
///
/// Concrete implementations decide *where* and *when* posted tasks run
/// (thread pool, main thread, Qt event loop, …).
///
/// An executor is an identity, so it is neither copyable nor movable: it owns
/// the core-cpp adapter (`coreExecutor()`) that holds a pointer back to it.
struct IExecutor {
    IExecutor() = default;
    IExecutor(const IExecutor&) = delete;
    IExecutor& operator=(const IExecutor&) = delete;
    IExecutor(IExecutor&&) = delete;
    IExecutor& operator=(IExecutor&&) = delete;
    virtual ~IExecutor() = default;
 
    /// @brief Schedules @p task for asynchronous execution.
    ///
    /// Thread-safe. The task is invoked at some point after this call returns.
    /// Any exception thrown by the task is silently swallowed unless the
    /// implementation documents otherwise.
    /// @param task Callable to execute.
    virtual void post(std::function<void()> task) = 0;
 
    /// @brief Whether this executor runs one task at a time, each after the
    ///        tasks posted before it: a strand, a GUI loop, a pumped
    ///        `MainThreadExecutor`.
    ///
    /// What an owner must be: a component whose state is touched only in its
    /// owner's tasks needs those tasks never to overlap. A pool runs several
    /// at once, and the inline executor runs each wherever it is posted from.
    /// @return `true` unless an override says otherwise.
    [[nodiscard]] virtual bool isSerial() const noexcept { return true; }
 
    /// @brief This executor as a core-cpp executor: the identity an
    ///        `core::async::ExecutorScope` names.
    /// @return The adapter that posts every resumption through `post()`. The
    ///         same object for the executor's whole life.
    [[nodiscard]] ::core::async::IExecutor& coreExecutor() noexcept { return _core; }
 
private:
    // Stores only a pointer to this object, so constructing it from `*this`
    // before this object is complete is sound.
    detail::CoreExecutorOver _core{*this};
};
 
inline void detail::CoreExecutorOver::submit(std::coroutine_handle<> handle) {
    _executor->post([handle] { handle.resume(); });
}
 
inline void detail::CoreExecutorOver::submit(::core::async::ParkedWork work) {
    _executor->post([work] {
        work.abandon.disarm();
        work.resume.resume();
    });
}
 
/// @brief Whether the calling thread is inside a task of @p executor.
///
/// True inside a task that @p executor's own worker is running, and inside
/// anything that task calls synchronously, including a task posted to
/// `detail::inlineExecutor()`, which states no scope of its own. Every scope in
/// force is consulted, not only the innermost: a strand's batch runs inside the
/// scope of the pool worker that drives it, and a task on that strand is still
/// a task of the pool.
///
/// Cheap enough for a debug-build assertion: a walk over the calling thread's
/// scope chain, which is a few entries deep.
/// @param executor The executor asked about.
/// @return True while a task of @p executor is running on the calling thread.
[[nodiscard]] inline bool runningOn(IExecutor& executor) noexcept {
    auto const* const identity = &executor.coreExecutor();
    return ::core::async::ExecutorScope::anyInForce(
        [identity](::core::async::ExecutorScope const& scope) { return &scope.executor() == identity; });
}
 
/// @brief Whether the calling thread is inside a task of the core-cpp
///        executor @p executor: a `core::net::EventLoop`'s turn, a core-cpp
///        strand's batch, or anything else that states an `ExecutorScope`.
///
/// The same walk as the overload above, for an owner that is a core-cpp
/// executor rather than a morph one — `IoLoop::loop()` is the one morph hands
/// out.
/// @param executor The executor asked about.
/// @return True while a task of @p executor is running on the calling thread.
[[nodiscard]] inline bool runningOn(::core::async::IExecutor const& executor) noexcept {
    return ::core::async::ExecutorScope::anyInForce(
        [&executor](::core::async::ExecutorScope const& scope) { return &scope.executor() == &executor; });
}
 
/// @brief Multi-threaded executor backed by a fixed-size thread pool.
///
/// Tasks are placed in a FIFO queue and consumed by worker threads.
/// Exceptions thrown by tasks are caught and logged via `morph::log` (they do
/// not propagate out of the worker or abort sibling tasks).
/// The destructor blocks until all worker threads have exited.
// NOLINTNEXTLINE(cppcoreguidelines-special-member-functions)
class ThreadPoolExecutor : public IExecutor {
public:
    /// @brief Constructs the pool with @p n worker threads.
    ///
    /// @p n is clamped to a minimum of 1: a pool with zero workers would accept
    /// posted tasks that could never run (every `post()` would hang forever), so
    /// `ThreadPoolExecutor(0)` yields a usable single-worker pool rather than a
    /// silently dead one.
    /// @param n Number of threads to spawn. Values below 1 are clamped to 1.
    explicit ThreadPoolExecutor(std::size_t n) {
        std::size_t const workers = n == 0 ? 1 : n;
        for (std::size_t i = 0; i < workers; ++i) {
            _workers.emplace_back([this] { loop(); });
        }
    }
 
    /// @brief Signals all workers to stop and joins them.
    ///
    /// Workers drain the queue before exiting: after `_stop` is set they keep
    /// popping and running already-queued tasks and only return once the queue is
    /// empty, so the join blocks until every task queued before destruction has
    /// run. Tasks `post()`ed concurrently with or after destruction race this and
    /// may be silently lost.
    ~ThreadPoolExecutor() override {
        {
            // notify_all() while still holding _m, not after releasing it: a
            // waiter woken by a notify issued after unlocking could reacquire
            // the lock, see _stop, return from wait(), and let this object be
            // destroyed while this thread is still physically inside the
            // notify call -- a real data race on the condition variable's
            // internal state (this pattern is what post()'s own notify_one(),
            // below, must avoid too, for the same reason).
            std::scoped_lock const lock{_m};
            _stop = true;
            _cv.notify_all();
        }
        for (auto& worker : _workers) {
            worker.join();
        }
    }
 
    /// @brief Enqueues @p task for execution on one of the pool threads.
    /// @param task Callable to execute. Thread-safe.
    void post(std::function<void()> task) override {
        std::scoped_lock const lock{_m};
        _q.push(std::move(task));
        _cv.notify_one();
    }
 
    /// @brief Not serial: the workers run tasks at once.
    /// @return `false`.
    [[nodiscard]] bool isSerial() const noexcept override { return false; }
 
private:
    void loop() {
        MORPH_THREAD_NAME("morph.pool");
        for (;;) {
            std::function<void()> task;
            {
                std::unique_lock lock{_m};
                _cv.wait(lock, [this] { return _stop || !_q.empty(); });
                if (_stop && _q.empty()) {
                    return;
                }
                task = std::move(_q.front());
                _q.pop();
            }
            try {
                // Stated per task, not once per worker: the scope ends with the
                // task, on a throw too, so a thread never carries this pool's
                // identity into anything it runs afterwards.
                ::core::async::ExecutorScope const scope{coreExecutor()};
                task();
            } catch (const std::exception& exc) {
                // A task failure must not kill the worker or unrelated tasks, but
                // it must not vanish either — log and carry on.
                ::morph::log::logError("[thread-pool] task threw: " + std::string{exc.what()});
            } catch (...) {
                ::morph::log::logError("[thread-pool] task threw unknown exception");
            }
        }
    }
    std::mutex _m;
    std::condition_variable _cv;
    std::queue<std::function<void()>> _q;
    std::vector<std::thread> _workers;
    bool _stop = false;
};
 
/// @brief Single-thread executor intended for use on the calling (main) thread.
///
/// Tasks posted from other threads are collected and dispatched only when
/// the caller invokes `runFor()`. Useful in tests or event loops that have no
/// native dispatcher.
class MainThreadExecutor : public IExecutor {
public:
    /// @brief Enqueues @p task to be run on the next `runFor()` call.
    ///
    /// Thread-safe. The task is *not* executed immediately.
    /// @param task Callable to execute.
    void post(std::function<void()> task) override {
        std::scoped_lock const lock{_m};
        _q.push(std::move(task));
        // Notified with the lock held: a consumer that polls (`runOnce()`) can
        // take the task the moment the lock is released and destroy this
        // executor, and a notify after that would touch a destroyed condition
        // variable.
        _cv.notify_all();
    }
 
    /// @brief Pumps the task queue for up to @p timeout.
    ///
    /// Runs queued tasks one by one. It does **not** return early when the queue
    /// drains: while time remains before the deadline it blocks waiting for
    /// newly posted tasks, and only returns once the deadline is reached (with
    /// any still-queued tasks left for a subsequent call — this is a pump, not a
    /// flush). Exceptions thrown by tasks are logged and execution continues with
    /// the next task.
    ///
    /// Must be called from the thread that "owns" this executor.
    /// @param timeout Maximum wall-clock time to spend draining.
    void runFor(std::chrono::milliseconds timeout) {
        auto deadline = std::chrono::steady_clock::now() + timeout;
        while (std::chrono::steady_clock::now() < deadline) {
            std::function<void()> task;
            {
                std::unique_lock lock{_m};
                if (!_cv.wait_until(lock, deadline, [this] { return !_q.empty(); })) {
                    return;
                }
                task = std::move(_q.front());
                _q.pop();
            }
            runTask(std::move(task));
        }
    }
 
    /// @brief Dequeues and runs at most one pending task, without blocking.
    ///
    /// Never waits for a task to appear: if the queue is empty, returns `false`
    /// immediately. Otherwise pops exactly one task, runs it, and returns `true`
    /// whether or not that task threw. Unlike `runFor()`, there is no wall-clock
    /// wait — this is the single-step primitive for callers (e.g. tests or
    /// external event loops) that want deterministic, one-task-at-a-time control
    /// instead of a timed pump.
    ///
    /// Must be called from the thread that "owns" this executor.
    /// @return `true` if a task was dequeued and run, `false` if the queue was empty.
    bool runOnce() {
        std::function<void()> task;
        {
            std::scoped_lock const lock{_m};
            if (_q.empty()) {
                return false;
            }
            task = std::move(_q.front());
            _q.pop();
        }
        runTask(std::move(task));
        return true;
    }
 
    /// @brief Runs tasks until the queue is empty, with no wall-clock timeout.
    ///
    /// Repeatedly calls the same dequeue-and-invoke step as `runOnce()` until the
    /// queue reports empty, then returns. It does **not** wait for tasks posted
    /// after draining begins: a task that posts new work while running extends
    /// the drain (the new task is still queued and gets consumed), but `drain()`
    /// never blocks waiting for an *externally* posted task the way `runFor()`
    /// does — it simply stops once the queue is observed empty.
    ///
    /// Must be called from the thread that "owns" this executor.
    void drain() {
        while (runOnce()) {
            // Keep consuming until the queue reports empty.
        }
    }
 
private:
    void runTask(std::function<void()> task) {
        try {
            // The pumping thread is the caller's own, so the scope must end
            // with the task: the caller runs other code on this thread between
            // pumps.
            ::core::async::ExecutorScope const scope{coreExecutor()};
            task();
        } catch (const std::exception& exc) {
            ::morph::log::logError("[main-thread] callback threw: " + std::string{exc.what()});
        } catch (...) {
            // Mirrors ThreadPoolExecutor::loop's own catch-all. Without it a
            // non-`std::exception` throw unwinds out of runFor()/runOnce()/
            // drain(), each of which documents that a throwing task is logged
            // and the pump continues -- runOnce() promises to return `true`
            // "whether or not that task threw", and would not return at all.
            ::morph::log::logError("[main-thread] callback threw unknown exception");
        }
    }
 
    std::mutex _m;
    std::condition_variable _cv;
    std::queue<std::function<void()>> _q;
};
 
namespace detail {
 
/// @brief Executor that runs each posted task on the posting thread, at once.
///
/// The base of an `OwnerStrand` that should run each task on the thread that
/// posts it (a test's owner with no pool). It is not serial
/// (`isSerial()` is `false`), so it cannot be a `Completion`'s owner: a
/// completion delivered "wherever the producer settled" has no one place its
/// callbacks are attached and run.
///
/// @warning Not a general-purpose executor. Posting to it re-enters the caller,
///          so a handler that takes a lock the posting frame already holds
///          self-deadlocks. Application code that wants "run it now" should
///          call the function instead of posting it.
///
/// States no `ExecutorScope`: the task runs on the caller's thread, so
/// whatever scope is already in force is the right answer to "where am I
/// running". A task posted here from inside a pool task is still on the pool
/// (`runningOn(pool)` stays true); one posted from a thread inside no
/// executor's task is on none.
// NOLINTNEXTLINE(cppcoreguidelines-special-member-functions)
class InlineExecutor : public IExecutor {
public:
    /// @brief Runs @p task immediately, on the calling thread.
    ///
    /// An empty @p task is ignored rather than invoked, since `std::function`'s
    /// empty state is a legal value to move around and calling it is undefined.
    /// @param task Callable to execute.
    void post(std::function<void()> task) override {
        if (task) {
            task();
        }
    }
 
    /// @brief Not serial: a task runs on whichever thread posts it, beside
    ///        any other running elsewhere.
    /// @return `false`.
    [[nodiscard]] bool isSerial() const noexcept override { return false; }
};
 
/// @brief The process-wide `InlineExecutor`.
///
/// A function-local static, so it is constructed on first use and outlives
/// everything built over it.
/// @return Reference to the shared inline executor.
inline IExecutor& inlineExecutor() {
    static InlineExecutor executor;
    return executor;
}
 
}  // namespace detail
 
}  // namespace morph::exec