♻️ Rollback start and stop to DriverInterface

darkwood-com · Sep 10, 2023 · 6948911 · 6948911
1 parent 8bc3f2d
commit 6948911
Show file tree

Hide file tree

Showing 17 changed files with 184 additions and 114 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -11,7 +11,6 @@
 - Update DX for Flow\DriverInterface
   - Update `async` that now $onResolve get called with async $callback result or Flow\Exception as first argument
   - Update `tick` that now return a Closure to cleanup tick interval
-  - Remove `start` and `stop`
 - Add Flow\Driver\FiberDriver from https://github.com/jolicode/castor/blob/main/src/functions.php
 - Upgrade to Symfony 6.3 and PHPUnit 10.3
 - Refactor docs structure

diff --git a/examples/flow.php b/examples/flow.php
@@ -17,8 +17,8 @@
 
 $driver = match (random_int(1, 5)) {
     1 => new AmpDriver(),
-    2 => new ReactDriver(),
-    3 => new FiberDriver(),
+    2 => new FiberDriver(),
+    3 => new ReactDriver(),
     4 => new RevoltDriver(),
     5 => new SwooleDriver(),
 };
@@ -28,7 +28,7 @@
     printf("*. #%d - Job 1 : Calculating %d + %d\n", $data->id, $data->number, $data->number);
 
     // simulating calculating some "light" operation from 0.1 to 1 seconds
-    $delay = random_int(1, 10) / 10;
+    $delay = random_int(1, 3);
     $driver->delay($delay);
     $result = $data->number;
     $result += $result;
@@ -88,5 +88,13 @@
 
 for ($i = 1; $i <= 5; $i++) {
     $ip = new Ip(new Data($i, $i));
+    $ipPool->offsetSet($ip, true);
     $flow($ip, static fn ($ip) => $ipPool->offsetUnset($ip));
 }
+
+$driver->tick(1, static function () use ($driver, $ipPool) {
+    if ($ipPool->count() === 0) {
+        $driver->stop();
+    }
+});
+$driver->start();
diff --git a/examples/server.php b/examples/server.php
@@ -17,8 +17,8 @@
 
 $driver = match (random_int(1, 5)) {
     1 => new AmpDriver(),
-    2 => new ReactDriver(),
-    3 => new FiberDriver(),
+    2 => new FiberDriver(),
+    3 => new ReactDriver(),
     4 => new RevoltDriver(),
     5 => new SwooleDriver(),
 };

diff --git a/src/Driver/AmpDriver.php b/src/Driver/AmpDriver.php
@@ -23,6 +23,11 @@
  */
 class AmpDriver implements DriverInterface
 {
+    /**
+     * @var array<string>
+     */
+    private array $ticksIds = [];
+
     public function __construct()
     {
         if (!function_exists('Amp\\async')) {
@@ -44,7 +49,7 @@ public function async(Closure $callback, Closure $onResolve = null): Closure
                         $onResolve(new RuntimeException($exception->getMessage(), $exception->getCode(), $exception));
                     }
                 }
-            }, $callback, $args, $onResolve)->await();
+            }, $callback, $args, $onResolve);
         };
     }
 
@@ -57,8 +62,29 @@ public function tick(int $interval, Closure $callback): Closure
     {
         $tickId = EventLoop::repeat($interval / 1000, $callback);
 
-        return static function () use ($tickId) {
+        $cancel = function () use ($tickId) {
+            unset($this->ticksIds[$tickId]);
             EventLoop::cancel($tickId);
         };
+
+        $this->ticksIds[$tickId] = $cancel;
+
+        return $cancel;
+    }
+
+    public function start(): void
+    {
+        if (!EventLoop::getDriver()->isRunning()) {
+            EventLoop::run();
+        }
+    }
+
+    public function stop(): void
+    {
+        foreach ($this->ticksIds as $cancel) {
+            $cancel();
+        }
+
+        EventLoop::getDriver()->stop();
     }
 }
diff --git a/src/Driver/FiberDriver.php b/src/Driver/FiberDriver.php
@@ -25,6 +25,11 @@ class FiberDriver implements DriverInterface
      */
     private array $fibers = [];
 
+    /**
+     * @var array<string>
+     */
+    private array $ticksIds = [];
+
     private bool $isLooping = false;
 
     public function async(Closure $callback, Closure $onResolve = null): Closure
@@ -45,7 +50,6 @@ public function async(Closure $callback, Closure $onResolve = null): Closure
             }
 
             $this->fibers[] = $fiberData;
