diff --git a/AGENTS.md b/AGENTS.md index b48a175..67fdf4d 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -245,9 +245,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 da5d665..dab6052 100644 --- a/README.md +++ b/README.md @@ -552,10 +552,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/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 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. *