#include "threadpool.h"
#include <iostream>

const int TASK_MAX_THRESHHOLD = INT32_MAX;
const int THREAD_MAX_THRESHHOLD = 10;
const int THREAD_MAX_IDLE_TIME = 10; // 单位:秒

// -------- Result 方法实现 --------
Result::Result(std::shared_ptr<Task> sp, bool isValid)
 : task_(sp)
 , isValid_(isValid)
{
 task_->setResult(this);
}
 
// setVal方法,获取任务执行完的返回值的
void Result::setVal(Any any) // Task->run() 执行之后才有
{
 this->any_ = std::move(any);
 sem_.post();
}

// 用户调用 get 方法获取 task 的返回值
Any Result::get()
{
 if (!isValid_) {
 return "";
 }
 sem_.wait();
 return std::move(any_); // 注意 Any 没有左值的拷贝构造,需要转为右值
}


// -------- Task 方法实现 --------
Task::Task()
 : result_(nullptr)
{}

void Task::exec()
{
 if (result_)
 result_->setVal(run()); // 这里发生多态调用
}

void Task::setResult(Result* res)
{
 result_ = res;
}


// -------- Thread 方法实现 --------
Thread::Thread(ThreadFunc func)
 : func_(func)
 , threadId_(generateId_ ++)
{
}

// 线程的创建中 start 和 submitTask 都是用户调用,不涉及多线程,因此不必原子类型
int Thread::generateId_ = 0;

Thread::~Thread()
{
}

// 启动线程
void Thread::start()
{
 // t 是线程对象,传入自定义的线程函数 func_,并且需要一个参数
 std::thread t(func_, threadId_); 
 t.detach(); // 设置分离线程,pthread_detach
}

int Thread::getId() const
{
 return threadId_;
}


// -------- ThreadPool 方法实现 --------

ThreadPool::ThreadPool()
 : initThreadSize_(10)
 , taskSize_(0)
 , curThreadSize_(0)
 , idleThreadSize_(0)
 , taskQueMaxThreshHold_(TASK_MAX_THRESHHOLD)
 , threadSizeThreshHold_(THREAD_MAX_THRESHHOLD)
 , poolMode_(PoolMode::MODE_FIXED)
 , isPoolRunning_(false)
{
}

ThreadPool::~ThreadPool()
{
 isPoolRunning_ = false;
 
 // 等待线程池里面所有的线程返回,两种状态:阻塞 | 正在执行任务中
 std::unique_lock<std::mutex> lock(taskQueMtx_);
 //!NOTE: 这个顺序很重要,访问 notEmpty_ 所在线程先抢到锁 wait 之后无法被 notify 而死锁
 // 不过MacOS上似乎不会出现这个死锁
 notEmpty_.notify_all();
 exitCond_.wait(lock, [&]() -> bool { return threads_.size() == 0; });
}

// 设置线程池的工作模式
void ThreadPool::setMode(PoolMode mode)
{ 
 if (checkRunningState())
 return ;
 poolMode_ = mode;
}

// 设置task任务队列上线阈值
void ThreadPool::setTaskQueMaxThreshHold(int threshhold)
{
 if (checkRunningState())
 return ;
 taskQueMaxThreshHold_ = threshhold;
}

// 设置线程池 cached 模式下线程阈值
void ThreadPool::setThreadSizeThreshHold(int threshhold)
{
 if (checkRunningState())
 return ;
 if (poolMode_ == PoolMode::MODE_CACHED) {
 threadSizeThreshHold_ = threshhold;
 }
}

bool ThreadPool::checkRunningState() const
{
 return isPoolRunning_;
}

