Sequentielle Abarbeitung mit Threads - Anregungen erwünscht

[DeddyH]

Ich habe gerade eine kleine Denkblockade :oops:. Folgende Situation: ich rufe Daten von einem REST-Server (auch von mir) ab. Da die Datenmengen auch mal etwas größer werden können, habe ich mir eine Art Pagination einfallen lassen, d.h. der Server schickt mir einen Response mit maximal 100 Datensätzen, welcher zusätzlich ein Objekt mit den Feldern offset, count und hasmoredata enthält. Damit kann ich am Client die passende Route errechnen, um die nächsten Datensätze abzurufen. Das funktioniert wie die Wutz, allerdings läuft das im Moment auf dem Client alles im Mainthread ab, der ist dann natürlich blockiert. Daher bin ich auf die Idee gekommen, jede Abfrage und anschließende Abarbeitung in einen Thread zu packen, habe aber Probleme bei der Implementation. Der derzeitige Code hat sinnbildlich den folgenden Aufbau:

Delphi-Quellcode:

procedure TMyClass.GetData;
var
  MoreData: Boolean;

  procedure ReceiveData;
  var
    Response: IResponse;
  begin
    Response := GetDataFromServer;
    DoSomeGreatWork;
    MoreData := Response.HasMoreData;
  end;

begin
  repeat
    ReceiveData;
  until not MoreData;
end;

Wie schon gesagt ist angedacht, ReceiveData in einen Thread auszulagern, aber ohne dass die Schleife vorzeitig verlassen wird. Mir fällt da einfach keine elegante Lösung ein, daher bitte ich um Vorschläge.

[Namenloser]

So?

Delphi-Quellcode:

TWorkThread = class(TThread);
  procedure Execute; override;
end;

procedure TWorkThread.Execute;
begin
  DoSomeGreatWork;
end;

procedure TMyClass.GetData;
var
  Threads: TList<TThread>;
  i: integer;
  procedure ReceiveData;
  var
    Response: IResponse;
    Thread: TWorkThread;
  begin
    Response := GetDataFromServer;
    Thread := TWorkThread.Create;
    Thread.Resume;
    Threads.Add(Thread);
    MoreData := Response.HasMoreData;
  end;
begin
  Threads := TList<TThread>.Create;
  repeat
    ReceiveData;
  until not MoreData;

  // Wait for threads to finish
  for i := 0 to Threads.Count - 1 do
  begin
    Threads[i].WaitFor;
    Threads[i].Free;
  end;

  // Do further work
  ...

  Threads.Free;
end;

(Keine Ahnung, ob ich die Generics richtig verwendet habe)

[Neutral General]

Du verlegst die Schleife in den gleichen Thread und feuerst ein Event wenn die Daten da sind.
GetData erstellt nur den Thread und kehrt danach direkt zurück.
Der Caller wartet dann auf das Event.

[DeddyH]

Mein Problem ist die zeitliche Abfolge. Ich möchte ja gerne das Abholen der Daten bereits in den Thread verlegen, weiß aber erst danach, ob es noch weitere Daten gibt. Ich müsste wohl also eine Art "geschicktes Locking" (TMonitor?) einbauen.

[Namenloser]

Aber wenn du eh auf die Abarbeitung der Schleife warten musst, dann hast du ja nichts gewonnen, wenn du das Abohlen der Daten mit in den Thread verlagerst. Dann kannst du es auch gleich im Hauptthread machen. Kannst du vielleicht noch mal genauer erklären, was du machen willst?

[TiGü]

Delphi-Quellcode:

program Project1;

{$APPTYPE CONSOLE}

{$R *.res}


uses
  System.SysUtils,
  System.Threading;

type
  IResponse = interface
    ['{E705E40C-9BC2-41EE-AE4B-2E1D514395E9}']
    function HasMoreData: Boolean;
  end;

  TResponse = class(TInterfacedObject, IResponse)
  private
    class var
      FCounter: Integer;
  public
    function HasMoreData: Boolean;
  end;

  TMyClass = class
    procedure GetData;
    function GetDataFromServer: IResponse;
    procedure DoSomeGreatWork;
  end;

procedure TMyClass.DoSomeGreatWork;
begin
  Writeln('Make Work Great Again!');
end;

