diff --git a/pageserver/src/layered_repository.rs b/pageserver/src/layered_repository.rs index decb6faf0c..b3f768ad6d 100644 --- a/pageserver/src/layered_repository.rs +++ b/pageserver/src/layered_repository.rs @@ -17,6 +17,7 @@ use bytes::Bytes; use lazy_static::lazy_static; use postgres_ffi::pg_constants::BLCKSZ; use tracing::*; +use zenith_utils::batch_fsync::BatchFsync; use std::collections::hash_map::Entry; use std::collections::HashMap; @@ -1145,6 +1146,8 @@ impl LayeredTimeline { // a lot of memory and/or aren't receiving much updates anymore. let mut disk_consistent_lsn = last_record_lsn; + let mut batch_fsync = BatchFsync::default(); + let mut layer_uploads = Vec::new(); while let Some((oldest_layer_id, oldest_layer, oldest_generation)) = layers.peek_oldest_open() @@ -1176,7 +1179,7 @@ impl LayeredTimeline { drop(layers); drop(write_guard); - let mut this_layer_uploads = self.evict_layer(oldest_layer_id)?; + let mut this_layer_uploads = self.evict_layer(oldest_layer_id, &mut batch_fsync)?; layer_uploads.append(&mut this_layer_uploads); write_guard = self.write_lock.lock().unwrap(); @@ -1198,9 +1201,11 @@ impl LayeredTimeline { // new layer files are durable let timeline_dir = File::open(self.conf.timeline_path(&self.timelineid, &self.tenantid))?; - timeline_dir.sync_all()?; + batch_fsync.add(timeline_dir)?; } + batch_fsync.done()?; + // If we were able to advance 'disk_consistent_lsn', save it the metadata file. // After crash, we will restart WAL streaming and processing from that point. let old_disk_consistent_lsn = self.disk_consistent_lsn.load(); @@ -1246,7 +1251,7 @@ impl LayeredTimeline { Ok(()) } - fn evict_layer(&self, layer_id: LayerId) -> Result> { + fn evict_layer(&self, layer_id: LayerId, batch_fsync: &mut BatchFsync) -> Result> { // Mark the layer as no longer accepting writes and record the end_lsn. // This happens in-place, no new layers are created now. // We call `get_last_record_lsn` again, which may be different from the @@ -1271,7 +1276,7 @@ impl LayeredTimeline { drop(layers); drop(write_guard); - let new_historics = oldest_layer.write_to_disk(self)?; + let new_historics = oldest_layer.write_to_disk(self, batch_fsync)?; write_guard = self.write_lock.lock().unwrap(); layers = self.layers.lock().unwrap(); @@ -1882,9 +1887,13 @@ pub fn evict_layer_if_needed(conf: &PageServerConf) -> Result<()> { let timeline = tenant_mgr::get_timeline_for_tenant(tenantid, timelineid)?; + let mut batch_fsync = BatchFsync::default(); + timeline .upgrade_to_layered_timeline() - .evict_layer(layer_id)?; + .evict_layer(layer_id, &mut batch_fsync)?; + + batch_fsync.done()?; global_layer_map = GLOBAL_LAYER_MAP.read().unwrap(); } diff --git a/pageserver/src/layered_repository/delta_layer.rs b/pageserver/src/layered_repository/delta_layer.rs index 182e3e6c5c..db2908a36e 100644 --- a/pageserver/src/layered_repository/delta_layer.rs +++ b/pageserver/src/layered_repository/delta_layer.rs @@ -48,6 +48,7 @@ use crate::{ZTenantId, ZTimelineId}; use anyhow::{bail, ensure, Result}; use log::*; use serde::{Deserialize, Serialize}; +use zenith_utils::batch_fsync::BatchFsync; use zenith_utils::vec_map::VecMap; // avoid binding to Write (conflicts with std::io::Write) // while being able to use std::fmt::Write's methods @@ -364,6 +365,7 @@ impl DeltaLayer { dropped: bool, page_versions: impl Iterator, relsizes: VecMap, + batch_fsync: &mut BatchFsync, ) -> Result { if seg.rel.is_blocky() { assert!(!relsizes.is_empty()); @@ -436,7 +438,8 @@ impl DeltaLayer { // This flushes the underlying 'buf_writer'. let writer = book.close()?; - writer.get_ref().sync_all()?; + let file = writer.into_inner()?; + batch_fsync.add(file)?; trace!("saved {}", &path.display()); diff --git a/pageserver/src/layered_repository/image_layer.rs b/pageserver/src/layered_repository/image_layer.rs index 179b66853e..64a87b9ff2 100644 --- a/pageserver/src/layered_repository/image_layer.rs +++ b/pageserver/src/layered_repository/image_layer.rs @@ -39,6 +39,7 @@ use std::fs::File; use std::io::{BufWriter, Write}; use std::path::{Path, PathBuf}; use std::sync::{Mutex, MutexGuard}; +use zenith_utils::batch_fsync::BatchFsync; use bookfile::{Book, BookWriter}; @@ -261,6 +262,7 @@ impl ImageLayer { seg: SegmentTag, lsn: Lsn, base_images: Vec, + batch_fsync: &mut BatchFsync, ) -> Result { let image_type = if seg.rel.is_blocky() { let num_blocks: u32 = base_images.len().try_into()?; @@ -320,7 +322,8 @@ impl ImageLayer { // This flushes the underlying 'buf_writer'. let writer = book.close()?; - writer.get_ref().sync_all()?; + let file = writer.into_inner()?; + batch_fsync.add(file)?; trace!("saved {}", path.display()); @@ -336,6 +339,7 @@ impl ImageLayer { timeline: &LayeredTimeline, src: &dyn Layer, lsn: Lsn, + batch_fsync: &mut BatchFsync, ) -> Result { let seg = src.get_seg_tag(); let timelineid = timeline.timelineid; @@ -364,7 +368,15 @@ impl ImageLayer { base_images.push(img); } - Self::create(conf, timelineid, timeline.tenantid, seg, lsn, base_images) + Self::create( + conf, + timelineid, + timeline.tenantid, + seg, + lsn, + base_images, + batch_fsync, + ) } /// diff --git a/pageserver/src/layered_repository/inmemory_layer.rs b/pageserver/src/layered_repository/inmemory_layer.rs index 028e72267c..ecf5483b00 100644 --- a/pageserver/src/layered_repository/inmemory_layer.rs +++ b/pageserver/src/layered_repository/inmemory_layer.rs @@ -19,6 +19,7 @@ use log::*; use std::path::PathBuf; use std::sync::atomic::Ordering; use std::sync::{Arc, RwLock}; +use zenith_utils::batch_fsync::BatchFsync; use zenith_utils::vec_map::VecMap; use zenith_utils::lsn::Lsn; @@ -601,7 +602,11 @@ impl InMemoryLayer { /// WAL records between start and end LSN. (The delta layer is not needed /// when a new relish is created with a single LSN, so that the start and /// end LSN are the same.) - pub fn write_to_disk(&self, timeline: &LayeredTimeline) -> Result { + pub fn write_to_disk( + &self, + timeline: &LayeredTimeline, + batch_fsync: &mut BatchFsync, + ) -> Result { trace!( "write_to_disk {} get_end_lsn is {}", self.filename().display(), @@ -631,6 +636,7 @@ impl InMemoryLayer { true, inner.page_versions.ordered_page_version_iter(None), inner.segsizes.clone(), + batch_fsync, )?; trace!( "freeze: created delta layer for dropped segment {} {}-{}", @@ -668,6 +674,7 @@ impl InMemoryLayer { false, page_versions, segsizes, + batch_fsync, )?; delta_layers.push(delta_layer); trace!( @@ -684,7 +691,7 @@ impl InMemoryLayer { // Write a new base image layer at the cutoff point let image_layer = - ImageLayer::create_from_src(self.conf, timeline, self, end_lsn_inclusive)?; + ImageLayer::create_from_src(self.conf, timeline, self, end_lsn_inclusive, batch_fsync)?; trace!( "freeze: created image layer {} at {}", self.seg, diff --git a/zenith_utils/src/batch_fsync.rs b/zenith_utils/src/batch_fsync.rs new file mode 100644 index 0000000000..28cdfe9661 --- /dev/null +++ b/zenith_utils/src/batch_fsync.rs @@ -0,0 +1,43 @@ +use std::{fs::File, io}; + +const MAX_PENDING_FILES: usize = 100; + +#[derive(Default)] +pub struct BatchFsync { + pending: Vec, + done: bool, +} + +impl BatchFsync { + pub fn add(&mut self, file: File) -> io::Result<()> { + if self.pending.len() == MAX_PENDING_FILES { + self.sync_batch()?; + } + + self.pending.push(file); + + Ok(()) + } + + /// Must be called before drop. + pub fn done(mut self) -> io::Result<()> { + self.done = true; + self.sync_batch() + } + + fn sync_batch(&mut self) -> io::Result<()> { + // TODO parallelize + for pending_file in self.pending.drain(..) { + pending_file.sync_all()?; + } + self.pending.clear(); + + Ok(()) + } +} + +impl Drop for BatchFsync { + fn drop(&mut self) { + assert!(self.done); + } +} diff --git a/zenith_utils/src/lib.rs b/zenith_utils/src/lib.rs index 912d8308e5..c4328f24e8 100644 --- a/zenith_utils/src/lib.rs +++ b/zenith_utils/src/lib.rs @@ -43,3 +43,6 @@ pub mod accum; // Utility for binding TcpListeners with proper socket options. pub mod tcp_listener; + +// Call fsync for many files at once. +pub mod batch_fsync;