Merge pull request #17 from reactphp-parallel/track-object-instances-in-threads

Track object instances in threads

Author: WyriHaximus

src/AbstractGeneratedProxy.php

@@ -8,6 +8,8 @@
 use ReactParallel\ObjectProxy\Message\Call;
 use ReactParallel\ObjectProxy\Message\Destruct;
 
+use function spl_object_hash;
+
 abstract class AbstractGeneratedProxy
 {
     private Channel $out;
@@ -30,6 +32,7 @@ final protected function proxyCallToMainThread(string $interface, string $method
         $call  = new Call(
             $input,
             $this->hash,
+            spl_object_hash($this),
             $interface,
             $method,
             $args,
@@ -44,7 +47,7 @@ final protected function proxyCallToMainThread(string $interface, string $method
     final protected function notifyMainThreadAboutDestruction(string $interface): void
     {
         try {
-            $this->out->send(new Destruct($this->hash, $interface));
+            $this->out->send(new Destruct($this->hash, spl_object_hash($this), $interface));
         } catch (Channel\Error\Closed $closed) {
             // @ignoreException
         }

src/Message/Call.php

@@ -11,6 +11,7 @@ final class Call
     private Channel $channel;
 
     private string $hash;
+    private string $objectHash;
     private string $interface;
 
     private string $method;
@@ -21,13 +22,14 @@ final class Call
     /**
      * @param mixed[] $args
      */
-    public function __construct(Channel $channel, string $hash, string $interface, string $method, array $args)
+    public function __construct(Channel $channel, string $hash, string $objectHash, string $interface, string $method, array $args)
     {
-        $this->channel   = $channel;
-        $this->hash      = $hash;
-        $this->interface = $interface;
-        $this->method    = $method;
-        $this->args      = $args;
+        $this->channel    = $channel;
+        $this->hash       = $hash;
+        $this->objectHash = $objectHash;
+        $this->interface  = $interface;
+        $this->method     = $method;
+        $this->args       = $args;
     }
 
     public function channel(): Channel
@@ -40,6 +42,11 @@ public function hash(): string
         return $this->hash;
     }
 
+    public function objectHash(): string
+    {
+        return $this->objectHash;
+    }
+
     public function interface(): string
     {
         return $this->interface;

src/Message/Destruct.php

@@ -7,19 +7,26 @@
 final class Destruct
 {
     private string $hash;
+    private string $objectHash;
     private string $interface;
 
-    public function __construct(string $hash, string $interface)
+    public function __construct(string $hash, string $objectHash, string $interface)
     {
-        $this->hash      = $hash;
-        $this->interface = $interface;
+        $this->hash       = $hash;
+        $this->objectHash = $objectHash;
+        $this->interface  = $interface;
     }
 
     public function hash(): string
     {
         return $this->hash;
     }
 
+    public function objectHash(): string
+    {
+        return $this->objectHash;
+    }
+
     public function interface(): string
     {
         return $this->interface;

src/Proxy.php

@@ -9,6 +9,7 @@
 use ReactParallel\ObjectProxy\Generated\ProxyList;
 use ReactParallel\ObjectProxy\Message\Call;
 use ReactParallel\ObjectProxy\Message\Destruct;
+use ReactParallel\ObjectProxy\Proxy\Instance;
 use WyriHaximus\Metrics\Label;
 use WyriHaximus\Metrics\Registry;
 use WyriHaximus\Metrics\Registry\Counters;
@@ -30,7 +31,7 @@ final class Proxy extends ProxyList
 
     private Channel $in;
 
-    /** @var array<string, object> */
+    /** @var array<string, Instance> */
     private array $instances = [];
 
     /** @var array<string, string|false> */
@@ -61,6 +62,7 @@ public function withMetrics(Registry $registry): self
         $self->counterDestruct = $registry->counter(
             'react_parallel_object_proxy_destruct',
             'Number of destroyed proxies by the garbage collector',
+            new Label\Name('class'),
             new Label\Name('interface'),
         );
         $self->in              = new Channel(Channel::Infinite);
@@ -80,14 +82,14 @@ public function create(object $object, string $interface): object
             throw NonExistentInterface::create($interface);
         }
 
-        if ($this->counterCreate instanceof Counters) {
-            $this->counterCreate->counter(new Label('class', get_class($object)), new Label('interface', $interface))->incr();
-        }
-
         $class = self::KNOWN_INTERFACE[$interface];
         $hash  = bin2hex(random_bytes(13));
 
-        $this->instances[$hash] = $object;
+        $this->instances[$hash] = new Instance($object, $interface);
+
+        if ($this->counterCreate instanceof Counters) {
+            $this->counterCreate->counter(new Label('class', $this->instances[$hash]->class()), new Label('interface', $interface))->incr();
+        }
 
         /** @psalm-suppress InvalidStringClass */
         return new $class($this->in, $hash);
@@ -117,13 +119,18 @@ private function setUpHandlers(): void
 
     private function handleCall(Call $call): void
     {
-        $object = $this->instances[$call->hash()];
+        if (! array_key_exists($call->hash(), $this->instances)) {
+            return;
+        }
+
+        $instance = $this->instances[$call->hash()];
+        $instance->reference($call->objectHash());
         if ($this->counterCall instanceof Counters) {
-            $this->counterCall->counter(new Label('class', get_class($object)), new Label('interface', $call->interface()))->incr();
+            $this->counterCall->counter(new Label('class', $instance->class()), new Label('interface', $instance->interface()))->incr();
         }
 
         /** @phpstan-ignore-next-line */
-        $outcome = $object->{$call->method()}(...$call->args());
+        $outcome = $instance->object()->{$call->method()}(...$call->args());
 
         if (is_object($outcome)) {
             $outcomeClass = get_class($outcome);
@@ -142,13 +149,21 @@ private function handleCall(Call $call): void
 
     private function handleDestruct(Destruct $destruct): void
     {
-        unset($this->instances[$destruct->hash()]);
+        if (! array_key_exists($destruct->hash(), $this->instances)) {
+            return;
+        }
+
+        $instance = $this->instances[$destruct->hash()];
+        $count    = $instance->dereference($destruct->objectHash());
+        if ($count === 0) {
+            unset($this->instances[$destruct->hash()]);
+        }
 
         if (! ($this->counterDestruct instanceof Counters)) {
             return;
         }
 
-        $this->counterDestruct->counter(new Label('interface', $destruct->interface()))->incr();
+        $this->counterDestruct->counter(new Label('class', $instance->class()), new Label('interface', $instance->interface()))->incr();
     }
 
     /** @phpstan-ignore-next-line */

src/Proxy/Instance.php

@@ -0,0 +1,52 @@
+<?php
+
+declare(strict_types=1);
+
+namespace ReactParallel\ObjectProxy\Proxy;
+
+use function count;
+use function get_class;
+
+final class Instance
+{
+    private object $object;
+    private string $class;
+    private string $interface;
+
+    /** @var array<string> */
+    private array $references = [];
+
+    public function __construct(object $object, string $interface)
+    {
+        $this->object    = $object;
+        $this->class     = get_class($object);
+        $this->interface = $interface;
+    }
+
+    public function object(): object
+    {
+        return $this->object;
+    }
+
+    public function class(): string
+    {
+        return $this->class;
+    }
+
+    public function interface(): string
+    {
+        return $this->interface;
+    }
+
+    public function reference(string $hash): void
+    {
+        $this->references[$hash] = $hash;
+    }
+
+    public function dereference(string $hash): int
+    {
+        unset($this->references[$hash]);
+
+        return count($this->references);
+    }
+}

tests/Message/CallTest.php

@@ -20,16 +20,18 @@ final class CallTest extends AsyncTestCase
      */
     public function getters(): void
     {
-        $channel   = new Channel(1);
-        $hash      = bin2hex(random_bytes(1024));
-        $interface = LoggerInterface::class;
-        $method    = 'hammer';
-        $args      = [time()];
+        $channel    = new Channel(1);
+        $hash       = bin2hex(random_bytes(1024));
+        $objectHash = bin2hex(random_bytes(1024));
+        $interface  = LoggerInterface::class;
+        $method     = 'hammer';
+        $args       = [time()];
 
-        $call = new Call($channel, $hash, $interface, $method, $args);
+        $call = new Call($channel, $hash, $objectHash, $interface, $method, $args);
 
         self::assertSame($channel, $call->channel());
         self::assertSame($hash, $call->hash());
+        self::assertSame($objectHash, $call->objectHash());
         self::assertSame($interface, $call->interface());
         self::assertSame($method, $call->method());
         self::assertSame($args, $call->args());

tests/ProxyTest.php

@@ -149,6 +149,6 @@ public function metricsDestructionTesting(): void
         self::assertStringContainsString('react_parallel_object_proxy_call_total{class="WyriHaximus\Metrics\InMemory\Registry",interface="WyriHaximus\Metrics\Registry"} 101', $txt);
         self::assertStringContainsString('react_parallel_object_proxy_call_total{class="WyriHaximus\Metrics\InMemory\Registry\Counters",interface="WyriHaximus\Metrics\Registry\Counters"} 101', $txt);
         self::assertStringContainsString('react_parallel_object_proxy_call_total{class="WyriHaximus\Metrics\InMemory\Counter",interface="WyriHaximus\Metrics\Counter"} 101', $txt);
-        self::assertStringContainsString('aaareact_parallel_object_proxy_call_total{class="WyriHaximus\Metrics\InMemory\Counter",interface="WyriHaximus\Metrics\Counter"} 101', $txt);
+        self::assertStringContainsString('react_parallel_object_proxy_destruct_total{class="WyriHaximus\Metrics\InMemory\Registry",interface="WyriHaximus\Metrics\Registry"} 13', $txt);
     }
 }