5

ミューテックスと同期用の2つの条件変数を使用して、スレッド化されたプロデューサー/コンシューマーシステムを実装するクラスがあります。プロデューサーは、使用するアイテムがあるときにコンシューマースレッドに通知し、コンシューマーは、アイテムを消費したときにプロデューサースレッドに通知します。スレッドは、デストラクタがブール変数を設定して終了するように要求するまで、生成と消費を続けます。どちらかのスレッドが条件変数を待機している可能性があるため、quit変数の2番目のチェックを実装する必要があります。

私は問題を次の例に減らしました（g++4.7でGNU/Linuxで動作しています）：

// 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 quit\n";
        }


        // 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 quit\n";
        }

        // 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スレッド関数を見ると、プロシューマーデストラクタによってquit変数が設定されるまでループしていることがわかりますが、循環バッファをロックした後、quit変数も再度チェックします。quit変数が設定されている場合、デッドロックを防ぐために相互に信号を送ります。

私が持っていた別のアイデアは、デストラクタからの条件変数でnotify_one（）を呼び出すことでしたが、それはより良い解決策でしょうか？

これを行うためのより良い方法はありますか？

更新1：この場合、スレッドの終了が要求されたときに、コンシューマーは循環バッファーに残っているデータを消費する必要がなく、プロデューサーがもう少し生成しても問題ないことを忘れました。それらが両方とも終了し、デッドロックしない限り、すべてがうまくいくでしょう。

4

2 に答える 2

3

私の意見では、joinを呼び出す前に、デストラクタの両方の条件変数でnotify_one（またはバッファを複数のプロデューサ/コンシューマに拡張する場合はnotify_all）を呼び出すことが、いくつかの理由で推奨されるソリューションです。

まず、これは条件変数が通常使用される方法と一致します。quit_を設定することにより、プロデューサー/コンシューマースレッドが関心を持って待機する状態を変更するため、状態の変更を通知する必要があります。

さらに、notify_oneは非常にコストのかかる操作であってはなりません。

また、より現実的なアプリケーションでは、2つの要素の生成の間に遅延がある場合があります。その場合、次の要素がキューに入れられるとすぐにキャンセルする必要があることにコンシューマーが気付くまで、デストラクタをブロックしたくない場合があります。サンプルコードでは、私が見る限り、それは発生しません。

于 2013-01-04T17:23:33.520 に答える
1

私の意見では、分離できる2つの機能があります。

  1. メッセージの受け渡しとディスパッチ
  2. 生産と消費

それらを実際に分離することは理にかなっています。「worker」スレッドは、「quit」または「do_work」を意味する「messages」を処理するだけです。

このようにして、実際の関数を集約する汎用の「worker」クラスを作成できます。produceとメソッドはconsumeクリーンなままで、workerクラスは作業を続けることだけに関心があります。

于 2013-01-04T17:38:06.470 に答える