From 4956e44e6087687398cbd3e7401ba54805739b0d Mon Sep 17 00:00:00 2001 From: hanez Date: Mon, 30 Mar 2026 14:53:11 +0200 Subject: [PATCH] Added some await pattern and added a scheduler to the stdlib. Broken! (0.39.13) --- CMakeLists.txt | 2 +- docs/asyncio.md | 78 +++++++++++++++++++ docs/examples/README.md | 1 + docs/examples/net/httpserver.md | 2 +- examples/io/await_http_client.fun | 111 ++++++++++++++++++++++++++++ examples/net/http_static_server.fun | 2 +- lib/async/scheduler.fun | 111 ++++++++++++++++++++++++++++ 7 files changed, 304 insertions(+), 3 deletions(-) create mode 100644 examples/io/await_http_client.fun create mode 100644 lib/async/scheduler.fun diff --git a/CMakeLists.txt b/CMakeLists.txt index 039818c..d213914 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -1,5 +1,5 @@ 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_REQUIRED ON) diff --git a/docs/asyncio.md b/docs/asyncio.md index fefb269..cace323 100644 --- a/docs/asyncio.md +++ b/docs/asyncio.md @@ -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. +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 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/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_cgi.fun — Server variant that dispatches CGI-like handlers - 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 ``` +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 + +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 - Always close descriptors with sock_close(fd) when finished or on error paths. diff --git a/docs/examples/README.md b/docs/examples/README.md index 9fa5fdd..32ca811 100644 --- a/docs/examples/README.md +++ b/docs/examples/README.md @@ -159,6 +159,7 @@ These require optional external libraries. See docs/external/. ## IO (./examples/io) - 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 - file_io.fun, read_write_file.fun — file operations - word_count.fun — classic WC example diff --git a/docs/examples/net/httpserver.md b/docs/examples/net/httpserver.md index 48ec535..cd4202e 100644 --- a/docs/examples/net/httpserver.md +++ b/docs/examples/net/httpserver.md @@ -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 - Run: - `./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) - File: `examples/net/http_server.fun` diff --git a/examples/io/await_http_client.fun b/examples/io/await_http_client.fun new file mode 100644 index 0000000..9a57a88 --- /dev/null +++ b/examples/io/await_http_client.fun @@ -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 + +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 +*/ diff --git a/examples/net/http_static_server.fun b/examples/net/http_static_server.fun index d7c603d..7a6057c 100755 --- a/examples/net/http_static_server.fun +++ b/examples/net/http_static_server.fun @@ -13,7 +13,7 @@ #include -port = 8088 +port = 8080 srv = TcpServer(port, 10) if (srv.listen() <= 0) print("HTTP static server: listen failed on :" + to_string(port)) diff --git a/lib/async/scheduler.fun b/lib/async/scheduler.fun new file mode 100644 index 0000000..db7bf59 --- /dev/null +++ b/lib/async/scheduler.fun @@ -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