improve jobpool

3 files changed, 145 insertions, 45 deletions

diff --git a/tests/modules/list/test.lua b/tests/modules/list/test.lua
index b1679a304..4a4e81fd0 100644
--- a/tests/modules/list/test.lua
+++ b/tests/modules/list/test.lua
@@ -86,6 +86,48 @@ function test_remove_last(t)
     end
 end
 
+function test_for_remove(t)
+    local d = list.new()
+    d:push({v = 1})
+    d:push({v = 2})
+    d:push({v = 3})
+    d:push({v = 4})
+    d:push({v = 5})
+    t:are_equal(d:first().v, 1)
+    t:are_equal(d:last().v, 5)
+    local idx = 1
+    local item = d:first()
+    while item ~= nil do
+        local next = d:next(item)
+        t:are_equal(item.v, idx)
+        d:remove(item)
+        item = next
+        idx = idx + 1
+    end
+    t:require(d:empty())
+end
+
+function test_rfor_remove(t)
+    local d = list.new()
+    d:push({v = 1})
+    d:push({v = 2})
+    d:push({v = 3})
+    d:push({v = 4})
+    d:push({v = 5})
+    t:are_equal(d:first().v, 1)
+    t:are_equal(d:last().v, 5)
+    local idx = 5
+    local item = d:last()
+    while item ~= nil do
+        local prev = d:prev(item)
+        t:are_equal(item.v, idx)
+        d:remove(item)
+        item = prev
+        idx = idx - 1
+    end
+    t:require(d:empty())
+end
+
 function test_insert_first(t)
     local d = list.new()
     d:push({v = 2})
diff --git a/xmake/modules/async/runjobs.lua b/xmake/modules/async/runjobs.lua
index 6d19efe17..1508a71d8 100644
--- a/xmake/modules/async/runjobs.lua
+++ b/xmake/modules/async/runjobs.lua
@@ -154,7 +154,6 @@ function main(name, jobs, opt)
     -- run jobs
     local index = 0
     local count = 0
-    local job_pending = nil
     local abort = false
     local abort_errors
     local progress_wrapper = {}
@@ -188,21 +187,18 @@ function main(name, jobs, opt)
             while index < total_max do
 
                 -- uses job pool?
+                local job
                 local jobname
                 local distccjob = false
                 if not jobs_cb then
 
-                    -- get job
-                    local job
-                    if job_pending then
-                        job = job_pending
-                    else
-                        job = jobs:pop()
-                    end
+                    -- get free job
+                    job = jobs:getfree()
                     if not job then
                         break
                     end
 
+                    -- TODO
                     -- we can only continue to run the job with distcc if local jobs are full
                     if distcc and index >= local_max then
                         if job.distcc then
@@ -216,7 +212,6 @@ function main(name, jobs, opt)
                     -- get run function
                     jobfunc = job.run
                     jobname = job.name
-                    job_pending = nil
                 else
                     jobname = tostring(index)
                 end
@@ -243,6 +238,9 @@ function main(name, jobs, opt)
                                 end
                                 count = count + 1
                                 jobfunc(i, total, {progress = progress_wrapper})
+                                if job then
+                                    jobs:remove(job)
+                                end
                                 print("finished", jobname)
                             end
                             running_jobs_indices[i] = nil
@@ -283,13 +281,8 @@ function main(name, jobs, opt)
             end
         end)
 
-        -- only need one job exited if be same priority
-        if priority_curr == priority_prev then
-            scheduler.co_group_wait(group_name, {limit = 1})
-        else
-            -- need to wait all running jobs exited first if be different priority
-            scheduler.co_group_wait(group_name)
-        end
+        -- wait for free jobs
+        scheduler.co_group_wait(group_name, {limit = 1})
     end
 
     -- wait all jobs exited
diff --git a/xmake/modules/private/async/jobpool.lua b/xmake/modules/private/async/jobpool.lua
index 10fe90bbe..51c76847f 100644
--- a/xmake/modules/private/async/jobpool.lua
+++ b/xmake/modules/private/async/jobpool.lua
@@ -24,7 +24,12 @@ import("core.base.list")
 import("core.base.hashset")
 
 -- define module
-local jobpool = jobpool or object {_init = {"_size", "_rootjob", "_leafjobs", "_poprefs"}}
+local jobpool = jobpool or object {_init = {"_size", "_rootjob", "_leafjobs"}}
+
+-- the job status
+local JOB_STATUS_FREE     = 1
+local JOB_STATUS_PENDING  = 2
+local JOB_STATUS_FINISHED = 3
 
 -- get jobs size
 function jobpool:size()
@@ -46,7 +51,7 @@ end
 --
 function jobpool:newjob(name, run, opt)
     opt = opt or {}
