Workerthread: Der Diener im Hintergrund

[alzaimar]

Sag mal, dein 'P4 HT' hat Hyperthreading? Hmmm. Dann gibt es vermutlich irgendwo einen Konflikt, bzw. einen Deadlock, der bei Einprozessorsystemen nicht auftritt.

Zitat:

Zitat von kalmi01

... und ich möchte mal behaupten, das die Version (subjektiv) langsamer ist als der Vorgänger.

Was ist langsamer? Da wird doch nicht viel gemacht?

Funktioniert denn der 'Terminate' Button? Kannst Du danach weiter testen oder hängt dann die SW wieder? Der Terminate-Button ruft 'Free' auf und instantiiert den Pool anschließend neu.

Was siehst Du eigentlich im Memo, wenn Du die Anwendung beendest?

[kalmi01]

Zitat:

Was ist langsamer? Da wird doch nicht viel gemacht?

Das ist mir wohl klar.
Auf dem einen PC rasen die Threads einfach nur so durch (Uni-Proz).
Auf dem anderen dauert es ca. 1 Sek. pro Thread.
Der Effekt tritt auch nur in der Pool-Variante auf.

Wenn ich das Prog aus der IDE starte, dauert es fast 2 Minuten, bis es mit der Initialisierung fertig ist.

Der normale Worker funktioniert auf beiden Systemen sauber !

Zitat:

Funktioniert denn der 'Terminate' Button?

Ja

Zitat:

Was siehst Du eigentlich im Memo, wenn Du die Anwendung beendest?

Was soll ich denn sehen ?
Bei einer Anwendung die hängt, passiert halt nicht viel.

Es gibt seltsamer Weise auch keinen Log von Eurekalog.

Laut Taskmanager liegt WorkerThreadPoolDemo auch nur gelangweilt in der Gegend rum.

[alzaimar]

Hi kalmi01,

Schön, das Du Dir die Zeit nimmst.

Ich erwarte bei der Pool-Demo, das alle 500 ms die Threads quasi gleichzeitig den nächsten Job ausführen. Denn der Job besteht ja nur aus einem 'Sleep(500)' und 'Log ins Memo'.

Wenn das auf dem HT-PC anders abläuft, dann wundert mich das (ich würde sogar postulieren, das Du ein anderes Programm ausführst :stupid: )

Was die Ausgabe anbelangt, bezog ich mich auf das Testprog, das ich Dir per PN zugeschickt habe.

Eine neue Version mit Testausgaben (per Kompilerdirektive abschaltbar) ist im 1.Post

[kalmi01]

Zitat:

Zitat von alzaimar

