Skip to content

Commit 6eb3f7c

Browse files
matyhtfmatyhtf
committed
fix(thread): harden pool shutdown, definition checks and runners
1 parent 8e5e00f commit 6eb3f7c

8 files changed

Lines changed: 448 additions & 72 deletions

File tree

‎.gitignore‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -16,7 +16,7 @@
1616
/.phplint-cache
1717
/.phplint.cache/*
1818

19-
# Ignore the script Swoole\Thread\Pool writes next to the class it runs.
19+
# Ignore runner files left by older Swoole\Thread\Pool versions.
2020
thread_runner.php
2121
thread_runner_*.php
2222

‎src/core/Thread/Pool.php‎

Lines changed: 103 additions & 69 deletions
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,7 @@
1212
namespace Swoole\Thread;
1313

1414
use PhpParser\Error;
15+
use PhpParser\Node\Stmt;
1516
use PhpParser\ParserFactory;
1617
use Swoole\Thread;
1718

@@ -95,54 +96,41 @@ public function start(): void
9596
}
9697
}
9798

98-
if ($this->autoloader) {
99-
$proxyDirectory = dirname($this->autoloader);
100-
} else {
101-
$proxyDirectory = dirname($this->classDefinitionFile);
99+
// Thread requests have their own working directory; resolve paths before launching them.
100+
if ($this->autoloader !== '') {
101+
$this->autoloader = realpath($this->autoloader) ?: $this->autoloader;
102102
}
103-
104-
$script = '<?php' . PHP_EOL;
105-
$script .= '$arguments = Swoole\Thread::getArguments();' . PHP_EOL;
106-
$script .= '$autoloader = $arguments[0];' . PHP_EOL;
107-
$script .= '$runnableClass = $arguments[1];' . PHP_EOL;
108-
$script .= '$queue = $arguments[2];' . PHP_EOL;
109-
$script .= '$index = $arguments[5];' . PHP_EOL;
110-
// PHP exit() and fatal errors skip finally, but still run request shutdown functions.
111-
$script .= 'register_shutdown_function(static function () use ($queue, $index): void {' . PHP_EOL;
112-
$script .= ' $queue->push($index, Swoole\Thread\Queue::NOTIFY_ONE);' . PHP_EOL;
113-
$script .= '});' . PHP_EOL;
114-
$script .= '$classDefinitionFile = $arguments[3];' . PHP_EOL;
115-
$script .= '$running = $arguments[4];' . PHP_EOL;
116-
$script .= '$threadArguments = array_slice($arguments, 6);' . PHP_EOL;
117-
$script .= 'if ($autoloader) require_once $autoloader;' . PHP_EOL;
118-
$script .= 'if ($classDefinitionFile) require_once $classDefinitionFile;' . PHP_EOL;
119-
$script .= '$runnable = new $runnableClass($running, $index);' . PHP_EOL;
120-
$script .= '$runnable->run($threadArguments);' . PHP_EOL;
121-
122-
// Existing installations may still have a runner using the old finally-based notification.
123-
$this->proxyFile = $proxyDirectory . '/thread_runner_' . sha1($script) . '.php';
124-
if (!is_file($this->proxyFile)) {
125-
file_put_contents($this->proxyFile, $script);
103+
if ($this->classDefinitionFile !== '') {
104+
$this->classDefinitionFile = realpath($this->classDefinitionFile) ?: $this->classDefinitionFile;
126105
}
127106

128-
$this->queue = new Queue();
129-
$this->running = new Atomic(1);
130-
131-
for ($index = 0; $index < $this->threadNum; $index++) {
132-
$this->createThread($index);
133-
}
107+
$this->queue = new Queue();
108+
$this->running = new Atomic(1);
109+
$this->proxyFile = $this->createRunner();
134110

135-
while ($this->running->get()) {
136-
$index = $this->queue->pop(-1);
137-
$thread = $this->threads[$index];
138-
$thread->join();
139-
unset($this->threads[$index]);
111+
try {
112+
for ($index = 0; $index < $this->threadNum; $index++) {
113+
$this->createThread($index);
114+
}
140115

141-
$this->createThread($index);
142-
}
116+
while ($this->running->get()) {
117+
$index = $this->queue->pop(-1);
118+
$thread = $this->threads[$index];
119+
$thread->join();
120+
unset($this->threads[$index]);
143121

144-
foreach ($this->threads as $thread) {
145-
$thread->join();
122+
$this->createThread($index);
123+
}
124+
} finally {
125+
$this->running->set(0);
126+
try {
127+
foreach ($this->threads as $thread) {
128+
$thread->join();
129+
}
130+
} finally {
131+
$this->threads = [];
132+
unlink($this->proxyFile);
133+
}
146134
}
147135
}
148136

@@ -161,42 +149,22 @@ protected function isValidPhpFile(string $filePath): bool
161149
throw new \Exception('The file of the class to run cannot be checked without the package nikic/php-parser. Install it, or set the file with withClassDefinitionFile().');
162150
}
163151

164-
$allowedNodeTypes = [
165-
\PhpParser\Node\Stmt\Class_::class,
166-
\PhpParser\Node\Stmt\Const_::class,
167-
\PhpParser\Node\Stmt\Use_::class,
168-
\PhpParser\Node\Stmt\Namespace_::class,
169-
\PhpParser\Node\Stmt\Declare_::class,
170-
];
171-
172152
$parser = (new ParserFactory())->createForNewestSupportedVersion();
173153
try {
174-
$code = file_get_contents($filePath);
175-
$stmts = $parser->parse($code);
176-
$skipLine = -1;
177-
foreach ($stmts as $stmt) {
178-
$isAllowed = false;
179-
foreach ($allowedNodeTypes as $allowedNodeType) {
180-
if ($stmt instanceof $allowedNodeType) {
181-
$isAllowed = true;
182-
break;
183-
}
184-
}
185-
if (!$isAllowed) {
186-
if ($stmt->getLine() == $skipLine) {
187-
continue;
188-
}
189-
return false;
190-
}
191-
}
154+
$code = file_get_contents($filePath);
155+
$stmts = $parser->parse($code);
156+
return $this->containsOnlyDeclarations($stmts ?? []);
192157
} catch (Error) {
193158
return false;
194159
}
195-
return true;
196160
}
197161

198162
protected function createThread(int $index): void
199163
{
164+
if (!$this->running->get()) {
165+
return;
166+
}
167+
200168
$this->threads[$index] = new Thread($this->proxyFile,
201169
$this->autoloader,
202170
$this->runnableClass,
@@ -207,4 +175,70 @@ protected function createThread(int $index): void
207175
...$this->arguments
208176
);
209177
}
178+
179+
private function createRunner(): string
180+
{
181+
$script = <<<'PHP'
182+
<?php
183+
$arguments = Swoole\Thread::getArguments();
184+
$autoloader = $arguments[0];
185+
$runnableClass = $arguments[1];
186+
$queue = $arguments[2];
187+
$index = $arguments[5];
188+
// PHP exit() and fatal errors skip finally, but still run request shutdown functions.
189+
register_shutdown_function(static function () use ($queue, $index): void {
190+
$queue->push($index, Swoole\Thread\Queue::NOTIFY_ONE);
191+
});
192+
$classDefinitionFile = $arguments[3];
193+
$running = $arguments[4];
194+
$threadArguments = array_slice($arguments, 6);
195+
// Preserve the working directory used when runners lived next to the autoloader or class file.
196+
chdir(dirname($autoloader ?: $classDefinitionFile));
197+
if ($autoloader) require_once $autoloader;
198+
if ($classDefinitionFile) require_once $classDefinitionFile;
199+
$runnable = new $runnableClass($running, $index);
200+
$runnable->run($threadArguments);
201+
PHP;
202+
203+
$file = @tempnam(sys_get_temp_dir(), 'swoole_thread_runner_');
204+
if ($file === false) {
205+
throw new \RuntimeException('Failed to create thread runner in the temporary directory.');
206+
}
207+
try {
208+
if (@file_put_contents($file, $script) !== strlen($script)) {
209+
throw new \RuntimeException("Failed to write thread runner '{$file}'.");
210+
}
211+
return $file;
212+
} catch (\Throwable $exception) {
213+
unlink($file);
214+
throw $exception;
215+
}
216+
}
217+
218+
/**
219+
* @param Stmt[] $statements
220+
*/
221+
private function containsOnlyDeclarations(array $statements): bool
222+
{
223+
foreach ($statements as $statement) {
224+
if ($statement instanceof Stmt\Namespace_ || $statement instanceof Stmt\Declare_) {
225+
if (!$this->containsOnlyDeclarations($statement->stmts ?? [])) {
226+
return false;
227+
}
228+
continue;
229+
}
230+
231+
// Class and function bodies are deferred; only inspect containers executed when loading the file.
232+
if ($statement instanceof Stmt\ClassLike
233+
|| $statement instanceof Stmt\Function_
234+
|| $statement instanceof Stmt\Const_
235+
|| $statement instanceof Stmt\Use_
236+
|| $statement instanceof Stmt\GroupUse
237+
|| $statement instanceof Stmt\Nop) {
238+
continue;
239+
}
240+
return false;
241+
}
242+
return true;
243+
}
210244
}

