impl coroutine semaphore

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