生産者と消費者の問題

コンピューティングにおいて、生産者-消費者問題(制限バッファ問題とも呼ばれる) は、1965 年以来エドガー W. ダイクストラによって説明されている一連の問題です。

ダイクストラは、 Electrologica X1 および X8 コンピュータのコンサルタントとして働いていたときに、生産者と消費者の問題の解決策を見つけました。「生産者と消費者の最初の使用は、部分的にはソフトウェアで、部分的にはハードウェアでした。ストアと周辺機器間の情報転送を担当するコンポーネントは、「チャネル」と呼ばれていました...同期は、現在生産者/消費者配置として知られている 2 つのカウント セマフォによって制御されました。キューの長さを示す 1 つのセマフォは、CPU によって増加 (V で) され、チャネルによって減少 (P で) され、未確認の完了数をカウントするもう 1 つのセマフォは、チャネルによって増加され、CPU によって減少されました。[2 番目のセマフォが正の場合、対応する割り込み フラグが立てられます。]」^[1]

ダイクストラは、無制限のバッファのケースについて次のように書いています。「ここでは、それぞれ「プロデューサー」と「コンシューマー」と呼ばれる 2 つのプロセスについて考えます。プロデューサーは循環プロセスであり、サイクルを経るたびに、コンシューマーが処理する必要がある一定量の情報を生成します。コンシューマーも循環プロセスであり、サイクルを経るたびに、プロデューサーによって生成された次の部分の情報を処理できます。この目的のために、2 つのプロセスは無制限の容量を持つバッファを介して接続されていると想定します。」^[2]

彼は、制限付きバッファのケースについて次のように書いている。「我々は、無制限の容量を持つバッファを介して結合された生産者と消費者を研究した。両者が有限サイズのバッファ、例えばN個の部分を介して結合されている場合、関係は対称的になる。」^[3]

複数の生産者と消費者のケースについては、「我々は、生産者と消費者のペアをいくつか考えます。ここで、ペア_{i は、n i}個の部分を含む情報ストリームを介して結合されています。我々は、すべてのストリームのすべての部分を含むはずの有限のバッファが、'tot' 個の部分の容量を持っていると仮定します。」^[4]

ペル・ブリンチ・ハンセンとニクラウス・ヴィルトはすぐにセマフォの問題に気付きました。「セマフォに関しては、高級言語には適していないという同じ結論に達しました。その代わりに、自然な同期イベントはメッセージの交換です。」^[5]

オリジナルのセマフォ境界バッファ ソリューションはALGOLスタイルで記述されました。バッファには N 個の部分または要素を保存できます。「キュー部分の数」セマフォはバッファ内の満たされた位置をカウントし、「空の位置の数」セマフォはバッファ内の空の位置をカウントし、セマフォ「バッファ操作」はバッファの put および get 操作のミューテックスとして機能します。バッファがいっぱいの場合、つまり空の位置の数がゼロの場合、プロデューサ スレッドは P(空の位置の数) 操作で待機します。バッファが空の場合、つまりキュー部分の数がゼロの場合、コンシューマ スレッドは P(キュー部分の数) 操作で待機します。V() 操作はセマフォを解放します。副作用として、スレッドは待機キューから準備完了キューに移動できます。P() 操作はセマフォ値をゼロまで減らします。V() 操作はセマフォ値を増やします。^[6]

begin integerキューイング部分の数、空の位置の数、バッファ操作;キューイング部分の数:= 0 ;空の位置の数:= N ;バッファ操作:= 1 ; parbeginプロデューサー:再度開始1 :次の部分を生成; P (空の位置の数) ; P (バッファ操作) ;部分をバッファに追加; V (バッファ操作) ; V (キューイング部分の数) ;再度1 へ移動end ;コンシューマー:再度開始2 : P (キューイング部分の数) ; P (バッファ操作) ;バッファから部分を取得; V (バッファ操作) ; V (空の位置の数) ;取得した部分を処理;再度2へ移動end parend end         
       
          
          
        
      
       
                  
                          
                        
                          
                        
                              
       
                   
                        
                          
                         
                          
                             
      

