Differential D17389 Diff 41829 src/daemon/PhutilDaemonOverseer.php

Changeset View

Standalone View

src/daemon/PhutilDaemonOverseer.php

<?php		<?php

/**		/**
* Oversees a daemon and restarts it if it fails.		* Oversees a daemon and restarts it if it fails.
		*
		* @task signals Signal Handling
*/		*/
final class PhutilDaemonOverseer extends Phobject {		final class PhutilDaemonOverseer extends Phobject {

private $argv;		private $argv;
private $moreArgs;
private $inAbruptShutdown;
private $inGracefulShutdown;
private static $instance;		private static $instance;

private $config;		private $config;
private $daemons = array();		private $pools = array();
private $traceMode;		private $traceMode;
private $traceMemory;		private $traceMemory;
private $daemonize;		private $daemonize;
private $piddir;		private $piddir;
private $log;		private $log;
private $libraries = array();		private $libraries = array();
private $modules = array();		private $modules = array();
private $verbose;		private $verbose;
private $err = 0;
private $lastPidfile;		private $lastPidfile;
private $startEpoch;		private $startEpoch;
private $autoscale = array();		private $autoscale = array();
private $autoscaleConfig = array();		private $autoscaleConfig = array();

		const SIGNAL_NOTIFY = 'signal/notify';
		const SIGNAL_RELOAD = 'signal/reload';
		const SIGNAL_GRACEFUL = 'signal/graceful';
		const SIGNAL_TERMINATE = 'signal/terminate';

