WaitForMultipleObjects "bWaitForSomeConditions"

[jensw_2000]

Wie kann man WaitForMultipleObjects sinnvoll "kaskadieren"?

Mein WorkerThread "rennt los", wenn eines von zwei Events gesetzt ist (SuspendEvent und TerminateEvent), oder wenn ein neuer Job in der Jobliste ist (Semaphore wird gesetzt und von einem WorkerThread aus dem Pool angenommen).

Ich würde den Thread gern "schlafen" lassen, wenn ein neuer Job (Semaphore) vorhanden ist, bis alle Voraussetzungen erfüllt sind, um den Job abzuarbeiten.

Zur Zeit löse ich das "schnell und schmutzig".

Ich nehme die Semaphore in CASE WAIT_OBJECT_0 + 0 an und "verbrauche" eine Semaphore damit.
Bevor ich mir den Job aus der Joblist hole, prüfe ich die Voraussetzungen zum Abarbeiten des Jobs. (Könnten sich ja geändert haben wärend der Thread geschlafen hat...).
Wenn die Voraussetzungen nicht erfüllt sind, hole ich den Job nicht aus der Liste und zähle die Semapohre der Joblist wieder um 1 hoch "ReleaseSemaphore(fJoblistSemaphore,1,nil)" damit jeder Job in der Liste wieder eine Semaphore hat.

Geht das Besser?
Wie kann ich das "bauen", dass das "WAIT_OBJECT_0 + 0" eine UND Verknüpfung von -"Semaphore da" UND "Voraussetzungen erfüllt"- darstellt?

Delphi-Quellcode:

     case WaitForMultipleObjects(3, @WeakupHandles[0], False, INFINITE) of
        WAIT_OBJECT_0 + 2: // "TerminateEvent gesetzt"
          begin
            WorkerThreadStatus := stOT_Terminating;
            BeforeWorkerThreadTerminate;
            Terminate;
            Break;
          end;
        WAIT_OBJECT_0 + 1: // "SuspendWorkersEvent gesetzt"
          begin
            ResetStatus;
            WorkerThreadStatus := stOT_Suspended;
            Suspend;
            Break;
          end;
        WAIT_OBJECT_0 + 0: // "neuer Job (Semaphore) vorhanden"
          begin
            WorkerThreadStatus := stOT_Working;
            
             .... arbeite los, wenn: 
             - DB Connection aktiv ist
             - Netzwerkverbindung aktiv
       
            VerarbeiteJob(Joblist.GetNextJob);

Grüße
Jens

[Furtbichler]

Wieso setzt Du die Semaphore nicht erst, wenn beide Bedingungen erfüllt sind?

[jensw_2000]

Der WorkerThreadPool hängt in einem Webservice.
Die Jobs kommen daher sporadisch und unkontrollierbar von aussen in die Jobliste.

Nach jedem JobList.Add(aJob) setze ich das Semaphore-Handle um 1 hoch.
Das Semaphore Handle legt ja im Endeffelt nur fest, wie oft WorkerThreads aus dem ThreadPool aufgeweckt werden müssen um alle Jobs abzuarbeiten.

Der, durch ein Semnaphore, aufgeweckte Thread greift sich dann immer den ältesten Job aus der Jobliste (FiFo). In der Zeit wo der Thread auf das Semaphore gewartet hat, können sich viele Zustände im System geändert haben. Das bekommt der Thread während des Wartens ja nicht mit.

Wenn ich nun die Jobs in die Liste schreiben würde, ohne ein Semaphore zu setzen, dann müsste ja wieder ein Thread "rotieren" und stänfig prüfen, ob es neue Jobs gibt und alle Voraussetzungen erfüllt sind um die Jobs abzuarbeiten, um dann Semaphore zu setzen, die dann von Workertheadpool erkannt werden ... Das wäre nicht zielführend. Der Service soll schlafen und erst aktiv werden, wenn es was zu erledigen gibt. Schlafen könnte dieser "ListenüberwachungsThread" dann ja nicht. Wenn ich den auf ein Semaphore warten lassen würde, dann drehe ich mich im Kreis :cyclops:

[Furtbichler]

Ach so. Zum Zeitpunkt des Ausführens müssen bestimmte Bedingungen erfüllt sein...
Wäre eine zweite Jobliste/Workerthread die Lösung? So nach dem Motto: "Ok, der Job soll jetzt, kann aber noch nicht.. Ach, sag mir Bescheid, wenn es geht".

[jensw_2000]

Der Ansatz bringt mich auf eine Idee.

Ich denke, dass ich meine Jobliste umbauen kann.

Wenn ein WorkerThread auf Grund einer "erhaltenen Semaphore" einen Job abrufen möchte, kann ich relativ umkompliziert ein paar Parameter mit den Zustandsinformationen der "Voraussetzungen" an die Joblisten Klasse übergeben. "Joblist.GetNextJob(DBready, NetworkReady... )".

