diff --git a/pageserver/src/page_cache.rs b/pageserver/src/page_cache.rs index fb1c5fc485..6169e8798a 100644 --- a/pageserver/src/page_cache.rs +++ b/pageserver/src/page_cache.rs @@ -80,6 +80,7 @@ use std::{ use anyhow::Context; use once_cell::sync::OnceCell; +use tokio::sync::{RwLock, RwLockReadGuard, RwLockWriteGuard}; use utils::{ id::{TenantId, TimelineId}, lsn::Lsn, @@ -451,7 +452,7 @@ impl PageCache { /// async fn try_lock_for_read(&self, cache_key: &mut CacheKey) -> Option { let cache_key_orig = cache_key.clone(); - if let Some(slot_idx) = self.search_mapping(cache_key) { + if let Some(slot_idx) = self.search_mapping(cache_key).await { // The page was found in the mapping. Lock the slot, and re-check // that it's still what we expected (because we released the mapping // lock already, another thread could have evicted the page) @@ -626,7 +627,7 @@ impl PageCache { /// returns. The caller is responsible for re-checking that the slot still /// contains the page with the same key before using it. /// - fn search_mapping(&self, cache_key: &mut CacheKey) -> Option { + async fn search_mapping(&self, cache_key: &mut CacheKey) -> Option { match cache_key { CacheKey::MaterializedPage { hash_key, lsn } => { let map = self.materialized_page_map.read().unwrap(); diff --git a/pageserver/src/tenant/block_io.rs b/pageserver/src/tenant/block_io.rs index 69d5b49c6d..a9934f8af3 100644 --- a/pageserver/src/tenant/block_io.rs +++ b/pageserver/src/tenant/block_io.rs @@ -150,52 +150,58 @@ pub struct FileBlockReader { file_id: page_cache::FileId, } -impl FileBlockReader -where - F: FileExt, -{ +impl FileBlockReader { pub fn new(file: F) -> Self { let file_id = page_cache::next_file_id(); FileBlockReader { file_id, file } } - - /// Read a page from the underlying file into given buffer. - fn fill_buffer(&self, buf: &mut [u8], blkno: u32) -> Result<(), std::io::Error> { - assert!(buf.len() == PAGE_SZ); - self.file.read_exact_at(buf, blkno as u64 * PAGE_SZ as u64) - } - /// Read a block. - /// - /// Returns a "lease" object that can be used to - /// access to the contents of the page. (For the page cache, the - /// lease object represents a lock on the buffer.) - pub async fn read_blk(&self, blknum: u32) -> Result { - let cache = page_cache::get(); - loop { - match cache - .read_immutable_buf(self.file_id, blknum) - .await - .map_err(|e| { - std::io::Error::new( - std::io::ErrorKind::Other, - format!("Failed to read immutable buf: {e:#}"), - ) - })? { - ReadBufResult::Found(guard) => break Ok(guard.into()), - ReadBufResult::NotFound(mut write_guard) => { - // Read the page from disk into the buffer - self.fill_buffer(write_guard.deref_mut(), blknum)?; - write_guard.mark_valid(); - - // Swap for read lock - continue; - } - }; - } - } } +macro_rules! impls { + (FileBlockReader<$ty:ty>) => { + impl FileBlockReader<$ty> { + /// Read a page from the underlying file into given buffer. + fn fill_buffer(&self, buf: &mut [u8], blkno: u32) -> Result<(), std::io::Error> { + assert!(buf.len() == PAGE_SZ); + self.file.read_exact_at(buf, blkno as u64 * PAGE_SZ as u64) + } + /// Read a block. + /// + /// Returns a "lease" object that can be used to + /// access to the contents of the page. (For the page cache, the + /// lease object represents a lock on the buffer.) + pub async fn read_blk(&self, blknum: u32) -> Result { + let cache = page_cache::get(); + loop { + match cache + .read_immutable_buf(self.file_id, blknum) + .await + .map_err(|e| { + std::io::Error::new( + std::io::ErrorKind::Other, + format!("Failed to read immutable buf: {e:#}"), + ) + })? { + ReadBufResult::Found(guard) => break Ok(guard.into()), + ReadBufResult::NotFound(mut write_guard) => { + // Read the page from disk into the buffer + self.fill_buffer(write_guard.deref_mut(), blknum)?; + write_guard.mark_valid(); + + // Swap for read lock + continue; + } + }; + } + } + } + }; +} + +impls!(FileBlockReader); +impls!(FileBlockReader); + impl BlockReader for FileBlockReader { fn block_cursor(&self) -> BlockCursor<'_> { BlockCursor::new(BlockReaderRef::FileBlockReaderFile(self)) diff --git a/pageserver/src/tenant/ephemeral_file.rs b/pageserver/src/tenant/ephemeral_file.rs index 31db3869d9..02ef7166e5 100644 --- a/pageserver/src/tenant/ephemeral_file.rs +++ b/pageserver/src/tenant/ephemeral_file.rs @@ -9,7 +9,6 @@ use std::cmp::min; use std::fs::OpenOptions; use std::io::{self, ErrorKind}; use std::ops::DerefMut; -use std::os::unix::prelude::FileExt; use std::path::PathBuf; use std::sync::atomic::AtomicU64; use tracing::*; diff --git a/pageserver/src/tenant/manifest.rs b/pageserver/src/tenant/manifest.rs index 1d2835114f..fef66abd4f 100644 --- a/pageserver/src/tenant/manifest.rs +++ b/pageserver/src/tenant/manifest.rs @@ -26,7 +26,7 @@ //! recovered from this file. This is tracked in //! -use std::io::{self, Read, Write}; +use std::io::{self, Write}; use crate::virtual_file::VirtualFile; use anyhow::Result; @@ -151,11 +151,12 @@ impl Manifest { /// Load a manifest. Returns the manifest and a list of operations. If the manifest is corrupted, /// the bool flag will be set to true and the user is responsible to reconstruct a new manifest and /// backup the current one. - pub fn load( - mut file: VirtualFile, + pub async fn load( + file: VirtualFile, ) -> Result<(Self, Vec, ManifestPartiallyCorrupted), ManifestLoadError> { let mut buf = vec![]; - file.read_to_end(&mut buf).map_err(ManifestLoadError::Io)?; + file.read_exact_at(&mut buf, 0) + .map_err(ManifestLoadError::Io)?; // Read manifest header let mut buf = Bytes::from(buf); @@ -241,8 +242,8 @@ mod tests { use super::*; - #[test] - fn test_read_manifest() { + #[tokio::test] + async fn test_read_manifest() { let testdir = crate::config::PageServerConf::test_repo_dir("test_read_manifest"); std::fs::create_dir_all(&testdir).unwrap(); let file = VirtualFile::create(&testdir.join("MANIFEST")).unwrap(); @@ -274,7 +275,7 @@ mod tests { .truncate(false), ) .unwrap(); - let (mut manifest, operations, corrupted) = Manifest::load(file).unwrap(); + let (mut manifest, operations, corrupted) = Manifest::load(file).await.unwrap(); assert!(!corrupted.0); assert_eq!(operations.len(), 2); assert_eq!( @@ -306,7 +307,7 @@ mod tests { .truncate(false), ) .unwrap(); - let (_manifest, operations, corrupted) = Manifest::load(file).unwrap(); + let (_manifest, operations, corrupted) = Manifest::load(file).await.unwrap(); assert!(!corrupted.0); assert_eq!(operations.len(), 3); assert_eq!(&operations[0], &Operation::Snapshot(snapshot, Lsn::from(0))); diff --git a/pageserver/src/tenant/storage_layer/delta_layer.rs b/pageserver/src/tenant/storage_layer/delta_layer.rs index b7afdadf92..fdcd89eac5 100644 --- a/pageserver/src/tenant/storage_layer/delta_layer.rs +++ b/pageserver/src/tenant/storage_layer/delta_layer.rs @@ -45,8 +45,8 @@ use pageserver_api::models::{HistoricLayerInfo, LayerAccessKind}; use rand::{distributions::Alphanumeric, Rng}; use serde::{Deserialize, Serialize}; use std::fs::{self, File}; +use std::io::SeekFrom; use std::io::{BufWriter, Write}; -use std::io::{Seek, SeekFrom}; use std::ops::Range; use std::os::unix::fs::FileExt; use std::path::{Path, PathBuf}; diff --git a/pageserver/src/tenant/storage_layer/image_layer.rs b/pageserver/src/tenant/storage_layer/image_layer.rs index 482f363aef..e0a6d576b7 100644 --- a/pageserver/src/tenant/storage_layer/image_layer.rs +++ b/pageserver/src/tenant/storage_layer/image_layer.rs @@ -42,8 +42,8 @@ use pageserver_api::models::{HistoricLayerInfo, LayerAccessKind}; use rand::{distributions::Alphanumeric, Rng}; use serde::{Deserialize, Serialize}; use std::fs::{self, File}; +use std::io::SeekFrom; use std::io::Write; -use std::io::{Seek, SeekFrom}; use std::ops::Range; use std::os::unix::prelude::FileExt; use std::path::{Path, PathBuf}; diff --git a/pageserver/src/virtual_file.rs b/pageserver/src/virtual_file.rs index a86b8fa2a6..8e4f103158 100644 --- a/pageserver/src/virtual_file.rs +++ b/pageserver/src/virtual_file.rs @@ -13,7 +13,7 @@ use crate::metrics::{STORAGE_IO_SIZE, STORAGE_IO_TIME}; use once_cell::sync::OnceCell; use std::fs::{self, File, OpenOptions}; -use std::io::{Error, ErrorKind, Read, Seek, SeekFrom, Write}; +use std::io::{Error, ErrorKind, Seek, SeekFrom, Write}; use std::os::unix::fs::FileExt; use std::path::{Path, PathBuf}; use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; @@ -321,54 +321,8 @@ impl VirtualFile { drop(self); std::fs::remove_file(path).expect("failed to remove the virtual file"); } -} -impl Drop for VirtualFile { - /// If a VirtualFile is dropped, close the underlying file if it was open. - fn drop(&mut self) { - let handle = self.handle.get_mut().unwrap(); - - // We could check with a read-lock first, to avoid waiting on an - // unrelated I/O. - let slot = &get_open_files().slots[handle.index]; - let mut slot_guard = slot.inner.write().unwrap(); - if slot_guard.tag == handle.tag { - slot.recently_used.store(false, Ordering::Relaxed); - // there is also operation "close-by-replace" for closes done on eviction for - // comparison. - STORAGE_IO_TIME - .with_label_values(&["close"]) - .observe_closure_duration(|| drop(slot_guard.file.take())); - } - } -} - -impl Read for VirtualFile { - fn read(&mut self, buf: &mut [u8]) -> Result { - let pos = self.pos; - let n = self.read_at(buf, pos)?; - self.pos += n as u64; - Ok(n) - } -} - -impl Write for VirtualFile { - fn write(&mut self, buf: &[u8]) -> Result { - let pos = self.pos; - let n = self.write_at(buf, pos)?; - self.pos += n as u64; - Ok(n) - } - - fn flush(&mut self) -> Result<(), std::io::Error> { - // flush is no-op for File (at least on unix), so we don't need to do - // anything here either. - Ok(()) - } -} - -impl Seek for VirtualFile { - fn seek(&mut self, pos: SeekFrom) -> Result { + pub fn seek(&mut self, pos: SeekFrom) -> Result { match pos { SeekFrom::Start(offset) => { self.pos = offset; @@ -392,10 +346,50 @@ impl Seek for VirtualFile { } Ok(self.pos) } -} -impl FileExt for VirtualFile { - fn read_at(&self, buf: &mut [u8], offset: u64) -> Result { + // Copied from https://doc.rust-lang.org/1.72.0/src/std/os/unix/fs.rs.html#117-135 + pub fn read_exact_at(&self, mut buf: &mut [u8], mut offset: u64) -> Result<(), Error> { + while !buf.is_empty() { + match self.read_at(buf, offset) { + Ok(0) => { + return Err(Error::new( + std::io::ErrorKind::UnexpectedEof, + "failed to fill whole buffer", + )) + } + Ok(n) => { + buf = &mut buf[n..]; + offset += n as u64; + } + Err(ref e) if e.kind() == std::io::ErrorKind::Interrupted => {} + Err(e) => return Err(e), + } + } + Ok(()) + } + + // Copied from https://doc.rust-lang.org/1.72.0/src/std/os/unix/fs.rs.html#219-235 + pub fn write_all_at(&self, mut buf: &[u8], mut offset: u64) -> Result<(), Error> { + while !buf.is_empty() { + match self.write_at(buf, offset) { + Ok(0) => { + return Err(Error::new( + std::io::ErrorKind::WriteZero, + "failed to write whole buffer", + )); + } + Ok(n) => { + buf = &buf[n..]; + offset += n as u64; + } + Err(ref e) if e.kind() == std::io::ErrorKind::Interrupted => {} + Err(e) => return Err(e), + } + } + Ok(()) + } + + pub fn read_at(&self, buf: &mut [u8], offset: u64) -> Result { let result = self.with_file("read", |file| file.read_at(buf, offset))?; if let Ok(size) = result { STORAGE_IO_SIZE @@ -405,7 +399,7 @@ impl FileExt for VirtualFile { result } - fn write_at(&self, buf: &[u8], offset: u64) -> Result { + pub fn write_at(&self, buf: &[u8], offset: u64) -> Result { let result = self.with_file("write", |file| file.write_at(buf, offset))?; if let Ok(size) = result { STORAGE_IO_SIZE @@ -416,6 +410,41 @@ impl FileExt for VirtualFile { } } +impl Drop for VirtualFile { + /// If a VirtualFile is dropped, close the underlying file if it was open. + fn drop(&mut self) { + let handle = self.handle.get_mut().unwrap(); + + // We could check with a read-lock first, to avoid waiting on an + // unrelated I/O. + let slot = &get_open_files().slots[handle.index]; + let mut slot_guard = slot.inner.write().unwrap(); + if slot_guard.tag == handle.tag { + slot.recently_used.store(false, Ordering::Relaxed); + // there is also operation "close-by-replace" for closes done on eviction for + // comparison. + STORAGE_IO_TIME + .with_label_values(&["close"]) + .observe_closure_duration(|| drop(slot_guard.file.take())); + } + } +} + +impl Write for VirtualFile { + fn write(&mut self, buf: &[u8]) -> Result { + let pos = self.pos; + let n = self.write_at(buf, pos)?; + self.pos += n as u64; + Ok(n) + } + + fn flush(&mut self) -> Result<(), std::io::Error> { + // flush is no-op for File (at least on unix), so we don't need to do + // anything here either. + Ok(()) + } +} + impl OpenFiles { fn new(num_slots: usize) -> OpenFiles { let mut slots = Box::new(Vec::with_capacity(num_slots));