🐛 debug

examples/yflow.php

@@ -17,7 +17,7 @@
 use Flow\Job\YJob;
 use Flow\JobInterface;
 
-$driver = match (random_int(1, 4)) {
+$driver = match (random_int(3, 3)) {
     1 => new AmpDriver(),
     2 => new ReactDriver(),
     3 => new FiberDriver(),
@@ -162,15 +162,15 @@ function factorialYMemo(int $n): int
 ) {
     yield [$factorialJob];
     yield [$factorialYJobBefore];
-    yield new YFlow($factorialYJob);
-    yield [$factorialYJobAfter];
-    yield [$factorialYMemoJob];
-    yield [$factorialYJobDeferBefore];
-    yield [$factorialYJobDefer, null, null, null, new DeferAsyncHandler()];
-    yield [$factorialYJobDeferAfter, null, null, null, new DeferAsyncHandler()];
+    //yield new YFlow($factorialYJob);
+    //yield [$factorialYJobAfter];
+    //yield [$factorialYMemoJob];
+    //yield [$factorialYJobDeferBefore];
+    //yield [$factorialYJobDefer, null, null, null, new DeferAsyncHandler()];
+    //yield [$factorialYJobDeferAfter, null, null, null, new DeferAsyncHandler()];
 }, ['driver' => $driver]);
 
-for ($i = 1; $i <= 5; $i++) {
+for ($i = 1; $i <= 1; $i++) {
     $ip = new Ip(new YFlowData($i, $i));
     $flow($ip);
 }

src/AsyncHandler/AsyncHandler.php

@@ -42,6 +42,8 @@ public function async(AsyncEvent $event): void
         $asyncJob = $async($event->getJob());
 
         $popIp = $this->ipPool->addIp($ip);
+
+        printf('async'."\n");
         $next = $asyncJob($ip->data);
         $next(static function ($data) use ($event, $popIp) {
             $event->getCallback()($data);
@@ -51,6 +53,7 @@ public function async(AsyncEvent $event): void
 
     public function pool(PoolEvent $event): void
     {
+
         $event->setIps($this->ipPool->getIps());
     }
 }

src/Driver/FiberDriver.php

@@ -138,7 +138,9 @@ public function await(array &$stream): void
             do {
                 foreach ($stream['dispatchers'] as $index => $dispatcher) {
                     $nextIp = $dispatcher->dispatch(new PullEvent(), Event::PULL)->getIp();
+
                     if ($nextIp !== null) {
+                        printf('nextip '.$this->countIps($stream['dispatchers'])."\n");
                         $stream['dispatchers'][$index]->dispatch(new AsyncEvent(static function (Closure|JobInterface $job) use ($async) {
                             return $async(false)($job);
                         }, static function (Closure|JobInterface $job) use ($defer) {
@@ -168,6 +170,7 @@ public function await(array &$stream): void
             }
 
             $tick++;
+            printf('count '.$this->countIps($stream['dispatchers'])."\n");
         } while ($this->countIps($stream['dispatchers']) > 0 or count($this->ticks) > 0);
     }
 

src/IpStrategy/LinearIpStrategy.php

@@ -37,6 +37,7 @@ public static function getSubscribedEvents(): array
      */
     public function push(PushEvent $event): void
     {
+        printf('push'."\n");
         $this->ips[] = $event->getIp();
     }
 
@@ -45,6 +46,7 @@ public function push(PushEvent $event): void
      */
     public function pull(PullEvent $event): void
     {
+        printf('pull'."\n");
         $event->setIp(array_shift($this->ips));
     }