threadpool.cpp

#include "threadpool.hpp"
#include <thread>
#include "Logger.hpp"
#include <iostream>
#include <sstream>
#include <chrono>
#include "result.hpp"
using namespace std;
using namespace placeholders;
const int TASK_MAX_THRESHHOLD = 1024;
const int THREAD_MAX_THRESHHOLD = 200;

threadPool::threadPool():
	initThreadSize_(4),
	taskSize_(0),
	taskQueMaxThreshHold_(TASK_MAX_THRESHHOLD),
	mode_(poolMode::POOL_FIXED),
	isPoolRunning_(false),
	idleThreadNum_(0),
	threadMaxThreshHold_(THREAD_MAX_THRESHHOLD),
	curThreadNum_(0)
{
}
threadPool::~threadPool(){
	isPoolRunning_ = false;
	
	unique_lock<mutex> lock(taskQueMtx_);
	notEmpty_.notify_all();
	exit_.wait(lock, [&]()->bool {
		return threads_.size() == 0;
		});
}

//start the thread pool
void threadPool::start(int size)
{
	initThreadSize_ = size;
	curThreadNum_=initThreadSize_;
	isPoolRunning_ = true;
	//create Thread class
	for (int i = 0; i < initThreadSize_; i++)
	{
		auto threadptr = make_unique<Thread>(std::bind(&threadPool::threadFunc, this ,_1));
		int id = threadptr->getid();
		threads_.emplace(id, std::move(threadptr));
	}

	//start threads
	for (int i = 0; i < initThreadSize_; i++)
	{
		threads_[i]->start();
		idleThreadNum_++;
		
	}
}

//submit the task
Result threadPool::submitTask(shared_ptr<Task> task)
{
	//lock the taskQue
	unique_lock<mutex> lock(taskQueMtx_);
	//wait until the size less than the taskQueMaxThreshHold
	while (taskQue_.size() >= (size_t)taskQueMaxThreshHold_)
	{
		if (cv_status::timeout == notFull_.wait_for(lock, std::chrono::seconds(1)))
		{
			LOG_ERROR("THREADPOOL::SUBMITTASK ERROR time out\n");
			return Result(task,false);
		}
	}
	//put into the queue
	taskQue_.emplace(task);
	taskSize_++;
	//notify the consumers
	notEmpty_.notify_all();
	//if the thread is not enough 
	if (mode_ == poolMode::POOL_CACHED && taskSize_ > idleThreadNum_ &&
		curThreadNum_ < threadMaxThreshHold_)
	{
		auto threadptr = make_unique<Thread>(std::bind(&threadPool::threadFunc, this,_1));
		int id = threadptr->getid();
		threads_.emplace(id, std::move(threadptr));
		curThreadNum_++;
		idleThreadNum_++;//?????
	}

	return Result(task, true);
}

static int threadIdToInt() {
	std::stringstream ss;
	ss << std::this_thread::get_id();
	int id;
	ss >> id;
	return id;
}
void threadPool::threadFunc(int threadid)
{
	LOG_INFO("THREADPOOL::THREADHANDLER>> TID: %d\n", threadIdToInt());
	auto lastTime = std::chrono::high_resolution_clock().now();
	//cout << "THREADPOOL::THREADHANDLER>> "<<std::this_thread::get_id() << endl;
	while (true)
	{
		shared_ptr<Task> taskp = nullptr;
		{
			unique_lock<mutex> lock(taskQueMtx_);
			//cached::judge the waiting time
			
			{
				while (taskQue_.size() <= 0)
				{
					if (mode_ == poolMode::POOL_CACHED)
					{
						if (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() >= 60 && curThreadNum_ > initThreadSize_)
							{
								//clean the thread
								threads_.erase(threadid);
								curThreadNum_--;
								idleThreadNum_--;

								LOG_INFO("thread %d is cleaned\n", threadid);
								return;
							}
						}
					}
					//fixed::always wait
					else {
						notEmpty_.wait(lock);
					}
					if (isPoolRunning_ == false)
					{
						threads_.erase(threadid);
						LOG_INFO("thread %d is exited\n", threadid);
						exit_.notify_all();
						return;
					}
				}
			}
			
			idleThreadNum_--;
			LOG_INFO("THREADPOOL::THREADHANDLER>> TID: %d\n get the task", threadIdToInt());
			taskp = taskQue_.front();
			taskQue_.pop();
			taskSize_--;

			if (taskQue_.size() > 0)
			{
				notEmpty_.notify_all();
			}
			notFull_.notify_all();
		}
		
		if (taskp) {
			taskp->exec();
		}
		
		idleThreadNum_++;
		lastTime = std::chrono::high_resolution_clock().now();
	}
	threads_.erase(threadid);
	LOG_INFO("thread %d is exited\n", threadid);
	exit_.notify_all();
	return;
}