support process for scheduler

3 files changed, 140 insertions, 7 deletions

diff --git a/tests/modules/process/sched_process.lua b/tests/modules/process/sched_process.lua
new file mode 100644
index 000000000..c9b629b03
--- /dev/null
+++ b/tests/modules/process/sched_process.lua
@@ -0,0 +1,15 @@
+import("core.base.scheduler")
+
+function _session(id, program, ...)
+    local proc = process.openv(program, table.pack(...))
+    local ok, status = proc:wait(-1)
+    print("%s/%d: %d, status: %d", proc, id, ok, status)
+    proc:close()
+end
+
+function main(program, ...)
+    for i = 1, 10 do
+        scheduler.co_start(_session, i, program, ...)
+    end
+    scheduler.runloop()
+end
diff --git a/xmake/core/base/process.lua b/xmake/core/base/process.lua
index 0d4cdbf26..99f756fdc 100644
--- a/xmake/core/base/process.lua
+++ b/xmake/core/base/process.lua
@@ -28,6 +28,7 @@ local path      = require("base/path")
 local utils     = require("base/utils")
 local string    = require("base/string")
 local coroutine = require("base/coroutine")
+local scheduler = require("base/scheduler")
 
 -- save original interfaces
 process._open       = process._open or process.open
@@ -70,24 +71,60 @@ end
 -- @return          ok, status
 --
 function _subprocess:wait(timeout)
-    if not self._PROC then
-        return -1, 0, string.format("subprocess(%s) has been closed!", self:name())
+
+    -- ensure opened
+    local ok, errors = self:_ensure_opened()
+    if not ok then
+        return -1, errors
+    end
+
+    -- wait events
+    local result = -1
+    local status_or_errors = nil
+    if scheduler:co_running() then
+        result, status_or_errors = scheduler:poller_waitproc(self, timeout or -1)
+    else
+        result, status_or_errors = process._wait(self:cdata(), timeout or -1)
     end
-    return process._wait(self._PROC, timeout or -1)
+    if result < 0 and status_or_errors then
+        status_or_errors = string.format("%s: %s", self, status_or_errors)
+    end
+    return result, status_or_errors
 end
 
 -- close subprocess
 function _subprocess:close(timeout)
-    if not self._PROC then
-        return false, string.format("subprocess(%s) has been closed!", self:name())
+
+    -- ensure opened
+    local ok, errors = self:_ensure_opened()
+    if not ok then
+        return false, errors
     end
-    local ok = process._close(self._PROC)
+
+    -- cancel pipe events from the scheduler
+    if scheduler:co_running() then
+        ok, errors = scheduler:poller_cancel(self)
+        if not ok then
+            return false, errors
+        end
+    end
+
+    -- close process
+    ok = process._close(self:cdata())
     if ok then
         self._PROC = nil
     end
     return ok
 end
 
+-- ensure the process is opened
+function _subprocess:_ensure_opened()
+    if not self:cdata() then
+        return false, string.format("%s: has been closed!", self)
+    end
+    return true
+end
+
 -- tostring(subprocess)
 function _subprocess:__tostring()
     return "<subprocess: " .. self:name() .. ">"
diff --git a/xmake/core/base/scheduler.lua b/xmake/core/base/scheduler.lua
index 6d966c4bb..1464ae2dc 100644
--- a/xmake/core/base/scheduler.lua
+++ b/xmake/core/base/scheduler.lua
@@ -156,6 +156,22 @@ function scheduler:_poller_events_cb(obj, events)
     local pollerdata = self:_poller_data(obj)
     assert(pollerdata, string.format("%s: cannot get poller data!", obj))
 
+    -- is process object?
+    if obj:otype() == poller.OT_PROC then
+
+        -- resume coroutine and return the process exit status
+        pollerdata.proc_status = events
+
+        -- waiting process? resume this coroutine
+        if pollerdata.co_waiting then
+            local co_waiting = pollerdata.co_waiting
+            pollerdata.co_waiting = nil
+            return self:_poller_resume_co(co_waiting, 1)
+        else 
+            pollerdata.proc_pending = 1
+        end
+        return ;
+    end
 
     -- get poller object events
     local events_prev_wait = pollerdata.poller_events_wait
@@ -401,13 +417,78 @@ function scheduler:poller_wait(obj, events, timeout)
     return self:co_suspend()
 end
 
+-- wait poller object/process status
+function scheduler:poller_waitproc(obj, timeout)
+
+    -- get the running coroutine
+    local running = self:co_running()
+    if not running then
+        return -1, "we must call poller_wait() in coroutine with scheduler!"
+    end
+
+    -- is stopped?
+    if not self._STARTED then
+        return -1, "the scheduler is stopped!"
+    end
+
+    -- check the object type
+    local otype = obj:otype()
+    if otype ~= poller.OT_PROC then
+        return -1, string.format("%s: invalid object type(%d)!", obj, otype)
+    end
+
+    -- get and allocate poller object data
+    local pollerdata = self:_poller_data(obj)
+    if not pollerdata then
+        pollerdata = {proc_pending = 0, proc_status = 0}
+        self:_poller_data_set(obj, pollerdata)
+    end
+
+    -- has pending process status?
+    if pollerdata.proc_pending ~= 0 then
+        pollerdata.proc_pending = 0
+        return 1, pollerdata.proc_status
+    end
+
+    -- insert poller object to poller for waiting process
+    local ok, errors = poller:insert(obj, 0, self._poller_events_cb)
+    if not ok then
+        return -1, errors
+    end
+
+    -- register timeout task to timer
+    local timer_task = nil
+    if timeout > 0 then
+        timer_task = self:_timer():post(function (cancel) 
+            if not cancel and running:is_suspended() then
+                pollerdata.co_waiting = nil
+                self:_co_tasks_suspended():remove(running)
+                self:co_resume(running, 0)
+            end
+        end, timeout)
+    end
+    running:_timer_task_set(timer_task)
+
+    -- set process status
+    pollerdata.proc_status  = 0
+    pollerdata.proc_pending = 0
+    pollerdata.co_waiting   = running
+
+    -- save the suspended coroutine
+    self:_co_tasks_suspended():insert(running)
+
+    -- wait
+    local ok = self:co_suspend()
+    return ok, pollerdata.proc_status
+end
+
 -- cancel poller object events
 function scheduler:poller_cancel(obj)
 
     -- reset the pollerdata data
     local pollerdata = self:_poller_data(obj)
     if pollerdata then
-        if pollerdata.poller_events_wait ~= 0 then
+        if pollerdata.poller_events_wait ~= 0 or obj:otype() == poller.OT_PROC then
             local ok, errors = poller:remove(obj)
             if not ok then
                 return false, errors