【C++11】 c++ - libc++abi.dylib:以 std::__1::system_error 类型的未捕获异常终止:互斥锁失败:参数无效

背景： 一个线程池的编写的时候 放在 windows使用的使用是正常的 ，但是放到 mac 乃至 类linux环境下就会异常 上面 c++ - libc++abi.dylib:以 std::__1::system_error 类型的未捕获异常终止:互斥锁失败:参数无效 这个错误就是 在mac电脑上报出来了 找半天

threadpool.h

#ifndef THREAD_POOL_H
#define THREAD_POOL_H#include <utility>
#include <vector>
#include <unordered_map>
#include <queue>
#include <memory>
#include <atomic>
#include <mutex>
#include <thread>
#include <condition_variable>
#include <functional>class Any {
public:Any() = default;~Any() = default;Any(const Any &) = delete;Any &operator=(const Any &) = delete;Any(Any &&) = default;Any &operator=(Any &&) = default;/// \brief 这个构造函数可以让Any类型接受任意其他的数据template<typename T>Any(T data): base_(std::make_unique<Derive < T> > (data)) {}public:template<typename T>T cast_() {Derive <T> *pd = dynamic_cast < Derive <T> * > (base_.get());if (nullptr == pd) {throw std::runtime_error("Type is unmatch!"); // 抛出标准异常}return pd->data_;}private:class Base {public:virtual ~Base() = default;};template<typename T>class Derive : public Base {public:Derive(T data): data_(data) {}public:T data_;};private:std::unique_ptr<Base> base_;
};class Semaphore {
public:Semaphore(int limit = 0): resLimit_(limit) {}~Semaphore() = default;public:/// \brief 获取一个信号量资源void wait() {std::unique_lock<std::mutex> lock(mtx_);cond_.wait(lock, [&]() -> bool {return resLimit_ > 0;});resLimit_--;}/// \brief 增加一个信号量资源void post() {std::unique_lock<std::mutex> lock(mtx_);resLimit_++;cond_.notify_all();}private:int resLimit_;std::mutex mtx_;std::condition_variable cond_;
};class Task;class Result {
public:Result(std::shared_ptr<Task> task, bool isVaild = true);~Result() = default;public:Any get();void setVal(Any any);private:Any any_;                           /// 存储任务的返回值Semaphore sem_;                     /// 线程通信信号量std::shared_ptr<Task> task_;        /// 指向对应获取返回值的任务对象std::atomic_bool isVaild_;  /// 是否有效
};/// 任务抽象基类
class Task {
public:Task();virtual ~Task() = default;public:virtual Any run() = 0;void exec();void setResult(Result *res);private:Result *result_;
};/// 线程池支持的模式
enum PoolMode {PM_FIXED = 0,     /// << 固定数量的线程PM_CACHED,    /// << 线程数量可动态增长
};/// 线程类型
class Thread {
public:using ThreadFunc = std::function<void(int)>;explicit Thread(ThreadFunc func);virtual ~Thread();public:void start();int getId() const;
private:ThreadFunc func_ = nullptr;static int generateId;int threadId_;
};/// 线程池类型
class ThreadPool {
public:explicit ThreadPool();~ThreadPool();public:/// \brief 设置线程池的工作模式void setModel(const PoolMode &mode);/// \brief 设置task任务队列上线阈值void setTaskQueMaxThreshHold(int threshHold);/// \brief 设置线程池cache模式下线程阈值void setThreadSizeThreadHold(int threshHold);/// \brief  给线程池提交任务/// \return 线程返回值Result submitTask(const std::shared_ptr<Task> sp);/// \brief 开启线程池void start(int threadSize = std::thread::hardware_concurrency());public:/// \brief 线程函数void threadFunc(int threadId);bool checkRunningState() const;public:/// 限制拷贝使用ThreadPool(const ThreadPool &other) = delete;ThreadPool &operator=(const ThreadPool &other) = delete;private:std::unordered_map<int, std::unique_ptr<Thread>> threads_;  ///线程列表size_t initThreadSize_;                         /// 初始的线程数量int threadSizeThreshHold_;                      /// 线程数量上限阈值std::atomic_int  curThreadSize_;                /// 记录当前线程池里面线程的总数量std::atomic_int idleThreadSize_;                /// 记录线程的数量std::queue<std::shared_ptr<Task>> taskQue_;     /// 任务队列std::atomic_int taskSize_;                      /// 任务数量int taskQueMaxThreshHold_;                      /// 任务队列数量上限阈值std::mutex taskQueMtx_;                         /// 保证任务队列的线程安全std::condition_variable notFull_;               /// 表示任务队列不满std::condition_variable notEmpty_;              /// 表示任务队列不空std::condition_variable exitCond_;              /// 等到线程资源全部回收PoolMode poolMode_;std::atomic_bool isRunning_;                    /// 表示当前线程池的启动状态
};#endif // THREAD_POOL

