Differential D17389 Diff 41813 src/daemon/PhutilDaemonHandle.php

Changeset View

Standalone View

src/daemon/PhutilDaemonHandle.php

<?php		<?php

final class PhutilDaemonHandle extends Phobject {		final class PhutilDaemonHandle extends Phobject {

const EVENT_DID_LAUNCH = 'daemon.didLaunch';		const EVENT_DID_LAUNCH = 'daemon.didLaunch';
const EVENT_DID_LOG = 'daemon.didLogMessage';		const EVENT_DID_LOG = 'daemon.didLogMessage';
const EVENT_DID_HEARTBEAT = 'daemon.didHeartbeat';		const EVENT_DID_HEARTBEAT = 'daemon.didHeartbeat';
const EVENT_WILL_GRACEFUL = 'daemon.willGraceful';		const EVENT_WILL_GRACEFUL = 'daemon.willGraceful';
const EVENT_WILL_EXIT = 'daemon.willExit';		const EVENT_WILL_EXIT = 'daemon.willExit';

private $overseer;		private $pool;
private $daemonClass;		private $properties;
		private $future;
private $argv;		private $argv;
private $config;
		private $restartAt;
		private $busyEpoch;

private $pid;		private $pid;
private $daemonID;		private $daemonID;
private $deadline;		private $deadline;
private $heartbeat;		private $heartbeat;
private $stdoutBuffer;		private $stdoutBuffer;
private $restartAt;
private $shouldRestart = true;		private $shouldRestart = true;
private $shouldShutdown;		private $shouldShutdown;
private $future;
private $traceMemory;

public function __construct(		private function __construct() {
PhutilDaemonOverseer $overseer,		// <empty>
$daemon_class,		}
array $argv,
array $config) {		public static function newFromConfig(array $config) {
		PhutilTypeSpec::checkMap(
$this->overseer = $overseer;		$config,
$this->daemonClass = $daemon_class;		array(
$this->argv = $argv;		'class' => 'string',
$this->config = $config;		'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->restartAt = time();

$this->daemonID = $this->generateDaemonID();
$this->dispatchEvent(		$this->dispatchEvent(
self::EVENT_DID_LAUNCH,		self::EVENT_DID_LAUNCH,
array(		array(
'argv' => $this->argv,		'argv' => $this->getCommandLineArguments(),
'explicitArgv' => idx($this->config, 'argv'),		'explicitArgv' => $this->getDaemonArguments(),
));		));

		return $this;
}		}

public function isRunning() {		public function isRunning() {
return (bool)$this->future;		return (bool)$this->future;
}		}

public function isDone() {		public function isDone() {
return (!$this->shouldRestart && !$this->isRunning());		return (!$this->shouldRestart && !$this->isRunning());
}		}

public function getFuture() {		public function getFuture() {
return $this->future;		return $this->future;
}		}

public function setTraceMemory($trace_memory) {
$this->traceMemory = $trace_memory;
return $this;
}

public function getTraceMemory() {
return $this->traceMemory;
}

public function update() {		public function update() {
$this->updateMemory();

if (!$this->isRunning()) {		if (!$this->isRunning()) {
if (!$this->shouldRestart) {		if (!$this->shouldRestart) {
return;		return;
}		}
if (!$this->restartAt \|\| (time() < $this->restartAt)) {		if (!$this->restartAt \|\| (time() < $this->restartAt)) {
return;		return;
}		}
if ($this->shouldShutdown) {		if ($this->shouldShutdown) {
Show All 31 Lines	if ($result !== null) {
} else {		} else {
$this->logMessage('DONE', pht('Process exited normally.'));		$this->logMessage('DONE', pht('Process exited normally.'));
}		}

$this->future = null;		$this->future = null;

if ($this->shouldShutdown) {		if ($this->shouldShutdown) {
$this->restartAt = null;		$this->restartAt = null;
		$this->dispatchEvent(self::EVENT_WILL_EXIT);
} else {		} else {
$this->scheduleRestart();		$this->scheduleRestart();
}		}
}		}

$this->updateHeartbeatEvent();		$this->updateHeartbeatEvent();
$this->updateHangDetection();		$this->updateHangDetection();
}		}
▲ Show 20 Lines • Show All 62 Lines • ▼ Show 20 Lines	final class PhutilDaemonHandle extends Phobject {
}		}

private function getDaemonCWD() {		private function getDaemonCWD() {
$root = dirname(phutil_get_library_root('phutil'));		$root = dirname(phutil_get_library_root('phutil'));
return $root.'/scripts/daemon/exec/';		return $root.'/scripts/daemon/exec/';
}		}

private function newExecFuture() {		private function newExecFuture() {
$class = $this->daemonClass;		$class = $this->getDaemonClass();
$argv = $this->argv;		$argv = $this->getCommandLineArguments();
$buffer_size = $this->getCaptureBufferSize();		$buffer_size = $this->getCaptureBufferSize();

// NOTE: PHP implements proc_open() by running 'sh -c'. On most systems this		// 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		// 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		// 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, 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		// "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		// 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,		// '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'		// 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		// process, and send it SIGUSR1 to keepalive which will terminate it
// immediately. We also won't be able to do process group management because		// 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		// the shell process won't properly posix_setsid() so the pgid of the child
// won't be meaningful.		// 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))		return id(new ExecFuture('exec ./exec_daemon.php %s %Ls', $class, $argv))
->setCWD($this->getDaemonCWD())		->setCWD($this->getDaemonCWD())
->setStdoutSizeLimit($buffer_size)		->setStdoutSizeLimit($buffer_size)
->setStderrSizeLimit($buffer_size)		->setStderrSizeLimit($buffer_size)
->write(json_encode($this->config));		->write($config);
}		}

/**		/**
* Dispatch an event to event listeners.		* Dispatch an event to event listeners.
*		*
* @param string Event type.		* @param string Event type.
* @param dict Event parameters.		* @param dict Event parameters.
* @return void		* @return void
*/		*/
private function dispatchEvent($type, array $params = array()) {		private function dispatchEvent($type, array $params = array()) {
$data = array(		$data = array(
'id' => $this->daemonID,		'id' => $this->getDaemonID(),
'daemonClass' => $this->daemonClass,		'daemonClass' => $this->getDaemonClass(),
'childPID' => $this->pid,		'childPID' => $this->getPID(),
) + $params;		) + $params;

$event = new PhutilEvent($type, $data);		$event = new PhutilEvent($type, $data);

try {		try {
PhutilEventEngine::dispatchEvent($event);		PhutilEventEngine::dispatchEvent($event);
} catch (Exception $ex) {		} catch (Exception $ex) {
phlog($ex);		phlog($ex);
}		}
}		}

private function annihilateProcessGroup() {		private function annihilateProcessGroup() {
$pid = $this->pid;		$pid = $this->getPID();

$pgid = posix_getpgid($pid);		$pgid = posix_getpgid($pid);
if ($pid && $pgid) {		if ($pid && $pgid) {
posix_kill(-$pgid, SIGTERM);		posix_kill(-$pgid, SIGTERM);
sleep($this->getKillDelay());		sleep($this->getKillDelay());
posix_kill(-$pgid, SIGKILL);		posix_kill(-$pgid, SIGKILL);
$this->pid = null;		$this->pid = null;
}		}
}		}

private function updateMemory() {
if ($this->traceMemory) {
$this->logMessage(
'RAMS',
pht(
'Overseer Memory Usage: %s KB',
new PhutilNumber(memory_get_usage() / 1024, 1)));
}
}

private function startDaemonProcess() {		private function startDaemonProcess() {
$this->logMessage('INIT', pht('Starting process.'));		$this->logMessage('INIT', pht('Starting process.'));

$this->deadline = time() + $this->getRequiredHeartbeatFrequency();		$this->deadline = time() + $this->getRequiredHeartbeatFrequency();
$this->heartbeat = time() + self::getHeartbeatEventFrequency();		$this->heartbeat = time() + self::getHeartbeatEventFrequency();
$this->stdoutBuffer = '';		$this->stdoutBuffer = '';

$this->future = $this->newExecFuture();		$this->future = $this->newExecFuture();
Show All 21 Lines	while (true) {
switch (idx($structure, 0)) {		switch (idx($structure, 0)) {
case PhutilDaemon::MESSAGETYPE_STDOUT:		case PhutilDaemon::MESSAGETYPE_STDOUT:
$this->logMessage('STDO', idx($structure, 1));		$this->logMessage('STDO', idx($structure, 1));
break;		break;
case PhutilDaemon::MESSAGETYPE_HEARTBEAT:		case PhutilDaemon::MESSAGETYPE_HEARTBEAT:
$this->deadline = time() + $this->getRequiredHeartbeatFrequency();		$this->deadline = time() + $this->getRequiredHeartbeatFrequency();
break;		break;
case PhutilDaemon::MESSAGETYPE_BUSY:		case PhutilDaemon::MESSAGETYPE_BUSY:
$this->overseer->didBeginWork($this);		if (!$this->busyEpoch) {
		$this->busyEpoch = time();
		}
break;		break;
case PhutilDaemon::MESSAGETYPE_IDLE:		case PhutilDaemon::MESSAGETYPE_IDLE:
$this->overseer->didBeginIdle($this);		$this->busyEpoch = null;
break;		break;
case PhutilDaemon::MESSAGETYPE_DOWN:		case PhutilDaemon::MESSAGETYPE_DOWN:
// The daemon is exiting because it doesn't have enough work and it		// 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.		// is trying to scale the pool down. We should not restart it.
$this->shouldRestart = false;		$this->shouldRestart = false;
$this->shouldShutdown = true;		$this->shouldShutdown = true;
break;		break;
default:		default:
// If we can't parse this or it isn't a message we understand, just		// If we can't parse this or it isn't a message we understand, just
// emit the raw message.		// emit the raw message.
$this->logMessage('STDO', pht('<Malformed> %s', $message));		$this->logMessage('STDO', pht('<Malformed> %s', $message));
break;		break;
}		}
}		}
}		}

public function didReceiveNotifySignal($signo) {		public function didReceiveNotifySignal($signo) {
$pid = $this->pid;		$pid = $this->getPID();
if ($pid) {		if ($pid) {
posix_kill($pid, $signo);		posix_kill($pid, $signo);
}		}
}		}

public function didReceiveReloadSignal($signo) {		public function didReceiveReloadSignal($signo) {
$signame = phutil_get_signal_name($signo);		$signame = phutil_get_signal_name($signo);
if ($signame) {		if ($signame) {
Show All 13 Lines	public function didReceiveReloadSignal($signo) {
// a new identical process once it exits". This can be used to update		// 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)		// daemons after code changes (the new processes will run the new code)
// without aborting any running tasks.		// without aborting any running tasks.

// We SIGINT the daemon but don't set the shutdown flag, so it will		// 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		// naturally be restarted after it exits, as though it had exited after an
// unhandled exception.		// unhandled exception.

posix_kill($this->pid, SIGINT);		posix_kill($this->getPID(), SIGINT);
}		}

public function didReceiveGracefulSignal($signo) {		public function didReceiveGracefulSignal($signo) {
$this->shouldShutdown = true;		$this->shouldShutdown = true;
$this->shouldRestart = false;		$this->shouldRestart = false;

$signame = phutil_get_signal_name($signo);		$signame = phutil_get_signal_name($signo);
if ($signame) {		if ($signame) {
$sigmsg = pht(		$sigmsg = pht(
'Graceful shutdown in response to signal %d (%s).',		'Graceful shutdown in response to signal %d (%s).',
$signo,		$signo,
$signame);		$signame);
} else {		} else {
$sigmsg = pht(		$sigmsg = pht(
'Graceful shutdown in response to signal %d.',		'Graceful shutdown in response to signal %d.',
$signo);		$signo);
}		}

$this->logMessage('DONE', $sigmsg, $signo);		$this->logMessage('DONE', $sigmsg, $signo);

posix_kill($this->pid, SIGINT);		posix_kill($this->getPID(), SIGINT);
}		}

public function didReceiveTerminalSignal($signo) {		public function didReceiveTerminateSignal($signo) {
$this->shouldShutdown = true;		$this->shouldShutdown = true;
$this->shouldRestart = false;		$this->shouldRestart = false;

$signame = phutil_get_signal_name($signo);		$signame = phutil_get_signal_name($signo);
if ($signame) {		if ($signame) {
$sigmsg = pht(		$sigmsg = pht(
'Shutting down in response to signal %s (%s).',		'Shutting down in response to signal %s (%s).',
$signo,		$signo,
$signame);		$signame);
} else {		} else {
$sigmsg = pht('Shutting down in response to signal %s.', $signo);		$sigmsg = pht('Shutting down in response to signal %s.', $signo);
}		}

$this->logMessage('EXIT', $sigmsg, $signo);		$this->logMessage('EXIT', $sigmsg, $signo);
$this->annihilateProcessGroup();		$this->annihilateProcessGroup();
}		}

private function logMessage($type, $message, $context = null) {		private function logMessage($type, $message, $context = null) {
$this->overseer->logMessage($type, $message, $context);		$this->getDaemonPool()->logMessage($type, $message, $context);

$this->dispatchEvent(		$this->dispatchEvent(
self::EVENT_DID_LOG,		self::EVENT_DID_LOG,
array(		array(
'type' => $type,		'type' => $type,
'message' => $message,		'message' => $message,
'context' => $context,		'context' => $context,
));		));
}		}

public function didRemoveDaemon() {		public function toDictionary() {
$this->dispatchEvent(self::EVENT_WILL_EXIT);		return array(
		'pid' => $this->getPID(),
		'id' => $this->getDaemonID(),
		'config' => $this->properties,
		);
}		}


}		}