No OneTemporary
Actions

Size

31 KB

Referenced Files

None

Subscribers

None

View Options

	diff --git a/src/applications/harbormaster/future/HarbormasterExecFuture.php b/src/applications/harbormaster/future/HarbormasterExecFuture.php
	index 9dc4e172fd..1e6d4a5c2c 100644
	--- a/src/applications/harbormaster/future/HarbormasterExecFuture.php
	+++ b/src/applications/harbormaster/future/HarbormasterExecFuture.php
	@@ -1,50 +1,57 @@
	<?php

	final class HarbormasterExecFuture
	extends Future {

	private $future;
	private $stdout;
	private $stderr;

	public function setFuture(ExecFuture $future) {
	$this->future = $future;
	return $this;
	}

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

	public function setLogs(
	HarbormasterBuildLog $stdout,
	HarbormasterBuildLog $stderr) {
	$this->stdout = $stdout;
	$this->stderr = $stderr;
	return $this;
	}

	public function isReady() {
	+ if ($this->hasResult()) {
	+ return true;
	+ }
	+
	$future = $this->getFuture();

	- $result = $future->isReady();
	+ $is_ready = $future->isReady();

	list($stdout, $stderr) = $future->read();
	$future->discardBuffers();

	if ($this->stdout) {
	$this->stdout->append($stdout);
	}

	if ($this->stderr) {
	$this->stderr->append($stderr);
	}

	- return $result;
	- }
	+ if ($future->hasResult()) {
	+ $this->setResult($future->getResult());
	+ }
	+
	+ // TODO: This should probably be implemented as a FutureProxy; it will
	+ // not currently propagate exceptions or sockets properly.

	- protected function getResult() {
	- return $this->getFuture()->getResult();
	+ return $is_ready;
	}

	}
	diff --git a/src/applications/repository/daemon/PhabricatorRepositoryPullLocalDaemon.php b/src/applications/repository/daemon/PhabricatorRepositoryPullLocalDaemon.php
	index 125ee833f8..48eabf1545 100644
	--- a/src/applications/repository/daemon/PhabricatorRepositoryPullLocalDaemon.php
	+++ b/src/applications/repository/daemon/PhabricatorRepositoryPullLocalDaemon.php
	@@ -1,595 +1,611 @@
	<?php

	/**
	* Run pull commands on local working copies to keep them up to date. This
	* daemon handles all repository types.
	*
	* By default, the daemon pulls every repository. If you want it to be
	* responsible for only some repositories, you can launch it with a list of
	* repositories:
	*
	* ./phd launch repositorypulllocal -- X Q Z
	*
	* You can also launch a daemon which is responsible for all //but// one or
	* more repositories:
	*
	* ./phd launch repositorypulllocal -- --not A --not B
	*
	* If you have a very large number of repositories and some aren't being pulled
	* as frequently as you'd like, you can either change the pull frequency of
	* the less-important repositories to a larger number (so the daemon will skip
	* them more often) or launch one daemon for all the less-important repositories
	* and one for the more important repositories (or one for each more important
	* repository).
	*
	* @task pull Pulling Repositories
	*/
	final class PhabricatorRepositoryPullLocalDaemon
	extends PhabricatorDaemon {

	private $statusMessageCursor = 0;

	/* -( Pulling Repositories )----------------------------------------------- */


	/**
	* @task pull
	*/
	protected function run() {
	$argv = $this->getArgv();
	array_unshift($argv, __CLASS__);
	$args = new PhutilArgumentParser($argv);
	$args->parse(
	array(
	array(
	'name' => 'no-discovery',
	'help' => pht('Pull only, without discovering commits.'),
	),
	array(
	'name' => 'not',
	'param' => 'repository',
	'repeat' => true,
	'help' => pht('Do not pull __repository__.'),
	),
	array(
	'name' => 'repositories',
	'wildcard' => true,
	'help' => pht('Pull specific __repositories__ instead of all.'),
	),
	));

	$no_discovery = $args->getArg('no-discovery');
	$include = $args->getArg('repositories');
	$exclude = $args->getArg('not');

	// Each repository has an individual pull frequency; after we pull it,
	// wait that long to pull it again. When we start up, try to pull everything
	// serially.
	$retry_after = array();

	$min_sleep = 15;
	$max_sleep = phutil_units('5 minutes in seconds');
	$max_futures = 4;
	$futures = array();
	$queue = array();

	+ $future_pool = new FuturePool();
	+
	+ $future_pool->getIteratorTemplate()
	+ ->setUpdateInterval($min_sleep);
	+
	$sync_wait = phutil_units('2 minutes in seconds');
	$last_sync = array();

	while (!$this->shouldExit()) {
	PhabricatorCaches::destroyRequestCache();
	$device = AlmanacKeys::getLiveDevice();

	$pullable = $this->loadPullableRepositories($include, $exclude, $device);

	// If any repositories have the NEEDS_UPDATE flag set, pull them
	// as soon as possible.
	$need_update_messages = $this->loadRepositoryUpdateMessages(true);
	foreach ($need_update_messages as $message) {
	$repo = idx($pullable, $message->getRepositoryID());
	if (!$repo) {
	continue;
	}

	$this->log(
	pht(
	'Got an update message for repository "%s"!',
	$repo->getMonogram()));

	$retry_after[$message->getRepositoryID()] = time();
	}

	if ($device) {
	$unsynchronized = $this->loadUnsynchronizedRepositories($device);
	$now = PhabricatorTime::getNow();
	foreach ($unsynchronized as $repository) {
	$id = $repository->getID();

	$this->log(
	pht(
	'Cluster repository ("%s") is out of sync on this node ("%s").',
	$repository->getDisplayName(),
	$device->getName()));

	// Don't let out-of-sync conditions trigger updates too frequently,
	// since we don't want to get trapped in a death spiral if sync is
	// failing.
	$sync_at = idx($last_sync, $id, 0);
	$wait_duration = ($now - $sync_at);
	if ($wait_duration < $sync_wait) {
	$this->log(
	pht(
	'Skipping forced out-of-sync update because the last update '.
	'was too recent (%s seconds ago).',
	$wait_duration));
	continue;
	}

	$last_sync[$id] = $now;
	$retry_after[$id] = $now;
	}
	}

	// If any repositories were deleted, remove them from the retry timer map
	// so we don't end up with a retry timer that never gets updated and
	// causes us to sleep for the minimum amount of time.
	$retry_after = array_select_keys(
	$retry_after,
	array_keys($pullable));

	// Figure out which repositories we need to queue for an update.
	foreach ($pullable as $id => $repository) {
	$now = PhabricatorTime::getNow();
	$display_name = $repository->getDisplayName();

	if (isset($futures[$id])) {
	$this->log(
	pht(
	'Repository "%s" is currently updating.',
	$display_name));
	continue;
	}

	if (isset($queue[$id])) {
	$this->log(
	pht(
	'Repository "%s" is already queued.',
	$display_name));
	continue;
	}

	$after = idx($retry_after, $id);
	if (!$after) {
	$smart_wait = $repository->loadUpdateInterval($min_sleep);
	$last_update = $this->loadLastUpdate($repository);

	$after = $last_update + $smart_wait;
	$retry_after[$id] = $after;

	$this->log(
	pht(
	'Scheduling repository "%s" with an update window of %s '.
	'second(s). Last update was %s second(s) ago.',
	$display_name,
	new PhutilNumber($smart_wait),
	new PhutilNumber($now - $last_update)));
	}

	if ($after > time()) {
	$this->log(
	pht(
	'Repository "%s" is not due for an update for %s second(s).',
	$display_name,
	new PhutilNumber($after - $now)));
	continue;
	}

	$this->log(
	pht(
	'Scheduling repository "%s" for an update (%s seconds overdue).',
	$display_name,
	new PhutilNumber($now - $after)));

	$queue[$id] = $after;
	}

	// Process repositories in the order they became candidates for updates.
	asort($queue);

	// Dequeue repositories until we hit maximum parallelism.
	while ($queue && (count($futures) < $max_futures)) {
	foreach ($queue as $id => $time) {
	$repository = idx($pullable, $id);
	if (!$repository) {
	$this->log(
	pht('Repository %s is no longer pullable; skipping.', $id));
	unset($queue[$id]);
	continue;
	}

	$display_name = $repository->getDisplayName();
	$this->log(
	pht(
	'Starting update for repository "%s".',
	$display_name));

	unset($queue[$id]);
	- $futures[$id] = $this->buildUpdateFuture(
	+
	+ $future = $this->buildUpdateFuture(
	$repository,
	$no_discovery);

	+ $futures[$id] = $future->getFutureKey();
	+
	+ $future_pool->addFuture($future);
	break;
	}
	}

	if ($queue) {
	$this->log(
	pht(
	'Not enough process slots to schedule the other %s '.
	'repository(s) for updates yet.',
	phutil_count($queue)));
	}

	- if ($futures) {
	- $iterator = id(new FutureIterator($futures))
	- ->setUpdateInterval($min_sleep);
	+ if ($future_pool->hasFutures()) {
	+ while ($future_pool->hasFutures()) {
	+ $future = $future_pool->resolve();

	- foreach ($iterator as $id => $future) {
	$this->stillWorking();

	if ($future === null) {
	$this->log(pht('Waiting for updates to complete...'));
	- $this->stillWorking();

	if ($this->loadRepositoryUpdateMessages()) {
	$this->log(pht('Interrupted by pending updates!'));
	break;
	}

	continue;
	}

	- unset($futures[$id]);
	- $retry_after[$id] = $this->resolveUpdateFuture(
	- $pullable[$id],
	+ $future_key = $future->getFutureKey();
	+ $repository_id = null;
	+ foreach ($futures as $id => $key) {
	+ if ($key === $future_key) {
	+ $repository_id = $id;
	+ unset($futures[$id]);
	+ break;
	+ }
	+ }
	+
	+ $retry_after[$repository_id] = $this->resolveUpdateFuture(
	+ $pullable[$repository_id],
	$future,
	$min_sleep);

	// We have a free slot now, so go try to fill it.
	break;
	}

	// Jump back into prioritization if we had any futures to deal with.
	continue;
	}

	$should_hibernate = $this->waitForUpdates($max_sleep, $retry_after);
	if ($should_hibernate) {
	break;
	}
	}

	}


	/**
	* @task pull
	*/
	private function buildUpdateFuture(
	PhabricatorRepository $repository,
	$no_discovery) {

	$bin = dirname(phutil_get_library_root('phabricator')).'/bin/repository';

	$flags = array();
	if ($no_discovery) {
	$flags[] = '--no-discovery';
	}

	$monogram = $repository->getMonogram();
	$future = new ExecFuture('%s update %Ls -- %s', $bin, $flags, $monogram);

	// Sometimes, the underlying VCS commands will hang indefinitely. We've
	// observed this occasionally with GitHub, and other users have observed
	// it with other VCS servers.

	// To limit the damage this can cause, kill the update out after a
	// reasonable amount of time, under the assumption that it has hung.

	// Since it's hard to know what a "reasonable" amount of time is given that
	// users may be downloading a repository full of pirated movies over a
	// potato, these limits are fairly generous. Repositories exceeding these
	// limits can be manually pulled with `bin/repository update X`, which can
	// just run for as long as it wants.

	if ($repository->isImporting()) {
	$timeout = phutil_units('4 hours in seconds');
	} else {
	$timeout = phutil_units('15 minutes in seconds');
	}

	$future->setTimeout($timeout);

	// The default TERM inherited by this process is "unknown", which causes PHP
	// to produce a warning upon startup. Override it to squash this output to
	// STDERR.
	$future->updateEnv('TERM', 'dumb');

	return $future;
	}


	/**
	* Check for repositories that should be updated immediately.
	*
	* With the `$consume` flag, an internal cursor will also be incremented so
	* that these messages are not returned by subsequent calls.
	*
	* @param bool Pass `true` to consume these messages, so the process will
	* not see them again.
	* @return list<wild> Pending update messages.
	*
	* @task pull
	*/
	private function loadRepositoryUpdateMessages($consume = false) {
	$type_need_update = PhabricatorRepositoryStatusMessage::TYPE_NEEDS_UPDATE;
	$messages = id(new PhabricatorRepositoryStatusMessage())->loadAllWhere(
	'statusType = %s AND id > %d',
	$type_need_update,
	$this->statusMessageCursor);

	// Keep track of messages we've seen so that we don't load them again.
	// If we reload messages, we can get stuck a loop if we have a failing
	// repository: we update immediately in response to the message, but do
	// not clear the message because the update does not succeed. We then
	// immediately retry. Instead, messages are only permitted to trigger
	// an immediate update once.

	if ($consume) {
	foreach ($messages as $message) {
	$this->statusMessageCursor = max(
	$this->statusMessageCursor,
	$message->getID());
	}
	}

	return $messages;
	}


	/**
	* @task pull
	*/
	private function loadLastUpdate(PhabricatorRepository $repository) {
	$table = new PhabricatorRepositoryStatusMessage();
	$conn = $table->establishConnection('r');

	$epoch = queryfx_one(
	$conn,
	'SELECT MAX(epoch) last_update FROM %T
	WHERE repositoryID = %d
	AND statusType IN (%Ls)',
	$table->getTableName(),
	$repository->getID(),
	array(
	PhabricatorRepositoryStatusMessage::TYPE_INIT,
	PhabricatorRepositoryStatusMessage::TYPE_FETCH,
	));

	if ($epoch) {
	return (int)$epoch['last_update'];
	}

	return PhabricatorTime::getNow();
	}

	/**
	* @task pull
	*/
	private function loadPullableRepositories(
	array $include,
	array $exclude,
	AlmanacDevice $device = null) {

	$query = id(new PhabricatorRepositoryQuery())
	->setViewer($this->getViewer());

	if ($include) {
	$query->withIdentifiers($include);
	}

	$repositories = $query->execute();
	$repositories = mpull($repositories, null, 'getPHID');

	if ($include) {
	$map = $query->getIdentifierMap();
	foreach ($include as $identifier) {
	if (empty($map[$identifier])) {
	throw new Exception(
	pht(
	'No repository "%s" exists!',
	$identifier));
	}
	}
	}

	if ($exclude) {
	$xquery = id(new PhabricatorRepositoryQuery())
	->setViewer($this->getViewer())
	->withIdentifiers($exclude);

	$excluded_repos = $xquery->execute();
	$xmap = $xquery->getIdentifierMap();

	foreach ($exclude as $identifier) {
	if (empty($xmap[$identifier])) {
	throw new Exception(
	pht(
	'No repository "%s" exists!',
	$identifier));
	}
	}

	foreach ($excluded_repos as $excluded_repo) {
	unset($repositories[$excluded_repo->getPHID()]);
	}
	}

	foreach ($repositories as $key => $repository) {
	if (!$repository->isTracked()) {
	unset($repositories[$key]);
	}
	}

	$viewer = $this->getViewer();

	$filter = id(new DiffusionLocalRepositoryFilter())
	->setViewer($viewer)
	->setDevice($device)
	->setRepositories($repositories);

	$repositories = $filter->execute();

	foreach ($filter->getRejectionReasons() as $reason) {
	$this->log($reason);
	}

	// Shuffle the repositories, then re-key the array since shuffle()
	// discards keys. This is mostly for startup, we'll use soft priorities
	// later.
	shuffle($repositories);
	$repositories = mpull($repositories, null, 'getID');

	return $repositories;
	}


	/**
	* @task pull
	*/
	private function resolveUpdateFuture(
	PhabricatorRepository $repository,
	ExecFuture $future,
	$min_sleep) {

	$display_name = $repository->getDisplayName();

	$this->log(pht('Resolving update for "%s".', $display_name));

	try {
	list($stdout, $stderr) = $future->resolvex();
	} catch (Exception $ex) {
	$proxy = new PhutilProxyException(
	pht(
	'Error while updating the "%s" repository.',
	$display_name),
	$ex);
	phlog($proxy);

	$smart_wait = $repository->loadUpdateInterval($min_sleep);
	return PhabricatorTime::getNow() + $smart_wait;
	}

	if (strlen($stderr)) {
	$stderr_msg = pht(
	'Unexpected output while updating repository "%s": %s',
	$display_name,
	$stderr);
	phlog($stderr_msg);
	}

	$smart_wait = $repository->loadUpdateInterval($min_sleep);

	$this->log(
	pht(
	'Based on activity in repository "%s", considering a wait of %s '.
	'seconds before update.',
	$display_name,
	new PhutilNumber($smart_wait)));

	return PhabricatorTime::getNow() + $smart_wait;
	}



	/**
	* Sleep for a short period of time, waiting for update messages from the
	*
	*
	* @task pull
	*/
	private function waitForUpdates($min_sleep, array $retry_after) {
	$this->log(
	pht('No repositories need updates right now, sleeping...'));

	$sleep_until = time() + $min_sleep;
	if ($retry_after) {
	$sleep_until = min($sleep_until, min($retry_after));
	}

	while (($sleep_until - time()) > 0) {
	$sleep_duration = ($sleep_until - time());

	if ($this->shouldHibernate($sleep_duration)) {
	return true;
	}

	$this->log(
	pht(
	'Sleeping for %s more second(s)...',
	new PhutilNumber($sleep_duration)));

	$this->sleep(1);

	if ($this->shouldExit()) {
	$this->log(pht('Awakened from sleep by graceful shutdown!'));
	return false;
	}

	if ($this->loadRepositoryUpdateMessages()) {
	$this->log(pht('Awakened from sleep by pending updates!'));
	break;
	}
	}

	return false;
	}

	private function loadUnsynchronizedRepositories(AlmanacDevice $device) {
	$viewer = $this->getViewer();
	$table = new PhabricatorRepositoryWorkingCopyVersion();
	$conn = $table->establishConnection('r');

	$our_versions = queryfx_all(
	$conn,
	'SELECT repositoryPHID, repositoryVersion FROM %R WHERE devicePHID = %s',
	$table,
	$device->getPHID());
	$our_versions = ipull($our_versions, 'repositoryVersion', 'repositoryPHID');

	$max_versions = queryfx_all(
	$conn,
	'SELECT repositoryPHID, MAX(repositoryVersion) maxVersion FROM %R
	GROUP BY repositoryPHID',
	$table);
	$max_versions = ipull($max_versions, 'maxVersion', 'repositoryPHID');

	$unsynchronized_phids = array();
	foreach ($max_versions as $repository_phid => $max_version) {
	$our_version = idx($our_versions, $repository_phid);
	if (($our_version === null) \|\| ($our_version < $max_version)) {
	$unsynchronized_phids[] = $repository_phid;
	}
	}

	if (!$unsynchronized_phids) {
	return array();
	}

	return id(new PhabricatorRepositoryQuery())
	->setViewer($viewer)
	->withPHIDs($unsynchronized_phids)
	->execute();
	}

	}
	diff --git a/src/infrastructure/daemon/PhutilDaemonOverseer.php b/src/infrastructure/daemon/PhutilDaemonOverseer.php
	index 44e0c41384..77f6f6a20a 100644
	--- a/src/infrastructure/daemon/PhutilDaemonOverseer.php
	+++ b/src/infrastructure/daemon/PhutilDaemonOverseer.php
	@@ -1,405 +1,410 @@
	<?php

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

	private $argv;
	private static $instance;

	private $config;
	private $pools = array();
	private $traceMode;
	private $traceMemory;
	private $daemonize;
	private $log;
	private $libraries = array();
	private $modules = array();
	private $verbose;
	private $startEpoch;
	private $autoscale = 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;

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

	$args = new PhutilArgumentParser($argv);
	$args->setTagline(pht('daemon overseer'));
	$args->setSynopsis(<<<EOHELP
	launch_daemon.php [__options__] __daemon__
	Launch and oversee an instance of __daemon__.
	EOHELP
	);
	$args->parseStandardArguments();
	$args->parse(
	array(
	array(
	'name' => 'trace-memory',
	'help' => pht('Enable debug memory tracing.'),
	),
	array(
	'name' => 'verbose',
	'help' => pht('Enable verbose activity logging.'),
	),
	array(
	'name' => 'label',
	'short' => 'l',
	'param' => 'label',
	'help' => pht(
	'Optional process label. Makes "%s" nicer, no behavioral effects.',
	'ps'),
	),
	));
	$argv = array();

	if ($args->getArg('trace')) {
	$this->traceMode = true;
	$argv[] = '--trace';
	}

	if ($args->getArg('trace-memory')) {
	$this->traceMode = true;
	$this->traceMemory = true;
	$argv[] = '--trace-memory';
	}
	$verbose = $args->getArg('verbose');
	if ($verbose) {
	$this->verbose = true;
	$argv[] = '--verbose';
	}

	$label = $args->getArg('label');
	if ($label) {
	$argv[] = '-l';
	$argv[] = $label;
	}

	$this->argv = $argv;

	if (function_exists('posix_isatty') && posix_isatty(STDIN)) {
	fprintf(STDERR, pht('Reading daemon configuration from stdin...')."\n");
	}
	$config = @file_get_contents('php://stdin');
	$config = id(new PhutilJSONParser())->parse($config);

	$this->libraries = idx($config, 'load');
	$this->log = idx($config, 'log');
	$this->daemonize = idx($config, 'daemonize');

	$this->config = $config;

	if (self::$instance) {
	throw new Exception(
	pht('You may not instantiate more than one Overseer per process.'));
	}

	self::$instance = $this;

	$this->startEpoch = time();

	if (!idx($config, 'daemons')) {
	throw new PhutilArgumentUsageException(
	pht('You must specify at least one daemon to start!'));
	}

	if ($this->log) {
	// NOTE: Now that we're committed to daemonizing, redirect the error
	// log if we have a `--log` parameter. Do this at the last moment
	// so as many setup issues as possible are surfaced.
	ini_set('error_log', $this->log);
	}

	if ($this->daemonize) {
	// We need to get rid of these or the daemon will hang when we TERM it
	// waiting for something to read the buffers. TODO: Learn how unix works.
	fclose(STDOUT);
	fclose(STDERR);
	ob_start();

	$pid = pcntl_fork();
	if ($pid === -1) {
	throw new Exception(pht('Unable to fork!'));
	} else if ($pid) {
	exit(0);
	}

	$sid = posix_setsid();
	if ($sid <= 0) {
	throw new Exception(pht('Failed to create new process session!'));
	}
	}

	$this->logMessage(
	'OVER',
	pht(
	'Started new daemon overseer (with PID "%s").',
	getmypid()));

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

	$this->installSignalHandlers();
	}

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

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

	+ $future_pool = $this->getFuturePool();
	+
	while (true) {
	if ($this->shouldReloadDaemons()) {
	$this->didReceiveSignal(SIGHUP);
	}

	- $futures = array();
	-
	$running_pools = false;
	foreach ($this->getDaemonPools() as $pool) {
	$pool->updatePool();

	if (!$this->shouldShutdown()) {
	if ($pool->isHibernating()) {
	if ($this->shouldWakePool($pool)) {
	$pool->wakeFromHibernation();
	}
	}
	}

	foreach ($pool->getFutures() as $future) {
	- $futures[] = $future;
	+ $future_pool->addFuture($future);
	}

	if ($pool->getDaemons()) {
	$running_pools = true;
	}
	}

	$this->updateMemory();

	- $this->waitForDaemonFutures($futures);
	+ if ($future_pool->hasFutures()) {
	+ $future_pool->resolve();
	+ } else {
	+ if (!$this->shouldShutdown()) {
	+ sleep(1);
	+ }
	+ }

	- if (!$futures && !$running_pools) {
	+ if (!$future_pool->hasFutures() && !$running_pools) {
	if ($this->shouldShutdown()) {
	break;
	}
	}
	}

	exit($this->err);
	}

	+ private function getFuturePool() {
	+ if (!$this->futurePool) {
	+ $pool = new FuturePool();

	- private function waitForDaemonFutures(array $futures) {
	- assert_instances_of($futures, 'ExecFuture');
	+ // TODO: This only wakes if any daemons actually exit, or 1 second
	+ // passes. It would be a bit cleaner to wait on any I/O, but Futures
	+ // currently can't do that.

	- if ($futures) {
	- // TODO: This only wakes if any daemons actually exit. It would be a bit
	- // cleaner to wait on any I/O with Channels.
	- $iter = id(new FutureIterator($futures))
	+ $pool->getIteratorTemplate()
	->setUpdateInterval(1);
	- foreach ($iter as $future) {
	- break;
	- }
	- } else {
	- if (!$this->shouldShutdown()) {
	- sleep(1);
	- }
	+
	+ $this->futurePool = $pool;
	}
	+ return $this->futurePool;
	}

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

	$forced_options = array(
	'load' => $this->libraries,
	'log' => $this->log,
	);

	foreach ($configs as $config) {
	$config = $forced_options + $config;

	$pool = PhutilDaemonPool::newFromConfig($config)
	->setOverseer($this)
	->setCommandLineArguments($this->argv);

	$this->pools[] = $pool;
	}
	}

	private function getDaemonPools() {
	return $this->pools;
	}

	private function updateMemory() {
	if (!$this->traceMemory) {
	return;
	}

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

	public function logMessage($type, $message, $context = null) {
	$always_log = false;
	switch ($type) {
	case 'OVER':
	case 'SGNL':
	case 'PIDF':
	$always_log = true;
	break;
	}

	if ($always_log \|\| $this->traceMode \|\| $this->verbose) {
	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) {
	$this->logMessage(
	'SGNL',
	pht(
	'Overseer ("%d") received signal %d ("%s").',
	getmypid(),
	$signo,
	phutil_get_signal_name($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;
	}

	private function shouldWakePool(PhutilDaemonPool $pool) {
	$modules = $this->getModules();

	$should_wake = false;
	foreach ($modules as $module) {
	try {
	if ($module->shouldWakePool($pool)) {
	$this->logMessage(
	'WAKE',
	pht(
	'Waking pool "%s" (triggered by overseer module "%s").',
	$pool->getPoolLabel(),
	get_class($module)));
	$should_wake = true;
	}
	} catch (Exception $ex) {
	phlog($ex);
	}
	}

	return $should_wake;
	}

	private function shouldShutdown() {
	return $this->inGracefulShutdown \|\| $this->inAbruptShutdown;
	}

	}

File Metadata

Mime Type: text/x-diff
Expires: Sun, Jul 27, 11:50 AM (6 d, 15 h ago)
Storage Engine: blob
Storage Format: Raw Data
Storage Handle: 185709
Default Alt Text: (31 KB)

No OneTemporaryActions

View Options

File Metadata

Event Timeline

No OneTemporary
Actions