技術

[xv6 #35] Chapter 3 – Locking – Code: Locks

テキストの45〜46ページ

本文

xv6は、spinlock構造体としてロックを表現する。
その構造体のなかで重要なフィールドはlockedであり、ロック可能なときにゼロとなり、すでにロックされているときにゼロ以外の値となる。
理論上、xv6は、以下のようなコードを実行することによってロックを獲得しなければならない。

void
acquire(struct spinlock *lk)
{
  for(;;) {
    if(!lk->locked) {
      lk->locked = 1;
      break;
    }
  }
}

残念ながら、この実装は、最近のマルチプロセッサ上では排他を保証しない。
2つ(もしくはそれ以上の)CPUが同時にif文に到達し、lockedがゼロなのを見て、そして両方ともif文の中のブロックを実行することによってロックをつかむ可能性がある。
この時点で、2つの別のCPUがロックを保持し、排他的特性を破ってしまう。
競合状態を避けるために役に立つどころか、このacquireの実装はそれ自身が競合状態を持つ。
この問題は、lockedのチェックとlockedへの代入(if(!lk->locked) と lk->locked = 1;の行)が別の文で行われているところにある。
上記のルーチンを正すため、lockedのチェックとlockedへの代入は、ひとつのアトミックな処理(これ以上分割できない)で実行しなければならない。

spinlock.h

// Mutual exclusion lock.
struct spinlock {
  uint locked;       // Is the lock held?
  
  // For debugging:
  char *name;        // Name of lock.
  struct cpu *cpu;   // The cpu holding the lock.
  uint pcs[10];      // The call stack (an array of program counters)
                     // that locked the lock.
};

それら2つの行をアトミックに実行するために、xv6はxchgという386の特殊なハードウェア命令に頼る。
ひとつのアトミックな操作で、xchgは、メモリ上の1ワード(ここでは多分32ビットかな)とレジスタの内容を入れ替える。
実際のacquire関数は、ループ内でこのxchg命令を繰り返す。
ループ内で毎回lk->lockedを読み取りながら、アトミックにそれに1をセットする。(acquire関数のwhile文のところ)
すでにロックが確保されていたら、lk->lockedはすでに1になっているだろう。
そしてxchgは1を返し、ループは継続される。
xchgが0を返した場合、acquireはロックの獲得に成功した状態であり(0だったlockedが1になった状態)、それによってループは終わる。
一度ロックが確保されたら、acquireは、デバッグのため、ロックを確保したCPUとスタックトレースを記録する。
プロセスが、ロックを確保しておきながら解放するのを忘れたとき、この情報は、原因を特定するのに役立つ。
それらデバッグ情報用のフィールドは、ロックによって保護され、ロックを保持してるときだけ変更出来る。

x86.hのxchgインライン関数(xchg命令をラップしている)

static inline uint
xchg(volatile uint *addr, uint newval)
{
  uint result;
  
  // The + in "+m" denotes a read-modify-write operand.
  asm volatile("lock; xchgl %0, %1" :
               "+m" (*addr), "=a" (result) :
               "1" (newval) :
               "cc");
  return result;
}

spinlock.cのacquire関数

// Acquire the lock.
// Loops (spins) until the lock is acquired.
// Holding a lock for a long time may cause
// other CPUs to waste time spinning to acquire it.
void
acquire(struct spinlock *lk)
{
  pushcli(); // disable interrupts to avoid deadlock.
  if(holding(lk))
    panic("acquire");

  // The xchg is atomic.
  // It also serializes, so that reads after acquire are not
  // reordered before it. 
  while(xchg(&lk->locked, 1) != 0)
    ;

  // Record info about lock acquisition for debugging.
  lk->cpu = cpu;
  getcallerpcs(&lk, lk->pcs);
}

release関数は、acquireの対になるものであり、デバッグ情報用のフィールドをクリアし、そしてロックを解放する。

spinlock.cのrelease関数

// Release the lock.
void
release(struct spinlock *lk)
{
  if(!holding(lk))
    panic("release");

  lk->pcs[0] = 0;
  lk->cpu = 0;

  // The xchg serializes, so that reads before release are 
  // not reordered after it.  The 1996 PentiumPro manual (Volume 3,
  // 7.2) says reads can be carried out speculatively and in
  // any order, which implies we need to serialize here.
  // But the 2007 Intel 64 Architecture Memory Ordering White
  // Paper says that Intel 64 and IA-32 will not move a load
  // after a store. So lock->locked = 0 would work here.
  // The xchg being asm volatile ensures gcc emits it after
  // the above assignments (and after the critical section).
  xchg(&lk->locked, 0);

  popcli();
}

感想

詳細なロックの実装についてです。

xchg命令のところが分かりづらいですが、lk->lockedが1だった場合(すでに他でロックが獲得されている場合)、値1と交換した結果、交換直前のlk->lockedの値である1が返ってきて、while文の条件部が真となり、ループが続く(ロックが解放されるまで待つ)ことになります。
lk->lockedが0だった場合(ロックが獲得可能な場合)、値1と交換した結果、交換直前のlk->lockedの値である0が返ってきて、while文の条件部が偽となり、ループを抜け、処理を続行できることになります。

スピンロックについてはスピンロック – Wikipediaにもう少し詳しく書いてあります。

release関数内に長めのコメントがありますが、多分Wikipediaのスピンロックのページの最適化のセクションに書かれてるような最適化出来る条件を満たさないので、そのままxchgを使ってますよという事だと思います。

コメントを残す

メールアドレスが公開されることはありません。



※画像をクリックして別の画像を表示

このサイトはスパムを低減するために Akismet を使っています。コメントデータの処理方法の詳細はこちらをご覧ください