/*
 * .============.
 * // M A K E / \
 * // C++ DEV / \
 * // E A S Y / \/ \
 * ++ ----------. \/\ .
 * \\ \ \ /\ /
 * \\ \ \ /
 * \\ \ \ /
 * -============'
 *
 * Copyright (c) 2018 Hevake and contributors, all rights reserved.
 *
 * This file is part of cpp-tbox (https://github.com/cpp-main/cpp-tbox)
 * Use of this source code is governed by MIT license that can be found
 * in the LICENSE file in the root of the source tree. All contributing
 * project authors may be found in the CONTRIBUTORS.md file in the root
 * of the source tree.
 */
#include "work_thread.h"

#include <cinttypes>
#include <set>
#include <deque>
#include <thread>
#include <mutex>
#include <algorithm>
#include <condition_variable>
#include <chrono>

#include <tbox/base/log.h>
#include <tbox/base/defines.h>
#include <tbox/base/cabinet.hpp>
#include <tbox/base/catch_throw.h>
#include <tbox/base/object_pool.hpp>
#include <tbox/base/assert.h>
#include <tbox/base/wrapped_recorder.h>
#include <tbox/event/loop.h>

#undef MODULE_ID
#define MODULE_ID "tbox.work_thread"

namespace tbox {
namespace eventx {

using Clock = std::chrono::steady_clock;

//! WorkThread 的私有数据
struct WorkThread::Data {
 event::Loop *default_main_loop = nullptr; //!< 主线程

 std::mutex lock; //!< 互斥锁
 std::condition_variable cond_var; //!< 条件变量

 std::thread work_thread;

 cabinet::Cabinet<Task> undo_tasks_cabinet;

 std::deque<TaskToken> undo_tasks_token_deque; //!< 排队中的任务队列
 std::set<TaskToken> doing_tasks_token; //!< 正在处理的任务集合

 ObjectPool<Task> task_pool{64};

 bool stop_flag = false; //!< 是否立即停止标记
};

/**
 * 任务项
 */
struct WorkThread::Task {
 TaskToken token;
 NonReturnFunc backend_task; //! 任务在工作线程中执行函数
 NonReturnFunc main_cb; //! 任务执行完成后由main_loop执行的回调函数
 event::Loop *main_loop = nullptr;
 Clock::time_point create_time_point;