Wenn das auf dem HT-PC anders abläuft, dann wundert mich das (ich würde sogar postulieren, das Du ein anderes Programm ausführst :stupid:

Da könntest Du nicht ganz unrecht haben :gruebel:
Mit dem neuen Source von Dir funktioniert es einwandfrei :thumb:

Der neu eingestellt Source hat 24 Unterschiede in der csWorkerThreadPool.pas, gegenüber dem per PN geschickten.

[alzaimar]

Na, Schlamperei meinerseits (und wie gesagt: Da war noch so eine Kleinigkeit :oops: )


Also, heute abend mach ich den Code nochmal sauber, kommentiere es noch ausführlich (wenn mir noch etwas einfällt) und stell eine überarbeitete Version hier ein.

Danke für die Geduld und die Tipps!

[x000x]

Moin moin,

ich denke ich habe noch eine Kleinigkeit gefunden?!
Sollte es jetzt nicht in TWorkerThreadPool einen Setter für fOnJobException
geben? Also ähnlich wie SetOnAction...

Dann hab ich noch ne Frage:
Angenommen, ich habe verschiedene Jobs die DB-Abfragen o.ä. erledigen.
Nach meinem Verständnis muss ich in jedem Job die TADOConnection erstellen.
Hab ich das jetzt richtig verstanden oder langt es, wenn ich Sie einmal
in z.B. einem Datenmodul einzufüge? (Wenn ja, dann hätte ich dafür gerne eine Erklärung)

[alzaimar]

Hi x000x,

Super! Da fehlte wirklich der Setter, der das Event an die einzelnen Threads weiterleitet... ist eingebaut.

ADO verwendet COM, und COM ist nicht ohne Weiteres threadübergreifend. Die Interfaces sind nur im Thread gültig. Willst Du sie einem anderen Thread zur Vergfügung stellen, musst Du sie über den Marshaller übergeben (ab in einen Stream, raus aus einem Stream). Irgendwie so.

ADO verwaltet jedoch einen Connection-Pool, der das vermutlich alles regelt.

Denk dran: TMyJob.UsesCOMObjects muss bei Verwendung von ADO auf True gesetzt werden, damit CoInitialize aufgerufen wird.

[kalmi01]

Moin moin,

Zitat:

Zitat von alzaimer

Da fehlte wirklich der Setter, der das Event an die einzelnen Threads weiterleitet... ist eingebaut.

Du lügst ja :wink:
Ich hätte da mal einen Vorschlag: Schreib doch beim Download in den Kommentar, von wann die Version ist :stupid:

Edit: Uuund ? Wo bleibt die Korrektur ? :mrgreen:

[alzaimar]

Zitat:

Zitat von kalmi01

Ich hätte da mal einen Vorschlag: Schreib doch beim Download in den Kommentar, von wann die Version ist :stupid:

Das würde ein Mindestmaß an Planung und Ordnung bedeuten. Das liegt außerhalb meiner Möglichkeiten. :freak:

[kalmi01]

Zitat:

Zitat von alzaimar

Zitat:

Zitat von kalmi01

Ich hätte da mal einen Vorschlag: Schreib doch beim Download in den Kommentar, von wann die Version ist :stupid:

Das würde ein Mindestmaß an Planung und Ordnung bedeuten. Das liegt außerhalb meiner Möglichkeiten. :freak:

Das Genie beherrscht das Chaos :cheers:

[stoxx]

siehe auch hier:

http://hallvards.blogspot.com/2008/0...n-threads.html

[Adamo]

Ich habe bei mir die Demo aus dem ersten Post kompiliert und gestartet (Delphi 5, Windows XP). Es läuft (ich musste lediglich die Unit Variants entfernen)
Doch leider tut sich irgendwie nichts.

In der Memo steht lediglich "500 Jobs generiert". Aber die Abarbeitung der Jobs startet nicht automatisch?


P.S. Hab es gerade auf einem anderen PC mit Delphi6 ausprobiert. Dort geht es. Aber warum geht es mit Delphi5 nicht?

[spaxxn]

Ist der/ sind die Threads vielleicht suspendiert?
Stichwort "Resume"

[alzaimar]

Bei D5 weiss ich es nicht, aber D6 hatte bis zu, 2.Update eine Macke im Synchronize.

Wieso friemelst du noch mit D5? :gruebel:

[Adamo]

Ich bin leider noch an Delphi 5 gebunden, weil wir ein altes und großes Projekt damit pflegen müssen.
Blöderweise brauche ich die Funktionalität in Delphi 5 :?

[x000x]

Moin moin,

Zitat:

Zitat von alzaimar

Wieso friemelst du noch mit D5? :gruebel:

Wieso "friemelt" man wenn man Delphi 5 benutzt :gruebel:

Meine Vermutung ist, dass die Sourcen im ersten Post nicht die aktuellste Version ist. Zum einen wird dort das setzen
der OnJobException ignoriert/nicht berücksichtigt - zum anderen wird dort nur "500 Jobs generiert" ausgegeben, weil die
Jobs eine Exception werfen: Abstakter Fehler...
(Bekommst du raus, wenn du den Quelltext so änderst, dass OnJobException auch zugewiesen wird)

Ich weiss(!), dass es auch mit D5 funktioniert hat, hatte halt damals den Thread verfolgt...
Konnte nur leider keine Sourcen mehr davon finden :-/

[Adamo]

an diser Stelle war ich auch gerade. Ich habe den Setter für fOnException in der Unit eingebaut und DebugUnit definiert.
Folgendes kommt also raus (bei 5 Threads):

meInfo
500 jobs generiert
Job received
Job received
Job received
Job received
Job received
Abstrakter Fehler
Abstrakter Fehler
Abstrakter Fehler
Abstrakter Fehler
Abstrakter Fehler


Die Exception kommt bei dieser Funktion. Beim Execute.

Delphi-Quellcode:

Procedure TWorkerThread.Run;
// Kleine Hilfsprozedur, die den aktuellen Job ausführt.
// Im Falle einer Exception wird (falls angegeben) ein Event aufgerufen.
Begin
  Try
    If fCurrentJob.UsesCOMObjects And Not fCOMInitialized Then Begin
      fCOMInitialized := True;
      CoInitialize(Nil);
    End;
    fCurrentJob.execute(Self);
  Except
    On E: Exception Do Begin
      fCurrentException := E;
      If Assigned(fOnJobException) Then
        Synchronize(DoOnException);
    End;
  End;
  fCurrentJob.Free;
End;

P.S. Ich habs :-)
Weil die Methode als virtual; abstract deklariert war, konnte sie nicht von der selben Klasse aufgerufen werden, in der sie deklariert war. Daher der abstrakte Fehler. Ich habe jetzt die Funktion als virtual deklariert und einen leeren Funktionsrumpf angelegt. Jetzt gehts.

