rewrite scheduler

4 files changed, 75 insertions, 234 deletions

diff --git a/tests/modules/scheduler/test.lua b/tests/modules/scheduler/test.lua
index d61b663e3..36d1d97cd 100644
--- a/tests/modules/scheduler/test.lua
+++ b/tests/modules/scheduler/test.lua
@@ -6,12 +6,10 @@ function test_runjobs(t)
     local task = function (a)
         t:are_equal(a, "xmake!")
         count = count + 1
-        coroutine.yield()
     end
     for i = 1, 100 do
-        scheduler.run(task, "xmake!")
+        scheduler.co_start(task, "xmake!")
     end
-    scheduler.runloop()
     t:are_equal(count, 100)
 end
 
diff --git a/xmake/core/base/scheduler.lua b/xmake/core/base/scheduler.lua
index 8d23e8ddb..d3b2fd4f3 100644
--- a/xmake/core/base/scheduler.lua
+++ b/xmake/core/base/scheduler.lua
@@ -23,16 +23,16 @@ local scheduler  = scheduler or {}
 local _coroutine = _coroutine or {}
 
 -- load modules
-local dlist     = require("base/dlist")
 local table     = require("base/table")
 local option    = require("base/option")
 local string    = require("base/string")
 local coroutine = require("base/coroutine")
 
 -- new a coroutine instance
-function _coroutine.new(rawco)
+function _coroutine.new(name, thread)
     local instance   = table.inherit(_coroutine)
-    instance._RAWCO  = rawco 
+    instance._NAME   = name
+    instance._THREAD = thread
     setmetatable(instance, _coroutine)
     return instance
 end
@@ -48,13 +48,13 @@ function _coroutine:name_set(name)
 end
 
 -- get the raw coroutine thread 
-function _coroutine:rawco()
-    return self._RAWCO
+function _coroutine:thread()
+    return self._THREAD
 end
 
 -- get the coroutine status
 function _coroutine:status()
-    return coroutine.status(self:rawco())
+    return coroutine.status(self:thread())
 end
 
 -- is dead?
@@ -68,266 +68,90 @@ function _coroutine:is_running()
 end
 
 -- is suspended?
-function _coroutine:is_suspended()
-    return self:status() == "suspended"
-end
-
--- resume the coroutine
-function _coroutine:resume(...)
-    return coroutine.resume(self:rawco(), ...)
-end
-
--- yield the coroutine
-function _coroutine:yield(...)
-    return coroutine.yield(self:rawco(), ...)
+function _coroutine:is_suspend()
+    return self:status() == "suspend"
 end
 
 -- tostring(socket)
 function _coroutine:__tostring()
-    return string.format("<co: %s/%s>", self:rawco(), self:name())
+    return string.format("<co: %s/%s>", self:thread(), self:name())
 end
 
 -- gc(coroutine)
 function _coroutine:__gc()
-    self._RAWCO = nil
-end
-
--- get the main loop coroutine
-function scheduler:_co_mainloop()
-    return self._CO_MAINLOOP
-end
-
--- create the coroutine instance
-function scheduler:_co_create(func)
-    local rawco = coroutine.create(func)
-    if not rawco then
-        return nil, "create coroutine failed!"
-    end
-    return _coroutine.new(rawco)
-end
-
--- suspend the current coroutine 
-function scheduler:_co_suspend(...)
-
-    -- get the next ready coroutine first
-    local co_next = self:_co_next_ready()
-
-    -- make the running coroutine as suspended
-    self:_co_mark_suspended(self:running())
-
-    -- switch to next coroutine 
-    if (co_next ~= self:running()) then
-        return self:_co_switch(co_next, ...)
-    end
-    
-    -- no more coroutine? switch to mainloop coroutine
-    return self:_co_switch(self:_co_mainloop(), ...)
-end
-
--- TODO
--- resume the given coroutine
-function scheduler:_co_resume()
+    self._THREAD = nil
 end
 
