Merge pull request #6891 from xmake-io/sem

Add coroutine semaphore

5 files changed, 210 insertions, 2 deletions

diff --git a/core/src/xmake/os/getenvs.c b/core/src/xmake/os/getenvs.c
index 23fdf677e..7e6d611d6 100644
--- a/core/src/xmake/os/getenvs.c
+++ b/core/src/xmake/os/getenvs.c
@@ -182,7 +182,5 @@ tb_int_t xm_os_getenvs(lua_State* lua)
         }
     }
 #endif
-
-    // ok
     return 1;
 }
diff --git a/tests/benchmarks/async/runjobs.lua b/tests/benchmarks/async/runjobs.lua
new file mode 100644
index 000000000..937697413
--- /dev/null
+++ b/tests/benchmarks/async/runjobs.lua
@@ -0,0 +1,33 @@
+import("async.runjobs")
+
+function test_run(total, comax)
+    local f = function () end
+    local t1 = os.mclock()
+    runjobs("test", f, {total = total, comax = comax})
+    t1 = os.mclock() - t1
+
+    local n = total
+    local t2 = os.mclock()
+    while n ~= 0 do
+        f()
+        n = n - 1
+    end
+    t2 = os.mclock() - t2
+    print("runjobs(%d/%d): %d ms, plain: %d ms", total, comax, t1, t2)
+end
+
+function test_run_proc(total, comax)
+    local f = function () os.runv(os.programfile(), {"--version"}) end
+    local t1 = os.mclock()
+    runjobs("test", f, {total = total, comax = comax})
+    t1 = os.mclock() - t1
+    print("runjobs_proc(%d/%d): %d ms", total, comax, t1)
+end
+
+function main()
+    test_run(10000, 1)
+    test_run(10000, 10)
+    test_run(10000, 100)
+    test_run_proc(1000, 10)
+end
+
diff --git a/tests/modules/scheduler/semaphore.lua b/tests/modules/scheduler/semaphore.lua
new file mode 100644
index 000000000..45cc43e1e
--- /dev/null
+++ b/tests/modules/scheduler/semaphore.lua
@@ -0,0 +1,33 @@
+import("core.base.scheduler")
+
+function _loop(semaphore, id)
+    print("[%d]: start", id)
+    while true do
+        print("[%d]: wait ..", id)
+        local value = semaphore:wait(-1)
+        print("[%d]:   -> triggered, value: %d ..", id, value)
+    end
+end
+
+function _input(semaphore)
+    while true do
+        if io.readable() then
+            local ch = io.read()
+            print("  -> post semaphore")
+            if ch then
+                semaphore:post(2)
+            end
+        else
+            os.sleep(1000)
+        end
+    end
+end
+
+function main()
+    local semaphore = scheduler.co_semaphore("", 1)
+    for i = 1, 10 do
+        scheduler.co_start(_loop, semaphore, i)
+    end
+    scheduler.co_start(_input, semaphore)
+end
+
diff --git a/xmake/core/base/scheduler.lua b/xmake/core/base/scheduler.lua
index 63c9fe009..5eaa60415 100644
--- a/xmake/core/base/scheduler.lua
+++ b/xmake/core/base/scheduler.lua
@@ -21,6 +21,7 @@
 -- define module: scheduler
 local scheduler  = scheduler or {}
 local _coroutine = _coroutine or {}
+local _semaphore = _semaphore or {}
 
 -- load modules
 local table     = require("base/table")
@@ -33,6 +34,102 @@ local hashset   = require("base/hashset")
 local coroutine = require("base/coroutine")
 local bit       = require("base/bit")
 
