Ein Versuch mit Omnithreadlibrary

[IMPEGA]

Hi Leute
Ich, schwacher Hobbyprogrammierer versuche mal wieder etwas Neues zu lernen.
Könnt Ihr mir sagen, ob das so korrekt ist? Das ist nur ein Beispiel.
Im Endeffekt habe ich eine Liste mit sehr vielen Daten die sollen in eine Liste einsortiert werden.
Threaded weil die Daten unterwegs angepasst werden müssen und Serial habe ich es schon versucht, es dauert unheimlich lange.
Hier ist mein Code. Bitte um Mitleid, ich versuche es nur.
Ich möchte nur wissen ob der Aufbau so weit OK ist, oder muss ich noch mehr beachten.
Leider ist mein Englisch sehr, sehr schlecht und mich durchlesen ist sehr problematisch.
Ich habe vieles probiert, das scheint mir am besten zu funktionieren.
Im Endeffekt möchte ich ein ThreadPool abarbeiten. Leider kriege ich es mit dem Threadpool nicht wirklich hin also muss die Notlösung her.
Hier ist mein Code.

Delphi-Quellcode:

unit Unit1;

interface

uses
  Windows, Messages, SysUtils, Variants, Classes, Graphics, Controls, Forms, Dialogs, StdCtrls,
  OtlComm, OtlSync, OtlCommon, OtlTask, OtlCollections;

type
  TForm1 = class(TForm)
    btnAsync: TButton;
    Memo1: TMemo;
    Button1: TButton;
    procedure btnAsyncClick(Sender: TObject);
    procedure Button1Click(Sender: TObject);
  private
    Liste : TStringList;
  public

  end;

var
  Form1: TForm1;

implementation

uses
  OtlTaskControl,
  OtlParallel;

{$R *.dfm}

var
 Abbruch : Boolean = False;

procedure TForm1.btnAsyncClick(Sender: TObject);
var
  Index  : Integer;
  Indx   : Integer;
  Tasks  : Word;
  Fertig : Word;
  Status : string;
begin

  Liste := TStringList.Create;
    for Index := 1 to 10000000 do
      begin
        Liste.Add('Ein Datensatz' + Index.ToString);
      end;


  Abbruch := False;
  Tasks   := 3;
  Fertig  := 0;

   {3 Parallele Tasks starten}
    for Index := 1 to Tasks do
      begin
        Parallel.Async(procedure var Indx : Integer;
            begin
             {Executed in background Thread }
              for Indx := 0 to (Liste.Count - 1) div Tasks do
                begin
                  if Abbruch = True then Break;
                    try
                      Sleep(Random(2000));
                        if Liste.Count > 0 then
                          begin
                            Status := Liste.Strings[0];
                            Liste.Delete(0);
                            Memo1.Lines.Add(Status);
                          end;
                    except
                    end;
                end;
            end,
          Parallel.TaskConfig.OnTerminated(
            procedure(const Task: IOmniTaskControl)
              begin
               {Executed in Main Thread }
                Inc(Fertig);
                  if Fertig = Tasks then FreeAndNil(Liste);
              end
          )
        );
      end;
end;

procedure TForm1.Button1Click(Sender: TObject);
begin
  Abbruch := True; (*Ich muss die Möglichkeit haben es abzubrechen*)
end;

end.

Bitte um Kritik und Verbesserungsvorschläge.

[himitsu]

Erstmal arbeitest du mit Threads, wobei du mehrmals "gleichzeitig" auf globale Objekte zugreist.

DU mußt also den Zugriff darauf irgendwie synchronisieren, z.B. mit einer CriticalSection oder passenden threadsicheren Komponenten ala TThreadList.


PS: seit paat Jährchen gibt es auch im Delphi eine immer größerwerdende Thread-Library,
http://docwiki.embarcadero.com/RADSt...amming_Library
wozu es auch einige Tutorials und YT-Videos gibt.


Wenn man keine Kontrolle braucht, wieviele Threads gleichzeitig laufen, oder Dergleichen,
dann reicht auch TThread, welches bereits seit Einführung der Generics nettze Funktionen besitzt.