C++ 20 以降、セマフォは言語の一部です。ダイクストラのソリューションは、現代の C++ で簡単に記述できます。変数 buffer_manipulation はミューテックスです。1 つのスレッドで取得し、別のスレッドで解放するセマフォ機能は必要ありません。lock() と unlock() のペアの代わりに lock_guard() ステートメントを使用するのは、C++ RAIIです。lock_guard デストラクタは、例外が発生した場合にロックを解放します。このソリューションは、複数のコンシューマ スレッドや複数のプロデューサ スレッドを処理できます。

#include <スレッド> #include <ミューテックス> #include <セマフォ> 
 
 

std :: counting_semaphore < N > number_of_queueing_portions { 0 }; std :: counting_semaphore < N > number_of_empty_positions { N }; std :: mutex buffer_manipulation ; 
 
 

void producer () { for (;;) { Portion part = produce_next_portion (); number_of_empty_positions . acquire (); { std :: lock_guard < std :: mutex > g ( buffer_manipulation ); add_portion_to_buffer ( part ); } number_of_queueing_portions . release (); } }  
    
       
    
    
       
      
    
    
  


void consumer () { for (;;) { number_of_queueing_portions . acquire (); Portion部分; { std :: lock_guard < std :: mutex > g ( buffer_manipulation ); part = take_portion_from_buffer (); } number_of_empty_positions . release (); process_portion_taken (部分); } }  
    
    
     
    
       
        
    
    
    
  


int main () { std :: thread t1 (プロデューサー); std :: thread t2 (コンシューマー); t1.join ( ) ; t2.join ( ) ; }  
   
   
  
  

^{[不適切な合成？ ]}

Per Brinch Hansen はモニターを定義しました。「モニター」という用語は、共有変数とそれに対する意味のある操作のセットを表すために使用します。モニターの目的は、特定のポリシーに従って個々のプロセス間のリソースのスケジュールを制御することです。^[7] Tony Hoare はモニターの理論的基礎を築きました。^[8]

境界付きバッファ:モニタ開始バッファ:部分の配列0 .. N - 1。先頭、末尾: 0 .. N - 1。カウント: 0 .. N。空でない、満杯でない:条件。手順append ( x :部分) ;開始count = Nの場合、満杯でない。待機。注意0 <= count < N。バッファ[末尾] := x 。末尾:= tail ( + ) 1。カウント: = count + 1。空でない。シグナル終了append 。手順remove (結果x :部分) ;開始count = 0の場合、空でない。待機。注意0 < count <= N。x : =バッファ[先頭] 。先頭:= head ( + ) 1。カウント: = count - 1。満杯でない。シグナル終了remove 。先頭: = 0 。tail := 0 ; count := 0 ;境界付きバッファの終了;  
      
      
     
      
    
          
           
        
          
          
      
     
      
          
           
        
          
          
      
     
          
  

モニターは、循環バッファを実現するための変数buffer、head、 、tail同期のための条件変数および 、および境界バッファにアクセスするためのメソッド および を含むオブジェクトです。モニター操作 wait はセマフォ操作 P または acquire に対応し、signal は V または release に対応します。丸で囲まれた操作 (+) は N を法として取られます。ここで示した Pascal スタイルの擬似コードはHoare モニターを示しています。Mesaモニターはの代わりにを使用します。プログラミング言語の C++ バージョンは次のとおりです。 countnonemptynonfullappendremovewhile countif count

