async await using libuv

lua/plenary/async_lib/async.lua

@@ -0,0 +1,150 @@
+local co = coroutine
+local uv = vim.loop
+
+local thread_loop = function(thread, callback)
+  local idle = uv.new_idle()
+  idle:start(function()
+    local success = co.resume(thread)
+    assert(success, "Coroutine failed")
+
+    if co.status(thread) == "dead" then
+      idle:stop()
+      callback()
+    end
+  end)
+end
+
+-- use with wrap
+local pong = function(func, callback)
+  assert(type(func) == "function", "type error :: expected func")
+  local thread = co.create(func)
+  local step
+  step = function(...)
+    local res = {co.resume(thread, ...)}
+    local stat = res[1]
+    local ret = {select(2, unpack(res))}
+    assert(stat, "Status should be true")
+    if co.status(thread) == "dead" then
+      (callback or function() end)(unpack(ret))
+    else
+      assert(#ret == 1, "expected a single return value")
+      assert(type(ret[1]) == "function", "type error :: expected func")
+      ret[1](step)
+    end
+  end
+  step()
+end
+
+-- use with pong, creates thunk factory
+local wrap = function(func)
+  assert(type(func) == "function", "type error :: expected func, got " .. type(func))
+
+  return function(...)
+    local params = {...}
+    return function(step)
+      table.insert(params, step)
+      return func(unpack(params))
+    end
+  end
+end
+
+local thread_loop_async = wrap(thread_loop)
+
+-- many thunks -> single thunk
+local join = function(thunks)
+  local len = #thunks
+  local done = 0
+  local acc = {}
+
+  local thunk = function(step)
+    if len == 0 then
+      return step()
+    end
+    for i, tk in ipairs(thunks) do
+      assert(type(tk) == "function", "thunk must be function")
+      local callback = function(...)
+        acc[i] = {...}
+        done = done + 1
+        if done == len then
+          step(unpack(acc))
+        end
+      end
+      tk(callback)
+    end
+  end
+  return thunk
+end
+
+local function run(future)
+  future()
+end
+
+local function run_all(futures)
+  for _, future in ipairs(futures) do
+    future()
+  end
+end
+
+-- sugar over coroutine
+local await = function(defer)
+  assert(type(defer) == "function", "type error :: expected func")
+  return co.yield(defer)
+end
+
+
+local await_all = function(defer)
+  assert(type(defer) == "table", "type error :: expected table")
+  return co.yield(join(defer))
+end
+
+local async = function(func)
+  return function(...)
+    local args = {...}
+    return wrap(pong)(function()
+      return func(unpack(args))
+    end)
+  end
+end
+
+local pong_loop = async(function(func, callback)
+  assert(type(func) == "function", "type error :: expected func")
+  local thread = co.create(func)
+
+  local _step
+  _step = function(...)
+    local res = {co.resume(thread, ...)}
+    local stat = res[1]
+    local ret = {select(2, unpack(res))}
+    assert(stat, "Status should be true")
+    if co.status(thread) == "dead" then
+      (callback or function() end)(unpack(ret))
+    else
+      assert(#ret == 1, "expected a single return value")
+      assert(type(ret[1]) == "function", "type error :: expected func")
+      -- yield before calling the next one
+      co.yield()
+      ret[1](_step)
+    end
+  end
+
+  local step = function()
+    thread_loop(co.create(_step))
+  end
+
+  step()
+end)
+
+--- because idle is a bad name
+local spawn = wrap(pong_loop)
+
+return {
+  async = async,
+  join = join,
+  await = await,
+  await_all = await_all,
+  run = run,
+  run_all = run_all,
+  spawn = spawn,
+  wrap = wrap,
+  wait_for_textlock = wrap(vim.schedule)
+} 

lua/plenary/async_lib/helpers.lua

@@ -0,0 +1,50 @@
+local M = {}
+
+VecDeque = {}
+VecDeque.__index = VecDeque
+
+function VecDeque.new()
+  return setmetatable({first = 0, last = -1}, VecDeque)
+end
+
+function VecDeque:pushleft(value)
+  local first = self.first - 1
+  self.first = first
+  self[first] = value
+end
+
+function VecDeque:pushright(value)
+  local last = self.last + 1
+  self.last = last
+  self[last] = value
+end
+
+function VecDeque:popleft()
+  local first = self.first
+  if first > self.last then return nil end
+  local value = self[first]
+  self[first] = nil        -- to allow garbage collection
+  self.first = first + 1
+  return value
+end
+
+function VecDeque:is_empty()
+  return self.first > self.last
+end
+
+function VecDeque:popright()
+  local last = self.last
+  if self.first > last then return nil end
+  local value = self[last]
+  self[last] = nil         -- to allow garbage collection
+  self.last = last - 1
+  return value
+end
+
+function VecDeque:len()
+  return self.last - self.first
+end
+
+M.VecDeque = VecDeque
+
+return M

lua/plenary/async_lib/init.lua

@@ -0,0 +1,7 @@
+local exports = require('plenary.async_lib.async')
+exports.uv = require('plenary.async_lib.uv_async')
+exports.utils = require('plenary.async_lib.utils')
+exports.lsp = require('plenary.async_lib.lsp')
+exports.work = require('plenary.async_lib.work')
+
+return exports

lua/plenary/async_lib/lsp.lua

@@ -0,0 +1,7 @@
+local a = require('plenary.async_lib.async')
+
+local M = {}
+
+M.buf_request = a.wrap(vim.lsp.buf_request)
+
+return M

lua/plenary/async_lib/utils.lua

@@ -0,0 +1,116 @@
+local a = require('plenary.async_lib.async')
+local async = a.async
+local co = coroutine
+local VecDeque = require('plenary.async_lib.helpers').VecDeque
+local uv = vim.loop
+
+local M = {}
+
+M.sleep = a.wrap(function(ms, callback)
+  local timer = uv.new_timer()
+  uv.timer_start(timer, ms, 0, function()
+    uv.timer_stop(timer)
+    uv.close(timer)
+    callback()
+  end)
+end)
+
+M.timer = function(ms)
+  return async(function()
+    a.wait(M.sleep(ms))
+  end)
+end
+
+M.id = async(function(...)
+  return ...
+end)
+
+M.thread_loop = function(thread, callback)
+  local idle = uv.new_idle()
+  idle:start(function()
+    local success = co.resume(thread)
+    assert(success, "Coroutine failed")
+
+    if co.status(thread) == "dead" then
+      idle:stop()
+      callback()
+    end
+  end)
+end
+
+M.thread_loop_async = a.wrap(M.thread_loop)
+
+M.yield_now = async(function()
+  a.wait(M.id())
+end)
+
+local Condvar = {}
+Condvar.__index = Condvar
+
+function Condvar.new()
+  return setmetatable({handles = {}}, Condvar)
+end
+
+--- async function
+--- blocks the thread until a notification is received
+Condvar.wait = a.wrap(function(self, callback)
+  -- not calling the callback will block the coroutine
+  table.insert(self.handles, callback)
+end)
+
+--- not an async function
+function Condvar:notify_all()
+  for _, callback in ipairs(self.handles) do
+    callback()
+  end
+  self.handles = {} -- reset all handles as they have been used up
+end
+
+--- not an async function
+function Condvar:notify_one()
+  if #self.handles == 0 then return end
+
+  local idx = math.random(#self.handles)
+  self.handles[idx]()
+  table.remove(self.handles, idx)
+end
+
+M.Condvar = Condvar
+
+M.channel = {}
+
+---comment
+---@return function
+---@return any
+M.channel.oneshot = function()
+  local val = nil
+  local saved_callback = nil
+
+  --- sender is not async
+  --- sends a value
+  local sender = function(t)
+    if val ~= nil then
+      error('Oneshot channel can only send one value!')
+    end
+
+    val = t
+    saved_callback(val)
+  end
+
+  --- receiver is async
+  --- blocks until a value is received
+  local receiver = a.wrap(function(callback)
+    if callback ~= nil then
+      error('Oneshot channel can only receive one value!')
+    end
+
+    saved_callback = callback
+  end)
+
+  return sender, receiver
+end
+
+M.channel.mpsc = function()
+end
+
+return M