diff --git a/.luarc.json b/.luarc.json index 9fd1e38a96..bca1c08b37 100644 --- a/.luarc.json +++ b/.luarc.json @@ -5,6 +5,7 @@ }, "workspace": { "ignoreDir": [ + "/lua/vim/async.lua", ".deps", "build" ], diff --git a/runtime/doc/lua-async.txt b/runtime/doc/lua-async.txt new file mode 100644 index 0000000000..f805090919 --- /dev/null +++ b/runtime/doc/lua-async.txt @@ -0,0 +1,603 @@ + +============================================================================== +Lua module: vim.async *lua-async* + +Structured async API for Lua code that waits on event-loop work. + +`vim.async` lets Lua code wait for timers, callbacks, and other tasks without +blocking Nvim's event loop. Async work runs inside tasks, which can pause at +checkpoints and manage child tasks created while they are running. + +Start async work with |vim.async.run()|. Inside a task, use +|vim.async.await()| to wait for callback-style APIs or other tasks without +blocking the event loop. Use |vim.async.pawait()| when an awaited operation +can fail and the current task should continue. + +Examples in this help use `local async = vim.async` for brevity. + +Example: run async work without blocking Nvim: >lua + local async = vim.async + + async.run(function() + vim.notify('waiting...') + async.sleep(1000) + vim.notify('done') + end) +< + +Example: await a callback-style API. Callback results are returned unchanged, +so an error-first callback still returns `err, value`: >lua + local async = vim.async + + async.run(function() + local err, stat = async.await(2, fs_stat, 'notes.txt') + if err then + error(err, 0) + end + print(('notes.txt is %d bytes'):format(stat.size)) + end) +< + +A task has two roles: +• it is a handle that can be awaited, waited for, or closed +• it is a scope for child tasks created while the task is running + +|vim.async.run()| creates a task. A top-level task starts immediately. A task +created while another task is running becomes a child of that task, and its +function starts when the parent reaches its next checkpoint. A parent task +finishes only after its attached children finish. If a child fails without +being handled, the parent fails and closes the remaining children. + +Use |Task:detach()| for background work that should keep running after the +current task finishes. A detached task becomes top-level work; the original +parent no longer waits for it or closes it. + +Awaiting a task observes that task's result; it does not attach the task to +the awaiter or change which task owns it. Ownership is decided when the task +is created. + +Scheduling is cooperative. When a task awaits a timer, I/O operation, +callback, or another task, `vim.async` saves the Lua stack and returns control +to the event loop. Other callbacks can run while the task is paused. Nothing +interrupts synchronous Lua code in the middle of a stack frame. + +Checkpoints are the places where a task can pause, start pending child tasks, +observe cancellation, and receive unhandled child failures. Inside a task, +these operations are checkpoints: +• `vim.async.await(...)` +• `vim.async.pawait(...)` +• `vim.async.checkpoint()` +• successful return from the task function, which is the final checkpoint for + child management + +Convenience APIs such as |vim.async.sleep()| and |vim.async.timeout()| can +also checkpoint because they call checkpointing APIs internally. + +Closing a task closes its attached children. Cancellation is cooperative: +|Task:close()| marks a task as closing, and the task observes that state at a +checkpoint. If a task is suspended on a closable operation such as a timer or +child task, `vim.async` closes that operation before reporting the +cancellation. + +Use |vim.async.await()| inside a task to suspend until work completes. It +accepts a task, a callback-taking function, or an argument position plus a +callback-taking function. `await(task)` returns the task result or raises the +task failure. |vim.async.pawait()| is the async counterpart to `pcall()` for +recoverable awaited-operation failures; it returns `ok, ...` instead of +failing the current task for that awaited operation. It does not suppress +cancellation or a failure already pending on the current task. + +From synchronous code, use |Task:wait()| or |Task:pwait()| to pump the event +loop until the task completes. Use |Task:on_complete()| to observe completion +without blocking. + +Coordination helpers work with task handles. |vim.async.iter()| yields +completed task handles in completion order, |vim.async.timeout()| awaits a +task with a deadline, and `vim.async.semaphore(permits)` creates a +|vim.async.Semaphore| that limits how many tasks can hold a permit for a +section at once. + + +*vim.async.Closable* + + Fields: ~ + • {close} (`fun(self, callback?: fun())`) + • {is_closing}? (`fun(self): boolean`) + +*vim.async.Semaphore* + A semaphore manages an internal permit counter. |Semaphore:acquire()| + consumes one permit and |Semaphore:release()| returns one permit. If no + permits are available, `acquire()` suspends the current task until another + task releases one. + + The preferred way to use a Semaphore is with the `with()` method, which + automatically acquires and releases the semaphore around a function call. + This is useful for limiting sections that start external work and then + await it, such as file reads, requests, or subprocesses. >lua + local async = vim.async + + async.run(function() + local limit = async.semaphore(4) + local tasks = {} + + for _, path in ipairs(paths) do + table.insert(tasks, async.run(function() + return limit:with(function() + return read_file(path) + end) + end)) + end + + local next_task = async.iter(tasks) + while true do + local task = next_task() + if task == nil then + break + end + async.await(task) + end + end) +< + + Fields: ~ + • {acquire} (`fun(self: vim.async.Semaphore)`) See + |Semaphore:acquire()|. + • {release} (`fun(self: vim.async.Semaphore)`) See + |Semaphore:release()|. + • {with} (`fun(self: vim.async.Semaphore, fn: async fun(): R...): R...`) + See |Semaphore:with()|. + +*vim.async.Task* + Extends: |vim.async.Closable| + + 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. + + Fields: ~ + • {close} (`fun(self: vim.async.Task, callback: fun()?)`) See + |Task:close()|. + • {completed} (`fun(self: vim.async.Task): boolean`) See + |Task:completed()|. + • {detach} (`fun(self: vim.async.Task): vim.async.Task`) See + |Task:detach()|. + • {name}? (`string`) Name of the task + • {on_complete} (`fun(self: vim.async.Task, callback: fun(err?: any, ...: R...)): fun()`) + See |Task:on_complete()|. + • {pwait} (`fun(self: vim.async.Task, timeout: integer?): boolean, R...`) + See |Task:pwait()|. + • {raise_on_error} (`fun(self: vim.async.Task): vim.async.Task`) See + |Task:raise_on_error()|. + • {status} (`fun(self: vim.async.Task): "running"|"awaiting"|"normal"|"completed"`) + See |Task:status()|. + • {traceback} (`fun(self: vim.async.Task, msg: string?, level: integer?): string`) + See |Task:traceback()|. + • {wait} (`fun(self: vim.async.Task, timeout: integer?): R...`) + See |Task:wait()|. + + +await({...}) *vim.async.await()* + 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) +< + + Parameters: ~ + • {...} (`any`) see overloads + + Overloads: ~ + • `async fun(func: (fun(callback: fun(...: R...)): vim.async.Closable?)): R...` + • `async fun(argc: integer, func: (fun(...: T..., callback: fun(...: R...)): vim.async.Closable?), ...: T...): R...` + • `async fun(task: vim.async.Task): R...` + + Return: ~ + (`R...`) + +checkpoint() *vim.async.checkpoint()* + 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 +< + +is_closing() *vim.async.is_closing()* + 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`) + +iter({tasks}) *vim.async.iter()* + Iterate completed tasks in completion order. + + The iterator yields task handles, not task results. Use + |vim.async.await()| or |vim.async.pawait()| to retrieve each result. The + tasks are observed in the order they complete, regardless of the order in + the input list. >lua + local async = vim.async + + async.run(function() + local tasks = { + async.run(function() return 'cache', read_cache() end):detach(), + async.run(function() return 'disk', read_file() end):detach(), + } + + for task in async.iter(tasks) do + local ok, source, text = async.pawait(task) + if ok then + for _, other in ipairs(tasks) do + if other ~= task then + other:close() + end + end + print(('loaded from %s'):format(source)) + return text + end + end + end) +< + + If code must support PUC Lua 5.1, use the direct-call form instead of a + generic `for` loop. The iterator may need to suspend while waiting for the + next completed task, and PUC Lua 5.1 cannot yield from a generic-for + iterator call. >lua + local next_task = async.iter(tasks) + while true do + local task = next_task() + if task == nil then + break + end + async.await(task) + end +< + + Parameters: ~ + • {tasks} (`vim.async.Task[]`) A list of tasks to wait for and + iterate over. + + Return: ~ + (`async fun(): vim.async.Task?`) iterator that yields each + completed task. + +pawait({...}) *vim.async.pawait()* + 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) +< + + Parameters: ~ + • {...} (`any`) see overloads + + Overloads: ~ + • `async fun(func: (fun(callback: fun(...: R...)): vim.async.Closable?)): boolean, R...` + • `async fun(argc: integer, func: (fun(...: T..., callback: fun(...: R...)): vim.async.Closable?), ...: T...): boolean, R...` + • `async fun(task: vim.async.Task): boolean, R...` + + Return (multiple): ~ + (`boolean`) ok + (`R...`) ... result or error + (`_overload`) true + (`R...`) + (`_overload`) false + (`any`) + +run({func}, {...}) *vim.async.run()* + 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()) +< + + Parameters: ~ + • {func} (`async fun(...: T...): R...`) + • {...} (`T...`) Arguments to pass to the function + + Overloads: ~ + • `fun(name: string, func: async fun(...: T...), ...: T...): vim.async.Task` + + Return: ~ + (`vim.async.Task`) + +Semaphore:acquire() *Semaphore:acquire()* + Acquire a semaphore permit. + + If the internal counter is greater than zero, decrement it by `1` and + return immediately. If it is `0`, wait until |Semaphore:release()| is + called. + +Semaphore:release() *Semaphore:release()* + Release a semaphore permit. + + Increments the internal counter by `1` and can wake a task waiting in + |Semaphore:acquire()|. + + Calling this more times than permits were acquired raises an error. + +Semaphore:with({fn}) *Semaphore:with()* + Executes a function while holding one semaphore permit. + + This acquires the semaphore before running the function and releases it + after the function completes, even if it errors or the current task is + closed. + + Parameters: ~ + • {fn} (`async fun(): R...`) Function to execute within the semaphore's + context. + + Return: ~ + (`R...`) Result(s) of the executed function. + +sleep({duration}) *vim.async.sleep()* + Asynchronously sleep for a given duration. + + Suspends the current task for the given duration without blocking the + event loop. After the delay and timer cleanup complete, `sleep()` returns + to its caller through the runtime's `schedule` hook. >lua + vim.async.run(function() + vim.async.sleep(100) + vim.notify('resumed later') + end) +< + + Parameters: ~ + • {duration} (`integer`) ms + +Task:close({callback}) *Task:close()* + 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"`. + + Parameters: ~ + • {callback} (`fun()?`) + +Task:completed() *Task:completed()* + Returns whether the Task has completed. + + Return: ~ + (`boolean`) + +Task:detach() *Task:detach()* + 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`) + +Task:on_complete({callback}) *Task:on_complete()* + 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. + + Parameters: ~ + • {callback} (`fun(err?: any, ...: R...)`) + + Return: ~ + (`fun()`) unsubscribe + +Task:pwait({timeout}) *Task:pwait()* + 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 +< + + Parameters: ~ + • {timeout} (`integer?`) + + Return (multiple): ~ + (`boolean`) + (`R...`) + +Task:raise_on_error() *Task:raise_on_error()* + 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`) self + +Task:status() *Task:status()* + 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"`) + +Task:traceback({msg}, {level}) *Task:traceback()* + Get the traceback of a task when it is not active. Will also get the + traceback of nested tasks. + + Parameters: ~ + • {msg} (`string?`) + • {level} (`integer?`) + + Return: ~ + (`string`) traceback + +Task:wait({timeout}) *Task:wait()* + 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 +< + + Parameters: ~ + • {timeout} (`integer?`) + + Return: ~ + (`R...`) + +timeout({duration}, {task}) *vim.async.timeout()* + Await a task with a timeout. + + If the task completes first, returns or raises the task result as + |vim.async.await()| would. If the deadline wins, closes the task and + raises `"timeout"` after the target task finishes cancellation cleanup. >lua + local async = vim.async + + async.run(function() + local task = async.run(read_file, 'notes.txt') + local text = async.timeout(5000, task) + show_buffer(text) + end) +< + + Parameters: ~ + • {duration} (`integer`) Timeout duration in milliseconds + • {task} (`vim.async.Task`) + + Return: ~ + (`R...`) + +wrap({argc}, {func}) *vim.async.wrap()* + Create an async function from a callback-style function. + + The callback is inserted at argument position `argc`. If `func` returns a + closable handle, it is closed when the awaiting task is cancelled. + + This is a reusable wrapper around |vim.async.await()|. Use it when the + same callback API is awaited from more than one place. >lua + local async = vim.async + local fs_stat = async.wrap(2, vim.uv.fs_stat) + + async.run(function() + local err, stat = fs_stat(vim.api.nvim_buf_get_name(0)) + if not err and stat then + print(stat.size) + end + end) +< + + Parameters: ~ + • {argc} (`integer`) + • {func} (`fun(...: T..., callback: fun(...: R...)): vim.async.Closable?`) + + Return: ~ + (`async fun(...: T...): R...`) + + + vim:tw=78:ts=8:sw=4:sts=4:et:ft=help:norl: diff --git a/runtime/doc/news.txt b/runtime/doc/news.txt index 3f0379a736..0bdcdec8ad 100644 --- a/runtime/doc/news.txt +++ b/runtime/doc/news.txt @@ -373,6 +373,9 @@ LSP LUA +• |lua-async|: `vim.async` provides structured concurrency primitives for Lua + code, including task handles, await helpers, cooperative cancellation, + completion-order iteration, and semaphores. • |vim.ui.img| can display images. Use `:checkhealth vim.health` to confirm your terminal supports it. • |vim.net.request()| can specify custom headers by passing `opts.headers`. diff --git a/runtime/lua/vim/_core/editor.lua b/runtime/lua/vim/_core/editor.lua index 66cc57f3d2..3fde7b4ef3 100644 --- a/runtime/lua/vim/_core/editor.lua +++ b/runtime/lua/vim/_core/editor.lua @@ -21,6 +21,7 @@ for k, v in pairs({ secure = true, snippet = true, pack = true, + async = true, _watch = true, net = true, pos = true, diff --git a/runtime/lua/vim/_core/util.lua b/runtime/lua/vim/_core/util.lua index 02e5e6af04..08cbf17f04 100644 --- a/runtime/lua/vim/_core/util.lua +++ b/runtime/lua/vim/_core/util.lua @@ -2,6 +2,27 @@ local M = {} +-- Generated from async.nvim/lua/async/_errors.lua: start +local nil_error = 'error(nil)' + +--- Normalize a failed Lua operation for error slots where `nil` means success. +--- @param err any +--- @return any +--- @private +function M._normalize_error(err) + return err == nil and nil_error or err +end + +--- Convert an error to a string without letting its metamethod interrupt cleanup. +--- @param err any +--- @return string +--- @private +function M._stringify_error(err) + local ok, message = pcall(tostring, err) + return ok and message or '' +end +-- Generated from async.nvim/lua/async/_errors.lua: end + --- Adds one or more blank lines above or below the cursor. --- @param above? boolean Place blank line(s) above the cursor local function add_blank(above) diff --git a/runtime/lua/vim/_meta.lua b/runtime/lua/vim/_meta.lua index 18dbe49c85..e285cd5a12 100644 --- a/runtime/lua/vim/_meta.lua +++ b/runtime/lua/vim/_meta.lua @@ -66,6 +66,8 @@ vim.uv = ... vim.F = require('vim.F') vim._watch = require('vim._watch') +---@diagnostic disable-next-line: no-unknown +vim.async = require('vim.async') vim.diagnostic = require('vim.diagnostic') vim.filetype = require('vim.filetype') vim.fs = require('vim.fs') diff --git a/runtime/lua/vim/async.lua b/runtime/lua/vim/async.lua new file mode 100644 index 0000000000..8483b45add --- /dev/null +++ b/runtime/lua/vim/async.lua @@ -0,0 +1,317 @@ +local core = require('vim.async._core') +local validate = vim.validate +local new_queue = require('vim.async._queue') +local runtime = require('vim.async._runtime') +local F = vim.F + +--- Structured async API for Lua code that waits on event-loop work. +--- +--- `vim.async` lets Lua code wait for timers, callbacks, and other tasks +--- without blocking Nvim's event loop. Async work runs inside tasks, which can +--- pause at checkpoints and manage child tasks created while they are running. +--- +--- Start async work with [vim.async.run()]. Inside a task, use +--- [vim.async.await()] to wait for callback-style APIs or other tasks without +--- blocking the event loop. Use [vim.async.pawait()] when an awaited operation +--- can fail and the current task should continue. +--- +--- Examples in this help use `local async = vim.async` for brevity. +--- +--- Example: run async work without blocking Nvim: +--- +--- ```lua +--- local async = vim.async +--- +--- async.run(function() +--- vim.notify('waiting...') +--- async.sleep(1000) +--- vim.notify('done') +--- end) +--- ``` +--- +--- Example: await a callback-style API. Callback results are returned +--- unchanged, so an error-first callback still returns `err, value`: +--- +--- ```lua +--- local async = vim.async +--- +--- async.run(function() +--- local err, stat = async.await(2, fs_stat, 'notes.txt') +--- if err then +--- error(err, 0) +--- end +--- print(('notes.txt is %d bytes'):format(stat.size)) +--- end) +--- ``` +--- +--- A task has two roles: +--- +--- - it is a handle that can be awaited, waited for, or closed +--- - it is a scope for child tasks created while the task is running +--- +--- [vim.async.run()] creates a task. A top-level task starts immediately. A +--- task created while another task is running becomes a child of that task, and +--- its function starts when the parent reaches its next checkpoint. A parent +--- task finishes only after its attached children finish. If a child fails +--- without being handled, the parent fails and closes the remaining children. +--- +--- Use [Task:detach()] for background work that should keep running after the +--- current task finishes. A detached task becomes top-level work; the original +--- parent no longer waits for it or closes it. +--- +--- Awaiting a task observes that task's result; it does not attach the task to +--- the awaiter or change which task owns it. Ownership is decided when the task +--- is created. +--- +--- Scheduling is cooperative. When a task awaits a timer, I/O operation, +--- callback, or another task, `vim.async` saves the Lua stack and returns +--- control to the event loop. Other callbacks can run while the task is paused. +--- Nothing interrupts synchronous Lua code in the middle of a stack frame. +--- +--- Checkpoints are the places where a task can pause, start pending child +--- tasks, observe cancellation, and receive unhandled child failures. Inside a +--- task, these operations are checkpoints: +--- +--- - `vim.async.await(...)` +--- - `vim.async.pawait(...)` +--- - `vim.async.checkpoint()` +--- - successful return from the task function, which is the final checkpoint +--- for child management +--- +--- Convenience APIs such as [vim.async.sleep()] and [vim.async.timeout()] can +--- also checkpoint because they call checkpointing APIs internally. +--- +--- Closing a task closes its attached children. Cancellation is cooperative: +--- [Task:close()] marks a task as closing, and the task observes that state at +--- a checkpoint. If a task is suspended on a closable operation such as a timer +--- or child task, `vim.async` closes that operation before reporting the +--- cancellation. +--- +--- Use [vim.async.await()] inside a task to suspend until work completes. It +--- accepts a task, a callback-taking function, or an argument position plus a +--- callback-taking function. `await(task)` returns the task result or raises +--- the task failure. [vim.async.pawait()] is the async counterpart to `pcall()` +--- for recoverable awaited-operation failures; it returns `ok, ...` instead of +--- failing the current task for that awaited operation. It does not suppress +--- cancellation or a failure already pending on the current task. +--- +--- From synchronous code, use [Task:wait()] or [Task:pwait()] to pump the event +--- loop until the task completes. Use [Task:on_complete()] to observe +--- completion without blocking. +--- +--- Coordination helpers work with task handles. [vim.async.iter()] yields +--- completed task handles in completion order, [vim.async.timeout()] awaits a +--- task with a deadline, and `vim.async.semaphore(permits)` creates a +--- [vim.async.Semaphore] that limits how many tasks can hold a permit for a +--- section at once. +--- @class vim.async: vim.async._core +local M = setmetatable({}, { __index = core }) + +M.semaphore = require('vim.async._semaphore') + +--- @param unsubscribe fun()[] +local function unsubscribe_all(unsubscribe) + for _, unsub in ipairs(unsubscribe) do + unsub() + end +end + +--- Create an async function from a callback-style function. +--- +--- The callback is inserted at argument position `argc`. If `func` returns a +--- closable handle, it is closed when the awaiting task is cancelled. +--- +--- This is a reusable wrapper around [vim.async.await()]. Use it when the same +--- callback API is awaited from more than one place. +--- +--- ```lua +--- local async = vim.async +--- local fs_stat = async.wrap(2, vim.uv.fs_stat) +--- +--- async.run(function() +--- local err, stat = fs_stat(vim.api.nvim_buf_get_name(0)) +--- if not err and stat then +--- print(stat.size) +--- end +--- end) +--- ``` +--- +--- @generic T, R +--- @param argc integer +--- @param func fun(...: T..., callback: fun(...: R...)): vim.async.Closable? +--- @return async fun(...: T...): R... +function M.wrap(argc, func) + validate('argc', argc, 'number') + validate('func', func, 'callable') + --- @async + return function(...) + return M.await(argc, func, ...) + end +end + +--- Iterate completed tasks in completion order. +--- +--- The iterator yields task handles, not task results. Use +--- [vim.async.await()] or [vim.async.pawait()] to retrieve each result. The +--- tasks are observed in the order they complete, regardless of the order in +--- the input list. +--- +--- ```lua +--- local async = vim.async +--- +--- async.run(function() +--- local tasks = { +--- async.run(function() return 'cache', read_cache() end):detach(), +--- async.run(function() return 'disk', read_file() end):detach(), +--- } +--- +--- for task in async.iter(tasks) do +--- local ok, source, text = async.pawait(task) +--- if ok then +--- for _, other in ipairs(tasks) do +--- if other ~= task then +--- other:close() +--- end +--- end +--- print(('loaded from %s'):format(source)) +--- return text +--- end +--- end +--- end) +--- ``` +--- +--- If code must support PUC Lua 5.1, use the direct-call form instead of a +--- generic `for` loop. The iterator may need to suspend while waiting for the +--- next completed task, and PUC Lua 5.1 cannot yield from a generic-for +--- iterator call. +--- +--- ```lua +--- local next_task = async.iter(tasks) +--- while true do +--- local task = next_task() +--- if task == nil then +--- break +--- end +--- async.await(task) +--- end +--- ``` +--- @async +--- @generic R +--- @param tasks vim.async.Task[] A list of tasks to wait for and iterate over. +--- @return async fun(): vim.async.Task? iterator that yields each completed task. +function M.iter(tasks) + validate('tasks', tasks, 'table') + + local remaining = #tasks + local queue = new_queue() + local unsubscribe = {} --- @type fun()[] + + if remaining == 0 then + queue:put_nowait() + else + for _, task in ipairs(tasks) do + unsubscribe[#unsubscribe + 1] = task:on_complete(function() + remaining = remaining - 1 + queue:put_nowait(task) + if remaining == 0 then + queue:put_nowait() + end + end) + end + end + + local proxy = newproxy(true) + getmetatable(proxy).__gc = function() + unsubscribe_all(unsubscribe) + end + + --- @async + return function() + local _ = proxy -- Keep the GC proxy alive with the iterator. + return queue:get() + end +end + +--- Asynchronously sleep for a given duration. +--- +--- Suspends the current task for the given duration without blocking the event +--- loop. After the delay and timer cleanup complete, `sleep()` returns to its +--- caller through the runtime's `schedule` hook. +--- +--- ```lua +--- vim.async.run(function() +--- vim.async.sleep(100) +--- vim.notify('resumed later') +--- end) +--- ``` +--- @async +--- @param duration integer ms +function M.sleep(duration) + validate('duration', duration, 'number') + M.await(function(callback) + local timer = runtime.new_timer() + timer:start(duration, 0, callback) + return timer + end) + -- Timer cleanup resumes this function directly from its close callback. + -- Yield once more so M.sleep() returns through the runtime scheduler. + M.await(runtime.schedule) +end + +--- Await a task with a timeout. +--- +--- If the task completes first, returns or raises the task result as +--- [vim.async.await()] would. If the deadline wins, closes the task and raises +--- `"timeout"` after the target task finishes cancellation cleanup. +--- +--- ```lua +--- local async = vim.async +--- +--- async.run(function() +--- local task = async.run(read_file, 'notes.txt') +--- local text = async.timeout(5000, task) +--- show_buffer(text) +--- end) +--- ``` +--- @async +--- @generic R +--- @param duration integer Timeout duration in milliseconds +--- @param task vim.async.Task +--- @return R... +function M.timeout(duration, task) + validate('duration', duration, 'number') + validate('task', task, 'table') + + local timed_out = false + local timer = M.run('__timeout', function() + M.sleep(duration) + timed_out = true + task:close() + end) + --- @diagnostic disable-next-line: invisible + timer._hidden = true + + local result = F.pack_len(M.pawait(task)) + timer:close() + M.pawait(timer) + + if timed_out then + error('timeout', 0) + end + + if not result[1] then + error(result[2], 0) + end + + return unpack(result, 2, result.n) +end + +if type(vim) == 'table' then + runtime.config({ + wait = vim.wait, + schedule = vim.schedule, + new_timer = vim.uv.new_timer, + }) +end + +return M diff --git a/runtime/lua/vim/async/_core.lua b/runtime/lua/vim/async/_core.lua new file mode 100644 index 0000000000..25837598a4 --- /dev/null +++ b/runtime/lua/vim/async/_core.lua @@ -0,0 +1,1128 @@ +-- LuaLS cannot model the generic annotations used by this vendored implementation. +---@diagnostic disable: no-unknown, undefined-doc-name, luadoc-miss-symbol, missing-return, missing-return-value, param-type-mismatch, return-type-mismatch, redundant-return-value, undefined-field, need-check-nil, await-in-sync + +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?> +local threads = setmetatable({}, { __mode = 'k' }) + +--- Returns the currently running task. +--- @return vim.async.Task? +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 +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: vim.async.Closable +--- @field package _thread thread +--- @field package _future vim.async.Future +--- @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 +--- @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?> +--- +--- 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 | 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 +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 + 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 + 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 + 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 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 + --- @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 + --- @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 + --- @param yielded vim.async.Task|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 + --- @param yielded vim.async.Task|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|vim.async.Closable? + + 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 + --- @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 + 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 +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 +--- @overload fun(name: string, func: async fun(...: T...), ...: T...): vim.async.Task +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 +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|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... +--- @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): 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 +--- @param prefix? string +--- @return string[] +local function inspect(parent, prefix) + local tasks = {} --- @type table?> + 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 diff --git a/runtime/lua/vim/async/_event.lua b/runtime/lua/vim/async/_event.lua new file mode 100644 index 0000000000..376def201b --- /dev/null +++ b/runtime/lua/vim/async/_event.lua @@ -0,0 +1,125 @@ +-- LuaLS cannot model the generic annotations used by this vendored implementation. +---@diagnostic disable: no-unknown, undefined-doc-name, luadoc-miss-symbol, missing-return, missing-return-value, param-type-mismatch, return-type-mismatch, redundant-return-value, undefined-field, need-check-nil, await-in-sync + +local async = require('vim.async._core') +local runtime = require('vim.async._runtime') + +--- An event can be used to notify multiple tasks that some event has +--- happened. An Event object manages an internal flag that can be set to true +--- with the `set()` method and reset to `false` with the `clear()` method. +--- The `wait()` method blocks until the flag is set to `true`. The flag is +--- set to `false` initially. +--- @class vim.async.Event +--- @field private _is_set boolean +--- @field private _waiters (function|false)[] +local Event = {} +Event.__index = Event + +--- @param waiters (function|false)[] +--- @return boolean +local function has_waiters(waiters) + for _, waiter in ipairs(waiters) do + if waiter then + return true + end + end + return false +end + +--- Set the event. +--- +--- All tasks waiting for event to be set will be awakened on a later event-loop +--- turn. +--- +--- If `max_woken` is provided, only up to `max_woken` waiters will be woken. +--- If waiters are woken this way, the event is reset because the signal is +--- reserved for those waiters. +--- @param max_woken? integer +function Event:set(max_woken) + if self._is_set then + return + end + + local limited = max_woken ~= nil + if not has_waiters(self._waiters) then + self._is_set = true + return + end + + self._is_set = true + if limited then + -- The signal is reserved for existing waiters and will be assigned on the + -- scheduled turn. New waiters must not consume it first. + self._is_set = false + end + + runtime.schedule(function() + local waiters = self._waiters + local waiters_to_notify = {} --- @type function[] + local limit = max_woken or math.huge + while #waiters > 0 and #waiters_to_notify < limit do + local waiter = table.remove(waiters, 1) + if waiter then + waiters_to_notify[#waiters_to_notify + 1] = waiter + end + end + + if limited and #waiters_to_notify == 0 and not has_waiters(waiters) then + self._is_set = true + end + + for _, waiter in ipairs(waiters_to_notify) do + waiter() + end + end) +end + +--- Wait until the event is set. +--- +--- If the event is set, return immediately. Otherwise block until another +--- task calls set(). +--- @async +function Event:wait() + async.await(function(callback) + if self._is_set then + callback() + else + table.insert(self._waiters, callback) + return { + close = function(_, on_close) + -- set() compacts the waiter list, so cancellation cannot rely on the + -- original insertion index still pointing at this callback. + for i, waiter in ipairs(self._waiters) do + if waiter == callback then + self._waiters[i] = false + break + end + end + if on_close then + on_close() + end + end, + } + end + end) +end + +--- Clear (unset) the event. +--- +--- Tasks awaiting on wait() will now block until the set() method is called +--- again. +function Event:clear() + self._is_set = false +end + +--- Create a new event. +--- +--- An event can signal to multiple listeners to resume execution. +--- The event can be set from a non-async context. +--- @return vim.async.Event +return function() + return setmetatable({ + _waiters = {}, + _is_set = false, + }, Event) +end diff --git a/runtime/lua/vim/async/_future.lua b/runtime/lua/vim/async/_future.lua new file mode 100644 index 0000000000..46b036b88c --- /dev/null +++ b/runtime/lua/vim/async/_future.lua @@ -0,0 +1,77 @@ +-- LuaLS cannot model the generic annotations used by this vendored implementation. +---@diagnostic disable: no-unknown, undefined-doc-name, luadoc-miss-symbol, missing-return, missing-return-value, param-type-mismatch, return-type-mismatch, redundant-return-value, undefined-field, need-check-nil, await-in-sync + +local F = vim.F +local util = require('vim._core.util') + +local Future = {} +Future.__index = Future + +function Future:completed() + return self._err ~= nil or self._result ~= nil +end + +function Future:result() + if not self:completed() then + error('Future has not completed', 2) + end + if self._err ~= nil then + return false, self._err + else + return true, F.unpack_len(self._result) + end +end + +function Future:on_complete(callback) + if self:completed() then + -- Already completed or closed + if self._err ~= nil then + callback(self._err) + else + callback(nil, F.unpack_len(self._result)) + end + return function() end + end + + local id = self._callback_pos + self._callback_pos = id + 1 + self._callbacks[id] = callback + + return function() + self._callbacks[id] = nil + end +end + +function Future:complete(err, ...) + if self:completed() then + error('Future is already completed', 2) + end + + if err ~= nil then + self._err = err + else + self._result = F.pack_len(...) + end + + local callbacks = self._callbacks + self._callbacks = {} + + local errs = {} -- Need to use pairs to avoid gaps caused by removed callbacks + for _, cb in pairs(callbacks) do + local ok, cb_err = pcall(cb, err, ...) + if not ok then + errs[#errs + 1] = util._stringify_error(util._normalize_error(cb_err)) + end + end + + if #errs > 0 then + error(table.concat(errs, '\n'), 0) + end +end + +return function() + return setmetatable({ + _callbacks = {}, + _callback_pos = 1, + }, Future) +end diff --git a/runtime/lua/vim/async/_queue.lua b/runtime/lua/vim/async/_queue.lua new file mode 100644 index 0000000000..5a2d9b4952 --- /dev/null +++ b/runtime/lua/vim/async/_queue.lua @@ -0,0 +1,107 @@ +-- LuaLS cannot model the generic annotations used by this vendored implementation. +---@diagnostic disable: no-unknown, undefined-doc-name, luadoc-miss-symbol, missing-return, missing-return-value, param-type-mismatch, return-type-mismatch, redundant-return-value, undefined-field, need-check-nil, await-in-sync + +local new_event = require('vim.async._event') + +--- An optionally bounded FIFO queue for passing values between async tasks. +--- +--- `put()` suspends the current task while the queue is full, and `get()` +--- suspends it while the queue is empty. The `put_nowait()` and `get_nowait()` +--- variants never suspend and raise an error when the operation cannot proceed. +--- @class vim.async.Queue +--- @field private _non_empty vim.async.Event +--- @field package _non_full vim.async.Event +--- @field private _max_size? integer +--- @field private _items R[] +--- @field private _right_i integer +--- @field private _left_i integer +local Queue = {} +Queue.__index = Queue + +--- Returns the number of items in the queue. +--- @return integer +function Queue:size() + return self._right_i - self._left_i +end + +--- Returns the maximum number of items in the queue. +--- @return integer? +function Queue:max_size() + return self._max_size +end + +--- Put an item into the queue. +--- +--- If the queue is full, wait until a free slot is available. +--- @async +--- @param value any +function Queue:put(value) + while self:size() == self:max_size() do + self._non_full:wait() + end + self:put_nowait(value) +end + +--- Get an item from the queue. +--- +--- If the queue is empty, wait until an item is available. +--- @async +--- @return any +function Queue:get() + while self:size() == 0 do + self._non_empty:wait() + end + return self:get_nowait() +end + +--- Get an item from the queue without blocking. +--- +--- If the queue is empty, raise an error. +--- @return any +function Queue:get_nowait() + if self:size() == 0 then + error('Queue is empty', 2) + end + -- TODO(lewis6991): For a long_running queue, _left_i might overflow. + self._left_i = self._left_i + 1 + local item = self._items[self._left_i] + self._items[self._left_i] = nil + if self._left_i == self._right_i then + self._non_empty:clear() + end + self._non_full:set(1) + return item +end + +--- Put an item into the queue without blocking. +--- If no free slot is immediately available, raise "Queue is full" error. +--- @param value any +function Queue:put_nowait(value) + if self:size() == self:max_size() then + error('Queue is full', 2) + end + self._right_i = self._right_i + 1 + self._items[self._right_i] = value + self._non_empty:set(1) + if self:size() == self:max_size() then + self._non_full:clear() + end +end + +--- Create a new FIFO queue with async support. +--- @param max_size? integer The maximum number of items in the queue, defaults to no limit +--- @return vim.async.Queue +return function(max_size) + local self = setmetatable({ + _items = {}, + _left_i = 0, + _right_i = 0, + _max_size = max_size, + _non_empty = new_event(), + _non_full = new_event(), + }, Queue) + + self._non_full:set() + + return self +end diff --git a/runtime/lua/vim/async/_runtime.lua b/runtime/lua/vim/async/_runtime.lua new file mode 100644 index 0000000000..d7a84f65af --- /dev/null +++ b/runtime/lua/vim/async/_runtime.lua @@ -0,0 +1,46 @@ +-- LuaLS cannot model the generic annotations used by this vendored implementation. +---@diagnostic disable: no-unknown, undefined-doc-name, luadoc-miss-symbol, missing-return, missing-return-value, param-type-mismatch, return-type-mismatch, redundant-return-value, undefined-field, need-check-nil, await-in-sync + +local validate = vim.validate + +--- @class vim.async.Timer: vim.async.Closable +--- @nodoc +--- @field start fun(self, timeout: integer, repeat_interval: integer, callback: fun()) + +--- @alias vim.async.TimerFactory fun(): vim.async.Timer +--- @nodoc + +--- @class vim.async.ConfigOpts +--- @nodoc +--- @field wait? fun(timeout: integer, predicate: fun(): boolean): boolean Run the event loop until the predicate succeeds or the timeout expires. +--- @field schedule? fun(callback: fun()) Queue a callback to run once on a later event-loop turn. +--- @field new_timer? vim.async.TimerFactory Create libuv-compatible timers for `sleep()` and `timeout()`. +--- @field debug? boolean Capture task creation metadata for debugging. + +--- @class vim.async.Runtime +--- @nodoc +--- @field wait fun(timeout: integer, predicate: fun(): boolean): boolean +--- @field schedule fun(callback: fun()) +--- @field new_timer vim.async.TimerFactory +--- @field debug boolean +local M = {} +M.debug = false + +--- @nodoc +--- @param opts vim.async.ConfigOpts +function M.config(opts) + validate('opts', opts, 'table') + validate('opts.wait', opts.wait, 'callable', true) + validate('opts.schedule', opts.schedule, 'callable', true) + validate('opts.new_timer', opts.new_timer, 'callable', true) + validate('opts.debug', opts.debug, 'boolean', true) + + M.wait = opts.wait or M.wait + M.schedule = opts.schedule or M.schedule + M.new_timer = opts.new_timer or M.new_timer + if opts.debug ~= nil then + M.debug = opts.debug + end +end + +return M diff --git a/runtime/lua/vim/async/_semaphore.lua b/runtime/lua/vim/async/_semaphore.lua new file mode 100644 index 0000000000..79091f42b9 --- /dev/null +++ b/runtime/lua/vim/async/_semaphore.lua @@ -0,0 +1,133 @@ +-- LuaLS cannot model the generic annotations used by this vendored implementation. +---@diagnostic disable: no-unknown, undefined-doc-name, luadoc-miss-symbol, missing-return, missing-return-value, param-type-mismatch, return-type-mismatch, redundant-return-value, undefined-field, need-check-nil, await-in-sync + +local F = vim.F +local new_event = require('vim.async._event') + +local pcall = pcall +do + local ok, coxpcall = pcall(require, 'coxpcall') + if ok and type(coxpcall) == 'table' and type(coxpcall.pcall) == 'function' then + pcall = coxpcall.pcall + end +end +local validate = vim.validate + +--- A semaphore manages an internal permit counter. [Semaphore:acquire()] +--- consumes one permit and [Semaphore:release()] returns one permit. If no +--- permits are available, `acquire()` suspends the current task until another +--- task releases one. +--- +--- The preferred way to use a Semaphore is with the `with()` method, which +--- automatically acquires and releases the semaphore around a function call. +--- This is useful for limiting sections that start external work and then +--- await it, such as file reads, requests, or subprocesses. +--- +--- ```lua +--- local async = vim.async +--- +--- async.run(function() +--- local limit = async.semaphore(4) +--- local tasks = {} +--- +--- for _, path in ipairs(paths) do +--- table.insert(tasks, async.run(function() +--- return limit:with(function() +--- return read_file(path) +--- end) +--- end)) +--- end +--- +--- local next_task = async.iter(tasks) +--- while true do +--- local task = next_task() +--- if task == nil then +--- break +--- end +--- async.await(task) +--- end +--- end) +--- ``` +--- @class vim.async.Semaphore +--- @field private _permits integer +--- @field private _max_permits integer +--- @field package _event vim.async.Event +local Semaphore = {} +Semaphore.__index = Semaphore + +--- Executes a function while holding one semaphore permit. +--- +--- This acquires the semaphore before running the function and releases it +--- after the function completes, even if it errors or the current task is +--- closed. +--- @async +--- @generic R +--- @param fn async fun(): R... # Function to execute within the semaphore's context. +--- @return R... # Result(s) of the executed function. +function Semaphore:with(fn) + self:acquire() + -- This pcall is only a try/finally guard for release(); all errors are + -- immediately rethrown so it is not an async recovery boundary. + local r = F.pack_len(pcall(fn)) + self:release() + local stat = r[1] + if not stat then + local err = r[2] + error(err, 0) + end + return unpack(r, 2, r.n) +end + +--- Acquire a semaphore permit. +--- +--- If the internal counter is greater than zero, decrement it by `1` and +--- return immediately. If it is `0`, wait until [Semaphore:release()] is +--- called. +--- @async +function Semaphore:acquire() + self._event:wait() + self._permits = self._permits - 1 + assert(self._permits >= 0, 'Semaphore value is negative') + if self._permits == 0 then + self._event:clear() + end +end + +--- Release a semaphore permit. +--- +--- Increments the internal counter by `1` and can wake a task waiting in +--- [Semaphore:acquire()]. +--- +--- Calling this more times than permits were acquired raises an error. +function Semaphore:release() + if self._permits >= self._max_permits then + error('Semaphore value is greater than max permits', 2) + end + self._permits = self._permits + 1 + self._event:set(1) +end + +--- Create an async semaphore that allows up to a given number of acquisitions. +--- +--- Prefer [Semaphore:with()] for most uses so permits are released reliably. +--- Use [Semaphore:acquire()] and [Semaphore:release()] directly only when the +--- acquire and release points cannot be expressed as one function call. +--- @param permits? integer (default: 1) +--- @return vim.async.Semaphore +local function new_semaphore(permits) + validate('permits', permits, 'number', true) + permits = permits or 1 + if permits < 1 or permits % 1 ~= 0 then + error('permits: expected positive integer', 2) + end + + local obj = setmetatable({ + _max_permits = permits, + _permits = permits, + _event = new_event(), + }, Semaphore) + obj._event:set() + return obj +end + +return new_semaphore diff --git a/src/gen/gen_vimdoc.lua b/src/gen/gen_vimdoc.lua index 4d92c94be8..a2a5043851 100755 --- a/src/gen/gen_vimdoc.lua +++ b/src/gen/gen_vimdoc.lua @@ -61,6 +61,9 @@ local INDENTATION = 4 --- @field fn_helptag_fmt? fun(fun: nvim.gen_vimdoc.HelptagTarget): string --- --- @field append_only? string[] +--- +--- Merge parsed files into the first section instead of rendering one section per file. +--- @field merge_files? boolean ---@alias nvim.gen_vimdoc.HelptagTarget ---| nvim.luacats.parser.fun @@ -359,6 +362,35 @@ local config = { return fmt('lsp-%s', name:lower()) end, }, + async = { + filename = 'lua-async.txt', + section_order = { + 'async.lua', + }, + merge_files = true, + files = { + 'runtime/lua/vim/async.lua', + 'runtime/lua/vim/async/_core.lua', + 'runtime/lua/vim/async/_semaphore.lua', + }, + section_fmt = function() + return 'Lua module: vim.async' + end, + helptag_fmt = function() + return 'lua-async' + end, + fn_xform = function(fun) + if fun.module == 'vim.async._core' or fun.module == 'vim.async._semaphore' then + fun.module = 'vim.async' + end + if fun.name == 'new_semaphore' then + fun.name = 'semaphore' + end + if fun.classvar == 'M' then + fun.classvar = nil + end + end, + }, diagnostic = { filename = 'diagnostic.txt', section_order = { @@ -1248,6 +1280,10 @@ local function gen_target(cfg) end end + local merged_classes = {} --- @type table + local merged_funs = {} --- @type nvim.luacats.parser.fun[] + local merged_briefs = {} --- @type string[] + for f, r in vim.spairs(file_results) do local classes, funs, briefs = r[1], r[2], r[3] @@ -1267,14 +1303,31 @@ local function gen_target(cfg) print(' Processing file:', f) - -- FIXME: Using f_base will confuse `_meta/protocol.lua` with `protocol.lua` - local f_base = vim.fs.basename(f) - sections[f_base] = make_section( - f_base, + if cfg.merge_files then + merged_classes = vim.tbl_extend('error', merged_classes, classes) + vim.list_extend(merged_funs, funs) + vim.list_extend(merged_briefs, briefs) + else + -- FIXME: Using f_base will confuse `_meta/protocol.lua` with `protocol.lua` + local f_base = vim.fs.basename(f) + sections[f_base] = make_section( + f_base, + cfg, + render_briefs(briefs, cfg), + render_funs(funs, all_classes, cfg), + render_classes(classes, funs, cfg) + ) + end + end + + if cfg.merge_files then + local section_file = cfg.section_order[1] + sections[section_file] = make_section( + section_file, cfg, - render_briefs(briefs, cfg), - render_funs(funs, all_classes, cfg), - render_classes(classes, funs, cfg) + render_briefs(merged_briefs, cfg), + render_funs(merged_funs, all_classes, cfg), + render_classes(merged_classes, merged_funs, cfg) ) end diff --git a/src/gen/util.lua b/src/gen/util.lua index 9791c4f19b..c1d94fa715 100644 --- a/src/gen/util.lua +++ b/src/gen/util.lua @@ -236,6 +236,8 @@ local function render_md(node, start_indent, indent, text_width, level, is_list) if ntype == 'text' then parts[#parts + 1] = node.text + elseif ntype == 'atx_heading' then + parts[#parts + 1] = ('*%s*'):format(node.heading_content.text) elseif ntype == 'html_tag' then error('html_tag: ' .. node.text) elseif ntype == 'inline_link' then diff --git a/src/nvim/CMakeLists.txt b/src/nvim/CMakeLists.txt index 70b314fb25..d4bf2f0487 100644 --- a/src/nvim/CMakeLists.txt +++ b/src/nvim/CMakeLists.txt @@ -998,6 +998,7 @@ file(GLOB API_SOURCES CONFIGURE_DEPENDS ${PROJECT_SOURCE_DIR}/src/nvim/api/*.c) file(GLOB LUA_SOURCES CONFIGURE_DEPENDS ${NVIM_RUNTIME_DIR}/lua/vim/*.lua + ${NVIM_RUNTIME_DIR}/lua/vim/async/*.lua ${NVIM_RUNTIME_DIR}/lua/vim/_meta/*.lua ${NVIM_RUNTIME_DIR}/lua/vim/filetype/*.lua ${NVIM_RUNTIME_DIR}/lua/vim/lsp/*.lua @@ -1014,6 +1015,7 @@ add_target(doc-vim ${NVIM_RUNTIME_DIR}/doc/api.txt ${NVIM_RUNTIME_DIR}/doc/diagnostic.txt ${NVIM_RUNTIME_DIR}/doc/lsp.txt + ${NVIM_RUNTIME_DIR}/doc/lua-async.txt ${NVIM_RUNTIME_DIR}/doc/lua.txt ${NVIM_RUNTIME_DIR}/doc/treesitter.txt ) diff --git a/test/functional/lua/async_spec.lua b/test/functional/lua/async_spec.lua new file mode 100644 index 0000000000..146f467fe5 --- /dev/null +++ b/test/functional/lua/async_spec.lua @@ -0,0 +1,2687 @@ +local t = require('test.testutil') +local n = require('test.functional.testnvim')() +local describe, it, before_each, after_each = t.describe, t.it, t.before_each, t.after_each +local exec_lua = n.exec_lua + +-- TODO: test error message has correct stack trace when: +-- task finishes with no continuation +-- task finishes with synchronous wait +-- nil in results + +-- TODO(lewis6991): test for cyclic await +-- - child awaiting an ancestor (not allowed) +-- - cyclic chain with detached tasks + +--- @param s string +--- @param f fun(...) +--- @param ... any +local function it_exec(s, f, ...) + local args = { ... } + it(s, function() + exec_lua(f, unpack(args)) + end) +end + +describe('async', function() + before_each(function() + n.clear() + exec_lua('package.path = ...', package.path) + + exec_lua(function() + _G.Async = require('vim.async') + _G.AsyncRuntime = require('vim.async._runtime') + local safe_pcall = pcall + local ok, coxpcall = pcall(require, 'coxpcall') + if ok and type(coxpcall) == 'table' and type(coxpcall.pcall) == 'function' then + safe_pcall = coxpcall.pcall + end + _G.pcall = safe_pcall + _G.await = Async.await + _G.run = Async.run + _G.wrap = Async.wrap + _G.uv_handles = setmetatable({}, { __mode = 'v' }) + + --- Keep track of uv handles so we can ensure they are closed + --- @generic T + --- @param name string + --- @param handle T? + --- @return T - ? + function _G.add_handle(name, handle) + uv_handles[name] = assert(handle) + return handle + end + + --- Check task eventually completes with an error + --- @param task vim.async.Task + --- @param pat string + --- @return string + function _G.check_task_err(task, pat) + local ok, err = task:pwait(100) + if ok then + error('Expected task to error, but it completed successfully', 2) + elseif not (err:match('^' .. pat .. '$') or (pat == 'closed' and is_closed_error(err))) then + error('Unexpected error: ' .. tostring(task:traceback(err)), 2) + end + return err + end + + --- @param s string + --- @return { [1]: string, pattern: boolean } + function _G.p(s) + return { s, pattern = true } + end + + --- @param err any + --- @return boolean + function _G.is_closed_error(err) + return err == 'closed' + or (type(err) == 'string' and err:match('^closed\nstack traceback:') ~= nil) + end + + --- @param err any + --- @return boolean + function _G.is_timeout_error(err) + return err == 'timeout' + or (type(err) == 'string' and err:match('^timeout\nstack traceback:') ~= nil) + end + + function _G.is_jit() + return package.loaded.jit ~= nil + end + + --- @param expected any + --- @param actual any + --- @param msg? string + function _G.eq(expected, actual, msg) + local match + if + type(expected) == 'table' + and type(expected[1]) == 'string' + and expected.pattern == true + then + match = actual:match(expected[1]) ~= nil + expected = expected[1] + else + match = vim.deep_equal(expected, actual) + end + + if not match then + if type(actual) == 'string' then + actual = '\n│ ' .. actual:gsub('\n', '\n│ ') + else + actual = vim.inspect(actual) + end + if type(expected) == 'string' then + expected = '\n│ ' .. expected:gsub('\n', '\n│ ') + else + expected = vim.inspect(expected) + end + error( + ('%s\n\nactual: %s\n\nexpected: %s'):format(msg or 'Mismatch:', actual, expected), + 2 + ) + end + end + + --- @async~ + function _G.eternity() + await(function(_cb) + -- Never call callback + return add_handle('timer', vim.uv.new_timer()) --[[@as vim.async.Closable]] + end) + end + end) + end) + + after_each(function() + exec_lua(function() + for k, v in pairs(uv_handles) do + assert(v:is_closing(), ('uv handle %s is not closing'):format(k)) + end + collectgarbage('collect') + assert(not next(uv_handles), 'Resources not collected') + end) + end) + + describe('basic operations', function() + it_exec('can error stack trace on sync wait', function() + local task = run(function() + error('SYNC ERR') + end) + check_task_err(task, '.*async_spec.lua:%d+: SYNC ERR') + end) + + it_exec('can await a uv callback function', function() + --- @param path string + --- @param options uv.spawn.options + --- @param on_exit fun(code: integer, signal: integer) + --- @return uv.uv_process_t handle + local function spawn(path, options, on_exit) + return add_handle('process', vim.uv.spawn(path, options, on_exit)) + end + + local done = run(function() + local code1 = await(3, spawn, 'echo', { args = { 'foo' } }) + assert(code1 == 0) + + local code2 = await(3, spawn, 'echo', { args = { 'bar' } }) + assert(code2 == 0) + await(vim.schedule) + + return true + end):wait(1000) + + eq(true, done) + end) + + it_exec('resumes sleep outside a fast event', function() + local in_fast_event = run(function() + Async.sleep(0) + return vim.in_fast_event() + end):wait(100) + + eq(false, in_fast_event) + end) + + it_exec('can await a run task', function() + local a = run(function() + return await(run(function() + await(vim.schedule) + return 'JJ' + end)) + end):wait(10) + + assert(a == 'JJ', 'GOT ' .. tostring(a)) + end) + + it_exec('can wait on an empty task', function() + local did_cb = false + local a = 1 + + local task = run(function() + -- task does not await anything, should complete immediately + a = a + 1 + end) + + task:on_complete(function() + did_cb = true + end) -- non-blocking + + task:wait(100) -- blocking + + assert(a == 2) + assert(did_cb) + end) + + it_exec('on_complete observes a pending child task without starting it', function() + local results = {} + + run(function() + local child = run(function() + results[#results + 1] = 'child_started' + return 'child_done' + end) + + child:on_complete(function(err, value) + assert(not err, tostring(err)) + results[#results + 1] = value + end) + results[#results + 1] = 'after_on_complete' + end):wait(100) + + eq({ + 'after_on_complete', + 'child_started', + 'child_done', + }, results) + end) + + it_exec('child tasks start when the parent reaches a checkpoint', function() + local results = {} + + run(function() + run(function() + results[#results + 1] = 'child_started' + end) + + results[#results + 1] = 'before_checkpoint' + await(vim.schedule) + results[#results + 1] = 'after_checkpoint' + end):wait(100) + + eq({ + 'before_checkpoint', + 'child_started', + 'after_checkpoint', + }, results) + end) + + it_exec('child tasks start at an explicit checkpoint', function() + local results = {} + + run(function() + run(function() + results[#results + 1] = 'child_started' + end) + + results[#results + 1] = 'before_checkpoint' + Async.checkpoint() + results[#results + 1] = 'after_checkpoint' + end):wait(100) + + eq({ + 'before_checkpoint', + 'child_started', + 'after_checkpoint', + }, results) + end) + + it_exec('handles tasks that complete', function() + local task = run(function() + -- should wait for 1 ms + await(function(callback) + local timer = add_handle('timer', vim.uv.new_timer()) + timer:start(1, 0, callback) + return timer --[[@as vim.async.Closable]] + end) + await(vim.schedule) + return nil, 1 + end) + + local r1, r2 = task:wait(10) + eq(r1, nil) + eq(r2, 1) + end) + + it_exec('can provide a traceback for nested tasks', function() + if not is_jit() then + return + end + + --- @async + local function t1() + await(run(function() + error('GOT HERE') + end)) + end + + local task = run(function() + await(run(function() + await(run(function() + await(run(function() + t1() + end)) + end)) + end)) + end) + + local err = check_task_err(task, '.*async_spec.lua:%d+: GOT HERE') + + local m = [[.*async_spec.lua:%d+: GOT HERE +stack traceback: + %[thread: 0x%x+%] %[C%]: in function 'error' + %[thread: 0x%x+%] .*async_spec.lua:%d+: in function <.*async_spec.lua:%d+> + %[thread: 0x%x+%] .*async_spec.lua:%d+: in function 't1' + %[thread: 0x%x+%] .*async_spec.lua:%d+: in function <.*async_spec.lua:%d+> + %[thread: 0x%x+%] .*async_spec.lua:%d+: in function <.*async_spec.lua:%d+> + %[thread: 0x%x+%] .*async_spec.lua:%d+: in function <.*async_spec.lua:%d+> + %[thread: 0x%x+%] .*async_spec.lua:%d+: in function <.*async_spec.lua:%d+>]] + + local tb = tostring(task:traceback(err) or ''):gsub('\t', ' ') + assert(tb:match(m), 'ERROR: ' .. tostring(tb)) + end) + + it_exec('does not keep completed awaited tasks in later tracebacks', function() + if not is_jit() then + return + end + + for _, await_child in ipairs({ + function() + await(run(function() + return 'done' + end)) + end, + function() + local ok = Async.pawait(run(function() + error('child error') + end)) + eq(false, ok) + end, + }) do + local task = run(function() + await_child() + error('parent error') + end) + + local err = check_task_err(task, '.*async_spec.lua:%d+: parent error') + local tb = tostring(task:traceback(err) or '') + + assert(tb:match("%[C%]: in function 'error'"), 'ERROR: ' .. tostring(tb)) + assert(not tb:match('child error'), 'ERROR: ' .. tostring(tb)) + assert(not tb:match('stack traceback:\nstack traceback:'), 'ERROR: ' .. tostring(tb)) + end + end) + + it_exec('does not print nil for tracebacks without a message', function() + if not is_jit() then + return + end + + local task = run(function() + await(function() end) + end) + + local tb = tostring(task:traceback() or '') + assert(not tb:match('^nil\n'), 'ERROR: ' .. tostring(tb)) + + task:close() + check_task_err(task, 'closed') + end) + + it_exec('does not need new stack frame for non-deferred continuations', function() + --- @async + local function deep(n) + if n == 0 then + return 'done' + end + await(function(cb) + cb() + end) + return deep(n - 1) + end + + local res = run(function() + return deep(10000) + end):wait() + assert(res == 'done') + end) + + it_exec('does not retain unused run arguments after task starts', function() + local unused = {} + local weak = setmetatable({ unused }, { __mode = 'v' }) + + local task = run(function(_) + Async.sleep(100) + end, 'used', unused) + + unused = nil + collectgarbage('collect') + collectgarbage('collect') + + local retained = weak[1] + task:close() + check_task_err(task, 'closed') + eq(nil, retained) + end) + end) + + describe('task cancellation and closing', function() + it_exec('can close tasks', function() + local task = run(eternity) + task:close() + check_task_err(task, 'closed') + end) + + it_exec('can close tasks which waiting on a wrapped callback function', function() + local wfn = wrap(1, function(_callback) + return add_handle('timer', vim.uv.new_timer()) --[[@as vim.async.Closable]] + end) + + local task = run(function() + wfn() + end) + + task:close() + check_task_err(task, 'closed') + end) + + it_exec('gracefully handles when closables are prematurely closed', function() + local result = run(function() + await(1, function(callback) + local timer = add_handle('timer', vim.uv.new_timer()) + timer:close(callback) + return timer --[[@as vim.async.Closable]] + end) + + return 'FINISH' + end):wait() + + eq('FINISH', result) + end) + + it_exec('callback function can be closed (nested)', function() + local child --- @type vim.async.Task + local task = run(function() + child = run(eternity) + await(child) + end) + + task:close() + + check_task_err(task, 'closed') + check_task_err(child, 'closed') + end) + + it_exec('can timeout tasks', function() + local task = run(eternity) + check_task_err(task, 'timeout') + task:close() + check_task_err(task, 'closed') + end) + + it_exec('can async timeout a test', function() + local task = run(eternity) + check_task_err(run(Async.timeout, 10, task), 'timeout') + end) + + it_exec('timeout waits for target cleanup before raising timeout', function() + local cleanup_done = false + + local task = run(function() + await(function() + return { + close = function(_, callback) + vim.schedule(function() + cleanup_done = true + callback() + end) + end, + } + end) + end) + + check_task_err(run(Async.timeout, 1, task), 'timeout') + eq(true, cleanup_done) + check_task_err(task, 'closed') + end) + + it_exec('timeout preserves target failure before the deadline', function() + local task = run(function() + Async.sleep(1) + error('TARGET_ERROR') + end) + + check_task_err(run(Async.timeout, 100, task), '.*async_spec.lua:%d+: TARGET_ERROR') + end) + + it_exec('returns when the task completes before the timeout', function() + local timeout_timer = { + closed = false, + close = function(self, callback) + self.closed = true + if callback then + callback() + end + end, + is_closing = function(self) + return self.closed + end, + start = function() end, + } + + AsyncRuntime.config({ + wait = vim.wait, + schedule = vim.schedule, + new_timer = function() + return timeout_timer + end, + }) + + local ok, err = pcall(function() + local task = run(function() + return 'FINISH' + end) + eq('FINISH', run(Async.timeout, 100, task):wait(10)) + assert(timeout_timer.closed) + end) + + AsyncRuntime.config({ + wait = vim.wait, + schedule = vim.schedule, + new_timer = vim.uv.new_timer, + }) + if not ok then + error(err, 0) + end + end) + + it_exec('closes detached child tasks', function() + local task1 = run(eternity) + task1:close() + + local task2 = run(function() + await(task1) + end) + + check_task_err(task2, 'closed') + end) + end) + + describe('error handling', function() + it_exec('handles tasks that error', function() + local task = run(function() + await(function(callback) + local timer = add_handle('timer', vim.uv.new_timer()) + timer:start(1, 0, callback) + return timer --[[@as vim.async.Closable]] + end) + await(vim.schedule) + error('GOT HERE') + end) + + check_task_err(task, '.*async_spec.lua:%d+: GOT HERE') + end) + + it_exec('can handle errors in wrapped functions', function() + local task = run(function() + await(function(_callback) + error('ERROR') + end) + end) + check_task_err(task, '.*async_spec.lua:%d+: ERROR') + end) + + it_exec('can pcall errors in wrapped functions', function() + local task = run(function() + return pcall(function() + await(function(_callback) + error('ERROR') + end) + end) + end) + local ok, msg = task:wait() + assert(not ok and msg, 'Expected error, got success') + assert(msg:match('^.*async_spec.lua:%d+: ERROR'), 'Got unexpected error: ' .. msg) + end) + + it_exec('handles when a floating child errors', function() + local parent = run(function() + local _child = run(function(...) + Async.sleep(5) + error('CHILD ERROR') + end) + end) + + check_task_err(parent, 'child error: .*async_spec.lua:%d+: CHILD ERROR') + end) + + it_exec('handles when a floating child errors and parent errors', function() + local parent = run(function() + local _child = run(function(...) + Async.sleep(5) + error('CHILD ERROR') + end) + error('PARENT ERROR') + end) + + check_task_err(parent, '.*async_spec.lua:%d+: PARENT ERROR') + end) + end) + + describe('task iteration', function() + it_exec('can iterate detached tasks', function() + local tasks = {} --- @type vim.async.Task[] + local expected = {} --- @type table[] + + for i = 1, 10 do + tasks[i] = run(function() + if i % 2 == 0 then + await(vim.schedule) + end + return 'FINISH', i + end) + expected[i] = { 'FINISH', i } + end + + local results = {} --- @type table[] + run(function() + local next_task = Async.iter(tasks) + while true do + local task = next_task() + if not task then + break + end + local r1, r2 = await(task) + results[r2] = { r1, r2 } + end + end):wait(1000) + + eq(expected, results) + end) + + it_exec('can inspect errors when iterating detached tasks', function() + local results = {} --- @type table[] + local tasks = {} --- @type vim.async.Task[] + local task_err --- @type any + + for i = 1, 10 do + tasks[i] = run(function() + await(vim.schedule) + if i == 3 then + error('ERROR IN TASK ' .. i) + end + return 'FINISH', i + end) + end + + run(function() + local next_task = Async.iter(tasks) + while true do + local task = next_task() + if not task then + break + end + local ok, r1, r2 = Async.pawait(task) + if not ok then + task_err = r1 + break + end + results[r2] = { r1, r2 } + end + end):wait(100) + + --- @cast task_err string + assert(task_err:match('.*async_spec.lua:%d+: ERROR IN TASK 3'), task_err) + + eq({ + { 'FINISH', 1 }, + { 'FINISH', 2 }, + }, results) + end) + + it_exec('iterates tasks in completion order', function() + --- @async + --- @param count integer + --- @param id integer + local function after_schedules(count, id) + for _ = 1, count do + await(vim.schedule) + end + return id + end + + local tasks = { + run(after_schedules, 3, 1), + run(after_schedules, 1, 2), + run(after_schedules, 2, 3), + } + + local order = {} + run(function() + local next_task = Async.iter(tasks) + while true do + local task = next_task() + if not task then + break + end + order[#order + 1] = await(task) + end + end):wait(100) + + eq({ 2, 3, 1 }, order) + end) + + it_exec('treats false task errors as errors when iterating', function() + local task = run(function() + await(vim.schedule) + error(false, 0) + end) + + run(function() + local completed = Async.iter({ task })() + local ok, err = Async.pawait(completed) + eq(false, ok) + eq(false, err) + end):wait(100) + end) + + it_exec('can iter tasks followed by error', function() + local task = run(function() + await(vim.schedule) + return 'FINISH', 1 + end) + + local expected = { { 'FINISH', 1 } } + local results = {} --- @type table[] + + local task2 = run(function() + local next_task = Async.iter({ task }) + while true do + local completed = next_task() + if not completed then + break + end + local r1, r2 = await(completed) + results[r2] = { r1, r2 } + end + error('GOT HERE') + end) + + check_task_err(task2, '.*async_spec.lua:%d+: GOT HERE') + eq(expected, results) + end) + + it_exec('can iter tasks with cancellation', function() + local tasks = {} --- @type vim.async.Task[] + + for i = 1, 4 do + tasks[i] = run(function() + if i == 2 then + eternity() + end + return 'FINISH', i + end) + end + + assert(tasks[2]):close() + + local results = {} --- @type table[] + local errs = {} --- @type any[] + run(function() + local next_task = Async.iter(tasks) + while true do + local task = next_task() + if not task then + break + end + local ok, r1, r2 = Async.pawait(task) + if ok then + results[r2] = { r1, r2 } + else + errs[#errs + 1] = r1 + end + end + end):wait(100) + + eq({ + [1] = { 'FINISH', 1 }, + [3] = { 'FINISH', 3 }, + [4] = { 'FINISH', 4 }, + }, results) + eq({ 'closed' }, errs) + end) + + it_exec('can iter tasks with garbage collection', function() + --- @param task vim.async.Task + --- @return integer + local function get_task_callback_count(task) + --- @diagnostic disable-next-line: invisible + return vim.tbl_count(task._future._callbacks) + end + + local task = run(eternity) + + run(function() + local itr = Async.iter({ task }) + eq(get_task_callback_count(task), 1, 'task should have one callback') + itr = nil + collectgarbage('collect') + eq(get_task_callback_count(task), 0, 'task should have no callbacks') + end):wait(100) + + task:close() + check_task_err(task, 'closed') + end) + + it_exec('handles empty task lists', function() + run(function() + eq(nil, Async.iter({})()) + end):wait(100) + end) + end) + + describe('child task management', function() + it_exec('does not close child tasks created outside of parent', function() + local t1 = run(Async.sleep, 10) + local t2 --- @type vim.async.Task + local t3 --- @type vim.async.Task + + local parent = run(function() + t2 = run(Async.sleep, 10) + t3 = run(Async.sleep, 10):detach() + await(t1) + end) + + parent:close() + + check_task_err(parent, 'closed') + t1:wait() + check_task_err(t2, 'closed') + t3:wait() + end) + + it_exec('stops observing external tasks when the waiter closes', function() + for _, wait in ipairs({ await, Async.pawait }) do + local external = run(eternity) + local waiter = run(function() + wait(external) + end) + + waiter:close() + check_task_err(waiter, 'closed') + + --- @diagnostic disable-next-line: invisible + local callback_count = vim.tbl_count(external._future._callbacks) + external:close() + check_task_err(external, 'closed') + + eq(0, callback_count) + end + end) + + it_exec('stops observing external tasks when a child fails', function() + for _, wait in ipairs({ await, Async.pawait }) do + local external = run(eternity) + local waiter = run(function() + run(function() + await(vim.schedule) + error('CHILD ERROR') + end) + wait(external) + end) + + check_task_err(waiter, 'child error: .*async_spec.lua:%d+: CHILD ERROR') + + --- @diagnostic disable-next-line: invisible + local callback_count = vim.tbl_count(external._future._callbacks) + external:close() + check_task_err(external, 'closed') + + eq(0, callback_count) + end + end) + + it_exec('stops observing children detached while being awaited', function() + for _, wait in ipairs({ await, Async.pawait }) do + local child --- @type vim.async.Task + local completions = 0 + local parent = run(function() + child = run(eternity) + wait(child) + end) + parent:on_complete(function() + completions = completions + 1 + end) + + child:detach() + parent:close() + check_task_err(parent, 'closed') + eq(false, child:completed()) + + --- @diagnostic disable-next-line: invisible + local callback_count = vim.tbl_count(child._future._callbacks) + child:close() + check_task_err(child, 'closed') + + eq(0, callback_count) + eq(1, completions) + end + end) + + it_exec('ignores extra callback awaitable results', function() + for _, wait in ipairs({ await, Async.pawait }) do + local external = run(eternity) + local waiter = run(function() + wait(function() + return external, function() + error('EXTRA_RETURN_CALLED') + end + end) + end) + + waiter:close() + check_task_err(waiter, 'closed') + external:close() + check_task_err(external, 'closed') + end + end) + + it_exec('detached pending child starts independently', function() + local results = {} + + run(function() + run(function() + results[#results + 1] = 'detached_started' + end):detach() + + results[#results + 1] = 'parent_done' + await(vim.schedule) + end):wait(100) + + eq({ + 'parent_done', + 'detached_started', + }, results) + end) + + it_exec('detached child failures do not fail the original parent', function() + local child --- @type vim.async.Task + + local parent = run(function() + child = run(function() + await(vim.schedule) + error('DETACHED_ERROR') + end):detach() + + await(vim.schedule) + return 'parent ok' + end) + + eq('parent ok', parent:wait(100)) + check_task_err(child, '.*async_spec.lua:%d+: DETACHED_ERROR') + end) + + it_exec('attaches tasks created from synchronous callbacks inside a task', function() + local release --- @type fun()? + local results = {} + + local parent = run(function() + local function call(callback) + callback() + end + + call(function() + run(function() + await(function(callback) + release = callback + end) + results[#results + 1] = 'child_done' + end) + end) + + results[#results + 1] = 'parent_body_done' + end) + + local ok, err = parent:pwait(10) + eq(false, ok) + assert(is_timeout_error(err), 'Expected timeout, got: ' .. tostring(err)) + eq({ 'parent_body_done' }, results) + assert(release, 'attached child was not started at parent finish') + + release() + parent:wait(50) + + eq({ 'parent_body_done', 'child_done' }, results) + end) + + it_exec('does not attach tasks created from event-loop callbacks', function() + local release --- @type fun()? + local child --- @type vim.async.Task + local results = {} + + local parent = run(function() + await(function(callback) + vim.schedule(function() + child = run(function() + await(function(child_callback) + release = child_callback + end) + results[#results + 1] = 'child_done' + end) + callback() + end) + end) + + results[#results + 1] = 'parent_done' + end) + + parent:wait(50) + + eq({ 'parent_done' }, results) + assert(child, 'event-loop callback did not create child task') + assert(release, 'top-level callback task was not started') + + release() + child:wait(50) + + eq({ 'parent_done', 'child_done' }, results) + end) + + it_exec('does not run pending children when parent errors before a checkpoint', function() + local child --- @type vim.async.Task + local child_ran = false + + local parent = run(function() + child = run(function() + child_ran = true + end) + + error('PARENT_ERROR') + end) + + check_task_err(parent, '.*async_spec.lua:%d+: PARENT_ERROR') + check_task_err(child, 'closed') + eq(false, child_ran) + end) + + it_exec('synchronous child wait starts only the waited child', function() + local results = {} + + run(function() + local child1 = run(function() + results[#results + 1] = 'child1' + end) + + run(function() + results[#results + 1] = 'child2' + end) + + child1:wait(100) + eq({ 'child1' }, results) + end):wait(100) + + eq({ 'child1', 'child2' }, results) + end) + + it_exec('does not wait for detached task children after sync wait times out', function() + local detached --- @type vim.async.Task + local release --- @type fun()? + local results = {} + + local parent = run(function() + await(vim.schedule) + + detached = run(function() + run(function() + await(function(callback) + release = callback + end) + results[#results + 1] = 'detached_child_done' + end) + end):detach() + + local ok, err = detached:pwait(10) + eq(false, ok) + assert(is_timeout_error(err), 'Expected timeout, got: ' .. tostring(err)) + + results[#results + 1] = 'parent_done' + end) + + parent:wait(50) + + eq({ 'parent_done' }, results) + assert(release, 'detached child was not started') + + release() + detached:wait(50) + + eq({ 'parent_done', 'detached_child_done' }, results) + end) + + it_exec('automatically awaits child tasks', function() + local child1, child2 --- @type vim.async.Task, vim.async.Task + local main = run(function() + child1 = run(Async.sleep, 10) + child2 = run(Async.sleep, 10) + end) + + main:wait() + assert(child1:completed()) + assert(child2:completed()) + end) + + it_exec('should not fail the parent task if children finish before parent', function() + local release_parent --- @type fun()? + local release_child1 --- @type fun()? + local release_child2 --- @type fun()? + local child1, child2 --- @type vim.async.Task, vim.async.Task + + local main = run(function() + child1 = run(function() + await(function(callback) + release_child1 = callback + end) + end) + child2 = run(function() + await(function(callback) + release_child2 = callback + end) + end) + + await(function(callback) + release_parent = callback + end) + end) + + assert(release_child1) + assert(release_child2) + assert(release_parent) + release_child1() + release_child2() + assert(child1:completed()) + assert(child2:completed()) + + release_parent() + main:wait() + end) + + it_exec('automatically closes suspended child tasks', function() + local forever_child --- @type vim.async.Task + + local main = run(function() + forever_child = run(function() + while true do + Async.sleep(1) + end + end) + Async.sleep(2) + end) + + eq(forever_child:status(), 'awaiting') + main:close() + check_task_err(main, 'closed') + check_task_err(forever_child, 'closed') + end) + + it_exec('child failure while parent is suspended closes siblings', function() + local sibling --- @type vim.async.Task + local continued = false + + local parent = run(function() + run(function() + Async.sleep(1) + error('CHILD_ERROR') + end) + + sibling = run(eternity) + Async.sleep(100) + continued = true + end) + + check_task_err(parent, 'child error: .*async_spec.lua:%d+: CHILD_ERROR') + check_task_err(sibling, 'closed') + eq(false, continued) + end) + + it_exec('should not close the parent task when child task is closed', function() + run(function() + run(eternity):close() + end):wait() + end) + end) + + describe('semaphore', function() + it_exec('rejects invalid permit counts', function() + for _, permits in ipairs({ 0, -1, 1.5, math.huge }) do + local ok, err = pcall(Async.semaphore, permits) + eq(false, ok) + --- @cast err string + assert( + err:match('permits: expected positive integer'), + 'Unexpected error: ' .. tostring(err) + ) + end + end) + + it_exec('runs', function() + local ret = {} + run(function() + local semaphore = Async.semaphore(3) + local tasks = {} --- @type vim.async.Task[] + for i = 1, 5 do + tasks[#tasks + 1] = run(function() + semaphore:with(function() + ret[#ret + 1] = 'start' .. i + await(vim.schedule) + ret[#ret + 1] = 'end' .. i + end) + end) + end + local next_task = Async.iter(tasks) + while true do + local task = next_task() + if not task then + break + end + await(task) + end + end):wait() + + eq({ + 'start1', + 'start2', + 'start3', + 'end1', + 'end2', + 'end3', + 'start4', + 'start5', + 'end4', + 'end5', + }, ret) + end) + + it_exec('ping pong', function() + local msgs = {} + local ball = { hits = 0 } + local max_hits = 10 + + --- @async + --- @param name string + --- @param sem vim.async.Semaphore + local function player(name, sem) + while ball.hits < max_hits do + local ok, err = pcall(sem.acquire, sem) + if not ok or ball.hits >= max_hits then + if not ok and not tostring(err):match('closed') then + error(err) + end + break + end + + ball.hits = ball.hits + 1 + msgs[#msgs + 1] = name + Async.sleep(2) + sem:release() + end + end + + run(function() + local sem = Async.semaphore(1) + local p1 = run(player, 'ping', sem) + local p2 = run(player, 'pong', sem) + local next_task = Async.iter({ p1, p2 }) + while true do + local task = next_task() + if not task then + break + end + await(task) + end + end):wait() + + eq({ 'ping', 'pong', 'ping', 'pong', 'ping', 'pong', 'ping', 'pong', 'ping', 'pong' }, msgs) + end) + + it_exec('does not lose a semaphore wake after closing a waiter', function() + local sem = Async.semaphore(1) + local second_acquired = false + + run(function() + sem:acquire() + + local first = run(function() + sem:acquire() + end) + + local second = run(function() + sem:acquire() + second_acquired = true + end) + + Async.checkpoint() + first:close() + Async.pawait(first) + + sem:release() + await(second) + end):wait(100) + + eq(true, second_acquired) + end) + + it_exec('releases semaphore permits when with errors', function() + run(function() + local sem = Async.semaphore(1) + + local ok, err = pcall(function() + sem:with(function() + error('WITH_ERROR') + end) + end) + + eq(false, ok) + --- @cast err string + assert(err:match('WITH_ERROR'), 'Expected WITH_ERROR, got: ' .. tostring(err)) + + sem:acquire() + sem:release() + end):wait(100) + end) + + it_exec('releases semaphore permits when with is cancelled', function() + local release --- @type fun()? + local sem = Async.semaphore(1) + + local task = run(function() + sem:with(function() + await(function(callback) + release = callback + end) + end) + end) + + run(function() + Async.checkpoint() + assert(release, 'semaphore body did not start') + + task:close() + Async.pawait(task) + + sem:acquire() + sem:release() + end):wait(100) + + check_task_err(task, 'closed') + end) + + it_exec('does not resume semaphore waiters inline on release', function() + local results = {} + + run(function() + local sem = Async.semaphore(1) + sem:acquire() + + run(function() + sem:acquire() + results[#results + 1] = 'waiter_acquired' + end) + + Async.checkpoint() + results[#results + 1] = 'before_release' + sem:release() + results[#results + 1] = 'after_release' + + eq({ 'before_release', 'after_release' }, results) + await(vim.schedule) + eq({ 'before_release', 'after_release', 'waiter_acquired' }, results) + end):wait(100) + end) + end) + + describe('queue', function() + it_exec('does not resume get waiters inline on put_nowait', function() + local new_queue = require('vim.async._queue') + local results = {} + + run(function() + local queue = new_queue() + + run(function() + local item = queue:get() + results[#results + 1] = 'got_' .. item + end) + + Async.checkpoint() + results[#results + 1] = 'before_put' + queue:put_nowait('item') + results[#results + 1] = 'after_put' + + eq({ 'before_put', 'after_put' }, results) + await(vim.schedule) + eq({ 'before_put', 'after_put', 'got_item' }, results) + end):wait(100) + end) + + it_exec('get waiters retry if a deferred item is consumed first', function() + local new_queue = require('vim.async._queue') + local results = {} + + run(function() + local queue = new_queue() + + run(function() + results[#results + 1] = queue:get() + end) + + Async.checkpoint() + queue:put_nowait('first') + eq('first', queue:get_nowait()) + + await(vim.schedule) + eq({}, results) + + queue:put_nowait('second') + await(vim.schedule) + eq({ 'second' }, results) + end):wait(100) + end) + + it_exec('does not resume put waiters inline on get_nowait', function() + local new_queue = require('vim.async._queue') + local results = {} + + run(function() + local queue = new_queue(1) + queue:put_nowait('first') + + run(function() + queue:put('second') + results[#results + 1] = 'put_second' + end) + + Async.checkpoint() + results[#results + 1] = 'before_get' + eq('first', queue:get_nowait()) + results[#results + 1] = 'after_get' + + eq({ 'before_get', 'after_get' }, results) + await(vim.schedule) + eq({ 'before_get', 'after_get', 'put_second' }, results) + end):wait(100) + end) + + it_exec('put waiters retry if a deferred slot is filled first', function() + local new_queue = require('vim.async._queue') + local results = {} + + run(function() + local queue = new_queue(1) + queue:put_nowait('first') + + run(function() + queue:put('second') + results[#results + 1] = 'put_second' + end) + + Async.checkpoint() + eq('first', queue:get_nowait()) + queue:put_nowait('interloper') + + await(vim.schedule) + eq({}, results) + + eq('interloper', queue:get_nowait()) + await(vim.schedule) + eq({ 'put_second' }, results) + eq('second', queue:get_nowait()) + end):wait(100) + end) + end) + + describe('coroutine safety', function() + it_exec('does not allow coroutine.yield', function() + local task = run(function() + coroutine.yield('This will cause an error.') + end) + check_task_err(task, 'Unexpected coroutine.yield().*') + end) + + it_exec('does not allow coroutine.resume', function() + local co --- @type thread + local task = run(function() + co = coroutine.running() + eternity() + end) + + local status, err = coroutine.resume(co) + assert(not status, 'Expected coroutine.resume to fail') + eq(err, 'Unexpected coroutine.resume()') + check_task_err(task, 'Unexpected coroutine.resume%(%)') + end) + + it_exec('does not allow coroutine.resume when awaiting detached task', function() + local t = run(eternity) + local co --- @type thread + local task = run(function() + co = coroutine.running() + await(t) + end) + + local status, err = coroutine.resume(co) + assert(not status, 'Expected coroutine.resume to fail') + eq(err, 'Unexpected coroutine.resume()') + check_task_err(task, 'Unexpected coroutine.resume%(%)') + t:close() + end) + + it_exec('preserves child errors after invalid coroutine.resume', function() + local blocker = run(eternity) + local co --- @type thread + local parent = run(function() + co = coroutine.running() + run(function() + await(vim.schedule) + error('CHILD ERROR') + end) + await(blocker) + end) + + local status, err = coroutine.resume(co) + assert(not status, 'Expected coroutine.resume to fail') + eq(err, 'Unexpected coroutine.resume()') + local check_ok, check_err = + pcall(check_task_err, parent, 'child error: .*async_spec.lua:%d+: CHILD ERROR') + + blocker:close() + check_task_err(blocker, 'closed') + + if not check_ok then + error(check_err, 0) + end + end) + end) + + describe('inspect_tree', function() + local outside_tree = t.dedent([=[ + parent %[awaiting%] + ├─ child1 %[awaiting%] + ├─ child2 %[awaiting%] + └─ child3 %[awaiting%] + ├─ sub_child1 %[awaiting%] + ├─ sub_child2 %[awaiting%] + └─ %[awaiting%]]=]) + local inside_tree = t.dedent([=[ + parent %[awaiting%] + ├─ child1 %[awaiting%] + ├─ child2 %[awaiting%] + └─ child3 %[running%] + ├─ sub_child1 %[awaiting%] + ├─ sub_child2 %[awaiting%] + └─ %[awaiting%]]=]) + local jit_tree = t.dedent([=[ + parent@.*async_spec.lua:%d+ %[awaiting%] + └─ child@.*async_spec.lua:%d+ %[awaiting%]]=]) + local puc_tree = t.dedent([=[ + parent=.* %[awaiting%] + └─ child=.* %[awaiting%]]=]) + + it_exec('outside of tasks', function(expected) + local parent = run('parent', function() + run('child1', eternity) + run('child2', eternity) + run('child3', function(...) + run('sub_child1', eternity) + run('sub_child2', eternity) + run(eternity) + end) + end) + + eq(p(expected), Async._inspect_tree()) + + parent:close() + check_task_err(parent, 'closed') + end, outside_tree) + + it_exec('inside a task', function(expected) + local inspect + local parent = run('parent', function() + run('child1', eternity) + run('child2', eternity) + run('child3', function(...) + run('sub_child1', eternity) + run('sub_child2', eternity) + run(eternity) + inspect = Async._inspect_tree() + end) + end) + + eq(p(expected), inspect) + + parent:close() + check_task_err(parent, 'closed') + end, inside_tree) + + it_exec('can show task creation locations in debug mode', function(jit_expected, puc_expected) + AsyncRuntime.config({ debug = true }) + local parent + local ok, err = pcall(function() + parent = run('parent', function() + run('child', eternity) + end) + + local expected = is_jit() and jit_expected or puc_expected + + eq(p(expected), Async._inspect_tree()) + end) + if parent then + parent:close() + end + AsyncRuntime.config({ debug = false }) + if parent then + check_task_err(parent, 'closed') + end + if not ok then + error(err, 0) + end + end, jit_tree, puc_tree) + end) + + describe('pcall and task-control errors', function() + it_exec('child errors remain terminal after pcall catches delivery', function() + local results = {} + local parent = run(function() + local _child = run(function() + Async.sleep(5) + error('CHILD ERROR') + end) + + local ok1, err1 = pcall(function() + Async.sleep(100) + end) + + if not ok1 then + results[#results + 1] = 'caught_first' + results[#results + 1] = err1:match('CHILD ERROR') and 'has_error' or 'no_error' + end + + local ok2, err2 = pcall(function() + Async.sleep(1) + end) + + if not ok2 then + results[#results + 1] = 'caught_second' + results[#results + 1] = err2:match('CHILD ERROR') and 'has_error' or 'no_error' + else + results[#results + 1] = 'no_second_error' + end + + results[#results + 1] = 'returned' + end) + + local ok, err = parent:pwait(200) + eq(false, ok) + --- @cast err string + assert(err:match('child error:.*CHILD ERROR'), 'Expected child error, got: ' .. tostring(err)) + + eq({ + 'caught_first', + 'has_error', + 'caught_second', + 'has_error', + 'returned', + }, results) + end) + + it_exec('awaited child errors remain terminal without child wrapper', function() + local results = {} + local parent = run(function() + local child = run(function() + error('AWAITED CHILD ERROR') + end) + + local ok1, err1 = pcall(function() + await(child) + end) + + if not ok1 then + results[#results + 1] = err1:match('AWAITED CHILD ERROR') and 'caught_child' or 'other' + results[#results + 1] = err1:match('child error:') and 'wrapped' or 'unwrapped' + end + + local ok2, err2 = pcall(function() + Async.sleep(1) + end) + + if not ok2 then + results[#results + 1] = err2:match('AWAITED CHILD ERROR') and 'caught_again' or 'other' + results[#results + 1] = err2:match('child error:') and 'wrapped' or 'unwrapped' + end + + results[#results + 1] = 'returned' + end) + + local ok, err = parent:pwait(200) + eq(false, ok) + --- @cast err string + assert( + err:match('.*async_spec.lua:%d+: AWAITED CHILD ERROR'), + 'Expected awaited child error, got: ' .. tostring(err) + ) + assert(not err:match('child error:'), 'Did not expect child wrapper, got: ' .. err) + + eq({ + 'caught_child', + 'unwrapped', + 'caught_again', + 'unwrapped', + 'returned', + }, results) + end) + + it_exec('false awaited child errors remain terminal after pcall catches delivery', function() + local parent = run(function() + local child = run(function() + error(false, 0) + end) + + local ok1, err1 = pcall(function() + await(child) + end) + + eq(false, ok1) + eq(false, err1) + + local ok2, err2 = pcall(function() + Async.sleep(1) + end) + + eq(false, ok2) + eq(false, err2) + end) + + local ok, err = parent:pwait(200) + eq(false, ok) + eq(false, err) + end) + + it_exec('pawait returns successful task results', function() + local parent = run(function() + local ok, a, b, c = Async.pawait(run(function() + Async.sleep(1) + return 1, 'two', true + end)) + + eq(true, ok) + eq(1, a) + eq('two', b) + eq(true, c) + + return 'parent ok' + end) + + eq('parent ok', parent:wait(100)) + end) + + it_exec('pawait accepts await callback overloads', function() + local parent = run(function() + local ok1, value = Async.pawait(function(callback) + vim.schedule(function() + callback('scheduled') + end) + end) + + local ok2, a, b = Async.pawait(2, function(prefix, callback) + vim.schedule(function() + callback(prefix, 'done') + end) + end, 'arg') + + eq({ true, 'scheduled' }, { ok1, value }) + eq({ true, 'arg', 'done' }, { ok2, a, b }) + + return 'parent ok' + end) + + eq('parent ok', parent:wait(100)) + end) + + it_exec('pawait returns awaitable setup errors as data', function() + local parent = run(function() + local ok, err = Async.pawait(function(_callback) + error() + end) + + eq(false, ok) + eq('error(nil)', err) + + return 'parent ok' + end) + + eq('parent ok', parent:wait(100)) + end) + + it_exec('pawait keeps synchronous results before a setup error', function() + local parent = run(function() + local function callback_then_error(callback) + callback('result') + error('LATE_SETUP_ERROR') + end + + local ok, result = Async.pawait(callback_then_error) + eq({ true, 'result' }, { ok, result }) + + local raw_ok, err = pcall(await, callback_then_error) + eq(false, raw_ok) + assert(tostring(err):match('LATE_SETUP_ERROR'), tostring(err)) + end) + + parent:wait(100) + end) + + it_exec('pawait ignores a late callback after setup fails', function() + local child --- @type vim.async.Task + local callback_ran = false + local parent = run(function() + child = run(function() + Async.sleep(5) + return 'child finished' + end) + + local ok, err = Async.pawait(function(callback) + vim.schedule(function() + callback_ran = true + callback('late result') + end) + error('SETUP_ERROR') + end) + + eq(false, ok) + assert(tostring(err):match('SETUP_ERROR'), tostring(err)) + return 'parent finished' + end) + + eq('parent finished', parent:wait(100)) + eq(true, callback_ran) + eq('child finished', child:wait(100)) + end) + + it_exec('pawait returns synchronous child errors as data', function() + local results = {} + local parent = run(function() + local child = run(function() + results[#results + 1] = 'child_started' + error('SYNC CHILD ERROR') + end) + + results[#results + 1] = 'after_run' + local ok, err = Async.pawait(child) + + eq(false, ok) + --- @cast err string + results[#results + 1] = err:match('SYNC CHILD ERROR') and 'got_error' or 'other' + results[#results + 1] = err:match('child error:') and 'wrapped' or 'unwrapped' + + Async.sleep(1) + results[#results + 1] = 'continued' + + return 'parent ok' + end) + + eq('parent ok', parent:wait(100)) + eq({ + 'after_run', + 'child_started', + 'got_error', + 'unwrapped', + 'continued', + }, results) + end) + + it_exec('pawait returns asynchronous child errors as data', function() + local results = {} + local parent = run(function() + local ok, err = Async.pawait(run(function() + Async.sleep(1) + error('ASYNC CHILD ERROR') + end)) + + eq(false, ok) + --- @cast err string + results[#results + 1] = err:match('ASYNC CHILD ERROR') and 'got_error' or 'other' + results[#results + 1] = err:match('child error:') and 'wrapped' or 'unwrapped' + + Async.sleep(1) + results[#results + 1] = 'continued' + + return 'parent ok' + end) + + eq('parent ok', parent:wait(100)) + eq({ + 'got_error', + 'unwrapped', + 'continued', + }, results) + end) + + it_exec('pawait does not protect current task cancellation', function() + local results = {} + local parent = run(function() + local ok, err = pcall(function() + Async.pawait(function(_callback) + return add_handle('pawait_current_cancellation_timer', vim.uv.new_timer()) + end) + end) + + eq(false, ok) + results[#results + 1] = is_closed_error(err) and 'caught_closed' or 'other_error' + results[#results + 1] = Async.is_closing() and 'is_closing' or 'not_closing' + results[#results + 1] = 'cleanup' + end) + + parent:close() + + check_task_err(parent, 'closed') + eq({ + 'caught_closed', + 'is_closing', + 'cleanup', + }, results) + end) + + it_exec('pawait does not protect unrelated current task errors', function() + local results = {} + local parent = run(function() + local _child = run(function() + Async.sleep(5) + error('CHILD ERROR') + end) + + local ok, err = pcall(function() + Async.pawait(function(callback) + local timer = add_handle('pending_child_error_timer', vim.uv.new_timer()) + timer:start(100, 0, function() + timer:close() + callback('done') + end) + return timer + end) + end) + + eq(false, ok) + --- @cast err string + results[#results + 1] = err:match('child error:.*CHILD ERROR') and 'child_error' + or 'other_error' + results[#results + 1] = Async.is_closing() and 'is_closing' or 'not_closing' + results[#results + 1] = 'cleanup' + end) + + local ok, err = parent:pwait(200) + eq(false, ok) + --- @cast err string + assert(err:match('child error:.*CHILD ERROR'), 'Expected child error, got: ' .. tostring(err)) + + eq({ + 'child_error', + 'not_closing', + 'cleanup', + }, results) + end) + + it_exec('cancellations are level-triggered (persist across catches)', function() + local results = {} + local task = run(function() + local ok1, err1 = pcall(function() + Async.sleep(100) + end) + + if not ok1 then + results[#results + 1] = 'caught_first' + results[#results + 1] = is_closed_error(err1) and 'is_closed' or 'other_error' + end + + local ok2, err2 = pcall(function() + Async.sleep(1) + end) + + if not ok2 then + results[#results + 1] = 'caught_second' + results[#results + 1] = is_closed_error(err2) and 'is_closed' or 'other_error' + end + + results[#results + 1] = 'should_not_reach' + end) + + task:close() + + check_task_err(task, 'closed') + + eq({ + 'caught_first', + 'is_closed', + 'caught_second', + 'is_closed', + 'should_not_reach', + }, results) + end) + + it_exec('checkpoint rethrows current task cancellation after cleanup', function() + local results = {} + local task = run(function() + local ok, err = pcall(function() + Async.sleep(100) + end) + + eq(false, ok) + results[#results + 1] = is_closed_error(err) and 'caught_closed' or 'other_error' + results[#results + 1] = 'cleanup' + + Async.checkpoint() + results[#results + 1] = 'after_checkpoint' + end) + + task:close() + + check_task_err(task, 'closed') + + eq({ + 'caught_closed', + 'cleanup', + }, results) + end) + + it_exec('checkpoint rethrows current task failure after cleanup', function() + local results = {} + local parent = run(function() + local _child = run(function() + Async.sleep(5) + error('CHILD ERROR') + end) + + local ok, err = pcall(function() + Async.sleep(100) + end) + + eq(false, ok) + --- @cast err string + results[#results + 1] = err:match('CHILD ERROR') and 'caught_child_error' or 'other_error' + results[#results + 1] = 'cleanup' + + Async.checkpoint() + results[#results + 1] = 'after_checkpoint' + end) + + check_task_err(parent, 'child error:.*CHILD ERROR') + + eq({ + 'caught_child_error', + 'cleanup', + }, results) + end) + + it_exec('can recover synchronous errors inside async tasks', function() + local results = {} + run(function() + local ok = pcall(function() + error('BAD CONFIG') + end) + + if not ok then + results[#results + 1] = 'error_caught' + end + + Async.sleep(1) + results[#results + 1] = 'finished' + end):wait(200) + + eq({ + 'error_caught', + 'finished', + }, results) + end) + + it_exec('cancellation persists even after pcall catches it', function() + local results = {} + local task = run(function() + for i = 1, 5 do + local ok, err = pcall(function() + Async.sleep(10) + end) + + if not ok then + if is_closed_error(err) then + results[#results + 1] = ('closed_iteration_%d'):format(i) + else + results[#results + 1] = ('error_iteration_%d'):format(i) + end + else + results[#results + 1] = ('success_iteration_%d'):format(i) + end + end + end) + + task:close() + + check_task_err(task, 'closed') + + eq({ + 'closed_iteration_1', + 'closed_iteration_2', + 'closed_iteration_3', + 'closed_iteration_4', + 'closed_iteration_5', + }, results) + end) + + it_exec('is_closing() reflects level-triggered cancellation state', function() + local results = {} + local task = run(function() + for _ = 1, 3 do + results[#results + 1] = ('is_closing_%d'):format(Async.is_closing() and 1 or 0) + + local ok = pcall(function() + Async.sleep(10) + end) + + if not ok then + results[#results + 1] = ('after_catch_is_closing_%d'):format( + Async.is_closing() and 1 or 0 + ) + end + end + end) + + task:close() + + check_task_err(task, 'closed') + + eq({ + 'is_closing_0', + 'after_catch_is_closing_1', + 'is_closing_1', + 'after_catch_is_closing_1', + 'is_closing_1', + 'after_catch_is_closing_1', + }, results) + end) + + it_exec('first child error remains pending across subsequent awaits', function() + local results = {} + local release_first --- @type fun()? + + -- Child failure, rather than this awaitable, resumes the parent. + local function wait_for_child_error() + await(function() end) + end + + local parent = run(function() + local _child1 = run(function() + await(function(callback) + release_first = callback + end) + error('ERROR_1') + end) + + local release_second --- @type fun()? + local _child2 = run(function() + await(function(callback) + release_second = callback + end) + error('ERROR_2') + end) + + local ok1, err1 = pcall(wait_for_child_error) + + if not ok1 then + results[#results + 1] = err1:match('ERROR_1') and 'got_error_1' or 'other' + end + + assert(release_second) + release_second() + + local ok2, err2 = pcall(wait_for_child_error) + + if not ok2 then + results[#results + 1] = err2:match('ERROR_1') and 'got_error_1_again' or 'other' + end + + results[#results + 1] = 'returned' + end) + + assert(release_first) + release_first() + + local ok, err = parent:pwait(200) + eq(false, ok) + --- @cast err string + assert( + err:match('child error:.*ERROR_1'), + 'Expected first child error, got: ' .. tostring(err) + ) + + eq({ + 'got_error_1', + 'got_error_1_again', + 'returned', + }, results) + end) + + it_exec('task error takes precedence over cancellation when both occur', function() + local task = run(function() + pcall(function() + Async.sleep(10) + end) + + error('TASK_ERROR') + end) + + task:close() + + local ok, err = task:pwait(100) + assert(not ok, 'Expected task to error') + eq(true, err:match('TASK_ERROR') ~= nil, 'Expected TASK_ERROR, got: ' .. tostring(err)) + end) + + it_exec( + 'cancellation takes precedence when task completes successfully while closing', + function() + local results = {} + local task = run(function() + local ok, err = pcall(function() + await(function(_callback) + return { + close = function(_, callback) + results[#results + 1] = 'close_called' + callback() + end, + } + end) + end) + + eq(false, ok) + eq(true, is_closed_error(err), 'Expected closed error, got: ' .. tostring(err)) + results[#results + 1] = 'caught_close' + results[#results + 1] = 'completed' + return 'SUCCESS' + end) + + eq('awaiting', task:status()) + task:close() + check_task_err(task, 'closed') + + eq({ + 'close_called', + 'caught_close', + 'completed', + }, results) + end + ) + end) + + describe('edge case tests', function() + it_exec('handles awaiting closable that is already closing', function() + -- Test for potential issue where is_closing() returns true + local close_count = 0 + local callback_called = false + + local closable = { + _closing = false, + is_closing = function(self) + return self._closing + end, + close = function(self, cb) + close_count = close_count + 1 + self._closing = true + if cb then + vim.schedule(cb) + end + end, + } + + local task = run(function() + -- Start closing the closable + closable:close() + + -- Now try to await something that returns this already-closing closable + local result = await(function(callback) + vim.schedule(function() + callback('RESULT') + end) + return closable + end) + + callback_called = true + return result + end) + + local result = task:wait(100) + eq('RESULT', result) + eq(true, callback_called) + -- The closable should only be closed once (by the explicit close call) + -- handle_close_awaiting should detect is_closing and not call close again + eq(1, close_count) + end) + + it_exec('child error during parent finalization is handled', function() + local parent = run(function() + local _child = run(function() + Async.sleep(5) + error('CHILD_ERROR') + end) + + -- Returning starts finalization, which waits for attached child work. + end) + + local ok, err = parent:pwait(100) + + eq(false, ok) + --- @cast err string + assert(err:match('child error:.*CHILD_ERROR'), 'Expected child error, got: ' .. tostring(err)) + end) + + it_exec('child error formatting cannot interrupt parent finalization', function() + local bad_error = setmetatable({}, { + __tostring = function() + error('TOSTRING_ERROR') + end, + }) + + local parent = run(function() + local _child = run(function() + Async.sleep(1) + error(bad_error, 0) + end) + end) + + check_task_err(parent, 'child error: ') + end) + + it_exec('child error during parent finalization completes once and closes siblings', function() + local completions = 0 + local sibling --- @type vim.async.Task + + local parent = run(function() + local _child = run(function() + Async.sleep(1) + error('CHILD_ERROR') + end) + + sibling = run(eternity) + end) + + parent:on_complete(function() + completions = completions + 1 + end) + + local ok, err = parent:pwait(100) + + eq(false, ok) + --- @cast err string + assert(err:match('child error:.*CHILD_ERROR'), 'Expected child error, got: ' .. tostring(err)) + eq(1, completions) + check_task_err(sibling, 'closed') + end) + + it_exec('later child error during parent finalization closes earlier siblings', function() + local completions = 0 + local sibling --- @type vim.async.Task + + local parent = run(function() + sibling = run(eternity) + + local _child = run(function() + Async.sleep(1) + error('CHILD_ERROR') + end) + end) + + parent:on_complete(function() + completions = completions + 1 + end) + + local ok, err = parent:pwait(100) + + eq(false, ok) + --- @cast err string + assert(err:match('child error:.*CHILD_ERROR'), 'Expected child error, got: ' .. tostring(err)) + eq(1, completions) + check_task_err(sibling, 'closed') + end) + + it_exec('child error during parent finalization waits for sibling cleanup', function() + local cleanup_done = false + local sibling --- @type vim.async.Task + + local parent = run(function() + sibling = run(function() + await(function() + return { + close = function(_, callback) + vim.schedule(function() + cleanup_done = true + callback() + end) + end, + } + end) + end) + + local _child = run(function() + Async.sleep(1) + error('CHILD_ERROR') + end) + end) + + local ok, err = parent:pwait(100) + + eq(false, ok) + --- @cast err string + assert(err:match('child error:.*CHILD_ERROR'), 'Expected child error, got: ' .. tostring(err)) + eq(true, cleanup_done) + check_task_err(sibling, 'closed') + end) + + it_exec('parent failure waits for every child cleanup', function() + local first_cleanup_done = false + local second_cleanup_done = false + + local parent = run(function() + run(function() + await(function() + return { + close = function(_, callback) + first_cleanup_done = true + callback() + end, + } + end) + end) + + run(function() + await(function() + return { + close = function(_, callback) + vim.schedule(function() + second_cleanup_done = true + callback() + end) + end, + } + end) + end) + + Async.checkpoint() + error('PARENT_ERROR') + end) + + local ok, err = parent:pwait(100) + + eq(false, ok) + --- @cast err string + assert(err:match('PARENT_ERROR'), 'Expected parent error, got: ' .. tostring(err)) + eq(true, first_cleanup_done) + eq(true, second_cleanup_done) + end) + + it_exec('does not resume a closing task before awaitable cleanup', function() + local operation_callback --- @type fun(...: any) + local body_ran = false + local cleanup_done = false + local closable = { closing = false } + + function closable:is_closing() + return self.closing + end + + function closable:close(callback) + self.closing = true + operation_callback('RESULT') + vim.schedule(function() + cleanup_done = true + callback() + end) + end + + local task = run(function() + Async.pawait(function(callback) + operation_callback = callback + return closable + end) + body_ran = true + end) + + task:close() + check_task_err(task, 'closed') + + eq(false, body_ran) + eq(true, cleanup_done) + end) + + it_exec('future complete is one-shot', function() + local future = require('vim.async._future')() + future:complete(nil, 'first') + + local ok, err = pcall(function() + future:complete(nil, 'second') + end) + + eq(false, ok) + --- @cast err string + assert(err:match('Future is already completed'), 'Unexpected error: ' .. tostring(err)) + + local stat, result = future:result() + eq(true, stat) + eq('first', result) + end) + + it_exec('future false error still completes', function() + local future = require('vim.async._future')() + future:complete(false) + + eq(true, future:completed()) + + local stat, err = future:result() + eq(false, stat) + eq(false, err) + end) + + it_exec('normalizes nil task errors', function() + check_task_err( + run(function() + error() + end), + 'error%(nil%)' + ) + end) + + it_exec('normalizes nil awaitable setup errors', function() + local task = run(function() + await(function() + error() + end) + end) + + check_task_err(task, 'error%(nil%)') + end) + + it_exec('normalizes nil close errors', function() + local task = run(function() + await(function() + return { + close = function() + error() + end, + } + end) + end) + + task:close() + check_task_err(task, 'error%(nil%)') + end) + + it_exec('normalizes nil future callback errors', function() + local future = require('vim.async._future')() + future:on_complete(function() + error() + end) + + local ok, err = pcall(function() + future:complete(nil, 'value') + end) + + eq(false, ok) + --- @cast err string + assert(err:match('error%(nil%)'), 'Unexpected error: ' .. tostring(err)) + end) + + it_exec('continues future callbacks after an unprintable error', function() + local future = require('vim.async._future')() + local observed = false + local unprintable = setmetatable({}, { + __tostring = function() + error('TOSTRING_ERROR') + end, + }) + + future:on_complete(function() + error(unprintable) + end) + future:on_complete(function() + observed = true + end) + + local ok, err = pcall(function() + future:complete(nil, 'value') + end) + + eq(false, ok) + eq(true, observed) + assert(tostring(err):match(''), 'Unexpected error: ' .. tostring(err)) + end) + + it_exec('callback called multiple times is handled gracefully', function() + -- Test that calling callback multiple times doesn't break things + local call_count = 0 + local results = {} + + local task = run(function() + local result = await(function(callback) + call_count = call_count + 1 + callback('FIRST_CALL') + + -- Try calling again (should be ignored) + vim.schedule(function() + call_count = call_count + 1 + callback('SECOND_CALL') + end) + end) + + table.insert(results, result) + return result + end) + + local final_result = task:wait(100) + + -- Should only get the first callback result + eq('FIRST_CALL', final_result) + eq(1, #results) + eq('FIRST_CALL', results[1]) + + -- Wait a bit for the second callback to potentially fire + run(function() + Async.sleep(20) + end):wait() + + -- Both callbacks should have been called + eq(2, call_count) + + -- But only the first one should have been processed + eq(1, #results) + end) + + it_exec('closable cleanup happens even if close() errors', function() + -- Test that if a closable's close() method errors, we handle it gracefully + local close_called = false + + local task = run(function() + local result = await(function(callback) + local closable = { + close = function() + close_called = true + error('CLOSE_ERROR') + end, + } + + vim.schedule(function() + callback('RESULT') + end) + + return closable + end) + + return result + end) + + task:close() -- This should trigger closing the closable + + -- The task should complete with the close error + local ok, err = task:pwait(100) + + eq(true, close_called, 'close() should have been called') + assert(not ok, 'Task should have errored') + assert(err:match('CLOSE_ERROR'), 'Expected CLOSE_ERROR, got: ' .. tostring(err)) + end) + end) +end)