+-- new a semaphore instance
+function _semaphore.new(name, value)
+    local instance    = table.inherit(_semaphore)
+    instance._NAME    = name
+    instance._VALUE   = value or 0
+    instance._WAITING = hashset.new()
+    setmetatable(instance, _semaphore)
+    return instance
+end
+
+-- get the semaphore name
+function _semaphore:name()
+    return self._NAME or "none"
+end
+
+-- post the semaphore value
+function _semaphore:post(value)
+    local new_value = self._VALUE + value
+    self._VALUE = new_value
+    if new_value > 0 then
+        local pending = {}
+        local waiting = self._WAITING
+        for item in waiting:items() do
+            if #pending < new_value then
+                table.insert(pending, item)
+            else
+                break
+            end
+        end
+        for _, item in ipairs(pending) do
+            scheduler:co_resume(item)
+        end
+    end
+    return new_value
+end
+
+-- wait the semaphore
+function _semaphore:wait(timeout)
+
+    -- get the running coroutine
+    local running = scheduler:co_running()
+    if not running then
+        return -1, "we must call semaphore:wait() in coroutine with scheduler!"
+    end
+
+    -- is stopped?
+    if not scheduler._STARTED then
+        return -1, "the scheduler is stopped!"
+    end
+
+    -- update value
+    local value = self._VALUE
+    if value > 0 then
+        self._VALUE = value - 1
+        return value
+    end
+
+    -- no signal? return immediately if timeout is zero
+    if timeout == 0 then
+        return 0
+    end
+
+    -- wait semaphore
+    self._WAITING:insert(running)
+    if timeout > 0 then
+        scheduler:_timer():post(function (cancel)
+            if running:is_suspended() then
+                return scheduler:co_resume(running, true)
+            end
+            return true
+        end, timeout)
+    end
+
+    while true do
+        local timeout = scheduler:co_suspend()
+
+        local value = self._VALUE
+        if value > 0 then
+            self._VALUE = value - 1
+            self._WAITING:remove(running)
+            return value
+        end
+
+        if timeout then
+            break
+        end
+    end
+    self._WAITING:remove(running)
+    return 0
+end
+
+-- tostring(semaphore)
+function _semaphore:__tostring()
+    return string.format("<co_semaphore: %s/%d>", self:name(), self._VALUE)
+end
+
 -- new a coroutine instance
 function _coroutine.new(name, thread)
     local instance   = table.inherit(_coroutine)
@@ -726,6 +823,11 @@ function scheduler:co_count()
     return self._CO_COUNT or 0
 end
 
+-- new a coroutine semaphore
+function scheduler:co_semaphore(name, value)
+    return _semaphore.new(name, value)
+end
+
 -- wait poller object io events, only for socket and pipe object
 function scheduler:poller_wait(obj, events, timeout)
 
diff --git a/xmake/core/sandbox/modules/import/core/base/scheduler.lua b/xmake/core/sandbox/modules/import/core/base/scheduler.lua
index b5d72286c..222bee391 100644
--- a/xmake/core/sandbox/modules/import/core/base/scheduler.lua
+++ b/xmake/core/sandbox/modules/import/core/base/scheduler.lua
@@ -20,6 +20,7 @@
 
 -- define module
 local sandbox_core_base_scheduler = sandbox_core_base_scheduler or {}
+local sandbox_core_base_scheduler_semaphore = sandbox_core_base_scheduler_semaphore or {}
 
 -- load modules
 local poller    = require("base/poller")
@@ -31,6 +32,41 @@ sandbox_core_base_scheduler.OT_SOCK = poller.OT_SOCK
 sandbox_core_base_scheduler.OT_PIPE = poller.OT_PIPE
 sandbox_core_base_scheduler.OT_PROC = poller.OT_PROC
 
+-- wrap semaphore
+function _semaphore_wrap(semaphore)
+
+    -- hook semaphore interfaces
+    local hooked = {}
+    for name, func in pairs(sandbox_core_base_scheduler_semaphore) do
+        if not name:startswith("_") and type(func) == "function" then
+            hooked["_" .. name] = semaphore["_" .. name] or semaphore[name]
+            hooked[name] = func
+        end
+    end
+    for name, func in pairs(hooked) do
+        semaphore[name] = func
+    end
+    return semaphore
+end
+
+-- post semaphore
+function sandbox_core_base_scheduler_semaphore.post(semaphore, value)
+    local result, errors = semaphore:_post(value)
+    if result < 0 and errors then
+        raise(errors)
+    end
+    return result
+end
+
+-- wait semaphore
+function sandbox_core_base_scheduler_semaphore.wait(semaphore, timeout)
+    local ok, errors = semaphore:_wait(timeout)
+    if ok < 0 and errors then
+        raise(errors)
+    end
+    return ok
+end
+
 -- start a new coroutine task
 function sandbox_core_base_scheduler.co_start(cotask, ...)
     local co, errors = scheduler:co_start(cotask, ...)
@@ -136,5 +172,11 @@ function sandbox_core_base_scheduler.co_count()
     return scheduler:co_count()
 end
 
+-- new a coroutine semaphore
+function sandbox_core_base_scheduler.co_semaphore(name, value)
+    local semaphore = scheduler:co_semaphore(name, value)
+    return _semaphore_wrap(semaphore)
+end
+
 -- return module
 return sandbox_core_base_scheduler