xtask/src/wireproxy.rs
//! `cargo xtask wireproxy` — a counting TCP relay, so three different git
//! servers are measured by **one** instrument.
//!
//! # Why this exists
//!
//! The forge bake-off puts stock `git`
//! against gunnar, gitea and Forgejo. Wall-clock milliseconds on a 32-core box
//! whose load has run 2–218 today are not a number anyone should stake a claim
//! on; **bytes on the wire are counted, not sampled**, and they are the thing a
//! git server is actually for.
//!
//! The obvious ways to count those bytes are all unfair in a way that is easy
//! to miss:
//!
//! - *Measure the received `.pack` on disk.* Works for clone. Does not work for
//! push: `receive-pack` explodes a pack below `transfer.unpackLimit` (100 by
//! default) into loose objects, so the pack that crossed the wire is gone
//! before anything can stat it. Two of the four operations would be measured
//! by a different rule than the other two.
//! - *Count inside each server.* Three instruments, three sets of bugs, and the
//! two Go servers would have to be patched and rebuilt.
//! - *`GIT_TRACE_PACKET`.* Counts pkt-lines after decoding, not bytes, and it
//! changes what the client does while you watch it.
//!
//! So: one relay in front of every server, on the same box, in the same
//! process, doing the same `read`/`write` loop. Whatever it miscounts, it
//! miscounts identically for all three, which is the only property a
//! comparison needs.
//!
//! # Usage
//!
//! ```text
//! cargo xtask wireproxy --listen 127.0.0.1:19001 --to 127.0.0.1:12222 \
//! --out /path/to/counts.tsv [--ready-fd-file /path/ready]
//! ```
//!
//! One line is appended per **closed** connection:
//!
//! ```text
//! <c2s_bytes>\t<s2c_bytes>\t<elapsed_ms>\t<connect_result>
//! ```
//!
//! `connect_result` is `ok` or `refused:<errno text>`. That last column is the
//! whole red-first story of this file: a server that never came up and a server that
//! transferred nothing both produce **zero bytes**, and without the column they
//! are the same row. The harness refuses to score any arm whose lines are not
//! all `ok`.
//!
//! # Zero dependencies
//!
//! `xtask` has none by construction (see `xtask/Cargo.toml`), so this is
//! `std::net` and `std::thread` and nothing else.
use std::fs::OpenOptions;
use std::io::{ErrorKind, Read, Write};
use std::net::{Shutdown, SocketAddr, TcpListener, TcpStream};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use std::time::Instant;
/// What one relayed connection cost, in bytes each way.
#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
pub struct ConnCount {
/// Client → server. For a push this is the pack; for a fetch it is the
/// want/have negotiation.
pub c2s: u64,
/// Server → client. For a fetch this is the pack.
pub s2c: u64,
}
/// How a connection ended. Kept separate from the counts precisely so that a
/// refused connect cannot be read as a cheap transfer.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ConnResult {
Ok,
Refused(String),
}
impl ConnResult {
pub fn as_column(&self) -> String {
match self {
ConnResult::Ok => "ok".into(),
ConnResult::Refused(why) => format!("refused:{why}"),
}
}
/// The only verdict the harness may score.
pub fn is_ok(&self) -> bool {
matches!(self, ConnResult::Ok)
}
}
/// Parse `--listen A --to B --out F`. Returns `Err` with a usage string rather
/// than panicking, because this runs inside a bench harness that must be able
/// to print *why* it could not start.
pub fn parse_args(args: &[String]) -> Result<(SocketAddr, SocketAddr, String), String> {
let mut listen = None;
let mut to = None;
let mut out = None;
let mut i = 0;
while i < args.len() {
let need = |i: usize| -> Result<&String, String> {
args.get(i + 1)
.ok_or_else(|| format!("{} needs a value", args[i]))
};
match args[i].as_str() {
"--listen" => {
listen = Some(need(i)?.clone());
i += 2;
}
"--to" => {
to = Some(need(i)?.clone());
i += 2;
}
"--out" => {
out = Some(need(i)?.clone());
i += 2;
}
other => return Err(format!("wireproxy: unknown argument `{other}`")),
}
}
let listen = listen.ok_or("wireproxy: --listen is required")?;
let to = to.ok_or("wireproxy: --to is required")?;
let out = out.ok_or("wireproxy: --out is required")?;
let listen: SocketAddr = listen
.parse()
.map_err(|e| format!("wireproxy: --listen {listen}: {e}"))?;
let to: SocketAddr = to
.parse()
.map_err(|e| format!("wireproxy: --to {to}: {e}"))?;
Ok((listen, to, out))
}
/// Copy `from` into `to` until EOF, adding every byte to `counter`.
///
/// Returns the total. A short write is looped over rather than assumed away:
/// `write_all` on a socket that has gone away returns an error, and treating
/// that as zero bytes would undercount exactly the case the harness cares about.
fn pump(mut from: TcpStream, mut to: TcpStream, counter: &AtomicU64) -> u64 {
let mut buf = vec![0u8; 64 * 1024];
let mut total = 0u64;
loop {
match from.read(&mut buf) {
Ok(0) => break,
Ok(n) => {
total += n as u64;
counter.fetch_add(n as u64, Ordering::Relaxed);
if to.write_all(&buf[..n]).is_err() {
break;
}
}
Err(ref e) if e.kind() == ErrorKind::Interrupted => continue,
Err(_) => break,
}
}
// Half-close so the peer sees EOF and its own pump can finish. Without this
// a `git fetch` that has sent its last `done` hangs until a timeout, and the
// measured millisecond column becomes the timeout rather than the transfer.
let _ = to.shutdown(Shutdown::Write);
total
}
/// Relay one accepted connection and return what it cost.
///
/// `Refused` is returned, not logged and swallowed: the caller writes it into
/// the TSV so the harness can see it.
pub fn relay(client: TcpStream, upstream_addr: SocketAddr) -> (ConnCount, ConnResult, u128) {
let started = Instant::now();
let upstream = match TcpStream::connect(upstream_addr) {
Ok(s) => s,
Err(e) => {
// A connection the proxy accepted but could not forward is the
// "container never came up" case. Zero bytes, and a result column
// that says so.
return (
ConnCount::default(),
ConnResult::Refused(e.to_string()),
started.elapsed().as_millis(),
);
}
};
let _ = client.set_nodelay(true);
let _ = upstream.set_nodelay(true);
let (c2s, s2c) = (Arc::new(AtomicU64::new(0)), Arc::new(AtomicU64::new(0)));
let (cr, ur) = match (client.try_clone(), upstream.try_clone()) {
(Ok(c), Ok(u)) => (c, u),
_ => {
return (
ConnCount::default(),
ConnResult::Refused("try_clone failed".into()),
started.elapsed().as_millis(),
)
}
};
let up_counter = Arc::clone(&c2s);
let up = std::thread::spawn(move || pump(cr, ur, &up_counter));
let down_total = pump(upstream, client, &s2c);
let up_total = up.join().unwrap_or(0);
(
ConnCount {
c2s: up_total,
s2c: down_total,
},
ConnResult::Ok,
started.elapsed().as_millis(),
)
}
/// Bind, then serve forever. Blocks; the harness kills it.
///
/// Binding happens **before** this prints `ready`, so a harness that waits for
/// that line cannot race the listener.
pub fn run(listen: SocketAddr, upstream: SocketAddr, out_path: &str) -> Result<(), String> {
let listener =
TcpListener::bind(listen).map_err(|e| format!("wireproxy: bind {listen}: {e}"))?;
let out = OpenOptions::new()
.create(true)
.append(true)
.open(out_path)
.map_err(|e| format!("wireproxy: open {out_path}: {e}"))?;
let out = Arc::new(Mutex::new(out));
println!("ready {listen} -> {upstream}");
let _ = std::io::stdout().flush();
for conn in listener.incoming() {
let client = match conn {
Ok(c) => c,
Err(_) => continue,
};
let out = Arc::clone(&out);
std::thread::spawn(move || {
let (count, result, ms) = relay(client, upstream);
if let Ok(mut f) = out.lock() {
let _ = writeln!(
f,
"{}\t{}\t{}\t{}",
count.c2s,
count.s2c,
ms,
result.as_column()
);
let _ = f.flush();
}
});
}
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
use std::net::TcpListener as L;
/// The claim: a refused upstream is distinguishable from a silent transfer.
///
/// Red-first: make [`relay`] return `ConnResult::Ok` on the connect error
/// path and this fails, because both rows become `0\t0\t…\tok`. That is the
/// exact confusion the brief calls out — "0 ops/s" and "container never came
/// up" must not look alike — and a byte count alone cannot tell them apart,
/// which is why the column exists at all.
#[test]
fn a_dead_upstream_is_not_a_cheap_transfer() {
// Bind and immediately drop, so the port is (almost certainly) closed.
let dead = {
let l = L::bind("127.0.0.1:0").unwrap();
l.local_addr().unwrap()
};
let front = L::bind("127.0.0.1:0").unwrap();
let front_addr = front.local_addr().unwrap();
let h = std::thread::spawn(move || {
let (c, _) = front.accept().unwrap();
relay(c, dead)
});
let _client = TcpStream::connect(front_addr).unwrap();
let (count, result, _ms) = h.join().unwrap();
assert_eq!(count, ConnCount::default(), "a refused relay moved bytes");
assert!(
!result.is_ok(),
"a refused upstream reported `ok` — this is the row that makes a \
dead server look like a fast one"
);
assert!(
result.as_column().starts_with("refused:"),
"the TSV column must name the failure, got {:?}",
result.as_column()
);
}
/// The counts are the bytes, both directions, and the two are not swapped.
///
/// Red-first: swap `c2s` and `s2c` in the returned [`ConnCount`] and this
/// fails. Asserting only the *total* would stay green through that swap,
/// and a fetch charged as a push is precisely the mistake that would make
/// gunnar look good for the wrong reason.
#[test]
fn each_direction_is_counted_separately_and_not_transposed() {
// An echo-ish upstream that reads 5 bytes then writes 11 back.
let upstream = L::bind("127.0.0.1:0").unwrap();
let up_addr = upstream.local_addr().unwrap();
std::thread::spawn(move || {
let (mut s, _) = upstream.accept().unwrap();
let mut b = [0u8; 5];
s.read_exact(&mut b).unwrap();
s.write_all(b"hello world").unwrap();
let _ = s.shutdown(Shutdown::Write);
});
let front = L::bind("127.0.0.1:0").unwrap();
let front_addr = front.local_addr().unwrap();
let h = std::thread::spawn(move || {
let (c, _) = front.accept().unwrap();
relay(c, up_addr)
});
let mut client = TcpStream::connect(front_addr).unwrap();
client.write_all(b"12345").unwrap();
// ⚠ WITHOUT THIS THE TEST HANGS FOREVER, and it took the whole repo's
// `cargo test -p xtask` with it — the run never reached a verdict, on
// any target, because this one lib test never finishes.
//
// The PRODUCT is not at fault and neither is `pump`, which half-closes
// correctly (see its tail). `relay` cannot return until BOTH pumps do,
// and the client→server pump ends only when the client's write side
// reaches EOF. A real client gets there by exiting; this one stays in
// scope until `h.join()`, so the relay waits on a socket that will
// never close while the test waits on the relay. Each is blocked on
// the other.
//
// Saying `done` by half-closing is also what a finished git client
// really does, so this makes the fixture more faithful, not less.
client.shutdown(Shutdown::Write).unwrap();
let mut got = Vec::new();
let _ = client.read_to_end(&mut got);
// The upward pump blocks on `read` until the client closes, so joining
// while this socket is still alive deadlocks. It did, the first time.
drop(client);
let (count, result, _) = h.join().unwrap();
assert!(result.is_ok());
assert_eq!(count.c2s, 5, "client→server byte count is wrong");
assert_eq!(count.s2c, 11, "server→client byte count is wrong");
assert_ne!(
count.c2s, count.s2c,
"the fixture cannot detect a transpose"
);
assert_eq!(&got, b"hello world");
}
#[test]
fn argument_parsing_rejects_what_it_cannot_honour() {
let a: Vec<String> = [
"--listen",
"127.0.0.1:1",
"--to",
"127.0.0.1:2",
"--out",
"f",
]
.iter()
.map(|s| s.to_string())
.collect();
let (l, t, o) = parse_args(&a).unwrap();
assert_eq!(l.port(), 1);
assert_eq!(t.port(), 2);
assert_eq!(o, "f");
// A missing --out must not default to somewhere: a harness that writes
// its counts into a file nobody reads is a green nobody has seen red.
let short: Vec<String> = ["--listen", "127.0.0.1:1", "--to", "127.0.0.1:2"]
.iter()
.map(|s| s.to_string())
.collect();
assert!(parse_args(&short).is_err());
assert!(parse_args(&["--listen".to_string(), "nope".to_string()]).is_err());
}
}