fastwebsockets: beat uWebSockets echo-server throughput

Cargo.toml

@@ -12,6 +12,46 @@ name = "echo_server"
 path = "examples/echo_server.rs"
 required-features = ["upgrade"]
 
+[[example]]
+name = "echo_server_low"
+path = "examples/echo_server_low.rs"
+required-features = ["upgrade"]
+
+# mio-driven echo server (Linux only) — tests whether the single-thread
+# gap to uWebSockets is in WebSocket framing/parsing or in Tokio/futures
+# runtime overhead. Uses fastwebsockets::ServerEngine for the framing.
+[[example]]
+name = "echo_server_mio"
+path = "examples/echo_server_mio.rs"
+required-features = ["upgrade"]
+
+# Tokio-based echo server using fastwebsockets::ServerEngine for the
+# per-frame hot path. Same async transport (TcpStream + hyper upgrade)
+# that the standard `echo_server` example uses, but the framing/unmask/
+# response synthesis runs synchronously inside the engine. This is the
+# "Deno-friendly" fast path.
+[[example]]
+name = "echo_server_tokio_fast"
+path = "examples/echo_server_tokio_fast.rs"
+required-features = ["upgrade"]
+
+# Bench-shape demo of the public `crate::reactor::Reactor` API.
+# Pure echo via `Reactor::run_echo()`; this is the binary that the
+# uWebSockets head-to-head benchmark targets. Linux-only.
+[[example]]
+name = "echo_server_reactor"
+path = "examples/echo_server_reactor.rs"
+required-features = ["reactor"]
+
+# End-to-end demo of the `Reactor` general API: Handler trait
+# (on_open / on_frame / on_close), Connection.send / .close, and
+# the cross-thread Sender (queued commands + waker). Implements a
+# broadcast chat broker. Linux-only.
+[[example]]
+name = "reactor_chat_broker"
+path = "examples/reactor_chat_broker.rs"
+required-features = ["reactor"]
+
 [[example]]
 name = "autobahn_client"
 path = "examples/autobahn_client.rs"
@@ -60,6 +100,14 @@ axum-core = { version = "0.5.0", optional = true }
 http = { version = "1", optional = true }
 async-trait = { version = "0.1", optional = true }
 
