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.

[Der schöne Günther]

┴┬┴┤( ͡° ͜ʖ├┬┴┬

[Fritzew]

Also mit dem Ansatz kommst Du nicht weit...

IFuture<t> funktioniert so nicht.
Wenn Du auf iFuture<t>.Value zugreifst wird gewartet bis der Thread beendet ist. Du müsstest auf jeden
Fall eine Abbruchbdingung haben.

In dem Fall würde ich mir einen TThread ableiten

[Der schöne Günther]

Richtig, deshalb hole ich mir doch Value auch nur wenn Status = Completed , ansonsten fasse ich das Ding nicht an. Das war so meine Idee dabei.

[Fritzew]

Ist dennoch der falsche Ansatz denke ich. Future ist nicht für sowas gedacht.....
Dann kommt noch dazu dass Du das System unnötig belastet, da alle tasks ja trotzdem bis zum ende weiterlaufen.
Baue Dir eine eigene Threadklasse die das sauber handelt.

[Glados]

Zitat:

Ist dennoch der falsche Ansatz denke ich. Future ist nicht für sowas gedacht.....

Wäre Future denn für so etwas geeignet?

Delphi-Quellcode:

 aFutureTasks := [TTask.Run(
    procedure
  begin
   Sleep(1000); // Arbeit 1
  end)];

 TTask.WaitForAll(aFutureTasks); // warten, bis alle fertig sind.

Ich verwende für solche Fälle immer eine Boolsche Variable die vor TTask.Run False ist, am Ende von TTask Run True und nach TTask run kommt eine while-Schleife.