[C#]Reactive Extensionsでasyncメソッドを順番に実行したい

Question

前提・実現したいこと

Reactive Extensionsで入力→変換→出力の"変換"の部分で非同期メソッドを使用するが、それを再入せず、順番に実行したいです。
また、変換→出力部分を後で停止できるようにもしたいです。

ここではサンプルとして、1秒毎に値を発行するObservable.Intervalを入力、文字列への変換をするメソッドConvertAsyncを変換、時刻などを付加してコンソールへプリントするメソッドWriteLineDebugを出力としています。

変換メソッドConvertAsyncは内部でawait Task.Delayをしていますが、実際に使用する場合はここがファイルやネットワークアクセスになります。
このConvertAsyncの開始・終了が前後しないように順番に行いたいです。

発生している問題・エラーメッセージ

SelectManyの場合

変換メソッドをSelectManyで呼んだ場合、変換メソッドConvertAsyncの順番が前後してしまいます。
ただし、変換→出力部分の停止はできています。

以下では0~3個目までは問題ないですが、3個目の開始Start ConvertAsync 3の後に、3個目の完了End ConvertAsync 3前に4個目が開始Start ConvertAsync 4してしまいます。

出力結果

00.065|Thread:1|Start Subscription
00.615|Thread:5|Start ConvertAsync 0
00.745|Thread:4|End   ConvertAsync 0
00.750|Thread:6|Subscrive inputValue:[Converted:0 delayMilliSec:100]
01.100|Thread:6|Start ConvertAsync 1
01.210|Thread:6|End   ConvertAsync 1
01.210|Thread:5|Subscrive inputValue:[Converted:1 delayMilliSec:100]
01.600|Thread:4|Start ConvertAsync 2
01.710|Thread:5|End   ConvertAsync 2
01.710|Thread:4|Subscrive inputValue:[Converted:2 delayMilliSec:100]
02.102|Thread:5|Start ConvertAsync 3
02.597|Thread:5|Start ConvertAsync 4
02.838|Thread:5|End   ConvertAsync 3
02.839|Thread:4|Subscrive inputValue:[Converted:3 delayMilliSec:723]
03.107|Thread:4|Start ConvertAsync 5
03.605|Thread:5|Start ConvertAsync 6
04.105|Thread:4|Start ConvertAsync 7
04.115|Thread:4|End   ConvertAsync 4
04.115|Thread:4|Subscrive inputValue:[Converted:4 delayMilliSec:1513]
04.145|Thread:4|End   ConvertAsync 5
04.145|Thread:4|Subscrive inputValue:[Converted:5 delayMilliSec:1029]
04.575|Thread:4|End   ConvertAsync 6
04.575|Thread:4|Subscrive inputValue:[Converted:6 delayMilliSec:948]

04.605|Thread:1|Dispose Subscription
04.605|Thread:5|Start ConvertAsync 8
06.090|Thread:5|End   ConvertAsync 7
06.289|Thread:5|End   ConvertAsync 8

Scan+Concatの場合

変換メソッドをScan+Concatで呼んだ場合、変換メソッドConvertAsyncの順番は保たれますが、
変換部分がすぐ停止しません。出力部分だけがすぐ停止されます。

以下では5個目完了ぐらいのタイミングで停止処理disposerA.Dispose()しても、12個目まで変換は続きます。
大体止めた個数の倍ぐらいまで変換が続くようです。20個目で停止→40個目ぐらいまで変換される。

出力結果

00.065|Thread:1|Start Subscription
00.613|Thread:4|Start ConvertAsync 0
00.745|Thread:4|End   ConvertAsync 0
00.751|Thread:5|Subscrive inputValue:[Converted:0 delayMilliSec:100]
01.121|Thread:5|Start ConvertAsync 1
01.233|Thread:5|End   ConvertAsync 1
01.233|Thread:5|Subscrive inputValue:[Converted:1 delayMilliSec:100]
01.614|Thread:6|Start ConvertAsync 2
01.725|Thread:4|End   ConvertAsync 2
01.725|Thread:4|Subscrive inputValue:[Converted:2 delayMilliSec:100]
02.112|Thread:4|Start ConvertAsync 3
03.978|Thread:6|End   ConvertAsync 3
03.979|Thread:6|Start ConvertAsync 4
03.979|Thread:5|Subscrive inputValue:[Converted:3 delayMilliSec:1858]
05.823|Thread:7|End   ConvertAsync 4
05.823|Thread:7|Start ConvertAsync 5
05.823|Thread:6|Subscrive inputValue:[Converted:4 delayMilliSec:1836]

06.920|Thread:1|Dispose Subscription
07.223|Thread:4|End   ConvertAsync 5
07.223|Thread:4|Start ConvertAsync 6
08.545|Thread:6|End   ConvertAsync 6
（略）
15.244|Thread:4|Start ConvertAsync 12
16.961|Thread:6|End   ConvertAsync 12

該当のソースコード

class Program
{
    static Random rand = new Random();
    static Stopwatch stopwatch = new Stopwatch();

    static async Task Main(string[] args)
    {
        stopwatch.Start();

        //[入力]　500msec間隔で{0,1,2,3,4,5,6,,,}と流れてくる
        var timer = Observable.Interval(TimeSpan.FromMilliseconds(500));

        WriteLineDebug($"Start Subscription");

        var disposerA = timer
        //[変換] 文字列に変えるが、3個目以降は遅くなる
        //Case1-SelectMany
        .SelectMany(l => ConvertAsync(l))
        //Case2-Scan+Concat
        //.Scan(Task.FromResult(""), async (previousTask, l) =>
        //{
        //    await previousTask.ConfigureAwait(false);
        //    return await ConvertAsync(l).ConfigureAwait(false);
        //})
        //.Concat()

        //[出力] 現在時刻などを追加してプリント
        .Subscribe(x =>
            WriteLineDebug($"Subscrive inputValue:[{x}]"));

        //Enter押下したら、購読を停止
        Console.ReadLine();

        WriteLineDebug("Dispose Subscription");
        disposerA.Dispose();

        //停止を確認
        Console.ReadLine();
    }

    private async static Task<string> ConvertAsync(long l)
    {
        WriteLineDebug($"Start ConvertAsync {l}");

        //実際はファイルアクセスやネットワークアクセスがあるとする
        int delayMilliSec = l >= 3 ? rand.Next(600, 2000) : 100;
        await Task.Delay(delayMilliSec).ConfigureAwait(false);

        WriteLineDebug($"End   ConvertAsync {l}");

        return $"Converted:{l} delayMilliSec:{delayMilliSec}";
    }

    private static void WriteLineDebug(string message) =>
        Console.WriteLine($"{stopwatch.ElapsedMilliseconds * 0.001:00.000}|Thread:{Thread.CurrentThread.ManagedThreadId}|{message}");
}

試したこと

以下のブログの目次を参考に、Switch()、スケージューラーの固定を検討しましたが、期待した動作にはなりませんでした。
Reactive Extensions再入門

補足情報（FW/ツールのバージョンなど）

C# 8.0
.NET 5.0
"System.Reactive" Version="5.0.0"
VisualStudio 2019

Comments

[Zuishin]

await していないからでは？

[soi013]

`.SelectMany(l => ConvertAsync(l))`
の中で、ということでしょうか？
以下の2つのケースを試してみましたが、どちらも結果は"SelectManyの場合"と同じになりました。

`.SelectMany(async l => await ConvertAsync(l))`
`.SelectMany(async l => await ConvertAsync(l).ConfigureAwait(false))`

[Zuishin]

今調べられないので引き続き緩いコメントですが、SelectMany はタスクを受け取る引数ありましたっけ？
IObservable に変換しないといけないんじゃないでしょうか？

[soi013]

はい、SelectManyは19個（！）のオーバーロードがあり、この場合は以下のシグネチャのメソッドが呼ばれます。

`IObservable<TResult> SelectMany<TSource, TResult>(this IObservable<TSource> source, Func<TSource, Task<TResult>> selector)`

なお、以下の場合ではTaskからIObservableになりますが、結果は同じです。もしかしたら上のメソッドはこれのシンタックスシュガーかもしれません。
`.SelectMany(l => ConvertAsync(l).ToObservable())`

[Zuishin]

それだと戻り値が一つのタスクだけになりそうなので、SelectMany の意味がないような。
Task オブジェクトを次に送ってるだけですよね。
SelectMany ではなく Select を使うんじゃないですか？

いや、よく考えると Select でもだめですね。

[soi013]

そうなんですよね。`SelectMany`なのに、全然"Many"じゃないんですよね。
Selectの場合は返り値が`IObservable<Task<string>>`になるので、Subscribeも少しく変わります。

```
.Select(async l => await ConvertAsync(l))
.Subscribe(async x =>
WriteLineDebug($"Subscrive inputValue:[{await x}]"));
```

結果はやはりSelectManyと同じです。
なお、`.Select(l => ConvertAsync(l))`でも同じです。

[soi013]

Scan+Concatのケースで変換が無限には続かなかったので、修正しました。

[Zuishin]

Select(l => ConvertAsync(l).Result) で待つことはできますが、これではだめなんですよね？
考えてみましたが、Convert を待つ間、タイマーからの出力をどこかに溜めておかなければいけないので、これ以外だと新しいオペレーターを作るか Subject を使うくらいしか思いつきませんでした。

[soi013]

やってみたところ、`Select(l => ConvertAsync(l).Result) `の結果は期待した動作になります！
ただ、`.Result`はスレッド待機させますし、デッドロックの原因になりそうで、できれば`await`したいですかね。

> 新しいオペレーターを作るか Subject を使う

この部分をもう少し詳しく教えてもらうことはできますでしょうか。
Timerからの入力を一度Subjectに入れて、そこから`SelectMany`などで変換につなぐイメージでしょうか。
Subjectに入れるということはHot変換だと思いますが、それが解決にどうつながるか、上手くイメージつきません。

[Zuishin]

デッドロックの心配であれば、Convert は l しか参照していないので大丈夫だと思いますが、Convert の仕様が変わったとしても ObserveOn や SubscribeOn でスケジューラをセットすれば大丈夫だと思います。それでデッドロックするなら await を使ってもデッドロックするでしょう。

> 新しいオペレーターを作るか Subject を使う

これは、time からの出力を受け取り次第待ち行列に入れ、待ち行列が空でない限り一つずつ取り出して await Convert を呼び出し、それが終了した時点で次のオブザーバーの OnNext を呼び出すという意味です。
それをオペレーターや Subject でできると思います。

[soi013]

一応いただいた情報をもとに自分で考えて作ってみましたが、
順番は保たれるが、Consoleからの入力を待てず（`while(true)`のせい）、3個目以降は飛び飛びになる（`FirstAsync`のせい）不思議な挙動になってしまいました。
なにか、ヒントなどありましたら、教えていただけると助かります。もちろん時間は後でも大丈夫です。

```
var subjectInput = new Subject<long>();

var disposer0 = timer
.Subscribe(l => subjectInput.OnNext(l));

var subjectConverted = new Subject<string>();

var disposerA = subjectConverted

//[出力] 現在時刻などを追加してプリント
.Subscribe(x =>
WriteLineDebug($"Subscrive inputValue:[{x}]"));

while (true)
{
var nextValue = await subjectInput.FirstAsync();
var convertedValue = await ConvertAsync(nextValue).ConfigureAwait(false);
subjectConverted.OnNext(convertedValue);
}
```

[Zuishin]

待ち行列に入れなければ取りこぼしてしまいます。
汚いですが、イメージ的にはこんな感じです。

var subject = new Subject<string>();
var disposerA = subject.Subscribe(x =>
WriteLineDebug($"Subscrive inputValue:[{x}]"));
var queue = new Queue<long>();
var query = timer
.Select(l => { queue.Enqueue(l); return Unit.Default; })
.Next()
.SelectMany(_ => Enumerable.Range(0, int.MaxValue).TakeWhile(_ => queue.Any()).Select(i => queue.Dequeue()));
foreach (var l in query)
{
subject.OnNext(await ConvertAsync(l));
}

[soi013]

具体的なコードありがとうございます。ただこれですと、`foreach (var l in query)`が無限ループになり、その後のConsole.ReadLine()の行に到達できず、したがって停止もできません。

少し変えて、以下のように`shouldStop = true;`のように止まるかどうかを別の変数に取るようにしたら、望む動作結果にはなりました。
ただ、すごくReactive感のないコードになってしまいます。。。あと、Task.Run()内でエラーが発生するとキャッチできない気がします。

```
bool shouldStop = false;
var subject = new Subject<string>();
var disposerA = subject.Subscribe(x =>
WriteLineDebug($"Subscrive inputValue:[{x}]"));

var queue = new Queue<long>();
var query = timer
.Select(l =>
{
queue.Enqueue(l);
return Unit.Default;
})
.Next()
.SelectMany(_ =>
Enumerable.Range(0, int.MaxValue)
.TakeWhile(_ => queue.Any())
.Select(i => queue.Dequeue()));

//Fire&Forget
var _ = Task.Run(async () =>
{
foreach (var l in query)
{
if (shouldStop)
return;

subject.OnNext(await ConvertAsync(l));
}
});

```

[Zuishin]

ちゃんとやるなら Replay 的なオペレーターを作るのがいいと思いますが、作っても私は使わなそうなので、手抜きで問題点だけを示しています。
問題点は、ConvertAsync を待っている間に time を取りこぼすことなので、これをどうにか解決しなければいけません。
キューにキャッシュするのが一番良さそうに思います。

[soi013]

そうですね、`ConvertAsync`の結果を待機させる以上は、入力を保持しておく必要があるので、何らかのコレクションに貯める必要はありますよね。
上手いことその辺を隠蔽して、かつキャプチャが無いように書きたいですかね。
ちょっと考えてみます。

入力に対してReactiveに反応して処理したい、でも実行順序は守って欲しい、という要望が矛盾しているような気もしてきました。

[soi013]

いただいた情報などを参考に色々調べた結果、Interactive.Asyncが使えそうだということが分かりました。よかったら回答を見て、コメントなどいただければ助かります。

Answer 1

Interactive.AsyncでPull/Pushの相互切り替え

いろいろ試行錯誤をした結果、Interactive.Asyncを使用して、Pull型(IAsyncEnumerable)とPush型（IObservable）の相互切り替えをするのがよいようです。

以下のコードでIObservableな入力timerからIAsyncEnumerableへ切り替え、非同期処理を順番に実行、またIObservableへ切り替えています。

timer
    .ToAsyncEnumerable()
    .SelectAwait(async x => await ConvertAsync(x).ConfigureAwait(false))
    .ToObservable()
    .Subscribe(x =>

RxとIx-Asyncってどう違うの？という説明は以下を見ていただくとわかりやすいです。

非同期ストリームを扱うときはRxとIx-Asyncを使い分ける - cactuaroid blog

ここでは上のコードのOperatorを順番に説明します。

ToAsyncEnumerable()

.ToAsyncEnumerable<T>()は受け取った、IObservable<T>をIAsyncEnumerable<T>に変換します。後段から列挙されるたびに、IObservable<T>の入力を待つか、すでに溜まっていた入力を返します。

後段の処理ConvertAsyncは順番が前後したくないので、使う側のタイミングで値を受け取りたい、つまりPull型です。@Zuishinさんがコメントで述べられていたように、これをするためには、IObservable<T>からの入力を貯める待ち行列が必要になります。

実際、Interactive.Asyncのソースコードを見てみると、入力のOnNextでConcurrentQueueにEnqueueします。後段から列挙MoveNextAsyncされると、すでに溜まっていたらTryDequeue、なければ待ちます。

SelectAwait

SelectAwait<TSource,TResult>は指定されたデリゲートFunc<TSource,ValueTask<TResult>>を受け取ったIAsyncEnumerable<TSource>に対して、順番に実行していきます。実際的には以下のコードのような動きです。

await foreach (var item in source)
{
    yield return await selector(item);
}

返り値はIAsyncEnumerable<Task<TResult>>ではなく、IAsyncEnumerable<TResult>なのがポイントです。

ToObservable

ToObservable<T>では.ToAsyncEnumerable<T>()とは逆にIAsyncEnumerable<T>をIObservable<T>に変換します。

ソースコードを見てみると、入力をMoveNextAsyncで列挙して、後段のObserverにOnNextで送っています。

出力結果

入力(1000msec)よりも変換が遅くなっても順番が前後することもなく、購読を停止後に変換処理も止まっています。

00.057|Thread:1|Start Subscription
00.642|Thread:4|Start ConvertAsync 0
00.744|Thread:4|End   ConvertAsync 0
00.746|Thread:4|Subscrive inputValue:[Converted:0 delayMilliSec:100]
01.120|Thread:4|Start ConvertAsync 1
01.228|Thread:5|End   ConvertAsync 1
01.229|Thread:5|Subscrive inputValue:[Converted:1 delayMilliSec:100]
01.628|Thread:6|Start ConvertAsync 2
01.735|Thread:4|End   ConvertAsync 2
01.735|Thread:4|Subscrive inputValue:[Converted:2 delayMilliSec:100]
02.119|Thread:6|Start ConvertAsync 3
03.952|Thread:6|End   ConvertAsync 3
03.952|Thread:6|Subscrive inputValue:[Converted:3 delayMilliSec:1821]
03.953|Thread:6|Start ConvertAsync 4
05.645|Thread:6|End   ConvertAsync 4
05.645|Thread:6|Subscrive inputValue:[Converted:4 delayMilliSec:1668]
05.645|Thread:6|Start ConvertAsync 5
06.851|Thread:6|End   ConvertAsync 5
06.852|Thread:6|Subscrive inputValue:[Converted:5 delayMilliSec:1196]
06.852|Thread:6|Start ConvertAsync 6

07.939|Thread:1|Dispose Subscription
08.883|Thread:6|End   ConvertAsync 6
08.884|Thread:6|Subscrive inputValue:[Converted:6 delayMilliSec:1915]

Main全体コード

質問文とたいぶ重複しますが、Main全体のコードも貼っておきます。

static async Task Main(string[] args)
{
    stopwatch.Start();

    //[入力]　500msec間隔で{0,1,2,3,4,5,6,,,}と流れてくる
    var timer = Observable.Interval(TimeSpan.FromMilliseconds(500));

    WriteLineDebug($"Start Subscription");

    //IX.NET方式
    //https://cactuaroid.hatenablog.com/entry/2018/09/10/011010
    var disposerA = timer
        //ストリームをPull型（IAsyncEnumerable）に変換
        .ToAsyncEnumerable()
        //[変換] 文字列に変えるが、3個目以降は遅くなる
        .SelectAwait(async x => await ConvertAsync(x).ConfigureAwait(false))
        //ストリームをPush型（IObservable）に変換
        .ToObservable()
        //[出力] 現在時刻などを追加してプリント
        .Subscribe(x =>
            WriteLineDebug($"Subscrive inputValue:[{x}]"), ex => WriteLineDebug($"Subscrive Error ex:{ex.Message}"), () => WriteLineDebug($"CompletedSelectMany"));

    //Enter押下したら、購読を停止
    Console.ReadLine();

    WriteLineDebug("Dispose Subscription");
    disposerA.Dispose();

    //停止を確認
    Console.ReadLine();
}

環境

"System.Interactive.Async" Version="5.0.0"
"System.Reactive" Version="5.0.0"

Comments

[Zuishin]

非常にスマートな解決法ですね。

[soi013]

Zuishinさんにそう言っていただければ安心です。ありがとうございます。

Answer 2

「発行側は自由なタイミングで発行するが、消費する側はPull型でタイミングを制御したい」場合にはRxよりもChannelsがいいかもしれません。

System.Threading.Channelsを使う
https://qiita.com/skitoy4321/items/c19ca3dc7624a7049fd5

もちろん非同期ロックなり待ち行列のキューを自前実装すればRxでも実現出来ると思いますが、実装ハードルが高く保守性もやや難のあるものが出来そうな気がします。

Channelsで巨人の方を借りつつ省力で実装しつつ、空いた時間でasync/awaitの特にキャンセル処理を行儀よく実装する方法を勉強していけるといいんじゃないでしょうか。

Comments

[soi013]

なるほど、`Channels`、昔さらっと読んだけど忘れていました。少し勉強して今回の解決に使えるか試してみます。

[soi013]

調べてみたのですが、ChannelsはPull型のストリームを扱うもので、特に生産:消費側が1:n or n:1 or n:nの柔軟な組み合わせが強みなようです。これはこれでRxとは別に使えそうです。

ただ今回の場合は全体としてはPush型でストリームを扱って、一部だけPullしたい、というケースでした。そしてChannelsを調べる過程で、そもそも必要なのは`IAsyncEnumerable<T>`との変換だという事に気づきました。
ですので、今回はInteractive.Asyncを使って解決するようにしました。
なお、ChannelsでもInteractive.Asyncと同様に`IAsyncEnumerable<T>`を入出力できます。
`Reader.ReadAllAsync()`とすると、`IAsyncEnumerable<T>`を受け取れます。

つまり、Push型の入力(Rx)から、並列可能な変換はRxのオペレーターで行い、
Interactive.AsyncでPull型のデータにして、
Channelsでn個の消費者から同時に消費される、なんてことが可能ですね。

[tor4kichi]

Linqの表現を維持したままIObservable->ToAsyncEnumerable() でより良く書けるんですね。Ixをちゃんと勉強してなかった上に知ったかぶりでChannelsを提示してしまってお恥ずかしい。

soi013さんが自己解決された回答の内容もよくまとまっていて勉強になります。

[soi013]

いえいえ、Channelsを理解する過程で、
やりたいことが、Pull型の非同期ストリームである、ということに気づくことが出来たので、tor4kichiさんのおかげです。
ありがとうございました。