Loading article…
コンピュータ分野において、生産者・消費者問題（バウンドバッファ問題とも呼ばれる）は、1965年以来エドガー・W・ダイクストラによって記述されてきた一連の問題である。
ダイクストラは、 Electrologica X1 および X8 コンピュータのコンサルタントとして働いていたときに、プロデューサー/コンシューマー問題の解決策を見つけました。「プロデューサー/コンシューマーの最初の使用は、部分的にはソフトウェア、部分的にはハードウェアでした。ストアと周辺機器間の情報転送を担当するコンポーネントは「チャネル」と呼ばれていました...同期は、現在プロデューサー/コンシューマー構成として知られている 2 つのカウントセマフォによって制御されていました。1 つのセマフォはキューの長さを示し、CPU によって (V で) インクリメントされ、チャネルによって (P で) デクリメントされ、もう 1 つのセマフォは未確認完了の数をカウントし、チャネルによってインクリメントされ、CPU によってデクリメントされました。[2 番目のセマフォが正の場合、対応する割り込みフラグが上がります。]」[ 1 ]
ダイクストラは、無制限バッファの場合について次のように書いています。「我々は、それぞれ「生産者」と「消費者」と呼ばれる2つのプロセスを考えます。生産者は循環プロセスであり、サイクルを一周するたびに、消費者が処理しなければならない一定量の情報を生成します。消費者もまた循環プロセスであり、サイクルを一周するたびに、生産者によって生成された次の情報部分を処理することができます...この目的のために、2つのプロセスは無制限の容量を持つバッファを介して接続されていると仮定します。」[ 2 ]
彼は、バッファが有限である場合について次のように書いています。「我々は、容量が無限のバッファを介して結合された生産者と消費者を研究しました...両者が有限サイズ、例えばN個の部分からなるバッファを介して結合されている場合、関係は対称になります」[ 3 ]
そして、複数の生産者・消費者のケースについて：「我々は、n i個の部分を含む情報ストリームを介して結合される、複数の生産者/消費者のペアを検討する。我々は、すべてのストリームのすべての部分を含むべき有限バッファの容量が「tot」個の部分であると仮定する。」[ 4 ]
ペル・ブリンチ・ハンセンとニクラウス・ヴィルトはすぐにセマフォの問題に気づきました。「セマフォに関しては、高水準言語には適さないという同じ結論に達しました。代わりに、自然な同期イベントはメッセージの交換です。」[ 5 ]
元のセマフォ境界バッファソリューションはALGOLスタイルで記述されました。バッファはN個の部分または要素を格納できます。「キューイング部分数」セマフォはバッファ内の満たされた位置をカウントし、「空き位置数」セマフォはバッファ内の空き位置をカウントし、セマフォ「バッファ操作」はバッファのputおよびget操作のミューテックスとして機能します。バッファが満杯の場合、つまり空き位置数がゼロの場合、プロデューサースレッドはP(空き位置数)操作で待機します。バッファが空の場合、つまりキューイング部分数がゼロの場合、コンシューマースレッドはP(キューイング部分数)操作で待機します。V()操作はセマフォを解放します。副作用として、スレッドは待機キューから準備完了キューに移動できます。P()操作はセマフォ値をゼロまで減らします。V()操作はセマフォ値を増やします。[ 6 ]
begin integer number of queueing portions , number of empty positions , buffer manipulation ; number of queueing portions := 0 ; number of empty positions := N ; buffer manipulation := 1 ; parbegin producer : begin again 1 : produce next portion ; P ( number of empty positions ) ; P ( buffer manipulation ) ; add portion to buffer ; V ( buffer manipulation ) ; V ( number of queueing portions ) ; goto again 1 end ; consumer : begin again 2 : P ( number of queueing portions ) ; P ( buffer manipulation ) ; take portion from buffer ; V ( buffer manipulation ) ; V ( number of empty positions ) ; process portion to process ; goto again 2 end parend endC++20以降、セマフォは言語の一部となっています。ダイクストラの解法は、現代のC++で簡単に記述できます。変数buffer_manipulationはミューテックスです。あるスレッドで取得し、別のスレッドで解放するというセマフォ機能は不要です。lock()とunlock()のペアの代わりにlock_guard()文を使用するのは、C++ RAIIに準拠しています。lock_guardデストラクタは、例外が発生した場合にロックが解放されることを保証します。この解法は、複数のコンシューマスレッドや複数のプロデューサースレッドを処理できます。
#include <thread> #include <mutex> #include <semaphore>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 portion = produce_next_portion (); number_of_empty_positions.acquire ( ); { std :: lock_guard < std :: mutex > g ( buffer_manipulation ) ; add_portion_to_buffer ( portion ); } number_of_queueing_portions.release () ; } }void consumer ( ) { for ( ;; ) { number_of_queueing_portions.acquire ( ); Portion portion ; { std :: lock_guard < std :: mutex > g ( buffer_manipulation ); portion = take_portion_from_buffer (); } number_of_empty_positions.release (); process_portion_taken ( portion ) ; } }int main ( ) { std :: thread t1 ( producer ); std :: thread t2 ( consumer ); t1.join ( ) ; t2.join ( ); }Per Brinch Hansen はモニタを次のように定義しました。「モニタという用語は、共有変数とそれに対する意味のある操作のセットを表すために使用します。モニタの目的は、特定のポリシーに従って個々のプロセス間でリソースのスケジューリングを制御することです。」[ 7 ] Tony Hoare はモニタの理論的基礎を築きました。[ 8 ]
境界バッファ:モニター開始バッファ:配列0 .. N - 1 of portion ; head 、tail : 0 .. N - 1 ; count : 0 .. N ; nonfull 、nonfull :条件; procedure append ( x : portion ) ; begin if count = N then nonfull.wait ; note 0 <= count < N ; buffer [ tail ] : = x ; tail : = tail ( + ) 1 ; count : = count + 1 ; nonempty.signal end append ; procedure remove ( result x : portion ) ; begin if count = 0 then nonempty.wait ; note 0 < count <= N ; x : = buffer [ head ] ; head : = head ( + ) 1 ; count : = count - 1 ; nonfull.signal end remove ; head : = 0 ; tail : = 0 ; count := 0 ;境界バッファの終了;モニタは、循環バッファを実現するための変数、、、、同期のための条件変数、および境界バッファにアクセスするためのメソッド、およびを含むオブジェクトです。モニタ操作waitbufferはセマフォ操作Pまたはacquireに対応し、signalはVまたはreleaseに対応します。丸で囲まれた操作(+)はNを法として計算されます。提示されたPascalスタイルの擬似コードはHoareモニタを示しています。Mesaモニタはの代わりにを使用します。プログラミング言語C++バージョンは次のとおりです。headtailcountnonemptynonfullappendremovewhile countif count
template < size_t N > class Bounded_buffer { Portion buffer [ N ]; // 0..N-1 size_t head = 0 , tail = 0 ; // 0..N-1 size_t size = 0 ; // 0..N std :: condition_variable non_empty , non_full ; std :: mutex mtx ;public : void append ( Portion portion ) { std :: unique_lock lck ( mtx ); non_full.wait ( lck , [ & ] { return size != N ; } ); assert ( size < N ); buffer [ tail ++ ] = std :: move ( portion ); tail %= N ; ++ size ; non_empty.notify_one ( ) ; }Portion remove () { std :: unique_lock lck ( mtx ); non_empty.wait ( lck , [ & ]{ return size != 0 ; } ) ; assert ( size <= N ) ; Portion portion = std :: move ( buffer [ head ++ ]); head % = N ; --size ; non_full.notify_one ( ); return portion ; } };C++版では、技術的な理由から追加のミューテックスが必要です。バッファの追加および削除操作の前提条件を強制するために、assert関数を使用しています。
Electrologica コンピュータの最初のプロデューサー/コンシューマー ソリューションでは「チャネル」が使用されました。 Hoare はチャネルを次のように定義しました。 送信元と宛先を明示的に命名する代わりに、通信が行われるポートに名前を付ける方法があります。 ポート名はプロセスにローカルであり、チャネルによってポートのペアを接続する方法は、並列コマンドの先頭で宣言できます。[ 9 ] Brinch Hansen は、プログラミング言語JoyceとSuper Pascalにチャネルを実装しました。 Plan 9 オペレーティングシステムのプログラミング言語Alef、Inferno オペレーティングシステムのプログラミング言語Limboにはチャネルがあります。 次の C ソース コードは、ユーザー スペースから Plan 9でコンパイルされます。
#include "uh" #include "libc.h" #include "thread.h"enum { STACK = 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 ; Channel * ch = chancreate ( sizeof ( ulong ), 1 ); mk ( producer , ch , STACK ); mk ( consumer , ch , STACK ); recvp ( chancreate ( sizeof ( void * ), 0 )); threadexitsall ( 0 ); }プログラムのエントリポイントは関数ですthreadmain。関数呼び出しによってch = chancreate(sizeof(ulong), 1)チャネルが作成され、関数呼び出しによってsendul(ch, i)チャネルに値が送信され、関数呼び出しによってp = recvul(ch)チャネルから値が受信されます。プログラミング言語Goにもチャネルがあります。Goの例を以下に示します。
パッケージメインimport ( "fmt" "math/rand" "time" )var sendMsg = 0func produceMessage () int { time.Sleep ( 400 * time.Millisecond ) sendMsg ++ fmt.Printf ( " sendMsg = %v\n" , sendMsg ) return sendMsg } func consumeMessage ( 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 { consumeMessage ( recvMsg ) } }Go のプロデューサー/コンシューマー ソリューションでは、コンシューマーにはメインの Go ルーチンを使用し、プロデューサーには新しい無名の Go ルーチンを作成します。 2 つの Go ルーチンはチャネル ch で接続されています。このチャネルは最大 3 つの int 値をキューに入れることができます。ステートメントはチャネルを作成し、ステートメントはチャネルに値を送信し、ステートメントはチャネルから値を受け取ります。[ 10 ]メモリ リソースの割り当て、処理リソースの割り当て、およびリソースの同期は、プログラミング言語によって自動的に行われます。ch := make(chan int, 3)ch <- produceMessage()recvMsg := range ch
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++ は、マルチ スレッド プログラミング用の変数と操作を提供します。次の 1 つのプロデューサーと 1 つのコンシューマーに対するビジー ウェイト C++11 ソリューションは、アトミック変数に対するアトミックatomic読み取り変更書き込み操作を使用します。fetch_addfetch_subcount
enum { N = 4 }; Message buffer [ N ]; std :: atomic < unsigned > count { 0 }; void producer () { unsigned tail { 0 }; for (;;) { Message 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 ); consumeMessage ( message ); } } int main () { std :: thread t1 ( producer ); std :: thread t2 ( consumer ); t1.join ( ) ; t2.join ( ) ; }循環バッファのインデックス変数headと はtailスレッドローカルであるため、メモリの一貫性には関係ありません。変数 は、countプロデューサー スレッドとコンシューマー スレッドのビジー ウェイトを制御します。