feat: expose tcp_nodelay for clients and servers

tonic-examples/Cargo.toml

@@ -80,7 +80,7 @@ tower = { git = "https:/tower-rs/tower" }
 # Required for routeguide
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
-rand = "0.7.2"
+rand = "0.6"
 
 # Required for wellknown types
 prost-types = "0.5"

tonic/Cargo.toml

@@ -76,7 +76,7 @@ rustls-native-certs = { version = "0.1", optional = true }
 [dev-dependencies]
 tokio = { version = "0.2", features = ["rt-core", "macros"] }
 static_assertions = "1.0"
-rand = "0.7.2"
+rand = "0.6"
 criterion = "0.3"
 
 [package.metadata.docs.rs]

tonic/src/transport/endpoint.rs

@@ -30,6 +30,7 @@ pub struct Endpoint {
  pub(super) init_stream_window_size: Option<u32>,
  pub(super) init_connection_window_size: Option<u32>,
  pub(super) tcp_keepalive: Option<Duration>,
+ pub(super) tcp_nodelay: bool,
 }
 
 impl Endpoint {
@@ -171,6 +172,14 @@ impl Endpoint {
  }
  }
 
+ /// Set the value of `TCP_NODELAY` option for accepted connections. Enabled by default.
+ pub fn tcp_nodelay(self, enabled: bool) -> Self {
+ Endpoint {
+ tcp_nodelay: enabled,
+ ..self
+ }
+ }
+
  /// Create a channel from this config.
  pub async fn connect(&self) -> Result<Channel, super::Error> {
  Channel::connect(self.clone()).await
@@ -191,6 +200,7 @@ impl From<Uri> for Endpoint {
  init_stream_window_size: None,
  init_connection_window_size: None,
  tcp_keepalive: None,
+ tcp_nodelay: true,
  }
  }
 }

tonic/src/transport/server.rs

@@ -56,6 +56,7 @@ pub struct Server {
  init_connection_window_size: Option<u32>,
  max_concurrent_streams: Option<u32>,
  tcp_keepalive: Option<Duration>,
+ tcp_nodelay: bool,
 }
 
 /// A stack based `Service` router.
@@ -77,7 +78,10 @@ pub trait ServiceName {
 impl Server {
  /// Create a new server builder that can configure a [`Server`].
  pub fn builder() -> Self {
- Default::default()
+ Server {
+ tcp_nodelay: true,
+ ..Default::default()
+ }
  }
 }
 
@@ -164,6 +168,14 @@ impl Server {
  }
  }
 
+ /// Set the value of `TCP_NODELAY` option for accepted connections. Enabled by default.
+ pub fn tcp_nodelay(self, enabled: bool) -> Self {
+ Server {
+ tcp_nodelay: enabled,
+ ..self
+ }
+ }
+
  /// Intercept the execution of gRPC methods.
  ///
  /// ```
@@ -221,12 +233,13 @@ impl Server {
  let init_connection_window_size = self.init_connection_window_size;
  let init_stream_window_size = self.init_stream_window_size;
  let max_concurrent_streams = self.max_concurrent_streams;
- let tcp_keepalive = self.tcp_keepalive;
  // let timeout = self.timeout.clone();
 
  let incoming = hyper::server::accept::from_stream::<_, _, crate::Error>(
  async_stream::try_stream! {
- let mut tcp = TcpIncoming::bind(addr, tcp_keepalive)?;
+ let mut tcp = TcpIncoming::bind(addr)?
+ .set_nodelay(self.tcp_nodelay)
+ .set_keepalive(self.tcp_keepalive);
 
  while let Some(stream) = tcp.try_next().await? {
  #[cfg(feature = "tls")]
@@ -418,13 +431,20 @@ struct TcpIncoming {
 }
 
 impl TcpIncoming {
- fn bind(addr: SocketAddr, tcp_keepalive: Option<Duration>) -> Result<Self, crate::Error> {
- let mut inner = conn::AddrIncoming::bind(&addr).map_err(Box::new)?;
- inner.set_nodelay(true);
- inner.set_keepalive(tcp_keepalive);
-
+ fn bind(addr: SocketAddr) -> Result<Self, crate::Error> {
+ let inner = conn::AddrIncoming::bind(&addr).map_err(Box::new)?;
  Ok(Self { inner })
  }
+
+ fn set_nodelay(mut self, enabled: bool) -> Self {
+ self.inner.set_nodelay(enabled);
+ self
+ }
+
+ fn set_keepalive(mut self, tcp_keepalive: Option<Duration>) -> Self {
+ self.inner.set_keepalive(tcp_keepalive);
+ self
+ }
 }
 
 impl Stream for TcpIncoming {

tonic/src/transport/service/connection.rs

@@ -28,10 +28,14 @@ pub(crate) struct Connection {
 impl Connection {
  pub(crate) async fn new(endpoint: Endpoint) -> Result<Self, crate::Error> {
  #[cfg(feature = "tls")]
- let connector = connector(endpoint.tls.clone(), endpoint.tcp_keepalive);
+ let connector = connector(endpoint.tls.clone())
+ .set_keepalive(endpoint.tcp_keepalive)
+ .set_nodelay(endpoint.tcp_nodelay);
 
  #[cfg(not(feature = "tls"))]
- let connector = connector(endpoint.tcp_keepalive);
+ let connector = connector()
+ .set_keepalive(endpoint.tcp_keepalive)
+ .set_nodelay(endpoint.tcp_nodelay);
 
  let settings = Builder::new()
  .http2_initial_stream_window_size(endpoint.init_stream_window_size)

tonic/src/transport/service/connector.rs

@@ -11,17 +11,13 @@ use tower_make::MakeConnection;
 use tower_service::Service;
 
 #[cfg(not(feature = "tls"))]
-pub(crate) fn connector(tcp_keepalive: Option<Duration>) -> HttpConnector {
- let mut http = HttpConnector::new();
- http.enforce_http(false);
- http.set_nodelay(true);
- http.set_keepalive(tcp_keepalive);
- http
+pub(crate) fn connector() -> Connector {
+ Connector::new()
 }
 
 #[cfg(feature = "tls")]
-pub(crate) fn connector(tls: Option<TlsConnector>, tcp_keepalive: Option<Duration>) -> Connector {
- Connector::new(tls, tcp_keepalive)
+pub(crate) fn connector(tls: Option<TlsConnector>) -> Connector {
+ Connector::new(tls)
 }
 
 pub(crate) struct Connector {
@@ -31,13 +27,35 @@ pub(crate) struct Connector {
 }
 
 impl Connector {
+ #[cfg(not(feature = "tls"))]
+ pub(crate) fn new() -> Self {
+ Self {
+ http: Self::http_connector(),
+ }
+ }
+
  #[cfg(feature = "tls")]
- pub(crate) fn new(tls: Option<TlsConnector>, tcp_keepalive: Option<Duration>) -> Self {
+ fn new(tls: Option<TlsConnector>) -> Self {
+ Self {
+ http: Self::http_connector(),
+ tls,
+ }
+ }
+
+ pub(crate) fn set_nodelay(mut self, enabled: bool) -> Self {
+ self.http.set_nodelay(enabled);
+ self
+ }
+
+ pub(crate) fn set_keepalive(mut self, duration: Option<Duration>) -> Self {
+ self.http.set_keepalive(duration);
+ self
+ }
+
+ fn http_connector() -> HttpConnector {
  let mut http = HttpConnector::new();
  http.enforce_http(false);
- http.set_nodelay(true);
- http.set_keepalive(tcp_keepalive);
- Self { http, tls }
+ http
  }
 }