From 9332d11d71852204226bcdb1a0b081bf001ccb3e Mon Sep 17 00:00:00 2001 From: Claude Fable 5 Date: Sun, 16 Aug 2026 09:54:27 +0000 Subject: [PATCH 1/3] feat(parallel): accept capacity 0 on a shared channel as a real rendezvous declareShared('x', SharedChannel::class, capacity: 0) used to be refused, because the substrate's handoff gate counted only receivers parked inside its own blocking recv() and this runtime never calls it - so a capacity-0 shared channel would have accepted a send approximately never. The substrate now names the missing half (registerReceiver()/cancelReceiver(), trySendTicket()/isTicketTaken()), and this is the consumer side of it. SharedChannel registers this process with the substrate while any local coroutine is waiting to receive, and withdraws when the last one leaves. The state is derived from the waiter list rather than counted up and down, because a count that drifts is exactly the stale registration the design has to prevent: one decrement too many closes a gate that should be open, one too few leaves siblings depositing values for a coroutine that has moved on. Every path that mutates the list - a park, a select registration, a select cancellation, a served wakeup - runs the same derivation. A rendezvous send returns only once the value has been TAKEN: the deposit hands back a ticket and the sender parks a second time on it. That is what keeps cancellation sound. A registration withdrawn between the deposit and the take leaves the record in the one ring slot, where the next receiver completes the handshake while the sender goes on waiting - the state it would have been in had it never found a partner. Nothing is lost and nothing is delivered twice, so a select loser never has to deliver a value it can no longer deliver. The cost is one narrower refusal in place of the old blanket one: a capacity-0 shared channel cannot be a select SEND case. A case must resolve without parking, and the deposit - the only non-parking moment - is one step too early to promise a take, so claiming there would quietly give that one case buffered semantics while send() on the same channel keeps rendezvous ones. It is refused with both remedies named. Receive cases compose with local channels as before, and canSend() answers false for an open rendezvous so select's fast path never walks into a send that parks. Two supporting details: serve() skips the coroutine that is currently running, since it is on its way into a park and has not suspended yet (unpark() would not schedule it, and a claimed select token would strand it); and shared channels are now created with a waiter table as large as the wake registry, so a registration can never be refused for a family that the notification plane itself can serve. Runtime CI stays red until the substrate PR merges - this depends on the new upstream methods. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01U5QRWPfmkZME7hDjU6i5EA --- AGENTS.md | 15 +- README.md | 11 +- src/Parallel/SharedArena.php | 26 +- src/Parallel/SharedChannel.php | 256 +++++++++++++++++- ...extWithdrawsItsRendezvousRegistration.phpt | 65 +++++ ...stsABoundedNumberOfWakeupsOnBothSides.phpt | 79 ++++++ ...nelComposesInASelectWithALocalChannel.phpt | 80 ++++++ ...tACapacityZeroSharedChannelIsAccepted.phpt | 51 ++++ ...zvousReceiverParksUntilASenderArrives.phpt | 62 +++++ ...WaitsUntilAnotherProcessTakesTheValue.phpt | 71 +++++ ...tARendezvousWaitIsNeverAFalseDeadlock.phpt | 64 +++++ ...NoRendezvousRegistrationForALaterSend.phpt | 71 +++++ ...OnACapacityZeroSharedChannelIsRefused.phpt | 64 +++++ ...aringASharedRootAfterTheForkIsRefused.phpt | 11 - ...fOneProcessRendezvousOnASharedChannel.phpt | 77 ++++++ tests/Support/shared.php | 130 +++++++++ 16 files changed, 1094 insertions(+), 39 deletions(-) create mode 100644 tests/Functional/testACancelledContextWithdrawsItsRendezvousRegistration.phpt create mode 100644 tests/Functional/testACapacityZeroHandoffCostsABoundedNumberOfWakeupsOnBothSides.phpt create mode 100644 tests/Functional/testACapacityZeroSharedChannelComposesInASelectWithALocalChannel.phpt create mode 100644 tests/Functional/testACapacityZeroSharedChannelIsAccepted.phpt create mode 100644 tests/Functional/testARendezvousReceiverParksUntilASenderArrives.phpt create mode 100644 tests/Functional/testARendezvousSendWaitsUntilAnotherProcessTakesTheValue.phpt create mode 100644 tests/Functional/testARendezvousWaitIsNeverAFalseDeadlock.phpt create mode 100644 tests/Functional/testASelectLoserLeavesNoRendezvousRegistrationForALaterSend.phpt create mode 100644 tests/Functional/testASelectSendCaseOnACapacityZeroSharedChannelIsRefused.phpt create mode 100644 tests/Functional/testTwoCoroutinesOfOneProcessRendezvousOnASharedChannel.phpt diff --git a/AGENTS.md b/AGENTS.md index 6f9213b..f045229 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -233,9 +233,18 @@ in every process. ### Known limits of the parallel surface, stated rather than discovered -- **A shared channel needs capacity ≥ 1.** The substrate's cross-process rendezvous only accepts a - send while a sibling is parked inside *its* blocking `recv()`, which this runtime never calls. - Capacity 0 is refused at declaration instead of delivered as a channel that usually does nothing. +- **A capacity-0 shared channel works, but not as a `select` send case.** The substrate's handoff + gate asks whether a receiver is waiting, and it used to count only receivers parked inside *its* + blocking `recv()` — which this runtime never calls. It now also counts a receiver **registered** + from here (`registerReceiver()`/`cancelReceiver()`), so a Fiber parked on this poller is a valid + rendezvous partner and `declareShared(..., capacity: 0)` is accepted. The registration is a claim + about presence, never about storage: the record still goes into the one ring slot a capacity-0 + channel allocates, which is what makes a withdrawal total — a select loser cancels and the record + it may have attracted simply waits in the ring for the next receiver while its sender stays + parked. Nothing is lost and nothing is delivered twice. `send()` therefore returns only once the + value has been **taken**, and that is exactly why a rendezvous cannot be a `select` **send** case: + a case must resolve without parking, and the deposit — the only non-parking moment — is one step + too early. That case is refused with both remedies named; receive cases compose as usual. - **Graphs are keyed per instance, and each unpublished spawn keeps its memory until teardown.** The substrate registers a persisted graph under a name minted from its own root address (`persistInstance()`), so any number of tasks of one class are in flight at once and none diff --git a/README.md b/README.md index e8eebd7..1e741df 100644 --- a/README.md +++ b/README.md @@ -492,10 +492,13 @@ z-engine requires it, and z-engine is a hard dependency of this package. whichever process filled it. Closures are shareable only by **pre-fork registration** (`registerSharedClosure()`); work created after the fork travels as a `Task`. Anything else throws `NotShareableValueException` naming the remedy. -- **A shared channel needs capacity ≥ 1.** A cross-process rendezvous only accepts a send while a - sibling is parked inside the substrate's own blocking `recv()`, and this runtime parks Fibers on its - poller instead — so capacity 0 is refused rather than delivered as a channel that usually does - nothing. +- **A capacity-0 shared channel is a real rendezvous, except as a `select` send case.** A + cross-process handoff is accepted while a receiver is waiting, and a Fiber parked on this + runtime's poller counts as one: the channel registers this process with the substrate while it has + a waiting receiver and withdraws when the last one leaves. `send()` returns once the value has + been **taken**, which is also why a rendezvous cannot be a `select` *send* case — a case has to + resolve without parking, and the deposit is one step too early to promise a take. That one case is + refused with the remedies named; receive cases mix with local channels as usual. - **`persist()` is per instance, roots are per name.** Two `RenderJob`s — or twenty — are twenty graphs, none superseding another, and two roots of one class are two roots. What a design pays for spawning arbitrary unpublished tasks is arena memory per spawn, held until teardown; diff --git a/src/Parallel/SharedArena.php b/src/Parallel/SharedArena.php index e23678e..24e657f 100644 --- a/src/Parallel/SharedArena.php +++ b/src/Parallel/SharedArena.php @@ -142,6 +142,16 @@ final class SharedArena /** One entry per attached process, holding its wake slot; the broadcast list. */ private readonly SharedArray $family; + /** + * Wake slots the registry was created with — the size of the largest family this arena serves. + * + * Also the waiter capacity every shared channel is created with: a capacity-0 channel registers + * one entry per process that has a receiver waiting, so a table smaller than the family could + * refuse a registration that is perfectly legitimate, and a table larger than it could never be + * filled. Pre-sized either way — no arena table ever grows. + */ + private readonly int $wakeSlots; + /** * Roots declared before the fork: name => descriptor. Every child inherits this table with the * rest of the parent's heap, so a worker resolving a root needs no lookup protocol at all. @@ -209,7 +219,8 @@ public function __construct( $slotCount, self::SLOTS_ROOT, ); - $this->family = SharedArray::create($this->allocator, $this->codec, $wakeSlots, self::FAMILY_ROOT); + $this->family = SharedArray::create($this->allocator, $this->codec, $wakeSlots, self::FAMILY_ROOT); + $this->wakeSlots = $wakeSlots; // The panic path persists one of these in whichever worker died, and the waiter attaches it // by address. Loaded here so the whole family agrees on its class entry. @@ -554,12 +565,15 @@ public function shared(string $name): mixed private function createRoot(string $name, string $class, int $capacity): array { if (is_a($class, SharedChannel::class, true) || is_a($class, SubstrateChannel::class, true)) { - if ($capacity < 1) { + if ($capacity < 0) { + // Capacity 0 is a rendezvous, exactly as it is for a local Channel: the substrate + // now lets a receiver parked on this runtime's poller count as the handoff partner + // (registerReceiver()), so nothing has to spin inside the substrate for it to work. throw new \InvalidArgumentException(sprintf( - 'shared channel "%s" needs a capacity of at least 1: a cross-process rendezvous ' - . 'accepts a send only while a sibling is parked inside the substrate\'s own ' - . 'blocking recv(), and this runtime parks Fibers on its poller instead', + 'shared channel "%s" cannot have a negative capacity, got %d; 0 is a ' + . 'cross-process rendezvous and a positive number buffers that many records', $name, + $capacity, )); } @@ -568,7 +582,7 @@ private function createRoot(string $name, string $class, int $capacity): array $this->codec, $this->wake, $capacity, - SubstrateChannel::DEFAULT_WAITERS, + max($this->wakeSlots, SubstrateChannel::DEFAULT_WAITERS), $name, ); diff --git a/src/Parallel/SharedChannel.php b/src/Parallel/SharedChannel.php index 479ce38..426df83 100644 --- a/src/Parallel/SharedChannel.php +++ b/src/Parallel/SharedChannel.php @@ -48,13 +48,28 @@ * another process got there first. That is what makes a spurious wakeup harmless and keeps the * whole design on the safe side of the level-triggered contract. * - * # Capacity 0 is not available here + * # Capacity 0 is a real cross-process rendezvous * - * The substrate's rendezvous handshake counts receivers that parked through *its* blocking `recv()`, - * and this class deliberately never calls it. A capacity-0 shared channel would therefore accept a - * send only while a sibling happened to be spinning inside the substrate, which is not a semantics - * anybody can build on — so it is refused at declaration rather than delivered as a channel that - * usually does not hand anything over. Cross-process rendezvous stays with the substrate's own API. + * The substrate's handshake gates a capacity-0 handoff on "is a receiver waiting", and it used to + * count only receivers parked inside *its* blocking `recv()` — which this runtime never calls. The + * substrate now names the other half: `registerReceiver()` announces a receiver that is parked in + * the consumer's own event loop, and `cancelReceiver()` withdraws it. This class registers exactly + * once per process while any local coroutine is waiting to receive, and withdraws the moment the + * last one is gone, so a sibling's `trySend()` has a partner precisely while this process has one. + * + * A rendezvous `send()` is over when the value has been **taken**, not when it was deposited: the + * deposit hands back a ticket, and the sender parks a second time until `isTicketTaken()`. That is + * what makes the semantics honest under cancellation — a registration that is withdrawn between the + * deposit and the take leaves the record in the ring for the next receiver, and the sender simply + * goes on waiting, exactly as it would have if it had never found a partner. Nothing is lost, and + * nothing is delivered twice. + * + * The one thing a rendezvous cannot do here is **lose a `select` race as a send case**. A select + * case has to resolve without parking, and the only point at which a rendezvous send could be + * declared complete without parking is the deposit — one step too early, since the partner it was + * deposited against may still walk away. Rather than quietly downgrading such a case to buffered + * semantics, {@see self::awaitSendable()} refuses it and names the two remedies. Receive cases are + * unaffected and compose with local channels as usual. * * The value type is deliberately `mixed`: what may travel is decided by the tag table, not by * a PHP generic, and a shared channel hands back exactly what the substrate's codec materialized @@ -67,12 +82,31 @@ final class SharedChannel implements ChannelInterface /** * Coroutines parked on this channel in this process. * - * @var list + * `ticket` belongs to a sender that already deposited a rendezvous record and is waiting for it + * to be taken — a different readiness question from "is there room", and the only one that can + * still be answered once the ring is full of this very sender's handoff. + * + * @var list */ private array $waiters = []; private bool $rechecking = false; + /** + * This process's rendezvous registration in the substrate, while it has a waiting receiver. + * + * One per process rather than one per coroutine: the registration says "somebody here is ready + * to take a value", and the wake slot it is filed under is a property of the process anyway. + */ + private ?int $receiverToken = null; + public function __construct( private readonly SharedArena $arena, private readonly SubstrateChannel $channel, @@ -93,15 +127,22 @@ public function send(mixed $value): void throw ClosedChannelException::onSend(); } - // Encoding happens inside trySend(), outside every lock: interning a string allocates - // arena memory and a value that cannot be shared must throw before a lock is taken. - if ($this->channel->trySend($value)) { + // Encoding happens inside trySendTicket(), outside every lock: interning a string + // allocates arena memory and a value that cannot be shared must throw before a lock is + // taken. + $ticket = $this->channel->trySendTicket($value); + + if ($ticket !== null) { $this->announce(WakeOpcode::Wake); + if ($this->isRendezvous()) { + $this->awaitHandoff($ticket); + } + return; } - $this->parkOn(true, sprintf('send on shared channel @0x%X', $this->address())); + $this->parkOn(true, null, sprintf('send on shared channel @0x%X', $this->address())); } } @@ -148,7 +189,19 @@ public function canSend(): bool { // A closed channel is "ready" to send: the send returns immediately, by throwing. Reporting // false would park a select on a channel that can never make progress. - return $this->channel->isClosed() || $this->channel->count() < $this->channel->capacity(); + if ($this->channel->isClosed()) { + return true; + } + + // A rendezvous send completes when the value has been TAKEN, and nothing observable right + // now can make that already true — even with a partner registered, the take is a second + // event that only a park can wait for. Answering anything else here would put `select`'s + // non-parking fast path into a send that parks. + if ($this->isRendezvous()) { + return false; + } + + return $this->channel->count() < $this->channel->capacity(); } public function canRecv(): bool @@ -156,6 +209,25 @@ public function canRecv(): bool return $this->channel->count() > 0 || $this->channel->isClosed(); } + /** Whether this channel hands values over directly rather than buffering them. */ + public function isRendezvous(): bool + { + return $this->channel->capacity() === 0; + } + + /** + * Whether any process of the family has a receiver waiting for a handoff right now. + * + * The gate a capacity-0 send passes, readable from anywhere: it is the shared count, not this + * process's waiter list, so a registration left behind by a select loser in *another* process + * shows up here too. That is what makes "no stale registration survives" testable across + * processes rather than only locally. + */ + public function hasWaitingReceiver(): bool + { + return $this->channel->parkedReceivers() > 0; + } + /** * The wake registry's socket — the descriptor readiness of this channel is signalled through. * @@ -178,17 +250,35 @@ public function awaitReceivable(SelectToken $token, int $caseIndex, CoroutineInt 'token' => $token, 'case' => $caseIndex, 'value' => null, + 'ticket' => null, ]; + + $this->syncRegistration(); } public function awaitSendable(SelectToken $token, int $caseIndex, CoroutineInterface $coroutine, mixed $value): void { + // A select case must resolve without parking, and a rendezvous send has no such moment: the + // deposit is the earliest point it could claim, and the partner it was deposited against + // can still walk away before taking the value. Reporting "sent" there would silently give + // this one case buffered semantics while send() on the same channel keeps rendezvous ones. + if ($this->isRendezvous() && !$this->channel->isClosed()) { + throw new \LogicException(sprintf( + 'a capacity-0 shared channel cannot be a select send case: a rendezvous send ' + . 'completes when the value is TAKEN, which a case cannot wait for without parking. ' + . 'Declare the channel with a capacity of at least 1, or drive the handoff from a ' + . 'coroutine of its own that calls send() on shared channel @0x%X', + $this->address(), + )); + } + $this->waiters[] = [ 'coroutine' => $coroutine, 'send' => true, 'token' => $token, 'case' => $caseIndex, 'value' => $value, + 'ticket' => null, ]; } @@ -198,6 +288,11 @@ public function cancelWait(SelectToken $token): void $this->waiters, static fn(array $waiter): bool => $waiter['token'] !== $token, )); + + // The losing case of a select is exactly the stale registration this has to avoid: a + // withdrawn waiter that still tells a sibling process a partner is present would make the + // next send deposit a value nobody in this process is coming for. + $this->syncRegistration(); } /** @@ -239,7 +334,29 @@ private function receive(): ?Delivery return $received[1] ? new Delivery($received[0]) : null; } - $this->parkOn(false, sprintf('recv on shared channel @0x%X', $this->address())); + $this->parkOn(false, null, sprintf('recv on shared channel @0x%X', $this->address())); + } + } + + /** + * The second half of a rendezvous send: wait until somebody has actually taken the record. + * + * The ticket is the ring position the value was deposited at, and the substrate reports it + * taken once its monotonic head has passed it — which is true no matter *who* took it. That is + * the whole reason this runtime does not need the deposit to bind one particular receiver: if + * the registration it was deposited against is withdrawn a moment later, the record stays in + * the ring, the next receiver completes the handshake, and this park simply lasts longer. + */ + private function awaitHandoff(int $ticket): void + { + while (!$this->channel->isTicketTaken($ticket)) { + if ($this->channel->isClosed()) { + // Closed with the handoff still in the ring: no receiver is coming for it, and a + // sender that waits for a take that cannot happen is a hang, not a rendezvous. + throw ClosedChannelException::whileParked(); + } + + $this->parkOn(true, $ticket, sprintf('handoff on shared channel @0x%X', $this->address())); } } @@ -277,11 +394,26 @@ private function serve(): void $scheduler = $this->arena->scheduler(); $remaining = []; + // The coroutine running right now is on its way INTO a park and has not suspended yet: + // unpark() would report nothing to do and a claimed select token would strand it with no + // scheduler entry. Skipping it is safe rather than merely cautious — the only thing that + // can have made it eligible in this window is another process, and that process's wake + // event is already sitting in this process's socket, so the poller re-runs this pass the + // moment the coroutine suspends. A local change cannot happen in the window at all: there + // is no suspension point between the caller's own poll and its park. + $current = $scheduler->current(); + foreach ($this->waiters as $waiter) { + if ($waiter['coroutine'] === $current) { + $remaining[] = $waiter; + + continue; + } + $token = $waiter['token']; if ($token === null) { - if (!($waiter['send'] ? $this->canSend() : $this->canRecv())) { + if (!$this->canProceed($waiter['send'], $waiter['ticket'])) { $remaining[] = $waiter; continue; @@ -314,6 +446,93 @@ private function serve(): void } $this->waiters = $remaining; + + $this->syncRegistration(); + } + + /** + * Whether a plain parked waiter can retry its operation now. + * + * Deliberately not {@see self::canSend()}: that answers `select`'s question ("would a send + * complete without parking?"), which is always no on a rendezvous channel. A parked sender asks + * the narrower one — can the value be *deposited* — and a sender that already deposited asks + * the narrower one still, whether its own record has been taken. + */ + private function canProceed(bool $forSend, ?int $ticket): bool + { + if (!$forSend) { + return $this->canRecv(); + } + + if ($ticket !== null) { + return $this->channel->isTicketTaken($ticket) || $this->channel->isClosed(); + } + + if ($this->channel->isClosed()) { + return true; + } + + if ($this->isRendezvous()) { + // The gate the substrate applies to the deposit itself, asked before parking again: + // an empty handoff slot and a receiver waiting somewhere in the family. + return $this->channel->count() === 0 && $this->channel->parkedReceivers() > 0; + } + + return $this->channel->count() < $this->channel->capacity(); + } + + /** + * Keep the substrate registration in step with whether this process has a waiting receiver. + * + * Derived from the waiter list rather than counted up and down, because a count that drifts is + * exactly the stale registration this exists to prevent: one extra decrement closes a gate that + * should be open, one missing one leaves a sibling depositing values for a coroutine that has + * long since moved on. Only a rendezvous channel needs it — a buffered send is gated by room in + * the ring, and registering there would buy nothing but a lock per park. + */ + private function syncRegistration(): void + { + if (!$this->isRendezvous()) { + return; + } + + $wanted = false; + foreach ($this->waiters as $waiter) { + if (!$waiter['send']) { + $wanted = true; + + break; + } + } + + if ($wanted === ($this->receiverToken !== null)) { + return; + } + + if (!$wanted) { + $token = $this->receiverToken; + $this->receiverToken = null; + $this->channel->cancelReceiver((int) $token); + + return; + } + + // Registering re-checks readiness inside the substrate's own critical section, so a record + // that arrived in the meantime is reported instead of registered for — and then nobody has + // to be woken for it, because whoever is parked here can be served on the spot. + $token = $this->channel->registerReceiver(); + + if ($token === null) { + $this->recheck(); + + return; + } + + $this->receiverToken = $token; + + // On a rendezvous channel the registration IS the state change a sender waits for: no + // record was published and no room was freed, so nothing else would ever tell it. + $this->announce(WakeOpcode::Wake); } /** @@ -372,7 +591,7 @@ private function completeSelectRecv(SelectToken $token, int $caseIndex): bool * to be written by a process that is not this one, so no amount of local scheduling could * produce the wakeup and a deadlock report must not count this coroutine as stuck. */ - private function parkOn(bool $forSend, string $what): void + private function parkOn(bool $forSend, ?int $ticket, string $what): void { $scheduler = $this->arena->scheduler(); $coroutine = $scheduler->current() ?? throw new \LogicException( @@ -385,9 +604,16 @@ private function parkOn(bool $forSend, string $what): void 'token' => null, 'case' => -1, 'value' => null, + 'ticket' => $ticket, ]; $coroutine->park($what, true); + + // Registered before the suspend, and the registration re-checks readiness inside the + // channel's own critical section: a sibling that deposits after this point necessarily + // sees the entry, so the wakeup cannot be lost. + $this->syncRegistration(); + $scheduler->suspend(SuspendCommand::BLOCKED); } diff --git a/tests/Functional/testACancelledContextWithdrawsItsRendezvousRegistration.phpt b/tests/Functional/testACancelledContextWithdrawsItsRendezvousRegistration.phpt new file mode 100644 index 0000000..6a987e5 --- /dev/null +++ b/tests/Functional/testACancelledContextWithdrawsItsRendezvousRegistration.phpt @@ -0,0 +1,65 @@ +--TEST-- +A rendezvous select abandoned by a cancelled context leaves no registration behind +--INI-- +ffi.enable=1 +opcache.jit=off +error_reporting=E_ALL & ~E_DEPRECATED +--FILE-- +declareShared('handoff', SharedChannel::class, 0); + +$probe = new ProbeRendezvousPartnerTask('handoff'); +$runtime->publishTask($probe); + +$runtime->run(static function (TaskRuntime $self) use ($probe): void { + Timer::after(20.0, static function (): void { + throw new RuntimeException('deadline: the cancellation never resolved the select'); + }); + + $scheduler = $self->scheduler(); + $shared = $self->shared('handoff'); + $context = Context::withCancel($scheduler); + + Coroutine::spawn(static function () use ($context): void { + Coroutine::sleep(0.05); + $context->cancel(); + }); + + $outcome = Select::on($scheduler) + ->recv($shared, static fn(mixed $value): string => 'handoff: ' . (string) $value) + ->recv($context->done(), static fn(): string => 'cancelled') + ->run(); + + echo $outcome, PHP_EOL; + echo 'a worker sees: ', $self->spawnParallel($probe)->await(), PHP_EOL; +}); + +echo 'children left: ', parallelChildrenLeft(), PHP_EOL; +?> +--EXPECT-- +cancelled +a worker sees: nobody is waiting +children left: none diff --git a/tests/Functional/testACapacityZeroHandoffCostsABoundedNumberOfWakeupsOnBothSides.phpt b/tests/Functional/testACapacityZeroHandoffCostsABoundedNumberOfWakeupsOnBothSides.phpt new file mode 100644 index 0000000..f67616f --- /dev/null +++ b/tests/Functional/testACapacityZeroHandoffCostsABoundedNumberOfWakeupsOnBothSides.phpt @@ -0,0 +1,79 @@ +--TEST-- +A run of cross-process rendezvous handoffs costs a bounded number of wakeups in both processes +--INI-- +ffi.enable=1 +opcache.jit=off +error_reporting=E_ALL & ~E_DEPRECATED +--FILE-- +declareShared('handoff', SharedChannel::class, 0); + +$sender = new RendezvousSendTask('handoff', HANDOFFS, 'h', 0.0, BOUND); +$runtime->publishTask($sender); + +$runtime->run(static function (TaskRuntime $self) use ($sender): void { + Timer::after(20.0, static function (): void { + throw new RuntimeException('deadline: the handoffs never completed'); + }); + + $channel = $self->shared('handoff'); + $handle = $self->spawnParallel($sender); + + $received = []; + + while (count($received) < HANDOFFS) { + [$value, $ok] = $channel->recvOk(); + + if (!$ok) { + break; + } + + $received[] = $value; + } + + echo 'received: ', implode(' ', $received), PHP_EOL; + echo 'the worker reported: ', $handle->await(), PHP_EOL; + + $wakeups = $self->arena()?->wakeups() ?? PHP_INT_MAX; + + echo 'this process woke a bounded number of times: ', + $wakeups <= BOUND ? 'yes' : 'NO (' . $wakeups . ')', + PHP_EOL; +}); + +echo 'children left: ', parallelChildrenLeft(), PHP_EOL; +?> +--EXPECT-- +received: h0 h1 h2 h3 h4 h5 h6 h7 +the worker reported: sent 8, handoff waited for a receiver, wakeups bounded +this process woke a bounded number of times: yes +children left: none diff --git a/tests/Functional/testACapacityZeroSharedChannelComposesInASelectWithALocalChannel.phpt b/tests/Functional/testACapacityZeroSharedChannelComposesInASelectWithALocalChannel.phpt new file mode 100644 index 0000000..c26d673 --- /dev/null +++ b/tests/Functional/testACapacityZeroSharedChannelComposesInASelectWithALocalChannel.phpt @@ -0,0 +1,80 @@ +--TEST-- +One select resolves a capacity-0 shared channel and a local channel, and unlinks the losers +--INI-- +ffi.enable=1 +opcache.jit=off +error_reporting=E_ALL & ~E_DEPRECATED +--FILE-- +declareShared('handoff', SharedChannel::class, 0); + +$sender = new RendezvousSendTask('handoff', 3, 'shared-'); +$runtime->publishTask($sender); + +$runtime->run(static function (TaskRuntime $self) use ($sender): void { + Timer::after(20.0, static function (): void { + throw new RuntimeException('deadline: the select never resolved four times'); + }); + + $shared = $self->shared('handoff'); + $local = new Channel($self->scheduler(), 1); + + $self->spawnParallel($sender); + + Coroutine::spawn(static function () use ($local): void { + Coroutine::sleep(0.3); + $local->send('local'); + }); + + $seen = []; + + for ($round = 0; $round < 4; ++$round) { + Select::on($self->scheduler()) + ->recv($shared, static function (mixed $value) use (&$seen): void { + $seen[] = 'shared:' . $value; + }) + ->recv($local, static function (mixed $value) use (&$seen): void { + $seen[] = 'local:' . $value; + }) + ->run(); + } + + sort($seen); + + echo implode(PHP_EOL, $seen), PHP_EOL; + echo 'the losing case left no registration: ', $shared->hasWaitingReceiver() ? 'NO' : 'yes', PHP_EOL; +}); + +echo 'children left: ', parallelChildrenLeft(), PHP_EOL; +?> +--EXPECT-- +local:local +shared:shared-0 +shared:shared-1 +shared:shared-2 +the losing case left no registration: yes +children left: none diff --git a/tests/Functional/testACapacityZeroSharedChannelIsAccepted.phpt b/tests/Functional/testACapacityZeroSharedChannelIsAccepted.phpt new file mode 100644 index 0000000..2b2cf7e --- /dev/null +++ b/tests/Functional/testACapacityZeroSharedChannelIsAccepted.phpt @@ -0,0 +1,51 @@ +--TEST-- +A shared channel can be declared with capacity 0, and reports itself as a rendezvous +--INI-- +ffi.enable=1 +opcache.jit=off +error_reporting=E_ALL & ~E_DEPRECATED +--FILE-- +declareShared('handoff', SharedChannel::class, 0); + +// A negative capacity is still nonsense, and still says so. +try { + $runtime->declareShared('broken', SharedChannel::class, -1); +} catch (InvalidArgumentException $refusal) { + echo $refusal->getMessage(), PHP_EOL; +} + +$runtime->run(static function (TaskRuntime $self): void { + $channel = $self->shared('handoff'); + + echo 'capacity: ', $channel->capacity(), PHP_EOL; + echo 'rendezvous: ', $channel->isRendezvous() ? 'yes' : 'no', PHP_EOL; + echo 'buffered right now: ', $channel->count(), PHP_EOL; +}); + +echo 'children left: ', parallelChildrenLeft(), PHP_EOL; +?> +--EXPECT-- +shared channel "broken" cannot have a negative capacity, got -1; 0 is a cross-process rendezvous and a positive number buffers that many records +capacity: 0 +rendezvous: yes +buffered right now: 0 +children left: none diff --git a/tests/Functional/testARendezvousReceiverParksUntilASenderArrives.phpt b/tests/Functional/testARendezvousReceiverParksUntilASenderArrives.phpt new file mode 100644 index 0000000..a95d004 --- /dev/null +++ b/tests/Functional/testARendezvousReceiverParksUntilASenderArrives.phpt @@ -0,0 +1,62 @@ +--TEST-- +A receive on a capacity-0 shared channel parks on the poller until another process sends +--INI-- +ffi.enable=1 +opcache.jit=off +error_reporting=E_ALL & ~E_DEPRECATED +--FILE-- +declareShared('handoff', SharedChannel::class, 0); + +$sender = new RendezvousSendTask('handoff', 1, 'late-', 0.0, 64, DELAY); +$runtime->publishTask($sender); + +$runtime->run(static function (TaskRuntime $self) use ($sender): void { + Timer::after(20.0, static function (): void { + throw new RuntimeException('deadline: the receive never woke'); + }); + + $channel = $self->shared('handoff'); + $handle = $self->spawnParallel($sender); + + $started = microtime(true); + $value = $channel->recv(); + $elapsed = microtime(true) - $started; + + echo 'received: ', $value, PHP_EOL; + echo 'the receive parked until the sender arrived: ', $elapsed >= FLOOR ? 'yes' : 'NO (' . round($elapsed, 3) . 's)', PHP_EOL; + echo 'the worker reported: ', $handle->await(), PHP_EOL; + echo 'no registration is left behind: ', $channel->hasWaitingReceiver() ? 'NO' : 'yes', PHP_EOL; +}); + +echo 'children left: ', parallelChildrenLeft(), PHP_EOL; +?> +--EXPECT-- +received: late-0 +the receive parked until the sender arrived: yes +the worker reported: sent 1, handoff waited for a receiver, wakeups bounded +no registration is left behind: yes +children left: none diff --git a/tests/Functional/testARendezvousSendWaitsUntilAnotherProcessTakesTheValue.phpt b/tests/Functional/testARendezvousSendWaitsUntilAnotherProcessTakesTheValue.phpt new file mode 100644 index 0000000..60a81f9 --- /dev/null +++ b/tests/Functional/testARendezvousSendWaitsUntilAnotherProcessTakesTheValue.phpt @@ -0,0 +1,71 @@ +--TEST-- +A send on a capacity-0 shared channel returns only once another process has taken the value +--INI-- +ffi.enable=1 +opcache.jit=off +error_reporting=E_ALL & ~E_DEPRECATED +--FILE-- +declareShared('handoff', SharedChannel::class, 0); + +$receiver = new SlowTakeRendezvousTask('handoff', BUSY); +$runtime->publishTask($receiver); + +$runtime->run(static function (TaskRuntime $self) use ($receiver): void { + Timer::after(20.0, static function (): void { + throw new RuntimeException('deadline: the rendezvous never completed'); + }); + + $channel = $self->shared('handoff'); + $handle = $self->spawnParallel($receiver); + + // Test scaffolding, not runtime behaviour: wait until the partner exists so the measurement + // below covers the take alone and not the wait for somebody to hand the value to. + for ($round = 0; $round < 2_000 && !$channel->hasWaitingReceiver(); ++$round) { + Coroutine::sleep(0.005); + } + + echo 'a partner is registered before the send: ', $channel->hasWaitingReceiver() ? 'yes' : 'no', PHP_EOL; + + $started = microtime(true); + $channel->send('one handoff'); + $elapsed = microtime(true) - $started; + + echo 'the send waited for the take: ', $elapsed >= FLOOR ? 'yes' : 'NO (' . round($elapsed, 3) . 's)', PHP_EOL; + echo 'the worker took: ', $handle->await(), PHP_EOL; + echo 'nothing is left in the handoff slot: ', $channel->count() === 0 ? 'yes' : 'no', PHP_EOL; +}); + +echo 'children left: ', parallelChildrenLeft(), PHP_EOL; +?> +--EXPECT-- +a partner is registered before the send: yes +the send waited for the take: yes +the worker took: one handoff +nothing is left in the handoff slot: yes +children left: none diff --git a/tests/Functional/testARendezvousWaitIsNeverAFalseDeadlock.phpt b/tests/Functional/testARendezvousWaitIsNeverAFalseDeadlock.phpt new file mode 100644 index 0000000..b112fe2 --- /dev/null +++ b/tests/Functional/testARendezvousWaitIsNeverAFalseDeadlock.phpt @@ -0,0 +1,64 @@ +--TEST-- +A coroutine parked on a capacity-0 shared channel is not reported as a deadlock +--INI-- +ffi.enable=1 +opcache.jit=off +error_reporting=E_ALL & ~E_DEPRECATED +--FILE-- +declareShared('handoff', SharedChannel::class, 0); + +$sender = new RendezvousSendTask('handoff', 2, 'v', 0.0, 64, 0.3); +$runtime->publishTask($sender); + +$runtime->run(static function (TaskRuntime $self) use ($sender): void { + $shared = $self->shared('handoff'); + $group = new WaitGroup($self->scheduler()); + + $self->spawnParallel($sender); + + $group->add(1); + + Coroutine::spawn(static function () use ($shared, $group): void { + // Parked here with an empty run queue, an empty timer heap and nothing local that could + // ever produce this value: the definition of the state the detector reports on. + echo 'the rendezvous answered: ', $shared->recv(), PHP_EOL; + echo 'and again: ', $shared->recv(), PHP_EOL; + $group->done(); + }); + + $group->wait(); + + echo 'no deadlock was reported: yes', PHP_EOL; +}); + +echo 'children left: ', parallelChildrenLeft(), PHP_EOL; +?> +--EXPECT-- +the rendezvous answered: v0 +and again: v1 +no deadlock was reported: yes +children left: none diff --git a/tests/Functional/testASelectLoserLeavesNoRendezvousRegistrationForALaterSend.phpt b/tests/Functional/testASelectLoserLeavesNoRendezvousRegistrationForALaterSend.phpt new file mode 100644 index 0000000..5ff9c18 --- /dev/null +++ b/tests/Functional/testASelectLoserLeavesNoRendezvousRegistrationForALaterSend.phpt @@ -0,0 +1,71 @@ +--TEST-- +A capacity-0 shared channel that lost a select is no longer a rendezvous partner, seen from a worker +--INI-- +ffi.enable=1 +opcache.jit=off +error_reporting=E_ALL & ~E_DEPRECATED +--FILE-- +declareShared('handoff', SharedChannel::class, 0); + +$probe = new ProbeRendezvousPartnerTask('handoff'); +$runtime->publishTask($probe); + +$runtime->run(static function (TaskRuntime $self) use ($probe): void { + Timer::after(20.0, static function (): void { + throw new RuntimeException('deadline: the select never resolved'); + }); + + $shared = $self->shared('handoff'); + $local = new Channel($self->scheduler(), 1); + + Coroutine::spawn(static function () use ($local): void { + Coroutine::sleep(0.05); + $local->send('local wins'); + }); + + // Nothing is ready when this runs, so the select genuinely parks — and parking is what puts a + // registration into the arena on behalf of the shared case. + $outcome = Select::on($self->scheduler()) + ->recv($shared, static fn(mixed $value): string => 'shared: ' . (string) $value) + ->recv($local, static fn(mixed $value): string => 'local: ' . (string) $value) + ->run(); + + echo $outcome, PHP_EOL; + echo 'this process still lists a waiter: ', $shared->hasWaitingReceiver() ? 'yes' : 'no', PHP_EOL; + echo 'a worker sees: ', $self->spawnParallel($probe)->await(), PHP_EOL; +}); + +echo 'children left: ', parallelChildrenLeft(), PHP_EOL; +?> +--EXPECT-- +local: local wins +this process still lists a waiter: no +a worker sees: nobody is waiting +children left: none diff --git a/tests/Functional/testASelectSendCaseOnACapacityZeroSharedChannelIsRefused.phpt b/tests/Functional/testASelectSendCaseOnACapacityZeroSharedChannelIsRefused.phpt new file mode 100644 index 0000000..f052f53 --- /dev/null +++ b/tests/Functional/testASelectSendCaseOnACapacityZeroSharedChannelIsRefused.phpt @@ -0,0 +1,64 @@ +--TEST-- +A capacity-0 shared channel cannot be a select send case, and the refusal names the remedies +--INI-- +ffi.enable=1 +opcache.jit=off +error_reporting=E_ALL & ~E_DEPRECATED +--FILE-- +declareShared('handoff', SharedChannel::class, 0); +$runtime->declareShared('buffered', SharedChannel::class, 1); + +$runtime->run(static function (TaskRuntime $self): void { + $rendezvous = $self->shared('handoff'); + $buffered = $self->shared('buffered'); + $local = new Channel($self->scheduler(), 1); + + try { + Select::on($self->scheduler()) + ->send($rendezvous, 'nowhere to commit', static fn(): string => 'sent') + ->recv($local, static fn(mixed $value): string => 'local') + ->run(); + } catch (LogicException $refusal) { + echo preg_replace('/@0x[0-9A-F]+/', '@ADDRESS', $refusal->getMessage()), PHP_EOL; + } + + // The same statement over a buffered shared channel is business as usual. + echo Select::on($self->scheduler()) + ->send($buffered, 'room for this one', static fn(): string => 'the buffered case sent') + ->recv($local, static fn(mixed $value): string => 'local') + ->run(), PHP_EOL; + + echo 'buffered now holds: ', $buffered->count(), PHP_EOL; +}); + +echo 'children left: ', parallelChildrenLeft(), PHP_EOL; +?> +--EXPECT-- +a capacity-0 shared channel cannot be a select send case: a rendezvous send completes when the value is TAKEN, which a case cannot wait for without parking. Declare the channel with a capacity of at least 1, or drive the handoff from a coroutine of its own that calls send() on shared channel @ADDRESS +the buffered case sent +buffered now holds: 1 +children left: none diff --git a/tests/Functional/testDeclaringASharedRootAfterTheForkIsRefused.phpt b/tests/Functional/testDeclaringASharedRootAfterTheForkIsRefused.phpt index 84bc5cc..84f89c0 100644 --- a/tests/Functional/testDeclaringASharedRootAfterTheForkIsRefused.phpt +++ b/tests/Functional/testDeclaringASharedRootAfterTheForkIsRefused.phpt @@ -9,7 +9,6 @@ error_reporting=E_ALL & ~E_DEPRECATED declare(strict_types=1); -use Lisachenko\NativePhpCoroutines\Parallel\SharedChannel; use Lisachenko\NativePhpCoroutines\Runtime; use Lisachenko\NativePhpCoroutines\TaskRuntime; use Lisachenko\NativePhpCoroutines\Tests\Support\SharedCounter; @@ -27,15 +26,6 @@ $runtime->declareShared('counter', SharedCounter::class); echo 'declared before the fork: ', $runtime->arena()?->hasRoot('counter') === true ? 'yes' : 'no', PHP_EOL; -// A capacity-0 shared channel is refused at declaration rather than delivered as one that usually -// fails to hand anything over — the cross-process rendezvous handshake counts receivers parked -// inside the substrate's own blocking recv(), which this runtime deliberately never calls. -try { - $runtime->declareShared('rendezvous', SharedChannel::class, 0); -} catch (InvalidArgumentException $refusal) { - echo $refusal->getMessage(), PHP_EOL; -} - $runtime->run(static function (TaskRuntime $self): void { // After the fork the workers already exist, so a root created now lives in this process alone. // That is not a late binding, and it is refused rather than silently made useless. @@ -56,7 +46,6 @@ echo 'children left: ', parallelChildrenLeft(), PHP_EOL; ?> --EXPECT-- declared before the fork: yes -shared channel "rendezvous" needs a capacity of at least 1: a cross-process rendezvous accepts a send only while a sibling is parked inside the substrate's own blocking recv(), and this runtime parks Fibers on its poller instead shared root "late" cannot be declared after the workers have forked: a root is inherited by address, so one created now exists only in this process. Declare every root before run() forks the pool closure "late" cannot be shared: the fork barrier has already been passed, and only a closure registered before it exists at the same address in every worker children left: none diff --git a/tests/Functional/testTwoCoroutinesOfOneProcessRendezvousOnASharedChannel.phpt b/tests/Functional/testTwoCoroutinesOfOneProcessRendezvousOnASharedChannel.phpt new file mode 100644 index 0000000..856329f --- /dev/null +++ b/tests/Functional/testTwoCoroutinesOfOneProcessRendezvousOnASharedChannel.phpt @@ -0,0 +1,77 @@ +--TEST-- +A capacity-0 shared channel hands values between two coroutines of the same process +--INI-- +ffi.enable=1 +opcache.jit=off +error_reporting=E_ALL & ~E_DEPRECATED +--FILE-- +declareShared('handoff', SharedChannel::class, 0); + +$runtime->run(static function (TaskRuntime $self): void { + Timer::after(20.0, static function (): void { + throw new RuntimeException('deadline: the same-process rendezvous never completed'); + }); + + $channel = $self->shared('handoff'); + $group = new WaitGroup($self->scheduler()); + $taken = []; + + $group->add(2); + + // The receiver goes first and parks, so the sender finds a partner already registered. + Coroutine::spawn(static function () use ($channel, $group, &$taken): void { + for ($round = 0; $round < 3; ++$round) { + $taken[] = $channel->recv(); + } + + $group->done(); + }); + + // The sender goes second and has to park on the handoff of each value being taken. + Coroutine::spawn(static function () use ($channel, $group): void { + for ($round = 0; $round < 3; ++$round) { + $channel->send('v' . $round); + } + + $group->done(); + }); + + $group->wait(); + + echo 'taken: ', implode(' ', $taken), PHP_EOL; + echo 'the handoff slot is empty: ', $channel->count() === 0 ? 'yes' : 'no', PHP_EOL; + echo 'no registration is left behind: ', $channel->hasWaitingReceiver() ? 'NO' : 'yes', PHP_EOL; +}); + +echo 'children left: ', parallelChildrenLeft(), PHP_EOL; +?> +--EXPECT-- +taken: v0 v1 v2 +the handoff slot is empty: yes +no registration is left behind: yes +children left: none diff --git a/tests/Support/shared.php b/tests/Support/shared.php index 0631d12..e6d0e35 100644 --- a/tests/Support/shared.php +++ b/tests/Support/shared.php @@ -236,6 +236,136 @@ public function run(TaskRuntime $runtime): mixed } } +/** + * Hands values over a **capacity-0** shared channel, one rendezvous at a time. + * + * Everything a rendezvous claim needs is measured here rather than in the parent, because the + * sender is the side that has to wait: `send()` on a capacity-0 channel deposits the record and + * then parks a second time until somebody has actually taken it, so the elapsed time is the proof + * that the handshake completed rather than merely started. The worker's own poller wakeup count + * comes back with it — a sender that regressed into a poll loop would blow the bound long before + * the values stopped arriving. + * + * The answer is a formatted string on purpose: it travels as an arena `zend_string`, and a test + * asserting on it fails with the offending number in the diff instead of a bare `false`. + */ +final class RendezvousSendTask implements Task +{ + public function __construct( + private readonly string $root, + private readonly int $count, + private readonly string $prefix = 'r', + private readonly float $floorSeconds = 0.0, + private readonly int $wakeupBound = 64, + private readonly float $delay = 0.0, + ) {} + + public function run(TaskRuntime $runtime): mixed + { + $channel = $runtime->shared($this->root); + + if (!$channel instanceof \Lisachenko\NativePhpCoroutines\ChannelInterface) { + throw new \LogicException('the shared channel root is not available in this worker'); + } + + if ($this->delay > 0.0) { + Coroutine::sleep($this->delay); + } + + $started = microtime(true); + + for ($index = 0; $index < $this->count; ++$index) { + $channel->send($this->prefix . $index); + } + + $elapsed = microtime(true) - $started; + + // A deliberate downcast: the wakeup counter is diagnostics, which the task surface + // intentionally does not carry. Real tasks never need the concrete runtime. + $arena = $runtime instanceof \Lisachenko\NativePhpCoroutines\Runtime ? $runtime->arena() : null; + $wakeups = $arena?->wakeups() ?? -1; + + return sprintf( + 'sent %d, handoff %s, wakeups %s', + $this->count, + $elapsed >= $this->floorSeconds ? 'waited for a receiver' : 'returned early', + $wakeups >= 0 && $wakeups <= $this->wakeupBound ? 'bounded' : 'UNBOUNDED (' . $wakeups . ')', + ); + } +} + +/** + * Registers as a rendezvous receiver at once, then refuses to run for a while before taking. + * + * The gap between the two is the whole point. A registered partner makes the sender's *deposit* + * possible immediately; holding this worker's only thread in a call-free loop makes the *take* + * impossible until the loop ends. A sender that treated the deposit as the end of the handshake + * would return during that window, and the measurement on the other side would show it. + */ +final class SlowTakeRendezvousTask implements Task +{ + public function __construct( + private readonly string $root, + private readonly float $busySeconds = 0.4, + ) {} + + public function run(TaskRuntime $runtime): mixed + { + $channel = $runtime->shared($this->root); + + if (!$channel instanceof \Lisachenko\NativePhpCoroutines\ChannelInterface) { + throw new \LogicException('the shared channel root is not available in this worker'); + } + + $taken = new \stdClass(); + $taken->value = null; + + Coroutine::spawn(static function () use ($channel, $taken): void { + $taken->value = $channel->recv(); + }); + + // One yield is enough to let the receiver reach its park, which is where it registers. + Coroutine::yield(); + + // A call-free loop owns this worker cooperatively: the receiver above cannot run, the + // poller cannot run, and the value can therefore be deposited but not taken. + $end = microtime(true) + $this->busySeconds; + + while (microtime(true) < $end) { + // deliberately empty + } + + for ($round = 0; $round < 500 && $taken->value === null; ++$round) { + Coroutine::sleep(0.01); + } + + return $taken->value ?? 'nothing was taken'; + } +} + +/** + * Asks, from another process, whether anybody is registered to take a handoff right now. + * + * This is what makes "a select loser leaves no stale registration" a cross-process claim instead of + * a local bookkeeping check: the registration lives in the arena, and the process that would act on + * it is not the process that made it. + */ +final class ProbeRendezvousPartnerTask implements Task +{ + public function __construct(private readonly string $root) {} + + public function run(TaskRuntime $runtime): mixed + { + $channel = $runtime->shared($this->root); + + if (!$channel instanceof \Lisachenko\NativePhpCoroutines\Parallel\SharedChannel) { + throw new \LogicException('the shared channel root is not available in this worker'); + } + + return $channel->hasWaitingReceiver() ? 'a partner is present' : 'nobody is waiting'; + } +} + /** * Awaits a result slot the *parent* opened, from inside a worker. * From 58cf7c39c82b3361aafd7e7d808ea0b67c437d6c Mon Sep 17 00:00:00 2001 From: Claude Fable 5 Date: Sun, 16 Aug 2026 10:18:26 +0000 Subject: [PATCH 2/3] test(tests): overlap a recycled result slot with a parked rendezvous ticket Slot recycling and a rendezvous handoff are disjoint by construction - separate arena allocations, separate mutexes, bump allocation that never reuses an address - so the only thing a released slot and a parked ticket share is the one wake socket this process owns. Both wakeups arrive as the same content-free poke and each waiter re-reads its own predicate, which means a slot event can wake a rendezvous sender spuriously and costs it a re-check of `head > ticket` and nothing more. That was an argument. This makes it observable: a sender parks on its ticket while a join handle takes a result slot's answer and hands the slot back with its generation bumped and its waiter words wiped. If a recycled slot could ever disturb a ticket wait, the send would return early or not at all. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01U5QRWPfmkZME7hDjU6i5EA --- ...endezvousSendIsStillParkedOnItsTicket.phpt | 95 +++++++++++++++++++ 1 file changed, 95 insertions(+) create mode 100644 tests/Functional/testASlotIsRecycledWhileARendezvousSendIsStillParkedOnItsTicket.phpt diff --git a/tests/Functional/testASlotIsRecycledWhileARendezvousSendIsStillParkedOnItsTicket.phpt b/tests/Functional/testASlotIsRecycledWhileARendezvousSendIsStillParkedOnItsTicket.phpt new file mode 100644 index 0000000..c0b96da --- /dev/null +++ b/tests/Functional/testASlotIsRecycledWhileARendezvousSendIsStillParkedOnItsTicket.phpt @@ -0,0 +1,95 @@ +--TEST-- +A result slot is claimed and given back while a rendezvous send waits for its value to be taken +--INI-- +ffi.enable=1 +opcache.jit=off +error_reporting=E_ALL & ~E_DEPRECATED +--FILE-- + ticket` and nothing +// else. This test is that argument made observable — if a recycled slot ever disturbed a ticket +// wait, the send would either return early or never return at all. +$runtime = new Runtime(workers: 2, arenaSize: 32 << 20, slots: 8); +$runtime->declareShared('handoff', SharedChannel::class, 0); + +$receiver = new SlowTakeRendezvousTask('handoff', 0.4); +$sleeper = new SleepingTask(0.15, 7); +$runtime->publishTask($receiver); +$runtime->publishTask($sleeper); + +$runtime->run(static function (TaskRuntime $self) use ($receiver, $sleeper): void { + Timer::after(20.0, static function (): void { + throw new RuntimeException('deadline: the overlapped waits never both completed'); + }); + + $channel = $self->shared('handoff'); + $group = new WaitGroup($self->scheduler()); + $order = []; + + $taker = $self->spawnParallel($receiver, 0); + + $group->add(2); + + Coroutine::spawn(static function () use ($channel, $group, &$order): void { + // Parks first for a partner, then a second time on the ticket, and the slot below is + // settled and released squarely inside that second park. + $channel->send('handed over'); + $order[] = 'the rendezvous completed'; + $group->done(); + }); + + Coroutine::spawn(static function () use ($self, $sleeper, $group, &$order): void { + $order[] = 'the join handle answered: ' . $self->spawnParallel($sleeper, 1)->await(); + $group->done(); + }); + + $group->wait(); + + $order[] = 'the receiver took: ' . $taker->await(); + + // Every slot settled and came back; the sleeper's did so while the ticket wait was still + // outstanding, which is the interleaving under test. + $table = $self->arena()?->slotTable(); + + echo implode(PHP_EOL, $order), PHP_EOL; + echo 'slots still out: ', $table?->outstanding() ?? -1, PHP_EOL; + echo 'the handoff slot is empty: ', $channel->count() === 0 ? 'yes' : 'no', PHP_EOL; + echo 'no registration is left behind: ', $channel->hasWaitingReceiver() ? 'NO' : 'yes', PHP_EOL; +}); + +echo 'children left: ', parallelChildrenLeft(), PHP_EOL; +?> +--EXPECT-- +the join handle answered: 7 +the rendezvous completed +the receiver took: handed over +slots still out: 0 +the handoff slot is empty: yes +no registration is left behind: yes +children left: none From 6e57cd830f32ef47108e4ddb8764d7260cc6fc0a Mon Sep 17 00:00:00 2001 From: Alexander Lisachenko <640114+lisachenko@users.noreply.github.com> Date: Sun, 16 Aug 2026 10:23:18 +0000 Subject: [PATCH 3/3] ci: rerun against the substrate's merged rendezvous registration The previous run installed the substrate from main before lisachenko/php-shared-data-extension#26 landed there, so the registration and ticket surface this branch consumes did not exist yet. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01U5QRWPfmkZME7hDjU6i5EA