// 给线程池「提交任务」
Result ThreadPool::submitTask(std::shared_ptr<Task> sp)
{
 // 获取锁
 std::unique_lock<std::mutex> lock(taskQueMtx_);
 
 // 线程的通信,等待任务队列有空余 wait, wait_for, wait_until
 /* while (taskQueue_.size() == taskQueMaxThreshHold_) {
 notFull_.wait(lock);
 } */
 /* notFull_.wait(lock, [&]() -> bool {
 return taskQueue_.size() < taskQueMaxThreshHold_;
 }); */

 // 用户提交任务,最长阻塞不能超过 1s,否则判断提交任务失败,返回
 if (!notFull_.wait_for(lock, std::chrono::seconds(1), [&]()->bool {
 return taskQueue_.size() < (size_t)taskQueMaxThreshHold_;
 })) {

 // 表示 notFull_ 等待了 1s,条件仍然没有满足
 std::cerr << "task queue is full, submit task failed.\n";

 // return task->getResult(); // 不合理:线程执行完task,task对象就被析构掉了
 return Result(sp, false);
 }

 // 如果有空余,把任务放入任务队列
 taskQueue_.emplace(sp);
 taskSize_ ++;

 // 因为放入了新任务,任务队列肯定不空,在 notEmpty_ 上进行通知
 notEmpty_.notify_all();
 
 // cached 模式需要根据任务数量和空闲线程的数量,判断是否需要创建新的线程出来
 // 创建新线程的条件:#1 任务数量多于 空闲线程; #2 当前线程数量小于上限阈值
 if (poolMode_ == PoolMode::MODE_CACHED && 
 taskSize_ > idleThreadSize_ &&
 curThreadSize_ < threadSizeThreshHold_)
 {
 // #1 创建新的线程对象
 auto ptr = std::make_unique<Thread>(std::bind(&ThreadPool::threadFunc, this, std::placeholders::_1));
 // auto ptr = std::make_unique<Thread>(std::bind(&ThreadPool::threadFunc_v0, this, std::placeholders::_1));
 // threads_.emplace_back(std::move(ptr));
 int threadId = ptr->getId();
 threads_.emplace(threadId, std::move(ptr));
 std::cout << ">>> create new thread... id = " << threadId << std::endl;

 // #2 启动线程
 threads_[threadId]->start();
 
 // #3 修改线程个数相关的变量
 curThreadSize_ ++;
 idleThreadSize_ ++;
 }
 
 return Result(sp, true);
}


// 开启线程池
void ThreadPool::start(int initThreadSize)
{
 isPoolRunning_ = true;
 // 记录初始线程个数
 initThreadSize_ = initThreadSize;
 curThreadSize_ = initThreadSize;

 // 创建线程对象
 for (int i = 0; i < initThreadSize_; ++ i) 
 {
 // 创建 thread 线程对象的时候,把线程函数给到 thread 线程对象
 auto ptr = std::make_unique<Thread>(Thread(std::bind(&ThreadPool::threadFunc, this, std::placeholders::_1)));
 // auto ptr = std::make_unique<Thread>(Thread(std::bind(&ThreadPool::threadFunc_v0, this, std::placeholders::_1)));
 // threads_.emplace_back(std::move(ptr));
 threads_.emplace(ptr->getId(), std::move(ptr));
 }

 // 启动所有线程 
 for (int i = 0; i < initThreadSize_; ++ i) 
 {
 threads_[i]->start(); // 执行线程函数
 idleThreadSize_ ++; // 记录空闲线程数量
 }
}


// 定义线程函数「消费任务」
void ThreadPool::threadFunc(int threadId)
{
 auto lastTime = std::chrono::high_resolution_clock().now();

 for (;;) 
 {
 std::shared_ptr<Task> task;
 {
 // 先获取锁【注意锁的粒度】
 std::unique_lock<std::mutex> lock(taskQueMtx_);
 std::cout << "tid: " << std::this_thread::get_id() << ",尝试获取任务...\n";

 while (taskQueue_.size() == 0)
 {
 // 如果没有任务了并且池已经析构就需要回收线程,然后退出
 if (!isPoolRunning_) {
 threads_.erase(threadId);
 std::cout << "~threadId: " << threadId << ' ' << std::this_thread::get_id() << " exit!\n";
 exitCond_.notify_all();
 return ;
 }

 // cached模式下,有可能已经创建了很多的线程,但是空闲时间超过60s,应该把多余的线程
 // 结束回收掉(超过initThreadSize_数量的线程要进行回收)
 // 当前时间 - 上一次线程执行的时间 > 60s
 
 // 每一秒中返回一次,如何区分超时返回还是有任务执行返回
 if (poolMode_ == PoolMode::MODE_CACHED)
 {
 // 条件变量,超时返回了
 if (std::cv_status::timeout == notEmpty_.wait_for(lock, std::chrono::seconds(1)))
 {
 auto now = std::chrono::high_resolution_clock().now();
 auto dur = std::chrono::duration_cast<std::chrono::seconds>(now - lastTime);

 if (dur.count() >= THREAD_MAX_IDLE_TIME &&
 curThreadSize_ > initThreadSize_)
 {
 threads_.erase(threadId);
 curThreadSize_ --;
 idleThreadSize_ --;

 std::cout << "threadId: " << threadId << ' ' << std::this_thread::get_id() << " exit!\n";
 return ;
 }
 }
 }
 else 
 {
 // 只要任务队列里面没有任务就要等待 notEmpty_ 条件
 notEmpty_.wait(lock);
 }
 }

 idleThreadSize_ --;
 std::cout << "tid: " << std::this_thread::get_id() << ", 获取成功.\n";

 // 从任务队列中取一个任务出来
 task = taskQueue_.front();
 taskQueue_.pop();
 taskSize_ --;

 // #通知1:如果依然有剩余任务,可以继续通知其他线程执行任务
 if (taskQueue_.size() > 0) {
 notEmpty_.notify_all();
 }

 // #通知2:取出一个任务进行通知,可以继续提交生产任务
 notFull_.notify_all();
 }

 // 当前线程负责执行这个任务
 if (task != nullptr) {
 // task->run(); 
 task->exec(); // #1 执行任务;#2 把任务的返回值 setVal 方法给到 Result
 }
 idleThreadSize_ ++; // 任务执行完了

 // 更新线程执行完任务的时间
 lastTime = std::chrono::high_resolution_clock().now();
 }
}