Delphi-Quellcode:

Procedure Execute(aThread: TWorkerThread); virtual;

Delphi-Quellcode:

Procedure TWorkerThreadJob.Execute;
Begin
 
End;

[x000x]

Moin moin,

denke mal der "Fehler" ist nicht die deklaration der Execute Methode als virtual; abstrakt;, denn diese soll ja in einer
abgeleiteten Klasse überschrieben werden.
Denke mal der Aufruf aus der TSampleJob.Execute in der wpMain mit inherited ist der Fehler.
Nimmst du das raus, so läuft es auch.

[Adamo]

Gut..es läuft :)

Wie finde ich am besten raus, wie viele Threads im Pool "gerade" beschäftigt sind?

[Assertor]

Hi alzaimar,

ein sehr schöne Kapselung für WorkerThreads. Tolle Arbeit!

Das einzige, was ich vermisse, ist eine Eigenschaft aThread.Terminated innerhalb der Jobs. Gerade wenn man länger dauernde Jobs laufen läßt, könnte man hiermit Verzögerungen beim Beenden vermeiden.

Edit: Das wär auch sehr einfach zu lösen, da der TWorkerThread ja ein Nachkomme von TThread ist. Einfach das property einfügen und gut ist.

Delphi-Quellcode:

TWorkerThread = Class(TThread)
  private
    fCurrentException: Exception;
    fCOMInitialized: Boolean;
    fCurrentJob: TWorkerThreadJob;
    fOnJobException: TJobExceptionEvent;
    fJobList: TWorkerThreadJobList;
    fID: Integer;
    fOnJobNotify: TJobMessageEvent;
    fJob: TWorkerThreadJob;
    fMessage: string;
{$IFDEF DebugUnit}
    fOnAction: TMessageEvent;
    fMsg: String;
    Procedure DoAction;
{$ENDIF}
    Procedure Run;
    Procedure DoOnException;
    procedure DoNotify;
  protected
    Procedure Execute; override;
  public
    Constructor Create(aID: Integer; aJobList: TWorkerThreadJobList);
    Destructor Destroy; override;
    Procedure Notify (aJob : TWorkerThreadJob; Const aMessage : String);
    Procedure Synchronize(Method: TThreadMethod);
    Property ID: Integer read fID;
    Property OnJobNotify : TJobMessageEvent read fOnJobNotify Write fOnJobNotify;
{$IFDEF DebugUnit}
    Property OnAction: TMessageEvent read fOnAction write fOnAction;
{$ENDIF}
    Property OnJobException: TJobExceptionEvent read fOnJobException write fOnJobException;
    Property Terminated; // eingefügt
  End;

Gruß Assertor

[alzaimar]

Hi Assertor,

Ich bau das mal ein und stelle es im 1.Post wieder rein.

Edit: Achso, äh.. :oops: Danke für das Kompliment.

[Assertor]

Hi Alzaimar,

Zitat:

Zitat von alzaimar

Edit: Achso, äh.. :oops: Danke für das Kompliment.

Bitte, gerne :)

:dp:

[Assertor]

Hi,

noch eine kleine Verbesserung, um das Debugging in neueren Delphi Versionen zu vereinfachen:

Delphi-Quellcode:

--- D:/Downloads/csWorkerThreadPool.pas   Sat Nov 29 15:30:22 2008
+++ D:/csWorkerThreadPool.pas   Sun May 17 01:38:13 2009
@@ -113,6 +113,7 @@
     fMsg: String;
     Procedure DoAction;
 {$ENDIF}
+    procedure SetName;
     Procedure Run;
     Procedure DoOnException;
     procedure DoNotify;
@@ -265,6 +266,35 @@
 
 { TWorkerThread }
 
