pthread_once()中的竞争条件？

Question

我有一个std::future在一个线程,正在等待std::promise在另一个线程中设置.

编辑: 使用示例应用程序更新了问题,该应用程序将永久阻止:

更新:如果我使用一个pthread_barrier代替,下面的代码并没有阻止.

我创建了一个测试应用程序,说明了这一点:

非常基本类foo创建一个在其run函数中thread设置a promise,并在构造函数中等待promise设置它.设置后,它会递增atomic计数

然后我创建了一堆这些foo对象,将它们拆掉,然后检查我的count.

#include <iostream>
#include <thread>
#include <atomic>
#include <future>
#include <list>
#include <unistd.h>

struct foo
{
    foo(std::atomic<int>& count)
        : _stop(false)
    {
        std::promise<void> p;
        std::future <void> f = p.get_future();

        _thread = std::move(std::thread(std::bind(&foo::run, this, std::ref(p))));

        // block caller until my thread has started 
        f.wait();

        ++count; // my thread has started, increment the count
    }
    void run(std::promise<void>& p)
    {
        p.set_value(); // thread has started, wake up the future

        while (!_stop)
            sleep(1);
    }
    std::thread _thread;
    bool _stop;
};

int main(int argc, char* argv[])
{
    if (argc != 2)
    {
        std::cerr << "usage: " << argv[0] << " num_threads" << std::endl;
        return 1;
    }
    int num_threads = atoi(argv[1]);
    std::list<foo*> threads;
    std::atomic<int> count(0); // count will be inc'd once per thread

    std::cout << "creating threads" << std::endl;
    for (int i = 0; i < num_threads; ++i)
        threads.push_back(new foo(count));

    std::cout << "stopping threads" << std::endl;
    for (auto f : threads)
        f->_stop = true;

    std::cout << "joining threads" << std::endl;
    for (auto f : threads)
    {
        if (f->_thread.joinable())
            f->_thread.join();
    }

    std::cout << "count=" << count << (num_threads == count ? " pass" : " fail!") << std::endl;
    return (num_threads == count);
}

如果我在一个包含1000个线程的循环中运行它,它只需要执行几次,直到一场比赛发生并且其中一个futures永远不会被唤醒,因此应用程序永远被卡住了.

# this loop never completes
$ for i in {1..1000}; do ./a.out 1000; done

如果我现在SIGABRT是应用程序,结果堆栈跟踪显示它卡在future::wait 堆栈跟踪如下:

// main thread
    pthread_cond_wait@@GLIBC_2.3.2 () from /lib64/libpthread.so.0
    __gthread_cond_wait (__mutex=<optimized out>, __cond=<optimized out>) at libstdc++-v3/include/x86_64-unknown-linux-gnu/bits/gthr-default.h:846
    std::condition_variable::wait (this=<optimized out>, __lock=...) at ../../../../libstdc++-v3/src/condition_variable.cc:56
    std::condition_variable::wait<std::__future_base::_State_base::wait()::{lambda()#1}>(std::unique_lock<std::mutex>&, std::__future_base::_State_base::wait()::{lambda()#1}) (this=0x93a050, __lock=..., __p=...) at include/c++/4.7.0/condition_variable:93
    std::__future_base::_State_base::wait (this=0x93a018) at include/c++/4.7.0/future:331
    std::__basic_future<void>::wait (this=0x7fff32587870) at include/c++/4.7.0/future:576
    foo::foo (this=0x938320, count=...) at main.cpp:18
    main (argc=2, argv=0x7fff32587aa8) at main.cpp:52


// foo thread
    pthread_once () from /lib64/libpthread.so.0
    __gthread_once (__once=0x93a084, __func=0x4378a0 <__once_proxy@plt>) at gthr-default.h:718
    std::call_once<void (std::__future_base::_State_base::*)(std::function<std::unique_ptr<std::__future_base::_Result_base, std::__future_base::_Result_base::_Deleter> ()>&, bool&), std::__future_base::_State_base* const, std::reference_wrapper<std::function<std::unique_ptr<std::__future_base::_Result_base, std::__future_base::_Result_base::_Deleter> ()> >, std::reference_wrapper<bool> >(std::once_flag&, void (std::__future_base::_State_base::*&&)(std::function<std::unique_ptr<std::__future_base::_Result_base, ...) at include/c++/4.7.0/mutex:819
    std::promise<void>::set_value (this=0x7fff32587880) at include/c++/4.7.0/future:1206
    foo::run (this=0x938320, p=...) at main.cpp:26

我很确定我的代码中没有出错,对吧？

这是pthread实现或std :: future/std :: promise实现的问题吗？

我的库版本是:

libstdc++.so.6
libc.so.6 (GNU C Library stable release version 2.11.1 (20100118))
libpthread.so.0 (Native POSIX Threads Library by Ulrich Drepper et al Copyright (C) 2006)

Answer 1

实际上,本地promise对象的析构函数之间存在竞争条件(在构造函数的末尾和set_value()来自线程的调用.也就是说,set_value()唤醒主要部分,然后下一个销毁promise对象,但set_value()函数没有但是已经完成,还有死锁.

阅读C++ 11标准,我不确定您是否允许使用:

void promise<void>::set_value();

效果:以原子方式将值r存储在共享状态中并使该状态准备就绪.

但在其他地方:

set_value,set_exception,set_value_at_thread_exit和set_exception_at_thread_exit成员函数的行为就像在更新promise对象时获取与promise对象关联的单个互斥锁一样.

set_value()对于其他函数,例如析构函数,调用是否应该是原子的？

恕我直言,我会说不.这些效果可以与销毁互斥锁​​相媲美,而其他线程仍在锁定它.结果未定义.

解决方案是使p线程更活跃.我能想到的两个解决方案:

1. 制作p类的成员,就像迈克尔·伯尔在其他答案建议.

2. 将promise转移到线程中.

在构造函数中:

std::promise<void> p;
std::future <void> f = p.get_future();
_thread = std::thread(&foo::run, this, std::move(p));

顺便说一下,你不需要调用bind,(线程构造函数已经过载),或者调用std::move移动线程(正确的值已经是一个r值).但是,std::move对承诺的要求是强制性的.

并且线程函数没有收到引用,但移动的承诺:

void run(std::promise<void> p)
{
    p.set_value();
}

我认为这正是C++ 11定义两个不同类的原因:promise并且future:将promise转移到线程中,但是您将继续保留未来以恢复结果.

Answer 2

尝试移动,std::promise<void> p;以便它不是构造函数的本地,而是它的成员struct foo:

struct foo
{
    foo(std::atomic<int>& count)
        : _stop(false)
    {
        // std::promise<void> p;    // <-- moved to be a member
        std::future <void> f = p.get_future();

        // ...same as before...
    }
    void run(std::promise<void>& p)
    {
        // ... same ...
    }

    std::promise<void> p;   // <---
    std::thread _thread;
    bool _stop;
};

我相信可能发生的事情是你进入一场在p构造函数中被销毁的竞赛,同时p.set_value()对其进行操作promise.在set_value()完成/清理时,内部会发生一些事情; 对已经被摧毁的人提起诉讼std::promise正在破坏pthread图书馆中的某些州.

这只是猜测 - 我目前还没有准备好访问再现问题的系统.但是,p成员将确保其生命周期延长到完成set_value()通话后.