**回転している間に：** *スピンロックでよく起こる問題をお聞きください*

イントロ

これまで1年以内に3回、スピンループで問題が発生したプロジェクトを経験しました。数年間にわたりスピニングスレッドと対峙してきましたが、正直言うと、私は攻撃側でも被害者側でもあったことがあります。
同じ問題を何度も目にするのは疲れますし、ブログを書いて人々に読んでもらい、他人が犯したミスを繰り返さないようにしたいと考えました。

実際、多くの方がこの件について書き、スピンロックに関わる様々な問題（速度、公平性、優先度反転、NUMA、破損コードなど）を取り上げています。
それでも「スピンロックは制御不能になりやすいのでOSプリミティブを使うべきだ」という事実に納得できないなら、続けて読んでください。自前のスピンロックを実装するときに絶対にやってはいけないことについて触れます。再度言いますが、今時はほとんどの場合スピンロックを使うべきではありません。もし使う場合は、本当に何をしているかを理解し、思わぬタイミングで問題に直面することを覚悟してください。

これは一般的なスピンループの話であり、多数存在するロッキングアルゴリズム（5）についてではありません。

壊れたスピンロック

まず基本から：自前のスピンロックを実装したいとします。

「簡単だ！ブール値、lock() と unlock() の関数だけで OK 」

そうです…

デモ用に

bool

int

class BrokenSpinLock {
    // 故意に int32_t を使っているだけで、気にしないでください。
    int32_t isLocked = 0;
public:
    void lock() {
        while (isLocked != 0) { /* まだロックされていればループ */ }
        isLocked = 1;   // Acquire
    }

    void unlock() {
        isLocked = 0;
    }
};

マルチスレッドを扱ったことがある人はすぐに問題点に気づくでしょう。複数のスレッドがこのロックを使おうとすると、

isLocked

以下の例では、2 つのスレッドが同時に

lock()

1. スレッド A が

   isLocked == 0

   を確認 | スレッド B が

   isLocked == 0

   を確認
2. 両方ともループから抜ける |
3. スレッド A が

   isLocked

   に 1 を書き込む | スレッド B が同じく 1 を書き込む

結果として、2 つのスレッドがロックを取得したと誤認します。

原子変数/操作に関する小話もあります。簡単に言えば：原子操作は他のスレッドが途中状態を見ることを防ぎ、レースコンディションを回避します（ただしそれ自体も危険で扱いが難しい）。

isLocked

std::atomic<int>

exchange()

void lock() {
    while (isLocked.exchange(1) != 0) {}
}

同時に実行しても原子性のおかげでどちらかが先に完了します。

CPUを焼き尽くすスピンロック

このスピンロックは「何もしない」ループです。
CPU は待機中であることを知る手段が無いため、頻度を高く保ち続けます。現代の CPU ではコア周波数を下げて電力と温度を抑える仕組みがありますが、スピンロックはそれを妨げます。特にモバイル／組込みデバイスでは望ましくありません。

説得力が足りない場合、以下のような性能低下も起きます。多くのスレッドがロック取得を競い合うと、一つだけが勝ち、残りはメモリ書き込みを行います。これは各コア間で同期される必要があります。

Intel の Optimization Reference Manual から引用：

現代マイクロプロセッサのスーパースカラー予測実行エンジンでは、このようなループが複数の同時読みリクエストを発生させます…

このペナルティは Core Solo/Duo で小さいですが、Xeon では約 25 倍重いです。さらに SMT（ハイパースレッディング）を有効にすると、スピン待ちループが実行帯域幅の大部分を消費します。

対策

最善の方法はスピンループを避け、CPU に「メモリ変更を通知されるまで待つ」ことです。x86 では

PAUSE

_mm_pause()

void cpu_pause() {
#if defined(__i386__) || defined(__x86_64__)
    _mm_pause();
#elif defined(__arm__) || defined(__aarch64__)
    __yield();
#else
    #error "unknown instruction set"
#endif
}

void lock() {
    while (isLocked.exchange(1) != 0)
        cpu_pause();
}

十分に待たないスピンロック

従来の手法ではバックオフ戦略を採用し、各試行で

PAUSE

void lock() {
    const int maxPauses = 64; // MAX_BACKOFF
    int nbPauses = 1;
    while (isLocked.exchange(1) != 0) {
        for (int i = 0; i < nbPauses; ++i)
            cpu_pause();
        nbPauses = nbPauses < maxBackoffs ? nbPauses * 2 : nbPauses;
    }
}

Intel は「

PAUSE

