ThreadPool 1.0.9 für Delphi 2010-XE

[WladiD]

Nachdem ich vergeblich nach einem Open-Source Thread-Pool für Delphi gesucht habe, welches meine Bedürfnisse abdeckt, musste ich mich dafür entschließen, einen eigenen zu schreiben. Diesen möchte ich euch hier vorstellen.

Vereinfacht kann man die Funktionsweise des ThreadPools wie folgt beschreiben:

Man verbindet sich aus einem beliebigen Thread mit dem spezifischen Manager (welcher ebenfalls ein eigenständiger Thread ist) und übergibt ihm eine Aufgabe. Der Manager übergibt die Aufgabe an einen neuen bzw. schlafenden Arbeiter oder packt sie auf die Warteliste, wenn bestimmte Grenzwerte es nicht zulassen die Aufgabe sofort abzuarbeiten. Der Arbeiter nimmt sich der Aufgabe an, arbeitet sie parallel ab und meldet sich über synchronisierte Ereignisse, wenn er fertig ist. Hat der Manager noch weitere Aufgaben anliegen, so bekommt der Arbeiter sofort den Nächsten. Sind alle Aufträge abgearbeitet, so kann der Manager eine bestimmte anzahl an Arbeitern schlafen legen (SpareWorkersCount) und die restlichen Arbeiter werden terminiert.


Hauptmerkmale

• Manager und Arbeiter sind eigenständige Threads
  
  
• Ein spezifischer Pool wird von den Basis-Klassen TPoolManager, TPoolWorker und TPoolTask abgeleitet, wobei nur wenige Methoden mit "Leben" gefüllt werden müssen.
  
  
• Für jeden abgeleiteten TPoolManager wird (nur) bei Bedarf eine eindeutige Instanz erstellt (verborgenes Singleton pattern)
  
  
• Die genaue Instanz eines spezifischen Managers ist unbekannt und muss über eine Verbindung mittels TMyPoolManager.Connect oder TMyPoolManager.Bind ermittelt werden
  
  
• Der Manager erstellt die Arbeiter gemäß der Auftragslage dynamisch unter berücksichtigung einer definierbaren Obergrenze (TMyPoolManager.ConcurrentWorkersCount)
  
  
• Die Arbeiter-Threads werden wiederverwendet. Das heißt, ist ein Arbeiter fertig und es liegen weitere Aufträge an, so bekommt er sofort den Nächsten.
  
  
• Anzahl an schlafenden Arbeitern kann über TMyPoolManager.SpareWorkersCount definiert werden
  
  
• Aufträge können einen Eigentümer haben
  
  
• Für jeden Eigentümer können Event-Handler an die Ereignisse OnTasksStatus und OnTasksComplete gebunden werden
  
  
• Aufträge können gezielt abgebrochen werden, z.B nach dem Eigentümer (TMyPoolManager.CancelTasksByOwner)
  
  
• Jeder Auftrag verfügt über die Basis-Ereignisse: OnStart, OnCancel, OnDone und kann um beliebige weitere ergänzt werden
  
  
• Aufträge können priorisiert werden
  
  
• Der Manager unterstützt den Demand-Mode: D.h. er beendet sich selbst, wenn keine Aufträge anliegen und keine Arbeiter auf Vorrat gehalten werden müssen.
  
  
• Alle instanzierten Thread-Pools (Manager und Arbeiter) können mit einer einzelnen Anweisung (z.B. bei Programmende) sauber terminiert werden: TPoolManager.TerminateAllRunningPools
  
  
• Die Freigabe eines zuvor registrierten Eigentümer kann an alle Thread-Pools mittels TPoolManager.DispatchOwnerDestroyed übermittelt werden
  
  
• Dynamsiche Anpassung von gleichzeitigen und auf Vorrat zu haltenden Arbeitern an das Ziel-System mittels TMyPoolManager.ConcurrentWorkersCountPerCPU und TMyPoolManager.SpareWorkersCountPerCPU
  
  
• Last but not least: Kein Polling.
  Die Manager und Arbeiter lauschen mittels WaitFor(INFINITE) an ihren Hauptsignalen, dadurch wird kostbare Rechnerkapazität nicht sinnlos vergeudet.

Download

Damit ich das Archiv nicht bei jedem Update hier aktualisieren muss, verlinke ich am liebsten zur Projektseite bei SourceForge:

ThreadPool for Delphi (Download | Git-Repository)