+{$IFDEF MSWINDOWS}
+type
+  TThreadNameInfo = record
+    FType: LongWord;     // must be 0x1000
+    FName: PAnsiChar;    // pointer to name (in user address space)
+    FThreadID: LongWord; // thread ID (-1 indicates caller thread)
+    FFlags: LongWord;    // reserved for future use, must be zero
+  end;
+{$ENDIF}
+
+procedure TWorkerThread.SetName;
+{$IFDEF MSWINDOWS}
+var
+  ThreadNameInfo: TThreadNameInfo;
+{$ENDIF}
+begin
+{$IFDEF MSWINDOWS}
+  ThreadNameInfo.FType := $1000;
+  ThreadNameInfo.FName := 'csWorkerThread';
+  ThreadNameInfo.FThreadID := $FFFFFFFF;
+  ThreadNameInfo.FFlags := 0;
+  try
+    RaiseException($406D1388, 0, sizeof(ThreadNameInfo) div sizeof(LongWord),
+      @ThreadNameInfo);
+  except
+  end;
+{$ENDIF}
+end;
+
 Constructor TWorkerThread.Create(aID: Integer; aJobList: TWorkerThreadJobList);
 Begin
   Inherited Create(True);
@@ -302,6 +332,7 @@
 
 Procedure TWorkerThread.Execute;
 Begin
+  SetName;
   While Not Terminated Do Begin
     fCurrentJob := fJobList.WaitForNextJob;
     If Assigned(fCurrentJob) Then Begin

Das benennt die Threads dann ordentlich, so daß man diese im Debugger auch erkennt :) Ich weiß aus dem Kopf nicht, ob die Exception für benannte Threads < Delphi 2006 Probleme macht, vielleich packst Du noch ein IFDEF rein, damit es nur > Delphi 2006 verwendet wird.

Gruß Assertor

[alzaimar]

:gruebel: Davon von den @@ Tags habe ich noch nie etwas gehört. Kannst Du mir kurz erklären, worum es dabei geht?

[Win32.API]

Das ist nur eine Darstellungsform von Unterschieden zwischen zwei Versionen.

Zitat:

++ -> Datei 1
-- -> Datei 2
+ -> Hinzugefügt
- -> Gelöscht
@@ -> Vieleicht Zeilen angaben?

Also alles, wo ein + vorsteht, ist neu und muss ergänzt werden.

--win32

[alzaimar]

Aha. Ich habe das mal eingebaut und geprüft. Delphi 6 meckert bei der Ausführung mit einer 'external Exception', läuft ansonsten durch. Ich habe das dann aber nur für Delphi 2006 und höher aktiviert.

Zur Ermittlung der Delphi-Version habe ich 'JEDI-SDL.INC' verwendet. Wenn es eine andere/bessere Möglichkeit gibt, die Delphi-Version zu ermitteln, immer her damit.

Update wie immer im 1.Post.

[Assertor]

Hi,

@Win32.API: Danke, warst schneller mit dem Erklären ;)

Zitat:

Zitat von alzaimar

Aha. Ich habe das mal eingebaut und geprüft. Delphi 6 meckert bei der Ausführung mit einer 'external Exception', läuft ansonsten durch. Ich habe das dann aber nur für Delphi 2006 und höher aktiviert.

Zur Ermittlung der Delphi-Version habe ich 'JEDI-SDL.INC' verwendet. Wenn es eine andere/bessere Möglichkeit gibt, die Delphi-Version zu ermitteln, immer her damit.

Ich nutze immer gerne die Compilers.inc von Mike Lischke und Jim Kueneman. Das ist die Include für die VirtualTreeView. Link zum Source (steht unter der MIT Lizenz): http://new.lischke-online.de:8080/br.../Compilers.inc

Dann einfach im Source um alle Änderungen ein

Delphi-Quellcode:

{$IFDEF COMPILER_6_UP}...{$ENDIF}

Das mit der Änderungsdatei war gestern einfacher, um die geänderten Stellen zu finden - war etwas faul ;) Tools wie WinMerge (Bild) erstellen sowas und sind eine erhebliche Arbeitserleichterung.

Gruß Assertor

[alzaimar]

Gut, diese JEDI-SDL.Inc macht im Prinzip das Gleiche...

[Assertor]

Hi,

Zitat:

Zitat von alzaimar

Gut, diese JEDI-SDL.Inc macht im Prinzip das Gleiche...

Ja, nur die Lizenz der Compilers.inc ist noch liberaler. Für Projekte von Entwicklern, die keine JEDIs einsetzen vielleicht besser.

Außerdem nimmt die JEDI-SDL.inc soweit ich sehen kann auch noch andere Änderungen an den Compiler Einstellungen vor (z.B. {$F+,Q-,R-,S-,I-,A+}). Das braucht der Diener im Hintegrund nicht und führt u.U. - wenn projektweit - zu Probleme oder ungewünschten Einstellungen.

