bucket-sort logo bucket-sort

プログラミングとインフラエンジニアリングの覚え書き

[C#] Task Parallel Library と PLINQ を理解する

C# , .NET bucket-sort

前回は、スレッド同期と ThreadPool を見ました。

今回は、ThreadPool の上に構築される高レベルな並列処理 API を扱います。
中心になるのは Task Parallel Library、つまり TPL です。

TPL を使うと、Thread を直接作らなくても、複数の処理を並列に実行できます。

System.Threading.Tasks 名前空間

TPL の主要な型は、System.Threading.Tasks 名前空間にあります。

using System.Threading.Tasks;

代表的な型は次の通りです。

Task

非同期または並列に実行される処理を表します。戻り値がない処理に使います。

Task

戻り値を持つ処理を表します。

Parallel

データ並列やタスク並列を簡単に書くための静的クラスです。

ParallelOptions

並列度やキャンセルなど、Parallel 実行時のオプションを指定します。

CancellationToken

キャンセル要求を伝えるための値です。

Parallel クラスの役割

Parallel クラスは、ループ処理を複数スレッドで実行するための API を提供します。

代表的なのは次の 3 つです。

  • Parallel.For
  • Parallel.ForEach
  • Parallel.Invoke

大量の独立した処理を並列に実行したいときに便利です。

Parallel.For で範囲を並列処理する

using System;
using System.Threading;
using System.Threading.Tasks;

Parallel.For(0, 10, i =>
{
    Console.WriteLine($"i={i}, Thread={Thread.CurrentThread.ManagedThreadId}");
});

for 文のように見えますが、実行順序は保証されません。
並列処理では、0, 1, 2, ... の順に処理されるとは限らない点に注意します。

Parallel.ForEach でコレクションを並列処理する

using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;

var files = new List<string>
{
    "a.txt",
    "b.txt",
    "c.txt",
    "d.txt"
};

Parallel.ForEach(files, file =>
{
    Console.WriteLine($"{file}: Thread={Thread.CurrentThread.ManagedThreadId}");
});

Parallel.ForEach は、コレクションの各要素を並列に処理します。
画像変換、ファイル解析、数値計算など、各要素が独立している処理に向いています。

最大並列度を制御する

並列処理は、並列度を上げれば必ず速くなるわけではありません。
CPU、I/O、DB、外部 API など、どこがボトルネックかによって最適な並列度は変わります。

using System;
using System.Threading;
using System.Threading.Tasks;

var options = new ParallelOptions
{
    MaxDegreeOfParallelism = 2
};

Parallel.For(0, 10, options, i =>
{
    Console.WriteLine($"i={i}, Thread={Thread.CurrentThread.ManagedThreadId}");
    Thread.Sleep(500);
});

MaxDegreeOfParallelism を使うと、同時に実行する処理数の上限を指定できます。

UI 要素へのアクセスに注意する

WPF、Windows Forms、MAUI などの UI アプリでは、UI 要素は基本的に UI スレッドから操作する必要があります。

次のような考え方が重要です。

  • 重い処理はバックグラウンドで実行する
  • UI の更新は UI スレッドに戻して行う
  • バックグラウンドスレッドから直接 UI コントロールを変更しない

WPF では Dispatcher、Windows Forms では Invoke / BeginInvoke を使って UI スレッドへ処理を戻します。

// WPF のイメージ
await Task.Run(() =>
{
    // 重い処理
});

// await 後は通常 UI スレッドに戻るため、UI 更新できる
StatusTextBlock.Text = "完了";

コンソールアプリではこの問題は目立ちませんが、UI アプリでは非常に重要です。

Task クラスの基本

Task は、ある処理の実行状態や完了を表すオブジェクトです。

using System;
using System.Threading;
using System.Threading.Tasks;

Task task = Task.Run(() =>
{
    Console.WriteLine($"Task started: {Thread.CurrentThread.ManagedThreadId}");
    Thread.Sleep(1000);
    Console.WriteLine("Task finished");
});

task.Wait();

Task.Run は、処理を ThreadPool 上で実行します。
Wait() は完了まで待機しますが、現在の .NET では後述する await を使う方が自然です。

Task で結果を返す

戻り値がある処理は Task<TResult> で表せます。

using System;
using System.Threading.Tasks;

Task<int> task = Task.Run(() =>
{
    int total = 0;

    for (int i = 1; i <= 100; i++)
    {
        total += i;
    }

    return total;
});

Console.WriteLine(task.Result);

Result は結果を取得しますが、完了していない場合はブロックします。
UI アプリや ASP.NET ではデッドロックや応答性低下の原因になることがあるため、可能なら await を使います。

キャンセル要求を処理する

長時間実行される処理には、キャンセルできる設計が必要です。
.NET では CancellationTokenSource と CancellationToken を使います。

using System;
using System.Threading;
using System.Threading.Tasks;

using var cts = new CancellationTokenSource();
CancellationToken token = cts.Token;

Task worker = Task.Run(() =>
{
    for (int i = 0; i < 100; i++)
    {
        token.ThrowIfCancellationRequested();

        Console.WriteLine($"Working... {i}");
        Thread.Sleep(200);
    }
}, token);

Thread.Sleep(1000);
cts.Cancel();

try
{
    worker.Wait();
}
catch (AggregateException ex)
{
    foreach (Exception inner in ex.InnerExceptions)
    {
        Console.WriteLine(inner.GetType().Name);
    }
}

キャンセルは、外部から強制的にスレッドを止める仕組みではありません。
処理側が CancellationToken を確認し、自分で中断する協調的な仕組みです。

Parallel でキャンセルを扱う

ParallelOptions に CancellationToken を指定できます。

using System;
using System.Threading;
using System.Threading.Tasks;

using var cts = new CancellationTokenSource();

var options = new ParallelOptions
{
    CancellationToken = cts.Token,
    MaxDegreeOfParallelism = 4
};

Task.Run(() =>
{
    Thread.Sleep(1000);
    cts.Cancel();
});

try
{
    Parallel.For(0, 100, options, i =>
    {
        options.CancellationToken.ThrowIfCancellationRequested();
        Console.WriteLine(i);
        Thread.Sleep(200);
    });
}
catch (OperationCanceledException)
{
    Console.WriteLine("キャンセルされました。");
}

並列ループでも、キャンセルは協調的に扱います。

Parallel.Invoke で複数処理を並列実行する

Parallel.Invoke は、複数のメソッドを並列に実行したいときに使えます。

using System;
using System.Threading;
using System.Threading.Tasks;

Parallel.Invoke(
    () =>
    {
        Thread.Sleep(500);
        Console.WriteLine("処理 A");
    },
    () =>
    {
        Thread.Sleep(500);
        Console.WriteLine("処理 B");
    },
    () =>
    {
        Thread.Sleep(500);
        Console.WriteLine("処理 C");
    });

複数の独立した処理をまとめて実行する用途に向いています。

PLINQ とは

PLINQ（Parallel LINQ）は、LINQ クエリを並列実行する仕組みです。

通常の LINQ は、基本的に 1 つの流れで処理されます。
PLINQ では、AsParallel() を付けることで、クエリを並列実行できます。

using System;
using System.Linq;
using System.Threading;

var numbers = Enumerable.Range(1, 20);

var results = numbers
    .AsParallel()
    .Where(n =>
    {
        Console.WriteLine($"n={n}, Thread={Thread.CurrentThread.ManagedThreadId}");
        return n % 2 == 0;
    })
    .Select(n => n * n)
    .ToList();

Console.WriteLine(string.Join(", ", results));

PLINQ は、CPU 負荷の高い変換やフィルタリングに向いています。
一方で、軽すぎる処理では並列化のコストの方が大きくなることがあります。

PLINQ に明示的に参加する

PLINQ では、AsParallel() を呼び出すことで並列化に参加します。

var query = Enumerable.Range(1, 1_000_000)
    .AsParallel()
    .Where(n => n % 3 == 0)
    .Select(n => n * 2);

Console.WriteLine(query.Count());

LINQ クエリがすべて PLINQ に向くわけではありません。
順序が重要な処理、共有状態を更新する処理、I/O 待ちが中心の処理では注意が必要です。

PLINQ をキャンセルする

PLINQ でも CancellationToken を使ってキャンセルできます。

using System;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;

using var cts = new CancellationTokenSource();

Task.Run(() =>
{
    Thread.Sleep(1000);
    cts.Cancel();
});

try
{
    var result = Enumerable.Range(1, 100_000_000)
        .AsParallel()
        .WithCancellation(cts.Token)
        .Where(n => n % 7 == 0)
        .Count();

    Console.WriteLine(result);
}
catch (OperationCanceledException)
{
    Console.WriteLine("PLINQ クエリがキャンセルされました。");
}

PLINQ のキャンセルも、CancellationToken を通じて協調的に行われます。

まとめ

TPL は、スレッドを直接扱わずに並列処理を書くための強力な仕組みです。

Parallel.For や Parallel.ForEach は、独立した大量の処理を並列化するのに向いています。
Task は、処理の実行と完了を表す基本単位です。
PLINQ は、LINQ クエリを並列化したいときに便利です。

次回は、Task をさらに自然に扱うための async / await を見ていきます。

Related Posts