feat(ext/net): reusePort for TCP on Linux (#16398)

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

@@ -1158,6 +1158,22 @@ declare namespace Deno {
     [Symbol.asyncIterator](): AsyncIterableIterator<[Uint8Array, Addr]>;
   }
 
+  /**
+   * @category Network
+   */
+  export interface TcpListenOptions extends ListenOptions {
+    /** When `true` the SO_REUSEPORT flag will be set on the listener. This
+     * allows multiple processes to listen on the same address and port.
+     *
+     * On Linux this will cause the kernel to distribute incoming connections
+     * across the different processes that are listening on the same address and
+     * port.
+     *
+     * This flag is only supported on Linux. It is silently ignored on other
+     * platforms. Defaults to `false`. */
+    reusePort?: boolean;
+  }
+
   /** **UNSTABLE**: New API, yet to be vetted.
    *
    * Unstable options which can be set when opening a Unix listener via

cli/tests/unit/net_test.ts

@@ -1005,3 +1005,48 @@ Deno.test(
     }
   },
 );
+
+Deno.test({
+  ignore: Deno.build.os !== "linux",
+  permissions: { net: true },
+}, async function netTcpListenReusePort() {
+  const port = 4003;
+  const listener1 = Deno.listen({ port, reusePort: true });
+  const listener2 = Deno.listen({ port, reusePort: true });
+  let p1;
+  let p2;
+  let listener1Recv = false;
+  let listener2Recv = false;
+  while (!listener1Recv || !listener2Recv) {
+    if (!p1) {
+      p1 = listener1.accept().then((conn) => {
+        conn.close();
+        listener1Recv = true;
+        p1 = undefined;
+      }).catch(() => {});
+    }
+    if (!p2) {
+      p2 = listener2.accept().then((conn) => {
+        conn.close();
+        listener2Recv = true;
+        p2 = undefined;
+      }).catch(() => {});
+    }
+    const conn = await Deno.connect({ port });
+    conn.close();
+    await Promise.race([p1, p2]);
+  }
+  listener1.close();
+  listener2.close();
+});
+
+Deno.test({
+  ignore: Deno.build.os === "linux",
+  permissions: { net: true },
+}, function netTcpListenReusePortDoesNothing() {
+  const listener1 = Deno.listen({ port: 4003, reusePort: true });
+  assertThrows(() => {
+    Deno.listen({ port: 4003, reusePort: true });
+  }, Deno.errors.AddrInUse);
+  listener1.close();
+});

cli/tests/unit/tls_test.ts

@@ -1410,3 +1410,69 @@ Deno.test(
     listener2.close();
   },
 );
+
+Deno.test({
+  ignore: Deno.build.os !== "linux",
+  permissions: { net: true },
+}, async function listenTlsReusePort() {
+  const hostname = "localhost";
+  const port = 4003;
+  const listener1 = Deno.listenTls({
+    hostname,
+    port,
+    cert,
+    key,
+    reusePort: true,
+  });
+  const listener2 = Deno.listenTls({
+    hostname,
+    port,
+    cert,
+    key,
+    reusePort: true,
+  });
+  let p1;
+  let p2;
+  let listener1Recv = false;
+  let listener2Recv = false;
+  while (!listener1Recv || !listener2Recv) {
+    if (!p1) {
+      p1 = listener1.accept().then((conn) => {
+        conn.close();
+        listener1Recv = true;
+        p1 = undefined;
+      }).catch(() => {});
+    }
+    if (!p2) {
+      p2 = listener2.accept().then((conn) => {
+        conn.close();
+        listener2Recv = true;
+        p2 = undefined;
+      }).catch(() => {});
+    }
+    const conn = await Deno.connectTls({ hostname, port, caCerts });
+    conn.close();
+    await Promise.race([p1, p2]);
+  }
+  listener1.close();
+  listener2.close();
+});
+
+Deno.test({
+  ignore: Deno.build.os === "linux",
+  permissions: { net: true },
+}, function listenTlsReusePortDoesNothing() {
+  const hostname = "localhost";
+  const port = 4003;
+  const listener1 = Deno.listenTls({
+    hostname,
+    port,
+    cert,
+    key,
+    reusePort: true,
+  });
+  assertThrows(() => {
+    Deno.listenTls({ hostname, port, cert, key, reusePort: true });
+  }, Deno.errors.AddrInUse);
+  listener1.close();
+});

ext/net/01_net.js

@@ -302,7 +302,7 @@
         const [rid, addr] = ops.op_net_listen_tcp({
           hostname: args.hostname ?? "0.0.0.0",
           port: args.port,
-        });
+        }, args.reusePort);
         addr.transport = "tcp";
         return new Listener(rid, addr);
       }

ext/net/02_tls.js

@@ -65,13 +65,14 @@
     hostname = "0.0.0.0",
     transport = "tcp",
     alpnProtocols = undefined,
+    reusePort = false,
   }) {
     if (transport !== "tcp") {
       throw new TypeError(`Unsupported transport: '${transport}'`);
     }
     const [rid, localAddr] = ops.op_net_listen_tls(
       { hostname, port },
-      { cert, certFile, key, keyFile, alpnProtocols },
+      { cert, certFile, key, keyFile, alpnProtocols, reusePort },
     );
     return new TlsListener(rid, localAddr);
   }

ext/net/lib.deno_net.d.ts

@@ -91,6 +91,11 @@ declare namespace Deno {
     hostname?: string;
   }
 
+  /** @category Network */
+  // deno-lint-ignore no-empty-interface
+  export interface TcpListenOptions extends ListenOptions {
+  }
+
   /** Listen announces on the local transport address.
    *
    * ```ts
@@ -106,11 +111,11 @@ declare namespace Deno {
    * @category Network
    */
   export function listen(
-    options: ListenOptions & { transport?: "tcp" },
+    options: TcpListenOptions & { transport?: "tcp" },
   ): Listener;
 
   /** @category Network */
-  export interface ListenTlsOptions extends ListenOptions {
+  export interface ListenTlsOptions extends TcpListenOptions {
     /** Server private key in PEM format */
     key?: string;
     /** Cert chain in PEM format */

ext/net/ops.rs

@@ -246,10 +246,14 @@ impl Resource for UdpSocketResource {
 fn op_net_listen_tcp<NP>(
   state: &mut OpState,
   addr: IpAddr,
+  reuse_port: bool,
 ) -> Result<(ResourceId, IpAddr), AnyError>
 where
   NP: NetPermissions + 'static,
 {
+  if reuse_port {
+    super::check_unstable(state, "Deno.listen({ reusePort: true })");
+  }
   state
     .borrow_mut::<NP>()
     .check_net(&(&addr.hostname, Some(addr.port)), "Deno.listen()")?;
@@ -264,6 +268,10 @@ where
   let socket = Socket::new(domain, Type::STREAM, None)?;
   #[cfg(not(windows))]
   socket.set_reuse_address(true)?;
+  if reuse_port {
+    #[cfg(target_os = "linux")]
+    socket.set_reuse_port(true)?;
+  }
   let socket_addr = socket2::SockAddr::from(addr);
   socket.bind(&socket_addr)?;
   socket.listen(128)?;

ext/net/ops_tls.rs

@@ -990,6 +990,7 @@ pub struct ListenTlsArgs {
   // TODO(kt3k): Remove this option at v2.0.
   key_file: Option<String>,
   alpn_protocols: Option<Vec<String>>,
+  reuse_port: bool,
 }
 
 #[op]
@@ -1001,6 +1002,10 @@ pub fn op_net_listen_tls<NP>(
 where
   NP: NetPermissions + 'static,
 {
+  if args.reuse_port {
+    super::check_unstable(state, "Deno.listenTls({ reusePort: true })");
+  }
+
   let cert_file = args.cert_file.as_deref();
   let key_file = args.key_file.as_deref();
   let cert = args.cert.as_deref();
@@ -1061,6 +1066,10 @@ where
   let socket = Socket::new(domain, Type::STREAM, None)?;
   #[cfg(not(windows))]
   socket.set_reuse_address(true)?;
+  if args.reuse_port {
+    #[cfg(target_os = "linux")]
+    socket.set_reuse_port(true)?;
+  }
   let socket_addr = socket2::SockAddr::from(bind_addr);
   socket.bind(&socket_addr)?;
   socket.listen(128)?;