Skip to main content

seq_runtime/
udp.rs

1//! UDP Socket Operations for Seq
2//!
3//! Provides non-blocking UDP datagram operations using May's
4//! coroutine-aware I/O. `udp.receive-from` yields the strand
5//! while waiting for a datagram instead of blocking the OS thread.
6//!
7//! These functions are exported with C ABI for LLVM codegen.
8//!
9//! ## Payloads are byte-clean
10//!
11//! Datagrams carry whatever bytes the wire delivered — no UTF-8
12//! validation. Binary protocols (DNS records, NTP packets, OSC
13//! int32 / float32 arguments, multicast TLV, MessagePack-over-UDP)
14//! round-trip through `udp.send-to` / `udp.receive-from` byte for
15//! byte. See `docs/design/STRING_BYTE_CLEANLINESS.md` for the
16//! `SeqString` design that makes this possible.
17
18use crate::stack::{Stack, pop, push};
19use crate::value::Value;
20use may::net::UdpSocket;
21use std::net::{IpAddr, SocketAddr};
22use std::sync::{Arc, Mutex};
23
24// Maximum number of concurrent sockets to prevent unbounded growth.
25// Same cap as `tcp.rs`.
26const MAX_SOCKETS: usize = 10_000;
27
28// Maximum bytes to read per datagram.
29//
30// UDP datagrams are protocol-capped at 65,507 bytes for IPv4 (the
31// `udp.length` header is 16-bit, minus IP+UDP headers), and 65,535
32// for IPv6 base-headered datagrams. We use the next power of two
33// (65,536) as the receive buffer size — anything larger cannot
34// arrive on the wire, so allocating more would be pure waste.
35//
36// This intentionally diverges from `tcp.rs`'s 1 MB cap, which makes
37// sense for streaming reads but not for one-datagram-per-call recv.
38const MAX_READ_SIZE: usize = 65_536;
39
40// Socket registry with ID reuse via free list.
41//
42// Slots hold `Arc<UdpSocket>` rather than the socket directly. Reasons:
43//
44// - `may::net::UdpSocket`'s I/O methods (`send_to`, `recv_from`,
45//   `local_addr`) all take `&self`, so multiple `Arc` clones across
46//   strands are safe without any further synchronisation.
47//
48// - I/O paths clone the `Arc` out of the registry under the lock, then
49//   drop the lock before doing the syscall. This is what the previous
50//   `take()`-and-restore pattern was reaching for, but with `Arc` we
51//   avoid the close-vs-in-flight race: `close` simply sets the slot to
52//   `None` (and frees the id) regardless of whether other strands
53//   currently hold an `Arc` clone. The in-flight strand's clone keeps
54//   the OS socket alive until its `recv_from` / `send_to` returns; the
55//   OS-level close only happens when the last `Arc` drops.
56//
57// - The id bookkeeping is now correct under all races: every successful
58//   `close` pushes the id to `free_ids`, even if the slot was being
59//   used for I/O.
60//
61// `tcp.rs` keeps the take-and-restore pattern because `TcpStream::read`
62// is `&mut self` — multiple strands cannot share a TcpStream the same
63// way. UDP's `&self`-only API is what makes the cleaner shape possible.
64struct SocketRegistry<T> {
65    sockets: Vec<Option<Arc<T>>>,
66    free_ids: Vec<usize>,
67}
68
69impl<T> SocketRegistry<T> {
70    const fn new() -> Self {
71        Self {
72            sockets: Vec::new(),
73            free_ids: Vec::new(),
74        }
75    }
76
77    fn allocate(&mut self, socket: T) -> Result<i64, &'static str> {
78        let socket = Arc::new(socket);
79        if let Some(id) = self.free_ids.pop() {
80            self.sockets[id] = Some(socket);
81            return Ok(id as i64);
82        }
83        if self.sockets.len() >= MAX_SOCKETS {
84            return Err("Maximum socket limit reached");
85        }
86        let id = self.sockets.len();
87        self.sockets.push(Some(socket));
88        Ok(id as i64)
89    }
90
91    /// Clone the `Arc` out of the slot so the caller can do I/O after
92    /// dropping the registry lock. Returns `None` if the slot is empty
93    /// (handle invalid, out of range, or already closed).
94    fn checkout(&self, id: usize) -> Option<Arc<T>> {
95        self.sockets.get(id).and_then(|slot| slot.clone())
96    }
97
98    /// Drop the slot's `Arc`. Returns whether the slot held a socket
99    /// (i.e. whether the close had any effect). Idempotent: a second
100    /// close on the same id returns `false`. Independent of any
101    /// in-flight I/O — those strands hold their own `Arc` clones.
102    fn free(&mut self, id: usize) -> bool {
103        if let Some(slot) = self.sockets.get_mut(id)
104            && slot.is_some()
105        {
106            *slot = None;
107            self.free_ids.push(id);
108            return true;
109        }
110        false
111    }
112}
113
114static SOCKETS: Mutex<SocketRegistry<UdpSocket>> = Mutex::new(SocketRegistry::new());
115
116/// Bind a UDP socket to a local port.
117///
118/// Stack effect: ( port -- socket bound-port Bool )
119///
120/// Binds to `0.0.0.0:port`. `port=0` lets the OS pick a free port; the
121/// returned `bound-port` is the actual bound port (equal to `port` if
122/// non-zero). On failure pushes `(0, 0, false)`.
123///
124/// # Safety
125/// Stack must have an Int (port) on top.
126#[unsafe(no_mangle)]
127pub unsafe extern "C" fn patch_seq_udp_bind(stack: Stack) -> Stack {
128    unsafe {
129        let (stack, port_val) = pop(stack);
130        let port = match port_val {
131            Value::Int(p) => p,
132            _ => return push_bind_failure(stack),
133        };
134
135        if !(0..=65535).contains(&port) {
136            return push_bind_failure(stack);
137        }
138
139        let addr = format!("0.0.0.0:{}", port);
140        let socket = match UdpSocket::bind(&addr) {
141            Ok(s) => s,
142            Err(_) => return push_bind_failure(stack),
143        };
144
145        // Capture the actual bound port before the registry takes ownership.
146        let bound_port = match socket.local_addr() {
147            Ok(addr) => addr.port() as i64,
148            Err(_) => return push_bind_failure(stack),
149        };
150
151        let mut sockets = SOCKETS.lock().unwrap();
152        match sockets.allocate(socket) {
153            Ok(socket_id) => {
154                let stack = push(stack, Value::Int(socket_id));
155                let stack = push(stack, Value::Int(bound_port));
156                push(stack, Value::Bool(true))
157            }
158            Err(_) => push_bind_failure(stack),
159        }
160    }
161}
162
163unsafe fn push_bind_failure(stack: Stack) -> Stack {
164    unsafe {
165        let stack = push(stack, Value::Int(0));
166        let stack = push(stack, Value::Int(0));
167        push(stack, Value::Bool(false))
168    }
169}
170
171/// Send a datagram to a host:port from a bound UDP socket.
172///
173/// Stack effect: ( bytes host port socket -- Bool )
174///
175/// Pops `socket`, `port`, `host`, `bytes` (in that order, so `bytes`
176/// is below all of them on entry). Returns `false` on type mismatch,
177/// invalid socket, address-resolution failure, or send error.
178///
179/// Host resolution goes through `dns::resolve` (the may-aware DNS
180/// worker pool from PR1). Previously this path used
181/// `format!("{host}:{port}")` + may's `ToSocketAddrs`, which silently
182/// called blocking `getaddrinfo` on the calling may carrier whenever
183/// `host` was a DNS name — a latent hazard that PR5 closes. IP
184/// literals still work; they round-trip through the resolver's
185/// numeric-host fast path. If resolution returns multiple addresses
186/// (e.g. localhost → ::1, 127.0.0.1) we try them in order and stop at
187/// the first `send_to` that doesn't error.
188///
189/// # Safety
190/// Stack must have Int (socket), Int (port), String (host),
191/// String (bytes) — top-down — on entry.
192#[unsafe(no_mangle)]
193pub unsafe extern "C" fn patch_seq_udp_send_to(stack: Stack) -> Stack {
194    unsafe {
195        let (stack, socket_val) = pop(stack);
196        // Reject negative ids before the `as usize` cast: a negative
197        // i64 wraps to usize::MAX, which would silently fall through
198        // to a benign `None` lookup. Catching it here is a clearer
199        // signal than the indirect not-found path.
200        let socket_id = match socket_val {
201            Value::Int(id) if id >= 0 => id as usize,
202            _ => return push(stack, Value::Bool(false)),
203        };
204
205        let (stack, port_val) = pop(stack);
206        let port = match port_val {
207            Value::Int(p) if (0..=65535).contains(&p) => p,
208            _ => return push(stack, Value::Bool(false)),
209        };
210
211        let (stack, host_val) = pop(stack);
212        let host = match host_val {
213            Value::String(s) => s,
214            _ => return push(stack, Value::Bool(false)),
215        };
216
217        let (stack, bytes_val) = pop(stack);
218        let bytes = match bytes_val {
219            Value::String(s) => s,
220            _ => return push(stack, Value::Bool(false)),
221        };
222
223        // Clone the Arc<UdpSocket> out of the registry. We don't hold
224        // the lock across the syscall, and a concurrent `close` is
225        // free to drop the registry's slot reference — our clone keeps
226        // the socket alive for the duration of this send.
227        let socket = {
228            let sockets = SOCKETS.lock().unwrap();
229            match sockets.checkout(socket_id) {
230                Some(s) => s,
231                None => return push(stack, Value::Bool(false)),
232            }
233        };
234
235        let hostname = host.as_str_or_empty();
236        if hostname.is_empty() {
237            return push(stack, Value::Bool(false));
238        }
239        let port_u16 = port as u16;
240        // Resolver only emits IP-string forms produced by
241        // `SocketAddr::ip().to_string()`; a parse failure here would
242        // mean a runtime invariant violation — skip rather than panic
243        // the carrier. Same pattern as `tcp::patch_seq_tcp_connect`.
244        let addrs: Vec<IpAddr> = crate::dns::resolve(hostname)
245            .iter()
246            .filter_map(|s| s.parse::<IpAddr>().ok())
247            .collect();
248        if addrs.is_empty() {
249            return push(stack, Value::Bool(false));
250        }
251        // Walk addresses; first send that doesn't error wins. UDP
252        // doesn't have a connect-handshake, so `send_to` failures are
253        // typically address-family mismatches (e.g. v6 IP on a v4-only
254        // socket) — the next address in the list usually clears it.
255        let sent = addrs.iter().any(|ip| {
256            socket
257                .send_to(bytes.as_bytes(), SocketAddr::new(*ip, port_u16))
258                .is_ok()
259        });
260        push(stack, Value::Bool(sent))
261    }
262}
263
264/// Receive one datagram from a UDP socket.
265///
266/// Stack effect: ( socket -- bytes host port Bool )
267///
268/// Yields the strand until a datagram arrives. On failure pushes
269/// `("", "", 0, false)` — invalid socket, recv error, datagram larger
270/// than `MAX_READ_SIZE`, or non-UTF-8 payload (see module doc).
271///
272/// # Safety
273/// Stack must have an Int (socket) on top.
274#[unsafe(no_mangle)]
275pub unsafe extern "C" fn patch_seq_udp_receive_from(stack: Stack) -> Stack {
276    unsafe {
277        let (stack, socket_val) = pop(stack);
278        let socket_id = match socket_val {
279            Value::Int(id) if id >= 0 => id as usize,
280            _ => return push_receive_failure(stack),
281        };
282
283        // Clone the Arc<UdpSocket> out of the registry. The receive
284        // strand keeps the socket alive even if another strand closes
285        // the handle while we're in `recv_from`. When close drops the
286        // registry's clone and ours returns, the OS-level close fires.
287        let socket = {
288            let sockets = SOCKETS.lock().unwrap();
289            match sockets.checkout(socket_id) {
290                Some(s) => s,
291                None => return push_receive_failure(stack),
292            }
293        };
294
295        let mut buffer = vec![0u8; MAX_READ_SIZE];
296        let recv_result = socket.recv_from(&mut buffer);
297
298        let (size, src) = match recv_result {
299            Ok(pair) => pair,
300            Err(_) => return push_receive_failure(stack),
301        };
302
303        buffer.truncate(size);
304        // The payload is whatever bytes the wire delivered. We no longer
305        // require UTF-8 — datagrams for OSC, DNS, NTP, MessagePack, etc.
306        // routinely include high-bit bytes from int32 / float32 / blob
307        // fields. The bytes go into a byte-clean SeqString unchanged.
308        let stack = push(stack, Value::String(crate::seqstring::global_bytes(buffer)));
309        let stack = push(stack, Value::String(src.ip().to_string().into()));
310        let stack = push(stack, Value::Int(src.port() as i64));
311        push(stack, Value::Bool(true))
312    }
313}
314
315unsafe fn push_receive_failure(stack: Stack) -> Stack {
316    unsafe {
317        let stack = push(stack, Value::String("".into()));
318        let stack = push(stack, Value::String("".into()));
319        let stack = push(stack, Value::Int(0));
320        push(stack, Value::Bool(false))
321    }
322}
323
324/// Close a UDP socket and free its handle.
325///
326/// Stack effect: ( socket -- Bool )
327///
328/// Returns `true` if the handle was open (the registry slot held a
329/// socket), `false` if it was already invalid (never allocated, or
330/// previously closed). Idempotent across redundant calls on the same
331/// id.
332///
333/// Concurrent I/O is safe: any strand mid-`send_to` / `recv_from`
334/// holds its own `Arc<UdpSocket>` clone, so closing the registry slot
335/// from another strand only drops the registry's reference. The
336/// in-flight syscall completes; the OS-level close fires when the
337/// last `Arc` is dropped. The id is recycled to the free list as
338/// soon as `close` returns, regardless of any in-flight strand.
339///
340/// # Safety
341/// Stack must have an Int (socket) on top.
342#[unsafe(no_mangle)]
343pub unsafe extern "C" fn patch_seq_udp_close(stack: Stack) -> Stack {
344    unsafe {
345        let (stack, socket_val) = pop(stack);
346        let socket_id = match socket_val {
347            Value::Int(id) if id >= 0 => id as usize,
348            _ => return push(stack, Value::Bool(false)),
349        };
350
351        let mut sockets = SOCKETS.lock().unwrap();
352        let existed = sockets.free(socket_id);
353        push(stack, Value::Bool(existed))
354    }
355}
356
357// Public re-exports with short names for in-module callers — the
358// `tests` submodule below imports them via `use super::*`. The
359// crate-root re-exports in `lib.rs` are the linker-facing aliases.
360pub use patch_seq_udp_bind as udp_bind;
361pub use patch_seq_udp_close as udp_close;
362pub use patch_seq_udp_receive_from as udp_receive_from;
363pub use patch_seq_udp_send_to as udp_send_to;
364
365#[cfg(test)]
366mod tests;