From ce8a897f985b2cfd1194f6c03fe5388f9bc6d6bf Mon Sep 17 00:00:00 2001 From: Lewis Russell Date: Tue, 25 Aug 2026 17:37:27 +0100 Subject: [PATCH] feat(lua): add vim.async Problem: Nvim has many Lua APIs that start callback-driven work: timers, jobs, libuv handles, and other event-loop tasks. Callers that need to sequence or cancel that work have to build their own coroutine wrappers, task bookkeeping, and cleanup rules. This makes async control flow hard to share, test, and document. Solution: Add `vim.async`, a structured-concurrency module vendored from async.nvim. It provides task handles, await/pawait helpers, sleep/timeout helpers, completion-order iteration, and semaphores on top of Nvim's event loop. The API follows the same broad model as Trio: async work has an owner, tasks are awaited explicitly, and cancellation is cooperative. Include generated vimdoc with an introductory overview and examples, a news entry, and functional tests for the new module. AI-assisted --- .luarc.json | 1 + runtime/doc/lua-async.txt | 603 ++++++ runtime/doc/news.txt | 3 + runtime/lua/vim/_core/editor.lua | 1 + runtime/lua/vim/_core/util.lua | 21 + runtime/lua/vim/_meta.lua | 2 + runtime/lua/vim/async.lua | 317 +++ runtime/lua/vim/async/_core.lua | 1128 +++++++++++ runtime/lua/vim/async/_event.lua | 125 ++ runtime/lua/vim/async/_future.lua | 77 + runtime/lua/vim/async/_queue.lua | 107 + runtime/lua/vim/async/_runtime.lua | 46 + runtime/lua/vim/async/_semaphore.lua | 133 ++ src/gen/gen_vimdoc.lua | 67 +- src/gen/util.lua | 2 + src/nvim/CMakeLists.txt | 2 + test/functional/lua/async_spec.lua | 2687 ++++++++++++++++++++++++++ 17 files changed, 5315 insertions(+), 7 deletions(-) create mode 100644 runtime/doc/lua-async.txt create mode 100644 runtime/lua/vim/async.lua create mode 100644 runtime/lua/vim/async/_core.lua create mode 100644 runtime/lua/vim/async/_event.lua create mode 100644 runtime/lua/vim/async/_future.lua create mode 100644 runtime/lua/vim/async/_queue.lua create mode 100644 runtime/lua/vim/async/_runtime.lua create mode 100644 runtime/lua/vim/async/_semaphore.lua create mode 100644 test/functional/lua/async_spec.lua 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)