Skip to content

Commit ab2e7b8

Browse files
WyriHaximusWyriHaximus
committed
Cancel fiber
1 parent ff11a7a commit ab2e7b8

File tree

4 files changed

+210
-9
lines changed

4 files changed

+210
-9
lines changed

‎composer.json‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -31,7 +31,8 @@
3131
"react/promise": "^2.8 || ^1.2.1"
3232
},
3333
"require-dev": {
34-
"phpunit/phpunit": "^9.3"
34+
"phpunit/phpunit": "^9.3",
35+
"react/promise-timer": "^1.8"
3536
},
3637
"autoload": {
3738
"psr-4": {

‎src/FiberMap.php‎

Lines changed: 51 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,51 @@
1+
<?php
2+
3+
namespace React\Async;
4+
5+
use Fiber;
6+
use React\Promise\PromiseInterface;
7+
8+
/**
9+
* @internal
10+
*/
11+
final class FiberMap
12+
{
13+
private array $status = [];
14+
private array $map = [];
15+
16+
public function register(Fiber $fiber): void
17+
{
18+
$this->status[spl_object_hash($fiber)] = false;
19+
$this->map[spl_object_hash($fiber)] = [];
20+
}
21+
22+
public function cancel(Fiber $fiber): void
23+
{
24+
$this->status[spl_object_hash($fiber)] = true;
25+
}
26+
27+
public function isCanceled(Fiber $fiber): bool
28+
{
29+
return $this->status[spl_object_hash($fiber)];
30+
}
31+
32+
public function attachPromise(Fiber $fiber, PromiseInterface $promise): void
33+
{
34+
$this->map[spl_object_hash($fiber)][spl_object_hash($promise)] = $promise;
35+
}
36+
37+
public function has(Fiber $fiber): bool
38+
{
39+
return array_key_exists(spl_object_hash($fiber), $this->map);
40+
}
41+
42+
public function getPromises(Fiber $fiber): array
43+
{
44+
return $this->map[spl_object_hash($fiber)];
45+
}
46+
47+
public function unregister(Fiber $fiber): void
48+
{
49+
unset($this->status[spl_object_hash($fiber)], $this->map[spl_object_hash($fiber)]);
50+
}
51+
}

‎src/functions.php‎

Lines changed: 64 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@
22

33
namespace React\Async;
44

5+
use Fiber;
56
use React\EventLoop\Loop;
67
use React\Promise\CancellablePromiseInterface;
78
use React\Promise\Deferred;
@@ -20,17 +21,47 @@
2021
*/
2122
function async(callable $function): callable
2223
{
23-
return static fn (mixed ...$args): PromiseInterface => new Promise(function (callable $resolve, callable $reject) use ($function, $args): void {
24-
$fiber = new \Fiber(function () use ($resolve, $reject, $function, $args): void {
25-
try {
26-
$resolve($function(...$args));
27-
} catch (\Throwable $exception) {
28-
$reject($exception);
24+
return static function (mixed ...$args) use ($function): PromiseInterface {
25+
$fiber = null;
26+
$promise = new Promise(function (callable $resolve, callable $reject) use ($function, $args, &$fiber): void {
27+
$fiber = new \Fiber(function () use ($resolve, $reject, $function, $args, &$fiber): void {
28+
if (fiberMap()->isCanceled($fiber)) {
29+
fiberMap()->unregister($fiber);
30+
$reject(new \Exception('Fiber has been cancelled before execution started'));
31+
return;
32+
}
33+
34+
try {
35+
$resolve($function(...$args));
36+
} catch (\Throwable $exception) {
37+
$reject($exception);
38+
} finally {
39+
fiberMap()->unregister($fiber);
40+
}
41+
});
42+
43+
fiberMap()->register($fiber);
44+
45+
Loop::futureTick(static fn() => $fiber->start());
46+
}, function () use (&$fiber): void {
47+
if ($fiber instanceof Fiber) {
48+
fiberMap()->cancel($fiber);
49+
foreach (fiberMap()->getPromises($fiber) as $promise) {
50+
if (method_exists($promise, 'cancel')) {
51+
$promise->cancel();
52+
}
53+
}
2954
}
3055
});
3156

32-
Loop::futureTick(static fn() => $fiber->start());
33-
});
57+
58+
$lowLevelFiber = \Fiber::getCurrent();
59+
if ($lowLevelFiber !== null) {
60+
fiberMap()->attachPromise($lowLevelFiber, $promise);
61+
}
62+
63+
return $promise;
64+
};
3465
}
3566

3667

@@ -82,6 +113,13 @@ function await(PromiseInterface $promise): mixed
82113
$rejected = false;
83114
$resolvedValue = null;
84115
$rejectedThrowable = null;
116+
$lowLevelFiber = \Fiber::getCurrent();
117+
118+
if ($lowLevelFiber !== null) {
119+
if (fiberMap()->isCanceled($lowLevelFiber) && $promise instanceof CancellablePromiseInterface) {
120+
$promise->cancel();
121+
}
122+
}
85123

86124
$promise->then(
87125
function (mixed $value) use (&$resolved, &$resolvedValue, &$fiber): void {
@@ -118,6 +156,10 @@ function (mixed $throwable) use (&$rejected, &$rejectedThrowable, &$fiber): void
118156
throw $rejectedThrowable;
119157
}
120158

159+
if ($lowLevelFiber !== null) {
160+
fiberMap()->attachPromise($lowLevelFiber, $promise);
161+
}
162+
121163
$fiber = FiberFactory::create();
122164

123165
return $fiber->suspend();
@@ -433,3 +475,17 @@ function waterfall(array $tasks): PromiseInterface
433475

434476
return $deferred->promise();
435477
}
478+
479+
/**
480+
* @internal
481+
*/
482+
function fiberMap(): FiberMap
483+
{
484+
static $wm = null;
485+
486+
if ($wm === null) {
487+
$wm = new FiberMap();
488+
}
489+
490+
return $wm;
491+
}

‎tests/AsyncTest.php‎

Lines changed: 93 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@
88
use function React\Async\async;
99
use function React\Async\await;
1010
use function React\Promise\all;
11+
use function React\Promise\Timer\sleep;
1112

1213
class AsyncTest extends TestCase
1314
{
@@ -84,4 +85,96 @@ public function testAsyncReturnsPromiseThatFulfillsWithValueWhenCallbackReturnsA
8485
$this->assertGreaterThan(0.1, $time);
8586
$this->assertLessThan(0.12, $time);
8687
}
88+
89+
public function testCancel()
90+
{
91+
self::expectOutputString('a');
92+
$this->expectException(\Exception::class);
93+
$this->expectExceptionMessage('Timer cancelled');
94+
95+
$promise = async(function (): int {
96+
echo 'a';
97+
await(sleep(2));
98+
echo 'b';
99+
100+
return time();
101+
})();
102+
103+
Loop::addTimer(0.001, function () use ($promise) {
104+
$promise->cancel();
105+
});
106+
await($promise);
107+
}
108+
109+
public function testNestedCancel()
110+
{
111+
self::expectOutputString('abc');
112+
$this->expectException(\Exception::class);
113+
$this->expectExceptionMessage('Timer cancelled');
114+
115+
$promise = async(function (): int {
116+
echo 'a';
117+
await(async(function(): void {
118+
echo 'b';
119+
await(async(function(): void {
120+
echo 'c';
121+
await(sleep(2));
122+
echo 'd';
123+
})());
124+
echo 'e';
125+
})());
126+
echo 'f';
127+
128+
return time();
129+
})();
130+
131+
Loop::addTimer(0.001, function () use ($promise) {
132+
$promise->cancel();
133+
});
134+
await($promise);
135+
}
136+
137+
public function testCancelFiberThatCatchesExceptions()
138+
{
139+
self::expectOutputString('ab');
140+
$this->expectException(\Exception::class);
141+
$this->expectExceptionMessage('Timer cancelled');
142+
143+
$promise = async(function (): int {
144+
echo 'a';
145+
try {
146+
await(sleep(2));
147+
} catch (\Throwable) {
148+
// No-Op
149+
}
150+
echo 'b';
151+
await(sleep(0.1));
152+
echo 'c';
153+
154+
return time();
155+
})();
156+
157+
Loop::addTimer(0.001, function () use ($promise) {
158+
$promise->cancel();
159+
});
160+
await($promise);
161+
}
162+
163+
public function testCancelNeverStartedFiber()
164+
{
165+
self::expectOutputString('');
166+
$this->expectException(\Exception::class);
167+
$this->expectExceptionMessage('Fiber has been cancelled before execution started');
168+
169+
$promise = async(function (): int {
170+
echo 'a';
171+
await(sleep(2));
172+
echo 'b';
173+
174+
return time();
175+
})();
176+
177+
$promise->cancel();
178+
await($promise);
179+
}
87180
}

0 commit comments

Comments
 (0)