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::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 let addrs = crate::dns::resolve_to_ips(hostname);
241 if addrs.is_empty() {
242 return push(stack, Value::Bool(false));
243 }
244 // Walk addresses; first send that doesn't error wins. UDP
245 // doesn't have a connect-handshake, so `send_to` failures are
246 // typically address-family mismatches (e.g. v6 IP on a v4-only
247 // socket) — the next address in the list usually clears it.
248 let sent = addrs.iter().any(|ip| {
249 socket
250 .send_to(bytes.as_bytes(), SocketAddr::new(*ip, port_u16))
251 .is_ok()
252 });
253 push(stack, Value::Bool(sent))
254 }
255}
256
257/// Receive one datagram from a UDP socket.
258///
259/// Stack effect: ( socket -- bytes host port Bool )
260///
261/// Yields the strand until a datagram arrives. On failure pushes
262/// `("", "", 0, false)` — invalid socket, recv error, datagram larger
263/// than `MAX_READ_SIZE`, or non-UTF-8 payload (see module doc).
264///
265/// # Safety
266/// Stack must have an Int (socket) on top.
267#[unsafe(no_mangle)]
268pub unsafe extern "C" fn patch_seq_udp_receive_from(stack: Stack) -> Stack {
269 unsafe {
270 let (stack, socket_val) = pop(stack);
271 let socket_id = match socket_val {
272 Value::Int(id) if id >= 0 => id as usize,
273 _ => return push_receive_failure(stack),
274 };
275
276 // Clone the Arc<UdpSocket> out of the registry. The receive
277 // strand keeps the socket alive even if another strand closes
278 // the handle while we're in `recv_from`. When close drops the
279 // registry's clone and ours returns, the OS-level close fires.
280 let socket = {
281 let sockets = SOCKETS.lock().unwrap();
282 match sockets.checkout(socket_id) {
283 Some(s) => s,
284 None => return push_receive_failure(stack),
285 }
286 };
287
288 let mut buffer = vec![0u8; MAX_READ_SIZE];
289 let recv_result = socket.recv_from(&mut buffer);
290
291 let (size, src) = match recv_result {
292 Ok(pair) => pair,
293 Err(_) => return push_receive_failure(stack),
294 };
295
296 buffer.truncate(size);
297 // The payload is whatever bytes the wire delivered. We no longer
298 // require UTF-8 — datagrams for OSC, DNS, NTP, MessagePack, etc.
299 // routinely include high-bit bytes from int32 / float32 / blob
300 // fields. The bytes go into a byte-clean SeqString unchanged.
301 let stack = push(stack, Value::String(crate::seqstring::global_bytes(buffer)));
302 let stack = push(stack, Value::String(src.ip().to_string().into()));
303 let stack = push(stack, Value::Int(src.port() as i64));
304 push(stack, Value::Bool(true))
305 }
306}
307
308unsafe fn push_receive_failure(stack: Stack) -> Stack {
309 unsafe {
310 let stack = push(stack, Value::String("".into()));
311 let stack = push(stack, Value::String("".into()));
312 let stack = push(stack, Value::Int(0));
313 push(stack, Value::Bool(false))
314 }
315}
316
317/// Close a UDP socket and free its handle.
318///
319/// Stack effect: ( socket -- Bool )
320///
321/// Returns `true` if the handle was open (the registry slot held a
322/// socket), `false` if it was already invalid (never allocated, or
323/// previously closed). Idempotent across redundant calls on the same
324/// id.
325///
326/// Concurrent I/O is safe: any strand mid-`send_to` / `recv_from`
327/// holds its own `Arc<UdpSocket>` clone, so closing the registry slot
328/// from another strand only drops the registry's reference. The
329/// in-flight syscall completes; the OS-level close fires when the
330/// last `Arc` is dropped. The id is recycled to the free list as
331/// soon as `close` returns, regardless of any in-flight strand.
332///
333/// # Safety
334/// Stack must have an Int (socket) on top.
335#[unsafe(no_mangle)]
336pub unsafe extern "C" fn patch_seq_udp_close(stack: Stack) -> Stack {
337 unsafe {
338 let (stack, socket_val) = pop(stack);
339 let socket_id = match socket_val {
340 Value::Int(id) if id >= 0 => id as usize,
341 _ => return push(stack, Value::Bool(false)),
342 };
343
344 let mut sockets = SOCKETS.lock().unwrap();
345 let existed = sockets.free(socket_id);
346 push(stack, Value::Bool(existed))
347 }
348}
349
350// Public re-exports with short names for in-module callers — the
351// `tests` submodule below imports them via `use super::*`. The
352// crate-root re-exports in `lib.rs` are the linker-facing aliases.
353pub use patch_seq_udp_bind as udp_bind;
354pub use patch_seq_udp_close as udp_close;
355pub use patch_seq_udp_receive_from as udp_receive_from;
356pub use patch_seq_udp_send_to as udp_send_to;
357
358#[cfg(test)]
359mod tests;