From 900e9ff362a4b245f72b20b8123e1000fd21f8cf Mon Sep 17 00:00:00 2001 From: Charles Sibbald Date: Tue, 3 Nov 2026 23:29:05 +0200 Subject: [PATCH 1/2] feat(server): phase timings and connection levels (feature op-timing) With the op-timing feature the server records, in op_duration_seconds {component="may_minihttp.server", op, outcome} (microscaler-timing): - accept_to_first_byte: connection accepted to its first request byte - read_parse: a request's first byte to the request fully parsed - service: the service call (err when it fails) - response_write: the service returned to the response fully written - request_total: first byte to response fully written and in op_gauge: open_connections, and on Linux accept_backlog and accept_backlog_limit, read from TCP_INFO on the listening socket at scrape time. Together they show whether time a client sees before the application's own timers is accept queueing, a slow request read, or the response write. Off by default; without the feature every hook is an empty inline function. --- Cargo.toml | 9 ++ src/http_server.rs | 57 +++++++---- src/lib.rs | 1 + src/server_timing.rs | 209 +++++++++++++++++++++++++++++++++++++++++ tests/server_timing.rs | 130 +++++++++++++++++++++++++ 5 files changed, 390 insertions(+), 16 deletions(-) create mode 100644 src/server_timing.rs create mode 100644 tests/server_timing.rs diff --git a/Cargo.toml b/Cargo.toml index bf1b51b..83926f2 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -12,6 +12,8 @@ categories = ["web-programming::http-server"] license = "MIT/Apache-2.0" [dependencies] +# Feature `op-timing`. Zero dependencies; branch main, as every other crate in a binary names it (one registry). +microscaler-timing = { git = "https://github.com/microscaler/microscaler-observability.git", branch = "main", optional = true } log = "0.4" itoa = "1" bytes = "1" @@ -54,11 +56,18 @@ flate2 = "1.1" tracing = "0.1" tracing-subscriber = { version = "0.3", default-features = false, features = ["registry", "std"] } +[target.'cfg(target_os = "linux")'.dependencies] +# Feature `op-timing`: TCP_INFO on the listening socket (accept backlog). +libc = { version = "0.2", optional = true } + [target.'cfg(unix)'.dev-dependencies] may_postgres = { git = "https://github.com/Xudong-Huang/may_postgres.git", default-features = false } [features] default = ["may/default"] +# Server phase timings and connection levels in op_duration_seconds / op_gauge (microscaler-timing, see +# src/server_timing.rs). Off by default; BRRTRouter turns it on. +op-timing = ["dep:microscaler-timing", "dep:libc"] # Native HTTP/1.1 client (drop-in for may_http::client). Opt-in for backwards compatibility. client = ["http", "may/io_timeout", "dep:generator", "dep:rustls", "dep:rustls-platform-verifier", "dep:url"] # Typed JSON request/response helpers. Kept optional so the core HTTP transport stays lean. diff --git a/src/http_server.rs b/src/http_server.rs index 6cb224a..d971236 100644 --- a/src/http_server.rs +++ b/src/http_server.rs @@ -57,6 +57,7 @@ pub trait HttpServiceFactory: Send + Sized + 'static { /// return a coroutine that you can cancel it when need to stop the service fn start(self, addr: L) -> io::Result> { let listener = TcpListener::bind(addr)?; + crate::server_timing::listening(&listener); go!( coroutine::Builder::new().name("TcpServerFac".to_owned()), move || { @@ -66,6 +67,7 @@ pub trait HttpServiceFactory: Send + Sized + 'static { use std::os::windows::io::AsRawSocket; for stream in listener.incoming() { let mut stream = t_c!(stream); + let accepted = std::time::Instant::now(); #[cfg(unix)] let id = stream.as_raw_fd() as usize; #[cfg(windows)] @@ -75,16 +77,15 @@ pub trait HttpServiceFactory: Send + Sized + 'static { let _ = stream.set_nodelay(true); let service = self.new_service(id); let builder = may::coroutine::Builder::new().id(id); - go!( - builder, - move || if let Err(e) = each_connection_loop(&mut stream, service) { - // Only log actual errors, not normal client disconnects - if !is_client_disconnect(&e) { - error!("service err = {e:?}"); - } - stream.shutdown(std::net::Shutdown::Both).ok(); + go!(builder, move || if let Err(e) = + each_connection_loop(&mut stream, service, accepted) + { + // Only log actual errors, not normal client disconnects + if !is_client_disconnect(&e) { + error!("service err = {e:?}"); } - ) + stream.shutdown(std::net::Shutdown::Both).ok(); + }) .unwrap(); } } @@ -176,15 +177,22 @@ pub struct HttpServer(pub T); pub struct HttpServerWithHeaders(pub T); #[cfg(unix)] -fn each_connection_loop(stream: &mut TcpStream, service: T) -> io::Result<()> { - each_connection_loop_with_headers::(stream, service) +fn each_connection_loop( + stream: &mut TcpStream, + service: T, + accepted: std::time::Instant, +) -> io::Result<()> { + each_connection_loop_with_headers::(stream, service, accepted) } #[cfg(unix)] fn each_connection_loop_with_headers( stream: &mut TcpStream, mut service: T, + accepted: std::time::Instant, ) -> io::Result<()> { + // op_duration_seconds / op_gauge with the `op-timing` feature; nothing without it. + let mut timing = crate::server_timing::ConnTiming::new(accepted); let mut req_buf = BytesMut::with_capacity(BUF_LEN); let mut rsp_buf = BytesMut::with_capacity(BUF_LEN); let mut body_buf = BytesMut::with_capacity(4096); @@ -194,7 +202,9 @@ fn each_connection_loop_with_headers( let close = Cell::new(false); loop { + let buffered = req_buf.len(); let read_blocked = nonblock_read(stream.inner_mut(), &mut req_buf)?; + timing.read(buffered, req_buf.len()); let mut refused = None; // prepare the requests, we should make sure the request is fully read @@ -213,9 +223,12 @@ fn each_connection_loop_with_headers( break; } }; + timing.parsed(); reserve_buf(&mut rsp_buf); let mut rsp = Response::new(&mut body_buf, &mut kv_buf); - match service.call(req, &mut rsp) { + let called = service.call(req, &mut rsp); + timing.served(called.is_ok()); + match called { Ok(()) => response::encode(rsp, &mut rsp_buf), Err(e) => { eprintln!("service err = {e:?}"); @@ -227,6 +240,7 @@ fn each_connection_loop_with_headers( } } + timing.incomplete(req_buf.len()); if let Some(e) = refused { return refuse(stream, &mut rsp_buf, e); } @@ -236,6 +250,7 @@ fn each_connection_loop_with_headers( // write out the responses nonblock_write(stream.inner_mut(), &mut rsp_buf)?; + timing.written(rsp_buf.is_empty()); // A large response or upload grows these buffers; a keep-alive connection would otherwise // hold that memory until it closes, however small its later requests are. @@ -290,14 +305,19 @@ fn close_after(stream: &mut TcpStream, rsp_buf: &[u8]) -> io::Result<()> { } #[cfg(not(unix))] -fn each_connection_loop(stream: &mut TcpStream, service: T) -> io::Result<()> { - each_connection_loop_with_headers::(stream, service) +fn each_connection_loop( + stream: &mut TcpStream, + service: T, + accepted: std::time::Instant, +) -> io::Result<()> { + each_connection_loop_with_headers::(stream, service, accepted) } #[cfg(not(unix))] fn each_connection_loop_with_headers( stream: &mut TcpStream, mut service: T, + _accepted: std::time::Instant, ) -> io::Result<()> { let mut req_buf = BytesMut::with_capacity(BUF_LEN); let mut rsp_buf = BytesMut::with_capacity(BUF_LEN); @@ -364,18 +384,21 @@ impl HttpServer { /// return a coroutine that you can cancel it when need to stop the service pub fn start(self, addr: L) -> io::Result> { let listener = TcpListener::bind(addr)?; + crate::server_timing::listening(&listener); let service = self.0; go!( coroutine::Builder::new().name("TcpServer".to_owned()), move || { for stream in listener.incoming() { let mut stream = t_c!(stream); + let accepted = std::time::Instant::now(); // Responses are written whole; Nagle would only hold a second write (a streamed // chunk) for the client's delayed ACK. Not fatal if it cannot be set. let _ = stream.set_nodelay(true); let service = service.clone(); go!( - move || if let Err(e) = each_connection_loop(&mut stream, service) { + move || if let Err(e) = each_connection_loop(&mut stream, service, accepted) + { // Only log actual errors, not normal client disconnects if !is_client_disconnect(&e) { error!("service err = {e:?}"); @@ -394,18 +417,20 @@ impl HttpServerW /// return a coroutine that you can cancel it when need to stop the service pub fn start(self, addr: L) -> io::Result> { let listener = TcpListener::bind(addr)?; + crate::server_timing::listening(&listener); let service = self.0; go!( coroutine::Builder::new().name("TcpServer".to_owned()), move || { for stream in listener.incoming() { let mut stream = t_c!(stream); + let accepted = std::time::Instant::now(); // Responses are written whole; Nagle would only hold a second write (a streamed // chunk) for the client's delayed ACK. Not fatal if it cannot be set. let _ = stream.set_nodelay(true); let service = service.clone(); go!(move || if let Err(e) = - each_connection_loop_with_headers::(&mut stream, service) + each_connection_loop_with_headers::(&mut stream, service, accepted) { // Only log actual errors, not normal client disconnects if !is_client_disconnect(&e) { diff --git a/src/lib.rs b/src/lib.rs index 65e7db1..dab474c 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -5,6 +5,7 @@ mod date; mod http_server; mod request; mod response; +mod server_timing; #[cfg(feature = "client")] pub mod client; diff --git a/src/server_timing.rs b/src/server_timing.rs new file mode 100644 index 0000000..488f637 --- /dev/null +++ b/src/server_timing.rs @@ -0,0 +1,209 @@ +//! Server-side phase timings in the shared `op_duration_seconds{component="may_minihttp.server", op, +//! outcome}` histogram (`microscaler-timing`, feature `op-timing`), and the levels beside them in +//! `op_gauge`. A host that renders that family (BRRTRouter does) shows them with no wiring. +//! +//! | op | from | to | +//! |---|---|---| +//! | `accept_to_first_byte` | the listener accepted the connection | its first request byte was read (once per connection: includes the client's send and this coroutine being scheduled) | +//! | `read_parse` | a request's first byte was read | the request was fully read and parsed | +//! | `service` | the request was handed to the service | the service returned (`err` if it failed) | +//! | `response_write` | the service returned | the response was fully written to the socket | +//! | `request_total` | a request's first byte was read | its response was fully written | +//! +//! Gauges: `open_connections`; on Linux, `accept_backlog` (connections the kernel has completed +//! but the server has not accepted yet, read from `TCP_INFO` on the listening socket at scrape +//! time) and `accept_backlog_limit` (the listen backlog). +//! +//! Without the feature every call here is an empty inline function. With it, recording is a few +//! atomic adds and an `Instant::now()`: nothing blocks the coroutine. + +#[cfg(feature = "op-timing")] +mod imp { + use std::time::Instant; + + use microscaler_timing::{Gauge, Outcome, Site}; + + static ACCEPT_TO_FIRST_BYTE: Site = Site::new("may_minihttp.server", "accept_to_first_byte"); + static READ_PARSE: Site = Site::new("may_minihttp.server", "read_parse"); + static SERVICE: Site = Site::new("may_minihttp.server", "service"); + static RESPONSE_WRITE: Site = Site::new("may_minihttp.server", "response_write"); + static REQUEST_TOTAL: Site = Site::new("may_minihttp.server", "request_total"); + static OPEN: Gauge = Gauge::new("may_minihttp.server", "open_connections"); + + /// One connection's timing state; lives in the connection coroutine. + pub(crate) struct ConnTiming { + accepted: Instant, + first_seen: bool, + /// When the first byte of the request being read arrived. + first_byte: Option, + /// The first byte of the request being served. + serving: Option, + call_started: Option, + /// (first byte, service returned) of responses not yet fully written. + pending: Vec<(Instant, Instant)>, + } + + impl ConnTiming { + pub(crate) fn new(accepted: Instant) -> Self { + OPEN.add(1); + Self { + accepted, + first_seen: false, + first_byte: None, + serving: None, + call_started: None, + pending: Vec::new(), + } + } + + /// After a read that took the buffer from `before` to `after` bytes. + #[inline] + pub(crate) fn read(&mut self, before: usize, after: usize) { + if after > before && self.first_byte.is_none() { + let now = Instant::now(); + self.first_byte = Some(now); + if !self.first_seen { + self.first_seen = true; + ACCEPT_TO_FIRST_BYTE.record(Outcome::Ok, now - self.accepted); + } + } + } + + /// A request was parsed (a pipelined request already buffered behind one counts from here). + #[inline] + pub(crate) fn parsed(&mut self) { + let now = Instant::now(); + let first = self.first_byte.take().unwrap_or(now); + READ_PARSE.record(Outcome::Ok, now - first); + self.serving = Some(first); + self.call_started = Some(now); + } + + /// No complete request is buffered; `buffered` bytes of the next one are. + #[inline] + pub(crate) fn incomplete(&mut self, buffered: usize) { + if buffered > 0 && self.first_byte.is_none() { + self.first_byte = Some(Instant::now()); + } + } + + /// The service returned. + #[inline] + pub(crate) fn served(&mut self, ok: bool) { + let now = Instant::now(); + if let Some(started) = self.call_started.take() { + SERVICE.record(if ok { Outcome::Ok } else { Outcome::Err }, now - started); + } + if let Some(first) = self.serving.take() { + self.pending.push((first, now)); + } + } + + /// After a write; `drained` when nothing is left to write. + #[inline] + pub(crate) fn written(&mut self, drained: bool) { + if !drained || self.pending.is_empty() { + return; + } + let now = Instant::now(); + for (first, served) in self.pending.drain(..) { + RESPONSE_WRITE.record(Outcome::Ok, now - served); + REQUEST_TOTAL.record(Outcome::Ok, now - first); + } + } + } + + impl Drop for ConnTiming { + fn drop(&mut self) { + OPEN.sub(1); + } + } + + #[cfg(target_os = "linux")] + mod backlog { + use std::sync::atomic::{AtomicI32, Ordering}; + + use microscaler_timing::GaugeFn; + + static LISTENER_FD: AtomicI32 = AtomicI32::new(-1); + static BACKLOG: GaugeFn = GaugeFn::new("may_minihttp.server", "accept_backlog", backlog); + static LIMIT: GaugeFn = GaugeFn::new("may_minihttp.server", "accept_backlog_limit", limit); + + /// `TCP_INFO` of the listening socket: for a listener, `tcpi_unacked` is the accept + /// queue's current length and `tcpi_sacked` its limit. + fn info() -> Option { + let fd = LISTENER_FD.load(Ordering::Relaxed); + if fd < 0 { + return None; + } + // SAFETY: `tcp_info` is plain old data; getsockopt writes at most `len` bytes into it. + let mut info: libc::tcp_info = unsafe { std::mem::zeroed() }; + let mut len = std::mem::size_of::() as libc::socklen_t; + // SAFETY: a valid out pointer and its length; a closed fd only makes the call fail. + let rc = unsafe { + libc::getsockopt( + fd, + libc::IPPROTO_TCP, + libc::TCP_INFO, + (&mut info as *mut libc::tcp_info).cast(), + &mut len, + ) + }; + (rc == 0).then_some(info) + } + + fn backlog() -> Option { + info().map(|i| i64::from(i.tcpi_unacked)) + } + + fn limit() -> Option { + info().map(|i| i64::from(i.tcpi_sacked)) + } + + pub(crate) fn listening(fd: i32) { + LISTENER_FD.store(fd, Ordering::Relaxed); + BACKLOG.register(); + LIMIT.register(); + } + } + + /// The server is listening on `listener`: report its accept queue. + pub(crate) fn listening(listener: &may::net::TcpListener) { + #[cfg(target_os = "linux")] + { + use std::os::fd::AsRawFd; + backlog::listening(listener.as_raw_fd()); + } + #[cfg(not(target_os = "linux"))] + let _ = listener; + } +} + +#[cfg(not(feature = "op-timing"))] +mod imp { + use std::time::Instant; + + pub(crate) struct ConnTiming; + + impl ConnTiming { + #[inline(always)] + pub(crate) fn new(_accepted: Instant) -> Self { + Self + } + #[inline(always)] + pub(crate) fn read(&mut self, _before: usize, _after: usize) {} + #[inline(always)] + pub(crate) fn parsed(&mut self) {} + #[inline(always)] + pub(crate) fn incomplete(&mut self, _buffered: usize) {} + #[inline(always)] + pub(crate) fn served(&mut self, _ok: bool) {} + #[inline(always)] + pub(crate) fn written(&mut self, _drained: bool) {} + } + + #[inline(always)] + pub(crate) fn listening(_listener: &may::net::TcpListener) {} +} + +pub(crate) use imp::{listening, ConnTiming}; diff --git a/tests/server_timing.rs b/tests/server_timing.rs new file mode 100644 index 0000000..d07d839 --- /dev/null +++ b/tests/server_timing.rs @@ -0,0 +1,130 @@ +//! Feature `op-timing`: the server's phases land in op_duration_seconds and its connection levels +//! in op_gauge (microscaler-timing). +#![cfg(feature = "op-timing")] + +use std::io::{self, Read, Write}; +use std::net::TcpStream; +use std::time::Duration; + +use may_minihttp::{HttpServer, HttpService, Request, Response}; +use microscaler_timing::{gauge_values, snapshot, Outcome}; + +#[derive(Clone)] +struct Slow; + +impl HttpService for Slow { + fn call(&mut self, req: Request, rsp: &mut Response) -> io::Result<()> { + if req.path() == "/fail" { + return Err(io::Error::other("fail")); + } + may::coroutine::sleep(Duration::from_millis(30)); + rsp.body("ok"); + Ok(()) + } +} + +fn count(op: &str, outcome: Outcome) -> u64 { + snapshot() + .into_iter() + .filter(|s| s.component == "may_minihttp.server" && s.op == op && s.outcome == outcome) + .map(|s| s.count()) + .sum() +} + +fn read_response(stream: &mut TcpStream) -> String { + let mut buf = [0u8; 1024]; + let mut out = Vec::new(); + while !String::from_utf8_lossy(&out).contains("\r\n\r\n") + || !out.ends_with(b"ok") && !String::from_utf8_lossy(&out).contains(" 500 ") + { + let n = stream.read(&mut buf).unwrap(); + if n == 0 { + break; + } + out.extend_from_slice(&buf[..n]); + } + String::from_utf8_lossy(&out).into_owned() +} + +#[test] +fn server_phases_and_connection_levels_are_recorded() { + may::config().set_stack_size(0x8000); + let port = std::net::TcpListener::bind("127.0.0.1:0") + .unwrap() + .local_addr() + .unwrap() + .port(); + let handle = HttpServer(Slow).start(("127.0.0.1", port)).unwrap(); + std::mem::forget(handle); + let mut stream = loop { + if let Ok(s) = TcpStream::connect(("127.0.0.1", port)) { + break s; + } + std::thread::sleep(Duration::from_millis(10)); + }; + stream + .set_read_timeout(Some(Duration::from_secs(3))) + .unwrap(); + // the client waits before sending: that is accept_to_first_byte + std::thread::sleep(Duration::from_millis(20)); + for _ in 0..2 { + stream + .write_all(b"GET /a HTTP/1.1\r\nHost: x\r\nContent-Length: 0\r\n\r\n") + .unwrap(); + assert!(read_response(&mut stream).contains("200")); + } + stream + .write_all(b"GET /fail HTTP/1.1\r\nHost: x\r\nContent-Length: 0\r\n\r\n") + .unwrap(); + let _ = read_response(&mut stream); + // the response is on the wire before the server's write returns: give it a moment + std::thread::sleep(Duration::from_millis(50)); + + assert_eq!(count("accept_to_first_byte", Outcome::Ok), 1); + assert_eq!(count("read_parse", Outcome::Ok), 3); + assert_eq!(count("service", Outcome::Ok), 2); + assert_eq!(count("service", Outcome::Err), 1); + assert_eq!(count("response_write", Outcome::Ok), 3); + assert_eq!(count("request_total", Outcome::Ok), 3); + let service = snapshot() + .into_iter() + .find(|s| { + s.component == "may_minihttp.server" && s.op == "service" && s.outcome == Outcome::Ok + }) + .unwrap(); + assert!( + service.sum_nanos >= 60_000_000, + "two 30 ms calls: {}", + service.sum_nanos + ); + let accept = snapshot() + .into_iter() + .find(|s| s.op == "accept_to_first_byte") + .unwrap(); + assert!(accept.sum_nanos >= 20_000_000, "{}", accept.sum_nanos); + + let gauges = gauge_values(); + let gauge = |op: &str| { + gauges + .iter() + .find(|(c, o, _)| *c == "may_minihttp.server" && *o == op) + .map(|g| g.2) + }; + assert_eq!(gauge("open_connections"), Some(1)); + #[cfg(target_os = "linux")] + { + assert_eq!(gauge("accept_backlog"), Some(0)); + assert!(gauge("accept_backlog_limit").unwrap() > 0); + } + drop(stream); + for _ in 0..100 { + if gauge_values() + .iter() + .any(|(c, o, v)| *c == "may_minihttp.server" && *o == "open_connections" && *v == 0) + { + return; + } + std::thread::sleep(Duration::from_millis(10)); + } + panic!("open_connections did not return to 0"); +} From b5fdc872b00faff4fb37a5b093533cfa7dfaf1d1 Mon Sep 17 00:00:00 2001 From: Charles Sibbald Date: Tue, 3 Nov 2026 23:31:19 +0200 Subject: [PATCH 2/2] fix(server): no dead-code warnings for the timing hooks on non-unix targets The connection loop that calls ConnTiming is unix-only, so Windows clippy (-D warnings) saw an unused struct and import. --- src/server_timing.rs | 3 +++ 1 file changed, 3 insertions(+) diff --git a/src/server_timing.rs b/src/server_timing.rs index 488f637..35a74a3 100644 --- a/src/server_timing.rs +++ b/src/server_timing.rs @@ -17,6 +17,9 @@ //! Without the feature every call here is an empty inline function. With it, recording is a few //! atomic adds and an `Instant::now()`: nothing blocks the coroutine. +// The connection loop that uses ConnTiming is unix-only; elsewhere only listening() is called. +#![cfg_attr(not(unix), allow(dead_code, unused_imports))] + #[cfg(feature = "op-timing")] mod imp { use std::time::Instant;