rust/neon
1
0
Fork 0
mirror of https://github.com/neondatabase/neon.git synced 2026-07-08 14:40:37 +00:00
Code Issues Packages Projects Releases Wiki Activity

Replace in-memory layers and OOM-triggered eviction with temp files.

Browse Source 
The "in-memory layer" is misnomer now, each in-memory layer is now actually
backed by a file. The files are ephemeral, in that they don't survive page
server crash or shutdown.

This is a bit slow, as we now access the file for every operation. The
next commit adds code to use the page cache for theses files.

This includes changes from 'inmemory-layer-chunks' branch to serialize /
the page versions when they are added to the open layer. The difference is
that they are not serialized to the expandable in-memory "chunk buffer", but
written out to the file.
This commit is contained in:
Heikki Linnakangas
2021-11-11 01:07:50 +02:00
parent 431d32756b
commit 04c4167e72
8 changed files with 255 additions and 270 deletions
Download Patch File Download Diff File Expand all files Collapse all files

54
pageserver/src/layered_repository.rs
View File

@@ -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<Arc<InMemoryLayer>> {
fn get_layer_for_write(&self, seg: SegmentTag, lsn: Lsn) -> Result<Arc<OpenLayer>> {
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<InMemoryLayer> = Arc::new(layer);
let layer_rc: Arc<OpenLayer> = 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(())
}

10
pageserver/src/layered_repository/blob.rs
View File

@@ -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<W: Write> BlobWriter<W> {
Self { writer, offset: 0 }
}
pub fn write_blob(&mut self, blob: &[u8]) -> Result<BlobRange> {
self.writer.write_all(blob)?;
pub fn write_blob_from_reader(&mut self, r: &mut impl Read) -> Result<BlobRange> {
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)
}

23
pageserver/src/layered_repository/delta_layer.rs
View File

@@ -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<Item = (u32, Lsn, &'a PageVersion)>,
page_versions: &PageVersions,
cutoff: Option<Lsn>,
relsizes: VecMap<Lsn, u32>,
) -> Result<DeltaLayer> {
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

78
pageserver/src/layered_repository/global_layer_map.rs
View File

@@ -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<OpenLayers> = 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<InMemoryLayer>),
Occupied(Arc<OpenLayer>),
/// Vacant slots form a linked list, the value is the index
/// of the next vacant slot in the list.
Vacant(Option<usize>),
@@ -56,14 +47,13 @@ struct Slot {
pub struct OpenLayers {
slots: Vec<Slot>,
num_occupied: usize,
next_victim: AtomicUsize,
// Head of free-slot list.
next_empty_slot_idx: Option<usize>,
}
impl OpenLayers {
pub fn insert(&mut self, layer: Arc<InMemoryLayer>) -> LayerId {
pub fn insert(&mut self, layer: Arc<OpenLayer>) -> 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<Arc<InMemoryLayer>> {
pub fn get(&self, layer_id: &LayerId) -> Option<Arc<OpenLayer>> {
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<InMemoryLayer>)> {
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<InMemoryLayer>)> {
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<u8> {
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];

196
pageserver/src/layered_repository/inmemory_layer.rs
View File

@@ -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<InMemoryLayerInner>,
inner: RwLock<OpenLayerInner>,
/// 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<Lsn>,
@@ -71,18 +70,9 @@ pub struct InMemoryLayerInner {
/// a non-blocky rel, 'segsizes' is not used and is always empty.
///
segsizes: VecMap<Lsn, u32>,
/// 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<ImageLayer>,
}
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<InMemoryLayer> {
) -> Result<OpenLayer> {
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<u32> {
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<u32> {
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<u32> {
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<InMemoryLayer> {
) -> Result<OpenLayer> {
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
);
}
}
}

34
pageserver/src/layered_repository/layer_map.rs
View File

@@ -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<Arc<InMemoryLayer>> {
pub fn get_open(&self, tag: &SegmentTag) -> Option<Arc<OpenLayer>> {
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<InMemoryLayer>) {
pub fn insert_open(&mut self, layer: Arc<OpenLayer>) {
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<InMemoryLayer>, u64)> {
pub fn peek_oldest_open(&mut self) -> Option<(LayerId, Arc<OpenLayer>, 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<InMemoryLayer>) -> LayerId {
pub fn update_open(&mut self, layer: Arc<OpenLayer>) -> 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<InMemoryLayer> {
) -> Arc<OpenLayer> {
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

124
pageserver/src/layered_repository/page_versions.rs
View File

@@ -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<u32, VecMap<Lsn, PageVersion>>);
const EMPTY_SLICE: &[(Lsn, u64)] = &[];
pub struct PageVersions {
map: HashMap<u32, VecMap<Lsn, u64>>,
/// 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<PageVersion>, usize) {
let map = self.0.entry(blknum).or_insert_with(VecMap::default);
map.append_or_update_last(lsn, page_version).unwrap()
) -> Result<Option<u64>> {
// 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<R: RangeBounds<Lsn>>(
&self,
blknum: u32,
range: R,
) -> &[(Lsn, PageVersion)] {
self.0
pub fn get_block_lsn_range<R: RangeBounds<Lsn>>(&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<Lsn>) -> OrderedPageVersionIter<'_> {
let mut ordered_blocks: Vec<u32> = self.0.keys().cloned().collect();
let mut ordered_blocks: Vec<u32> = 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<PageVersionReader, std::io::Error> {
// 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<PageVersion> {
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<usize, std::io::Error> {
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<Lsn>,
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<Self::Item> {
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(())
}
}

6
pageserver/src/walreceiver.rs
View File

@@ -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)
}