Worker::stdout() - Code Metrics - Inspection of "Merge pull request #287 from chaplean/master" - brianium/paratest - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( a815d4...30f01d )

by Julian

created 2018-01-18 12:09 UTC

Worker::stdout() A

↳ Parent: Worker

Complexity

Conditions	1
Paths	1

Size

Total Lines	4
Code Lines	2

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	0
CRAP Score	2

Importance

Changes

Metric	Value
dl	0
loc	4
ccs	0
cts	2
cp	0
rs	10
c	0
b	0
f	0
cc	1
eloc	2
nc	1
nop	0
crap	2

<?php

declare(strict_types=1);

namespace ParaTest\Runners\PHPUnit;

use Exception;
use Symfony\Component\Process\PhpExecutableFinder;

class Worker
{
    private static $descriptorspec = [
       0 => ['pipe', 'r'],
       1 => ['pipe', 'w'],
       2 => ['pipe', 'w'],
    ];
    private $proc;
    private $pipes;
    private $inExecution = 0;
    private $isRunning = false;
    private $exitCode = null;
    private $commands = [];
    private $chunks = '';
    private $alreadyReadOutput = '';
    /**
     * @var ExecutableTest
     */
    private $currentlyExecuting;

    public function start(string $wrapperBinary, $token = 1, $uniqueToken = null)
    {
        $bin = 'PARATEST=1 ';
        if (is_numeric($token)) {
            $bin .= "TEST_TOKEN=$token ";
        }
        if ($uniqueToken) {
            $bin .= "UNIQUE_TEST_TOKEN=$uniqueToken ";
        }
        $finder = new PhpExecutableFinder();
        $bin .= $finder->find() . " \"$wrapperBinary\"";
        $pipes = [];
        $this->proc = proc_open($bin, self::$descriptorspec, $pipes);
        $this->pipes = $pipes;
        $this->isRunning = true;
    }

    public function stdout()
    {
        return $this->pipes[1];
    }

    public function execute(string $testCmd)
    {
        $this->checkStarted();
        $this->commands[] = $testCmd;
        fwrite($this->pipes[0], $testCmd . "\n");
        ++$this->inExecution;
    }

    public function assign(ExecutableTest $test, string $phpunit, array $phpunitOptions)
    {
        if ($this->currentlyExecuting !== null) {
            throw new Exception('Worker already has a test assigned - did you forget to call reset()?');
        }
        $this->currentlyExecuting = $test;
        $this->execute($test->command($phpunit, $phpunitOptions));
    }

    public function printFeedback(ResultPrinter $printer)
    {
        if ($this->currentlyExecuting !== null) {
            $printer->printFeedback($this->currentlyExecuting);
        }
    }

    public function reset()
    {
        $this->currentlyExecuting = null;
    }

    public function isStarted(): bool
    {
        return $this->proc !== null && $this->pipes !== null;
    }

    private function checkStarted()
    {
        if (!$this->isStarted()) {
            throw new \RuntimeException('You have to start the Worker first!');
        }
    }

    public function stop()
    {
        fwrite($this->pipes[0], "EXIT\n");
        fclose($this->pipes[0]);
    }

    /**
     * This is an utility function for tests.
     * Refactor or write it only in the test case.
     */
    public function waitForFinishedJob()
    {
        if ($this->inExecution === 0) {
            return;
        }
        $tellsUsItHasFinished = false;
        stream_set_blocking($this->pipes[1], true);
        while ($line = fgets($this->pipes[1])) {
            if (strstr($line, "FINISHED\n")) {
                $tellsUsItHasFinished = true;
                --$this->inExecution;
                break;
            }
        }
        if (!$tellsUsItHasFinished) {
            throw new \RuntimeException('The Worker terminated without finishing the job.');
        }
    }

    public function isFree(): bool
    {
        $this->checkNotCrashed();
        $this->updateStateFromAvailableOutput();

        return $this->inExecution === 0;
    }

    /**
     * @deprecated
     * This function consumes a lot of CPU while waiting for
     * the worker to finish. Use it only in testing paratest
     * itself.
     */
    public function waitForStop()
    {
        $status = proc_get_status($this->proc);
        while ($status['running']) {
            $status = proc_get_status($this->proc);
            $this->setExitCode($status);
        }
    }

    public function getCoverageFileName()
    {
        if ($this->currentlyExecuting !== null) {
            return $this->currentlyExecuting->getCoverageFileName();
        }
    }

    private function setExitCode(array $status)
    {
        if (!$status['running']) {
            if ($this->exitCode === null) {
                $this->exitCode = $status['exitcode'];
            }
        }
    }

    public function isRunning(): bool
    {
        $this->checkNotCrashed();
        $this->updateStateFromAvailableOutput();

        return $this->isRunning;
    }

