add thread queue

2 files changed, 223 insertions, 0 deletions

diff --git a/xmake/core/base/thread.lua b/xmake/core/base/thread.lua
index c924724b8..2abed1df0 100644
--- a/xmake/core/base/thread.lua
+++ b/xmake/core/base/thread.lua
@@ -24,6 +24,7 @@ local _thread    = _thread or {}
 local _mutex     = _mutex or {}
 local _event     = _event or {}
 local _semaphore = _semaphore or {}
+local _queue     = _queue or {}
 
 -- load modules
 local io      = require("base/io")
@@ -109,6 +110,10 @@ function _thread:start()
         elseif type(arg) == "table" and arg._SEMAPHORE and arg.cdata then
             thread.semaphore_incref(arg:cdata())
             arg = {semaphore = true, name = arg:name(), caddr = libc.dataptr(arg:cdata())}
+        -- is queue? we can only pass cdata address
+        elseif type(arg) == "table" and arg._QUEUE and arg.cdata then
+            thread.queue_incref(arg:cdata())
+            arg = {queue = true, name = arg:name(), caddr = libc.dataptr(arg:cdata())}
         end
         table.insert(argv, arg)
     end
@@ -494,6 +499,137 @@ function _semaphore:__gc()
     end
 end
 
+-- new an queue
+function _queue.new(name, cdata)
+    local queue = table.inherit(_queue)
+    queue._NAME = name
+    queue._QUEUE = cdata
+    setmetatable(queue, _queue)
+    return queue
+end
+
+-- get the queue name
+function _queue:name()
+    return self._NAME
+end
+
+-- get the cdata
+function _queue:cdata()
+    return self._QUEUE
+end
+
+-- get queue size
+function _queue:size()
+    local ok, errors = self:_ensure_opened()
+    if not ok then
+        return nil, errors
+    end
+
+    return thread.queue_size(self:cdata())
+end
+
+-- is empty queue?
+function _queue:empty()
+    local ok, errors = self:_ensure_opened()
+    if not ok then
+        return nil, errors
+    end
+
+    return thread.queue_size(self:cdata()) == 0
+end
+
+-- clear queue
+function _queue:clear()
+    local ok, errors = self:_ensure_opened()
+    if not ok then
+        return false, errors
+    end
+
+    local ok, errors = thread.queue_clear(self:cdata())
+    if not ok then
+        return false, string.format("%s: clear failed, errors: %s!", self, errors or "unknown")
+    end
+    return ok
+end
+
+-- push queue item
+function _queue:push(value)
+    local ok, errors = self:_ensure_opened()
+    if not ok then
+        return false, errors
+    end
+
+    if type(value) == "table" then
+        value = string.serialize(value, {strip = true, indent = false})
+        if value == nil then
+            return false, string.format("%s: cannot serialize value: %s", self, value)
+        end
+        value = "__table_" .. value
+    end
+
+    local ok, errors = thread.queue_push(self:cdata(), value)
+    if not ok then
+        return false, string.format("%s: push item failed, errors: %s!", self, errors or "unknown")
+    end
+    return ok
+end
+
+-- pop queue item
+function _queue:pop()
+    local ok, errors = self:_ensure_opened()
+    if not ok then
+        return nil, errors or "unknown"
+    end
+
+    local value, errors = thread.queue_pop(self:cdata())
+    if value == nil and errors then
+        return nil, string.format("%s: push item failed, errors: %s!", self, errors or "unknown")
+    end
+
+    if type(value) == "string" and value:startswith("__table_") then
+        value = value:sub(9)
+        value, errors = string.deserialize(value)
+        if not value then
+            return nil, string.format("invalid queue item, %s!", errors or "unknown")
+        end
+    end
+    return value
+end
+
+-- close queue
+function _queue:close()
+    local ok, errors = self:_ensure_opened()
+    if not ok then
+        return false, errors
+    end
+
+    ok = thread.queue_exit(self:cdata())
+    if ok then
+        self._QUEUE = nil
+    end
+    return ok
+end
+
+-- ensure the file is opened
+function _queue:_ensure_opened()
+    if not self:cdata() then
+        return false, string.format("%s: has been closed!", self)
+    end
+    return true
+end
+
+-- tostring(queue)
+function _queue:__tostring()
+    return "<queue: " .. (self:name() or tostring(self:cdata())) .. ">"
+end
+
+-- gc(queue)
+function _queue:__gc()
+    if self:cdata() and thread.queue_exit(self:cdata()) then
+        self._QUEUE = nil
+    end
+end
+
 -- new a thread
 --
 -- @param callback      the thread callback
