fix thread wait

Author: waruqi

xmake/core/base/private/async_task.lua

@@ -81,8 +81,6 @@ function async_task._loop(event, queue, mutex, is_stopped, is_diagnosis)
     local thread = require("base/thread")
 
     local function dprint(...)
-        -- TODO
-        is_diagnosis = true
         if is_diagnosis then
             print(...)
         end
@@ -194,6 +192,7 @@ function async_task._start()
 
             -- Perhaps the thread hasn't started yet.
             -- Let's wait a while and let it finish executing the tasks in the current queue.
+            utils.dprint("async_task: wait the pending tasks(%d) for exiting ..", task_queue:size())
             task_mutex:lock()
             local is_empty = task_queue:empty()
             task_mutex:unlock()
@@ -204,7 +203,7 @@ function async_task._start()
             task_is_stopped:set(true)
             task_event:post()
 
-            utils.dprint("async_task: wait the pending tasks(%d) for exiting ..", task_queue:size())
+            utils.dprint("async_task: wait thread for exiting ..")
             task_thread:wait(-1)
             utils.dprint("async_task: wait finished")
         end

xmake/core/base/thread.lua

@@ -113,15 +113,18 @@ function _thread:start()
     end
 
     -- init callback info
+    local is_internal = self._INTERNAL
     local callback = string._dump(self._CALLBACK)
-    local callinfo = {name = self:name(), argv = argv, internal = self._INTERNAL}
+    local callinfo = {name = self:name(), argv = argv, internal = is_internal}
 
     -- we need a pipe pair to wait and listen thread exit event
-    local rpipe, wpipe = pipe.openpair("AA")
-    self._RPIPE = rpipe
-    callinfo.wpipe = libc.dataptr(wpipe:cdata(), {ffi = false})
-    -- we need to suppress gc to free it, because it has been transfer to thread in another lua state instance
-    wpipe._PIPE = nil
+    if not is_internal then
+        local rpipe, wpipe = pipe.openpair("AA")
+        self._RPIPE = rpipe
+        callinfo.wpipe = libc.dataptr(wpipe:cdata(), {ffi = false})
+        -- we need to suppress gc to free it, because it has been transfer to thread in another lua state instance
+        wpipe._PIPE = nil
+    end
 
     -- serialize and pass callback and arguments to this thread
     -- we do not use string.serialize to serialize callback, because it's slower (deserialize)
@@ -182,7 +185,7 @@ function _thread:wait(timeout)
 
     local ok, errors
     local rpipe = self._RPIPE
-    if rpipe then
+    if rpipe and scheduler:co_running() then
         local buff = bytes(16)
         local read, data_or_errors = rpipe:read(buff, 1, {block = true, timeout = timeout})
         if read > 0 then
@@ -192,7 +195,7 @@ function _thread:wait(timeout)
             errors = data_or_errors
         end
     end
-    if not scheduler:co_running() then
+    if not rpipe then
         local waitok, wait_errors = thread.thread_wait(self:cdata(), timeout)
         if ok == nil or ok > 0 then
             ok = waitok
@@ -857,7 +860,9 @@ function thread._run_thread(callback_str, callinfo_str)
             argv = callinfo.argv
             threadname = callinfo.name
             is_internal = callinfo.internal
-            wpipe = pipe.new(libc.ptraddr(callinfo.wpipe, {ffi = false}))
+            if callinfo.wpipe then
+                wpipe = pipe.new(libc.ptraddr(callinfo.wpipe, {ffi = false}))
+            end
         end
     end