MITのxv6を読もう - 生産者/消費者モデルでデッドロックを避ける方法 -

sleepとwakeupの例として、生産者/消費者モデルが紹介されている。

生産者 : キューにデータを書き込んでいく。一定数までキューにデータを書き込むとスリープする。
消費者 : キューからデータを読み込む。キューが空になるまで読み込むとスリープする。

このときに問題になるのがキューの管理である。キューに対して生産者プロセスと消費者プロセスが同時にアクセスをするため、矛盾の無いように処理をする必要がある。

201 void*
202 send(struct q *q, void *p)
203 {
204     while(q->ptr != 0)
205     ;
206     q->ptr = p;
207     wakeup(q); /* wake recv */
208 }
209
210 void*
211 recv(struct q *q)
212 {
213     void *p;
214     
215     while((p = q->ptr) == 0)
216     sleep(q);
217     q->ptr = 0;
218     return p;
219 }

安直に作った場合、

生産者 : キューが空で無い間は待つ。キューが空になると、データを挿入してwakeupにより自分はスリープ状態に入り、recvプロセスを起こす。
消費者 : キューが空の間は待つ。この間sleep状態になる。生産者によって起こされると、キューを消費する。

さて、このモデルでデッドロックが発生するケースというのは、

f:id:msyksphinz:20150620095443j:plain

消費者がキューの状態を確認して、キューが空なのでスリープ状態に入るが、実はその直後に生産者がキューへデータを書き込んでしまったという状態。消費者がスリープ状態に入る前に生産者がwakeupを実行しても、スリープ状態のプロセスはまだ存在していないためどのプロセスも起こすことができない。従って、消費者はスリープ状態に入り、生産者は何も出来ずにデッドロックが発生してしまう。

そもそもこの問題は、消費者がキューを確認してから、スリープに入るまでの間に生産者が動作できてしまったのが問題。では、その間にロックを挿入してみるとどうか。

300 struct q {
301     struct spinlock lock;
302     void *ptr;
303 };
304
305 void*
306 send(struct q *q, void *p)
307 {
308     acquire(&q->lock);
309     while(q->ptr != 0)
310         ;
311     q->ptr = p;
312     wakeup(q);
313     release(&q->lock);
314 }
315
316 void*
317 recv(struct q *q)
318 {
319     void *p;
320     
321     acquire(&q->lock);
322     while((p = q->ptr) == 0)
323     sleep(q);
324     q->ptr = 0;
325     release(&q->lock);
326     return p;
327 }

f:id:msyksphinz:20150620095815j:plain

そもそも安直にロックを挿入してしまうと、消費者はロックを獲得したままスリープ状態に入ってしまうため、生産者はもうどうしようもできなくなる。

では、スリープに細工を加えよう。スリープする直前に、任意のロックを解放できるようにする。ここで任意のロックは、引数で指定する。

400 struct q {
401     struct spinlock lock;
402     void *ptr;
403 };
404
405 void*
406 send(struct q *q, void *p)
407 {
408     acquire(&q->lock);
409     while(q->ptr != 0)
410         ;
411     q->ptr = p;
412     wakeup(q);
413     release(&q->lock);
414 }
415
416 void*
417 recv(struct q *q)
418 {
419     void *p;
420     
421     acquire(&q->lock);
422     while((p = q->ptr) == 0)
423     sleep(q, &q->lock);
424     q->ptr = 0;
425     release(&q->lock);
426     return p;
427 }

f:id:msyksphinz:20150620095943j:plain