    public function isCrashed(): bool
    {
        if (!$this->isStarted()) {
            return false;
        }
        $status = proc_get_status($this->proc);

        $this->updateStateFromAvailableOutput();
        if (!$this->isRunning) {
            return false;
        }

        $this->setExitCode($status);
        if ($this->exitCode === null) {
            return false;
        }

        return $this->exitCode !== 0;
    }

    private function checkNotCrashed()
    {
        if ($this->isCrashed()) {
            throw new \RuntimeException(
                'This worker has crashed. Last executed command: ' . end($this->commands) . PHP_EOL
                . 'Output:' . PHP_EOL
                . '----------------------' . PHP_EOL
                . $this->alreadyReadOutput . PHP_EOL
                . '----------------------' . PHP_EOL
                . $this->readAllStderr()
            );
        }
    }

    private function readAllStderr()
    {
        return stream_get_contents($this->pipes[2]);
    }

    /**
     * Have to read even incomplete lines to play nice with stream_select()
     * Otherwise it would continue to non-block because there are bytes to be read,
     * but fgets() won't pick them up.
     */
    private function updateStateFromAvailableOutput()
    {
        if (isset($this->pipes[1])) {
            stream_set_blocking($this->pipes[1], false);
            while ($chunk = fread($this->pipes[1], 4096)) {
                $this->chunks .= $chunk;
                $this->alreadyReadOutput .= $chunk;
            }
            $lines = explode("\n", $this->chunks);
            // last element is not a complete line,
            // becomes part of a line completed later
            $this->chunks = $lines[count($lines) - 1];
            unset($lines[count($lines) - 1]);
            // delivering complete lines to this Worker
            foreach ($lines as $line) {
                $line .= "\n";
                if (strstr($line, "FINISHED\n")) {
                    --$this->inExecution;
                }
                if (strstr($line, "EXITED\n")) {
                    $this->isRunning = false;
                }
            }
            stream_set_blocking($this->pipes[1], true);
        }
    }
}