-            $this->loop();
         };
     }
 
@@ -59,50 +63,61 @@ public function delay(float $seconds): void
      */
     public function tick(int $interval, Closure $callback): Closure
     {
-        $closure = static fn () => $callback();
+        $tickId = uniqid('flow_fiber_tick_id');
 
+        $closure = static fn () => $callback();
         register_tick_function($closure);
 
-        return static fn () => unregister_tick_function($closure);
+        $cancel = function () use ($tickId, $closure) {
+            unset($this->ticksIds[$tickId]);
+            unregister_tick_function($closure);
+        };
+
+        return $cancel;
     }
 
-    private function loop(): void
+    public function start(): void
     {
-        if ($this->isLooping === false) {
-            $this->isLooping = true;
+        $this->isLooping = true;
 
-            $isRunning = true;
+        $isRunning = true;
 
-            while ($isRunning) {
-                $isRunning = false;
+        while ($this->isLooping || $isRunning) { /** @phpstan-ignore-line */
+            $isRunning = false;
 
-                foreach ($this->fibers as $i => $fiber) {
-                    $isRunning = $isRunning || !$fiber['fiber']->isTerminated();
+            foreach ($this->fibers as $i => $fiber) {
+                $isRunning = $isRunning || !$fiber['fiber']->isTerminated();
 
-                    if (!$fiber['fiber']->isTerminated() and $fiber['fiber']->isSuspended()) {
-                        try {
-                            $fiber['fiber']->resume();
-                        } catch (Throwable $exception) {
-                            $this->fibers[$i]['exception'] = $exception;
-                        }
-                    } else {
-                        if ($fiber['onResolve']) {
-                            if ($fiber['exception'] === null) {
-                                $fiber['onResolve']($fiber['fiber']->getReturn());
-                            } else {
-                                $fiber['onResolve'](new RuntimeException($fiber['exception']->getMessage(), $fiber['exception']->getCode(), $fiber['exception']));
-                            }
+                if (!$fiber['fiber']->isTerminated() and $fiber['fiber']->isSuspended()) {
+                    try {
+                        $fiber['fiber']->resume();
+                    } catch (Throwable $exception) {
+                        $this->fibers[$i]['exception'] = $exception;
+                    }
+                } else {
+                    if ($fiber['onResolve']) {
+                        if ($fiber['exception'] === null) {
+                            $fiber['onResolve']($fiber['fiber']->getReturn());
+                        } else {
+                            $fiber['onResolve'](new RuntimeException($fiber['exception']->getMessage(), $fiber['exception']->getCode(), $fiber['exception']));
                         }
-                        unset($this->fibers[$i]);
                     }
+                    unset($this->fibers[$i]);
                 }
+            }
 
-                if (Fiber::getCurrent()) {
-                    Fiber::suspend();
-                    usleep(1_000);
-                }
+            if (Fiber::getCurrent()) {
+                Fiber::suspend();
+                usleep(1_000);
             }
         }
+    }
+
+    public function stop(): void
+    {
+        foreach ($this->ticksIds as $cancel) {
+            $cancel();
+        }
 
         $this->isLooping = false;
     }

diff --git a/src/Driver/ReactDriver.php b/src/Driver/ReactDriver.php
@@ -26,7 +26,10 @@ class ReactDriver implements DriverInterface
 {
     private LoopInterface $eventLoop;
 
-    private int $counter = 0;
+    /**
+     * @var array<string>
+     */
+    private array $ticksIds = [];
 
     public function __construct(LoopInterface $eventLoop = null)
     {
@@ -39,8 +42,8 @@ public function __construct(LoopInterface $eventLoop = null)
 
     public function async(Closure $callback, Closure $onResolve = null): Closure
     {
-        return function (...$args) use ($callback, $onResolve): void {
-            async(function () use ($callback, $onResolve, $args) {
+        return static function (...$args) use ($callback, $onResolve): void {
+            async(static function () use ($callback, $onResolve, $args) {
                 try {
                     $return = $callback(...$args, ...($args = []));
                     if ($onResolve) {
@@ -50,11 +53,8 @@ public function async(Closure $callback, Closure $onResolve = null): Closure
                     if ($onResolve) {
                         $onResolve(new RuntimeException($exception->getMessage(), $exception->getCode(), $exception));
                     }
-                } finally {
-                    $this->pop();
                 }
             })();
-            $this->push();
         };
     }
 
@@ -65,28 +65,27 @@ public function delay(float $seconds): void
 
     public function tick(int $interval, Closure $callback): Closure
     {
-        $tickId = $this->eventLoop->addPeriodicTimer($interval, $callback);
-        $this->push();
+        $tickId = uniqid('flow_react_tick_id');
 
-        return function () use ($tickId) {
-            $this->eventLoop->cancelTimer($tickId);
-            $this->pop();
+        $timer = $this->eventLoop->addPeriodicTimer($interval, $callback);
+
+        return function () use ($tickId, $timer) {
+            unset($this->ticksIds[$tickId]);
+            $this->eventLoop->cancelTimer($timer);
         };
     }
 
-    private function push(): void
+    public function start(): void
     {
-        if ($this->counter === 0) {
-            $this->eventLoop->run();
-        }
-        $this->counter++;
+        $this->eventLoop->run();
     }
 
-    private function pop(): void
+    public function stop(): void
     {
-        $this->counter--;
-        if ($this->counter === 0) {
-            $this->eventLoop->stop();
+        foreach ($this->ticksIds as $cancel) {
+            $cancel();
         }
+
+        $this->eventLoop->stop();
     }
 }
diff --git a/src/Driver/RevoltDriver.php b/src/Driver/RevoltDriver.php
@@ -20,7 +20,10 @@
  */
 class RevoltDriver implements DriverInterface
 {
-    private int $counter = 0;
+    /**
+     * @var array<string>
+     */
+    private array $ticksIds = [];
 
     public function __construct(Driver $driver = null)
     {
@@ -35,8 +38,8 @@ public function __construct(Driver $driver = null)
 
     public function async(Closure $callback, Closure $onResolve = null): Closure
     {
-        return function (...$args) use ($callback, $onResolve): void {
-            EventLoop::queue(function (Closure $callback, array $args, Closure $onResolve = null) {
+        return static function (...$args) use ($callback, $onResolve): void {
+            EventLoop::queue(static function (Closure $callback, array $args, Closure $onResolve = null) {
                 try {
                     $return = $callback(...$args, ...($args = []));
                     if ($onResolve) {
@@ -46,11 +49,8 @@ public function async(Closure $callback, Closure $onResolve = null): Closure
                     if ($onResolve) {
                         $onResolve(new RuntimeException($exception->getMessage(), $exception->getCode(), $exception));
                     }
-                } finally {
-                    $this->pop();
                 }
             }, $callback, $args, $onResolve);
-            $this->push();
         };
     }
 
@@ -69,27 +69,26 @@ public function delay(float $seconds): void
     public function tick(int $interval, Closure $callback): Closure
     {
         $tickId = EventLoop::repeat($interval / 1000, $callback);
-        $this->push();
 
-        return function () use ($tickId) {
+        $cancel = function () use ($tickId) {
+            unset($this->ticksIds[$tickId]);
             EventLoop::cancel($tickId);
-            $this->pop();
         };
+
+        $this->ticksIds[$tickId] = $cancel;
+
+        return $cancel;
     }
 
-    private function push(): void
+    public function start(): void
     {
-        if (/* $this->counter === 0 || */ !EventLoop::getDriver()->isRunning()) {
+        if (!EventLoop::getDriver()->isRunning()) {
             EventLoop::run();
         }
-        $this->counter++;
     }
 
-    private function pop(): void
+    public function stop(): void
     {
-        $this->counter--;
-        if ($this->counter === 0) {
-            EventLoop::getDriver()->stop();
-        }
+        EventLoop::getDriver()->stop();
     }
 }
diff --git a/src/Driver/SwooleDriver.php b/src/Driver/SwooleDriver.php
@@ -22,6 +22,11 @@
  */
 class SwooleDriver implements DriverInterface
 {
+    /**
+     * @var array<int, callable(): void>
+     */
+    private array $ticksIds = [];
+
     public function __construct()
     {
         if (!extension_loaded('openswoole')) {
@@ -59,8 +64,20 @@ public function tick(int $interval, Closure $callback): Closure
     {
         $tickId = Timer::tick($interval, $callback);
 
-        return static function () use ($tickId) {
+        return function () use ($tickId) {
+            unset($this->ticksIds[$tickId]);
             Timer::clear($tickId); // @phpstan-ignore-line
         };
     }
+
+    public function start(): void
+    {
+    }
+
+    public function stop(): void
+    {
+        foreach ($this->ticksIds as $cancel) {
+            $cancel();
+        }
+    }
 }