Something went wrong. Try again.
@recaptime-dev's working patches + fork for Phorge, a community fork of Phabricator. (Upstream dev and stable branches are at upstream/main hq.recaptime.dev/wiki/Phorge
phorge phabricator
Something went wrong. Try again.
14 kB · 537 lines
PHP
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538<?php
final class PhutilDaemonHandle extends Phobject {
const EVENT_DID_LAUNCH = 'daemon.didLaunch'; const EVENT_DID_LOG = 'daemon.didLogMessage'; const EVENT_DID_HEARTBEAT = 'daemon.didHeartbeat'; const EVENT_WILL_GRACEFUL = 'daemon.willGraceful'; const EVENT_WILL_EXIT = 'daemon.willExit';
private $pool; private $properties; private $future; private $argv;
private $restartAt; private $busyEpoch;
private $daemonID; private $deadline; private $heartbeat; private $stdoutBuffer; private $shouldRestart = true; private $shouldShutdown; private $hibernating = false; private $shouldSendExitEvent = false;
private function __construct() { // <empty> }
public static function newFromConfig(array $config) { PhutilTypeSpec::checkMap( $config, array( 'class' => 'string', 'argv' => 'optional list<string>', 'load' => 'optional list<string>', 'log' => 'optional string|null', 'down' => 'optional int', ));
$config = $config + array( 'argv' => array(), 'load' => array(), 'log' => null, 'down' => 15, );
$daemon = new self(); $daemon->properties = $config; $daemon->daemonID = $daemon->generateDaemonID();
return $daemon; }
public function setDaemonPool(PhutilDaemonPool $daemon_pool) { $this->pool = $daemon_pool; return $this; }
public function getDaemonPool() { return $this->pool; }
public function getBusyEpoch() { return $this->busyEpoch; }
public function getDaemonClass() { return $this->getProperty('class'); }
private function getProperty($key) { return idx($this->properties, $key); }
public function setCommandLineArguments(array $arguments) { $this->argv = $arguments; return $this; }
public function getCommandLineArguments() { return $this->argv; }
public function getDaemonArguments() { return $this->getProperty('argv'); }
public function didLaunch() { $this->restartAt = time(); $this->shouldSendExitEvent = true;
$this->dispatchEvent( self::EVENT_DID_LAUNCH, array( 'argv' => $this->getCommandLineArguments(), 'explicitArgv' => $this->getDaemonArguments(), ));
return $this; }
public function isRunning() { return (bool)$this->getFuture(); }
public function isHibernating() { return !$this->isRunning() && !$this->isDone() && $this->hibernating; }
public function wakeFromHibernation() { if (!$this->isHibernating()) { return $this; }
$this->logMessage( 'WAKE', pht( 'Process is being awakened from hibernation.'));
$this->restartAt = time(); $this->update();
return $this; }
public function isDone() { return (!$this->shouldRestart && !$this->isRunning()); }
public function update() { if (!$this->isRunning()) { if (!$this->shouldRestart) { return; } if (!$this->restartAt || (time() < $this->restartAt)) { return; } if ($this->shouldShutdown) { return; } $this->startDaemonProcess(); }
$future = $this->getFuture();
$result = null; $caught = null; if ($future->canResolve()) { $this->future = null; try { $result = $future->resolve(); } catch (Exception $ex) { $caught = $ex; } catch (Throwable $ex) { $caught = $ex; } }
list($stdout, $stderr) = $future->read(); $future->discardBuffers();
if (strlen($stdout)) { $this->didReadStdout($stdout); }
$stderr = trim($stderr); if (strlen($stderr)) { foreach (phutil_split_lines($stderr, false) as $line) { $this->logMessage('STDE', $line); } }
if ($result !== null || $caught !== null) {
if ($caught) { $message = pht( 'Process failed with exception: %s', $caught->getMessage()); $this->logMessage('FAIL', $message); } else { list($err) = $result;
if ($err) { $this->logMessage('FAIL', pht('Process exited with error %s.', $err)); } else { $this->logMessage('DONE', pht('Process exited normally.')); } }
if ($this->shouldShutdown) { $this->restartAt = null; } else { $this->scheduleRestart(); } }
$this->updateHeartbeatEvent(); $this->updateHangDetection(); }
private function updateHeartbeatEvent() { if ($this->heartbeat > time()) { return; }
$this->heartbeat = time() + $this->getHeartbeatEventFrequency(); $this->dispatchEvent(self::EVENT_DID_HEARTBEAT); }
private function updateHangDetection() { if (!$this->isRunning()) { return; }
if (time() > $this->deadline) { $this->logMessage('HANG', pht('Hang detected. Restarting process.')); $this->annihilateProcessGroup(); $this->scheduleRestart(); } }
private function scheduleRestart() { // Wait a minimum of a few sceconds before restarting, but we may wait // longer if the daemon has initiated hibernation. $default_restart = time() + self::getWaitBeforeRestart(); if ($default_restart >= $this->restartAt) { $this->restartAt = $default_restart; }
$this->logMessage( 'WAIT', pht( 'Waiting %s second(s) to restart process.', new PhutilNumber($this->restartAt - time()))); }
/** * Generate a unique ID for this daemon. * * @return string A unique daemon ID. */ private function generateDaemonID() { return substr(getmypid().':'.Filesystem::readRandomCharacters(12), 0, 12); }
public function getDaemonID() { return $this->daemonID; }
private function getFuture() { return $this->future; }
private function getPID() { $future = $this->getFuture();
if (!$future) { return null; }
if (!$future->hasPID()) { return null; }
return $future->getPID(); }
private function getCaptureBufferSize() { return 65535; }
private function getRequiredHeartbeatFrequency() { return 86400; }
public static function getWaitBeforeRestart() { return 5; }
public static function getHeartbeatEventFrequency() { return 120; }
private function getKillDelay() { return 3; }
private function getDaemonCWD() { $root = dirname(phutil_get_library_root('phabricator')); return $root.'/scripts/daemon/exec/'; }
private function newExecFuture() { $class = $this->getDaemonClass(); $argv = $this->getCommandLineArguments(); $buffer_size = $this->getCaptureBufferSize();
// NOTE: PHP implements proc_open() by running 'sh -c'. On most systems this // is bash, but on Ubuntu it's dash. When you proc_open() using bash, you // get one new process (the command you ran). When you proc_open() using // dash, you get two new processes: the command you ran and a parent // "dash -c" (or "sh -c") process. This means that the child process's PID // is actually the 'dash' PID, not the command's PID. To avoid this, use // 'exec' to replace the shell process with the real process; without this, // the child will call posix_getppid(), be given the pid of the 'sh -c' // process, and send it SIGUSR1 to keepalive which will terminate it // immediately. We also won't be able to do process group management because // the shell process won't properly posix_setsid() so the pgid of the child // won't be meaningful.
$config = $this->properties; unset($config['class']); $config = phutil_json_encode($config);
return id(new ExecFuture('exec ./exec_daemon.php %s %Ls', $class, $argv)) ->setCWD($this->getDaemonCWD()) ->setStdoutSizeLimit($buffer_size) ->setStderrSizeLimit($buffer_size) ->write($config); }
/** * Dispatch an event to event listeners. * * @param string $type Event type. * @param array $params (optional) Event parameters. * @return void */ private function dispatchEvent($type, array $params = array()) { $data = array( 'id' => $this->getDaemonID(), 'daemonClass' => $this->getDaemonClass(), 'childPID' => $this->getPID(), ) + $params;
$event = new PhutilEvent($type, $data);
try { PhutilEventEngine::dispatchEvent($event); } catch (Exception $ex) { phlog($ex); } }
private function annihilateProcessGroup() { $pid = $this->getPID(); if ($pid) { $pgid = posix_getpgid($pid); if ($pgid) { posix_kill(-$pgid, SIGTERM); sleep($this->getKillDelay()); posix_kill(-$pgid, SIGKILL); } } }
private function startDaemonProcess() { $this->logMessage('INIT', pht('Starting process.'));
$this->deadline = time() + $this->getRequiredHeartbeatFrequency(); $this->heartbeat = time() + self::getHeartbeatEventFrequency(); $this->stdoutBuffer = ''; $this->hibernating = false;
$future = $this->newExecFuture(); $this->future = $future;
$pool = $this->getDaemonPool(); $overseer = $pool->getOverseer(); $overseer->addFutureToPool($future); }
private function didReadStdout($data) { $this->stdoutBuffer .= $data; while (true) { $pos = strpos($this->stdoutBuffer, "\n"); if ($pos === false) { break; } $message = substr($this->stdoutBuffer, 0, $pos); $this->stdoutBuffer = substr($this->stdoutBuffer, $pos + 1);
try { $structure = phutil_json_decode($message); } catch (PhutilJSONParserException $ex) { $structure = array(); }
switch (idx($structure, 0)) { case PhutilDaemon::MESSAGETYPE_STDOUT: $this->logMessage('STDO', idx($structure, 1)); break; case PhutilDaemon::MESSAGETYPE_HEARTBEAT: $this->deadline = time() + $this->getRequiredHeartbeatFrequency(); break; case PhutilDaemon::MESSAGETYPE_BUSY: if (!$this->busyEpoch) { $this->busyEpoch = time(); } break; case PhutilDaemon::MESSAGETYPE_IDLE: $this->busyEpoch = null; break; case PhutilDaemon::MESSAGETYPE_DOWN: // The daemon is exiting because it doesn't have enough work and it // is trying to scale the pool down. We should not restart it. $this->shouldRestart = false; $this->shouldShutdown = true; break; case PhutilDaemon::MESSAGETYPE_HIBERNATE: $config = idx($structure, 1); $duration = (int)idx($config, 'duration', 0); $this->restartAt = time() + $duration; $this->hibernating = true; $this->busyEpoch = null; $this->logMessage( 'ZZZZ', pht( 'Process is preparing to hibernate for %s second(s).', new PhutilNumber($duration))); break; default: // If we can't parse this or it isn't a message we understand, just // emit the raw message. $this->logMessage('STDO', pht('<Malformed> %s', $message)); break; } } }
public function didReceiveNotifySignal($signo) { $pid = $this->getPID(); if ($pid) { posix_kill($pid, $signo); } }
public function didReceiveReloadSignal($signo) { $signame = phutil_get_signal_name($signo); if ($signame) { $sigmsg = pht( 'Reloading in response to signal %d (%s).', $signo, $signame); } else { $sigmsg = pht( 'Reloading in response to signal %d.', $signo); }
$this->logMessage('RELO', $sigmsg, $signo);
// This signal means "stop the current process gracefully, then launch // a new identical process once it exits". This can be used to update // daemons after code changes (the new processes will run the new code) // without aborting any running tasks.
// We SIGINT the daemon but don't set the shutdown flag, so it will // naturally be restarted after it exits, as though it had exited after an // unhandled exception.
$pid = $this->getPID(); if ($pid) { posix_kill($pid, SIGINT); } }
public function didReceiveGracefulSignal($signo) { $this->shouldShutdown = true; $this->shouldRestart = false;
$signame = phutil_get_signal_name($signo); if ($signame) { $sigmsg = pht( 'Graceful shutdown in response to signal %d (%s).', $signo, $signame); } else { $sigmsg = pht( 'Graceful shutdown in response to signal %d.', $signo); }
$this->logMessage('DONE', $sigmsg, $signo);
$pid = $this->getPID(); if ($pid) { posix_kill($pid, SIGINT); } }
public function didReceiveTerminateSignal($signo) { $this->shouldShutdown = true; $this->shouldRestart = false;
$signame = phutil_get_signal_name($signo); if ($signame) { $sigmsg = pht( 'Shutting down in response to signal %s (%s).', $signo, $signame); } else { $sigmsg = pht('Shutting down in response to signal %s.', $signo); }
$this->logMessage('EXIT', $sigmsg, $signo); $this->annihilateProcessGroup(); }
private function logMessage($type, $message, $context = null) { $this->getDaemonPool()->logMessage($type, $message, $context);
$this->dispatchEvent( self::EVENT_DID_LOG, array( 'type' => $type, 'message' => $message, 'context' => $context, )); }
public function didExit() { if ($this->shouldSendExitEvent) { $this->dispatchEvent(self::EVENT_WILL_EXIT); $this->shouldSendExitEvent = false; }
return $this; }
}