@@ -576,6 +712,8 @@ function thread._run_thread(callback_str, callinfo_str)
                 arg = _event.new(arg.name, libc.ptraddr(arg.caddr))
             elseif type(arg) == "table" and arg.semaphore and arg.caddr then
                 arg = _semaphore.new(arg.name, libc.ptraddr(arg.caddr))
+            elseif type(arg) == "table" and arg.queue and arg.caddr then
+                arg = _queue.new(arg.name, libc.ptraddr(arg.caddr))
             end
             table.insert(newargv, arg)
         end
@@ -616,6 +754,16 @@ function thread.semaphore(name, value)
     end
 end
 
+-- open a queue
+function thread.queue(name)
+    local queue = thread.queue_init()
+    if queue then
+        return _queue.new(name, queue)
+    else
+        return nil, string.format("cannot open queue: %s", os.strerror())
+    end
+end
+
 -- return module
 return thread
 
diff --git a/xmake/core/sandbox/modules/import/core/base/thread.lua b/xmake/core/sandbox/modules/import/core/base/thread.lua
index cb8de9ac6..a10dc1511 100644
--- a/xmake/core/sandbox/modules/import/core/base/thread.lua
+++ b/xmake/core/sandbox/modules/import/core/base/thread.lua
@@ -30,6 +30,7 @@ local sandbox_core_base_thread_instance  = sandbox_core_base_thread_instance or
 local sandbox_core_base_thread_mutex     = sandbox_core_base_thread_mutex or {}
 local sandbox_core_base_thread_event     = sandbox_core_base_thread_event or {}
 local sandbox_core_base_thread_semaphore = sandbox_core_base_thread_semaphore or {}
+local sandbox_core_base_thread_queue     = sandbox_core_base_thread_queue or {}
 
 -- export the thread status
 sandbox_core_base_thread.STATUS_READY     = thread.STATUS_READY
@@ -161,6 +162,59 @@ function sandbox_core_base_thread_semaphore.close(semaphore)
     end
 end
 
+-- get queue size
+function sandbox_core_base_thread_queue.size(queue)
+    local size, errors = queue:_size()
+    if not size then
+        raise(errors)
+    end
+    return size
+end
+
+-- is empty queue?
+function sandbox_core_base_thread_queue.empty(queue)
+    local ok, errors = queue:_empty()
+    if ok == nil then
+        raise(errors)
+    end
+    return ok
+end
+
+-- clear queue
+function sandbox_core_base_thread_queue.clear(queue)
+    local ok, errors = queue:_clear()
+    if not ok then
+        raise(errors)
+    end
+    return ok
+end
+
+-- push queue item
+function sandbox_core_base_thread_queue.push(queue, value)
+    local ok, errors = queue:_push(value)
+    if not ok then
+        raise(errors)
+    end
+    return ok
+end
+
+-- pop queue item
+function sandbox_core_base_thread_queue.pop(queue)
+    local value, errors = queue:_pop()
+    if value == nil and errors then
+        raise(errors)
+    end
+    return value
+end
+
+-- close queue
+function sandbox_core_base_thread_queue.close(queue)
+    local ok, errors = queue:_close()
+    if not ok then
+        raise(errors)
+    end
+end
+
 -- new thread
 function sandbox_core_base_thread.new(callback, opt)
     local instance, errors = thread.new(callback, opt)
@@ -257,6 +311,27 @@ function sandbox_core_base_thread.semaphore(name, value)
     return semaphore
 end
 
+-- open a queue
+function sandbox_core_base_thread.queue(name)
+    local queue, errors = thread.queue(name)
+    if not queue then
+        raise(errors)
+    end
+
+    -- hook filequeue interfaces
+    local hooked = {}
+    for name, func in pairs(sandbox_core_base_thread_queue) do
+        if not name:startswith("_") and type(func) == "function" then
+            hooked["_" .. name] = queue["_" .. name] or queue[name]
+            hooked[name] = func
+        end
+    end
+    for name, func in pairs(hooked) do
+        queue[name] = func
+    end
+    return queue
+end
+
 -- return module
 return sandbox_core_base_thread