support to schedule thread and coroutine

7 files changed, 66 insertions, 18 deletions

diff --git a/core/src/xmake/io/pipe_close.c b/core/src/xmake/io/pipe_close.c
index 6f50cbf67..062e7009d 100644
--- a/core/src/xmake/io/pipe_close.c
+++ b/core/src/xmake/io/pipe_close.c
@@ -41,11 +41,11 @@ tb_int_t xm_io_pipe_close(lua_State* lua)
     tb_assert_and_check_return_val(lua, 0);
 
     // check pipe?
-    if (!xm_lua_ispointer(lua, 1))
+    if (!xm_pipe_file_is_valid(lua, 1))
         return 0;
 
     // get the pipe file
-    tb_pipe_file_ref_t pipefile = (tb_pipe_file_ref_t)xm_lua_topointer(lua, 1);
+    tb_pipe_file_ref_t pipefile = xm_pipe_file_get(lua, 1);
     tb_check_return_val(pipefile, 0);
 
     // exit pipe file
diff --git a/core/src/xmake/io/pipe_read.c b/core/src/xmake/io/pipe_read.c
index d75547714..cb3457fec 100644
--- a/core/src/xmake/io/pipe_read.c
+++ b/core/src/xmake/io/pipe_read.c
@@ -41,7 +41,7 @@ tb_int_t xm_io_pipe_read(lua_State* lua)
     tb_assert_and_check_return_val(lua, 0);
 
     // check pipe file
-    if (!xm_lua_ispointer(lua, 1))
+    if (!xm_pipe_file_is_valid(lua, 1))
     {
         lua_pushinteger(lua, -1);
         lua_pushliteral(lua, "invalid pipe file!");
@@ -49,7 +49,7 @@ tb_int_t xm_io_pipe_read(lua_State* lua)
     }
 
     // get pipe file
-    tb_pipe_file_ref_t pipefile = (tb_pipe_file_ref_t)xm_lua_topointer(lua, 1);
+    tb_pipe_file_ref_t pipefile = xm_pipe_file_get(lua, 1);
     tb_check_return_val(pipefile, 0);
 
     // get data
diff --git a/core/src/xmake/io/pipe_wait.c b/core/src/xmake/io/pipe_wait.c
index 7de310425..f39220d41 100644
--- a/core/src/xmake/io/pipe_wait.c
+++ b/core/src/xmake/io/pipe_wait.c
@@ -41,11 +41,11 @@ tb_int_t xm_io_pipe_wait(lua_State* lua)
     tb_assert_and_check_return_val(lua, 0);
 
     // check pipe?
-    if (!xm_lua_ispointer(lua, 1))
+    if (!xm_pipe_file_is_valid(lua, 1))
         return 0;
 
     // get pipe file
-    tb_pipe_file_ref_t pipefile = (tb_pipe_file_ref_t)xm_lua_topointer(lua, 1);
+    tb_pipe_file_ref_t pipefile = xm_pipe_file_get(lua, 1);
     tb_check_return_val(pipefile, 0);
 
     // get events
diff --git a/core/src/xmake/io/pipe_write.c b/core/src/xmake/io/pipe_write.c
index 2ea0c69fe..140894428 100644
--- a/core/src/xmake/io/pipe_write.c
+++ b/core/src/xmake/io/pipe_write.c
@@ -41,7 +41,7 @@ tb_int_t xm_io_pipe_write(lua_State* lua)
     tb_assert_and_check_return_val(lua, 0);
 
     // check pipe
-    if (!xm_lua_ispointer(lua, 1))
+    if (!xm_pipe_file_is_valid(lua, 1))
     {
         lua_pushinteger(lua, -1);
         lua_pushliteral(lua, "invalid pipe file!");
@@ -49,7 +49,7 @@ tb_int_t xm_io_pipe_write(lua_State* lua)
     }
 
     // get pipe file
-    tb_pipe_file_ref_t pipefile = (tb_pipe_file_ref_t)xm_lua_topointer(lua, 1);
+    tb_pipe_file_ref_t pipefile = xm_pipe_file_get(lua, 1);
     tb_check_return_val(pipefile, 0);
 
     // get data and size
diff --git a/core/src/xmake/io/prefix.h b/core/src/xmake/io/prefix.h
index 7d8a8405b..ec2cc7216 100644
--- a/core/src/xmake/io/prefix.h
+++ b/core/src/xmake/io/prefix.h
@@ -94,6 +94,21 @@ typedef struct __xm_io_file_t
 
 } xm_io_file_t;
 
