single thread event loop runtime

This commit is contained in:
shoney.arickathil 2026-05-04 23:16:10 +02:00
parent 21b3f1d894
commit 2525eaf39b
11 changed files with 710 additions and 0 deletions

1
Cargo.lock generated
View file

@ -417,6 +417,7 @@ version = "0.1.0"
dependencies = [
"anyhow",
"axum",
"libc",
"serde",
"serde_json",
"tempfile",

View file

@ -19,6 +19,7 @@ serde_json = "1"
tokio = { version = "1", features = ["rt", "macros", "net", "signal", "sync", "time"] }
axum = "0.7"
tower = "0.4"
libc = "0.2" # phase 02 — direct epoll/eventfd/timerfd/signalfd syscalls
[dev-dependencies]
tempfile = "3"

View file

@ -9,6 +9,7 @@ pub mod compile;
pub mod engine;
pub mod lexer;
pub mod parser;
pub mod runtime;
pub mod server;
pub mod token;

View file

@ -0,0 +1,66 @@
//! `eventfd(2)` wrapper — counter semaphore exposed as a file descriptor.
//!
//! Used as the cross-flow wake-up primitive: any code path that needs the
//! event loop to come back and run something writes a `1` to the eventfd,
//! which becomes readable on the loop's next `wait_once`.
//!
//! Ported from `reference/crates/wo-event/src/eventfd.rs` with an added
//! `AsRawFd` impl so callers can drop the fd straight into `EventLoop`.
use std::io;
use std::os::unix::io::{AsRawFd, RawFd};
pub struct EventFd {
fd: RawFd,
}
impl EventFd {
/// Create a non-blocking, close-on-exec eventfd with initial counter 0.
pub fn new() -> io::Result<Self> {
let fd = unsafe { libc::eventfd(0, libc::EFD_NONBLOCK | libc::EFD_CLOEXEC) };
if fd < 0 { return Err(io::Error::last_os_error()); }
Ok(Self { fd })
}
/// Add `val` to the counter. Wakes any waiter when the counter goes 0→non-zero.
pub fn write(&self, val: u64) -> io::Result<()> {
let buf = val.to_ne_bytes();
let ret = unsafe {
libc::write(self.fd, buf.as_ptr() as *const libc::c_void, 8)
};
if ret < 0 { Err(io::Error::last_os_error()) } else { Ok(()) }
}
/// Atomically read and zero the counter. Returns `EAGAIN` if the
/// counter is already 0 (since we set `EFD_NONBLOCK`).
pub fn read(&self) -> io::Result<u64> {
let mut buf = [0u8; 8];
let ret = unsafe {
libc::read(self.fd, buf.as_mut_ptr() as *mut libc::c_void, 8)
};
if ret < 0 { Err(io::Error::last_os_error()) } else { Ok(u64::from_ne_bytes(buf)) }
}
}
impl AsRawFd for EventFd {
fn as_raw_fd(&self) -> RawFd { self.fd }
}
impl Drop for EventFd {
fn drop(&mut self) {
unsafe { libc::close(self.fd) };
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn writes_accumulate() {
let efd = EventFd::new().unwrap();
efd.write(5).unwrap();
efd.write(3).unwrap();
assert_eq!(efd.read().unwrap(), 8);
}
}

View file

@ -0,0 +1,22 @@
//! Single-threaded event loop on Linux kernel primitives.
//!
//! Phase 02 of the runtime plan — see `docs/plan/02-event-loop-epoll.md`.
//! Wraps `epoll`, `eventfd`, `timerfd`, and `signalfd` directly via `libc`,
//! with no `tokio` / `mio` / `nix` involvement. Every fd is registered
//! edge-triggered (`EPOLLET`); the loop reads to `EAGAIN`.
//!
//! Lives alongside the tokio-backed `server` module until phase 04 cuts
//! the binary over.
//!
//! Filename convention mirrors Go's `src/runtime/netpoll_<flavour>.go`;
//! when phase 03 adds io_uring it lands as `netpoll_io_uring.rs` next door.
mod eventfd;
mod netpoll_epoll;
mod signalfd;
mod timerfd;
pub use eventfd::EventFd;
pub use netpoll_epoll::{Event, EventLoop, Interest, Token};
pub use signalfd::SignalFd;
pub use timerfd::TimerFd;

View file

@ -0,0 +1,170 @@
//! `epoll`-backed event loop. Single-threaded, edge-triggered.
//!
//! Ported from `reference/crates/wo-event/src/epoll.rs`. Differences:
//! * `Token` is a newtype rather than a `u64` alias.
//! * `Interest` is a struct exposing `READABLE`, `WRITABLE`, `READ_WRITE`
//! constants, matching the API in `docs/plan/02-event-loop-epoll.md`.
//! * Every registration sets `EPOLLET`; callers must read to `EAGAIN`.
//! * `wait_once` reuses an internal event buffer instead of allocating
//! a fresh `[epoll_event; 64]` per call.
use std::io;
use std::os::unix::io::RawFd;
use std::time::Duration;
/// Caller-assigned identifier for a registered file descriptor.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub struct Token(pub u64);
/// Interest flags for epoll registration.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Interest(u32);
impl Interest {
pub const READABLE: Interest = Interest(libc::EPOLLIN as u32);
pub const WRITABLE: Interest = Interest(libc::EPOLLOUT as u32);
pub const READ_WRITE: Interest = Interest((libc::EPOLLIN | libc::EPOLLOUT) as u32);
fn bits(self) -> u32 { self.0 }
}
/// One readiness event from the loop.
#[derive(Debug, Clone)]
pub struct Event {
token: Token,
pub readable: bool,
pub writable: bool,
pub error: bool,
pub hangup: bool,
}
impl Event {
pub fn token(&self) -> Token { self.token }
}
const EVENT_BUF: usize = 64;
/// Single-threaded event loop built on `epoll_create1` + `epoll_wait`.
pub struct EventLoop {
epoll_fd: RawFd,
events: Vec<libc::epoll_event>,
}
impl EventLoop {
pub fn new() -> io::Result<Self> {
let fd = unsafe { libc::epoll_create1(libc::EPOLL_CLOEXEC) };
if fd < 0 { return Err(io::Error::last_os_error()); }
Ok(Self {
epoll_fd: fd,
events: vec![libc::epoll_event { events: 0, u64: 0 }; EVENT_BUF],
})
}
pub fn register(&self, fd: RawFd, interest: Interest, token: Token) -> io::Result<()> {
self.ctl(libc::EPOLL_CTL_ADD, fd, interest, token)
}
pub fn modify(&self, fd: RawFd, interest: Interest, token: Token) -> io::Result<()> {
self.ctl(libc::EPOLL_CTL_MOD, fd, interest, token)
}
pub fn deregister(&self, fd: RawFd) -> io::Result<()> {
let ret = unsafe {
libc::epoll_ctl(self.epoll_fd, libc::EPOLL_CTL_DEL, fd, std::ptr::null_mut())
};
if ret < 0 { Err(io::Error::last_os_error()) } else { Ok(()) }
}
fn ctl(&self, op: libc::c_int, fd: RawFd, interest: Interest, token: Token) -> io::Result<()> {
let mut ev = libc::epoll_event {
events: interest.bits() | libc::EPOLLET as u32 | libc::EPOLLRDHUP as u32,
u64: token.0,
};
let ret = unsafe { libc::epoll_ctl(self.epoll_fd, op, fd, &mut ev) };
if ret < 0 { Err(io::Error::last_os_error()) } else { Ok(()) }
}
/// Block until at least one fd is ready or `timeout` expires.
/// `None` blocks indefinitely. Returns up to 64 events per call.
pub fn wait_once(&mut self, timeout: Option<Duration>) -> io::Result<Vec<Event>> {
let timeout_ms: i32 = match timeout {
None => -1,
Some(d) => d.as_millis().min(i32::MAX as u128) as i32,
};
let n = unsafe {
libc::epoll_wait(
self.epoll_fd,
self.events.as_mut_ptr(),
self.events.len() as i32,
timeout_ms,
)
};
if n < 0 {
let err = io::Error::last_os_error();
if err.raw_os_error() == Some(libc::EINTR) { return Ok(vec![]); }
return Err(err);
}
Ok((0..n as usize)
.map(|i| {
let bits = self.events[i].events;
Event {
token: Token(self.events[i].u64),
readable: bits & libc::EPOLLIN as u32 != 0,
writable: bits & libc::EPOLLOUT as u32 != 0,
error: bits & libc::EPOLLERR as u32 != 0,
hangup: bits & (libc::EPOLLHUP | libc::EPOLLRDHUP) as u32 != 0,
}
})
.collect())
}
pub fn fd(&self) -> RawFd { self.epoll_fd }
}
impl Drop for EventLoop {
fn drop(&mut self) {
unsafe { libc::close(self.epoll_fd) };
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::runtime::EventFd;
use std::os::unix::io::AsRawFd;
#[test]
fn wait_once_returns_eventfd_token() {
let mut eloop = EventLoop::new().unwrap();
let efd = EventFd::new().unwrap();
eloop.register(efd.as_raw_fd(), Interest::READABLE, Token(42)).unwrap();
efd.write(1).unwrap();
let events = eloop.wait_once(Some(Duration::from_millis(100))).unwrap();
assert_eq!(events.len(), 1);
assert_eq!(events[0].token(), Token(42));
assert!(events[0].readable);
assert_eq!(efd.read().unwrap(), 1);
}
#[test]
fn wait_once_times_out_with_no_events() {
let mut eloop = EventLoop::new().unwrap();
let events = eloop.wait_once(Some(Duration::from_millis(10))).unwrap();
assert!(events.is_empty());
}
#[test]
fn deregister_silences_fd() {
let mut eloop = EventLoop::new().unwrap();
let efd = EventFd::new().unwrap();
eloop.register(efd.as_raw_fd(), Interest::READABLE, Token(1)).unwrap();
eloop.deregister(efd.as_raw_fd()).unwrap();
efd.write(1).unwrap();
let events = eloop.wait_once(Some(Duration::from_millis(10))).unwrap();
assert!(events.is_empty());
}
}

View file

@ -0,0 +1,52 @@
//! `signalfd(2)` wrapper — POSIX signals delivered as fd reads.
//!
//! Replaces `tokio::signal::unix::signal` for graceful shutdown: the loop
//! gets `SIGINT` / `SIGTERM` as just another readable fd it can poll.
//!
//! Blocks the captured signals in the calling thread's mask, so the
//! kernel routes them to the signalfd instead of running default handlers.
//! Ported from `reference/crates/wo-event/src/signalfd.rs`.
use std::io;
use std::os::unix::io::{AsRawFd, RawFd};
pub struct SignalFd {
fd: RawFd,
}
impl SignalFd {
/// Capture `SIGINT` and `SIGTERM`. Both are blocked process-wide.
pub fn new() -> io::Result<Self> {
let mut mask: libc::sigset_t = unsafe { std::mem::zeroed() };
unsafe {
libc::sigemptyset(&mut mask);
libc::sigaddset(&mut mask, libc::SIGINT);
libc::sigaddset(&mut mask, libc::SIGTERM);
let ret = libc::pthread_sigmask(libc::SIG_BLOCK, &mask, std::ptr::null_mut());
if ret != 0 { return Err(io::Error::from_raw_os_error(ret)); }
}
let fd = unsafe { libc::signalfd(-1, &mask, libc::SFD_NONBLOCK | libc::SFD_CLOEXEC) };
if fd < 0 { return Err(io::Error::last_os_error()); }
Ok(Self { fd })
}
/// Read one pending signal. Returns the signal number (e.g. `SIGINT = 2`).
pub fn read(&self) -> io::Result<i32> {
let mut info: libc::signalfd_siginfo = unsafe { std::mem::zeroed() };
let size = std::mem::size_of::<libc::signalfd_siginfo>();
let ret = unsafe {
libc::read(self.fd, &mut info as *mut _ as *mut libc::c_void, size)
};
if ret < 0 { Err(io::Error::last_os_error()) } else { Ok(info.ssi_signo as i32) }
}
}
impl AsRawFd for SignalFd {
fn as_raw_fd(&self) -> RawFd { self.fd }
}
impl Drop for SignalFd {
fn drop(&mut self) {
unsafe { libc::close(self.fd) };
}
}

View file

@ -0,0 +1,102 @@
//! `timerfd_create(2)` wrapper — timers as file descriptors.
//!
//! Both one-shot and periodic timers are armed with `timerfd_settime`. The
//! fd becomes readable when the timer expires; reading drains the
//! expiration count.
//!
//! Ported from `reference/crates/wo-event/src/timerfd.rs`. Adds `oneshot`
//! and `periodic` constructors that match the API in the phase-02 plan.
use std::io;
use std::os::unix::io::{AsRawFd, RawFd};
use std::time::Duration;
pub struct TimerFd {
fd: RawFd,
}
impl TimerFd {
/// Create a disarmed monotonic timer fd (non-blocking, close-on-exec).
pub fn new() -> io::Result<Self> {
let fd = unsafe {
libc::timerfd_create(
libc::CLOCK_MONOTONIC,
libc::TFD_NONBLOCK | libc::TFD_CLOEXEC,
)
};
if fd < 0 { return Err(io::Error::last_os_error()); }
Ok(Self { fd })
}
/// Convenience: a fresh fd armed to fire once after `after`.
pub fn oneshot(after: Duration) -> io::Result<Self> {
let t = Self::new()?;
t.set(after, Duration::ZERO)?;
Ok(t)
}
/// Convenience: a fresh fd that fires every `every` (first tick at +`every`).
pub fn periodic(every: Duration) -> io::Result<Self> {
let t = Self::new()?;
t.set(every, every)?;
Ok(t)
}
/// Arm: fire once after `initial`, then repeat every `interval`.
/// Pass `Duration::ZERO` for `interval` to make it one-shot.
pub fn set(&self, initial: Duration, interval: Duration) -> io::Result<()> {
let spec = libc::itimerspec {
it_interval: timespec(interval),
it_value: timespec(initial),
};
let ret = unsafe {
libc::timerfd_settime(self.fd, 0, &spec, std::ptr::null_mut())
};
if ret < 0 { Err(io::Error::last_os_error()) } else { Ok(()) }
}
/// Read the number of expirations since the previous read.
pub fn read(&self) -> io::Result<u64> {
let mut buf = [0u8; 8];
let ret = unsafe {
libc::read(self.fd, buf.as_mut_ptr() as *mut libc::c_void, 8)
};
if ret < 0 { Err(io::Error::last_os_error()) } else { Ok(u64::from_ne_bytes(buf)) }
}
}
impl AsRawFd for TimerFd {
fn as_raw_fd(&self) -> RawFd { self.fd }
}
impl Drop for TimerFd {
fn drop(&mut self) {
unsafe { libc::close(self.fd) };
}
}
fn timespec(d: Duration) -> libc::timespec {
libc::timespec {
tv_sec: d.as_secs() as libc::time_t,
tv_nsec: d.subsec_nanos() as libc::c_long,
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::runtime::{EventLoop, Interest, Token};
#[test]
fn oneshot_fires_within_window() {
let mut eloop = EventLoop::new().unwrap();
let timer = TimerFd::oneshot(Duration::from_millis(100)).unwrap();
eloop.register(timer.as_raw_fd(), Interest::READABLE, Token(99)).unwrap();
let events = eloop.wait_once(Some(Duration::from_millis(500))).unwrap();
assert!(!events.is_empty(), "expected timer event within 500ms");
assert_eq!(events[0].token(), Token(99));
assert!(timer.read().unwrap() >= 1);
}
}

173
docs/examples/blog/api.rest Normal file
View file

@ -0,0 +1,173 @@
###############################################################################
# blog/api.rest — exercise the `.wo` runtime against this directory.
#
# Start the server first (from repo root):
# cargo run --bin wo -- run docs/examples/blog
#
# Then in VS Code (REST Client extension) or JetBrains (HTTP Client) click
# "Send Request" on each block top to bottom. `# @name foo` lets later blocks
# pick up ids minted by earlier ones.
#
# A more annotated, side-by-side cousin of this file (with the same requests
# but heavier commentary on Stage-3+ stubs) lives at reference/rest/blog.rest.
###############################################################################
@host = http://127.0.0.1:8080
### Runtime info — 200
GET {{host}}/
### Liveness probe — 200 "ok"
GET {{host}}/healthz
###############################################################################
# Author — exposes: list, get, me, subscribe (no create)
###############################################################################
### List authors — 200 [] on a fresh boot (Stage 2 has no startup seeding)
GET {{host}}/api/authors
### Author create not exposed — 405
POST {{host}}/api/authors
Content-Type: application/json
{ "email": "alice@example.com", "handle": "alice", "display": "Alice" }
### /me — Stage 3 session layer; 501
GET {{host}}/api/authors/me
### LIVE subscribe — Stage 3; 501
GET {{host}}/api/authors/live
###############################################################################
# Article — exposes: list, get, create, update, delete, subscribe
###############################################################################
### Create an article — 201
# @name createArticle
POST {{host}}/api/articles
Content-Type: application/json
{
"slug": "hello-writeonce",
"title": "Hello, writeonce",
"author": 1,
"published": true,
"meta": {
"excerpt": "First post on the new runtime.",
"body_md": "# Hi\n\nHello from the `.wo` runtime. The server, the database, and this HTTP API are all one binary.\n"
}
}
### Create a draft — 201
# @name createDraft
POST {{host}}/api/articles
Content-Type: application/json
{
"slug": "second-draft",
"title": "Second Post (draft)",
"author": 1,
"published": false,
"meta": { "excerpt": "", "body_md": "WIP." }
}
### List articles — 200 with 2 rows
GET {{host}}/api/articles
### Get one article — 200
GET {{host}}/api/articles/{{createArticle.response.body.id}}
### PATCH the title — 200
PATCH {{host}}/api/articles/{{createArticle.response.body.id}}
Content-Type: application/json
{ "title": "Hi, writeonce!" }
### PATCH an embedded-doc field (Stage 2 = shallow merge — re-send the whole `meta`)
PATCH {{host}}/api/articles/{{createArticle.response.body.id}}
Content-Type: application/json
{
"meta": { "excerpt": "Updated excerpt.", "body_md": "# Hi\n\nUpdated body." }
}
### Publish the draft — 200 (`on update` trigger that sets published_at is Stage 3+)
PATCH {{host}}/api/articles/{{createDraft.response.body.id}}
Content-Type: application/json
{ "published": true }
### Delete the draft — 204
DELETE {{host}}/api/articles/{{createDraft.response.body.id}}
### Re-fetch the deleted id — 404
GET {{host}}/api/articles/{{createDraft.response.body.id}}
### LIVE subscribe — Stage 3; 501
GET {{host}}/api/articles/live
###############################################################################
# Tag — exposes: list, get, subscribe
###############################################################################
### List tags — 200 []
GET {{host}}/api/tags
### Tag create not exposed — 405
POST {{host}}/api/tags
Content-Type: application/json
{ "slug": "rust", "label": "Rust" }
### LIVE subscribe — Stage 3; 501
GET {{host}}/api/tags/live
###############################################################################
# Comment — exposes: list, get, create, update, delete, subscribe
###############################################################################
### Create a comment — 201
# @name createComment
POST {{host}}/api/comments
Content-Type: application/json
{
"article": {{createArticle.response.body.id}},
"author": 1,
"body": "Nice post. Runs on one binary which is still weird to me."
}
### List comments — 200 with 1 row
GET {{host}}/api/comments
### Get one comment — 200
GET {{host}}/api/comments/{{createComment.response.body.id}}
### Update body — 200 (the `set self.edited_at = now()` trigger lands in Stage 3+)
PATCH {{host}}/api/comments/{{createComment.response.body.id}}
Content-Type: application/json
{ "body": "Edited: really, one binary? Neat." }
### Delete the comment — 204
DELETE {{host}}/api/comments/{{createComment.response.body.id}}
### LIVE subscribe — Stage 3; 501
GET {{host}}/api/comments/live
###############################################################################
# Final state — one updated article, no drafts, no comments.
###############################################################################
### Final article list — 200 with 1 row
GET {{host}}/api/articles
### Final comment list — 200 []
GET {{host}}/api/comments

View file

@ -0,0 +1,122 @@
###############################################################################
# ecommerce/api.rest — exercise the `.wo` runtime against this directory.
#
# Start the server first (from repo root):
# cargo run --bin wo -- run docs/examples/ecommerce
#
# This sample leans on features that land in later stages:
# * orders are minted by `fn checkout(...)` — Stage 3/4
# * customers/products/orders are seeded by `on startup do: seed()` — Stage 3+
# * Order-status lifecycle triggers — Stage 3+
#
# So this file documents what Stage 2 *does* serve: route wiring, empty-list
# reads, 405 for un-exposed methods, and the Stage-3 stubs that respond 501.
# A heavier annotated cousin lives at reference/rest/ecommerce.rest.
###############################################################################
@host = http://127.0.0.1:8080
### Runtime info — 200
GET {{host}}/
### Liveness probe — 200 "ok"
GET {{host}}/healthz
###############################################################################
# Product — exposes: list, get, subscribe (no create — admin seeds inventory)
###############################################################################
### List products — 200 [] (no startup seed yet)
GET {{host}}/api/products
### Product create not exposed — 405
POST {{host}}/api/products
Content-Type: application/json
{
"sku": "SKU-WIDGET",
"name": "Widget",
"price": 1999,
"meta": { "description": "A widget.", "images": [], "attributes": { "colour": "blue" } },
"inventory": { "on_hand": 50, "reserved": 0, "reorder_at": 10 }
}
### Get by id — 404 (nothing exists)
GET {{host}}/api/products/1
### LIVE subscribe — Stage 3; 501
GET {{host}}/api/products/live
###############################################################################
# Customer — exposes: get, me, update, subscribe (no list, no create)
###############################################################################
### List not exposed — 404 (no route at /api/customers at all)
GET {{host}}/api/customers
### Customer create not exposed — 404 (same reason)
POST {{host}}/api/customers
Content-Type: application/json
{ "email": "carol@shop.test", "name": "Carol", "role": "Customer" }
### Get by id — 404 (nothing exists)
GET {{host}}/api/customers/1
### Update by id — 404 (would be 200 if the row existed)
PATCH {{host}}/api/customers/1
Content-Type: application/json
{ "name": "Carol Updated" }
### /me — Stage 3 session layer; 501
GET {{host}}/api/customers/me
### LIVE subscribe — Stage 3; 501
GET {{host}}/api/customers/live
###############################################################################
# Order — exposes: list, get, subscribe (use fn checkout to create)
###############################################################################
### List orders — 200 []
GET {{host}}/api/orders
### Order create not exposed (use fn checkout) — 405
POST {{host}}/api/orders
Content-Type: application/json
{ "customer": 1, "status": "Pending", "line_items": [] }
### LIVE subscribe — the WebSocket the Stage-6 `##ui #admin-orders` board
### will open. Stage 2 returns 501.
GET {{host}}/api/orders/live
###############################################################################
# fn checkout — Stage 3/4 (transactional functions)
#
# Becomes the canonical cross-paradigm ACID test once it lands: one call
# updates the product's inventory doc, inserts an Order row, and creates a
# Purchase graph edge inside one BEGIN ... COMMIT.
# See docs/examples/ecommerce/logic/checkout.wo.
###############################################################################
### Stage 2 — 404 (route not registered yet)
POST {{host}}/api/fn/checkout
Content-Type: application/json
{ "customer": 1, "product": 1, "qty": 2 }
###############################################################################
# Purchase — link type, no `service rest` block
# Edges are created by fn checkout and traversed via Customer.purchased.
###############################################################################
### Purchase list not exposed — 404
GET {{host}}/api/purchases