From 74fb1120256eecdeb3ecd875a8025760521f55c0 Mon Sep 17 00:00:00 2001 From: Claude Fable 5 Date: Sun, 16 Aug 2026 09:13:13 +0000 Subject: [PATCH 1/3] feat(ipc): let a receiver parked elsewhere be a rendezvous partner A capacity-0 SharedChannel accepts a value only while a receiver is waiting, and until now the only way to be one was to be inside the blocking recv() spin loop. A consumer with its own scheduler never calls it - its whole invariant is that a worker blocks in exactly one place - so a cross-process rendezvous was unusable from a coroutine runtime rather than merely inconvenient (native-php-coroutines#14). registerReceiver()/cancelReceiver(), and their sender mirrors, make the registration a named operation: announce the interest, park the Fiber on notificationStream(), withdraw on unpark. trySendTicket() and isTicketTaken() expose the two halves a rendezvous is made of, since the deposit and the take are separate events and only the second completes the handshake. A registration is a claim about presence, never about storage. The record still goes into the single ring slot max($capacity, 1) already allocates, so there is no per-registration cell to keep consistent: cancellation can always succeed because it never has a value in its hands, and a record deposited against a registration withdrawn a moment later waits in the ring for the next receiver while its sender stays parked. The whole handshake stays under the channel's own mutex, so the happens-before edge is the release/acquire pair it always was. Registrations can outlive their process, unlike a waiter that parks inside a call and releases its own entry on the way out. Each waiter word now packs `owner pid << 32 | wake slot + 1` - one aligned word, because two words are a 16-byte record and those tear - and a rendezvous deposit reaps the entries whose owner is gone before it reads the gate. Liveness is posix_kill(pid, 0) plus the wake registry still naming that pid as the slot's owner, since a dead owner's slot is recycled to the next claimer. Also fixes a counter leak found next door: parkForRoom()/parkForRecord() incremented the parked count even when the waiter table was full, and unpark() has nothing to give back for an entry that was never taken, so the count stayed permanently high - which on a rendezvous channel is exactly the gate a handoff passes. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01U5QRWPfmkZME7hDjU6i5EA --- README.md | 17 ++ docs/shared-memory-model.md | 17 ++ src/Ipc/IpcException.php | 12 + src/Ipc/SharedChannel.php | 327 +++++++++++++++++++++++++++- src/Ipc/WaiterTable.php | 88 +++++++- src/Ipc/WakeRegistry.php | 44 ++++ tests/Ipc/SharedChannelForkTest.php | 284 ++++++++++++++++++++++++ 7 files changed, 773 insertions(+), 16 deletions(-) diff --git a/README.md b/README.md index 0baa1fd..b7b651f 100644 --- a/README.md +++ b/README.md @@ -372,6 +372,23 @@ a package with no scheduler can offer: every primitive also exposes its non-bloc (`trySend()`/`tryRecv()`/`tryLock()`/`readSlot()`) plus `notificationStream()`, so a coroutine runtime can park a Fiber in its own event loop instead. +A capacity-0 channel needs one thing more, because its gate is "is a receiver waiting" and a +consumer with its own scheduler is never inside `recv()`: + +```php +$token = $channel->registerReceiver(); // null => a record is already there, take it now +// ... park the Fiber on notificationStream() in the consumer's own event loop ... +$channel->cancelReceiver($token); // on unpark, whatever woke it + +$ticket = $channel->trySendTicket($value); // deposits only while a receiver is waiting +$done = $channel->isTicketTaken($ticket); // the handshake completes when it is TAKEN +``` + +The registration is a claim about presence, never about storage — the record goes into the +one ring slot a capacity-0 channel allocates — so `cancelReceiver()` never has a value in its +hands and can always succeed. A registration can outlive its process, so each waiter entry +carries its owner pid and a rendezvous deposit reaps the dead ones before it reads the gate. + ### Shared closures (registered before the fork) A closure compiled **before the fork** is valid in every worker: the family inherited the diff --git a/docs/shared-memory-model.md b/docs/shared-memory-model.md index d4173b7..4e6df17 100644 --- a/docs/shared-memory-model.md +++ b/docs/shared-memory-model.md @@ -263,6 +263,23 @@ sentence. dedicated mutex (a structure locked on every operation does not belong on a shared stripe). Head and tail are monotonic counters, so fill level is a subtraction; capacity 0 is a true cross-process rendezvous; `close()` crosses processes; + - a rendezvous accepts a value only while a receiver is waiting, and a consumer with its own + scheduler is never inside `recv()` — so `registerReceiver()`/`cancelReceiver()` (and their + sender mirrors) let a receiver parked in someone else's event loop count as the partner. + The registration is a claim about **presence, never about storage**: the record still goes + into the single ring slot a capacity-0 channel allocates, so a cancellation can always + succeed — it never has a value in its hands — and a record deposited against a + registration that is withdrawn a moment later simply waits in the ring for the next + receiver while its sender stays parked. The whole handshake (register, re-check, deposit, + cancel) happens under the channel's own mutex, so the happens-before edge is the same + release/acquire pair it always was; + - a registration outlives the call that made it, and can therefore outlive its process. Each + waiter entry packs `owner pid << 32 | wake slot + 1` into one aligned word (two words would + be a 16-byte record, and those tear — §5), and a rendezvous deposit reaps the entries whose + owner is gone before it reads the gate, so a dead worker cannot go on standing in for a + partner. Liveness is `posix_kill(pid, 0)` **plus** the wake registry still naming that pid + as the slot's owner, because a dead owner's slot is recycled to the next process that + claims one; - `SharedArray` — fixed-capacity vector of records, per-instance stripe: the container a `zend_array` cannot be (§4); - `ResultSlotTable` — futures. A slot settles exactly once, carrying either a value record or diff --git a/src/Ipc/IpcException.php b/src/Ipc/IpcException.php index d50b56c..43332db 100644 --- a/src/Ipc/IpcException.php +++ b/src/Ipc/IpcException.php @@ -67,6 +67,18 @@ public static function slotAlreadyCompleted(int $id): self )); } + public static function waiterTableFull(string $role, int $capacity): self + { + return new self(sprintf( + 'The %s waiter table of this channel is full: all %d entries hold a registration. Waiter ' . + 'tables are pre-sized in the arena and never grow - a grown table would be reallocated ' . + 'into one process\'s private heap - so create the channel with a larger waiterCapacity ' . + 'before the workers fork.', + $role, + $capacity, + )); + } + public static function invalidCapacity(string $structure, int $capacity): self { return new self(sprintf('%s capacity must be a positive number of records, got %d', $structure, $capacity)); diff --git a/src/Ipc/SharedChannel.php b/src/Ipc/SharedChannel.php index 7ae3e80..8453f07 100644 --- a/src/Ipc/SharedChannel.php +++ b/src/Ipc/SharedChannel.php @@ -56,6 +56,34 @@ * the other half of the API: trySend()/tryRecv() plus notificationStream(), so it can park a * Fiber in its own event loop and never block the process. Both halves observe the same * waiter tables, so a Fiber-parked consumer and a spin-blocked one wake identically. + * + * ## Rendezvous with a receiver that is parked somewhere else + * + * The gate on a capacity-0 handoff is "is a receiver waiting", and until registerReceiver() + * existed the only way to be one was to be inside recv() - which a scheduler-driven consumer + * never calls, because its whole invariant is that a worker blocks in exactly one place. That + * made a rendezvous unusable from a coroutine runtime rather than merely inconvenient, so the + * registration is now a named operation of its own: + * + * ```php + * $token = $channel->registerReceiver(); // null => a record is already there + * // ... park the Fiber on notificationStream() in the consumer's own event loop ... + * $channel->cancelReceiver($token); // on unpark, whatever woke it + * ``` + * + * A registration is a claim about presence, never about storage: the handed-off record still + * goes into the one ring slot a capacity-0 channel allocates (`max($capacity, 1)`), so there + * is no per-registration cell to keep consistent and no way for a value to belong to a waiter + * that walked away. That is what makes cancellation total - cancelReceiver() can always + * succeed, because it never has a value in its hands. A record deposited against a + * registration that is cancelled a moment later simply stays in the ring for the next + * receiver, and the sender stays parked until somebody actually takes it, which is exactly + * the state it would have been in had it never deposited at all. + * + * Registrations outlive the call that made them, so unlike a waiter parked inside recv() they + * can outlive their process. Each entry records its owner pid; a rendezvous deposit reaps the + * ones whose owner is gone before it reads the gate (see reapDeadWaiters()), so a dead + * worker's registration cannot go on telling senders that a partner is present. */ final class SharedChannel { @@ -219,6 +247,26 @@ public function isClosed(): bool return $this->word(self::WORD_CLOSED) !== 0; } + /** + * Receivers currently waiting on this channel, parked inside recv() or registered + * + * On a rendezvous channel this is the gate a handoff passes: a non-zero count is what + * makes trySend() accept a value. It is a hint for everybody else - by the time a caller + * reads it, a waiter may have taken a record or cancelled. + */ + public function parkedReceivers(): int + { + return $this->word(self::WORD_RECEIVERS_PARKED); + } + + /** + * Senders currently waiting on this channel, parked inside send() or registered + */ + public function parkedSenders(): int + { + return $this->word(self::WORD_SENDERS_PARKED); + } + /** * The descriptor a scheduler parks on; drain it and re-poll tryRecv()/trySend() * @@ -240,21 +288,49 @@ public function wasLockRecovered(): bool /** * Sends without ever blocking; false means "no room right now" * - * On a rendezvous channel this succeeds only while a receiver is already parked, and it - * returns as soon as the record is deposited - the synchronous half of the handshake - * (waiting until the value is actually taken) is what send() adds on top. + * On a rendezvous channel this succeeds only while a receiver is waiting - one parked + * inside recv() or one that announced itself with registerReceiver() - and it returns as + * soon as the record is deposited. The synchronous half of the handshake (waiting until + * the value is actually taken) is what send() adds on top, and what a consumer with its + * own scheduler builds out of trySendTicket() and isTicketTaken(). */ public function trySend(mixed $value): bool + { + return $this->trySendTicket($value) !== null; + } + + /** + * trySend(), returning the ticket the record was deposited at instead of a flag + * + * The ticket is the monotonic position in the ring, and `head > ticket` is the whole + * definition of "this record has been taken" - see isTicketTaken(). A consumer that parks + * its own waiters needs exactly that: trySend() alone cannot express a rendezvous, because + * the deposit and the take are two events and only the second one completes the handshake. + * + * @return int|null Ticket of the deposited record, or null when it could not be deposited + */ + public function trySendTicket(mixed $value): ?int { [$tag, $payload] = $this->codec->encode($value); $ticket = $this->offer($tag, $payload, requireParkedReceiver: $this->isRendezvous()); if ($ticket === null) { - return false; + return null; } $this->wakeReceivers($tag, $payload); - return true; + return $ticket; + } + + /** + * Whether the record deposited at $ticket has been taken by a receiver + * + * A single aligned word read of a monotonic counter, so no lock: head only ever grows, and + * an 8-byte load never tears (EPIC #15, correction #2). + */ + public function isTicketTaken(int $ticket): bool + { + return $this->word(self::WORD_HEAD) > $ticket; } /** @@ -365,6 +441,174 @@ public function close(): void $this->wake->notifyAll($this->senders->occupants(), $event); } + /** + * Announces a receiver that is parked somewhere other than inside recv() + * + * This is the half of the rendezvous handshake a consumer with its own scheduler could not + * express before: it makes the channel count this process as a waiting receiver, so a + * sibling's trySend() on a capacity-0 channel has a partner to hand its value to, while + * the Fiber that will take the value sits in the consumer's own event loop on + * notificationStream(). + * + * The registration and the readiness re-check happen in ONE critical section, which is the + * only thing that makes the wakeup safe: a sender that deposits after this point + * necessarily sees this entry, and a record that arrived before it is reported here rather + * than waited for. A caller that gets null must NOT park - it retries tryRecv() at once. + * + * @return int|null Token for cancelReceiver(), or null when a record (or a close) is + * already there and nothing was registered + * + * @throws IpcException When every entry of the receivers table is taken + */ + public function registerReceiver(): ?int + { + // Claimed before the lock: slot() may take the wake registry's own mutex, and two + // arena locks held at once is a lock order nobody else in this package obeys + $wakeSlot = $this->wake->slot(); + $owner = (int) getmypid(); + + $recovered = $this->arena->lockMutexAt($this->mutex); + + $ready = $this->word(self::WORD_TAIL) > $this->word(self::WORD_HEAD) + || $this->word(self::WORD_CLOSED) !== 0; + $entry = null; + if (!$ready) { + $entry = $this->receivers->register($wakeSlot, $owner); + if ($entry !== null) { + $this->setWord(self::WORD_RECEIVERS_PARKED, $this->word(self::WORD_RECEIVERS_PARKED) + 1); + } + } + + $this->arena->unlockMutexAt($this->mutex); + + $this->recoveredLock = $this->recoveredLock || $recovered; + + if ($ready) { + return null; + } + if ($entry === null) { + throw IpcException::waiterTableFull('receivers', $this->waiterCapacity); + } + + // On a rendezvous channel the registration IS the state change a sender is waiting + // for - there is no room to free and no record to publish - so it has to be announced + // like any other, after the lock is gone and never while holding it + if ($this->isRendezvous()) { + $this->wakeSenders(); + } + + return $entry; + } + + /** + * Withdraws a registerReceiver() registration + * + * Always succeeds and never hands anything back, because a registration never owned a + * value: a record deposited against it is in the ring, where the next receiver takes it + * and the sender goes on waiting until one does. That is what lets a select loser or a + * cancelled context unwind without having to deliver a value it can no longer deliver. + */ + public function cancelReceiver(int $token): void + { + $this->unpark($this->receivers, $token, self::WORD_RECEIVERS_PARKED); + } + + /** + * Announces a sender that is parked somewhere other than inside send() + * + * The mirror of registerReceiver(), and the same one-critical-section rule: what "ready" + * means depends on what the sender is waiting for. + * + * @param int|null $ticket Ticket of a record this sender already deposited and is waiting + * to see taken (the second half of a rendezvous send); null when + * it is waiting for somewhere to put a value in the first place + * + * @return int|null Token for cancelSender(), or null when the sender can already proceed + * and nothing was registered + * + * @throws IpcException When every entry of the senders table is taken + */ + public function registerSender(?int $ticket = null): ?int + { + $wakeSlot = $this->wake->slot(); + $owner = (int) getmypid(); + $limit = max($this->capacity, 1); + + $recovered = $this->arena->lockMutexAt($this->mutex); + + $head = $this->word(self::WORD_HEAD); + $ready = $this->word(self::WORD_CLOSED) !== 0; + if (!$ready) { + $ready = $ticket !== null + ? $head > $ticket + : $this->word(self::WORD_TAIL) - $head < $limit + && (!$this->isRendezvous() || $this->word(self::WORD_RECEIVERS_PARKED) > 0); + } + $entry = null; + if (!$ready) { + $entry = $this->senders->register($wakeSlot, $owner); + if ($entry !== null) { + $this->setWord(self::WORD_SENDERS_PARKED, $this->word(self::WORD_SENDERS_PARKED) + 1); + } + } + + $this->arena->unlockMutexAt($this->mutex); + + $this->recoveredLock = $this->recoveredLock || $recovered; + + if ($ready) { + return null; + } + if ($entry === null) { + throw IpcException::waiterTableFull('senders', $this->waiterCapacity); + } + + return $entry; + } + + /** + * Withdraws a registerSender() registration + */ + public function cancelSender(int $token): void + { + $this->unpark($this->senders, $token, self::WORD_SENDERS_PARKED); + } + + /** + * Releases registrations whose owning process is gone, and reports how many + * + * A waiter parked inside recv() or send() takes its entry back on the way out, so it can + * never go stale; a registration made from a consumer's event loop can, and on a + * rendezvous channel a stale one would keep telling senders that a partner is present. A + * deposit reaps before it reads the gate, so this is normally invisible - it is public for + * a supervisor that wants to reclaim a dead worker's entries on its own schedule, and for + * the tests that prove the reaping happens at all. + * + * Cheap it is not: the survey asks the operating system whether each owner still exists. + * It runs entirely OUTSIDE the lock, and only what it found is re-verified and released + * inside one. + */ + public function reapDeadWaiters(): int + { + $deadReceivers = $this->surveyDead($this->receivers); + $deadSenders = $this->surveyDead($this->senders); + + if ($deadReceivers === [] && $deadSenders === []) { + return 0; + } + + $recovered = $this->arena->lockMutexAt($this->mutex); + + $reaped = $this->releaseSurveyed($this->receivers, $deadReceivers, self::WORD_RECEIVERS_PARKED) + + $this->releaseSurveyed($this->senders, $deadSenders, self::WORD_SENDERS_PARKED); + + $this->arena->unlockMutexAt($this->mutex); + + $this->recoveredLock = $this->recoveredLock || $recovered; + + return $reaped; + } + /** * Writes one record into the ring if it fits, and returns the ticket it was written at * @@ -377,8 +621,19 @@ private function offer(ValueTag $tag, int $payload, bool $requireParkedReceiver) { $limit = max($this->capacity, 1); + // The gate below trusts the parked count, and a registration can outlive the process + // that made it - so on a handoff the entries whose owner is gone are surveyed here, + // outside the lock where the syscalls belong, and released inside the very critical + // section that then reads the count. Without that a dead worker's leftover entry would + // make every later send believe a partner is present + $dead = $requireParkedReceiver && $this->word(self::WORD_RECEIVERS_PARKED) > 0 + ? $this->surveyDead($this->receivers) + : []; + $recovered = $this->arena->lockMutexAt($this->mutex); + $this->releaseSurveyed($this->receivers, $dead, self::WORD_RECEIVERS_PARKED); + $closed = $this->word(self::WORD_CLOSED); $head = $this->word(self::WORD_HEAD); $tail = $this->word(self::WORD_TAIL); @@ -422,7 +677,13 @@ private function parkForRoom(?float $deadline): bool // Registered and re-checked in ONE critical section: a receiver that frees a slot // after this point necessarily sees this entry, so the wakeup cannot be lost $entry = $this->senders->register($wakeSlot); - $this->setWord(self::WORD_SENDERS_PARKED, $this->word(self::WORD_SENDERS_PARKED) + 1); + if ($entry !== null) { + // Counted only when an entry was actually taken: unpark() has nothing to give + // back for a full table, so counting a failed registration would leave the + // parked count permanently too high - and on a rendezvous channel that count + // is the gate a handoff passes + $this->setWord(self::WORD_SENDERS_PARKED, $this->word(self::WORD_SENDERS_PARKED) + 1); + } } $this->arena->unlockMutexAt($this->mutex); @@ -453,7 +714,11 @@ private function parkForRecord(?float $deadline): bool $entry = null; if (!$ready) { $entry = $this->receivers->register($wakeSlot); - $this->setWord(self::WORD_RECEIVERS_PARKED, $this->word(self::WORD_RECEIVERS_PARKED) + 1); + if ($entry !== null) { + // See parkForRoom(): a registration that found no free entry must not be + // counted, or the count never comes back down + $this->setWord(self::WORD_RECEIVERS_PARKED, $this->word(self::WORD_RECEIVERS_PARKED) + 1); + } } $this->arena->unlockMutexAt($this->mutex); @@ -487,6 +752,54 @@ private function awaitTaken(int $ticket, ?float $deadline): bool return true; } + /** + * Entries of $table whose owning process is gone, read without the lock + * + * Makes one liveness syscall per occupied entry, which is why no caller runs it inside a + * critical section. The raw word travels with the entry index so the release side can tell + * "still the registration I surveyed" from "released and re-taken since". + * + * @return list + */ + private function surveyDead(WaiterTable $table): array + { + $dead = []; + foreach ($table->entries() as $occupant) { + if (!$this->wake->isOwnerAlive($occupant['slot'], $occupant['pid'])) { + $dead[] = ['entry' => $occupant['entry'], 'word' => $occupant['word']]; + } + } + + return $dead; + } + + /** + * Releases what surveyDead() found; the caller holds this channel's lock + * + * @param list $dead + * + * @return int Entries actually released + */ + private function releaseSurveyed(WaiterTable $table, array $dead, int $counterWord): int + { + $released = 0; + foreach ($dead as $stale) { + if ($table->wordAt($stale['entry']) !== $stale['word']) { + // Released and re-taken between the survey and the lock: the entry now belongs + // to somebody alive, and reclaiming it would unregister a live waiter + continue; + } + $table->release($stale['entry']); + $released++; + } + + if ($released > 0) { + $this->setWord($counterWord, max($this->word($counterWord) - $released, 0)); + } + + return $released; + } + /** * Deregisters a waiter entry, under the lock, and drops the parked counter with it */ diff --git a/src/Ipc/WaiterTable.php b/src/Ipc/WaiterTable.php index 07eff36..27fd121 100644 --- a/src/Ipc/WaiterTable.php +++ b/src/Ipc/WaiterTable.php @@ -18,22 +18,45 @@ /** * Who to poke: a fixed table of wake-registry slots parked on one structure * - * One word per entry, holding `wake slot + 1` so that a zero word means "free" without - * costing a separate occupancy flag. The table is not a queue and has no ordering: waking - * is level-triggered, so notifying everybody parked is always correct and notifying one - * more than necessary costs a socket write and a re-poll. + * One word per entry, holding `owner pid << 32 | wake slot + 1` so that a zero word means + * "free" without costing a separate occupancy flag. The table is not a queue and has no + * ordering: waking is level-triggered, so notifying everybody parked is always correct and + * notifying one more than necessary costs a socket write and a re-poll. + * + * ## Why the owner pid rides in the same word + * + * A waiter that parks inside a blocking call always takes its entry back on the way out, so + * for those the entry can never outlive its owner. A waiter REGISTERED from a consumer's own + * event loop ({@see SharedChannel::registerReceiver()}) is a different lifetime: if that + * process dies while registered, the entry stays, and on a rendezvous channel a stale entry + * is the difference between "a partner is present" and "nobody is there". Recording the pid + * next to the slot is what lets {@see SharedChannel::reapDeadWaiters()} tell a live + * registration from the remains of a dead worker - and the recycled-slot case with it, since + * the wake registry hands a dead owner's slot to the next process that claims one. + * + * The pid is packed into the high half of the SAME aligned word rather than into a second + * one, because two words are a 16-byte record and a 16-byte record tears (EPIC #15, + * correction #1). Linux pids fit comfortably in 32 bits, so one word carries both and every + * read of an entry stays a single non-tearing load. * * ## Locking * * register() and release() MUTATE the table and must be called with the owning structure's * lock held - two processes scanning for a free word without it could pick the same entry, - * and the loser would park with nobody knowing about it. occupants() only READS single - * aligned words, which never tear (EPIC #15, correction #2), and is deliberately used + * and the loser would park with nobody knowing about it. occupants() and entries() only READ + * single aligned words, which never tear (EPIC #15, correction #2), and are deliberately used * without the lock: a notifier reads the table AFTER publishing its record and releasing the * mutex, so it never holds a lock while writing to a socket. */ final class WaiterTable { + /** + * Low half of an entry word: the wake slot, biased by one so that zero means "free" + */ + private const int SLOT_MASK = 0xFFFFFFFF; + + private const int PID_SHIFT = 32; + public function __construct( private readonly Arena $arena, private readonly int $address, @@ -52,14 +75,22 @@ public static function bytesFor(int $capacity): int /** * Parks a wake slot; the caller holds the structure's lock * + * @param int $ownerPid Process the entry belongs to, so a registration that outlives its + * owner can be recognized; 0 records no owner, which is what a + * waiter parked inside a blocking call wants - it always releases + * its own entry, so there is nothing to reap + * * @return int|null Entry index to hand to release(), or null when the table is full * (the caller then falls back to polling with a bounded timeout) */ - public function register(int $wakeSlot): ?int + public function register(int $wakeSlot, int $ownerPid = 0): ?int { for ($entry = 0; $entry < $this->capacity; $entry++) { if ($this->arena->readWord($this->address + $entry * 8) === 0) { - $this->arena->writeWord($this->address + $entry * 8, $wakeSlot + 1); + $this->arena->writeWord( + $this->address + $entry * 8, + ($ownerPid << self::PID_SHIFT) | ($wakeSlot + 1), + ); return $entry; } @@ -89,10 +120,49 @@ public function occupants(): array for ($entry = 0; $entry < $this->capacity; $entry++) { $parked = $this->arena->readWord($this->address + $entry * 8); if ($parked !== 0) { - $slots[] = $parked - 1; + $slots[] = ($parked & self::SLOT_MASK) - 1; } } return $slots; } + + /** + * Every occupied entry with the word that occupies it, read without the lock + * + * The raw word comes back with the decoded parts because a reaper has to re-verify it + * under the lock before releasing anything: between the scan and the critical section the + * owner may have released the entry and a third process may have taken it for itself. + * + * @return list + */ + public function entries(): array + { + $entries = []; + for ($entry = 0; $entry < $this->capacity; $entry++) { + $word = $this->arena->readWord($this->address + $entry * 8); + if ($word !== 0) { + $entries[] = [ + 'entry' => $entry, + 'slot' => ($word & self::SLOT_MASK) - 1, + 'pid' => $word >> self::PID_SHIFT, + 'word' => $word, + ]; + } + } + + return $entries; + } + + /** + * The raw word at $entry, for a reaper re-verifying its scan under the lock + */ + public function wordAt(int $entry): int + { + if ($entry < 0 || $entry >= $this->capacity) { + return 0; + } + + return $this->arena->readWord($this->address + $entry * 8); + } } diff --git a/src/Ipc/WakeRegistry.php b/src/Ipc/WakeRegistry.php index 98a0a1e..cfbf505 100644 --- a/src/Ipc/WakeRegistry.php +++ b/src/Ipc/WakeRegistry.php @@ -189,6 +189,50 @@ public function stream() return $this->readers[$slot] ?? throw IpcException::wakeRegistryNotInherited(); } + /** + * The pid that currently owns $slot, or 0 when the slot is free + * + * A single aligned word read and no lock: the claim table is one word per entry, and an + * aligned 8-byte load never tears (EPIC #15, correction #2), so the answer is either the + * old owner or the new one and never a mixture of the two. + */ + public function ownerOf(int $slot): int + { + if ($slot < 0 || $slot >= $this->capacity) { + return 0; + } + + return (int) $this->arena->readWord($this->entryAddress($slot)); + } + + /** + * Whether $slot is still held by $pid and $pid is still running + * + * This is the liveness test a structure holding LONG-LIVED registrations needs: a waiter + * parked inside a blocking call always takes its entry back on the way out, but a + * registration made from a consumer's own event loop survives the process that made it, + * and on a rendezvous channel a surviving registration is the difference between "a + * partner is present" and "nobody is there". + * + * Both halves of the check matter. The pid may be gone; or the pid may be gone AND its + * slot already recycled to a new worker, which is a live process that never registered + * anywhere - so the claim table has to agree that this pid still owns this slot. + * + * **Makes a syscall, so it is never called from inside a critical section.** Callers scan + * lock-free, then re-verify what they found under the structure's lock before acting on it + * (the same shape claim() uses for its own scan). + */ + public function isOwnerAlive(int $slot, int $pid): bool + { + if ($pid <= 0) { + // No owner was recorded: the entry belongs to a caller that releases it itself, + // so there is nothing here that could ever go stale + return true; + } + + return $this->ownerOf($slot) === $pid && self::isAlive($pid); + } + /** * Sends one event to a parked process */ diff --git a/tests/Ipc/SharedChannelForkTest.php b/tests/Ipc/SharedChannelForkTest.php index 07530f9..7d89ac6 100644 --- a/tests/Ipc/SharedChannelForkTest.php +++ b/tests/Ipc/SharedChannelForkTest.php @@ -150,6 +150,290 @@ public function testRendezvousTrySendOnlySucceedsWhileAReceiverIsParked(): void $this->assertSame(self::OK, $this->await($receiver)); } + public function testARegisteredReceiverIsARendezvousPartnerWithoutEverBeingInsideRecv(): void + { + $channel = $this->channel(0); + $wake = $this->wake(); + $ready = AtomicInt::create($this->arena()); + + $receiver = $this->fork(static function () use ($channel, $wake, $ready): int { + // The shape a runtime with its own scheduler uses: announce the interest, then + // wait in the event loop it already has. recv() is never called + $token = $channel->registerReceiver(); + if ($token === null) { + return self::WRONG_STATE; + } + $ready->set(1); + + $waits = 0; + $end = microtime(true) + 5.0; + while ($channel->count() === 0 && microtime(true) < $end) { + $waits++; + $wake->wait(5.0); + } + $channel->cancelReceiver($token); + + $received = $channel->tryRecv(); + if ($received === null || $received[0] !== 'across the gate') { + return self::WRONG_VALUE; + } + + // Bounded, not exact: the deposit may already have landed by the time this loop is + // reached, so zero waits is a legitimate outcome and so is one. What must never happen + // is a stream of them - a consumer that had to poll for the handoff would come back + // here over and over, and this bound is what makes that a failure instead of a + // slowdown + return $waits <= 2 ? self::OK : self::TIMED_OUT; + }); + + $this->assertTrue($this->awaitWord($ready->address(), 1), 'the receiver never registered'); + $this->assertSame(1, $channel->parkedReceivers()); + + // The gate is open although nobody is inside recv() anywhere in this family + $ticket = $channel->trySendTicket('across the gate'); + $this->assertNotNull($ticket, 'a registered receiver was not accepted as a rendezvous partner'); + + $this->assertSame(self::OK, $this->await($receiver), 'the registered receiver disagreed'); + $this->assertTrue($channel->isTicketTaken($ticket), 'the handoff was never taken'); + } + + public function testRegisteringAReceiverWakesASenderParkedOnTheNotificationSocket(): void + { + $channel = $this->channel(0); + $wake = $this->wake(); + $parked = AtomicInt::create($this->arena()); + + $sender = $this->fork(static function () use ($channel, $wake, $parked): int { + // Nobody is waiting, so there is nowhere to put the value yet + if ($channel->trySendTicket('late partner') !== null) { + return self::WRONG_STATE; + } + + $token = $channel->registerSender(); + if ($token === null) { + return self::WRONG_STATE; + } + $parked->set(1); + + // A registration is the only state change that can help this sender: no record is + // published, no room is freed. If registerReceiver() did not wake parked senders, + // this single bounded wait would come back empty and the test would fail rather + // than be rescued by a re-poll + $woken = $wake->wait(5.0) !== []; + $channel->cancelSender($token); + if (!$woken) { + return self::TIMED_OUT; + } + + $ticket = $channel->trySendTicket('late partner'); + if ($ticket === null) { + return self::WRONG_STATE; + } + + $end = microtime(true) + 5.0; + while (!$channel->isTicketTaken($ticket) && microtime(true) < $end) { + $wake->wait(1.0); + } + + return $channel->isTicketTaken($ticket) ? self::OK : self::TIMED_OUT; + }); + + $this->assertTrue($this->awaitWord($parked->address(), 1), 'the sender never parked'); + + $token = $channel->registerReceiver(); + $this->assertNotNull($token); + + $received = null; + $end = microtime(true) + 5.0; + while ($received === null && microtime(true) < $end) { + $wake->wait(1.0); + $received = $channel->tryRecv(); + } + $channel->cancelReceiver($token); + + $this->assertNotNull($received, 'the sender never deposited after being woken'); + $this->assertSame('late partner', $received[0]); + $this->assertSame(self::OK, $this->await($sender), 'the parked sender disagreed'); + } + + public function testACancelledRegistrationStopsBeingARendezvousPartner(): void + { + $channel = $this->channel(0); + + $token = $channel->registerReceiver(); + $this->assertNotNull($token); + $this->assertSame(1, $channel->parkedReceivers()); + $this->assertTrue($channel->trySend('while registered')); + + // Drain, so the refusal below is about the partner and not about the one ring slot + $this->assertNotNull($channel->tryRecv()); + + $channel->cancelReceiver($token); + + $this->assertSame(0, $channel->parkedReceivers()); + $this->assertFalse($channel->trySend('after cancelling'), 'a cancelled registration still gated a send'); + } + + public function testARecordDepositedAgainstACancelledRegistrationStaysForTheNextReceiver(): void + { + $channel = $this->channel(0); + + $token = $channel->registerReceiver(); + $this->assertNotNull($token); + + // The exact race a select loser runs into: the deposit lands, and the registration it + // landed against is withdrawn before anybody took the value + $ticket = $channel->trySendTicket('deposited then abandoned'); + $this->assertNotNull($ticket); + $channel->cancelReceiver($token); + + // Nothing is lost and nothing is owed: the record is in the ring, the sender's ticket + // is still untaken, and the next receiver completes the handshake + $this->assertFalse($channel->isTicketTaken($ticket), 'the abandoned handoff counted as taken'); + $this->assertSame(1, $channel->count()); + + [$value, $ok] = $channel->recv(5.0); + $this->assertTrue($ok); + $this->assertSame('deposited then abandoned', $value); + $this->assertTrue($channel->isTicketTaken($ticket)); + } + + public function testADeadWorkersRegistrationDoesNotMakeALaterSendBelieveAPartnerIsPresent(): void + { + $channel = $this->channel(0); + + $registrant = $this->fork(static function () use ($channel): int { + // Registers and dies holding the entry, which is what a killed worker does + return $channel->registerReceiver() === null ? self::WRONG_STATE : self::OK; + }); + $this->assertSame(self::OK, $this->await($registrant)); + + // The entry outlived its owner, exactly as a stale registration would + $this->assertSame(1, $channel->parkedReceivers()); + + $this->assertFalse( + $channel->trySend('nobody is really there'), + 'a dead worker\'s registration was accepted as a rendezvous partner', + ); + $this->assertSame(0, $channel->parkedReceivers(), 'the dead registration was not reaped'); + $this->assertSame(0, $channel->count(), 'a value was deposited for a receiver that no longer exists'); + } + + public function testTheReceiversTableRefusesARegistrationPastItsCapacityAndNamesIt(): void + { + $channel = $this->channel(0, waiters: 1); + + $this->assertNotNull($channel->registerReceiver()); + + $this->expectException(IpcException::class); + $this->expectExceptionMessageMatches('/receivers waiter table of this channel is full/'); + + $channel->registerReceiver(); + } + + public function testRegistrationsCancelledUnderContentionNeverLoseOrDuplicateAHandoff(): void + { + $channel = $this->channel(0); + $wake = $this->wake(); + $rounds = 100; + $taken = AtomicInt::create($this->arena()); + $sum = AtomicInt::create($this->arena()); + + $receiver = $this->fork(static function () use ($channel, $wake, $rounds, $taken, $sum): int { + $got = 0; + $lap = 0; + $deadline = microtime(true) + 25.0; + + while ($got < $rounds) { + if (microtime(true) > $deadline) { + return self::TIMED_OUT; + } + $lap++; + + $token = $channel->registerReceiver(); + if ($token === null) { + // A record was already there, so nothing was registered + $received = $channel->tryRecv(); + if ($received !== null && $received[1]) { + $got++; + $taken->add(1); + $sum->add($received[0]); + } + + continue; + } + + if ($lap % 3 === 0) { + // The select-loser interleaving, driven on purpose: withdraw the moment + // after registering, so a deposit lands against a partner that is leaving + $channel->cancelReceiver($token); + + continue; + } + + $wake->wait(0.2); + $channel->cancelReceiver($token); + + $received = $channel->tryRecv(); + if ($received !== null && $received[1]) { + $got++; + $taken->add(1); + $sum->add($received[0]); + } + } + + return self::OK; + }); + + $sender = $this->fork(static function () use ($channel, $wake, $rounds): int { + $deadline = microtime(true) + 25.0; + + for ($value = 1; $value <= $rounds; $value++) { + $ticket = null; + while ($ticket === null) { + if (microtime(true) > $deadline) { + return self::TIMED_OUT; + } + $ticket = $channel->trySendTicket($value); + if ($ticket === null) { + $token = $channel->registerSender(); + if ($token !== null) { + $wake->wait(0.2); + $channel->cancelSender($token); + } + } + } + + // A rendezvous send is only over once the record has been TAKEN, so the next + // value is only offered after this one completed its handshake + while (!$channel->isTicketTaken($ticket)) { + if (microtime(true) > $deadline) { + return self::TIMED_OUT; + } + $token = $channel->registerSender($ticket); + if ($token !== null) { + $wake->wait(0.2); + $channel->cancelSender($token); + } + } + } + + return self::OK; + }); + + $this->awaitAll([$receiver, $sender], 'a rendezvous partner gave up under contention'); + + $this->assertSame($rounds, $taken->get(), 'a handoff was lost while a registration was cancelled'); + $this->assertSame( + $rounds * ($rounds + 1) / 2, + $sum->get(), + 'a handoff was delivered twice, or a different value than the one sent', + ); + $this->assertSame(0, $channel->count(), 'the rendezvous slot is not empty after every value was taken'); + $this->assertSame(0, $channel->parkedReceivers(), 'a registration outlived the exchange'); + $this->assertSame(0, $channel->parkedSenders(), 'a sender registration outlived the exchange'); + } + public function testCloseCrossesProcessesAndReceiversDrainWhatIsLeft(): void { $channel = $this->channel(8); From 5815118f6a3a17d06b8e963efc8d0ff2c3baa1c7 Mon Sep 17 00:00:00 2001 From: Claude Fable 5 Date: Sun, 16 Aug 2026 10:00:16 +0000 Subject: [PATCH 2/3] fix(ipc): leave an already-free waiter entry alone when a token is cancelled A registration token outlives the call that produced it, which makes a repeated cancel a mistake a caller can actually make - unlike a waiter parked inside recv(), which holds its entry for the length of one function. Counting the same entry down twice would drive the parked count below the truth, and on a rendezvous channel that count is the gate a handoff passes: too low and the channel stops accepting values with a receiver standing right there. unpark() now releases only an entry that is still occupied. It is not a licence to cancel twice - by then the entry may hold a NEW registration, which a second cancel would withdraw - so the docblock says that outright and the token still has to be dropped once used. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01U5QRWPfmkZME7hDjU6i5EA --- src/Ipc/SharedChannel.php | 13 +++++++++++-- tests/Ipc/SharedChannelForkTest.php | 19 +++++++++++++++++++ 2 files changed, 30 insertions(+), 2 deletions(-) diff --git a/src/Ipc/SharedChannel.php b/src/Ipc/SharedChannel.php index 8453f07..d94949d 100644 --- a/src/Ipc/SharedChannel.php +++ b/src/Ipc/SharedChannel.php @@ -802,6 +802,13 @@ private function releaseSurveyed(WaiterTable $table, array $dead, int $counterWo /** * Deregisters a waiter entry, under the lock, and drops the parked counter with it + * + * An entry that is already free is left alone rather than counted down again: a token + * outlives the call that made it, so a repeated cancel is a mistake a caller can actually + * make, and a parked count driven below the truth would close a rendezvous gate that should + * be open. This is not a licence to cancel twice - a token that has been cancelled must be + * dropped, because the entry it names may by then hold a NEW registration, which a second + * cancel would withdraw. */ private function unpark(WaiterTable $table, ?int $entry, int $counterWord): void { @@ -811,8 +818,10 @@ private function unpark(WaiterTable $table, ?int $entry, int $counterWord): void $recovered = $this->arena->lockMutexAt($this->mutex); - $table->release($entry); - $this->setWord($counterWord, max($this->word($counterWord) - 1, 0)); + if ($table->wordAt($entry) !== 0) { + $table->release($entry); + $this->setWord($counterWord, max($this->word($counterWord) - 1, 0)); + } $this->arena->unlockMutexAt($this->mutex); diff --git a/tests/Ipc/SharedChannelForkTest.php b/tests/Ipc/SharedChannelForkTest.php index 7d89ac6..6f42ea5 100644 --- a/tests/Ipc/SharedChannelForkTest.php +++ b/tests/Ipc/SharedChannelForkTest.php @@ -274,6 +274,25 @@ public function testACancelledRegistrationStopsBeingARendezvousPartner(): void $this->assertFalse($channel->trySend('after cancelling'), 'a cancelled registration still gated a send'); } + public function testCancellingAnAlreadyFreeRegistrationDoesNotDriveTheParkedCountNegative(): void + { + $channel = $this->channel(0); + + $token = $channel->registerReceiver(); + $this->assertNotNull($token); + + $channel->cancelReceiver($token); + $channel->cancelReceiver($token); + + $this->assertSame(0, $channel->parkedReceivers()); + + // And the channel still works: the count was not corrupted by the repeat + $again = $channel->registerReceiver(); + $this->assertNotNull($again); + $this->assertSame(1, $channel->parkedReceivers()); + $this->assertTrue($channel->trySend('still a partner')); + } + public function testARecordDepositedAgainstACancelledRegistrationStaysForTheNextReceiver(): void { $channel = $this->channel(0); From 8157a5cc87d5d767a5140c6f92fa46a486980fc5 Mon Sep 17 00:00:00 2001 From: Claude Fable 5 Date: Sun, 16 Aug 2026 10:09:58 +0000 Subject: [PATCH 3/3] docs(ipc): name WaiterTable as the owner of the wait-group waiter encoding The header comment described the waiter region as "wake slots parked on zero", which was the whole story when a waiter word held nothing but a biased slot. It now packs an owner pid alongside it, and the encoding belongs to WaiterTable rather than to any of its users - a wait group records no owner, so its words are unchanged, but the comment should point at the structure that decides rather than restate one case of it. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01U5QRWPfmkZME7hDjU6i5EA --- src/Ipc/SharedWaitGroup.php | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/Ipc/SharedWaitGroup.php b/src/Ipc/SharedWaitGroup.php index 2a3cf6e..d8ad643 100644 --- a/src/Ipc/SharedWaitGroup.php +++ b/src/Ipc/SharedWaitGroup.php @@ -25,7 +25,7 @@ * * ```text * header (4 words) counter | mutex address | waiter capacity | waiters parked - * waiters waiter capacity words - wake slots parked on zero + * waiters waiter capacity words - a {@see WaiterTable}, zero meaning free * ``` * * A negative counter is a hard error rather than a clamp: done() called more often than