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.

[thkerkmann]

Hi Alzaimar,

geniale Klasse.

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.

oder fehlt mir da was ?

Gruss

[alzaimar]

stimmt auffallend...

Ich habe z.B. eine 'TDisplayJob' Klasse, die bestimmte Ergebnisse, die ein anderer Job ermittelt hat, irgendwo reinzeichnet. Der eine Berechnungsjob befüllt also die Jobliste mit einem Synchronized-Job zur Darstellung.

Um das in einem Job zu erledigen, müsste ich ein paar Sachen umstricken (Synchronized-Methode des worker threads als public deklarieren und Parameterliste der TJob.Execute-Methode ändern).

Ich mach mir darüber aber mal Gedanken, obwohl die Arbeitsteilung (Berechnung: Synchronized = False, Darstellung: Synchronized = True) ganz elegant ist.

Danke für den Hinweis.

[mschaefer]

Eigentlich hört sich das ganz interesssant an. Werd mir das morgen mal genauer anschauen..

Grüße // Martin

[Luckie]

Warum so kompliziert. Man kann auch Windows das ganze machen lassen: QueueUserWorkItem.

Delphi-Quellcode:

function Thread(p: Pointer): Integer; stdcall;
var
  LBIndex                : Cardinal;
  Cnt                    : Cardinal;
begin
  LBIndex := PThreadParams(p)^.LBIndex;
  TListbox(frmReceiver.FindComponent('Listbox' + IntToStr(LBIndex))).Items.Clear;
  for cnt := 0 to 29 do
  begin
    TListbox(frmReceiver.FindComponent('Listbox' + IntToStr(LBIndex))).Items.Add(IntToStr(cnt));
    sleep(100);
  end;
  result := 0;
end;

procedure TfrmReceiver.WMCOPYDATA(var msg: TWMCopyData);
var
  idx                    : Integer;
  ThreadParams           : PThreadParams;
begin
  idx := PMyRecord(msg.CopyDataStruct.lpData)^.idx;
  New(ThreadParams);
  ThreadParams.LBIndex := idx;
  if QueueUserWorkItem(@Thread, ThreadParams, WT_EXECUTEDEFAULT or 2 shl 16) = 0 then
    ShowMessage(SysErrorMessage(GetLastError));
end;

[WS1976]

Hallo alzaimar,

ich finde deinen Code Klasse! Ob das Ganze nötig ist oder nicht muss man sicherlich im Einzelfall überprüfen. Schon alleine um dahinterzukommen wie Threads funktionieren hat sich die Arbeit gelohnt.
sehr guter Code.
sehr gut kommentiert.
( aber schnapp jetzt bitte nicht über )

Gruss
Rainer

[alzaimar]

Zitat von WS1976:

ich finde deinen Code Klasse!

Danke

Zitat von WS1976:

Ob das Ganze nötig ist oder nicht muss man sicherlich im Einzelfall überprüfen.

Wir haben hier ein Messaging-System, wobei die Clients per UDP Änderungen an einem Datenbestand mitgeteilt bekommen. Das Problem ist, das die Daten sowohl in eine interne Struktur eingepflegt, als auch visualisiert werden müssen. Teilweise rattern 100te von Paketen pro Sekunde ein. Damit die Anwendung nicht einfriert, schiebt der Empfangs-Event des UDP-Clients die vorzunehmenden Änderungen in die Jobliste. Die Visualisierung geschieht im OnIdle der Anwendung. Weiterhin müssen die Pakete jedoch an andere Programmteile weitergeleitet werden: auch dafür haben wir einen Job definiert. So ist gewährleistet, das es zu keiner Zeit zu irgendwelchen Verzögerungen kommt, auch wenn es kurzzeitig zu UDP-Bursts kommt.

Wir hatten vorher den Workerthread im Rahmen einer garbage collection im Einsatz, wo Zigtausende von Objekten freigegeben werden mussten. Das dauert doch ein paar Sekunden. Also haben wir Jobs definiert, die jeweils einige hundert Objekte freigeben. Damit läuft die Anwendung wieder flüssig.

Zitat von WS1976:

Schon alleine um dahinterzukommen wie Threads funktionieren hat sich die Arbeit gelohnt.

Du hast mich durchschaut. Das war die Motivation für mich, diesen Code hier zu posten, weil viele hier Probleme mit Threads haben. Insbesondere die Verwendung der Semaphore als wirkungsvollere Alternative zum 'Resume/Suspend'-Mechanismus sowie das externe Antriggern des Threads ist es Wert, mal analysiert zu werden.

Weiterhin habe ich nur unglaublich komplexe Units zum Thema 'Worker threads' gefunden. Da dachte ich mir eben: Nö. Muss einfacher gehen.

