テキストの46〜47ページ

本文

xv6は、競合状態を避けるためにロックを利用して慎重にプログラムされている。
単純な例は、IDEドライバである。
この章のはじめで言及したように、iderw関数は、ディスクへのリクエストのキューを持ち、プロセッサは同時にそのリストへ新しいリクエストを追加するだろう。（iderw関数のDOC:insert−queueとコメントが付いてる部分）
このリストとその他のインバリアントを保護するために、iderwはidelockを獲得し（acquireし）、関数の終わりでそれを解放する。
キューの操作の後にacquireの呼び出しを移動する事によって、この章のはじめのほうで見たような競合状態を引き起こす方法を調べるという練習問題1がある。
その練習問題は試す価値がある。
なぜなら、競合を引き起こすことは簡単ではないということがハッキリするからである。
つまりそれは、競合状態を引き起こすバグを探すのは難しいということを意味する。
xv6が、そんないくつかの競合を持つ可能性はなくもない。

ide.c

// Simple PIO-based (non-DMA) IDE driver code.

#include "types.h"
#include "defs.h"
#include "param.h"
#include "memlayout.h"
#include "mmu.h"
#include "proc.h"
#include "x86.h"
#include "traps.h"
#include "spinlock.h"
#include "buf.h"

#define IDE_BSY       0x80
#define IDE_DRDY      0x40
#define IDE_DF        0x20
#define IDE_ERR       0x01

#define IDE_CMD_READ  0x20
#define IDE_CMD_WRITE 0x30

// idequeue points to the buf now being read/written to the disk.
// idequeue->qnext points to the next buf to be processed.
// You must hold idelock while manipulating queue.

static struct spinlock idelock;
static struct buf *idequeue;

static int havedisk1;
static void idestart(struct buf*);

// Wait for IDE disk to become ready.
static int
idewait(int checkerr)
{
  int r;

  while(((r = inb(0x1f7)) & (IDE_BSY|IDE_DRDY)) != IDE_DRDY) 
    ;
  if(checkerr && (r & (IDE_DF|IDE_ERR)) != 0)
    return -1;
  return 0;
}

void
ideinit(void)
{
  int i;

  initlock(&idelock, "ide");
  picenable(IRQ_IDE);
  ioapicenable(IRQ_IDE, ncpu - 1);
  idewait(0);
  
  // Check if disk 1 is present
  outb(0x1f6, 0xe0 | (1<<4));
  for(i=0; i<1000; i++){
    if(inb(0x1f7) != 0){
      havedisk1 = 1;
      break;
    }
  }
  
  // Switch back to disk 0.
  outb(0x1f6, 0xe0 | (0<<4));
}

// Start the request for b.  Caller must hold idelock.
static void
idestart(struct buf *b)
{
  if(b == 0)
    panic("idestart");

  idewait(0);
  outb(0x3f6, 0);  // generate interrupt
  outb(0x1f2, 1);  // number of sectors
  outb(0x1f3, b->sector & 0xff);
  outb(0x1f4, (b->sector >> 8) & 0xff);
  outb(0x1f5, (b->sector >> 16) & 0xff);
  outb(0x1f6, 0xe0 | ((b->dev&1)<<4) | ((b->sector>>24)&0x0f));
  if(b->flags & B_DIRTY){
    outb(0x1f7, IDE_CMD_WRITE);
    outsl(0x1f0, b->data, 512/4);
  } else {
    outb(0x1f7, IDE_CMD_READ);
  }
}

// Interrupt handler.
void
ideintr(void)
{
  struct buf *b;

  // Take first buffer off queue.
  acquire(&idelock);
  if((b = idequeue) == 0){
    release(&idelock);
    // cprintf("spurious IDE interrupt\n");
    return;
  }
  idequeue = b->qnext;

  // Read data if needed.
  if(!(b->flags & B_DIRTY) && idewait(1) >= 0)
    insl(0x1f0, b->data, 512/4);
  
  // Wake process waiting for this buf.
  b->flags |= B_VALID;
  b->flags &= ~B_DIRTY;
  wakeup(b);
  
  // Start disk on next buf in queue.
  if(idequeue != 0)
    idestart(idequeue);

  release(&idelock);
}

