TaskExecutor.hpp

Go to the documentation of this file.

#pragma once
#ifndef _LOGIT_TASK_EXECUTOR_HPP_INCLUDED
#define _LOGIT_TASK_EXECUTOR_HPP_INCLUDED
 
#include <thread>
#include <queue>
#include <mutex>
#include <functional>
#include <condition_variable>
 
namespace logit {

    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;
            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_one();
            if (m_worker_thread.joinable()) {
                m_worker_thread.join();
            }
        }
 
    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;                                 

        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();
                task();
            }
        }

        TaskExecutor() : m_stop_flag(false) {
            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;
    };
 
}; // namespace logit
 
#endif // _LOGIT_TASK_EXECUTOR_HPP_INCLUDED