Track active loaders and promise adapters

docs/adr/0001-track-active-loaders-and-promise-adapters.md

@@ -0,0 +1,30 @@
+# Track Active Loaders and Promise Adapters
+
+## Status
+
+Accepted
+
+## Context
+
+`DataLoader::await()` must dispatch queued loaders and use the promise adapter
+that owns the supplied promise. A strong global registry kept every loader and
+its cache alive. Selecting the first registered loader also sent promises to
+the wrong adapter when an application used more than one promise library.
+
+## Decision
+
+Keep weak references to live loader instances. Retain a loader strongly only
+while it has queued work, and release it when dispatch starts. Keep a weak map
+from load promises to their owning adapters. For promises without recorded
+ownership, ask live adapters whether they recognize the promise before using a
+compatibility fallback.
+
+A loader destructor cancels only that loader's queued promises. It does not
+dispatch global work.
+
+## Consequences
+
+Idle and dispatched loaders can be garbage-collected. Inline temporary loaders
+remain alive until their queued work is dispatched. Mixed promise libraries use
+the correct adapter when ownership is known or recognized. A loader with queued
+work remains retained until the application dispatches or cancels that work.

src/DataLoader.php

@@ -35,10 +35,20 @@ class DataLoader implements DataLoaderInterface
      */
     private $queue = [];
 
+    /**
+     * @var \WeakMap<self, null>|null
+     */
+    private static $instances;
+
     /**
      * @var self[]
      */
-    private static $instances = [];
+    private static $activeInstances = [];
+
+    /**
+     * @var \WeakMap<object, PromiseAdapterInterface>|null
+     */
+    private static $promiseAdapters;
 
     /**
      * @var PromiseAdapterInterface
@@ -51,7 +61,8 @@ public function __construct(callable $batchLoadFn, PromiseAdapterInterface $prom
         $this->promiseAdapter = $promiseFactory;
         $this->options = $options ?: new Option();
         $this->promiseCache = $this->options->getCacheMap();
-        self::$instances[] = $this;
+        self::$instances ??= new \WeakMap();
+        self::$instances[$this] = null;
     }
 
     /**
@@ -69,27 +80,29 @@ public function load($key)
         if ($shouldCache) {
             $cachedPromise = $this->promiseCache->get($cacheKey);
             if ($cachedPromise) {
-                return $cachedPromise;
+                return $this->registerCachedPromiseAdapter($cachedPromise);
             }
         }
 
         // Otherwise, produce a new Promise for this value.
         $promise = $this->getPromiseAdapter()->create(
             $resolve,
             $reject,
-            function () {
+            static function () {
                 // Cancel/abort any running operations like network connections, streams etc.
 
                 throw new \RuntimeException('DataLoader destroyed before promise complete.');
             }
         );
+        $this->registerPromiseAdapter($promise);
 
         $this->queue[] = [
             'key' => $key,
             'resolve' => $resolve,
             'reject' => $reject,
             'promise' => $promise,
         ];
+        self::$activeInstances[spl_object_id($this)] = $this;
 
         // Determine if a dispatch of this queue should be scheduled.
         // A single dispatch should be scheduled per queue at the time when the
@@ -119,12 +132,12 @@ public function loadMany($keys)
         if ($keys instanceof \Traversable) {
             $keys = iterator_to_array($keys, false);
         }
-        return $this->getPromiseAdapter()->createAll(array_map(
+        return $this->registerPromiseAdapter($this->getPromiseAdapter()->createAll(array_map(
             function ($key) {
                 return $this->load($key);
             },
             $keys
-        ));
+        )));
     }
 
     /**
@@ -163,6 +176,7 @@ public function prime($key, $value)
             // Cache a rejected promise if the value is a Throwable, in order to match
             // the behavior of load(key).
             $promise = $value instanceof \Throwable ? $this->getPromiseAdapter()->createRejected($value) : $this->getPromiseAdapter()->createFulfilled($value);
+            $this->registerPromiseAdapter($promise);
 
             $this->promiseCache->set($cacheKey, $promise);
         }
@@ -172,21 +186,21 @@ public function prime($key, $value)
 
     public function __destruct()
     {
+        unset(self::$activeInstances[spl_object_id($this)]);
+
         if ($this->needProcess()) {
-            foreach ($this->queue as $data) {
+            $queue = $this->queue;
+            $this->queue = [];
+
+            foreach ($queue as $data) {
                 try {
                     $this->getPromiseAdapter()->cancel($data['promise']);
                 } catch (\Throwable $e) {
                     // no need to do nothing if cancel failed
                 }
             }
-            $this->await();
-        }
-        foreach (self::$instances as $i => $instance) {
-            if ($this !== $instance) {
-                continue;
-            }
-            unset(self::$instances[$i]);
+
+            $this->getPromiseAdapter()->await();
         }
     }
 
@@ -208,6 +222,31 @@ protected function getPromiseAdapter()
         return $this->promiseAdapter;
     }
 
+    private function registerPromiseAdapter($promise)
+    {
+        if (is_object($promise)) {
+            self::$promiseAdapters ??= new \WeakMap();
+            self::$promiseAdapters[$promise] = $this->getPromiseAdapter();
+        }
+
+        return $promise;
+    }
+
+    private function registerCachedPromiseAdapter($promise)
+    {
+        if (!is_object($promise) || (null !== self::$promiseAdapters && isset(self::$promiseAdapters[$promise]))) {
+            return $promise;
+        }
+
+        $promiseAdapter = $this->getPromiseAdapter();
+        if ($promiseAdapter->isPromise($promise, true)) {
+            self::$promiseAdapters ??= new \WeakMap();
+            self::$promiseAdapters[$promise] = $promiseAdapter;
+        }
+
+        return $promise;
+    }
+
     /**
      * {@inheritdoc}
      */