1		<?php
2
3		declare(strict_types=1);
4
5		namespace ParaTest\Runners\PHPUnit;
6
7		use Exception;
8		use Symfony\Component\Process\PhpExecutableFinder;
9
10		class Worker
11		{
12		private static $descriptorspec = [
13		0 => ['pipe', 'r'],
14		1 => ['pipe', 'w'],
15		2 => ['pipe', 'w'],
16		];
17		private $proc;
18		private $pipes;
19		private $inExecution = 0;
20		private $isRunning = false;
21		private $exitCode = null;
22		private $commands = [];
23		private $chunks = '';
24		private $alreadyReadOutput = '';
25		/**
26		* @var ExecutableTest
27		*/
28		private $currentlyExecuting;
29
30	5	public function start(string $wrapperBinary, $token = 1, $uniqueToken = null)
31		{
32	5	$bin = 'PARATEST=1 ';
33	5	if (is_numeric($token)) {
34	5	$bin .= "TEST_TOKEN=$token ";
35		}
36	5	if ($uniqueToken) {
37		$bin .= "UNIQUE_TEST_TOKEN=$uniqueToken ";
38		}
39	5	$finder = new PhpExecutableFinder();
40	5	$bin .= $finder->find() . " \"$wrapperBinary\"";
41	5	$pipes = [];
42	5	$this->proc = proc_open($bin, self::$descriptorspec, $pipes);
43	5	$this->pipes = $pipes;
44	5	$this->isRunning = true;
45	5	}
46
47		public function stdout()
48		{
49		return $this->pipes[1];
50		}
51
52	4	public function execute(string $testCmd)
53		{
54	4	$this->checkStarted();
55	4	$this->commands[] = $testCmd;
56	4	fwrite($this->pipes[0], $testCmd . "\n");
57	4	++$this->inExecution;
58	4	}
59
60		public function assign(ExecutableTest $test, string $phpunit, array $phpunitOptions)
61		{
62		if ($this->currentlyExecuting !== null) {
63		throw new Exception('Worker already has a test assigned - did you forget to call reset()?');
64		}
65		$this->currentlyExecuting = $test;
66		$this->execute($test->command($phpunit, $phpunitOptions));
67		}
68
69		public function printFeedback(ResultPrinter $printer)
70		{
71		if ($this->currentlyExecuting !== null) {
72		$printer->printFeedback($this->currentlyExecuting);
73		}
74		}
75
76		public function reset()
77		{
78		$this->currentlyExecuting = null;
79		}
80
81	6	public function isStarted(): bool
82		{
83	6	return $this->proc !== null && $this->pipes !== null;
84		}
85
86	4	private function checkStarted()
87		{
88	4	if (!$this->isStarted()) {
89		throw new \RuntimeException('You have to start the Worker first!');
90		}
91	4	}
92
93	3	public function stop()
94		{
95	3	fwrite($this->pipes[0], "EXIT\n");
96	3	fclose($this->pipes[0]);
97	3	}
98
99		/**
100		* This is an utility function for tests.
101		* Refactor or write it only in the test case.
102		*/
103	2	public function waitForFinishedJob()
104		{
105	2	if ($this->inExecution === 0) {
106		return;
107		}
108	2	$tellsUsItHasFinished = false;
109	2	stream_set_blocking($this->pipes[1], true);
110	2	while ($line = fgets($this->pipes[1])) {
111	2	if (strstr($line, "FINISHED\n")) {
112	2	$tellsUsItHasFinished = true;
113	2	--$this->inExecution;
114	2	break;
115		}
116		}
117	2	if (!$tellsUsItHasFinished) {
118		throw new \RuntimeException('The Worker terminated without finishing the job.');
119		}
120	2	}
121
122	1	public function isFree(): bool
123		{
124	1	$this->checkNotCrashed();
125	1	$this->updateStateFromAvailableOutput();
126
127	1	return $this->inExecution === 0;
128		}
129
130		/**
131		* @deprecated
132		* This function consumes a lot of CPU while waiting for
133		* the worker to finish. Use it only in testing paratest
134		* itself.
135		*/
136	3	public function waitForStop()
137		{
138	3	$status = proc_get_status($this->proc);
139	3	while ($status['running']) {
140	3	$status = proc_get_status($this->proc);
141	3	$this->setExitCode($status);
142		}
143	3	}
144
145		public function getCoverageFileName()
146		{
147		if ($this->currentlyExecuting !== null) {
148		return $this->currentlyExecuting->getCoverageFileName();
149		}
150		}
151
152	4	private function setExitCode(array $status)
153		{
154	4	if (!$status['running']) {
155	3	if ($this->exitCode === null) {
156	3	$this->exitCode = $status['exitcode'];
157		}
158		}
159	4	}
160
161	1	public function isRunning(): bool
162		{
163	1	$this->checkNotCrashed();
164	1	$this->updateStateFromAvailableOutput();
165
166	1	return $this->isRunning;
167		}
168
169	3	public function isCrashed(): bool
170		{
171	3	if (!$this->isStarted()) {
172	1	return false;
173		}
174	3	$status = proc_get_status($this->proc);
175
176	3	$this->updateStateFromAvailableOutput();
177	3	if (!$this->isRunning) {
178	2	return false;
179		}
180
181	2	$this->setExitCode($status);
182	2	if ($this->exitCode === null) {
183	2	return false;
184		}
185
186		return $this->exitCode !== 0;
187		}
188
189	2	private function checkNotCrashed()
190		{
191	2	if ($this->isCrashed()) {
192		throw new \RuntimeException(
193		'This worker has crashed. Last executed command: ' . end($this->commands) . PHP_EOL
194		. 'Output:' . PHP_EOL
195		. '----------------------' . PHP_EOL
196		. $this->alreadyReadOutput . PHP_EOL
197		. '----------------------' . PHP_EOL
198		. $this->readAllStderr()
199		);
200		}
201	2	}
202
203		private function readAllStderr()
204		{
205		return stream_get_contents($this->pipes[2]);
206		}
207
208		/**
209		* Have to read even incomplete lines to play nice with stream_select()
210		* Otherwise it would continue to non-block because there are bytes to be read,
211		* but fgets() won't pick them up.
212		*/
213	3	private function updateStateFromAvailableOutput()
214		{
215	3	if (isset($this->pipes[1])) {
216	2	stream_set_blocking($this->pipes[1], false);
217	2	while ($chunk = fread($this->pipes[1], 4096)) {
218	1	$this->chunks .= $chunk;
219	1	$this->alreadyReadOutput .= $chunk;
220		}
221	2	$lines = explode("\n", $this->chunks);
222		// last element is not a complete line,
223		// becomes part of a line completed later
224	2	$this->chunks = $lines[count($lines) - 1];
225	2	unset($lines[count($lines) - 1]);
226		// delivering complete lines to this Worker
227	2	foreach ($lines as $line) {
228	1	$line .= "\n";
229	1	if (strstr($line, "FINISHED\n")) {
230		--$this->inExecution;
231		}
232	1	if (strstr($line, "EXITED\n")) {
233	1	$this->isRunning = false;
234		}
235		}
236	2	stream_set_blocking($this->pipes[1], true);
237		}
238	3	}
239		}
240

brianium / paratest

Push — master ( a815d4...30f01d )

Worker::stdout() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like