threadpool.cpp

#include "ThreadPool.h"#include <thread>
#include <iostream>constexpr int TASK_MAX_THRESHHOLD = INT32_MAX;
constexpr int THREAD_MAX_THRESHHOLD = 1024;
constexpr int THREAD_MAX_IDLE_TIME = 60;ThreadPool::ThreadPool(): initThreadSize_(0), taskSize_(0), curThreadSize_(0), idleThreadSize_(0),taskQueMaxThreshHold_(TASK_MAX_THRESHHOLD), threadSizeThreshHold_(THREAD_MAX_THRESHHOLD),poolMode_(PoolMode::PM_FIXED), isRunning_(false) {}ThreadPool::~ThreadPool() {isRunning_ = false;/// 等待线程池里面所有的线程返回  有两种状态: 阻塞 & 正在执行任务中std::unique_lock<std::mutex> lock(taskQueMtx_);notEmpty_.notify_all();exitCond_.wait(lock, [&]() -> bool { return threads_.size() == 0; });}void ThreadPool::start(int threadSize) {isRunning_ = true;/// 记录初始线程个数initThreadSize_ = threadSize;curThreadSize_ = threadSize;/// 创建线程对象for (int i = 0; i < initThreadSize_; ++i) {auto thd = std::make_unique<Thread>([this](auto &&PH1) { threadFunc(std::forward<decltype(PH1)>(PH1)); });int threadId = thd->getId();threads_.emplace(threadId, std::move(thd));}/// 启动所有线程for (int i = 0; i < initThreadSize_; ++i) {threads_[i]->start();idleThreadSize_++;}
}void ThreadPool::setModel(const PoolMode &mode) {if (checkRunningState())return;poolMode_ = mode;
}void ThreadPool::setTaskQueMaxThreshHold(int threshHold) {if (checkRunningState())return;taskQueMaxThreshHold_ = threshHold;
}Result ThreadPool::submitTask(const std::shared_ptr<Task> sp) {std::unique_lock<std::mutex> lock(taskQueMtx_);if (!notFull_.wait_for(lock, std::chrono::seconds(1),[&]() -> bool { return taskQue_.size() < taskQueMaxThreshHold_; })) {std::cerr << "task queue is full sunmit task fail." << std::endl;return Result(sp, false);}taskQue_.emplace(sp);taskSize_++;notEmpty_.notify_all();/// cache model 需要根据任务数量和空闲线程的数量, 判断是否需要创建新的线程出来/// 任务处理比较紧急 小而快的任务if (PoolMode::PM_CACHED == poolMode_&& taskSize_ > idleThreadSize_&& curThreadSize_ < threadSizeThreshHold_) {std::cout << ">>> create new threadID " << std::endl;/// 创建新的线程auto thd = std::make_unique<Thread>(std::bind(&ThreadPool::threadFunc, this, std::placeholders::_1));int threadId = thd->getId();threads_.emplace(threadId, std::move(thd));/// 启动线程threads_[threadId]->start();/// 修改线程个数curThreadSize_++;idleThreadSize_++;}return Result(sp);
}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;/// cache模式下 有可能已经创建了很多的线程, 但是空闲时间超过了60s 应该把多余的线程/// 结束回收掉???(超过initThreadSize_数量的线程要进行回收)/// 当前时间 - 上一次线程执行的时间 > 60swhile (taskSize_ == 0) {/// 线程池要结束 回收线程资源if (!isRunning_) {threads_.erase(threadId);std::cout << "threadID: " << std::this_thread::get_id() << " exit!" << std::endl;exitCond_.notify_all();return;}if (PoolMode::PM_CACHED == poolMode_) {/// 每一秒中返回一次 怎么区分: 超时返回? 还是有任务待执行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: " << std::this_thread::get_id() << " exit!" << std::endl;return;}}} else {/// 等待 notEmpty_ 条件notEmpty_.wait(lock);}}idleThreadSize_--;std::cout << "tid " << std::this_thread::get_id()<< " 获取任务成功" << std::endl;task = taskQue_.front();taskQue_.pop();taskSize_--;if (!taskQue_.empty()) {notEmpty_.notify_all();}/// 取出任务 进行通知notFull_.notify_all();}if (task != nullptr) {task->exec();}lastTime = std::chrono::high_resolution_clock::now(); /// 更新时间idleThreadSize_++;}
}bool ThreadPool::checkRunningState() const {return isRunning_;
}void ThreadPool::setThreadSizeThreadHold(int threshHold) {if (checkRunningState())return;if (PoolMode::PM_CACHED == poolMode_)threadSizeThreshHold_ = threshHold;
}线程方法实现/
int Thread::generateId = 0;Thread::Thread(Thread::ThreadFunc func): func_(func), threadId_(generateId++) {}Thread::~Thread() = default;void Thread::start() {std::thread t(func_, threadId_);t.detach();
}int Thread::getId() const {return threadId_;
}Result::Result(std::shared_ptr<Task> task, bool isVaild): task_(std::move(task)), isVaild_(isVaild) {task_->setResult(this);
}Any Result::get() {if (!isVaild_) {return {};}sem_.wait();return std::move(any_);
}void Result::setVal(Any any) {any_ = std::move(any);sem_.post();
}Task::Task() : result_(nullptr) {}void Task::exec() {if (result_) {result_->setVal(run());}
}void Task::setResult(Result *res) {result_ = res;
}

