Prepare for Rust 1.50+ by boxing large futures (#1003)

Version 1.50 of the Rust compiler introduced a regression
(rust-lang/rust#84873) that results in the compiler using extremely
large amounts of memory (and eventually getting OOM killed) when
compiling code involving very large nested types. This regression is
triggered by a number of future types in the proxy.

This change adds several `BoxService` layers--primarily within
`switch` layers--to reduce the size of future types so that the proxy
can successfully be compiled on Rust 1.50+.

This change does not update the Rust toolchain version.

Co-authored-by: Oliver Gould <[email protected]>

Author: hawkw

linkerd/app/gateway/src/lib.rs

@@ -218,8 +218,7 @@ where
         .push(detect::NewDetectService::layer(
             detect_protocol_timeout,
             http::DetectHttp::default(),
-        ))
-        .into_inner();
+        ));
 
     // When a transported connection is received, use the header's target to
     // drive routing.
@@ -232,6 +231,8 @@ where
         )
         .push_http_server()
         .into_stack()
+        .push_on_response(svc::BoxService::layer())
+        .push(svc::BoxNewService::layer())
         .push_switch(
             |GatewayTransportHeader {
                  target,
@@ -248,14 +249,19 @@ where
                 })),
                 None => Ok::<_, Never>(svc::Either::B(target)),
             },
-            tcp.into_inner(),
+            tcp.push_on_response(svc::BoxService::layer())
+                .push(svc::BoxNewService::layer())
+                .into_inner(),
         )
         .push_switch(
             |gw| match gw {
                 GatewayConnection::TransportHeader(t) => Ok::<_, Never>(svc::Either::A(t)),
                 GatewayConnection::Legacy(c) => Ok(svc::Either::B(c)),
             },
-            legacy_http,
+            legacy_http
+                .push_on_response(svc::BoxService::layer())
+                .push(svc::BoxNewService::layer())
+                .into_inner(),
         )
         .into_inner()
 }

linkerd/app/inbound/src/lib.rs

@@ -234,6 +234,7 @@ where
             .push_http_router(profiles)
             .push_http_server()
             .stack
+            .push_on_response(svc::BoxService::layer())
             .push_map_target(HttpAccept::from)
             .push(svc::UnwrapOr::layer(
                 // When HTTP detection fails, forward the connection to the
@@ -242,6 +243,8 @@ where
                     .push_tcp_forward(server_port)
                     .stack
                     .push_map_target(TcpEndpoint::from)
+                    .push_on_response(svc::BoxService::layer())
+                    .push(svc::BoxNewService::layer())
                     .into_inner(),
             ))
             .push_map_target(detect::allow_timeout)
@@ -257,6 +260,7 @@ where
                 config.detect_protocol_timeout,
             ))
             .instrument(|_: &_| debug_span!("proxy"))
+            .push_on_response(svc::BoxService::layer())
             .push_switch(
                 disable_detect,
                 self.clone()
@@ -267,6 +271,8 @@ where
                     .push_map_target(TcpAccept::port_skipped)
                     .check_new_service::<T, _>()
                     .instrument(|_: &T| debug_span!("forward"))
+                    .push_on_response(svc::BoxService::layer())
+                    .push(svc::BoxNewService::layer())
                     .into_inner(),
             )
             .check_new_service::<T, I>()
@@ -276,6 +282,8 @@ where
                     .push_direct(gateway)
                     .stack
                     .instrument(|_: &_| debug_span!("direct"))