@@ -247,23 +286,45 @@ function ($reason) use (&$isPromiseCompleted, &$rejectedReason) {
 
                 return $resolvedValue;
             }
+        } else {
+            throw new \InvalidArgumentException(sprintf('The "%s" method must be called with a Promise ("then" method).', __METHOD__));
+        }
+
+        if (null !== self::$promiseAdapters && isset(self::$promiseAdapters[$promise])) {
+            return self::$promiseAdapters[$promise]->await($promise, $unwrap);
         }
 
-        if (empty(self::$instances)) {
+        $fallbackPromiseAdapter = null;
+        if (null !== self::$instances) {
+            foreach (self::$instances as $dataLoader => $unused) {
+                $promiseAdapter = $dataLoader->getPromiseAdapter();
+                $fallbackPromiseAdapter = $fallbackPromiseAdapter ?: $promiseAdapter;
+
+                if ($promiseAdapter->isPromise($promise, true)) {
+                    return $promiseAdapter->await($promise, $unwrap);
+                }
+            }
+        }
+
+        if (null === $fallbackPromiseAdapter) {
             throw new \RuntimeException('Found no active DataLoader instance.');
         }
 
-        return self::$instances[0]->getPromiseAdapter()->await($promise, $unwrap);
+        return $fallbackPromiseAdapter->await($promise, $unwrap);
     }
 
     private static function awaitInstances()
     {
+        if (empty(self::$activeInstances)) {
+            return;
+        }
+
         do {
             $wait = false;
-            $dataLoaders = self::$instances;
+            $dataLoaders = self::$activeInstances;
 
             foreach ($dataLoaders as $dataLoader) {
-                if (!$dataLoader || !$dataLoader->needProcess()) {
+                if (!$dataLoader->needProcess()) {
                     $wait |= false;
                     continue;
                 }
@@ -305,6 +366,8 @@ protected function checkKey($key, $method)
      */
     private function dispatchQueue()
     {
+        unset(self::$activeInstances[spl_object_id($this)]);
+
         // Take the current loader queue, replacing it with an empty queue.
         $queue = $this->queue;
         $this->queue = [];

tests/DataLoadTestCase.php

@@ -865,18 +865,70 @@ public function testOnDestructionAllPromiseInQueueShouldBeCancelled()
         $loader->load('A1')->then(null, function ($reason) use (&$exception) {
             $exception = $reason;
         });
+
+        $unrelatedLoadCalls = new \ArrayObject();
+        $unrelatedLoader = new DataLoader(function ($keys) use ($unrelatedLoadCalls) {
+            $unrelatedLoadCalls[] = $keys;
+
+            return self::$promiseAdapter->createFulfilled($keys);
+        }, self::$promiseAdapter);
+        $unrelatedPromise = $unrelatedLoader->load('B1');
+
         $loader->__destruct();
         unset($loader);
 
         $this->assertInstanceOf(\RuntimeException::class, $exception);
         $this->assertEquals($exception->getMessage(), 'DataLoader destroyed before promise complete.');
+        $this->assertSame([], $unrelatedLoadCalls->getArrayCopy());
+        $this->assertSame('B1', DataLoader::await($unrelatedPromise));
+    }
+
+    public function testDataLoaderCanBeGarbageCollectedAfterLosingItsLastExternalReference()
+    {
+        $loader = new DataLoader(function ($keys) {
+            return self::$promiseAdapter->createFulfilled($keys);
+        }, self::$promiseAdapter);
+        $loaderReference = \WeakReference::create($loader);
+
+        unset($loader);
+        gc_collect_cycles();
+
+        $this->assertNull($loaderReference->get());
+    }
+
+    public function testDataLoaderIsRetainedWhileWorkIsPendingAndCollectedAfterDispatch()
+    {
+        $loader = new DataLoader(function ($keys) {
+            return self::$promiseAdapter->createFulfilled($keys);
+        }, self::$promiseAdapter);
+        $loaderReference = \WeakReference::create($loader);
+        $promise = $loader->load('A');
+
+        unset($loader);
+        gc_collect_cycles();
+
+        $this->assertNotNull($loaderReference->get());
+        $this->assertSame('A', DataLoader::await($promise));
+
+        gc_collect_cycles();
+
+        $this->assertNull($loaderReference->get());
     }
 
     public function testCallingAwaitFunctionWhenNoInstanceOfDataLoaderShouldNotThrowError()
     {
         self::assertNull(DataLoader::await());
     }
 
+    public function testAwaitSupportsAnInlineTemporaryDataLoader()
+    {
+        $value = DataLoader::await((new DataLoader(function ($keys) {
+            return self::$promiseAdapter->createFulfilled($keys);
+        }, self::$promiseAdapter))->load(1));
+
+        $this->assertSame(1, $value);
+    }
+
     public function testAwaitAlsoAwaitsNewlyCreatedDataloaders()
     {
         $firstComplete = false;

tests/ReactDataLoadTest.php

@@ -11,7 +11,10 @@
 
 namespace Overblog\DataLoader\Test;
 
+use Overblog\DataLoader\CacheMap;
 use Overblog\DataLoader\DataLoader;
+use Overblog\DataLoader\Option;
+use Overblog\PromiseAdapter\Adapter\GuzzleHttpPromiseAdapter;
 use Overblog\PromiseAdapter\Adapter\ReactPromiseAdapter;
 use Overblog\PromiseAdapter\PromiseAdapterInterface;
 
@@ -22,6 +25,67 @@ protected function createPromiseAdapter(): PromiseAdapterInterface
         return new ReactPromiseAdapter();
     }
 
+    public function testAwaitUsesTheAdapterThatRecognizesThePromiseAndDispatchesAllLiveLoaders()
+    {
+        $reactAdapter = new ReactPromiseAdapter();
+        $reactLoadCalls = new \ArrayObject();
+        $reactLoader = new DataLoader(function ($keys) use ($reactAdapter, $reactLoadCalls) {
+            $reactLoadCalls[] = $keys;
+
+            return $reactAdapter->createFulfilled($keys);
+        }, $reactAdapter);
+
+        $guzzleAdapter = new GuzzleHttpPromiseAdapter();
+        $guzzleLoadCalls = new \ArrayObject();
+        $guzzleLoader = new DataLoader(function ($keys) use ($guzzleAdapter, $guzzleLoadCalls) {
+            $guzzleLoadCalls[] = $keys;
+
+            return $guzzleAdapter->createFulfilled($keys);
+        }, $guzzleAdapter);
+
+        $reactLoader->load('react');
+        $guzzlePromise = $guzzleLoader->load('guzzle');
+        $guzzleLoaderReference = \WeakReference::create($guzzleLoader);
+        unset($guzzleLoader);
+
+        $this->assertSame('guzzle', DataLoader::await($guzzlePromise));
+        $this->assertSame([['react']], $reactLoadCalls->getArrayCopy());
+        $this->assertSame([['guzzle']], $guzzleLoadCalls->getArrayCopy());
+
+        gc_collect_cycles();
+
+        $this->assertNull($guzzleLoaderReference->get());
+    }
+
+    public function testForeignCacheHitDoesNotReplaceThePromiseAdapter()
+    {
+        $cacheMap = new CacheMap();
+
+        $guzzleAdapter = new GuzzleHttpPromiseAdapter();
+        $guzzleLoadCalls = new \ArrayObject();
+        $guzzleLoader = new DataLoader(function ($keys) use ($guzzleAdapter, $guzzleLoadCalls) {
+            $guzzleLoadCalls[] = $keys;
+
+            return $guzzleAdapter->createFulfilled($keys);
+        }, $guzzleAdapter, new Option(['cacheMap' => $cacheMap]));
+
+        $reactAdapter = new ReactPromiseAdapter();
+        $reactLoadCalls = new \ArrayObject();
+        $reactLoader = new DataLoader(function ($keys) use ($reactAdapter, $reactLoadCalls) {
+            $reactLoadCalls[] = $keys;
+
+            return $reactAdapter->createFulfilled($keys);
+        }, $reactAdapter, new Option(['cacheMap' => $cacheMap]));
+
+        $guzzlePromise = $guzzleLoader->load('shared');
+        $cachedPromise = $reactLoader->load('shared');
+
+        $this->assertSame($guzzlePromise, $cachedPromise);
+        $this->assertSame('shared', DataLoader::await($cachedPromise));
+        $this->assertSame([['shared']], $guzzleLoadCalls->getArrayCopy());
+        $this->assertSame([], $reactLoadCalls->getArrayCopy());
+    }
+
     /**
      * @runInSeparateProcess
      */