		private $err = 0;
		private $inAbruptShutdown;
		private $inGracefulShutdown;

public function __construct(array $argv) {		public function __construct(array $argv) {
PhutilServiceProfiler::getInstance()->enableDiscardMode();		PhutilServiceProfiler::getInstance()->enableDiscardMode();

$args = new PhutilArgumentParser($argv);		$args = new PhutilArgumentParser($argv);
$args->setTagline(pht('daemon overseer'));		$args->setTagline(pht('daemon overseer'));
$args->setSynopsis(<<<EOHELP		$args->setSynopsis(<<<EOHELP
launch_daemon.php [__options__] __daemon__		launch_daemon.php [__options__] __daemon__
Launch and oversee an instance of __daemon__.		Launch and oversee an instance of __daemon__.
▲ Show 20 Lines • Show All 106 Lines • ▼ Show 20 Lines	if ($this->daemonize) {
throw new Exception(pht('Unable to fork!'));		throw new Exception(pht('Unable to fork!'));
} else if ($pid) {		} else if ($pid) {
exit(0);		exit(0);
}		}
}		}

$this->modules = PhutilDaemonOverseerModule::getAllModules();		$this->modules = PhutilDaemonOverseerModule::getAllModules();

pcntl_signal(SIGUSR2, array($this, 'didReceiveNotifySignal'));		$this->installSignalHandlers();

pcntl_signal(SIGHUP, array($this, 'didReceiveReloadSignal'));
pcntl_signal(SIGINT, array($this, 'didReceiveGracefulSignal'));
pcntl_signal(SIGTERM, array($this, 'didReceiveTerminalSignal'));
}		}

public function addLibrary($library) {		public function addLibrary($library) {
$this->libraries[] = $library;		$this->libraries[] = $library;
return $this;		return $this;
}		}

public function run() {		public function run() {
$this->daemons = array();		$this->createDaemonPools();

foreach ($this->config['daemons'] as $config) {
$config += array(
'argv' => array(),
'autoscale' => array(),
);

$daemon = new PhutilDaemonHandle(
$this,
$config['class'],
$this->argv,
array(
'log' => $this->log,
'argv' => $config['argv'],
'load' => $this->libraries,
'autoscale' => $config['autoscale'],
));

$daemon->setTraceMemory($this->traceMemory);

$this->addDaemon($daemon, $config);

$group = idx($config['autoscale'], 'group');
if (strlen($group)) {
if (isset($this->autoscaleConfig[$group])) {
throw new Exception(
pht(
'Two daemons are part of the same autoscale group ("%s"). '.
'Each daemon autoscale group must be unique.',
$group));
}
$this->autoscaleConfig[$group] = $config;
}
}

$should_reload = false;

while (true) {		while (true) {
foreach ($this->modules as $module) {		if ($this->shouldReloadDaemons()) {
try {		$this->didReceiveSignal(SIGHUP);
if ($module->shouldReloadDaemons()) {
$this->logMessage(
'RELO',
pht(
'Reloading daemons (triggered by overseer module "%s").',
get_class($module)));
$should_reload = true;
}
} catch (Exception $ex) {
phlog($ex);
}
}

if ($should_reload) {
$this->didReceiveReloadSignal(SIGHUP);
$should_reload = false;
}		}

$futures = array();		$futures = array();
foreach ($this->getDaemonHandles() as $daemon) {		foreach ($this->getDaemonPools() as $pool) {
$daemon->update();		$pool->updatePool();
if ($daemon->isRunning()) {
$futures[] = $daemon->getFuture();
}

if ($daemon->isDone()) {		foreach ($pool->getFutures() as $future) {
$this->removeDaemon($daemon);		$futures[] = $future;
}		}
}		}

$this->updatePidfile();		$this->updatePidfile();
$this->updateAutoscale();		$this->updateMemory();

if ($futures) {		$this->waitForDaemonFutures($futures);
$iter = id(new FutureIterator($futures))
->setUpdateInterval(1);		if (!$futures) {
foreach ($iter as $future) {
break;
}
} else {
if ($this->inGracefulShutdown) {		if ($this->inGracefulShutdown) {
break;		break;
}		}
sleep(1);
}		}
}		}

exit($this->err);		exit($this->err);
}		}

private function addDaemon(PhutilDaemonHandle $daemon, array $config) {
$id = $daemon->getDaemonID();
$this->daemons[$id] = array(
'handle' => $daemon,
'config' => $config,
);

$autoscale_group = $this->getAutoscaleGroup($daemon);
if ($autoscale_group) {
$this->autoscale[$autoscale_group][$id] = true;
}

return $this;
}

private function removeDaemon(PhutilDaemonHandle $daemon) {
$id = $daemon->getDaemonID();

$autoscale_group = $this->getAutoscaleGroup($daemon);
if ($autoscale_group) {
unset($this->autoscale[$autoscale_group][$id]);
}

unset($this->daemons[$id]);

$daemon->didRemoveDaemon();

return $this;
}

private function getAutoscaleGroup(PhutilDaemonHandle $daemon) {
$id = $daemon->getDaemonID();
$autoscale = $this->daemons[$id]['config']['autoscale'];
return idx($autoscale, 'group');
}

private function getAutoscaleProperty($group_key, $key, $default = null) {
$config = $this->autoscaleConfig[$group_key]['autoscale'];
return idx($config, $key, $default);
}

public function didBeginWork(PhutilDaemonHandle $daemon) {
$id = $daemon->getDaemonID();
$busy = idx($this->daemons[$daemon->getDaemonID()], 'busy');
if (!$busy) {
$this->daemons[$id]['busy'] = time();
}
}

public function didBeginIdle(PhutilDaemonHandle $daemon) {
$id = $daemon->getDaemonID();
unset($this->daemons[$id]['busy']);
}

public function updateAutoscale() {
if ($this->inGracefulShutdown) {
return;
}

foreach ($this->autoscale as $group => $daemons) {		private function waitForDaemonFutures(array $futures) {
$scaleup_duration = $this->getAutoscaleProperty($group, 'up', 2);		assert_instances_of($futures, 'ExecFuture');
$max_pool_size = $this->getAutoscaleProperty($group, 'pool', 8);
$reserve = $this->getAutoscaleProperty($group, 'reserve', 0);

// Don't scale a group if it is already at the maximum pool size.		if ($futures) {
if (count($daemons) >= $max_pool_size) {		// TODO: This only wakes if any daemons actually exit. It would be a bit
continue;		// cleaner to wait on any I/O with Channels.
}		$iter = id(new FutureIterator($futures))
		->setUpdateInterval(1);
$should_scale = true;		foreach ($iter as $future) {
foreach ($daemons as $daemon_id => $ignored) {
$busy = idx($this->daemons[$daemon_id], 'busy');
if (!$busy) {
// At least one daemon in the group hasn't reported that it has
// started work.
$should_scale = false;
break;
}

if ((time() - $busy) < $scaleup_duration) {
// At least one daemon in the group was idle recently, so we have
// not fullly
$should_scale = false;
break;		break;
}		}
		} else {
		if (!$this->inGracefulShutdown) {
		sleep(1);
}		}

// If we have a configured memory reserve for this pool, it tells us that
// we should not scale up unless there's at least that much memory left
// on the system (for example, a reserve of 0.25 means that 25% of system
// memory must be free to autoscale).
if ($should_scale && $reserve) {
// On some systems this may be slightly more expensive than other
// checks, so only do it once we're prepared to scale up.
$memory = PhutilSystem::getSystemMemoryInformation();
$free_ratio = ($memory['free'] / $memory['total']);

// If we don't have enough free memory, don't scale.
if ($free_ratio <= $reserve) {
continue;
}		}
}		}

if ($should_scale) {		private function createDaemonPools() {
$config = $this->autoscaleConfig[$group];		$configs = $this->config['daemons'];

$config['autoscale']['clone'] = true;

$clone = new PhutilDaemonHandle(		$forced_options = array(
$this,
$config['class'],
$this->argv,
array(
'log' => $this->log,
'argv' => $config['argv'],
'load' => $this->libraries,		'load' => $this->libraries,
'autoscale' => $config['autoscale'],		'log' => $this->log,
));		);

$this->logMessage(
'AUTO',
pht(
'Scaling pool "%s" up to %s daemon(s).',
$group,
new PhutilNumber(count($daemons) + 1)));

$this->addDaemon($clone, $config);

// Don't scale more than one pool up per iteration. Otherwise, we could
// break the memory barrier if we have a lot of pools and scale them
// all up at once.
return;
}
}
}

public function didReceiveNotifySignal($signo) {
foreach ($this->getDaemonHandles() as $daemon) {
$daemon->didReceiveNotifySignal($signo);
}
}

public function didReceiveReloadSignal($signo) {		foreach ($configs as $config) {
foreach ($this->getDaemonHandles() as $daemon) {		$config = $forced_options + $config;
$daemon->didReceiveReloadSignal($signo);
}
}

public function didReceiveGracefulSignal($signo) {		$pool = PhutilDaemonPool::newFromConfig($config)
// If we receive SIGINT more than once, interpret it like SIGTERM.		->setOverseer($this)
if ($this->inGracefulShutdown) {		->setCommandLineArguments($this->argv);
return $this->didReceiveTerminalSignal($signo);
}
$this->inGracefulShutdown = true;

foreach ($this->getDaemonHandles() as $daemon) {		$this->pools[] = $pool;
$daemon->didReceiveGracefulSignal($signo);
}		}
}		}

