LoginSignup
12
2

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

DelphiAdvent Calendar 2021

Day 10
@ht_deko(Hideaki Tominaga)

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

Last updated at Posted at 2021-12-09

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

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

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

4.1. スレッドプール

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

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

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

See also:

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:

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:

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:

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:

4.6. TParallelArray.Sort

TParallelArray.Sort() はスレッドを使ってパラメータに指定された配列を高速にソートするパラメータ化メソッドです 2。

ParallelArray1.dpr
program ParallelArray1;
{$APPTYPE CONSOLE}
uses
  System.SysUtils, System.Generics.Collections, System.Threading, System.Diagnostics;
  
var
  IntArr1, IntArr2: array of Integer;
  T: TStopWatch;
begin
  SetLength(IntArr1, 10000000);
  SetLength(IntArr2, 10000000);
  for var i:=Low(IntArr1) to High(IntArr1) do
  begin
    IntArr1[i] := Random(MAXINT);
    IntArr2[i] := IntArr1[i];
  end;  

  // TArray
  T := TStopWatch.StartNew;
  TArray.Sort<Integer>(IntArr1);
  Writeln(T.Elapsed.ToString);

  // TParallelArray
  T := TStopWatch.StartNew;
  TParallelArray.Sort<Integer>(IntArr2);
  Writeln(T.Elapsed.ToString);
end.

コンペアラ (比較者) をパラメータに指定可能なオーバーロードされたメソッドもあります。

ParallelArray2.dpr
program ParallelArray2;
{$APPTYPE CONSOLE}
uses
  System.SysUtils, System.Generics.Collections, System.Generics.Defaults, System.Threading;
  
var
  IntArr: array of Integer;
  Comparer: IComparer<Integer>;
begin
  SetLength(IntArr, 10);
  for var i:=Low(IntArr) to High(IntArr) do
    IntArr[i] := Random(100);

  // Sort (昇順)
  TParallelArray.Sort<Integer>(IntArr);
  for var i:=Low(IntArr) to High(IntArr) do
    Writeln('#1: ', IntArr[i]);
  
  // Sort with Comparer (降順)
  Comparer := TDelegatedComparer<Integer>.Create(
    function(const Left, Right: Integer): Integer
    begin
      result := Right - Left; 
    end);
  TParallelArray.Sort<Integer>(IntArr, Comparer);
  for var i:=Low(IntArr) to High(IntArr) do
    Writeln('#2: ', IntArr[i]);  
end.

See also:

4.7. TParallelArray.For

TParallelArray.For() は配列データをスライスしてスレッド実行するパラメータ化メソッドです 2。

最もシンプルなものは次のようなパラメータを持ちます。実行する手続きは TParallelArrayForProc 型です。

TParallelArray.For<T>(T 型の配列, 実行する手続きへの参照型);

例えば整数型の配列を伴って処理を行うには次のようになります。

uses
  ..., System.SyncObjs, System.Threading;

  ...

procedure TForm1.Button1Click(Sender: TObject);
var
  IntArr: array of Integer;
begin
  SetLength(IntArr, 1000000);
  for var i:=Low(IntArr) to High(IntArr) do
    IntArr[i] := Random(100);

  // パラレル実行
  var Count := 0;
  var Value := 0;
  TParallelArray.For<Integer>(IntArr,
    procedure (const AValues: array of Integer; AFrom, ATo: NativeInt)
    begin
      TInterlocked.Add(Count, 1);  // この手続きが呼ばれた回数
      var v := 0;
      for var i := AFrom to ATo do // AFrom / ATo は毎回異なる
        Inc(v, AValues[i]);
      TInterlocked.Add(Value, v);
    end);
  Memo1.Lines.Add('Value: ' + Value.ToString);
  Memo1.Lines.Add('Count: ' + Count.ToString);
end;

TParallelArray.For() もメインスレッドで実行されると TThread.Synchronize() をブロックするので、処理によっては TTask を併用する必要があります。

uses
  ..., System.SyncObjs, System.Threading;

  ...

procedure TForm1.Button1Click(Sender: TObject);
var
  IntArr: array of Integer;
begin
  SetLength(IntArr, 1000000);
  for var i:=Low(IntArr) to High(IntArr) do
    IntArr[i] := Random(100);

  // パラレル実行
  var Count := 0;
  var Value := 0;
  TTask.Run(procedure
    begin
      TParallelArray.For<Integer>(IntArr,
        procedure (const AValues: array of Integer; AFrom, ATo: NativeInt)
        begin
          TInterlocked.Add(Count, 1);  // この手続きが呼ばれた回数
          TThread.Synchronize(nil,
            procedure
            begin
              Memo1.Lines.Add(AFrom.ToString + ':' + ATo.ToString);
            end);
          var v := 0;
          for var i := AFrom to ATo do // AFrom / ATo は毎回異なる
            Inc(v, AValues[i]);
          TInterlocked.Add(Value, v);
        end);
      TThread.Synchronize(nil,
        procedure
        begin
          Memo1.Lines.Add('Value: ' + Value.ToString);
          Memo1.Lines.Add('Count: ' + Count.ToString);
        end);
    end);
end;

See also:

参考

索引

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

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

  2. TParallelArray は Delphi 12.2 Athens 以降で利用可能です。 ↩ ↩2

12
2
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
12
2

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?