Delphi-Quellcode:

TThread.CreateAnonymousThread(procedure
  begin
    // im Thread
  end);

TThread.Synchronize(procedure
  begin
    // jetzt im Hauptthread (auf Ende warten, ähnlich SendMessage)
  end);

TThread.Queue(procedure  // hier besser TThread.ForceQueue nutzen
  begin
    // später im Hauptthread (ohne zu warten, ähnlich PostMessage)
  end);

[IMPEGA]

Zitat:

DU mußt also den Zugriff darauf irgendwie synchronisieren, z.B. mit einer CriticalSection oder passenden threadsicheren Komponenten ala TThreadList.

Ja, das stimmt. Den Code habe ich aber aus einem Beispiel von OmniThreadLibrary abgeleitet. Test_46.
Dort ist es auch so deklariert. Ich habe eine interne Schleife zusätzlich eingebaut und paar Sachen die für mich relevant sind dazu geschrieben.
Die Übergabe an den Hauptthread wird dort aber genau so angegeben. Darum frage ich die Profis was Sache ist.
Bin etwas durcheinander. Ich kenne es eigentlich auch so, dass man den Thread synchronisiert und nicht direkt anspricht.
Hier möchte ich noch etwas hinzufügen.
Ich kann mehr oder weniger mit Threads umgehen,bis jetzt habe ich die Class von Delphi-Praxis benutzt. Hat so weit alles gut geklappt.
Mir geht es um das Lernen. OmniThreadLibrary soll wohl sehr gut sein, deshalb versuche ich ein wenig zu testen.
Das ist das Original aus der Hilfe von OmniThreadLibrary

Delphi-Quellcode:

uses
  OtlTaskControl,
  OtlParallel;

{$R *.dfm}

procedure TfrmDemoParallelAsync.btnAsyncClick(Sender: TObject);
begin
  btnAsync.Enabled := false;

  Parallel.Async(
    procedure
    begin
      // executed in background thread
      Sleep(500);
      MessageBeep($FFFFFFFF);
    end,
    Parallel.TaskConfig.OnTerminated(
      procedure (const task: IOmniTaskControl)
      begin
        // executed in main thread
        btnAsync.Enabled := true;
      end
    )
  );
end;

Ok, ich denke hier ist einiges falsch.

Delphi-Quellcode:

                            Status := Liste.Strings[0];
                            Liste.Delete(0);
                            Memo1.Lines.Add(Status);

Meine Frage dazu.
Die ersten zwei Zeilen. Kann ich so auf die Liste zugreifen?
Ich denke, dass die dritte Zeile nun irgendwie synchronisiert werden muss.
Oder sind alle 3 Zeilen in der Form nicht akzeptabel?
Die Threads greifen nämlich auf die globale Liste zu. Darf es so bleiben?

PS: Die Hilfe bei http://docwiki.embarcadero.com ist vorsichtig ausgedrückt sehr mager. Ein Anfänger wird garantiert nie eine Thread basierte Anwendung danach schreiben können.

[himitsu]

Ich nutze die OTL nicht, aber ich bin mir sehr sicher, dass in der Dokumentation oder in Demos niemals derartig auf ungeschützte Objekte zugegriffen wird. (das würde mich jedenfalls sehr wundern, wenn dort derart grob fahrlässige Fehler drin wären)


Klar kann man so darauf zugreifen, aber niemals "ungeschützt" in einem Thread, es sei denn dieser eine Thread ist der Einzige, welcher darauf zugreift.

Der Haupt-/MainThrad ist ja auch ein Thread und so lange etwas ausschließlich darin läuft, kann es z.B. auch problemlos auf die VCL (GUI) zugreifen.

Zitat:

Delphi-Quellcode:

Status := Liste.Strings[0];
Liste.Delete(0);

Wenn ein anderer Thread zwichen diesen Beiden Befehlen die Liste verändert (Add/Append/Delete), dann rate mal was hier wohl schief laufen kann.
Noch schlimmer ist es, dass der andere Thread auch "mitten" in einem der beiden Befehle einschlagen kann ... dann wird nicht nur dieser eine Index betroffen, sondern es kann auch das gesamte Programm zerstören, indem es beim Speichermanagement was schrottet.


