WIP: fallocate files before writing

pageserver/src/metrics.rs

@@ -1290,6 +1290,7 @@ pub(crate) enum StorageIoOperation {
     Fsync,
     Metadata,
     SetLen,
+    Fallocate,
 }
 
 impl StorageIoOperation {
@@ -1305,6 +1306,7 @@ impl StorageIoOperation {
             StorageIoOperation::Fsync => "fsync",
             StorageIoOperation::Metadata => "metadata",
             StorageIoOperation::SetLen => "set_len",
+            StorageIoOperation::Fallocate => "fallocate",
         }
     }
 }

pageserver/src/tenant/ephemeral_file.rs

@@ -88,6 +88,10 @@ impl EphemeralFile {
             gate.enter()?,
         );
 
+        file.fallocate_keep_size(0, 1 * 1024 * 1024 * 1024, ctx)
+            .await
+            .unwrap();
+
         let page_cache_file_id = page_cache::next_file_id(); // XXX get rid, we're not page-caching anymore
 
         Ok(EphemeralFile {

pageserver/src/tenant/remote_timeline_client/download.rs

@@ -76,6 +76,8 @@ pub async fn download_layer_file<'a>(
         layer_metadata.generation,
     );
 
+    let expected = layer_metadata.file_size;
+
     let (bytes_amount, temp_file) = download_retry(
         || async {
             // TempVirtualFile requires us to never reuse a filename while an old
@@ -103,6 +105,16 @@ pub async fn download_layer_file<'a>(
                 .map_err(DownloadError::Other)?,
                 gate.enter().map_err(|_| DownloadError::Cancelled)?,
             );
+            if let Ok(file_size) = TryInto::<i64>::try_into(layer_metadata.file_size.next_multiple_of(
+64 * 1024 /* TODO this is the max roundtup size by the buffered writer set_len_then_truncate */
+
+            )) {
+                temp_file.fallocate_keep_size(
+                    0,
+                    file_size,
+                    ctx,
+                ).await.unwrap();
+            };
             download_object(storage, &remote_path, temp_file, gate, cancel, ctx).await
         },
         &format!("download {remote_path:?}"),
@@ -110,7 +122,6 @@ pub async fn download_layer_file<'a>(
     )
     .await?;
 
-    let expected = layer_metadata.file_size;
     if expected != bytes_amount {
         return Err(DownloadError::Other(anyhow!(
             "According to layer file metadata should have downloaded {expected} bytes but downloaded {bytes_amount} bytes into file {:?}",

pageserver/src/tenant/storage_layer/delta_layer.rs

@@ -440,6 +440,10 @@ impl DeltaLayerWriterInner {
             gate.enter()?,
         );
 
+        file.fallocate_keep_size(0, 1 * 1024 * 1024 * 1024, ctx)
+            .await
+            .unwrap();
+
         // Start at PAGE_SZ, make room for the header block
         let blob_writer = BlobWriter::new(
             file,

pageserver/src/tenant/storage_layer/image_layer.rs

@@ -806,6 +806,10 @@ impl ImageLayerWriterInner {
             gate.enter()?,
         );
 
+        file.fallocate_keep_size(0, 1 * 1024 * 1024 * 1024, ctx)
+            .await
+            .unwrap();
+
         // Start at `PAGE_SZ` to make room for the header block.
         let blob_writer = BlobWriter::new(
             file,

pageserver/src/virtual_file.rs

@@ -161,6 +161,15 @@ impl VirtualFile {
         self.inner.set_len(len, ctx).await
     }
 
+    pub async fn fallocate_keep_size(
+        &self,
+        offset: i64,
+        size: i64,
+        ctx: &RequestContext,
+    ) -> Result<(), Error> {
+        self.inner.fallocate_keep_size(offset, size, ctx).await
+    }
+
     pub async fn metadata(&self) -> Result<Metadata, Error> {
         self.inner.metadata().await
     }
@@ -638,6 +647,20 @@ impl VirtualFileInner {
         })
     }
 
+    pub async fn fallocate_keep_size(
+        &self,
+        offset: i64,
+        size: i64,
+        _ctx: &RequestContext,
+    ) -> Result<(), Error> {
+        with_file!(self, StorageIoOperation::Fallocate, |file_guard| {
+            let (_file_guard, res) = io_engine::get()
+                .fallocate_keep_size(file_guard, offset, size)
+                .await;
+            res.maybe_fatal_err("fallocate") // TODO haven't thought about this
+        })
+    }
+
     /// Helper function internal to `VirtualFile` that looks up the underlying File,
     /// opens it and evicts some other File if necessary. The passed parameter is
     /// assumed to be a function available for the physical `File`.

pageserver/src/virtual_file/io_engine.rs

@@ -11,6 +11,7 @@
 
 #[cfg(target_os = "linux")]
 pub(super) mod tokio_epoll_uring_ext;
+use nix::fcntl::{FallocateFlags, fallocate};
 
 use tokio_epoll_uring::IoBuf;
 use tracing::Instrument;
@@ -109,6 +110,7 @@ pub(crate) fn get() -> IoEngine {
     }
 }
 
+use std::os::fd::AsRawFd;
 use std::os::unix::prelude::FileExt;
 use std::sync::atomic::{AtomicU8, Ordering};
 
@@ -230,6 +232,35 @@ impl IoEngine {
         }
     }
 
+    pub(super) async fn fallocate_keep_size(
+        &self,
+        file_guard: FileGuard,
+        offset: i64,
+        len: i64,
+    ) -> (FileGuard, std::io::Result<()>) {
+        // TODO io_uring implementation
+        match self {
+            IoEngine::NotSet => panic!("not initialized"),
+            IoEngine::StdFs => {
+                unimplemented!()
+            }
+            #[cfg(target_os = "linux")]
+            IoEngine::TokioEpollUring => {
+                // TODO: fallocate op for tokio-epoll-uring
+                file_guard.with_std_file(|std_file| {
+                    fallocate(
+                        std_file.as_raw_fd(),
+                        FallocateFlags::FALLOC_FL_KEEP_SIZE,
+                        offset,
+                        len,
+                    )
+                    .expect("TODO")
+                });
+                (file_guard, Ok(()))
+            }
+        }
+    }
+
     pub(super) async fn write_at<B: IoBuf + Send>(
         &self,
         file_guard: FileGuard,