Zitat von WS1976:

( aber schnapp jetzt bitte nicht über )

Geht nicht. Zu alt.

@Luckie: Oder so. Da lernt man aber nix über Threads.

[shmia]

Ist es nicht so, dass Exceptions innnerhalb von Threads ins Nirvana gehen?
Man könnte diese aber über ein Event an die Applikation weiterreichen.

Delphi-Quellcode:

Procedure TWorkerThread.Run;
// Kleine Hilfsprozedur, die den aktuellen Job ausführt. 
Begin
  Try
    If fCurrentJob.UsesCOMObjects and not fCOMInitialized Then Begin
      fCOMInitialized := True;
      CoInitialize(Nil)
    End;
    fCurrentJob.execute(Self);
  Except
     // die Exception wurde nicht innerhalb des Jobs behandelt
     if Assigned(FOnException) then
        // Event auslösen
        FOnException(fCurrentJob, ExceptObject);
  End;
  fCurrentJob.Free;
End;

Man könnte den TNopJob vermeiden, wenn man so ändert:

Delphi-Quellcode:

Procedure TWorkerThread.Execute;
// Arbeitsschleife des Workerthreads. Er wartet, bis die Semaphore aktiv ist.
// Dann wird der nächste Job aus der Jobliste extrahiert und abgearbeitet.
Var
  L: TList;
Begin
  While Not Terminated Do
    If WaitForSingleObject(fSemaphore, INFINITE) = WAIT_OBJECT_0 Then 
  Begin
     if Terminated then Exit; // <===== Ausstieg, falls Kill aufgerufen wurde
      L := fJobs.LockList;
      Try
   If l.count > 0 Then 
         Begin
     fCurrentJob := TWorkerThreadJob(L[0]);
     l.Delete(0);
   End;
      Finally
   fJobs.UnlockList;
      End;
       If fCurrentJob.Synchronized Then
   Synchronize(Run)
       Else
   Run;
    End;
End;

Procedure TWorkerThread.Kill;
Begin
  ReleaseSemaphore(fSemaphore, 1, Nil)   // <=====
  Terminate;
End;

[alzaimar]

Hi shmia,

Das mit dem Beenden ist so eine Sache: Deine Idee ist gut, hat aber eine Schwachstelle:
Du rufst Kill auf.
Kill ruft ReleaseSemaphore auf. Dann wird in der Execute-Methode WaitForSingleObjects verlassen und kein Job ausgeführt. Wenn nun wieder WaitForSingleObjects aufgerufen wird, *bevor* in der Kill-Methode die 2.Zeile ausgeführt wurde, hängt die Chose immer noch.

Wenn überhaupt, dann anders herum: Also erst Terminate und dann ReleaseSemaphore.
Ich habe mich jedoch dagegen entschieden:
1. TNopJob ist ein nettes Beispiel, wie man einen Job deklariert.
2. Ich mag es nicht, zentrale Funktionsaufrufe (wie hier: ReleaseSemaphore) an zwei Stellen im Code aufzurufen.
3. Die Abfrage 'If Terminated' ist immer (bis auf 1x) überflüssig.


Besser wäre es (finde ich), das Resultat von 'WaitForSingleObject' auszuwerten:

Delphi-Quellcode:

Procedure TWorkerThread.Execute;
...
Begin
  While Not Terminated Do
    Case WaitForSingleObject(fSemaphore, INFINITE) Of
      WAIT_ABANDONED :  // Passiert, wenn das Handle geschlossen wurde
        Break;
      WAIT_OBJECT_0 :  Begin
      ... // Hier Job extrahieren und ausführen
      End
  End;
End;

Procedure TWorkerThread.Kill;
Begin
  Terminate;
  CloseHandle (fSemaphore);
End;

Und den Destruktor anpassen. Ich habe es noch etwas komfortabler gemacht (siehe oben).

Das wäre eventuell akzeptabel.

Die Sache mit den Exceptions habe ich bewusst dem Job überlassen. Der soll mit Exceptions umgehen und weiss am Besten, was zu tun ist. Wenn ihn das nicht interessiert, dann muss er keinerlei Vorkehrungen treffen, den der Workerthread macht das.

Es schadet nicht, und wer's braucht... Also... wird eingepflegt.

*einpfleg* hmmm.. Wenn der Job nicht synchronisiert ist, aber in der Exception-Routine die VCL zur Darstellung verwendet wird, dann hängt die Anwendung bzw. schmiert irgendwann ab. Also muss man den Aufruf des Events synchronisieren. *hack* Fettich.

[kalmi01]

Zitat von alzaimar:

Also muss man den Aufruf des Events synchronisieren. *hack* Fettich.

Und wo iss "Fettich" ?