TThreadPool funktioniert nicht korrekt bei meinem Test

[NoGAD]

Hallo, ich habe hier ein kleines Testprogramm und finde den Fehler nicht, warum es immer zum Absturz kommt.

Aus einer ListBox werden die Items durchlaufen und per Internetabfrage ein JSON gelesen, welches ausgewertet wird.

Ich möchte die Abfrage gerne im mehreren Threads laufen lassen. Nach der Abfrage muss unbedingt gewartet werden, bis alle Threads fertig sind, damit ich das Array verarbiten kann.

Finet bitte jemand den/die Fehler?

Delphi-Quellcode:

    try
      Dummy_ThreadPool.Default.QueueWorkItem(
        procedure
        begin
          SearchGoogleSingleISBNTask(Dummy_ItemString, MyJSONArray[Dummy_Int], Dummy_Semaphore);
        end);
    except
      on E: Exception do
        showmessage(E.message);
    end;
  end;

  Dummy_Semaphore.Acquire;

[Sinspin]

Hallo, hast du erstmal ganz einfach getestet ob es klappt aus der Liste zu lesen (in der Thread Procedure ein bissschen zu warten) und dann ein Ergebnis zu schreiben?
Also ohne den Internet Zugriff.

[Der schöne Günther]

Hallo-

Ich habe das jetzt nicht im Detail durchgearbeitet, aber das ist echt wild. Ich verstehe nicht, warum diese exotisschen Dinge wie

Delphi-Quellcode:

TThreadPool::QueueWorkItem(..)

oder dein selbst erfundenes

Delphi-Quellcode:

TSimpleSemaphore

notwendig sind.

Macht doch einfach deinen eigenen

Delphi-Quellcode:

TThreadPool

auf, mache so oft

Delphi-Quellcode:

TTask.Run(..)

wie du Einträge in der Liste hast und sage am Schluss

Delphi-Quellcode:

TTask::WaitForAll(..)

.

[Der schöne Günther]

PS: Das eigentliche Abholen der Daten lässt sich sehr stark vereinfachen:

1. Wirf einen TRestClient auf dein Formular
2. Setze BaseURL auf https://www.googleapis.com
3. Wirf ein TRestRequest auf dein Formular
4. Setze Resource auf books/v1/volumes?q=isbn:{ISBN}
5. Füge einen Param hinzu, setze Kind auf pkURLSEGMENT und Name auf isbn

In deinem Code kannst du dann einfach sagen

Delphi-Quellcode:

   for var isbn in ['978-3-77073-431-3', '978-0-59600-224-4'] do
      begin
         RESTRequest1.Params[0].Value := isbn;
         RESTRequest1.Execute();
         ShowMessage( RESTRequest1.Response.JSONText );
      end;

Das ist das gesamte Abholen via Http.

[NoGAD]

Danke für die Anregungen.

Das ist mein erster Versuch, daher ist alles durcheinander.

Zitat:

Zitat von Sinspin (Beitrag 1535271)

Hallo, hast du erstmal ganz einfach getestet ob es klappt aus der Liste zu lesen (in der Thread Procedure ein bissschen zu warten) und dann ein Ergebnis zu schreiben?
Also ohne den Internet Zugriff.

Ohne Thread klappt das, also in einer normalen Schleife, wenn alle Einträge nacheinander abgearbeitet werden.

Zitat:

Zitat von Der schöne Günther (Beitrag 1535285)

Ich verstehe nicht, warum diese exotisschen Dinge wie

Delphi-Quellcode:

TThreadPool::QueueWorkItem(..)

oder dein selbst erfundenes

Delphi-Quellcode:

TSimpleSemaphore

notwendig sind.

Bei Delphi 10.4 gibt es scheinbar keine Möglichkeit, mittels TThreadPool die Anzahl der Threads zu begrenzen, daher der Versuch über TSemaphore. Da aber auch TSemaphore scheinbar keine Erstellung mit einem Maximalparameter für die Begrenzung der Threads zulässt, habe ich TSimpleSemaphore erstellt.

Zitat:

Zitat von Der schöne Günther (Beitrag 1535285)

