Merge pull request #25 from reactphp-parallel/destruction-edge-cases

Destruction edge cases

Author: WyriHaximus

composer-require-checker.json

@@ -7,7 +7,8 @@
     "WyriHaximus\\Constants\\Numeric\\ZERO", "WyriHaximus\\getIn", "WyriHaximus\\Constants\\Numeric\\TWO",
     "Composer\\Composer", "Composer\\Config", "Composer\\IO\\IOInterface", "Composer\\Script\\Event",
     "Composer\\EventDispatcher\\EventSubscriberInterface", "Composer\\Package\\RootPackageInterface",
-    "Composer\\Plugin\\PluginInterface", "Composer\\Script\\ScriptEvents"
+    "Composer\\Plugin\\PluginInterface", "Composer\\Script\\ScriptEvents",
+    "Safe\\chmod", "Safe\\file_get_contents", "Safe\\file_put_contents", "Safe\\sprintf"
   ],
   "php-core-extensions" : [
     "Core",

composer.json

@@ -21,14 +21,14 @@
         "psr/log": "^1.1",
         "react-parallel/react-parallel": "^1.0",
         "rx/operator-extras": "^2.1",
-        "thecodingmachine/safe": "^1.2",
         "wyrihaximus/constants": "^1.5",
         "wyrihaximus/iterator-or-array-to-array": "^1.1",
         "wyrihaximus/metrics": "dev-master",
         "wyrihaximus/string-get-in": "^1.0"
     },
     "require-dev": {
         "monolog/monolog": "^2.1",
+        "thecodingmachine/safe": "^1.3",
         "wyrihaximus/async-test-utilities": "^2.3",
         "yuloh/container": "^1.0"
     },

composer.lock

@@ -7,6 +7,7 @@
 use parallel\Channel;
 use ReactParallel\ObjectProxy\Message\Call;
 use ReactParallel\ObjectProxy\Message\Destruct;
+use ReactParallel\ObjectProxy\Message\Existence;
 use ReactParallel\ObjectProxy\Message\Notify;
 
 use function spl_object_hash;
@@ -65,4 +66,13 @@ final protected function notifyMainThreadAboutDestruction(): void
             // @ignoreException
         }
     }
+
+    final public function notifyMainThreadAboutOurExistence(): void
+    {
+        try {
+            $this->out->send(new Existence($this->hash, spl_object_hash($this)));
+        } catch (Channel\Error\Closed $closed) {
+            // @ignoreException
+        }
+    }
 }

src/AbstractGeneratedProxy.php

@@ -0,0 +1,27 @@
+<?php
+
+declare(strict_types=1);
+
+namespace ReactParallel\ObjectProxy\Message;
+
+final class Existence
+{
+    private string $hash;
+    private string $objectHash;
+
+    public function __construct(string $hash, string $objectHash)
+    {
+        $this->hash       = $hash;
+        $this->objectHash = $objectHash;
+    }
+
+    public function hash(): string
+    {
+        return $this->hash;
+    }
+
+    public function objectHash(): string
+    {
+        return $this->objectHash;
+    }
+}

src/Message/Existence.php

@@ -9,6 +9,7 @@
 use ReactParallel\ObjectProxy\Generated\ProxyList;
 use ReactParallel\ObjectProxy\Message\Call;
 use ReactParallel\ObjectProxy\Message\Destruct;
+use ReactParallel\ObjectProxy\Message\Existence;
 use ReactParallel\ObjectProxy\Message\Notify;
 use ReactParallel\ObjectProxy\Proxy\Instance;
 use WyriHaximus\Metrics\Label;
@@ -20,6 +21,7 @@
 use function get_class;
 use function is_object;
 use function random_bytes;
+use function var_export;
 
 final class Proxy extends ProxyList
 {
@@ -111,12 +113,22 @@ public function __destruct()
     private function setUpHandlers(): void
     {
         $this->factory->streams()->channel($this->in)->subscribe(function (object $message): void {
+            if ($message instanceof Existence) {
+                $this->handleExistence($message);
+
+                return;
+            }
+
             if ($message instanceof Notify) {
                 $this->handleNotify($message);
+
+                return;
             }
 
             if ($message instanceof Call) {
                 $this->handleCall($message);
+
+                return;
             }
 
             if (! ($message instanceof Destruct)) {
@@ -129,6 +141,18 @@ private function setUpHandlers(): void
         });
     }
 
+    private function handleExistence(Existence $existence): void
+    {
+        if (! array_key_exists($existence->hash(), $this->instances)) {
+            var_export($existence);
+
+            return;
+        }
+
+        $instance = $this->instances[$existence->hash()];
+        $instance->reference($existence->objectHash());
+    }
+
     private function handleNotify(Notify $notify): void
     {
         if (! array_key_exists($notify->hash(), $this->instances)) {

src/Proxy.php

@@ -12,6 +12,7 @@
 use React\Promise\PromiseInterface;
 use React\Promise\Timer\TimeoutException;
 use ReactParallel\Factory;
+use ReactParallel\ObjectProxy\Generated\WyriHaximus__Metrics_RegistryProxy;
 use ReactParallel\ObjectProxy\NonExistentInterface;
 use ReactParallel\ObjectProxy\Proxy;
 use stdClass;
@@ -26,6 +27,7 @@
 use function explode;
 use function range;
 use function React\Promise\all;
+use function Safe\sleep;
 use function time;
 
 final class ProxyTest extends AsyncTestCase
@@ -169,4 +171,46 @@ public function metricsDestructionTesting(): void
             (int) explode("\n", explode('react_parallel_object_proxy_destruct_total{class="WyriHaximus\Metrics\InMemory\Registry",interface="WyriHaximus\Metrics\Registry"}', $txt)[1])[0]
         );
     }
+
+    /**
+     * @test
+     */
+    public function destructionEdgeCases(): void
+    {
+        $loop          = EventLoopFactory::create();
+        $factory       = new Factory($loop);
+        $registry      = new InMemmoryRegistry();
+        $proxy         = (new Proxy($factory))->withMetrics($registry);
+        $limitedPool   = $factory->limitedPool(13);
+        $registryProxy = $proxy->create($registry, Registry::class);
+        $fn            = static function (int $int, WyriHaximus__Metrics_RegistryProxy $registryProxy, int $sleep): int {
+            $registryProxy->notifyMainThreadAboutOurExistence();
+            sleep($sleep);
+            $registryProxy->counter('counter', 'bla bla bla', new Label\Name('name'))->counter(new Label('name', 'value'))->incr();
+
+            return $int;
+        };
+
+        $promises = [];
+        foreach (range(0, 3) as $i) {
+            $promises[] = $limitedPool->run($fn, [$i, $registryProxy, 0]);
+        }
+
+        $leet = $this->await(
+            // @phpstan-ignore-next-line
+            all($promises)->then(static function (array $v) use ($factory): PromiseInterface {
+                return new Promise(static function (callable $resolve) use ($v, $factory): void {
+                    $factory->loop()->addTimer(3, static function () use ($resolve, $v): void {
+                        $resolve($v);
+                    });
+                });
+            })->then(static function () use ($fn, $registryProxy, $limitedPool): PromiseInterface {
+                return $limitedPool->run($fn, [1337, $registryProxy, 20]);
+            })->always(static function () use ($limitedPool): void {
+                $limitedPool->close();
+            }),
+            $loop
+        );
+        self::assertSame(1337, $leet);
+    }
 }