perf: really, really stop nagle's algorithm

src/dns/server.rs

@@ -106,9 +106,9 @@ impl Server {
             .tcp_bind(addr)
             .map_err(|e| Error::Bind(addr, e))?;
         // Save the bound address.
-        let tcp_addr = tcp_listener.local_addr().unwrap();
+        let tcp_addr = tcp_listener.local_addr();
         server.register_listener(
-            tcp_listener,
+            tcp_listener.inner(),
             Duration::from_secs(DEFAULT_TCP_REQUEST_TIMEOUT),
         );
 

src/inpod/config.rs

@@ -12,8 +12,9 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use crate::config;
+use crate::{config, socket};
 use std::sync::Arc;
+use tokio::net::TcpSocket;
 
 use super::netns::InpodNetns;
 
@@ -79,17 +80,27 @@ impl InPodSocketFactory {
 
 impl crate::proxy::SocketFactory for InPodSocketFactory {
     fn new_tcp_v4(&self) -> std::io::Result<tokio::net::TcpSocket> {
-        self.configure(tokio::net::TcpSocket::new_v4)
+        self.configure(|| {
+            TcpSocket::new_v4().and_then(|s| {
+                s.set_nodelay(true)?;
+                Ok(s)
+            })
+        })
     }
 
     fn new_tcp_v6(&self) -> std::io::Result<tokio::net::TcpSocket> {
-        self.configure(tokio::net::TcpSocket::new_v6)
+        self.configure(|| {
+            TcpSocket::new_v6().and_then(|s| {
+                s.set_nodelay(true)?;
+                Ok(s)
+            })
+        })
     }
 
-    fn tcp_bind(&self, addr: std::net::SocketAddr) -> std::io::Result<tokio::net::TcpListener> {
+    fn tcp_bind(&self, addr: std::net::SocketAddr) -> std::io::Result<socket::Listener> {
         let std_sock = self.configure(|| std::net::TcpListener::bind(addr))?;
         std_sock.set_nonblocking(true)?;
-        tokio::net::TcpListener::from_std(std_sock)
+        tokio::net::TcpListener::from_std(std_sock).map(socket::Listener::new)
     }
 
     fn udp_bind(&self, addr: std::net::SocketAddr) -> std::io::Result<tokio::net::UdpSocket> {
@@ -119,7 +130,7 @@ impl crate::proxy::SocketFactory for InPodSocketPortReuseFactory {
         self.sf.new_tcp_v6()
     }
 
-    fn tcp_bind(&self, addr: std::net::SocketAddr) -> std::io::Result<tokio::net::TcpListener> {
+    fn tcp_bind(&self, addr: std::net::SocketAddr) -> std::io::Result<socket::Listener> {
         let sock = self.sf.configure(|| match addr {
             std::net::SocketAddr::V4(_) => tokio::net::TcpSocket::new_v4(),
             std::net::SocketAddr::V6(_) => tokio::net::TcpSocket::new_v6(),
@@ -130,7 +141,7 @@ impl crate::proxy::SocketFactory for InPodSocketPortReuseFactory {
         }
 
         sock.bind(addr)?;
-        sock.listen(128)
+        sock.listen(128).map(socket::Listener::new)
     }
 
     fn udp_bind(&self, addr: std::net::SocketAddr) -> std::io::Result<tokio::net::UdpSocket> {
@@ -210,7 +221,7 @@ mod test {
 
         let sock_addr: std::net::SocketAddr = "127.0.0.1:8080".parse().unwrap();
         {
-            let s = sf.tcp_bind(sock_addr).unwrap();
+            let s = sf.tcp_bind(sock_addr).unwrap().inner();
 
             // make sure mark nad port re-use are set
             let sock_ref = socket2::SockRef::from(&s);
@@ -257,7 +268,7 @@ mod test {
 
         let sock_addr: std::net::SocketAddr = "127.0.0.1:8080".parse().unwrap();
         {
-            let s = sf.tcp_bind(sock_addr).unwrap();
+            let s = sf.tcp_bind(sock_addr).unwrap().inner();
 
             // make sure mark nad port re-use are set
             let sock_ref = socket2::SockRef::from(&s);

src/proxy.rs

@@ -59,7 +59,7 @@ pub trait SocketFactory {
 
     fn new_tcp_v6(&self) -> std::io::Result<TcpSocket>;
 
-    fn tcp_bind(&self, addr: SocketAddr) -> std::io::Result<TcpListener>;
+    fn tcp_bind(&self, addr: SocketAddr) -> std::io::Result<socket::Listener>;
 
     fn udp_bind(&self, addr: SocketAddr) -> std::io::Result<tokio::net::UdpSocket>;
 }
@@ -69,17 +69,23 @@ pub struct DefaultSocketFactory;
 
 impl SocketFactory for DefaultSocketFactory {
     fn new_tcp_v4(&self) -> std::io::Result<TcpSocket> {
-        TcpSocket::new_v4()
+        TcpSocket::new_v4().and_then(|s| {
+            s.set_nodelay(true)?;
+            Ok(s)
+        })
     }
 
     fn new_tcp_v6(&self) -> std::io::Result<TcpSocket> {
-        TcpSocket::new_v6()
+        TcpSocket::new_v6().and_then(|s| {
+            s.set_nodelay(true)?;
+            Ok(s)
+        })
     }
 
-    fn tcp_bind(&self, addr: SocketAddr) -> std::io::Result<TcpListener> {
+    fn tcp_bind(&self, addr: SocketAddr) -> std::io::Result<socket::Listener> {
         let std_sock = std::net::TcpListener::bind(addr)?;
         std_sock.set_nonblocking(true)?;
-        TcpListener::from_std(std_sock)
+        TcpListener::from_std(std_sock).map(socket::Listener::new)
     }
 
     fn udp_bind(&self, addr: SocketAddr) -> std::io::Result<tokio::net::UdpSocket> {
@@ -411,12 +417,12 @@ impl TryFrom<&str> for TraceParent {
 
 pub(super) fn maybe_set_transparent(
     pi: &ProxyInputs,
-    listener: &TcpListener,
+    listener: &socket::Listener,
 ) -> Result<bool, Error> {
     Ok(match pi.cfg.enable_original_source {
         Some(true) => {
             // Explicitly enabled. Return error if we cannot set it.
-            socket::set_transparent(listener)?;
+            listener.set_transparent()?;
             true
         }
         Some(false) => {
@@ -425,7 +431,7 @@ pub(super) fn maybe_set_transparent(
         }
         None => {
             // Best effort
-            socket::set_transparent(listener).is_ok()
+            listener.set_transparent().is_ok()
         }
     })
 }

src/proxy/inbound.rs

@@ -24,7 +24,7 @@ use futures::stream::StreamExt;
 
 use http::{Method, Response, StatusCode};
 
-use tokio::net::{TcpListener, TcpStream};
+use tokio::net::TcpStream;
 
 use tracing::{debug, info, instrument, trace_span, Instrument};
 
@@ -41,7 +41,7 @@ use crate::socket::to_canonical;
 use crate::state::service::Service;
 use crate::state::workload::address::Address;
 use crate::state::workload::application_tunnel::Protocol as AppProtocol;
-use crate::{assertions, copy, proxy, strng, tls};
+use crate::{assertions, copy, proxy, socket, strng, tls};
 
 use crate::proxy::h2;
 use crate::state::workload::{self, NetworkAddress, Workload};
@@ -50,14 +50,14 @@ use crate::strng::Strng;
 use crate::tls::TlsError;
 
 pub(super) struct Inbound {
-    listener: TcpListener,
+    listener: socket::Listener,
     drain: Watch,
     pi: ProxyInputs,
 }
 
 impl Inbound {
     pub(super) async fn new(mut pi: ProxyInputs, drain: Watch) -> Result<Inbound, Error> {
-        let listener: TcpListener = pi
+        let listener = pi
             .socket_factory
             .tcp_bind(pi.cfg.inbound_addr)
             .map_err(|e| Error::Bind(pi.cfg.inbound_addr, e))?;
@@ -69,7 +69,7 @@ impl Inbound {
             pi.cfg = Arc::new(cfg);
         }
         info!(
-            address=%listener.local_addr().expect("local_addr available"),
+            address=%listener.local_addr(),
             component="inbound",
             transparent,
             "listener established",
@@ -82,7 +82,7 @@ impl Inbound {
     }
 
     pub(super) fn address(&self) -> SocketAddr {
-        self.listener.local_addr().expect("local_addr available")
+        self.listener.local_addr()
     }
 
     pub(super) async fn run(self, illegal_ports: Arc<HashSet<u16>>) {
@@ -91,7 +91,10 @@ impl Inbound {
             cert_manager: self.pi.cert_manager.clone(),
             network: strng::new(&self.pi.cfg.network),
         };
-        let stream = crate::hyper_util::tls_server(acceptor, self.listener);
+
+        // Safety: we set nodelay directly in tls_server, so it is safe to convert to a normal listener.
+        // Although, that is *after* the TLS handshake; in theory we may get some benefits to setting it earlier.
+        let stream = crate::hyper_util::tls_server(acceptor, self.listener.inner());
         let mut stream = stream.take_until(Box::pin(self.drain.signaled()));
 
         let (sub_drain_signal, sub_drain) = drain::channel();
@@ -289,12 +292,8 @@ impl Inbound {
         };
 
         let orig_src = enable_original_source.then_some(source_ip);
-        let stream = super::freebind_connect(orig_src, upstream_addr, pi.socket_factory.as_ref())
-            .await
-            .and_then(|s| {
-                s.set_nodelay(true)?;
-                Ok(s)
-            });
+        let stream =
+            super::freebind_connect(orig_src, upstream_addr, pi.socket_factory.as_ref()).await;
         let mut stream = match stream {
             Err(err) => {
                 result_tracker.record(Err(err));

src/proxy/inbound_passthrough.rs

@@ -18,7 +18,7 @@ use std::sync::Arc;
 use std::time::Instant;
 
 use drain::Watch;
-use tokio::net::{TcpListener, TcpStream};
+use tokio::net::TcpStream;
 
 use tracing::{error, info, trace, Instrument};
 
@@ -32,7 +32,7 @@ use crate::{assertions, copy, rbac, strng};
 use crate::{proxy, socket};
 
 pub(super) struct InboundPassthrough {
-    listener: TcpListener,
+    listener: socket::Listener,
     pi: ProxyInputs,
     drain: Watch,
 }
@@ -42,7 +42,7 @@ impl InboundPassthrough {
         mut pi: ProxyInputs,
         drain: Watch,
     ) -> Result<InboundPassthrough, Error> {
-        let listener: TcpListener = pi
+        let listener = pi
             .socket_factory
             .tcp_bind(pi.cfg.inbound_plaintext_addr)
             .map_err(|e| Error::Bind(pi.cfg.inbound_plaintext_addr, e))?;
@@ -56,7 +56,7 @@ impl InboundPassthrough {
         }
 
         info!(
-            address=%listener.local_addr().expect("local_addr available"),
+            address=%listener.local_addr(),
             component="inbound plaintext",
             transparent,
             "listener established",
@@ -69,7 +69,7 @@ impl InboundPassthrough {
     }
 
     pub(super) fn address(&self) -> SocketAddr {
-        self.listener.local_addr().expect("local_addr available")
+        self.listener.local_addr()
     }
 
     pub(super) async fn run(self, illegal_ports: Arc<HashSet<u16>>) {

src/proxy/outbound.rs

@@ -22,7 +22,7 @@ use drain::Watch;
 
 use hyper::header::FORWARDED;
 
-use tokio::net::{TcpListener, TcpStream};
+use tokio::net::TcpStream;
 
 use tracing::{debug, error, info, info_span, trace_span, warn, Instrument};
 
@@ -43,12 +43,12 @@ use crate::{assertions, copy, proxy, socket, strng};
 pub struct Outbound {
     pi: ProxyInputs,
     drain: Watch,
-    listener: TcpListener,
+    listener: socket::Listener,
 }
 
 impl Outbound {
     pub(super) async fn new(mut pi: ProxyInputs, drain: Watch) -> Result<Outbound, Error> {
-        let listener: TcpListener = pi
+        let listener = pi
             .socket_factory
             .tcp_bind(pi.cfg.outbound_addr)
             .map_err(|e| Error::Bind(pi.cfg.outbound_addr, e))?;
@@ -61,7 +61,7 @@ impl Outbound {
         }
 
         info!(
-            address=%listener.local_addr().expect("local_addr available"),
+            address=%listener.local_addr(),
             component="outbound",
             transparent,
             "listener established",
@@ -74,7 +74,7 @@ impl Outbound {
     }
 
     pub(super) fn address(&self) -> SocketAddr {
-        self.listener.local_addr().expect("local_addr available")
+        self.listener.local_addr()
     }
 
     pub(super) async fn run(self) {
@@ -104,6 +104,7 @@ impl Outbound {
                             id: TraceParent::new(),
                             pool: pool.clone(),
                         };
+                        stream.set_nodelay(true).unwrap();
                         let span = info_span!("outbound", id=%oc.id);
                         let serve_outbound_connection = (async move {
                             debug!(dur=?start_outbound_instant.elapsed(), id=%oc.id, "outbound spawn START");

src/proxy/pool.rs

@@ -95,7 +95,6 @@ impl ConnSpawner {
         let connector = cert.outbound_connector(key.dst_id.clone())?;
         let tcp_stream =
             super::freebind_connect(local, key.dst, self.socket_factory.as_ref()).await?;
-        tcp_stream.set_nodelay(true)?;
         let tls_stream = connector.connect(tcp_stream).await?;
         trace!("connector connected, handshaking");
         let sender =

src/proxy/socks5.rs

@@ -21,7 +21,7 @@ use std::sync::Arc;
 
 use tokio::io::AsyncReadExt;
 use tokio::io::AsyncWriteExt;
-use tokio::net::{TcpListener, TcpStream};
+use tokio::net::TcpStream;
 use tracing::{error, info};
 
 use crate::proxy::outbound::OutboundConnection;
@@ -30,19 +30,19 @@ use crate::socket;
 
 pub(super) struct Socks5 {
     pi: ProxyInputs,
-    listener: TcpListener,
+    listener: socket::Listener,
     drain: Watch,
 }
 
 impl Socks5 {
     pub(super) async fn new(pi: ProxyInputs, drain: Watch) -> Result<Socks5, Error> {
-        let listener: TcpListener = pi
+        let listener = pi
             .socket_factory
             .tcp_bind(pi.cfg.socks5_addr.unwrap())
             .map_err(|e| Error::Bind(pi.cfg.socks5_addr.unwrap(), e))?;
 
         info!(
-            address=%listener.local_addr().expect("local_addr available"),
+            address=%listener.local_addr(),
             component="socks5",
             "listener established",
         );
@@ -55,7 +55,7 @@ impl Socks5 {
     }
 
     pub(super) fn address(&self) -> SocketAddr {
-        self.listener.local_addr().expect("local_addr available")
+        self.listener.local_addr()
     }
 
     pub async fn run(self) {