class Bounded_buffer { Portion buffer [ N ]; // 0..N-1 unsigned head 、tail ; // 0..N-1 unsigned count ; // 0..N std :: condition_variable nonempty 、nonfull ; std :: mutex mtx ; public : void append ( Portion x ) { std :: unique_lock < std :: mutex > lck ( mtx ); nonfull . wait ( lck , [ & ]{ return ! ( N == count ); }); assert ( 0 <= count && count < N ); buffer [ tail ++ ] = x ; tail %= N ; ++ count ; nonempty . notification_one (); }部分remove () { std :: unique_lock < std :: mutex > lck ( mtx ); nonempty . wait ( lck , [ & ]{ return ! ( 0 == count ); }); assert ( 0 < count && count <= N );部分x = buffer [ head ++ ]; head %= N ; -- count ; nonfull . notification_one (); return x ; } Bounded_buffer () { head = 0 ; tail = 0 ; count = 0 ; } };  
       
      
          
    
   

     
     
          
          
      
      
    
    
  
    
     
          
          
       
       
    
    
     
  
   
            
  

C++ バージョンでは、技術的な理由により追加のミューテックスが必要です。バッファーの追加および削除操作の 前提条件を強制するために assert を使用します。

エレクトロロジカ コンピュータの最初の生産者と消費者のソリューションは「チャネル」を使用していました。Hoare はチャネルを定義しました: 送信元と送信先を明示的に命名する代わりに、通信が行われるポートに名前を付けることができます。ポート名はプロセスに対してローカルであり、ポートのペアをチャネルで接続する方法は、並列コマンドの先頭で宣言できます。^[9] Brinch Hansen は、プログラミング言語JoyceとSuper Pascalでチャネルを実装しました。Plan 9 オペレーティング システムのプログラミング言語Alefと Inferno オペレーティング システムのプログラミング言語Limboにはチャネルがあります。次の C ソース コードは、ユーザー空間から Plan 9でコンパイルされます:

#include "uh" #include "libc.h" #include "thread.h" 
 
 

列挙型{スタック= 8192 };     

void producer ( void * v ) { Channel * ch = v ; for ( uint i = 1 ; ; ++ i ) { sleep ( 400 ); print ( "p %d \n " , i ); sendul ( ch , i ); } } void consumer ( void * v ) { Channel * ch = v ; for (;;) { uint p = recvul ( ch ); print ( " \t\t c %d \n " , p ); sleep ( 200 + nrand ( 600 )); } } void threadmain ( int argc , char ** argv ) { int ( * mk )( void ( * fn )( void * ), void * arg , uint stack ); mk = threadcreate ;チャネル* ch = chancreate ( sizeof ( ulong ), 1 ); mk ( producer , ch , STACK ); mk ( consumer , ch , STACK ); recvp ( chancreate ( sizeof ( void * ), 0 )); threadexitsall ( 0 ); }   
     
         
    
     
     
  

   
     
    
       
     
      
  

     
        
    
      
    
    
   
  

プログラムのエントリ ポイントは function にありますthreadmain。関数呼び出しch = chancreate(sizeof(ulong), 1)によってチャネルが作成され、関数呼び出しによってsendul(ch, i)チャネルに値が送信され、関数呼び出しによってp = recvul(ch)チャネルから値が受信されます。プログラミング言語Goにもチャネルがあります。Go の例:

パッケージメイン 

インポート( "fmt" "math/rand" "time" ) 
	
	
	


var送信メッセージ= 0   

func produceMessage ( ) int { time.Sleep ( 400 * time.Millisecond ) sendMsg ++ fmt.Printf ( " sendMsg = %v\n" , sendMsg ) return sendMsg } func consumerMessage ( recvMsg int ) { fmt.Printf ( " \t\trecvMsg = %v \ n " , recvMsg ) time.Sleep ( time.Duration ( 200 + rand.Intn ( 600 ) ) * time.Millisecond ) } func main ( ) { ch : = make ( chan int , 3 ) go func ( ) { for { ch < - produceMessage ( ) } } ( ) for recvMsg : = range ch { consumerMessage ( recvMsg ) } }   
	  
	
	 
	 

   
	 
	  

  
	    
	  
		 
			  
		
	
	     
		
	

Goのプロデューサー・コンシューマーソリューションでは、コンシューマー用のメインのGoルーチンを使用し、プロデューサー用に新しい名前のないGoルーチンを作成します。2つのGoルーチンはチャネルchに接続されています。このチャネルは最大3つのint値をキューに入れることができます。ステートメントはch := make(chan int, 3)チャネルを作成し、ステートメントはch <- produceMessage()チャネルに値を送信し、ステートメントはrecvMsg := range chチャネルから値を受信します。^[10]メモリリソースの割り当て、処理リソースの割り当て、およびリソースの同期は、プログラミング言語によって自動的に行われます。

Leslie Lamport は、 1 人のプロデューサーと 1 人のコンシューマーのための境界付きバッファー プロデューサー-コンシューマー ソリューションを文書化しました。バッファーには最大で b 個のメッセージ (b >= 1) を保持できると仮定します。このソリューションでは、k を b より大きい定数とし、s と r を 0 から k-1 までの値を取る整数変数とします。最初は s=r で、バッファーは空であると仮定します。k を b の倍数に選択することで、バッファーを配列 B [0: b - 1] として実装できます。プロデューサーは新しいメッセージをそれぞれ B[s mod b] に格納し、コンシューマーは B[r mod b] から各メッセージを取得します。^[11] 無限 k に対して一般化されたアルゴリズムを以下に示します。

プロデューサー: 
L : if ( s - r ) mod k = b then goto L fi ;メッセージをバッファーに格納します。s := ( s + 1 ) mod k ; goto L ;コンシューマー: L : if ( s - r ) mod k = 0 then goto L fi ;バッファーからメッセージを取得します。r := ( r + 1 ) mod k ; goto L ;               
         
            
       

               
         
            
       

Lamport のソリューションでは、スケジューラで待機する代わりに、スレッドでビジー待機を使用します。このソリューションでは、都合の悪いときにスケジューラのスレッド切り替えが及ぼす影響は無視されます。最初のスレッドがメモリから変数の値を読み取った場合、スケジューラは変数の値を変更する 2 番目のスレッドに切り替え、その後、スケジューラは最初のスレッドに戻り、最初のスレッドは変数の現在の値ではなく古い値を使用します。アトミックな読み取り、変更、書き込みにより、この問題が解決されます。最新の C++ では、atomicマルチスレッド プログラミング用の変数と操作が提供されています。1 つのプロデューサーと 1 つのコンシューマーに対する次のビジー待機 C++11 ソリューションでは、アトミックな読み取り、変更、書き込み操作fetch_addとfetch_subアトミック変数を使用していますcount。

enum { N = 4 };メッセージバッファ[ N ]; std :: atomic < unsigned > count { 0 }; void producer () { unsigned tail { 0 }; for (;;) {メッセージmessage = produceMessage (); while ( N == count ) ; // ビジー待機中buffer [ tail ++ ] = message ; tail %= N ; count . fetch_add ( 1 , std :: memory_order_relaxed ); } } void consumer () { unsigned head { 0 }; for (;;) { while ( 0 == count ) ; // ビジー待機中Message message = buffer [ head ++ ]; head %= N ; count . fetch_sub ( 1 , std :: memory_order_relaxed ); consumerMessage ( message ); } } int main () { std :: thread t1 ( producer ); std ::スレッドt2 (コンシューマー); t1 . join (); t2 . join (); }    
 
  
  
    
    
       
        
       
      
      
     
  

  
    
    
        
       
       
      
     
    
  

  
   
   
  
  

循環バッファ インデックス変数はスレッド ローカルであるため、メモリの一貫性には関係ありません。変数は、headプロデューサースレッドとコンシューマー スレッドのビジー待機を制御します。 tailcount

• アトミック操作
• デザインパターン
• 先入れ先出し
• パイプライン
• チャネル
• Java での実装: Java メッセージ サービス

• Mark Grand Patterns in Java、第 1 巻、UML で図解された再利用可能なデザイン パターンのカタログ
• C/C++ ユーザーズ ジャーナル (Dr.Dobb's) 2004 年 1 月号の「C++ プロデューサー/コンシューマー同時実行テンプレート ライブラリ」は、Ted Yuan 著で、すぐに使用できる C++ テンプレート ライブラリです。小さなテンプレート ライブラリのソース コードとサンプルは、こちらで入手できます。
• Ioan Tinca、Java における生産者-消費者問題の進化