From 3bdbdd4d23ed7d487605b919387084fa73b704c4 Mon Sep 17 00:00:00 2001
From: nvms <jon@pye.rs>
Date: Mon, 14 Sep 2026 10:55:39 -0400
Subject: [PATCH 1/3] worker threads: Zphp\Channel

a bounded queue of transferable values shared between threads by identity:
a process-wide registry maps ids to channels, every vm that holds one
binds its own wrapper, and a wrapper crosses to a worker as its id through
native __serialize/__unserialize. send and recv block with optional
timeouts, trySend refuses instead of waiting, close drains then ends
every foreach, and channels ride inside task arguments, results, and
other channels.

everything that crosses a thread is now a payload: the bytes plus a
retained reference to each channel inside, so a channel created in a
worker survives the worker's teardown until the caller binds it. a result
that fails the transfer check settles the task with the exception. cross
thread allocations go through libc malloc in release: smp_allocator keeps
a freelist per thread and a producer/consumer pair grew rss without bound.

an exception thrown from __unserialize or __wakeup now propagates instead
of turning into the parse warning, and unserialize reuses the class
table's key for the object's class name instead of copying it into the
request arena per object.
---
 README.md                          |  39 ++-
 src/runtime/value.zig              |   1 +
 src/runtime/vm.zig                 |   2 +
 src/stdlib/channel.zig             | 404 +++++++++++++++++++++++++++++
 src/stdlib/native_params.zig       |   4 +
 src/stdlib/serialize.zig           |  29 ++-
 src/stdlib/workers.zig             | 111 +++++---
 tests/unserialize_magic_throws.php |  16 ++
 tests/workers/basic.expected       |   2 +-
 tests/workers/channel.expected     |  48 ++++
 tests/workers/channel.php          |  92 +++++++
 tests/workers/memory.php           |   7 +
 tests/workers/run                  |   2 +-
 tests/workers/worker.php           |  18 ++
 14 files changed, 732 insertions(+), 43 deletions(-)
 create mode 100644 src/stdlib/channel.zig
 create mode 100644 tests/unserialize_magic_throws.php
 create mode 100644 tests/workers/channel.expected
 create mode 100644 tests/workers/channel.php

diff --git a/README.md b/README.md
index 6be2c2cb..60775f2c 100644
--- a/README.md
+++ b/README.md
@@ -117,7 +117,44 @@ A task is a named callable: a function name, `'Class::method'`, or `[$class, $me
 
 `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.
