eee

Abstracts/HotReload.php

@@ -0,0 +1,171 @@
+<?php
+
+namespace Kiri\Server\Abstracts;
+
+use Kiri\Di\Inject\Container;
+use Kiri\Error\StdoutLogger;
+use Kiri\Events\EventProvider;
+use Kiri\Router\Router;
+use Kiri\Server\Events\OnWorkerStart;
+use Kiri\Server\ServerInterface;
+use Psr\Log\LoggerInterface;
+use Swoole\Event;
+use Swoole\Process;
+use Kiri\Server\Processes\AbstractProcess;
+
+class HotReload extends AbstractProcess
+{
+
+
+    /**
+     * @var mixed
+     */
+    protected mixed $pipe;
+
+
+    /**
+     * @var LoggerInterface|StdoutLogger
+     */
+    #[Container(LoggerInterface::class)]
+    public StdoutLogger|LoggerInterface $logger;
+
+
+    /**
+     * @var bool
+     */
+    protected bool $enable_coroutine = false;
+
+
+    /**
+     * @var array
+     */
+    protected array $watches = [];
+
+
+    /**
+     * @var bool
+     */
+    protected bool $enable_queue = false;
+
+
+    /**
+     * @var bool
+     */
+    protected bool $reloading = false;
+
+
+    /**
+     * @throws \Exception
+     */
+    public function __construct()
+    {
+        di(EventProvider::class)->on(OnWorkerStart::class, [di(Router::class), 'scan_build_route']);
+    }
+
+
+    /**
+     * @return string
+     */
+    public function getName(): string
+    {
+        return 'hotReload';
+    }
+
+
+    /**
+     * @return void
+     */
+    public function onSigterm(): void
+    {
+        // TODO: Implement onSigterm() method.
+        $this->stop();
+    }
+
+    /**
+     * @param ?Process $process
+     */
+    public function process(Process|null $process): void
+    {
+        $this->pipe = inotify_init();
+        $this->addListen();
+        Event::add($this->pipe, function () use ($process) {
+            $read = inotify_read($this->pipe);
+            if (count($read) > 0) {
+                $this->reload();
+            }
+        });
+        Event::cycle(function (): void {
+            if ($this->isStop()) {
+                Event::exit();
+            }
+        });
+        Event::wait();
+    }
+
+
+    /**
+     * @return void
+     */
+    public function reload(): void
+    {
+        if ($this->reloading) {
+            return;
+        }
+        $this->reloading = true;
+
+        di(StdoutLogger::class)->println('reloading server[' . \config('id', 'system-service') . '], please waite.');
+
+        $this->clear();
+        di(ServerInterface::class)->reload();
+        $this->addListen();
+
+        $this->reloading = false;
+    }
+
+
+    /**
+     * @return void
+     */
+    protected function addListen(): void
+    {
+        foreach (config('reload.listen') as $value) {
+            $this->readDirectory($value);
+        }
+    }
+
+
+    /**
+     * @return void
+     */
+    protected function clear(): void
+    {
+        $this->watches = [];
+    }
+
+    /**
+     * @param string $directory
+     * @return void
+     */
+    public function readFile(string $directory): void
+    {
+        if (str_ends_with($directory, '.php') === true) {
+            inotify_add_watch($this->pipe, $directory, IN_MODIFY | IN_MOVE | IN_CREATE | IN_DELETE);
+        }
+    }
+
+
+    /**
+     * @param string $directory
+     * @return void
+     */
+    public function readDirectory(string $directory): void
+    {
+        foreach (glob($directory . '/*') as $data) {
+            if (is_dir($data)) {
+                $this->readDirectory($data);
+            } else {
+                $this->readFile($data);
+            }
+        }
+    }
+}

Handler/OnPipeMessage.php

@@ -2,6 +2,7 @@
 
 namespace Kiri\Server\Handler;
 