-    return {name = name, run = run, distcc = opt.distcc}
+    return {name = name, run = run, distcc = opt.distcc, status = JOB_STATUS_FREE}
 end
 
 -- add run job to the given job node
@@ -61,7 +66,7 @@ end
 --
 function jobpool:addjob(name, run, opt)
     opt = opt or {}
-    return self:add({name = name, run = run, distcc = opt.distcc}, opt.rootjob)
+    return self:add({name = name, run = run, distcc = opt.distcc, status = JOB_STATUS_FREE}, opt.rootjob)
 end
 
 -- add job to the given job node
@@ -96,27 +101,80 @@ function jobpool:add(job, rootjob)
     return job
 end
 
--- pop job without deps at leaf node
-function jobpool:pop()
+-- has free jobs?
+function jobpool:hasfree()
+    if self:size() == 0 then
+        return
+    end
+
+    -- peak a free job from the leaf jobs
+    local leafjobs = self:_getleafjobs()
+    if not leafjobs:empty() then
+        if self._nextfree then
+            return true
+        end
+        local job = leafjobs:last()
+        while job ~= nil do
+            local prevjob = leafjobs:prev(job)
+            if self:_isfree(job) then
+                self._nextfree = job
+                return true
+            elseif job.group or job.status == JOB_STATUS_FINISHED then
+                self:remove(job)
+            end
+            job = prevjob
+        end
+    end
+end
 
-    -- no jobs?
+-- get a free job from the leaf jobs
+function jobpool:getfree()
     if self:size() == 0 then
         return
     end
 
-    -- init leaf jobs first
-    local leafjobs = self._leafjobs
-    if leafjobs:empty() then
-        local refs = {}
-        self:_genleafjobs(self:rootjob(), leafjobs, refs)
+    -- get a free job from the leaf jobs
+    local leafjobs = self:_getleafjobs()
+    if not leafjobs:empty() then
+        if self._nextfree then
+            local job = self._nextfree
+            local nextfree = leafjobs:prev(job)
+            if nextfree ~= job and self:_isfree(nextfree) then
+                self._nextfree = nextfree
+            else
+                self._nextfree = nil
+            end
+            job.status = JOB_STATUS_PENDING
+            return job
+        end
+        local job = leafjobs:last()
+        while job ~= nil do
+            local prevjob = leafjobs:prev(job)
+            if self:_isfree(job) then
+                local nextfree = prevjob
+                if nextfree ~= job and self:_isfree(nextfree) then
+                    self._nextfree = nextfree
+                end
+                job.status = JOB_STATUS_PENDING
+                return job
+            elseif job.group or job.status == JOB_STATUS_FINISHED then
+                self:remove(job)
+            end
+            job = prevjob
+        end
     end
+end
 
-    -- pop a job from the leaf jobs
+-- remove the given job from the leaf jobs
+function jobpool:remove(job)
+    assert(self:size() > 0)
+    local leafjobs = self:_getleafjobs()
     if not leafjobs:empty() then
+        assert(job ~= self._nextfree)
 
-        -- get job
-        local job = leafjobs:last()
-        leafjobs:remove_last()
+        -- remove this job from leaf jobs
+        job.status = JOB_STATUS_FINISHED
+        leafjobs:remove(job)
 
         -- get parents node
         local parents = assert(job._parents, "invalid job without parents node!")
@@ -133,19 +191,6 @@ function jobpool:pop()
                 end
             end
         end
-
-        -- is group node or referenced node (it has been popped once) ?
-        local poprefs = self._poprefs
-        local jobkey = tostring(job)
-        if job.group or poprefs[jobkey] then
-            -- pop the next real job
-            return self:pop()
-        else
-            -- pop this job
-            self._size = self._size - 1
-            poprefs[jobkey] = true
-            return job
-        end
     end
 end
 
@@ -177,6 +222,26 @@ function jobpool:group_leave()
     end
 end
 
+-- is free job?
+-- we need to ignore group node (empty job) and referenced node (finished job)
+function jobpool:_isfree(job)
+    if job and job.status == JOB_STATUS_FREE and not job.group then
+        return true
+    end
+end
+
+-- get leaf jobs
+function jobpool:_getleafjobs()
+    local leafjobs = self._leafjobs
+    if leafjobs == nil then
+        leafjobs = list.new()
+        local refs = {}
+        self:_genleafjobs(self:rootjob(), leafjobs, refs)
+        self._leafjobs = leafjobs
+    end
+    return leafjobs
+end
+
 -- generate all leaf jobs from the given job
 function jobpool:_genleafjobs(job, leafjobs, refs)
     local deps = job._deps
@@ -231,5 +296,5 @@ end
 
 -- new a jobpool
 function new()
-    return jobpool {0, {name = "root"}, list.new(), {}}
+    return jobpool {0, {name = "root"}, nil}
 end