Ok, lässt so langsam beginnen ...PHP pThreads failing, wenn sie von cron
ich ein pThreads Skript ausgeführt wird und für mich arbeiten, getestet und funktioniert 100% der Zeit, als ich es manuell vom Befehl ausführen Linie über ssh. Das Skript ist wie folgt, wobei der Hauptthread-Prozesscode so eingestellt ist, dass er die Laufzeit des Zufallsprozesses simuliert.
class ProcessingPool extends Worker {
public function run(){}
}
class LongRunningProcess extends Threaded implements Collectable {
public function __construct($id,$data) {
$this->id = $id;
$this->data = $data;
}
public function run() {
$data = $this->data;
$this->garbage = true;
$this->result = 'START TIME:'.time().PHP_EOL;
// Here is our actual logic which will be handled within a single thread (obviously simulated here instead of the real functionality)
sleep(rand(1,100));
$this->result .= 'ID:'.$this->id.' RESULT: '.print_r($this->data,true).PHP_EOL;
$this->result .= 'END TIME:'.time().PHP_EOL;
$this->finished = time();
}
public function __destruct() {
$Finished = 'EXITED WITHOUT FINISHING';
if($this->finished > 0) {
$Finished = 'FINISHED';
}
if ($this->id === null) {
print_r("nullified thread $Finished!");
} else {
print_r("Thread w/ ID {$this->id} $Finished!");
}
}
public function isGarbage() : bool { return $this->garbage; }
public function getData() {
return $this->data;
}
public function getResult() {
return $this->result;
}
protected $id;
protected $data;
protected $result;
private $garbage = false;
private $finished = 0;
}
$LoopDelay = 500000; // microseconds
$MinimumRunTime = 300; // seconds (5 minutes)
// So we setup our pthreads pool which will hold our collection of threads
$pool = new Pool(4, ProcessingPool::class, []);
$Count = 0;
$StillCollecting = true;
$CountCollection = 0;
do {
// Grab all items from the conversion_queue which have not been processed
$result = $DB->prepare("SELECT * FROM `processing_queue` WHERE `processed` = 0 ORDER BY `queue_id` ASC");
$result->execute();
$rows = $result->fetchAll(PDO::FETCH_ASSOC);
if(!empty($rows)) {
// for each of the rows returned from the queue, and allow the workers to run and return
foreach($rows as $id => $row) {
$update = $DB->prepare("UPDATE `processing_queue` SET `processed` = 1 WHERE `queue_id` = ?");
$update->execute([$row['queue_id']]);
$pool->submit(new LongRunningProcess($row['fqueue_id'],$row));
$Count++;
}
} else {
// 0 Rows To Add To Pool From The Queue, Do Nothing...
}
// Before we allow the loop to move on to the next part, lets try and collect anything that finished
$pool->collect(function ($Processed) use(&$CountCollection) {
global $DB;
$data = $Processed->getData();
$result = $Processed->getResult();
$update = $DB->prepare("UPDATE `processing_queue` SET `processed` = 2 WHERE `queue_id` = ?");
$update->execute([$data['queue_id']]);
$CountCollection++;
return $Processed->isGarbage();
});
print_r('Collecting Loop...'.$CountCollection.'/'.$Count);
// If we have collected the same total amount as we have processed then we can consider ourselves done collecting everything that has been added to the database during the time this script started and was running
if($CountCollection == $Count) {
$StillCollecting = false;
print_r('Done Collecting Everything...');
}
// If we have not reached the full MinimumRunTime that this cron should run for, then lets continue to loop
$EndTime = microtime(true);
$TimeElapsed = ($EndTime - $StartTime);
if(($TimeElapsed/($LoopDelay/1000000)) < ($MinimumRunTime/($LoopDelay/1000000))) {
$StillCollecting = true;
print_r('Ended To Early, Lets Force Another Loop...');
}
usleep($LoopDelay);
} while($StillCollecting);
$pool->shutdown();
So, während das obige Skript über eine Befehlszeile läuft (die den Grund Beispiel eingestellt wurden, und ein detaillierter Verarbeitungscode in dem obigen Beispiel simuliert), der folgende Befehl ergibt sich ein anderes Ergebnis, wenn von einem cron Setup für alle 5 Minuten ...
/opt/php7zts/bin/php -q /home/account/cron-entry.php file=every-5-minutes/processing-queue.php
Das obige Skript, bei der Verwendung der oben genannten Kommandozeilenaufruf ausführen, wird loo p während der Laufzeit des Skripts immer wieder und sammelt alle neuen Elemente aus der DB-Warteschlange und fügt sie in den Pool ein, wodurch jeweils vier Prozesse ausgeführt und beendet werden können, die dann gesammelt und die Warteschlange aktualisiert wird eine weitere Schleife passiert, ziehen Sie alle neuen Elemente aus der DB. Dieses Skript wird ausgeführt, bis alle Prozesse in der Warteschlange während der Ausführung des Skripts verarbeitet und erfasst wurden. Wenn das Skript nicht für den vollen erwarteten Zeitraum von 5 Minuten ausgeführt wurde, wird die Schleife gezwungen, die Warteschlange weiter zu überprüfen. Wenn das Skript über die 5-Minuten-Markierung ausgeführt wurde, können alle aktuellen Threads vor dem Schließen abgearbeitet werden. Beachten Sie, dass der oben genannte Code auch eine code-basierte "Flock" -Funktionalität enthält, die zukünftige Crons dieser Leerlaufschleife erzeugt und beendet oder startet, sobald die Sperre aufgehoben wurde, um sicherzustellen, dass die Warteschlange und die Threads nicht aneinander stoßen. Auch hier funktioniert alles von der Befehlszeile über SSH.
Sobald ich den obigen Befehl nehmen und es in einen cron für alle 5 Minuten zu laufen setzen, mich im Wesentlichen eine nicht enden wollenden Schleife zu geben, während der Speicher beibehalten wird, erhalte ich ein anderes Ergebnis ...
Dieses Ergebnis wird wie folgt beschrieben ... Das Skript startet, überprüft die Herde und fährt fort, wenn die Sperre nicht da ist, erstellt es die Sperre und führt das obige Skript aus. Die Elemente werden aus der Warteschlange in der Datenbank entnommen und in den Pool eingefügt. Der Pool löst die 4 Threads wie erwartet aus. Aber Das unerwartete Ergebnis ist, dass der Befehl run() nicht ausgeführt wird und stattdessen die __destruct-Funktion und ein "Thread w/ID 2 FERTIG!" Art der Nachricht wird an die Ausgabe zurückgegeben. Dies wiederum bedeutet, dass die Sammlungsseite der Dinge nichts sammelt, und das initiierende Skript (das Cron-Skript selbst /home/account/cron-entry.php file = alle-5-Minuten/processing-queue.php) beendet, nachdem alles in den Pool gelegt und zerstört wurde. Was den Cron-Job vorzeitig "beendet", da nichts anderes zu tun ist als Schleifen und Ziehen nichts Neues aus der Warteschlange, da sie als verarbeitet betrachtet werden, wenn == 1 in der Warteschlange.
Die Frage wird dann schließlich ... Wie mache ich das Cron-Skript auf die Threads aufmerksam machen, in denen launched und run() sie ohne den Pool zu schließen, bevor sie etwas tun können?
(beachten Sie ..., wenn Sie Kopieren/Einfügen der bereitgestellten Skript, beachten Sie, dass ich es nicht testen, nachdem die detaillierte Logik zu entfernen, so dass es einige einfache Korrekturen benötigen ...Bitte wählen Sie den Code nicht aus, da der Schlüssel hier ist, dass Pthreads funktioniert, wenn das Skript von der Befehlszeile aus ausgeführt wird, aber nicht ordnungsgemäß ausgeführt wird, wenn das Skript von einem CRON ausgeführt wird. Wenn Sie vorhaben, mit nicht-konstruktiver Kritik zu kommentieren, benutzen Sie bitte Ihre Finger, um etwas anderes zu tun!)
Joe Watkins! Ich brauche deine Brillanz! Danke im Voraus!