#154: ThreadPool/ThreadChannel — $this/method-as-closure coverage + fix worker crash on exit()/die()

CHANGELOG.md

@@ -14,6 +14,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - **`Scope::allowZombies(): Scope`** — opt back into safe disposal on a scope (sets `DISPOSE_SAFELY`); returns `$this` for chaining. Inverse of the existing `asNotSafely()`. Use after `new Scope()` when the scope is expected to outlive coroutines parked in `delay()`/`recv()`/etc., turning them into zombies instead of cancelling them at dispose time.
 
 ### Fixed
+- **#154 `ThreadPool` worker crash on `exit()`/`die()` in a task or bootloader** — a graceful `exit()`/`die()` inside a submitted task or the bootloader threw an unwind-exit token that the worker either passed to `reject()` or re-raised via `zend_bailout()` — both crash the worker fiber with `"Error transfer requires a throwable value"` (assert + ASAN stack-use-after-return at `zend_fibers.c:491`). Fixed in `thread_pool.c`: the sync task call and the bootloader call now run under a `zend_try`, and the worker checks `zend_is_unwind_exit()`/`zend_is_graceful_exit()` on `EG(exception)`. Behaviour: `exit()`/`die()` in a **task** is graceful "this task is done" — the worker's request survives it, so the task's future resolves to **`null`** and the worker keeps serving subsequent tasks (verified mixing `exit()` with throwing and normal tasks on a single worker). A real fatal error (e.g. OOM `zend_bailout`) or `exit()`/`die()` in the **bootloader** instead delivers an `Async\ThreadTransferException` to pending awaiters and tears the pool down — the worker can't safely continue. Regression tests `tests/thread_pool/061-bootloader_exit.phpt`, `062-task_exit_sync.phpt`, `064-task_exit_worker_survives.phpt`.
+- **#154 `ThreadPool` swallowed the real error when a `$this`-bound bootloader could not load on the worker** — a bootloader bound to `$this` of a class not defined on the worker (or whose body threw) had its exception converted to an `E_WARNING` and discarded, while pending tasks were rejected with a generic `"task was cancelled before execution"` cancellation — hiding the cause. `thread_pool_drain_tasks` gained a `reject_with` parameter so the worker now propagates the actual error (e.g. `Cannot load transferred object: class "C" not found`, or the thrown exception) to every awaiter. Regression tests `tests/thread_pool/060-bootloader_this_transfer_error.phpt`, `063-bootloader_exception.phpt`.
 - **#146 Thread-pool task freed under a still-running worker (cross-thread UAF)** — `libuv_queue_task` took no reference on the `zend_async_task_t` it handed to `uv_queue_work`; the work wrapper held only a raw pointer. A coroutine awaiting the task that was **cancelled while the worker thread was still inside the task's `run()`** (e.g. blocked in a contended `flock()`) released its refs and freed the task — and its inline-tail data — out from under the detached worker. When the worker's syscall returned it wrote into freed memory (`heap-use-after-free in php_stdiop_flock_task_run`, `main/streams/plain_wrapper.c:1208`). Only reproducible under real multi-core scheduling (CI Linux x64 ASAN), not single-host WSL2. Fixed in `ext/async/libuv_reactor.c`: the in-flight work now owns a reference — `ZEND_ASYNC_EVENT_ADD_REF` after `uv_queue_work`, matched by `ZEND_ASYNC_EVENT_RELEASE` in `libuv_task_after_work_cb` (which libuv runs only after the worker returns). The task now outlives its worker regardless of coroutine cancellation — a general fix for every thread-pool task, not just `flock()`. This completes the earlier #146 work (inline-tail task data + pin-across-SUSPEND in php-src `plain_wrapper.c`), which was necessary but did not cover the cancel-vs-blocked-worker race. Regression backstop: the cancel-mid-flock scenarios in `fuzzy-tests/io/flock_chaos.feature`.
 - **#139 Late await() on coroutine that finished with an exception no longer double-throws** — when a coroutine body threw without an awaiter subscribed at that moment, `coroutine.c` rethrew the exception immediately into the parent frame *and* stored it on the handle. A subsequent `await($coro)` then delivered a second copy through future-replay, the catch ran, but the first copy surfaced as `Uncaught Fatal` once the catch unwound. Reworked to Future-style semantics: the exception is held on the coroutine and only surfaced (a) immediately when nobody can ever observe it — refcount ≤ 1 or `{main}` — or (b) at handle destruction via fire-and-forget safety net. Sticky observation tracked via the existing `EXC_CAUGHT` event flag, lifted from per-NOTIFY `EXCEPTION_HANDLED` after `CALLBACKS_NOTIFY` and set by `zend_async_resume_when` on every await subscription. Edge case verified: when a scope dies before its coroutine throws, child coroutines receive `AsyncCancellation` (handled by spec), so the refcount heuristic cannot misfire. Regression test `tests/coroutine/039-await_after_finished_with_exception.phpt`.
 - **#143 Async\iterate: heap-use-after-free with refcounted values/keys (e.g. generator yielding fresh strings)** — `iterate()`'s userland-callback path had two refcount bugs colliding in `ext/async/iterator.c`. (1) After `zval_ptr_dtor(&fci.params[0])` the slot was not reset to `IS_UNDEF`, so the post-loop cleanup at line 487 dtor'd the same slot a second time → UAF on the just-freed string. (2) `ZVAL_COPY_VALUE(&fci.params[1], &key)` aliased the key into params[1] without bumping refcount, but both `&key` (line 450) and `&fci.params[1]` (line 472) were dtor'd → double-free on string keys. Both bugs were silent for int values / interned-string keys (the existing 009-iterate_generator test used `'a'/'b'/'c' => 100/200/300`), so neither was caught until the chaos suite started yielding `FAST_CONCAT` strings. Surfaced as `heap-use-after-free in zend_gc_delref` from `zend_generator_free_storage` under ASAN. Fix: `ZVAL_UNDEF(&fci.params[0])` after dtor, and `ZVAL_COPY` (not `_VALUE`) for params[1] — symmetric with params[0]. Regression test `tests/iterate/015-iterate_generator_refcounted_values.phpt`.

