#include <mutex>
#include <iostream>

#include "util/AgThreadPool.h"
#include "util/Log.h"

AgThreadPool::AgThreadPool(const char *name, int maximumPoolSize, int workQueueCapacity)
    : name(name), maximumPoolSize(maximumPoolSize), workQueueCapacity(workQueueCapacity), debugInfoTimeout(0), _terminate(false), oneWarnFullWorkQueue(true)
{
    lastDebugInfoTime = Util::getCurrentMilliseconds();
    poolManager = Thread(&AgThreadPool::managePool, this);
}

AgThreadPool::~AgThreadPool()
{
    _terminate = true;
    notEmptyTaskQueue.notify_all();
    poolManager.join();
}

void AgThreadPool::execute(Task task)
{
    std::unique_lock<std::mutex> lock(taskQueueMutex);
    bool wasEmpty = taskQueue.empty();
    if (wasEmpty) {
        taskQueue.push(task);
        notEmptyTaskQueue.notify_one();
    } else {
        if (taskQueue.size() < workQueueCapacity) {
            taskQueue.push(task);
            notEmptyTaskQueue.notify_one();
        } else {
            if (oneWarnFullWorkQueue) {
                oneWarnFullWorkQueue = false;
                LOG_CORE_WARN("Full task queue in AgThreadPool: " << name);
            }

            // Wait until taskQueue dequeued ...
            std::cv_status result = notFullTaskQueue.wait_for(lock, std::chrono::seconds(30));
            if (result == std::cv_status::timeout) {
                LOG_CORE_ERROR("Error dequeued tasks in AgThreadPool: " << name);
                return;
            }
            taskQueue.push(task);
        }
    }

    lock.unlock();

    if (debugInfoTimeout) {
        milliseconds current = Util::getCurrentMilliseconds();
        if (current.count() - lastDebugInfoTime.count() > debugInfoTimeout) {
           lastDebugInfoTime = current;
           std::cout << std::endl << "++++++++++++++++++++++++++++++++++++++++" << std::endl;
           lock.lock();
           std::cout << "AgThreadPool: " << name << ", Queue size: " << taskQueue.size() << std::endl;
           lock.unlock();
           std::cout << std::endl << "++++++++++++++++++++++++++++++++++++++++" << std::endl;

           LOG_OPC_SERVER_INFO("AgThreadPool: " << name << ", Queue size: " << taskQueue.size());
       }
    }
}

void AgThreadPool::managePool()
{
    LOG_CORE_DEBUG("Starting in AgThreadPool: " << name << ", ManagePool, id: " << std::this_thread::get_id());

    for (int i = 0; i < maximumPoolSize; i++)
        threadPoolTasks.push_back(Thread(&AgThreadPool::worker, this));

    for (auto &thread : threadPoolTasks)
        thread.join();
}

long AgThreadPool::getTasksCount()
{
    std::unique_lock<std::mutex> lock(taskQueueMutex);
    return taskQueue.size();
}

void AgThreadPool::worker()
{
    LOG_CORE_DEBUG("Starting in AgThreadPool: " << name << ", new Worker thread, id: " << std::this_thread::get_id());

    while (!_terminate)
    {
        std::unique_lock<std::mutex> lock(taskQueueMutex);
        while (taskQueue.empty() && !_terminate)
            notEmptyTaskQueue.wait(lock);

        if (_terminate) break;

        Task task = taskQueue.front();
        taskQueue.pop();
        if (taskQueue.size() == workQueueCapacity - 1)
            notFullTaskQueue.notify_one();

        lock.unlock();

        try
        {
            task();
        }
        catch (const std::exception& e)
        {
            LOG_AG_ERROR("AgThreadPool: " << name << ", error: " << e.what());
        }
        catch ( ... )
        {
            LOG_CORE_ERROR("Unknown error in AgThreadPool: " << name);
        }
    }

    LOG_CORE_DEBUG("Stopping in AgThreadPool: " << name << ", Worker thread, id: " << std::this_thread::get_id());
}

void AgThreadPool::setDebugInfoTimeout(long timeoutSec)
{
    debugInfoTimeout = timeoutSec * 1000;
}

void AgThreadPool::terminate()
{
    _terminate = true;
}

bool AgThreadPool::isTerminated()
{
    return _terminate;
}

