TaskExecutor.hpp

Go to the documentation of this file.

#pragma once
#ifndef LOGIT_CPP_HEADER_LOGIT_CPP_LOGIT_DETAIL_TASKEXECUTOR_HPP_INCLUDED
#define LOGIT_CPP_HEADER_LOGIT_CPP_LOGIT_DETAIL_TASKEXECUTOR_HPP_INCLUDED
 
 
#include <functional>
#include <atomic>
#if defined(__EMSCRIPTEN__) && !defined(__EMSCRIPTEN_PTHREADS__)
  #include <deque>
  #include <mutex>
  #include <emscripten/emscripten.h>
    #else
        #include <thread>
        #include <deque>
        #include <mutex>
        #include <condition_variable>
        #include <chrono>
#endif
 
// Enable lock-free MPSC ring integration (non-Emscripten) by defining:
//   #define LOGIT_USE_MPSC_RING
 
#if !defined(__EMSCRIPTEN__) || defined(__EMSCRIPTEN_PTHREADS__)
  #ifdef LOGIT_USE_MPSC_RING
    #include "MpscRingAny.hpp"
  #endif
#endif
 
#include "QueuePolicy.hpp"
 
namespace logit { namespace detail {
 
#   if defined(__EMSCRIPTEN__) && !defined(__EMSCRIPTEN_PTHREADS__)
    
    class TaskExecutor {
    public:
        static TaskExecutor& get_instance() {
            static TaskExecutor instance;
            return instance;
        }
 
        void add_task(std::function<void()> task) {
            if (!task) return;
            bool schedule = false;
            for (;;) {
                schedule = false;
                {
                    std::lock_guard<std::mutex> lk(m_mutex);
                    if (m_max_queue_size > 0 && m_tasks.size() >= m_max_queue_size) {
                        switch (m_overflow_policy) {
                            case QueuePolicy::DropNewest:
                                ++m_dropped_tasks;
                                return;
                            case QueuePolicy::DropOldest:
                                if (!m_tasks.empty()) {
                                    m_tasks.pop_front();
                                    ++m_dropped_tasks;
                                }
                                break;
                            case QueuePolicy::Block:
                                break; // handled after unlocking
                        }
                        if (m_overflow_policy == QueuePolicy::Block && m_tasks.size() >= m_max_queue_size) {
                            // fall through to drain outside lock
                        } else {
                            m_tasks.emplace_back(std::move(task));
                            schedule = !m_scheduled;
                            m_scheduled = m_scheduled || schedule;
                            break;
                        }
                    } else {
                        m_tasks.emplace_back(std::move(task));
                        schedule = !m_scheduled;
                        m_scheduled = m_scheduled || schedule;
                        break;
                    }
                }
                drain();
            }
            if (schedule) {
                emscripten_async_call(&TaskExecutor::drain_thunk, this, 0);
            }
        }
 
        void wait() { drain(); }
        void shutdown() { drain(); }
 
        void set_max_queue_size(std::size_t size) {
            std::lock_guard<std::mutex> lk(m_mutex);
            m_max_queue_size = size;
        }
 
        void set_queue_policy(QueuePolicy policy) {
            std::lock_guard<std::mutex> lk(m_mutex);
            m_overflow_policy = policy;
        }
 
        std::size_t dropped_tasks() const noexcept {
            return m_dropped_tasks.load(std::memory_order_relaxed);
        }
        void reset_dropped_tasks() noexcept {
            m_dropped_tasks.store(0, std::memory_order_relaxed);
        }
    
    private:
        TaskExecutor()
            : m_max_queue_size(0),
              m_overflow_policy(QueuePolicy::Block),
              m_dropped_tasks(0),
              m_scheduled(false) {}
        ~TaskExecutor() = default;
        TaskExecutor(const TaskExecutor&) = delete;
        TaskExecutor& operator=(const TaskExecutor&) = delete;
        TaskExecutor(TaskExecutor&&) = delete;
        TaskExecutor& operator=(TaskExecutor&&) = delete;
    
