<4> 並列プログラミングライブラリ (PPL) (Delphi コンカレントプログラミング)

4. 並列プログラミングライブラリ (PPL)

クロスプラットフォームで使える並列プログラミングライブラリ (Parallel Programming Library / PPL) は System.Threading を uses に加えることで利用可能になります ¹。

この並列プログラミングライブラリはマルチスレッドを抽象化しており、TThread を使う事なくマルチスレッドを利用可能です。

4.1. スレッドプール

例えば 1 万ファイルをスレッドで処理するのに、1 万スレッド生成するのは効率の面で問題があります。生成されるスレッドの上限を決めておき、同時実行されるスレッドの管理を行う機構がスレッドプールです。いわゆる Worker Thread です。

Delphi の PPL では、CPU の負荷に応じて自動的にワーカースレッドが作成されます。 ワーカースレッドは System.Threading.TThreadPool クラスが管理しています。

TThreadPool クラスをそのまま使う事はまずないかと思います。

See also:

• System.Threading.TThreadPool (DocWiki)
• 並列プログラミングでのスレッドプーリングの概要 (Support Wiki)
• スレッドプール (docs.microsoft.com)
• マネージドスレッドプール (docs.microsoft.com)
• Thread pool (Wikipedia: en)

4.2. TTask

コードブロックを簡単にスレッドで実行することができます。

uses
  ..., System.Threading;

...

  TTask.Run(procedure
    begin
      // 何かの処理
      TThread.Synchronize(nil,
        procedure
        begin
          // 何かの処理
        end);
    end);

コンストラクタと Start() メソッドを使って TThread.CreateAnonymousThread のように使う事もできます。

  TTask.Create(procedure
    begin
      // 何かの処理
      TThread.Synchronize(nil,
        procedure
        begin
          // 何かの処理
        end);
    end).Start;

インターフェイス型変数に取って処理する事もできます。次のコードの Task 変数はインターフェイス型変数なので、明示的な破棄は不要です。

var
  Task: ITask; 

...

  Task := TTask.Create(procedure
    begin
      // 何かの処理
      TThread.Synchronize(nil,
        procedure
        begin
          // 何かの処理
        end);
    end);
  Task.Start;

WaitForAll() を使うと、ITask 配列で処理されるスレッドの終了待ちを行う事ができます。

var
  TaskArr: array of ITask;
begin
  Setlength (TaskArr ,3);
  TaskArr[0] := TTask.Run(procedure begin {何かの処理} end);
  TaskArr[1] := TTask.Run(procedure begin {何かの処理} end);
  TaskArr[2] := TTask.Run(procedure begin {何かの処理} end);
  TTask.WaitForAll(TaskArr);
end;

もっとシンプルに書くこともできます。

  TTask.WaitForAll([
    TTask.Run(procedure begin {何かの処理} end),
    TTask.Run(procedure begin {何かの処理} end),
    TTask.Run(procedure begin {何かの処理} end)
  ]);

タスクを停止させるには、Cancel() メソッドを使います。

procedure TForm1.Button1Click(Sender: TObject);
begin
  var Task := TTask.Run(
    procedure
    begin
      TThread.Sleep(2000);
      TThread.Synchronize(nil,
        procedure
        begin
          Button1.Caption := 'Done.'; // ボタンを押して 2 秒後にボタンのキャプションが...
        end);
    end);
  Task.Cancel; // 書き変わらない (w
end;

See also:

• System.Threading.TTask (DocWiki)
• System.Threading.TTask.Run (DocWiki)
• System.Threading.TTask.Create (DocWiki)
• System.Threading.ITask.Start (DocWiki)
• System.Threading.ITask.Cancel (DocWiki)
• System.Threading.TTask.WaitForAll (DocWiki)
• チュートリアル：並列プログラミング ライブラリのタスクを使用する (DocWiki)
• Delphi でコントロール配列 [小ネタ] (Qiita)

4.3. TTask.Future

いわゆる Future です。意訳すると "引換券" です。焼き芋屋さんに「焼いといて！」とお願いして、引換券をもらった感じです。すぐに引き換えに行っても「まだ焼けてないよ！」と待たされることになりますが、ちょっと経ってから行くとすぐに焼き芋を貰えます。

