TaskExecutor.hpp

Go to the documentation of this file.

#pragma once
#ifndef _LOGIT_DETAIL_TASK_EXECUTOR_HPP_INCLUDED
#define _LOGIT_DETAIL_TASK_EXECUTOR_HPP_INCLUDED

 
#include <thread>
#include <queue>
#include <mutex>
#include <functional>
#include <condition_variable>
#include <iostream>
#include <chrono>
 
namespace logit { namespace detail {

    enum class QueuePolicy { Drop, Block };
 
#if defined(__EMSCRIPTEN__)

    class TaskExecutor {
    public:
        static TaskExecutor& get_instance() {
            static TaskExecutor instance;
            return instance;
        }
 
        void add_task(std::function<void()> task) {
            if (task) task();
        }
 
        void wait() {}
        void shutdown() {}
 
        void set_max_queue_size(std::size_t) {}
        void set_queue_policy(QueuePolicy) {}
 
    private:
        TaskExecutor()  = default;
        ~TaskExecutor() = default;
        TaskExecutor(const TaskExecutor&) = delete;
        TaskExecutor& operator=(const TaskExecutor&) = delete;
        TaskExecutor(TaskExecutor&&) = delete;
        TaskExecutor& operator=(TaskExecutor&&) = delete;
    };
 
#else

    class TaskExecutor {
    public:
        static TaskExecutor& get_instance() {
            static TaskExecutor* instance = new TaskExecutor();
            return *instance;
        }

        void add_task(std::function<void()> task) {
            std::unique_lock<std::mutex> lock(m_queue_mutex);
            if (m_stop_flag) return;
            if (m_max_queue_size > 0 && m_tasks_queue.size() >= m_max_queue_size) {
                if (m_overflow_policy == QueuePolicy::Drop) {
                    return;
                }
                m_queue_condition.wait(lock, [this]() {
                    return m_tasks_queue.size() < m_max_queue_size || m_stop_flag;
                });
                if (m_stop_flag) return;
            }
            m_tasks_queue.push(std::move(task));
            lock.unlock();
            m_queue_condition.notify_one();
        }

        void wait() {
            m_queue_condition.notify_one();
            for (;;) {
                std::unique_lock<std::mutex> lock(m_queue_mutex);
                if (m_tasks_queue.empty() || m_stop_flag) break;
                lock.unlock();
                std::this_thread::yield();
                std::this_thread::sleep_for(std::chrono::milliseconds(1));
            }
        }

        void shutdown() {
            std::unique_lock<std::mutex> lock(m_queue_mutex);
            m_stop_flag = true;
            lock.unlock();
            m_queue_condition.notify_all();
            if (m_worker_thread.joinable()) {
                m_worker_thread.join();
            }
        }

        void set_max_queue_size(std::size_t size) {
            std::lock_guard<std::mutex> lock(m_queue_mutex);
            m_max_queue_size = size;
        }

        void set_queue_policy(QueuePolicy policy) {
            std::lock_guard<std::mutex> lock(m_queue_mutex);
            m_overflow_policy = policy;
        }
 
    private:
        std::queue<std::function<void()>> m_tasks_queue;  
        mutable std::mutex m_queue_mutex;                 
        std::condition_variable m_queue_condition;        
        std::thread m_worker_thread;                      
        bool m_stop_flag;                                 
        std::size_t m_max_queue_size;                     
        QueuePolicy m_overflow_policy;                    

        void worker_function() {
            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;
                });
                if (m_stop_flag && m_tasks_queue.empty()) {
                    break;
                }
                task = std::move(m_tasks_queue.front());
                m_tasks_queue.pop();
                lock.unlock();
                m_queue_condition.notify_one();
                task();
            }
        }

        TaskExecutor() : m_stop_flag(false), m_max_queue_size(0), m_overflow_policy(QueuePolicy::Block) {
            m_worker_thread = std::thread(&TaskExecutor::worker_function, this);
        }

        ~TaskExecutor() {
            shutdown();
        }
 
        // Delete copy constructor and assignment operators to enforce singleton usage.
        TaskExecutor(const TaskExecutor&) = delete;
        TaskExecutor& operator=(const TaskExecutor&) = delete;
        TaskExecutor(TaskExecutor&&) = delete;
        TaskExecutor& operator=(TaskExecutor&&) = delete;
    };
 
#endif // defined(__EMSCRIPTEN__)
 
}} // namespace logit::detail
 
#endif // _LOGIT_DETAIL_TASK_EXECUTOR_HPP_INCLUDED