Parallel.ForEachでのawaitは可能でしょうか

Question

概要

Parallel.ForEachのdelegate上でawaitを使いたいのですが、
すべての処理が実行されずにParallel.ForEachが抜けてしまいます。
Parallel.ForEachのdelegate上では同期処理を書かなくてはならないのでしょうか。

背景

listのデータに対して順序にかかわらず上限のある並列数(1～)で実行したい。
システム全体でのスレッドの上限があるため、極力節約したい。
※目くじら立てる程でもないのかもしれませんが、ほとんどの処理をAsync化してしまったので、wait等は避けようとしてどうしてこうなった。
(追記 2020/01/15 21:15 JST)
なぜParallel.ForEachを使い、並列数を指定しているのかの、説明ができていませんでした。
ループ処理内でネットワーク通信を挟んでいるためです：
ネットワーク通信先では同時接続数の上限もあるため、同時アクセス数の制限を設けたい。(Server busyとなってしまう)
なお、ネットワーク接続先は複数箇所であるため、クライアント側のソケットも使い果たすとマズイため、一部処理を並列化し、終了を簡単に待つことができる機構を求めた結果、この質問になってしまいました。

環境

• .NET Framework 4.5.2
• VisualStudio 2015 Community Edition
• Windows 10 home

もしくは

• .NET Core 2.2
• VisualStudio 2017 Professional
• Windows 10 Pro

備考（オフトピック）

コード確認はできていませんが、Task.Wait()メソッド(引数無し)では通常、呼び出し元スレッドを維持してしまう認識であっているか、教えていただける幸いです。

検証

コード

using System;
using Microsoft.VisualStudio.TestTools.UnitTesting;
using System.Collections.Generic;
using System.Threading.Tasks;
using System.Threading;
using System.Diagnostics;
using System.Collections;
using System.Linq;

namespace ParallelTest
{
    [TestClass]
    public class UnitTest1
    {
        [TestMethod]
        public void TestMethod1()
        {
            var list = Enumerable.Range(1, 30).ToList();
            ParallelFor(10, list, async (item) =>
            {
                Trace.TraceInformation("Inner start {0}", item);
                await Task.Delay(100);
                Trace.TraceInformation("Inner end {0}", item);
            });

        }
        /// <summary>
        /// 同時実行数制限付き並列動作
        /// </summary>
        /// <typeparam name="T">リストのデータタイプ</typeparam>
        /// <param name="degree">並列数</param>
        /// <param name="list">並列処理するリスト</param>
        /// <param name="task">並列処理内容</param>
        static void ParallelFor<T>(int degree, IList<T> list, Func<T,Task> task)
        {
            Trace.TraceInformation("Enter");
            Parallel.ForEach(list, new ParallelOptions (){ MaxDegreeOfParallelism = degree }, async (data) =>
            {
                Trace.TraceInformation("Task start {0}", data);
                await task(data);
                Trace.TraceInformation("Task end  {0}", data);
            });
            Trace.TraceInformation("End");
        }
    }
}

結果

Task endが呼び出されることなく、終了してしまう。

Enter
Task start 10
Task start 4
Inner start 10
Task start 7
Inner start 7
Task start 13
Task start 1
Inner start 1
Inner start 4
Task start 8
Inner start 8
Task start 9
Inner start 9
Task start 14
Inner start 14
Task start 11
Inner start 11
Task start 12
Inner start 12
Inner start 13
Task start 19
Inner start 19
Task start 20
Inner start 20
Task start 21
Inner start 21
Task start 22
Inner start 22
Task start 16
Inner start 16
Task start 17
Inner start 17
Task start 18
Inner start 18
Task start 25
Inner start 25
Task start 26
Inner start 26
Task start 27
Inner start 27
Task start 28
Inner start 28
Task start 2
Inner start 2
Task start 3
Inner start 3
Task start 23
Inner start 23
Task start 29
Inner start 29
Task start 30
Inner start 30
Task start 5
Inner start 5
Task start 6
Inner start 6
Task start 24
Inner start 24
Task start 15
Inner start 15
End

