#254: stop the worker from pinning every task's nested closure

CHANGELOG.md

@@ -10,6 +10,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Fixed
 
 - **Every closure a worker materialized was freed by a rule that holds only on 64-bit** (#255). `op_array_to_emalloc` laid the literals inside the opcode block, which is the layout `pass_two()` leaves and `destroy_op_array` expects there. Under `ZEND_USE_ABS_CONST_ADDR` — 32-bit — the two are separate allocations and `destroy_op_array` frees `op_array->literals` on its own, so the allocator was handed a pointer into the middle of a block it never issued, and the block was then freed a second time. The copy now follows the same rule as the compiler on both platforms.
+- **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.

tests/thread_pool/102-worker_releases_nested_closures.phpt

@@ -0,0 +1,75 @@
+--TEST--
+ThreadPool: a task declaring a nested closure leaves nothing behind in the worker
+--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;
+
+// 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:
+// 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
+// it is the price of the fix, not a detail.
+spawn(function() {
+    $pool = new ThreadPool(workers: 1, coroutine: true);
+
+    $task = static function() { $f = static function () { return 1; }; return $f(); };
+    $probe = static function() { return memory_get_usage(); };
+
+    $run = static function(int $n) use ($pool, $task) {
+        for ($i = 0; $i < $n; $i++) {
+            await($pool->submit($task));
+        }
+    };
+
+    $run(200);
+    $first = await($pool->submit($probe));
+
+    $run(1000);
+    $second = await($pool->submit($probe));
+
+    $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);
+    }
+
+    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

thread.c

@@ -2534,6 +2534,23 @@ static void op_array_to_emalloc(zend_op_array *op_array)
 
 		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;
+		}
+
 		if (op_array->last_literal) {
 			for (uint32_t i = 0; i < op_array->last_literal; i++) {
 				/* Deep-copy refcounted literals into the worker's heap.