update: opz the main co state

Author: cclilshy

examples/08-process.php

@@ -1,4 +1,5 @@
 <?php declare(strict_types=1);
+
 /**
  * Copyright © 2024 cclilshy
  * Email: jingnigg@gmail.com
@@ -13,6 +14,8 @@
 use Ripple\Process;
 use Ripple\Time;
 
+use function Co\go;
+
 require_once __DIR__ . '/../vendor/autoload.php';
 
 
@@ -22,8 +25,24 @@
 
 $child = Process::fork(function () {
     \var_dump('is child');
-    Time::sleep(10);
+    Time::sleep(1);
     exit(127);
 });
 
+echo "child pid > {$child} \n";
+
 \var_dump(Process::wait($child));
+
+go(function () {
+    $child = Process::fork(function () {
+        \var_dump('is child');
+        Time::sleep(1);
+        exit(127);
+    });
+
+    echo "child pid > {$child} \n";
+
+    \var_dump(Process::wait($child));
+});
+
+\Co\wait();

src/Coroutine.php

@@ -101,7 +101,6 @@ abstract public function start(): mixed;
      * 暂停协程执行
      * @param mixed $value 暂停时传递的值
      * @return mixed 恢复时传入的值
-     * @throws Throwable 当协程暂停过程中发生异常时抛出
      */
     abstract public function suspend(mixed $value = null): mixed;
 

src/Process.php

@@ -14,6 +14,7 @@
 
 use Closure;
 use Ripple\Runtime\Exception\CoroutineStateException;
+use Ripple\Runtime\MainCoroutine;
 use Ripple\Runtime\Scheduler;
 use Ripple\Runtime\Support\Stdin;
 use Throwable;
@@ -85,46 +86,58 @@ public static function fork(Closure $callback): int
         }
 
         $owner = \Co\current();
+        if ($owner instanceof MainCoroutine) {
+            return self::spawn($callback);
+        }
+
         Scheduler::nextTick(static function () use ($callback, $owner) {
-            // 子进程执行
-            $pid = pcntl_fork();
+            Scheduler::resume($owner, self::spawn($callback));
+        });
 
-            if ($pid === -1 || $pid > 0) {
-                Scheduler::resume($owner, $pid);
-                return;
-            }
+        try {
+            return $owner->suspend();
+        } catch (Throwable $e) {
+            throw new RuntimeException($e->getMessage(), $e->getCode(), $e);
+        }
+    }
 
-            // 清理调度器和事件监听器
-            Scheduler::clear();
-            Runtime::watcher()->forked();
+    /**
+     * @param Closure $callback
+     * @return int
+     */
+    private static function spawn(Closure $callback): int
+    {
+        // 子进程执行
+        $pid = pcntl_fork();
 
-            // 保存并清空 fork 回调列表
-            $forked = self::$forked;
-            self::$forked = [];
-            self::$children = [];
+        if ($pid === -1 || $pid > 0) {
+            return $pid;
+        }
 
-            // 执行所有 fork 回调
-            foreach ($forked as $forkedCallback) {
-                call_user_func($forkedCallback);
-            }
+        // 清理调度器和事件监听器
+        Scheduler::clear();
+        Runtime::watcher()->forked();
 
-            // 执行用户回调
-            try {
-                $callback();
-            } catch (Throwable $e) {
-                Stdin::println($e->getMessage());
-            }
+        // 保存并清空 fork 回调列表
+        $forked = self::$forked;
+        self::$forked = [];
+        self::$children = [];
 
-            // 等待所有协程完成并退出
-            wait();
-            exit(0);
-        });
+        // 执行所有 fork 回调
+        foreach ($forked as $forkedCallback) {
+            call_user_func($forkedCallback);
+        }
 
+        // 执行用户回调
         try {
-            return $owner->suspend();
+            $callback();
         } catch (Throwable $e) {
-            throw new RuntimeException($e->getMessage(), $e->getCode(), $e);
+            Stdin::println($e->getMessage());
         }
