summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authorruki <[email protected]>2025-08-28 23:14:13 +0800
committerruki <[email protected]>2025-08-28 23:14:13 +0800
commit628371d0edea22eb177fa2b1a6b9f28e8c0cff2d (patch)
tree2c8e834e76870f3cfa2e2b2b3b3342d69b0f83bf
parent8a331ab64fe560924b8bc3ab3dc4ac2831dd8920 (diff)
add thread queue
-rw-r--r--core/src/xmake/engine.c16
-rw-r--r--core/src/xmake/thread/prefix.h43
-rw-r--r--core/src/xmake/thread/queue_clear.c47
-rw-r--r--core/src/xmake/thread/queue_exit.c55
-rw-r--r--core/src/xmake/thread/queue_incref.c46
-rw-r--r--core/src/xmake/thread/queue_init.c87
-rw-r--r--core/src/xmake/thread/queue_pop.c92
-rw-r--r--core/src/xmake/thread/queue_push.c96
-rw-r--r--core/src/xmake/thread/queue_size.c46
-rw-r--r--tests/modules/thread/queue.lua30
-rw-r--r--xmake/core/base/thread.lua148
-rw-r--r--xmake/core/sandbox/modules/import/core/base/thread.lua75
12 files changed, 781 insertions, 0 deletions
diff --git a/core/src/xmake/engine.c b/core/src/xmake/engine.c
index 850be59f4..d3d48c7cc 100644
--- a/core/src/xmake/engine.c
+++ b/core/src/xmake/engine.c
@@ -368,6 +368,15 @@ tb_int_t xm_thread_semaphore_post(lua_State* lua);
tb_int_t xm_thread_semaphore_wait(lua_State* lua);
tb_int_t xm_thread_semaphore_incref(lua_State* lua);
+// the thread/queue functions
+tb_int_t xm_thread_queue_init(lua_State* lua);
+tb_int_t xm_thread_queue_exit(lua_State* lua);
+tb_int_t xm_thread_queue_size(lua_State* lua);
+tb_int_t xm_thread_queue_clear(lua_State* lua);
+tb_int_t xm_thread_queue_incref(lua_State* lua);
+tb_int_t xm_thread_queue_push(lua_State* lua);
+tb_int_t xm_thread_queue_pop(lua_State* lua);
+
// open cjson
__tb_extern_c_enter__
tb_int_t luaopen_cjson(lua_State *l);
@@ -678,6 +687,13 @@ static luaL_Reg const g_thread_functions[] =
, { "semaphore_post", xm_thread_semaphore_post }
, { "semaphore_wait", xm_thread_semaphore_wait }
, { "semaphore_incref", xm_thread_semaphore_incref }
+, { "queue_init", xm_thread_queue_init }
+, { "queue_exit", xm_thread_queue_exit }
+, { "queue_size", xm_thread_queue_size }
+, { "queue_clear", xm_thread_queue_clear }
+, { "queue_incref", xm_thread_queue_incref }
+, { "queue_push", xm_thread_queue_push }
+, { "queue_pop", xm_thread_queue_pop }
, { tb_null, tb_null }
};
diff --git a/core/src/xmake/thread/prefix.h b/core/src/xmake/thread/prefix.h
index be06d952e..8fc303b10 100644
--- a/core/src/xmake/thread/prefix.h
+++ b/core/src/xmake/thread/prefix.h
@@ -63,6 +63,40 @@ typedef struct __xm_thread_semaphore_t
}xm_thread_semaphore_t;
+// the thread queue type
+typedef struct __xm_thread_queue_t
+{
+ tb_queue_ref_t handle;
+ tb_atomic_t refn;
+
+}xm_thread_queue_t;
+
+// the thread queue item kind
+typedef enum __xm_thread_queue_item_kind_e
+{
+ XM_THREAD_QUEUE_ITEM_NIL = 0,
+ XM_THREAD_QUEUE_ITEM_BOOL = 1,
+ XM_THREAD_QUEUE_ITEM_INT = 2,
+ XM_THREAD_QUEUE_ITEM_NUM = 3,
+ XM_THREAD_QUEUE_ITEM_STR = 4
+
+}xm_thread_queue_item_kind_e;
+
+// the thread queue item type
+typedef struct __xm_thread_queue_item_t
+{
+ tb_uint32_t kind : 3;
+ tb_uint32_t size : 29;
+ union
+ {
+ tb_char_t* string;
+ tb_bool_t boolean;
+ lua_Integer integer;
+ lua_Number number;
+ } u;
+
+}xm_thread_queue_item_t;
+
// get the thread event from arguments
static __tb_inline__ xm_thread_event_t* xm_thread_event_get(lua_State* lua, tb_int_t index)
{
@@ -90,6 +124,15 @@ static __tb_inline__ xm_thread_semaphore_t* xm_thread_semaphore_get(lua_State* l
return thread_semaphore;
}
+// get the thread queue from arguments
+static __tb_inline__ xm_thread_queue_t* xm_thread_queue_get(lua_State* lua, tb_int_t index)
+{
+ xm_thread_queue_t* thread_queue = tb_null;
+ if (xm_lua_isinteger(lua, index)) thread_queue = (xm_thread_queue_t*)(tb_size_t)(tb_long_t)lua_tointeger(lua, index);
+ else if (xm_lua_ispointer(lua, index)) thread_queue = (xm_thread_queue_t*)xm_lua_topointer(lua, index);
+ return thread_queue;
+}
+
#endif
diff --git a/core/src/xmake/thread/queue_clear.c b/core/src/xmake/thread/queue_clear.c
new file mode 100644
index 000000000..bf70d12e1
--- /dev/null
+++ b/core/src/xmake/thread/queue_clear.c
@@ -0,0 +1,47 @@
+/*!A cross-platform build utility based on Lua
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ * Copyright (C) 2015-present, Xmake Open Source Community.
+ *
+ * @author ruki
+ * @file thread_queue_lock.c
+ *
+ */
+
+/* //////////////////////////////////////////////////////////////////////////////////////
+ * trace
+ */
+#define TB_TRACE_MODULE_NAME "thread_queue"
+#define TB_TRACE_MODULE_DEBUG (0)
+
+/* //////////////////////////////////////////////////////////////////////////////////////
+ * includes
+ */
+#include "prefix.h"
+
+/* //////////////////////////////////////////////////////////////////////////////////////
+ * implementation
+ */
+tb_int_t xm_thread_queue_clear(lua_State* lua)
+{
+ tb_assert_and_check_return_val(lua, 0);
+
+ xm_thread_queue_t* thread_queue = xm_thread_queue_get(lua, 1);
+ tb_assert_and_check_return_val(thread_queue && thread_queue->handle, 0);
+
+ tb_queue_clear(thread_queue->handle);
+ lua_pushboolean(lua, tb_true);
+ return 1;
+}
+
diff --git a/core/src/xmake/thread/queue_exit.c b/core/src/xmake/thread/queue_exit.c
new file mode 100644
index 000000000..8178ea22f
--- /dev/null
+++ b/core/src/xmake/thread/queue_exit.c
@@ -0,0 +1,55 @@
+/*!A cross-platform build utility based on Lua
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ * Copyright (C) 2015-present, Xmake Open Source Community.
+ *
+ * @author ruki
+ * @file thread_queue_exit.c
+ *
+ */
+
+/* //////////////////////////////////////////////////////////////////////////////////////
+ * trace
+ */
+#define TB_TRACE_MODULE_NAME "thread_queue"
+#define TB_TRACE_MODULE_DEBUG (0)
+
+/* //////////////////////////////////////////////////////////////////////////////////////
+ * includes
+ */
+#include "prefix.h"
+
+/* //////////////////////////////////////////////////////////////////////////////////////
+ * implementation
+ */
+tb_int_t xm_thread_queue_exit(lua_State* lua)
+{
+ tb_assert_and_check_return_val(lua, 0);
+
+ xm_thread_queue_t* thread_queue = xm_thread_queue_get(lua, 1);
+ tb_assert_and_check_return_val(thread_queue && thread_queue->handle, 0);
+
+ if (tb_atomic_fetch_and_sub(&thread_queue->refn, 1) == 1)
+ {
+ if (thread_queue->handle)
+ {
+ tb_queue_exit(thread_queue->handle);
+ thread_queue->handle = tb_null;
+ }
+ tb_free(thread_queue);
+ }
+ lua_pushboolean(lua, tb_true);
+ return 1;
+}
+
diff --git a/core/src/xmake/thread/queue_incref.c b/core/src/xmake/thread/queue_incref.c
new file mode 100644
index 000000000..4f06807c4
--- /dev/null
+++ b/core/src/xmake/thread/queue_incref.c
@@ -0,0 +1,46 @@
+/*!A cross-platform build utility based on Lua
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ * Copyright (C) 2015-present, Xmake Open Source Community.
+ *
+ * @author ruki
+ * @file thread_queue_incref.c
+ *
+ */
+
+/* //////////////////////////////////////////////////////////////////////////////////////
+ * trace
+ */
+#define TB_TRACE_MODULE_NAME "thread_queue"
+#define TB_TRACE_MODULE_DEBUG (0)
+
+/* //////////////////////////////////////////////////////////////////////////////////////
+ * includes
+ */
+#include "prefix.h"
+
+/* //////////////////////////////////////////////////////////////////////////////////////
+ * implementation
+ */
+tb_int_t xm_thread_queue_incref(lua_State* lua)
+{
+ tb_assert_and_check_return_val(lua, 0);
+
+ xm_thread_queue_t* thread_queue = xm_thread_queue_get(lua, 1);
+ tb_assert_and_check_return_val(thread_queue && thread_queue->handle, 0);
+
+ lua_pushboolean(lua, tb_atomic_fetch_and_add(&thread_queue->refn, 1) >= 1);
+ return 1;
+}
+
diff --git a/core/src/xmake/thread/queue_init.c b/core/src/xmake/thread/queue_init.c
new file mode 100644
index 000000000..4c4e9a9f6
--- /dev/null
+++ b/core/src/xmake/thread/queue_init.c
@@ -0,0 +1,87 @@
+/*!A cross-platform build utility based on Lua
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ * Copyright (C) 2015-present, Xmake Open Source Community.
+ *
+ * @author ruki
+ * @file queue_init.c
+ *
+ */
+
+/* //////////////////////////////////////////////////////////////////////////////////////
+ * trace
+ */
+#define TB_TRACE_MODULE_NAME "thread_queue"
+#define TB_TRACE_MODULE_DEBUG (0)
+
+/* //////////////////////////////////////////////////////////////////////////////////////
+ * includes
+ */
+#include "prefix.h"
+
+/* //////////////////////////////////////////////////////////////////////////////////////
+ * private implementation
+ */
+static tb_void_t xm_thread_queue_item_free(tb_element_ref_t element, tb_pointer_t buff)
+{
+ xm_thread_queue_item_t* item = (xm_thread_queue_item_t*)buff;
+ if (item)
+ {
+ if (item->kind == XM_THREAD_QUEUE_ITEM_STR)
+ {
+ if (item->u.string) tb_free((tb_pointer_t)item->u.string);
+ item->u.string = tb_null;
+ }
+ item->size = 0;
+ }
+}
+
+/* //////////////////////////////////////////////////////////////////////////////////////
+ * implementation
+ */
+tb_int_t xm_thread_queue_init(lua_State* lua)
+{
+ tb_assert_and_check_return_val(lua, 0);
+
+ tb_bool_t ok = tb_false;
+ xm_thread_queue_t* thread_queue = tb_null;
+ do
+ {
+ thread_queue = tb_malloc0_type(xm_thread_queue_t);
+ tb_assert_and_check_break(thread_queue);
+
+ thread_queue->refn = 1;
+ thread_queue->handle = tb_queue_init(0, tb_element_mem(sizeof(xm_thread_queue_item_t), xm_thread_queue_item_free, tb_null));
+ tb_assert_and_check_break(thread_queue->handle);
+
+ xm_lua_pushpointer(lua, (tb_pointer_t)thread_queue);
+ ok = tb_true;
+
+ } while (0);
+
+ if (!ok)
+ {
+ if (thread_queue)
+ {
+ if (thread_queue->handle)
+ {
+ tb_queue_exit(thread_queue->handle);
+ thread_queue->handle = tb_null;
+ }
+ tb_free(thread_queue);
+ }
+ lua_pushnil(lua);
+ }
+ return 1;
+}
diff --git a/core/src/xmake/thread/queue_pop.c b/core/src/xmake/thread/queue_pop.c
new file mode 100644
index 000000000..4e205bab4
--- /dev/null
+++ b/core/src/xmake/thread/queue_pop.c
@@ -0,0 +1,92 @@
+/*!A cross-platform build utility based on Lua
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ * Copyright (C) 2015-present, Xmake Open Source Community.
+ *
+ * @author ruki
+ * @file thread_queue_unlock.c
+ *
+ */
+
+/* //////////////////////////////////////////////////////////////////////////////////////
+ * trace
+ */
+#define TB_TRACE_MODULE_NAME "thread_queue"
+#define TB_TRACE_MODULE_DEBUG (0)
+
+/* //////////////////////////////////////////////////////////////////////////////////////
+ * includes
+ */
+#include "prefix.h"
+
+/* //////////////////////////////////////////////////////////////////////////////////////
+ * implementation
+ */
+tb_int_t xm_thread_queue_pop(lua_State* lua)
+{
+ tb_assert_and_check_return_val(lua, 0);
+
+ xm_thread_queue_t* thread_queue = xm_thread_queue_get(lua, 1);
+ tb_assert_and_check_return_val(thread_queue && thread_queue->handle, 0);
+
+ if (tb_queue_null(thread_queue->handle))
+ {
+ lua_pushnil(lua);
+ lua_pushliteral(lua, "the thread queue is empty");
+ return 2;
+ }
+
+ xm_thread_queue_item_t* item = (xm_thread_queue_item_t*)tb_queue_get(thread_queue->handle);
+ tb_assert_and_check_return_val(item, 0);
+
+ tb_bool_t ok = tb_false;
+ switch (item->kind)
+ {
+ case XM_THREAD_QUEUE_ITEM_STR:
+ if (item->size)
+ lua_pushlstring(lua, item->u.string, item->size);
+ else lua_pushliteral(lua, "");
+ ok = tb_true;
+ break;
+ case XM_THREAD_QUEUE_ITEM_INT:
+ lua_pushinteger(lua, item->u.integer);
+ ok = tb_true;
+ break;
+ case XM_THREAD_QUEUE_ITEM_NUM:
+ lua_pushnumber(lua, item->u.number);
+ ok = tb_true;
+ break;
+ case XM_THREAD_QUEUE_ITEM_BOOL:
+ lua_pushboolean(lua, item->u.boolean);
+ ok = tb_true;
+ break;
+ case XM_THREAD_QUEUE_ITEM_NIL:
+ lua_pushnil(lua);
+ ok = tb_true;
+ break;
+ default:
+ break;
+ }
+
+ if (!ok)
+ {
+ lua_pushnil(lua);
+ lua_pushliteral(lua, "invalid thread queue item");
+ return 2;
+ }
+
+ tb_queue_pop(thread_queue->handle);
+ return 1;
+}
+
diff --git a/core/src/xmake/thread/queue_push.c b/core/src/xmake/thread/queue_push.c
new file mode 100644
index 000000000..510a6b89b
--- /dev/null
+++ b/core/src/xmake/thread/queue_push.c
@@ -0,0 +1,96 @@
+/*!A cross-platform build utility based on Lua
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ * Copyright (C) 2015-present, Xmake Open Source Community.
+ *
+ * @author ruki
+ * @file thread_queue_lock.c
+ *
+ */
+
+/* //////////////////////////////////////////////////////////////////////////////////////
+ * trace
+ */
+#define TB_TRACE_MODULE_NAME "thread_queue"
+#define TB_TRACE_MODULE_DEBUG (0)
+
+/* //////////////////////////////////////////////////////////////////////////////////////
+ * includes
+ */
+#include "prefix.h"
+
+/* //////////////////////////////////////////////////////////////////////////////////////
+ * implementation
+ */
+tb_int_t xm_thread_queue_push(lua_State* lua)
+{
+ tb_assert_and_check_return_val(lua, 0);
+
+ xm_thread_queue_t* thread_queue = xm_thread_queue_get(lua, 1);
+ tb_assert_and_check_return_val(thread_queue && thread_queue->handle, 0);
+
+ if (tb_queue_full(thread_queue->handle))
+ {
+ lua_pushboolean(lua, tb_false);
+ lua_pushliteral(lua, "the thread queue is full");
+ return 2;
+ }
+
+ xm_thread_queue_item_t item;
+ if (lua_isstring(lua, 2))
+ {
+ size_t data_size = 0;
+ tb_char_t const* data = luaL_checklstring(lua, 2, &data_size);
+ tb_assert_and_check_return_val(data, 0);
+
+ item.kind = (tb_uint32_t)XM_THREAD_QUEUE_ITEM_STR;
+ item.size = (tb_uint32_t)data_size;
+ if (data_size)
+ {
+ item.u.string = tb_malloc_cstr(data_size);
+ tb_assert_and_check_return_val(item.u.string, 0);
+ tb_memcpy(item.u.string, data, data_size);
+ }
+ }
+ else if (lua_isinteger(lua, 2))
+ {
+ item.kind = (tb_uint32_t)XM_THREAD_QUEUE_ITEM_INT;
+ item.u.integer = lua_tointeger(lua, 2);
+ }
+ else if (lua_isnumber(lua, 2))
+ {
+ item.kind = (tb_uint32_t)XM_THREAD_QUEUE_ITEM_NUM;
+ item.u.number = lua_tonumber(lua, 2);
+ }
+ else if (lua_isboolean(lua, 2))
+ {
+ item.kind = (tb_uint32_t)XM_THREAD_QUEUE_ITEM_BOOL;
+ item.u.boolean = lua_toboolean(lua, 2);
+ }
+ else if (lua_isnil(lua, 2))
+ {
+ item.kind = (tb_uint32_t)XM_THREAD_QUEUE_ITEM_NIL;
+ }
+ else
+ {
+ lua_pushboolean(lua, tb_false);
+ lua_pushliteral(lua, "unsupported thread queue item");
+ return 2;
+ }
+
+ tb_queue_put(thread_queue->handle, &item);
+ lua_pushboolean(lua, tb_true);
+ return 1;
+}
+
diff --git a/core/src/xmake/thread/queue_size.c b/core/src/xmake/thread/queue_size.c
new file mode 100644
index 000000000..559a1dceb
--- /dev/null
+++ b/core/src/xmake/thread/queue_size.c
@@ -0,0 +1,46 @@
+/*!A cross-platform build utility based on Lua
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ * Copyright (C) 2015-present, Xmake Open Source Community.
+ *
+ * @author ruki
+ * @file thread_queue_lock.c
+ *
+ */
+
+/* //////////////////////////////////////////////////////////////////////////////////////
+ * trace
+ */
+#define TB_TRACE_MODULE_NAME "thread_queue"
+#define TB_TRACE_MODULE_DEBUG (0)
+
+/* //////////////////////////////////////////////////////////////////////////////////////
+ * includes
+ */
+#include "prefix.h"
+
+/* //////////////////////////////////////////////////////////////////////////////////////
+ * implementation
+ */
+tb_int_t xm_thread_queue_size(lua_State* lua)
+{
+ tb_assert_and_check_return_val(lua, 0);
+
+ xm_thread_queue_t* thread_queue = xm_thread_queue_get(lua, 1);
+ tb_assert_and_check_return_val(thread_queue && thread_queue->handle, 0);
+
+ lua_pushinteger(lua, (tb_int_t)tb_queue_size(thread_queue->handle));
+ return 1;
+}
+
diff --git a/tests/modules/thread/queue.lua b/tests/modules/thread/queue.lua
new file mode 100644
index 000000000..36034993f
--- /dev/null
+++ b/tests/modules/thread/queue.lua
@@ -0,0 +1,30 @@
+import("core.base.thread")
+
+function callback(event, queue)
+ import("core.base.thread")
+ print("%s: starting ..", thread.running())
+ while true do
+ print("%s: waiting ..", thread.running())
+ if event:wait(-1) > 0 then
+ print("%s: triggered", thread.running())
+ end
+ while not queue:empty() do
+ print("%s: get %s", thread.running(), queue:pop())
+ end
+ end
+end
+
+function main()
+ local event = thread.event()
+ local queue = thread.queue()
+ local t = thread.start_named("keyboard", callback, event, queue)
+ while true do
+ local ch = io.read()
+ if ch then
+ queue:push(ch)
+ event:post()
+ end
+ end
+ t:wait(-1)
+end
+
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