/*
 * .============.
 * // 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 "common_loop.h"

#include <unistd.h>
#include <inttypes.h>
#include <algorithm>
#include <tbox/base/log.h>
#include <tbox/base/assert.h>
#include <tbox/base/wrapped_recorder.h>

namespace tbox {
namespace event {

using namespace std::chrono;

CommonLoop::RunFuncItem::RunFuncItem(RunId i, Func &&f, const std::string &w)
 : id(i)
 , commit_time_point(steady_clock::now())
 , func(std::move(f))
 , what(w)
{ }

Loop::RunId CommonLoop::allocRunInLoopId()
{
 run_in_loop_id_alloc_ += 2;
 if (run_in_loop_id_alloc_ == 0) //! 确保分配到的ID一定不为0
 run_in_loop_id_alloc_ += 2;

 return run_in_loop_id_alloc_;
}

Loop::RunId CommonLoop::allocRunNextId()
{
 run_next_id_alloc_ += 2;
 return run_next_id_alloc_;
}

Loop::RunId CommonLoop::runInLoop(Func &&func, const std::string &what)
{
 RECORD_SCOPE();
 std::lock_guard<std::recursive_mutex> g(lock_);

 RunId run_id = allocRunInLoopId();
 run_in_loop_func_queue_.emplace_back(RunFuncItem(run_id, std::move(func), what));

 if (sp_run_read_event_ != nullptr)
 commitRunRequest();

 auto queue_size = run_in_loop_func_queue_.size();
 if (queue_size > water_line_.run_in_loop_queue_size)
 LogNotice("run_in_loop_queue_size: %u", queue_size);

 if (queue_size > run_in_loop_peak_num_)
 run_in_loop_peak_num_ = queue_size;

 return run_id;
}

Loop::RunId CommonLoop::runInLoop(const Func &func, const std::string &what)
{
 Func func_copy(func);
 return runInLoop(std::move(func_copy), what);
}

Loop::RunId CommonLoop::runNext(Func &&func, const std::string &what)
{
 RECORD_SCOPE();
 RunId run_id = allocRunNextId();
 run_next_func_queue_.emplace_back(RunFuncItem(run_id, std::move(func), what));

 auto queue_size = run_next_func_queue_.size();
 if (queue_size > water_line_.run_next_queue_size)
 LogNotice("run_next_queue_size: %u", queue_size);

 if (queue_size > run_next_peak_num_)
 run_next_peak_num_ = queue_size;

 return run_id;
}

Loop::RunId CommonLoop::runNext(const Func &func, const std::string &what)
{
 Func func_copy(func);
 return runNext(std::move(func_copy), what);
}

Loop::RunId CommonLoop::run(Func &&func, const std::string &what)
{
 RECORD_SCOPE();
 bool can_run_next = true;
 {
 std::lock_guard<std::recursive_mutex> g(lock_);
 if (isRunningLockless() && !isInLoopThreadLockless())
 can_run_next = false;
 }

 if (can_run_next)
 return runNext(std::move(func), what);
 else
 return runInLoop(std::move(func), what);
}

Loop::RunId CommonLoop::run(const Func &func, const std::string &what)
{
 Func func_copy(func);
 return run(std::move(func_copy), what);
}

//! 从队列中删除指定run_id的项
bool CommonLoop::RemoveRunFuncItemById(RunFuncQueue &run_deqeue, RunId run_id)
{
 auto is_run_id_match = [run_id] (RunFuncItem &item) { return item.id == run_id; };

 auto end_iter = run_deqeue.end();
 auto iter = std::remove_if(run_deqeue.begin(), end_iter, is_run_id_match);
 if (iter != end_iter) {
 run_deqeue.erase(iter, end_iter);
 return true;
 }
 return false;
};

bool CommonLoop::cancel(RunId run_id)
{
 if (run_id == 0)
 return false;

 //! 先从正在执行的任务队列里删
 if (RemoveRunFuncItemById(tmp_func_queue_, run_id))
 return true;

 if (run_id & 1) { //! 奇数为runNext()的任务
 return RemoveRunFuncItemById(run_next_func_queue_, run_id);
 } else { //! 偶数为runInLoop()的任务
 std::lock_guard<std::recursive_mutex> g(lock_);
 return RemoveRunFuncItemById(run_in_loop_func_queue_, run_id);
 }
}

void CommonLoop::handleNextFunc()
{
 RECORD_SCOPE();
 /**
 * 这里使用 tmp_func_queue_ 将 run_next_func_queue_ 中的内容交换出去。然后再从
 * tmp_func_queue_ 逐一取任务出来执行。其目的在于腾空 run_next_func_queue_,让
 * 新 runNext() 的任务则会在下一轮循环中执行。从而防止无限 runNext() 引起的死循
 * 环,导致其它事件得不到处理。
 *
 * 为什么tmp_func_queue_被定义成成员变量,而不是栈上的临时变量呢?
 * 如果使用的是临时变量,那么run_next_func_queue_被swap()之后任务,就无法被cancel
 * 这不符合cancel的功能要求。于是将其作为成员变量,方便cancel()的时候,被swap
 * 的任务也可以被cancel。
 */
 run_next_func_queue_.swap(tmp_func_queue_);

 while (!tmp_func_queue_.empty()) {
 auto item = tmp_func_queue_.front();
 tmp_func_queue_.pop_front();

 auto now = steady_clock::now();
 auto delay = now - item.commit_time_point;
 if (delay > water_line_.run_next_delay)
 LogNotice("run_next_delay: %" PRIu64 " us, what: '%s'",
 delay.count()/1000, item.what.c_str());

 if (item.func) {
 RECORD_SCOPE();
 ++cb_level_;
 item.func();
 --cb_level_;
 }

 auto cost = steady_clock::now() - now;
 if (cost > water_line_.run_cb_cost)
 LogNotice("run_cb_cost: %" PRIu64 " us, what: '%s'",
 cost.count()/1000, item.what.c_str());
 }
}