VCL フォームアプリケーションを新規作成し、フォームにボタンを置いて、次のようなコードを書きます。

unit Unit1;

interface

uses
  Winapi.Windows, Winapi.Messages, System.SysUtils, System.Variants, System.Classes, Vcl.Graphics,
  Vcl.Controls, Vcl.Forms, Vcl.Dialogs, Vcl.StdCtrls, System.Threading;

type
  TForm1 = class(TForm)
    Button1: TButton;
    procedure Button1Click(Sender: TObject);
  private
    { Private 宣言 }
  public
    { Public 宣言 }
  end;

var
  Form1: TForm1;
  Future: IFuture<Integer>;

implementation

{$R *.dfm}

procedure TForm1.Button1Click(Sender: TObject);
begin
  ShowMessage(Future.Value.ToString)
end;

initialization
  Future := TTask.Future<Integer>(
    function: Integer
    begin
      TThread.Sleep(5000);
      Result := 100;
  end);
end.

アプリケーションを起動してすぐにボタンを押すと結果が表示されるまでちょっと待たされますが、アプリケーションを起動して 5 秒以上待ってからボタンを押すと結果がすぐに表示されます。

コード中の Future はインターフェイス型変数なので、明示的な破棄は不要です。

『Delphi クイックリファレンス』には TThread を使った TFuture の実装が載っています。
Ray Lischner (著), 光田 秀, 竹田 知生 (訳) (2001).「Delphi クイックリファレンス」オライリージャパン. pp.174-184.

See also:

• System.Threading.TTask.Future (DocWiki)
• System.Threading.IFuture (DocWiki)
• チュートリアル：並列プログラミング ライブラリのフューチャを使用する (DocWiki)
• Future / Promise / Delay パターン (Wikipedia)

4.4. TParallel.For

ものすごくざっくり言えば for to do 文の並列実行版です。一つのコードブロックを繰り返し並列実行します。

オーバーロードされたメソッドが多いので混乱しますが、最もシンプルなものは次のようなパラメータを持ちます。

TParallel.For(開始値, 終了値, 実行する手続きへの参照型);

「実行する手続き」のうち最もシンプルなのは TProc<Integer> です。i: Integer のような Integer 型のパラメータを一つ持つ手続きです。パラメータ名は i でなくとも任意の識別子で構いません。

