死锁问题分析和解决——资源回收时

1.描述问题

在完成线程池核心功能功能时，没有遇到太大的问题（Any,Result,Semfore的设计），在做线程池资源回收时，遇到了死锁的问题        

1、在ThreadPool的资源回收，等待线程池所有线程退出时，发生死锁问题，导致进程无法退出

死锁代码：

#include "threadpool.h"#include <thread>
#include <iostream>const int TASK_MAX_THRESHHOLD = INT32_MAX;
const int THREAD_MAX_THRESHHOLD = 100;
const int THREAD_MAX_IDLE_TIME = 60;//单位：秒//线程池构造
ThreadPool::ThreadPool(): initThreadSize_(0), taskSize_(0), idleThreadSize_(0)//刚开始时还没有线程, curThreadSize_(0), taskQueMaxThreshHold_(TASK_MAX_THRESHHOLD), threadSizeThreshHold_(THREAD_MAX_THRESHHOLD), poolMode_(PoolMode::MODE_FIXED), isPoolRunning_(false)
{}//线程池析构
ThreadPool::~ThreadPool()
{isPoolRunning_ = false;notEmpty_.notify_all();//等待线程池里面所有的线程返回  有两种状态：阻塞 & 正在执行任务中std::unique_lock<std::mutex> lock(taskQueMtx_);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;}
}// 给线程池提交任务  用户调用该接口，传入任务对象，生产任务
Result ThreadPool::submitTask(std::shared_ptr<Task> sp)
{//获取锁std::unique_lock<std::mutex> lock(taskQueMtx_);//线程的通信  等待任务队列有空余// 用户提交任务，最长不能阻塞超过1s,否则判断提交任务失败，返回//while (taskQue_.size() == taskQueMaxThreshHold_)//{//	notFull_.wait(lock);//}/** wait:直到等待满足条件（第二个参数lamada）才返回* wait_for:满足条件返回真,到了约定的时间段（5s）返回假* wait_until:满足条件返回真,到了约定的时间点（下周一）返回假*/if (!notFull_.wait_for(lock, std::chrono::seconds(1),[&]()->bool {return taskQue_.size() < (size_t)taskQueMaxThreshHold_; }))//等同于上面的语句，参数：需要释放的锁  函数对象(要能满足条件变量）//任务队列中的任务数小于上限的阈值，否则就阻塞在这句{//表示notFull_等待1s，条件依然没有满足std::cerr << "task queue is full,submit task fail." << std::endl;//return task->getResult(); //Task  Result  线程执行完task，task对象就被析构掉了return Result(sp, false);//返回临时对象，应该自动匹配右值的资源转移，如果编译不通过，把C++标准调高一点}//如果有空余，把任务放入任务队列中taskQue_.emplace(sp);taskSize_++;//因为新放了任务，任务队列肯定不空了，在notEmpty_上进行通知,赶快分配线程执行任务notEmpty_.notify_all();//cached模式 任务处理比较紧急 场景：小而快的任务 需要根据任务数量和空闲线程的数量，判断是否需要创建新的线程出来if (poolMode_ == PoolMode::MODE_CACHED&& taskSize_ > idleThreadSize_&& curThreadSize_ < threadSizeThreshHold_){std::cout << ">>> create new thread..." << std::this_thread::get_id() << " exit!" << std::endl;//创建新的线程对象auto ptr = std::make_unique<Thread>(std::bind(&ThreadPool::threadFunc, this, std::placeholders::_1));int threadId = ptr->getId();threads_.emplace(threadId, std::move(ptr));//threads_.emplace_back(std::move(ptr));//启动线程threads_[threadId]->start();//修改线程个数相关的变量curThreadSize_++;idleThreadSize_++;}//返回任务的Result对象return Result(sp);// return task->getResult();
}//开启线程池
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>(std::bind(&ThreadPool::threadFunc, this, std::placeholders::_1));int threadId = ptr->getId();threads_.emplace(threadId, std::move(ptr));//threads_.emplace_back(std::move(ptr));//unique_ptr将左值引用的拷贝构造和赋值都delete了，需要右值（进行资源转移）}//启动所有线程 std::vector<Thread*> threads_;for (int i = 0; i < initThreadSize_; i++){threads_[i]->start(); //需要去执行一个线程函数idleThreadSize_++;//记录初始空闲线程的数量}
}//定义线程函数  线程池的所有线程从任务队列里面消费任务
void ThreadPool::threadFunc(int threadid) //线程函数返回，相应的线程也就结束了
{/*std::cout << "begin threadFunc tid:" << std::this_thread::get_id() << std::endl;std::cout << "end threadFunc tid:" << std::this_thread::get_id() << std::endl;*/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()<< "尝试获取任务..." << std::endl;//cached模式下，有可能已经创建了很多的线程，但是空闲时间超过60s,应该把多余的线程结束回收掉(超过initThreadSize_数量的线程要进行回收)//当前时间-上一次线程执行的时间>60s//每一秒中返回一次  怎么区分：超时返回？还是有任务待执行返回while (taskQue_.size() == 0){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_){//开始回收当前线程//记录线程数量的相关变量的值修改//把线程对象从线程列表容器中删除  没有办法  threadFunc  <=>thread对象//threadid=>thread对象=》删除threads_.erase(threadid);// 这个id不是std::this_thread::getid()  是自己生成的，我们自定义的curThreadSize_--;idleThreadSize_--;std::cout << "threadid:" << std::this_thread::get_id() << "exit!" << std::endl;return;}}}else{//等待notEmpty条件notEmpty_.wait(lock);}//线程池结束，回收线程资源if (!isPoolRunning_){threads_.erase(threadid);// 这个id不是std::this_thread::getid()  是自己生成的，我们自定义的std::cout << "threadid:" << std::this_thread::get_id() << "exit!" << std::endl;exitCond_.notify_all();return;}}idleThreadSize_--;//唤醒线程工作，空闲线程-1std::cout << "tid:" << std::this_thread::get_id()<< "获取任务成功..." << std::endl;//从任务队列中取一个任务出来task = taskQue_.front();taskQue_.pop();taskSize_--;//如果依然有剩余任务，继续通知其它的线程执任务if (taskQue_.size() > 0){notEmpty_.notify_all();}//取出一个任务，进行通知，通知可以继续提交生产任务notFull_.notify_all();}//就应该把锁释放掉//当前线程负责执行这个任务if (task != nullptr){//task->run();//执行任务；把任务的返回值setVal方法给到Resulttask->exec();}idleThreadSize_++;//线程执行完任务，空闲线程+1lastTime = std::chrono::high_resolution_clock().now();//更新线程执行完任务的时间}threads_.erase(threadid);// 这个id不是std::this_thread::getid()  是自己生成的，我们自定义的std::cout << "threadid:" << std::this_thread::get_id() << "exit!" << std::endl;exitCond_.notify_all();
}bool ThreadPool::checkRunningState() const
{return isPoolRunning_;
}///   线程方法实现
int Thread::generateId_ = 0;//线程构造
Thread::Thread(ThreadFunc func):func_(func),threadId_(generateId_++)
{}//线程析构
Thread::~Thread(){}//启动线程
void Thread::start()
{//创建一个线程来执行一个线程函数std::thread t(func_, threadId_);//C++11来说 线程对象t  和线程函数func_t.detach();//设置分离线程，线程对象t出作用域会析构，但是线程函数不能结束否则程序会挂掉，所以要将线程分离出去，做到二者互不影响//pthread_detach  pthread_t设置成分离线程//主线程要用pthread_join回收线程，防止孤儿线程的出现}	//获取线程id
int Thread::getId()const
{return threadId_;
}///   Task方法实现
Task::Task():result_(nullptr)
{}void Task::exec()
{result_->setVal(run());//这里发生多态调用
}void Task::setResult(Result* res)
{result_ = res;
}///   Result方法的实现
Result::Result(std::shared_ptr<Task> task, bool isValid):isValid_(isValid),task_(task)
{task_->setResult(this);
}Any Result::get() // 用户调用的
{if (!isValid_){return "";}//task任务如果没有执行完，这里会阻塞用户的线程sem_.wait();//用户调用get时，如果任务在线程池中，还没有被执行完，那么调用get方法的线程就会阻塞住return std::move(any_);//右值引用
}void Result::setVal(Any any)//谁调用的呢？？
{//存储task的返回值this->any_ = std::move(any);sem_.post();//已经获取的任务的返回值，增加信号量资源
}

 

我们的资源回收代码如下：

//线程池析构
ThreadPool::~ThreadPool()
{isPoolRunning_ = false;notEmpty_.notify_all();//等待线程池里面所有的线程返回  有两种状态：阻塞 & 正在执行任务中std::unique_lock<std::mutex> lock(taskQueMtx_);exitCond_.wait(lock, [&]()->bool {return threads_.size() == 0; });
}

现在，有的线程没有被回收，线程队列中还有线程，所以就一直阻塞等待了。
线程池的那个线程为什么没有被回收掉？
（时而出现，时而不出现的问题）

我们通过在windows上调试：

我们通过在Linux上进行gdb调试

主要通过gdb attach到正在运行的进程，通过info threads，thread tid，bt等命令查看各个线程的调用堆栈信息，结合项目代码，定位到发生死锁的代码片段，分析死锁问题发生的原因

2.分析问题

原先针对上面的2种情况的处理方法如下：

第3种情况：
有的线程执行完任务，又进入while循环了

在这里有2种情况：
1、pool线程先获取到锁，线程池的线程获取不到锁，阻塞。
此时pool线程看wait条件，size>0，不满足条件，就进入等待wait状态了，并且把互斥锁mutex释放掉。
线程池的线程就获取到锁了，发现任务队列没有任务了，这个任务就在notEmpty条件变量上wait，但是此时pool线程没有办法再对这个条件变量notify了。
发生死锁了！！！

2、线程池里的线程先获取到锁，发生任务队列为空，在条件变量notEmpty上wait了，释放锁，然后pool线程抢到锁，只是看exitCond条件变量的wait条件，看size还是大于0，还是死锁了。

解决方法：pool线程获取到锁后再notify

//线程池析构
ThreadPool::~ThreadPool()
{isPoolRunning_ = false;//等待线程池里面所有的线程返回  有两种状态：阻塞 & 正在执行任务中std::unique_lock<std::mutex> lock(taskQueMtx_);notEmpty_.notify_all();exitCond_.wait(lock, [&]()->bool {return threads_.size() == 0; });
}

 我们在消费者线程进行锁+双重判断：

//定义线程函数   线程池的所有线程从任务队列里面消费任务
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()<< "尝试获取任务..." << std::endl;//cached模式下，有可能已经创建了很多的线程，但是空闲时间超过60s，应该把多余的线程//结束回收掉（超过initThreadSize_数量的线程要进行回收）//当前时间 - 上一次线程执行的时间 > 60s//每一秒中返回一次   怎么区分：超时返回？还是有任务待执行返回//锁 + 双重判断while (taskQue_.size() == 0){//线程池要结束，回收线程资源if (!isPoolRunning_){threads_.erase(threadid);//std::this_thread::getid()std::cout << "threadid:" << std::this_thread::get_id() << " exit!"<< std::endl;exitCond_.notify_all();return;//线程函数结束，线程结束}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_)//任务数量大于空闲线程数量{//开始回收当前线程//记录线程数量的相关变量的值修改//把线程对象从线程列表容器中删除   没有办法 threadFunc《=》thread对象//通过threadid => thread对象 => 删除threads_.erase(threadid);//std::this_thread::getid()curThreadSize_--;idleThreadSize_--;std::cout << "threadid:" << std::this_thread::get_id() << " exit!"<< std::endl;return;}}}else{//等待notEmpty条件notEmpty_.wait(lock);}//if (!isPoolRunning_)//{//	threads_.erase(threadid);//std::this_thread::getid()//	std::cout << "threadid:" << std::this_thread::get_id() << " exit!"//		<< std::endl;//	exitCond_.notify_all();//	return;//结束线程函数，就是结束当前线程了!//}}idleThreadSize_--;std::cout << "tid:" << std::this_thread::get_id()<< "获取任务成功..." << std::endl;//从任务队列种取一个任务出来task = taskQue_.front();taskQue_.pop();taskSize_--;//如果依然有剩余任务，继续通知其它得线程执行任务if (taskQue_.size() > 0){notEmpty_.notify_all();}//取出一个任务，进行通知，通知可以继续提交生产任务notFull_.notify_all();} //就应该把锁释放掉//当前线程负责执行这个任务if (task != nullptr){//task->run();//执行任务；把任务的返回值setVal方法给到Result，基类指针调用派生类对象的同名覆盖方法task->exec();//用户还是使用run方法}idleThreadSize_++;lastTime = std::chrono::high_resolution_clock().now();//更新线程执行完任务的时间}
}