C++条件变量wait_until超时异常排查与修复咨询

我正在学习C++条件变量与线程的配合使用，目标是让主线程创建若干指定时长的Countdown任务，子线程通过condition_variable的超时机制在时间到期后通知调用者。但运行示例代码时，每次结果不同，有时报错，且子线程实际超时远短于设定时长。我知晓其他线程计时方案，但希望保留基于条件变量无需轮询的实现，怀疑问题出在Holder类中map添加Countdown对象时的移动构造部分，请求正确实现建议。

示例代码 #

#include <chrono>
#include <condition_variable>
#include <functional>
#include <iostream>
#include <map>
#include <mutex>
#include <thread>

using Time = std::chrono::system_clock;
using Seconds = std::chrono::seconds;
using Timepoint = Time::time_point;

class Countdown {
private:
    Timepoint               target;
    std::thread             t;
    std::condition_variable cv;
    std::mutex              cv_m;
    unsigned int            guid;
    std::string             name;

public:
    Countdown() 
    { // Needed to compile, but doesn't appear to run 
        std::cout << "empty Countdown constructor" << std::endl;
    }

    Countdown(unsigned int guid_, std::string name_, unsigned int waitFor, std::function<void(unsigned int)> callback)
        : guid(guid_)
        , name(name_)
        , target(Time::now() + Seconds(waitFor))
    {

        auto exec_run = [this, guid_, waitFor, callback]() mutable {
            std::unique_lock<std::mutex> lk(cv_m);
            std::cout << "[Thread " << guid_ << "] waiting for " << waitFor << " seconds." << std::endl;
 
            Timepoint before = Time::now();
            if (cv.wait_until(lk, target) == std::cv_status::timeout)
            {
                Timepoint after = Time::now();
                std::chrono::duration<float> difference = after - before;
                std::cout << "[Thread " << guid_ << "] Elapsed " << difference.count() << " seconds." << std::endl;
                callback(guid_);
            } 
        };
        
        t = std::thread(exec_run);
    }

    Countdown(Countdown &&from) // move constructor
    {
        //std::cout << "Countdown move constructor" << std::endl; 
        target = from.target;
        t = std::move(from.t);
        name = from.name;
        guid = from.guid;
    }

    ~Countdown()
    {
        //std::cout << "~Countdown()" << std::endl; 
        if (t.joinable()) t.join();
    }
};

class Holder {
private:
    std::map<unsigned int, Countdown>   waitlist;
    unsigned int                        id;
    std::vector<unsigned int>           completed;

public:
    Holder()
        : id(0)
    { }

    // Create a new task with a name for WaitFor (s) period of time
    unsigned int addTask(std::string name, unsigned int waitFor) {
        id++;
        waitlist.emplace(std::pair(id, Countdown(id, name, waitFor, 
                                  std::bind(&Holder::taskComplete, this, std::placeholders::_1))));

        return id;
    }

    void taskComplete(unsigned int id)
    {
        std::cout << "[Thread " << id << "] taskComplete" << std::endl;
        // Add task id to the completed list to be picked up by main thread
        completed.push_back(id);
    }

    void cleanupCompleted()
    {
        // Purge the completed entries from the waitlist
        for (auto& id : completed)
        {
            std::cout << "[Main] Erasing task: " << id << std::endl;
            waitlist.erase(id);
        }

        // Empty the completed list
        completed.clear();
    }
};

int main()
{
    Holder *h = new Holder();
    // Create a task which spawns a thread, which notifies us when complete
    unsigned int id1 = h->addTask("fluffy", 1); // 1 second task
    unsigned int id2 = h->addTask("woof", 4);   // 4 second task
    std::cout << "[Main]: Done adding tasks.." << std::endl;
 
    // Rest a while..
    std::this_thread::sleep_for(Seconds(5));
    h->cleanupCompleted();

    // Just to show the main thread continues on.
    std::cout << "[Main]: Doing other stuff.." << std::endl;
    delete(h);

    return 0;
}

实际/预期结果 #

