diff --git a/src/Core/Process/Process.php b/src/Core/Process/Process.php index 3cdd494..798f6ca 100644 --- a/src/Core/Process/Process.php +++ b/src/Core/Process/Process.php @@ -48,7 +48,7 @@ use function call_user_func; use function P\cancel; use function P\cancelAll; -use function P\defer; +use function P\getIdentities; use function P\promise; use function P\tick; use function pcntl_fork; @@ -235,16 +235,11 @@ public function task(Closure $closure): Task|false /** * It is necessary to ensure that the final closure cannot be escaped by any means. */ - - // forget all events cancelAll(); $this->forked(); - // mouth user func - defer(function () use ($closure, $args) { - call_user_func($closure, ...$args); - }); + call_user_func($closure, ...$args); // Whether it belongs to the PRipple coroutine space if(Coroutine::Coroutine()->isCoroutine()) { @@ -256,20 +251,6 @@ public function task(Closure $closure): Task|false tick(); exit(0); } - - // // call user mount - // try { - // call_user_func($closure, ...$args); - // } catch (Throwable) { - // exit(1); - // } - // - // // Determine whether the event list is empty - // if(count(getIdentities()) === 0) { - // exit(0); - // } - // - // throw new EscapeException('The process is abnormal.'); } if(empty($this->process2runtime)) { diff --git a/src/Worker/Worker.php b/src/Worker/Worker.php index 075d439..91cbdfc 100644 --- a/src/Worker/Worker.php +++ b/src/Worker/Worker.php @@ -43,6 +43,7 @@ use Psc\Utils\Output; use Psc\Utils\Serialization\Zx7e; +use function P\delay; use function P\promise; use function socket_create_pair; use function socket_export_stream; @@ -192,7 +193,7 @@ public function commandToAll(Command $command): void public function syncId(): Promise { return promise(function (Closure $resolve, Closure $reject) { - $id = spl_object_hash($resolve); + $id = spl_object_hash($resolve); $command = Command::make(Worker::COMMAND_SYNC_ID, ['id' => $id]); $this->command($command); @@ -217,10 +218,10 @@ private function _onCommand(Command $workerCommand): void $this->onReload(); break; case Worker::COMMAND_SYNC_ID: - $id = $workerCommand->arguments['id']; + $id = $workerCommand->arguments['id']; $sync = $workerCommand->arguments['sync']; - if($callback = $this->queue[$id] ?? null) { + if ($callback = $this->queue[$id] ?? null) { unset($this->queue[$id]); $callback['resolve']($sync); } @@ -305,10 +306,12 @@ private function guard(Manager $manager, int $index): bool unset($this->streams[$index]); } - if(isset($this->runtimes[$index])) { + if (isset($this->runtimes[$index])) { unset($this->runtimes[$index]); } - $this->guard($manager, $index); + \P\delay(function () use ($manager, $index) { + $this->guard($manager, $index); + }, 0.1); }); return true; }