+                    .push_on_response(svc::BoxService::layer())
+                    .push(svc::BoxNewService::layer())
                     .into_inner(),
             )
             .instrument(|a: &T| {

linkerd/app/outbound/src/discover.rs

@@ -27,10 +27,11 @@ impl<N> Outbound<N> {
         N: svc::NewService<(Option<profiles::Receiver>, tcp::Accept), Service = NSvc>
             + Clone
             + Send
+            + Sync
             + 'static,
         NSvc: svc::Service<SensorIo<I>, Response = (), Error = Error> + Send + 'static,
         NSvc::Future: Send,
-        P: profiles::GetProfile<profiles::LookupAddr> + Clone + Send + 'static,
+        P: profiles::GetProfile<profiles::LookupAddr> + Clone + Send + Sync + 'static,
         P::Future: Send,
         P::Error: Send,
     {
@@ -80,6 +81,9 @@ impl<N> Outbound<N> {
             .push_cache(config.proxy.cache_max_idle_age)
             .instrument(|a: &tcp::Accept| info_span!("server", orig_dst = %a.orig_dst))
             .push_request_filter(|t: T| tcp::Accept::try_from(t.param()))
+            // Boxing is necessary purely to limit the link-time overhead of
+            // having enormous types.
+            .push(svc::BoxNewService::layer())
             .check_new_service::<T, I>();
 
         Outbound {

linkerd/app/outbound/src/endpoint.rs

@@ -205,7 +205,8 @@ impl<S> Outbound<S> {
         S::Response:
             tls::HasNegotiatedProtocol + io::AsyncRead + io::AsyncWrite + Send + Unpin + 'static,
         S::Future: Send + Unpin,
-        I: io::AsyncRead + io::AsyncWrite + io::PeerAddr + fmt::Debug + Send + Unpin + 'static,
+        I: io::AsyncRead + io::AsyncWrite + io::PeerAddr,
+        I: fmt::Debug + Send + Sync + Unpin + 'static,
     {
         let http = self
             .clone()

linkerd/app/outbound/src/http/detect.rs

@@ -21,12 +21,14 @@ impl<N> Outbound<N> {
             > + Clone,
     >
     where
-        I: io::AsyncRead + io::AsyncWrite + io::PeerAddr + std::fmt::Debug + Send + Unpin + 'static,
-        N: svc::NewService<T, Service = NSvc> + Clone + Send + 'static,
-        NSvc: svc::Service<io::EitherIo<I, io::PrefixedIo<I>>, Response = ()> + Send + 'static,
+        I: io::AsyncRead + io::AsyncWrite + io::PeerAddr,
+        I: std::fmt::Debug + Send + Sync + Unpin + 'static,
+        N: svc::NewService<T, Service = NSvc> + Clone + Send + Sync + 'static,
+        NSvc:
+            svc::Service<io::EitherIo<I, io::PrefixedIo<I>>, Response = ()> + Send + Sync + 'static,
         NSvc::Error: Into<Error>,
         NSvc::Future: Send,
-        H: svc::NewService<U, Service = HSvc> + Clone + Send + 'static,
+        H: svc::NewService<U, Service = HSvc> + Clone + Send + Sync + 'static,
         HSvc: svc::Service<http::Request<http::BoxBody>, Response = http::Response<http::BoxBody>>,
         HSvc: Clone + Send + Sync + Unpin + 'static,
         HSvc::Error: Into<Error>,
@@ -82,7 +84,10 @@ impl<N> Outbound<N> {
                 },
                 skipped,
             )
-            .check_new_service::<T, _>();
+            .check_new_service::<T, _>()
+            // Boxing is necessary purely to limit the link-time overhead of
+            // having enormous types.
+            .push(svc::BoxNewService::layer());
 
         Outbound {
             config,

linkerd/app/outbound/src/http/endpoint.rs

@@ -19,7 +19,10 @@ impl<C> Outbound<C> {
                     Error = Error,
                     Future = impl Send,
                 >,
-            > + Clone,
+            > + Clone
+            + Send
+            + Sync
+            + 'static,
     >
     where
         T: Clone + Send + Sync + 'static,
@@ -33,7 +36,7 @@ impl<C> Outbound<C> {
         C: svc::Service<T> + Clone + Send + Sync + Unpin + 'static,
         C::Response: io::AsyncRead + io::AsyncWrite + Send + Unpin,
         C::Error: Into<Error>,
-        C::Future: Send + Unpin,
+        C::Future: Send + Unpin + 'static,
     {
         let Self {
             config,

linkerd/app/outbound/src/http/logical.rs

@@ -30,13 +30,14 @@ impl<E> Outbound<E> {
     where
         B: http::HttpBody<Error = Error> + std::fmt::Debug + Default + Send + 'static,
         B::Data: Send + 'static,
-        E: svc::NewService<Endpoint, Service = ESvc> + Clone + Send + 'static,
+        E: svc::NewService<Endpoint, Service = ESvc> + Clone + Send + Sync + 'static,
         ESvc: svc::Service<http::Request<http::BoxBody>, Response = http::Response<http::BoxBody>>
             + Send
             + 'static,
         ESvc::Error: Into<Error>,
         ESvc::Future: Send,
-        R: Resolve<ConcreteAddr, Error = Error, Endpoint = Metadata> + Clone + Send + 'static,
+        R: Resolve<ConcreteAddr, Error = Error, Endpoint = Metadata>,
+        R: Clone + Send + Sync + 'static,
         R::Resolution: Send,
         R::Future: Send + Unpin,
     {
@@ -152,7 +153,11 @@ impl<E> Outbound<E> {
                     .push(http::strip_header::request::layer(DST_OVERRIDE_HEADER))
                     .push(http::BoxResponse::layer()),
             )
-            .instrument(|l: &Logical| debug_span!("logical", dst = %l.logical_addr));
+            .instrument(|l: &Logical| debug_span!("logical", dst = %l.logical_addr))
+            // Boxing is necessary purely to limit the link-time overhead of
+            // having enormous types.
+            .push(svc::BoxNewService::layer())
+            .push_on_response(svc::BoxService::layer());
 
         Outbound {
             config,

linkerd/app/outbound/src/http/server.rs

@@ -17,7 +17,7 @@ impl<N> Outbound<N> {
     >
     where
         T: svc::Param<http::normalize_uri::DefaultAuthority>,
-        N: svc::NewService<T, Service = NSvc> + Clone + Send + 'static,
+        N: svc::NewService<T, Service = NSvc> + Clone + Send + Sync + 'static,
         NSvc: svc::Service<http::Request<http::BoxBody>, Response = http::Response<http::BoxBody>>,
         NSvc: Send + 'static,
         NSvc::Error: Into<Error>,
@@ -62,7 +62,10 @@ impl<N> Outbound<N> {
             .push(http::NewNormalizeUri::layer())
             // Record when a HTTP/1 URI originated in absolute form
             .push_on_response(http::normalize_uri::MarkAbsoluteForm::layer())
-            .check_new_service::<T, http::Request<http::BoxBody>>();
+            .check_new_service::<T, http::Request<http::BoxBody>>()
+            // Boxing is necessary purely to limit the link-time overhead of
+            // having enormous types.
+            .push(svc::BoxNewService::layer());
 
         Outbound {
             config,

linkerd/app/outbound/src/ingress.rs

@@ -161,6 +161,7 @@ impl<H> Outbound<H> {
         // Boxing is necessary purely to limit the link-time overhead of
         // having enormous types.
         .push(svc::BoxNewService::layer())
+        .push_on_response(svc::BoxService::layer())
         .check_new_service::<T, I>()
         .into_inner()
     }

linkerd/app/outbound/src/logical.rs

@@ -131,10 +131,11 @@ impl<C> Outbound<C> {
             tls::HasNegotiatedProtocol + io::AsyncRead + io::AsyncWrite + Send + Unpin + 'static,
         C::Future: Send + Unpin,
         R: Clone + Send + 'static,
-        R: Resolve<ConcreteAddr, Endpoint = Metadata, Error = Error>,
+        R: Resolve<ConcreteAddr, Endpoint = Metadata, Error = Error> + Sync,
         R::Resolution: Send,
         R::Future: Send + Unpin,
-        I: io::AsyncRead + io::AsyncWrite + io::PeerAddr + fmt::Debug + Send + Unpin + 'static,
+        I: io::AsyncRead + io::AsyncWrite + io::PeerAddr,
+        I: fmt::Debug + Send + Sync + Unpin + 'static,
     {
         let http = self
             .clone()

linkerd/app/outbound/src/tcp/logical.rs

@@ -31,9 +31,14 @@ where
     >
     where
         I: io::AsyncRead + io::AsyncWrite + std::fmt::Debug + Send + Unpin + 'static,
-        R: Resolve<ConcreteAddr, Endpoint = Metadata, Error = Error> + Clone + Send + 'static,
+        R: Resolve<ConcreteAddr, Endpoint = Metadata, Error = Error>
+            + Clone
+            + Send
+            + Sync
+            + 'static,
         R::Resolution: Send,
         R::Future: Send + Unpin,
+        C: Send + Sync + 'static,
     {
         let Self {
             config,
@@ -101,7 +106,10 @@ where
             .push_cache(cache_max_idle_age)
             .check_new_service::<Logical, I>()
             .instrument(|_: &Logical| debug_span!("tcp"))
-            .check_new_service::<Logical, I>();
+            .check_new_service::<Logical, I>()
+            // Boxing is necessary purely to limit the link-time overhead of
+            // having enormous types.
+            .push(svc::BoxNewService::layer());
 
         Outbound {
             config,