Merge pull request #44 from fechan/concurrency-fix

Fix concurrency issues again

computercraft/sigils.lua

@@ -4,6 +4,7 @@ local Pipe = require('sigils.pipe')
 local WebSocket = require('sigils.websocket')
 local Utils = require('sigils.utils')
 local Logging = require('sigils.logging')
+local Concurrent = require('sigils.concurrent')
 
 local DEFAULT_SERVER_URL = 'wss://sigils.fredchan.org'
 
@@ -81,6 +82,7 @@ local function init ()
     function () WebSocket.doWebSocket(wsContext) end,
     function () Controller.listenForCcpipesEvents(wsContext, factory) end,
     function () Pipe.processAllPipesForever(factory) end,
+    function () Concurrent.default_runner.run_forever() end,
     function () waitForQuitKey(wsContext) end,
     function () while true do os.sleep(0.05) end end -- forces the OS not to lock up
   )

computercraft/sigils/ItemDetailAndLimitCache.lua

@@ -30,6 +30,7 @@ function ItemDetailAndLimitCache.new (missingPeriphs, initialMap)
   ---@param groups Group[] List of groups to fulfill item limits and details for
   function o:Fulfill(groups)
     local runner = Concurrent.default_runner
+    local parallelTasks = {}
 
     for _, group in pairs(groups) do
       for _, slot in pairs(group.slots) do
@@ -41,7 +42,8 @@ function ItemDetailAndLimitCache.new (missingPeriphs, initialMap)
           end
 
           -- fulfill itemDetail
-          runner.spawn(
+          table.insert(
+            parallelTasks,
             function ()
               local periph = peripheral.wrap(slot.periphId)
               if periph and o.map[slotId].itemDetail == nil then
@@ -52,7 +54,8 @@ function ItemDetailAndLimitCache.new (missingPeriphs, initialMap)
           )
 
           -- fulfill itemLimit
-          runner.spawn(
+          table.insert(
+            parallelTasks,
             function ()
               local periph = peripheral.wrap(slot.periphId)
               if periph and o.map[slotId].itemLimit == nil then
@@ -65,7 +68,7 @@ function ItemDetailAndLimitCache.new (missingPeriphs, initialMap)
       end
     end
 
-    runner.run_until_done()
+    runner.await_batch_tasks(parallelTasks)
   end
 
   ---Get the item limit of the given Slot

computercraft/sigils/concurrent.lua

@@ -110,13 +110,30 @@ local function create_runner(max_size)
     end
   end
 
+  ---Run an array of functions in parallel and wait for all of them to finish
+  ---@param fns function[] List of functions to wait for the completion of
+  function await_batch_tasks(fns)
+    local awaits = {}
+    for _, task in pairs(fns) do
+      local resolve, await = create_future()
+      table.insert(awaits, await)
+      spawn(function ()
+        task()
+        resolve()
+      end)
+    end
+
+    parallel.waitForAll(table.unpack(awaits))
+  end
+
   --- A coroutine executor.
   -- @type Runner
   return {
     spawn = spawn,
     has_work = has_work,
     run_until_done = run_until_done,
     run_forever = run_forever,
+    await_all_tasks = await_all_tasks,
   }
 end