feat(ext/net): Refactor TCP socket listeners for future clustering mode (#23037)

Changes:

- Implements a TCP socket listener that will allow for round-robin
load-balancing in-process.
 - Cleans up the raw networking code to make it easier to work with.

1 files changed, 34 insertions, 57 deletions

diff --git a/ext/net/ops_tls.rs b/ext/net/ops_tls.rs
index 874f795f2..c0ac31586 100644
--- a/ext/net/ops_tls.rs
+++ b/ext/net/ops_tls.rs
@@ -3,8 +3,10 @@
 use crate::io::TcpStreamResource;
 use crate::ops::IpAddr;
 use crate::ops::TlsHandshakeInfo;
+use crate::raw::NetworkListenerResource;
 use crate::resolve_addr::resolve_addr;
 use crate::resolve_addr::resolve_addr_sync;
+use crate::tcp::TcpListener;
 use crate::DefaultTlsOptions;
 use crate::NetPermissions;
 use crate::UnsafelyIgnoreCertificateErrors;
@@ -36,9 +38,6 @@ use deno_tls::TlsKeys;
 use rustls_tokio_stream::TlsStreamRead;
 use rustls_tokio_stream::TlsStreamWrite;
 use serde::Deserialize;
-use socket2::Domain;
-use socket2::Socket;
-use socket2::Type;
 use std::borrow::Cow;
 use std::cell::RefCell;
 use std::convert::From;
@@ -47,13 +46,13 @@ use std::fs::File;
 use std::io::BufReader;
 use std::io::ErrorKind;
 use std::io::Read;
+use std::net::SocketAddr;
 use std::num::NonZeroUsize;
 use std::path::Path;
 use std::rc::Rc;
 use std::sync::Arc;
 use tokio::io::AsyncReadExt;
 use tokio::io::AsyncWriteExt;
-use tokio::net::TcpListener;
 use tokio::net::TcpStream;
 
 pub use rustls_tokio_stream::TlsStream;
@@ -61,6 +60,23 @@ pub use rustls_tokio_stream::TlsStream;
 pub(crate) const TLS_BUFFER_SIZE: Option<NonZeroUsize> =
   NonZeroUsize::new(65536);
 
+pub struct TlsListener {
+  pub(crate) tcp_listener: TcpListener,
+  pub(crate) tls_config: Arc<ServerConfig>,
+}
+
+impl TlsListener {
+  pub async fn accept(&self) -> std::io::Result<(TlsStream, SocketAddr)> {
+    let (tcp, addr) = self.tcp_listener.accept().await?;
+    let tls =
+      TlsStream::new_server_side(tcp, self.tls_config.clone(), TLS_BUFFER_SIZE);
+    Ok((tls, addr))
+  }
+  pub fn local_addr(&self) -> std::io::Result<SocketAddr> {
+    self.tcp_listener.local_addr()
+  }
+}
+
 #[derive(Debug)]
 pub struct TlsStreamResource {
   rd: AsyncRefCell<TlsStreamRead>,
@@ -399,22 +415,6 @@ fn load_private_keys_from_file(
   load_private_keys(&key_bytes)
 }
 
-pub struct TlsListenerResource {
-  pub(crate) tcp_listener: AsyncRefCell<TcpListener>,
-  pub(crate) tls_config: Arc<ServerConfig>,
-  cancel_handle: CancelHandle,
-}
-
-impl Resource for TlsListenerResource {
-  fn name(&self) -> Cow<str> {
-    "tlsListener".into()
-  }
-
-  fn close(self: Rc<Self>) {
-    self.cancel_handle.cancel();
-  }
-}
-
 #[derive(Deserialize)]
 #[serde(rename_all = "camelCase")]
 pub struct ListenTlsArgs {
@@ -470,31 +470,14 @@ where
   let bind_addr = resolve_addr_sync(&addr.hostname, addr.port)?
     .next()
     .ok_or_else(|| generic_error("No resolved address found"))?;
-  let domain = if bind_addr.is_ipv4() {
-    Domain::IPV4
-  } else {
-    Domain::IPV6
-  };
-  let socket = Socket::new(domain, Type::STREAM, None)?;
-  #[cfg(not(windows))]
-  socket.set_reuse_address(true)?;
-  if args.reuse_port {
-    #[cfg(any(target_os = "android", target_os = "linux"))]
-    socket.set_reuse_port(true)?;
-  }
-  let socket_addr = socket2::SockAddr::from(bind_addr);
-  socket.bind(&socket_addr)?;
-  socket.listen(128)?;
-  socket.set_nonblocking(true)?;
-  let std_listener: std::net::TcpListener = socket.into();
-  let tcp_listener = TcpListener::from_std(std_listener)?;
+
+  let tcp_listener = TcpListener::bind_direct(bind_addr, args.reuse_port)?;
   let local_addr = tcp_listener.local_addr()?;
 
-  let tls_listener_resource = TlsListenerResource {
-    tcp_listener: AsyncRefCell::new(tcp_listener),
-    tls_config: Arc::new(tls_config),
-    cancel_handle: Default::default(),
-  };
+  let tls_listener_resource = NetworkListenerResource::new(TlsListener {
+    tcp_listener,
+    tls_config: tls_config.into(),
+  });
 
   let rid = state.resource_table.add(tls_listener_resource);
 
@@ -510,16 +493,16 @@ pub async fn op_net_accept_tls(
   let resource = state
     .borrow()
     .resource_table
-    .get::<TlsListenerResource>(rid)
+    .get::<NetworkListenerResource<TlsListener>>(rid)
     .map_err(|_| bad_resource("Listener has been closed"))?;
 
-  let cancel_handle = RcRef::map(&resource, |r| &r.cancel_handle);
-  let tcp_listener = RcRef::map(&resource, |r| &r.tcp_listener)
+  let cancel_handle = RcRef::map(&resource, |r| &r.cancel);
+  let listener = RcRef::map(&resource, |r| &r.listener)
     .try_borrow_mut()
     .ok_or_else(|| custom_error("Busy", "Another accept task is ongoing"))?;
 
-  let (tcp_stream, remote_addr) =
-    match tcp_listener.accept().try_or_cancel(&cancel_handle).await {
+  let (tls_stream, remote_addr) =
+    match listener.accept().try_or_cancel(&cancel_handle).await {
       Ok(tuple) => tuple,
       Err(err) if err.kind() == ErrorKind::Interrupted => {
         // FIXME(bartlomieju): compatibility with current JS implementation.
@@ -528,14 +511,7 @@ pub async fn op_net_accept_tls(
       Err(err) => return Err(err.into()),
     };
 
-  let local_addr = tcp_stream.local_addr()?;
-
-  let tls_stream = TlsStream::new_server_side(
-    tcp_stream,
-    resource.tls_config.clone(),
-    TLS_BUFFER_SIZE,
-  );
-
+  let local_addr = tls_stream.local_addr()?;
   let rid = {
     let mut state_ = state.borrow_mut();
     state_
@@ -555,6 +531,7 @@ pub async fn op_tls_handshake(
   let resource = state
     .borrow()
     .resource_table
-    .get::<TlsStreamResource>(rid)?;
+    .get::<TlsStreamResource>(rid)
+    .map_err(|_| bad_resource("Listener has been closed"))?;
   resource.handshake().await
 }