Use the paged memory pool in Mountpoint

Signed-off-by: Alessandro Passaro <alexpax@amazon.co.uk>

Author: passaro

mountpoint-s3-fs/examples/mount_from_config.rs

@@ -19,6 +19,7 @@ use mountpoint_s3_fs::{
     },
     logging::{LoggingConfig, LoggingHandle, error_logger::FileErrorLogger, init_logging},
     manifest::{Manifest, ingest_manifest},
+    memory::PagedPool,
     metrics::{self, MetricsSinkHandle},
     prefix::Prefix,
     s3::config::{ClientConfig, PartConfig, Region, S3Path, TargetThroughputSetting},
@@ -234,8 +235,9 @@ fn mount_filesystem(
 
     // Create the client and runtime
     let client_config = config.build_client_config()?;
+    let pool = PagedPool::new([client_config.part_config.read_size_bytes]);
     let client = client_config
-        .create_client(None)
+        .create_client(pool.clone(), None)
         .context("Failed to create S3 client")?;
     let runtime = Runtime::new(client.event_loop_group());
 

mountpoint-s3-fs/src/s3/config.rs

@@ -11,6 +11,7 @@ use mountpoint_s3_client::{ObjectClient, S3CrtClient, S3RequestError};
 use regex::Regex;
 use thiserror::Error;
 
+use crate::memory::PagedPool;
 use crate::prefix::{Prefix, PrefixError};
 
 /// Configuration for the S3 Client to use in Mountpoint.
@@ -273,15 +274,20 @@ const INITIAL_READ_WINDOW_SIZE: usize = 1024 * 1024 + 128 * 1024;
 
 impl ClientConfig {
     /// Create an [S3CrtClient]
-    pub fn create_client(self, validate_on_s3_path: Option<&S3Path>) -> anyhow::Result<S3CrtClient> {
+    pub fn create_client(
+        self,
+        memory_pool: PagedPool,
+        validate_on_s3_path: Option<&S3Path>,
+    ) -> anyhow::Result<S3CrtClient> {
         let mut client_config = S3ClientConfig::new()
             .auth_config(self.auth_config)
             .throughput_target_gbps(self.throughput_target.value())
             .read_part_size(self.part_config.read_size_bytes)
             .write_part_size(self.part_config.write_size_bytes)
             .read_backpressure(true)
             .initial_read_window(INITIAL_READ_WINDOW_SIZE)
-            .user_agent(self.user_agent);
+            .user_agent(self.user_agent)
+            .memory_pool(memory_pool);
         if let Some(interfaces) = self.bind {
             client_config = client_config.network_interface_names(interfaces);
         }

mountpoint-s3/src/bin/mock-mount-s3.rs

@@ -19,6 +19,7 @@ use mountpoint_s3_client::mock_client::throughput_client::ThroughputMockClient;
 use mountpoint_s3_client::mock_client::{MockClient, MockObject};
 use mountpoint_s3_client::types::ETag;
 use mountpoint_s3_fs::Runtime;
+use mountpoint_s3_fs::memory::PagedPool;
 use mountpoint_s3_fs::s3::S3Personality;
 use mountpoint_s3_fs::s3::config::{ClientConfig, S3Path, TargetThroughputSetting};
 
@@ -29,6 +30,7 @@ fn main() -> anyhow::Result<()> {
 
 pub fn create_mock_client(
     client_config: ClientConfig,
+    _pool: PagedPool,
     s3_path: &S3Path,
     personality: Option<S3Personality>,
 ) -> anyhow::Result<(Arc<ThroughputMockClient>, Runtime, S3Personality)> {

mountpoint-s3/src/run.rs

@@ -10,6 +10,7 @@ use mountpoint_s3_client::{ObjectClient, S3CrtClient};
 use mountpoint_s3_fs::data_cache::{DataCacheConfig, ManagedCacheDir};
 use mountpoint_s3_fs::fuse::session::FuseSession;
 use mountpoint_s3_fs::logging::init_logging;
+use mountpoint_s3_fs::memory::PagedPool;
 use mountpoint_s3_fs::s3::S3Personality;
 use mountpoint_s3_fs::s3::config::{ClientConfig, S3Path};
 use mountpoint_s3_fs::{MountpointConfig, Runtime, metrics};
@@ -174,8 +175,17 @@ fn mount(args: CliArgs, client_builder: impl ClientBuilder) -> anyhow::Result<Fu
 
     let client_config = args.client_config(build_info::FULL_VERSION);
 
+    // Set up a paged memory pool
+    let pool = PagedPool::new([
+        1024 * 1024,
+        client_config.part_config.read_size_bytes,
+        client_config.part_config.write_size_bytes,
+    ]);
+    pool.schedule_trim(Duration::from_secs(60));
+
     let s3_path = args.s3_path()?;
-    let (client, runtime, s3_personality) = client_builder.build(client_config, &s3_path, args.personality())?;
+    let (client, runtime, s3_personality) =
+        client_builder.build(client_config, pool.clone(), &s3_path, args.personality())?;
 
     let bucket_description = args.bucket_description()?;
     tracing::debug!("using S3 personality {s3_personality:?} for {bucket_description}");
@@ -208,36 +218,39 @@ pub trait ClientBuilder {
     fn build(
         self,
         client_config: ClientConfig,
+        pool: PagedPool,
         s3_path: &S3Path,
         personality: Option<S3Personality>,
     ) -> anyhow::Result<(Self::Client, Runtime, S3Personality)>;
 }
 
 impl<F, C> ClientBuilder for F
 where
-    F: FnOnce(ClientConfig, &S3Path, Option<S3Personality>) -> anyhow::Result<(C, Runtime, S3Personality)>,
+    F: FnOnce(ClientConfig, PagedPool, &S3Path, Option<S3Personality>) -> anyhow::Result<(C, Runtime, S3Personality)>,
     C: ObjectClient + Clone + Send + Sync + 'static,
 {
     type Client = C;
 
     fn build(
         self,
         client_config: ClientConfig,
+        pool: PagedPool,
         s3_path: &S3Path,
         personality: Option<S3Personality>,
     ) -> anyhow::Result<(Self::Client, Runtime, S3Personality)> {
-        self(client_config, s3_path, personality)
+        self(client_config, pool, s3_path, personality)
     }
 }
 
 // Create a real S3 client
 pub fn create_s3_client(
     client_config: ClientConfig,
+    pool: PagedPool,
     s3_path: &S3Path,
     personality: Option<S3Personality>,
 ) -> anyhow::Result<(S3CrtClient, Runtime, S3Personality)> {
     let client = client_config
-        .create_client(Some(s3_path))
+        .create_client(pool, Some(s3_path))
         .context("Failed to create S3 client")?;
 
     let runtime = Runtime::new(client.event_loop_group());