From 76292b38d3d210bf85942a391b44c7bd3f2671cb Mon Sep 17 00:00:00 2001 From: Edmond <1571649+edmonddantes@users.noreply.github.com> Date: Thu, 20 Aug 2026 11:00:52 +0000 Subject: [PATCH 1/2] #254: stop the worker from pinning every task's nested closure DECLARE_LAMBDA_FUNCTION memoizes the Closure it creates and pins it in EG(lambda_cache), which is drained when the request ends. A worker's request outlives every task it runs, so the pinned object held the task's nested body at refcount 1 and destroy_op_array left the whole task op_array behind. Measured at about 930 bytes per task, linear: 6 MB over 6000 submits, flat after the change. Releasing the pinned closures at task end would not do: the handler reads the cache slot before creating anything, so a closure that outlived its task would hand out a freed object on the next call. The memo is turned off instead, which is what zend_compile_func_decl already does for the top level of a script, for the same reason and with the same trade: the same closure literal evaluated twice yields two objects. --- CHANGELOG.md | 1 + .../102-worker_releases_nested_closures.phpt | 60 +++++++++++++++++++ thread.c | 17 ++++++ 3 files changed, 78 insertions(+) create mode 100644 tests/thread_pool/102-worker_releases_nested_closures.phpt diff --git a/CHANGELOG.md b/CHANGELOG.md index d95defb..1f59c1d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- **A `ThreadPool` worker held on to every task that declared a nested closure** (#254). `ZEND_DECLARE_LAMBDA_FUNCTION` memoizes the `Closure` it creates in a run-time cache slot and pins it in `EG(lambda_cache)`, a stack drained when the request ends. A worker's request outlives every task it runs, so the pinned object kept the task's nested body at refcount 1 and `destroy_op_array` left the whole task op_array behind: about 930 bytes per task, growing without a ceiling — 6 MB over 6000 submits, and a long-lived pool reached `memory_limit` on nothing but a `static function () {}` inside its tasks. The memo is now off in a materialized op_array, which is the decision the compiler already makes for the top level of a script and for the same reason. It buys nothing in a worker in any case: each task gets an op_array and a cache of its own. One visible consequence: inside a task the same closure literal evaluated twice yields two objects, where in the submitting thread it yields one. - **`ThreadPool::getWorkerCount()` reported the number of workers the pool was constructed with, whatever became of the threads afterwards** (#231). It now counts the workers that are running, so a closed pool reports 0 once its threads have drained, and no test could state "no worker died" before, because the value that says so did not exist. `reload()` sizes its cohort from that count instead of the constructed one, so a pool that lost a thread no longer waits for an exit token nobody will post, and it reports a rotation that left the pool with no worker at all. `submit()` and `map()` on an open pool with no live worker throw `Async\ThreadPoolException` instead of accepting a task whose Future never settles. - **A `ThreadPool` task in coroutine mode left its un-awaited children running on the worker** (#245). The per-task scope inherited `DISPOSE_SAFELY`, so a leftover child was zombified rather than cancelled, and nothing reached that scope's disposal anyway while the child was alive: the awaiter had its result and the child went on running, out of the active count and past the worker's drain. The sync path already cancelled at task end; both modes now answer the same. - **`pdo_mysql/009-pdo_cancellation` cancelled on a wall clock and failed whenever the runner was busy** (#247). The test slept a fixed 100 ms before cancelling, so its expected output claimed that connecting to MySQL finishes inside that budget; on CI it does not, and the cancellation landed during connect. The coroutine now says when the query is under way and the caller waits for that. Test only. diff --git a/tests/thread_pool/102-worker_releases_nested_closures.phpt b/tests/thread_pool/102-worker_releases_nested_closures.phpt new file mode 100644 index 0000000..50d51d9 --- /dev/null +++ b/tests/thread_pool/102-worker_releases_nested_closures.phpt @@ -0,0 +1,60 @@ +--TEST-- +ThreadPool: a task declaring a nested closure leaves nothing behind in the worker +--SKIPIF-- + +--FILE-- +submit($task)); + } + + $base = await($pool->submit($probe)); + + for ($i = 0; $i < 2000; $i++) { + await($pool->submit($task)); + } + + $grown = await($pool->submit($probe)) - $base; + + // Before the fix this run grew by about 1.9 MB. + var_dump($grown < 100000); + + var_dump(await($pool->submit(static function() { + $a = []; + for ($i = 0; $i < 2; $i++) { $a[] = static function () { return 1; }; } + return $a[0] === $a[1]; + }))); + + $pool->close(); + echo "Done\n"; +}); + +?> +--EXPECT-- +bool(true) +bool(false) +Done diff --git a/thread.c b/thread.c index 036a2a6..d95bdfa 100644 --- a/thread.c +++ b/thread.c @@ -2513,6 +2513,23 @@ static void op_array_to_emalloc(zend_op_array *op_array) zend_op *new_opcodes = (zend_op *) emalloc(opcodes_size + literals_size); memcpy(new_opcodes, orig_opcodes, sizeof(zend_op) * op_array->last); + for (uint32_t i = 0; i < op_array->last; i++) { + if (new_opcodes[i].opcode != ZEND_DECLARE_LAMBDA_FUNCTION) { + continue; + } + + /* The cache slot of this opcode memoizes the Closure object and + * pins it in EG(lambda_cache), which is drained when the request + * ends. A worker's request outlives every task it runs, so the + * pinned object holds the task's nested body at refcount 1 and + * destroy_op_array leaves it behind. The memo is worthless here + * anyway: a task gets a copy of the op_array with a cache of its + * own, so nothing carries over to the next one. This is the + * decision zend_compile_func_decl already makes for the top level + * of a script, and for the same reason. */ + new_opcodes[i].extended_value = (uint32_t) -1; + } + zval *new_literals = NULL; if (op_array->last_literal) { new_literals = (zval *) ((char *) new_opcodes + opcodes_size); From a0047dc30a4f9ad56bf9facf9f3bc90c19a545be Mon Sep 17 00:00:00 2001 From: Edmond <1571649+edmonddantes@users.noreply.github.com> Date: Thu, 20 Aug 2026 11:45:17 +0000 Subject: [PATCH 2/2] #254: assert that the growth does not scale with the number of tasks The test compared the worker's memory against a reading taken 200 tasks in, so anything a worker allocates once on its way up counted against the bound, and the release build on CI crossed it. Two windows of a thousand tasks are compared instead: a one-off allocation lands in the first, a per-task leak shows in both. The numbers are printed when the bound is crossed, so a failure says how much rather than only that. --- .../102-worker_releases_nested_closures.phpt | 33 ++++++++++++++----- 1 file changed, 24 insertions(+), 9 deletions(-) diff --git a/tests/thread_pool/102-worker_releases_nested_closures.phpt b/tests/thread_pool/102-worker_releases_nested_closures.phpt index 50d51d9..1075afe 100644 --- a/tests/thread_pool/102-worker_releases_nested_closures.phpt +++ b/tests/thread_pool/102-worker_releases_nested_closures.phpt @@ -18,6 +18,11 @@ use function Async\await; // at refcount 1 and destroy_op_array left the whole task op_array behind: // about 930 bytes per task, growing without a ceiling. // +// What is asserted is that the growth does not scale with the number of tasks, +// not that the worker holds some absolute number of bytes. Two windows of a +// thousand tasks each are compared: whatever a worker allocates once on its way +// up lands in the first, and a per-task leak shows in both. +// // The memo is off in a materialized op_array, so the same literal evaluated // twice inside one task yields two objects, as it does at the top level of a // script and as it did before the memo existed. That is asserted here because @@ -28,19 +33,29 @@ spawn(function() { $task = static function() { $f = static function () { return 1; }; return $f(); }; $probe = static function() { return memory_get_usage(); }; - for ($i = 0; $i < 200; $i++) { - await($pool->submit($task)); - } + $run = static function(int $n) use ($pool, $task) { + for ($i = 0; $i < $n; $i++) { + await($pool->submit($task)); + } + }; - $base = await($pool->submit($probe)); + $run(200); + $first = await($pool->submit($probe)); - for ($i = 0; $i < 2000; $i++) { - await($pool->submit($task)); - } + $run(1000); + $second = await($pool->submit($probe)); - $grown = await($pool->submit($probe)) - $base; + $run(1000); + $third = await($pool->submit($probe)); + + // With the leak each window grew by about 930 KB. + $grown = $third - $second; + + if ($grown >= 100000) { + printf("second window grew by %d bytes (first %d, second %d, third %d)\n", + $grown, $first, $second, $third); + } - // Before the fix this run grew by about 1.9 MB. var_dump($grown < 100000); var_dump(await($pool->submit(static function() {