Skip to content

Commit 1ac7688

Browse files
ondrejmirtesclaude
ondrejmirtes
and
claude
committed
Introduce Process interface and ProcessBase for parallel workers
Splits the parallel worker into a process-creation-agnostic half (ProcessBase: the TCP/NDJSON connection and timeout timer) and the process-creation half. The Process interface is what ParallelAnalyser and ProcessPool talk to; the existing react/child-process implementation is renamed to SpawnedProcess. Sole implementation for now — a pcntl_fork() based one follows. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
1 parent f8dd187 commit 1ac7688

5 files changed

Lines changed: 239 additions & 134 deletions

File tree

‎build/baseline-7.4.neon‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -11,9 +11,9 @@ parameters:
1111
path: ../src/Parallel/ParallelAnalyser.php
1212

1313
-
14-
message: "#^Class PHPStan\\\\Parallel\\\\Process has an uninitialized property \\$process\\. Give it default value or assign it in the constructor\\.$#"
14+
message: "#^Class PHPStan\\\\Parallel\\\\SpawnedProcess has an uninitialized property \\$process\\. Give it default value or assign it in the constructor\\.$#"
1515
count: 1
16-
path: ../src/Parallel/Process.php
16+
path: ../src/Parallel/SpawnedProcess.php
1717
-
1818
message: "#^Class PHPStan\\\\PhpDoc\\\\ResolvedPhpDocBlock has an uninitialized property \\$phpDocNodes\\. Give it default value or assign it in the constructor\\.$#"
1919
count: 1

‎src/Parallel/ParallelAnalyser.php‎

Lines changed: 23 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -190,13 +190,13 @@ public function analyse(
190190
$commandOptions[] = escapeshellarg($insteadOfFile);
191191
}
192192