Macht doch einfach deinen eigenen

Delphi-Quellcode:

TThreadPool

auf, mache so oft

Delphi-Quellcode:

TTask.Run(..)

wie du Einträge in der Liste hast und sage am Schluss

Delphi-Quellcode:

TTask::WaitForAll(..)

.

Das würde ich gerne, habe bisher aber nicht verstanden, wie das umgesetzt wird. :-(

LG Mathias


PS:

Zitat:

Zitat von Der schöne Günther

PS: Das eigentliche Abholen der Daten lässt sich sehr stark vereinfachen:

Ist das ein Thread?

[Olli73]

Zitat:

Zitat von NoGAD (Beitrag 1535287)

Bei Delphi 10.4 gibt es scheinbar keine Möglichkeit, mittels TThreadPool die Anzahl der Threads zu begrenzen, daher der Versuch über TSemaphore. Da aber auch TSemaphore scheinbar keine Erstellung mit einem Maximalparameter für die Begrenzung der Threads zulässt, habe ich TSimpleSemaphore erstellt.

https://docwiki.embarcadero.com/Libr...xWorkerThreads

???

[Der schöne Günther]

Zitat:

Zitat von NoGAD (Beitrag 1535287)

Bei Delphi 10.4 gibt es scheinbar keine Möglichkeit, mittels TThreadPool die Anzahl der Threads zu begrenzen, daher der Versuch über TSemaphore.

Habe ich auch nicht verstanden.
Genau das tust du doch schon in deinem

Delphi-Quellcode:

Dummy_ThreadPool.SetMaxWorkerThreads(Dummy_MaxTasks);

Ist das nicht genau, was du willst?

Zitat:

Zitat von NoGAD (Beitrag 1535287)

Zitat:

Zitat von Der schöne Günther

PS: Das eigentliche Abholen der Daten lässt sich sehr stark vereinfachen:

Ist das ein Thread?

Nein, das sollte nur zeigen, wie stark sich das eigentliche Abholen vom Server vereinfachen lässt, damit man sich nicht mehr mit Http-Returncodes und sonstwas herumschlagen muss. Anfrage raus, Json zurück, fertig.

Aber natürlich kann man das auch parallelisieren.
Hierzu würde ich aber einmal das Thema hier überfliegen, insbesondere die Beiträge des immer äußerst lesenswerten Herrn Raabe:
https://en.delphipraxis.net/topic/54...s-in-parallel/


Ich habe das einmal so runtergetippt wie ich das machen würde, hier fehlt natürlich noch deine Konvertierung des Json-Strings von der Google-API in deinen eigenen Typ

Delphi-Quellcode:

TJSon_Items

.

Delphi-Quellcode:

unit Unit1;

interface uses
   System.SysUtils,
   System.Classes,
   System.Threading,

   Data.Bind.Components,
   Data.Bind.ObjectScope,

   Vcl.Graphics,
   Vcl.Controls,
   Vcl.Forms,
   Vcl.Dialogs,
   Vcl.StdCtrls,

   REST.Types,
   REST.Client;

type
   TForm1 = class(TForm)
      button_serial: TButton;
      button_parallel: TButton;
      procedure button_serialClick(Sender: TObject);
      function getIsbnJson(const isbn: String): String;
      function getIsbnJsonAsync(const isbn: String): IFuture<String>;
      procedure button_parallelClick(Sender: TObject);
   end;

var
   Form1: TForm1;

const
   ISBNs: TArray<String> = [
      '978-3-77073-431-3',
      '978-0-59600-224-4',
      '3-630-61957-6',
      '978-3-89086-120-3',
      '978-3-89086-138-8'
   ];

implementation uses System.Diagnostics;

{$R *.dfm}

procedure TForm1.button_serialClick(Sender: TObject);
begin
   const stopwatch = TStopwatch.StartNew();
   for var isbn in ISBNs do
      getIsbnJson(isbn);
   stopwatch.Stop();
   ShowMessageFmt('Ms taken: %d ms', [stopwatch.ElapsedMilliseconds]);
end;

procedure TForm1.button_parallelClick(Sender: TObject);
begin
   var tasks: TArray<IFuture<String>> := [];

   const stopwatch = TStopwatch.StartNew();
   for var isbn in ISBNs do
      tasks := tasks + [getIsbnJsonAsync(isbn).Start()];
   for var task in tasks do
      task.Wait();
   stopwatch.Stop();

   ShowMessageFmt('Ms taken: %d ms', [stopwatch.ElapsedMilliseconds]);
end;

function TForm1.getIsbnJson(const isbn: String): String;
begin
   const request = TRESTRequest.Create(nil);
   try
      request.Resource := 'books/v1/volumes?q=isbn:{ISBN}';
      request.Params.AddItem('isbn', isbn, TRESTRequestParameterKind.pkURLSEGMENT);

      const client = TRESTClient.Create({Owner:}request);
      client.BaseURL := 'https://www.googleapis.com';
      client.SynchronizedEvents := False;
      request.Client := client;
      request.SynchronizedEvents := False;

      request.Execute();
      Result := request.Response.JSONText;
   finally
      request.Destroy();
   end;
end;

function TForm1.getIsbnJsonAsync(const isbn: String): IFuture<String>;
begin
   Result := TTask.Future<String>(
      function(): String
      begin
         Result := getIsbnJson(isbn);
      end
   );
end;

end.

[Olli73]

Dieser Code tut nicht das, was du willst:

Delphi-Quellcode:

for Dummy_Int := Dummy_ItemCount downto 0 do
  begin
    Dummy_ItemString := ListBox1.Items[Dummy_Int];
    MyJSONArray[Dummy_Int].SyncObj := TCriticalSection.Create;

    try
      Dummy_ThreadPool.Default.QueueWorkItem(
        procedure
        begin
          SearchGoogleSingleISBNTask(Dummy_ItemString, MyJSONArray[Dummy_Int], Dummy_Semaphore);
        end);
    except
      on E: Exception do
        showmessage(E.message);
    end;
  end;

"Dummy_Int" wird im Thread erst ausgewertet, wenn der Thread ausgeführt wird. Da du aber Dummy_Int (im Hauptthrad) herunterzählst, ist es zu diesem Zeitpunkt evtl. schon 0 oder irgendwas dazwischen.

Dur brauchst sowas:

Delphi-Quellcode:

var X: Integer;


X := 0;
for Dummy_Int := Dummy_ItemCount downto 0 do
  begin
    Dummy_ItemString := ListBox1.Items[Dummy_Int];
    MyJSONArray[Dummy_Int].SyncObj := TCriticalSection.Create;

    try
      Dummy_ThreadPool.Default.QueueWorkItem(
        procedure
        var
          Y: Integer;
        begin
          // Hier synchronisiert(!) ein Inc(X) machen und ein Y := X;
          SearchGoogleSingleISBNTask(Dummy_ItemString, MyJSONArray[Y], Dummy_Semaphore); // Und hier dann Y statt Dummy_Int 
        end);
    except
      on E: Exception do
        showmessage(E.message);
    end;
  end;

[NoGAD]

Ich danke euch. Durch die Hilfen werde ich mich erst einmal drucharbeiten.

LG Mathias :-)

