Added some await pattern and added a scheduler to the stdlib. Broken! (0.39.13)
This commit is contained in:
parent
ffdb9fffcb
commit
4956e44e60
7 changed files with 304 additions and 3 deletions
|
|
@ -1,5 +1,5 @@
|
||||||
cmake_minimum_required(VERSION 3.10)
|
cmake_minimum_required(VERSION 3.10)
|
||||||
project(fun VERSION 0.39.12 LANGUAGES C)
|
project(fun VERSION 0.39.13 LANGUAGES C)
|
||||||
|
|
||||||
set(CMAKE_C_STANDARD 99)
|
set(CMAKE_C_STANDARD 99)
|
||||||
set(CMAKE_C_STANDARD_REQUIRED ON)
|
set(CMAKE_C_STANDARD_REQUIRED ON)
|
||||||
|
|
|
||||||
|
|
@ -13,6 +13,8 @@ This lets a single Fun script handle many concurrent connections efficiently wit
|
||||||
|
|
||||||
There is no special syntax (like async/await) — you compose ordinary control flow with a few focused opcodes and stdlib functions.
|
There is no special syntax (like async/await) — you compose ordinary control flow with a few focused opcodes and stdlib functions.
|
||||||
|
|
||||||
|
However, for more ergonomic, "await-like" workflows without changing the VM, a tiny cooperative scheduler is provided in the stdlib at lib/async/scheduler.fun. It lets you write small step functions that advance per tick and use await_read/await_write wrappers for readability.
|
||||||
|
|
||||||
## Building blocks
|
## Building blocks
|
||||||
|
|
||||||
Core helpers wired into the VM (see src/vm/os/*):
|
Core helpers wired into the VM (see src/vm/os/*):
|
||||||
|
|
@ -93,6 +95,7 @@ Because Fun keeps the primitives low-level and explicit, you can build simple co
|
||||||
## Examples in the repository
|
## Examples in the repository
|
||||||
|
|
||||||
- examples/io/async_http_client.fun — Minimal HTTP GET over non-blocking TCP using fd_poll_* helpers
|
- examples/io/async_http_client.fun — Minimal HTTP GET over non-blocking TCP using fd_poll_* helpers
|
||||||
|
- examples/io/await_http_client.fun — Same goal, but written using lib/async/scheduler.fun with await_* helpers
|
||||||
- examples/net/http_mt_server.fun — Multi-tenant HTTP server scaffold (compare patterns for concurrency)
|
- examples/net/http_mt_server.fun — Multi-tenant HTTP server scaffold (compare patterns for concurrency)
|
||||||
- examples/net/http_mt_server_cgi.fun — Server variant that dispatches CGI-like handlers
|
- examples/net/http_mt_server_cgi.fun — Server variant that dispatches CGI-like handlers
|
||||||
- lib/net/http_cgi_server.fun — Library helpers used by the server examples
|
- lib/net/http_cgi_server.fun — Library helpers used by the server examples
|
||||||
|
|
@ -107,6 +110,81 @@ If installed system-wide, just:
|
||||||
fun /usr/share/fun/examples/io/async_http_client.fun
|
fun /usr/share/fun/examples/io/async_http_client.fun
|
||||||
```
|
```
|
||||||
|
|
||||||
|
Or to try the await-style client using the cooperative scheduler:
|
||||||
|
```
|
||||||
|
FUN_LIB_DIR=./lib ./build/fun examples/io/await_http_client.fun
|
||||||
|
```
|
||||||
|
|
||||||
|
## Cooperative scheduler helpers (library-level)
|
||||||
|
|
||||||
|
The file lib/async/scheduler.fun provides a minimal cooperative scheduler built on the existing primitives. There is no VM-level suspension: each task is a small state machine advanced one step per tick. API summary:
|
||||||
|
|
||||||
|
- task_spawn(step_fn, state_map) → task_handle
|
||||||
|
- Registers a task. step_fn is a function that takes a Map state; mutate state and set state.done = 1 when complete.
|
||||||
|
- run_once() → 1
|
||||||
|
- Performs one scheduling tick over all runnable tasks.
|
||||||
|
- run_until_done() → 1
|
||||||
|
- Repeats run_once() with a tiny sleep_ms(1) until all tasks finish.
|
||||||
|
- await_read(fd, timeout_ms) → int
|
||||||
|
- Wrapper over fd_poll_read; returns 1 if readable, 0 on timeout/EOF, -1 on error.
|
||||||
|
- await_write(fd, timeout_ms) → int
|
||||||
|
- Wrapper over fd_poll_write; returns 1 if writable, 0 on timeout, -1 on error.
|
||||||
|
- yield() → 1
|
||||||
|
- No-op helper to make intent explicit in step functions.
|
||||||
|
- async_sleep_mark(state, ms) → 1
|
||||||
|
- Mark the task to be skipped for roughly ms milliseconds; cleared automatically when it wakes.
|
||||||
|
|
||||||
|
Example skeleton using the scheduler:
|
||||||
|
```
|
||||||
|
#include <async/scheduler.fun>
|
||||||
|
|
||||||
|
fun my_task_step(t)
|
||||||
|
if (t.phase == nil)
|
||||||
|
t.phase = 0
|
||||||
|
|
||||||
|
if (t.phase == 0)
|
||||||
|
t.fd = tcp_connect("example.org", 80)
|
||||||
|
if (t.fd == 0)
|
||||||
|
t.done = 1
|
||||||
|
return
|
||||||
|
fd_set_nonblock(t.fd, 1)
|
||||||
|
t.phase = 1
|
||||||
|
return
|
||||||
|
|
||||||
|
if (t.phase == 1)
|
||||||
|
if (await_write(t.fd, 50) == 1)
|
||||||
|
sock_send(t.fd, "GET / HTTP/1.1\r\nHost: example.org\r\n\r\n")
|
||||||
|
t.buf = ""
|
||||||
|
t.phase = 2
|
||||||
|
return
|
||||||
|
|
||||||
|
if (t.phase == 2)
|
||||||
|
rd = await_read(t.fd, 100)
|
||||||
|
if (rd < 0)
|
||||||
|
t.done = 1
|
||||||
|
return
|
||||||
|
if (rd == 0)
|
||||||
|
// try a small read to detect EOF
|
||||||
|
data = sock_recv(t.fd, 4096)
|
||||||
|
if (len(data) == 0)
|
||||||
|
t.done = 1
|
||||||
|
else
|
||||||
|
t.buf = t.buf + data
|
||||||
|
return
|
||||||
|
// readable
|
||||||
|
data = sock_recv(t.fd, 4096)
|
||||||
|
if (len(data) == 0)
|
||||||
|
t.done = 1
|
||||||
|
else
|
||||||
|
t.buf = t.buf + data
|
||||||
|
return
|
||||||
|
|
||||||
|
task = task_spawn(my_task_step, {})
|
||||||
|
run_until_done()
|
||||||
|
```
|
||||||
|
|
||||||
|
This approach is 100% compatible with current runtimes and serves as a stepping stone towards potential future VM-level async/await opcodes.
|
||||||
|
|
||||||
## Error handling and cleanup
|
## Error handling and cleanup
|
||||||
|
|
||||||
- Always close descriptors with sock_close(fd) when finished or on error paths.
|
- Always close descriptors with sock_close(fd) when finished or on error paths.
|
||||||
|
|
|
||||||
|
|
@ -159,6 +159,7 @@ These require optional external libraries. See docs/external/.
|
||||||
|
|
||||||
## IO (./examples/io)
|
## IO (./examples/io)
|
||||||
- async_http_client.fun — non‑blocking HTTP client
|
- async_http_client.fun — non‑blocking HTTP client
|
||||||
|
- await_http_client.fun — non‑blocking HTTP client using lib/async/scheduler.fun (await-style helpers)
|
||||||
- csv_reader.fun — parse CSV files
|
- csv_reader.fun — parse CSV files
|
||||||
- file_io.fun, read_write_file.fun — file operations
|
- file_io.fun, read_write_file.fun — file operations
|
||||||
- word_count.fun — classic WC example
|
- word_count.fun — classic WC example
|
||||||
|
|
|
||||||
|
|
@ -35,7 +35,7 @@ See docs/stdlib.md → net/ for a quick index of these modules.
|
||||||
- Sends the response with `sock_send()` and closes the client
|
- Sends the response with `sock_send()` and closes the client
|
||||||
- Run:
|
- Run:
|
||||||
- `./examples/net/http_static_server.fun`
|
- `./examples/net/http_static_server.fun`
|
||||||
- Open http://127.0.0.1:8088/
|
- Open http://127.0.0.1:8080/
|
||||||
|
|
||||||
## Example: http_server.fun (static + simple .fun CGI)
|
## Example: http_server.fun (static + simple .fun CGI)
|
||||||
- File: `examples/net/http_server.fun`
|
- File: `examples/net/http_server.fun`
|
||||||
|
|
|
||||||
111
examples/io/await_http_client.fun
Normal file
111
examples/io/await_http_client.fun
Normal file
|
|
@ -0,0 +1,111 @@
|
||||||
|
#!/usr/bin/env fun
|
||||||
|
|
||||||
|
/*
|
||||||
|
* Async-style HTTP GET using lib/async/scheduler.fun helpers
|
||||||
|
*
|
||||||
|
* This demonstrates building an "await-like" workflow without VM-level
|
||||||
|
* async/await. We use a cooperative task that advances a small state
|
||||||
|
* machine each tick, calling await_write/await_read to probe readiness.
|
||||||
|
*/
|
||||||
|
|
||||||
|
#include <async/scheduler.fun>
|
||||||
|
|
||||||
|
host = "example.org"
|
||||||
|
port = 80
|
||||||
|
|
||||||
|
// Task state and step function
|
||||||
|
fun http_get_step(t)
|
||||||
|
// phases: 0=connect, 1=set nonblock, 2=send, 3=recv, 4=done
|
||||||
|
if (t.phase == nil)
|
||||||
|
t.phase = 0
|
||||||
|
|
||||||
|
if (t.phase == 0)
|
||||||
|
t.fd = tcp_connect(host, port)
|
||||||
|
if (t.fd == 0)
|
||||||
|
print("connect failed")
|
||||||
|
t.done = 1
|
||||||
|
return
|
||||||
|
t.phase = 1
|
||||||
|
return
|
||||||
|
|
||||||
|
if (t.phase == 1)
|
||||||
|
ok = fd_set_nonblock(t.fd, 1)
|
||||||
|
if (ok == 0)
|
||||||
|
print("failed to set nonblocking")
|
||||||
|
sock_close(t.fd)
|
||||||
|
t.done = 1
|
||||||
|
return
|
||||||
|
t.req = "GET / HTTP/1.1\r\nHost: " + host + "\r\nConnection: close\r\n\r\n"
|
||||||
|
t.sent = 0
|
||||||
|
t.phase = 2
|
||||||
|
return
|
||||||
|
|
||||||
|
if (t.phase == 2)
|
||||||
|
// send request in chunks when writable
|
||||||
|
if (t.sent < len(t.req))
|
||||||
|
// short timeout to keep loop responsive
|
||||||
|
wr = await_write(t.fd, 50)
|
||||||
|
if (wr < 0)
|
||||||
|
print("poll write error")
|
||||||
|
sock_close(t.fd)
|
||||||
|
t.done = 1
|
||||||
|
return
|
||||||
|
if (wr == 0)
|
||||||
|
// not yet writable; yield to others
|
||||||
|
return
|
||||||
|
chunk = substr(t.req, t.sent, len(t.req) - t.sent)
|
||||||
|
n = sock_send(t.fd, chunk)
|
||||||
|
if (n < 0)
|
||||||
|
print("send error")
|
||||||
|
sock_close(t.fd)
|
||||||
|
t.done = 1
|
||||||
|
return
|
||||||
|
t.sent = t.sent + n
|
||||||
|
return
|
||||||
|
// all sent; switch to receive
|
||||||
|
t.buf = ""
|
||||||
|
t.phase = 3
|
||||||
|
return
|
||||||
|
|
||||||
|
if (t.phase == 3)
|
||||||
|
rd = await_read(t.fd, 100)
|
||||||
|
if (rd < 0)
|
||||||
|
print("poll read error")
|
||||||
|
sock_close(t.fd)
|
||||||
|
t.phase = 4
|
||||||
|
t.done = 1
|
||||||
|
return
|
||||||
|
if (rd == 0)
|
||||||
|
// try a small read to detect close, else wait more
|
||||||
|
data = sock_recv(t.fd, 4096)
|
||||||
|
if (len(data) == 0)
|
||||||
|
sock_close(t.fd)
|
||||||
|
t.phase = 4
|
||||||
|
t.done = 1
|
||||||
|
return
|
||||||
|
t.buf = t.buf + data
|
||||||
|
return
|
||||||
|
// readable
|
||||||
|
data = sock_recv(t.fd, 4096)
|
||||||
|
if (len(data) == 0)
|
||||||
|
sock_close(t.fd)
|
||||||
|
t.phase = 4
|
||||||
|
t.done = 1
|
||||||
|
return
|
||||||
|
t.buf = t.buf + data
|
||||||
|
return
|
||||||
|
|
||||||
|
if (t.phase == 4)
|
||||||
|
// print first 200 chars and finish
|
||||||
|
print(substr(t.buf, 0, 200))
|
||||||
|
t.done = 1
|
||||||
|
return
|
||||||
|
|
||||||
|
|
||||||
|
// Spawn task and run
|
||||||
|
task = task_spawn(http_get_step, {})
|
||||||
|
run_until_done()
|
||||||
|
|
||||||
|
/*
|
||||||
|
Expected: prints the beginning of an HTTP response from example.org
|
||||||
|
*/
|
||||||
|
|
@ -13,7 +13,7 @@
|
||||||
|
|
||||||
#include <io/socket.fun>
|
#include <io/socket.fun>
|
||||||
|
|
||||||
port = 8088
|
port = 8080
|
||||||
srv = TcpServer(port, 10)
|
srv = TcpServer(port, 10)
|
||||||
if (srv.listen() <= 0)
|
if (srv.listen() <= 0)
|
||||||
print("HTTP static server: listen failed on :" + to_string(port))
|
print("HTTP static server: listen failed on :" + to_string(port))
|
||||||
|
|
|
||||||
111
lib/async/scheduler.fun
Normal file
111
lib/async/scheduler.fun
Normal file
|
|
@ -0,0 +1,111 @@
|
||||||
|
/*
|
||||||
|
* Cooperative asyncio helpers (library-level) for Fun
|
||||||
|
*
|
||||||
|
* This module provides a tiny, user-space scheduler built on top of the
|
||||||
|
* existing non-blocking FD helpers (fd_set_nonblock, fd_poll_read, fd_poll_write).
|
||||||
|
*
|
||||||
|
* There is no VM-level suspension: tasks must be written as small step
|
||||||
|
* functions that advance a state machine a little and then return so the
|
||||||
|
* scheduler can run other tasks. Use await_read/await_write to probe IO
|
||||||
|
* readiness with short timeouts and yield() to voluntarily give up control.
|
||||||
|
*
|
||||||
|
* API (minimal):
|
||||||
|
* - task_spawn(step_fn, state_map) -> task_handle (a Map)
|
||||||
|
* - run_until_done() -> runs all spawned tasks until every task has state.done == 1
|
||||||
|
* - run_once() -> performs one scheduling tick over all tasks
|
||||||
|
* - await_read(fd, timeout_ms) -> 1 if readable else 0; -1 on error
|
||||||
|
* - await_write(fd, timeout_ms) -> 1 if writable else 0; -1 on error
|
||||||
|
* - yield() -> 1 (hint to return to scheduler)
|
||||||
|
* - async_sleep_mark(state, ms) -> marks state to sleep for ms; scheduler will skip until wake
|
||||||
|
*/
|
||||||
|
|
||||||
|
__tasks = []
|
||||||
|
|
||||||
|
fun __now_ms()
|
||||||
|
return time_now_ms()
|
||||||
|
|
||||||
|
/* Spawn a cooperative task.
|
||||||
|
* step_fn: function taking a single Map parameter (the task object itself)
|
||||||
|
* state_map: optional Map to seed task fields; may contain 'done' flag initially 0
|
||||||
|
*/
|
||||||
|
fun task_spawn(step_fn, state_map)
|
||||||
|
t = {}
|
||||||
|
t.fn = step_fn
|
||||||
|
if (typeof(state_map) == "Map")
|
||||||
|
/* shallow copy of provided state (no "for in" syntax in Fun parser) */
|
||||||
|
_ks = keys(state_map)
|
||||||
|
_i = 0
|
||||||
|
_n = len(_ks)
|
||||||
|
while (_i < _n)
|
||||||
|
_k = _ks[_i]
|
||||||
|
t[_k] = state_map[_k]
|
||||||
|
_i = _i + 1
|
||||||
|
/* normalize done flag to 0/1 */
|
||||||
|
_d = to_number(t.done)
|
||||||
|
if (_d == 0 || _d == 1)
|
||||||
|
t.done = _d
|
||||||
|
else
|
||||||
|
t.done = 0
|
||||||
|
t._sleep_until = 0
|
||||||
|
push(__tasks, t)
|
||||||
|
return t
|
||||||
|
|
||||||
|
/* Mark the task state to sleep (skip execution) for ms milliseconds */
|
||||||
|
fun async_sleep_mark(state, ms)
|
||||||
|
state._sleep_until = __now_ms() + to_number(ms)
|
||||||
|
return 1
|
||||||
|
|
||||||
|
/* Voluntary cooperative yield helper (avoid reserved keyword name) */
|
||||||
|
fun co_yield()
|
||||||
|
return 1
|
||||||
|
|
||||||
|
/* Probe for readability. Returns 1 if readable, 0 on timeout or EOF, -1 on error. */
|
||||||
|
fun await_read(fd, timeout_ms)
|
||||||
|
return fd_poll_read(to_number(fd), to_number(timeout_ms))
|
||||||
|
|
||||||
|
/* Probe for writability. Returns 1 if writable, 0 on timeout, -1 on error. */
|
||||||
|
fun await_write(fd, timeout_ms)
|
||||||
|
return fd_poll_write(to_number(fd), to_number(timeout_ms))
|
||||||
|
|
||||||
|
/* Execute one scheduling tick: iterate over all tasks and invoke their step
|
||||||
|
* function if not done and if not sleeping. Removes finished tasks at the end.
|
||||||
|
*/
|
||||||
|
fun run_once()
|
||||||
|
i = 0
|
||||||
|
n = len(__tasks)
|
||||||
|
now = __now_ms()
|
||||||
|
while (i < n)
|
||||||
|
t = __tasks[i]
|
||||||
|
if (typeof(t) != "Map")
|
||||||
|
i = i + 1
|
||||||
|
continue
|
||||||
|
if (to_number(t.done) != 1)
|
||||||
|
if (to_number(t._sleep_until) > now)
|
||||||
|
i = i + 1
|
||||||
|
continue
|
||||||
|
else
|
||||||
|
t._sleep_until = 0
|
||||||
|
/* Call step function if present */
|
||||||
|
if (t.fn != nil)
|
||||||
|
t.fn(t)
|
||||||
|
i = i + 1
|
||||||
|
|
||||||
|
/* Rebuild task list keeping only unfinished tasks (single pass) */
|
||||||
|
tmp = []
|
||||||
|
j = 0
|
||||||
|
m = len(__tasks)
|
||||||
|
while (j < m)
|
||||||
|
tt = __tasks[j]
|
||||||
|
if (typeof(tt) != "Map" || to_number(tt.done) != 1)
|
||||||
|
push(tmp, tt)
|
||||||
|
j = j + 1
|
||||||
|
__tasks = tmp
|
||||||
|
return 1
|
||||||
|
|
||||||
|
/* Run until all tasks report done == 1. To avoid busy spinning, sleep 1ms per tick. */
|
||||||
|
fun run_until_done()
|
||||||
|
while (len(__tasks) > 0)
|
||||||
|
run_once()
|
||||||
|
/* Tiny pause to reduce CPU when nothing is ready */
|
||||||
|
sleep_ms(1)
|
||||||
|
return 1
|
||||||
Loading…
Add table
Add a link
Reference in a new issue