public function didReceiveTerminalSignal($signo) {		private function getDaemonPools() {
$this->err = 128 + $signo;		return $this->pools;
if ($this->inAbruptShutdown) {
exit($this->err);
}		}
$this->inAbruptShutdown = true;

foreach ($this->getDaemonHandles() as $daemon) {
$daemon->didReceiveTerminalSignal($signo);
}
}

private function getDaemonHandles() {
return ipull($this->daemons, 'handle');
}

/**		/**
* Identify running daemons by examining the process table. This isn't		* Identify running daemons by examining the process table. This isn't
* completely reliable, but can be used as a fallback if the pid files fail		* completely reliable, but can be used as a fallback if the pid files fail
* or we end up with stray daemons by other means.		* or we end up with stray daemons by other means.
*		*
* Example output (array keys are process IDs):		* Example output (array keys are process IDs):
*		*
▲ Show 20 Lines • Show All 52 Lines • ▼ Show 20 Lines	public static function findRunningDaemons() {
return $results;		return $results;
}		}

private function updatePidfile() {		private function updatePidfile() {
if (!$this->piddir) {		if (!$this->piddir) {
return;		return;
}		}

$daemons = array();		$pidfile = $this->toDictionary();

foreach ($this->daemons as $daemon) {		if ($pidfile !== $this->lastPidfile) {
$handle = $daemon['handle'];		$this->lastPidfile = $pidfile;
$config = $daemon['config'];		$pidfile_path = $this->piddir.'/daemon.'.getmypid();
		Filesystem::writeFile($pidfile_path, phutil_json_encode($pidfile));
		}
		}

if (!$handle->isRunning()) {		public function toDictionary() {
		$daemons = array();
		foreach ($this->getDaemonPools() as $pool) {
		foreach ($pool->getDaemons() as $daemon) {
		if (!$daemon->isRunning()) {
continue;		continue;
}		}

$daemons[] = array(		$daemons[] = $daemon->toDictionary();
'pid' => $handle->getPID(),		}
'id' => $handle->getDaemonID(),
'config' => $config,
);
}		}

$pidfile = array(		return array(
'pid' => getmypid(),		'pid' => getmypid(),
'start' => $this->startEpoch,		'start' => $this->startEpoch,
'config' => $this->config,		'config' => $this->config,
'daemons' => $daemons,		'daemons' => $daemons,
);		);
		}

if ($pidfile !== $this->lastPidfile) {		private function updateMemory() {
$this->lastPidfile = $pidfile;		if (!$this->traceMemory) {
$pidfile_path = $this->piddir.'/daemon.'.getmypid();		return;
Filesystem::writeFile($pidfile_path, json_encode($pidfile));
}		}

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

public function logMessage($type, $message, $context = null) {		public function logMessage($type, $message, $context = null) {
if ($this->traceMode \|\| $this->verbose) {		if ($this->traceMode \|\| $this->verbose) {
error_log(date('Y-m-d g:i:s A').' ['.$type.'] '.$message);		error_log(date('Y-m-d g:i:s A').' ['.$type.'] '.$message);
}		}
}		}


		/* -( Signal Handling )---------------------------------------------------- */


		/**
		* @task signals
		*/
		private function installSignalHandlers() {
		$signals = array(
		SIGUSR2,
		SIGHUP,
		SIGINT,
		SIGTERM,
		);

		foreach ($signals as $signal) {
		pcntl_signal($signal, array($this, 'didReceiveSignal'));
		}
		}


		/**
		* @task signals
		*/
		public function didReceiveSignal($signo) {
		switch ($signo) {
		case SIGUSR2:
		$signal_type = self::SIGNAL_NOTIFY;
		break;
		case SIGHUP:
		$signal_type = self::SIGNAL_RELOAD;
		break;
		case SIGINT:
		// If we receive SIGINT more than once, interpret it like SIGTERM.
		if ($this->inGracefulShutdown) {
		return $this->didReceiveSignal(SIGTERM);
		}

		$this->inGracefulShutdown = true;
		$signal_type = self::SIGNAL_GRACEFUL;
		break;
		case SIGTERM:
		// If we receive SIGTERM more than once, terminate abruptly.
		$this->err = 128 + $signo;
		if ($this->inAbruptShutdown) {
		exit($this->err);
		}

		$this->inAbruptShutdown = true;
		$signal_type = self::SIGNAL_TERMINATE;
		break;
		default:
		throw new Exception(
		pht(
		'Signal handler called with unknown signal type ("%d")!',
		$signo));
		}

		foreach ($this->getDaemonPools() as $pool) {
		$pool->didReceiveSignal($signal_type, $signo);
		}
		}


		/* -( Daemon Modules )----------------------------------------------------- */


		private function getModules() {
		return $this->modules;
		}

		private function shouldReloadDaemons() {
		$modules = $this->getModules();

		$should_reload = false;
		foreach ($modules as $module) {
		try {
		// NOTE: Even if one module tells us to reload, we call the method on
		// each module anyway to make calls a little more predictable.

		if ($module->shouldReloadDaemons()) {
		$this->logMessage(
		'RELO',
		pht(
		'Reloading daemons (triggered by overseer module "%s").',
		get_class($module)));
		$should_reload = true;
		}
		} catch (Exception $ex) {
		phlog($ex);
		}
		}

		return $should_reload;
		}


}		}