        std::deque<std::function<void()>> m_tasks;
        std::mutex m_mutex;
        std::size_t m_max_queue_size;
        QueuePolicy m_overflow_policy;
        std::atomic<std::size_t> m_dropped_tasks;
        bool m_scheduled;
    
        static void drain_thunk(void* arg) {
            static_cast<TaskExecutor*>(arg)->drain();
        }
    
        void drain() {
            for (;;) {
                std::function<void()> task;
                {
                    std::lock_guard<std::mutex> lk(m_mutex);
                    if (m_tasks.empty()) {
                        m_scheduled = false;
                        break;
                    }
                    task = std::move(m_tasks.front());
                    m_tasks.pop_front();
                }
                task();
            }
        }
    };
    
#   else // !Emscripten or pthreads
    
    class TaskExecutor {
    public:
        static TaskExecutor& get_instance() {
            static TaskExecutor* instance = new TaskExecutor();
            return *instance;
        }
 
        void add_task(std::function<void()> task) {
            if (!task) return;
#        ifndef LOGIT_USE_MPSC_RING
            std::unique_lock<std::mutex> lock(m_queue_mutex);
            if (m_stop_flag.load(std::memory_order_acquire)) return;
            if (m_max_queue_size > 0 && m_tasks_queue.size() >= m_max_queue_size) {
                switch (m_overflow_policy.load(std::memory_order_relaxed)) {
                    case QueuePolicy::DropNewest:
                        ++m_dropped_tasks;
                        return;
                    case QueuePolicy::DropOldest:
                        if (!m_tasks_queue.empty()) {
                            m_tasks_queue.pop_front();
                            ++m_dropped_tasks;
                        }
                        break;
                    case QueuePolicy::Block:
                        m_queue_condition.wait(lock, [this]() {
                            return m_tasks_queue.size() < m_max_queue_size ||
                                   m_stop_flag.load(std::memory_order_acquire);
                        });
                        if (m_stop_flag.load(std::memory_order_acquire)) return;
                        break;
                }
            }
            m_tasks_queue.push_back(std::move(task));
            lock.unlock();
            m_queue_condition.notify_one();
#        else
            enter_producer_();
    
            std::function<void()> local_task = std::move(task);
            bool done = false;
    
            while (!done) {
                if (m_stop_flag.load(std::memory_order_acquire)) {
                    break;
                }
    
                const auto policy = m_overflow_policy.load(std::memory_order_relaxed);
    
                // Backpressure relies on the count of in-flight tasks.
                if (policy == QueuePolicy::Block &&
                    m_max_queue_size > 0 &&
                    m_active_tasks.load(std::memory_order_relaxed) >= m_max_queue_size)
                {
                    std::unique_lock<std::mutex> lk(m_cv_mutex);
                    m_cv.wait_for(lk, std::chrono::microseconds(LOGIT_TASK_EXECUTOR_BLOCK_WAIT_USEC));
                    continue;
                }
    
                // Try to push into the ring buffer.
                // Move into the ring: std::function's move construction is
                // noexcept, while copying may allocate and throw.  The ring
                // deliberately accepts only no-throw construction so a
                // producer can never leave a claimed cell unpublished.
                if (m_mpsc_queue.try_push(std::move(local_task))) {
                    m_cv.notify_one(); // wake the worker
                    break;
                }
 
                // Apply the configured overflow policy when the ring is full.
                switch (policy) {
                    case QueuePolicy::DropNewest:
                        m_dropped_tasks.fetch_add(1, std::memory_order_relaxed);
                        done = true;
                        break;
 
                    case QueuePolicy::DropOldest:
                        // Safe MPSC behaviour: drop the incoming task.
                        // Preserves ordering and avoids producer/consumer deadlocks.
                        m_dropped_tasks.fetch_add(1, std::memory_order_relaxed);
                        done = true;
                        break;
    
                    case QueuePolicy::Block: {
                        std::unique_lock<std::mutex> lk(m_cv_mutex);
                        m_cv.wait_for(lk, std::chrono::microseconds(LOGIT_TASK_EXECUTOR_BLOCK_WAIT_USEC));
                        break;
                    }
                }
            }
            leave_producer_();
#        endif
        }
    