+# Linux mio-driven reactor (opt-in via the `reactor` feature). Wraps
+# many WebSocket sessions on one thread / one event loop, sharing one
+# scratch buffer — the framing path that closes the high-fd / high-
+# payload gap to uWebSockets without spinning per-connection tokio
+# tasks. See `src/reactor.rs` and `examples/echo_server_reactor.rs`.
+mio = { version = "1.0", features = ["net", "os-poll"], optional = true }
+slab = { version = "0.4", optional = true }
+
 [features]
 default = ["simd"]
 upgrade = [
@@ -74,6 +122,8 @@ simd = ["simdutf8"]
 unstable-split = []
 # Axum integration
 with_axum = ["axum-core", "http", "async-trait"]
+# Linux mio-driven server-side reactor. See `crate::reactor`.
+reactor = ["mio", "slab", "base64", "sha1"]
 
 [dev-dependencies]
 tokio = { version = "1.25.0", features = ["full", "macros"] }
@@ -89,6 +139,13 @@ anyhow = "1.0.71"
 webpki-roots = "0.23.0"
 bytes = "1.4.0"
 axum = "0.8.1"
+# Used by examples/echo_server.rs to set SO_REUSEPORT on per-worker listener
+# sockets when FWS_WORKERS > 1. Tokio's TcpListener::bind does not expose
+# SO_REUSEPORT; we build the socket via socket2 and convert.
+socket2 = "0.5"
+mio = { version = "1.0", features = ["net", "os-poll"] }
+slab = "0.4"
+libc = "0.2"
 
 [[test]]
 name = "upgrade"

benches/unmask.rs

@@ -1,16 +1,16 @@
 use criterion::*;
 
 fn benchmark(c: &mut Criterion) {
-  const STREAM_SIZE: usize = 64 << 20;
-
-  let mut data: Vec<u8> = (0..STREAM_SIZE).map(|_| rand::random()).collect();
-  let mut group = c.benchmark_group("unmask2");
-  group.throughput(Throughput::Bytes(STREAM_SIZE as u64));
-  group.bench_function("unmask 64 << 20", |b| {
-    b.iter(|| {
-      fastwebsockets::unmask(black_box(&mut data), [1, 2, 3, 4]);
+  let mut group = c.benchmark_group("unmask");
+  for &size in &[64usize, 1024, 16 * 1024, 64 << 20] {
+    let mut data: Vec<u8> = (0..size).map(|_| rand::random()).collect();
+    group.throughput(Throughput::Bytes(size as u64));
+    group.bench_function(format!("len={}", size), |b| {
+      b.iter(|| {
+        fastwebsockets::unmask(black_box(&mut data), [1, 2, 3, 4]);
+      });
     });
-  });
+  }
   group.finish();
 }
 

examples/echo_server.rs

@@ -13,7 +13,10 @@
 // limitations under the License.
 
 use fastwebsockets::upgrade;
+use fastwebsockets::FragmentCollector;
 use fastwebsockets::OpCode;
+use fastwebsockets::Role;
+use fastwebsockets::WebSocket;
 use fastwebsockets::WebSocketError;
 use http_body_util::Empty;
 use hyper::body::Bytes;
@@ -22,11 +25,19 @@ use hyper::server::conn::http1;
 use hyper::service::service_fn;
 use hyper::Request;
 use hyper::Response;
+use hyper_util::rt::TokioIo;
 use tokio::net::TcpListener;
+use tokio::net::TcpStream;
 
-async fn handle_client(fut: upgrade::UpgradeFut) -> Result<(), WebSocketError> {
-  let mut ws = fastwebsockets::FragmentCollector::new(fut.await?);
-
+async fn echo_loop<S>(ws: WebSocket<S>) -> Result<(), WebSocketError>
+where
+  S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin,
+{
+  // The bench load_test.c never fragments, but the Autobahn suite does and
+  // expects cross-fragment UTF-8 validation. Wrap with FragmentCollector so
+  // the example stays protocol-compliant; FragmentCollector is a thin
+  // pass-through for non-fragmented frames (one match per frame).
+  let mut ws = FragmentCollector::new(ws);
   loop {
     let frame = ws.read_frame().await?;
     match frame.opcode {
@@ -37,9 +48,47 @@ async fn handle_client(fut: upgrade::UpgradeFut) -> Result<(), WebSocketError> {
       _ => {}
     }
   }
+  Ok(())
+}
 
+async fn handle_client(fut: upgrade::UpgradeFut) -> Result<(), WebSocketError> {
+  // Drive hyper's upgrade future, then downcast to the underlying TcpStream so
+  // the steady-state echo loop runs without hyper's read-buffer + trait-object
+  // indirection on every read/write.
+  let upgraded = fut.upgraded().await?;
+  match upgraded.downcast::<TokioIo<TcpStream>>() {
+    Ok(parts) => {
+      // hyper may have buffered bytes the client sent right after the upgrade
+      // request. Carry them into the WebSocket's framing buffer.
+      let stream = parts.io.into_inner();
+      let _ = stream.set_nodelay(true);
+      let ws = WebSocket::after_handshake_with_buffer(
+        stream,
+        Role::Server,
+        &parts.read_buf,
+      );
+      echo_loop(ws).await
+    }
+    Err(upgraded) => {
+      // Some other transport (TLS, h2c) — fall back to the generic path.
+      let ws = WebSocket::after_handshake(TokioIo::new(upgraded), Role::Server);
+      echo_loop(ws).await
+    }
+  }
+}
+
+async fn handle_client_tcp(stream: TcpStream) -> Result<(), WebSocketError> {
+  let _ = stream.set_nodelay(true);
+  let io = TokioIo::new(stream);
+  let conn_fut = http1::Builder::new()
+    .serve_connection(io, service_fn(server_upgrade))
+    .with_upgrades();
+  if let Err(e) = conn_fut.await {
+    eprintln!("An error occurred: {:?}", e);
+  }
   Ok(())
 }
+
 async fn server_upgrade(
   mut req: Request<Incoming>,
 ) -> Result<Response<Empty<Bytes>>, WebSocketError> {
@@ -54,27 +103,82 @@ async fn server_upgrade(
   Ok(response)
 }
 
-fn main() -> Result<(), WebSocketError> {
+fn make_reuseport_listener(addr: &str) -> std::io::Result<TcpListener> {
+  use socket2::{Domain, Protocol, Socket, Type};
+  let parsed: std::net::SocketAddr = addr.parse().map_err(|e| {
+    std::io::Error::new(
+      std::io::ErrorKind::InvalidInput,
+      format!("bad addr: {}", e),
+    )
+  })?;
+  let domain = if parsed.is_ipv6() {
+    Domain::IPV6
+  } else {
+    Domain::IPV4
+  };
+  let sock = Socket::new(domain, Type::STREAM, Some(Protocol::TCP))?;
+  sock.set_reuse_address(true)?;
+  #[cfg(any(target_os = "linux", target_os = "freebsd"))]
+  sock.set_reuse_port(true)?;
+  sock.set_nonblocking(true)?;
+  sock.bind(&parsed.into())?;
+  sock.listen(1024)?;
+  TcpListener::from_std(sock.into())
+}
+
+fn run_worker(
+  worker_id: usize,
+  addr: String,
+) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
   let rt = tokio::runtime::Builder::new_current_thread()
     .enable_io()
-    .build()
-    .unwrap();
-
+    .build()?;
   rt.block_on(async move {
-    let listener = TcpListener::bind("127.0.0.1:8080").await?;
-    println!("Server started, listening on {}", "127.0.0.1:8080");
+    let listener = make_reuseport_listener(&addr)?;
+    eprintln!("[worker {}] listening on {}", worker_id, addr);
     loop {
       let (stream, _) = listener.accept().await?;
-      println!("Client connected");
       tokio::spawn(async move {
-        let io = hyper_util::rt::TokioIo::new(stream);
-        let conn_fut = http1::Builder::new()
-          .serve_connection(io, service_fn(server_upgrade))
-          .with_upgrades();
-        if let Err(e) = conn_fut.await {
-          println!("An error occurred: {:?}", e);
+        if let Err(e) = handle_client_tcp(stream).await {
+          eprintln!("connection error: {}", e);
         }
       });
     }
   })
 }
+
+fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+  let workers = std::env::var("FWS_WORKERS")
+    .ok()
+    .and_then(|s| s.parse::<usize>().ok())
+    .unwrap_or(1);
+
+  let addr =
+    std::env::var("FWS_ADDR").unwrap_or_else(|_| "127.0.0.1:8080".to_string());
+
+  if workers <= 1 {
+    return run_worker(0, addr).map_err(|e| e.into());
+  }
+
+  // Multi-worker: each thread runs its own current_thread runtime and binds
+  // a SO_REUSEPORT listener on the same port. The kernel load-balances
+  // accept() across the listeners, so each connection lives entirely inside
+  // one worker (no cross-thread task migration). This is the same model
+  // uWebSockets recommends for scaling beyond one core.
+  let mut handles = Vec::with_capacity(workers);
+  for i in 0..workers {
+    let addr = addr.clone();
+    let h = std::thread::Builder::new()
+      .name(format!("fws-worker-{}", i))
+      .spawn(move || {
+        if let Err(e) = run_worker(i, addr) {
+          eprintln!("[worker {}] exiting: {}", i, e);
+        }
+      })?;
+    handles.push(h);
+  }
+  for h in handles {
+    let _ = h.join();
+  }
+  Ok(())
+}