Bei der Compilers.inc werden wirklich nur die Compiler festgestellt, kein Rumdrehen an Einstellungen.

Gruß Assertor

[sundance]

@alzaimar:
Erst mal vielen Dank für deine Mühe, die du in diese Unit gesteckt hast. Ich bin gerade dabei, in einem von meinen kleinen "Projekten" davon zu profitieren.... ;-)

Ein paar Fragen/Anmerkungen hätte ich aber doch:

- Was bezweckst du mit der Methode TWorkerThreadPool.Notify?

- Wie würdest du es anstellen, auf die Abarbeitung aller Jobs in der List zu warten, um z.B. irgendwelche Controls zu aktualisieren oder eine Statusmeldung auszugeben? (Polling auf TWorkerThreadPool.PendingJobCount = 0 ist ja wohl nicht der beste Weg...)

- In diesem Thread war mal eine Anmerkung:

Zitat:

Eins hab ich aber nicht ganz verstanden:
Wenn man Synchronized auf true setzt, wird der ganze Job über Synchronize(Run) ausgeführt. Wird das nicht zu einer blockierenden Operation dann ? Dann kann ich doch eigentlich diese Arbeit gleich im Haupt-VCL-thread ausführen lassen.

Zitat:

stimmt auffallend...

Nun ist das aber immer noch so drin:

Delphi-Quellcode:

If fCurrentJob.Synchronized Then
  Synchronize(Run)
Else
  Run;

Läuft der synchronisierte Thread nun autark oder nicht?
(na ja, vielleicht habe ich deine Source auch noch nicht ganz verstanden...)

.sundance.

[alzaimar]

Hi,

der Job wird dann natürlich im Hauptthread ausgeführt. Ich verwende das z.B. für Jobs, die meine Controls aktualisieren: Ein Job (der im Hintergrund läuft) erzeugt dann zum Schluss einen synchronisierten Job, der irgendwelche Aktualisierungen im Formular durchführt.

Um das Ende aller Jobs mitzubekommen, kannst Du doch zum Schluss einen 'Ende-Job' in die Queue einfügen. Wenn das nicht geht, würde ich pollen. Woher soll der Pool auch wissen, das alle Jobs abgearbeitet wurden?

Mit Hilfe der TWorkerThread.Notify-Methode kann ein Job eine synchronisierte Nachricht verschicken. Diese wird über das ThreadPool.OnNotify-Ereignis verschickt. Siehe Demo.

[Viktorii]

Moin zusammen.

Ich versuche gerade die Funktionsweise nachzuvollziehen und habe da ein paar Problem.

Ich hoffe die Frage ist jetzt nicht allzu blöd.

Wie kann die Execute Methode von TWorkerThread Parameter haben?

Execute wird ja nie direkt aufgerufen. Und trotzdem 'kennt' die Execute Methode aThread. Wie kann das also funktionieren?

Vielen Dank

[alzaimar]

Zitat:

Zitat von Viktorii

Wie kann die Execute Methode von TWorkerThread Parameter haben?

Die Execute-Methode des Workerthreads hat keine Parameter, sondern nur der Job. Der Job ist einfach nur eine Klasse, um eine einzelne Aufgabe zu kapseln.

[doctor-x]

Hallo zusammen,

erst mal ein dickes Lob; TWorkerThreadPool finde ich eine echt klasse und einfach zu verwendende sache! 8-)

Habe dazu aber mal eine frage...

Wenn ich einen Job erstelle und starte, wie kann ich im Job auf ein schließen des Programms mit einem abbruch des Jobs reagieren ohne das ganze mit "Synchronized := True;" auszuführen?

Habe es mal mit "Synchronized := True;" und der abfage "Application.Terminated" gemacht, aber bei "Synchronized := False;" hab ich ja keinen sicheren zugriff auf den Main-Thread.

Habe ich irgenwie auf den Thread der meinen Job bearbeitet zugriff? (für die abfrage "Thread.Terminatd")

Danke schon mal im voraus!

[Thom]

Hallo alzaimar,

Danke für die Arbeit!
Da die Units aber nicht mehr ganz taufrisch sind, werden in der inc-Datei Delphi 2010 und Delphi XE (noch) nicht berücksichtigt, so daß keine Benennung der Threads erfolgt. Das hat zwar keinen Einfluß auf die Arbeit des Pools - mir ist das nur beim Debuggen meines Programmes aufgefallen, daß da was fehlt... :wink: