Worker threads: Zphp\Pool, Future, and Task

.github/workflows/ci.yml

@@ -169,6 +169,8 @@ jobs:
           sudo apt-get install -y libpcre2-dev libsqlite3-dev zlib1g-dev libmysqlclient-dev libpq-dev libssl-dev libnghttp2-dev libcurl4-openssl-dev libxml2-dev libicu-dev libgmp-dev libgd-dev libsodium-dev libldap2-dev
       - run: zig build -Doptimize=ReleaseFast
       - run: python3 ./tests/memory_soak
+      - name: worker pool retention
+        run: ./zig-out/bin/zphp run tests/workers/memory.php
 
   php-compat:
     needs: changes
@@ -425,6 +427,8 @@ jobs:
         run: ZPHP=./zig-out/bin/zphp.exe ./tests/run
       - name: serve
         run: unset OPENSSL_CONF; ZPHP=./zig-out/bin/zphp.exe ./tests/serve_test
+      - name: workers
+        run: ZPHP=./zig-out/bin/zphp.exe ./tests/workers/run
 
   extensions:
     needs: changes
@@ -449,6 +453,8 @@ jobs:
         run: STATIC=1 ./tests/extensions/run
       - name: Run ./tests/ini_test
         run: ./tests/ini_test
+      - name: Run ./tests/workers/run
+        run: ./tests/workers/run
 
   compile:
     needs: changes

Makefile

@@ -36,7 +36,10 @@ bench: ## Run runtime benchmarks (ReleaseFast)
 bench-compare: ## Time this tree against its merge base with main on this machine (ReleaseFast, interleaved); pass BASE=<ref> to pick the base
 	python3 ./benchmarks/compare $(if $(BASE),--base $(BASE),)
 
-.PHONY: fuzz
+.PHONY: workers fuzz
+workers: build ## Run the worker pool suite (tests/workers/run)
+	./tests/workers/run
+
 fuzz: build ## Mutation-fuzz the pipeline and decoders on the Debug build (FUZZ_SECONDS per fuzzer, default 120)
 	python3 ./tests/fuzz/run all $(or $(FUZZ_SECONDS),120)
 

README.md

@@ -96,6 +96,29 @@ zig build -Doptimize=ReleaseFast -Dextension=hello.c   # static, compiled into z
 
 `ZPHP_EXTENSION_DIR` names a directory whose libraries load automatically. Extensions get module, worker, and request lifecycle hooks, a request-local and a worker-local data slot, and a destructor per resource type that runs when the PHP value is unset, goes out of scope, unwinds through an exception, or the request ends. Values cross the boundary as opaque handles, so the runtime's internals can change without breaking compiled extensions; an ABI version in the descriptor rejects mismatches at load time. The static musl release binaries cannot load shared libraries and take static extensions only. `tests/extensions/demo.c` exercises the whole API.
 
+## Worker threads
+
+`Zphp\Pool` runs PHP functions on a fixed set of OS threads, each with its own isolated interpreter. A pool takes a bootstrap script that every worker runs once, so functions, classes, and worker-local state are ready before the first task arrives.
+
+```php
+$pool = new Zphp\Pool(workers: 4, bootstrap: __DIR__ . '/worker.php');
+
+$futures = [];
+foreach ($pages as $page) {
+    $futures[] = $pool->submit('render', [$page]);
+}
+foreach ($futures as $future) {
+    echo $future->await();
+}
+$pool->shutdown();
+```
+
+A task is a named callable: a function name, `'Class::method'`, or `[$class, $method]`, defined by the bootstrap or built in. Arguments and results are copied between interpreters, so they must be null, bool, int, float, string, arrays of those, or objects of classes both sides define. Closures, generators, and handle-backed objects such as PDO connections are refused when submitted, with the offending path in the message.
+
+`await()` returns the result, or rethrows the task's exception as the same class when the caller has it. A queued task can be cancelled; a running one sees `Zphp\Task::cancelled()` and stops when it chooses, since nothing is ever killed. The queue is bounded: `submit()` blocks when it is full and `trySubmit()` returns null instead. `collect()` hands back completed futures in completion order, and `readiness()` is a stream that becomes readable when one is waiting, for use with `stream_select()`. `shutdown()` stops accepting work, cancels what is queued, and waits for running tasks; the pool's destructor does the same.
+
+Channels and submitting closures are planned.
+
 ## Related projects
 
 - [zphp-bindings](https://github.com/nexxii04/zphp-bindings): Zig bindings for the extension ABI, so extensions can be written in Zig without C.

src/runtime/value.zig

@@ -801,6 +801,8 @@ pub const NativeHandle = struct {
         websocket,
         xml_reader,
         xml_writer,
+        pool,
+        future,
         _,
     };
 

src/runtime/vm.zig

@@ -1451,6 +1451,7 @@ pub const VM = struct {
         try @import("../stdlib/xml_parser.zig").register(vm, allocator);
         try @import("../stdlib/intl.zig").register(vm, allocator);
         try @import("../stdlib/gmp.zig").register(vm, allocator);
+        try @import("../stdlib/workers.zig").register(vm, allocator);
         try @import("../stdlib/bcmath.zig").register(vm, allocator);
         try @import("../stdlib/gd.zig").register(vm, allocator);
         try @import("../stdlib/soap.zig").register(vm, allocator);
@@ -2430,6 +2431,7 @@ pub const VM = struct {
         @import("../stdlib/xmlwriter.zig").cleanupResources(self.objects);
         @import("../stdlib/intl.zig").cleanupResources(self.objects);
         @import("../stdlib/gmp.zig").cleanupResources(self.objects);
+        @import("../stdlib/workers.zig").cleanupResources(self.objects);
         extension.cleanupResources(self.objects);
         @import("../stdlib/gd.zig").cleanupResources(self.objects);
         @import("../stdlib/ftp.zig").cleanupResources(self.objects);

src/stdlib/native_params.zig

@@ -6,6 +6,12 @@ const std = @import("std");
 
 pub const map = std.StaticStringMap([]const []const u8).initComptime(.{
     .{ "Deprecated::__construct", &.{ "$message", "$since" } },
+    .{ "Zphp\\Pool::__construct", &.{ "$workers", "$bootstrap", "$queue" } },
+    .{ "Zphp\\Pool::submit", &.{ "$callable", "$args" } },
+    .{ "Zphp\\Pool::trySubmit", &.{ "$callable", "$args" } },
+    .{ "Zphp\\Pool::collect", &.{"$timeout"} },
+    .{ "Zphp\\Pool::shutdown", &.{"$timeout"} },
+    .{ "Zphp\\Future::await", &.{"$timeout"} },
     .{ "substr", &.{ "$string", "$offset", "$length" } },
     .{ "str_replace", &.{ "$search", "$replace", "$subject", "$count" } },
     .{ "str_ireplace", &.{ "$search", "$replace", "$subject", "$count" } },

src/stdlib/network.zig

@@ -208,7 +208,7 @@ fn native_stream_socket_pair(ctx: *NativeContext, _: []const Value) RuntimeError
     return NativeResult.borrowed(.{ .array = arr });
 }
 
-fn socketStream(ctx: *NativeContext, sock: std.posix.socket_t) !*PhpObject {
+pub fn socketStream(ctx: *NativeContext, sock: std.posix.socket_t) !*PhpObject {
     const obj = try ctx.allocator.create(PhpObject);
     obj.* = .{ .class_name = "FileHandle" };
     try ctx.vm.objects.append(ctx.allocator, obj);