accelerator/src/thread_pool.cpp

//
// SPDX-FileCopyrightText: Hadad <hadad@linuxmail.org>
// SPDX-License-Identifier: Apache-2.0
//

#include "thread_pool.hpp"

namespace pocket_tts_accelerator {

ThreadPool::ThreadPool(std::size_t number_of_threads)
    : should_stop(false)
    , is_stopped(false)
    , thread_count(number_of_threads) {
    
    worker_threads.reserve(number_of_threads);
    
    for (std::size_t thread_index = 0; thread_index < number_of_threads; ++thread_index) {
        worker_threads.emplace_back(&ThreadPool::worker_thread_function, this);
    }
}

ThreadPool::~ThreadPool() {
    shutdown();
}

void ThreadPool::shutdown() {
    {
        std::unique_lock<std::mutex> lock(queue_mutex);
        
        if (is_stopped.load()) {
            return;
        }
        
        should_stop.store(true);
    }
    
    task_available_condition.notify_all();
    
    for (std::thread& worker_thread : worker_threads) {
        if (worker_thread.joinable()) {
            worker_thread.join();
        }
    }
    
    is_stopped.store(true);
}

bool ThreadPool::is_running() const {
    return !should_stop.load() && !is_stopped.load();
}

std::size_t ThreadPool::get_pending_task_count() const {
    std::unique_lock<std::mutex> lock(queue_mutex);
    return task_queue.size();
}

std::size_t ThreadPool::get_thread_count() const {
    return thread_count;
}

void ThreadPool::worker_thread_function() {
    while (true) {
        std::function<void()> task_to_execute;
        
        {
            std::unique_lock<std::mutex> lock(queue_mutex);
            
            task_available_condition.wait(lock, [this] {
                return should_stop.load() || !task_queue.empty();
            });
            
            if (should_stop.load() && task_queue.empty()) {
                return;
            }
            
            task_to_execute = std::move(task_queue.front());
            task_queue.pop();
        }
        
        task_to_execute();
    }
}

}