Das Archiv enthält bereits zwei ausführliche Beispiele (auch in kompilierter Form), sodass man sowohl die Implementierung eines spezialisierten Thread-Pools nachvollziehen kann als auch dessen Verwendung.

Voraussetzung

Die Unit ist ab Delphi 2010 lauffähig.

Lizenz

Wie ich in der Einleitung geschrieben hatte, suchte ich nach einer Open-Source Lösung, die es nicht gab. Aus diesem Grunde stelle ich meine unter die MPL 1.1.

mfg

[WladiD]

Habe eine ungeteste Anpassung für Delphi 2009 eingebaut und ins Repository eingespielt:

ThreadPool (Revision 11)

Musste die bewährte "Compilers.inc" von Mike Lischke einbinden.

Kann das jemand testen und mir mitteilen, ob es sich damit jetzt compilieren lässt?

Dank im Voraus!

[Bernerbaer]

Danke für Deine Änderungen, aber:

ThreadPool lässt sich nun ohne Fehlermeldung kompilieren.

Die beiliegenden Demos stürzen jedoch ab, im folgenden die ersten Fehlermeldungen mit dem angezeigten Codeabschnitt:

Code:

function IsTaskValid(Task:TPoolTask):Boolean;
   begin
      Result:=Owner.Tasks.IndexOf(Task) >= 0;
   end;

->Erste Gelegenheit für Exception bei $76EDFBAE. Exception-Klasse EAccessViolation mit Meldung 'Zugriffsverletzung bei Adresse 00475AF1 in Modul 'PrimePoolProject.exe'. Lesen von Adresse 00000000'. Prozess PrimePoolProject.exe (4432)

Code:

procedure TPoolManager.WorkerTerminated(TerminatedWorker:TPoolWorker);
begin
   BeginWriteWorkers;
   try
      Workers.Remove(TerminatedWorker);
   finally
      EndWriteWorkers;
      MainSignal.SetEvent;
   end;
end;

-> Erste Gelegenheit für Exception bei $76EDFBAE. Exception-Klasse EAccessViolation mit Meldung 'Zugriffsverletzung bei Adresse 00475AF1 in Modul 'PrimePoolProject.exe'. Lesen von Adresse 00000000'. Prozess PrimePoolProject.exe (4432)

[WladiD]

Zitat von Bernerbaer:

Die beiliegenden Demos stürzen jedoch ab...

Kommt der Fehler sofort nach dem Programmstart oder erst wenn Sie auf den Button "Add tasks" klicken. Gibt es irgendeine Ausgabe im Memo, kommt die AV sporadisch oder immer an der selben Stelle. Oder treten die Fehler gar in meinen Kompilaten auf?

Fragen über Fragen. Hätte ich mir doch gleich denken können, dass man für keine "Plattform" programmiert, die man selbst nicht hat/testen kann.

[Bernerbaer]

Antworten

1. Der Fehler tritt erst auf wenn Add Task geklickt wird, der Fehler tritt erst am beim letzen (?) Durchlauf der Schleife for c:= 0 to Steps - 1 auf
2. Nein, im Memo werden keine Einträge angelegt
3. Die beiliegenden Kompilate der Demos laufen problemlos
4. Der Fehler tritt bei den vom mir kompilierten Demos immer auf.

[WladiD]

Zitat von Bernerbaer:

Antworten

1. Der Fehler tritt erst auf wenn Add Task geklickt wird, der Fehler tritt erst am beim letzen (?) Durchlauf der Schleife for c:= 0 to Steps - 1 auf
2. Nein, im Memo werden keine Einträge angelegt
3. Die beiliegenden Kompilate der Demos laufen problemlos
4. Der Fehler tritt bei den vom mir kompilierten Demos immer auf.

Generics/Closures waren neu in D2009 = alpha
Generics/Closures wurden in D2010 verbessert = beta
Generics/Closures wurden in DXE weiter verbessert = stable?

So oder so ähnlich wird es wohl sein. Genau da vermute ich den Fehler. Daher werde ich das bisher versuchte wieder droppen und nicht weiter probieren etliche Workarounds einzubauen, um mit D2009 kompatibel zu sein. Der ganze ThreadPool sollte ursprünglich komplett auf Generics basieren, welchen man nur die spezielle Task-Klasse übergibt, doch dort scheiterte ich, weil der Support der Generics noch nicht fehlerfrei ist, damit will ich sagen, dass ich schon mit 2010er meine Probleme hatte. So ist das mit neuen Möglichkeiten, doch irgendwann muss man sie nutzen um einen Nutzen zu erlangen.

