diff options
Diffstat (limited to 'xmake/core/base/thread.lua')
| -rw-r--r-- | xmake/core/base/thread.lua | 137 |
1 files changed, 92 insertions, 45 deletions
diff --git a/xmake/core/base/thread.lua b/xmake/core/base/thread.lua index 65e161050..91dd2a64d 100644 --- a/xmake/core/base/thread.lua +++ b/xmake/core/base/thread.lua @@ -103,26 +103,10 @@ function _thread:start() local argv = {} for _, arg in ipairs(self._ARGV) do if type(arg) == "table" then - -- is mutex? we can only pass cdata address - if arg._MUTEX and arg.cdata then - thread.mutex_incref(arg:cdata()) - arg = {mutex = true, name = arg:name(), caddr = libc.dataptr(arg:cdata(), {ffi = false})} - -- is event? we can only pass cdata address - elseif arg._EVENT and arg.cdata then - thread.event_incref(arg:cdata()) - arg = {event = true, name = arg:name(), caddr = libc.dataptr(arg:cdata(), {ffi = false})} - -- is semaphore? we can only pass cdata address - elseif arg._SEMAPHORE and arg.cdata then - thread.semaphore_incref(arg:cdata()) - arg = {semaphore = true, name = arg:name(), caddr = libc.dataptr(arg:cdata(), {ffi = false})} - -- is queue? we can only pass cdata address - elseif arg._QUEUE and arg.cdata then - thread.queue_incref(arg:cdata()) - arg = {queue = true, name = arg:name(), caddr = libc.dataptr(arg:cdata(), {ffi = false})} - -- is sharedata? we can only pass cdata address - elseif arg._SHAREDATA and arg.cdata then - thread.sharedata_incref(arg:cdata()) - arg = {sharedata = true, name = arg:name(), caddr = libc.dataptr(arg:cdata())} + -- try to serialize thread object (mutex, event, semaphore, queue, sharedata) + local serialized = thread._serialize_object(arg) + if serialized then + arg = serialized end end table.insert(argv, arg) @@ -782,23 +766,71 @@ function _sharedata:__gc() end end --- new a thread --- --- @param callback the thread callback --- @param opt the thread options, e.g. {name = "", argv = {}, stacksize = 8192} --- --- @return the thread instance --- -function thread.new(callback, opt) - if callback == nil then - return nil, "invalid thread, callback is nil" +-- serialize thread object for passing through queue or table (private helper) +-- this is used when you need to pass thread objects (mutex, event, semaphore, queue, sharedata) +-- through a queue or embed them in a table +-- returns a table with serialized caddr that can be pushed to queue +function thread._serialize_object(obj) + if not obj or type(obj) ~= "table" or not obj.cdata then + return nil end - return _thread.new(callback, opt) + + local result = {} + -- detect object type by checking internal marker + if obj._MUTEX then + thread.mutex_incref(obj:cdata()) + result.mutex = true + result.name = obj:name() + result.caddr = libc.dataptr(obj:cdata(), {ffi = false}) + elseif obj._EVENT then + thread.event_incref(obj:cdata()) + result.event = true + result.name = obj:name() + result.caddr = libc.dataptr(obj:cdata(), {ffi = false}) + elseif obj._SEMAPHORE then + thread.semaphore_incref(obj:cdata()) + result.semaphore = true + result.name = obj:name() + result.caddr = libc.dataptr(obj:cdata(), {ffi = false}) + elseif obj._QUEUE then + thread.queue_incref(obj:cdata()) + result.queue = true + result.name = obj:name() + result.caddr = libc.dataptr(obj:cdata(), {ffi = false}) + elseif obj._SHAREDATA then + thread.sharedata_incref(obj:cdata()) + result.sharedata = true + result.name = obj:name() + result.caddr = libc.dataptr(obj:cdata(), {ffi = false}) + else + return nil + end + + return result end --- get the running thread name -function thread.running() - return thread._RUNNING +-- deserialize thread object from serialized data (private helper) +-- this is used to restore thread objects (mutex, event, semaphore, queue, sharedata) +-- from serialized caddr received through queue or from table +function thread._deserialize_object(data) + if not data or type(data) ~= "table" or not data.caddr then + return nil + end + + local cdata = libc.ptraddr(data.caddr, {ffi = false}) + if data.mutex then + return _mutex.new(data.name, cdata) + elseif data.event then + return _event.new(data.name, cdata) + elseif data.semaphore then + return _semaphore.new(data.name, cdata) + elseif data.queue then + return _queue.new(data.name, cdata) + elseif data.sharedata then + return _sharedata.new(data.name, cdata) + end + + return nil end -- run thread @@ -861,20 +893,16 @@ function thread._run_thread(callback_str, callinfo_str) -- save the running thread name thread._RUNNING = threadname - -- translate arguments (mutex, ...) + -- translate arguments (mutex, event, semaphore, queue, sharedata, ...) if argv then local newargv = {} for _, arg in ipairs(argv) do - if type(arg) == "table" and arg.mutex and arg.caddr then - arg = _mutex.new(arg.name, libc.ptraddr(arg.caddr, {ffi = false})) - elseif type(arg) == "table" and arg.event and arg.caddr then - arg = _event.new(arg.name, libc.ptraddr(arg.caddr, {ffi = false})) - elseif type(arg) == "table" and arg.semaphore and arg.caddr then - arg = _semaphore.new(arg.name, libc.ptraddr(arg.caddr, {ffi = false})) - elseif type(arg) == "table" and arg.queue and arg.caddr then - arg = _queue.new(arg.name, libc.ptraddr(arg.caddr, {ffi = false})) - elseif type(arg) == "table" and arg.sharedata and arg.caddr then - arg = _sharedata.new(arg.name, libc.ptraddr(arg.caddr, {ffi = false})) + if type(arg) == "table" and arg.caddr then + -- try to deserialize thread object + local obj = thread._deserialize_object(arg) + if obj then + arg = obj + end end table.insert(newargv, arg) end @@ -895,6 +923,25 @@ function thread._run_thread(callback_str, callinfo_str) return ok, errors end +-- new a thread +-- +-- @param callback the thread callback +-- @param opt the thread options, e.g. {name = "", argv = {}, stacksize = 8192} +-- +-- @return the thread instance +-- +function thread.new(callback, opt) + if callback == nil then + return nil, "invalid thread, callback is nil" + end + return _thread.new(callback, opt) +end + +-- get the running thread name +function thread.running() + return thread._RUNNING +end + -- open a mutex function thread.mutex(name) local mutex = thread.mutex_init() |