Die Jobliste kann an Hand der Zustandinfos feststellen, ob der Thread den Job ausführen kann oder nicht. Falls nicht, kann sie den Thread via Suspend "hart" einschlafen lassen.
Der Thread fragt dann nie wieder nach einem Job, bis er von extern per Resume wieder fortgesetzt wird ....
Die Joblist gibt dem Thread in diesem Fall natürlich keinen Job und wirft wieder eine Semapohre in die Runde, weil der Job ja nicht vergeben wurde ..

Am Ende müsste meine Workerthread-Pool Klasse nur noch von Zeit zu Zeit prüfen, ob suspendierte Workerthreads auf Grund "positiver Statusinfos" wieder fortgesetzt werden können.

Das wäre natürlich nur ein Workaround.

Wenn einige meine User nicht noch am Server 2003 hängen würden, könnte man das Problem sauber über Conditions lösen. Die unterstützt Windows aber erst seit dem Vista/WS2008 Kernel.

Ich hoffe noch, dass es dafür eine WinXP/WS2003 kompatibele Ausweichlösung gibt.

[Furtbichler]

Nie mit Resume/Suspend arbeiten!

Ich würde es so machen:

Also:
1. Hole Job aus der Liste
2. Frage, ob Startbedingung erfüllt sind
3. Warte auf Antwort

Bei (2) kannst Du ein Mutex (oder ne 1-Semaphore) verwenden, das Du dem 'Startbedingungsevaluierungsthread' übergibst. Der prüft und gibt das Mutex zurück (3), aber erst, wenn alles OK ist, worauf der WT dann weiterläuft.

Falls es soetwas wie ein Timeout gibt, dann musst Du die Antwort noch kodieren (OK, TIMEOUT, SHUTDOWN).

[jensw_2000]

Zitat:

1. Hole Job aus der Liste
2. Frage, ob Startbedingung erfüllt sind
3. Warte auf Antwort

Wenn ich das richtig verstehe, dann soll also ein "WorkerThreadStarterThread" auf die Semaphore der Joblist reagieren und wenn alle Voraussetzungen passen, ein Mutex oder eine Semaphore setzen, die dann den Workerthread startet.

Besser "warten auf erfüllte Voraussetzungen" anstatt Suspend ist vernünftig. Nehme ich :thumb:

Ich versuche mal einen Weg ohne den zusätzlichen "StarterThread"....
Etwa so sollte das Thema ohne Suspend und zusätzliche Threads doch klappen oder?

Delphi-Quellcode:

WeakupHandles[0] := JobList.JobSemaphore;
WeakupHandles[1] := SuspendThreadsEvent.Handle;
WeakupHandles[2] := TerminateThreadsEvent.Handle;

case WaitForMultipleObjects(3, @WeakupHandles[0], False, INFINITE) of
         WAIT_OBJECT_0 + 2: // "TerminateEvent gesetzt"
           begin
             WorkerThreadStatus := stOT_Terminating;
             BeforeWorkerThreadTerminate;
             Terminate;
             Break;
           end;
         WAIT_OBJECT_0 + 1: // "SuspendWorkersEvent gesetzt"
           begin
             ResetStatus;
             WorkerThreadStatus := stOT_Suspended;
             "Flag setzen auf der der Thread am Ende von Execute reagiert";
             Break;
           end;
         WAIT_OBJECT_0 + 0: // "neuer Job (Semaphore) vorhanden"
           begin

             WorkerThreadStatus := stOT_WaitingForPrerequisites;

             hBedingungenErfuellt := CheckBedingunen(); // noch zu bauen              
             
             WeakupHandles[0] := JobList.JobSemaphore;
             WeakupHandles[1] := SuspendThreadsEvent.Handle;
             WeakupHandles[2] := hBedingungenErfuellt; // noch zu bauen

             
             case WaitForMultipleObjects(3, @WeakupHandles[0], False, INFINITE) of
               WAIT_OBJECT_0 + 2: // "TerminateEvent gesetzt"
               begin
                 WorkerThreadStatus := stOT_Terminating;
                 BeforeWorkerThreadTerminate;
                 Terminate;
                 Break;
               end;
               WAIT_OBJECT_0 + 1: // "SuspendWorkersEvent gesetzt"
               begin
                 ResetStatus;
                 WorkerThreadStatus := stOT_Suspended;
                 ReleaseSemaphore(JobList.JobSemaphore,1,nil); // Semaphore zurückgeben  
                 "Flag setzen auf der der Thread am Ende von Execute reagiert";
                 Break;
               end;
               WAIT_OBJECT_0 // hBedigungnenErfuellt ist gesetzt
               begin
                 WorkerThreadStatus := stOT_Working;
                 VerarbeiteJob(Joblist.GetNextJob); 
                 break;
               end; 
             end;  
             break; 
           end;
...