* eine CriticalSection um das Auslesen+Delete (alternativ gehen auch andere Synchronisiermethoden, wie z.B. TMonitor.Enter oder siehe SyncObjs)
das Benutzten des "Status" im Anschluss muß nicht da drin sein, da es die "globale" Liste nicht mehr betrifft.

PS: statt Delete kann man auch Remove benuten, dann ist es nur noch ein Befehl.
Aber auch dieser ist nicht "atomar" und muß demach abgesichert werden.

* Rate mal, warum ich die TThreadList erwähnt hab, denn Jene hat bereits eine Sicherung eingebaut, welche Thradübergreifende Konflikte abfängt.


Beispiel für atomare Befehle:

Delphi-Quellcode:

Inc(i);

hat Probleme, wenn mehrere Threads gleichzeitig den selben Speicher verändern wollen.
Darum gibt es InterlockedIncrement / InterlockedIncrement, bzw. inzwischen auch AtomicIncrement, als platformübergreifende Variante. (nicht nur Windows, sondern auch Android, iOS usw.)

[IMPEGA]

Danke erstmal.
OK, also nicht so prickelnd.
Gut erklärt, danke, sehe ich ein. Nun schaue ich mir TThreadList an.
Mal schauen was ich da basteln kann.

Falls mir jemand ein einfaches Beispiel in TThreadList zeigen könnte wäre es supi.
Ich fange an zu lesen, es dauert halt immer sehr lange bis ich es verstanden habe. Einfacher Beispiel würde es mir enorm erleichtern.
Wie schon gesagt. Meine Englisch -kenntnisse sind ein Problem.

Im Endeffekt möchte ich eine sehr lange Liste in mehreren Threads abarbeiten.
Daten werden ausgewertet, in eine andere Tabelle eingetragen und die Zeile muss aus der Liste verschwinden.
Vielleicht hat jemand etwas Ähnliches schon parat. Es muss nicht exakt die Lösung für mich sein, es geht um das Verstehen der Verläufe.
Mein Code schreibe ich mir eh selbst.

[IMPEGA]

Neuer Anfang.

Frage an die Profis.
Bin ich damit im Ansatz OK?

Delphi-Quellcode:

    { Private-Deklarationen }
    Fpool : TThreadpool;

...

procedure TForm1.Button5Click(Sender: TObject);
begin
  Memo1.Clear;

  if FPool = nil then
    begin
      Fpool := TThreadPool.Create;
      Fpool.SetMaxWorkerThreads(10);
      Fpool.SetMinWorkerThreads(1);
    end;

  TTask.Run(procedure
    begin
      TParallel.&For(0, 100,  {von 1 to 10}
        procedure(Index : Integer)
          begin
            Sleep(1000);
            Memo1.Lines.Add('TThread.Ergebnis : ' + TTask.CurrentTask.Id.ToString + ' : ' + Index.ToString);

           // KANN ICH HIER AUF EINE GLOBALE STRINGLIST ZUGREIFEN ?????

          end, FPool);
    end);
end;

[supermuckl]

Schon diese Zeile müsste mit dem Hauptthread synchronisiert werden:

Delphi-Quellcode:

Memo1.Lines.Add('TThread.Ergebnis : ' + TTask.CurrentTask.Id.ToString + ' : ' + Index.ToString);

Weil die VCL das Memo1 verwaltet und ggf. neu zeichen will, während ein Pool-Thread gerade die Lines verändert.

[TurboMagic]

Wie soll denn auf die Stringliste zugegriffen werden?
Von allen zum Zeitpunkt der Thread Ausführung aktiven Parteien
nur lesend?

Sonst braucht es z. B. eine TCriticalSection mit Aquire und Release
um die Zugriffe herum.

[IMPEGA]