+use Kiri;
 use Kiri\Server\Abstracts\Server;
 use Kiri\Server\Contract\OnPipeMessageInterface;
 
@@ -20,13 +21,17 @@ class OnPipeMessage extends Server
      */
     public function onPipeMessage(\Swoole\Server $server, int $src_worker_id, mixed $message): void
     {
-        if (is_string($message)) {
-            $message = unserialize($message);
+        try {
+            if (is_string($message)) {
+                $message = unserialize($message);
+            }
+            if (!is_object($message) || !($message instanceof OnPipeMessageInterface)) {
+                return;
+            }
+            call_user_func([$message, 'process'], $server, $src_worker_id);
+        } catch (\Throwable $throwable) {
+            Kiri::getLogger()->error(throwable($throwable));
         }
-        if (!is_object($message) || !($message instanceof OnPipeMessageInterface)) {
-            return;
-        }
-        call_user_func([$message, 'process'], $server, $src_worker_id);
     }
 
 

Handler/OnServerWorker.php

@@ -3,7 +3,10 @@
 namespace Kiri\Server\Handler;
 
 use Kiri;
+use Kiri\Abstracts\CoordinatorManager;
+use Kiri\Coordinator;
 use Kiri\Core\Help;
+use Kiri\Di\Inject\Container;
 use Kiri\Events\EventDispatch;
 use Kiri\Server\Events\OnAfterWorkerStart;
 use Kiri\Server\Events\OnBeforeWorkerStart;
@@ -12,6 +15,7 @@ use Kiri\Server\Events\OnWorkerError;
 use Kiri\Server\Events\OnWorkerExit;
 use Kiri\Server\Events\OnWorkerStart;
 use Kiri\Server\Events\OnWorkerStop;
+use Psr\Http\Message\RequestInterface;
 use Swoole\Server;
 use Swoole\Timer;
 use Throwable;
@@ -25,6 +29,14 @@ use function config;
 class OnServerWorker extends Kiri\Server\Abstracts\Server
 {
 
+
+    /**
+     * @var EventDispatch
+     */
+    #[Container(EventDispatch::class)]
+    public EventDispatch $dispatch;
+
+
     /**
      * @return void
      */
@@ -59,14 +71,15 @@ class OnServerWorker extends Kiri\Server\Abstracts\Server
      */
     public function onWorkerStart(Server $server, int $workerId): void
     {
-        $dispatch = Kiri::getDi()->get(EventDispatch::class);
-        $dispatch->dispatch(new OnBeforeWorkerStart($server, $workerId));
+        $this->dispatch->dispatch(new OnBeforeWorkerStart($server, $workerId));
         if ($workerId < $server->setting['worker_num']) {
-            $dispatch->dispatch(new OnWorkerStart($server, $workerId));
+            CoordinatorManager::utility(Coordinator::WORKER_START)->waite();
+
+            $this->dispatch->dispatch(new OnWorkerStart($server, $workerId));
         } else {
-            $dispatch->dispatch(new OnTaskerStart($server, $workerId));
+            $this->dispatch->dispatch(new OnTaskerStart($server, $workerId));
         }
-        $dispatch->dispatch(new OnAfterWorkerStart($server, $workerId));
+        $this->dispatch->dispatch(new OnAfterWorkerStart($server, $workerId));
     }
 
 
@@ -117,9 +130,16 @@ class OnServerWorker extends Kiri\Server\Abstracts\Server
         event(new OnWorkerError($server, $worker_id, $worker_pid, $exit_code, $signal));
 
         debug_print_backtrace(DEBUG_BACKTRACE_PROVIDE_OBJECT);
-        $message = sprintf('Worker#%d::%d error stop. signal %d, exit_code %d, msg %s', $worker_id, $worker_pid, $signal, $exit_code, swoole_strerror(swoole_last_error(), $signal));
 
-        error($message);
+        /** @var RequestInterface $context */
+        $context = Kiri\Di\Context::get(RequestInterface::class);
+        if (is_null($context)) {
+            $message = sprintf('Worker#%d::%d error stop. signal %d, exit_code %d, msg %s', $worker_id, $worker_pid, $signal, $exit_code, swoole_strerror(swoole_last_error(), $signal));
+        } else {
+            $message = sprintf('Worker#%d::%d error stop. signal %d, exit_code %d, msg %s, method: %s, path: %s, query: %s', $worker_id, $worker_pid, $signal, $exit_code, swoole_strerror(swoole_last_error(), $signal),
+                $context->getMethod(), $context->getUri()->getPath(), $context->getUri()->getQuery());
+        }
+        error($message . PHP_EOL);
 
         $this->system_mail($message);
     }

Processes/AbstractProcess.php

@@ -5,6 +5,8 @@ namespace Kiri\Server\Processes;
 
 use Swoole\Coroutine;
 use Swoole\Process;
+use const SIGHUP;
+use const SIGTERM;
 
 /**
  *
@@ -125,26 +127,29 @@ abstract class AbstractProcess implements OnProcessInterface
     {
         $this->process = $process;
         if ($this->enable_coroutine) {
-            Coroutine::set([
-                'enable_deadlock_check'        => false,
-                'deadlock_check_disable_trace' => false,
-                'exit_condition'               => function () {
-                    return Coroutine::stats()['coroutine_num'] === 0;
-                }
-            ]);
-            Coroutine::create(fn () => $this->coroutineWaitSignal());
+            $array['enable_deadlock_check']        = false;
+            $array['deadlock_check_disable_trace'] = false;
+            $array['exit_condition']               = [$this, 'exit_condition'];
+            Coroutine::set($array);
+            Coroutine::create(fn() => $this->coroutineWaitSignal());
         } else {
-            pcntl_signal(SIGTERM, [$this, 'pointWaitSignal']);
+            $process::signal(SIGTERM | SIGINT | SIGUSR1, [$this, 'pointWaitSignal']);
         }
         return $this;
     }
 
 
+    public function exit_condition(): bool
+    {
+        return Coroutine::stats()['coroutine_num'] === 0;
+    }
+
+
     /**
-     * @param $data
+     * @param $signal
      * @return void
      */
-    public function pointWaitSignal($data): void
+    public function pointWaitSignal($signal): void
     {
         $this->stop = true;
 

ServerCommand.php

@@ -6,16 +6,21 @@ namespace Kiri\Server;
 
 use Exception;
 use Kiri;
+use Kiri\Server\Events\OnWorkerStart;
+use Kiri\Events\EventProvider;
 use Kiri\Events\EventDispatch;
 use Kiri\Router\Router;
 use Kiri\Server\Abstracts\AsyncServer;
 use Kiri\Server\Events\OnShutdown;
+use Psr\Container\ContainerInterface;
 use Symfony\Component\Console\Command\Command;
 use Symfony\Component\Console\Input\InputArgument;
 use Symfony\Component\Console\Input\InputInterface;
 use Symfony\Component\Console\Input\InputOption;
 use Symfony\Component\Console\Output\OutputInterface;
 use function config;
+use Kiri\Server\Abstracts\HotReload;
+use Kiri\Di\Inject\Container;
 
 defined('ROUTER_TYPE_HTTP') or define('ROUTER_TYPE_HTTP', 'http');
 defined('PID_PATH') or define('PID_PATH', APP_PATH . 'storage/server.pid');
@@ -31,6 +36,34 @@ class ServerCommand extends Command
     public State $state;
 
 
+    /**
+     * @var ContainerInterface
+     */
+    #[Container(ContainerInterface::class)]
+    public ContainerInterface $container;
+
+
+    /**
+     * @var AsyncServer
+     */
+    #[Container(AsyncServer::class)]
+    public AsyncServer $asyncServer;
+
+
+    /**
+     * @var EventDispatch
+     */
+    #[Container(EventDispatch::class)]
+    public EventDispatch $eventDispatch;
+
+
+    /**
+     * @var EventProvider
+     */
+    #[Container(EventProvider::class)]
+    public EventProvider $eventProvider;
+
+
     /**
      * @param string|null $name
      * @throws Exception
@@ -92,12 +125,11 @@ class ServerCommand extends Command
     protected function stop(): int
     {
         $configs   = config('server', []);
-        $instances = Kiri::getDi()->get(AsyncServer::class)->sortService($configs['ports'] ?? []);
+        $instances = $this->asyncServer->sortService($configs['ports'] ?? []);
         foreach ($instances as $config) {
             $this->state->exit($config->port);
         }
-        $dispatch = Kiri::getDi()->get(EventDispatch::class);
-        $dispatch->dispatch(new OnShutdown());
+        $this->eventDispatch->dispatch(new OnShutdown());
         return 1;
     }
 
@@ -109,12 +141,14 @@ class ServerCommand extends Command
      */
     protected function start(InputInterface $input): int
     {
-        $manager = Kiri::getDi()->get(AsyncServer::class);
-        $router  = Kiri::getDi()->get(Router::class);
-        $router->scan_build_route();
-        $manager->addProcess(config('processes', []));
-        $manager->initCoreServers(config('server', []), (int)$input->getOption('daemon'));
-        $manager->start();
+        $this->asyncServer->addProcess(config('processes', []));
+        if (\config('reload.hot', false) === true) {
+            $this->asyncServer->addProcess([HotReload::class]);
+        } else {
+            di(Router::class)->scan_build_route();
+        }
+        $this->asyncServer->initCoreServers(config('server', []), (int)$input->getOption('daemon'));
+        $this->asyncServer->start();
         return 1;
     }
 

Task/OnTaskFinish.php

@@ -6,12 +6,12 @@ class OnTaskFinish
 {
 
 
-	/**
-	 * @param int $task_id
-	 * @param mixed $data
-	 */
-	public function __construct(int $task_id, mixed $data)
-	{
-	}
+    /**
+     * @param int $task_id
+     * @param mixed $data
+     */
+    public function __construct(public int $task_id, public mixed $data)
+    {
+    }
 
 }

Task/OnTaskInterface.php

@@ -0,0 +1,17 @@
+<?php
+
+namespace Kiri\Server\Task;
+
+interface OnTaskInterface
+{
+
+
+    /**
+     * @param int $task_id
+     * @param int $src_worker_id
+     * @return mixed
+     */
+    public function process(int $task_id, int $src_worker_id): mixed;
+
+
+}

Task/Task.php

@@ -67,20 +67,24 @@ class Task
      * @param int $task_id
      * @param int $src_worker_id
      * @param mixed $data
-     * @return mixed
+     * @return void
      * @throws
      */
-    public function onTask(Server $server, int $task_id, int $src_worker_id, mixed $data): mixed
+    public function onTask(Server $server, int $task_id, int $src_worker_id, mixed $data): void
     {
         try {
-            $data = json_decode($data, true);
-            if (is_null($data)) {
-                return null;
+            [$handler, $params] = [$data[0], $data[1]];
+
+            $handler = Kiri::getDi()->make($handler, $params);
+            if (!($handler instanceof OnTaskInterface)) {
+                throw new \Exception('Task process must implements ' . OnTaskInterface::class);
             }
-            $data[0] = Kiri::getDi()->get($data[0]);
-            return call_user_func($data, $task_id, $src_worker_id);
+
+            $response = call_user_func([$handler, 'process'], $task_id, $src_worker_id);
         } catch (\Throwable $throwable) {
-            return $this->exception->emit($throwable, response());
+            $response = throwable($throwable);
+        } finally {
+            $server->finish($response);
         }
     }
 

Task/TaskExecute.php

@@ -2,22 +2,50 @@
 
 namespace Kiri\Server\Task;
 
+use Exception;
 use Kiri\Server\ServerInterface;
 
 class TaskExecute implements TaskInterface
 {
 
+
     /**
+     * @param string $handler
+     * @param mixed $data
+     * @param int $dstWorkerId
+     * @param callable|null $finishFinishCallback
+     * @return void
+     * @throws
+     */
+    public function task(string $handler, mixed $data, int $dstWorkerId = -1, ?callable $finishFinishCallback = null): void
+    {
+        $array = class_implements($handler, true);
+        if (!in_array(OnTaskInterface::class, $array, true)) {
+            throw new Exception('Task is not implement OnTaskInterface');
+        }
+        $server = \Kiri::getDi()->get(ServerInterface::class);
+
+        $server->task([$handler, $data], $dstWorkerId, $finishFinishCallback);
+    }
+
+
+    /**
+     * @param string $handler
      * @param mixed $data
      * @param float $timeout
      * @param int $dstWorkerId
      * @return mixed
+     * @throws
      */
-    public function taskWait(mixed $data, float $timeout = 0.5, int $dstWorkerId = -1): mixed
+    public function taskWait(string $handler, mixed $data, float $timeout = 0.5, int $dstWorkerId = -1): mixed
     {
+        $array = class_implements($handler, true);
+        if (!in_array(OnTaskInterface::class, $array, true)) {
+            throw new Exception('Task is not implement OnTaskInterface');
+        }
         $server = \Kiri::getDi()->get(ServerInterface::class);
 
-        return $server->taskwait($data, $timeout, $dstWorkerId);
+        return $server->taskwait([$handler, $data], $timeout, $dstWorkerId);
     }
 
 
@@ -45,6 +73,4 @@ class TaskExecute implements TaskInterface
 
         return $server->taskWaitMulti($tasks, $timeout);
     }
-
-
-}
+}

Task/TaskInterface.php

@@ -7,11 +7,24 @@ interface TaskInterface
 
 
     /**
-     * @param array $tasks
-     * @param float $timeout
-     * @return false|array
+     * @param string $handler
+     * @param mixed $data
+     * @param int $dstWorkerId
+     * @param callable|null $finishFinishCallback
+     * @return void
      */
-    public function taskWaitMulti(array $tasks, float $timeout = 0.5): false|array;
+    public function task(string $handler, mixed $data, int $dstWorkerId = -1, ?callable $finishFinishCallback = null): void;
+
+
+    /**
+     * @param string $handler
+     * @param mixed $data
+     * @param float $timeout
+     * @param int $dstWorkerId
+     * @return mixed
+     */
+    public function taskWait(string $handler, mixed $data, float $timeout = 0.5, int $dstWorkerId = -1): mixed;
+
 
     /**
      * @param array $tasks
@@ -20,12 +33,12 @@ interface TaskInterface
      */
     public function taskCo(array $tasks, float $timeout = 0.5): false|array;
 
+
     /**
-     * @param mixed $data
+     * @param array $tasks
      * @param float $timeout
-     * @param int $dstWorkerId
-     * @return mixed
+     * @return false|array
      */
-    public function taskWait(mixed $data, float $timeout = 0.5, int $dstWorkerId = -1): mixed;
+    public function taskWaitMulti(array $tasks, float $timeout = 0.5): false|array;
 
 }

Task/TestTask.php

@@ -0,0 +1,24 @@
+<?php
+
+namespace Kiri\Server\Task;
+
+class TestTask implements OnTaskInterface
+{
+
+
+    public function __construct($user, $friend)
+    {
+    }
+
+
+    /**
+     * @param int $task_id
+     * @param int $src_worker_id
+     * @return int
+     */
+    public function process(int $task_id, int $src_worker_id): int
+    {
+        return 0;
+    }
+
+}