// 初始版本的线程函数,不等待线程执行任务结束
void ThreadPool::threadFunc_v0(int threadId)
{
 auto lastTime = std::chrono::high_resolution_clock().now();

 while (isPoolRunning_) 
 {
 std::shared_ptr<Task> task;
 {
 // 先获取锁【注意锁的粒度】
 std::unique_lock<std::mutex> lock(taskQueMtx_);
 std::cout << "tid: " << std::this_thread::get_id() << ",尝试获取任务...\n";

 // 锁 + 双重判断
 while (isPoolRunning_ && taskQueue_.size() == 0)
 {
 // cached模式下,有可能已经创建了很多的线程,但是空闲时间超过60s,应该把多余的线程
 // 结束回收掉(超过initThreadSize_数量的线程要进行回收)
 // 当前时间 - 上一次线程执行的时间 > 60s
 
 // 每一秒中返回一次,如何区分超时返回还是有任务执行返回
 if (poolMode_ == PoolMode::MODE_CACHED)
 {
 // 条件变量,超时返回了
 if (std::cv_status::timeout == notEmpty_.wait_for(lock, std::chrono::seconds(1)))
 {
 auto now = std::chrono::high_resolution_clock().now();
 auto dur = std::chrono::duration_cast<std::chrono::seconds>(now - lastTime);

 if (dur.count() >= THREAD_MAX_IDLE_TIME &&
 curThreadSize_ > initThreadSize_)
 {
 threads_.erase(threadId);
 curThreadSize_ --;
 idleThreadSize_ --;

 std::cout << "threadId: " << threadId << ' ' << std::this_thread::get_id() << " exit!\n";
 return ;
 }
 }
 }
 else 
 {
 // 只要任务队列里面没有任务就要等待 notEmpty_ 条件
 notEmpty_.wait(lock);
 }

 // 如果没有任务了并且池已经析构就需要回收线程,然后退出
 // if (!isPoolRunning_) {
 // threads_.erase(threadId);
 // std::cout << "~threadId: " << threadId << ' ' << std::this_thread::get_id() << " 111exit!\n";
 // exitCond_.notify_all();
 // return ;
 // }
 }

 if (!isPoolRunning_) {
 break;
 }

 idleThreadSize_ --;
 std::cout << "tid: " << std::this_thread::get_id() << ", 获取成功.\n";

 // 从任务队列中取一个任务出来
 task = taskQueue_.front();
 taskQueue_.pop();
 taskSize_ --;

 // #通知1:如果依然有剩余任务,可以继续通知其他线程执行任务
 if (taskQueue_.size() > 0) {
 notEmpty_.notify_all();
 }

 // #通知2:取出一个任务进行通知,可以继续提交生产任务
 notFull_.notify_all();
 }

 // 当前线程负责执行这个任务
 if (task != nullptr) {
 // task->run(); 
 task->exec(); // #1 执行任务;#2 把任务的返回值 setVal 方法给到 Result
 }
 idleThreadSize_ ++; // 任务执行完了

 // 更新线程执行完任务的时间
 lastTime = std::chrono::high_resolution_clock().now();
 }

 threads_.erase(threadId);
 std::cout << "~threadId: " << threadId << ' ' << std::this_thread::get_id() << " 222exit!\n";
 exitCond_.notify_all();
}