#include "taskpool/TaskPool.hpp"


namespace otx {
namespace flash {

TaskPool::TaskPool(int parallelSize){
  this->mParallelSize = parallelSize;
  if (this->mParallelSize <= 0) {
    this->mParallelSize = 5;
  }
  this->mCurrentParallelSize = 0;
  this->mShutdown.store(false, std::memory_order::memory_order_relaxed);
}
TaskPool::~TaskPool() {
  this->shutdown();
}

void TaskPool::addTask(Task *task){
  std::lock_guard<std::mutex> lock(this->mTaskLock);
  if (this->mShutdown.load(std::memory_order::memory_order_relaxed)) {
    return;
  }
  // 并发数没到的话，就会直接开新线程。
  if(this->mCurrentParallelSize < this->mParallelSize){
    this->mCurrentParallelSize++;
    // // 任务执行完成后，由内部自动释放。
    // ExecuteContext *context = new ExecuteContext();
    // context->mPool = this;
    // context->mTask = task;
    // std::thread *t = new std::thread(TaskPool::threadWork, context);
    // this->mThreads.push_back(t);
    
    // 任务执行完成后，由内部自动释放。
    ExecuteContext *context = new ExecuteContext();
    context->mPool = this;
    context->mTask = nullptr; // not used.
    std::thread *t = new std::thread(TaskPool::threadWork, context);
    this->mThreads.push_back(t);
  }
  // 等待调度执行。
  this->mTasks.push(task);
}
void TaskPool::cancelAll(){
  // 只能取消在队列中的，不能取消正在执行的。
  std::lock_guard<std::mutex> lock(this->mTaskLock);
  int size = this->mTasks.size();
  for (int i = 0; i < size;i++){
    Task *item = this->mTasks.front();
    this->mTasks.pop();
    item->cancel();// 用户端在此处应该释放掉task内存资源。
  }
}
void TaskPool::shutdown(){
  if (this->mShutdown.load(std::memory_order::memory_order_relaxed))
    return;

  // 结束所有任务线程。
  this->mShutdown.store(true, std::memory_order::memory_order_relaxed);

  // 取消所有待执行的任务。
  cancelAll();

  std::unique_lock<std::mutex> lock(this->mTaskLock);
  
  int size = this->mThreads.size();
  for (int i = 0; i < size;i++){
    std::thread *t = this->mThreads[i];
    if(t->joinable()){
      t->join();
    }
    delete t;
  }
  this->mThreads.clear();
  lock.unlock();
}

void TaskPool::threadWork(ExecuteContext *context){
  printf("---------------Task thread start %p.\r\n", context);
  // 先执行Context中的task，往后会自动从队列中拉取。
  if(context->mTask){// 不使用了，mTask用于为nullptr。
    context->mTask->execute();
    // 注意：Task在execute中可能就被释放了，因此不能再被访问。
    context->mTask = nullptr; // 显式设置为null。
  }
  
  // 这个线程还需继续运行
  while (!context->mPool->mShutdown.load(std::memory_order::memory_order_relaxed)) {
    otx::flash::Task *task = nullptr;
    int queueSize = 0;

    std::unique_lock<std::mutex> lock(context->mPool->mTaskLock);
    queueSize = context->mPool->mTasks.size();
    if (queueSize > 0) {
      task = context->mPool->mTasks.front();
      context->mPool->mTasks.pop(); // 这里不能delete task，按照约定它由用户释放，通常是在execute末尾处。
    }
    lock.unlock();

    if (task && !context->mPool->mShutdown.load(std::memory_order::memory_order_relaxed)) {
      task->execute();
      // 注意：Task在execute中可能就被释放了，因此不能再被访问。
      task = nullptr; // 显式设置为null。
    } else if (queueSize == 0 && !context->mPool->mShutdown.load(std::memory_order::memory_order_relaxed)) {
      std::this_thread::sleep_for(std::chrono::milliseconds(20));
    }
  }

  printf("---------------Task thread end %p.\r\n",context);
  // delete context when thread end.
  delete context;
}

} // namespace flash
} // namespace otx