Nichtsdestotrotz ist es Open Source, also wenn das jemand hinmoddeld, dass es funktioniert, dem ist ein Eintrag unter Credits/Copyrights sicher.

[WladiD]

Soeben habe ich die neue Version 1.0.1 veröffentlicht. (Links siehe 1. Beitrag)

Neben einigen Optimierungen enthält es jetzt die Unterstützung für priorisierbare Tasks. Ableitende Tasks können die Eigenschaften TPoolTask.Priority oder TPoolTask.PriorityRaw veröffentlichen und der entsprechende Manager muss die Eigenschaft TPoolManager.SortTasks auf TRUE setzen um die Sortierung zu aktivieren.

Wenn mehr Aufgaben hinzugefügt werden, als gleichzeitig abgearbeitet werden können, so kommen die Tasks mit der höheren Priorität eher dran.

Ach ja, als Voraussetzung bleibt: ab Delphi 2010

[mquadrat]

Was vielleicht noch für den einen odere anderen nützlich sein könnte, wären Abhängigkeiten der Aufträge, so dass B erst nach A ausgeführt wird.

Prinzipiell ist die Sache super. Für mich zwar mangeld D2010/XE nicht nutzbar, aber es zeigt, dass die Community zumindest noch nich ganz kapituliert hat

[WladiD]

Zitat von mquadrat:

Was vielleicht noch für den einen odere anderen nützlich sein könnte, wären Abhängigkeiten der Aufträge, so dass B erst nach A ausgeführt wird.

Das ist schon ohne weiteres Zutun möglich:

Delphi-Quellcode:

var
  Task:TMyTask;
begin
  Task:=TMyTask.Create(Owner);
  // Task mit Daten füllen
  Task.OnDone:=procedure(Sender:TObject)
  var
    DependTask:TMyTask;
  begin
    if TMyTask(Sender).State <> tsSuccess then
      Exit;
    DependTask:=TMyTask.Create(Owner);
    // DependTask mit Daten füllen

    TMyPoolManager.AddTask(DependTask);
  end;
  
  TMyPoolManager.AddTask(Task);
end;

Zitat von mquadrat:

... die Community zumindest noch nich ganz kapituliert hat"

Ich hoffe, dass diese Aussage sich nicht auf die These "Delphi ist tot" bezieht. Native (abwertend auch unmanaged) Code wird auch in 20 Jahren eine Rolle spielen. Beides muss sich nicht zwangsweise ausschließen. Hier ist eine ausführlichere Antwort: www.isdelphidead.com ?

[mquadrat]

Auf die Idee mit dem Event-Handler war ich auch schon gekommen. Die Abhängigkeit mit in das Interface zu integrieren, würde weniger Glue-Code produzieren und man könnte Workflows fast 1:1 abbilden.

Natürlich wird Native-Code noch lange eine Rolle spielen. Aber IMHO werden es immer weniger Community-Projekte und 3rd-Party-Bibliotheken/Komponenten. Und schließlich waren die ganzen Projekte rund um Delphi lange ein Alleinstellungsmerkmal.

[WladiD]

Hallo alle zusammen,

heute habe ich die neue Version 1.0.3 des ThreadPool veröffentlicht (Download-Links wie gehabt im 1. Beitrag).

Die ThreadPool-Unit wurde massiv rationalisiert. Es werden jetzt weniger Locks benötigt und die ganze Unit konnte (trotz höherer Versionsnummer) etwas schrumpfen.

In der Version 1.0.2, die ich hier nicht vorstellte, gab es eine CodeSite-Integration, die mittels Compiler-Conditionals in der beiliegenden Compile.inc aktiviert werden kann. Es ist doch ziemlich schwierig soetwas mittels Breakpoints/OutputDebugString zu debuggen.

Desweiteren kam ein DUnit-Testprojekt hinzu, in dem —unter anderem– diverse Stress-Tests stattfinden, die vor jedem Release bestanden werden müssen. Es ist ebenfalls dem Archiv beigelegt.


Ich wünsche allen ein erfolgreiches neues Jahr!

P.S.: Ja, ich bin auf Delphi XE umgestiegen und bin sehr glücklich damit . Da ich D2010 parallel betreibe, ist (und hoffentlich bleibt) der ThreadPool weiterhin damit kompatibel.