prepare for Rust 1.50+ by boxing large futures

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 branch introduces several `BoxService` layers, primarily around
`Switch` layers, to reduce the size of future types, so that the proxy
can successfully be compiled on Rust 1.50+.

This branch does *not* update the toolchain version. The Rust 1.51
toolchain also introduces some new clippy lints, which trigger on some
code patterns that are very common in the current proxy. Therefore, the
diff from a compiler update will be much larger than just adding
additional boxing. I'll land the compiler update in a separate branch
after this merges.

Author: hawkw

linkerd/app/gateway/src/lib.rs

@@ -117,6 +117,7 @@ where
         .push_tcp_endpoint()
         .push_tcp_logical(resolve.clone())
         .into_stack()
+        // .push_on_response(svc::BoxService::layer())
         .push_request_filter(
             |(p, _): (Option<profiles::Receiver>, _)| -> Result<_, Error> {
                 let profile = p.ok_or_else(discovery_rejected)?;
@@ -229,6 +230,8 @@ where
         )
         .push_http_server()
         .into_stack()
+        .push_on_response(svc::BoxService::layer())
+        .push(svc::BoxNewService::layer())
         .push_switch(
             |GatewayTransportHeader {
                  target,
@@ -254,6 +257,8 @@ where
             },
             legacy_http,
         )
+        .push_on_response(svc::BoxService::layer())
+        .push(svc::BoxNewService::layer())
         .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,13 +243,16 @@ 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)
             .push(detect::NewDetectService::layer(
                 config.detect_protocol_timeout,
                 http::DetectHttp::default(),
             ))
+            // .push_on_response(svc::BoxService::layer())
             .push_request_filter(require_id)
             .push(self.runtime.metrics.transport.layer_accept())
             .push_request_filter(TcpAccept::try_from)
@@ -257,6 +261,7 @@ where
                 config.detect_protocol_timeout,
             ))
             .instrument(|_: &_| debug_span!("proxy"))
+            .push_on_response(svc::BoxService::layer())
             .push_switch(
                 disable_detect,
                 self.clone()
@@ -267,6 +272,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,12 +283,15 @@ where
                     .push_direct(gateway)
                     .stack
                     .instrument(|_: &_| debug_span!("direct"))
+                    .push_on_response(svc::BoxService::layer())
+                    .push(svc::BoxNewService::layer())
                     .into_inner(),
             )
             .instrument(|a: &T| {
                 let OrigDstAddr(target_addr) = a.param();
                 info_span!("server", port = target_addr.port())
             })
+            // .push_on_response(svc::BoxService::layer())
             .into_inner()
     }
 }

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,
     {
@@ -74,6 +75,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,14 @@ 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
+            + fmt::Debug
+            + Send
+            + Sync
+            + Unpin
+            + 'static,
     {
         let http = self
             .clone()

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

@@ -21,12 +21,20 @@ 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
+            + 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 +90,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,
@@ -70,7 +73,9 @@ impl<C> Outbound<C> {
                 "host",
                 CANONICAL_DST_HEADER,
             ]))
-            .push_on_response(http::BoxResponse::layer());
+            .push_on_response(http::BoxResponse::layer())
+            // .push_on_response(svc::BoxService::layer())
+            ;
 
         Outbound {
             config,

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

@@ -30,13 +30,17 @@ 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>
+            + Clone
+            + Send
+            + Sync
+            + 'static,
         R::Resolution: Send,
         R::Future: Send + Unpin,
     {
@@ -152,7 +156,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

@@ -148,6 +148,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,17 @@ 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
+            + 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,