Как реализовать параллельный цикл в Delphi?

Question

Как реализовать параллельный цикл в Delphi (Delphi 200X, Delphi XE)? Как это сделать наилучшим образом? Есть ли универсальное решение?

С примерами, пожалуйста.

Answer 1

Возможно, лучшим решением на данный момент является конструкция Parallel For Loop в OmniThreadLibrary. Вы передаете ему коллекцию или пару целых чисел, представляющих нижнюю и верхнюю границы, и анонимный метод, представляющий тело цикла, и использует пул потоков для параллельного цикла цикла for.

Обратите внимание, что это будет работать только в том случае, если метод тела петли может стоять сам по себе. Если он изменяет любые внешние переменные или полагается на значение вычисления, сделанного ранее в цикле, то он не может быть распараллелен.

Введение в параллельную библиотеку OmniThreadLibrary можно найти here. Например, простой для петли Перебора чисел выглядят следующим образом:

Parallel.ForEach(1, testSize).Execute(
    procedure (const elem: integer) 
    begin 
    // do something with 'elem' 
    end); 

Answer 2

Зависит от того, что вы подразумеваете под параллельным циклом и приложением/реализацией.

Посмотрите на TThread и TMultiReadExclusiveWriteSynchronizer.

Answer 3

Если вам нужно только ParallelFor вы можете использовать этот код:

interface 

uses 
    Classes, SysUtils; 

type 
    TParallelProc = reference to procedure(i: Integer; ThreadID: Integer); 

    TParallel = class(TThread) 
    private 
    FProc: TParallelProc; 
    FThreadID: Integer; //current thread ID 
    protected 
    procedure Execute; override; 
    function GetNextValue: Integer; 
    public 
    constructor Create; 
    destructor Destroy; override; 

    property Proc: TParallelProc 
     read FProc write FProc; 
    class var 
     CurrPos: Integer; //current loop index 
     MaxPos: Integer; //max loops index 
     cs: TCriticalSection; 
     ThCount: Integer; //thread counter - how much threads have finished execution 
    end; 


{** ParallelFor Loop - all iterations will be performed in chosen threads 
@param nMin - Loop min value (first iteration) 
@param nMax - Loop max value (last iteration) 
@param nThreads - how much threads to use 
@param aProc - anonymous procedure which will be performed in loop thread 
} 
procedure ParallelFor(nMin, nMax, nThreads: Integer; aProc: TParallelProc); overload; 
{** ParallelFor Loop - all iterations will be performed in max cpu cores 
@param nMin - Loop min value (first iteration) 
@param nMax - Loop max value (last iteration) 
@param aProc - anonymous procedure which will be performed in loop thread 
} 
procedure ParallelFor(nMin, nMax: Integer; aProc: TParallelProc); overload; 

implementation 

uses 
    {$IFDEF MSWINDOWS} 
    Windows, 
    {$ENDIF} 
    SyncObjs; 

procedure ParallelFor(nMin, nMax, nThreads: Integer; aProc: TParallelProc); 
var 
    threads: array of TParallel; 
    I: Integer; 
begin 
    if nMin > nMax then 
    Exit; 
    // initialize TParallel class data 
    TParallel.CurrPos := nMin; 
    TParallel.MaxPos := nMax; 
    TParallel.cs := TCriticalSection.Create; 
    TParallel.ThCount := 0; 

    // create the threads 
    SetLength (threads, nThreads); 
    for I := 0 to nThreads - 1 do 
    begin 
    threads[I] := TParallel.Create; // suspended 
    threads[I].FThreadID := I; 
    threads[I].Proc := aProc; 
    threads[I].Start; 
    end; 

    for I := 0 to nThreads - 1 do 
    begin 
    threads[I].WaitFor; 
    end; 

    for I := 0 to nThreads - 1 do 
    begin 
    threads[I].Free; 
    end; 

    TParallel.cs.Free; 
end; 

procedure ParallelFor(nMin, nMax: Integer; aProc: TParallelProc); 
begin 
    ParallelFor(nMin, nMax, CPUCount, aProc); 
end; 

{ TParallel } 

constructor TParallel.Create; 
begin 
    inherited Create(True); // suspended 
    InterlockedIncrement(ThCount); 
    FreeOnTerminate := False; 
    FThreadID := 0; 
end; 

destructor TParallel.Destroy; 
begin 
    InterlockedDecrement(ThCount); 
    inherited; 
end; 

procedure TParallel.Execute; 
var 
    nCurrent: Integer; 
begin 
    nCurrent := GetNextValue; 
    while nCurrent <= MaxPos do 
    begin 
    Proc(nCurrent, FThreadID); 
    nCurrent := GetNextValue; 
    end; 
end; 

function TParallel.GetNextValue: Integer; 
begin 
    cs.Acquire; 
    try 
    Result := CurrPos; 
    Inc(CurrPos); 
    finally 
    cs.Release; 
    end; 
end; 

Однако, если вам нужно больше потокового «материал» Вам следует рассмотреть возможность использования сторонних библиотек.