--- switch to the given coroutine
-function scheduler:_co_switch(co, ...)
-
-    -- mark it as the running coroutine
-    self._RUNNING = co
-
-    -- switch to this coroutine
-    return co:resume(...)
-end
-
--- TODO we need support socket/pipe io and processes as same time
-function scheduler:_poller_loop()
-    print("poller loop")
-end
-
--- ensure to start the poller loop
-function scheduler:_poller_loop_ensure()
-
-    -- ensure to run on coroutine with scheduler
-    if not self:running() then
-        return false, "please wait events on coroutine with scheduler!"
-    end
-
-    -- start the poller loop
-    if not self._POLLER_STARTED then
-        local co, errors = self:run(self._poller_loop, self)
-        if not co then
-            return false, "start poller loop failed!"
-        end
-        co:name_set("poller_loop")
-        self._POLLER_STARTED = true
-    end
-    return true
-end
-
--- the scheduler loop coroutine
-function scheduler:_mainloop(opt)
-
-    -- run loop
-    local co_list_ready = self:_co_list_ready()
-    while co_list_ready:size() > 0 do
-
-        -- get the first ready coroutine
-        local co_ready = co_list_ready:first()
-
-        -- switch to this coroutine
-        local ok, result_or_errors = self:_co_switch(co_ready)
-        if not ok then
-            os.raise(result_or_errors)
-        end
-           
-        -- this coroutine has been finished? we remove it from the ready queue
-        if co_ready:is_dead() then
-            self:_co_mark_dead(co_ready)
-        end
-    end
+-- wait the current coroutine
+function scheduler:_co_wait()
+    -- TODO
 end
 
--- start scheduler loop
-function scheduler:_startloop()
-    self._RUNNING = nil
-    self._CO_MAINLOOP = nil
-    self._POLLER_STARTED = false
+-- wake the given coroutine
+function scheduler:_co_wake(co)
+    -- TODO
 end
 
--- stop scheduler loop
-function scheduler:_stoploop()
-    self._RUNNING = nil
-    self._CO_MAINLOOP = nil
-    self._POLLER_STARTED = false
+-- start a new coroutine task
+function scheduler:co_start(cotask, ...)
+    return self:co_start_named(nil, cotask, ...)
 end
 
--- get all ready coroutines list
--- 
--- ready: ready -> ready -> .. -> running -> .. -> ready -> ..->
---         |                                                    |   <-> poller_loop
---          ---------------------------<------------------------
---                                |        |
---                                 mainloop
---
-function scheduler:_co_list_ready()
-    local co_list_ready = self._CO_LIST_READY
-    if not co_list_ready then
-        co_list_ready = dlist()
-        self._CO_LIST_READY = co_list_ready
-    end
-    return co_list_ready
-end
-
--- get all suspended coroutines list
-function scheduler:_co_list_suspended()
-    local co_list_suspended = self._CO_LIST_SUSPENDED
-    if not co_list_suspended then
-        co_list_suspended = dlist()
-        self._CO_LIST_SUSPENDED = co_list_suspended
+-- start a new named coroutine task
+function scheduler:co_start_named(coname, cotask, ...)
+    local co = _coroutine.new(coname, coroutine.create(cotask))
+    self:co_tasks()[co:thread()] = co
+    local ok, errors = scheduler:co_resume(co, ...)
+    self:co_tasks()[co:thread()] = nil
+    if not ok then
+        return nil, errors
     end
-    return co_list_suspended
+    return co
 end
 
--- get the next ready coroutine
-function scheduler:_co_next_ready()
-    return self:_co_list_ready():next(self:running())
+-- resume the given coroutine
+function scheduler:co_resume(co, ...)
+    return coroutine.resume(co:thread(), ...)
 end
 
--- mark the given coroutine as suspended
-function scheduler:_co_mark_suspended(co)
-
-    -- cannot be mainloop coroutine
-    assert(co ~= self:_co_mainloop())
-
-    -- remove this coroutine from the ready coroutines
-    self:_co_list_ready():remove(co)
-
-    -- append this coroutine to suspended coroutines
-    self:_co_list_suspended():push(co)
+-- suspend the current coroutine
+function scheduler:co_suspend(...)
+    return coroutine.yield(...)
 end
 
--- mark the given coroutine as dead
-function scheduler:_co_mark_dead(co)
-
-    -- cannot be mainloop coroutine
-    assert(co ~= self:_co_mainloop())
-
-    -- remove this coroutine from the ready coroutines
-    self:_co_list_ready():remove(co)
+-- get the current running coroutine 
+function scheduler:co_running()
+    local running = coroutine.running()
+    return running and self:co_tasks()[running] or nil 
 end
 
--- run new coroutine function, it will insert to the pending queue
-function scheduler:run(func, ...)
-    local argv = table.pack(...)
-    local co, errors = self:_co_create(function () return func(table.unpack(argv)) end)
-    if not co then
-        return nil, errors 
+-- get all coroutine tasks
+function scheduler:co_tasks()
+    local cotasks = self._CO_TASKS
+    if not cotasks then
+        cotasks = {}
+        self._CO_TASKS = cotasks
     end
