C#非同期メソッドを指定したスレッド上で動かす

内容

C#の非同期メソッドが実行されるスレッドはThreadPoolクラスが生成したワーカースレッドですが、これを指定したスレッドに切り替える方法を紹介します。

スレッド切り替えの様子

非同期メソッド中でSwitchToメソッドを呼び出すことでそれ以降の動作スレッドを切り替えます。

切り替えの様子：

async Task SwitchToTestAsync()
{
    Console.WriteLine(".NET ThreadPoolクラスのワーカースレッドまたは起動元スレッド上");
    
    await mainThreadQueue.SwitchTo();
    Console.WriteLine("メインスレッド上");

    await userThreadPool.SwitchTo();
    Console.WriteLine("自前スレッドプールのワーカースレッド上");

    await Task.Yield();
    Console.WriteLine(".NET ThreadPoolクラスのワーカースレッド上");
}

実装方法

次の２組を切り替えたいスレッド種別毎に用意します。例えばメインスレッド向け、自前スレッドプール向けなどです。

• Action型を入れられるキューおよびそのキューからActionを取り出して実行するスレッド
• SwitchToメソッド

メインスレッドへの切り替え実装

アクション消費スレッドとキュー

キュー経由でActionを取り出してメインスレッドで実行する処理を用意します。次のようなコードで実装できます。RunAsyncはAction追加例です。

static void Main(string[] args)
{
    var mainThreadQueue = new BlockingCollection<Action>();
    RunAsync(mainThreadQueue);
    while (EnableFlag)
    {
        var action = mainThreadQueue.Take();
        action();
    }
}

static async void RunAsync(BlockingCollection<Action> mainThreadQueue)
{
    ...
    mainThreadQueue.Add(() => Console.WriteLine("Now on main thread."));
    ...
}

SwitchToメソッド

SwitchToメソッドのコード例です。先に実装したアクション消費スレッドが BlockingCollection 型を利用しているのでこの拡張メソッドでも参照してしまっていますが、専用クラスに置き換えるなどして影響範囲をより狭くできます。ここではコード行数を減らすためにそのまま参照しています。

仕組みを説明します。C#では非同期メソッド中でawaitするとそれ以降の処理をActionとして取得できます。取得したActionをキューに入れて別スレッドで取り出し実行することで動作スレッド切り替えを実現しています。本来は何かしらのI/O処理が終わるのを待ってそのActionを呼び出すことで後続処理を再開するための仕組みのようです。

public static class ThreadSwitcher
{
    public static ThreadNotifyCompletion SwitchTo(this BlockingCollection<Action> actionQueue)
    {
        return new FiberNotifyCompletion(actionQueue);
    }
}

public struct ThreadNotifyCompletion : INotifyCompletion
{
    private readonly BlockingCollection<Action> _actionQueue;

    public ThreadNotifyCompletion(BlockingCollection<Action> actionQueue)
    {
        _actionQueue = actionQueue;
    }

    /// <summary>
    /// await enabling.
    /// </summary>
    /// <returns></returns>
    public ThreadNotifyCompletion GetAwaiter()
    {
        return this;
    }

    /// <summary>
    /// Always false, to have the completion process performed.
    /// </summary>
    public bool IsCompleted { get { return false; } }

    /// <summary>
    /// Called to resume subsequent processing at the end of await.
    /// </summary>
    /// <param name="action"></param>
    public void OnCompleted(Action action)
    {
        _actionQueue.Add(action);
    }

    /// <summary>
    /// Do nothing.
    /// </summary>
    public void GetResult()
    {
    }
}

自前スレッドプールへの切り替え実装

アクション消費スレッドとキュー

自前スレッドプールを作成します。実装上の要件は Action をキューイングできることだけです。

実装するには最低限下記のようなメソッド構成が要るはずです。内部動作的には new Thread によっていくつかスレッドを作り、共通の BlockingCollection<Action> からActionを取り出して実行する処理をそれぞれで実行します。キューを外部に見せる代わりにキューイング用メソッドを用意することで .NET標準の ThreadPool.QueueUserWorkItem(WaitCallback) メソッドと似たような使い勝手にそろえることができます。

メソッド構成例：

public class UserThreadPool
{
    public void Queue(Action action) { }
    public void Start() { }
    public void Stop() { }
    public Task Join() { }
}

長くなるので詳細な例は割愛しますが、参考コードをリンクします。

自前スレッドプール参考コード：

リンク先のコードでは直接キューを見せずにキューイング用のメソッドを提供しています。 void Queue(WaitCallback callback) がそれです。

SwitchToメソッド

メインスレッド向けの実装とほぼ同じですが、直接キュー追加メソッド BlockingCollection<Action>.Add を呼び出していたところをキューイング用メソッド呼び出しに変えます。SwitchToメソッドの実装方法はいくつか考えられますが、下記コード例はインタフェース定義して拡張メソッドから参照させたものです。

public class UserThreadPool : IThreadPool
{ ... }

public interface IThreadPool
{
    void Queue(Action action);
}

public static class ThreadPoolSwitcher
{
    public static ThreadPoolNotifyCompletion SwitchTo(this IThreadPool threadPool)
    {
        return new ThreadPoolNotifyCompletion(threadPool);
    }
}

public class ThreadPoolNotifyCompletion
{ ... }

参考資料

ライブラリ実装

今回紹介したSwitchToメソッドを含む、C#向けスレッディングライブラリを開発しています。もともとはRetlangというライブラリがあってそれを趣味のゲームプログラミングに使う目的でリファクタリングしているものとなります。自分が使うこと以外考えておらず突然破壊的な変更をすることもあるために実用には適しませんが、もうすこしSwitchTo周りの実装コード例が見たいという場合にはご覧ください。

SwitchToメソッドを含むC#向けスレッディングライブラリ(GitHub)：

テストコードでのUserThreadPool.SwitchTo利用箇所：