mirror of
https://github.com/neovim/neovim.git
synced 2026-11-06 13:26:57 +00:00
1129 lines
34 KiB
Lua
1129 lines
34 KiB
Lua
local util = require('vim._core.util')
|
|
local future = require('vim.async._future')
|
|
local runtime = require('vim.async._runtime')
|
|
|
|
local is_callable = vim.is_callable
|
|
local validate = vim.validate
|
|
local pcall = pcall
|
|
local coroutine_running = coroutine.running
|
|
do
|
|
local ok, coxpcall = pcall(require, 'coxpcall')
|
|
if ok and type(coxpcall) == 'table' then
|
|
if type(coxpcall.pcall) == 'function' then
|
|
pcall = coxpcall.pcall
|
|
end
|
|
if type(coxpcall.running) == 'function' then
|
|
coroutine_running = coxpcall.running
|
|
end
|
|
end
|
|
end
|
|
local maxint = 2 ^ 32 - 1
|
|
local pack_len = vim.F.pack_len
|
|
local unpack_len = vim.F.unpack_len
|
|
|
|
--- @class vim.async._core
|
|
--- @nodoc
|
|
local M = {}
|
|
|
|
--- Weak table to keep track of running tasks
|
|
--- @type table<thread, vim.async.Task<any>?>
|
|
local threads = setmetatable({}, { __mode = 'k' })
|
|
|
|
--- Returns the currently running task.
|
|
--- @return vim.async.Task<any>?
|
|
local function running()
|
|
--- @diagnostic disable-next-line: invisible, undefined-field
|
|
local task = threads[coroutine_running()]
|
|
if task and not task:completed() then
|
|
return task
|
|
end
|
|
end
|
|
|
|
--- Internal marker used to identify that a yielded value is an asynchronous yielding.
|
|
local yield_marker = {}
|
|
local resume_marker = {}
|
|
|
|
local resume_error = 'Unexpected coroutine.resume()'
|
|
local yield_error = 'Unexpected coroutine.yield()'
|
|
|
|
--- @return vim.async.Task<any>
|
|
local function current_task()
|
|
return (assert(running(), 'Not in async context'))
|
|
end
|
|
|
|
--- Checks the arguments of a `coroutine.resume`.
|
|
--- This is used to ensure that a resume is expected.
|
|
--- @generic T
|
|
--- @param marker any
|
|
--- @param err? any
|
|
--- @param ... T...
|
|
--- @return T...
|
|
local function check_yield(marker, err, ...)
|
|
if marker ~= resume_marker then
|
|
current_task():_raise(resume_error)
|
|
-- Return an error to the caller. This will also leave the task in a dead
|
|
-- and unfinished state.
|
|
error(resume_error, 0)
|
|
elseif err ~= nil then
|
|
error(err, 0)
|
|
end
|
|
return ...
|
|
end
|
|
|
|
--- @class vim.async.Closable
|
|
--- @field close fun(self, callback?: fun())
|
|
--- @field is_closing? fun(self): boolean
|
|
|
|
--- A coroutine-backed async operation and concurrency scope.
|
|
---
|
|
--- Use [vim.async.run()] to create tasks. A task may be awaited by more than
|
|
--- one waiter. When a task is created inside another task, it is attached to
|
|
--- that parent and becomes part of the parent's concurrency scope.
|
|
---
|
|
--- @class vim.async.Task<R>: vim.async.Closable
|
|
--- @field package _thread thread
|
|
--- @field package _future vim.async.Future<R>
|
|
--- @field package _closing boolean
|
|
--- @field package _error? any
|
|
--- @field package _finalizing_children boolean
|
|
--- @field package _started boolean
|
|
---
|
|
--- Reference to parent to handle attaching/detaching.
|
|
--- @field package _parent? vim.async.Task<any>
|
|
--- @field package _parent_children_idx? integer
|
|
---
|
|
--- Name of the task
|
|
--- @field name? string
|
|
---
|
|
--- Hide implementation tasks from user-facing inspection output.
|
|
--- @field package _hidden? boolean
|
|
---
|
|
--- The source line that created this task, used for inspect().
|
|
--- @field package _caller? string
|
|
---
|
|
--- Maintain children as an array to preserve closure order.
|
|
--- @field package _children table<integer, vim.async.Task<any>?>
|
|
---
|
|
--- Pointer to last child in children
|
|
--- @field package _children_idx integer
|
|
---
|
|
--- Tasks can await other async functions (task of callback functions)
|
|
--- when we are waiting on a child, we store the handle to it here so we can
|
|
--- close it.
|
|
--- @field package _awaiting? vim.async.Task<any> | vim.async.Closable
|
|
--- Removes the completion callback when a Task await is abandoned.
|
|
--- @field package _awaiting_unsubscribe? fun()
|
|
local Task = {}
|
|
|
|
--- @return_cast x vim.async.Task<any>
|
|
--- @param x any
|
|
local function is_task(x)
|
|
return getmetatable(x) == Task
|
|
end
|
|
|
|
do --- Task
|
|
Task.__index = Task
|
|
|
|
--- @package
|
|
--- @param name? string
|
|
--- @param func async fun(...: any)
|
|
--- @return vim.async.Task<any>
|
|
function Task._new(name, func, ...)
|
|
local func_args = pack_len(...) --[[@as any[]? ]]
|
|
local thread = coroutine.create(function(marker, err)
|
|
-- Drop the packed vararg table before user code can suspend; otherwise
|
|
-- the coroutine closure retains it for the task lifetime.
|
|
local args = func_args
|
|
func_args = nil
|
|
check_yield(marker, err)
|
|
return func(unpack_len(args))
|
|
end)
|
|
|
|
local self = setmetatable({
|
|
name = name,
|
|
_closing = false,
|
|
_finalizing_children = false,
|
|
_started = false,
|
|
_thread = thread,
|
|
_future = future(),
|
|
_children = {},
|
|
_children_idx = 0,
|
|
}, Task)
|
|
|
|
threads[thread] = self
|
|
|
|
return self
|
|
end
|
|
|
|
--- Returns whether the Task has completed.
|
|
--- @return boolean
|
|
function Task:completed()
|
|
return self._future:completed()
|
|
end
|
|
|
|
--- Add a callback to be run when the Task has completed.
|
|
---
|
|
--- If the Task is already done when this method is called, the callback is
|
|
--- called immediately with the results.
|
|
---
|
|
--- This only observes completion. It does not start a pending task.
|
|
--- @param callback fun(err?: any, ...: R...)
|
|
--- @return fun() unsubscribe
|
|
function Task:on_complete(callback)
|
|
validate('callback', callback, 'callable')
|
|
return self._future:on_complete(callback)
|
|
end
|
|
|
|
--- Synchronously wait for the Task to complete.
|
|
---
|
|
--- If a timeout is provided, waits for the given time in milliseconds before
|
|
--- failing with `"timeout"`. With no timeout, waits indefinitely.
|
|
---
|
|
--- This is for synchronous code. Inside a task, prefer [vim.async.await()] so
|
|
--- the current task suspends instead of pumping the event loop itself.
|
|
---
|
|
--- ```lua
|
|
--- local result = task:wait(10) -- wait for 10ms or raise "timeout"
|
|
---
|
|
--- local result = task:wait() -- wait indefinitely
|
|
--- ```
|
|
--- @param timeout integer?
|
|
--- @return R...
|
|
function Task:wait(timeout)
|
|
validate('timeout', timeout, 'number', true)
|
|
self:_start()
|
|
|
|
if not runtime.wait(timeout or maxint, function()
|
|
return self:completed()
|
|
end) then
|
|
error('timeout', 2)
|
|
end
|
|
local res = pack_len(self._future:result())
|
|
|
|
assert(self:status() == 'completed' or res[2] == yield_error)
|
|
|
|
if not res[1] then
|
|
error(res[2], 2)
|
|
end
|
|
|
|
return unpack(res, 2, res.n)
|
|
end
|
|
|
|
--- Protected-call version of [Task:wait()].
|
|
---
|
|
--- Equivalent to `pcall(task.wait, task, timeout)`.
|
|
---
|
|
--- ```lua
|
|
--- local ok, result_or_err = task:pwait(1000)
|
|
--- if not ok then
|
|
--- vim.notify(tostring(result_or_err), vim.log.levels.ERROR)
|
|
--- end
|
|
--- ```
|
|
--- @param timeout integer?
|
|
--- @return boolean, R...
|
|
function Task:pwait(timeout)
|
|
validate('timeout', timeout, 'number', true)
|
|
return pcall(self.wait, self, timeout)
|
|
end
|
|
|
|
--- @package
|
|
--- @param parent? vim.async.Task<any>
|
|
function Task:_attach(parent)
|
|
if parent then
|
|
-- Attach to parent
|
|
parent._children_idx = parent._children_idx + 1
|
|
parent._children[parent._children_idx] = self
|
|
|
|
-- Keep track of the parent and this tasks index so we can detach
|
|
self._parent = parent
|
|
self._parent_children_idx = parent._children_idx
|
|
end
|
|
end
|
|
|
|
--- Remove this task from its parent without changing execution state.
|
|
--- @private
|
|
--- @return boolean removed
|
|
function Task:_detach()
|
|
if not self._parent then
|
|
return false
|
|
end
|
|
|
|
self._parent._children[self._parent_children_idx] = nil
|
|
self._parent = nil
|
|
self._parent_children_idx = nil
|
|
return true
|
|
end
|
|
|
|
--- Detach a task from its parent.
|
|
---
|
|
--- The task becomes a top-level task.
|
|
--- If it was waiting for a parent checkpoint, it is scheduled to start.
|
|
---
|
|
--- Use this for background work that should not be cancelled when the
|
|
--- current task finishes. Detached task failures no longer fail the original
|
|
--- parent, so observe them explicitly with [Task:on_complete()],
|
|
--- [Task:wait()], or [vim.async.await()].
|
|
---
|
|
--- ```lua
|
|
--- vim.async.run(function()
|
|
--- while true do
|
|
--- refresh_index()
|
|
--- vim.async.sleep(1000)
|
|
--- end
|
|
--- end):detach()
|
|
--- ```
|
|
--- @return vim.async.Task<R>
|
|
function Task:detach()
|
|
local should_start = self._parent and not self._started and not self:completed()
|
|
self:_detach()
|
|
if should_start then
|
|
runtime.schedule(function()
|
|
self:_start()
|
|
end)
|
|
end
|
|
return self
|
|
end
|
|
|
|
--- Get the traceback of a task when it is not active.
|
|
--- Will also get the traceback of nested tasks.
|
|
---
|
|
--- @param msg? string
|
|
--- @param level? integer
|
|
--- @return string traceback
|
|
function Task:traceback(msg, level)
|
|
level = level or 0
|
|
|
|
local thread = '[' .. tostring(self._thread) .. '] '
|
|
|
|
local awaiting = self._awaiting
|
|
if is_task(awaiting) then
|
|
msg = awaiting:traceback(msg, level + 1)
|
|
end
|
|
|
|
local tblvl = is_task(awaiting) and 2 or nil
|
|
local tb = debug.traceback(self._thread, '', tblvl) or ''
|
|
msg = (msg == nil and '' or tostring(msg)) .. tb:gsub('\n\t', '\n\t' .. thread)
|
|
|
|
if level == 0 then
|
|
--- @type string
|
|
msg = msg
|
|
:gsub('\nstack traceback:\n', '\nSTACK TRACEBACK:\n', 1)
|
|
:gsub('\nstack traceback:\n', '\n')
|
|
:gsub('\nSTACK TRACEBACK:\n', '\nstack traceback:\n', 1)
|
|
end
|
|
|
|
return msg
|
|
end
|
|
|
|
--- Raise this task's error when it completes.
|
|
---
|
|
--- Use this for detached or top-level fire-and-forget tasks whose completion
|
|
--- will not otherwise be observed. Attached task errors already propagate to
|
|
--- their parent.
|
|
---
|
|
--- Detached tasks do not raise errors automatically. Detaching changes
|
|
--- ownership only; their completion can still be handled with
|
|
--- [vim.async.await()], [Task:wait()], [Task:pwait()], or [Task:on_complete()].
|
|
--- @return vim.async.Task<R> self
|
|
function Task:raise_on_error()
|
|
self:on_complete(function(err)
|
|
if err ~= nil then
|
|
error(self:traceback(err), 0)
|
|
end
|
|
end)
|
|
return self
|
|
end
|
|
|
|
--- @package
|
|
function Task:_start()
|
|
if self._started or self:completed() then
|
|
return
|
|
end
|
|
|
|
self._started = true
|
|
self:_resume()
|
|
end
|
|
|
|
--- Start children whose first resume was deferred by `run()`.
|
|
---
|
|
--- Deferring child start lets `await()` or `pawait()` claim the task boundary
|
|
--- before child code runs. At await checkpoints and successful parent finish,
|
|
--- any remaining pending children must start so implicit waits, cancellation,
|
|
--- and inspection see the full task tree.
|
|
--- @package
|
|
function Task:_start_pending_children()
|
|
for i = 1, self._children_idx do
|
|
local child = self._children[i]
|
|
if child then
|
|
child:_start()
|
|
end
|
|
end
|
|
end
|
|
|
|
--- @private
|
|
function Task:_close_children()
|
|
for i = 1, self._children_idx do
|
|
local child = self._children[i]
|
|
if child then
|
|
child:close()
|
|
end
|
|
end
|
|
end
|
|
|
|
--- Keep the first task error. The error can be any non-nil Lua value.
|
|
--- @package
|
|
--- @param err any
|
|
--- @return any
|
|
function Task:_set_error(err)
|
|
if self._error == nil then
|
|
self._error = err
|
|
end
|
|
return self._error
|
|
end
|
|
|
|
--- @package
|
|
--- @param err any
|
|
function Task:_raise(err)
|
|
if self:status() == 'running' then
|
|
-- A running coroutine cannot be resumed recursively, so deliver the
|
|
-- error on a later event-loop turn after the current stack unwinds.
|
|
runtime.schedule(function()
|
|
if not self:completed() then
|
|
self:_resume(err)
|
|
end
|
|
end)
|
|
else
|
|
self:_resume(err)
|
|
end
|
|
end
|
|
|
|
--- Request cooperative close for the task and all of its children.
|
|
---
|
|
--- The optional callback observes task completion and may run immediately if
|
|
--- the task has already completed.
|
|
---
|
|
--- Closing is cooperative. The task observes the close request at a
|
|
--- checkpoint such as [vim.async.await()] or [vim.async.checkpoint()]. If the
|
|
--- task is suspended on an owned closable operation, that operation is closed
|
|
--- before the task reports `"closed"`.
|
|
---
|
|
--- @param callback? fun()
|
|
function Task:close(callback)
|
|
if not self:completed() and not self._closing then
|
|
self._closing = true
|
|
self:_raise('closed')
|
|
end
|
|
if callback then
|
|
self:on_complete(function()
|
|
callback()
|
|
end)
|
|
end
|
|
end
|
|
|
|
--- Record a child failure on this task and either deliver it to a live
|
|
--- parent or close sibling children during finalization.
|
|
--- @package
|
|
--- @param child vim.async.Task<any>
|
|
--- @param err any
|
|
--- @param child_was_awaited boolean?
|
|
function Task:_child_failed(child, err, child_was_awaited)
|
|
-- A parent close turns child "closed" results into cleanup, not failure.
|
|
if child._closing or child_was_awaited or (self._closing and err == 'closed') then
|
|
return
|
|
end
|
|
|
|
local task_err = self:_set_error('child error: ' .. util._stringify_error(err))
|
|
if self._finalizing_children then
|
|
-- The parent coroutine is already dead, so sibling cleanup must wake the
|
|
-- finalizer instead of trying to resume the parent.
|
|
self:_close_children()
|
|
else
|
|
self:_raise(task_err)
|
|
end
|
|
end
|
|
|
|
--- Checks if an object is closable, i.e., has a `close` method.
|
|
--- @param obj any
|
|
--- @return boolean
|
|
--- @return_cast obj vim.async.Closable
|
|
local function is_closable(obj)
|
|
local ty = type(obj)
|
|
return (ty == 'table' or ty == 'userdata') and is_callable(obj.close)
|
|
end
|
|
|
|
do -- Task:_resume()
|
|
--- Complete this task with an error and propagate it to the parent if the
|
|
--- parent did not explicitly await this task.
|
|
--- @param parent? vim.async.Task<any>
|
|
--- @param err any
|
|
function Task:_finish_error(parent, err)
|
|
if err == nil then
|
|
err = self._error
|
|
end
|
|
err = util._normalize_error(err)
|
|
if parent then
|
|
parent:_child_failed(self, err, parent._awaiting == self)
|
|
end
|
|
self._future:complete(err)
|
|
end
|
|
|
|
--- @private
|
|
--- @param stat boolean
|
|
--- @param ... R... result
|
|
function Task:_finish(stat, ...)
|
|
if self:completed() then
|
|
return
|
|
end
|
|
|
|
local parent = self._parent
|
|
self:_detach()
|
|
|
|
threads[self._thread] = nil
|
|
|
|
if not stat then
|
|
self:_finish_error(parent, ...)
|
|
else
|
|
if self._error ~= nil then
|
|
self:_finish_error(parent, self._error)
|
|
else
|
|
self._future:complete(nil, ...)
|
|
end
|
|
end
|
|
end
|
|
|
|
--- @package
|
|
--- @param stat boolean
|
|
--- @param ... R... result
|
|
function Task:_finalize(stat, ...)
|
|
if next(self._children) == nil then
|
|
self:_finish(stat, ...)
|
|
return
|
|
end
|
|
|
|
local finish_args = pack_len(stat, ...)
|
|
self._finalizing_children = true
|
|
-- Only spawn the helper after the no-child path; an empty helper task
|
|
-- would otherwise finalize by recursively spawning another helper.
|
|
local await_children = Task._new('await_children', function()
|
|
-- TODO(lewis6991): should we collect all errors?
|
|
local close_remaining = not stat
|
|
|
|
if close_remaining then
|
|
self:_close_children()
|
|
else
|
|
self:_start_pending_children()
|
|
end
|
|
|
|
for i = 1, self._children_idx do
|
|
local child = self._children[i]
|
|
if child then
|
|
-- Child failures are recorded on `self`. Protect this helper so one
|
|
-- failed child cannot stop it awaiting the remaining cleanup.
|
|
local ok, err = M.pawait(child)
|
|
-- A close can arrive while normal finalization is awaiting
|
|
-- children; from that point child errors are cleanup results.
|
|
if not close_remaining and not self._closing and not ok and not child._closing then
|
|
self:_set_error('child error: ' .. util._stringify_error(err))
|
|
close_remaining = true
|
|
self:_close_children()
|
|
end
|
|
end
|
|
end
|
|
|
|
self._finalizing_children = false
|
|
if stat and self._closing and self._error == nil then
|
|
self:_finish(false, 'closed')
|
|
else
|
|
self:_finish(unpack_len(finish_args))
|
|
end
|
|
end)
|
|
await_children._hidden = true
|
|
await_children:_start()
|
|
end
|
|
|
|
--- Resume a task with the raw or protected result of an await.
|
|
--- @param task vim.async.Task<any>
|
|
--- @param yielded vim.async.Task<any>|fun(callback: fun(err?: any, ...: any)): vim.async.Closable?
|
|
--- @param protected boolean?
|
|
--- @param err? any
|
|
--- @param ... any
|
|
local function resume_from_await(task, yielded, protected, err, ...)
|
|
-- An error from `await(task)` also marks the waiting task as failed, even
|
|
-- if the error raised by `await()` is caught.
|
|
if not protected and is_task(yielded) and err ~= nil then
|
|
task:_set_error(err)
|
|
end
|
|
|
|
if protected then
|
|
if err ~= nil then
|
|
return task:_resume(nil, false, err)
|
|
end
|
|
return task:_resume(nil, true, ...)
|
|
end
|
|
|
|
return task:_resume(err, ...)
|
|
end
|
|
|
|
--- Begin waiting on a yielded awaitable.
|
|
--- @param task vim.async.Task<any>
|
|
--- @param yielded vim.async.Task<any>|fun(callback: fun(err?: any, ...: any)): vim.async.Closable?
|
|
--- @param protected boolean?
|
|
local function start_await(task, yielded, protected)
|
|
-- TODO(#36): Defer task control until setup returns its cleanup handle.
|
|
-- The first callback or setup failure settles the await.
|
|
-- Ignore any callback that arrives afterwards.
|
|
local settled = false
|
|
local setup_ok --- @type boolean?
|
|
|
|
-- Await setup may invoke the callback before `_awaiting` is installed.
|
|
-- Save those arguments so the task resumes only after setup finishes.
|
|
-- `sync_args` has three states:
|
|
-- - `nil`: the callback did not fire during setup;
|
|
-- - `false`: `callback(nil)`, the common no-error/no-result case,
|
|
-- avoiding a table allocation;
|
|
-- - a table: every other argument list, packed to preserve nils.
|
|
local sync_args --- @type false|{[integer]: any, n: integer}?
|
|
local awaiting --- @type vim.async.Task<any>|vim.async.Closable?
|
|
|
|
--- @param err? any
|
|
--- @param ... any
|
|
local function complete_await(err, ...)
|
|
-- Cancellation and child failures resume through `_raise()`. Ignore a
|
|
-- racing result so `_resume()` can finish awaitable cleanup first.
|
|
if settled or task._closing or task._error ~= nil then
|
|
return
|
|
end
|
|
settled = true
|
|
|
|
if setup_ok == nil then
|
|
if err == nil and select('#', ...) == 0 then
|
|
sync_args = false
|
|
else
|
|
sync_args = pack_len(err, ...)
|
|
end
|
|
else
|
|
-- The callback has fired. Keep `_awaiting` so `_resume()` can close a
|
|
-- callback-style handle or retain a failed Task for its traceback.
|
|
task._awaiting_unsubscribe = nil
|
|
|
|
if not task:completed() then
|
|
return resume_from_await(task, yielded, protected, err, ...)
|
|
end
|
|
end
|
|
end
|
|
|
|
local unsubscribe
|
|
local task_await --- @type boolean
|
|
local setup_result
|
|
-- Either call below may invoke `complete_await()` before returning. In that
|
|
-- case, `unsubscribe` or `awaiting` has not yet received the returned cleanup
|
|
-- handle. While `setup_ok` is nil, `complete_await()` saves its arguments
|
|
-- in `sync_args`. This lets the code below handle setup errors and install the
|
|
-- cleanup state before using the buffered result.
|
|
if is_task(yielded) then
|
|
task_await = true
|
|
--- @diagnostic disable-next-line: cast-local-type
|
|
awaiting = yielded
|
|
setup_ok, setup_result = pcall(yielded._future.on_complete, yielded._future, complete_await)
|
|
else
|
|
task_await = false
|
|
-- Callback setup has one result: the optional closable.
|
|
--- @type fun(callback: fun(err?: any, ...: any)): vim.async.Closable?
|
|
local awaitable = yielded
|
|
setup_ok, setup_result = pcall(awaitable, complete_await)
|
|
end
|
|
|
|
if not setup_ok then
|
|
local err = util._normalize_error(setup_result)
|
|
if protected and settled then
|
|
-- The first synchronous callback wins over a later setup error.
|
|
awaiting = nil
|
|
else
|
|
settled = true
|
|
return resume_from_await(task, yielded, protected, err)
|
|
end
|
|
elseif task_await then
|
|
unsubscribe = setup_result
|
|
else
|
|
awaiting = setup_result
|
|
end
|
|
|
|
if not is_closable(awaiting) then
|
|
awaiting = nil
|
|
end
|
|
--- @diagnostic disable-next-line: assign-type-mismatch
|
|
task._awaiting = awaiting
|
|
|
|
if is_task(awaiting) then
|
|
if not settled and unsubscribe then
|
|
--- @diagnostic disable-next-line: assign-type-mismatch
|
|
task._awaiting_unsubscribe = unsubscribe
|
|
end
|
|
awaiting:_start()
|
|
end
|
|
|
|
if task:completed() then
|
|
return
|
|
end
|
|
|
|
task:_start_pending_children()
|
|
|
|
if sync_args == false then
|
|
return resume_from_await(task, yielded, protected)
|
|
elseif sync_args then
|
|
return resume_from_await(task, yielded, protected, unpack_len(sync_args))
|
|
end
|
|
end
|
|
|
|
--- Finalize a completed coroutine or start its yielded await.
|
|
--- Keep results in varargs to preserve nils without packing them.
|
|
--- @param task vim.async.Task<any>
|
|
--- @param stat boolean
|
|
--- @param ... any
|
|
local function handle_resume(task, stat, ...)
|
|
if coroutine.status(task._thread) == 'dead' then
|
|
-- The coroutine finished during resume. A normal return must not
|
|
-- overwrite a pending task failure.
|
|
if task._error ~= nil and stat then
|
|
task:_finalize(false, task._error)
|
|
elseif task._closing and stat then
|
|
task:_finalize(false, 'closed')
|
|
else
|
|
task:_finalize(stat, ...)
|
|
end
|
|
return
|
|
end
|
|
|
|
local marker, yielded, protected = ...
|
|
if marker ~= yield_marker or (not is_task(yielded) and not is_callable(yielded)) then
|
|
task:_finalize(false, yield_error)
|
|
return
|
|
end
|
|
|
|
return start_await(task, yielded, protected)
|
|
end
|
|
|
|
--- Clear an await boundary and remove its Task completion callback.
|
|
--- @param task vim.async.Task<any>
|
|
local function clear_awaiting(task)
|
|
local unsubscribe = task._awaiting_unsubscribe
|
|
task._awaiting = nil
|
|
task._awaiting_unsubscribe = nil
|
|
if unsubscribe then
|
|
unsubscribe()
|
|
end
|
|
end
|
|
|
|
--- @package
|
|
--- @param err? any
|
|
--- @param ... any resume values
|
|
function Task:_resume(err, ...)
|
|
-- Clear self._awaiting when either:
|
|
-- - this task resumes before a non-child finishes, so its callback
|
|
-- cannot retain this task; or
|
|
-- - there is no raw error needing its traceback frames and
|
|
-- self._awaiting is finished.
|
|
if
|
|
is_task(self._awaiting)
|
|
and (
|
|
(self._awaiting._parent ~= self and self._awaiting_unsubscribe)
|
|
or (err == nil and self._awaiting:completed())
|
|
)
|
|
then
|
|
clear_awaiting(self)
|
|
end
|
|
|
|
local awaiting = self._awaiting
|
|
-- Only close awaitables owned by this task; external tasks are observed.
|
|
if awaiting and (not is_task(awaiting) or awaiting._parent == self) then
|
|
local already_closing = false
|
|
if type(awaiting.is_closing) == 'function' then
|
|
already_closing = awaiting:is_closing()
|
|
end
|
|
|
|
if already_closing then
|
|
clear_awaiting(self)
|
|
return self:_resume(err, ...)
|
|
end
|
|
|
|
local args = pack_len(err, ...)
|
|
-- We must close the closable child before we resume to ensure
|
|
-- all resources are collected.
|
|
--- @diagnostic disable-next-line: param-type-mismatch
|
|
local close_ok, close_err = pcall(awaiting.close, awaiting, function()
|
|
clear_awaiting(self)
|
|
return self:_resume(unpack_len(args))
|
|
end)
|
|
|
|
if close_ok then
|
|
return
|
|
end
|
|
clear_awaiting(self)
|
|
return self:_resume(util._normalize_error(close_err))
|
|
end
|
|
|
|
-- An external coroutine.resume() may have already killed the coroutine.
|
|
-- Finalize its pending failure instead of trying to resume it again.
|
|
if coroutine.status(self._thread) == 'dead' then
|
|
self:_finalize(false, err, ...)
|
|
return
|
|
end
|
|
|
|
return handle_resume(self, coroutine.resume(self._thread, resume_marker, err, ...))
|
|
end
|
|
end
|
|
|
|
--- @package
|
|
function Task:_log(...)
|
|
print(tostring(self._thread), ...)
|
|
end
|
|
|
|
--- Returns the status of the task:
|
|
---
|
|
--- - `"running"`: task is currently executing Lua code
|
|
--- - `"normal"`: task is active but another coroutine is running
|
|
--- - `"awaiting"`: task is suspended at a checkpoint or waiting for children
|
|
--- - `"completed"`: task and all attached children have completed
|
|
--- @return "running" | "awaiting" | "normal" | "completed"
|
|
function Task:status()
|
|
if self:completed() then
|
|
return 'completed'
|
|
end
|
|
|
|
local co_status = coroutine.status(self._thread)
|
|
if co_status == 'dead' then
|
|
return 'awaiting'
|
|
elseif co_status == 'suspended' then
|
|
return 'awaiting'
|
|
elseif co_status == 'normal' then
|
|
-- TODO(lewis6991): This state is a bit ambiguous. If all tasks
|
|
-- are started from the main thread, then we can remove this state.
|
|
-- Though it still may be possible if the user resumes a non-task
|
|
-- coroutine.
|
|
return 'normal'
|
|
end
|
|
assert(co_status == 'running')
|
|
return 'running'
|
|
end
|
|
end
|
|
|
|
--- @generic T, R
|
|
--- @param name? string
|
|
--- @param func async fun(...: T...): R... Function to run in an async context
|
|
--- @param ... T... Arguments to pass to the function
|
|
--- @return vim.async.Task<R...>
|
|
local function run(name, func, ...)
|
|
validate('func', func, 'callable')
|
|
local task = Task._new(name, func, ...)
|
|
task:_attach(running())
|
|
if runtime.debug then
|
|
local info = debug.getinfo(2, 'Sl')
|
|
if info and info.currentline then
|
|
task._caller = ('%s:%d'):format(info.source, info.currentline)
|
|
end
|
|
end
|
|
|
|
-- Top-level tasks have no parent checkpoint to start them, so they start
|
|
-- immediately. Attached children start when their parent next reaches an
|
|
-- await checkpoint, or when the parent finishes successfully and implicitly
|
|
-- waits for its children. If the parent errors or closes, pending children
|
|
-- are closed without running user code.
|
|
if not task._parent then
|
|
task:_start()
|
|
end
|
|
|
|
return task
|
|
end
|
|
|
|
--- Create a task from an async function.
|
|
---
|
|
--- Top-level tasks start immediately. Child tasks are attached immediately and
|
|
--- first run when their parent reaches a checkpoint.
|
|
---
|
|
--- Creating a task decides ownership. Awaiting the task later only observes
|
|
--- its result; it does not attach the task to the awaiter.
|
|
---
|
|
--- ```lua
|
|
--- local async = vim.async
|
|
---
|
|
--- async.run(function()
|
|
--- local child = async.run(function()
|
|
--- return read_file('notes.txt')
|
|
--- end)
|
|
---
|
|
--- local text = async.await(child)
|
|
--- show_buffer(text)
|
|
--- end)
|
|
--- ```
|
|
---
|
|
--- A task created from synchronous code is top-level:
|
|
---
|
|
--- ```lua
|
|
--- local task = vim.async.run(function()
|
|
--- vim.async.sleep(100)
|
|
--- return 'done'
|
|
--- end)
|
|
---
|
|
--- print(task:wait())
|
|
--- ```
|
|
--- @generic T, R
|
|
--- @param func async fun(...: T...): R...
|
|
--- @param ... T... Arguments to pass to the function
|
|
--- @return vim.async.Task<R...>
|
|
--- @overload fun(name: string, func: async fun(...: T...), ...: T...): vim.async.Task<R...>
|
|
function M.run(func, ...)
|
|
if type(func) == 'string' then
|
|
return run(func, ...)
|
|
elseif is_callable(func) then
|
|
return run(nil, func, ...)
|
|
end
|
|
error('Invalid arguments')
|
|
end
|
|
|
|
--- @generic T, R
|
|
--- @param argc integer
|
|
--- @param fun fun(...: T..., callback: fun(...: R...))
|
|
--- @param ... T... func arguments
|
|
--- @return fun(callback: fun(...: R...))
|
|
local function norm_cb_fun(argc, fun, ...)
|
|
if argc == 1 and select('#', ...) == 0 then
|
|
-- Avoid allocating an empty argument table for the common await(func) shape.
|
|
local cb_fun = fun
|
|
--- @cast cb_fun fun(callback: fun(...: any)): any?
|
|
--- @param callback fun(...: any)
|
|
--- @return any?
|
|
return function(callback)
|
|
return cb_fun(function(...)
|
|
callback(nil, ...)
|
|
end)
|
|
end
|
|
end
|
|
|
|
local args = pack_len(...)
|
|
|
|
--- @param callback fun(...: any)
|
|
--- @return any?
|
|
return function(callback)
|
|
args[argc] = function(...)
|
|
callback(nil, ...)
|
|
end
|
|
args.n = math.max(args.n, argc)
|
|
return fun(unpack_len(args))
|
|
end
|
|
end
|
|
|
|
--- Get the current task, failing before an operation yields if it is closing or
|
|
--- failed.
|
|
--- @return vim.async.Task<any>
|
|
local function check_current_task()
|
|
local task = current_task()
|
|
|
|
if task._closing then
|
|
error('closed', 0)
|
|
elseif task._error ~= nil then
|
|
error(task._error, 0)
|
|
end
|
|
|
|
return task
|
|
end
|
|
|
|
--- Convert the public await forms into a Task or callback awaitable.
|
|
---
|
|
--- Callback-style APIs do not have an error slot, so `norm_cb_fun()` inserts
|
|
--- `nil`; the scheduler observes Task futures directly.
|
|
--- @param ... any
|
|
--- @return vim.async.Task<any>|fun(callback: fun(err?: any, ...: any)): vim.async.Closable?
|
|
local function to_awaitable(...)
|
|
local arg1 = select(1, ...)
|
|
|
|
if type(arg1) == 'number' then
|
|
return norm_cb_fun(...)
|
|
elseif type(arg1) == 'function' then
|
|
return norm_cb_fun(1, arg1)
|
|
elseif is_task(arg1) then
|
|
return arg1
|
|
else
|
|
error('Invalid arguments, expected Task or (argc, func) got: ' .. tostring(arg1), 2)
|
|
end
|
|
end
|
|
|
|
--- Suspend the current task until an awaitable completes.
|
|
---
|
|
--- Accepts a task, a callback-taking function, or an argument position plus a
|
|
--- callback-taking function. Raises awaited errors and current task-control
|
|
--- errors.
|
|
---
|
|
--- The callback forms return callback arguments unchanged:
|
|
---
|
|
--- ```lua
|
|
--- local async = vim.async
|
|
---
|
|
--- async.run(function()
|
|
--- local err, stat = async.await(2, vim.uv.fs_stat, 'notes.txt')
|
|
--- if err then
|
|
--- error(err, 0)
|
|
--- end
|
|
--- print(stat.size)
|
|
--- end)
|
|
--- ```
|
|
---
|
|
--- If a callback API starts cancellable work, return a closable handle from the
|
|
--- await callback. [Task:close()] will close that handle if cancellation
|
|
--- arrives while the task is suspended there.
|
|
---
|
|
--- ```lua
|
|
--- local async = vim.async
|
|
---
|
|
--- async.run(function()
|
|
--- local lines = async.await(function(done)
|
|
--- return start_read_lines('notes.txt', done)
|
|
--- end)
|
|
--- render(lines)
|
|
--- end)
|
|
--- ```
|
|
--- @async
|
|
--- @generic T, R
|
|
--- @param ... any see overloads
|
|
--- @overload async fun(func: (fun(callback: fun(...: R...)): vim.async.Closable?)): R...
|
|
--- @overload async fun(argc: integer, func: (fun(...: T..., callback: fun(...: R...)): vim.async.Closable?), ...: T...): R...
|
|
--- @overload async fun(task: vim.async.Task<R>): R...
|
|
--- @return R...
|
|
function M.await(...)
|
|
check_current_task()
|
|
return check_yield(coroutine.yield(yield_marker, to_awaitable(...)))
|
|
end
|
|
|
|
--- Protected await.
|
|
---
|
|
--- Async counterpart to `pcall()`. Accepts the same forms as
|
|
--- [vim.async.await()], but returns a leading `ok` boolean for
|
|
--- awaited-operation failures.
|
|
---
|
|
--- Use this when the awaited task or operation is allowed to fail and the
|
|
--- current task should continue. Cancellation or already pending failure from
|
|
--- the current task is not protected.
|
|
---
|
|
--- ```lua
|
|
--- local async = vim.async
|
|
---
|
|
--- async.run(function()
|
|
--- local ok, text_or_err = async.pawait(async.run(read_file, 'notes.txt'))
|
|
--- if not ok then
|
|
--- text_or_err = ''
|
|
--- end
|
|
---
|
|
--- show_buffer(text_or_err)
|
|
--- end)
|
|
--- ```
|
|
--- @async
|
|
--- @generic T, R
|
|
--- @param ... any see overloads
|
|
--- @overload async fun(func: (fun(callback: fun(...: R...)): vim.async.Closable?)): boolean, R...
|
|
--- @overload async fun(argc: integer, func: (fun(...: T..., callback: fun(...: R...)): vim.async.Closable?), ...: T...): boolean, R...
|
|
--- @overload async fun(task: vim.async.Task<R>): boolean, R...
|
|
--- @return boolean ok
|
|
--- @return R... ... result or error
|
|
--- @return_overload true, R...
|
|
--- @return_overload false, any
|
|
function M.pawait(...)
|
|
check_current_task()
|
|
return check_yield(coroutine.yield(yield_marker, to_awaitable(...), true))
|
|
end
|
|
|
|
--- Start pending child tasks and deliver pending cancellation or task failure
|
|
--- from the current task.
|
|
---
|
|
--- This does not yield to the event loop.
|
|
---
|
|
--- Use this after cleanup code that catches an async failure or close signal,
|
|
--- so persistent task state is delivered again before normal execution
|
|
--- continues.
|
|
---
|
|
--- ```lua
|
|
--- local ok, err = pcall(cleanup_sensitive_work)
|
|
--- cleanup_resources()
|
|
--- vim.async.checkpoint()
|
|
--- if not ok then
|
|
--- error(err, 0)
|
|
--- end
|
|
--- ```
|
|
--- @async
|
|
function M.checkpoint()
|
|
M.await(function(callback)
|
|
callback()
|
|
end)
|
|
end
|
|
|
|
--- Returns true if the current task has been closed.
|
|
---
|
|
--- Can be used in an async function to do cleanup when a task is closing.
|
|
---
|
|
--- ```lua
|
|
--- while not vim.async.is_closing() do
|
|
--- poll_once()
|
|
--- vim.async.sleep(1000)
|
|
--- end
|
|
--- ```
|
|
--- @return boolean
|
|
function M.is_closing()
|
|
local task = running()
|
|
return task and task._closing or false
|
|
end
|
|
|
|
--- @private
|
|
--- @param parent? vim.async.Task<any>
|
|
--- @param prefix? string
|
|
--- @return string[]
|
|
local function inspect(parent, prefix)
|
|
local tasks = {} --- @type table<any, vim.async.Task<any>?>
|
|
if parent then
|
|
for _, task in pairs(parent._children) do
|
|
if not task._hidden then
|
|
tasks[#tasks + 1] = task
|
|
end
|
|
end
|
|
else
|
|
-- Gather for all detached tasks
|
|
for _, task in pairs(threads) do
|
|
if not task._parent and not task._hidden then
|
|
tasks[#tasks + 1] = task
|
|
end
|
|
end
|
|
end
|
|
|
|
local r = {} --- @type string[]
|
|
for i, task in ipairs(tasks) do
|
|
local last = i == #tasks
|
|
local label = task.name or ''
|
|
if task._caller then
|
|
label = label .. task._caller
|
|
end
|
|
if label ~= '' then
|
|
label = label .. ' '
|
|
end
|
|
r[#r + 1] = ('%s%s%s[%s]'):format(
|
|
prefix or '',
|
|
parent and (last and '└─ ' or '├─ ') or '',
|
|
label,
|
|
task:status()
|
|
)
|
|
local child_prefix = (prefix or '') .. (parent and (last and ' ' or '│ ') or '')
|
|
for _, line in ipairs(inspect(task, child_prefix)) do
|
|
r[#r + 1] = line
|
|
end
|
|
end
|
|
return r
|
|
end
|
|
|
|
--- Inspect the current async task tree.
|
|
---
|
|
--- Returns a string representation of the task tree, showing the names and
|
|
--- statuses of each task.
|
|
--- @return string
|
|
function M._inspect_tree()
|
|
-- Inspired by https://docs.python.org/3.14/whatsnew/3.14.html#asyncio-introspection-capabilities
|
|
return table.concat(inspect(), '\n')
|
|
end
|
|
|
|
return M
|