 Task *next = nullptr;
};

/////////////////////////////////////////////////////////////////////////////////

WorkThread::WorkThread(event::Loop *main_loop) :
 d_(new Data)
{
 d_->default_main_loop = main_loop;
 d_->work_thread = std::thread(std::bind(&WorkThread::threadProc, this));
 d_->stop_flag = false;
}

WorkThread::~WorkThread()
{
 cleanup();
}

ThreadExecutor::TaskToken WorkThread::execute(NonReturnFunc &&backend_task)
{
 return execute(std::move(backend_task), nullptr, nullptr);
}

ThreadExecutor::TaskToken WorkThread::execute(const NonReturnFunc &backend_task)
{
 NonReturnFunc backend_task_copy(backend_task);
 return execute(std::move(backend_task_copy), nullptr, nullptr);
}

ThreadExecutor::TaskToken WorkThread::execute(NonReturnFunc &&backend_task, NonReturnFunc &&main_cb) {
 return execute(std::move(backend_task), std::move(main_cb), nullptr);
}

ThreadExecutor::TaskToken WorkThread::execute(const NonReturnFunc &backend_task, const NonReturnFunc &main_cb) {
 NonReturnFunc backend_task_copy(backend_task);
 NonReturnFunc main_cb_copy(main_cb);
 return execute(std::move(backend_task_copy), std::move(main_cb_copy), nullptr);
}

ThreadExecutor::TaskToken WorkThread::execute(NonReturnFunc &&backend_task, NonReturnFunc &&main_cb, event::Loop *main_loop)
{
 RECORD_SCOPE();
 TaskToken token;

 if (d_ == nullptr) {
 LogWarn("WorkThread has been cleanup");
 return token;
 }

 {
 std::lock_guard<std::mutex> lg(d_->lock);

 Task *item = d_->task_pool.alloc();
 item->backend_task = std::move(backend_task);
 item->main_cb = std::move(main_cb);
 item->main_loop = (main_loop != nullptr) ? main_loop : d_->default_main_loop;
 item->create_time_point = Clock::now();
 item->token = token = d_->undo_tasks_cabinet.alloc(item);

 d_->undo_tasks_token_deque.push_back(token);
 }

 LogDbg("create task %u", token.id());
 d_->cond_var.notify_one();

 return token;
}

ThreadExecutor::TaskToken WorkThread::execute(const NonReturnFunc &backend_task, const NonReturnFunc &main_cb, event::Loop *main_loop)
{
 NonReturnFunc backend_task_copy(backend_task);
 NonReturnFunc main_cb_copy(main_cb);
 return execute(std::move(backend_task_copy), std::move(main_cb_copy), main_loop);
}

ThreadExecutor::TaskStatus WorkThread::getTaskStatus(TaskToken task_token) const
{
 if (d_ == nullptr) {
 LogWarn("WorkThread has been cleanup");
 return TaskStatus::kCleanup;
 }

 std::lock_guard<std::mutex> lg(d_->lock);

 if (d_->undo_tasks_cabinet.at(task_token) != nullptr)
 return TaskStatus::kWaiting;

 if (d_->doing_tasks_token.find(task_token) != d_->doing_tasks_token.end())
 return TaskStatus::kExecuting;

 return TaskStatus::kNotFound;
}

ThreadExecutor::CancelResult WorkThread::cancel(TaskToken token)
{
 RECORD_SCOPE();
 if (d_ == nullptr) {
 LogWarn("WorkThread has been cleanup");
 return CancelResult::kCleanup;
 }

 std::lock_guard<std::mutex> lg(d_->lock);

 //! 如果正在执行
 if (d_->doing_tasks_token.find(token) != d_->doing_tasks_token.end())
 return CancelResult::kExecuting; //! 返回正在执行

 //! 从高优先级向低优先级遍历,找出优先级最高的任务
 if (!d_->undo_tasks_token_deque.empty()) {
 auto iter = std::find(d_->undo_tasks_token_deque.begin(), d_->undo_tasks_token_deque.end(), token);
 if (iter != d_->undo_tasks_token_deque.end()) {
 d_->undo_tasks_token_deque.erase(iter);
 d_->task_pool.free(d_->undo_tasks_cabinet.free(token));
 return CancelResult::kSuccess;
 }
 }

 return CancelResult::kNotFound; //! 返回没有找到
}

void WorkThread::threadProc()
{
 LogDbg("thread start");

 while (true) {
 Task* item = nullptr;
 {
 std::unique_lock<std::mutex> lk(d_->lock);

 //! 等待任务
 d_->cond_var.wait(lk, std::bind(&WorkThread::shouldThreadExitWaiting, this));

 /**
 * 有两种情况会从 cond_var.wait() 退出
 * 1. 任务队列中有任务需要执行时
 * 2. 析构时要求所有工作线程退出时
 *
 * 所以,下面检查 stop_flag 看是不是请求退出
 */
 if (d_->stop_flag) {
 LogDbg("thread will exit, stop flag.");
 break;
 }

 item = popOneTask(); //! 从任务队列中取出优先级最高的任务
 }

 //! 后面就是去执行任务,不需要再加锁了
 if (item != nullptr) {
 RECORD_SCOPE();
 {
 std::lock_guard<std::mutex> lg(d_->lock);
 d_->doing_tasks_token.insert(item->token);
 }

 LogDbg("thread pick task %u", item->token.id());

 auto exec_time_point = Clock::now();
 auto wait_time_cost = exec_time_point - item->create_time_point;

 {
 RECORD_SCOPE();
 CatchThrow(item->backend_task, "tbox::eventx::WorkThread", true);
 }

 auto exec_time_cost = Clock::now() - exec_time_point;

 LogDbg("thread finish task %u, cost %" PRIu64 " + %" PRIu64 " us",
 item->token.id(),
 wait_time_cost.count() / 1000,
 exec_time_cost.count() / 1000);

 /**
 * 有时在妥托给WorkThread执行动作时,会在lamda中捕获智能指针,它所指向的
 * 对象的析构函数是有动作的,如:http的sp_ctx要在析构中发送HTTP回复,如果
 * 析构函数在子线程中执行,则会出现不希望见到的多线程竞争。为此,我们在main_cb
 * 中也让它持有这个智能指针,希望智能指针所指的对象只在主线程中析构。
 *
 * 为了保证main_cb中的持有的对象能够在main_loop线程中被析构,
 * 所以这里要先task_pool.free(),然后再runInLoop(std::move(main_cpp))
 */

 auto main_cb = std::move(item->main_cb);
 auto main_loop = item->main_loop;

 {
 std::lock_guard<std::mutex> lg(d_->lock);
 d_->doing_tasks_token.erase(item->token);
 d_->task_pool.free(item);
 }

 if (main_cb && main_loop != nullptr) {
 RECORD_SCOPE();
 main_loop->runInLoop(std::move(main_cb), "WorkThread::threadProc, invoke main_cb");
 }
 }
 }

 LogDbg("thread exit");
}

bool WorkThread::shouldThreadExitWaiting() const
{
 return d_->stop_flag || !d_->undo_tasks_token_deque.empty();
}

WorkThread::Task* WorkThread::popOneTask()
{
 if (!d_->undo_tasks_token_deque.empty()) {
 TaskToken token = d_->undo_tasks_token_deque.front();
 d_->undo_tasks_token_deque.pop_front();
 return d_->undo_tasks_cabinet.free(token);
 }
 return nullptr;
}

void WorkThread::cleanup()
{
 if (d_ == nullptr)
 return;

 {
 std::lock_guard<std::mutex> lg(d_->lock);
 //! 清空task中的任务
 while (!d_->undo_tasks_token_deque.empty()) {
 auto token = d_->undo_tasks_token_deque.front();
 d_->task_pool.free(d_->undo_tasks_cabinet.free(token));
 d_->undo_tasks_token_deque.pop_front();
 }
 }

 d_->stop_flag = true;
 d_->cond_var.notify_all();

 d_->work_thread.join();

 CHECK_DELETE_RESET_OBJ(d_);
}

}
}