[Main]: Done adding tasks..
[Thread 2] waiting for 4 seconds.
[Thread 1] waiting for 1 seconds.
[Thread 2] Elapsed 8.243e-06 seconds. **(预期应为 ~4 秒)**
[Thread 2] taskComplete
[Thread 1] Elapsed 0.000124505 seconds. **(预期应为 ~1 秒)**
[Thread 1] taskComplete
[Main] Erasing task: 2
[Main] Erasing task: 1
[Main]: Doing other stuff..

核心问题 #

1. 移动构造未处理条件变量与互斥锁：std::condition_variable和std::mutex不可移动、不可复制，原移动构造函数仅复制了线程、时间点等成员，未处理这两个核心同步对象。当Countdown被移动到map中时，原对象的锁和条件变量被销毁，新对象使用的是未初始化的同步对象，导致wait_until直接超时返回。
2. lambda捕获的this指针失效：构造函数中lambda捕获的this指向临时对象，当临时对象被移动到map后，原对象销毁，lambda中的this变为野指针，访问target等成员会触发未定义行为。
3. 线程安全问题：Holder的completed向量被主线程和子线程同时读写，存在数据竞争。

修复步骤 #

1. 调整同步对象的存储方式 #

将condition_variable和mutex改为动态分配（用std::unique_ptr），这样移动时仅转移指针所有权，避免同步对象失效：

// Countdown类成员修改
private:
    Timepoint               target;
    std::thread             t;
    std::unique_ptr<std::condition_variable> cv;
    std::unique_ptr<std::mutex>              cv_m;
    unsigned int            guid;
    std::string             name;

2. 修正构造函数与lambda捕获 #

构造函数中初始化智能指针，lambda直接捕获时间点和智能指针，不再依赖this：

Countdown(unsigned int guid_, std::string name_, unsigned int waitFor, std::function<void(unsigned int)> callback)
    : guid(guid_)
    , name(std::move(name_))
    , target(Time::now() + Seconds(waitFor))
    , cv(std::make_unique<std::condition_variable>())
    , cv_m(std::make_unique<std::mutex>())
{
    auto exec_run = [target=this->target, cv=std::move(this->cv), cv_m=std::move(this->cv_m), 
                     guid_, waitFor, callback]() mutable {
        std::unique_lock<std::mutex> lk(*cv_m);
        std::cout << "[Thread " << guid_ << "] waiting for " << waitFor << " seconds." << std::endl;
 
        Timepoint before = Time::now();
        if (cv->wait_until(lk, target) == std::cv_status::timeout)
        {
            Timepoint after = Time::now();
            std::chrono::duration<float> difference = after - before;
            std::cout << "[Thread " << guid_ << "] Elapsed " << difference.count() << " seconds." << std::endl;
            callback(guid_);
        } 
    };
    
    t = std::thread(std::move(exec_run));
}

3. 修复线程安全问题 #

给Holder的completed向量添加互斥锁保护：

// Holder类添加成员
private:
    std::mutex completed_mtx;

// 修改taskComplete方法
void taskComplete(unsigned int id)
{
    std::cout << "[Thread " << id << "] taskComplete" << std::endl;
    std::lock_guard<std::mutex> lk(completed_mtx);
    completed.push_back(id);
}

// 修改cleanupCompleted方法
void cleanupCompleted()
{
    std::lock_guard<std::mutex> lk(completed_mtx);
    std::vector<unsigned int> temp_completed = std::move(completed);
    completed.clear();

    for (auto id : temp_completed)
    {
        std::cout << "[Main] Erasing task: " << id << std::endl;
        waitlist.erase(id);
    }
}

4. 优化map插入逻辑 #

使用std::piecewise_construct直接在map中构造Countdown，避免不必要的移动操作：

unsigned int addTask(std::string name, unsigned int waitFor) {
    id++;
    waitlist.emplace(std::piecewise_construct,
                     std::forward_as_tuple(id),
                     std::forward_as_tuple(id, std::move(name), waitFor, 
                                           std::bind(&Holder::taskComplete, this, std::placeholders::_1)));
    return id;
}