//PAGEBREAK!
// Sync buf with disk. 
// If B_DIRTY is set, write buf to disk, clear B_DIRTY, set B_VALID.
// Else if B_VALID is not set, read buf from disk, set B_VALID.
void
iderw(struct buf *b)
{
  struct buf **pp;

  if(!(b->flags & B_BUSY))
    panic("iderw: buf not busy");
  if((b->flags & (B_VALID|B_DIRTY)) == B_VALID)
    panic("iderw: nothing to do");
  if(b->dev != 0 && !havedisk1)
    panic("iderw: ide disk 1 not present");

  acquire(&idelock);  // DOC:acquire-lock

  // Append b to idequeue.
  b->qnext = 0;
  for(pp=&idequeue; *pp; pp=&(*pp)->qnext)  // DOC:insert-queue
    ;
  *pp = b;
  
  // Start disk if necessary.
  if(idequeue == b)
    idestart(b);
  
  // Wait for request to finish.
  // Assuming will not sleep too long: ignore proc->killed.
  while((b->flags & (B_VALID|B_DIRTY)) != B_VALID){
    sleep(b, &idelock);
  }

  release(&idelock);
}

ロックを使うことの難しい部分は、どうやって多くのロックを使うか、そしてどのロックがどのデータとインバリアントを保護するのか決めることである。
そこにはいくつかの基本的な原則がある。
最初に、他のCPUが読み書きできるような変数はいつでも、同時にあるCPUから書き込まれる可能性がある。
ロックは、その2つの操作がオーバーラップしないように導くべきである。
次に、ロックはインバリアントを保護するということを忘れるな。
あるインバリアントが複数のデータ構造にまたがる場合、典型的には、インバリアントを保護することを確実にするために、一つのロックによって、その全てのデータ構造は保護されなければならない。

上記のルールは、ロックが必要なときの話であり、ロックが不必要な場合については何も語ってはいない。
そして、効率上重要な点であるが、ロックは並列性を損なうので、ロックは使い過ぎないようにしなければならない。
効率が重要でない場合、ユニプロセッサのコンピュータとして使えば、ロックについては気にする必要はなくなる。
カーネルのデータ構造を保護するためなら、カーネルに入るときに獲得され、カーネルを出るときに解放されるようなロックが一つあればいい。
多くのユニプロセッサ用のOSのは、ときどきジャイアントロック（giant kernel lock）と呼ばれるこのやり方を用いてマルチプロセッサ上で実行出来るよう移行してきた。
しかしこの方法は、本物の同時実行性を犠牲にする。
一度にひとつのCPUだけが、カーネルで実行出来るのみである。
カーネルが何か重い計算を行う場合、より細かい粒度のロックの大きなセットを扱うのが、より効率的である。
もしそうなら、カーネルは複数のCPU上で同時に実行出来る。

究極的には、ロック粒度の選択は、並列プログラミングにおける課題である。
xv6は、少なく粒度が粗いデータ構造特有のロックを使う。
例えばxv6は、プロセステーブルとそのインバリアントを保護するために一つのロックを使う。
（プロセステーブルについては第4章で説明する。）
より細かい粒度の方法では、プロセステーブルの個別のエントリ上で動いているスレッドが、並行して続行出来るようにするために、プロセステーブル内のひとつのエントリ毎にロックを持つだろう。
しかしながら、プロセステーブル全体に渡るインバリアントを持つ操作は、いくつかのロックを持ち出さなきゃならないので、複雑になる。
xv6の例が、ロックの使い方を伝える手助けになることを願う。

感想

おもむろに練習問題1が出てきました。
もし試すなら、決まったパターンをファイルに書き込むプログラムを作って、それを複数同時実行するのが手っ取り早いかなと思います。
もし競合状態が発生したら、ファイルの内容に欠落が起きるはずです。
ただ、確率的になかなり低そうなのでどのくらいやったら観測可能かは分かりません。
iderwではキューの最後まで辿って、すかさずそのキューの最後に新しいリクエストを追加してます（*pp = b;の行）が、あるCPUにおけるその行の処理が、他のCPUがキューの最後まで辿った直後かつ追加の直前に行われたら競合状態になるはずです。
多分panicしたりとかデッドロックしたりとかにはならないはずなので、表面上は何も問題ないように見えるかもしれませんね。

プログラマ界隈の笑い話として、「謎のprint文に、この行をコメントアウトしたら何故か他の箇所で問題が起きるので注意。なんていうコメントが付いてた」なんてのがありますが、ここまで分かるようになると、（その原因を放置してるという意味では異常ですが）コンピュータの仕組み的には特に不思議でもなんでもないですね。
そういう場合メモリ管理に問題がある可能性もありますが、ちょっとしたタイミングの違いによって潜在的な競合状態が表面化するかしないかの差が出てるだけってのがしっくり来ます。

並列性を上げるには、ロック粒度を細かくして数を増やすのが有効とも書いてありますね。
似たような例として、DBのテーブルロックと行ロックの並列性についての話がありますね。