(C#) lock フリーで高速なスレッドセーフ操作

C# で排他制御をしたい場合いくつか方法があります。何も考えずに雑にlock構文を書くだけで同期は取れるのですが、より高度にパフォーマンスチューニングしたい場合のため、効率的な方法を紹介します。

例えば以下のようなものがあるとします。

// スレッドセーフではないもの
public static class Foo
{
    private static int[] _array = new int[10];
    private static int _updateCount;

    public static void SetValue(int index, int value)
    {
        _array[index] = value;
        _updateCount++;
    }

    public static (int value, int updateCount) GetValue(int index)
    {
        return (_array[index], _updateCount);
    }
}

上記のコードのSetValueメソッドは、内部の配列の指定のインデックスに値をセットし、更新回数を記録しています。GetValueメソッドは指定のインデックスの値と更新回数を返します。これらのメソッドはスレッドセーフではないため、マルチスレッドで正しく動作する保証はありません。今回はこれらをスレッドセーフにする方法をいくつか見ていきます。なお、説明用の例として簡単のために、範囲外のインデックスは指定されないという前提にします。

lockによる同期

// lock でスレッドセーフ化したもの
public static class Foo
{
    private static readonly object _lockObj = new object();
    private static int[] _array = new int[10];
    private static int _updateCount;

    public static void SetValue(int index, int value)
    {
        lock(_lockObj) {
            _array[index] = value;
            _updateCount++;
        }
    }

    public static (int value, int updateCount) GetValue(int index)
    {
        lock(_lockObj) {
            return (_array[index], _updateCount);
        }
    }
}

最初の例にlockを追加しただけですが、これだけでスレッドセーフになります。簡単ですね。lock(_lockObj)している間、他のスレッドはlock(_lockObj)内に入ることができないため、その内部ではマルチスレッドのことを考える必要がありません。難しいことを考えたくない、かつ特にパフォーマンスチューニングも求めていない場合、これで十分です。

ThreadStatic 属性によるスレッドセーフ

static フィールドにThreadStaticAttributeをつけると、簡単にスレッドセーフにすることができます。これは同期的に操作をするというより、そもそも同期をとる必要がない構造にするというものです。

// ThreadStatic による実装
public static class Foo
{
    private const int ArrayLength = 10;
    [ThreadStatic]
    private static int[]? _array;
    [ThreadStatic]
    private static int _updateCount;

    public static void SetValue(int index, int value)
    {
        _array ??= new int[ArrayLength];
        _array[index] = value;
        _updateCount++;
    }

    public static (int value, int updateCount) GetValue(int index)
    {
        return (_array == null) ? (0, 0) : (_array[index], _updateCount);
    }
}

最初のコードと違うのは、[ThreadStatic]が付いていることです。この属性が付いているフィールドはスレッドごとに別のフィールドが存在している状態になります。スレッドローカルなメモリにしかアクセスしていないため、マルチスレッドによる問題は発生しません。スレッド間での同期が不要なため高速です。

ただし、スレッド A でSetValueで値をセットしたものは、スレッドBでGetValueを呼んでも取得できません。そのため、この方法が使えるのはスレッドごとに独立していても問題がないものだけです。

注意点としては、[ThreadStatic]をつけるフィールドは static である必要があります。また、このフィールドは初期値をセットしたり、static コンストラクタで初期化してはいけません。初期化をしても、最初に呼ばれたスレッドでしか初期化が走らないためおかしくなります。そのため、各スレッドで_arrayにアクセスする時に配列を作成する必要があります。

ThreadStatic属性を使った方法は、現実的な場面ではSystem.Buffers.ArrayPool<T>.Sharedなどで使われています。(詳細は過去記事を参照)

SpinWait による軽量な同期

まず次のようなFastSpinLockという構造体を作ります。(中身は後で説明します)

public struct FastSpinLock
{
    private const int SYNC_ENTER = 1;
    private const int SYNC_EXIT = 0;
    private int _syncFlag;

    [MethodImpl(MethodImplOptions.AggressiveInlining)]
    public void Enter()
    {
        if(Interlocked.CompareExchange(ref _syncFlag, SYNC_ENTER, SYNC_EXIT) == SYNC_ENTER) {
            Spin();
        }
        return;
    }

    [MethodImpl(MethodImplOptions.AggressiveInlining)]
    public void Exit() => Volatile.Write(ref _syncFlag, SYNC_EXIT);

    [MethodImpl(MethodImplOptions.NoInlining)]
    private void Spin()
    {
        var spinner = new SpinWait();
        spinner.SpinOnce();
        while(Interlocked.CompareExchange(ref _syncFlag, SYNC_ENTER, SYNC_EXIT) == SYNC_ENTER) {
            spinner.SpinOnce();
        }
    }
}

これを使って、最初のコードをスレッドセーフにします。

// SpinWait による軽量な同期
public static class Foo
{
    private static FastSpinLock _spinLock;
    private static int[] _array = new int[10];
    private static int _updateCount;

    public static void SetValue(int index, int value)
    {
        try {
            _spinLock.Enter();
            _array[index] = value;
            _updateCount++;
        }
        finally {
            _spinLock.Exit();
        }
    }

    public static (int value, int updateCount) GetValue(int index)
    {
        try {
            _spinLock.Enter();
            return (_array[index], _updateCount);
        }
        finally {
            _spinLock.Exit();
        }
    }
}

まずは使い方ですが、try-finally で EnterメソッドとExitメソッドを呼びます。EnterからExitまでの間はlockと同じく単一のスレッドしか入れないため同期が取れます。注意点としては、このFastSpinLock構造体は mutable であるため、readonlyにしてはいけません。

どうしてこれで同期が取れるかというFastSpinLock構造体の原理を説明します。

同期を取るためのフラグが_syncFlagで、誰も制御を取っていない場合はこのフラグが0です。Enterメソッドを呼ぶと_syncFlagが0の場合、1になります。この時、このフラグの操作はInterlocked.CompareExchangeによってアトミックに行われます。0から1に変わった時のみ、そのままEnterメソッドを抜けます。(=スレッド競合が発生していない状態)

Enterメソッドで_syncFlagが1だった場合、スピンウェイトをします。スピンウェイトとは無限ループによってフラグが0になるのを待つことで同期を取る方法です。この時、ただ無限ループするだけではCPUに負荷がかかるだけで無駄な待機をしてしまうため、フラグが1だった場合は現在のスレッドの残りのタイムスライスを放棄して、次のコンテキストスイッチを待ちます。また、複数回ループを待機してもフラグを取得できなかった場合、待機時間を長くしてみるべきです。これらの操作を自分で正しく書くのは面倒なため、System.Threading.SpinWaitという構造体が用意されており、自動で行ってくれます。これでフラグを取得できるまでスピンウェイトします。

Exitメソッドではフラグを0に戻しています。_syncFlagはint型であるため、書き込みはアトミックであることが保証されています。ただし、このフラグはマルチスレッドから読み書きされるため、確実にメモリに書き込まれている必要があります。通常の代入ではコンパイラの最適化によって、シングルスレッドでは問題が起きない範囲で順番が前後したり、CPUキャッシュ上だけで操作が行われたりするため、この最適化をしないようにVolatile.Writeによって確実に書き込みます。

上記の実装は、ほとんどの場合競合が発生せず待機が行われない、かつ同期を取得している時間も非常に短いという想定において最速になるよう実装しています。Enterメソッドではホットパスのコード量を最小にしてインライン展開し、スピンウェイトの待機処理はインライン展開しないようにしています。スレッドの競合が発生しない場合、スレッドセーフではないコードに比べてオーバーヘッドが非常に少ないです。

ちなみに、このFastSpinLockとだいたい同じ実装をしているSystem.Threading.SpinLockという構造体があるのですが、System.Threading.SpinLockは上記の実装よりもう少しリッチで複雑な処理をしています。それを使ってもいいのですが、前述の条件で最小かつ最速は上記の実装です。

ベンチマーク

ベンチマークです。各手法それぞれ以下のようなコードの速度を計測しています。スレッド間の競合が発生しない場合での速度を計測しています。 (ベンチマークコードの全体はページ最後に gist を貼っています)

[Benchmark]
public int NotThreadSafe()
{
    for (int i = 0; i < 10; i++)
    {
        Foo.SetValue(i, i);
    }
    int result = 0;
    for (int i = 0; i < 10; i++)
    {
        result = Foo.GetValue(i).value;
    }
    return result;
}

BenchmarkDotNet=v0.13.0, OS=Windows 10.0.19042.1110 (20H2/October2020Update)
Intel Core i7-10700 CPU 2.90GHz, 1 CPU, 16 logical and 8 physical cores
.NET SDK=6.0.100-preview.4.21255.9
  [Host]     : .NET 5.0.6 (5.0.621.22011), X64 RyuJIT
  Job-DFQJUV : .NET 5.0.6 (5.0.621.22011), X64 RyuJIT

Jit=RyuJit  Platform=X64  IterationCount=100  

結果

Method	Mean	Error	StdDev	Median	Ratio	RatioSD	Rank	Gen 0	Gen 1	Gen 2	Allocated
not thread safe	51.11 ns	0.044 ns	0.129 ns	51.09 ns	1.00	0.00	1	-	-	-	-
lock	303.88 ns	0.399 ns	1.158 ns	303.73 ns	5.95	0.03	4	-	-	-	-
ThreadStatic	172.30 ns	0.128 ns	0.371 ns	172.18 ns	3.37	0.01	3	-	-	-	-
FastSpinLock	149.94 ns	0.158 ns	0.461 ns	149.82 ns	2.93	0.01	2	-	-	-	-

スレッドセーフでないコードが最も早いのは当然として、ThreadStatic属性による方法よりスピンロックの方が速いのが驚きです。おそらくThreadStatic属性の制約上、フィールドを初期化できないため毎回初期化されているかをチェックしている分だけ遅くなっています。逆に言うと条件分岐一回分で差が出るぐらいFastSpinLockの実装がガチガチです。

別にlockが遅いわけではないんですが、他が速すぎて相対的に遅いですね。

結論としては lock より速い方法はありますが、同期が必要な操作にかかる時間やスレッド競合の頻度などでかなりシビアな判断が必要になります。ここまで書いておいてなんですが、何を使えばいいかわからなければlockしてください。一番簡単です。

ベンチマークのコード

gist.github.com