        void wait() {
#        ifndef LOGIT_USE_MPSC_RING
            std::unique_lock<std::mutex> lock(m_queue_mutex);
            m_queue_condition.wait(lock, [this]() {
                return ((m_tasks_queue.empty() &&
                        m_active_tasks.load(std::memory_order_relaxed) == 0) ||
                        m_stop_flag.load(std::memory_order_acquire));
            });
#        else
            std::unique_lock<std::mutex> lock(m_queue_mutex);
            m_queue_condition.wait(lock, [this]() {
                return ((queue_empty_() &&
                        m_active_tasks.load(std::memory_order_relaxed) == 0) ||
                        m_stop_flag.load(std::memory_order_acquire));
            });
#        endif
        }
    
        void shutdown() {
            std::lock_guard<std::mutex> lifecycle_lock(m_lifecycle_mutex);
#        ifndef LOGIT_USE_MPSC_RING
            std::unique_lock<std::mutex> lock(m_queue_mutex);
            m_stop_flag.store(true, std::memory_order_release);
            lock.unlock();
            m_queue_condition.notify_all();
            if (m_worker_thread.joinable()) {
                m_worker_thread.join();
            }
#        else
            {
                std::lock_guard<std::mutex> lock(m_queue_mutex);
                m_stop_flag.store(true, std::memory_order_release);
            }
            m_cv.notify_all();
            m_queue_condition.notify_all();
            if (m_worker_thread.joinable()) {
                m_worker_thread.join();
            }
#        endif
        }
    
        void set_max_queue_size(std::size_t size) {
            std::lock_guard<std::mutex> lifecycle_lock(m_lifecycle_mutex);
            if (m_stop_flag.load(std::memory_order_acquire)) return;
#       ifdef LOGIT_USE_MPSC_RING
            // Tell producers to pause before any wait()/stop conditions run.
            m_resizing.store(true, std::memory_order_release);
            const auto deadline = std::chrono::steady_clock::now() +
                                  std::chrono::seconds(1);
            // Existing producers may be blocked by backpressure, so do not
            // wait forever for the resize barrier.
            if (!wait_until_producers_paused_(deadline)) {
                m_resizing.store(false, std::memory_order_release);
                m_resize_cv.notify_all();
                return;
            }
 
            // Drain the queue completely, but do not wait forever if the worker
            // is stalled (e.g., blocked sink or backpressure keeping
            // m_active_tasks > 0). If we fail to drain before the deadline,
            // abort the resize and re-open the barrier so producers can
            // continue.
            if (!wait_until_idle_(deadline)) {
                m_resizing.store(false, std::memory_order_release);
                m_resize_cv.notify_all();
                return;
            }
 
            // Stop the worker so it cannot touch m_mpsc_queue during the resize.
            std::unique_lock<std::mutex> lk(m_queue_mutex);
            m_stop_flag.store(true, std::memory_order_relaxed);
            lk.unlock();
 
            m_cv.notify_all();
            m_queue_condition.notify_all();
            if (m_worker_thread.joinable()) {
                m_worker_thread.join();
            }
 
            // Reinitialise the parameters and the ring on a single thread.
            lk.lock();
            m_max_queue_size = size;
            const std::size_t cap =
                    (m_max_queue_size == 0 ? m_default_ring_cap : m_max_queue_size);
            m_mpsc_queue = MpscRingAny<std::function<void()>>(cap);
            // Reset counters (except drops) because the queue is empty.
            m_active_tasks.store(0, std::memory_order_relaxed);
            // Keep m_dropped_tasks untouched; tests manage it via macros.
            lk.unlock();
 
            // Clear the stop flag and restart the worker thread.
            m_stop_flag.store(false, std::memory_order_relaxed);
            m_worker_thread = std::thread(&TaskExecutor::worker_function, this);
 
            // Re-open the barrier so producers can continue.
            m_resizing.store(false, std::memory_order_release);
            m_resize_cv.notify_all();
#       else
            std::lock_guard<std::mutex> lock(m_queue_mutex);
            m_max_queue_size = size;
#       endif
        }
    