tests/thread_channel/038-this_bound_closure.phpt

@@ -0,0 +1,44 @@
+--TEST--
+ThreadChannel: $this-bound closure transferred to a worker thread
+--SKIPIF--
+<?php
+if (!PHP_ZTS) die('skip ZTS required');
+if (!function_exists('Async\spawn_thread')) die('skip spawn_thread not available');
+?>
+--FILE--
+<?php
+
+use Async\ThreadChannel;
+use function Async\spawn;
+use function Async\spawn_thread;
+use function Async\await;
+
+class C {
+    public int $n = 21;
+
+    public function makeClosure(): Closure {
+        return function() {
+            return $this->n + 100;
+        };
+    }
+}
+
+$boot = function() {
+    eval('class C { public int $n = 21; }');
+};
+
+spawn(function() use ($boot) {
+    $ch = new ThreadChannel(1);
+
+    $t = spawn_thread(function() use ($ch) {
+        $cl = $ch->recv();
+        echo "worker got: ", $cl(), "\n";
+    }, bootloader: $boot);
+
+    $c = new C();
+    $ch->send($c->makeClosure());
+    await($t);
+});
+?>
+--EXPECT--
+worker got: 121

tests/thread_channel/039-method_as_closure.phpt

@@ -0,0 +1,42 @@
+--TEST--
+ThreadChannel: first-class callable method ($obj->method(...)) transferred to a worker thread
+--SKIPIF--
+<?php
+if (!PHP_ZTS) die('skip ZTS required');
+if (!function_exists('Async\spawn_thread')) die('skip spawn_thread not available');
+?>
+--FILE--
+<?php
+
+use Async\ThreadChannel;
+use function Async\spawn;
+use function Async\spawn_thread;
+use function Async\await;
+
+class C {
+    public int $n = 30;
+
+    public function work(): int {
+        return $this->n * 2;
+    }
+}
+
+$boot = function() {
+    eval('class C { public int $n = 30; function work(): int { return $this->n * 2; } }');
+};
+
+spawn(function() use ($boot) {
+    $ch = new ThreadChannel(1);
+
+    $t = spawn_thread(function() use ($ch) {
+        $cl = $ch->recv();
+        echo "worker got: ", $cl(), "\n";
+    }, bootloader: $boot);
+
+    $c = new C();
+    $ch->send($c->work(...));
+    await($t);
+});
+?>
+--EXPECT--
+worker got: 60

