fix: condvar not properly removing handles (#384)

fix #315

Author: GustavoKatel

README.md

@@ -97,6 +97,52 @@ end
 
 Please use `plenary.async` instead. This was version 1 and is just here for compatibility reasons.
 
+### plenary.async.control.channel.oneshot
+
+Creates a oneshot channel. It can only send data one time.
+
+The sender is not async while the receiver is.
+
+Example:
+
+```lua
+local a = require'plenary.async'
+local tx, rx = a.control.channel.oneshot()
+
+a.run(function()
+    local ret = long_running_fn()
+    tx(ret)
+end)
+
+local ret = rx()
+```
+
+### plenary.async.control.channel.mpsc
+
+Creates a multiple producer single consumer channel.
+
+Example:
+
+```lua
+local a = require'plenary.async'
+local sender, receiver = a.control.channel.mpsc()
+
+a.run(function()
+  sender.send(10)
+  sender.send(20)
+end)
+
+a.run(function()
+  sender.send(30)
+  sender.send(40)
+end)
+
+for _ = 1, 4 do
+  local value = receiver.recv()
+  print('received:', value)
+end
+```
+
 ### plenary.job
 
 A Lua module to interact with system processes. Pass in your `command`, the desired `args`, `env` and `cwd`.

lua/plenary/async/control.lua

@@ -21,9 +21,20 @@ end, 2)
 
 ---notify everyone that is waiting on this Condvar
 function Condvar:notify_all()
+  local len = #self.handles
   for i, callback in ipairs(self.handles) do
+    if i > len then
+      -- this means that more handles were added while we were notifying
+      -- if we don't break we can get starvation notifying as soon as new handles are added
+      break
+    end
+
     callback()
-    self.handles[i] = nil
+  end
+
+  for i = 1, len do
+    -- table.remove will ensure that indexes are correct and make "ipairs" safe, which is not the case for "self.handles[i] = nil"
+    table.remove(self.handles, i)
   end
 end
 
@@ -207,7 +218,7 @@ M.channel.mpsc = function()
     if deque:is_empty() then
       condvar:wait()
     end
-    local val = deque:popright()
+    local val = deque:popleft()
     deque:clear()
     return unpack(val or {})
   end

tests/plenary/async/channel_spec.lua

@@ -84,6 +84,101 @@ describe("channel", function()
     end)
   end)
 
+  describe("mpsc", function()
+    a.it("should wait multiple recv before any send", function()
+      local sender, receiver = channel.mpsc()
+
+      local expected_count = 10
+
+      a.run(function()
+        for i = 1, expected_count do
+          a.util.sleep(250)
+          sender.send(i)
+        end
+      end)
+
+      local receive_count = 0
+      while receive_count < expected_count do
+        receive_count = receive_count + 1
+        local i = receiver.recv()
+        eq(receive_count, i)
+      end
+    end)
+
+    a.it("should queues multiple sends before any read", function()
+      local sender, receiver = channel.mpsc()
+
+      local counter = 0
+
+      a.run(function()
+        counter = counter + 1
+        sender.send(10)
+
+        counter = counter + 1
+        sender.send(20)
+      end)
+
+      a.util.sleep(1000)
+
+      eq(10, receiver.recv())
+      eq(20, receiver.recv())
+      eq(2, counter)
+    end)
+
+    a.it("should queues multiple sends from multiple producers before any read", function()
+      local sender, receiver = channel.mpsc()
+
+      local counter = 0
+
+      a.run(function()
+        counter = counter + 1
+        sender.send(10)
+
+        counter = counter + 1
+        sender.send(20)
+      end)
+
+      a.run(function()
+        counter = counter + 1
+        sender.send(30)
+
+        counter = counter + 1
+        sender.send(40)
+      end)
+
+      a.util.sleep(1000)
+
+      local read_counter = 0
+      a.util.block_on(function()
+        for _ = 1, 4 do
+          receiver.recv()
+          read_counter = read_counter + 1
+        end
+      end, 1000)
+      eq(4, counter)
+      eq(4, read_counter)
+    end)
+
+    a.it("should read only the last value", function()
+      local sender, receiver = channel.mpsc()
+
+      local counter = 0
+
+      a.run(function()
+        counter = counter + 1
+        sender.send(10)
+
+        counter = counter + 1
+        sender.send(20)
+      end)
+
+      a.util.sleep(1000)
+
+      eq(20, receiver.last())
+      eq(2, counter)
+    end)
+  end)
+
   describe("counter", function()
     a.it("should work", function()
       local tx, rx = channel.counter()