feat 优化channel的信号响应方案

Author: chaz6chez

README.md

@@ -273,4 +273,27 @@
   Cache::ChRemoveListener('test', '1', false);
   ```
   
+- **实验性功能：信号通知**
+  - 由于共享内存无法使用事件监听，所以底层使用Timer定时器进行轮询，实验性功能可以开启使用系统信号来监听数据的变化
+  ```php
+  // 设置信号
+  // 因为event等事件循环库是对标准信号的监听，所以不能使用自定实时信号SIGRTMIN ~ SIGRTMAX
+  // 默认暂时使用SIGPOLL，异步IO监听信号，可能影响异步文件IO相关的触发
+  Future::$signal = \SIGPOLL;
+  // 开启信号监听，这时候开启的监听会触发之前的回调和通道回调，不会影响之前的回调
+  Cache::channelUseSignalEnable(true)
+  ```
+  - 当使用的监听信号存在已注册的回调产生回调冲突时，可以手动设置回调事件共享
+  ```php
+  // 设置信号
+  // 因为event等事件循环库是对标准信号的监听，所以不能使用自定实时信号SIGRTMIN ~ SIGRTMAX
+  // 默认暂时使用SIGPOLL
+  Future::$signal = \SIGPOLL;
+  // 假设\SIGPOLL存在一个已注册的回调，YourEventLoop::getCallback(\SIGPOLL)可以获取该事件在当前进程注册的回调响应
+  // 设置回调
+  Future::setSignalCallback(YourEventLoop::getCallback(\SIGPOLL));
+  // 开启信号监听，这时候开启的监听会触发之前的回调和通道回调，不会影响之前的回调
+  Cache::channelUseSignalEnable(true)
+  ```
+  > 通道信号监听维系了一个事件队列，多次触发信号时，回调只会根据事件队列是否存在事件消费标记而执行事件回调
 ### 其他功能具体可以参看代码注释和测试用例

composer.json

@@ -25,7 +25,8 @@
     "webman/console": "^1.0 | ^2.0"
   },
   "suggest": {
-    "webman/console": "Webman-CLI support. "
+    "webman/console": "Webman-CLI support. ",
+    "ext-posix": "Channel signal support. "
   },
   "autoload": {
     "psr-4": {

src/Future.php

@@ -33,6 +33,20 @@ class Future
      */
     protected static array $_futures = [];
 
+    /**
+     * @var Closure|null
+     */
+    protected static ?Closure $_signalCallback = null;
+
+    /**
+     * @param Closure|null $func
+     * @return void
+     */
+    public static function setSignalCallback(?Closure $func): void
+    {
+        self::$_signalCallback = $func;
+    }
+
     /**
      * @param Closure $func
      * @param array $args
@@ -63,13 +77,18 @@ public static function add(Closure $func, array $args = [], float|int $interval
                 break;
             # 信号
             case self::DRIVER_SIGNAL:
-                $id = false;
+                $func = function () use ($func, $args) {
+                    // 触发信号原回调
+                    if (self::$_signalCallback) {
+                        call_user_func(self::$_signalCallback);
+                    }
+                    // 触发信号通道回调
+                    call_user_func($func, $args);
+                };
                 if (method_exists(Worker::$globalEvent, 'onSignal')) {
-                    Worker::$globalEvent->onSignal(self::$signal, function () use ($func, $args) {
-                        call_user_func($func, $args);
-                    });
+                    Worker::$globalEvent->onSignal(self::$signal, $func);
                 } else {
-                    Worker::$globalEvent->add(self::$signal, EventInterface::EV_SIGNAL, $func, $args);
+                    Worker::$globalEvent->add(self::$signal, EventInterface::EV_SIGNAL, $func);
                 }
                 self::$_futures[$id = 0] = $func;
                 break;
@@ -120,10 +139,19 @@ public static function del(int|null $id = null): void
                     break;
                 case self::DRIVER_SIGNAL:
                     if ($id === 0) {
-                        if (method_exists(Worker::$globalEvent, 'offSignal')) {
-                            Worker::$globalEvent->offSignal(self::$signal);
+                        // 如果有信号回调，则恢复信号回调
+                        if (self::$_signalCallback) {
+                            if (method_exists(Worker::$globalEvent, 'onSignal')) {
+                                Worker::$globalEvent->onSignal(self::$signal, self::$_signalCallback);
+                            } else {
+                                Worker::$globalEvent->add(self::$signal, EventInterface::EV_SIGNAL, self::$_signalCallback);
+                            }
                         } else {
-                            Worker::$globalEvent->del(self::$signal, EventInterface::EV_SIGNAL);
+                            if (method_exists(Worker::$globalEvent, 'offSignal')) {
+                                Worker::$globalEvent->offSignal(self::$signal);
+                            } else {
+                                Worker::$globalEvent->del(self::$signal, EventInterface::EV_SIGNAL);
+                            }
                         }
                     }
                     break;

src/Traits/ChannelMethods.php

@@ -23,6 +23,9 @@ trait ChannelMethods
     /** @var string 通道pid列表 */
     protected static string $_CHANNEL_PID_LIST = '#ChannelPidList#';
 
+    /** @var string 通道事件列表 */
+    protected static string $_CHANNEL_EVENT_LIST = '#ChannelEventList#';
+
     /**
      * @var array = [channelKey => futureId]
      */
@@ -152,7 +155,14 @@ protected static function _ChPublish(string $key, mixed $message, bool $store =
             if (self::isChannelUseSignal()) {
                 $list = self::_Get(self::$_CHANNEL_PID_LIST, []);
                 foreach ($list as $pid) {
-                    @posix_kill($pid, Future::$signal);
+                    self::_Atomic(self::$_CHANNEL_EVENT_LIST, function () use ($pid) {
+                        // 设置通道事件标记
+                        $channelEventList = self::_Get(self::$_CHANNEL_EVENT_LIST, []);
+                        $channelEventList[$pid][] = 1;
+                        self::_Set(self::$_CHANNEL_EVENT_LIST, $channelEventList);
+                        // 发送信号通知进程
+                        @posix_kill($pid, Future::$signal);
+                    });
                 }
             }
             return [
@@ -202,9 +212,25 @@ protected static function _ChCreateListener(string $key, string|int $workerId, C
              * ]
              */
             $channel = self::_Get($channelName = self::GetChannelKey($key), []);
+            // 监听器回调函数
             $callback = function () use ($key, $workerId, $listener) {
                 // 原子性执行
                 self::_Atomic($key, function () use ($key, $workerId, $listener) {
+                    // 信号监听
+                    if (self::isChannelUseSignal()) {
+                        $pid = posix_getpid();
+                        // 获取通道事件标记列表
+                        $channelEventList = self::_Get(self::$_CHANNEL_EVENT_LIST, []);
+                        $events = $channelEventList[$pid] ?? [];
+                        // 如果没有事件标记则跳过
+                        if (!array_pop($events)) {
+                            return;
+                        }
+                        // 更新通道事件标记
+                        $channelEventList[$pid] = $events;
+                        self::_Set(self::$_CHANNEL_EVENT_LIST, $channelEventList);
+                    }
+                    // 数据回调
                     $channel = self::_Get($channelName = self::GetChannelKey($key), []);
                     if ((!empty($value = $channel[$workerId]['value'] ?? []))) {
                         // 先进先出
@@ -257,11 +283,19 @@ protected static function _ChRemoveListener(string $key, string|int $workerId, b
                 Future::del($id);
                 // 信号监听则注册pid
                 if (self::isChannelUseSignal()) {
+                    $pid = posix_getpid();
+                    // 移除pid
                     $channelPidList = self::_Get(self::$_CHANNEL_PID_LIST, []);
-                    if ($channelPidList[$pid = posix_getpid()] ?? null) {
+                    if ($channelPidList[$pid] ?? null) {
                         unset($channelPidList[$pid]);
                         self::_Set(self::$_CHANNEL_PID_LIST, $channelPidList);
                     }
+                    // 移除事件标记
+                    $channelEventList = self::_Get(self::$_CHANNEL_EVENT_LIST, []);
+                    if ($channelEventList[$pid] ?? null) {
+                        unset($channelEventList[$pid]);
+                        self::_Set(self::$_CHANNEL_EVENT_LIST, $channelEventList);
+                    }
                 }
                 if ($remove) {
                     /**