Asynchron Dinge sammeln, in bestimmter Zeit

[Der schöne Günther]

Als Vorwarnung: Ich bin nicht gut was parallele Programmierung angeht. Die PPL aus Delphi habe ich nur ein paar mal oberflächlich verwendet, die OmniThreadLibrary habe ich mir nur kurz angeschaut, nie in der Praxis verwendet (sie sieht aber extrem geil aus).

Ich möchte in einer bestimmten Zeit Datendinger sammeln. Das Sammeln kann lange dauern, es ist aber wichtig dass mein Aufruf nach einer bestimmten Zeit wieder zurückkehrt. Notfalls hat er nichts gesammelt, das ist ok.

Hier einmal ein simples Beispiel für ein Datending mittels IFuture<T>:

Delphi-Quellcode:

uses
   System.SysUtils,
   System.Threading;

function produceData(): TBytes;
begin
   Sleep(5000);
   Result := [42];
end;

function collectData(const timeoutMs: Word): TBytes;
var
   collected: IFuture<TBytes>;
begin
   collected := TTask.Future<TBytes>(produceData);
   if collected.Wait(timeoutMs) then
      Result := collected.Value
   else
      Result := [];
end;

Ist das schon im Ansatz falsch?


Wenn nicht: Ich möchte das jetzt aufblähen auf nicht ein TBytes, sondern mehrere: collectData(..) soll jetzt ein TArray<TBytes> statt TBytes liefern.


Ich tue mich jetzt schwer wie ich das richtig löse. Da ein IFuture<T> sich von ITask ableitet kann ich sagen TTask.WaitForAll(meineTasks, gewünschterTimeout) . Ich würde dann auf alle Futures warten, aber nicht länger als X. Ich nehme alle bislang gesammelten Ergebnisse. Alle anderen Futures die nicht rechtzeitig fertig wurden können ihren Kram noch in Ruhe zu Ende machen, aber es interessiert niemanden mehr.

Wäre das so richtig umgesetzt? Gibt es unter System.Threading (oder anderswo) das schon vorgefertigt oder vereinfacht? Hätte ich konkret etwas davon meine Futures hier in einem eigenen TThreadPool ablaufen zu lassen?

Delphi-Quellcode:

uses
   System.SysUtils,
   System.Threading;

function produceData(): TBytes;
begin
   Sleep( Random(5000) );
   Result := [Random(100)];
end;

function collectData(const timeoutMs: Word): TArray<TBytes>;
type
   TFuture = IFuture<TBytes>;
var
   futures:   TArray<ITask>;
   task:      ITask;
begin
   futures := [
      TTask.Future<TBytes>(produceData),
      TTask.Future<TBytes>(produceData),
      TTask.Future<TBytes>(produceData)
   ];

   TTask.WaitForAll(futures, timeoutMs);
   Result := [];
   for task in futures do
      if (task.Status = TTaskStatus.Completed) then
         Result := Result + [TFuture(task).Value];
end;

Vielen Dank im Voraus, ich bin gespannt.