initial 2.0.0-a4 commit

src/ZM/Utils/CoMessage.php

@@ -0,0 +1,80 @@
+<?php
+
+
+namespace ZM\Utils;
+
+
+use Co;
+use Exception;
+use ZM\Store\LightCacheInside;
+use ZM\Store\Lock\SpinLock;
+use ZM\Store\ZMAtomic;
+
+class CoMessage
+{
+    /**
+     * @param array $hang
+     * @param array $compare
+     * @param int $timeout
+     * @return bool
+     * @throws Exception
+     */
+    public static function yieldByWS(array $hang, array $compare, $timeout = 600) {
+        $cid = Co::getuid();
+        $api_id = ZMAtomic::get("wait_msg_id")->add(1);
+        $hang["compare"] = $compare;
+        $hang["coroutine"] = $cid;
+        $hang["worker_id"] = server()->worker_id;
+        $hang["result"] = null;
+        SpinLock::lock("wait_api");
+        $wait = LightCacheInside::get("wait_api", "wait_api");
+        $wait[$api_id] = $hang;
+        LightCacheInside::set("wait_api", "wait_api", $wait);
+        SpinLock::unlock("wait_api");
+        $id = swoole_timer_after($timeout * 1000, function () use ($api_id) {
+            $r = LightCacheInside::get("wait_api", "wait_api")[$api_id] ?? null;
+            if (is_array($r)) {
+                Co::resume($r["coroutine"]);
+            }
+        });
+        Co::suspend();
+        SpinLock::lock("wait_api");
+        $sess = LightCacheInside::get("wait_api", "wait_api");
+        $result = $sess[$api_id]["result"];
+        unset($sess[$api_id]);
+        LightCacheInside::set("wait_api", "wait_api", $sess);
+        SpinLock::unlock("wait_api");
+        if (isset($id)) swoole_timer_clear($id);
+        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) {
+            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;
+        }
+    }
+}

src/ZM/Utils/ZMUtil.php

@@ -10,15 +10,18 @@ use Swoole\Event;
 use Swoole\Timer;
 use ZM\Console\Console;
 use ZM\Store\LightCache;
+use ZM\Store\LightCacheInside;
+use ZM\Store\ZMAtomic;
 use ZM\Store\ZMBuf;
 
 class ZMUtil
 {
     public static function stop() {
         Console::warning(Console::setColor("Stopping server...", "red"));
+        LightCache::savePersistence();
         if (ZMBuf::$terminal !== null)
             Event::del(ZMBuf::$terminal);
-        ZMBuf::atomic("stop_signal")->set(1);
+        ZMAtomic::get("stop_signal")->set(1);
         try {
             LightCache::set('stop', 'OK');
         } catch (Exception $e) {
@@ -30,9 +33,8 @@ class ZMUtil
     public static function reload($delay = 800) {
         Console::info(Console::setColor("Reloading server...", "gold"));
         usleep($delay * 1000);
-        foreach (LightCache::getAll() as $k => $v) {
-            if (mb_substr($k, 0, 8) == "wait_api")
-                if ($v["result"] === null) Co::resume($v["coroutine"]);
+        foreach ((LightCacheInside::get("wait_api", "wait_api") ?? []) as $k => $v) {
+            if ($v["result"] === null && isset($v["coroutine"])) Co::resume($v["coroutine"]);
         }
         foreach (server()->connections as $v) {
             server()->close($v);
@@ -50,4 +52,8 @@ class ZMUtil
             return ZMBuf::$instance[$class];
         }
     }
+
+    public static function sendActionToWorker($target_id, $action, $data) {
+        server()->sendMessage(json_encode(["action" => $action, "data" => $data]), $target_id);
+    }
 }

src/ZM/Utils/terminal_listener.php

@@ -1,33 +0,0 @@
-<?php
-
-use Swoole\Coroutine\Http\Client;
-
-Co\run(function (){
-    hello:
-    global $terminal_id, $port;
-    $client = new Client("127.0.0.1", $port);
-    $client->set(['websocket_mask' => true]);
-    $client->setHeaders(["x-terminal-id" => $terminal_id, 'x-pid' => posix_getppid()]);
-    $ret = $client->upgrade("/?type=terminal");
-    if ($ret) {
-        while (true) {
-            $line = fgets(STDIN);
-            if ($line !== false) {
-                $r = $client->push(trim($line));
-                if (trim($line) == "reload" || trim($line) == "r" || trim($line) == "stop") {
-                    break;
-                }
-                if($r === false) {
-                    echo "Unable to connect framework terminal, connection closed. Trying to reconnect after 5s.\n";
-                    sleep(5);
-                    goto hello;
-                }
-            } else {
-                break;
-            }
-        }
-    } else {
-        echo "Unable to connect framework terminal. port: $port\n";
-    }
-});
-