mutual exclusion ってなんかかっこいい

Program

なんか必殺技ぽい。

まぁそれはそうとお勉強会で本当は SELinux の勉強をするつもりだったんだけど、電車の中でふと mutex 的なものを書いたことないなーと思い出したのでちょっと書いてみました。

元のプログラムはカウンタ。こんなやつ。

int cnt;

void* count_up(void* idp) {
    int id = (int)idp;
    int i;
    printf("thread %d start\n", id);
    for (i = 0; i < 10000000; i++) {
        cnt++;
    }
    printf("thread %d end\n", id);
    return NULL;
}

int main() {
    int i;
    pthread_t th[NUM_THREADS];
    for (i = 0; i < NUM_THREADS; i++) {
        pthread_create(&th[i], NULL, count_up, (void*)i);
    }

    void* dummy;
    for (i = 0; i < NUM_THREADS; i++) {
        pthread_join(th[i], &dummy);
    }

    printf("%d\n", cnt);
}

これをちゃんとした結果が出るようにしようという話。ちなみに -O とかつけるとそれだけで正しい答えが出るようになってしまったりします。いやー GCC というのはえらいですね。というわけで -O は無しで。

以下のコードをまとめたものは github に置いてあります。

http://github.com/shinh/test/blob/e9f989532c3e07482ce7ddd5b7901ddaf1dd7f3e/mutex.c

pthread_mutex_t

        pthread_mutex_lock(&mu);
        cnt++;
        pthread_mutex_unlock(&mu);

普通。

pthread_spinlock_t

        pthread_spin_lock(&mu);
        cnt++;
        pthread_spin_unlock(&mu);

普通。 pthread_mutex_t は経験上ゼロ初期化でも使える (PTHREAD_MUTEX_INITIALIZER を使った方がいいのは火を見るより明らか) けど、 pthread_spinlock_t はゼロではダメみたいだった。 spinlock 用の PTHREAD_SPINLOCK_INITIALIZER とかは見つからんかったので、 pthread_spinlock_init しろということのようだ。

Compare and Swap

        while (__sync_val_compare_and_swap(&mu, 0, 1)) {
            sched_yield();
        }
        cnt++;
        mu = 0;

CAS とか略するらしい。 x86 とかだと cmpxchg だけど、 GCC だとまぁ簡単にやれる。 __sync_val_compare_and_swap の semantics はよく忘れるんだけど、以下と同じことが atomic に行なえる、と思えば良いらしい。上記では、 2つのスレッドが critical section に入らないように、 mu が 1 になってたら他の thread に処理をあけわたして、 0 だったら 1 入れて critical section に入るというようなかんじ。

template<typename T>
inline T sync_val_compare_and_swap(T* ptr, T oldval, T newval) {
  T ret = *ptr;
  if (ret == oldval) {
    *ptr = newval;
  }
  return ret;
}

カウンタみたいな簡単なものだったら critical section を作ってやらなくても、カウントアップ自体を atomic にやってやれば良くて、 CAS でやるなら、

        while (1) {
            int v = cnt;
            if (__sync_val_compare_and_swap(&cnt, v, v+1) == v) break;
            sched_yield();
        }

とかでできる。

こっち系の処理をする時は ABA problem というのに気をつけた方が良いらしい。最初に load した時に取得した値は A だったんだけど、その後に他のスレッドが B に変更してさらに他のスレッドが A に戻した時に、変わってないから大丈夫ーと他の値を書いてしまうような問題。カウンタの場合は Pentium 1 ペタヘルツとかを使っていて、 load した後に int がオーバーフローして一周したりしたら起こるかもしれないけど、いずれにせよ値が一致してるなら代入すべき次の値も一緒なので特に問題は起きない。

lock prefix

カウンタの場合、もっと手っ取り早く lock prefix つけてやればいい。 GCC だと超簡単。

        __sync_add_and_fetch(&cnt, 1);

さて最近の GCC は LL/SC 使ってこれを実装するのもできるようになってるらしいんだけど、最近の GCC が使える PPC マシンが無かったので試せてない。玄箱を lenny とかにすればいいと思うんだけど、 kernel のバージョン上げないといけないっぽいんだよな…

