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 TThreadPool::QueueWorkItem(..) oder dein selbst erfundenes TSimpleSemaphore notwendig sind.

Macht doch einfach deinen eigenen TThreadPool auf, mache so oft TTask.Run(..) wie du Einträge in der Liste hast und sage am Schluss 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 von 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.

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

Zitat von Der schöne Günther:

Ich verstehe nicht, warum diese exotisschen Dinge wie TThreadPool::QueueWorkItem(..) oder dein selbst erfundenes 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 von Der schöne Günther:

Macht doch einfach deinen eigenen TThreadPool auf, mache so oft TTask.Run(..) wie du Einträge in der Liste hast und sage am Schluss TTask::WaitForAll(..) .

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

LG Mathias


PS:

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 von NoGAD:

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 von NoGAD:

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
Dummy_ThreadPool.SetMaxWorkerThreads(Dummy_MaxTasks); Ist das nicht genau, was du willst?

Zitat von NoGAD:

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 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 von Der schöne Günther:

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