feat(#92): grouped future waits (awaitAll/onReady) and range read-ahead

CHANGELOG.md

@@ -27,6 +27,18 @@
   `Snapshot::$parentTransaction` is documented as a GC anchor.
 
 ### Added
+- [#92] Non-blocking future support: `Future::awaitAll()` resolves N futures
+  with a single grouped wait (all requests in flight at once — total latency
+  of the slowest future instead of the sum of all of them, results and errors
+  resolved in input order); `Future::onReady()` registers a completion hook
+  that always runs on the PHP thread (never on the FDB network thread); and
+  `RangeResult` iteration now prefetches the next page before the consumer
+  finishes the current one, removing one round trip of latency per page of a
+  range scan. The awaitAll/poll model and its limits (no
+  `fdb_future_set_callback` binding yet, no Fiber suspension) are documented
+  in `docs/advanced.md`. Covered by
+  `tests/Unit/FutureAwaitAllTest.php` and
+  `tests/Unit/RangeResultTest.php`.
 - [#96] Upstream FoundationDB binding tester support: a PHP stack machine
   driver (`tests/bindingtester/tester.php`) implementing the upstream
   protocol (API operations, tuple operations, directory layer extension,

docs/advanced.md

@@ -2,7 +2,7 @@
 
 ## Overview
 
-Advanced features: Locality API, Key Utilities, Database Monitoring, Connection Strings, Explicit Lifecycle Management.
+Advanced features: Locality API, Key Utilities, Database Monitoring, Connection Strings, Explicit Lifecycle Management, Grouped Future Waits & Range Read-Ahead.
 
 ## Locality API
 
@@ -177,3 +177,58 @@ $future->isReady();  // check without blocking
 $future->cancel();   // cancel the operation
 $value = $future->await(); // block until ready
 ```
+
+### Grouped waits: `Future::awaitAll()`
+
+Futures can be fanned out and awaited as a group. All N requests are already
+in flight, so the total wait is driven by the slowest future (roughly one
+round trip) instead of N round trips:
+
+```php
+$values = Future::awaitAll([
+    'a' => $tr->get('a'),
+    'b' => $tr->get('b'),
+    'c' => $tr->get('c'),
+]);
+// ['a' => '...', 'b' => '...', 'c' => '...']
+```
+
+- Results keep the input keys; resolution order matches the input order.
+- Errors are still thrown from `await()` — after all futures are ready, in
+  input order, so a failure in an early future never cancels the rest.
+- Implemented by polling the non-blocking `fdb_future_is_ready()` for every
+  future at once (1 ms spin interval). No PHP code ever runs on the FDB
+  network thread.
+
+### Completion hooks: `Future::onReady()`
+
+```php
+$future = $tr->get('key');
+$future->onReady(function ($f) {
+    echo "ready!\n";
+});
+$value = $future->await(); // hook fires just before the value is read
+```
+
+- The hook runs exactly once, on the PHP thread that resolved the future —
+  never on the FDB network thread.
+- Hooks registered after the future has already been resolved fire
+  immediately. Exceptions thrown from a hook propagate to the caller.
+
+### Range read-ahead
+
+`RangeResult` iteration overlaps network round trips with consumption: while
+the consumer is processing page N, the request for page N+1 is already in
+flight. For typical multi-page range scans this removes one round trip of
+latency per page. Nothing to configure — it is the default behavior of
+`foreach ($range as $keyValue)` and `RangeResult::toArray()`.
+
+### Limits of the current async model
+
+- `await()` still blocks the calling (PHP) thread until the future is ready —
+  there is no callback binding (`fdb_future_set_callback`) yet, because PHP
+  code cannot safely run on the FDB network thread. Grouped waits
+  (`awaitAll()`) and range read-ahead recover most of the practical
+  parallelism without it.
+- Fibers are not suspended by `await()`, so the library does not yet compose
+  with Fiber-based event loops (Revolt/AMPHP/ReactPHP).

src/Future/Future.php

@@ -9,6 +9,9 @@
 
 abstract class Future
 {
+    /** Poll interval (microseconds) used by awaitAll() while spinning. */
+    private const AWAIT_ALL_POLL_INTERVAL_US = 1000;
+
     protected bool $resolved = false;
 
     /**
@@ -17,6 +20,9 @@ abstract class Future
      */
     protected ?bool $errorState = null;
 
+    /** @var list<callable(static): void> */
+    private array $onReadyHooks = [];
+
     public function __construct(
         protected CData $fpointer,
         protected readonly NativeClient $client,
@@ -56,6 +62,70 @@ public function cancel(): void
         $this->client->fdb->fdb_future_cancel($this->fpointer);
     }
 
+    /**
+     * Registers a hook that runs exactly once, after the future has become
+     * ready (before its payload is read by await()).
+     *
+     * The hook always executes on the PHP thread that resolved the future —
+     * never on the FDB network thread. Under the current blocking model the
+     * hook fires when the future is resolved via await() (or by awaitAll()).
+     * Hooks registered after the future has already been resolved fire
+     * immediately. Exceptions thrown from a hook propagate to the caller.
+     *
+     * @param callable(static): void $fn
+     */
+    public function onReady(callable $fn): void
+    {
+        if ($this->resolved) {
+            $fn($this);
+
+            return;
+        }
+
+        $this->onReadyHooks[] = $fn;
+    }
+
+    /**
+     * Awaits many futures together instead of serializing them: all pending
+     * requests are already in flight, so total wait time is driven by the
+     * slowest future (roughly one round trip) rather than the sum of all of
+     * them. N futures awaited one-by-one with await() cost N round trips.
+     *
+     * Polls the non-blocking fdb_future_is_ready() for every future at once
+     * (no PHP code ever runs on the FDB network thread), then resolves the
+     * results in the original order. Futures are resolved strictly after all
+     * of them are ready, so an error in an early future does not cancel the
+     * wait for the rest; errors are still thrown from await() in input order.
+     *
+     * @param array<Future> $futures
+     * @return array<int|string, mixed> results of every future, keyed like the input
+     */
+    public static function awaitAll(array $futures): array
+    {
+        $pending = [];
+        foreach ($futures as $key => $future) {
+            if (!$future->resolved && !$future->isReady()) {
+                $pending[$key] = $future;
+            }
+        }
+
+        while ($pending !== []) {
+            usleep(self::AWAIT_ALL_POLL_INTERVAL_US);
+            foreach ($pending as $key => $future) {
+                if ($future->isReady()) {
+                    unset($pending[$key]);
+                }
+            }
+        }
+
+        $results = [];
+        foreach ($futures as $key => $future) {
+            $results[$key] = $future->await();
+        }
+
+        return $results;
+    }
+
     abstract public function await(): mixed;
 
     protected function blockUntilReady(): void
@@ -68,11 +138,25 @@ protected function blockUntilReady(): void
         // after the memory has been released.
         $errorCode = $this->client->fdb->fdb_future_get_error($this->fpointer);
         $this->errorState = $errorCode !== 0;
+        $this->fireOnReadyHooks();
         if ($errorCode !== 0) {
             $this->client->checkError($errorCode);
         }
     }
 
+    private function fireOnReadyHooks(): void
+    {
+        if ($this->onReadyHooks === []) {
+            return;
+        }
+
+        $hooks = $this->onReadyHooks;
+        $this->onReadyHooks = [];
+        foreach ($hooks as $hook) {
+            $hook($this);
+        }
+    }
+
     protected function releaseMemory(): void
     {
         $this->client->fdb->fdb_future_release_memory($this->fpointer);

src/Future/FutureKeyValueArray.php

@@ -7,7 +7,7 @@
 use CrazyGoat\FoundationDB\KeyValue;
 use FFI;
 
-final class FutureKeyValueArray extends Future
+final class FutureKeyValueArray extends Future implements KvsFuture
 {
     private ?FutureKvResult $cachedResult = null;
 

src/Future/FutureKvResult.php

@@ -6,7 +6,7 @@
 
 use CrazyGoat\FoundationDB\KeyValue;
 
-final readonly class FutureKvResult
+final readonly class FutureKvResult implements KvsFuture
 {
     /**
      * @param list<KeyValue> $kvs
@@ -18,4 +18,13 @@ public function __construct(
         public bool $more,
     ) {
     }
+
+    /**
+     * Already resolved — this method simply returns $this so that
+     * FutureKvResult can be used anywhere a KvsFuture is expected.
+     */
+    public function await(): FutureKvResult
+    {
+        return $this;
+    }
 }

src/Future/KvsFuture.php

@@ -0,0 +1,20 @@
+<?php
+
+declare(strict_types=1);
+
+namespace CrazyGoat\FoundationDB\Future;
+
+/**
+ * A pending (or already resolved) batch of key-value pairs that can be
+ * resolved with await().
+ *
+ * Implemented by FutureKeyValueArray (a real FDB future backed by
+ * `fdb_transaction_get_range()`) and by FutureKvResult (an already-resolved
+ * value, e.g. from unit-test stubs). RangeResult::paginate() accepts either,
+ * which lets callers hand in un-awaited futures so that the next page can be
+ * prefetched while the current one is being consumed.
+ */
+interface KvsFuture
+{
+    public function await(): FutureKvResult;
+}

src/RangeResult.php

@@ -7,7 +7,7 @@
 use Closure;
 use CrazyGoat\FoundationDB\Enum\StreamingMode;
 use CrazyGoat\FoundationDB\Future\FutureKeyValueArray;
-use CrazyGoat\FoundationDB\Future\FutureKvResult;
+use CrazyGoat\FoundationDB\Future\KvsFuture;
 
 /** @implements \IteratorAggregate<int, KeyValue> */
 final readonly class RangeResult implements \IteratorAggregate
@@ -38,15 +38,20 @@ public function getIterator(): \Generator
                 StreamingMode $mode,
                 int $iteration,
                 bool $reverse,
-            ): FutureKvResult => $this->getRangeRaw($begin, $end, $limit, $mode, $iteration, $reverse)->await(),
+            ): FutureKeyValueArray => $this->getRangeRaw($begin, $end, $limit, $mode, $iteration, $reverse),
         );
     }
 
     /**
      * Iterates a range across server batches, advancing the (exclusive) endpoint
      * strictly past the last key of the previous batch so that no key is yielded twice.
      *
-     * @param Closure(KeySelector, KeySelector, int, StreamingMode, int, bool): FutureKvResult $fetcher
+     * Read-ahead: while the current batch is being consumed, the request for the
+     * next batch is already in flight (the fetcher returns an un-awaited
+     * future), so page N+1's network round trip overlaps with the consumption
+     * of page N instead of serializing after it.
+     *
+     * @param Closure(KeySelector, KeySelector, int, StreamingMode, int, bool): KvsFuture $fetcher
      * @return \Generator<int, KeyValue>
      */
     public static function paginate(
@@ -66,27 +71,30 @@ public static function paginate(
             return;
         }
 
-        while (true) {
-            $currentLimit = $limit !== null ? $limit - $fetched : 0;
+        $future = $fetcher($beginSelector, $endSelector, $limit ?? 0, $mode, $iteration, $reverse);
 
-            $result = $fetcher($beginSelector, $endSelector, $currentLimit, $mode, $iteration, $reverse);
+        while (true) {
+            $result = $future->await();
             $kvs = $result->kvs;
             $count = $result->count;
+            $fetched += $count;
 
-            foreach ($kvs as $kv) {
-                yield $kv;
-                $fetched++;
-            }
+            $exhausted = $count === 0
+                || !$result->more
+                || ($limit !== null && $fetched >= $limit);
 
-            if ($count === 0 || !$result->more) {
-                break;
-            }
+            if ($exhausted) {
+                foreach ($kvs as $kv) {
+                    yield $kv;
+                }
 
-            if ($limit !== null && $fetched >= $limit) {
                 break;
             }
 
+            // Prefetch the next batch BEFORE yielding the current one, so its
+            // round trip overlaps with the consumer processing this batch.
             $lastKey = $kvs[$count - 1]->key;
+            $nextLimit = $limit !== null ? $limit - $fetched : 0;
 
             if ($reverse) {
                 $endSelector = KeySelector::firstGreaterOrEqual($lastKey);
@@ -95,6 +103,11 @@ public static function paginate(
             }
 
             $iteration++;
+            $future = $fetcher($beginSelector, $endSelector, $nextLimit, $mode, $iteration, $reverse);
+
+            foreach ($kvs as $kv) {
+                yield $kv;
+            }
         }
     }