Comments

[Zuishin]

ParallelFor を非同期にしないのに意味はありますか？

[testset]

listの処理が終わった後、同期処理を行うため、非同期にしていません。

なぜParallel.ForEachを使い、並列数を指定しているのかの、説明ができていませんでした。
ループ処理内でネットワーク通信を挟んでいるためです：
ネットワーク通信先では同時接続数の上限もあるため、同時アクセス数の制限を設けたい。(Server busyとなってしまう)
なお、ネットワーク接続先は複数箇所であるため、クライアント側のソケットも使い果たすとマズイため、一部処理を並列化し、終了を簡単に待つことができる機構を求めた結果、この質問になってしまいました。

[testset]

この質問を投稿すべきか数週間悩んでいましたが、
結果として、いろいろな方面からご回答いただき、非常に勉強になりました。
PLINQのパターンや、Taskを単純にキックする物を後日(未定)対応し、
質問本文にでも記載させていただきたく存じます。
→ [記載しました](#reply-342783)

Answer 1

こんにちは。

Parallel.ForEach に渡す各処理を非同期メソッドにしたいという質問だと認識しました。
おそらくですが、それをやるのは無意味です。
Parallel.ForEach は指定数のスレッドを確保し、その中でそれぞれの処理を並列に実行するものですが、非同期メソッド (Task) は実処理がスレッドから切り離されているため、正しく並行性の管理ができません。

この場合、何も考えずに通常のループなどで非同期メソッドを全て起動してしまえば、Task はスレッドプールを利用するため、スレッドが際限なく増えるなどは起きず、システムが持つスレッドを良い具合に使ってくれます。

Parallel は、大量の (比較的) 短い「同期」処理を、指定した量のスレッドを予め確保して最速で走らせるためのメソッドだと考えるのが良いと思います。

Comments

[testset]

APIドキュメントにもFunc<TLocal,Task>のようなオーバーロード(?)も存在しないため、「非同期メソッド (Task) は実処理がスレッドから切り離されて」しまうため、期待する動作とならないと認識しました。
https://docs.microsoft.com/ja-jp/dotnet/api/system.threading.tasks.parallel.foreach?view=netframework-4.5.2

Taskを全て起動させる方向で考えいます。

Answer 2

この例みたいに、
同時実行数制御をしたいのなら、
https://qiita.com/TsuyoshiUshio@github/items/79ad787899cddaa3ac1c
の記事みたいに、
await semaphoreSlim.WaitAsync();
semaphoreSlim.Release();
を使うとできるよ。
PLINQでは、適切にできている保証がないわ。

Answer 3

        [TestMethod]
        public void MyTestMethod30()
        {
            System.Console.WriteLine("S");
            var list = Enumerable.Range(1, 30)
                .AsParallel()
                .WithDegreeOfParallelism(4)
                .Select(async n =>
             {
                 System.Console.WriteLine($"Start{n}");
                 await Task.Delay(100);
                 System.Console.WriteLine($"End{n}");
             }).ToArray();

            Task.WaitAll(list);
            System.Console.WriteLine("E");

        }

やりたいことなら、PLINQを使って、こんな感じでいいのでは。

Comments

[testset]

PLinqは思い浮かびませんでした。
試してみます。

Answer 4

どこかにぶら下げようかと思いましたが、コード量が多いので、
回答として新たに記載させていただきます。
ご回答いただいた皆様、誠にありがとうございます。

誤った実装

※期待した実装は下に記載しています。
次のコードで正しく動作しているように見えましたが、対応を誤っていました。
理由は、10個ずつしか処理をしていません。

[TestMethod]
public void TestMethod3()
{
    int count = 0;
    var list = Enumerable.Range(1, 30).ToList();
    ParallelForDoNot(10, list, async (item) =>
    {
        Trace.TraceInformation("Inner start {0}", item);
        await Task.Delay(100);
        Trace.TraceInformation("Inner end {0}", item);
        Interlocked.Increment(ref count);
    });
    Assert.AreEqual(list.Count, count);
}
static void ParallelForDoNot<T>(int degree, IList<T> list, Func<T, Task> task)
{
    foreach (var items in list
        .Select((item, i) => new { Item = item, Index = i })
        .GroupBy(g => g.Index / degree)
        )
    {
        Task.WaitAll(items.Select(item => task(item.Item)).ToArray());
    }
}

実行結果

Inner start 1
Inner start 2
Inner start 3
Inner start 4
Inner start 5
Inner start 6
Inner start 7
Inner start 8
Inner start 9
Inner start 10
Inner end 10
Inner end 8
Inner end 7
Inner end 5
Inner end 9
Inner end 3
Inner end 6
Inner end 1
Inner end 4
Inner end 2
Inner start 11
Inner start 12
Inner start 13
Inner start 14
Inner start 15
Inner start 16
Inner start 17
Inner start 18
Inner start 19
Inner start 20
Inner end 20
Inner end 17
Inner end 16
Inner end 15
Inner end 14
Inner end 13
Inner end 12
Inner end 11
Inner end 19
Inner end 18
Inner start 21
Inner start 22
Inner start 23
Inner start 24
Inner start 25
Inner start 26
Inner start 27
Inner start 28
Inner start 29
Inner start 30
Inner end 29
Inner end 28
Inner end 26
Inner end 25
Inner end 30
Inner end 23
Inner end 21
Inner end 24
Inner end 27
Inner end 22

期待した実装

10個分の並列稼働を実現します。
念のためasyncを使っていない場合のためのオーバーロードも用意しています。

[TestMethod]
public void TestMethod1()
{
    int count = 0;
    var list = Enumerable.Range(1, 30).ToList();
    ParallelFor(10, list, async (item) =>
    {
        Trace.TraceInformation("Inner start {0}", item);
        await Task.Delay(100);
        Trace.TraceInformation("Inner end {0}", item);
        Interlocked.Increment(ref count);
    });
    Assert.AreEqual(list.Count, count);
}

[TestMethod]
public void TestMethod2()
{
    int count = 0;
    var list = Enumerable.Range(1, 30).ToList();
    ParallelFor(10, list, (item) =>
    {
        Trace.TraceInformation("Inner start {0}", item);
        Thread.Sleep(100);
        Trace.TraceInformation("Inner end {0}", item);
        Interlocked.Increment(ref count);
    });
    Assert.AreEqual(list.Count, count);
}

/// <summary>
/// 同時実行数制限付き並列動作
/// </summary>
/// <typeparam name="T">リストのデータタイプ</typeparam>
/// <param name="degree">並列数</param>
/// <param name="list">並列処理するリスト</param>
/// <param name="task">並列処理内容</param>
static void ParallelFor<T>(int degree, IEnumerable<T> list, Func<T, Task> task)
{
    using (var sem = new SemaphoreSlim(degree))
    {
        // async記述により、Taskが返される
        // Taskの配列をWaitする
        Task.WaitAll(list.Select(async (item) =>
        {
            await sem.WaitAsync();
            try
            {
                await task(item);
            }
            finally
            {
                sem.Release();
            }
        }).ToArray());
    }
}
/// <summary>
/// 同時実行数制限付き並列動作
/// </summary>
/// <typeparam name="T">リストのデータタイプ</typeparam>
/// <param name="degree">並列数</param>
/// <param name="list">並列処理するリスト</param>
/// <param name="task">並列処理内容</param>
static void ParallelFor<T>(int degree, IEnumerable<T> list, Action<T> task)
{
    Parallel.ForEach(list, new ParallelOptions() { MaxDegreeOfParallelism = degree }, task);
}

実行結果

テスト名:	TestMethod1
テストの完全名:	ParallelTest.UnitTest1.TestMethod1
テスト成果:	成功
テスト継続時間:	0:00:00.3635226

Inner start 1
Inner start 2
Inner start 3
Inner start 4
Inner start 5
Inner start 6
Inner start 7
Inner start 8
Inner start 9
Inner start 10
Inner end 8
Inner end 7
Inner start 12
Inner end 6
Inner start 13
Inner end 5
Inner start 14
Inner end 4
Inner start 15
Inner end 3
Inner start 16
Inner end 2
Inner start 17
Inner end 1
Inner start 18
Inner start 11
Inner end 10
Inner start 19
Inner end 9
Inner start 20
Inner end 11
Inner start 21
Inner end 17
Inner start 22
Inner end 14
Inner start 23
Inner end 15
Inner start 24
Inner end 12
Inner start 25
Inner end 13
Inner start 26
Inner end 18
Inner start 27
Inner end 16
Inner start 28
Inner end 20
Inner start 29
Inner end 19
Inner start 30
Inner end 28
Inner end 27
Inner end 25
Inner end 23
Inner end 22
Inner end 21
Inner end 24
Inner end 26
Inner end 30
Inner end 29

Answer 5

Parallel.ForEach, Parallel.For メソッドのオーバロードの中には引数に ParallelOptions を取るものがあるので、その MaxDegreeOfParallelism プロパティを使って望むこと (上限のある並列数で実行) はできませんか？

(UI への応答性を高めるためのバックグラウンド処理をしたいのか、マルチコアを利用して処理時間の短縮を図りたいのか目的が分かりません。なんとなく見当違いのことのように思いますが、それはとりあえず置いときます)

【追記】

上に書いたことを検証してみました。

listのデータに対して順序にかかわらず上限のある並列数(1～)で実行したい。システム全体でのスレッドの上限があるため、極力節約したい。

その目的のために await を使う意味が分かりません。ParallelOptions.MaxDegreeOfParallelism を設定して、普通に以下のコードのようにすれば上記の目的は果たせるはずですが？

var list = Enumerable.Range(1, 30).ToList();
ParallelOptions option = new ParallelOptions();
option.MaxDegreeOfParallelism = 2;
Parallel.ForEach(list, option, i => 
{
    Console.WriteLine($"Processing {i} on thread {Thread.CurrentThread.ManagedThreadId}");
});

上記のコード（MaxDegreeOfParallelism は 2 に制限）の実行結果は以下の通りとなります。質問者さんのコードでは 10 に制限しているように見えますが、スレッドの数はそのようになりませんか？

Processing 1 on thread 1
Processing 16 on thread 3
Processing 17 on thread 3
Processing 18 on thread 3
Processing 19 on thread 3
Processing 20 on thread 3
Processing 21 on thread 3
Processing 22 on thread 3
Processing 23 on thread 3
Processing 24 on thread 3
Processing 25 on thread 3
Processing 26 on thread 3
Processing 27 on thread 3
Processing 28 on thread 3
Processing 29 on thread 3
Processing 30 on thread 3
Processing 4 on thread 3
Processing 5 on thread 3
Processing 6 on thread 3
Processing 7 on thread 3
Processing 8 on thread 3
Processing 9 on thread 3
Processing 10 on thread 3
Processing 11 on thread 3
Processing 12 on thread 3
Processing 13 on thread 3
Processing 14 on thread 3
Processing 2 on thread 1
Processing 3 on thread 1
Processing 15 on thread 3

ちなみに、ParallelOptions を設定しない場合は以下のようになります。

Processing 1 on thread 1
Processing 2 on thread 1
Processing 16 on thread 7
Processing 7 on thread 4
Processing 8 on thread 4
Processing 9 on thread 4
Processing 13 on thread 6
Processing 14 on thread 6
Processing 15 on thread 6
Processing 19 on thread 9
Processing 4 on thread 3
Processing 10 on thread 5
Processing 26 on thread 5
Processing 27 on thread 5
Processing 28 on thread 5
Processing 29 on thread 5
Processing 30 on thread 5
Processing 22 on thread 8
Processing 11 on thread 4
Processing 20 on thread 6
Processing 25 on thread 10
Processing 3 on thread 1
Processing 17 on thread 7
Processing 18 on thread 7
Processing 21 on thread 6
Processing 12 on thread 4
Processing 5 on thread 3
Processing 6 on thread 3
Processing 23 on thread 9
Processing 24 on thread 9

Comments

[YAmaGNZ]

使い方が希望している動作と一致しているかは分かりませんが、提示されているコードでMaxDegreeOfParallelismプロパティを使用していますよ。

[testset]

ご確認のほどありがとうございます。
私が記載したコードでは、Parallel.ForEachに渡している処理がasyncであるため、TaskのStartだけを行い、次の処理を行ってしまっているようです。そのため、早々にParallel.ForEachが終了し、親タスクが破棄される(実は裏で動いているかもしれない)ため、質問に記載したログのようになっている認識です。

[退会済みユーザー]

> Parallel.ForEachに渡している処理がasyncであるため

意味が分かりません。何故そういうことをする必要があるのか目的・理由を具体的に説明いただけませんか。どう考えても「システム全体でのスレッドの上限があるため、極力節約したい」ということとは違うと思うのですが。tamoto さんが言われるように余計なことをせず OS に任せておくか、最大やってもMaxDegreeOfParallelism の設定だけでいいのではないですか。

[testset]

全く持って、おっしゃる通りです。OSに任せるのがよいと、今では思っています。
なぜそうしたのか、ということは質問本文に記載させていただきました。

[退会済みユーザー]

なぜそうしたのかの理由は質問文に追記した以下と理解しています。

> ループ処理内でネットワーク通信を挟んでいるためです：ネットワーク通信先では同時接続数の上限もあるため、同時アクセス数の制限を設けたい。(Server busyとなってしまう)

その理由で Parallell.ForEach とか await とかの話が出てくるのが解せません。

相手先サーバーが Server Busy とならないようにするためにクライアント側でできることは、質問者さんが書かれているように「同時アクセス数の制限」ぐらいしかなさそうですが、とすると、そもそも Parallel.ForEach を使うことが間違いだと思います。

ASP.NET のようにサーバー側の話であれば、非同期で実行すると、スレッドプールから取得したスレッドをプールに戻して、非同期実行が完了すると再びスレッドプールからスレッドを取得して処理を完了し応答を返すということが可能（すなわち、スレッドプールのスレッドが枯渇して Server Busy とならないように対処することが可能）なのですが。

サーバー側の話とごっちゃになっているような気がします。違いますか？

[退会済みユーザー]

同時アクセス数を制限しつつ並列に複数データを送っていくなら(暗黙の了解でサーバーに対するクライアントの同時アクセス数が決まっているなら)一応スレッド数上限を設けたParallelでも分からなくないですが
awaitが出てきた理由と結論がTask全部起動なのは未だに謎。
本人が解決したと思ってるなら良いけれど、意図するところがよく分からない話でした。

[testset]

ASP.NETを使えればよいのですが、該当処理はレガシーシステムの間に入り込む通信処理で、上位からは複数の要求があり、下位へは厳しい同時アクセス制限があります。さらに、下位の通信環境は当初酷いもの(ダイアルアップなど)だと聞いていたため、並列化を考えないといけないため、このような質問が出てきました。
Taskを全部起動するというのは説明が足りていませんでしたが、SemaphoreSlimで制限をかけたうえで起動するという方法で解決させる見込みでした。

Answer 6

所望の動作が厳密に分からないですが自前のParallelForからParallel.Foreachを呼び出したりせずasyncもawaitもTask.Waitも何もなしで並列化自体はできるかと。
(あれこれやらずに普通にlistにParallel.ForなりParallel.Foreachなりを使う)
その場合全てのinner endのあとにendが来ます。

Parallel.ForEachのdelegate上でawaitを使いたいのですが、
すべての処理が実行されずにParallel.ForEachが抜けてしまいます。

そもそもawait自体Task.Waitみたいに同期とって返す物ではないのでそういうものかと。

Comments

[testset]

確かにおっしゃる通りです。
asyncを行わなければ、期待する動作をすることは認識しております。
早々にご回答ありがとうございます。