BlockingCollectionでproducer-consumerパターンの上手い実装方法を教えて下さい。

Question

以下の要件でバックグラウンド非同期のファイル読み込みをしたいと考えています。
そこでproducer-consumerパターンで実装してみたのですが上手く行かずつまづいています。

・読み込みタスクはずっと稼働し続ける。
・読み込みタスクのオーダーは途中でキャンセルされる場合がある。
(その時あったリクエストは全て破棄)
・キャンセル後は再びオーダー待ちに戻る。
・オーダーはいつ入るか分からない。
・読み込み終わったらその段階でUIスレッドに結果を渡す。

オーダーが不定期なため、CPUリソースを抑えるためにBlockingCollection<T>を使用しています。
しかし逆にこれがキャンセルとの相性を悪くしており、読み込みタスク制御が上手く行きません。
Cancel()でタスクを止めたいのですが、BlockingCollectionが文字通りブロッキングしている時があるため、キャンセルをかけても受け付けない状態(WatingForActivattion)が出てきます。
orderをCompleteAdding()かけて動かそうとしても、今度はかかるタイミングでUIのデッドロックが発生してしまい、結局止められないケースが出ています。

BlockingCollectionを使用しなければ比較的簡単に解決できると思いますが、ずっとCPUリソースを消費してしまうため最終手段としたいと思っています。

非同期プログラミングに不慣れなので、良い実装方法を探しています。
いい解決法などありましたらよろしくおねがいします。

以下に簡易ソースを書きます。

//UIスレッドにデータを返すためのインターフェイス
public interface IFileDataSetter
{
    Task SetFileDataAsync(byte[] data);
}

//読み込みタスク側
public class AsyncLoader
{
    private IFileDataSetter setter;    //実装先はForm
    private BlockingCollection<string> order = new BlokingCollection<string>();
    private CancellationTokenSource cts = new CancellationTokenSource();
    private Task loadTask;

    public void LoadStart()
    {
        loadTask = loadBackground();
    }

    private Task loadBackground()
    {
        return Task.Run(() =>
        {
            foreach (var fileName in order.GetConsumingEnumerable())
            {
                //ファイルを読み込みUIスレッドに結果を渡す
                await setter.SetFileDataAsync(data);
            }
        }, cts.Token);
    }

    public void Order(string fileName)
    {
        order.Add(fileName);
    }

    public void Cancel()
    {
        cts.Cancel(true);    //読み込みタスクをキャンセル
        while (!loadTask.IsCanceled) //←ブロッキングされてる時はずっとキャンセルできない
        {
            Thread.Sleep(100);
        }
        loadTask = loadBackground();
    }
}

Answer 1

こんにちは

BlockingCollectionを使ったことがありませんので、見当違いな回答でしたらすみません。

loadBackground()のタスク内で、以下のサイトのように、
タスクがキャンセルされたときの処理を追加して、タスクを終わらせてあげればよいのではないでしょうか。
タスクのキャンセル

Comments

[退会済みユーザー]

回答ありがとうございます。

BlockingCollectionはタスク内の動作を『ブロッキング』する機能を持っています。
(具体的にはforeachのGetConsumingEnumerable()の部分)
このため別スレッドでorderにAddされるまでforeachのアイテム取得状態でタスクが停止したままになります。
この停止中にCancelを送ろうが、そもそもタスクが動いていないのでCancelを取得できないという事です。

もちろんorderにAddやCompleteAddingをすればforeachが流れだすのでそこで捕まえられます。
しかし、そこまでしないとキャンセル制御が成り立たないのかというところが問題としているところでした。

ここまで面倒なら普通にConcurrentQueueなどを使ってwhileループした方が楽です。
しかし、whileループをするのはCPUを消費するので出来ればしたくありません。

となれば、最終的に仕様そのもの(ずっと動作している状態)を見直した方がいい気がしてきました。

ありがとうございました。

[daichan]

なるほど。詳しい解説ありがとうございます。

GetConsumingEnumerable()メソッドにCancellationTokenオブジェクトを渡せそうですけど、これで解決できませんか？

[退会済みユーザー]

再度回答ありがとうございます。

確かにありますね。見落としていました。
これなら上手く解決できそうな感じです。

ありがとうございました。