WorkerThread erweitern

[SvB]

Hi, ich arbeite mit dem WorkerThread und es funktioniert alles wunderbar. Nun habe ich folgende Besonderheit, die ich implementieren möchte und wollte erst hier mal anfragen, ob meine Überlegungen funktionieren könnten, bevor ich rum bastele. Ich habe mir jetzt einige Zeit die Unit genau angesehen und ich denke, dass ich es einigermaßen verstanden habe, was da passiert.

Ich habe einen WorkerThreadPool mit einer PoolSize von 5, d.h. es werden 5 Threads erzeugt, die "gleichzeitig" 5 Jobs abarbeiten können (habe ich verstanden). Nun benötige ich aber, dass ein bestimmter Typ von Job nur einmal "gleichzeitig" abgearbeitet werden kann und die anderen Jobs gleichen Typs solange in der Warteschlange bleiben, bis der aktuelle Job abgearbeitet worden ist. Hier meine Überlegung:

Variante 1:
Ich erweitere TWorkerThreadJob um eine Feld fMaxSimultaneous und setze das beim Erzeugen meins speziellen Jobs auf 1, der Standardwert ist 0 und dann läuft alles wie gehabt.
Dann erzeuge ich mir eine Klasse "TWorkerThreadJobListWorking", in der die Jobs drinne stehen, die gerade abgearbeitet werden. Dadurch, dass in "WaitForNextJob" der Job aus der Liste gelöscht wird, der als nächstes abgearbeitet werden soll, füge ich diesen Job in "TWorkerThread.Execute" in "TWorkerThreadJobListWorking" hinten ran. Bevor ich den hinten ran hänge, prüfe ich, ob ein Job selben Typs da schon drin ist, wenn nein, dann füge ich ihn hinzu, wenn ja, dann füge ich den Job wieder ans Ende der "normalen" Job List. Wenn "Run" abgearbeitet wurde, dann nehme ich den Job aus "TWorkerThreadJobListWorking" wieder raus. Der Nachteil hierbei wäre, dass ein Job, der in der Liste oben steht, dann auf einmal wieder ganz unten steht.

Variante 2:
Ich erweitere TWorkerThreadJob um eine Feld fMaxSimultaneous und setze das beim Erzeugen meins speziellen Jobs auf 1, der Standardwert ist 0 und dann läuft alles wie gehabt. Dann noch ein Feld fWorking: Boolean.
In WaitForNextJob wird der "neue" Job nicht mehr aus der Liste gelöscht. Ich durchlaufe per Schleife die Liste und prüfe, welcher der nächste Job ist der fWorking = False hat. Dann prüfe ich, ob dieser Job fMaxSimultaneous = 1 hat und prüfe dann noch mal die Liste ob schon ein Job selben Typs fWorking = True hat. Wenn das nicht der Fall ist, dann wird fWorking auf True gesetzt und der Job zurück gegeben. Wenn ich "TWorkerThread.Execute" das "Run" abgearbeitet wurde, dann wird der Job aus der Liste entfernt.

Was denkt Ihr, was der bessere Ansatz wäre? Oder habt Ihr eventuell noch eine andere Idee?

Danke

[Assarbad]

Warum nicht Mutex oder kritischen Abschnitt benutzen?

[SvB]

Tja, da fängt es dann schon an, ich bin nicht unbedingt der Crack.
Da muss ich dann mal schauen, für was das gut ist und was ich damit machen kann, um zu meiner Lösung zu kommen.
Das werde ich dann aber wohl erst morgen machen.

[Assarbad]

Okay, kurz angerissen. Du würdest bspw. TryEnterCriticalSection (ab NT4) benutzen um einen kritischen Abschnitt zu betreten. Schlägt es fehl, würde der Thread nix machen, wäre also für den nächsten Auftrag frei. Funktioniert es, wärst du sicher allein in dem kritischen Abschnitt zu sein, so wie du es wolltest.

Nähmen wir mal an, daß du mehr als eine Auftrag brauchst, aber die Anzahl dennoch unter der Anzahl der Threads halten willst, würden sich Semaphoren anbieten.

[SvB]

Danke erst mal. Da das für mich Neuland ist, muss ich mich in die Materie erst mal etwas einlesen.

