提升条件变量问题

Question

以下较大程序的最小代码示例将命令从客户端线程发送到 asio io_service 对象。 io_service 对象（在 Ios 类中）使用一个线程运行。当命令发送时，客户端线程一直等待，直到 Ios 对象（通过 Cmd::NotifyFinish()）通知它它已完成。

这个示例似乎可以在 Linux Ubuntu 11.04 上运行，boost 1.46 很好，但在 Windows 7 boost 1.46 上它断言。

我怀疑这与 Cmd::NotifyFinish() 中的锁定有关。当我将锁移出嵌套范围时，当在锁的范围内调用 waitConditionVariable_.notify_one() 时，它不会在 Windows 7 上崩溃。但是，boost::thread 文档指出 notify_one() 不需要在锁内调用。

堆栈跟踪（下）显示它在调用 notify_one() 时正在断言。就好像在调用 notify 之前 cmd 对象已经消失了一样……

如何使这个线程安全而不是断言？

#include <boost/asio.hpp>
#include <boost/thread/thread.hpp>
#include <boost/thread/mutex.hpp>
#include <boost/thread/locks.hpp>
#include <boost/thread/condition_variable.hpp>
#include <boost/bind.hpp>
#include <iostream>

class Cmd
{
public:
    Cmd() :   cnt_(0), waitPred_(false), waiting_(false)
    {
    }
    virtual ~Cmd()
    {
    }
    void BindInfo(int CmdSeq)
    {
        cnt_ = CmdSeq;
    }
    void NotifyFinish()
    {
        // call by service thread...
        {
            boost::mutex::scoped_lock lock(waitMutex_);
            waitPred_ = true;
            if (!waiting_)
            {
                // don't need to notify as isn't waiting
                return;
            }
        }
        waitConditionVariable_.notify_one();
    }
    void Wait()
    {
        // called by worker threads
        boost::mutex::scoped_lock lock(waitMutex_);
        waiting_ = true;
        while (!waitPred_)
            waitConditionVariable_.wait(lock);
    }
    int cnt_;
private:

    boost::mutex waitMutex_;
    boost::condition_variable waitConditionVariable_;
    bool waitPred_;
    bool waiting_;
};


class Ios
{
public:
    Ios() : timer_(ios_), cnt_(0), thread_(boost::bind(&Ios::Start, this))
    {
    }
    void Start()
    {
        timer_.expires_from_now(boost::posix_time::seconds(5));
        timer_.async_wait(boost::bind(&Ios::TimerHandler, this, _1));
        ios_.run();
    }
    void RunCmd(Cmd& C)
    {
        ios_.post(boost::bind(&Ios::RunCmdAsyn, this, boost::ref(C)));
    }

private:
    void RunCmdAsyn(Cmd& C)
    {
        C.BindInfo(cnt_++);
        C.NotifyFinish();
    }
    void TimerHandler(const boost::system::error_code& Ec)
    {
        if (!Ec)
        {
            std::cout << cnt_ << "\n";
            timer_.expires_from_now(boost::posix_time::seconds(5));
            timer_.async_wait(boost::bind(&Ios::TimerHandler, this, _1));
        }
        else
            exit(0);
    }

    boost::asio::io_service ios_;
    boost::asio::deadline_timer timer_;
    int cnt_;
    boost::thread thread_;
};

static Ios ios;

void ThreadFn()
{
    while (1)
    {
        Cmd c;
        ios.RunCmd(c);
        c.Wait();
        //std::cout << c.cnt_ << "\n";
    }
}

int main()
{
    std::cout << "Starting\n";
    boost::thread_group threads;
    const int num = 5;

    for (int i = 0; i < num; i++)
    {
        // Worker threads
        threads.create_thread(ThreadFn);
    }
    threads.join_all();

}

堆栈跟踪