测试代码 main.cpp

#include "ThreadPool.h"#include <iostream>
#include <chrono>
#include <thread>using uLong = unsigned long long;class MyTask : public Task {
public:MyTask(int begin, int end): begin_(begin), end_(end) {}~MyTask() override = default;public:Any run() override {std::cout << "tid: " << std::this_thread::get_id()<< " begin!" << std::endl;std::this_thread::sleep_for(std::chrono::seconds(3));uLong sum = 0;for (int i = begin_; i < end_; ++i) {sum += i;}std::cout << "tid: " << std::this_thread::get_id()<< " end!" << std::endl;return {sum};}private:int begin_;int end_;
};int main(int argc, char *argv[]) {{ThreadPool pool;pool.setModel(PoolMode::PM_CACHED);pool.start(4);Result res1 = pool.submitTask(std::make_shared<MyTask>(1, 100000000));Result res2 = pool.submitTask(std::make_shared<MyTask>(100000001, 200000000));Result res3 = pool.submitTask(std::make_shared<MyTask>(200000001, 300000000));pool.submitTask(std::make_shared<MyTask>(200000001, 300000000));auto sum1 = res1.get().cast_<uLong>();auto sum2 = res2.get().cast_<uLong>();auto sum3 = res3.get().cast_<uLong>();std::cout << " slave:" << (sum1 + sum2 + sum3) << std::endl;std::cout << " slave:" << (sum1 + sum2 + sum3) << std::endl;}//    user_ulong_t sum = 0;
//    for (int i = 0; i < 300000000; ++i) {
//        sum += i;
//    }
//
//    std::cout << " master:" << (sum) << std::endl;getchar();
}

问题原点

这个崩溃 是可以追踪的 每次都崩溃在

    /// \brief 增加一个信号量资源void post() {std::unique_lock<std::mutex> lock(mtx_);resLimit_++;cond_.notify_all();}

这个cond_.notify_all();通知的地方 ，原来是 windows 平台下 或者说是 msvc sdk下的标准库的 条件变量 在析构的时候 会自己释放资源 但是另外两个平台下不会