+// check pipe file
+static __tb_inline__ tb_bool_t xm_pipe_file_is_valid(lua_State* lua, tb_int_t index)
+{
+    return xm_lua_ispointer(lua, index) || xm_lua_isinteger(lua, index);
+}
+
+// get the pipe file from arguments
+static __tb_inline__ tb_pipe_file_ref_t xm_pipe_file_get(lua_State* lua, tb_int_t index)
+{
+    tb_pipe_file_ref_t pipe_file = tb_null;
+    if (xm_lua_isinteger(lua, index)) pipe_file = (tb_pipe_file_ref_t)(tb_size_t)(tb_long_t)lua_tointeger(lua, index);
+    else if (xm_lua_ispointer(lua, index)) pipe_file = (tb_pipe_file_ref_t)xm_lua_topointer(lua, index);
+    return pipe_file;
+}
+
 #endif
 
 
diff --git a/xmake/core/base/pipe.lua b/xmake/core/base/pipe.lua
index 48d36df9e..4b122ed9e 100644
--- a/xmake/core/base/pipe.lua
+++ b/xmake/core/base/pipe.lua
@@ -285,6 +285,11 @@ function _instance:__gc()
     end
 end
 
+-- new a pipe
+function pipe.new(cdata, name)
+    return _instance.new(cdata, name)
+end
+
 -- open a named pipe file
 --
 -- 1. named pipe (server-side):
diff --git a/xmake/core/base/thread.lua b/xmake/core/base/thread.lua
index 5bf08d11e..3697c580a 100644
--- a/xmake/core/base/thread.lua
+++ b/xmake/core/base/thread.lua
@@ -27,12 +27,14 @@ local _semaphore = _semaphore or {}
 local _queue     = _queue or {}
 
 -- load modules
-local io      = require("base/io")
-local libc    = require("base/libc")
-local bytes   = require("base/bytes")
-local table   = require("base/table")
-local string  = require("base/string")
-local sandbox = require("sandbox/sandbox")
+local io        = require("base/io")
+local libc      = require("base/libc")
+local pipe      = require("base/pipe")
+local bytes     = require("base/bytes")
+local table     = require("base/table")
+local string    = require("base/string")
+local scheduler = require("base/scheduler")
+local sandbox   = require("sandbox/sandbox")
 
 -- the thread status
 thread.STATUS_READY     = 1
@@ -118,11 +120,20 @@ function _thread:start()
         table.insert(argv, arg)
     end
 
+    -- init callback info
+    local callback = string._dump(self._CALLBACK)
+    local callinfo = {name = self:name(), argv = argv}
+
+    -- we need a pipe pair to wait and listen thread exit event
+    local rpipe, wpipe = pipe.openpair("BA") -- rpipe (block)
+    self._RPIPE = rpipe
+    callinfo.wpipe = libc.dataptr(wpipe:cdata())
+    -- we need to suppress gc to free it, because it has been transfer to thread in another lua state instance
+    wpipe._PIPE = nil
+
     -- serialize and pass callback and arguments to this thread
     -- we do not use string.serialize to serialize callback, because it's slower (deserialize)
     -- and we cannot strip function debug info, we need to reserve _ENV, and other upvalue names
-    local callback = string._dump(self._CALLBACK)
-    local callinfo = {name = self:name(), argv = argv}
     callinfo = string.serialize(callinfo, {strip = true, indent = false})
 
     -- init and start thread
@@ -177,7 +188,13 @@ function _thread:wait(timeout)
     end
     assert(self:cdata())
 
-    local ok, errors = thread.thread_wait(self:cdata(), timeout)
+    local ok, errors
+    local rpipe = self._RPIPE
+    if rpipe and scheduler:co_running() then
+        ok, errors = rpipe:wait(pipe.EV_READ, timeout)
+    else
+        ok, errors = thread.thread_wait(self:cdata(), timeout)
+    end
     if ok < 0 then
         return -1, errors or string.format("%s: failed to resume thread!", self)
     end
@@ -656,6 +673,7 @@ function thread._run_thread(callback_str, callinfo_str)
     local callinfo
     local argv
     local threadname
+    local wpipe
     if callinfo_str then
         local result, errors = string.deserialize(callinfo_str)
         if not result then
@@ -665,6 +683,7 @@ function thread._run_thread(callback_str, callinfo_str)
         if callinfo then
             argv = callinfo.argv
             threadname = callinfo.name
+            wpipe = pipe.new(libc.ptraddr(callinfo.wpipe))
         end
     end
 
@@ -719,7 +738,16 @@ function thread._run_thread(callback_str, callinfo_str)
     end
 
     -- do callback
-    return sandbox.load(sandbox_inst:script(), table.unpack(argv or {}))
+    local ok, errors = sandbox.load(sandbox_inst:script(), table.unpack(argv or {}))
+
+    -- thread is finished, we need to notify the waited thread
+    if wpipe then
+        local ok, errors = wpipe:write("exited")
+        if ok == nil then
+            return false, errors
+        end
+    end
+    return ok, errors
 end
 
 -- open a mutex