impl runloop for scheduler

3 files changed, 65 insertions, 6 deletions

diff --git a/core/src/xmake/io/poller_wait.c b/core/src/xmake/io/poller_wait.c
index 995815319..2589025c2 100644
--- a/core/src/xmake/io/poller_wait.c
+++ b/core/src/xmake/io/poller_wait.c
@@ -58,10 +58,6 @@ tb_int_t xm_io_poller_wait(lua_State* lua)
     // check
     tb_assert_and_check_return_val(lua, 0);
 
-    // is user data?
-    if (!lua_isuserdata(lua, 1)) 
-        return 0;
-
     // get timeout
     tb_long_t timeout = (tb_long_t)luaL_checknumber(lua, 1);
 
diff --git a/tests/modules/scheduler/test.lua b/tests/modules/scheduler/test.lua
index 36d1d97cd..364b87d2d 100644
--- a/tests/modules/scheduler/test.lua
+++ b/tests/modules/scheduler/test.lua
@@ -13,4 +13,18 @@ function test_runjobs(t)
     t:are_equal(count, 100)
 end
 
+function test_sleep(t)
+
+    --[[
+    local task = function (a)
+        local dt = os.mclock()
+        scheduler.sleep(500)
+        dt = os.mclock() - dt
+        t:require(dt > 100 and dt < 1000)
+    end
+    for i = 1, 3 do
+        scheduler.co_start(task)
+    end
+    scheduler.runloop()]]
+end
 
diff --git a/xmake/core/base/scheduler.lua b/xmake/core/base/scheduler.lua
index d3b2fd4f3..a2745c6e6 100644
--- a/xmake/core/base/scheduler.lua
+++ b/xmake/core/base/scheduler.lua
@@ -26,6 +26,7 @@ local _coroutine = _coroutine or {}
 local table     = require("base/table")
 local option    = require("base/option")
 local string    = require("base/string")
+local poller    = require("base/poller")
 local coroutine = require("base/coroutine")
 
 -- new a coroutine instance
@@ -85,6 +86,7 @@ end
 -- wait the current coroutine
 function scheduler:_co_wait()
     -- TODO
+    self:co_suspend()
 end
 
 -- wake the given coroutine
@@ -145,13 +147,60 @@ end
 -- sleep some times (ms)
 function scheduler:sleep(ms)
 
-    -- TODO
-    self:wait()
+    print("sleep", ms)
+
+    self:_co_wait()
     return true
 end
 
+-- the loop is started?
+function scheduler:is_started()
+    return self._STARTED
+end
+
+-- stop the scheduler loop
+function scheduler:stop()
+    -- TODO post a kill signal to poller
+    self._STARTED = false
+end
+
 -- run loop, schedule coroutine with socket/io and sub-processes
 function scheduler:runloop(opt)
+
+    -- start loop
+    self._STARTED = true
+
+    -- run loop
+    opt = opt or {}
+    local ok = true
+    local errors = nil
+    local timeout = opt.timeout or -1
+    while self._STARTED do 
+
+        -- wait events
+        local count, events = poller:wait(timeout)
+        if count < 0 then
+            ok = false
+            errors = events
+            break
+        end
+
+        -- spank the timer and trigger all timeout tasks
+        -- TODO
+
+        -- resume all suspended tasks with events
+        -- TODO
+    end
+
+    -- mark the loop as stopped first
+    self._STARTED = false
+
+    -- TODO resume all suspended tasks
+    -- we cannot suspend them now, all tasks will be exited directly and free all resources.
+    -- ...
+
+    -- finished
+    return ok, errors
 end
 
 -- return module: scheduler