Ich bin noch nicht ganz fertig, allerdings schon sehr müde davon.
Videos auf YT? Zu meinem Thema nicht eins was man als brauchbar bezeichnen kann. Sprich, ja ich habe mehrere auf Englisch geschaut.
Englische bringen auch was, das Problem ist nur, da kann ich gleich Muke dabei hören. Verstehen tue ich vielleicht 10 Prozent.
Ich kann nur das was ich sehe umsetzen.
Dokumentation beinhaltet heutzutage so gut wie gar keine Vorlagen, die man analysieren könnte.
Für einen Anfänger ist es sehr schwer zu Recht zu kommen.
Bitte schreibt mir keine Verweise auf YT oder Emba. Die Hilfe habe ich schon gesucht.

Neuer Versuch.
Das funktioniert zwar schon, allerdings ich verliere einige Ausgaben. Es fehlen Sachen, so als nicht alle Threads geliefert hätten.
Ich habe kein Plan, ob es am Server liegt der nicht antwortet, oder mein Code fehlerhaft ist. Momentan bin ich so überladen mit den Infos dass ich mich schon leicht verliere.

Ich kann es nicht in Echt testen, das ist mir zu riskant, ich habe mir auf Freespace halt eine Liste mit 1000 Datensätzen erstellt.
Darauf greife ich zu und versuche mein Vorhaben zu simulieren. Später wird es local auf dem Server passieren. Das kann ich aber noch nicht testen.

Delphi-Quellcode:

(* uses System.Threading;
   private
     FPool : TThreadpool;  *)
procedure TForm1.Button6Click(Sender: TObject);
var
  Task      : ITask;
  Korrektur : string;
  Liste     : TStringList;
begin
  Liste      := TStringList.Create;
  Liste.LoadFromFile('LocaleDatei.dat');

  if FPool = nil then
    begin
      Fpool := TThreadPool.Create;
      Fpool.SetMaxWorkerThreads(10);
      Fpool.SetMinWorkerThreads(1);
    end;

  TParallel.For(0, Liste.Count - 1, procedure (Index: Integer)
    begin
      Task := TTask.Create(
        procedure
          begin
            if TTask.CurrentTask.Status = TTaskStatus.Canceled then
               exit;
           Korrektur := GetHTTP; // (function GetHTTP : string;) HIER VERBINDE ICH MICH PER GET AUF EINEN SERVER.  HOLE EIN SATZ ALS STRING REIN

            if TTask.CurrentTask.Status <> TTaskStatus.Canceled then
              begin
                TThread.Queue(TThread.CurrentThread,
                  procedure
                    begin
                      if Assigned(Memo1) then
                        begin
                          Memo1.Lines.Add(Liste[Index]); // HIER WIRD SCHON FALSCH NUMMERIERT. ICH DENKE ABER WEIL ES NICHT SERIAL LÄUFT
                          Korrektur := StringReplace(Korrektur, 'Alter Wert', 'Neuer Wert' ,[rfReplaceAll, rfIgnoreCase]); //HIER WIRD NOCH MEHR PASSIEREN, DAS DIENNT NUR ALS MUSTER.
                          Memo1.Lines.Add(Korrektur); // HIER WIRD SPETER NOCH EIN UPLOAD GEMACHT. ERSTMAL UNWICHTIG. DIENT KONTROLLE
                        end;
                    end);
              end;
          end);
      Task.Start;
    end, FPool);
end;

Es sind sehr viele kleine Änderungen nötig. Serial dauert es ewig lange. Es muss schon mindestens 50-100 Mal parallel laufen. Am besten wählbar weil die Datenbanken unterschiedlich sind und der Rechner nicht immer unter Volllast laufen soll.
Meine Frage wäre eben ob es in der Form mit den Threads funktionieren kann.
Statt Queue könnte ich glaube ich Synchronize nehmen. Keine Ahnung was davon in meinem Fall besser ist.
Bitte um erneute Rückmeldung.

[generic]

Das mit dem Youtube Videos können wir ändern ;-)
Ich werde mal schauen, dass ich den Vortrag von den Delphi-Tagen 2011 in ein Video verpacke.
Wenn ich ein Video mache, werde ich es in "Coding BOTT" veröffentlichen.

Die Folien hänge ich mal an, aber ohne die "Live"-Demos etwas ohne Informationen.