[NoGAD]

Zitat:

Zitat von Der schöne Günther (Beitrag 1535289)

Delphi-Quellcode:

procedure TForm1.button_parallelClick(Sender: TObject);
begin
   var tasks: TArray<IFuture<String>> := [];

   const stopwatch = TStopwatch.StartNew();
   for var isbn in ISBNs do
      tasks := tasks + [getIsbnJsonAsync(isbn).Start()];
   for var task in tasks do
      task.Wait();
   stopwatch.Stop();

   ShowMessageFmt('Ms taken: %d ms', [stopwatch.ElapsedMilliseconds]);
end;

Hallo Günther,

bei wenigen Abfragen funktioniert Dein Code. Da ich aber die Anzahl der Abfragen begrenzen muss, müsste ich die Anzahl der Tasks begrenzen. Wie wäre das möglich?

Ich hatte mir überlegt, den Code zu ändern, um die Tasks erst zu sammeln und im Anschluss zu starten:

Delphi-Quellcode:

  for var isbn in ISBNs do
    tasks := tasks + [getIsbnJsonAsync(isbn)];

  for var task in tasks do
  begin
{ Hier könnte ich mittels Integer-Variable prüfen, wie viele Tasks gestartet wurden }
    task.Start();
{ Hier müsste ich prüfen, ob aktuell laufende Tasks schon beendet wurden }
  end;