msvcp100d.dll!std::_Debug_message(const wchar_t * message, const wchar_t * file, unsigned int line)  Line 15    C++
iosthread.exe!std::_Vector_const_iterator<std::_Vector_val<boost::intrusive_ptr<boost::detail::basic_cv_list_entry>,std::allocator<boost::intrusive_ptr<boost::detail::basic_cv_list_entry> > > >::_Compat(const std::_Vector_const_iterator<std::_Vector_val<boost::intrusive_ptr<boost::detail::basic_cv_list_entry>,std::allocator<boost::intrusive_ptr<boost::detail::basic_cv_list_entry> > > > & _Right)  Line 238 + 0x17 bytes   C++
iosthread.exe!std::_Vector_const_iterator<std::_Vector_val<boost::intrusive_ptr<boost::detail::basic_cv_list_entry>,std::allocator<boost::intrusive_ptr<boost::detail::basic_cv_list_entry> > > >::operator==(const std::_Vector_const_iterator<std::_Vector_val<boost::intrusive_ptr<boost::detail::basic_cv_list_entry>,std::allocator<boost::intrusive_ptr<boost::detail::basic_cv_list_entry> > > > & _Right)  Line 203 C++
iosthread.exe!std::_Vector_const_iterator<std::_Vector_val<boost::intrusive_ptr<boost::detail::basic_cv_list_entry>,std::allocator<boost::intrusive_ptr<boost::detail::basic_cv_list_entry> > > >::operator!=(const std::_Vector_const_iterator<std::_Vector_val<boost::intrusive_ptr<boost::detail::basic_cv_list_entry>,std::allocator<boost::intrusive_ptr<boost::detail::basic_cv_list_entry> > > > & _Right)  Line 208 + 0xc bytes C++
iosthread.exe!std::_Debug_range2<std::_Vector_iterator<std::_Vector_val<boost::intrusive_ptr<boost::detail::basic_cv_list_entry>,std::allocator<boost::intrusive_ptr<boost::detail::basic_cv_list_entry> > > > >(std::_Vector_iterator<std::_Vector_val<boost::intrusive_ptr<boost::detail::basic_cv_list_entry>,std::allocator<boost::intrusive_ptr<boost::detail::basic_cv_list_entry> > > > _First, std::_Vector_iterator<std::_Vector_val<boost::intrusive_ptr<boost::detail::basic_cv_list_entry>,std::allocator<boost::intrusive_ptr<boost::detail::basic_cv_list_entry> > > > _Last, const wchar_t * _File, unsigned int _Line, std::random_access_iterator_tag __formal)  Line 715 + 0xc bytes  C++
iosthread.exe!std::_Debug_range<std::_Vector_iterator<std::_Vector_val<boost::intrusive_ptr<boost::detail::basic_cv_list_entry>,std::allocator<boost::intrusive_ptr<boost::detail::basic_cv_list_entry> > > > >(std::_Vector_iterator<std::_Vector_val<boost::intrusive_ptr<boost::detail::basic_cv_list_entry>,std::allocator<boost::intrusive_ptr<boost::detail::basic_cv_list_entry> > > > _First, std::_Vector_iterator<std::_Vector_val<boost::intrusive_ptr<boost::detail::basic_cv_list_entry>,std::allocator<boost::intrusive_ptr<boost::detail::basic_cv_list_entry> > > > _Last, const wchar_t * _File, unsigned int _Line)  Line 728 + 0x6c bytes    C++
iosthread.exe!std::find_if<std::_Vector_iterator<std::_Vector_val<boost::intrusive_ptr<boost::detail::basic_cv_list_entry>,std::allocator<boost::intrusive_ptr<boost::detail::basic_cv_list_entry> > > >,bool (__cdecl*)(boost::intrusive_ptr<boost::detail::basic_cv_list_entry> const &)>(std::_Vector_iterator<std::_Vector_val<boost::intrusive_ptr<boost::detail::basic_cv_list_entry>,std::allocator<boost::intrusive_ptr<boost::detail::basic_cv_list_entry> > > > _First, std::_Vector_iterator<std::_Vector_val<boost::intrusive_ptr<boost::detail::basic_cv_list_entry>,std::allocator<boost::intrusive_ptr<boost::detail::basic_cv_list_entry> > > > _Last, bool (const boost::intrusive_ptr<boost::detail::basic_cv_list_entry> &)* _Pred)  Line 92 + 0x54 bytes    C++
iosthread.exe!std::remove_if<std::_Vector_iterator<std::_Vector_val<boost::intrusive_ptr<boost::detail::basic_cv_list_entry>,std::allocator<boost::intrusive_ptr<boost::detail::basic_cv_list_entry> > > >,bool (__cdecl*)(boost::intrusive_ptr<boost::detail::basic_cv_list_entry> const &)>(std::_Vector_iterator<std::_Vector_val<boost::intrusive_ptr<boost::detail::basic_cv_list_entry>,std::allocator<boost::intrusive_ptr<boost::detail::basic_cv_list_entry> > > > _First, std::_Vector_iterator<std::_Vector_val<boost::intrusive_ptr<boost::detail::basic_cv_list_entry>,std::allocator<boost::intrusive_ptr<boost::detail::basic_cv_list_entry> > > > _Last, bool (const boost::intrusive_ptr<boost::detail::basic_cv_list_entry> &)* _Pred)  Line 1848 + 0x58 bytes    C++
iosthread.exe!boost::detail::basic_condition_variable::notify_one()  Line 267 + 0xb4 bytes  C++
iosthread.exe!Cmd::NotifyFinish()  Line 41  C++

Answer 1

问题在于条件变量是Cmd 对象的成员，该对象由客户端线程创建，并在等待完成时被该客户端线程销毁。

所以你有一个竞争条件：

• boost::condition_variable::notify_one() 在“服务线程”上调用
• 解除阻塞等待该条件变量的客户端线程
• 然后，客户端线程可以销毁服务线程在调用notify_one 时仍在使用的条件变量。

因此，我认为您观察到“好像 cmd 对象在调用 notify 之前就消失了”几乎就是发生了什么。除了Cmd 对象在notify_one() 被调用之前没有消失，它在notify_one() 工作时消失了。您的另一条注释是“boost::thread 文档声明 notify_one() 不需要在锁内调用”是正确的，但这并不意味着条件变量可以在 notify_one() 返回之前被销毁。

您需要管理 Cmd 对象的生命周期，以便服务线程在它被销毁之前使用它 - 在调用 notify_one() 时持有 Cmd 对象中的互斥锁是一种方法那个（正如你所注意到的）。或者您可以将条件变量从Cmd 对象中提取出来，使其生命周期独立于Cmd 对象（也许shared_ptr<> 可以提供帮助）。

另外，请注意，我认为 Cmd 类的 waiting_ 成员是多余的 - 当条件变量上没有服务员时，您可以调用 notify_one() 或 notify_all() - 它已经在检查对你来说（我不认为它会伤害任何东西，只是它的复杂性不需要在 Cmd 类中）。

Answer 2

void ThreadFn()
{
    while (1)
    {
        Cmd c;
        ios.RunCmd(c);
         c.Wait();
        //std::cout << c.cnt_ << "\n";
    }
}

既然这个循环是无限的，为什么不直接放 Cmd c;超出了 while(1) 的范围，因此它在 while(1) 的每次迭代中都重用了“c”？

void ThreadFn()
{
    Cmd c;

    while (1)
    {   
        ios.RunCmd(c);
        c.Wait();
        //std::cout << c.cnt_ << "\n";
    }
}