Loading article…
並行プログラミングにおいて、モニタはスレッドが共有オブジェクトの状態に同時にアクセスすることを防止し、状態の変化を待機できるようにする同期構造です。モニタは、スレッドが排他アクセスを一時的に放棄して何らかの条件が満たされるのを待機し、その後排他アクセスを回復してタスクを再開するためのメカニズムを提供します。モニタは、ミューテックス（ロック）と少なくとも1つの条件変数で構成されます。条件変数は、オブジェクトの状態が変更されると明示的に「シグナル」され、ミューテックスは一時的に条件変数を「待機」している別のスレッドに渡されます。
モニターのもう一つの定義は、複数のスレッドがメソッドや変数に安全にアクセスできるようにミューテックスを含むスレッドセーフなオブジェクト、クラス、またはモジュールです。モニターの決定的な特徴は、そのメソッドが相互排他的に実行されることです。つまり、どの時点においても、モニターのメソッドを実行できるスレッドは最大で1つだけです。1つ以上の条件変数を使用することで、スレッドが特定の条件を待機する機能も提供できます（つまり、最初の「モニター」の定義を使用します）。この記事の残りの部分では、この意味での「モニター」を「スレッドセーフなオブジェクト/クラス/モジュール」と呼びます。
モニタはPer Brinch Hansen [ 1 ]とCAR Hoare [ 2 ]によって発明され、最初にBrinch HansenのConcurrent Pascal言語で実装されました。[ 3 ]
スレッドがスレッドセーフなオブジェクトのメソッドを実行している間、そのオブジェクトはミューテックス（ロック）を保持することで占有されていると言われます。スレッドセーフなオブジェクトは、いかなる時点においても、オブジェクトを占有できるスレッドは最大で1つだけとなるように実装されています。ロックは最初はロック解除されていますが、各パブリックメソッドの開始時にロックされ、各パブリックメソッドからの戻り時にロック解除されます。
いずれかのメソッドを呼び出す際、スレッドは、他のスレッドがスレッドセーフなオブジェクトのメソッドをいずれも実行していない状態になるまで待機してから、そのメソッドの実行を開始する必要があります。この相互排他がない場合、2 つのスレッドがデータ競合や論理エラーを引き起こす可能性があることに注意してください。たとえば、アカウントから 1000 を引き出す 2 つのスレッドは両方とも true を返し、残高は 1000 しか減らない可能性があります。これは、次のようになります。まず、両方のスレッドが現在の残高を取得し、それが 1000 より大きいことを確認して、そこから 1000 を減算します。次に、両方のスレッドが残高を保存して戻ります。
多くのアプリケーションでは、相互排他だけでは不十分です。操作を試みているスレッドは、何らかの条件Pが真になるまで待機する必要がある場合があります。ビジー待機ループ
( P )でない間はスキップする
相互排他では、他のスレッドがモニターに入って条件を真にすることができないため、この方法は機能しません。モニターのロックを解除し、一定時間待機し、モニターをロックして条件Pをチェックするループなど、他の「解決策」も存在します。理論的には、この方法は機能し、デッドロックは発生しませんが、問題が発生します。適切な待機時間を決定するのは困難です。短すぎるとスレッドが CPU を占有し、長すぎると明らかに応答しなくなります。必要なのは、条件Pが真である（または真になる可能性がある）ときにスレッドに通知する方法です。
古典的な並行処理問題として、最大サイズが制限されたプロデューサー/コンシューマー問題があります。この問題では、タスクのキューまたはリングバッファがあり、1つ以上のスレッドがタスクをキューに追加する「プロデューサー」スレッド、1つ以上の他のスレッドがキューからタスクを取り出す「コンシューマー」スレッドとなります。キュー自体はスレッドセーフではないと想定され、空、満杯、または空と満杯の中間の状態になります。キューがタスクで満杯の場合、コンシューマースレッドがタスクを取り出すことで空きができるまで、プロデューサースレッドはブロックする必要があります。一方、キューが空の場合、プロデューサースレッドがタスクを追加することで利用可能になるまで、コンシューマースレッドはブロックする必要があります。
キューはスレッド間で共有される並行オブジェクトであるため、キューへのアクセスはアトミックに行う必要があります。なぜなら、キューへのアクセス中にキューが矛盾した状態になる可能性があり、その状態がスレッド間で公開されてはならないからです。したがって、キューにアクセスするコードはすべてクリティカルセクションを構成し、相互排他によって同期する必要があります。キューにアクセスするコードのクリティカルセクション内のコードとプロセッサ命令が、同じプロセッサ上のスレッド間の任意のコンテキストスイッチ、または複数のプロセッサで同時に実行されるスレッドによってインターリーブされる可能性がある場合、矛盾した状態が公開され、競合状態が発生するリスクがあります。
単純なアプローチとしては、ビジーウェイト方式で同期処理を行わないコードを設計する方法がありますが、これは競合状態を引き起こす可能性があります。
global RingBuffer queue ; // スレッドセーフではないタスクのリングバッファ。// 各プロデューサースレッドの動作を表すメソッド: public method producer () { while ( true ) { task myTask = ...; // プロデューサーは追加する新しいタスクを作成します。while ( queue . isFull ()) {} // キューが満杯でなくなるまで待機します。queue . enqueue ( myTask ); // タスクをキューに追加します。} }// 各コンシューマースレッドの動作を表すメソッド: public method consumer () { while ( true ) { while ( queue . isEmpty ()) {} // キューが空でなくなるまで待機します。myTask = queue . dequeue (); // キューからタスクを取り出します。doStuff ( myTask ); // タスクを使って何か処理を行います。} }このコードには、キューへのアクセスが中断され、他のスレッドによるキューへのアクセスと混在する可能性があるという深刻な問題があります。queue.enqueue メソッドとqueue.dequeueメソッドには、キューのサイズ、開始位置と終了位置、キュー要素の割り当てと割り当てなど、キューのメンバ変数を更新する命令が含まれている可能性があります。さらに、queue.isEmpty()メソッドとqueue.isFull()メソッドもこの共有状態を読み取ります。プロデューサー/コンシューマー スレッドが enqueue/dequeue の呼び出し中に混在することを許容すると、キューの状態が不整合になり、競合状態が発生する可能性があります。また、あるコンシューマーがビジー ウェイトを終了してから dequeue を呼び出すまでの間にキューを空にした場合、2 番目のコンシューマーは空のキューからデキューしようとしてエラーが発生します。同様に、あるプロデューサーがビジー ウェイトを終了してから enqueue を呼び出すまでの間にキューを満杯にした場合、2 番目のプロデューサーは満杯のキューに要素を追加しようとしてエラーが発生します。
前述したように、同期を実現するための単純なアプローチの1つは、「スピン待機」を使用することです。これは、ミューテックスを使用してコードの重要なセクションを保護し、ビジー待機を引き続き使用し、各ビジー待機チェックの間にロックを取得および解放します。
global RingBuffer queue ; // タスクのスレッドセーフでないリングバッファ。global Lock queueLock ; // タスクのリングバッファのミューテックス。// 各プロデューサー スレッドの動作を表すメソッド: public method producer () { while ( true ) { task myTask = ...; // プロデューサーは追加する新しいタスクを作成します。queueLock.acquire (); // 初期ビジーウェイトチェックのためにロックを取得します。while ( queue.isFull ( ) ) { // キューが満杯でなくなるまでビジーウェイトします。queueLock.release ( ); // キューロックを必要とする他のスレッドが実行できるように、一時的にロックを解放します。 // これにより、コンシューマーがタスクを取得できるようになります。queueLock.acquire ( ) ; // 次の「queue.isFull()」呼び出しのためにロックを再取得します。}queue.enqueue ( myTask ); // タスクをキューに追加します。queueLock.release ( ); // 次のタスクを追加するために再び必要になるまで、キューのロックを解除します。} }// 各コンシューマースレッドの動作を表すメソッド: public method consumer () { while ( true ) { queueLock . acquire (); // 初期ビジーウェイトチェックのためにロックを取得します。while ( queue . isEmpty ()) { // キューが空でなくなるまでビジーウェイトします。queueLock . release (); //プロデューサーがタスクを追加できるように、queueLock を必要とする他のスレッドが実行できるように、一時的にロックを解除します。 queueLock . acquire (); // 次の "queue.isEmpty()" 呼び出しのためにロックを再取得します。} myTask = queue . dequeue (); // キューからタスクを取り出します。queueLock . release (); // 次のタスクを取り出すために再び必要になるまで、キューロックを解除します。doStuff ( myTask ); // タスクを使って何かを実行します。} }この方法は、矛盾した状態が発生しないことを保証しますが、不要なビジーウェイトのためにCPUリソースを浪費します。キューが空で、プロデューサースレッドが長時間追加するものがない場合でも、コンシューマースレッドは常に不必要にビジーウェイト状態になります。同様に、コンシューマーが現在のタスクの処理で長時間ブロックされ、キューが満杯の場合でも、プロデューサーは常にビジーウェイト状態になります。これは無駄なメカニズムです。必要なのは、プロデューサースレッドがキューが満杯でなくなるまでブロックし、コンシューマースレッドがキューが空でなくなるまでブロックするようにする方法です。
（注：ミューテックス自体もスピンロックになる可能性があり、ロックを取得するためにビジーウェイトが発生しますが、CPUリソースの無駄遣いという問題を解決するために、queueLockはスピンロックではなく、ブロッキングロックキューを適切に使用するものと仮定します。）
解決策は条件変数を使用することです。概念的には、条件変数はミューテックスに関連付けられたスレッドのキューであり、スレッドはそこで何らかの条件が真になるのを待機できます。したがって、各条件変数cはアサーションP cに関連付けられます。スレッドが条件変数で待機している間、そのスレッドはモニタを占有しているとはみなされないため、他のスレッドがモニタに入り、モニタの状態を変更できます。ほとんどのタイプのモニタでは、これらの他のスレッドは条件変数cにシグナルを送信して、現在の状態でアサーションP cが真であることを示すことができます。
したがって、条件変数には主に3つの操作があります。
wait c, mここで、cは条件変数、 はモニタに関連付けられたミューテックス（ロック）mです。この操作は、アサーションP c が真になるまで待機してから処理を進める必要があるスレッドによって呼び出されます。スレッドが待機している間、モニタは占有されません。「wait」操作の機能と基本的な契約は、以下の手順を実行することです。 m、c「待機キュー」（別名「スリープキュー」）に移動して、m。c待機キューに入っている間、次に実行されるプログラムカウンタはステップ2、つまり「wait」関数/サブルーチンの途中にあります。したがって、スレッドはスリープ状態になり、その後「wait」操作の途中でウェイクアップします。cのスリープ キューにあり、ミューテックスを解放している可能性がありますが、スレッドがスリープ状態になる前にプリエンプティブ スレッド スイッチが発生し、別のスレッドがc最初のスレッドを のキューから移動する際にシグナル操作 (下記参照) を呼び出しますc。問題の最初のスレッドが に戻されるとすぐに、そのプログラム カウンタはステップ 1c になり、スリープ状態になり、再びウェイクアップできなくなります。これは、スリープ時に のスリープ キューにあるべきだったという不変条件に違反します。その他の競合状態は、ステップ 1a と 1b の順序と、コンテキスト スイッチがc発生する場所に依存します。signal c（別名 ）はnotify c、アサーションP c が真であることを示すためにスレッドによって呼び出されます。モニタのタイプと実装に応じて、これにより のcスリープ キューから 1 つ以上のスレッドが「準備完了キュー」または実行可能な別のキューに移動します。mに関連付けられたミューテックスを解放する前に「シグナル」操作を実行するのがベスト プラクティスとされていますcが、コードが並行処理用に適切に設計され、スレッドの実装によっては、シグナルの前にロックを解放することも許容される場合が多くあります。スレッドの実装によっては、この順序がスケジューリングの優先順位に影響を与える可能性があります。（一部の著者は、シグナルの前にロックを解放することを推奨しています。）スレッドの実装では、この順序に関する特別な制約を文書化する必要があります。broadcast c、 とも呼ばれる はnotifyAll c、c の待機キュー内のすべてのスレッドを起動する同様の操作です。これにより、待機キューが空になります。一般的に、複数の述語条件が同じ条件変数に関連付けられている場合、アプリケーションはシグナルではなくブロードキャストを必要とします。これは、間違った条件を待機しているスレッドが起動され、正しい条件を待機しているスレッドを起動せずにすぐにスリープ状態に戻ってしまう可能性があるためです。それ以外の場合、述語条件がそれに関連付けられている条件変数と 1 対 1 である場合は、ブロードキャストよりもシグナルの方が効率的になる可能性があります。設計規則として、複数の条件変数を同じミューテックスに関連付けることはできますが、その逆はできません。（これは一対多の対応です。）これは、述語P c がモニターを使用するすべてのスレッドで同じであり、条件を変更する可能性のある他のすべてのスレッド、または問題のスレッドが条件を変更している間にそれを読み取る可能性のある他のすべてのスレッドから相互排他で保護する必要があるためです。しかし、同じミューテックスの使用を必要とする同じ変数に対して異なる条件を待機したい異なるスレッドが存在する可能性があります。上記のプロデューサー・コンシューマーの例では、キューは一意のミューテックスオブジェクトによって保護される必要があります。「プロデューサー」スレッドは、ロックと条件変数mを使用してモニターで待機します。mmこれはキューが満杯でなくなるまでブロックします。「コンシューマー」スレッドは、同じミューテックスを使用するものの条件変数が異なる別のモニターで待機します。キューが空でなくなるまでブロックします。同じ条件変数に異なるミューテックスを使用することは（通常）意味がありませんが、この古典的な例は、同じミューテックスを使用する複数の条件変数を持つことがしばしば確かに意味がある理由を示しています。1 つ以上の条件変数（1 つ以上のモニター）で使用されるミューテックスは、条件変数を使用しないコード（待機/シグナル操作なしで取得/解放するだけのコード）と共有することもできます。ただし、これらのクリティカル セクションが、同時実行データに対する特定の条件を待つ必要がない場合に限ります。
モニターの正しい基本的な使い方は以下のとおりです。
acquire ( m ); // このモニターのロックを取得します。while ( ! p ) { // 待機している条件/述語/アサーションが真でない間... wait ( m , cv ); // このモニターのロックと条件変数で待機します。} // ... 重要なコードセクションはここにあります ... signal ( cv2 ); // または: broadcast(cv2); // cv2 は cv と同じか異なる可能性があります。release ( m ); // このモニターのロックを解放します。以下は、何が起こっているかをより分かりやすく説明するために、より詳細なコメントを追加した擬似コードです。
// ... (前のコード) // モニターに入ろうとしています。 //スレッド間で共有される同時実行データに関連付けられたアドバイザリ ミューテックス (ロック) を取得し、 // 同じ同時実行データを読み書きするクリティカル セクションで実行中に、 // 2 つのスレッドがプリエンプティブにインターリーブされたり、異なるコアで同時に実行されたりしないようにします。// 別のスレッドがこのミューテックスを保持している場合、このスレッドはスリープ状態 (ブロック) になり、m のスリープ キューに配置されます。 (ミューテックス "m"はスピン ロックであってはなりません。) acquire ( m ); // これでロックを保持し、初めて条件を確認できます。// 上記の「acquire」の後に初めて while ループの条件を実行するとき、 // 「待機している条件/述語/アサーションは既に真ですか？」と尋ねています。while ( ! p ()) // "p" は、条件をチェックしてブール値に評価される任意の式 (変数または // 関数呼び出し) です。 // これはクリティカル セクションなので、この "while" ループ条件を実行するときはロックを保持している必要があります。 // "while" 条件がチェックされるのが初めてでない場合、// 「このモニターを使用している別のスレッドが通知して私を起こし、コンテキスト スイッチで戻ってきたので、// 待機している条件 / 述語 / アサーションは、私が起こされてから、このループの最後の反復の "wait" 呼び出し内でロックを再取得するまでの間、 // 真のままだったのか、それとも、// その間に他のスレッドが条件を再び偽にして、このウェイクアップを偽のものにしてしまったのか」という質問をしています。{ // これがループの最初の反復である場合、答えは// 「いいえ」です。条件はまだ準備できていません。それ以外の場合は、答えは次のようになります。// 後者です。これは誤ったウェイクアップであり、他のスレッドが最初に発生し、 // 条件が再び偽になったため、// 再度待機する必要があります。wait ( m , cv ); // 一時的に、任意のコア上の他のスレッドがm または cv に対して操作を実行できないようにします。// release(m) // ロック "m" をアトミックに解放して、// // この並行データを使用する他のコードが操作できるようにします。 // // このスレッドを cv の待機キューに移動して、// // 条件が真になったときに通知されるようにし、 // // このスレッドをスリープします。 // //他のスレッドとコアがm と cv に対して操作を実行できるようにします。// // このコアでコンテキスト スイッチが発生します。// // 将来のある時点で、待機している条件が// 真になり、このモニタ (m、cv) を使用する別のスレッドが、// このスレッドをウェイクアップするシグナル、または// ウェイクアップするブロードキャストを実行します。 // これは、私たちが cv の待機キューから取り出されたことを意味します。 // // この間、他のスレッドによって条件が再び偽になる場合、 // 条件が 1 回以上切り替わる場合、// または、たまたま真のままになる場合があります。// // このスレッドは、あるコアで再び実行されます。// // acquire(m) // ロック「m」が再取得されます。// このループの反復処理を終了し、「while」ループの条件を再確認して、// 述語がまだ真であることを確認します。}// 待機している条件は真です！ // モニターに入る前か、または// 「wait」の最後の実行から、ロックを保持しています。// ここに重要なコードセクションが入ります。このセクションには、述語が真であるという前提条件があります。// このコードは、cv の条件を偽にしたり、他の条件変数の述語を真にしたりする可能性があります。//どの条件変数の述語（ミューテックス m を共有する）が真になったか、または真になる可能性があるか、および使用されているモニタのセマンティック タイプに応じて、シグナルまたはブロードキャストを呼び出します。for ( cv_x in cvs_to_signal ) { signal ( cv_x ); // または: broadcast(cv_x); } // 1 つ以上のスレッドが起動されましたが、m を取得しようとするとすぐにブロックされます。// 通知されたスレッドなどがクリティカルセクションに入れるように、ミューテックスを解放します。release ( m );条件変数の使用方法を紹介したので、それを使って古典的な境界付き生産者/消費者問題を改めて検討し、解決してみましょう。古典的な解決策は、キュー上の1つのロックを共有する2つの条件変数からなる2つのモニターを使用することです。
global volatile RingBuffer queue ; // タスクのスレッドセーフでないリングバッファ。global Lock queueLock ; // タスクのリングバッファのミューテックス。(スピンロックではありません。) global CV queueEmptyCV ; // キューが空でなくなるのを待っているコンシューマスレッドの条件変数。 // 関連付けられたロックは "queueLock" です。 global CV queueFullCV ; // キューが満杯でなくなるのを待っているプロデューサースレッドの条件変数。 // 関連付けられたロックも "queueLock" です。// 各プロデューサー スレッドの動作を表すメソッド: public method producer () { while ( true ) { // プロデューサーは追加する新しいタスクを作成します。task myTask = ...;// 初期述語チェックのために「queueLock」を取得します。queueLock.acquire ( ) ;// キューが満杯でないかどうかをチェックするクリティカルセクション。while ( queue . isFull ()) { // "queueLock" を解放し、このスレッドを "queueFullCV" にエンキューして、このスレッドをスリープ状態にする。wait ( queueLock , queueFullCV ); // このスレッドが起動したら、次の述語チェックのために "queueLock" を再取得する。}// タスクをキューに追加する重要なセクション（「queueLock」を保持していることに注意してください）。queue.enqueue ( myTask ) ;// キューが空でないことが保証されたので、待機しているコンシューマ スレッドを 1 つまたはすべて起動して、コンシューマ スレッドがタスクを引き受けます。signal ( queueEmptyCV ); // または: broadcast(queueEmptyCV); // クリティカル セクションの終了。// 次のタスクを追加するために再び必要になるまで、「queueLock」を解放します。queueLock.release ( ); } }// 各コンシューマ スレッドの動作を表すメソッド: public method consumer () { while ( true ) { // 初期述語チェックのために "queueLock" を取得します。queueLock . acquire ();// キューが空でないかどうかをチェックするクリティカルセクション。while ( queue . isEmpty ()) { // "queueLock" を解放し、このスレッドを "queueEmptyCV" にエンキューして、このスレッドをスリープさせる。wait ( queueLock , queueEmptyCV ); // このスレッドが起動したら、次の述語チェックのために "queueLock" を再取得する。}// キューからタスクを取り出す重要なセクション（「queueLock」を保持していることに注意してください）。myTask = queue.dequeue ( ) ;// キューが満杯でないことが保証されたので、キューが満杯でないのを待っているプロデューサー スレッドを 1 つまたはすべて起動し、プロデューサー スレッドがタスクを追加します。signal ( queueFullCV ); // または: broadcast(queueFullCV); // クリティカル セクションの終了。// 次のタスクを実行するために再び必要になるまで、「queueLock」を解放します。queueLock.release ( );// タスクに対して何らかの処理を実行します。doStuff ( myTask ); } }これにより、タスクキューを共有するプロデューサースレッドとコンシューマースレッド間の並行性が確保され、前述のスピンロックを使用したアプローチのようにビジーウェイトするのではなく、何もすることがないスレッドがブロックされます。
この解決策のバリエーションとして、プロデューサーとコンシューマーの両方に単一の条件変数を使用する方法があります。例えば、「queueFullOrEmptyCV」または「queueSizeChangedCV」という名前です。この場合、条件変数には複数の条件が関連付けられ、個々のスレッドがチェックする条件よりも弱い条件を表します。条件変数は、キューが満杯でないことを待っているスレッドと、キューが空でないことを待っているスレッドを表します。ただし、これを行うには、条件変数を使用するすべてのスレッドでブロードキャストを使用する必要があり、通常のシグナルは使用できません。これは、通常のシグナルによって、条件がまだ満たされていない間違ったタイプのスレッドが起動され、正しいタイプのスレッドにシグナルが送られることなく、そのスレッドがスリープ状態に戻ってしまう可能性があるためです。たとえば、プロデューサーがキューを満杯にして、コンシューマーではなく別のプロデューサーを起動し、起動されたプロデューサーがスリープ状態に戻る可能性があります。逆に、コンシューマーがキューを空にして、プロデューサーではなく別のコンシューマーを起動し、コンシューマーがスリープ状態に戻る可能性もあります。ブロードキャストを使用することで、適切なタイプのスレッドが問題文で想定されているとおりに処理を進めることが保証されます。
以下は、条件変数1つとブロードキャストのみを使用したバリアントです。
global volatile RingBuffer queue ; // スレッドセーフではないタスクのリングバッファ。global Lock queueLock ; // タスクのリングバッファ用のミューテックス。(スピンロックではありません。) global CV queueFullOrEmptyCV ; // キューがどのスレッドにも準備できていない場合の単一の条件変数 //つまり、キューが満杯でない状態になるのを待っているプロデューサー スレッドと、キューが空でない状態になるのを待っているコンシューマー スレッドの場合です。 // 関連付けられているロックは "queueLock" です。 // 通常の "signal" は複数の述語条件 (アサーション)に関連付けられているため、使用するのは安全ではありません。// 各プロデューサー スレッドの動作を表すメソッド: public method producer () { while ( true ) { // プロデューサーは追加する新しいタスクを作成します。task myTask = ...;// 初期述語チェックのために「queueLock」を取得します。queueLock.acquire ( ) ;// キューが満杯でないかどうかをチェックするクリティカルセクション。while ( queue . isFull ()) { // "queueLock" を解放し、このスレッドを "queueFullOrEmptyCV" にエンキューして、このスレッドをスリープ状態にする。wait ( queueLock , queueFullOrEmptyCV ); // このスレッドが起動したら、次の述語チェックのために "queueLock" を再取得する。}// タスクをキューに追加する重要なセクション（「queueLock」を保持していることに注意してください）。queue.enqueue ( myTask ) ;// キューがそれぞれ満杯でも空でもなくなるのを待っているプロデューサー スレッドとコンシューマー スレッドをすべて起動します。 // コンシューマー スレッドがタスクを引き受けます。 broadcast ( queueFullOrEmptyCV ); // 「signal」は使用しないでください (別のプロデューサー スレッドのみを起動する可能性があるため)。// クリティカル セクションの終了。// 次のタスクを追加するために再び必要になるまで、「queueLock」を解放します。queueLock.release ( ); } }// 各コンシューマ スレッドの動作を表すメソッド: public method consumer () { while ( true ) { // 初期述語チェックのために "queueLock" を取得します。queueLock . acquire ();// キューが空でないかどうかをチェックするクリティカルセクション。while ( queue . isEmpty ()) { // "queueLock" を解放し、このスレッドを "queueFullOrEmptyCV" にエンキューして、このスレッドをスリープ状態にする。wait ( queueLock , queueFullOrEmptyCV ); // このスレッドが起動したら、次の述語チェックのために "queueLock" を再取得する。}// キューからタスクを取り出す重要なセクション（「queueLock」を保持していることに注意してください）。myTask = queue.dequeue ( ) ;// キューがそれぞれ満杯でも空でもなくなるのを待っているプロデューサー スレッドとコンシューマー スレッドをすべて起動します。 // プロデューサー スレッドがタスクを追加します。 broadcast ( queueFullOrEmptyCV ); // 「signal」は使用しないでください (別のコンシューマー スレッドのみを起動する可能性があるため)。// クリティカル セクションの終了。// 次のタスクを実行するために再び必要になるまで、「queueLock」を解放します。queueLock.release ( );// タスクに対して何らかの処理を実行します。doStuff ( myTask ); } }モニタは、アトミックな読み出し・変更・書き込みプリミティブと待機プリミティブを使用して実装されます。読み出し・変更・書き込みプリミティブ（通常はテストアンドセットまたはコンペアアンドスワップ）は、 ISAによって提供されるメモリロック命令の形式をとることが多いですが、割り込みが無効になっているシングルプロセッサデバイスでは、ロックしない命令で構成することもできます。待機プリミティブは、ビジーウェイトループ、またはスレッドが実行準備が整うまでスケジューリングされないようにするOS提供のプリミティブのいずれかです。
以下は、テストアンドセット方式と先着順方式を用いた、スレッドシステムの一部、ミューテックス、およびMesaスタイルの条件変数の擬似コード実装例です。
// スレッドシステムの基本部分: // "ThreadQueue" はランダムアクセスをサポートするものとします。public volatile ThreadQueue readyQueue ; // 準備完了スレッドのスレッド安全でないキュー。要素は (Thread*) です。public volatile global Thread * currentThread ; // この変数はコアごとに割り当てられるものとします。(その他は共有されます。)// スレッドシステム自体の同期状態のみにスピンロックを実装します。// これは、同期プリミティブとしてテストアンドセットで使用されます。public volatile global bool threadingSystemBusy = false ;// コンテキストスイッチ割り込みサービスルーチン (ISR): // 現在の CPU コアで、別のスレッドにプリエンプティブに切り替えます。public method contextSwitchISR () { if ( testAndSet ( threadingSystemBusy )) { return ; // 現時点ではコンテキストを切り替えることができません。}// この割り込みが再び発生してコンテキストスイッチを妨害しないようにします: systemCall_disableInterrupts ();// 現在実行中のプロセスのすべてのレジスタを取得します。// プログラムカウンタ (PC) には、 // 下記の "resume" ラベルの命令位置が必要です。 レジスタ値の取得はプラットフォームに依存し、// 現在のスタック フレーム、JMP/CALL 命令などの読み取りが含まれる場合があります。 (詳細はこの範囲外です。) currentThread -> registers = getAllRegisters (); // メモリ内の "currentThread" オブジェクトにレジスタを格納します。currentThread -> registers . PC = resume ; // このメソッドで、次の PC を下記の "resume" ラベルに設定します。readyQueue.enqueue ( currentThread ); // このスレッドを後で実行できるように準備完了キューに戻します。Thread * otherThread = readyQueue.dequeue ( ); // 準備完了キューから削除し、次に実行するスレッドを取得します。currentThread = otherThread ; //グローバル current-thread ポインタ値を置き換えて、次のスレッドの準備ができるようにします。// currentThread/otherThread からレジスタを復元します。これには、他のスレッドの保存された PC へのジャンプも含まれます(以下の "resume" で)。繰り返しますが、この処理の詳細については、この範囲外です。restoreRegisters ( otherThread . registers );// *** 現在「otherThread」（現在は「currentThread」）を実行中です！元のスレッドは現在「スリープ状態」です。***resume : // ここでコンテキストを切り替えるときに、別の contextSwitch() 呼び出しで PC を設定する必要があります。// otherThreadが終了していた場所に戻ります。threadingSystemBusy = false ; // アトミックな代入である必要があります。systemCall_enableInterrupts (); // このコアでプリエンプティブスイッチングを再度有効にします。}// スレッドスリープ メソッド: // 現在の CPU コアで、現在のスレッドを準備完了キューに入れずに、同期的に別のスレッドにコンテキスト スイッチします。 // このメソッドが、contextSwitchISR() を呼び出すスレッド切り替えタイマーによって中断されないように、「threadingSystemBusy」を保持し、割り込みを無効にする必要があります。 // このメソッドから戻った後、「threadingSystemBusy」をクリアする必要があります。public method threadSleep () { // 現在実行中のプロセスのすべてのレジスタを取得します。// プログラム カウンタ (PC) には、以下の「resume」ラベルの命令位置が必要です。 // レジスタ値の取得はプラットフォームに依存し、現在のスタック フレーム、JMP/CALL 命令などの読み取りが含まれる場合があります。 (詳細はこの範囲外です。) currentThread -> registers = getAllRegisters (); // メモリ内の「currentThread」オブジェクトのレジスタを格納します。currentThread -> registers . PC = resume ; // このメソッドで、次の PC を以下の「resume」ラベルに設定します。// contextSwitchISR() とは異なり、currentThread を readyQueue に戻しません。// 代わりに、すでにミューテックスまたは条件変数のキューに配置されています。Thread * otherThread = readyQueue . dequeue (); // ready キューから削除して、次に実行するスレッドを取得します。currentThread = otherThread ; // グローバル current-thread ポインタ値を置き換えて、次のスレッドの準備ができるようにします。// currentThread/otherThread からレジスタを復元します。これには、他のスレッドの保存された PC へのジャンプも含まれます(以下の "resume" で)。繰り返しますが、この処理の詳細については、この範囲外です。restoreRegisters ( otherThread . registers );// *** 現在「otherThread」（現在は「currentThread」）を実行中です！元のスレッドは現在「スリープ状態」です。***resume : // ここでコンテキストを切り替えるときに、別の contextSwitch() 呼び出しで PC を設定する必要があります。// otherThreadが終了していた場所に戻る。}public method wait ( Mutex m , ConditionVariable c ) { // 他のスレッドが任意のコアでこのオブジェクトの// "held" と "threadQueue"、または "readyQueue" にアクセスしている間、内部スピンロックします。while ( testAndSet ( threadingSystemBusy )) {} // 注: "threadingSystemBusy" は現在 true です。// このコアでの割り込みを無効にするシステム呼び出し。これにより、threadSleep() は、contextSwitchISR() を呼び出すこのコアのスレッド切り替えタイマーによって中断されません。// 効率を上げるために threadSleep() の外で実行します。これにより、このスレッドは条件変数キューに入った直後にスリープされます。systemCall_disableInterrupts (); assert m . held ; // (具体的には、このスレッドがそれを保持している必要があります。) m . release (); c . waitingThreads . enqueue ( currentThread ); threadSleep (); // スレッドがスリープします... スレッドはシグナル/ブロードキャストから起動されます。threadingSystemBusy = false ; // アトミックな代入である必要があります。systemCall_enableInterrupts (); // このコアでプリエンプティブスイッチングを再度有効にします。// Mesa スタイル: // ここでコンテキストスイッチが発生する可能性があり、クライアント呼び出し元の述語が false になります。m.acquire (); }public method signal ( ConditionVariable c ) { // 他のスレッドが任意のコアでこのオブジェクトの// "held" と "threadQueue"、または "readyQueue" にアクセスしている間、内部スピンロックします。while ( testAndSet ( threadingSystemBusy )) {} // 注: "threadingSystemBusy" が true になります。 // このコアでの割り込みを無効にするシステム呼び出し。これにより、threadSleep()がこのコアのスレッド切り替えタイマーによって中断され、 contextSwitchISR() が呼び出されるのを防ぎます。 // 効率を上げるために threadSleep() の外で実行します。これにより、このスレッドは条件変数キューに入った直後にスリープされます。systemCall_disableInterrupts (); if ( ! c . waitingThreads . isEmpty ()) { wokenThread = c . waitingThreads . dequeue (); readyQueue . enqueue ( wokenThread ); } threadingSystemBusy = false ; // アトミックな代入である必要があります。systemCall_enableInterrupts (); // このコアでプリエンプティブスイッチングを再度有効にします。// Mesa スタイル: // 起動したスレッドには優先度が与えられません。}public method broadcast ( ConditionVariable c ) { // 他のスレッドが任意のコアでこのオブジェクトの// "held" と "threadQueue"、または "readyQueue" にアクセスしている間、内部スピンロックします。while ( testAndSet ( threadingSystemBusy )) {} // 注: "threadingSystemBusy" が true になります。 // このコアでの割り込みを無効にするシステム呼び出し。これにより、threadSleep()がこのコアのスレッド切り替えタイマーによって中断され、 contextSwitchISR() が呼び出されるのを防ぎます。 // 効率を上げるために threadSleep() の外で実行します。これにより、このスレッドは条件変数キューに入った直後にスリープされます。systemCall_disableInterrupts (); while ( ! c . waitingThreads . isEmpty ()) { wokenThread = c . waitingThreads . dequeue (); readyQueue . enqueue ( wokenThread ); } threadingSystemBusy = false ; // アトミックな代入である必要があります。systemCall_enableInterrupts (); // このコアでプリエンプティブスイッチングを再度有効にします。// Mesa スタイル: // 起動されたスレッドには優先度が与えられません。}class Mutex { protected volatile bool held = false ; private volatile ThreadQueue blockingThreads ; // ブロックされたスレッドのスレッド安全でないキュー。要素は (Thread*) です。public method acquire () { // 任意のコア上の他のスレッドがこのオブジェクトの// "held" と "threadQueue"、または "readyQueue" にアクセスしている間、内部スピンロックします。while ( testAndSet ( threadingSystemBusy )) {} // 注: "threadingSystemBusy" は現在 true です。 // このコアでの割り込みを無効にするシステム呼び出し。これにより、threadSleep()がこのコア上のスレッド切り替えタイマーによって中断され、contextSwitchISR() が呼び出されるのを防ぎます。 // 効率を高めるために threadSleep() の外で実行し、このスレッドがロックキューに入った直後にスリープされるようにします。systemCall_disableInterrupts ();assert ! blockingThreads.contains ( currentThread ) ;if ( held ) { // "currentThread" をこのロックのキューに追加して、// このロックで "sleeping" とみなされるようにします。// "currentThread" は、threadSleep() で処理する必要があることに注意してください。readyQueue . remove ( currentThread ); blockingThreads . enqueue ( currentThread ); threadSleep (); // これで、"held" が false になったため、スレッドが起動しました。assert ! held ; assert ! blockingThreads . contains ( currentThread ); } held = true ; threadingSystemBusy = false ; // アトミックな代入である必要があります。systemCall_enableInterrupts (); // このコアでプリエンプティブ切り替えを再び有効にします。} public method release () { // 他のコアのスレッドがこのオブジェクトの// "held" と "threadQueue"、または "readyQueue" にアクセスしている間、内部スピンロックします。while ( testAndSet ( threadingSystemBusy )) {} // 注: "threadingSystemBusy" は現在 true です。// 効率化のためにこのコアの割り込みを無効にするシステムコール。systemCall_disableInterrupts (); assert held ; // (ロックが保持されている間のみ解放を実行する必要があります。)held = false ; if ( ! blockingThreads.isEmpty ( )) { Thread * unblockedThread = blockingThreads.dequeue (); readyQueue.enqueue ( unblockedThread ) ; } threadingSystemBusy = false ; //アトミックな代入である必要があります。systemCall_enableInterrupts ( ); // このコアでプリエンプティブ切り替えを再度有効にします。} }struct ConditionVariable { volatile ThreadQueue waitingThreads ; }CAR HoareとPer Brinch Hansenによる当初の提案は、ブロッキング条件変数に関するものでした。ブロッキング条件変数を使用すると、シグナルを発信するスレッドは、シグナルを受け取ったスレッドが戻るか、または条件変数で再度待機することによってモニターの占有を放棄するまで、モニターの外で（少なくとも）待機する必要があります。ブロッキング条件変数を使用するモニターは、Hoareスタイルのモニターまたはシグナルアンド緊急待機モニターと呼ばれることがよくあります。
a2 つの条件変数とを持つ Hoare スタイルのモニター。Buhrらbによる。各モニターオブジェクトには、2つのスレッドキューが関連付けられていると仮定します。
e入場待ち列sこれは、シグナルを発信したスレッドのキューです。さらに、各条件変数cに対してキューが存在すると仮定します。
c.qこれは、条件変数cを待機しているスレッドのキューです。すべてのキューは通常、公平であることが保証されており、実装によっては先入れ先出しが保証されている場合もあります。
各操作の実装は以下のとおりです。（各操作は相互に排他的に実行されるものと想定します。したがって、再起動されたスレッドは、操作が完了するまで実行を開始しません。）
モニターに入る： メソッドを入力してください モニターがロックされている場合 このスレッドをeに追加する このスレッドをブロックする それ以外 モニターをロックする モニターを離れる: スケジュール メソッドから戻るcを待ってください: このスレッドをc .qに追加します スケジュール このスレッドをブロックする シグナルc : cを待機しているスレッドがある場合。c からそのようなスレッド t を 1 つ選択して削除します。 （これは「シグナルスレッド」と呼ばれます） このスレッドをsに追加する 再起動 （つまり、次にモニターに表示されるのはtです） このスレッドをブロックする スケジュール： s にスレッドがある場合 sから1つのスレッドを選択して削除し、再起動します。 （このスレッドが次にモニターに表示されます） そうでなければ、 e にスレッドがある場合 e から 1 つのスレッドを選択して削除し、再起動します。 （このスレッドが次にモニターに表示されます） それ以外 モニターのロックを解除 （モニターは使用されていない状態になります）
このscheduleルーチンは、モニターを占有する次のスレッドを選択するか、候補となるスレッドがない場合はモニターのロックを解除します。
その結果として生じるシグナリング方式は「シグナル・アンド・アージャント・ウェイト」と呼ばれ、シグナラーは待機する必要があるものの、エントランスキュー上のスレッドよりも優先される。代替案として「シグナル・アンド・ウェイト」があり、こちらはキューが存在せずs、シグナラーがeキュー上で待機する。
一部の実装では、シグナル送信とプロシージャからの復帰を組み合わせたシグナルおよび復帰操作が提供されています。
cをシグナルし、 cを待機しているスレッドがある場合は、 .qを返します。 c からそのようなスレッド t を 1 つ選択して削除します。 （これは「シグナルスレッド」と呼ばれます） 再起動 （つまり、次にモニターに表示されるのはtです） それ以外 スケジュール メソッドから戻る
どちらの場合（「シグナルと緊急待機」または「シグナルと待機」）でも、条件変数がシグナルされ、その条件変数で待機しているスレッドが少なくとも 1 つある場合、シグナルを発信するスレッドは、その間に他のスレッドが占有権を取得できないように、占有権をシグナルされたスレッドにシームレスに引き渡します。各シグナルc操作の開始時にP c が真であれば、各待機c操作の終了時にも P cは真になります。これは、次の契約で要約されます。これらの契約では、Iはモニタの不変量です。
モニターに入る: 事後条件I モニターを離れる: 前提条件Iwait c : 前提条件I はモニターの状態 を変更します。事後条件P cとI信号c : 事前条件P cとI は モニタの状態 を変更する事後条件I信号cと戻り値: 前提条件P cとI
これらの契約では、IとP cはキューの内容や長さに依存しないことが前提とされています。
（条件変数に対してキューで待機しているスレッド数を照会できる場合、より高度な契約を与えることができます。たとえば、不変条件を確立せずに占有率を渡せるようにする便利な契約のペアは次のとおりです。）
wait c : 前提条件I はモニターの状態 を変更します。事後条件P c信号c の事前条件(空でない( c )かつP c )または(空である( c )かつI ) はモニタの状態 を変更する事後条件I
（詳細はハワード[ 4 ]およびブールら[ 5 ]を参照。）
アサーションP cは完全にプログラマー次第です。プログラマーは、それが何であるかについて一貫性を保つ必要があります。
このセクションの最後に、境界付きスレッドセーフスタックを実装するブロッキングモニターを使用したスレッドセーフクラスの例を示します。
モニタークラスSharedStack { private const capacity := 10 private int [capacity] A private int size := 0 invariant 0 <= sizeおよびsize <= capacity private BlockingCondition theStackIsNotEmpty /* 0 < sizeおよびsize <= capacityに関連付けられています*/ private BlockingCondition theStackIsNotFull /* 0 <= sizeおよびsize < capacityに関連付けられています*/public method push( int value) { size = capacityの場合、 theStackIsNotFull を待機し、0 <= sizeかつsize < capacity であることを確認します。 A[size] := value ; size := size + 1 assert 0 < sizeかつsize <= capacity signal theStackIsNotEmpty and return } パブリックメソッドint pop() { size = 0の場合、 theStackIsNotEmpty を待機し、 0 < sizeかつsize <= capacityであることを確認します。 size := size - 1 ; assert 0 <= sizeかつsize < capacity の場合 、 theStackIsNotFullをシグナルし、 A[size]を返します。 } }この例では、スレッドセーフなスタックが内部的にミューテックスを提供しており、これは以前のプロデューサー/コンシューマーの例と同様に、同じ同時実行データに対して異なる条件をチェックする2つの条件変数で共有されています。唯一の違いは、プロデューサー/コンシューマーの例では通常の非スレッドセーフなキューを想定し、モニターの詳細を抽象化せずにスタンドアロンのミューテックスと条件変数を使用していた点です。この例では、「wait」操作が呼び出されると、何らかの方法でスレッドセーフなスタックのミューテックスが提供される必要があり、「wait」操作が「モニター クラス」に統合されている場合などが該当します。このような抽象化された機能とは別に、「raw」モニターを使用する場合は、常にミューテックスと条件変数を含める必要があり、各条件変数には一意のミューテックスが割り当てられます。
非ブロッキング条件変数（「Mesaスタイル」条件変数、または「シグナルアンドコンティニュー」条件変数とも呼ばれる）では、シグナルを送っても、シグナルを送ったスレッドがモニターの占有を失うことはありません。代わりに、シグナルを受け取ったスレッドはeキューに移動されます。キューは必要ありませんs。
aとb非ブロッキング条件変数の場合、シグナル操作は「notify」と呼ばれることが多く、ここではその用語に従います。また、条件変数で待機しているすべてのスレッドをキューに移動させる「notify all」操作を提供するのも一般的ですe。
ここでは、さまざまな操作の意味を説明します。（各操作は互いに排他的に実行されるものと想定します。したがって、再起動されたスレッドは、操作が完了するまで実行を開始しません。）
モニターに入る： メソッドを入力してください モニターがロックされている場合 このスレッドをeに追加する このスレッドをブロックする それ以外 モニターをロックする モニターを離れる: スケジュール メソッドから戻るcを待ってください: このスレッドをc .qに追加します スケジュール このスレッドをブロックする cに通知します:c .q でスレッドが待機している場合c から 1 つのスレッド t を選択して削除します。 （これは「通知スレッド」と呼ばれます） tをeに移動 全員に通知するc :c .q で待機しているすべてのスレッドをe に移動します スケジュール ： e にスレッドがある場合 e から 1 つのスレッドを選択して削除し、再起動します。 それ以外の場合は モニターのロックを解除します
この方式のバリエーションとして、通知されたスレッドは、 と呼ばれるキューに移動される可能性がありw、 は よりも優先されます。詳細については、eHoward [ 4 ]および Buhr ら[ 5 ]を参照してください。
各条件変数cにアサーションP cを関連付けることで、から戻ったときにP c が必ず真となるようにすることができます。ただし、通知スレッドが占有を放棄してから通知されたスレッドがモニターに再入するように選択されるまでの間、P cが維持されるようにする必要があります。この間には、他の占有者によるアクティビティが発生する可能性があります。そのため、 P c が単に真となることはよくあります。waitc
このため、通常は各待機操作を次のようなループで囲む必要があります。
( P )でない間は待機するここで、PはP cよりも強い条件です。操作および は、待機中のスレッドに対してP が真である可能性があるという「ヒント」として扱われます。このようなループの最初の反復以降、各反復は通知の損失を表します。したがって、ノンブロッキングモニターでは、通知が失われすぎないように注意する必要があります。notifycnotify allc
「ヒント」の例として、銀行口座を考えてみましょう。引き出しスレッドは、口座に十分な資金が貯まるまで待機してから処理を進めます。
モニタークラスAccount { private int balance := 0 invariant balance >= 0 private NonblockingCondition balanceMayBeBigEnough public メソッドwithdraw( int amount) 前提条件amount >= 0 { while balance < amount do wait balanceMayBeBigEnough assert balance >= amount 残高 := 残高 - 金額 } public メソッドdeposit( int amount) 前提条件amount >= 0 { 残高 := 残高 + 金額 すべてのバランスに通知します。十分大きいかもしれません。 } }この例では、待機条件は引き出し金額の関数であるため、入金スレッドがそのような条件を満たしたかどうかを知ることは不可能です。この場合、待機中の各スレッドを（一度に1つずつ）モニターに接続し、アサーションが真であるかどうかを確認するのが理にかなっています。
Java言語では、各オブジェクトをモニターとして使用できます。相互排他を必要とするメソッドは、synchronizedキーワードで明示的にマークする必要があります。コードブロックもsynchronizedでマークできます。[ 6 ]
明示的な条件変数を持つ代わりに、各モニター（つまりオブジェクト）には、入口キューに加えて単一の待機キューが備えられています。すべての待機はこの単一の待機キューで行われ、すべてのnotifyおよびnotifyAll操作はこのキューに適用されます。[ 7 ]このアプローチは、 C#などの他の言語でも採用されています。
シグナル伝達のもう1つの方法は、シグナル操作を省略することです。スレッドがモニターから離れるたび（戻るか待機するたび）、待機中のすべてのスレッドのアサーションが評価され、いずれかが真となるまで続きます。このようなシステムでは、条件変数は不要ですが、アサーションは明示的にコーディングする必要があります。待機の契約は次のとおりです。
wait P : 前提条件I はモニターの状態 を変更します。事後条件PとI
ブリンチ・ハンセンとホアは、1970年代初頭に、彼ら自身とエドガー・ダイクストラの以前のアイデアに基づいて、モニタの概念を開発しました。[ 8 ]ブリンチ・ハンセンは、 Simula 67のクラス概念を採用して最初のモニタ表記を発表し、[ 1 ]キューイング機構を発明しました。[ 9 ]ホアは、プロセスの再開のルールを改良しました。[ 2 ]ブリンチ・ハンセンは、 Concurrent Pascalでモニタの最初の実装を作成しました。[ 8 ]ホアは、それらがセマフォと同等であることを証明しました。
モニター（および Concurrent Pascal）はすぐにSolo オペレーティングシステムでプロセス同期の構造化に使用されました。[ 10 ] [ 11 ]
モニターをサポートしているプログラミング言語には、以下のようなものがあります。
ネイティブでモニターをサポートしていない言語でもモニターを構築できるようにするライブラリが数多く開発されています。ライブラリ呼び出しを使用する場合、プログラマは実行されるコードの開始と終了を相互排他で明示的にマークする必要があります。Pthreadsはそのようなライブラリの1つです。