tests/thread_pool/042-submit_this_basic.phpt

@@ -0,0 +1,46 @@
+--TEST--
+ThreadPool: submit() $this-bound closure transfers $this as a deep copy
+--SKIPIF--
+<?php
+if (!PHP_ZTS) die('skip ZTS required');
+if (!class_exists('Async\ThreadPool')) die('skip ThreadPool not available');
+?>
+--FILE--
+<?php
+
+use Async\ThreadPool;
+use function Async\spawn;
+use function Async\await;
+
+class Runner {
+    public int $n = 0;
+    public string $tag = "hi";
+
+    public function go(ThreadPool $pool): string {
+        // Return worker state instead of echoing from the worker thread:
+        // cross-thread stdout ordering relative to the main coroutine is not deterministic.
+        $f = $pool->submit(function() {
+            $before = "n={$this->n} tag={$this->tag} class=" . get_class($this);
+            $this->n = 999;
+            return $before . " -> n={$this->n}";
+        });
+        return await($f);
+    }
+}
+
+$boot = function() {
+    eval('class Runner { public int $n = 0; public string $tag = "hi"; }');
+};
+
+spawn(function() use ($boot) {
+    $pool = new ThreadPool(2, 0, $boot);
+    $r = new Runner();
+    $r->n = 42;
+    echo $r->go($pool), "\n";
+    echo "parent n={$r->n}\n";
+    $pool->close();
+});
+?>
+--EXPECT--
+n=42 tag=hi class=Runner -> n=999
+parent n=42

tests/thread_pool/043-submit_this_private_protected.phpt

@@ -0,0 +1,38 @@
+--TEST--
+ThreadPool: submit() $this-bound closure can access protected/private properties
+--SKIPIF--
+<?php
+if (!PHP_ZTS) die('skip ZTS required');
+if (!class_exists('Async\ThreadPool')) die('skip ThreadPool not available');
+?>
+--FILE--
+<?php
+
+use Async\ThreadPool;
+use function Async\spawn;
+use function Async\await;
+
+class C {
+    private int $secret = 7;
+    protected string $p = "prot";
+
+    public function go(ThreadPool $pool): string {
+        $f = $pool->submit(function() {
+            return "secret={$this->secret} p={$this->p} => " . ($this->secret * 2);
+        });
+        return await($f);
+    }
+}
+
+$boot = function() {
+    eval('class C { private int $secret = 7; protected string $p = "prot"; }');
+};
+
+spawn(function() use ($boot) {
+    $pool = new ThreadPool(2, 0, $boot);
+    echo (new C)->go($pool), "\n";
+    $pool->close();
+});
+?>
+--EXPECT--
+secret=7 p=prot => 14

tests/thread_pool/044-submit_this_self_cycle.phpt

@@ -0,0 +1,39 @@
+--TEST--
+ThreadPool: submit() $this with a self-cycle ($this->self === $this)
+--SKIPIF--
+<?php
+if (!PHP_ZTS) die('skip ZTS required');
+if (!class_exists('Async\ThreadPool')) die('skip ThreadPool not available');
+?>
+--FILE--
+<?php
+
+use Async\ThreadPool;
+use function Async\spawn;
+use function Async\await;
+
+class C {
+    public $self;
+    public int $n = 5;
+
+    public function go(ThreadPool $pool): string {
+        $this->self = $this;
+        $f = $pool->submit(function() {
+            return var_export($this->self === $this, true) . ":" . $this->n;
+        });
+        return await($f);
+    }
+}
+
+$boot = function() {
+    eval('class C { public $self; public int $n = 5; }');
+};
+
+spawn(function() use ($boot) {
+    $pool = new ThreadPool(2, 0, $boot);
+    echo (new C)->go($pool), "\n";
+    $pool->close();
+});
+?>
+--EXPECT--
+true:5

tests/thread_pool/045-submit_this_readonly.phpt