‎tests/fixtures/Thread/RunnerPoolWorker.php‎

Lines changed: 27 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,27 @@
1+
<?php
2+
/**
3+
* This file is part of Swoole.
4+
*
5+
* @link https://www.swoole.com
6+
* @contact team@swoole.com
7+
* @license https://github.com/swoole/library/blob/master/LICENSE
8+
*/
9+
10+
declare(strict_types=1);
11+
12+
namespace Swoole\Tests\Fixtures;
13+
14+
use Swoole\Thread\Runnable;
15+
16+
class RunnerPoolWorker extends Runnable
17+
{
18+
public function run(array $args): void
19+
{
20+
[$state, $name] = $args;
21+
$runner = get_included_files()[0];
22+
$state[$name] = $runner;
23+
$state[$name . '_exists'] = is_file($runner);
24+
$state[$name . '_cwd'] = getcwd();
25+
$this->shutdown();
26+
}
27+
}

‎tests/fixtures/Thread/ShutdownPoolWorker.php‎

Lines changed: 40 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,40 @@
1+
<?php
2+
/**
3+
* This file is part of Swoole.
4+
*
5+
* @link https://www.swoole.com
6+
* @contact team@swoole.com
7+
* @license https://github.com/swoole/library/blob/master/LICENSE
8+
*/
9+
10+
declare(strict_types=1);
11+
12+
namespace Swoole\Tests\Fixtures;
13+
14+
use Swoole\Thread;
15+
use Swoole\Thread\Atomic;
16+
use Swoole\Thread\Runnable;
17+
18+
class ShutdownPoolWorker extends Runnable
19+
{
20+
public function __construct(Atomic $running, int $index)
21+
{
22+
parent::__construct($running, $index);
23+
$arguments = Thread::getArguments();
24+
$arguments[6]->incr('constructed');
25+
if ($arguments[7] === 'shutdown_during_startup') {
26+
$this->shutdown();
27+
$arguments[6]['stopped'] = true;
28+
}
29+
}
30+
31+
public function run(array $args): void
32+
{
33+
$args[0]->incr('runs');
34+
if ($args[1] === 'shutdown') {
35+
// Let the manager enter its completion wait before requesting shutdown.
36+
usleep(50000);
37+
$this->shutdown();
38+
}
39+
}
40+
}