bool CommonLoop::hasNextFunc() const
{
 return !run_next_func_queue_.empty();
}

void CommonLoop::handleRunInLoopFunc()
{
 RECORD_SCOPE();
 {
 //! 同handleNextFunc()的说明
 std::lock_guard<std::recursive_mutex> g(lock_);
 run_in_loop_func_queue_.swap(tmp_func_queue_);
 finishRunRequest();
 }

 while (!tmp_func_queue_.empty()) {
 auto item = tmp_func_queue_.front();
 tmp_func_queue_.pop_front();

 auto now = steady_clock::now();
 auto delay = now - item.commit_time_point;
 if (delay > water_line_.run_in_loop_delay)
 LogNotice("run_in_loop_delay: %" PRIu64 " us, what: '%s'",
 delay.count()/1000, item.what.c_str());

 if (item.func) {
 RECORD_SCOPE();
 ++cb_level_;
 item.func();
 --cb_level_;
 }

 auto cost = steady_clock::now() - now;
 if (cost > water_line_.run_cb_cost)
 LogNotice("run_cb_cost: %" PRIu64 " us, what: '%s'",
 cost.count()/1000, item.what.c_str());
 }
}

//! 清理 run_in_loop_func_queue_ 与 run_next_func_queue_ 中的任务
void CommonLoop::cleanupDeferredTasks()
{
 int remain_loop_count = 100; //! 限定次数,防止出现 runNext() 递归导致无法退出循环的问题
 while ((!run_in_loop_func_queue_.empty() || !run_next_func_queue_.empty()) && remain_loop_count-- > 0) {

 RunFuncQueue run_next_tasks = std::move(run_next_func_queue_);
 RunFuncQueue run_in_loop_tasks = std::move(run_in_loop_func_queue_);

 while (!run_next_tasks.empty()) {
 RunFuncItem &item = run_next_tasks.front();
 if (item.func) {
 RECORD_SCOPE();
 ++cb_level_;
 item.func();
 --cb_level_;
 }
 run_next_tasks.pop_front();
 }

 while (!run_in_loop_tasks.empty()) {
 RunFuncItem &item = run_in_loop_tasks.front();
 if (item.func) {
 RECORD_SCOPE();
 ++cb_level_;
 item.func();
 --cb_level_;
 }
 run_in_loop_tasks.pop_front();
 }
 }

 if (remain_loop_count == 0)
 LogWarn("found recursive actions, force quit");
}

void CommonLoop::commitRunRequest()
{
 RECORD_SCOPE();
 if (!has_commit_run_req_) {
 uint64_t one = 1;
 ssize_t wsize = write(run_event_fd_, &one, sizeof(one));
 if (wsize != sizeof(one))
 LogErr("write error");

 has_commit_run_req_ = true;
 request_stat_start_ = steady_clock::now();
 }
}

void CommonLoop::finishRunRequest()
{
 auto delay = loop_stat_start_ - request_stat_start_;
 if (delay > water_line_.wake_delay)
 LogNotice("wake_delay: %" PRIu64 " us", delay.count()/1000);

 uint64_t one = 1;
 ssize_t rsize = read(run_event_fd_, &one, sizeof(one));
 if (rsize != sizeof(one))
 LogErr("read error");

 has_commit_run_req_ = false;
}

}
}