Ich muss halt sicherstellen, dass folgendes funktioniert: ThreadPool mit z.B. 5 Threads. Dann habe ich JobA, JobB, JobC. JobA macht was in der Datenbank, JobB ließt Dateien aus einem Ordner analysiert diese und schreibt bestimmte Daten in die DB, JobC macht auch irgend etwas.
Jetzt ist es so, dass ich 2x JobA erzeuge und in die JobListe hinzufüge. Der WorkerThread verteilt die dann ja auf zwei Threads und diese werden bearbeitet. Dann schiebe ich noch 5x JobB in die JobListe und normalerweise würden dann drei davon auf die verbleibenden wartenden drei Threads aufgeteilt, die zwei restlichen Jobs bleiben in der Liste, bis wieder ein Thread frei ist. Es soll aber so sein, dass JobB nicht auf die 3 verbleibenden Thread aufgeteilt werden, sondern festgestellt wird, dass JobB immer nur einzeln abgearbeitet werden, d.h. es wird nur ein weiterer Thread mit einem Auftrag belegt. Es bleiben dann noch zwei wartende Threads übrig, die dann noch weitere eintreffende JobsA oder JobsC abarbeiten könnten. Wenn dann der eine JobB abgearbeitet wurde, kann den nächste JobB aus der Liste geholt werden usw.

Ich habe mal grob Deine Anhaltspunkte durchgestöbert und ich denke, dass ich mit den Semaphoren da am besten weiterkomme. Ich setze mich einfach mal dran und probiere das aus.

[SvB]

Wie gesagt, für mich ist es im Moment schwierig zu entscheiden, wie ich vorgehen könnte, da ich bisher wede etwas mit Mutex, CriticalSection, Semaphoren gemacht habe.
Ich muss mich erst mal in das Thema einarbeiten.

[Assarbad]

Zitat von SvB:

Wie gesagt, für mich ist es im Moment schwierig zu entscheiden, wie ich vorgehen könnte, da ich bisher wede etwas mit Mutex, CriticalSection, Semaphoren gemacht habe.
Ich muss mich erst mal in das Thema einarbeiten.

Okay. Wenn du C/C++ kannst, empfehle ich "Windows System Programming" und wenn du ein wenig mehr Theorie haben willst "The Art of Concurrency: A Thread Monkey's Guide to Writing Parallel Applications".

[SvB]

Jetzt habe ich mir mal TCriticalSection angesehen und in der Delphi Hilfe steht folgendes:

Zitat:

Mit TCriticalSection können Operationen abgesichert werden, die nicht durch den Ausführung eines bestimmten Codeabschnitts durch einen anderen Thread unterbrochen werden dürfen. Kritische Abschnitt arbeiten wie Gates, die jeweils nur einen einzelnen Thread durchlassen. Da die Ausführung anderer Threads blockiert wird, verringert sich die Leistung der Anwendung erheblich, wenn kritische Abschnitte zu häufig eingesetzt werden.

Das bedeutet, dass wenn mein JobB gestartet werden würde, dass die anderen Threads dann solange unterbrochen werden, bis der JobB abgearbeitet ist. Das wäre schlecht, denn die anderen Threads sollen in dieser Zeit noch JobA und JobB abarbeiten.

Dann schaue ich mir mal das nächste an.

[Assarbad]

Zitat von SvB:

Das bedeutet, dass wenn mein JobB gestartet werden würde, dass die anderen Threads dann solange unterbrochen werden, bis der JobB abgearbeitet ist. Das wäre schlecht, denn die anderen Threads sollen in dieser Zeit noch JobA und JobB abarbeiten.

Nein, das bedeutet es nicht. Es bedeutet, daß alle Jobs die sich den kritischen Abschnitt teilen jeweils nur einzeln ausgeführt werden. Zusätzlich ist zu bemerken, daß du eben TryEnter... und nicht Enter... benutzen solltest.

[SvB]

Ich habe mich heute fast den ganzen Tag mit meinem Problem beschäftigt. Zuerst habe ich mit Mutex probiert, da sind aber andauernd Fehler gekommen und ich habe es irgendwie nicht rihtig verstanden. Dann habe ich mit CriticalSection und TryCiritcalSection beschäftigt, aber irgendwo hat mir auch da de Ansatz gefehlt.

Ich habe aber jetzt eine Lösung gefunden, die auch anscheinend funktioniert und ich glaube, dass ich es anders nicht hinbekommen hätte. Mit meiner aktuellen Lösung kann ich sogar einem Job mitteilen, wie oft er gleichzeitig ausgeführt werden soll, z.B. sage ich meinem JobA nichts, dann kann er beliebig oft ausgeführt werden, meinem JobB sage ich 2, dann wird er bei 10 Threads nur 2 mal gleichzeitig ausgeführt.
Ich habe das über eine zusätzliche TThreadList gemacht, in der ich die aktuell ausgeführten Jobs hineinlege.

Wen es interessiert, ich habe den Quellcode und das Beispiel beigefügt. Ich habe es unter Delphi XE angepasst und getestet.