Worker threads: closures as task callables

README.md

@@ -113,7 +113,19 @@ foreach ($futures as $future) {
 $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.
+A task is a closure or 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. Generators and handle-backed objects such as PDO connections are refused when submitted, with the offending path in the message.
+
+A closure travels as its compiled code plus its captures: `use` variables, the variables an arrow function reads from the submitting scope, and `$this` when it has one. Each worker loads the code once and runs every later submit of the same closure against it. Captures follow the transfer rules above, and a closure that captures by reference is refused, since nothing can be shared between threads.
+
+```php
+$scale = 0.5;
+$thumbnails = [];
+foreach ($paths as $path) {
+    $thumbnails[] = $pool->submit(function (string $path) use ($scale) {
+        return resize($path, $scale);
+    }, [$path]);
+}
+```
 
 `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.
 
@@ -154,7 +166,6 @@ function resize_images(Zphp\Channel $jobs, Zphp\Channel $results): void
 
 `send()` blocks while the channel is full and `recv()` blocks while it is empty; both take an optional timeout in seconds and throw `Zphp\TimeoutException` when it passes. `trySend()` returns false instead of waiting. `close()` lets buffered values drain and then ends every `foreach`, while `send()` and `recv()` on a closed channel throw `Zphp\ChannelException`. Values follow the same transfer rules as task arguments, and a channel can carry other channels. A channel stays alive while any thread holds it or a value in flight names it.
 
-Submitting closures is planned.
 
 ## Related projects
 

src/pipeline/compiler.zig

@@ -17,14 +17,25 @@ const compiler_class = @import("compiler_class.zig");
 const Allocator = std.mem.Allocator;
 const Error = Allocator.Error || error{CompileError};
 
-var global_closure_counter: u32 = 0;
+// closure and anonymous class names carry a process-wide id: worker threads
+// compile their own scripts and receive closures compiled on other threads,
+// so two compile units must never mint the same name
+var next_closure_id = std.atomic.Value(u32).init(0);
 
 pub fn closureCounter() u32 {
-    return global_closure_counter;
+    return next_closure_id.load(.monotonic);
 }
 
+// raises the counter past ids baked into loaded bytecode; never lowers it
 pub fn setClosureCounter(value: u32) void {
-    global_closure_counter = value;
+    var current = next_closure_id.load(.monotonic);
+    while (value > current) {
+        current = next_closure_id.cmpxchgWeak(current, value, .monotonic, .monotonic) orelse return;
+    }
+}
+
+pub fn allocClosureId() u32 {
+    return next_closure_id.fetchAdd(1, .monotonic);
 }
 
 pub const TypeHint = struct {
@@ -112,7 +123,10 @@ pub fn compileWithPath(ast: *const Ast, allocator: Allocator, file_path: []const
         .break_jumps = .{},
         .continue_jumps = .{},
         .file_path = file_path,
-        .closure_count = global_closure_counter,
+        // seeds the closures-so-far heuristic behind the locals-only frame
+        // fast path exactly as the old process counter did; names come from
+        // allocClosureId
+        .closure_count = closureCounter(),
     };
     errdefer {
         c.chunk.deinit(allocator);
@@ -166,7 +180,6 @@ pub fn compileWithPath(ast: *const Ast, allocator: Allocator, file_path: []const
     const slot_names = try c.buildSlotNames();
     const local_count = c.next_slot;
     c.local_slots.deinit(allocator);
-    global_closure_counter = c.closure_count;
     const strict = detectStrictTypes(ast.source);
     for (c.functions.items) |*f| f.strict_types = strict;
     return .{ .chunk = c.chunk, .functions = c.functions, .string_allocs = c.string_allocs, .allocator = allocator, .local_count = local_count, .slot_names = slot_names, .type_hints = c.type_hints, .function_attrs = c.function_attrs, .new_defaults = c.new_defaults, .deferred_exprs = c.deferred_exprs, .source = ast.source, .file_path = file_path, .strict_types = strict };

src/pipeline/compiler_class.zig

@@ -1,4 +1,5 @@
 const std = @import("std");
+const compiler = @import("compiler.zig");
 const Compiler = @import("compiler.zig").Compiler;
 const TypeHint = @import("compiler.zig").TypeHint;
 const FunctionAttrEntry = @import("compiler.zig").FunctionAttrEntry;
@@ -891,7 +892,7 @@ pub fn compileFunction(self: *Compiler, node: Ast.Node) Error!void {
 }
 
 pub fn compileClosure(self: *Compiler, node: Ast.Node) Error!void {
-    const id = self.closure_count;
+    const id = compiler.allocClosureId();
     self.closure_count += 1;
 
     var name_buf: [32]u8 = undefined;
@@ -1899,7 +1900,7 @@ pub fn compileClassDecl(self: *Compiler, node: Ast.Node) Error!void {
 }
 
 pub fn compileAnonymousClass(self: *Compiler, node: Ast.Node) Error!void {
-    const anon_name = try std.fmt.allocPrint(self.allocator, "class@anonymous_{d}", .{self.closure_count});
+    const anon_name = try std.fmt.allocPrint(self.allocator, "class@anonymous_{d}", .{compiler.allocClosureId()});
     try self.string_allocs.append(self.allocator, anon_name);
     self.closure_count += 1;
 

src/runtime/vm.zig

@@ -534,6 +534,7 @@ pub const VM = struct {
     fiber_suspend_value: Value = .null,
     captures: std.ArrayListUnmanaged(CaptureEntry) = .{},
     capture_index: std.StringHashMapUnmanaged(CaptureRange) = .{},
+    interned_names: std.StringHashMapUnmanaged(void) = .{},
     cycle_closures: std.AutoArrayHashMapUnmanaged(*Value.String.Owner, i64) = .{},
     collecting_cycles: bool = false,
     closure_instance_count: u32 = 0,
@@ -2637,6 +2638,7 @@ pub const VM = struct {
         self.persistent_strings.deinit(self.allocator);
         self.captures.deinit(self.allocator);
         self.capture_index.deinit(self.allocator);
+        self.interned_names.deinit(self.allocator);
         self.free_closure_names.deinit(self.allocator);
         self.php_constants.deinit(self.allocator);
         self.user_constants.deinit(self.allocator);
@@ -2829,6 +2831,7 @@ pub const VM = struct {
         self.cycle_array_candidates.clearRetainingCapacity();
         self.captures.clearRetainingCapacity();
         self.capture_index.clearRetainingCapacity();
+        self.interned_names.clearRetainingCapacity();
         self.ob_stack.clearRetainingCapacity();
         self.request_vars.clearRetainingCapacity();
         self.response_code = 200;
@@ -13132,6 +13135,47 @@ pub const VM = struct {
         return .{ .string = .{ .ptr = name.ptr, .len = name.len, .owner = owner } };
     }
 
+    // a closure instance for a function this vm knows, with captures given
+    // up front the way closure_use adds them one call at a time; a closure
+    // that arrived from another thread is rebuilt through here
+    pub fn bindClosureInstance(self: *VM, compile_name: []const u8, names: []const []const u8, values: []const Value) !Value {
+        const func = self.functions.get(compile_name) orelse return error.RuntimeError;
+        const instance = try self.newClosureInstance(compile_name, func);
+        const name = instance.string.bytes();
+        for (names, values) |var_name, val| {
+            const kept = try self.internName(var_name);
+            retainValue(val);
+            try self.captures.append(self.allocator, .{ .closure_name = name, .var_name = kept, .value = val });
+            self.capture_index.getPtr(name).?.len += 1;
+        }
+        return instance;
+    }
+
+    // request-lifetime bytes for a name that recurs across calls, kept once
+    pub fn internName(self: *VM, name: []const u8) ![]const u8 {
+        if (self.interned_names.getKey(name)) |kept| return kept;
+        const kept = try self.allocator.dupe(u8, name);
+        errdefer self.allocator.free(kept);
+        try self.strings.append(self.allocator, kept);
+        try self.interned_names.put(self.allocator, kept, {});
+        return kept;
+    }
+
+    // a variable of the frame below the current one: what a native sees as
+    // its caller's scope. reference cells win, then the slot, then the map
+    pub fn callerVar(self: *VM, name: []const u8) ?Value {
+        if (self.frame_count < 2) return null;
+        const frame = &self.frames[self.frame_count - 2];
+        if (frame.ref_slots.get(name)) |cell| return cell.*;
+        const slot_names = if (frame.slot_names.len > 0) frame.slot_names else if (frame.func) |f| f.slot_names else self.global_slot_names;
+        for (slot_names, 0..) |sn, si| {
+            if (!std.mem.eql(u8, sn, name)) continue;
+            if (si < frame.locals.len and frame.locals[si] != .null) return frame.locals[si];
+            break;
+        }
+        return frame.vars.get(name);
+    }
+
     pub fn retainClosureByName(self: *VM, name: []const u8) void {
         const range = self.capture_index.getPtr(name) orelse return;
         if (range.owner) |owner| owner.refcount += 1;