#include "tile/base/internal/thread_pool.h"

#include "tile/base/thread/scoped_lock.h"
#include "tile/base/thread/unique_lock.h"

namespace tile {
namespace internal {
static std::atomic<uint64_t> worker_id{1};

ThreadPool::ThreadPool(std::size_t workers, const std::vector<int> &affinity, int nice_value)
{
    TILE_CHECK_GE(workers, 1);
    workers_.resize(workers);
    int id = 1;
    for (auto &&t : workers_) {
        t = std::thread([=] {
            if (nice_value) {
                // set nice
            }

            if (!affinity.empty()) {
                // SetCurrentThreadAffinity
            }

            int id = worker_id.fetch_add(1);
            // TILE_LOG_INFO("Worker {} started.", id);

            WorkerProc();

            // TILE_LOG_INFO("Worker {} exited.", id);
        });
    }
}

void
ThreadPool::QueueJob(JobType &&job)
{
    TILE_CHECK(job != nullptr);
    ScopedLock _(lock_);
    jobs_.push(std::move(job));
    cv_.NotifyOne();
}

void
ThreadPool::Stop()
{
    ScopedLock _(lock_);
    exiting_.store(true, std::memory_order_relaxed);
    cv_.NotifyAll();
}

void
ThreadPool::Join()
{
    for (auto &&t : workers_) { t.join(); }
}

void
ThreadPool::WorkerProc()
{
    while (true) {
        UniqueLock<Mutex> lk(lock_);
        while (!exiting_.load() && jobs_.empty()) {
            cv_.Wait(lk, [&] { return exiting_.load() || !jobs_.empty(); });
        }

        if (exiting_.load()) { break; }

        TILE_CHECK(!jobs_.empty());
        JobType job = std::move(jobs_.front());
        jobs_.pop();

        lk.Unlock();
        TILE_CHECK(job != nullptr);
        job();
    }
}

}// namespace internal
}// namespace tile
