feat(ext/flash): add reuseport option on Linux (#16022)

Cargo.lock

@@ -1094,6 +1094,7 @@ dependencies = [
  "rustls",
  "rustls-pemfile 1.0.1",
  "serde",
+ "socket2",
  "tokio",
 ]
 

cli/bench/http/deno_http_flash.js

@@ -8,4 +8,4 @@ function handler() {
   return new Response("Hello World");
 }
 
-serve(handler, { hostname, port });
+serve(handler, { hostname, port, reusePort: true });

cli/bench/http/deno_http_flash_ops.js

@@ -11,7 +11,7 @@ const {
 } = Deno;
 const addr = Deno.args[0] || "127.0.0.1:4500";
 const [hostname, port] = addr.split(":");
-const serverId = op_flash_serve({ hostname, port });
+const serverId = op_flash_serve({ hostname, port, reuseport: true });
 const serverPromise = opAsync("op_flash_drive_server", serverId);
 
 const fastOps = op_flash_make_request();

cli/bench/http/deno_http_flash_ops_spawn.js

@@ -0,0 +1,16 @@
+if (Deno.build.os !== "linux") {
+  throw new Error("SO_REUSEPORT is only supported on Linux");
+}
+
+const executable = Deno.execPath();
+const path = new URL("./deno_http_flash_ops.js", import.meta.url).pathname;
+// single flash instance runs on ~1.8 cores
+const cpus = navigator.hardwareConcurrency / 2;
+const processes = new Array(cpus);
+for (let i = 0; i < cpus; i++) {
+  const proc = Deno.run({
+    cmd: [executable, "run", "-A", "--unstable", path, Deno.args[0]],
+  });
+  processes.push(proc.status());
+}
+await Promise.all(processes);

cli/bench/http/deno_http_flash_spawn.js

@@ -0,0 +1,16 @@
+if (Deno.build.os !== "linux") {
+  throw new Error("SO_REUSEPORT is only supported on Linux");
+}
+
+const executable = Deno.execPath();
+const path = new URL("./deno_http_flash.js", import.meta.url).pathname;
+// single flash instance runs on ~1.8 cores
+const cpus = navigator.hardwareConcurrency / 2;
+const processes = new Array(cpus);
+for (let i = 0; i < cpus; i++) {
+  const proc = Deno.run({
+    cmd: [executable, "run", "-A", "--unstable", path, Deno.args[0]],
+  });
+  processes.push(proc.status());
+}
+await Promise.all(processes);

cli/dts/lib.deno.unstable.d.ts

@@ -1352,6 +1352,9 @@ declare namespace Deno {
     /** An AbortSignal to close the server and all connections. */
     signal?: AbortSignal;
 
+    /** Sets SO_REUSEPORT on Linux. */
+    reusePort?: boolean;
+
     /** The handler to invoke when route handlers throw an error. */
     onError?: (error: unknown) => Response | Promise<Response>;
 

ext/flash/01_http.js

@@ -459,6 +459,7 @@
     const listenOpts = {
       hostname: options.hostname ?? "127.0.0.1",
       port: options.port ?? 9000,
+      reuseport: options.reusePort ?? false,
     };
     if (options.cert || options.key) {
       if (!options.cert || !options.key) {

ext/flash/Cargo.toml

@@ -26,4 +26,5 @@ mio = { version = "0.8.1", features = ["os-poll", "net"] }
 rustls = { version = "0.20" }
 rustls-pemfile = "1.0"
 serde = { version = "1.0.136", features = ["derive"] }
+socket2 = "0.4.7"
 tokio = { version = "1.21", features = ["full"] }

ext/flash/lib.rs

@@ -37,6 +37,7 @@ use mio::Poll;
 use mio::Token;
 use serde::Deserialize;
 use serde::Serialize;
+use socket2::Socket;
 use std::cell::RefCell;
 use std::cell::UnsafeCell;
 use std::collections::HashMap;
@@ -806,6 +807,7 @@ pub struct ListenOpts {
   key: Option<String>,
   hostname: String,
   port: u16,
+  reuseport: bool,
 }
 
 fn run_server(
@@ -815,8 +817,29 @@ fn run_server(
   addr: SocketAddr,
   maybe_cert: Option<String>,
   maybe_key: Option<String>,
+  reuseport: bool,
 ) -> Result<(), AnyError> {
-  let mut listener = TcpListener::bind(addr)?;
+  let domain = if addr.is_ipv4() {
+    socket2::Domain::IPV4
+  } else {
+    socket2::Domain::IPV6
+  };
+  let socket = Socket::new(domain, socket2::Type::STREAM, None)?;
+
+  #[cfg(not(windows))]
+  socket.set_reuse_address(true)?;
+  if reuseport {
+    #[cfg(target_os = "linux")]
+    socket.set_reuse_port(true)?;
+  }
+
+  let socket_addr = socket2::SockAddr::from(addr);
+  socket.bind(&socket_addr)?;
+  socket.listen(128)?;
+  socket.set_nonblocking(true)?;
+  let std_listener: std::net::TcpListener = socket.into();
+  let mut listener = TcpListener::from_std(std_listener);
+
   let mut poll = Poll::new()?;
   let token = Token(0);
   poll
@@ -875,6 +898,7 @@ fn run_server(
                 .registry()
                 .register(&mut socket, token, Interest::READABLE)
                 .unwrap();
+
               let socket = match tls_context {
                 Some(ref tls_conf) => {
                   let connection =
@@ -1156,8 +1180,17 @@ where
   let tx = ctx.tx.clone();
   let maybe_cert = opts.cert;
   let maybe_key = opts.key;
+  let reuseport = opts.reuseport;
   let join_handle = tokio::task::spawn_blocking(move || {
-    run_server(tx, listening_tx, close_rx, addr, maybe_cert, maybe_key)
+    run_server(
+      tx,
+      listening_tx,
+      close_rx,
+      addr,
+      maybe_cert,
+      maybe_key,
+      reuseport,
+    )
   });
   let flash_ctx = state.borrow_mut::<FlashContext>();
   let server_id = flash_ctx.next_server_id;