procedure TMyClass.GetData;
var
  MoreData: Boolean;

  procedure ReceiveData;
  var
    Response: IResponse;
  begin
    Response := GetDataFromServer;
    DoSomeGreatWork;
    MoreData := Response.HasMoreData;
  end;

begin
  repeat
    ReceiveData;
    Writeln('HasMoreData: ', MoreData);
  until not MoreData;

  Writeln('Fix und fertig!');
end;

function TMyClass.GetDataFromServer: IResponse;
begin
  Result := TResponse.Create;
  Sleep(50);
end;

function TResponse.HasMoreData: Boolean;
begin
  Result := True;
  Inc(FCounter);
  if FCounter = 100 then
    Result := False;
end;

procedure Main;
begin
  TTask.Run(
    procedure
    var
      MyObject: TMyClass;
    begin
      MyObject := TMyClass.Create;
      try
        MyObject.GetData;
      finally
        MyObject.Free;
      end
    end);
end;

begin
  try
    Main;
    Readln;
  except
    on E: Exception do
      Writeln(E.ClassName, ': ', E.Message);
  end;
end.

[Neutral General]

Zitat:

Zitat von DeddyH (Beitrag 1388165)

Mein Problem ist die zeitliche Abfolge. Ich möchte ja gerne das Abholen der Daten bereits in den Thread verlegen, weiß aber erst danach, ob es noch weitere Daten gibt. Ich müsste wohl also eine Art "geschicktes Locking" (TMonitor?) einbauen.

Hast du gesehn was ich geschrieben hab? Damit erledigt sich das doch.

[TiGü]

Das Daten holen und Verarbeiten kannst du natürlich auch Trennen.
Das Verarbeiten wiederum kann nochmals parallelisiert werden, falls notwendig.

Delphi-Quellcode:

program Project1;

{$APPTYPE CONSOLE}

{$R *.res}


uses
  System.SysUtils,
  System.Threading,
  System.Generics.Collections;

type
  IResponse = interface
    ['{E705E40C-9BC2-41EE-AE4B-2E1D514395E9}']
    function HasMoreData: Boolean;
  end;

  TResponse = class(TInterfacedObject, IResponse)
  private
    class var
      FCounter: Integer;
  public
    function HasMoreData: Boolean;
  end;

  TMyClass = class
  private
    FRespones: TList<IResponse>;
  public
    constructor Create();
    destructor Destroy; override;
    procedure GetData;
    function GetDataFromServer: IResponse;
    procedure DoSomeGreatWork(const AResponse: IResponse);
  end;

constructor TMyClass.Create;
begin
  inherited;
  FRespones := TList<IResponse>.Create;
end;

destructor TMyClass.Destroy;
begin
  FRespones.Free;
  inherited;
end;

procedure TMyClass.DoSomeGreatWork;
begin
  Writeln('Make Work Great Again!');
end;

procedure TMyClass.GetData;
var
  MoreData: Boolean;
  Response: IResponse;
begin
  repeat
    Response := GetDataFromServer;
    FRespones.Add(Response);
    MoreData := Response.HasMoreData;
    Writeln('HasMoreData: ', MoreData);
  until not MoreData;

  for Response in FRespones do
  begin
    DoSomeGreatWork(Response);
  end;

  Writeln('Fix und fertig!');
end;

function TMyClass.GetDataFromServer: IResponse;
begin
  Result := TResponse.Create;
  Sleep(50);
end;

function TResponse.HasMoreData: Boolean;
begin
  Result := True;
  Inc(FCounter);
  if FCounter = 100 then
    Result := False;
end;

procedure Main;
begin
  TTask.Run(
    procedure
    var
      MyObject: TMyClass;
    begin
      MyObject := TMyClass.Create;
      try
        MyObject.GetData;
      finally
        MyObject.Free;
      end
    end);
end;

begin
  try
    Main;
    Readln;
  except
    on E: Exception do
      Writeln(E.ClassName, ': ', E.Message);
  end;
end.

[DeddyH]

Sorry, wie gesagt: ich bin heute irgendwie verpeilt. Primär geht es mir lediglich um eine reagierende Oberfläche, ich schaue mir den Code jetzt erst einmal genauer an. Danke an alle.

[Mavarik]

Pipeline?

Soll jetzt keine Werbung sein...

Aber ich mache es so: LINK

Mavarik