        void set_queue_policy(QueuePolicy policy) {
            std::lock_guard<std::mutex> lifecycle_lock(m_lifecycle_mutex);
            if (m_stop_flag.load(std::memory_order_acquire)) return;
            std::lock_guard<std::mutex> lock(m_queue_mutex);
            m_overflow_policy.store(policy, std::memory_order_relaxed);
        }
 
        std::size_t dropped_tasks() const noexcept {
            return m_dropped_tasks.load(std::memory_order_relaxed);
        }
        void reset_dropped_tasks() noexcept {
            m_dropped_tasks.store(0, std::memory_order_relaxed);
        }
    
    private:
        mutable std::mutex m_lifecycle_mutex;      
    #ifndef LOGIT_USE_MPSC_RING
        std::deque<std::function<void()>> m_tasks_queue;
        mutable std::mutex m_queue_mutex;
        std::condition_variable m_queue_condition;
        std::thread m_worker_thread;
        std::atomic<bool> m_stop_flag;
        std::size_t m_max_queue_size;
        std::atomic<QueuePolicy> m_overflow_policy;
        std::atomic<std::size_t> m_dropped_tasks;
        std::atomic<std::size_t> m_active_tasks;
    #else
        mutable std::mutex m_queue_mutex;          
        std::condition_variable m_queue_condition; 
 
        std::condition_variable m_cv;              
        std::mutex m_cv_mutex;                     
 
        std::atomic<bool> m_resizing;              
        std::condition_variable m_resize_cv;       
        std::atomic<std::size_t> m_active_producers; 
    
        std::thread m_worker_thread;
        std::atomic<bool> m_stop_flag;
        std::size_t m_max_queue_size;
        std::atomic<QueuePolicy> m_overflow_policy;
        std::atomic<std::size_t> m_dropped_tasks;
        std::atomic<std::size_t> m_active_tasks;
    
        const std::size_t m_default_ring_cap = LOGIT_TASK_EXECUTOR_DEFAULT_RING_CAPACITY;
        MpscRingAny<std::function<void()>> m_mpsc_queue;
    #endif
    
        void worker_function() {
    #ifndef LOGIT_USE_MPSC_RING
            for (;;) {
                std::function<void()> task;
                std::unique_lock<std::mutex> lock(m_queue_mutex);
                m_queue_condition.wait(lock, [this]() {
                    return !m_tasks_queue.empty() || m_stop_flag.load(std::memory_order_acquire);
                });
                if (m_stop_flag.load(std::memory_order_acquire) && m_tasks_queue.empty()) {
                    break;
                }
                task = std::move(m_tasks_queue.front());
                m_tasks_queue.pop_front();
                m_active_tasks.fetch_add(1, std::memory_order_relaxed);
                lock.unlock();
                m_queue_condition.notify_one();
    
                task();
    
                lock.lock();
                m_active_tasks.fetch_sub(1, std::memory_order_relaxed);
                if (m_tasks_queue.empty() && m_active_tasks.load(std::memory_order_relaxed) == 0) {
                    m_queue_condition.notify_all();
                }
                lock.unlock();
            }
    #else
            for (;;) {
                bool drained_any = false;
                std::function<void()> task;
    
                int budget = LOGIT_TASK_EXECUTOR_DRAIN_BUDGET;
                while (budget--) {
                    // Count the pop attempt as active so wait() cannot observe
                    // an empty ring between try_pop() freeing a cell and the
                    // dequeued task starting execution.
                    m_active_tasks.fetch_add(1, std::memory_order_relaxed);
                    if (!m_mpsc_queue.try_pop(task)) {
                        m_active_tasks.fetch_sub(1, std::memory_order_relaxed);
                        break;
                    }
 
                    drained_any = true;
    
                    task();
    
                    m_active_tasks.fetch_sub(1, std::memory_order_relaxed);
                    m_cv.notify_one(); // freed an in-flight slot
                }
    
                if (queue_empty_() && m_active_tasks.load(std::memory_order_relaxed) == 0) {
                    std::unique_lock<std::mutex> lock(m_queue_mutex);
                    m_queue_condition.notify_all(); // notify wait()
                    m_cv.notify_all();              // wake producers blocked on Block
                    if (m_stop_flag.load(std::memory_order_acquire)) {
                        break;
                    }
                }
    
                if (!drained_any) {
                    std::unique_lock<std::mutex> lk(m_cv_mutex);
                    if (m_stop_flag.load(std::memory_order_acquire) && queue_empty_()) {
                        break;
                    }
                    m_cv.wait_for(lk, std::chrono::milliseconds(1));
                }
            }
    #endif
        }
    