Load-linked / Store-conditional

PPC とかは CAS じゃなくて LL/SC とか略される、こっちのタイプをサポートしているらしい。要は load した時にそのアドレスに linked flag を立てておいて、それが汚れてなければ store conditional は成功する、っていう感じのものらしい。

クリティカルセクション作る実装:

        int* mup;
        asm(" b .yielded;\n"
            ".yield:\n"
            " bl sched_yield;\n"
            ".yielded:\n");
        mup = &mu;
        asm(".lock_cnt:\n"
            " lwarx 7, 0, %0;\n"
            " cmpwi 7, 0;\n"
            " bne .yield;\n"
            " addi 7, 7, 1;\n"
            " stwcx. 7, 0, %0;\n"
            " bne- .lock_cnt;\n"
            :"=r"(mup)::"7");
        cnt++;
        mu = 0;

stwcx. の最後のピリオドが必要だと気付いてなくてハマった。 RISC はメモリアドレスをレジスタにロードするコードを GCC に突っ込んでもらいたい場合はどうすればいいんだろうなー。

カウンタを load linked で読む実装:

        int* cntp = &cnt;
        asm(".lock_cnt:\n"
            " lwarx 7, 0, %0;\n"
            " addi 7, 7, 1;\n"
            " stwcx. 7, 0, %0;\n"
            " bne- .lock_cnt;\n"
            :"=r"(cntp)::"7");

Load linked で load した後に、 context switch なんかが入ると特に値がいじられてなくても store conditional は失敗し、 weak LL/SC とかいうらしい。現実的な CPU はまず weak らしい。まぁそうだよな。 LL の後で別の LL が入ったりしても SC は失敗するので、 LL 2回して SC 2回すればお手軽に自爆デッドロックができちゃうから注意、って感じらしい。ふーん。やってみよう。

int main() {
    int i, j;
    int* ip = &i;
    int* jp = &j;
    asm(".lock_cnt:\n"
        " lwarx 7, 0, %0;\n"
        ".lock_cnt2:\n"
        " lwarx 8, 0, %1;\n"
        " stwcx. 8, 0, %1;\n"
        " bne- .lock_cnt2;\n"
        " stwcx. 7, 0, %0;\n"
        " bne- .lock_cnt;\n"
        :"=&r"(ip),"=&r"(jp)::"7","8");
}

C++0x の atomic_compare_exchange_(weak|strong)

LL/SC についてよくわかってなかった段階ではなんで weak と strong を用意せにゃならんのかよくわかってなかったんですが、 LL/SC わかっちゃうと当然ですね。

http://www.open-std.org/jtc1/sc22/wg21/docs/papers/2008/n2748.html

要は、

  ll r1, object;
  if (r1 != *expected) {
      *expected = r1;
      return false;
  }
  sc r1, memory;
  return <store succeeded>;

って感じで weak の方を実装してやるって話なんだろう。

さて CAS を LL/SC で実装するってのは LL/SC なマシンで最適なパフォーマンスを出す手段として適切なんだろうか。どうもそうじゃない気がするんだよな。例えば上の mutex みたいなやつだと、 SC に失敗した時の fail ならもう一度 LL してやれば成功する可能性がそれなりにあるから busy loop してやった方がいいけど、 expected が一致しない fail なら lock 取られちゃってるんだから他のスレッドに処理明け渡して unlock してもらった方がいい、みたいなことがあると思う。そいう処理したければ expected が一致してるかどうかをチェックしにゃならんと思うんだけど、それって明らかに無駄だよねえというような。

答え: x86 以外の CPU は絶滅するのでもうどうでもいい。

追記:

いくつか mu = 0 でロックを放してるものがありますが、それだと WAW の順序保証が無い環境だと cnt++ と入れ変わっちゃうんじゃーというようなことを Cryolite さんに指摘いただきました。全くだと思います。ただ再現環境とか無いのが大変な感じですね。

http://twitter.com/Cryolite/status/8852507725

なにかあれば下記メールアドレスへ。
shinichiro.hamaji _at_ gmail.com
shinichiro.h