Vielleicht wäre es einfacher, das über eine Message laufen zu lassen?


LG Mathias

[Der schöne Günther]

Zitat:

Zitat von NoGAD (Beitrag 1535306)

Da ich aber die Anzahl der Abfragen begrenzen muss, müsste ich die Anzahl der Tasks begrenzen. Wie wäre das möglich?

Klar- Das wäre eigentlich genauso, wie du es in deinem ersten Versuch bereits drin hattest: Du erstellst dir erst einmal deinen eigenen

Delphi-Quellcode:

TThreadPool

und setzt dann dort die Anzahl der Worker-Threads.

In Aufrufe wie

Delphi-Quellcode:

TTask.Run(..)

oder

Delphi-Quellcode:

TTask.Future(..)

gibst du als zusätzlichen Parameter dann einfach noch deinen selbst erstellten Threadpool an, die Tasks/Futures laufen dann in diesem Threadpool, in welchem nie mehr als X Tasks gleichzeitig ausgeführt werden.

PS: Rein aus Interesse- Was schwebt dir denn als maximale Anzahl der gleichzeitig laufenden Threads vor?

[NoGAD]

Zitat:

Zitat von Der schöne Günther (Beitrag 1535308)

PS: Rein aus Interesse- Was schwebt dir denn als maximale Anzahl der gleichzeitig laufenden Threads vor?

5-20.

Manchmal ist bei 8 schon Schluss, dann gibt es einen 429 (Too Many Requests) Fehler. Ich weiß noch nicht damit umzugehen.

Die Threads möchte ich ja nur für die Internetabfrage nutzen, um parallele Abfragen zu machen.

LG Mathias


PS: die tasks haben ja eine fortlaufene ID. Ich habe bemerkt, dass diese nicht zurückgesetzt wird, wenn ein task beendet wurde.

Somit erhöht sich die id bei einem weiteren Durchlauf.

Delphi-Quellcode:

  for var task in tasks do
    task._Release;

Löscht die tasks leider auch nicht.

[Uwe Raabe]

Ich finde, das ist ein klassischer Anwendungsfall für TParallel.For.
Über den zurückgegeben TLoopResult kannst du prüfen, ob alles abgearbeitet ist.
Wenn du einen eigenen ThreadPool übergibst, kannst du mit TThreadPool.SetMaxWorkerThreads die Anzahl gleichzeitig aktiver Tasks eingrenzen.

[Der schöne Günther]

Ich kann Google verstehen, acht ist schon viel. Vom Gefühl her hätte ich den Threadpool auf 4-8 begrenzt. Weiterhin muss es nicht einzig an "zu vielen gleichzeitig" liegen, es könnte genauso gut sein, dass "zu viele Einträge pro Zeit" übertragen wurden.

Die Entscheidung, ab wann der Server sagt "Junge, nimm mal den Fuß vom Gas" liegt allein beim Anbieter (Google). Du könntest in deiner Anwendung entweder die fehlgeschlagenen ISBNs in einer neuen Liste sammeln mit der man es dann noch einmal versucht oder nur die erfolgreichen aus der "To Do"-Liste entfernen.

Zitat:

Zitat von Uwe Raabe (Beitrag 1535310)

Ich finde, das ist ein klassischer Anwendungsfall für TParallel.For.
(...)

Stimmt, das hatte ich nicht auf dem Schirm. Das macht es noch einfacher, als von Hand Tasks/Futures in einem Array zu sammeln. 👍

[NoGAD]

Zitat:

Zitat von Uwe Raabe (Beitrag 1535310)

TLoopResult kannst du prüfen, ob alles abgearbeitet ist.

Welche unit benötige ich für TLoopResult?