rdtsc()

struct Yielder {
    static const int maxPauses = 64; // MAX_BACKOFF
    int nbPauses = 1;
    void do_yield() {
        const int jitter = static_cast<int>(__rdtsc() & (nbPauses - 1));
        const int nbPausesThisLoop = nbPauses - jitter;
        for (int i = 0; i < nbPausesThisLoop; ++i)
            cpu_pause();
        nbPauses = nbPauses < maxBackoffs ? nbPauses * 2 : nbPauses;
    }
};

void lock() {
    Yielder yielder;
    while (isLocked.exchange(1) != 0)
        yielder.do_yield();
}

待ちすぎるスピンロック

maxBackoff

PAUSE

TSC 周期で

PAUSE

static inline bool before(uint64_t a, uint64_t b) {
    return ((int64_t)b - (int64_t)a) > 0;
}

void pollDelay(uint32_t clocks) {
    uint64_t endTime = _rdtsc() + clocks;
    for (; before(_rdtsc(), endTime); )
        cpu_pause();
}

Yielder

struct Yielder {
    static const int maxPauses = 64; // MAX_BACKOFF
    int nbPauses = 1;
    const int maxCycles = /* Some value */;

    void do_yield() {
        uint64_t beginTSC = __rdtsc();
        uint64_t endTSC   = beginTSC + maxCycles;
        const int jitter  = static_cast<int>(beginTSC & (nbPauses - 1));
        const int nbPausesThisLoop = nbPauses - jitter;
        for (int i = 0; i < nbPausesThisLoop && before(__rdtsc(), endTSC); ++i)
            cpu_pause();
        nbPauses = nbPauses < maxBackoffs ? nbPauses * 2 : nbPauses;
    }
};

void lock() {
    Yielder yield;
    while (isLocked.exchange(1) != 0)
        yield.do_yield();
}

優先度反転を扱わないスピンロック

優先度反転はスピンロックで最悪のケースです。低優先度スレッドがロックを取得し、高優先度スレッドが待機すると、OS スケジューラが低優先度スレッドをプリエンプトしてしまい、高優先度スレッドは CPU サイクルを無駄に消費します。

一般的な対策は指数バックオフと

std::this_thread::yield()

OS に通知するスピンロック

より堅牢な解決策は、Linux の

futex

WaitOnAddress

WakeByAddressSingle

FUTEX_WAKE

void do_yield(int32_t* addr, int32_t cmpVal, uint32_t timeoutMs) {
    // 指数バックオフ + jitter...
    if (nbPauses >= maxBackoffs) {
        WaitOnAddress(addr, &cmpVal, sizeof(cmpVal), timeoutMs);
        nbPauses = 1;
    }
}

void lock() {
    Yielder yield;
    while (isLocked.exchange(1, std::memory_order_acquire) != 0)
        do_yield(&isLocked, 1 /*while locked*/, 1 /*ms*/);
}

void unlock() {
    isLocked = 0;
    WakeByAddressSingle(&isLocked); // 可能なスレッドを通知
}

不公平なスピンロック

トークンロックなど一部のアルゴリズムは不公平です。競合時に高速なスレッドが常に優先され、他のスレッドが永遠に待つことがあります。不公平を避けたいなら OS プリミティブ（futex、セマフォ等）を利用する方が安全です。

偽共有を起こすスピンロック

異なる変数を書き込んでいても同じキャッシュラインに存在すると 偽共有 となり性能が低下します。ロックオブジェクトはアラインまたはパディングして別々のキャッシュラインに配置しましょう。

alignas(std::hardware_destructive_interference_size) MyLock lock1;
alignas(std::hardware_destructive_interference_size) MyLock lock2;

特殊命令

MWAIT

TPAUSE

WaitOnAddress

monitorx

mwaitx

結論

ほとんどの場合、自前のロックを作る必要はありません。最良のロックは「使わない」ものです。自前でスピンロックを実装する場合、以下のベストプラクティスに従ってください。

• 低い競合：クリティカルセクションが小さく、競合が稀なときのみ使用
• 正しいメモリオーダリング：取得/解放で acquire/release を適用
• OS に通知：

  futex

  /

  WaitOnAddress

  /

  std::atomic_wait

  があれば利用
• 偽共有を避ける：アライン／パディングで別キャッシュラインに配置

実際に失敗したプロジェクト例：

sched_yield

Sleep(0)

Sleep

WaitOnAddress

結論：自前のロックを実装するのは難しく、テスト済みのプラットフォームプリミティブを利用しましょう。