uses
  ..., System.Threading;

  ...

  // シーケンシャル実行
  for var i:=1 to 10 do
  begin
    // ここのコードが繰り返し (直列) 実行される。
  end;

  // パラレル実行
  TTask.Run(procedure
    begin
      TParallel.For(1, 10, 
        procedure (i: Integer)
        begin
          // ここのコードがスレッドで並列実行される。
        end);
    end;

TParallel.For() を TTask 内で実行している理由の一つは TParallel.For() はメインスレッドで実行されると TThread.Synchronize() をブロックするからです。

// 本当に固まるので実行注意！
procedure TForm1.Button1Click(Sender: TObject);
begin
  Memo1.Clear;
  Memo1.Update;
  TParallel.For(1, 10,
    procedure (i: Integer)
    begin
      TThread.Sleep(5000);
      TThread.Synchronize(nil,
        procedure
        begin
          Memo1.Lines.Add(i.ToString);
        end);
    end);
end;

TThread.Queue() だとブロックされないのですが、すべてのスレッドを実行し終わるまで (終了ではありません) はフォームを移動できません。

procedure TForm1.Button1Click(Sender: TObject);
begin
  Memo1.Clear;
  Memo1.Update;
  TParallel.For(1, 10,
    procedure (i: Integer)
    begin
      TThread.Sleep(5000);
      TThread.Queue(nil,
        procedure
        begin
          Memo1.Lines.Add(i.ToString);
        end);
    end);
end;

...なので、TParallel.For() は基本的に TTask または、

procedure TForm1.Button1Click(Sender: TObject);
begin
  Memo1.Clear;
  Memo1.Update;
  TTask.Run(procedure
    begin
      TParallel.For(1, 10,
        procedure (i: Integer)
        begin
          TThread.Sleep(5000);
          TThread.Synchronize(nil,
            procedure
            begin
              Memo1.Lines.Add(i.ToString);
            end);
        end)
    end);
end;

TThread と併用する必要があります。

procedure TForm1.Button1Click(Sender: TObject);
begin
  Memo1.Clear;
  Memo1.Update;
  TThread.CreateAnonymousThread(procedure
    begin
      TParallel.For(1, 10,
        procedure (i: Integer)
        begin
          TThread.Sleep(5000);
          TThread.Synchronize(nil,
            procedure
            begin
              Memo1.Lines.Add(i.ToString);
            end);
        end)
    end).Start;
end;

10.1 Berlin と 10.2 Tokyo には TParallel.For() が最初の 2 ループを同じスレッドで実行してしまうというバグがあります。

See also:

• System.Threading.TParallel.For (DocWiki)
• 並列プログラミング ライブラリの TParallel.For の使用 (DocWiki)
• チュートリアル：並列プログラミング ライブラリの for ループを使用する (DocWiki)
• RSP-20695: Parallel For or Join with n iterations runs in n-1 threads (Quality Portal)

4.5. TParallel.Join

TParallel.Join は、複数のコードブロックを並列実行します。

最もシンプルなものは次のようなパラメータを持ちます。

TParallel.Join(実行する手続きへの参照型の配列);

TTask.WaitForAll() もそうでしたが、TParallel.Join() を無名メソッドで書く事はあまりないかもしれません。

procedure TForm1.Button1Click(Sender: TObject);
begin
  Memo1.Clear;
  Memo1.Update;
  TParallel.Join([
    procedure
    begin
      TThread.Synchronize(nil,
        procedure
        begin
          Memo1.Lines.Add('A');
        end);
    end,
    procedure
    begin
      TThread.Synchronize(nil,
        procedure
        begin
          Memo1.Lines.Add('B');
        end);
    end,
    procedure
    begin
      TThread.Synchronize(nil,
        procedure
        begin
          Memo1.Lines.Add('C');
        end);
    end
  ]);
end;

好みの問題でしょうが、個人的には手続きは独立させておいた方が視認性が良いと感じます。

procedure TForm1.A;
begin
  TThread.Synchronize(nil,
    procedure
    begin
      Memo1.Lines.Add('A');
    end);
end;

procedure TForm1.B;
begin
  TThread.Synchronize(nil,
    procedure
    begin
      Memo1.Lines.Add('B');
    end);
end;

procedure TForm1.C;
begin
  TThread.Synchronize(nil,
    procedure
    begin
      Memo1.Lines.Add('C');
    end);
end;

procedure TForm1.Button1Click(Sender: TObject);
begin
  Memo1.Clear;
  Memo1.Update;
  TParallel.Join([A, B, C]);
end;

10.1 Berlin と 10.2 Tokyo には TParallel.Join() が最初の 2 つのタスクを同じスレッドで実行してしまうというバグがあります。

See also:

• System.Threading.TParallel.Join (DocWiki)
• RSP-19557: TParallel.Join does not create enough threads (Quality Portal)

参考

• 並列プログラミング ライブラリの使用 (DocWiki)
• RAD Studio XE7 - the new Parallel Programming Library (Youtube)
• Parallel Programming Library (Delphi) (Youtube)
• RAD Studio XE7の新機能 - パラレルライブラリ・動的配列 (Youtube)
• コードサンプル: ライフゲーム (VCL) (docwiki)
• コードサンプル: ライフゲーム (FireMonkey) (docwiki)
• タスク並列ライブラリ (TPL) (docs.microsoft.com)
• CodeRage 9: Parallel Programming Library: Create Responsive Object Pascal Apps (Youtube)
• Delphi - Multi threading (Chau Chee Yang Technical Blog)
• Delphi の TTask でも ContinueWith がしたい！ (Qiita: @sagiri )

索引

[ ← 3. スレッドローカル変数 (スレッドローカルストレージ) ] [ ↑ 目次へ ] [ → 5. サードパーティ製ライブラリ ]

1. System.Threading は Delphi XE7 以降で利用可能です。 ↩