parent
d79581530b
commit
b61bf7b155
@ -0,0 +1,174 @@ |
||||
# Concurrency: coroutines and `task` |
||||
|
||||
`a` runs every script on an async core ([tokio](https://tokio.rs/)). The async |
||||
stdlib calls — `os.sleep`, the `http` client, `sqlite` queries — don't block the |
||||
OS thread while they wait; they suspend and let other work run. This page |
||||
explains how that interacts with Lua coroutines, and how to run several pieces of |
||||
work concurrently with `task.join`. |
||||
|
||||
## TL;DR |
||||
|
||||
- Plain Lua coroutines work exactly as in stock Lua. |
||||
- Async stdlib calls (`os.sleep`, `http.*`, `sqlite` …) suspend on the runtime — |
||||
they don't block the thread. |
||||
- Use **`task.join`** to run multiple coroutines concurrently. N tasks that each |
||||
wait T seconds finish in ~T, not ~N·T. |
||||
- **Don't** call async stdlib functions inside a coroutine you drive yourself |
||||
with `coroutine.resume` / `coroutine.wrap` — see [the gotcha](#the-gotcha). |
||||
|
||||
## Plain coroutines |
||||
|
||||
Stock Lua coroutines are pure VM machinery and behave normally: |
||||
|
||||
```lua |
||||
local function squares(n) |
||||
for i = 1, n do |
||||
coroutine.yield(i * i) |
||||
end |
||||
end |
||||
|
||||
local next = coroutine.wrap(squares) |
||||
print(next(3), next(3), next(3)) --> 1 4 9 |
||||
``` |
||||
|
||||
This is ordinary cooperative scheduling: nothing runs concurrently, and the |
||||
tokio runtime is never involved. |
||||
|
||||
## Async calls suspend, they don't block |
||||
|
||||
An async stdlib call at the top level of your script (or inside a `task` |
||||
coroutine) suspends until it's ready, without tying up the thread: |
||||
|
||||
```lua |
||||
local t0 = os.microtime() |
||||
os.sleep(0.10) |
||||
print(string.format("waited %.3fs", os.microtime() - t0)) --> waited ~0.100s |
||||
``` |
||||
|
||||
While that sleep is pending, any sibling tasks (see below) keep making progress. |
||||
|
||||
## `task.join` — run work concurrently |
||||
|
||||
``` |
||||
task.join(fn1, fn2, ...) -> r1, r2, ... |
||||
``` |
||||
|
||||
Runs each function as its own coroutine, drives them **concurrently** on the |
||||
runtime, and returns each one's first result positionally once all have |
||||
finished. If any task raises an error, `task.join` re-raises the first one. |
||||
|
||||
Because async calls suspend instead of blocking, the tasks overlap: |
||||
|
||||
```lua |
||||
local function worker(name, secs) |
||||
return function() |
||||
os.sleep(secs) -- suspends; siblings run meanwhile |
||||
return name |
||||
end |
||||
end |
||||
|
||||
local t0 = os.microtime() |
||||
local a, b, c = task.join( |
||||
worker("slow", 0.30), |
||||
worker("med", 0.20), |
||||
worker("fast", 0.10) |
||||
) |
||||
print(a, b, c) --> slow med fast |
||||
print(string.format("%.3fs", os.microtime() - t0)) --> ~0.300s, not 0.600s |
||||
``` |
||||
|
||||
The three sleeps run at the same time, so the wall-clock time is the *longest* |
||||
single task, not the sum. |
||||
|
||||
### Real-world example: concurrent HTTP fetches |
||||
|
||||
```lua |
||||
local function fetch(url) |
||||
return function() |
||||
return http.get(url).status |
||||
end |
||||
end |
||||
|
||||
local s1, s2, s3 = task.join( |
||||
fetch("https://httpbingo.org/delay/1"), |
||||
fetch("https://httpbingo.org/delay/1"), |
||||
fetch("https://httpbingo.org/delay/1") |
||||
) |
||||
print(s1, s2, s3) --> 200 200 200, in ~1s total instead of ~3s |
||||
``` |
||||
|
||||
### Passing results back |
||||
|
||||
Each task returns its first value to the corresponding slot: |
||||
|
||||
```lua |
||||
local me, repos = task.join( |
||||
function() return http.getJSON("https://api.github.com/users/torvalds") end, |
||||
function() return http.getJSON("https://api.github.com/users/torvalds/repos") end |
||||
) |
||||
print(me.name, #repos) -- both fetched concurrently |
||||
``` |
||||
|
||||
### Nesting |
||||
|
||||
`task.join` suspends like any other async call, so a task can itself call |
||||
`task.join`: |
||||
|
||||
```lua |
||||
local total = task.join( |
||||
function() |
||||
local x, y = task.join( |
||||
function() os.sleep(0.05); return 21 end, |
||||
function() os.sleep(0.05); return 21 end |
||||
) |
||||
return x + y -- 42 |
||||
end, |
||||
function() os.sleep(0.10); return "sibling" end |
||||
) |
||||
print(total) --> 42 |
||||
``` |
||||
|
||||
## The gotcha |
||||
|
||||
Async stdlib functions only suspend correctly when the runtime is driving the |
||||
coroutine — i.e. at the top level of your script, or inside a `task.join` |
||||
coroutine. If you drive a coroutine **yourself**, the async call does *not* wait: |
||||
|
||||
```lua |
||||
-- DON'T do this: |
||||
local co = coroutine.wrap(function() |
||||
os.sleep(0.50) |
||||
return "awoke" |
||||
end) |
||||
|
||||
local v = co() -- returns immediately with an opaque value; the 0.5s wait |
||||
-- never happens, because `coroutine.wrap` can't drive the |
||||
-- runtime. |
||||
``` |
||||
|
||||
Under the hood, an async call yields a private marker that only the runtime's |
||||
scheduler understands; a hand-written `resume`/`wrap` loop just receives that |
||||
marker and moves on. The rule of thumb: |
||||
|
||||
> Use plain `coroutine.*` for pure-Lua generators. The moment a coroutine needs |
||||
> to `os.sleep`, hit the network, or touch the database, run it through |
||||
> `task.join` instead. |
||||
|
||||
## How it works (and its limits) |
||||
|
||||
`task.join` wraps each function in a Lua coroutine and polls them all |
||||
concurrently on the tokio runtime. The concurrency comes from the **reactor** — |
||||
timers and I/O yielding control while they wait — not from extra threads: |
||||
|
||||
- Great for **I/O-bound** work: sleeps, HTTP requests, database queries all |
||||
overlap. |
||||
- A **CPU-bound** task (a tight compute loop with no async calls) will *not* |
||||
yield, so it blocks its siblings until it finishes or hits an async call. |
||||
|
||||
In other words this is **concurrency, not parallelism**: one thread, many |
||||
in-flight operations. |
||||
|
||||
## See also |
||||
|
||||
- Runnable demo: [`lua/coroutines-demo.lua`](../lua/coroutines-demo.lua) |
||||
(`a lua/coroutines-demo.lua`) |
||||
@ -0,0 +1,99 @@ |
||||
-- Proof-of-concept: how Lua coroutines interact with the tokio runtime, and |
||||
-- how the async os.sleep behaves in each setting. |
||||
-- |
||||
-- Run with: cargo run -- lua/coroutines-demo.lua |
||||
|
||||
local function banner(s) print("\n=== " .. s .. " ===") end |
||||
|
||||
---------------------------------------------------------------------- |
||||
banner("1. Pure Lua coroutines (no async involved)") |
||||
---------------------------------------------------------------------- |
||||
-- Plain cooperative coroutines work exactly as in stock Lua: this is pure VM |
||||
-- machinery and never touches tokio. |
||||
local function counter(n) |
||||
for i = 1, n do |
||||
coroutine.yield(i * i) |
||||
end |
||||
return "done" |
||||
end |
||||
|
||||
local co = coroutine.wrap(counter) |
||||
print("squares:", co(3), co(3), co(3)) -- 1 4 9 |
||||
|
||||
local raw = coroutine.create(counter) |
||||
print("status fresh:", coroutine.status(raw)) -- suspended |
||||
coroutine.resume(raw, 1) |
||||
coroutine.resume(raw, 1) -- runs to `return` |
||||
print("status after return:", coroutine.status(raw)) -- dead |
||||
|
||||
---------------------------------------------------------------------- |
||||
banner("2. Async os.sleep at the top level") |
||||
---------------------------------------------------------------------- |
||||
-- The main chunk is itself run by mlua's async executor (exec_async), so an |
||||
-- async call here is driven correctly and really suspends on the tokio timer. |
||||
local t0 = os.microtime() |
||||
os.sleep(0.10) |
||||
print(string.format("slept ~0.10s, measured %.3fs", os.microtime() - t0)) |
||||
|
||||
---------------------------------------------------------------------- |
||||
banner("3. GOTCHA: async os.sleep inside a *manually resumed* coroutine") |
||||
---------------------------------------------------------------------- |
||||
-- mlua implements async functions by yielding a private sentinel to whatever is |
||||
-- driving the coroutine. mlua's own executor understands it; a plain |
||||
-- coroutine.resume / coroutine.wrap does NOT. So driving an async call yourself |
||||
-- does not actually wait -- the sleep is not performed by your resume loop. |
||||
local sleeper = coroutine.wrap(function() |
||||
os.sleep(0.50) |
||||
return "awoke" |
||||
end) |
||||
|
||||
local t1 = os.microtime() |
||||
local first = sleeper() -- returns immediately with the sentinel |
||||
print(string.format("first resume returned %s after only %.3fs (did NOT wait 0.5s)", |
||||
tostring(first), os.microtime() - t1)) |
||||
print("--> lesson: don't hand-drive coroutines that call async stdlib functions.") |
||||
|
||||
---------------------------------------------------------------------- |
||||
banner("4. Real concurrency: task.join drives coroutines on tokio") |
||||
---------------------------------------------------------------------- |
||||
-- task.join runs each function as its own coroutine and lets the tokio reactor |
||||
-- interleave them. Three tasks that each sleep concurrently finish in ~the |
||||
-- longest single sleep, not the sum -- proof the sleeps overlap on one thread. |
||||
local function worker(name, secs) |
||||
return function() |
||||
print(string.format(" [%s] start", name)) |
||||
os.sleep(secs) |
||||
print(string.format(" [%s] woke after %.2fs", name, secs)) |
||||
return name .. ":" .. secs |
||||
end |
||||
end |
||||
|
||||
local t2 = os.microtime() |
||||
local a, b, c = task.join( |
||||
worker("slow", 0.30), |
||||
worker("med", 0.20), |
||||
worker("fast", 0.10) |
||||
) |
||||
local elapsed = os.microtime() - t2 |
||||
print(string.format("results: %s, %s, %s", a, b, c)) |
||||
print(string.format("wall time: %.3fs", elapsed)) |
||||
print(string.format("sequential would have been ~0.60s; concurrent ~0.30s => %s", |
||||
elapsed < 0.45 and "CONCURRENT (overlapped on tokio)" or "serialized?!")) |
||||
|
||||
---------------------------------------------------------------------- |
||||
banner("5. Nested + return values") |
||||
---------------------------------------------------------------------- |
||||
-- task.join itself yields, so it composes: a joined task can join again. |
||||
local outer = task.join( |
||||
function() |
||||
local x, y = task.join( |
||||
function() os.sleep(0.05); return 21 end, |
||||
function() os.sleep(0.05); return 21 end |
||||
) |
||||
return x + y |
||||
end, |
||||
function() os.sleep(0.10); return "sibling" end |
||||
) |
||||
print("nested join result:", outer) -- 42 |
||||
|
||||
print("\nAll demos finished.") |
||||
@ -0,0 +1,46 @@ |
||||
use futures::future::join_all; |
||||
use mlua::prelude::{LuaResult, LuaValue}; |
||||
use mlua::{Function, Lua, MultiValue, Variadic}; |
||||
|
||||
/// Proof-of-concept concurrency primitive built on Lua coroutines + tokio.
|
||||
///
|
||||
/// `task.join(f1, f2, ...)` runs each function as its own Lua coroutine and
|
||||
/// drives them *concurrently* on the tokio runtime, returning each one's first
|
||||
/// result positionally once all have finished. Because async stdlib calls like
|
||||
/// `os.sleep` yield to the tokio reactor rather than blocking the OS thread,
|
||||
/// sibling coroutines make progress while one is sleeping — so N tasks that each
|
||||
/// sleep T seconds finish in ~T, not ~N*T.
|
||||
///
|
||||
/// Note: the coroutines are NOT `tokio::spawn`ed (the Lua state is `!Send`);
|
||||
/// they are polled cooperatively on the current thread via `join_all`. The
|
||||
/// concurrency comes from the reactor, not from extra threads.
|
||||
pub(super) fn install(lua: &Lua) -> LuaResult<()> { |
||||
let task = lua.create_table()?; |
||||
|
||||
task.raw_set( |
||||
"join", |
||||
lua.create_async_function(|lua, funcs: Variadic<Function>| async move { |
||||
// Wrap each function in its own coroutine and turn it into a future.
|
||||
let mut threads = Vec::with_capacity(funcs.len()); |
||||
for f in funcs.iter() { |
||||
let thread = lua.create_thread(f.clone())?; |
||||
threads.push(thread.into_async::<LuaValue>(())); |
||||
} |
||||
let threads: Vec<_> = threads.into_iter().collect::<LuaResult<_>>()?; |
||||
|
||||
// Poll them all concurrently. The await point is where the tokio
|
||||
// reactor gets to interleave the sleeping coroutines.
|
||||
let results = join_all(threads).await; |
||||
|
||||
// Collect first-return-values positionally; propagate the first error.
|
||||
let mut out = Vec::with_capacity(results.len()); |
||||
for r in results { |
||||
out.push(r?); |
||||
} |
||||
Ok(MultiValue::from_vec(out)) |
||||
})?, |
||||
)?; |
||||
|
||||
lua.globals().raw_set("task", task)?; |
||||
Ok(()) |
||||
} |
||||
Loading…
Reference in new issue