feat(causal): instrument the request hot path — five sites and a progress point
RFC 007 causal-site coverage for the Phase 2 bench (and any downstream
profiling): parse (head parsing attempts in read_head), router (dispatch
matching only — the winning handler and the next fall-through run outside
the guard, so the site measures dispatch, not what it dispatches to;
call() restructured match-then-dispatch for that, semantics preserved
incl. first-path+method-match-wins and the 405 two-pass), pipeline (the
plug chain inside catch_unwind; nested sites like router take over
attribution during their span, so this reads as chain overhead + handler
code outside inner sites), serialize (response head+bytes-body
serialisation; chunked stream framing is not covered — it interleaves
with writes in pump_stream and the bench doesn't stream), and
socket-write (all of write_all, writability parks included: a park
inside a site is exactly what the park-gated resume credit attributes).
progress!("responses") fires once per response fully on the wire, at
both completion points (the common path and the HTTP/1.0 EOF-stream
early return).
Site guards are #[cfg(feature = "smarm-causal")]-gated: the macros are
no-ops featureless, but binding the unit expansion would trip clippy's
let_unit_value. progress! is bare — its featureless expansion is an
empty block, free and lint-clean. Featureless build byte-behavior is
unchanged; both configs clippy-clean, full suite green both ways.
This commit is contained in:
+23
-2
@@ -209,6 +209,8 @@ pub fn run_connection(
|
|||||||
// Catch panics at the actor boundary — a panicking handler should
|
// Catch panics at the actor boundary — a panicking handler should
|
||||||
// not take down the whole connection silently with no response.
|
// not take down the whole connection silently with no response.
|
||||||
let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
|
let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
|
||||||
|
#[cfg(feature = "smarm-causal")]
|
||||||
|
let _g = smarm::causal_site!("pipeline");
|
||||||
pipeline.run(conn)
|
pipeline.run(conn)
|
||||||
}));
|
}));
|
||||||
|
|
||||||
@@ -273,7 +275,11 @@ pub fn run_connection(
|
|||||||
let keep_alive = keep_alive
|
let keep_alive = keep_alive
|
||||||
&& !(version == HttpVersion::Http10 && (is_stream || is_error));
|
&& !(version == HttpVersion::Http10 && (is_stream || is_error));
|
||||||
|
|
||||||
let head_bytes = parser::serialise_response(&response_conn, keep_alive);
|
let head_bytes = {
|
||||||
|
#[cfg(feature = "smarm-causal")]
|
||||||
|
let _g = smarm::causal_site!("serialize");
|
||||||
|
parser::serialise_response(&response_conn, keep_alive)
|
||||||
|
};
|
||||||
if write_all(raw, &head_bytes, Instant::now() + limits.write_timeout).is_err() {
|
if write_all(raw, &head_bytes, Instant::now() + limits.write_timeout).is_err() {
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
@@ -285,10 +291,15 @@ pub fn run_connection(
|
|||||||
}
|
}
|
||||||
if !chunked {
|
if !chunked {
|
||||||
// EOF delimits the HTTP/1.0 stream body.
|
// EOF delimits the HTTP/1.0 stream body.
|
||||||
|
smarm::progress!("responses");
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// A full response (head + body, streamed or not) is on the wire:
|
||||||
|
// the unit of useful work for causal profiling (RFC 007).
|
||||||
|
smarm::progress!("responses");
|
||||||
|
|
||||||
// ----- 5. Loop or close. -----
|
// ----- 5. Loop or close. -----
|
||||||
if !keep_alive {
|
if !keep_alive {
|
||||||
return;
|
return;
|
||||||
@@ -358,7 +369,12 @@ fn read_head(
|
|||||||
// Try to parse what we already have. On the first iteration of a
|
// Try to parse what we already have. On the first iteration of a
|
||||||
// fresh keep-alive cycle, `buf` may already hold the next request.
|
// fresh keep-alive cycle, `buf` may already hold the next request.
|
||||||
if !buf.is_empty() {
|
if !buf.is_empty() {
|
||||||
match parser::parse_head(buf, limits.max_headers) {
|
let head = {
|
||||||
|
#[cfg(feature = "smarm-causal")]
|
||||||
|
let _g = smarm::causal_site!("parse");
|
||||||
|
parser::parse_head(buf, limits.max_headers)
|
||||||
|
};
|
||||||
|
match head {
|
||||||
Ok(h) => {
|
Ok(h) => {
|
||||||
let deadline = request_deadline
|
let deadline = request_deadline
|
||||||
.unwrap_or_else(|| Instant::now() + limits.request_timeout);
|
.unwrap_or_else(|| Instant::now() + limits.request_timeout);
|
||||||
@@ -634,6 +650,11 @@ fn try_write_once(fd: RawFd, buf: &[u8]) {
|
|||||||
// wait_writable forever (the write-side twin of slowloris).
|
// wait_writable forever (the write-side twin of slowloris).
|
||||||
|
|
||||||
pub(crate) fn write_all(fd: RawFd, mut buf: &[u8], deadline: Instant) -> io::Result<()> {
|
pub(crate) fn write_all(fd: RawFd, mut buf: &[u8], deadline: Instant) -> io::Result<()> {
|
||||||
|
// The whole loop — writability parks included — runs under the
|
||||||
|
// `socket-write` causal site: a park inside a site is exactly what
|
||||||
|
// RFC 007's park-gated resume credit exists to attribute.
|
||||||
|
#[cfg(feature = "smarm-causal")]
|
||||||
|
let _g = smarm::causal_site!("socket-write");
|
||||||
while !buf.is_empty() {
|
while !buf.is_empty() {
|
||||||
// Park on writability before each syscall, bounded by the budget.
|
// Park on writability before each syscall, bounded by the budget.
|
||||||
let remaining = deadline.saturating_duration_since(Instant::now());
|
let remaining = deadline.saturating_duration_since(Instant::now());
|
||||||
|
|||||||
+19
-6
@@ -135,16 +135,29 @@ impl Plug for Router {
|
|||||||
// path matches but a same-path-different-method does, return 405.
|
// path matches but a same-path-different-method does, return 405.
|
||||||
// Otherwise pass through to `next` so outer pipelines can layer a
|
// Otherwise pass through to `next` so outer pipelines can layer a
|
||||||
// 404 handler (or skip and let the connection actor emit a default).
|
// 404 handler (or skip and let the connection actor emit a default).
|
||||||
|
//
|
||||||
|
// Matching runs under the `router` causal site (RFC 007); the
|
||||||
|
// winning handler and the `next` fall-through run outside it, so
|
||||||
|
// the site measures dispatch, not what it dispatches to.
|
||||||
let mut path_seen = false;
|
let mut path_seen = false;
|
||||||
for route in &self.routes {
|
let mut hit = None;
|
||||||
if let Some(params) = route.pattern.match_path(&conn.path) {
|
{
|
||||||
if route.method == conn.method {
|
#[cfg(feature = "smarm-causal")]
|
||||||
let conn = conn.put_params(params);
|
let _g = smarm::causal_site!("router");
|
||||||
return route.handler.call(conn, next);
|
for (i, route) in self.routes.iter().enumerate() {
|
||||||
|
if let Some(params) = route.pattern.match_path(&conn.path) {
|
||||||
|
if route.method == conn.method {
|
||||||
|
hit = Some((i, params));
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
path_seen = true;
|
||||||
}
|
}
|
||||||
path_seen = true;
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
if let Some((i, params)) = hit {
|
||||||
|
let conn = conn.put_params(params);
|
||||||
|
return self.routes[i].handler.call(conn, next);
|
||||||
|
}
|
||||||
if path_seen {
|
if path_seen {
|
||||||
// Path is known, method isn't — RFC 7231 §6.5.5.
|
// Path is known, method isn't — RFC 7231 §6.5.5.
|
||||||
conn.put_status(405)
|
conn.put_status(405)
|
||||||
|
|||||||
Reference in New Issue
Block a user