c++ - 如何反转 set_value() 并“停用”承诺？

Question

我在这里有一个关于同步的总 n00b 问题。我有一个'writer'线程，它在每次迭代时为一个promise分配一个不同的值'p'。我需要'reader'线程等待这个值的shared_futures然后处理它们，我的问题是我如何使用future/promise来确保reader线程在执行他们的处理任务之前等待'p'的新更新每次迭代？非常感谢。

Answer 1

一个promise/future对被设计为只携带一个值（或异常）。要执行您所描述的操作，您可能希望采用不同的工具。

如果您希望多个线程（您的读者）都停在一个共同点上，您可以考虑使用barrier.

Answer 2

您可以通过将其分配给空白承诺来“重置”承诺。

myPromise = promise< int >();

一个更完整的例子：

promise< int > myPromise;

void writer()
{
    for( int i = 0; i < 10; ++i )
    {
        cout << "Setting promise.\n";
        myPromise.set_value( i );

        myPromise = promise< int >{};       // Reset the promise.

        cout << "Waiting to set again...\n";
        this_thread::sleep_for( chrono::seconds( 1 ));
    }
}

void reader()
{
    int result;
    do
    {
        auto myFuture = myPromise.get_future();
        cout << "Waiting to receive result...\n";
        result = myFuture.get();
        cout << "Received " << result << ".\n";
    } while( result < 9 );
}

int main()
{
    std::thread write( writer );
    std::thread read( reader );

    write.join();
    read.join();

    return 0;
}

然而，这种方法的一个问题是，两个线程之间的同步可能导致编写promise::set_value()器在读取器调用之间多次调用future::get()，或者future::get()在重置承诺时被调用。这些问题可以小心避免（例如，在调用之间进行适当的休眠），但这会将我们带入黑客和猜测的领域，而不是逻辑上正确的并发。

因此，尽管可以通过将其分配给新的 Promise 来重置 Promise，但这样做往往会引发更广泛的同步问题。

Answer 3

以下代码演示了如何使用future和实现生产者/消费者模式promise。

有两个promise变量，由生产者和消费者线程使用。每个线程重置两个promise变量之一并等待另一个。

#include <iostream>
#include <future>
#include <thread>
using namespace std;

// produces integers from 0 to 99
void producer(promise<int>& dataready, promise<void>& consumed)
{
    for (int i = 0; i < 100; ++i) {
        // do some work here ...
        consumed = promise<void>{};      // reset
        dataready.set_value(i);          // make data available
        consumed.get_future().wait();    // wait for the data to be consumed
    
    }
    dataready.set_value(-1);                     // no more data
}

// consumes integers
void consumer(promise<int>& dataready, promise<void>& consumed)
{
    for (;;) {
        int n = dataready.get_future().get();    // wait for data ready
        if (n >= 0) {
            std::cout << n << ",";
            dataready = promise<int>{};  // reset
            consumed.set_value();        // mark data as consumed
            // do some work here ...
        }
        else
            break;
    }
}

int main(int argc, const char*argv[])
{
    promise<int> dataready{};
    promise<void> consumed{};

    thread th1([&] {producer(dataready, consumed); });
    thread th2([&] {consumer(dataready, consumed); });

    th1.join();
    th2.join();
    std::cout  << "\n";

    return 0;
}