master

PhabricatorRepositoryPullLocalDaemon.php
View Options

			<?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]);

			$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 ($future_pool->hasFutures()) {
			while ($future_pool->hasFutures()) {
			$future = $future_pool->resolve();

			$this->stillWorking();

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

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

			continue;
			}

			$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 $consume (optional) 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();
			}

			}

Owner Packages

No Owners

phorge/src/applications/repository/daemon/PhabricatorRepositoryPullLocalDaemon.phpe81eff68e581HEAD -> master

PhabricatorRepositoryPullLocalDaemon.phpView Options

Owner Packages

phorge/src/applications/repository/daemon/PhabricatorRepositoryPullLocalDaemon.php
e81eff68e581HEAD -> master

PhabricatorRepositoryPullLocalDaemon.php
View Options