‎tests/fixtures/Thread/worker_runner.php‎

Lines changed: 100 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,100 @@
1+
<?php
2+
/**
3+
* This file is part of Swoole.
4+
*
5+
* @link https://www.swoole.com
6+
* @contact team@swoole.com
7+
* @license https://github.com/swoole/library/blob/master/LICENSE
8+
*/
9+
10+
declare(strict_types=1);
11+
12+
use Swoole\Tests\Fixtures\RunnerPoolWorker;
13+
use Swoole\Thread\Map;
14+
use Swoole\Thread\Pool;
15+
16+
$mode = $argv[1];
17+
$directory = $argv[2];
18+
$autoload = var_export(dirname(__DIR__, 3) . '/vendor/autoload.php', true);
19+
file_put_contents($directory . '/autoload.php', "<?php\nif (!defined('SWOOLE_LIBRARY')) { require {$autoload}; }\n");
20+
copy(__DIR__ . '/RunnerPoolWorker.php', $directory . '/RunnerPoolWorker.php');
21+
require $directory . '/autoload.php';
22+
23+
$state = new Map();
24+
$pool = new class(RunnerPoolWorker::class, 1, $mode, $state, $directory) extends Pool {
25+
public string $runner;
26+
27+
public function __construct(string $runnableClass, int $threadNum, private string $mode, private Map $state, private string $directory)
28+
{
29+
parent::__construct($runnableClass, $threadNum);
30+
}
31+
32+
protected function createThread(int $index): void
33+
{
34+
$this->runner = (new ReflectionProperty(Pool::class, 'proxyFile'))->getValue($this);
35+
if (!is_file($this->runner)) {
36+
throw new RuntimeException('Runner was not published before starting the worker.');
37+
}
38+
if ($this->mode === 'runner_overlapping_pools') {
39+
(new Pool(RunnerPoolWorker::class, 1))
40+
->withAutoloader($this->directory . '/autoload.php')
41+
->withClassDefinitionFile($this->directory . '/RunnerPoolWorker.php')
42+
->withArguments($this->state, 'inner')
43+
->start()
44+
;
45+
if ($this->state['inner'] === $this->runner || !is_file($this->runner)) {
46+
throw new RuntimeException('Overlapping pools shared or removed each other\'s runner.');
47+
}
48+
if (is_file($this->state['inner'])) {
49+
throw new RuntimeException('The inner pool did not clean up its runner.');
50+
}
51+
}
52+
parent::createThread($index);
53+
if ($this->mode === 'runner_startup_failure') {
54+
throw new RuntimeException('Simulated startup failure after creating a worker.');
55+
}
56+
}
57+
};
58+
59+
chdir($directory);
60+
$relative = $mode === 'runner_relative_paths';
61+
$pool->withAutoloader($relative ? './autoload.php' : $directory . '/autoload.php')
62+
->withClassDefinitionFile($relative ? './RunnerPoolWorker.php' : $directory . '/RunnerPoolWorker.php')
63+
->withArguments($state, 'outer')
64+
;
65+
66+
try {
67+
if ($mode === 'runner_read_only') {
68+
chmod($directory, 0555);
69+
}
70+
try {
71+
$pool->start();
72+
if ($mode === 'runner_startup_failure' || $mode === 'runner_creation_failure') {
73+
throw new LogicException('Expected startup to fail.');
74+
}
75+
} catch (RuntimeException $exception) {
76+
if ($mode === 'runner_startup_failure') {
77+
if ($exception->getMessage() !== 'Simulated startup failure after creating a worker.') {
78+
throw $exception;
79+
}
80+
} elseif ($mode === 'runner_creation_failure') {
81+
if (!str_contains($exception->getMessage(), 'Failed to create thread runner')) {
82+
throw $exception;
83+
}
84+
echo "RECOVERED\n";
85+
exit;
86+
} else {
87+
throw $exception;
88+
}
89+
}
90+
clearstatcache();
91+
if (is_file($pool->runner) || !$state['outer_exists'] || $state['outer'] !== $pool->runner) {
92+
throw new RuntimeException('Runner must exist until the worker exits, then be removed.');
93+
}
94+
if (realpath($state['outer_cwd']) !== realpath($directory)) {
95+
throw new RuntimeException('Runner changed the worker working directory.');
96+
}
97+
echo "RECOVERED\n";
98+
} finally {
99+
chmod($directory, 0700);
100+
}

0 commit comments

Comments
 (0)