如何正确退出可能正在等待std::condition_variable的std::线程
How to correctly exit a std::thread that might be waiting on a std::condition_variable?
我有一个类,它实现了一个线程生产者/消费者系统,使用一个互斥锁和两个条件变量进行同步。当有物品需要使用时,生产者向消费者线程发出信号,而消费者在消费了物品后向生产者线程发出信号。线程继续生产和消费,直到析构函数通过设置一个布尔变量请求它们退出。因为任何一个线程都可能在等待一个条件变量,所以我必须实现对quit变量的第二次检查,这感觉是错误的和混乱的…
我将问题简化为以下示例(在GNU/Linux上使用g++4.7):
// C++11and Boost required.
#include <cstdlib> // std::rand()
#include <cassert>
#include <boost/circular_buffer.hpp>
#include <atomic>
#include <chrono>
#include <condition_variable>
#include <iostream>
#include <mutex>
#include <thread>
#include <vector>
// Creates a single producer and single consumer thread.
class prosumer
{
public:
// Create the circular buffer and start the producer and consumer thread.
prosumer()
: quit_{ false }
, buffer_{ circular_buffer_capacity }
, producer_{ &prosumer::producer_func, this }
, consumer_{ &prosumer::consumer_func, this }
{}
// Set the quit flag and wait for the threads to exit.
~prosumer()
{
quit_ = true;
producer_.join();
consumer_.join();
}
private:
// Thread entry point for the producer.
void producer_func()
{
// Value to add to the ringbuffer to simulate data.
int counter = 0;
while ( quit_ == false )
{
// Simulate the production of some data.
std::vector< int > produced_items;
const auto items_to_produce = std::rand() % circular_buffer_capacity;
for ( int i = 0; i < items_to_produce; ++i )
{
produced_items.push_back( ++counter );
}
// Get a lock on the circular buffer.
std::unique_lock< std::mutex > lock( buffer_lock_ );
// Wait for the buffer to be emptied or the quit flag to be set.
buffer_is_empty_.wait( lock, [this]()
{
return buffer_.empty() == true || quit_ != false;
} );
// Check if the thread was requested to quit.
if ( quit_ != false )
{
// Don't let the consumer deadlock.
buffer_has_data_.notify_one();
break;
}
// The buffer is locked by this thread. Put the data into it.
buffer_.insert( std::end( buffer_ ), std::begin( produced_items ), std::end( produced_items ) );
// Notify the consumer that the buffer has some data in it.
buffer_has_data_.notify_one();
}
std::cout << "producer thread quitn";
}
// Thread entry for the consumer.
void consumer_func()
{
int counter_check = 0;
while ( quit_ == false )
{
std::unique_lock< std::mutex > lock( buffer_lock_ );
// Wait for the buffer to have some data before trying to read from it.
buffer_has_data_.wait( lock, [this]()
{
return buffer_.empty() == false || quit_ != false;
} );
// Check if the thread was requested to quit.
if ( quit_ != false )
{
// Don't let the producer deadlock.
buffer_is_empty_.notify_one();
break;
}
// The buffer is locked by this thread. Simulate consuming the data.
for ( auto i : buffer_ ) assert( i == ++counter_check );
buffer_.clear();
// Notify the producer thread that the buffer is empty.
buffer_is_empty_.notify_one();
}
std::cout << "consumer thread quitn";
}
// How many items the circular buffer can hold.
static const int circular_buffer_capacity = 64;
// Flag set in the destructor to signal the threads to stop.
std::atomic_bool quit_;
// Circular buffer to hold items and a mutex for synchronization.
std::mutex buffer_lock_;
boost::circular_buffer< int > buffer_;
// Condition variables for the threads to signal each other.
std::condition_variable buffer_has_data_;
std::condition_variable buffer_is_empty_;
std::thread producer_;
std::thread consumer_;
};
int main( int argc, char **argv )
{
(void)argc; (void) argv;
prosumer test;
// Let the prosumer work for a little while.
std::this_thread::sleep_for( std::chrono::seconds( 3 ) );
return EXIT_SUCCESS;
}
如果你看一下producer_func和consumer_func线程函数,你可以看到它们循环,直到退出变量被prosumer析构函数设置,但它们也会在锁定循环缓冲区后再次检查退出变量。如果设置了quit变量,它们会互相发送信号以防止死锁。
我的另一个想法是在析构函数的条件变量上调用notify_one(),这是一个更好的解决方案吗?
有更好的方法吗?
更新1:我忘了提一下,在这种情况下,当线程被请求退出时,消费者不需要消耗循环缓冲区中的任何剩余数据,如果生产者产生更多一点也没关系。只要它们都退出并且不死锁,就可以了。
在我看来,在调用join之前,在析构函数中的两个条件变量上调用notify_one(或者应该是notify_all,如果您要将缓冲区扩展到多个生产者/消费者)将是首选的解决方案,原因如下:
首先,这与条件变量的典型使用方式相匹配:通过设置quit_,您更改了生产者/消费者线程感兴趣并等待的状态,因此您应该通知它们状态更改。
此外,notify_one不应该是一个开销很大的操作。
同样,在更现实的应用中,在两个元素的生产之间可能存在延迟;在这种情况下,你可能不希望在析构函数中阻塞,直到消费者注意到它必须在下一个元素进入队列时立即取消;在示例代码中,据我所见,这种情况不会发生。
在我看来,有两个功能是可以分开的:
- 消息传递和调度
- 生产和消费
把它们分开是有意义的:'worker'线程只是处理'message ',可能意味着'quit'或'do_work'。
这样你就可以创建一个泛型的'worker'类来聚合实际的函数。produce
和consume
方法保持干净,而worker
类只关心保持工作进行。
- 使用std::multimap迭代器创建std::list
- C++中std::resize(n)和std::shrink_to_fit之间的区别
- 来自 std::list 的迭代器 .end() 按预期返回"0xcdcdcdcdcdcdcdcd"但 .begin()
- C++17复制构造函数,在std::unordereded_map上进行深度复制
- 如何导出包含具有"std::unique_ptr"值的"std::map"属性的
- 从持续时间构造std::chrono::system_clock::time_point
- std::具有相同基类的类的变体
- std::向量与传递值的动态数组
- 使用std::vector的OpenCL矩阵乘法
- std::map<struct,struct>::find 找不到匹配项,但是如果我循环通过 begin() 到 end(),我在那里看到匹配项
- std::condition_variable::wait()如何评估给定的谓词
- 如何获取std::result_of函数的返回类型
- std::原子加载和存储都需要吗
- 将对象移动到std::shared_ptr
- POCO::PostgreSQL:如何将std::vector支持添加到`Binder::bind`
- 使用一个考虑到std::map中键值的滚动或换行的键
- 如何从 std::atomic 中提取指针 T<T>?
- 为什么 std::unique 不调用 std::sort?
- 使用std::函数映射对象方法
- 我需要std::condition,但有两个以上的选择