diff --git a/AGENTS.md b/AGENTS.md index ed0f52e..eebe713 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -226,12 +226,17 @@ in every process. - **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. -- **The registry is keyed by class**, so `persist()` and the arena route of `ArenaTaskDirectory` hold - one live graph per class. A second concurrent task of one class is refused with the remedy named - (publish before the fork, or use distinct classes) rather than silently superseding a graph a - worker is still reading. -- **One `SharedError` per store.** A second panic replaces the first, so `ParallelTaskException` - promises the panic it was handed and not a history of them. +- **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 + supersedes a graph a worker is still reading; shared *roots* are filed under the name they were + declared with, so one class serves many roots. The cost sits where the arena's economics already + are: every `spawnParallel()` of an unpublished task clones its graph into the arena and that + memory lives until the family tears down — a steady-state workload publishes its tasks before + the fork, which allocates nothing per spawn. +- **One `SharedError` per panic.** Each capture is its own instance graph, so two workers failing + near-simultaneously each leave an error their waiter can still attach by the address its own + slot carries. - **Result slots are bump-allocated from a pre-sized table and never given back.** They are a bounded supply for the life of the arena, which is what `soak-arena-watermark.php` reports rather than assumes. diff --git a/README.md b/README.md index 7fc890d..056f960 100644 --- a/README.md +++ b/README.md @@ -496,8 +496,10 @@ z-engine requires it, and z-engine is a hard dependency of this package. 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. -- **`persist()` and shared roots are keyed by class.** One live graph per class: a second instance - supersedes the first, so a design that needs several gives them distinct classes or a `SharedArray`. +- **`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; + `publishTask()` before the fork allocates nothing per spawn. - **The arena is a bump allocator with no free list.** Blocks are reclaimed when the region dies with the creating process, and rewriting a shared string property costs a block per write. Size for it, and watch the watermark **plateau** rather than expecting it to fall. diff --git a/src/Parallel/ArenaTaskDirectory.php b/src/Parallel/ArenaTaskDirectory.php index e270139..02c7427 100644 --- a/src/Parallel/ArenaTaskDirectory.php +++ b/src/Parallel/ArenaTaskDirectory.php @@ -30,14 +30,16 @@ * * Either way the task itself never travels: only the integer does, on a fixed-size record. * - * # The one-graph-per-class limit of route 2, said out loud + * # Route 2 is per instance, and each spawn keeps its memory until teardown * - * The substrate's registry is keyed by **class name**, so persisting a second instance of a class is - * an upsert that supersedes the first — and superseding a task another worker is still running would - * release the graph under its feet. This class therefore tracks what is in flight per class and - * refuses the second concurrent task of one class with the remedy named: publish tasks before the - * fork, or give the two tasks distinct classes. It is a refusal rather than a silent replacement - * because the silent version corrupts memory in a process that is not the one making the mistake. + * A persisted task is an **instance graph** ({@see SharedArena::persist()} rides the substrate's + * `persistInstance()`): its registry entry is named by its own root address, so a second task of + * the same class is a second entry, never an upsert — any number of `new RenderJob(...)` are in + * flight at once, and none supersedes a graph a worker is still reading. What route 2 costs + * instead is arena memory per spawn: an instance graph lives until the family tears down, which + * is the arena's ordinary leak-until-teardown economics and is what the watermark soak reports. + * A steady-state workload spawning the same tasks forever wants route 1, which allocates nothing + * per spawn. */ final class ArenaTaskDirectory implements TaskDirectory { @@ -64,13 +66,6 @@ final class ArenaTaskDirectory implements TaskDirectory */ private array $publishedAddresses = []; - /** - * Class name => arena address of the task graph currently in flight under that key. - * - * @var array - */ - private array $inFlight = []; - /** * Deliberately not 0 and deliberately spaced. A published address is opaque and is never * dereferenced; keeping it far from anything the arena hands out makes a confusion of the two @@ -112,18 +107,6 @@ public function addressOf(Task $task): int return $this->publishedAddresses[$key]; } - $class = $task::class; - - if (isset($this->inFlight[$class])) { - throw new \LogicException(sprintf( - '%s is already running in a worker, and the shared registry is keyed by class: ' - . 'persisting a second instance of it would release the graph the running worker ' - . 'is reading. Publish tasks before the fork with register(), or give the two ' - . 'tasks distinct classes', - $class, - )); - } - try { $shared = $this->arena->persist($task); } catch (SubstrateRefusal | NotPersistableException $refused) { @@ -131,16 +114,12 @@ public function addressOf(Task $task): int 'the task %s cannot be cloned into the shared arena: %s. A task carries only values ' . 'that can cross a worker boundary — scalars, shared objects, SharedArray — and ' . 'never a plain array property, a resource or a post-fork closure', - $class, + $task::class, $refused->getMessage(), ), 0, $refused); } - $address = $this->arena->addressOf($shared); - - $this->inFlight[$class] = $address; - - return $address; + return $this->arena->addressOf($shared); } public function taskAt(int $address): Task @@ -160,18 +139,4 @@ public function taskAt(int $address): Task )); } - /** - * Report that the task under this class has finished, so the key is free again. - * - * Called by the supervisor when a slot settles. Without it the second spawn of a class would be - * refused forever, which is a leak of the refusal rather than of memory. - */ - public function releaseInFlight(int $address): void - { - foreach ($this->inFlight as $class => $registered) { - if ($registered === $address) { - unset($this->inFlight[$class]); - } - } - } } diff --git a/src/Parallel/SharedArena.php b/src/Parallel/SharedArena.php index 5d41b40..5042ab7 100644 --- a/src/Parallel/SharedArena.php +++ b/src/Parallel/SharedArena.php @@ -429,7 +429,10 @@ public function persist(object $object): object // happens before the fork, which is what gives the whole family one class entry for it. class_exists($object::class); - return $this->store->persist($object::class, $object, true); + // Per instance, not per class: each graph's registry entry is named by its own root + // address, so persisting a second instance of one class never supersedes the first — + // which is what lets two tasks of one class be in flight at once. + return $this->store->persistInstance($object, true); } /** The arena address of a shared instance — the only identity that means anything across a fork. */ @@ -579,8 +582,10 @@ private function createRoot(string $name, string $class, int $capacity): array return [self::KIND_ARRAY, $array->address()]; } + // Keyed by the ROOT NAME, not the class: two roots of one class are two entries, and the + // name the application declared is exactly the name the registry files the graph under. $instance = new $class(); - $shared = $this->store->persist($class, $instance, true); + $shared = $this->store->persist($name, $instance, true); $address = $this->store->sharedIdOf($shared); $this->arena->putRoot($name, $address); diff --git a/src/Parallel/WorkerSupervisor.php b/src/Parallel/WorkerSupervisor.php index 56e2cb6..85fe13e 100644 --- a/src/Parallel/WorkerSupervisor.php +++ b/src/Parallel/WorkerSupervisor.php @@ -63,13 +63,6 @@ final class WorkerSupervisor /** @var list */ private array $crashes = []; - /** - * Slot id => the arena address of the task dispatched under it, while it is still in flight. - * - * @var array - */ - private array $dispatched = []; - private readonly SlotTable $slots; private int $cursor = 0; @@ -224,8 +217,6 @@ public function spawn(Task $task, ?int $worker = null): JoinHandleInterface $slot = $this->slots->open($target->id()); - $this->dispatched[$slot->id] = $address; - try { $target->dispatch($slot->id, $address); } catch (\Throwable $failure) { @@ -347,7 +338,6 @@ private function apply(ProcessWorker $worker, ControlRecord $record): void // makes it a real PHP value rather than something rebuilt from bytes on a socket. if ($this->arena !== null) { $this->slots->refresh(); - $this->releaseTask($record->slotId); return; } @@ -371,22 +361,6 @@ private function apply(ProcessWorker $worker, ControlRecord $record): void // over. } - /** Let the directory reuse a class key once the task under it has finished. */ - private function releaseTask(int $slotId): void - { - $address = $this->dispatched[$slotId] ?? null; - - if ($address === null) { - return; - } - - unset($this->dispatched[$slotId]); - - if ($this->tasks instanceof ArenaTaskDirectory) { - $this->tasks->releaseInFlight($address); - } - } - /** * Turn a `PANIC` record's tag into the exception the waiter sees. * diff --git a/tests/Functional/testATaskPanicSurfacesWithItsClassMessageAndTrace.phpt b/tests/Functional/testATaskPanicSurfacesWithItsClassMessageAndTrace.phpt index 57115f6..814cc45 100644 --- a/tests/Functional/testATaskPanicSurfacesWithItsClassMessageAndTrace.phpt +++ b/tests/Functional/testATaskPanicSurfacesWithItsClassMessageAndTrace.phpt @@ -30,7 +30,7 @@ $runtime = new Runtime(workers: 1, arenaSize: 32 << 20); $task = new SharedPanicTask(); $runtime->publishTask($task); -$runtime->run(static function (TaskRuntime $self) use ($task): void { +$runtime->run(static function (TaskRuntime $self) use ($task, $runtime): void { Timer::after(15.0, static function (): void { throw new RuntimeException('deadline: the panic never reached the waiter'); }); @@ -46,9 +46,10 @@ $runtime->run(static function (TaskRuntime $self) use ($task): void { echo 'worker: ', $panic->workerId(), PHP_EOL; } - // The pool is unharmed: a panicking task is an ordinary outcome, not a lost worker. + // The pool is unharmed: a panicking task is an ordinary outcome, not a lost worker. The + // supervisor is diagnostics, so it is read off the concrete runtime, not the task surface. echo 'the worker is still alive: ', - $self->supervisor()?->worker(0)->isAlive() === true ? 'yes' : 'no', + $runtime->supervisor()?->worker(0)->isAlive() === true ? 'yes' : 'no', PHP_EOL; }); diff --git a/tests/Functional/testEachPanicKeepsItsOwnSharedError.phpt b/tests/Functional/testEachPanicKeepsItsOwnSharedError.phpt new file mode 100644 index 0000000..a8bab69 --- /dev/null +++ b/tests/Functional/testEachPanicKeepsItsOwnSharedError.phpt @@ -0,0 +1,55 @@ +--TEST-- +Two panicking workers each keep their own error, and each waiter reads its own +--INI-- +ffi.enable=1 +opcache.jit=off +error_reporting=E_ALL & ~E_DEPRECATED +--FILE-- +run(static function (TaskRuntime $self): void { + Timer::after(30.0, static function (): void { + throw new RuntimeException('deadline: the two panics never reached their waiters'); + }); + + $one = $self->spawnParallel(new PanicWithMessageTask('the first worker exploded'), 0); + $two = $self->spawnParallel(new PanicWithMessageTask('the second worker exploded'), 1); + + foreach (['first' => $one, 'second' => $two] as $which => $handle) { + try { + $handle->await(); + + echo $which, ': the await returned, which it must not', PHP_EOL; + } catch (ParallelTaskException $panic) { + echo $which, ' waiter got its own panic: ', + str_contains($panic->getMessage(), "the {$which} worker exploded") ? 'yes' : 'no', PHP_EOL; + } + } +}); + +echo 'children left: ', parallelChildrenLeft(), PHP_EOL; +?> +--EXPECT-- +first waiter got its own panic: yes +second waiter got its own panic: yes +children left: none diff --git a/tests/Functional/testTwoSharedRootsOfOneClassAreDistinct.phpt b/tests/Functional/testTwoSharedRootsOfOneClassAreDistinct.phpt new file mode 100644 index 0000000..b3a73b7 --- /dev/null +++ b/tests/Functional/testTwoSharedRootsOfOneClassAreDistinct.phpt @@ -0,0 +1,57 @@ +--TEST-- +Two shared roots of one class are two graphs, each mutated on its own identity +--INI-- +ffi.enable=1 +opcache.jit=off +error_reporting=E_ALL & ~E_DEPRECATED +--FILE-- +declareShared('left', SharedCounter::class); +$runtime->declareShared('right', SharedCounter::class); + +$runtime->run(static function (TaskRuntime $self): void { + Timer::after(30.0, static function (): void { + throw new RuntimeException('deadline: the root mutations never came back'); + }); + + $self->spawnParallel(new MutateSharedTask('left', 41, 'sinister'))->await(); + $self->spawnParallel(new MutateSharedTask('right', 42, 'dexter'))->await(); + + $left = $self->shared('left'); + $right = $self->shared('right'); + + echo 'the roots are distinct objects: ', $left !== $right ? 'yes' : 'no', PHP_EOL; + echo 'left: ', $left->value, ' ', $left->label, PHP_EOL; + echo 'right: ', $right->value, ' ', $right->label, PHP_EOL; +}); + +echo 'children left: ', parallelChildrenLeft(), PHP_EOL; +?> +--EXPECT-- +the roots are distinct objects: yes +left: 41 sinister +right: 42 dexter +children left: none diff --git a/tests/Functional/testTwoTasksOfOneClassAreInFlightAtOnce.phpt b/tests/Functional/testTwoTasksOfOneClassAreInFlightAtOnce.phpt new file mode 100644 index 0000000..e53dd9e --- /dev/null +++ b/tests/Functional/testTwoTasksOfOneClassAreInFlightAtOnce.phpt @@ -0,0 +1,65 @@ +--TEST-- +Two tasks of one class are in flight at once, and each graph survives the other +--INI-- +ffi.enable=1 +opcache.jit=off +error_reporting=E_ALL & ~E_DEPRECATED +--FILE-- +persist(new NapThenEchoTask(0.25, 'graph-one')); +$second = $runtime->persist(new NapThenEchoTask(0.25, 'graph-two')); + +$arena = $runtime->arena(); + +echo 'both instances are shared at distinct addresses: ', + $arena !== null && $arena->addressOf($first) !== $arena->addressOf($second) ? 'yes' : 'no', PHP_EOL; + +$runtime->run(static function (TaskRuntime $self) use ($first, $second): void { + Timer::after(30.0, static function (): void { + throw new RuntimeException('deadline: the concurrent same-class tasks never finished'); + }); + + // Pinned to different workers and both napping, so their lifetimes genuinely overlap. + $one = $self->spawnParallel($first, 0); + $two = $self->spawnParallel($second, 1); + + // Asserted, not assumed: while both are running, each graph is read back through shared + // memory and still carries its own payload — nothing was mutated or freed by the sibling. + echo 'graph one intact while both run: ', $first->payload === 'graph-one' ? 'yes' : 'no', PHP_EOL; + echo 'graph two intact while both run: ', $second->payload === 'graph-two' ? 'yes' : 'no', PHP_EOL; + + echo 'first awaits its own result: ', $one->await() === 'graph-one' ? 'yes' : 'no', PHP_EOL; + echo 'second awaits its own result: ', $two->await() === 'graph-two' ? 'yes' : 'no', PHP_EOL; +}); + +echo 'children left: ', parallelChildrenLeft(), PHP_EOL; +?> +--EXPECT-- +both instances are shared at distinct addresses: yes +graph one intact while both run: yes +graph two intact while both run: yes +first awaits its own result: yes +second awaits its own result: yes +children left: none diff --git a/tests/Support/shared.php b/tests/Support/shared.php index 97979d8..1f893c2 100644 --- a/tests/Support/shared.php +++ b/tests/Support/shared.php @@ -161,6 +161,40 @@ public function run(TaskRuntime $runtime): mixed } } +/** + * Parks briefly on the worker's scheduler, then hands back the string it was built with. + * + * The payload is the graph: it lives in this task's own arena clone as an arena string, so the + * value coming back intact is direct evidence the graph was neither superseded nor released while + * a second instance of this same class was persisted and run concurrently. The properties are + * public so the spawner can read them back through shared memory while the task is still running. + */ +final class NapThenEchoTask implements Task +{ + public function __construct( + public readonly float $seconds, + public readonly string $payload, + ) {} + + public function run(TaskRuntime $runtime): mixed + { + Coroutine::sleep($this->seconds); + + return $this->payload; + } +} + +/** Panics with the message it was built with, so each instance's panic is distinguishable. */ +final class PanicWithMessageTask implements Task +{ + public function __construct(public readonly string $message) {} + + public function run(TaskRuntime $runtime): mixed + { + throw new \RuntimeException($this->message); + } +} + /** Pushes a bounded number of values onto a named shared channel. */ final class SharedSendTask implements Task {