Workerthread: Der Diener im Hintergrund

[alzaimar]

Hallo,

Threads sind eine tolle Sache. Wenn man sie versteht. Wenn nicht, bringen sie einen an den Rand des Wahnsinns. Oder ein Stück weiter. Ich weiss, wovon ich rede.

Ich benötige für meine Applikationen immer mal wieder einen Thread, der ab und zu einzelne Jobs im Hintergrund ausführen muss. Wenn man z.B. auf bestimmte Ereignisse reagieren muss, diese Reaktion aber zeitaufwändig ist, dann kann die Ereignisbehandlung während der Ausführung des Eventhandlers blockiert sein.

Also habe ich einen generischen Thread implementiert, der einzelne Aufgaben (Jobs) im Hintergrund abarbeitet. Ich muss dazu nur den Job selbst definieren, alles Andere macht dieser Thread. Diese Threads nennen sich auch 'worker threads', weil sie eben ankommende Arbeiten ausführen.

So ein Job kann z.B. das Einlesen einer Tabelle von einer Datenbank sein, oder auch das Abspeichern von Daten.

Die Unit instantiiert einen Workerthread 'PendingJobs', den man sofort mit Arbeit befüllen kann. Dazu ruft man einfach die 'AddJob'-Methode auf:

Delphi-Quellcode:

...
PendingJob.AddJob (TMyJob.Create);
...

Die Anweisung schiebt den Job ans Ende der Jobliste und aktiviert ggf. den Thread (wenn der nicht sowieso schon aktiv ist). Der TMyJob wird dann 'demnächst' im Hintergrund ausgeführt.

Wenn man einen Job definiert, der mit COM-Objekten arbeitet (z.B. ADO) dann setzt man die Eigenschaft 'UsesCOMObjects' des Jobs auf True.


Der Thread verwendet eine Semaphore sowie eine Threadliste (die Liste der noch nicht ausgeführten Jobs). Auf Schnickschnack habe ich verzichtet, das kann jeder selbst einbauen.

Ich habe die Unit sowie eine kleine bescheuerte Demo bereitgestellt, die jedoch zeigt, wie einfach man Hintergrundarbeiten in seiner Anwendung implementieren kann.

Kritik, Verbesserungs- und Erweiterungsvorschläge sind ausdrücklich erwünscht.

Viel Spaß damit.

Update: Die Idee von thkerkmann wurde aufgenommen: Die Synchronize-Methode des Workerthreads ist nun public und kann innerhalb der TJob.Execute-Methode aufgrufen werden.

Update: Idee von shmia: Exception-Notification Event, und eine verbesserte Beendigung des Workerthreads. Man kann unn steuern, ob vor Beendigung alle ausstehenden Jobs noch abgearbeitet werden sollen oder nicht.

Update: Aus dem Thread wurde ein Threadpool (Idee von shmia). Er ist in einer wackeligen Version fertig. Er läuft in der Demo stabil, aber was heißt das schon. Ich würde Euch bitten, mal kräftig gegen die Klasse zu treten. Mal sehen, was sie aushält.

Der Threadpool verwaltet mehrere Threads (Anzahl einstellbar, auch zur Laufzeit), die gemeinsam die oben erwähnte Jobliste abarbeiten.

Update vom Threadpool: Schleifenvariablen als Cardinal deklariert, wenn Schleifenende = 'X-1', und X=0, bombt es.

Update vom Threadpool: Memoryleak beim Beenden beseitigt. Weiterhin stellt die Unit keine Instanz 'PendingJobs' mehr bereit. Denn das wurde im Finalisierungsabschnitt der Unit wieder freigegeben. Wenn die Classes-Unit ihren Finalisierungsabschnitt schon durchlaufen hat, gibt es Probleme.

Update vom Threadpool: Das Herabsetzen der Poolsize zur Laufzeit führte zu einem Speicherleck, ein fehlendes 'UnlockList' im der Clear-Methode der JobList und noch ein paar Kommentare.

Update vom Threadpool: x000x hat einen Bug gefunden: Die Setter-Methode des OnJobException-Events muss den Methodenzeiger natürlich an die einzelnen Threads weiterleiten. Weiterhin habe ich die Workerthread-Unit rausgenommen. Man braucht sie nicht mehr.

Update vom Threadpool: Die aktuelle Version läuft stabil, sie ist in einer Kundenanwendung im Einsatz.

Update vom Threadpool: Neue Methode, um aus einem Job heraus Nachrichten zu verschicken (siehe beiliegende Demo).

Update: Eigenschaft 'Terminated' des TWorkerThread wird publiziert, kleiner Fehler in der Clear-Methode beseitigt.

Update: Kleiner Hack, um den Threadnamen zu setzen (Debugginghilfe). Dank an Assertor.

[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?

[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 von alzaimar:

Wieso friemelst du noch mit D5?

Wieso "friemelt" man wenn man Delphi 5 benutzt

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.


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