diff options
| author | ruki <[email protected]> | 2025-08-28 23:14:13 +0800 |
|---|---|---|
| committer | ruki <[email protected]> | 2025-08-28 23:14:13 +0800 |
| commit | 628371d0edea22eb177fa2b1a6b9f28e8c0cff2d (patch) | |
| tree | 2c8e834e76870f3cfa2e2b2b3b3342d69b0f83bf | |
| parent | 8a331ab64fe560924b8bc3ab3dc4ac2831dd8920 (diff) | |
add thread queue
| -rw-r--r-- | core/src/xmake/engine.c | 16 | ||||
| -rw-r--r-- | core/src/xmake/thread/prefix.h | 43 | ||||
| -rw-r--r-- | core/src/xmake/thread/queue_clear.c | 47 | ||||
| -rw-r--r-- | core/src/xmake/thread/queue_exit.c | 55 | ||||
| -rw-r--r-- | core/src/xmake/thread/queue_incref.c | 46 | ||||
| -rw-r--r-- | core/src/xmake/thread/queue_init.c | 87 | ||||
| -rw-r--r-- | core/src/xmake/thread/queue_pop.c | 92 | ||||
| -rw-r--r-- | core/src/xmake/thread/queue_push.c | 96 | ||||
| -rw-r--r-- | core/src/xmake/thread/queue_size.c | 46 | ||||
| -rw-r--r-- | tests/modules/thread/queue.lua | 30 | ||||
| -rw-r--r-- | xmake/core/base/thread.lua | 148 | ||||
| -rw-r--r-- | xmake/core/sandbox/modules/import/core/base/thread.lua | 75 |
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 |
