Parallel Map

At my latest parallel programming presentation a participant suggested that I should extend the OmniThreadLibrary with a parallel mapping abstraction. Dear sir, here is a gift for you.

var
  numbers: TArray<integer>;
  odds   : TArray<string>;
begin
  //initialize the `numbers` array (not shown)
  odds := Parallel.Map<integer,string>(numbers,
    function (const source: integer; var dest: string): boolean
    begin
      Result := Odd(source);
      if Result then
        dest := IntTostr(source);
    end);
  //do something with the `odds` array (not shown)
end;

Since today, OTL SVN contains an implementation of a parallel mapping algorithm, accessible in two different ways. Parallel.Map<T1,T2> returns the IOmniParallelMapper<T1,T2> interface while the longer version (shown above) contains a synchronous, easy-to-use wrapper.

IOmniParallelMapper<T1,T2> = interface
  function  Execute(mapper: TMapProc<T1,T2>): 
    IOmniParallelMapper<T1,T2>;
  function  NoWait: IOmniParallelMapper<T1,T2>;
  function  NumTasks(numTasks: integer): 
    IOmniParallelMapper<T1,T2>;
  function  OnStop(stopCode: TProc): 
    IOmniParallelMapper<T1,T2>; overload;
  function  OnStop(stopCode: TOmniTaskStopDelegate): 
    IOmniParallelMapper<T1,T2>; overload;
  function  Result: TArray<T2>;
  function  Source(const data: TArray<T1>; 
    makeCopy: boolean = false): IOmniParallelMapper<T1,T2>;
  function  TaskConfig(const config: IOmniTaskConfig): 
    IOmniParallelMapper<T1,T2>;
  function  WaitFor(maxWait_ms: cardinal): boolean;
end;

class function Map<T1,T2>: IOmniParallelMapper<T1,T2>; overload;
class function Map<T1,T2>(const source: TArray<T1>;
  mapper: TMapProc<T1,T2>): TArray<T2>; overload;

class function Parallel.Map<T1, T2>(const source: TArray<T1>; mapper: TMapProc<T1,T2>):
  TArray<T2>;
var
  map: IOmniParallelMapper<T1,T2>;
begin
  map := Parallel.Map<T1,T2>.Source(source);
  map.Execute(mapper);
  map.WaitFor(INFINITE);
  Result := map.Result;
end;

Parallel.Map takes an array of some type, then runs a user-provided mapper/filter on that array and returns an array of mapped values. Source array can be grabbed by a reference or copied into an internal field if required (as in the example below where source data is prepared in a local array which will be destroy before Parallel.Map finishes execution). Special care is given to the performance so there’s as little data copying and memory manipulations as possible.

Mapper function receives one element from the internal array and produces zero or one element of the output type. In the former case it should set Result to False; in the latter case it should set Result to True and assign appropriate value to the target parameter.

TMapProc<T1,T2> = reference to function(const source: T1; 
  var target: T2): boolean;

Elements are, of course, not processed in the same order as they appear in the source array, but the implementation guarantees that the order will not change during mapping. In other words, if the source array contains elements el1, el2, … elN, then the target array will contain elements map(el1), map(el2), … map(elN), in that order (but with potentially missing elements if mapping function returns False for some input values).

The new demo 60_Map also demonstrates asynchronous use.

procedure TfrmTestParallelMap.btnMap2Click(Sender: TObject);
var
  i      : integer;
  numbers: TArray<integer>;
begin
  SetLength(numbers, CSourceSize);
  for i := Low(numbers) to High(numbers) do
    numbers[i] := i;

  FMapper := Parallel.Map<integer,string>;
  FMapper.Source(numbers, true);
  FMapper.NumTasks(4); //just for testing
  FMapper.NoWait;
  FMapper.OnStop(ShowResult);
  FMapper.Execute(MapOdds);
end;

function TfrmTestParallelMap.MapOdds(const source: integer; 
  var dest: string): boolean;
begin
  Result := Odd(source);
  if Result then
    dest := IntTostr(source);
end;

procedure TfrmTestParallelMap.ShowResult(const task: IOmniTask);
begin
  //we are still in a background thread so schedule work 
  //for the main thread
  task.Invoke(
    procedure
    begin
      lbLog.Items.Add(ToString(FMapper.Result));
      FMapper := nil;
    end);
end;

There’s not much more to say about the parallel map. It is a simple feature with a simple interface and relatively simple implementation (TOmniParallelMapper<T1,T2>.Execute does most of the work).