+### Channels
+
+`Zphp\Channel` is a bounded queue that workers and the main thread share. A channel passed to a task binds to the same queue on the other side, so a producer and its consumers can run on different threads without sharing PHP memory.
+
+```php
+$jobs = new Zphp\Channel(capacity: 16);
+$results = new Zphp\Channel(capacity: 256);
+
+$consumers = [];
+for ($i = 0; $i < 4; $i++) {
+    $consumers[] = $pool->submit('resize_images', [$jobs, $results]);
+}
+foreach (glob('uploads/*.jpg') as $path) {
+    $jobs->send($path);
+}
+$jobs->close();
+foreach ($consumers as $future) {
+    $future->await();
+}
+$results->close();
+foreach ($results as $thumbnail) {
+    echo $thumbnail, "\n";
+}
+```
+
+```php
+// worker.php
+function resize_images(Zphp\Channel $jobs, Zphp\Channel $results): void
+{
+    foreach ($jobs as $path) {
+        $results->send(resize($path));
+    }
+}
+```
+
+`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
 
diff --git a/src/runtime/value.zig b/src/runtime/value.zig
index dd4cfd8d..b0e6d3a7 100644
--- a/src/runtime/value.zig
+++ b/src/runtime/value.zig
@@ -803,6 +803,7 @@ pub const NativeHandle = struct {
         xml_writer,
         pool,
         future,
+        channel,
         _,
     };
 
diff --git a/src/runtime/vm.zig b/src/runtime/vm.zig
index d2498333..b85ef2bb 100644
--- a/src/runtime/vm.zig
+++ b/src/runtime/vm.zig
@@ -1452,6 +1452,7 @@ pub const VM = struct {
         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/channel.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);
@@ -2432,6 +2433,7 @@ pub const VM = struct {
         @import("../stdlib/intl.zig").cleanupResources(self.objects);
         @import("../stdlib/gmp.zig").cleanupResources(self.objects);
         @import("../stdlib/workers.zig").cleanupResources(self.objects);
+        @import("../stdlib/channel.zig").cleanupResources(self.objects);
         extension.cleanupResources(self.objects);
         @import("../stdlib/gd.zig").cleanupResources(self.objects);
         @import("../stdlib/ftp.zig").cleanupResources(self.objects);
diff --git a/src/stdlib/channel.zig b/src/stdlib/channel.zig
new file mode 100644
index 00000000..e1ae608a
--- /dev/null
+++ b/src/stdlib/channel.zig
@@ -0,0 +1,404 @@
+// a Zphp\Channel is a bounded queue of serialized values shared between
+// threads by identity: every VM that holds one binds its own wrapper object
+// to the same Channel, and a wrapper crosses to a worker as the channel's id
+const std = @import("std");
+const value_mod = @import("../runtime/value.zig");
+const Value = value_mod.Value;
+const PhpObject = value_mod.PhpObject;
+const vm_mod = @import("../runtime/vm.zig");
+const VM = vm_mod.VM;
+const NativeContext = vm_mod.NativeContext;
+const ClassDef = vm_mod.ClassDef;
+const NativeResult = @import("../runtime/native_result.zig").NativeResult;
+const RuntimeError = error{ RuntimeError, OutOfMemory };
+const workers = @import("workers.zig");
+const platform = @import("../platform.zig");
+
+pub const channel_class = "Zphp\\Channel";
+const channel_exception = "Zphp\\ChannelException";
+const timeout_exception = "Zphp\\TimeoutException";
+
+// ---------------------------------------------------------------------------
+// the shared queue
+
+const Wait = union(enum) { none, forever, until: i128 };
+
+fn waitFor(cond: *std.Thread.Condition, mutex: *std.Thread.Mutex, wait: Wait) bool {
+    switch (wait) {
+        .none => return false,
+        .forever => {
+            cond.wait(mutex);
+            return true;
+        },
+        .until => |deadline| {
+            const now = std.time.nanoTimestamp();
+            if (now >= deadline) return false;
+            cond.timedWait(mutex, @intCast(deadline - now)) catch {};
+            return true;
+        },
+    }
+}
+
+pub const Channel = struct {
+    allocator: std.mem.Allocator,
+    id: u64,
+    items: []workers.Payload,
+    head: usize = 0,
+    len: usize = 0,
+    closed: bool = false,
+    mutex: std.Thread.Mutex = .{},
+    not_empty: std.Thread.Condition = .{},
+    not_full: std.Thread.Condition = .{},
+    // one reference per wrapper object, across every vm in the process
+    refs: std.atomic.Value(u32) = std.atomic.Value(u32).init(1),
+
+    const SendResult = enum { ok, full, closed, timeout };
+    const RecvResult = union(enum) { value: workers.Payload, empty, closed, timeout };
+
+    fn send(ch: *Channel, payload: workers.Payload, wait: Wait) SendResult {
+        ch.mutex.lock();
+        defer ch.mutex.unlock();
+        while (!ch.closed and ch.len == ch.items.len) {
+            if (!waitFor(&ch.not_full, &ch.mutex, wait)) return if (wait == .none) .full else .timeout;
+        }
+        if (ch.closed) return .closed;
+        ch.items[(ch.head + ch.len) % ch.items.len] = payload;
+        ch.len += 1;
+        ch.not_empty.signal();
+        return .ok;
+    }
+
+    fn recv(ch: *Channel, wait: Wait) RecvResult {
+        ch.mutex.lock();
+        defer ch.mutex.unlock();
+        while (ch.len == 0 and !ch.closed) {
+            if (!waitFor(&ch.not_empty, &ch.mutex, wait)) return if (wait == .none) .empty else .timeout;
+        }
+        if (ch.len == 0) return .closed;
+        const payload = ch.items[ch.head];
+        ch.head = (ch.head + 1) % ch.items.len;
+        ch.len -= 1;
+        ch.not_full.signal();
+        return .{ .value = payload };
+    }
+
+    // buffered values stay receivable; senders and empty receivers are released
+    fn close(ch: *Channel) void {
+        ch.mutex.lock();
+        defer ch.mutex.unlock();
+        ch.closed = true;
+        ch.not_empty.broadcast();
+        ch.not_full.broadcast();
+    }
+
+    fn count(ch: *Channel) usize {
+        ch.mutex.lock();
+        defer ch.mutex.unlock();
+        return ch.len;
+    }
+
+    fn isClosed(ch: *Channel) bool {
+        ch.mutex.lock();
+        defer ch.mutex.unlock();
+        return ch.closed;
+    }
+
+    pub fn retain(ch: *Channel) void {
+        _ = ch.refs.fetchAdd(1, .acq_rel);
+    }
+
+    // the last wrapper takes the channel out of the registry under its lock,
+    // so a lookup racing the release either retains a live channel or misses
+    pub fn release(ch: *Channel) void {
+        registry_mutex.lock();
+        if (ch.refs.fetchSub(1, .acq_rel) != 1) {
+            registry_mutex.unlock();
+            return;
+        }
+        _ = registry.remove(ch.id);
+        registry_mutex.unlock();
+        ch.free();
+    }
+
+    fn free(ch: *Channel) void {
+        var i: usize = 0;
+        while (i < ch.len) : (i += 1) ch.items[(ch.head + i) % ch.items.len].free(ch.allocator);
+        ch.allocator.free(ch.items);
+        ch.allocator.destroy(ch);
+    }
+};
+
+// ---------------------------------------------------------------------------
+// the process-wide registry that resolves an id back to its channel
+
+var registry_mutex: std.Thread.Mutex = .{};
+var registry: std.AutoHashMapUnmanaged(u64, *Channel) = .{};
+var next_id: u64 = 1;
+const registry_allocator = std.heap.page_allocator;
+
+fn create(allocator: std.mem.Allocator, capacity: usize) !*Channel {
+    const ch = try allocator.create(Channel);
+    errdefer allocator.destroy(ch);
+    const items = try allocator.alloc(workers.Payload, capacity);
+    errdefer allocator.free(items);
+    registry_mutex.lock();
+    defer registry_mutex.unlock();
+    ch.* = .{ .allocator = allocator, .id = next_id, .items = items };
+    try registry.put(registry_allocator, ch.id, ch);
+    next_id += 1;
+    return ch;
+}
+
+fn lookup(id: u64) ?*Channel {
+    registry_mutex.lock();
+    defer registry_mutex.unlock();
+    const ch = registry.get(id) orelse return null;
+    ch.retain();
+    return ch;
+}
+
+// ---------------------------------------------------------------------------
+// php surface
+
+fn getThis(ctx: *NativeContext) ?*PhpObject {
+    const v = ctx.vm.currentFrame().vars.get("$this") orelse return null;
+    if (v != .object) return null;
+    return v.object;
+}
+
+fn throwNamed(ctx: *NativeContext, class_name: []const u8, comptime fmt: []const u8, args: anytype) RuntimeError {
+    const msg = try std.fmt.allocPrint(ctx.allocator, fmt, args);
+    try ctx.vm.strings.append(ctx.allocator, msg);
+    try ctx.vm.setPendingException(class_name, msg);
+    return error.RuntimeError;
+}
+
+fn channelOf(ctx: *NativeContext, obj: *PhpObject) RuntimeError!*Channel {
+    return obj.native.get(Channel, .channel) orelse throwNamed(ctx, channel_exception, "the channel is not open", .{});
+}
+
+fn bind(obj: *PhpObject, ch: *Channel) void {
+    obj.native = .{ .kind = .channel, .ptr = @intFromPtr(ch) };
+}
+
+fn waitArg(args: []const Value, index: usize) Wait {
+    if (index >= args.len or args[index] == .null) return .forever;
+    const ns = workers.optionalSeconds(args[index]) orelse return .forever;
+    return .{ .until = std.time.nanoTimestamp() + @as(i128, ns) };
+}
+
+fn flushOutput(vm: *VM) void {
+    if (vm.output.items.len == 0) return;
+    platform.writeStdout(vm.output.items);
+    vm.output.clearRetainingCapacity();
+}
+
+fn channelConstruct(ctx: *NativeContext, args: []const Value) RuntimeError!NativeResult {
+    const obj = getThis(ctx) orelse return NativeResult.scalar(.null);
+    var capacity: usize = 1;
+    if (args.len >= 1 and args[0] != .null) {
+        if (args[0] != .int or args[0].int < 1) return throwNamed(ctx, channel_exception, "capacity must be a positive integer", .{});
+        capacity = @intCast(args[0].int);
+    }
+    const ch = create(workers.transferAllocator(ctx.allocator), capacity) catch return throwNamed(ctx, channel_exception, "could not create the channel", .{});
+    bind(obj, ch);
+    return NativeResult.scalar(.null);
+}
+
+fn sendValue(ctx: *NativeContext, args: []const Value, wait: Wait) RuntimeError!Channel.SendResult {
+    const obj = getThis(ctx) orelse return .closed;
+    const ch = try channelOf(ctx, obj);
+    if (args.len < 1) return throwNamed(ctx, channel_exception, "send() needs a value", .{});
+    const payload = try workers.pack(ctx, args[0], "value", ch.allocator);
+    flushOutput(ctx.vm);
+    const result = ch.send(payload, wait);
+    if (result != .ok) payload.free(ch.allocator);
+    return result;
+}
+
+fn channelSend(ctx: *NativeContext, args: []const Value) RuntimeError!NativeResult {
+    switch (try sendValue(ctx, args, waitArg(args, 1))) {
+        .ok, .full => return NativeResult.scalar(.null),
+        .closed => return throwNamed(ctx, channel_exception, "the channel is closed", .{}),
+        .timeout => return throwNamed(ctx, timeout_exception, "the channel did not accept the value in time", .{}),
+    }
+}
+
+fn channelTrySend(ctx: *NativeContext, args: []const Value) RuntimeError!NativeResult {
+    switch (try sendValue(ctx, args, .none)) {
+        .ok => return NativeResult.scalar(.{ .bool = true }),
+        .full, .timeout => return NativeResult.scalar(.{ .bool = false }),
+        .closed => return throwNamed(ctx, channel_exception, "the channel is closed", .{}),
+    }
+}
+
+fn unpackValue(ctx: *NativeContext, ch: *Channel, payload: workers.Payload) RuntimeError!Value {
+    return workers.unpack(ctx, payload, ch.allocator) orelse throwNamed(ctx, channel_exception, "the value did not transfer", .{});
+}
+
+fn receive(ctx: *NativeContext, ch: *Channel, wait: Wait) Channel.RecvResult {
+    flushOutput(ctx.vm);
+    return ch.recv(wait);
+}
+
+fn channelRecv(ctx: *NativeContext, args: []const Value) RuntimeError!NativeResult {
+    const obj = getThis(ctx) orelse return NativeResult.scalar(.null);
+    const ch = try channelOf(ctx, obj);
+    switch (receive(ctx, ch, waitArg(args, 0))) {
+        .value => |payload| return NativeResult.transfer(try unpackValue(ctx, ch, payload)),
+        .closed => return throwNamed(ctx, channel_exception, "the channel is closed", .{}),
+        .empty, .timeout => return throwNamed(ctx, timeout_exception, "no value arrived in time", .{}),
+    }
+}
+
+fn channelClose(ctx: *NativeContext, _: []const Value) RuntimeError!NativeResult {
+    const obj = getThis(ctx) orelse return NativeResult.scalar(.null);
+    const ch = try channelOf(ctx, obj);
+    ch.close();
+    return NativeResult.scalar(.null);
+}
+
+fn channelIsClosed(ctx: *NativeContext, _: []const Value) RuntimeError!NativeResult {
+    const obj = getThis(ctx) orelse return NativeResult.scalar(.null);
+    const ch = try channelOf(ctx, obj);
+    return NativeResult.scalar(.{ .bool = ch.isClosed() });
+}
+
+fn channelCount(ctx: *NativeContext, _: []const Value) RuntimeError!NativeResult {
+    const obj = getThis(ctx) orelse return NativeResult.scalar(.null);
+    const ch = try channelOf(ctx, obj);
+    return NativeResult.scalar(.{ .int = @intCast(ch.count()) });
+}
+
+fn channelCapacity(ctx: *NativeContext, _: []const Value) RuntimeError!NativeResult {
+    const obj = getThis(ctx) orelse return NativeResult.scalar(.null);
+    const ch = try channelOf(ctx, obj);
+    return NativeResult.scalar(.{ .int = @intCast(ch.items.len) });
+}
+
+fn channelId(ctx: *NativeContext, _: []const Value) RuntimeError!NativeResult {
+    const obj = getThis(ctx) orelse return NativeResult.scalar(.null);
+    const ch = try channelOf(ctx, obj);
+    return NativeResult.scalar(.{ .int = @intCast(ch.id) });
+}
+
+// ---------------------------------------------------------------------------
+// iteration: foreach pulls values until the channel is closed and drained.
+// the pending value and its position live on the wrapper, so each consumer
+// iterates independently
+
+fn channelRewind(_: *NativeContext, _: []const Value) RuntimeError!NativeResult {
+    return NativeResult.scalar(.null);
+}
+
+fn channelValid(ctx: *NativeContext, _: []const Value) RuntimeError!NativeResult {
+    const obj = getThis(ctx) orelse return NativeResult.scalar(.{ .bool = false });
+    const ch = try channelOf(ctx, obj);
+    if (obj.get("__ready") == .bool and obj.get("__ready").bool) return NativeResult.scalar(.{ .bool = true });
+    switch (receive(ctx, ch, .forever)) {
+        .value => |payload| {
+            const v = try unpackValue(ctx, ch, payload);
+            try obj.set(ctx.allocator, "__current", v);
+            if (v == .string) v.string.release();
+            try obj.set(ctx.allocator, "__ready", .{ .bool = true });
+            return NativeResult.scalar(.{ .bool = true });
+        },
+        else => return NativeResult.scalar(.{ .bool = false }),
+    }
+}
+
+fn channelCurrent(ctx: *NativeContext, _: []const Value) RuntimeError!NativeResult {
+    const obj = getThis(ctx) orelse return NativeResult.scalar(.null);
+    return NativeResult.share(obj.get("__current"));
+}
+
+fn channelKey(ctx: *NativeContext, _: []const Value) RuntimeError!NativeResult {
+    const obj = getThis(ctx) orelse return NativeResult.scalar(.null);
+    const key = obj.get("__key");
+    return NativeResult.scalar(.{ .int = if (key == .int) key.int else 0 });
+}
+
+fn channelNext(ctx: *NativeContext, _: []const Value) RuntimeError!NativeResult {
+    const obj = getThis(ctx) orelse return NativeResult.scalar(.null);
+    const key = obj.get("__key");
+    try obj.set(ctx.allocator, "__key", .{ .int = if (key == .int) key.int + 1 else 1 });
+    try obj.set(ctx.allocator, "__ready", .{ .bool = false });
+    try obj.set(ctx.allocator, "__current", .null);
+    return NativeResult.scalar(.null);
+}
+
+// ---------------------------------------------------------------------------
+// transfer: a channel serializes as its id and binds to the same channel
+// wherever it is unserialized in this process
+
+fn channelSerialize(ctx: *NativeContext, _: []const Value) RuntimeError!NativeResult {
+    const obj = getThis(ctx) orelse return NativeResult.scalar(.null);
+    const ch = try channelOf(ctx, obj);
+    const arr = try ctx.createArray();
+    try arr.set(ctx.allocator, .{ .string = Value.String.borrowed("id") }, .{ .int = @intCast(ch.id) });
+    return NativeResult.borrowed(.{ .array = arr });
+}
+
+fn channelUnserialize(ctx: *NativeContext, args: []const Value) RuntimeError!NativeResult {
+    const obj = getThis(ctx) orelse return NativeResult.scalar(.null);
+    if (obj.native.kind != .none) return NativeResult.scalar(.null);
+    const id: u64 = blk: {
+        if (args.len < 1 or args[0] != .array) break :blk 0;
+        const v = args[0].array.get(.{ .string = Value.String.borrowed("id") });
+        break :blk if (v == .int and v.int > 0) @intCast(v.int) else 0;
+    };
+    const ch = lookup(id) orelse return throwNamed(ctx, channel_exception, "channel {d} does not exist in this process", .{id});
+    bind(obj, ch);
+    return NativeResult.scalar(.null);
+}
+
+// ---------------------------------------------------------------------------
+// lifetimes
+
+fn cleanupChannel(obj: *PhpObject) bool {
+    const ch = obj.native.get(Channel, .channel) orelse return true;
+    obj.native = .{};
+    ch.release();
+    return true;
+}
+
+pub fn cleanupResources(objects: std.ArrayListUnmanaged(*PhpObject)) void {
+    for (objects.items) |obj| {
+        if (obj.pooled) continue;
+        if (obj.native.kind == .channel) _ = cleanupChannel(obj);
+    }
+}
+
+// ---------------------------------------------------------------------------
+// registration
+
+const methods = [_]struct { name: []const u8, arity: u8, native: vm_mod.NativeFn }{
+    .{ .name = "__construct", .arity = 1, .native = channelConstruct },
+    .{ .name = "send", .arity = 2, .native = channelSend },
+    .{ .name = "trySend", .arity = 1, .native = channelTrySend },
+    .{ .name = "recv", .arity = 1, .native = channelRecv },
+    .{ .name = "close", .arity = 0, .native = channelClose },
+    .{ .name = "isClosed", .arity = 0, .native = channelIsClosed },
+    .{ .name = "count", .arity = 0, .native = channelCount },
+    .{ .name = "capacity", .arity = 0, .native = channelCapacity },
+    .{ .name = "id", .arity = 0, .native = channelId },
+    .{ .name = "rewind", .arity = 0, .native = channelRewind },
+    .{ .name = "valid", .arity = 0, .native = channelValid },
+    .{ .name = "current", .arity = 0, .native = channelCurrent },
+    .{ .name = "key", .arity = 0, .native = channelKey },
+    .{ .name = "next", .arity = 0, .native = channelNext },
+    .{ .name = "__serialize", .arity = 0, .native = channelSerialize },
+    .{ .name = "__unserialize", .arity = 1, .native = channelUnserialize },
+};
+
+pub fn register(vm: *VM, a: std.mem.Allocator) !void {
+    var def = ClassDef{ .name = channel_class, .is_final = true, .native_cleanup = cleanupChannel };
+    try def.interfaces.append(a, "Iterator");
+    try def.interfaces.append(a, "Countable");
+    inline for (methods) |m| {
+        try def.methods.put(a, m.name, .{ .name = m.name, .arity = m.arity });
+        try vm.native_fns.put(a, channel_class ++ "::" ++ m.name, m.native);
+    }
+    try vm.classes.put(a, channel_class, def);
+    try vm.classes.put(a, channel_exception, ClassDef{ .name = channel_exception, .parent = "Exception" });
+}
diff --git a/src/stdlib/native_params.zig b/src/stdlib/native_params.zig
index 38b06f9d..480fd10e 100644
--- a/src/stdlib/native_params.zig
+++ b/src/stdlib/native_params.zig
@@ -12,6 +12,10 @@ pub const map = std.StaticStringMap([]const []const u8).initComptime(.{
     .{ "Zphp\\Pool::collect", &.{"$timeout"} },
     .{ "Zphp\\Pool::shutdown", &.{"$timeout"} },
     .{ "Zphp\\Future::await", &.{"$timeout"} },
+    .{ "Zphp\\Channel::__construct", &.{"$capacity"} },
+    .{ "Zphp\\Channel::send", &.{ "$value", "$timeout" } },
+    .{ "Zphp\\Channel::trySend", &.{"$value"} },
+    .{ "Zphp\\Channel::recv", &.{"$timeout"} },
     .{ "substr", &.{ "$string", "$offset", "$length" } },
     .{ "str_replace", &.{ "$search", "$replace", "$subject", "$count" } },
     .{ "str_ireplace", &.{ "$search", "$replace", "$subject", "$count" } },
diff --git a/src/stdlib/serialize.zig b/src/stdlib/serialize.zig
index a0a87eda..eb408b2d 100644
--- a/src/stdlib/serialize.zig
+++ b/src/stdlib/serialize.zig
@@ -180,6 +180,22 @@ pub fn serializeToString(ctx: *NativeContext, val: Value) RuntimeError!NativeRes
 }
 
 // a string result carries one reference the caller must release or store
+// the object keeps its class name, so it must outlive this call: the class
+// table's own key does, and only an unknown class needs a copy. a script that
+// unserializes objects in a loop would otherwise grow the request arena by a
+// class name per object
+fn keptClassName(ctx: *NativeContext, class_allowed: bool, orig_class: []const u8) ![]const u8 {
+    if (!class_allowed) return "__PHP_Incomplete_Class";
+    return ctx.vm.classes.getKey(orig_class) orelse try ctx.createString(orig_class);
+}
+
+// a declared property is stored by slot and the name is not kept; a dynamic
+// one is stored by name and needs request-lifetime bytes
+fn keptPropertyName(ctx: *NativeContext, obj: *PhpObject, name: []const u8) ![]const u8 {
+    if (obj.getSlotIndex(name) != null) return name;
+    return ctx.createString(name);
+}
+
 pub fn unserializeFromString(ctx: *NativeContext, s: []const u8) ?Value {
     var uctx = UnserCtx{};
     defer uctx.deinit(ctx.allocator);
@@ -540,7 +556,9 @@ fn native_unserialize(ctx: *NativeContext, args: []const Value) RuntimeError!Nat
         if (md == .int and md.int >= 0) uctx.max_depth = @intCast(md.int);
     }
     const result = unserializeValue(ctx, &uctx, s, 0) catch {
-        if (uctx.threw) return error.RuntimeError;
+        // an exception from __unserialize, __wakeup, or Serializable::unserialize
+        // propagates; only a parse failure is a warning
+        if (uctx.threw or ctx.vm.pending_exception != null) return error.RuntimeError;
         // emit PHP's parse-failure warning. depth-exceeded gets its own
         // message that names the option + ini setting; generic parse errors
         // get 'Error at offset N of M bytes'
@@ -711,7 +729,7 @@ fn unserializeValue(ctx: *NativeContext, uctx: *UnserCtx, s: []const u8, pos: us
             if (colon1 + 2 + name_len + 1 >= s.len) return error.RuntimeError;
             const orig_class = s[colon1 + 2 .. colon1 + 2 + name_len];
             const class_allowed = uctx.classAllowed(orig_class) and ctx.vm.classes.contains(orig_class);
-            const class_name = try ctx.createString(if (class_allowed) orig_class else "__PHP_Incomplete_Class");
+            const class_name = try keptClassName(ctx, class_allowed, orig_class);
             var p = colon1 + 2 + name_len + 2;
             const count_end = std.mem.indexOfPos(u8, s, p, ":") orelse return error.RuntimeError;
             const prop_count = std.fmt.parseInt(usize, s[p..count_end], 10) catch return error.RuntimeError;
@@ -787,10 +805,7 @@ fn unserializeValue(ctx: *NativeContext, uctx: *UnserCtx, s: []const u8, pos: us
                 } else if (fixed_data) |arr| {
                     if (key_result.value == .int) try arr.set(ctx.allocator, .{ .int = key_result.value.int }, val_result.value);
                 } else if (key_result.value == .string) {
-                    // property names are kept by the object as given, so
-                    // they need request-lifetime bytes rather than the
-                    // counted key that dies with this iteration
-                    const stripped = try ctx.createString(stripVisibilityPrefix(key_result.value.string.bytes()));
+                    const stripped = try keptPropertyName(ctx, obj, stripVisibilityPrefix(key_result.value.string.bytes()));
                     // when restoring into a kept class, assigning an
                     // __PHP_Incomplete_Class value to a typed property whose
                     // declared type isn't compatible is a TypeError in PHP
@@ -836,7 +851,7 @@ fn unserializeValue(ctx: *NativeContext, uctx: *UnserCtx, s: []const u8, pos: us
             if (colon1 + 2 + name_len + 1 >= s.len) return error.RuntimeError;
             const orig_class = s[colon1 + 2 .. colon1 + 2 + name_len];
             const class_allowed = uctx.classAllowed(orig_class) and ctx.vm.classes.contains(orig_class);
-            const class_name = try ctx.createString(if (class_allowed) orig_class else "__PHP_Incomplete_Class");
+            const class_name = try keptClassName(ctx, class_allowed, orig_class);
             var p = colon1 + 2 + name_len + 2;
             const len_end = std.mem.indexOfPos(u8, s, p, ":") orelse return error.RuntimeError;
             const data_len = std.fmt.parseInt(usize, s[p..len_end], 10) catch return error.RuntimeError;
diff --git a/src/stdlib/workers.zig b/src/stdlib/workers.zig
index b6d75303..4ad5d24a 100644
--- a/src/stdlib/workers.zig
+++ b/src/stdlib/workers.zig
@@ -3,6 +3,7 @@
 // Zphp\Future carries the result or the exception back. values cross as
 // serialized bytes and are materialized in the receiving VM
 const std = @import("std");
+const builtin = @import("builtin");
 const value_mod = @import("../runtime/value.zig");
 const Value = value_mod.Value;
 const PhpObject = value_mod.PhpObject;
@@ -18,6 +19,7 @@ const serialize = @import("serialize.zig");
 const network = @import("network.zig");
 const platform = @import("../platform.zig");
 const extension = @import("../extension.zig");
+const channel = @import("channel.zig");
 
 const pool_class = "Zphp\\Pool";
 const future_class = "Zphp\\Future";
@@ -30,6 +32,16 @@ const transfer_exception = "Zphp\\TransferException";
 
 const default_queue: usize = 1024;
 
+// memory that one thread allocates and another frees. the release build's
+// smp_allocator keeps a freelist per thread and reclaims another thread's
+// list only when its own runs dry, so a producer that allocates on one
+// thread for a consumer that frees on another keeps mapping fresh slabs;
+// libc malloc balances that. the Debug build keeps its leak-checking
+// allocator for everything
+pub fn transferAllocator(vm_allocator: std.mem.Allocator) std.mem.Allocator {
+    return if (builtin.mode == .Debug) vm_allocator else std.heap.c_allocator;
+}
+
 // ---------------------------------------------------------------------------
 // tasks
 
@@ -47,9 +59,9 @@ const Task = struct {
     id: u64,
     pool: *Pool,
     callable: []u8,
-    args: []u8,
+    args: Payload,
     state: TaskState = .queued,
-    result: ?[]u8 = null,
+    result: ?Payload = null,
     failure: ?Failure = null,
     fatal: ?[]u8 = null,
     cancel_requested: std.atomic.Value(bool) = std.atomic.Value(bool).init(false),
@@ -68,8 +80,8 @@ const Task = struct {
         const pool = t.pool;
         const a = pool.allocator;
         a.free(t.callable);
-        a.free(t.args);
-        if (t.result) |r| a.free(r);
+        t.args.free(a);
+        if (t.result) |r| r.free(a);
         if (t.failure) |f| {
             a.free(f.class_name);
             a.free(f.message);
@@ -186,6 +198,8 @@ const StartState = enum { starting, running, failed };
 
 const Pool = struct {
     allocator: std.mem.Allocator,
+    // the worker VMs allocate and free on their own thread
+    vm_allocator: std.mem.Allocator,
     owner: *VM,
     workers: []Worker,
     queue: Queue,
@@ -214,7 +228,8 @@ const Pool = struct {
         if (pool.refs.fetchSub(1, .acq_rel) == 1) pool.free();
     }
 
-    fn create(allocator: std.mem.Allocator, owner: *VM, workers: usize, bootstrap_path: ?[]const u8, queue_size: usize) !*Pool {
+    fn create(vm_allocator: std.mem.Allocator, owner: *VM, workers: usize, bootstrap_path: ?[]const u8, queue_size: usize) !*Pool {
+        const allocator = transferAllocator(vm_allocator);
         const pool = try allocator.create(Pool);
         errdefer allocator.destroy(pool);
         const items = try allocator.alloc(*Task, queue_size);
@@ -224,6 +239,7 @@ const Pool = struct {
         const wake = try platform.socketPair();
         pool.* = .{
             .allocator = allocator,
+            .vm_allocator = vm_allocator,
             .owner = owner,
             .workers = slots,
             .queue = .{ .items = items },
@@ -405,18 +421,18 @@ fn markCancelled(task: *Task) void {
 fn workerMain(w: *Worker) void {
     const pool = w.pool;
     current_worker = @intCast(w.index);
-    const vm = VM.initOnHeap(pool.allocator) catch {
+    const vm = VM.initOnHeap(pool.vm_allocator) catch {
         pool.reportStart("worker: out of memory");
         return;
     };
     var boot_result: ?*@import("../pipeline/compiler.zig").CompileResult = null;
     defer if (boot_result) |r| {
         r.deinit();
-        pool.allocator.destroy(r);
+        pool.vm_allocator.destroy(r);
     };
     defer {
         vm.deinit();
-        pool.allocator.destroy(vm);
+        pool.vm_allocator.destroy(vm);
     }
     vm.file_loader = pool.file_loader;
     vm.installHooks();
@@ -443,7 +459,7 @@ fn bootstrap(vm: *VM, pool: *Pool, path: []const u8) ?*@import("../pipeline/comp
         flushOutput(vm);
         pool.reportStart(msg);
         result.deinit();
-        pool.allocator.destroy(result);
+        pool.vm_allocator.destroy(result);
         return null;
     };
     flushOutput(vm);
@@ -500,7 +516,9 @@ fn runTask(vm: *VM, task: *Task) void {
 
 fn execute(ctx: *NativeContext, task: *Task) void {
     const callable = hold(serialize.unserializeFromString(ctx, task.callable) orelse return settleFatal(task, "the callable did not transfer"));
-    const args_value = hold(serialize.unserializeFromString(ctx, task.args) orelse return settleFatal(task, "the arguments did not transfer"));
+    const args_payload = task.args;
+    task.args = Payload.empty;
+    const args_value = hold(unpack(ctx, args_payload, task.pool.allocator) orelse return settleFatal(task, "the arguments did not transfer"));
     var args: [64]Value = undefined;
     var count: usize = 0;
     if (args_value == .array) {
@@ -520,15 +538,13 @@ fn execute(ctx: *NativeContext, task: *Task) void {
     // native result: strings need one more retain, containers are ours
     const owned = NativeResult.share(result).value;
     defer ctx.vm.releaseValue(owned);
-    const bytes = serialize.serializeToString(ctx, result) catch {
-        settleFatal(task, "the result could not be transferred back");
+    const payload = pack(ctx, result, "result", task.pool.allocator) catch {
+        settleFailure(ctx.vm, task, callable);
         return;
     };
-    defer bytes.value.string.release();
-    const copy = task.pool.allocator.dupe(u8, bytes.value.string.bytes()) catch return settleFatal(task, "out of memory");
     task.mutex.lock();
     defer task.mutex.unlock();
-    task.result = copy;
+    task.result = payload;
     task.state = .done;
     task.finished.broadcast();
 }
@@ -604,14 +620,31 @@ fn settleFailure(vm: *VM, task: *Task, callable: Value) void {
 }
 
 // ---------------------------------------------------------------------------
-// transfer rules, checked in the caller so the error names the path
+// transfer: a value crosses threads as serialized bytes plus a reference to
+// every channel inside it, so a channel that only the bytes name stays alive
+// until the receiver has bound its own wrapper
+
+pub const Payload = struct {
+    bytes: []u8,
+    channels: []*channel.Channel,
+
+    pub const empty = Payload{ .bytes = &.{}, .channels = &.{} };
 
+    pub fn free(p: Payload, a: std.mem.Allocator) void {
+        for (p.channels) |ch| ch.release();
+        a.free(p.channels);
+        a.free(p.bytes);
+    }
+};
+
+// the transfer rules are checked in the sender so the error names the path
 const TransferCheck = struct {
     ctx: *NativeContext,
     path: std.ArrayListUnmanaged(u8) = .{},
+    channels: std.ArrayListUnmanaged(*channel.Channel) = .{},
 
     fn refuse(self: *TransferCheck, what: []const u8) RuntimeError {
-        const msg = try std.fmt.allocPrint(self.ctx.allocator, "{s} cannot be transferred to a worker (at {s})", .{ what, self.path.items });
+        const msg = try std.fmt.allocPrint(self.ctx.allocator, "{s} cannot be transferred between threads (at {s})", .{ what, self.path.items });
         try self.ctx.vm.strings.append(self.ctx.allocator, msg);
         try self.ctx.vm.setPendingException(transfer_exception, msg);
         return error.RuntimeError;
@@ -635,6 +668,7 @@ const TransferCheck = struct {
                 }
             },
             .object => |obj| {
+                if (obj.native.get(channel.Channel, .channel)) |ch| return self.channels.append(self.ctx.allocator, ch);
                 if (obj.native.kind != .none) return self.refuse("an object backed by a native handle");
                 if (std.mem.eql(u8, obj.class_name, pool_class) or std.mem.eql(u8, obj.class_name, future_class)) return self.refuse("a pool or future");
                 const mark = self.path.items.len;
@@ -653,17 +687,30 @@ const TransferCheck = struct {
     }
 };
 
-fn checkTransferable(ctx: *NativeContext, v: Value, root: []const u8) RuntimeError!void {
+fn serializedCopy(ctx: *NativeContext, v: Value, allocator: std.mem.Allocator) RuntimeError![]u8 {
+    const bytes = try serialize.serializeToString(ctx, v);
+    defer bytes.value.string.release();
+    return allocator.dupe(u8, bytes.value.string.bytes());
+}
+
+pub fn pack(ctx: *NativeContext, v: Value, root: []const u8, allocator: std.mem.Allocator) RuntimeError!Payload {
     var tc = TransferCheck{ .ctx = ctx };
     defer tc.path.deinit(ctx.allocator);
+    defer tc.channels.deinit(ctx.allocator);
     try tc.path.appendSlice(ctx.allocator, root);
     try tc.check(v);
+    const bytes = try serializedCopy(ctx, v, allocator);
+    errdefer allocator.free(bytes);
+    const channels = try allocator.dupe(*channel.Channel, tc.channels.items);
+    for (channels) |ch| ch.retain();
+    return .{ .bytes = bytes, .channels = channels };
 }
 
-fn serializedCopy(ctx: *NativeContext, v: Value, allocator: std.mem.Allocator) RuntimeError![]u8 {
-    const bytes = try serialize.serializeToString(ctx, v);
-    defer bytes.value.string.release();
-    return allocator.dupe(u8, bytes.value.string.bytes());
+// materializes the value in this vm and drops the payload; the wrappers the
+// unserializer bound hold their own channel references
+pub fn unpack(ctx: *NativeContext, payload: Payload, allocator: std.mem.Allocator) ?Value {
+    defer payload.free(allocator);
+    return serialize.unserializeFromString(ctx, payload.bytes);
 }
 
 // ---------------------------------------------------------------------------
@@ -692,7 +739,7 @@ fn taskOf(obj: *PhpObject) ?*Task {
     return obj.native.get(Task, .future);
 }
 
-fn optionalSeconds(v: Value) ?u64 {
+pub fn optionalSeconds(v: Value) ?u64 {
     return switch (v) {
         .int => |i| if (i < 0) null else @as(u64, @intCast(i)) * std.time.ns_per_s,
         .float => |f| if (f < 0) null else @as(u64, @intFromFloat(f * @as(f64, @floatFromInt(std.time.ns_per_s)))),
@@ -747,7 +794,8 @@ fn submitTask(ctx: *NativeContext, args: []const Value, block: bool) RuntimeErro
     if (!valid) return throwNamed(ctx, transfer_exception, "tasks are named callables: a function name, 'Class::method', or [class, method]", .{});
     const task_args: Value = if (args.len >= 2) args[1] else .{ .array = try ctx.createArray() };
     if (task_args != .array) return throwNamed(ctx, pool_exception, "arguments must be an array", .{});
-    try checkTransferable(ctx, task_args, "args");
+    const packed_args = try pack(ctx, task_args, "args", pool.allocator);
+    errdefer packed_args.free(pool.allocator);
     pool.mutex.lock();
     const closed = pool.shutting_down;
     pool.mutex.unlock();
@@ -755,26 +803,24 @@ fn submitTask(ctx: *NativeContext, args: []const Value, block: bool) RuntimeErro
 
     const task = try pool.allocator.create(Task);
     errdefer pool.allocator.destroy(task);
-    task.* = .{ .id = pool.nextId(), .pool = pool, .callable = &.{}, .args = &.{} };
+    task.* = .{ .id = pool.nextId(), .pool = pool, .callable = &.{}, .args = packed_args };
     pool.retain();
     errdefer pool.release();
     task.callable = try serializedCopy(ctx, callable, pool.allocator);
     errdefer pool.allocator.free(task.callable);
-    task.args = try serializedCopy(ctx, task_args, pool.allocator);
-    errdefer pool.allocator.free(task.args);
 
     switch (pool.queue.push(task, block)) {
         .ok => {},
         .full => {
             pool.allocator.free(task.callable);
-            pool.allocator.free(task.args);
+            packed_args.free(pool.allocator);
             pool.allocator.destroy(task);
             pool.release();
             return NativeResult.scalar(.null);
         },
         .closed => {
             pool.allocator.free(task.callable);
-            pool.allocator.free(task.args);
+            packed_args.free(pool.allocator);
             pool.allocator.destroy(task);
             pool.release();
             return throwNamed(ctx, pool_exception, "the pool is shutting down", .{});
@@ -854,13 +900,12 @@ fn futureAwait(ctx: *NativeContext, args: []const Value) RuntimeError!NativeResu
     switch (state) {
         .done => {
             // materialized once; later awaits read the value kept on the future
-            if (task.result) |bytes| {
-                const v = serialize.unserializeFromString(ctx, bytes) orelse return throwNamed(ctx, task_exception, "the result did not transfer", .{});
+            if (task.result) |payload| {
+                task.result = null;
+                const v = unpack(ctx, payload, task.pool.allocator) orelse return throwNamed(ctx, task_exception, "the result did not transfer", .{});
                 // the store takes its own reference; the one unserialize handed over goes
                 try obj.set(ctx.allocator, "__result", v);
                 if (v == .string) v.string.release();
-                task.pool.allocator.free(bytes);
-                task.result = null;
             }
             return NativeResult.share(obj.get("__result"));
         },
diff --git a/tests/unserialize_magic_throws.php b/tests/unserialize_magic_throws.php
new file mode 100644
index 00000000..59c8d4aa
--- /dev/null
+++ b/tests/unserialize_magic_throws.php
@@ -0,0 +1,16 @@
+<?php
+// an exception thrown while unserialize() rebuilds an object propagates to
+// the caller; only a malformed payload is a warning plus false
+class Rebuilt {
+    public function __serialize(): array { return ['n' => 1]; }
+    public function __unserialize(array $data): void { throw new RuntimeException("rebuild refused " . $data['n']); }
+}
+class Woken {
+    public $n = 2;
+    public function __wakeup(): void { throw new LogicException("wake refused"); }
+}
+try { unserialize(serialize(new Rebuilt)); echo "no exception\n"; } catch (RuntimeException $e) { echo get_class($e), ": ", $e->getMessage(), "\n"; }
+try { unserialize(serialize(new Woken)); echo "no exception\n"; } catch (LogicException $e) { echo get_class($e), ": ", $e->getMessage(), "\n"; }
+var_dump(@unserialize('a:1:{i:0;s:3:"ab"}'));
+$nested = unserialize(serialize(['ok' => 1]));
+var_dump($nested);
diff --git a/tests/workers/basic.expected b/tests/workers/basic.expected
index 19eee832..504ddf5a 100644
--- a/tests/workers/basic.expected
+++ b/tests/workers/basic.expected
@@ -5,7 +5,7 @@ int(9)
 InvalidArgumentException: bad input 7
 Error: Call to undefined function missing_fn()
 transfer: tasks are named callables: a function name, 'Class::method', or [class, method]
-transfer: an object backed by a native handle cannot be transferred to a worker (at args[0])
+transfer: an object backed by a native handle cannot be transferred between threads (at args[0])
 int(2470)
 timeout
 bool(false)
diff --git a/tests/workers/channel.expected b/tests/workers/channel.expected
new file mode 100644
index 00000000..cecd7012
--- /dev/null
+++ b/tests/workers/channel.expected
@@ -0,0 +1,48 @@
+int(3)
+int(0)
+bool(false)
+bool(true)
+bool(true)
+int(3)
+bool(false)
+string(1) "a"
+array(1) {
+}
+NULL
+recv timeout: no value arrived in time
+bool(true)
+string(4) "last"
+recv closed: the channel is closed
+send closed: the channel is closed
+trySend closed: the channel is closed
+bad capacity: capacity must be a positive integer
+transfer: a Closure cannot be transferred between threads (at value)
+Error: Trying to clone an uncloneable object of class Zphp\Channel
+send timeout: the channel did not accept the value in time
+bool(false)
+string(1) "x"
+bool(true)
+string(1) "y"
+0 => 10
+1 => 20
+2 => 30
+bool(true)
+bool(true)
+stale: channel 999999 does not exist in this process
+string(14) "through a copy"
+int(20)
+int(2870)
+int(2)
+int(50)
+int(1)
+int(50)
+string(11) "produced 50"
+string(5) "x,y,z"
+string(7) "timeout"
+string(8) "got late"
+string(6) "closed"
+bool(true)
+string(17) "hello from worker"
+int(3)
+result: a Closure cannot be transferred between threads (at result)
+end
diff --git a/tests/workers/channel.php b/tests/workers/channel.php
new file mode 100644
index 00000000..8f90ccf1
--- /dev/null
+++ b/tests/workers/channel.php
@@ -0,0 +1,92 @@
+<?php
+// single thread: a bounded queue with value semantics for what crosses it
+$ch = new Zphp\Channel(capacity: 3);
+var_dump($ch->capacity(), count($ch), $ch->isClosed(), $ch instanceof Iterator, $ch instanceof Countable);
+$ch->send("a"); $ch->send(["k" => 2]); $ch->send(null);
+var_dump(count($ch), $ch->trySend(4));
+var_dump($ch->recv(), $ch->recv(), $ch->recv());
+try { $ch->recv(0.05); } catch (Zphp\TimeoutException $e) { echo "recv timeout: ", $e->getMessage(), "\n"; }
+$ch->send("last");
+$ch->close();
+var_dump($ch->isClosed(), $ch->recv());
+try { $ch->recv(); } catch (Zphp\ChannelException $e) { echo "recv closed: ", $e->getMessage(), "\n"; }
+try { $ch->send(1); } catch (Zphp\ChannelException $e) { echo "send closed: ", $e->getMessage(), "\n"; }
+try { $ch->trySend(1); } catch (Zphp\ChannelException $e) { echo "trySend closed: ", $e->getMessage(), "\n"; }
+try { new Zphp\Channel(0); } catch (Zphp\ChannelException $e) { echo "bad capacity: ", $e->getMessage(), "\n"; }
+try { (new Zphp\Channel)->send(fn() => 1); } catch (Zphp\TransferException $e) { echo "transfer: ", $e->getMessage(), "\n"; }
+try { clone $ch; } catch (Error $e) { echo get_class($e), ": ", $e->getMessage(), "\n"; }
+
+// a full channel refuses a non-blocking send and times out a bounded one
+$full = new Zphp\Channel(1);
+$full->send("x");
+try { $full->send("y", 0.05); } catch (Zphp\TimeoutException $e) { echo "send timeout: ", $e->getMessage(), "\n"; }
+var_dump($full->trySend("y"), $full->recv(), $full->trySend("y"), $full->recv());
+
+// iteration drains a closed channel, keys count from zero per consumer
+$it = new Zphp\Channel(5);
+foreach ([1, 2, 3] as $n) $it->send($n * 10);
+$it->close();
+foreach ($it as $k => $v) echo "$k => $v\n";
+
+// a channel crosses as its id and binds to the same channel; a stale id is an error
+$again = unserialize(serialize($it));
+var_dump($again->isClosed(), $again->id() === $it->id());
+try { unserialize('O:12:"Zphp\Channel":1:{s:2:"id";i:999999;}'); } catch (Zphp\ChannelException $e) { echo "stale: ", $e->getMessage(), "\n"; }
+$carrier = new Zphp\Channel(1);
+$inner = new Zphp\Channel(1);
+$carrier->send(['inner' => $inner]);
+unset($inner);
+$got = $carrier->recv();
+$got['inner']->send("through a copy");
+var_dump($got['inner']->recv());
+
+// across threads
+$pool = new Zphp\Pool(workers: 2, bootstrap: __DIR__ . "/worker.php");
+
+// fan out jobs to two consumers, collect on a second channel
+$jobs = new Zphp\Channel(4);
+$results = new Zphp\Channel(100);
+$a = $pool->submit('consume', [$jobs, $results]);
+$b = $pool->submit('consume', [$jobs, $results]);
+for ($i = 1; $i <= 20; $i++) $jobs->send($i);
+$jobs->close();
+var_dump($a->await() + $b->await());
+$results->close();
+$sum = 0; $workers = []; foreach ($results as $r) { $sum += $r['sq']; $workers[$r['worker']] = true; }
+var_dump($sum, count($workers));
+
+// a worker produces, main iterates, capacity one keeps them in step
+$stream = new Zphp\Channel(1);
+$p = $pool->submit('produce', [$stream, 50]);
+$seen = []; foreach ($stream as $k => $v) $seen[$k] = $v;
+var_dump(count($seen), $seen[0], $seen[49], $p->await());
+
+// a pipeline of three stages
+$s1 = new Zphp\Channel(2); $s2 = new Zphp\Channel(2); $s3 = new Zphp\Channel(2);
+$pool->submit('forward', [$s1, $s2]); $pool->submit('forward', [$s2, $s3]);
+foreach (["x", "y", "z"] as $v) $s1->send($v);
+$s1->close();
+var_dump(implode(",", iterator_to_array($s3, false)));
+
+// a worker blocked in recv sees the timeout, then a value, then the close
+$c = new Zphp\Channel(1);
+var_dump($pool->submit('wait_recv', [$c, 0.05])->await());
+$w = $pool->submit('wait_recv', [$c, 5]); usleep(50000); $c->send("late"); var_dump($w->await());
+$w = $pool->submit('wait_recv', [$c, 5]); usleep(50000); $c->close(); var_dump($w->await());
+
+// a channel created in a worker comes back in a result and outlives the worker's wrapper
+$r = $pool->submit('make_channel')->await();
+var_dump($r['ch'] instanceof Zphp\Channel, $r['ch']->recv());
+
+// main drops its wrapper while a worker still consumes
+$d = new Zphp\Channel(2);
+$f = $pool->submit('slow_drain', [$d]);
+$d->send(1); $d->send(2); $d->send(3); $d->close();
+unset($d);
+var_dump($f->await());
+
+// a result that cannot cross fails the task with the path
+try { $pool->submit('bad_result')->await(); } catch (Zphp\TransferException $e) { echo "result: ", $e->getMessage(), "\n"; }
+
+$pool->shutdown();
+echo "end\n";
diff --git a/tests/workers/memory.php b/tests/workers/memory.php
index 86738125..dd526139 100644
--- a/tests/workers/memory.php
+++ b/tests/workers/memory.php
@@ -12,6 +12,13 @@ function big(int $n): array { return array_fill(0, $n, str_repeat("x", 64)); }
     for ($i = 0; $i < 200; $i++) { $pool->submit('str_repeat', ['x', 8000]); }
     while ($pool->collect(5.0)) {}
 }
+$stream = new Zphp\Channel(8);
+for ($batch = 0; $batch < 10; $batch++) {
+    $f = $pool->submit('produce_big', [$stream, 300]);
+    for ($i = 0; $i < 300; $i++) { $stream->recv(); }
+    $f->await();
+}
+for ($i = 0; $i < 2000; $i++) { $carrier = new Zphp\Channel(1); $carrier->send(['inner' => new Zphp\Channel(1)]); $carrier->recv()['inner']->trySend("x"); }
 $growth = memory_get_usage() - $before;
 echo $growth < 16 * 1024 * 1024 ? "memory bounded\n" : "memory grew by $growth\n";
 exit($growth < 16 * 1024 * 1024 ? 0 : 1);
diff --git a/tests/workers/run b/tests/workers/run
index 55b34d0a..d5f35449 100755
--- a/tests/workers/run
+++ b/tests/workers/run
@@ -19,7 +19,7 @@ check() {
     fi
 }
 
-for script in basic edge; do
+for script in basic edge channel; do
     actual="$("$ZPHP" run "$SCRIPT_DIR/$script.php" 2>&1 | tr -d '\r' | grep -v '^error(gpa)' | grep -v '^\s' | grep -v '^/')" || true
     check "$script.php" "$(tr -d '\r' < "$SCRIPT_DIR/$script.expected")" "$actual"
 done
diff --git a/tests/workers/worker.php b/tests/workers/worker.php
index 9b0d1163..f3032925 100644
--- a/tests/workers/worker.php
+++ b/tests/workers/worker.php
@@ -12,3 +12,21 @@ function until_cancelled(): string { while (!Zphp\Task::cancelled()) usleep(1000
 function counter(): int { static $n = 0; return ++$n; }
 function big(int $n): array { return array_fill(0, $n, str_repeat("x", 64)); }
 function echoes(string $s): void { echo $s, "\n"; }
+function consume(Zphp\Channel $jobs, Zphp\Channel $results): int {
+    $n = 0;
+    foreach ($jobs as $job) { $results->send(['job' => $job, 'worker' => Zphp\Task::worker(), 'sq' => $job * $job]); $n++; }
+    return $n;
+}
+function produce(Zphp\Channel $out, int $count): string {
+    for ($i = 1; $i <= $count; $i++) $out->send($i);
+    $out->close();
+    return "produced $count";
+}
+function wait_recv(Zphp\Channel $ch, float $t): string {
+    try { return "got " . $ch->recv($t); } catch (Zphp\TimeoutException $e) { return "timeout"; } catch (Zphp\ChannelException $e) { return "closed"; }
+}
+function make_channel(): array { $c = new Zphp\Channel(3); $c->send("hello from worker"); return ['ch' => $c]; }
+function slow_drain(Zphp\Channel $ch): int { $n = 0; foreach ($ch as $v) { usleep(20000); $n++; } return $n; }
+function forward(Zphp\Channel $in, Zphp\Channel $out): void { foreach ($in as $v) $out->send($v); $out->close(); }
+function bad_result(): Closure { return fn() => 1; }
+function produce_big(Zphp\Channel $out, int $n): void { for ($i = 0; $i < $n; $i++) $out->send(big(100)); }

From f982e17d2699df43dfd4545770fe87d19fd4202d Mon Sep 17 00:00:00 2001
From: nvms <jon@pye.rs>
Date: Mon, 14 Sep 2026 11:27:25 -0400
Subject: [PATCH 2/3] worker pool: a future awaited directly never leaves a
 wake byte behind

futureAwait can take delivery between the task settling and the worker's
complete(), which then appended the task and wrote a wake byte nobody
read. linux charges a one-byte unix socket send hundreds of bytes of
buffer, so a few hundred such races filled it and every worker blocked in
send forever; the memory soak hung on ci while passing on macos. complete
now skips a task that is already delivered, and both ends of the wake are
non-blocking so a backlog nobody collects cannot stall a worker either.
the retention step gets a five minute limit
---
 .github/workflows/ci.yml |  1 +
 src/stdlib/workers.zig   | 17 +++++++++++++++++
 2 files changed, 18 insertions(+)

diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 431cec13..32ae5a49 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -170,6 +170,7 @@ jobs:
       - run: zig build -Doptimize=ReleaseFast
       - run: python3 ./tests/memory_soak
       - name: worker pool retention
+        timeout-minutes: 5
         run: ./zig-out/bin/zphp run tests/workers/memory.php
 
   php-compat:
diff --git a/src/stdlib/workers.zig b/src/stdlib/workers.zig
index 4ad5d24a..1cf7987f 100644
--- a/src/stdlib/workers.zig
+++ b/src/stdlib/workers.zig
@@ -236,7 +236,12 @@ const Pool = struct {
         errdefer allocator.free(items);
         const slots = try allocator.alloc(Worker, workers);
         errdefer allocator.free(slots);
+        // the wake is a level: one byte per pending completion, written
+        // without blocking so a backlog nobody collects can never stall a
+        // worker, and read without blocking so a missing byte is not waited on
         const wake = try platform.socketPair();
+        try platform.setNonBlocking(wake[0], true);
+        try platform.setNonBlocking(wake[1], true);
         pool.* = .{
             .allocator = allocator,
             .vm_allocator = vm_allocator,
@@ -287,8 +292,20 @@ const Pool = struct {
         return id;
     }
 
+    // a future awaited directly may take delivery between the task settling
+    // and this call; then the completion list and the wake must not see it,
+    // or the byte nobody reads accumulates until the socket buffer stalls
+    // every worker
     fn complete(pool: *Pool, task: *Task) void {
         pool.mutex.lock();
+        task.mutex.lock();
+        const delivered = task.delivered;
+        task.mutex.unlock();
+        if (delivered) {
+            pool.mutex.unlock();
+            task.release();
+            return;
+        }
         pool.completed.append(pool.allocator, task) catch {};
         pool.changed.broadcast();
         pool.mutex.unlock();

From 41ced6da0a1b5f33e033d2c1f3f69544dedc0df8 Mon Sep 17 00:00:00 2001
From: nvms <jon@pye.rs>
Date: Mon, 14 Sep 2026 11:45:00 -0400
Subject: [PATCH 3/3] worker pool: worker vm heaps on libc malloc in release

on the four-cpu ci runner the soak grew 17 to 93 MB on some runs and stayed
flat on others with no cross-thread frees involved: smp_allocator starts
every thread on slot zero and moves it only on contention, so two busy
worker vms scattered their freelists across slots and kept mapping slabs.
the whole pool, worker vm heaps included, now uses the transfer allocator;
the scaling benchmark is unchanged
---
 src/stdlib/workers.zig | 25 ++++++++++++-------------
 1 file changed, 12 insertions(+), 13 deletions(-)

diff --git a/src/stdlib/workers.zig b/src/stdlib/workers.zig
index 1cf7987f..a9f16a3d 100644
--- a/src/stdlib/workers.zig
+++ b/src/stdlib/workers.zig
@@ -32,12 +32,14 @@ const transfer_exception = "Zphp\\TransferException";
 
 const default_queue: usize = 1024;
 
-// memory that one thread allocates and another frees. the release build's
-// smp_allocator keeps a freelist per thread and reclaims another thread's
-// list only when its own runs dry, so a producer that allocates on one
-// thread for a consumer that frees on another keeps mapping fresh slabs;
-// libc malloc balances that. the Debug build keeps its leak-checking
-// allocator for everything
+// everything the pool touches, worker VM heaps included. the release build's
+// smp_allocator keeps a freelist per thread slot, reclaims another slot's
+// list only when its own runs dry, and hands every thread slot zero until
+// contention moves it, so a producer that allocates on one thread for a
+// consumer that frees on another keeps mapping fresh slabs, and two busy
+// worker VMs on a four-cpu box scatter their frees across slots and grew
+// rss by up to 90 MB over a soak that other runs finished flat. libc malloc
+// balances both. the Debug build keeps its leak-checking allocator
 pub fn transferAllocator(vm_allocator: std.mem.Allocator) std.mem.Allocator {
     return if (builtin.mode == .Debug) vm_allocator else std.heap.c_allocator;
 }
@@ -198,8 +200,6 @@ const StartState = enum { starting, running, failed };
 
 const Pool = struct {
     allocator: std.mem.Allocator,
-    // the worker VMs allocate and free on their own thread
-    vm_allocator: std.mem.Allocator,
     owner: *VM,
     workers: []Worker,
     queue: Queue,
@@ -244,7 +244,6 @@ const Pool = struct {
         try platform.setNonBlocking(wake[1], true);
         pool.* = .{
             .allocator = allocator,
-            .vm_allocator = vm_allocator,
             .owner = owner,
             .workers = slots,
             .queue = .{ .items = items },
@@ -438,18 +437,18 @@ fn markCancelled(task: *Task) void {
 fn workerMain(w: *Worker) void {
     const pool = w.pool;
     current_worker = @intCast(w.index);
-    const vm = VM.initOnHeap(pool.vm_allocator) catch {
+    const vm = VM.initOnHeap(pool.allocator) catch {
         pool.reportStart("worker: out of memory");
         return;
     };
     var boot_result: ?*@import("../pipeline/compiler.zig").CompileResult = null;
     defer if (boot_result) |r| {
         r.deinit();
-        pool.vm_allocator.destroy(r);
+        pool.allocator.destroy(r);
     };
     defer {
         vm.deinit();
-        pool.vm_allocator.destroy(vm);
+        pool.allocator.destroy(vm);
     }
     vm.file_loader = pool.file_loader;
     vm.installHooks();
@@ -476,7 +475,7 @@ fn bootstrap(vm: *VM, pool: *Pool, path: []const u8) ?*@import("../pipeline/comp
         flushOutput(vm);
         pool.reportStart(msg);
         result.deinit();
-        pool.vm_allocator.destroy(result);
+        pool.allocator.destroy(result);
         return null;
     };
     flushOutput(vm);