In System.Threading gibt es das nicht unter Delphi 10.4

[jaenicke]

Zitat:

Zitat von NoGAD (Beitrag 1535328)

Welche unit benötige ich für TLoopResult?

In System.Threading gibt es das nicht unter Delphi 10.4

Doch, gibt es. Du hast offensichtlich in der Unit nicht danach gesucht. Es ist dort innerhalb von TParallel als nested type deklariert, sprich ist nach außen verfügbar als TParallel.TLoopResult.

// EDIT:
Es geht aber auch ohne...

Delphi-Quellcode:

  var LoopResult := TParallel.For(1, 10,
    procedure(i: Integer; State: TParallel.TLoopState)
    begin
      if i > 5 then
        State.Break;
    end);
  if LoopResult.Completed then
    ShowMessage('Durchgelaufen')
  else
  begin
    var BreakIndex: string := LoopResult.LowestBreakIteration;
    ShowMessage('Abgebrochen bei: ' + BreakIndex);
  end;

[NoGAD]

Hallo an euch alle.

Im Moment ist mir das einfach zu schwer, korrekt umzusetzen. Ich werde mir das für später aufheben und erst einmal zurück zur Einzelabfrage gehen, will heißen: Item für Item aus der ListBox.

Damit umgehe ich auch "Too many Requests".


Vielen lieben Dank für eure Hilfe! :-)

LG Mathias

[himitsu]

vielleicht auch mal eine ganz abwägige Idee?

mehrere ISBNs in einer Anfrage zusammenfassen und mit nur einem Aufruf abfragen. :stupid:

[Der schöne Günther]

Zitat:

Zitat von NoGAD (Beitrag 1535332)

Im Moment ist mir das einfach zu schwer, korrekt umzusetzen. Ich werde mir das für später aufheben und erst einmal zurück zur Einzelabfrage gehen, will heißen: Item für Item aus der ListBox.

Kann man absolut verstehen, würde ich ebenso machen- Performance-Optimierung ganz zum Schluss.

Wer behauptet, Programme zu parallelisieren sei einfach hat es ganz sicher selbst noch nicht verstanden.

[jaenicke]

Zitat:

Zitat von Der schöne Günther (Beitrag 1535339)

Wer behauptet, Programme zu parallelisieren sei einfach hat es ganz sicher selbst noch nicht verstanden.

Wobei es am schwierigsten ist, es hinterher zu ergänzen.

Wenn man konsequent Funktionalität kapselt, so dass diese möglichst keine Abhängigkeiten hat, hat man es später einfacher damit.

[Der schöne Günther]

Zitat:

Zitat von jaenicke (Beitrag 1535340)

Zitat:

Zitat von Der schöne Günther (Beitrag 1535339)

Wer behauptet, Programme zu parallelisieren sei einfach hat es ganz sicher selbst noch nicht verstanden.

Wobei es am schwierigsten ist, es hinterher zu ergänzen.

Dem würde ich jetzt schon, aber nur begrenzt zustimmen-

Klar ist das ein Aufwand, wenn man z.B. vorher damit zufrieden war, dass die gesamte Anwendung hängt solange sie Daten aus dem Netz zieht und verarbeitet, und jetzt eine Anwendung haben möchte, die immer reagiert und jeden einzelnen ISBN-Eintrag anzeigt, ob er schon abgearbeitet ist.

Aber Code ohne Verzahnung mit Benutzer-Oberfläche später mit Parallelisierung zu optimieren (beispielsweise Berechnungen oder aufeinanderfolgende Kommunikation mit unterschiedlichen Teilnehmern) macht eigentlich sogar relativ viel Spaß 😊

[NoGAD]

Zitat:

Zitat von Der schöne Günther (Beitrag 1535341)

Aber Code ohne Verzahnung mit Benutzer-Oberfläche [..] macht eigentlich sogar relativ viel Spaß 😊

Dem stimme ich zu. Es ärgert mich aber, dass ich manches nicht mehr (so schnell wie früher) verstehe und dann stundenlang an einem Problem sitze, obwohl ich diese Zeit in weitere Programmierung stecken könnte ;-)

LG Mathias