improve thread

2 files changed, 205 insertions, 62 deletions

diff --git a/xmake/core/base/private/async_task.lua b/xmake/core/base/private/async_task.lua
index 676f2b226..9cd51926e 100644
--- a/xmake/core/base/private/async_task.lua
+++ b/xmake/core/base/private/async_task.lua
@@ -39,6 +39,8 @@ local task_mutex = nil
 -- the asynchronous task loop
 function async_task._loop(event, queue, mutex, is_stopped, is_diagnosis)
     local os = require("base/os")
+    local try = require("sandbox/modules/try")
+    local thread = require("base/thread")
 
     local function dprint(...)
         if is_diagnosis then
@@ -46,6 +48,17 @@ function async_task._loop(event, queue, mutex, is_stopped, is_diagnosis)
         end
     end
 
+    -- restore thread objects from serialized format
+    local function _restore_thread_objects(cmd)
+        -- use thread helper to deserialize thread objects from queue data
+        if cmd.event_data then
+            cmd.event = thread._deserialize_object(cmd.event_data)
+        end
+        if cmd.result_data then
+            cmd.result = thread._deserialize_object(cmd.result_data)
+        end
+    end
+
     local function _runcmd_cp(cmd)
         os.cp(cmd.srcpath, cmd.dstpath)
     end
@@ -61,16 +74,39 @@ function async_task._loop(event, queue, mutex, is_stopped, is_diagnosis)
         rmdir = _runcmd_rmdir
     }
     local function _runcmd(cmd)
+        local ok = true
+        local errors
+
+        -- restore thread objects if needed
+        _restore_thread_objects(cmd)
+
         local runop = runops[cmd.kind]
         if runop then
-            runop(cmd)
+            try
+            {
+                function ()
+                    runop(cmd)
+                end,
+                catch
+                {
+                    function (errs)
+                        ok = false
+                        errors = tostring(errs)
+                    end
+                }
+            }
+        end
+        -- notify completion if event is provided
+        if cmd.event and cmd.result then
+            cmd.result:set({ok = ok, errors = errors})
+            cmd.event:post()
         end
     end
 
     dprint("async_task: started")
     while not is_stopped:get() do
         if event:wait(-1) > 0 then
-            
+
             -- fetch all tasks from queue at once
             local cmds = {}
             mutex:lock()
@@ -150,25 +186,45 @@ function async_task.cp(srcpath, dstpath, opt)
         return false, errors
     end
 
-    -- TODO
-    assert(opt.detach)
-
     -- post task
     srcpath = path.absolute(tostring(srcpath))
     dstpath = path.absolute(tostring(dstpath))
+
+    local cmd = {kind = "cp", srcpath = srcpath, dstpath = dstpath}
+    local cmd_event, cmd_result
+
+    -- create event and result for non-detach mode
+    if not opt.detach then
+        cmd_event = thread.event()
+        cmd_result = thread.sharedata()
+
+        -- serialize thread objects for passing to worker thread
+        cmd.event_data = thread._serialize_object(cmd_event)
+        cmd.result_data = thread._serialize_object(cmd_result)
+    end
+
     task_mutex:lock()
-    task_queue:push({kind = "cp", srcpath = srcpath, dstpath = dstpath})
+    task_queue:push(cmd)
     local queue_size = task_queue:size()
     task_mutex:unlock()
+
     if opt.detach then
         -- We cache some tasks before executing them to avoid frequent thread switching.
         if queue_size > 10 then
             task_event:post()
         end
+        return true
     else
+        -- wait for completion
         task_event:post()
+        cmd_event:wait(-1)
+        local result = cmd_result:get()
+        if result and result.ok then
+            return true
+        else
+            return false, result and result.errors or "unknown error"
+        end
     end
-    return true
 end
 
 -- remove files or directories
@@ -179,24 +235,44 @@ function async_task.rm(filepath, opt)
         return false, errors
     end
 
-    -- TODO
-    assert(opt.detach)
-
     -- post task
     filepath = path.absolute(tostring(filepath))
+
+    local cmd = {kind = "rm", filepath = filepath}
+    local cmd_event, cmd_result
+
+    -- create event and result for non-detach mode
+    if not opt.detach then
+        cmd_event = thread.event()
+        cmd_result = thread.sharedata()
+
+        -- serialize thread objects for passing to worker thread
+        cmd.event_data = thread._serialize_object(cmd_event)
+        cmd.result_data = thread._serialize_object(cmd_result)
+    end
+
     task_mutex:lock()
-    task_queue:push({kind = "rm", filepath = filepath})
+    task_queue:push(cmd)
     local queue_size = task_queue:size()
     task_mutex:unlock()
+
     if opt.detach then
         -- We cache some tasks before executing them to avoid frequent thread switching.
         if queue_size > 10 then
             task_event:post()
         end
+        return true
     else
+        -- wait for completion
         task_event:post()
+        cmd_event:wait(-1)
+        local result = cmd_result:get()
+        if result and result.ok then
+            return true
+        else
+            return false, result and result.errors or "unknown error"
+        end
     end
-    return true
 end
 
 -- remove directories
@@ -207,24 +283,44 @@ function async_task.rmdir(dir, opt)
         return false, errors
     end
 
-    -- TODO
-    assert(opt.detach)
-
     -- post task
     dir = path.absolute(tostring(dir))
+
+    local cmd = {kind = "rmdir", dir = dir}
+    local cmd_event, cmd_result
+
+    -- create event and result for non-detach mode
+    if not opt.detach then
+        cmd_event = thread.event()
+        cmd_result = thread.sharedata()
+
+        -- serialize thread objects for passing to worker thread
+        cmd.event_data = thread._serialize_object(cmd_event)
+        cmd.result_data = thread._serialize_object(cmd_result)
+    end
+
     task_mutex:lock()
-    task_queue:push({kind = "rmdir", dir = dir})
+    task_queue:push(cmd)
     local queue_size = task_queue:size()
     task_mutex:unlock()
+
     if opt.detach then
         -- We cache some tasks before executing them to avoid frequent thread switching.
         if queue_size > 10 then
             task_event:post()
         end
+        return true
     else
+        -- wait for completion
         task_event:post()
+        cmd_event:wait(-1)
+        local result = cmd_result:get()
+        if result and result.ok then
+            return true
+        else
+            return false, result and result.errors or "unknown error"
+        end
     end
-    return true
 end
 
 -- return module: async_task
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()