[vfs] Reduce the size of idle connections

Future objects in rust can easily grow to be several KBs. If we have a
thousand open connections each being handled by a multi KB future then
we're using MBs of memory. Most of this memory is wasted because the
connections sit idle most of the time and it's unlikely that more than a
dozen connections will ever be handling a request at the same time.

This change introduces a RequestListener that is optimized to reduce the
memory usage of idle connections. The RequestListener dynamically heap
allocates the futures for handling requests so the memory of these
futures isn't consumed while the connections are idle.

RequestListener manually implements Future to work around
rust-lang/rust#62958 and
rust-lang/rust#108906. This further reduces
the size of an idle RequestListener to typically less than 80 bytes.

RequestListener only reduces the idle memory overhead of the futures
downstream from itself. There's still work to be done to reduce the size
of futures upstream from the RequestListeners.

Bug: 398886282
Change-Id: I30bd18099de05bd57639df69d88d011f21c03335
Reviewed-on: https://fuchsia-review.googlesource.com/c/fuchsia/+/1198690
Commit-Queue: Chris Drouillard <[email protected]>
Reviewed-by: Chris Suter <[email protected]>

Author: Chris Drouillard

src/storage/lib/vfs/rust/BUILD.gn

@@ -70,6 +70,7 @@ common_vfs_args = {
     "src/pseudo_directory.rs",
     "src/remote.rs",
     "src/remote/tests.rs",
+    "src/request_handler.rs",
     "src/service.rs",
     "src/service/tests.rs",
     "src/service/tests/direct_connection.rs",

src/storage/lib/vfs/rust/src/directory/immutable/connection.rs

@@ -9,13 +9,14 @@ use crate::directory::connection::{BaseConnection, ConnectionState};
 use crate::directory::entry_container::Directory;
 use crate::execution_scope::ExecutionScope;
 use crate::node::OpenNode;
+use crate::request_handler::{RequestHandler, RequestListener};
 use crate::{ObjectRequestRef, ProtocolsExt};
 
 use fidl_fuchsia_io as fio;
 use fio::DirectoryRequest;
-use futures::TryStreamExt;
 use std::future::Future;
-use std::pin::pin;
+use std::ops::ControlFlow;
+use std::pin::Pin;
 use std::sync::Arc;
 use zx_status::Status;
 
@@ -24,19 +25,6 @@ pub struct ImmutableConnection<DirectoryType: Directory> {
 }
 
 impl<DirectoryType: Directory> ImmutableConnection<DirectoryType> {
-    async fn handle_requests<RS>(mut self, mut requests: RS)
-    where
-        RS: futures::stream::Stream<Item = Result<DirectoryRequest, fidl::Error>>,
-    {
-        let mut requests = pin!(requests);
-        while let Ok(Some(request)) = requests.try_next().await {
-            let _guard = self.base.scope.active_guard();
-            if !matches!(self.base.handle_request(request).await, Ok(ConnectionState::Alive)) {
-                break;
-            }
-        }
-    }
-
     pub fn create(
         scope: ExecutionScope,
         directory: Arc<DirectoryType>,
@@ -81,8 +69,24 @@ impl<DirectoryType: Directory> ImmutableConnection<DirectoryType> {
         let object_request = object_request.take();
         Ok(async move {
             if let Ok(requests) = object_request.into_request_stream(&connection.base).await {
-                connection.handle_requests(transform(requests)).await;
+                RequestListener::new(transform(requests), connection).await;
             }
         })
     }
 }
+
+impl<DirectoryType: Directory> RequestHandler for ImmutableConnection<DirectoryType> {
+    type Request = Result<DirectoryRequest, fidl::Error>;
+
+    async fn handle_request(self: Pin<&mut Self>, request: Self::Request) -> ControlFlow<()> {
+        let this = self.get_mut();
+        let _guard = this.base.scope.active_guard();
+        match request {
+            Ok(request) => match this.base.handle_request(request).await {
+                Ok(ConnectionState::Alive) => ControlFlow::Continue(()),
+                Ok(ConnectionState::Closed) | Err(_) => ControlFlow::Break(()),
+            },
+            Err(_) => ControlFlow::Break(()),
+        }
+    }
+}

src/storage/lib/vfs/rust/src/directory/mutable/connection.rs

@@ -14,22 +14,21 @@ use crate::execution_scope::ExecutionScope;
 use crate::name::validate_name;
 use crate::node::OpenNode;
 use crate::path::Path;
+use crate::request_handler::{RequestHandler, RequestListener};
 use crate::token_registry::{TokenInterface, TokenRegistry, Tokenizable};
 use crate::{ObjectRequestRef, ProtocolsExt};
 
 use anyhow::Error;
 use fidl::endpoints::ServerEnd;
 use fidl::Handle;
 use fidl_fuchsia_io as fio;
-use futures::{pin_mut, TryStreamExt as _};
-use pin_project::pin_project;
 use std::future::Future;
+use std::ops::ControlFlow;
 use std::pin::Pin;
 use std::sync::Arc;
 use storage_trace::{self as trace, TraceFutureExt};
 use zx_status::Status;
 
-#[pin_project]
 pub struct MutableConnection<DirectoryType: MutableDirectory> {
     base: BaseConnection<DirectoryType>,
 }
@@ -51,7 +50,7 @@ impl<DirectoryType: MutableDirectory> MutableConnection<DirectoryType> {
         let object_request = object_request.take();
         Ok(async move {
             if let Ok(requests) = object_request.into_request_stream(&connection.base).await {
-                connection.handle_requests(requests).await
+                RequestListener::new(requests, Tokenizable::new(connection)).await;
             }
         })
     }
@@ -62,7 +61,7 @@ impl<DirectoryType: MutableDirectory> MutableConnection<DirectoryType> {
     ) -> Result<ConnectionState, Error> {
         match request {
             fio::DirectoryRequest::Unlink { name, options, responder } => {
-                let result = this.as_mut().handle_unlink(name, options).await;
+                let result = this.handle_unlink(name, options).await;
                 responder.send(result.map_err(Status::into_raw))?;
             }
             fio::DirectoryRequest::GetToken { responder } => {
@@ -78,7 +77,6 @@ impl<DirectoryType: MutableDirectory> MutableConnection<DirectoryType> {
             }
             fio::DirectoryRequest::SetAttr { flags, attributes, responder } => {
                 let status = match this
-                    .as_mut()
                     .handle_update_attributes(io1_to_io2_attrs(flags, attributes))
                     .await
                 {
@@ -99,8 +97,7 @@ impl<DirectoryType: MutableDirectory> MutableConnection<DirectoryType> {
                     responder.send(Err(Status::INVALID_ARGS.into_raw()))?;
                 } else {
                     responder.send(
-                        this.as_mut()
-                            .base
+                        this.base
                             .directory
                             .create_symlink(name, target, connection)
                             .await
@@ -148,10 +145,7 @@ impl<DirectoryType: MutableDirectory> MutableConnection<DirectoryType> {
             fio::DirectoryRequest::UpdateAttributes { payload, responder } => {
                 async move {
                     responder.send(
-                        this.as_mut()
-                            .handle_update_attributes(payload)
-                            .await
-                            .map_err(Status::into_raw),
+                        this.handle_update_attributes(payload).await.map_err(Status::into_raw),
                     )
                 }
                 .trace(trace::trace_future_args!(c"storage", c"Directory::UpdateAttributes"))
@@ -165,7 +159,7 @@ impl<DirectoryType: MutableDirectory> MutableConnection<DirectoryType> {
     }
 
     async fn handle_update_attributes(
-        self: Pin<&mut Self>,
+        &self,
         attributes: fio::MutableNodeAttributes,
     ) -> Result<(), Status> {
         if !self.base.options.rights.contains(fio::Operations::UPDATE_ATTRIBUTES) {
@@ -176,11 +170,7 @@ impl<DirectoryType: MutableDirectory> MutableConnection<DirectoryType> {
         self.base.directory.update_attributes(attributes).await
     }
 
-    async fn handle_unlink(
-        self: Pin<&mut Self>,
-        name: String,
-        options: fio::UnlinkOptions,
-    ) -> Result<(), Status> {
+    async fn handle_unlink(&self, name: String, options: fio::UnlinkOptions) -> Result<(), Status> {
         if !self.base.options.rights.contains(fio::Rights::MODIFY_DIRECTORY) {
             return Err(Status::BAD_HANDLE);
         }
@@ -277,18 +267,23 @@ impl<DirectoryType: MutableDirectory> MutableConnection<DirectoryType> {
     async fn handle_remove_extended_attribute(&self, name: Vec<u8>) -> Result<(), Status> {
         self.base.directory.remove_extended_attribute(name).await
     }
+}
+
+impl<DirectoryType: MutableDirectory> RequestHandler
+    for Tokenizable<MutableConnection<DirectoryType>>
+{
+    type Request = Result<fio::DirectoryRequest, fidl::Error>;
 
-    async fn handle_requests(self, mut requests: fio::DirectoryRequestStream) {
-        let this = Tokenizable::new(self);
-        pin_mut!(this);
-        while let Ok(Some(request)) = requests.try_next().await {
-            let _guard = this.base.scope.active_guard();
-            if !matches!(
-                Self::handle_request(Pin::as_mut(&mut this), request).await,
-                Ok(ConnectionState::Alive)
-            ) {
-                break;
+    async fn handle_request(self: Pin<&mut Self>, request: Self::Request) -> ControlFlow<()> {
+        let _guard = self.base.scope.active_guard();
+        match request {
+            Ok(request) => {
+                match MutableConnection::<DirectoryType>::handle_request(self, request).await {
+                    Ok(ConnectionState::Alive) => ControlFlow::Continue(()),
+                    Ok(ConnectionState::Closed) | Err(_) => ControlFlow::Break(()),
+                }
             }
+            Err(_) => ControlFlow::Break(()),
         }
     }
 }