@@ -0,0 +1,37 @@
+--TEST--
+ThreadPool: submit() $this with a readonly property
+--SKIPIF--
+<?php
+if (!PHP_ZTS) die('skip ZTS required');
+if (!class_exists('Async\ThreadPool')) die('skip ThreadPool not available');
+?>
+--FILE--
+<?php
+
+use Async\ThreadPool;
+use function Async\spawn;
+use function Async\await;
+
+class C {
+    public function __construct(public readonly int $n = 11) {}
+
+    public function go(ThreadPool $pool): int {
+        $f = $pool->submit(function() {
+            return $this->n;
+        });
+        return await($f);
+    }
+}
+
+$boot = function() {
+    eval('class C { public function __construct(public readonly int $n = 11) {} }');
+};
+
+spawn(function() use ($boot) {
+    $pool = new ThreadPool(2, 0, $boot);
+    echo "result=", (new C(99))->go($pool), "\n";
+    $pool->close();
+});
+?>
+--EXPECT--
+result=99

tests/thread_pool/046-submit_this_enum_instance.phpt

@@ -0,0 +1,38 @@
+--TEST--
+ThreadPool: submit() $this is a BackedEnum case; access through $this works
+--SKIPIF--
+<?php
+if (!PHP_ZTS) die('skip ZTS required');
+if (!class_exists('Async\ThreadPool')) die('skip ThreadPool not available');
+?>
+--FILE--
+<?php
+
+use Async\ThreadPool;
+use function Async\spawn;
+use function Async\await;
+
+enum Suit: string {
+    case H = 'h';
+    case S = 's';
+
+    public function go(ThreadPool $pool): string {
+        $f = $pool->submit(function() {
+            return $this->value . ":" . $this->name;
+        });
+        return await($f);
+    }
+}
+
+$boot = function() {
+    eval('enum Suit: string { case H = "h"; case S = "s"; }');
+};
+
+spawn(function() use ($boot) {
+    $pool = new ThreadPool(2, 0, $boot);
+    echo "result=", Suit::H->go($pool), "\n";
+    $pool->close();
+});
+?>
+--EXPECT--
+result=h:H

tests/thread_pool/047-submit_this_enum_property.phpt

@@ -0,0 +1,44 @@
+--TEST--
+ThreadPool: submit() $this has an enum-typed property; value preserved across transfer
+--SKIPIF--
+<?php
+if (!PHP_ZTS) die('skip ZTS required');
+if (!class_exists('Async\ThreadPool')) die('skip ThreadPool not available');
+?>
+--FILE--
+<?php
+
+use Async\ThreadPool;
+use function Async\spawn;
+use function Async\await;
+
+enum Color: string {
+    case R = 'r';
+    case G = 'g';
+}
+
+class C {
+    public Color $c = Color::R;
+
+    public function go(ThreadPool $pool): string {
+        $f = $pool->submit(function() {
+            return $this->c->value;
+        });
+        return await($f);
+    }
+}
+
+$boot = function() {
+    eval('enum Color: string { case R = "r"; case G = "g"; } class C { public Color $c = Color::R; }');
+};
+
+spawn(function() use ($boot) {
+    $pool = new ThreadPool(2, 0, $boot);
+    $c = new C();
+    $c->c = Color::G;
+    echo "result=", $c->go($pool), "\n";
+    $pool->close();
+});
+?>
+--EXPECT--
+result=g

tests/thread_pool/048-submit_this_weakref.phpt

@@ -0,0 +1,47 @@
+--TEST--
+ThreadPool: submit() $this with a WeakReference property; identity preserved when target reachable
+--SKIPIF--
+<?php
+if (!PHP_ZTS) die('skip ZTS required');
+if (!class_exists('Async\ThreadPool')) die('skip ThreadPool not available');
+?>
+--FILE--
+<?php
+
+use Async\ThreadPool;
+use function Async\spawn;
+use function Async\await;
+
+class T {
+    public int $v = 99;
+}
+
+class C {
+    public \WeakReference $w;
+    public T $strong;
+
+    public function go(ThreadPool $pool) {
+        $f = $pool->submit(function() {
+            $t = $this->w->get();
+            return $t === null ? "dead" : $t->v;
+        });
+        return await($f);
+    }
+}
+
+$boot = function() {
+    eval('class T { public int $v = 99; } class C { public \WeakReference $w; public T $strong; }');
+};
+
+spawn(function() use ($boot) {
+    $pool = new ThreadPool(2, 0, $boot);
+    $c = new C();
+    $t = new T();
+    $c->strong = $t;
+    $c->w = \WeakReference::create($t);
+    echo "result=", $c->go($pool), "\n";
+    $pool->close();
+});
+?>
+--EXPECT--
+result=99