+
+        // 等待所有协程完成并退出
+        wait();
+        exit(0);
     }
 
     /**

src/Runtime.php

@@ -93,15 +93,6 @@ public static function watcher(): WatchAbstract
         return self::$watcher;
     }
 
-    /**
-     * 获取事件桥
-     * @return Bridge
-     */
-    public static function bridge(): Bridge
-    {
-        return self::$bridge;
-    }
-
     /**
      * 运行回调并进入主循环
      * @param Closure|null $callback

src/Runtime/MainCoroutine.php

@@ -13,7 +13,6 @@
 namespace Ripple\Runtime;
 
 use Closure;
-use Ripple\Runtime\Exception\CoroutineStateException;
 use Ripple\Coroutine;
 use Ripple\Watch\Interface\WatchAbstract;
 use Throwable;
@@ -37,6 +36,11 @@ final class MainCoroutine extends Coroutine
      */
     private bool $hasMessage = false;
 
+    /**
+     * @var mixed
+     */
+    private mixed $suspendValue = null;
+
     /**
      * 设置主协程为可运行状态
      * @return void
@@ -52,6 +56,7 @@ public function runnable(): void
      */
     public function start(): bool
     {
+        $this->setState(Coroutine::STATE_WAITING);
         call_user_func($this->callback);
         return true;
     }
@@ -64,45 +69,60 @@ public function start(): bool
      */
     public function suspend(mixed $value = null): mixed
     {
-        if ($this->state === Coroutine::STATE_CREATED) {
-            Scheduler::start($this);
-            return $this->result;
-        }
-
-        if ($this->state === Coroutine::STATE_RUNNABLE) {
-            Scheduler::tick();
-            return $this->result;
-        }
+        $this->suspendValue = $value;
+        try {
+            if ($this->state === Coroutine::STATE_CREATED) {
+                Scheduler::start($this);
+                $this->suspendValue = null;
+                return $this->result;
+            }
 
-        if ($this->state === Coroutine::STATE_RUNNING) {
+            $this->setState(Coroutine::STATE_WAITING);
             Scheduler::tick();
+            $this->suspendValue = null;
             return $this->result;
+        } finally {
+            $this->setState(Coroutine::STATE_RUNNING);
         }
-
-        throw new CoroutineStateException('Cannot suspend a main coroutine in the current state.');
     }
 
     /**
      * 恢复主协程执行
      * @param ?mixed $value 恢复时传入的值
      * @return bool 恢复是否成功
+     * @throws Throwable
      */
-    public function resume(mixed $value = null): bool
+    public function resume(mixed $value = null): mixed
     {
         $this->hasMessage = true;
         $this->result = $value;
+
+        $owner = \Co\current();
+        if ($owner instanceof FiberCoroutine) {
+            Scheduler::nextTick(fn () => Scheduler::resume($owner, $this->suspendValue));
+            return $owner->suspend();
+        }
+
         return true;
     }
 
     /**
      * 向主协程抛出异常
      * @param Throwable $exception 要抛出的异常对象
      * @return bool 异常设置是否成功
+     * @throws Throwable
      */
-    public function throw(Throwable $exception): bool
+    public function throw(Throwable $exception): mixed
     {
         $this->hasMessage = true;
         $this->hasExcept = $exception;
+
+        $owner = \Co\current();
+        if ($owner instanceof FiberCoroutine) {
+            Scheduler::nextTick(fn () => Scheduler::resume($owner, $this->suspendValue));
+            return $owner->suspend();
+        }
+
         return true;
     }
 

src/Stream.php

@@ -253,7 +253,7 @@ public function flush(float|int|null $timeout = null): void
         $timer = null;
         if ($timeout && $timeout >= 0) {
             $timer = Time::afterFunc($timeout, function () use ($timeout) {
-                $this->wco->isSuspended() && Scheduler::throw($this->wco, new ConnectionException('Write timeout'));
+                Scheduler::throw($this->wco, new ConnectionException('Write timeout'));
             });
         }
 

src/Sync/Channel.php

@@ -185,7 +185,6 @@ public function receive(): mixed
             return $owner->suspend();
         } catch (Throwable $e) {
             throw new RuntimeException($e->getMessage(), $e->getCode(), $e);
-
         }
     }
 

tests/Sync/ChannelTest.php

@@ -255,8 +255,6 @@ public function testMultipleSendersReceivers(): void
      */
     public function testChannelFIFO(): void
     {
-
-
         $channel = new Channel(3);
         $received = [];
 
@@ -273,7 +271,7 @@ public function testChannelFIFO(): void
             }
         });
 
-        Time::sleep(0.1);
+        \Co\wait();
 
         $this->assertCount(5, $received);
         $this->assertEquals("Message 1", $received[0]);