完整修复后的代码 #

#include <chrono>
#include <condition_variable>
#include <functional>
#include <iostream>
#include <map>
#include <mutex>
#include <thread>
#include <memory>

using Time = std::chrono::system_clock;
using Seconds = std::chrono::seconds;
using Timepoint = Time::time_point;

class Countdown {
private:
    Timepoint               target;
    std::thread             t;
    std::unique_ptr<std::condition_variable> cv;
    std::unique_ptr<std::mutex>              cv_m;
    unsigned int            guid;
    std::string             name;

public:
    Countdown() = delete; // 禁用默认构造函数，避免误用

    Countdown(unsigned int guid_, std::string name_, unsigned int waitFor, std::function<void(unsigned int)> callback)
        : guid(guid_)
        , name(std::move(name_))
        , target(Time::now() + Seconds(waitFor))
        , cv(std::make_unique<std::condition_variable>())
        , cv_m(std::make_unique<std::mutex>())
    {
        auto exec_run = [target=this->target, cv=std::move(this->cv), cv_m=std::move(this->cv_m), 
                         guid_, waitFor, callback]() mutable {
            std::unique_lock<std::mutex> lk(*cv_m);
            std::cout << "[Thread " << guid_ << "] waiting for " << waitFor << " seconds." << std::endl;
 
            Timepoint before = Time::now();
            if (cv->wait_until(lk, target) == std::cv_status::timeout)
            {
                Timepoint after = Time::now();
                std::chrono::duration<float> difference = after - before;
                std::cout << "[Thread " << guid_ << "] Elapsed " << difference.count() << " seconds." << std::endl;
                callback(guid_);
            } 
        };
        
        t = std::thread(std::move(exec_run));
    }

    Countdown(Countdown &&from) = default;

    ~Countdown()
    {
        if (t.joinable()) 
            t.join();
    }

    Countdown(const Countdown&) = delete;
    Countdown& operator=(const Countdown&) = delete;
};

class Holder {
private:
    std::map<unsigned int, Countdown>   waitlist;
    unsigned int                        id;
    std::vector<unsigned int>           completed;
    std::mutex                          completed_mtx;

public:
    Holder()
        : id(0)
    { }

    unsigned int addTask(std::string name, unsigned int waitFor) {
        id++;
        waitlist.emplace(std::piecewise_construct,
                         std::forward_as_tuple(id),
                         std::forward_as_tuple(id, std::move(name), waitFor, 
                                               std::bind(&Holder::taskComplete, this, std::placeholders::_1)));
        return id;
    }

    void taskComplete(unsigned int id)
    {
        std::cout << "[Thread " << id << "] taskComplete" << std::endl;
        std::lock_guard<std::mutex> lk(completed_mtx);
        completed.push_back(id);
    }

    void cleanupCompleted()
    {
        std::lock_guard<std::mutex> lk(completed_mtx);
        std::vector<unsigned int> temp_completed = std::move(completed);
        completed.clear();

        for (auto id : temp_completed)
        {
            std::cout << "[Main] Erasing task: " << id << std::endl;
            waitlist.erase(id);
        }
    }
};

int main()
{
    Holder h;
    unsigned int id1 = h.addTask("fluffy", 1);
    unsigned int id2 = h.addTask("woof", 4);
    std::cout << "[Main]: Done adding tasks.." << std::endl;
 
    std::this_thread::sleep_for(Seconds(5));
    h.cleanupCompleted();

    std::cout << "[Main]: Doing other stuff.." << std::endl;

    return 0;
}

修复后预期结果 #

[Main]: Done adding tasks..
[Thread 1] waiting for 1 seconds.
[Thread 2] waiting for 4 seconds.
[Thread 1] Elapsed 1.000xx seconds.
[Thread 1] taskComplete
[Thread 2] Elapsed 4.000xx seconds.
[Thread 2] taskComplete
[Main] Erasing task: 1
[Main] Erasing task: 2
[Main]: Doing other stuff..

内容的提问来源于stack exchange，提问作者Paul