diff --git a/src/daemon/PhutilDaemon.php b/src/daemon/PhutilDaemon.php index 71fe6c1..aff636c 100644 --- a/src/daemon/PhutilDaemon.php +++ b/src/daemon/PhutilDaemon.php @@ -1,392 +1,392 @@ shouldExit()) { * if (work_available()) { * $this->willBeginWork(); * do_work(); * $this->sleep(0); * } else { * $this->willBeginIdle(); * $this->sleep(1); * } * } * * In particular, call @{method:willBeginWork} before becoming busy, and * @{method:willBeginIdle} when no work is available. If the daemon is launched * into an autoscale pool, this will cause the pool to automatically scale up * when busy and down when idle. * * See @{class:PhutilHighIntensityIntervalDaemon} for an example of a simple * autoscaling daemon. * * Launching a daemon which does not make these callbacks into an autoscale * pool will have no effect. * * @task overseer Communicating With the Overseer * @task autoscale Autoscaling Daemon Pools * * @stable */ abstract class PhutilDaemon { const MESSAGETYPE_STDOUT = 'stdout'; const MESSAGETYPE_HEARTBEAT = 'heartbeat'; const MESSAGETYPE_BUSY = 'busy'; const MESSAGETYPE_IDLE = 'idle'; const MESSAGETYPE_DOWN = 'down'; const WORKSTATE_BUSY = 'busy'; const WORKSTATE_IDLE = 'idle'; private $argv; private $traceMode; private $traceMemory; private $verbose; private $notifyReceived; private $inGracefulShutdown; private $workState = null; private $idleSince = null; private $autoscaleProperties = array(); final public function setVerbose($verbose) { $this->verbose = $verbose; return $this; } final public function getVerbose() { return $this->verbose; } private static $sighandlerInstalled; final public function __construct(array $argv) { declare(ticks = 1); $this->argv = $argv; if (!self::$sighandlerInstalled) { self::$sighandlerInstalled = true; pcntl_signal(SIGTERM, __CLASS__.'::exitOnSignal'); } pcntl_signal(SIGINT, array($this, 'onGracefulSignal')); pcntl_signal(SIGUSR2, array($this, 'onNotifySignal')); // Without discard mode, this consumes unbounded amounts of memory. Keep // memory bounded. PhutilServiceProfiler::getInstance()->enableDiscardMode(); $this->beginStdoutCapture(); } final public function __destruct() { $this->endStdoutCapture(); } final public function stillWorking() { $this->emitOverseerMessage(self::MESSAGETYPE_HEARTBEAT, null); if ($this->traceMemory) { $memuse = number_format(memory_get_usage() / 1024, 1); $daemon = get_class($this); fprintf(STDERR, '%s', " {$daemon} Memory Usage: {$memuse} KB\n"); } } final public function shouldExit() { return $this->inGracefulShutdown; } final protected function sleep($duration) { $this->notifyReceived = false; $this->willSleep($duration); $this->stillWorking(); $is_autoscale = $this->isClonedAutoscaleDaemon(); $scale_down = $this->getAutoscaleDownDuration(); $max_sleep = 60; if ($is_autoscale) { $max_sleep = min($max_sleep, $scale_down); } if ($is_autoscale) { if ($this->workState == self::WORKSTATE_IDLE) { $dur = (time() - $this->idleSince); $this->log(pht('Idle for %s seconds.', $dur)); } } while ($duration > 0 && !$this->notifyReceived && !$this->shouldExit()) { // If this is an autoscaling clone and we've been idle for too long, // we're going to scale the pool down by exiting and not restarting. The // DOWN message tells the overseer that we don't want to be restarted. if ($is_autoscale) { if ($this->workState == self::WORKSTATE_IDLE) { if ($this->idleSince && ($this->idleSince + $scale_down < time())) { $this->inGracefulShutdown = true; $this->emitOverseerMessage(self::MESSAGETYPE_DOWN, null); $this->log( pht( 'Daemon was idle for more than %s seconds, scaling pool '. 'down.', $scale_down)); break; } } } sleep(min($duration, $max_sleep)); $duration -= $max_sleep; $this->stillWorking(); } } protected function willSleep($duration) { return; } public static function exitOnSignal($signo) { // Normally, PHP doesn't invoke destructors when existing in response to // a signal. This forces it to do so, so we have a fighting chance of // releasing any locks, leases or resources on our way out. exit(128 + $signo); } final protected function getArgv() { return $this->argv; } final public function execute() { $this->willRun(); $this->run(); } abstract protected function run(); final public function setTraceMemory() { $this->traceMemory = true; return $this; } final public function getTraceMemory() { return $this->traceMemory; } final public function setTraceMode() { $this->traceMode = true; PhutilServiceProfiler::installEchoListener(); PhutilConsole::getConsole()->getServer()->setEnableLog(true); $this->didSetTraceMode(); return $this; } final public function getTraceMode() { return $this->traceMode; } final public function onGracefulSignal($signo) { $this->inGracefulShutdown = true; } final public function onNotifySignal($signo) { $this->notifyReceived = true; $this->onNotify($signo); } protected function onNotify($signo) { // This is a hook for subclasses. } protected function willRun() { // This is a hook for subclasses. } protected function didSetTraceMode() { // This is a hook for subclasses. } final protected function log($message) { if ($this->verbose) { $daemon = get_class($this); fprintf(STDERR, '%s', " {$daemon} {$message}\n"); } } /* -( Communicating With the Overseer )------------------------------------ */ private function beginStdoutCapture() { ob_start(array($this, 'didReceiveStdout'), 2); } private function endStdoutCapture() { ob_end_flush(); } public function didReceiveStdout($data) { if (!strlen($data)) { return ''; } return $this->encodeOverseerMessage(self::MESSAGETYPE_STDOUT, $data); } private function encodeOverseerMessage($type, $data) { $structure = array($type); if ($data !== null) { $structure[] = $data; } return json_encode($structure)."\n"; } private function emitOverseerMessage($type, $data) { $this->endStdoutCapture(); echo $this->encodeOverseerMessage($type, $data); $this->beginStdoutCapture(); } public static function errorListener($event, $value, array $metadata) { // If the caller has redirected the error log to a file, PHP won't output // messages to stderr, so the overseer can't capture them. Install a // listener which just echoes errors to stderr, so the overseer is always // aware of errors. $console = PhutilConsole::getConsole(); $message = idx($metadata, 'default_message'); if ($message) { $console->writeErr("%s\n", $message); } if (idx($metadata, 'trace')) { $trace = PhutilErrorHandler::formatStacktrace($metadata['trace']); $console->writeErr("%s\n", $trace); } } /* -( Autoscaling )-------------------------------------------------------- */ /** * Prepare to become busy. This may autoscale the pool up. * * This notifies the overseer that the daemon has become busy. If daemons * that are part of an autoscale pool are continuously busy for a prolonged * period of time, the overseer may scale up the pool. * * @return this * @task autoscale */ protected function willBeginWork() { if ($this->workState != self::WORKSTATE_BUSY) { $this->workState = self::WORKSTATE_BUSY; $this->idleSince = null; $this->emitOverseerMessage(self::MESSAGETYPE_BUSY, null); } return $this; } /** * Prepare to idle. This may autoscale the pool down. * * This notifies the overseer that the daemon is no longer busy. If daemons * that are part of an autoscale pool are idle for a prolonged period of time, * they may exit to scale the pool down. * * @return this * @task autoscale */ protected function willBeginIdle() { if ($this->workState != self::WORKSTATE_IDLE) { $this->workState = self::WORKSTATE_IDLE; $this->idleSince = time(); $this->emitOverseerMessage(self::MESSAGETYPE_IDLE, null); } return $this; } /** * Determine if this is a clone or the original daemon. * * @return bool True if this is an cloned autoscaling daemon. * @task autoscale */ private function isClonedAutoscaleDaemon() { return (bool)$this->getAutoscaleProperty('clone', false); } /** * Get the duration (in seconds) which a daemon must be continuously idle * for before it should exit to scale the pool down. * * @return int Duration, in seconds. * @task autoscale */ private function getAutoscaleDownDuration() { return $this->getAutoscaleProperty('down', 15); } /** * Configure autoscaling for this daemon. * * @param map Map of autoscale properties. * @return this * @task autoscale */ public function setAutoscaleProperties(array $autoscale_properties) { PhutilTypeSpec::checkMap( $autoscale_properties, array( 'group' => 'optional string', 'up' => 'optional int', 'down' => 'optional int', 'pool' => 'optional int', 'clone' => 'optional bool', - 'reserve' => 'optional float', + 'reserve' => 'optional int|float', )); $this->autoscaleProperties = $autoscale_properties; return $this; } /** * Read autoscaling configuration for this daemon. * * @param string Property to read. * @param wild Default value to return if the property is not set. * @return wild Property value, or `$default` if one is not set. * @task autoscale */ private function getAutoscaleProperty($key, $default = null) { return idx($this->autoscaleProperties, $key, $default); } } diff --git a/src/daemon/PhutilDaemonHandle.php b/src/daemon/PhutilDaemonHandle.php index 82a5cea..3e935d3 100644 --- a/src/daemon/PhutilDaemonHandle.php +++ b/src/daemon/PhutilDaemonHandle.php @@ -1,399 +1,404 @@ overseer = $overseer; $this->daemonClass = $daemon_class; $this->argv = $argv; $this->config = $config; $this->restartAt = time(); $this->daemonID = $this->generateDaemonID(); $this->dispatchEvent( self::EVENT_DID_LAUNCH, array( 'argv' => $this->argv, 'explicitArgv' => idx($this->config, 'argv'), )); } public function isRunning() { return (bool)$this->future; } public function isDone() { return (!$this->shouldRestart && !$this->isRunning()); } public function getFuture() { return $this->future; } public function setSilent($silent) { $this->silent = $silent; return $this; } public function getSilent() { return $this->silent; } public function setTraceMemory($trace_memory) { $this->traceMemory = $trace_memory; return $this; } public function getTraceMemory() { return $this->traceMemory; } public function update() { $this->updateMemory(); if (!$this->isRunning()) { if (!$this->shouldRestart) { return; } if (!$this->restartAt || (time() < $this->restartAt)) { return; } if ($this->shouldShutdown) { return; } $this->startDaemonProcess(); } $future = $this->future; $result = null; if ($future->isReady()) { $result = $future->resolve(); } list($stdout, $stderr) = $future->read(); $future->discardBuffers(); if (strlen($stdout)) { $this->didReadStdout($stdout); } $stderr = trim($stderr); if (strlen($stderr)) { $this->logMessage('STDE', $stderr); } if ($result !== null) { list($err) = $result; if ($err) { $this->logMessage('FAIL', pht('Process exited with error %s', $err)); } else { $this->logMessage('DONE', pht('Process exited normally.')); } $this->future = null; if ($this->shouldShutdown) { $this->restartAt = null; $this->dispatchEvent(self::EVENT_WILL_EXIT); } 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() { $this->logMessage('WAIT', pht('Waiting to restart process.')); $this->restartAt = time() + self::getWaitBeforeRestart(); } /** * 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; } public function getPID() { return $this->pid; } 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('phutil')); return $root.'/scripts/daemon/exec/'; } private function newExecFuture() { $class = $this->daemonClass; $argv = $this->argv; $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. return id(new ExecFuture('exec ./exec_daemon.php %s %Ls', $class, $argv)) ->setCWD($this->getDaemonCWD()) ->setStdoutSizeLimit($buffer_size) ->setStderrSizeLimit($buffer_size) ->write(json_encode($this->config)); } /** * Dispatch an event to event listeners. * * @param string Event type. * @param dict Event parameters. * @return void */ private function dispatchEvent($type, array $params = array()) { $data = array( 'id' => $this->daemonID, 'daemonClass' => $this->daemonClass, 'childPID' => $this->pid, ) + $params; $event = new PhutilEvent($type, $data); try { PhutilEventEngine::dispatchEvent($event); } catch (Exception $ex) { phlog($ex); } } private function annihilateProcessGroup() { $pid = $this->pid; $pgid = posix_getpgid($pid); if ($pid && $pgid) { // NOTE: On Ubuntu, 'kill' does not recognize the use of "--" to // explicitly delineate PID/PGIDs from signals. We don't actually need it, // so use the implicit "kill -TERM -pgid" form instead of the explicit // "kill -TERM -- -pgid" form. exec("kill -TERM -{$pgid}"); sleep($this->getKillDelay()); // On OSX, we'll get a permission error on stderr if the SIGTERM was // successful in ending the life of the process group, presumably because // all that's left is the daemon itself as a zombie waiting for us to // reap it. However, we still need to issue this command for process // groups that resist SIGTERM. Rather than trying to figure out if the // process group is still around or not, just SIGKILL unconditionally and // ignore any error which may be raised. exec("kill -KILL -{$pgid} 2>/dev/null"); $this->pid = null; } } private function gracefulProcessGroup() { $pid = $this->pid; $pgid = posix_getpgid($pid); if ($pid && $pgid) { exec("kill -INT -{$pgid}"); } } private function updateMemory() { if ($this->traceMemory) { $memuse = number_format(memory_get_usage() / 1024, 1); $this->logMessage('RAMS', 'Overseer Memory Usage: '.$memuse.' KB'); } } private function startDaemonProcess() { $this->logMessage('INIT', pht('Starting process.')); $this->deadline = time() + $this->getRequiredHeartbeatFrequency(); $this->heartbeat = time() + self::getHeartbeatEventFrequency(); $this->stdoutBuffer = ''; $this->future = $this->newExecFuture(); $this->future->start(); $this->pid = $this->future->getPID(); } 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); $structure = @json_decode($message, true); if (!is_array($structure)) { $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: $this->overseer->didBeginWork($this); break; case PhutilDaemon::MESSAGETYPE_IDLE: $this->overseer->didBeginIdle($this); 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; default: // If we can't parse this or it isn't a message we understand, just // emit the raw message. $this->logMessage('STDO', pht(' %s', $message)); break; } } } public function didReceiveNotifySignal($signo) { $pid = $this->pid; if ($pid) { posix_kill($pid, $signo); } } public function didReceiveGracefulSignal($signo) { $this->shouldShutdown = true; + if (!$this->isRunning()) { + // If we aren't running a daemon, emit this event now. Otherwise, we'll + // emit it when the daemon exits. + $this->dispatchEvent(self::EVENT_WILL_EXIT); + } $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); $this->gracefulProcessGroup(); } public function didReceiveTerminalSignal($signo) { $signame = phutil_get_signal_name($signo); if ($signame) { $sigmsg = "Shutting down in response to signal {$signo} ({$signame})."; } else { $sigmsg = "Shutting down in response to signal {$signo}."; } $this->logMessage('EXIT', $sigmsg, $signo); $this->annihilateProcessGroup(); $this->dispatchEvent(self::EVENT_WILL_EXIT); } private function logMessage($type, $message, $context = null) { if (!$this->getSilent()) { echo date('Y-m-d g:i:s A').' ['.$type.'] '.$message."\n"; } $this->dispatchEvent( self::EVENT_DID_LOG, array( 'type' => $type, 'message' => $message, 'context' => $context, )); } }