PhageAgentAction.php
796cb1c2ee27stable

PhageAgentAction.php
View Options

			<?php

			abstract class PhageAgentAction
			extends PhageAction {

			private $future;
			private $channel;
			private $commands = array();
			private $commandKey = 0;

			private $isExiting = false;
			private $isActive = false;

			private $limit = 0;
			private $throttle = 0;
			private $waitUntil;

			private $queued = array();
			private $active = array();

			final public function isContainerAction() {
			return true;
			}

			protected function willAddAction(PhageAction $action) {
			// TODO: You should probably be able to add agents to an agent, is that
			// explicit or by making agents execute actions?

			if (!($action instanceof PhageExecuteAction)) {
			throw new Exception(
			pht(
			'Can only add execute actions to an agent.'));
			}

			if ($this->isExiting) {
			throw new Exception(
			pht(
			'You can not add new actions to an exiting agent.'));
			}

			$key = 'command/'.$this->commandKey++;

			$this->commands[$key] = array(
			'key' => $key,
			'command' => $action,
			);

			$this->queued[$key] = $key;
			}

			public function isActiveAgent() {
			return $this->isActive;
			}

			final public function setLimit($limit) {
			$this->limit = $limit;
			return $this;
			}

			final public function getLimit() {
			return $this->limit;
			}

			final public function setThrottle($throttle) {
			$this->throttle = $throttle;
			return $this;
			}

			final public function getThrottle() {
			return $this->throttle;
			}

			abstract protected function newAgentFuture(PhutilCommandString $command);

			protected function getAllWaitingChannels() {
			$channels = array();

			if ($this->isActiveAgent()) {
			$channels[] = $this->channel;
			}

			return $channels;
			}

			public function startAgent() {
			$bootloader = new PhagePHPAgentBootloader();

			$future = $this->newAgentFuture($bootloader->getBootCommand());

			$future->write($bootloader->getBootSequence(), $keep_open = true);

			$channel = new PhutilExecChannel($future);
			$channel->setStderrHandler(array($this, 'didReadAgentStderr'));

			$channel = new PhutilJSONProtocolChannel($channel);

			$this->future = $future;
			$this->channel = $channel;
			$this->isActive = true;
			}

			private function updateQueue() {
			// If we don't have anything waiting in queue, we have nothing to do.
			if (!$this->queued) {
			return false;
			}

			$now = microtime(true);

			// If we're throttling commands and recently started one, don't start
			// another one yet.
			$throttle = $this->getThrottle();
			if ($throttle) {
			if ($this->waitUntil && ($now < $this->waitUntil)) {
			return false;
			}
			}

			// If we're limiting parallelism and the active list is full, don't
			// start anything else yet.
			$limit = $this->getLimit();
			if ($limit) {
			if (count($this->active) >= $limit) {
			return false;
			}
			}

			// Move a command out of the queue and tell the agent to execute it.
			$key = head($this->queued);
			unset($this->queued[$key]);

			$this->active[$key] = $key;
			$command = $this->commands[$key]['command'];

			$channel = $this->getChannel();

			$channel->write(
			array(
			'type' => 'EXEC',
			'key' => $key,
			'command' => $command->getCommand()->getUnmaskedString(),
			'timeout' => $command->getTimeout(),
			));

			if ($throttle) {
			$this->waitUntil = ($now + $throttle);
			}

			return true;
			}

			private function getChannel() {
			return $this->channel;
			}

			public function updateAgent() {
			if (!$this->isActiveAgent()) {
			return;
			}

			$channel = $this->channel;

			while (true) {
			do {
			$did_update = $this->updateQueue();
			} while ($did_update);

			$is_open = $channel->update();

			$message = $channel->read();
			if ($message !== null) {
			switch ($message['type']) {
			case 'TEXT':
			$key = $message['key'];
			$this->writeOutput($key, $message['kind'], $message['text']);
			break;
			case 'RSLV':
			$key = $message['key'];
			$command = $this->commands[$key]['command'];

			$this->writeOutput($key, 'stdout', $message['stdout']);
			$this->writeOutput($key, 'stderr', $message['stderr']);

			$exit_code = $message['err'];

			$command->setExitCode($exit_code);
			$command->setDidTimeout((bool)idx($message, 'timeout'));

			if ($exit_code != 0) {
			$exit_code = $this->formatOutput(
			pht(
			'Command ("%s") exited nonzero ("%s")!',
			$command->getCommand(),
			$exit_code),
			$command->getLabel());

			fprintf(STDOUT, '%s', $exit_code);
			}

			unset($this->active[$key]);

			if (!$this->active && !$this->queued) {
			$channel->write(
			array(
			'type' => 'EXIT',
			'key' => 'exit',
			));

			$this->isExiting = true;
			break;
			}
			}
			}

			if (!$is_open) {
			if ($this->isExiting) {
			$this->isActive = false;
			break;
			} else {
			throw new Exception(pht('Channel closed unexpectedly!'));
			}
			}

			if ($message === null) {
			break;
			}
			}
			}

			private function writeOutput($key, $kind, $text) {
			if (!strlen($text)) {
			return;
			}

			switch ($kind) {
			case 'stdout':
			$target = STDOUT;
			break;
			case 'stderr':
			$target = STDERR;
			break;
			default:
			throw new Exception(pht('Unknown output kind "%s".', $kind));
			}

			$command = $this->commands[$key]['command'];

			$label = $command->getLabel();
			if (!strlen($label)) {
			$label = pht('Unknown Command');
			}

			$text = $this->formatOutput($text, $label);
			fprintf($target, '%s', $text);
			}

			private function formatOutput($output, $context) {
			$output = phutil_split_lines($output, false);
			foreach ($output as $key => $line) {
			$output[$key] = tsprintf("[%s] %R\n", $context, $line);
			}
			$output = implode('', $output);

			return $output;
			}

			public function didReadAgentStderr($channel, $stderr) {
			throw new Exception(
			pht(
			'Unexpected output on agent stderr: %s.',
			$stderr));
			}

			}

libphutil/src/phage/action/PhageAgentAction.php796cb1c2ee27stable

PhageAgentAction.phpView Options

libphutil/src/phage/action/PhageAgentAction.php
796cb1c2ee27stable

PhageAgentAction.php
View Options