background_task_lane.hpp

Go to the documentation of this file.

#pragma once
 
#include <atomic>
#include <chrono>
#include <cstddef>
#include <cstdint>
#include <thread>
#include <type_traits>
#include <utility>
 
#include <pulp/runtime/exceptions.hpp>
#include <pulp/runtime/fixed_function.hpp>
#include <pulp/runtime/spsc_queue.hpp>
#include <pulp/runtime/triple_buffer.hpp>
 
namespace pulp::format {
 
enum class BackgroundTaskPolicy { Ordered, Latest };

template <typename Task, std::size_t Capacity = 64>
class BackgroundTaskLane {
    static_assert(std::is_trivially_copyable_v<Task>,
                  "RT task payloads must be trivially copyable");
public:
    using Handler = void (*)(void*, const Task&);
    using HandlerFunction = runtime::FixedFunction<void(const Task&), 48>;
 
    BackgroundTaskLane() = default;
    ~BackgroundTaskLane() { stop(); }
    BackgroundTaskLane(const BackgroundTaskLane&) = delete;
    BackgroundTaskLane& operator=(const BackgroundTaskLane&) = delete;
 
    bool start(Handler handler, void* context = nullptr,
               BackgroundTaskPolicy policy = BackgroundTaskPolicy::Ordered) {
        if (!handler)
            return false;
        return start(
            HandlerFunction([handler, context](const Task& task) { handler(context, task); }),
            policy);
    }
 
    bool start(HandlerFunction handler,
               BackgroundTaskPolicy policy = BackgroundTaskPolicy::Ordered) {
        if (!handler || running_.load(std::memory_order_acquire)) return false;
        handler_ = std::move(handler);
        policy_ = policy;
        latest_generation_.store(0, std::memory_order_relaxed);
        stopping_.store(false, std::memory_order_release);
        PULP_TRY {
            worker_ = std::thread([this] { worker_loop(); });
        } PULP_CATCH_ALL {
            handler_.reset();
            return false;
        }
        running_.store(true, std::memory_order_release);
        return true;
    }
 
    void stop() noexcept {
        stopping_.store(true, std::memory_order_release);
        if (worker_.joinable()) worker_.join();
        running_.store(false, std::memory_order_release);
    }
 
    bool try_spawn(const Task& task) noexcept {
        if (!running_.load(std::memory_order_acquire)) return false;
        if (policy_ == BackgroundTaskPolicy::Latest) {
            latest_.write(task);
            latest_generation_.fetch_add(1, std::memory_order_release);
            return true;
        }
        return ordered_.try_push(task);
    }
 
    bool running() const noexcept { return running_.load(std::memory_order_acquire); }
    std::uint64_t overflow_count() const noexcept { return ordered_.overflow_count(); }
 
private:
    void worker_loop() noexcept {
        std::uint64_t consumed_generation = 0;
        while (!stopping_.load(std::memory_order_acquire)) {
            bool handled = false;
            if (policy_ == BackgroundTaskPolicy::Latest) {
                const auto generation = latest_generation_.load(std::memory_order_acquire);
                if (generation != consumed_generation) {
                    invoke(latest_.read());
                    consumed_generation = generation;
                    handled = true;
                }
            } else if (auto task = ordered_.try_pop()) {
                invoke(*task);
                handled = true;
            }
            if (!handled) std::this_thread::sleep_for(std::chrono::milliseconds(1));
        }
        if (policy_ == BackgroundTaskPolicy::Ordered) {
            while (auto task = ordered_.try_pop()) invoke(*task);
        } else {
            const auto generation = latest_generation_.load(std::memory_order_acquire);
            if (generation != consumed_generation) invoke(latest_.read());
        }
    }
 
    void invoke(const Task& task) noexcept {
        PULP_TRY {
            handler_(task);
        }
        PULP_CATCH_ALL {}
    }
 
    runtime::SpscQueue<Task, Capacity> ordered_;
    runtime::TripleBuffer<Task> latest_;
    std::atomic<std::uint64_t> latest_generation_{0};
    std::atomic<bool> running_{false};
    std::atomic<bool> stopping_{false};
    HandlerFunction handler_;
    BackgroundTaskPolicy policy_ = BackgroundTaskPolicy::Ordered;
    std::thread worker_;
};
 
} // namespace pulp::format