From bad7ab0d8b2f433c7864e5b211fac3686a2aa306 Mon Sep 17 00:00:00 2001 From: Christian Schwarz Date: Thu, 31 Aug 2023 08:50:51 +0000 Subject: [PATCH] git merge --squash of https://github.com/neondatabase/neon/pull/5121 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit commit edbe3d2f760efa03bf12acf704b86f81d85a34be Author: Arpad Müller Date: Mon Aug 28 11:49:56 2023 +0200 Remove Read impl that was only used in one place commit 0d9fa95454bc0df66c900dd4f3fe97e92c49efd3 Author: Arpad Müller Date: Mon Aug 28 11:31:43 2023 +0200 Move used FileExt functions to inherent impls commit e983b3cc2e48be9f5618982ef99cce5e31a44ef1 Author: Arpad Müller Date: Mon Aug 28 11:05:06 2023 +0200 Don't use generics bounded by trait commit a362ab9169c063b5ff9fdf215e482dcc3d901123 Author: Arpad Müller Date: Mon Aug 28 10:27:12 2023 +0200 Move VirtualFile::seek to inherent function commit 0cfc9edcb870a0324e2bc1276e1e8d8a8100a0d9 Author: Arpad Müller Date: Thu Aug 17 00:02:37 2023 +0200 Make read_blk and parts of the page cache async The returned PageReadGuard is not Send so we change the locks used for the SlotInner's in the page cache to the ones from tokio. Also, make read_blk async. --- pageserver/src/page_cache.rs | 5 +- pageserver/src/tenant/block_io.rs | 84 ++++++----- pageserver/src/tenant/ephemeral_file.rs | 1 - pageserver/src/tenant/manifest.rs | 17 +-- .../src/tenant/storage_layer/delta_layer.rs | 2 +- .../src/tenant/storage_layer/image_layer.rs | 2 +- pageserver/src/virtual_file.rs | 133 +++++++++++------- 7 files changed, 140 insertions(+), 104 deletions(-) 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));