    #ifdef LOGIT_USE_MPSC_RING
        bool queue_empty_() const noexcept {
            return m_mpsc_queue.empty();
        }
 
        bool wait_until_idle_(std::chrono::steady_clock::time_point deadline) {
            std::unique_lock<std::mutex> lock(m_queue_mutex);
            return m_queue_condition.wait_until(lock, deadline, [this]() {
                return ((queue_empty_() &&
                         m_active_tasks.load(std::memory_order_relaxed) == 0) ||
                        m_stop_flag.load(std::memory_order_acquire));
            });
        }
 
        void enter_producer_() {
            for (;;) {
                if (m_resizing.load(std::memory_order_acquire)) {
                    std::unique_lock<std::mutex> lk(m_cv_mutex);
                    m_resize_cv.wait(lk, [this]() {
                        return !m_resizing.load(std::memory_order_acquire);
                    });
                    continue;
                }
 
                m_active_producers.fetch_add(1, std::memory_order_acq_rel);
                if (!m_resizing.load(std::memory_order_acquire)) {
                    return;
                }
                leave_producer_();
            }
        }
 
        void leave_producer_() {
            if (m_active_producers.fetch_sub(1, std::memory_order_acq_rel) == 1 &&
                m_resizing.load(std::memory_order_acquire)) {
                m_resize_cv.notify_all();
            }
        }
 
        bool wait_until_producers_paused_(std::chrono::steady_clock::time_point deadline) {
            std::unique_lock<std::mutex> lk(m_cv_mutex);
            return m_resize_cv.wait_until(lk, deadline, [this]() {
                return m_active_producers.load(std::memory_order_acquire) == 0;
            });
        }
    #endif
    
        TaskExecutor()
    #ifndef LOGIT_USE_MPSC_RING
            : m_stop_flag(false),
              m_max_queue_size(0),
              m_overflow_policy(QueuePolicy::Block),
              m_dropped_tasks(0),
              m_active_tasks(0)
    #else
            : m_resizing(false),
              m_active_producers(0),
              m_worker_thread(),
              m_stop_flag(false),
              m_max_queue_size(0),
              m_overflow_policy(QueuePolicy::Block),
              m_dropped_tasks(0),
              m_active_tasks(0),
              m_mpsc_queue(m_default_ring_cap)
    #endif
        {
            m_worker_thread = std::thread(&TaskExecutor::worker_function, this);
        }
    
        ~TaskExecutor() {
            shutdown();
        }
    
        TaskExecutor(const TaskExecutor&) = delete;
        TaskExecutor& operator=(const TaskExecutor&) = delete;
        TaskExecutor(TaskExecutor&&) = delete;
        TaskExecutor& operator=(TaskExecutor&&) = delete;
    };
 
#endif // Emscripten split
 
}} // namespace logit::detail
 
#endif // LOGIT_CPP_HEADER_LOGIT_CPP_LOGIT_DETAIL_TASKEXECUTOR_HPP_INCLUDED