diff --git a/pageserver/src/layered_repository.rs b/pageserver/src/layered_repository.rs index 99220f6c21..ecde1005b0 100644 --- a/pageserver/src/layered_repository.rs +++ b/pageserver/src/layered_repository.rs @@ -68,7 +68,7 @@ use delta_layer::DeltaLayer; use image_layer::ImageLayer; use global_layer_map::{LayerId, GLOBAL_LAYER_MAP}; -use inmemory_layer::InMemoryLayer; +use inmemory_layer::OpenLayer; use layer_map::LayerMap; use storage_layer::{ Layer, PageReconstructData, PageReconstructResult, SegmentTag, RELISH_SEG_SIZE, @@ -1025,7 +1025,7 @@ impl LayeredTimeline { /// /// Get a handle to the latest layer for appending. /// - fn get_layer_for_write(&self, seg: SegmentTag, lsn: Lsn) -> Result> { + fn get_layer_for_write(&self, seg: SegmentTag, lsn: Lsn) -> Result> { let mut layers = self.layers.lock().unwrap(); assert!(lsn.is_aligned()); @@ -1056,14 +1056,8 @@ impl LayeredTimeline { lsn ); - layer = InMemoryLayer::create( - self.conf, - self.timelineid, - self.tenantid, - seg, - lsn, - lsn, - )?; + layer = + OpenLayer::create(self.conf, self.timelineid, self.tenantid, seg, lsn, lsn)?; } else { return Ok(open_layer); } @@ -1099,7 +1093,7 @@ impl LayeredTimeline { prev_layer.get_start_lsn(), prev_layer.get_end_lsn() ); - layer = InMemoryLayer::create_successor_layer( + layer = OpenLayer::create_successor_layer( self.conf, prev_layer, self.timelineid, @@ -1116,11 +1110,10 @@ impl LayeredTimeline { lsn ); - layer = - InMemoryLayer::create(self.conf, self.timelineid, self.tenantid, seg, lsn, lsn)?; + layer = OpenLayer::create(self.conf, self.timelineid, self.tenantid, seg, lsn, lsn)?; } - let layer_rc: Arc = Arc::new(layer); + let layer_rc: Arc = Arc::new(layer); layers.insert_open(Arc::clone(&layer_rc)); Ok(layer_rc) @@ -1774,7 +1767,7 @@ impl<'a> TimelineWriter for LayeredTimelineWriter<'a> { let seg = SegmentTag::from_blknum(rel, blknum); let layer = self.tl.get_layer_for_write(seg, lsn)?; - let delta_size = layer.put_wal_record(lsn, blknum, rec); + let delta_size = layer.put_wal_record(lsn, blknum, rec)?; self.tl .increase_current_logical_size(delta_size * BLCKSZ as u32); Ok(()) @@ -1793,7 +1786,7 @@ impl<'a> TimelineWriter for LayeredTimelineWriter<'a> { let seg = SegmentTag::from_blknum(rel, blknum); let layer = self.tl.get_layer_for_write(seg, lsn)?; - let delta_size = layer.put_page_image(blknum, lsn, img); + let delta_size = layer.put_page_image(blknum, lsn, img)?; self.tl .increase_current_logical_size(delta_size * BLCKSZ as u32); @@ -1943,32 +1936,3 @@ fn rename_to_backup(path: PathBuf) -> anyhow::Result<()> { path )) } - -//----- Global layer management - -/// Check if too much memory is being used by open layers. If so, evict -pub fn evict_layer_if_needed(conf: &PageServerConf) -> Result<()> { - // Keep evicting layers until we are below the memory threshold. - let mut global_layer_map = GLOBAL_LAYER_MAP.read().unwrap(); - while let Some((layer_id, layer)) = global_layer_map.find_victim_if_needed(conf.open_mem_limit) - { - drop(global_layer_map); - let tenantid = layer.get_tenant_id(); - let timelineid = layer.get_timeline_id(); - - let _entered = - info_span!("global evict", timeline = %timelineid, tenant = %tenantid).entered(); - info!("evicting {}", layer.filename().display()); - drop(layer); - - let timeline = tenant_mgr::get_timeline_for_tenant(tenantid, timelineid)?; - - timeline - .upgrade_to_layered_timeline() - .evict_layer(layer_id)?; - - global_layer_map = GLOBAL_LAYER_MAP.read().unwrap(); - } - - Ok(()) -} diff --git a/pageserver/src/layered_repository/blob.rs b/pageserver/src/layered_repository/blob.rs index 592e279f4f..7bf6052983 100644 --- a/pageserver/src/layered_repository/blob.rs +++ b/pageserver/src/layered_repository/blob.rs @@ -1,4 +1,4 @@ -use std::io::Write; +use std::io::{Read, Write}; use std::os::unix::prelude::FileExt; use anyhow::Result; @@ -29,14 +29,14 @@ impl BlobWriter { Self { writer, offset: 0 } } - pub fn write_blob(&mut self, blob: &[u8]) -> Result { - self.writer.write_all(blob)?; + pub fn write_blob_from_reader(&mut self, r: &mut impl Read) -> Result { + let len = std::io::copy(r, &mut self.writer)?; let range = BlobRange { offset: self.offset, - size: blob.len(), + size: len as usize, }; - self.offset += blob.len() as u64; + self.offset += len as u64; Ok(range) } diff --git a/pageserver/src/layered_repository/delta_layer.rs b/pageserver/src/layered_repository/delta_layer.rs index f1aaf95d75..e0547743da 100644 --- a/pageserver/src/layered_repository/delta_layer.rs +++ b/pageserver/src/layered_repository/delta_layer.rs @@ -39,6 +39,7 @@ //! use crate::layered_repository::blob::BlobWriter; use crate::layered_repository::filename::{DeltaFileName, PathOrConf}; +use crate::layered_repository::page_versions::PageVersions; use crate::layered_repository::storage_layer::{ Layer, PageReconstructData, PageReconstructResult, PageVersion, SegmentTag, }; @@ -373,14 +374,14 @@ impl DeltaLayer { } /// Create a new delta file, using the given page versions and relsizes. - /// The page versions are passed by an iterator; the iterator must return - /// page versions in blknum+lsn order. + /// The page versions are passed in a PageVersions struct. If 'cutoff' is + /// given, only page versions with LSN < cutoff are included. /// - /// This is used to write the in-memory layer to disk. The in-memory layer uses the same - /// data structure with two btreemaps as we do, so passing the btreemaps is currently - /// expedient. + /// This is used to write the in-memory layer to disk. The page_versions and + /// relsizes are thus passed in the same format as they are in the in-memory + /// layer, as that's expedient. #[allow(clippy::too_many_arguments)] - pub fn create<'a>( + pub fn create( conf: &'static PageServerConf, timelineid: ZTimelineId, tenantid: ZTenantId, @@ -388,7 +389,8 @@ impl DeltaLayer { start_lsn: Lsn, end_lsn: Lsn, dropped: bool, - page_versions: impl Iterator, + page_versions: &PageVersions, + cutoff: Option, relsizes: VecMap, ) -> Result { if seg.rel.is_blocky() { @@ -427,9 +429,10 @@ impl DeltaLayer { let mut page_version_writer = BlobWriter::new(book, PAGE_VERSIONS_CHAPTER); - for (blknum, lsn, page_version) in page_versions { - let buf = PageVersion::ser(page_version)?; - let blob_range = page_version_writer.write_blob(&buf)?; + let page_versions_iter = page_versions.ordered_page_version_iter(cutoff); + for (blknum, lsn, pos) in page_versions_iter { + let blob_range = + page_version_writer.write_blob_from_reader(&mut page_versions.reader(pos)?)?; inner .page_version_metas diff --git a/pageserver/src/layered_repository/global_layer_map.rs b/pageserver/src/layered_repository/global_layer_map.rs index 9ad4bd6f44..9ad0df7ddc 100644 --- a/pageserver/src/layered_repository/global_layer_map.rs +++ b/pageserver/src/layered_repository/global_layer_map.rs @@ -10,10 +10,10 @@ //! The ID can be used to relocate the layer later, without having to hold locks. //! -use std::sync::atomic::{AtomicU8, AtomicUsize, Ordering}; +use std::sync::atomic::{AtomicU8, Ordering}; use std::sync::{Arc, RwLock}; -use super::inmemory_layer::InMemoryLayer; +use super::inmemory_layer::OpenLayer; use lazy_static::lazy_static; @@ -23,15 +23,6 @@ lazy_static! { pub static ref GLOBAL_LAYER_MAP: RwLock = RwLock::new(OpenLayers::default()); } -/// -/// How much memory is being used by all the open layers? This is used to trigger -/// freezing and evicting an open layer to disk. -/// -/// This is only a rough approximation, it leaves out a lot of things like malloc() -/// overhead. But as long there is enough "slop" and it's not set too close to the RAM -/// size on the system, it's good enough. -pub static GLOBAL_OPEN_MEM_USAGE: AtomicUsize = AtomicUsize::new(0); - // TODO these types can probably be smaller #[derive(PartialEq, Eq, Clone, Copy)] pub struct LayerId { @@ -40,7 +31,7 @@ pub struct LayerId { } enum SlotData { - Occupied(Arc), + Occupied(Arc), /// Vacant slots form a linked list, the value is the index /// of the next vacant slot in the list. Vacant(Option), @@ -56,14 +47,13 @@ struct Slot { pub struct OpenLayers { slots: Vec, num_occupied: usize, - next_victim: AtomicUsize, // Head of free-slot list. next_empty_slot_idx: Option, } impl OpenLayers { - pub fn insert(&mut self, layer: Arc) -> LayerId { + pub fn insert(&mut self, layer: Arc) -> LayerId { let slot_idx = match self.next_empty_slot_idx { Some(slot_idx) => slot_idx, None => { @@ -101,7 +91,7 @@ impl OpenLayers { } } - pub fn get(&self, layer_id: &LayerId) -> Option> { + pub fn get(&self, layer_id: &LayerId) -> Option> { let slot = self.slots.get(layer_id.index)?; // TODO should out of bounds indexes just panic? if slot.tag != layer_id.tag { return None; @@ -125,64 +115,6 @@ impl OpenLayers { } } - /// Find a victim layer to evict, if the total memory usage of all open layers - /// is larger than 'limit' - pub fn find_victim_if_needed(&self, limit: usize) -> Option<(LayerId, Arc)> { - let mem_usage = GLOBAL_OPEN_MEM_USAGE.load(Ordering::Relaxed); - - if mem_usage > limit { - self.find_victim() - } else { - None - } - } - - pub fn find_victim(&self) -> Option<(LayerId, Arc)> { - if self.num_occupied == 0 { - return None; - } - - // Run the clock algorithm. - // - // FIXME: It's theoretically possible that a constant stream of get() requests - // comes in faster than we advance the clock hand, so that this never finishes. - loop { - // FIXME: Because we interpret the clock hand variable modulo slots.len(), the - // hand effectively jumps to a more or less random place whenever the array is - // expanded. That's relatively harmless, it just leads to a non-optimal choice - // of victim. Also, in a server that runs for long enough, the array should reach - // a steady-state size and not grow anymore. - let next_victim = self.next_victim.fetch_add(1, Ordering::Relaxed) % self.slots.len(); - - let slot = &self.slots[next_victim]; - - if let SlotData::Occupied(data) = &slot.data { - fn update_fn(old_usage_count: u8) -> Option { - if old_usage_count > 0 { - Some(old_usage_count - 1) - } else { - None - } - } - - if slot - .usage_count - .fetch_update(Ordering::Relaxed, Ordering::Relaxed, update_fn) - .is_err() - { - // Found a slot with usage_count == 0. Return it. - return Some(( - LayerId { - index: next_victim, - tag: slot.tag, - }, - Arc::clone(data), - )); - } - } - } - } - // TODO this won't be a public API in the future pub fn remove(&mut self, layer_id: &LayerId) { let slot = &mut self.slots[layer_id.index]; diff --git a/pageserver/src/layered_repository/inmemory_layer.rs b/pageserver/src/layered_repository/inmemory_layer.rs index 1555f9677a..d673765af2 100644 --- a/pageserver/src/layered_repository/inmemory_layer.rs +++ b/pageserver/src/layered_repository/inmemory_layer.rs @@ -1,9 +1,8 @@ -//! +//! FIXME //! An in-memory layer stores recently received page versions in memory. The page versions //! are held in a BTreeMap, and there's another BTreeMap to track the size of the relation. //! use crate::layered_repository::filename::DeltaFileName; -use crate::layered_repository::global_layer_map::GLOBAL_OPEN_MEM_USAGE; use crate::layered_repository::storage_layer::{ Layer, PageReconstructData, PageReconstructResult, PageVersion, SegmentTag, RELISH_SEG_SIZE, }; @@ -11,21 +10,21 @@ use crate::layered_repository::LayeredTimeline; use crate::layered_repository::ZERO_PAGE; use crate::layered_repository::{DeltaLayer, ImageLayer}; use crate::repository::WALRecord; +use crate::virtual_file::VirtualFile; use crate::PageServerConf; use crate::{ZTenantId, ZTimelineId}; use anyhow::{ensure, Result}; use bytes::Bytes; use log::*; +use std::fs::OpenOptions; use std::path::PathBuf; -use std::sync::atomic::Ordering; use std::sync::{Arc, RwLock}; -use zenith_utils::vec_map::VecMap; - use zenith_utils::lsn::Lsn; +use zenith_utils::vec_map::VecMap; use super::page_versions::PageVersions; -pub struct InMemoryLayer { +pub struct OpenLayer { conf: &'static PageServerConf, tenantid: ZTenantId, timelineid: ZTimelineId, @@ -42,14 +41,14 @@ pub struct InMemoryLayer { /// The above fields never change. The parts that do change are in 'inner', /// and protected by mutex. - inner: RwLock, + inner: RwLock, /// Predecessor layer might be needed? incremental: bool, } -pub struct InMemoryLayerInner { - /// Frozen in-memory layers have an exclusive end LSN. +pub struct OpenLayerInner { + /// Frozen layers have an exclusive end LSN. /// Writes are only allowed when this is None end_lsn: Option, @@ -71,18 +70,9 @@ pub struct InMemoryLayerInner { /// a non-blocky rel, 'segsizes' is not used and is always empty. /// segsizes: VecMap, - - /// Approximate amount of memory used by this layer. - /// - /// TODO: This is currently a very crude metric, we don't take into account allocator - /// overhead, memory fragmentation, memory used by the VecMaps, nor many other things. - /// Just the actual # of bytes of a page image (8 kB) or the size of a WAL record. - /// - /// Whenever this is changed, you must also modify GLOBAL_OPEN_MEM_USAGE accordingly! - mem_usage: usize, } -impl InMemoryLayerInner { +impl OpenLayerInner { fn assert_writeable(&self) { assert!(self.end_lsn.is_none()); } @@ -100,17 +90,9 @@ impl InMemoryLayerInner { } } -impl Drop for InMemoryLayerInner { - fn drop(&mut self) { - if self.mem_usage > 0 { - GLOBAL_OPEN_MEM_USAGE.fetch_sub(self.mem_usage, Ordering::Relaxed); - self.mem_usage = 0; - } - } -} - -impl Layer for InMemoryLayer { - // An in-memory layer doesn't really have a filename as it's not stored on disk, +impl Layer for OpenLayer { + // FIXME + // An open layer doesn't really have a filename as it's not stored on disk, // but we construct a filename as if it was a delta layer fn filename(&self) -> PathBuf { let inner = self.inner.read().unwrap(); @@ -185,7 +167,7 @@ impl Layer for InMemoryLayer { .get_block_lsn_range(blknum, ..=lsn) .iter() .rev(); - for (entry_lsn, pv) in iter { + for (entry_lsn, pos) in iter { match &cached_img_lsn { Some(cached_lsn) if entry_lsn <= cached_lsn => { return Ok(PageReconstructResult::Cached) @@ -193,9 +175,10 @@ impl Layer for InMemoryLayer { _ => {} } + let pv = inner.page_versions.get_page_version(*pos)?; match pv { PageVersion::Page(img) => { - reconstruct_data.page_img = Some(img.clone()); + reconstruct_data.page_img = Some(img); need_image = false; break; } @@ -293,7 +276,8 @@ impl Layer for InMemoryLayer { println!("segsizes {}: {}", k, v); } - for (blknum, lsn, pv) in inner.page_versions.ordered_page_version_iter(None) { + for (blknum, lsn, pos) in inner.page_versions.ordered_page_version_iter(None) { + let pv = inner.page_versions.get_page_version(pos)?; let pv_description = match pv { PageVersion::Page(_img) => "page", PageVersion::Wal(_rec) => "wal", @@ -312,7 +296,7 @@ pub struct LayersOnDisk { pub image_layers: Vec, } -impl InMemoryLayer { +impl OpenLayer { /// Return the oldest page version that's stored in this layer pub fn get_oldest_pending_lsn(&self) -> Lsn { self.oldest_pending_lsn @@ -328,7 +312,7 @@ impl InMemoryLayer { seg: SegmentTag, start_lsn: Lsn, oldest_pending_lsn: Lsn, - ) -> Result { + ) -> Result { trace!( "initializing new empty InMemoryLayer for writing {} on timeline {} at {}", seg, @@ -342,7 +326,23 @@ impl InMemoryLayer { segsizes.append(start_lsn, 0).unwrap(); } - Ok(InMemoryLayer { + let delta_filename = DeltaFileName { + seg, + start_lsn, + end_lsn: Lsn(u64::MAX), + dropped: false, + } + .to_string(); + let filename = conf + .timeline_path(&timelineid, &tenantid) + .join(PathBuf::from(format!("open-{}", delta_filename))); + + let file = VirtualFile::open_with_options( + &filename, + OpenOptions::new().read(true).write(true).create(true), + )?; + + Ok(OpenLayer { conf, timelineid, tenantid, @@ -350,12 +350,11 @@ impl InMemoryLayer { start_lsn, oldest_pending_lsn, incremental: false, - inner: RwLock::new(InMemoryLayerInner { + inner: RwLock::new(OpenLayerInner { end_lsn: None, dropped: false, - page_versions: PageVersions::default(), + page_versions: PageVersions::new(file), segsizes, - mem_usage: 0, }), }) } @@ -363,18 +362,18 @@ impl InMemoryLayer { // Write operations /// Remember new page version, as a WAL record over previous version - pub fn put_wal_record(&self, lsn: Lsn, blknum: u32, rec: WALRecord) -> u32 { + pub fn put_wal_record(&self, lsn: Lsn, blknum: u32, rec: WALRecord) -> Result { self.put_page_version(blknum, lsn, PageVersion::Wal(rec)) } /// Remember new page version, as a full page image - pub fn put_page_image(&self, blknum: u32, lsn: Lsn, img: Bytes) -> u32 { + pub fn put_page_image(&self, blknum: u32, lsn: Lsn, img: Bytes) -> Result { self.put_page_version(blknum, lsn, PageVersion::Page(img)) } /// Common subroutine of the public put_wal_record() and put_page_image() functions. /// Adds the page version to the in-memory tree - pub fn put_page_version(&self, blknum: u32, lsn: Lsn, pv: PageVersion) -> u32 { + pub fn put_page_version(&self, blknum: u32, lsn: Lsn, pv: PageVersion) -> Result { assert!(self.seg.blknum_in_seg(blknum)); trace!( @@ -388,14 +387,7 @@ impl InMemoryLayer { inner.assert_writeable(); - let mut mem_usage = 0; - mem_usage += match &pv { - PageVersion::Page(img) => img.len(), - PageVersion::Wal(rec) => rec.rec.len(), - }; - - let (old, delta_size) = inner.page_versions.append_or_update_last(blknum, lsn, pv); - mem_usage += delta_size; + let old = inner.page_versions.append_or_update_last(blknum, lsn, pv)?; if old.is_some() { // We already had an entry for this LSN. That's odd.. @@ -405,8 +397,6 @@ impl InMemoryLayer { ); } - let mut delta_logical_size = 0; - // Also update the relation size, if this extended the relation. if self.seg.rel.is_blocky() { let newsize = blknum - self.seg.segno * RELISH_SEG_SIZE + 1; @@ -439,10 +429,9 @@ impl InMemoryLayer { gapblknum, blknum ); - let (old, delta_size) = inner + let old = inner .page_versions - .append_or_update_last(gapblknum, lsn, zeropv); - mem_usage += delta_size; + .append_or_update_last(gapblknum, lsn, zeropv)?; // We already had an entry for this LSN. That's odd.. if old.is_some() { @@ -453,18 +442,12 @@ impl InMemoryLayer { } } - let (_old, delta_size) = - inner.segsizes.append_or_update_last(lsn, newsize).unwrap(); - mem_usage += delta_size; - - delta_logical_size = newsize - oldsize; + inner.segsizes.append_or_update_last(lsn, newsize).unwrap(); + return Ok(newsize - oldsize); } } - inner.mem_usage += mem_usage; - GLOBAL_OPEN_MEM_USAGE.fetch_add(mem_usage, Ordering::Relaxed); - - delta_logical_size + Ok(0) } /// Remember that the relation was truncated at given LSN @@ -481,9 +464,7 @@ impl InMemoryLayer { let oldsize = inner.get_seg_size(lsn); assert!(segsize < oldsize); - let (old, delta_size) = inner.segsizes.append_or_update_last(lsn, segsize).unwrap(); - inner.mem_usage += delta_size; - GLOBAL_OPEN_MEM_USAGE.fetch_add(delta_size, Ordering::Relaxed); + let (old, _delta_size) = inner.segsizes.append_or_update_last(lsn, segsize).unwrap(); if old.is_some() { // We already had an entry for this LSN. That's odd.. @@ -505,7 +486,7 @@ impl InMemoryLayer { } /// - /// Initialize a new InMemoryLayer for, by copying the state at the given + /// Initialize a new OpenLayer for, by copying the state at the given /// point in time from given existing layer. /// pub fn create_successor_layer( @@ -515,14 +496,14 @@ impl InMemoryLayer { tenantid: ZTenantId, start_lsn: Lsn, oldest_pending_lsn: Lsn, - ) -> Result { + ) -> Result { let seg = src.get_seg_tag(); assert!(oldest_pending_lsn.is_aligned()); assert!(oldest_pending_lsn >= start_lsn); trace!( - "initializing new InMemoryLayer for writing {} on timeline {} at {}", + "initializing new OpenLayer for writing {} on timeline {} at {}", seg, timelineid, start_lsn, @@ -535,7 +516,23 @@ impl InMemoryLayer { segsizes.append(start_lsn, size).unwrap(); } - Ok(InMemoryLayer { + let delta_filename = DeltaFileName { + seg, + start_lsn, + end_lsn: Lsn(u64::MAX), + dropped: false, + } + .to_string(); + let filename = conf + .timeline_path(&timelineid, &tenantid) + .join(PathBuf::from(format!("open-{}", delta_filename))); + + let file = VirtualFile::open_with_options( + &filename, + OpenOptions::new().read(true).write(true).create(true), + )?; + + Ok(OpenLayer { conf, timelineid, tenantid, @@ -543,12 +540,11 @@ impl InMemoryLayer { start_lsn, oldest_pending_lsn, incremental: true, - inner: RwLock::new(InMemoryLayerInner { + inner: RwLock::new(OpenLayerInner { end_lsn: None, dropped: false, - page_versions: PageVersions::default(), + page_versions: PageVersions::new(file), segsizes, - mem_usage: 0, }), }) } @@ -578,16 +574,6 @@ impl InMemoryLayer { for (_blk, lsn, _pv) in inner.page_versions.ordered_page_version_iter(None) { assert!(lsn <= end_lsn); } - - // It's a bit premature to subtract the global mem usage here already. - // This layer consumes memory until it's written out to disk and dropped. - // But GLOBAL_OPEN_MEM_USAGE is used to trigger layer eviction, if there are - // too many open layers, and from that point of view this should no longer be - // counted against the global mem usage. - if inner.mem_usage > 0 { - GLOBAL_OPEN_MEM_USAGE.fetch_sub(inner.mem_usage, Ordering::Relaxed); - inner.mem_usage = 0; - } } } @@ -627,7 +613,8 @@ impl InMemoryLayer { self.start_lsn, end_lsn_exclusive, true, - inner.page_versions.ordered_page_version_iter(None), + &inner.page_versions, + None, inner.segsizes.clone(), )?; trace!( @@ -644,13 +631,9 @@ impl InMemoryLayer { // Since `end_lsn` is inclusive, subtract 1. // We want to make an ImageLayer for the last included LSN, - // so the DeltaLayer should exlcude that LSN. + // so the DeltaLayer should exclude that LSN. let end_lsn_inclusive = Lsn(end_lsn_exclusive.0 - 1); - let mut page_versions = inner - .page_versions - .ordered_page_version_iter(Some(end_lsn_inclusive)); - let mut delta_layers = Vec::new(); if self.start_lsn != end_lsn_inclusive { @@ -664,7 +647,8 @@ impl InMemoryLayer { self.start_lsn, end_lsn_inclusive, false, - page_versions, + &inner.page_versions, + Some(end_lsn_inclusive), segsizes, )?; delta_layers.push(delta_layer); @@ -675,7 +659,11 @@ impl InMemoryLayer { end_lsn_inclusive ); } else { - assert!(page_versions.next().is_none()); + assert!(inner + .page_versions + .ordered_page_version_iter(None) + .next() + .is_none()); } drop(inner); @@ -695,3 +683,27 @@ impl InMemoryLayer { }) } } + +impl Drop for OpenLayer { + fn drop(&mut self) { + let delta_filename = DeltaFileName { + seg: self.seg, + start_lsn: self.start_lsn, + end_lsn: Lsn(u64::MAX), + dropped: false, // even if it dropped now, that's not reflected in the filename + } + .to_string(); + let filename = self + .conf + .timeline_path(&self.timelineid, &self.tenantid) + .join(PathBuf::from(format!("open-{}", delta_filename))); + + if let Err(err) = std::fs::remove_file(&filename) { + error!( + "could not unlink open layer file {}: {}", + filename.display(), + err + ); + } + } +} diff --git a/pageserver/src/layered_repository/layer_map.rs b/pageserver/src/layered_repository/layer_map.rs index 45d5a90fe0..d2c03d9e9e 100644 --- a/pageserver/src/layered_repository/layer_map.rs +++ b/pageserver/src/layered_repository/layer_map.rs @@ -11,7 +11,7 @@ use crate::layered_repository::interval_tree::{IntervalItem, IntervalIter, IntervalTree}; use crate::layered_repository::storage_layer::{Layer, SegmentTag}; -use crate::layered_repository::InMemoryLayer; +use crate::layered_repository::OpenLayer; use crate::relish::*; use anyhow::Result; use lazy_static::lazy_static; @@ -67,7 +67,7 @@ impl LayerMap { /// Get the open layer for given segment for writing. Or None if no open /// layer exists. /// - pub fn get_open(&self, tag: &SegmentTag) -> Option> { + pub fn get_open(&self, tag: &SegmentTag) -> Option> { let segentry = self.segs.get(tag)?; segentry @@ -78,7 +78,7 @@ impl LayerMap { /// /// Insert an open in-memory layer /// - pub fn insert_open(&mut self, layer: Arc) { + pub fn insert_open(&mut self, layer: Arc) { let segentry = self.segs.entry(layer.get_seg_tag()).or_default(); let layer_id = segentry.update_open(Arc::clone(&layer)); @@ -214,7 +214,7 @@ impl LayerMap { } /// Return the oldest in-memory layer, along with its generation number. - pub fn peek_oldest_open(&mut self) -> Option<(LayerId, Arc, u64)> { + pub fn peek_oldest_open(&mut self) -> Option<(LayerId, Arc, u64)> { let global_map = GLOBAL_LAYER_MAP.read().unwrap(); while let Some(oldest_entry) = self.open_layers.peek() { @@ -279,7 +279,7 @@ impl IntervalItem for dyn Layer { /// Per-segment entry in the LayerMap::segs hash map. Holds all the layers /// associated with the segment. /// -/// The last layer that is open for writes is always an InMemoryLayer, +/// The last layer that is open for writes is always an OpenLayer, /// and is kept in a separate field, because there can be only one for /// each segment. The older layers, stored on disk, are kept in an /// IntervalTree. @@ -323,7 +323,7 @@ impl SegEntry { // Set new open layer for a SegEntry. // It's ok to rewrite previous open layer, // but only if it is not writeable anymore. - pub fn update_open(&mut self, layer: Arc) -> LayerId { + pub fn update_open(&mut self, layer: Arc) -> LayerId { if let Some(prev_open_layer_id) = &self.open_layer_id { if let Some(prev_open_layer) = GLOBAL_LAYER_MAP.read().unwrap().get(prev_open_layer_id) { @@ -414,15 +414,15 @@ mod tests { forknum: 0, }); - /// Construct a dummy InMemoryLayer for testing - fn dummy_inmem_layer( + /// Construct a dummy OpenLayer for testing + fn dummy_open_layer( conf: &'static PageServerConf, segno: u32, start_lsn: Lsn, oldest_pending_lsn: Lsn, - ) -> Arc { + ) -> Arc { Arc::new( - InMemoryLayer::create( + OpenLayer::create( conf, ZTimelineId::from_str("00000000000000000000000000000000").unwrap(), ZTenantId::from_str("00000000000000000000000000000000").unwrap(), @@ -439,20 +439,20 @@ mod tests { #[test] fn test_open_layers() -> Result<()> { - let conf = PageServerConf::dummy_conf(PageServerConf::test_repo_dir("dummy_inmem_layer")); + let conf = PageServerConf::dummy_conf(PageServerConf::test_repo_dir("dummy_open_layer")); let conf = Box::leak(Box::new(conf)); let mut layers = LayerMap::default(); let gen1 = layers.increment_generation(); - layers.insert_open(dummy_inmem_layer(conf, 0, Lsn(0x100), Lsn(0x100))); - layers.insert_open(dummy_inmem_layer(conf, 1, Lsn(0x100), Lsn(0x200))); - layers.insert_open(dummy_inmem_layer(conf, 2, Lsn(0x100), Lsn(0x120))); - layers.insert_open(dummy_inmem_layer(conf, 3, Lsn(0x100), Lsn(0x110))); + layers.insert_open(dummy_open_layer(conf, 0, Lsn(0x100), Lsn(0x100))); + layers.insert_open(dummy_open_layer(conf, 1, Lsn(0x100), Lsn(0x200))); + layers.insert_open(dummy_open_layer(conf, 2, Lsn(0x100), Lsn(0x120))); + layers.insert_open(dummy_open_layer(conf, 3, Lsn(0x100), Lsn(0x110))); let gen2 = layers.increment_generation(); - layers.insert_open(dummy_inmem_layer(conf, 4, Lsn(0x100), Lsn(0x110))); - layers.insert_open(dummy_inmem_layer(conf, 5, Lsn(0x100), Lsn(0x100))); + layers.insert_open(dummy_open_layer(conf, 4, Lsn(0x100), Lsn(0x110))); + layers.insert_open(dummy_open_layer(conf, 5, Lsn(0x100), Lsn(0x100))); // A helper function (closure) to pop the next oldest open entry from the layer map, // and assert that it is what we'd expect diff --git a/pageserver/src/layered_repository/page_versions.rs b/pageserver/src/layered_repository/page_versions.rs index 4f27b59981..5fe942d7f4 100644 --- a/pageserver/src/layered_repository/page_versions.rs +++ b/pageserver/src/layered_repository/page_versions.rs @@ -1,40 +1,78 @@ +//! +//! Data structure to ingest incoming WAL into an append-only file. +//! +//! - The file is considered temporary, and will be discarded on crash +//! - based on a B-tree +//! + +use std::os::unix::fs::FileExt; use std::{collections::HashMap, ops::RangeBounds, slice}; +use anyhow::Result; + +use std::cmp::min; +use std::io::Seek; + use zenith_utils::{lsn::Lsn, vec_map::VecMap}; use super::storage_layer::PageVersion; +use crate::virtual_file::VirtualFile; -const EMPTY_SLICE: &[(Lsn, PageVersion)] = &[]; +use zenith_utils::bin_ser::LeSer; -#[derive(Debug, Default)] -pub struct PageVersions(HashMap>); +const EMPTY_SLICE: &[(Lsn, u64)] = &[]; + +pub struct PageVersions { + map: HashMap>, + + /// The PageVersion structs are stored in a serialized format in this file. + /// Each serialized PageVersion is preceded by a 'u32' length field. + /// The 'map' stores offsets into this file. + file: VirtualFile, +} impl PageVersions { + pub fn new(file: VirtualFile) -> PageVersions { + PageVersions { + map: HashMap::new(), + file, + } + } + pub fn append_or_update_last( &mut self, blknum: u32, lsn: Lsn, page_version: PageVersion, - ) -> (Option, usize) { - let map = self.0.entry(blknum).or_insert_with(VecMap::default); - map.append_or_update_last(lsn, page_version).unwrap() + ) -> Result> { + // remember starting position + let pos = self.file.stream_position()?; + + // make room for the 'length' field by writing zeros as a placeholder. + self.file.seek(std::io::SeekFrom::Start(pos + 4)).unwrap(); + + page_version.ser_into(&mut self.file).unwrap(); + + // write the 'length' field. + let len = self.file.stream_position()? - pos - 4; + let lenbuf = u32::to_ne_bytes(len as u32); + self.file.write_all_at(&lenbuf, pos)?; + + let map = self.map.entry(blknum).or_insert_with(VecMap::default); + Ok(map.append_or_update_last(lsn, pos as u64).unwrap().0) } /// Get all [`PageVersion`]s in a block - pub fn get_block_slice(&self, blknum: u32) -> &[(Lsn, PageVersion)] { - self.0 + fn get_block_slice(&self, blknum: u32) -> &[(Lsn, u64)] { + self.map .get(&blknum) .map(VecMap::as_slice) .unwrap_or(EMPTY_SLICE) } /// Get a range of [`PageVersions`] in a block - pub fn get_block_lsn_range>( - &self, - blknum: u32, - range: R, - ) -> &[(Lsn, PageVersion)] { - self.0 + pub fn get_block_lsn_range>(&self, blknum: u32, range: R) -> &[(Lsn, u64)] { + self.map .get(&blknum) .map(|vec_map| vec_map.slice_range(range)) .unwrap_or(EMPTY_SLICE) @@ -43,7 +81,7 @@ impl PageVersions { /// Iterate through [`PageVersion`]s in (block, lsn) order. /// If a [`cutoff_lsn`] is set, only show versions with `lsn < cutoff_lsn` pub fn ordered_page_version_iter(&self, cutoff_lsn: Option) -> OrderedPageVersionIter<'_> { - let mut ordered_blocks: Vec = self.0.keys().cloned().collect(); + let mut ordered_blocks: Vec = self.map.keys().cloned().collect(); ordered_blocks.sort_unstable(); let slice = ordered_blocks @@ -59,6 +97,40 @@ impl PageVersions { cur_slice_iter: slice.iter(), } } + + /// Returns a 'Read' that reads the page version at given offset. + pub fn reader(&self, pos: u64) -> Result { + // read length + let mut lenbuf = [0u8; 4]; + self.file.read_exact_at(&mut lenbuf, pos)?; + let len = u32::from_ne_bytes(lenbuf); + + Ok(PageVersionReader { + file: &self.file, + pos: pos + 4, + end_pos: pos + 4 + len as u64, + }) + } + + pub fn get_page_version(&self, pos: u64) -> Result { + let mut reader = self.reader(pos)?; + Ok(PageVersion::des_from(&mut reader)?) + } +} + +pub struct PageVersionReader<'a> { + file: &'a VirtualFile, + pos: u64, + end_pos: u64, +} + +impl<'a> std::io::Read for PageVersionReader<'a> { + fn read(&mut self, buf: &mut [u8]) -> Result { + let len = min(buf.len(), (self.end_pos - self.pos) as usize); + let n = self.file.read_at(&mut buf[..len], self.pos)?; + self.pos += n as u64; + Ok(n) + } } pub struct OrderedPageVersionIter<'a> { @@ -69,7 +141,7 @@ pub struct OrderedPageVersionIter<'a> { cutoff_lsn: Option, - cur_slice_iter: slice::Iter<'a, (Lsn, PageVersion)>, + cur_slice_iter: slice::Iter<'a, (Lsn, u64)>, } impl OrderedPageVersionIter<'_> { @@ -83,14 +155,14 @@ impl OrderedPageVersionIter<'_> { } impl<'a> Iterator for OrderedPageVersionIter<'a> { - type Item = (u32, Lsn, &'a PageVersion); + type Item = (u32, Lsn, u64); fn next(&mut self) -> Option { loop { - if let Some((lsn, page_version)) = self.cur_slice_iter.next() { + if let Some((lsn, pos)) = self.cur_slice_iter.next() { if self.is_lsn_before_cutoff(lsn) { let blknum = self.ordered_blocks[self.cur_block_idx]; - return Some((blknum, *lsn, page_version)); + return Some((blknum, *lsn, *pos)); } } @@ -109,8 +181,12 @@ mod tests { use super::*; #[test] - fn test_ordered_iter() { - let mut page_versions = PageVersions::default(); + fn test_ordered_iter() -> Result<()> { + let test_file_path = crate::PageServerConf::test_repo_dir("test_ordered_iter"); + let file = VirtualFile::create(&test_file_path)?; + + let mut page_versions = PageVersions::new(file); + const BLOCKS: u32 = 1000; const LSNS: u64 = 50; @@ -119,11 +195,11 @@ mod tests { for blknum in 0..BLOCKS { for lsn in 0..LSNS { - let (old, _delta_size) = page_versions.append_or_update_last( + let old = page_versions.append_or_update_last( blknum, Lsn(lsn), empty_page_version.clone(), - ); + )?; assert!(old.is_none()); } } @@ -150,5 +226,7 @@ mod tests { } assert!(iter.next().is_none()); assert!(iter.next().is_none()); // should be robust against excessive next() calls + + Ok(()) } } diff --git a/pageserver/src/walreceiver.rs b/pageserver/src/walreceiver.rs index d6212a5f54..1f84ed8507 100644 --- a/pageserver/src/walreceiver.rs +++ b/pageserver/src/walreceiver.rs @@ -5,7 +5,6 @@ //! //! We keep one WAL receiver active per timeline. -use crate::layered_repository; use crate::relish::*; use crate::restore_local_repo; use crate::tenant_mgr; @@ -176,7 +175,7 @@ fn thread_main(conf: &'static PageServerConf, timelineid: ZTimelineId, tenantid: } fn walreceiver_main( - conf: &PageServerConf, + _conf: &PageServerConf, timelineid: ZTimelineId, wal_producer_connstr: &str, tenantid: ZTenantId, @@ -296,9 +295,6 @@ fn walreceiver_main( caught_up = true; } - // Release memory if needed - layered_repository::evict_layer_if_needed(conf)?; - Some(endlsn) }