feat(parallel): let any number of same-class tasks be in flight at once

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.

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.

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<class-string, int>
-     */
-    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,35 +107,19 @@ 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) {
             throw new NotShareableValueException(sprintf(
                 '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]);
-            }
-        }
-    }
 }

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);

src/Parallel/WorkerSupervisor.php

@@ -63,13 +63,6 @@ final class WorkerSupervisor
     /** @var list<WorkerCrashedException> */
     private array $crashes = [];
 
-    /**
-     * Slot id => the arena address of the task dispatched under it, while it is still in flight.
-     *
-     * @var array<int, int>
-     */
-    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.
      *

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;
 });
 

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--
+<?php
+
+declare(strict_types=1);
+
+use Lisachenko\NativePhpCoroutines\Exception\ParallelTaskException;
+use Lisachenko\NativePhpCoroutines\Runtime;
+use Lisachenko\NativePhpCoroutines\TaskRuntime;
+use Lisachenko\NativePhpCoroutines\Tests\Support\PanicWithMessageTask;
+use Lisachenko\NativePhpCoroutines\Timer;
+
+use function Lisachenko\NativePhpCoroutines\Tests\Support\parallelChildrenLeft;
+
+include __DIR__ . '/../../vendor/autoload.php';
+include __DIR__ . '/../Support/parallel.php';
+include __DIR__ . '/../Support/shared.php';
+
+// A captured panic used to be persisted under SharedError's class name, so the second worker's
+// capture superseded the first worker's registry entry — a waiter attaching the first error's
+// address after that found nothing. Each capture is now its own instance graph: two workers
+// failing near-simultaneously each leave an error, and each waiter reads exactly its own.
+$runtime = new Runtime(workers: 2, arenaSize: 32 << 20);
+
+$runtime->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

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--
+<?php
+
+declare(strict_types=1);
+
+use Lisachenko\NativePhpCoroutines\Runtime;
+use Lisachenko\NativePhpCoroutines\TaskRuntime;
+use Lisachenko\NativePhpCoroutines\Tests\Support\MutateSharedTask;
+use Lisachenko\NativePhpCoroutines\Tests\Support\SharedCounter;
+use Lisachenko\NativePhpCoroutines\Timer;
+
+use function Lisachenko\NativePhpCoroutines\Tests\Support\parallelChildrenLeft;
+
+include __DIR__ . '/../../vendor/autoload.php';
+include __DIR__ . '/../Support/parallel.php';
+include __DIR__ . '/../Support/shared.php';
+
+// The same class-keying that limited tasks limited roots: declaring a second root of one class
+// used to upsert the first root's registry entry. Roots are now filed under the NAME they were
+// declared with, so one class serves as many roots as the application wants.
+//
+// The two mutating tasks are also two unpublished instances of ONE task class, spawned through
+// the persist route — the second half of what issue #15 unlocked.
+$runtime = new Runtime(workers: 1, arenaSize: 32 << 20);
+
+$runtime->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