193-
$process = new Process(ProcessHelper::getWorkerCommand(
193+
$process = $this->createProcess(
194+
$loop,
194195
$mainScript,
195-
'worker',
196196
$projectConfigFile,
197197
$commandOptions,
198198
$input,
199-
), $loop, $this->processTimeout);
199+
);
200200
$process->start(function (array $json) use ($process, &$internalErrors, &$errors, &$filteredPhpErrors, &$allPhpErrors, &$locallyIgnoredErrors, &$linesToIgnore, &$unmatchedLineIgnores, &$collectedData, &$dependencies, &$usedTraitDependencies, &$exportedNodes, &$peakMemoryUsages, &$jobs, $postFileCallback, &$internalErrorsCount, &$reachedInternalErrorsCountLimit, $processIdentifier, $onFileAnalysisHandler, &$allProcessedFiles): void {
201201
$fileErrors = [];
202202
foreach ($json['errors'] as $jsonError) {
@@ -355,4 +355,24 @@ public function analyse(
355355
return $deferred->promise();
356356
}
357357

358+
/**
359+
* @param string[] $commandOptions
360+
*/
361+
private function createProcess(
362+
LoopInterface $loop,
363+
string $mainScript,
364+
?string $projectConfigFile,
365+
array $commandOptions,
366+
InputInterface $input,
367+
): Process
368+
{
369+
return new SpawnedProcess(ProcessHelper::getWorkerCommand(
370+
$mainScript,
371+
'worker',
372+
$projectConfigFile,
373+
$commandOptions,
374+
$input,
375+
), $loop, $this->processTimeout);
376+
}
377+
358378
}

‎src/Parallel/Process.php‎

Lines changed: 14 additions & 129 deletions
Original file line numberDiff line numberDiff line change
@@ -2,151 +2,36 @@
22

33
namespace PHPStan\Parallel;
44

5-
use PHPStan\ShouldNotHappenException;
6-
use React\EventLoop\LoopInterface;
7-
use React\EventLoop\TimerInterface;
85
use React\Stream\ReadableStreamInterface;
96
use React\Stream\WritableStreamInterface;
107
use Throwable;
11-
use function fclose;
12-
use function rewind;
13-
use function sprintf;
14-
use function stream_get_contents;
15-
use function tmpfile;
168

17-
final class Process
9+
/**
10+
* A parallel analysis worker as seen by ParallelAnalyser / ProcessPool.
11+
*
12+
* Implementations differ only in how the worker process comes to life:
13+
* SpawnedProcess spawns a fresh PHP process via react/child-process,
14+
* ForkedProcess forks the already-booted main process via pcntl_fork(). Both
15+
* then speak the same TCP + NDJSON protocol, so request()/quit()/
16+
* bindConnection() behave identically and live in ProcessBase.
17+
*/
18+
interface Process
1819
{
1920

20-
private \React\ChildProcess\Process $process;
21-
22-
private ?WritableStreamInterface $in = null;
23-
24-
/** @var resource */
25-
private $stdOut;
26-
27-
/** @var resource */
28-
private $stdErr;
29-
30-
/** @var callable(mixed[] $json) : void */
31-
private $onData;
32-
33-
/** @var callable(Throwable $exception): void */
34-
private $onError;
35-
36-
private ?TimerInterface $timer = null;
37-
38-
public function __construct(
39-
private string $command,
40-
private LoopInterface $loop,
41-
private float $timeoutSeconds,
42-
)
43-
{
44-
}
45-
4621
/**
4722
* @param callable(mixed[] $json) : void $onData
4823
* @param callable(Throwable $exception): void $onError
4924
* @param callable(?int $exitCode, string $output) : void $onExit
5025
*/
51-
public function start(callable $onData, callable $onError, callable $onExit): void
52-
{
53-
$tmpStdOut = tmpfile();
54-
if ($tmpStdOut === false) {
55-
throw new ShouldNotHappenException('Failed creating temp file for stdout.');
56-
}
57-
$tmpStdErr = tmpfile();
58-
if ($tmpStdErr === false) {
59-
throw new ShouldNotHappenException('Failed creating temp file for stderr.');
60-
}
61-
$this->stdOut = $tmpStdOut;
62-
$this->stdErr = $tmpStdErr;
63-
$this->process = new \React\ChildProcess\Process($this->command, fds: [
64-
1 => $this->stdOut,
65-
2 => $this->stdErr,
66-
]);
67-
$this->process->start($this->loop);
68-
$this->onData = $onData;
69-
$this->onError = $onError;
70-
$this->process->on('exit', function ($exitCode) use ($onExit): void {
71-
$this->cancelTimer();
72-
73-
$output = '';
74-
rewind($this->stdOut);
75-
$output .= stream_get_contents($this->stdOut);
76-
77-
rewind($this->stdErr);
78-
$output .= stream_get_contents($this->stdErr);
79-
80-
$onExit($exitCode, $output);
81-
fclose($this->stdOut);
82-
fclose($this->stdErr);
83-
});
84-
}
85-
86-
private function cancelTimer(): void
87-
{
88-
if ($this->timer === null) {
89-
return;
90-
}
91-
92-
$this->loop->cancelTimer($this->timer);
93-
$this->timer = null;
94-
}
26+
public function start(callable $onData, callable $onError, callable $onExit): void;
9527

9628
/**
9729
* @param mixed[] $data
9830
*/
99-
public function request(array $data): void
100-
{
101-
$this->cancelTimer();
102-
if ($this->in === null) {
103-
throw new ShouldNotHappenException();
104-
}
105-
$this->in->write($data);
106-
$this->timer = $this->loop->addTimer($this->timeoutSeconds, function (): void {
107-
$onError = $this->onError;
108-
$onError(new ProcessTimedOutException(sprintf('Child process timed out after %.1f seconds. Try making it longer with parallel.processTimeout setting.', $this->timeoutSeconds)));
109-
});
110-
}
111-
112-
public function quit(): void
113-
{
114-
$this->cancelTimer();
115-
if (!$this->process->isRunning()) {
116-
return;
117-
}
118-
119-
foreach ($this->process->pipes as $pipe) {
120-
$pipe->close();
121-
}
122-
123-
if ($this->in === null) {
124-
return;
125-
}
126-
127-
$this->in->end();
128-
}
31+
public function request(array $data): void;
12932

130-
public function bindConnection(ReadableStreamInterface $out, WritableStreamInterface $in): void
131-
{
132-
$out->on('data', function (array $json): void {
133-
$this->cancelTimer();
134-
if ($json['action'] !== 'result') {
135-
return;
136-
}
33+
public function quit(): void;
13734

138-
$onData = $this->onData;
139-
$onData($json['result']);
140-
});
141-
$this->in = $in;
142-
$out->on('error', function (Throwable $error): void {
143-
$onError = $this->onError;
144-
$onError($error);
145-
});
146-
$in->on('error', function (Throwable $error): void {
147-
$onError = $this->onError;
148-
$onError($error);
149-
});
150-
}
35+
public function bindConnection(ReadableStreamInterface $out, WritableStreamInterface $in): void;
15136

15237
}

‎src/Parallel/ProcessBase.php‎

Lines changed: 107 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,107 @@
1+
<?php declare(strict_types = 1);
2+
3+
namespace PHPStan\Parallel;
4+
5+
use PHPStan\ShouldNotHappenException;
6+
use React\EventLoop\LoopInterface;
7+
use React\EventLoop\TimerInterface;
8+
use React\Stream\ReadableStreamInterface;
9+
use React\Stream\WritableStreamInterface;
10+
use Throwable;
11+
use function sprintf;
12+
13+
/**
14+
* Process-creation-agnostic half of a parallel worker: the TCP/NDJSON
15+
* connection plus the per-request timeout timer. Subclasses only implement
16+
* start() (how the worker process is created) and quit() (how it is torn down).
17+
*/
18+
abstract class ProcessBase implements Process
19+
{
20+
21+
private ?WritableStreamInterface $in = null;
22+
23+
/** @var callable(mixed[] $json) : void */
24+
private $onData;
25+
26+
/** @var callable(Throwable $exception): void */
27+
private $onError;
28+
29+
private ?TimerInterface $timer = null;
30+
31+
public function __construct(
32+
protected LoopInterface $loop,
33+
protected float $timeoutSeconds,
34+
)
35+
{
36+
}
37+
38+
/**
39+
* @param callable(mixed[] $json) : void $onData
40+
* @param callable(Throwable $exception): void $onError
41+
*/
42+
protected function setCallbacks(callable $onData, callable $onError): void
43+
{
44+
$this->onData = $onData;
45+
$this->onError = $onError;
46+
}
47+
48+
protected function cancelTimer(): void
49+
{
50+
if ($this->timer === null) {
51+
return;
52+
}
53+
54+
$this->loop->cancelTimer($this->timer);
55+
$this->timer = null;
56+
}
57+
58+
/** Cancels the timeout timer and ends the writable side of the connection. */
59+
protected function endConnection(): void
60+
{
61+
$this->cancelTimer();
62+
if ($this->in === null) {
63+
return;
64+
}
65+
66+
$this->in->end();
67+
}
68+
69+
/**
70+
* @param mixed[] $data
71+
*/
72+
public function request(array $data): void
73+
{
74+
$this->cancelTimer();
75+
if ($this->in === null) {
76+
throw new ShouldNotHappenException();
77+
}
78+
$this->in->write($data);
79+
$this->timer = $this->loop->addTimer($this->timeoutSeconds, function (): void {
80+
$onError = $this->onError;
81+
$onError(new ProcessTimedOutException(sprintf('Child process timed out after %.1f seconds. Try making it longer with parallel.processTimeout setting.', $this->timeoutSeconds)));
82+
});
83+
}
84+
85+
public function bindConnection(ReadableStreamInterface $out, WritableStreamInterface $in): void
86+
{
87+
$out->on('data', function (array $json): void {
88+
$this->cancelTimer();
89+
if ($json['action'] !== 'result') {
90+
return;
91+
}
92+
93+
$onData = $this->onData;
94+
$onData($json['result']);
95+
});
96+
$this->in = $in;
97+
$out->on('error', function (Throwable $error): void {
98+
$onError = $this->onError;
99+
$onError($error);
100+
});
101+
$in->on('error', function (Throwable $error): void {
102+
$onError = $this->onError;
103+
$onError($error);
104+
});
105+
}
106+
107+
}

0 commit comments

Comments
 (0)