-    self:_co_list_ready():push(co)
-    return co
-end
-
--- get the current running coroutine in scheduler
-function scheduler:running()
-    return self._RUNNING 
+    return cotasks
 end
 
 -- wait socket events
 function scheduler:waitsock(sock, events, timeout)
 
-    -- ensure the poller loop
-    local ok, errors = self:_poller_loop_ensure()
-    if not ok then
-        return -1, errors
-    end
-
     -- TODO
-    self:_co_suspend()
     return 0
 end
 
 -- sleep some times (ms)
 function scheduler:sleep(ms)
 
-    -- ensure the poller loop
-    local ok, errors = self:_poller_loop_ensure()
-    if not ok then
-        return false, errors
-    end
-
     -- TODO
-    self:_co_suspend()
+    self:wait()
     return true
 end
 
 -- run loop, schedule coroutine with socket/io and sub-processes
 function scheduler:runloop(opt)
-
-    -- ensure only one scheduler
-    if self._RUNNING then
-        return false, "there is already a running scheduler!"
-    end
-
-    -- start scheduler
-    self:_startloop()
-
-    -- create the main loop
-    local co_mainloop, errors = self:_co_create(self._mainloop)
-    if not co_mainloop then
-        return false, errors 
-    end
-    co_mainloop:name_set("mainloop")
-    self._CO_MAINLOOP = co_mainloop
-
-    -- start the main loop
-    local ok, errors = self:_co_switch(co_mainloop, self, opt)
-    if not ok then
-        return false, errors
-    end
-
-    -- stop scheduler
-    self:_stoploop()
-    return true
 end
 
 -- return module: scheduler
diff --git a/xmake/core/base/socket.lua b/xmake/core/base/socket.lua
index f86ca9bc1..ff362a132 100644
--- a/xmake/core/base/socket.lua
+++ b/xmake/core/base/socket.lua
@@ -476,7 +476,7 @@ function _instance:wait(events, timeout)
     -- wait events
     local events = -1
     local errors = nil
-    if scheduler:running() then
+    if scheduler:co_running() then
         events, errors = scheduler:waitsock(self._SOCK, events, timeout or -1)
     else
         events, errors = io.socket_wait(self._SOCK, events, timeout or -1)
diff --git a/xmake/core/sandbox/modules/import/core/base/scheduler.lua b/xmake/core/sandbox/modules/import/core/base/scheduler.lua
index 020199c7d..ff6576793 100644
--- a/xmake/core/sandbox/modules/import/core/base/scheduler.lua
+++ b/xmake/core/sandbox/modules/import/core/base/scheduler.lua
@@ -25,15 +25,39 @@ local sandbox_core_base_scheduler = sandbox_core_base_scheduler or {}
 local scheduler = require("base/scheduler")
 local raise     = require("sandbox/modules/raise")
 
--- run a new coroutine function 
-function sandbox_core_base_scheduler.run(func, ...)
-    local co, errors = scheduler:run(func, ...)
+-- start a new coroutine task 
+function sandbox_core_base_scheduler.co_start(cotask, ...)
+    local co, errors = scheduler:co_start(cotask, ...)
     if not co then
         raise(errors)
     end
     return co
 end
 
+-- start a new named coroutine task 
+function sandbox_core_base_scheduler.co_start_named(coname, cotask, ...)
+    local co, errors = scheduler:co_start_named(coname, cotask, ...)
+    if not co then
+        raise(errors)
+    end
+    return co
+end
+
+-- resume the given coroutine
+function sandbox_core_base_scheduler.co_resume(co, ...)
+    return scheduler:resume(co:thread(), ...)
+end
+
+-- suspend the current coroutine
+function sandbox_core_base_scheduler.co_suspend(...)
+    return scheduler:co_suspend(...)
+end
+
+-- get the current running coroutine 
+function sandbox_core_base_scheduler.co_running()
+    return scheduler:co_running()
+end
+
 -- sleep some times (ms)
 function sandbox_core_base_scheduler.sleep(ms)
     local ok, errors = scheduler:sleep(ms)
@@ -50,10 +74,5 @@ function sandbox_core_base_scheduler.runloop(opt)
     end
 end
 
--- get the current running coroutine instance
-function sandbox_core_base_scheduler.running()
-    return scheduler:running()
-end
-
 -- return module
 return sandbox_core_base_scheduler