From f5cb55bc3ea8333c64e45bdbae2890b6ca1cd3c4 Mon Sep 17 00:00:00 2001 From: ruki Date: Sat, 4 Oct 2025 22:02:39 +0800 Subject: add runjobs benchmark --- core/src/xmake/os/getenvs.c | 2 -- tests/benchmarks/async/runjobs.lua | 33 +++++++++++++++++++++++++++++++++ 2 files changed, 33 insertions(+), 2 deletions(-) create mode 100644 tests/benchmarks/async/runjobs.lua 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 + -- cgit v1.3.1 From 1705c0f34f8910c399b680f66dd60f453f91904c Mon Sep 17 00:00:00 2001 From: ruki Date: Sat, 4 Oct 2025 22:55:08 +0800 Subject: add coroutine semaphore --- xmake/core/base/scheduler.lua | 42 ++++++++++++++++++++++ .../sandbox/modules/import/core/base/scheduler.lua | 42 ++++++++++++++++++++++ 2 files changed, 84 insertions(+) diff --git a/xmake/core/base/scheduler.lua b/xmake/core/base/scheduler.lua index 63c9fe009..28c58f3c8 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,42 @@ 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 + setmetatable(instance, _semaphore) + return instance +end + +-- get the semaphore name +function _semaphore:name() + return self._NAME or "none" +end + +-- get the semaphore value +function _semaphore:value() + return self._VALUE or 0 +end + +-- post the semaphore value +function _semaphore:post(value) + value = self._VALUE + value + self._VALUE = value + return value +end + +-- wait the semaphore +function _semaphore:wait(timeout) + return 0 +end + +-- tostring(semaphore) +function _semaphore:__tostring() + return string.format("", self:name(), self:value()) +end + -- new a coroutine instance function _coroutine.new(name, thread) local instance = table.inherit(_coroutine) @@ -726,6 +763,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 -- cgit v1.3.1 From e1adff76f2efecf0c82bd7a9db9ab60497be6926 Mon Sep 17 00:00:00 2001 From: ruki Date: Sat, 4 Oct 2025 23:07:02 +0800 Subject: add semaphore test --- tests/modules/scheduler/semaphore.lua | 34 ++++++++++++++++++++++++++++++++++ 1 file changed, 34 insertions(+) create mode 100644 tests/modules/scheduler/semaphore.lua diff --git a/tests/modules/scheduler/semaphore.lua b/tests/modules/scheduler/semaphore.lua new file mode 100644 index 000000000..097b0e360 --- /dev/null +++ b/tests/modules/scheduler/semaphore.lua @@ -0,0 +1,34 @@ +import("core.base.scheduler") + +function _loop(semaphore, id) + print("[%d]: start", id) + while true do + print("[%d]: wait ..", id) + --semaphore:wait(-1) + os.sleep(1000) + end + print("[%d]: end", id) +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 + -- cgit v1.3.1 From 6de0d5453f0d7a85f3d4fc363eb960751fe2709e Mon Sep 17 00:00:00 2001 From: ruki Date: Sat, 4 Oct 2025 23:52:51 +0800 Subject: impl coroutine semaphore --- tests/modules/scheduler/semaphore.lua | 4 +- xmake/core/base/scheduler.lua | 73 +++++++++++++++++++++++++++++++---- 2 files changed, 67 insertions(+), 10 deletions(-) diff --git a/tests/modules/scheduler/semaphore.lua b/tests/modules/scheduler/semaphore.lua index 097b0e360..788650019 100644 --- a/tests/modules/scheduler/semaphore.lua +++ b/tests/modules/scheduler/semaphore.lua @@ -4,8 +4,8 @@ function _loop(semaphore, id) print("[%d]: start", id) while true do print("[%d]: wait ..", id) - --semaphore:wait(-1) - os.sleep(1000) + local value = semaphore:wait(-1) + print("[%d]: -> triggered, value: %d ..", id, value) end print("[%d]: end", id) end diff --git a/xmake/core/base/scheduler.lua b/xmake/core/base/scheduler.lua index 28c58f3c8..e840e2445 100644 --- a/xmake/core/base/scheduler.lua +++ b/xmake/core/base/scheduler.lua @@ -36,9 +36,10 @@ 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 + local instance = table.inherit(_semaphore) + instance._NAME = name + instance._VALUE = value or 0 + instance._WAITING = hashset.new() setmetatable(instance, _semaphore) return instance end @@ -48,20 +49,76 @@ function _semaphore:name() return self._NAME or "none" end --- get the semaphore value -function _semaphore:value() - return self._VALUE or 0 -end - -- post the semaphore value function _semaphore:post(value) value = self._VALUE + value self._VALUE = value + if value > 0 then + local pending = {} + local waiting = self._WAITING + for item in waiting:items() do + if #pending < value then + table.insert(pending, item) + end + end + for _, item in ipairs(pending) do + scheduler:co_resume(item) + end + end return 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 return 0 end -- cgit v1.3.1 From af8d455a20ebf6d64333737979d13d41abfaf920 Mon Sep 17 00:00:00 2001 From: ruki Date: Sat, 4 Oct 2025 23:58:52 +0800 Subject: fix waiting list --- xmake/core/base/scheduler.lua | 1 + 1 file changed, 1 insertion(+) diff --git a/xmake/core/base/scheduler.lua b/xmake/core/base/scheduler.lua index e840e2445..a81b4652b 100644 --- a/xmake/core/base/scheduler.lua +++ b/xmake/core/base/scheduler.lua @@ -119,6 +119,7 @@ function _semaphore:wait(timeout) break end end + self._WAITING:remove(running) return 0 end -- cgit v1.3.1 From 49e117984150d88b8151f317c866935024e2367c Mon Sep 17 00:00:00 2001 From: ruki Date: Sun, 5 Oct 2025 00:12:27 +0800 Subject: fix semaphore tostring --- xmake/core/base/scheduler.lua | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/xmake/core/base/scheduler.lua b/xmake/core/base/scheduler.lua index a81b4652b..aaeeed5a0 100644 --- a/xmake/core/base/scheduler.lua +++ b/xmake/core/base/scheduler.lua @@ -125,7 +125,7 @@ end -- tostring(semaphore) function _semaphore:__tostring() - return string.format("", self:name(), self:value()) + return string.format("", self:name(), self._VALUE) end -- new a coroutine instance -- cgit v1.3.1 From 137a9d0875ac96e3636410479d1551f89acf3632 Mon Sep 17 00:00:00 2001 From: ruki Date: Sun, 5 Oct 2025 00:15:43 +0800 Subject: improve semaphore post --- tests/modules/scheduler/semaphore.lua | 1 - xmake/core/base/scheduler.lua | 12 +++++++----- 2 files changed, 7 insertions(+), 6 deletions(-) diff --git a/tests/modules/scheduler/semaphore.lua b/tests/modules/scheduler/semaphore.lua index 788650019..45cc43e1e 100644 --- a/tests/modules/scheduler/semaphore.lua +++ b/tests/modules/scheduler/semaphore.lua @@ -7,7 +7,6 @@ function _loop(semaphore, id) local value = semaphore:wait(-1) print("[%d]: -> triggered, value: %d ..", id, value) end - print("[%d]: end", id) end function _input(semaphore) diff --git a/xmake/core/base/scheduler.lua b/xmake/core/base/scheduler.lua index aaeeed5a0..5eaa60415 100644 --- a/xmake/core/base/scheduler.lua +++ b/xmake/core/base/scheduler.lua @@ -51,21 +51,23 @@ end -- post the semaphore value function _semaphore:post(value) - value = self._VALUE + value - self._VALUE = value - if value > 0 then + 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 < value then + 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 value + return new_value end -- wait the semaphore -- cgit v1.3.1