diff --git a/composer.json b/composer.json index 788f9497..64feb465 100644 --- a/composer.json +++ b/composer.json @@ -3,7 +3,7 @@ "description": "High performance QQ robot and web server development framework", "minimum-stability": "stable", "license": "Apache-2.0", - "version": "2.1.6", + "version": "2.2.0", "extra": { "exclude_annotate": [ "src/ZM" diff --git a/config/global.php b/config/global.php index ef75bf04..c3f4f37d 100644 --- a/config/global.php +++ b/config/global.php @@ -43,6 +43,11 @@ $config['light_cache'] = [ 'auto_save_interval' => 900 ]; +$config["worker_cache"] = [ + "worker" => 0, + "transaction_timeout" => 30000 +]; + /** MySQL数据库连接信息，host留空则启动时不创建sql连接池 */ $config['sql_config'] = [ 'sql_host' => '', diff --git a/src/Module/Example/Hello.php b/src/Module/Example/Hello.php index 5ee91712..ba871a3d 100644 --- a/src/Module/Example/Hello.php +++ b/src/Module/Example/Hello.php @@ -2,6 +2,8 @@ namespace Module\Example; +use Swoole\Coroutine; +use ZM\Annotation\Http\Controller; use ZM\Annotation\Http\Middleware; use ZM\Annotation\Swoole\OnCloseEvent; use ZM\Annotation\Swoole\OnOpenEvent; @@ -11,12 +13,16 @@ use ZM\Console\Console; use ZM\Annotation\CQ\CQCommand; use ZM\Annotation\Http\RequestMapping; use ZM\Event\EventDispatcher; +use ZM\Store\Lock\SpinLock; +use ZM\Store\WorkerCache; +use ZM\Utils\ProcessManager; use ZM\Utils\ZMUtil; /** * Class Hello * @package Module\Example * @since 2.0 + * @Controller("/sdd") */ class Hello { @@ -79,6 +85,7 @@ class Hello * @RequestMapping("/") */ public function index() { + Console::info("Called!!"); return "Hello Zhamao!"; } @@ -89,7 +96,42 @@ class Hello * @return string */ public function paramGet($param) { - return "Your name: {$param["name"]}"; + //WorkerCache::set("ss_cnt", 0); + return json_encode(Coroutine::stats()); + } + + /** + * @param $param + * @RequestMapping("/wc/set/{key}/{value}") + * @return string + */ + public function wcSet($param) { + WorkerCache::set($param["key"], $param["value"], true); + //SpinLock::lock("ss"); + WorkerCache::add("ss_cnt", 1); + //SpinLock::unlock("ss"); + return json_encode(Coroutine::stats()); + } + + /** + * @param $param + * @return mixed|string + * @RequestMapping("/wc/get/{key}") + * @Middleware("timer") + */ + public function wcGet($param) { + $obj = WorkerCache::get($param["key"]); + if(is_string($obj)) return $obj; + return json_encode($obj); + } + + /** + * @return mixed + * @RequestMapping("/task") + */ + public function runTask() { + $r = ProcessManager::runOnTask(["class" => Hello::class, "method" => "index", "params" => []]); + return $r; } /** diff --git a/src/ZM/Annotation/Swoole/OnPipeMessageEvent.php b/src/ZM/Annotation/Swoole/OnPipeMessageEvent.php new file mode 100644 index 00000000..76cf0c4c --- /dev/null +++ b/src/ZM/Annotation/Swoole/OnPipeMessageEvent.php @@ -0,0 +1,24 @@ +fd . ": " . TermColor::ITALIC . $frame->data . TermColor::RESET); - unset(Context::$context[\Swoole\Coroutine::getCid()]); + unset(Context::$context[Coroutine::getCid()]); $conn = ManagerGM::get($frame->fd); set_coroutine_params(["server" => $server, "frame" => $frame, "connection" => $conn]); $dispatcher1 = new EventDispatcher(OnMessageEvent::class); $dispatcher1->setRuleFunction(function ($v) { + /** @noinspection PhpUnreachableStatementInspection */ return ctx()->getConnection()->getName() == $v->connect_type && eval("return " . $v->getRule() . ";"); }); @@ -480,38 +484,73 @@ class ServerEventHandler * @param $server * @param $src_worker_id * @param $data + * @throws InterruptException */ public function onPipeMessage(Server $server, $src_worker_id, $data) { //var_dump($data, $server->worker_id); //unset(Context::$context[Co::getCid()]); $data = json_decode($data, true); switch ($data["action"] ?? '') { - case "resume_ws_message": - $obj = $data["data"]; - Co::resume($obj["coroutine"]); + case "getWorkerCache": + $r = WorkerCache::get($data["key"]); + $action = ["action" => "returnWorkerCache", "cid" => $data["cid"], "value" => $r]; + $server->sendMessage(json_encode($action, 256), $src_worker_id); break; - case "stop": - Console::verbose('正在清理 #' . $server->worker_id . ' 的计时器'); - Timer::clearAll(); + case "setWorkerCache": + $r = WorkerCache::set($data["key"], $data["value"]); + $action = ["action" => "returnWorkerCache", "cid" => $data["cid"], "value" => $r]; + $server->sendMessage(json_encode($action, 256), $src_worker_id); break; - case "terminate": - $server->stop(); + case "asyncAddWorkerCache": + WorkerCache::add($data["key"], $data["value"], true); break; - case 'echo': - Console::success('接收到来自 #' . $src_worker_id . ' 的消息'); + case "asyncSubWorkerCache": + WorkerCache::sub($data["key"], $data["value"], true); break; - case 'send': - $server->sendMessage(json_encode(["action" => "echo"]), $data["target"]); + case "asyncSetWorkerCache": + WorkerCache::set($data["key"], $data["value"], true); + break; + case "addWorkerCache": + $r = WorkerCache::add($data["key"], $data["value"]); + $action = ["action" => "returnWorkerCache", "cid" => $data["cid"], "value" => $r]; + $server->sendMessage(json_encode($action, 256), $src_worker_id); + break; + case "subWorkerCache": + $r = WorkerCache::sub($data["key"], $data["value"]); + $action = ["action" => "returnWorkerCache", "cid" => $data["cid"], "value" => $r]; + $server->sendMessage(json_encode($action, 256), $src_worker_id); + break; + case "returnWorkerCache": + WorkerCache::$transfer[$data["cid"]] = $data["value"]; + zm_resume($data["cid"]); break; default: - echo $data . PHP_EOL; + $dispatcher = new EventDispatcher(OnPipeMessageEvent::class); + $dispatcher->setRuleFunction(function (OnPipeMessageEvent $v) use ($data) { + return $v->action == $data["action"]; + }); + $dispatcher->dispatchEvents($data); + break; } } /** * @SwooleHandler("task") + * @param Server|null $server + * @param Server\Task $task + * @return mixed */ - public function onTask() { + public function onTask(?Server $server, Server\Task $task) { + $data = $task->data; + switch($data["action"]) { + case "runMethod": + $c = $data["class"]; + $ss = new $c(); + $method = $data["method"]; + $ps = $data["params"]; + $task->finish($ss->$method(...$ps)); + } + return null; } /** diff --git a/src/ZM/Http/RouteManager.php b/src/ZM/Http/RouteManager.php index 0aa842e7..4fd4a0b4 100644 --- a/src/ZM/Http/RouteManager.php +++ b/src/ZM/Http/RouteManager.php @@ -8,6 +8,7 @@ use Symfony\Component\Routing\Route; use Symfony\Component\Routing\RouteCollection; use ZM\Annotation\Http\Controller; use ZM\Annotation\Http\RequestMapping; +use ZM\Console\Console; class RouteManager { @@ -25,8 +26,9 @@ class RouteManager break; } } - - $route_name = $prefix."/".$vss->route; + $tail = trim($vss->route, "/"); + $route_name = $prefix.($tail === "" ? "" : "/").$tail; + Console::debug("添加路由：".$route_name); $route = new Route($route_name, ['_class' => $class, '_method' => $method]); $route->setMethods($vss->request_method); diff --git a/src/ZM/Store/LightCache.php b/src/ZM/Store/LightCache.php index c28c61e5..b9ff8cf6 100644 --- a/src/ZM/Store/LightCache.php +++ b/src/ZM/Store/LightCache.php @@ -7,6 +7,7 @@ namespace ZM\Store; use Exception; use Swoole\Table; use ZM\Console\Console; +use ZM\Exception\ZMException; class LightCache { @@ -19,6 +20,11 @@ class LightCache public static $last_error = ''; + /** + * @param $config + * @return bool|mixed + * @throws Exception + */ public static function init($config) { self::$config = $config; self::$kv_table = new Table($config["size"], $config["hash_conflict_proportion"]); @@ -50,11 +56,11 @@ class LightCache /** * @param string $key - * @return null|string - * @throws Exception + * @return null|mixed + * @throws ZMException */ public static function get(string $key) { - if (self::$kv_table === null) throw new Exception("not initialized LightCache"); + if (self::$kv_table === null) throw new ZMException("not initialized LightCache"); self::checkExpire($key); $r = self::$kv_table->get($key); return $r === false ? null : self::parseGet($r); @@ -63,10 +69,10 @@ class LightCache /** * @param string $key * @return mixed|null - * @throws Exception + * @throws ZMException */ public static function getExpire(string $key) { - if (self::$kv_table === null) throw new Exception("not initialized LightCache"); + if (self::$kv_table === null) throw new ZMException("not initialized LightCache"); self::checkExpire($key); $r = self::$kv_table->get($key, "expire"); return $r === false ? null : $r - time(); @@ -77,10 +83,10 @@ class LightCache * @param string|array|int $value * @param int $expire * @return mixed - * @throws Exception + * @throws ZMException */ public static function set(string $key, $value, int $expire = -1) { - if (self::$kv_table === null) throw new Exception("not initialized LightCache"); + if (self::$kv_table === null) throw new ZMException("not initialized LightCache"); if (is_array($value)) { $value = json_encode($value, JSON_UNESCAPED_UNICODE); if (strlen($value) >= self::$config["max_strlen"]) return false; @@ -93,7 +99,7 @@ class LightCache $data_type = "bool"; $value = json_encode($value); } else { - throw new Exception("Only can set string, array and int"); + throw new ZMException("Only can set string, array and int"); } try { return self::$kv_table->set($key, [ @@ -110,10 +116,10 @@ class LightCache * @param string $key * @param $value * @return bool|mixed - * @throws Exception + * @throws ZMException */ public static function update(string $key, $value) { - if (self::$kv_table === null) throw new Exception("not initialized LightCache."); + if (self::$kv_table === null) throw new ZMException("not initialized LightCache."); if (is_array($value)) { $value = json_encode($value, JSON_UNESCAPED_UNICODE); if (strlen($value) >= self::$config["max_strlen"]) return false; @@ -123,7 +129,7 @@ class LightCache } elseif (is_int($value)) { $data_type = "int"; } else { - throw new Exception("Only can set string, array and int"); + throw new ZMException("Only can set string, array and int"); } try { if (self::$kv_table->get($key) === false) return false; diff --git a/src/ZM/Store/WorkerCache.php b/src/ZM/Store/WorkerCache.php new file mode 100644 index 00000000..e1d15cdb --- /dev/null +++ b/src/ZM/Store/WorkerCache.php @@ -0,0 +1,83 @@ +worker_id) { + return self::$store[$key] ?? null; + } else { + $action = ["action" => "getWorkerCache", "key" => $key, "cid" => zm_cid()]; + server()->sendMessage(json_encode($action, JSON_UNESCAPED_UNICODE), $config["worker"]); + zm_yield(); + $p = self::$transfer[zm_cid()] ?? null; + unset(self::$transfer[zm_cid()]); + return $p; + } + } + + public static function set($key, $value, $async = false) { + $config = self::$config ?? ZMConfig::get("global", "worker_cache"); + if ($config["worker"] === server()->worker_id) { + self::$store[$key] = $value; + return true; + } else { + $action = ["action" => $async ? "asyncSetWorkerCache" : "setWorkerCache", "key" => $key, "value" => $value, "cid" => zm_cid()]; + $ss = server()->sendMessage(json_encode($action, JSON_UNESCAPED_UNICODE), $config["worker"]); + if(!$ss) return false; + if ($async) return true; + zm_yield(); + $p = self::$transfer[zm_cid()] ?? null; + unset(self::$transfer[zm_cid()]); + return $p; + } + } + + public static function add($key, int $value, $async = false) { + $config = self::$config ?? ZMConfig::get("global", "worker_cache"); + if ($config["worker"] === server()->worker_id) { + if(!isset(self::$store[$key])) self::$store[$key] = 0; + self::$store[$key] += $value; + return true; + } else { + $action = ["action" => $async ? "asyncAddWorkerCache" : "addWorkerCache", "key" => $key, "value" => $value, "cid" => zm_cid()]; + $ss = server()->sendMessage(json_encode($action, JSON_UNESCAPED_UNICODE), $config["worker"]); + // if(!$ss) return false; + if ($async) return true; + zm_yield(); + $p = self::$transfer[zm_cid()] ?? null; + unset(self::$transfer[zm_cid()]); + return $p; + } + } + + public static function sub($key, int $value, $async = false) { + $config = self::$config ?? ZMConfig::get("global", "worker_cache"); + if ($config["worker"] === server()->worker_id) { + if(!isset(self::$store[$key])) self::$store[$key] = 0; + self::$store[$key] -= $value; + return true; + } else { + $action = ["action" => $async ? "asyncSubWorkerCache" : "subWorkerCache", "key" => $key, "value" => $value, "cid" => zm_cid()]; + $ss = server()->sendMessage(json_encode($action, JSON_UNESCAPED_UNICODE), $config["worker"]); + // if(!$ss) return false; + if ($async) return true; + zm_yield(); + $p = self::$transfer[zm_cid()] ?? null; + unset(self::$transfer[zm_cid()]); + return $p; + } + } +} \ No newline at end of file diff --git a/src/ZM/Utils/CoMessage.php b/src/ZM/Utils/CoMessage.php index 637f56f7..b19fb307 100644 --- a/src/ZM/Utils/CoMessage.php +++ b/src/ZM/Utils/CoMessage.php @@ -48,34 +48,4 @@ class CoMessage if ($result === null) return false; return $result; } - - public static function resumeByWS() { - $dat = ctx()->getData(); - $last = null; - SpinLock::lock("wait_api"); - $all = LightCacheInside::get("wait_api", "wait_api") ?? []; - foreach ($all as $k => $v) { - if(!isset($v["compare"])) continue; - foreach ($v["compare"] as $vs) { - if ($v[$vs] != ($dat[$vs] ?? null)) { - continue 2; - } - } - $last = $k; - } - if($last !== null) { - $all[$last]["result"] = $dat; - LightCacheInside::set("wait_api", "wait_api", $all); - SpinLock::unlock("wait_api"); - if ($all[$last]["worker_id"] != server()->worker_id) { - ZMUtil::sendActionToWorker($all[$k]["worker_id"], "resume_ws_message", $all[$last]); - } else { - Co::resume($all[$last]["coroutine"]); - } - return true; - } else { - SpinLock::unlock("wait_api"); - return false; - } - } } diff --git a/src/ZM/Utils/ProcessManager.php b/src/ZM/Utils/ProcessManager.php new file mode 100644 index 00000000..13ef40f8 --- /dev/null +++ b/src/ZM/Utils/ProcessManager.php @@ -0,0 +1,17 @@ +taskwait([ + "action" => "runMethod", + "class" => $param["class"], + "method" => $param["method"], + "params" => $param["params"] + ], $timeout, $dst_worker_id); + } +} \ No newline at end of file diff --git a/src/ZM/Utils/ZMUtil.php b/src/ZM/Utils/ZMUtil.php index a4a3e949..8056f9bc 100644 --- a/src/ZM/Utils/ZMUtil.php +++ b/src/ZM/Utils/ZMUtil.php @@ -51,8 +51,4 @@ class ZMUtil return ZMBuf::$instance[$class]; } } - - public static function sendActionToWorker($target_id, $action, $data) { - server()->sendMessage(json_encode(["action" => $action, "data" => $data]), $target_id); - } } diff --git a/test/usage_test.php b/test/usage_test.php new file mode 100644 index 00000000..ef8584ee --- /dev/null +++ b/test/usage_test.php @@ -0,0 +1,10 @@ +