From df3e403967bd6a5cad486bd38f38401327e08261 Mon Sep 17 00:00:00 2001 From: Heikki Linnakangas Date: Sun, 11 Jul 2021 11:40:48 +0300 Subject: [PATCH] Introduce a new "layered" repository implementation. This replaces the RocksDB based implementation with an approach using "snapshot files" on disk, and in-memory btreemaps to hold the recent changes. This make the repository implementation a configuration option. You can choose 'layered' or 'rocksdb' in the "pageserver init" call, but there is no corresponding --repository-formt option in 'zenith init', so in practice you have to change the default in pageserver.rs if you want to test different implementations. The unit tests have been refactored to exercise both implementations, though. 'layered' is now the default. TODOs: - Push/pull is not implemented, causing 'test_history_inmemory' test in 'cargo test' to fail. - Garbage collection has not been implemented yet. The 'test_gc' test is failing because of that. - Unlinking relations has not been implemented either. (That has no user visible effect until garbage collection is implemented) --- pageserver/src/bin/pageserver.rs | 30 +- pageserver/src/branches.rs | 26 + pageserver/src/layered_repository.rs | 717 ++++++++++++++++++ .../src/layered_repository/inmemory_layer.rs | 371 +++++++++ .../src/layered_repository/snapshot_layer.rs | 486 ++++++++++++ .../src/layered_repository/storage_layer.rs | 75 ++ pageserver/src/lib.rs | 9 + pageserver/src/page_cache.rs | 20 +- pageserver/src/repository.rs | 91 ++- 9 files changed, 1803 insertions(+), 22 deletions(-) create mode 100644 pageserver/src/layered_repository.rs create mode 100644 pageserver/src/layered_repository/inmemory_layer.rs create mode 100644 pageserver/src/layered_repository/snapshot_layer.rs create mode 100644 pageserver/src/layered_repository/storage_layer.rs diff --git a/pageserver/src/bin/pageserver.rs b/pageserver/src/bin/pageserver.rs index d31f7dc90d..32fa2bc125 100644 --- a/pageserver/src/bin/pageserver.rs +++ b/pageserver/src/bin/pageserver.rs @@ -17,7 +17,10 @@ use anyhow::Result; use clap::{App, Arg, ArgMatches}; use daemonize::Daemonize; -use pageserver::{branches, logger, page_cache, page_service, PageServerConf}; +use slog::{Drain, FnValue}; + +use pageserver::{branches, page_cache, page_service}; +use pageserver::{PageServerConf, RepositoryFormat}; const DEFAULT_LISTEN_ADDR: &str = "127.0.0.1:64000"; @@ -33,6 +36,7 @@ struct CfgFileParams { gc_horizon: Option, gc_period: Option, pg_distrib_dir: Option, + repository_format: Option, } impl CfgFileParams { @@ -47,6 +51,7 @@ impl CfgFileParams { gc_horizon: get_arg("gc_horizon"), gc_period: get_arg("gc_period"), pg_distrib_dir: get_arg("postgres-distrib"), + repository_format: get_arg("repository-format"), } } @@ -58,6 +63,7 @@ impl CfgFileParams { gc_horizon: self.gc_horizon.or(other.gc_horizon), gc_period: self.gc_period.or(other.gc_period), pg_distrib_dir: self.pg_distrib_dir.or(other.pg_distrib_dir), + repository_format: self.repository_format.or(other.repository_format), } } @@ -86,6 +92,21 @@ impl CfgFileParams { anyhow::bail!("Can't find postgres binary at {:?}", pg_distrib_dir); } + // + // FIXME: This pageserver --repository-format option is pretty useless as it + // isn't exposed as an option to "zenith init". But you can change the default + // here if you want to test the rocksdb implementation: + // + let repository_format = match self.repository_format.as_ref() { + Some(repo_format_str) if repo_format_str == "rocksdb" => RepositoryFormat::RocksDb, + Some(repo_format_str) if repo_format_str == "layered" => RepositoryFormat::Layered, + Some(repo_format_str) => anyhow::bail!( + "invalid --repository-format '{}', must be 'rocksdb' or 'layered'", + repo_format_str + ), + None => RepositoryFormat::Layered, // default + }; + Ok(PageServerConf { daemonize: false, @@ -98,6 +119,7 @@ impl CfgFileParams { workdir: PathBuf::from("."), pg_distrib_dir, + repository_format, }) } } @@ -157,6 +179,12 @@ fn main() -> Result<()> { .help("Create tenant during init") .requires("init"), ) + .arg( + Arg::with_name("repository-format") + .long("repository-format") + .takes_value(true) + .help("Which repository implementation to use, 'rocksdb' or 'layered'"), + ) .get_matches(); let workdir = Path::new(arg_matches.value_of("workdir").unwrap_or(".zenith")); diff --git a/pageserver/src/branches.rs b/pageserver/src/branches.rs index a33597bb2d..e488d7f5bc 100644 --- a/pageserver/src/branches.rs +++ b/pageserver/src/branches.rs @@ -179,8 +179,34 @@ fn bootstrap_timeline( info!("bootstrap_timeline {:?} at lsn {}", pgdata_path, lsn); + // We don't use page_cache here, because we don't want to spawn the WAL redo thread during + // repository initialization. + // + // FIXME: That caused trouble, because the WAL redo thread launched initdb in the background, + // and it kept running even after the "zenith init" had exited. In tests, we started the + // page server immediately after that, so that initdb was still running in the background, + // and we failed to run initdb again in the same directory. This has been solved for the + // rapid init+start case now, but the general race condition remains if you restart the + // server quickly. + let walredo_mgr = std::sync::Arc::new(crate::walredo::DummyRedoManager {}); + let repo: Box = match conf.repository_format { + crate::RepositoryFormat::Layered => Box::new( + crate::layered_repository::LayeredRepository::new(conf, walredo_mgr), + ), + crate::RepositoryFormat::RocksDb => { + let storage = crate::rocksdb_storage::RocksObjectStore::create(conf)?; + + Box::new(crate::object_repository::ObjectRepository::new( + conf, + std::sync::Arc::new(storage), + walredo_mgr, + )) + } + }; + let timeline = repo.create_empty_timeline(tli, lsn)?; restore_local_repo::import_timeline_from_postgres_datadir(&pgdata_path, &*timeline, lsn)?; + timeline.checkpoint()?; let wal_dir = pgdata_path.join("pg_wal"); restore_local_repo::import_timeline_wal(&wal_dir, &*timeline, timeline.get_last_record_lsn())?; diff --git a/pageserver/src/layered_repository.rs b/pageserver/src/layered_repository.rs new file mode 100644 index 0000000000..a86f3041d1 --- /dev/null +++ b/pageserver/src/layered_repository.rs @@ -0,0 +1,717 @@ +//! +//! Zenith repository implementation that keeps old data in "snapshot files", and +//! the recent changes in memory. See layered_repository/snapshot_layer.rs and +//! layered_repository/inmemory_layer.rs, respectively. The functions here are +//! responsible for locating the correct layer for the get/put call, tracing +//! timeline branching history as needed. +//! +//! The snapshot files are stored in the .zenith/timelines/ directory. +//! In addition to the snapshot files, there is a metadata file in the +//! same directory that contains information about the timline, in particular its +//! parent timeline, and the last LSN that has been written to disk. +//! +//! This is based on the design at https://github.com/zenithdb/rfcs/pull/8. +//! Some notable differences and details not covered by the RFC: +//! +//! - A snapshot layer doesn't contain a snapshot at a specific LSN, but all page +//! versions in a range of LSNs. So each snapshot file has a start and end LSN. +//! + +use anyhow::{bail, Context, Result}; +use bytes::Bytes; +use log::*; +use serde::{Deserialize, Serialize}; + +use std::collections::HashSet; +use std::collections::{BTreeMap, HashMap}; +use std::fs::File; +use std::io::Write; +use std::ops::Bound::Included; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +use crate::repository::{GcResult, History, RelTag, Repository, Timeline, WALRecord}; +use crate::restore_local_repo::import_timeline_wal; +use crate::walredo::WalRedoManager; +use crate::PageServerConf; +use crate::ZTimelineId; + +use zenith_utils::bin_ser::BeSer; +use zenith_utils::lsn::{AtomicLsn, Lsn}; +use zenith_utils::seqwait::SeqWait; + +mod inmemory_layer; +mod snapshot_layer; +mod storage_layer; + +use inmemory_layer::InMemoryLayer; +use snapshot_layer::SnapshotLayer; +use storage_layer::Layer; + +// Timeout when waiting or WAL receiver to catch up to an LSN given in a GetPage@LSN call. +static TIMEOUT: Duration = Duration::from_secs(60); + +/// +/// Repository consists of multiple timelines. Keep them in a hash table. +/// +pub struct LayeredRepository { + conf: &'static PageServerConf, + timelines: Mutex>>, + + walredo_mgr: Arc, +} + +/// Public interface +impl Repository for LayeredRepository { + fn get_timeline(&self, timelineid: ZTimelineId) -> Result> { + let mut timelines = self.timelines.lock().unwrap(); + + Ok(self.get_timeline_locked(timelineid, &mut timelines)?) + } + + fn create_empty_timeline( + &self, + timelineid: ZTimelineId, + start_lsn: Lsn, + ) -> Result> { + let mut timelines = self.timelines.lock().unwrap(); + + std::fs::create_dir_all(self.conf.timeline_path(timelineid))?; + + // Write initial metadata. + let metadata = TimelineMetadata { + last_valid_lsn: start_lsn, + last_record_lsn: start_lsn, + ancestor_timeline: None, + ancestor_lsn: start_lsn, + }; + Self::save_metadata(self.conf, timelineid, &metadata)?; + + let timeline = LayeredTimeline::new( + self.conf, + metadata, + None, + timelineid, + self.walredo_mgr.clone(), + )?; + + let timeline_rc = Arc::new(timeline); + let r = timelines.insert(timelineid, timeline_rc.clone()); + assert!(r.is_none()); + Ok(timeline_rc) + } + + /// Branch a timeline + fn branch_timeline(&self, src: ZTimelineId, dst: ZTimelineId, start_lsn: Lsn) -> Result<()> { + // just to check the source timeline exists + let _src_timeline = self.get_timeline(src)?; + + // Create the metadata file, noting the ancestor of th new timeline. There is initially + // no data in it, but all the read-calls know to look into the ancestor. + let metadata = TimelineMetadata { + last_valid_lsn: start_lsn, + last_record_lsn: start_lsn, + ancestor_timeline: Some(src), + ancestor_lsn: start_lsn, + }; + std::fs::create_dir_all(self.conf.timeline_path(dst))?; + Self::save_metadata(self.conf, dst, &metadata)?; + + info!("branched timeline {} from {} at {}", dst, src, start_lsn); + + Ok(()) + } +} + +/// Private functions +impl LayeredRepository { + // Implementation of the public `get_timeline` function. This differs from the public + // interface in that the caller must already hold the mutex on the 'timelines' hashmap. + fn get_timeline_locked( + &self, + timelineid: ZTimelineId, + timelines: &mut HashMap>, + ) -> Result> { + match timelines.get(&timelineid) { + Some(timeline) => Ok(timeline.clone()), + None => { + let metadata = Self::load_metadata(self.conf, timelineid)?; + + let ancestor = if let Some(ancestor_timelineid) = metadata.ancestor_timeline { + Some(self.get_timeline_locked(ancestor_timelineid, timelines)?) + } else { + None + }; + + let timeline = LayeredTimeline::new( + self.conf, + metadata, + ancestor, + timelineid, + self.walredo_mgr.clone(), + )?; + + // Load any new WAL after the last checkpoint into memory. + info!( + "Loading WAL for timeline {} starting at {}", + timelineid, + timeline.get_last_record_lsn() + ); + let wal_dir = self.conf.timeline_path(timelineid).join("wal"); + import_timeline_wal(&wal_dir, &timeline, timeline.get_last_record_lsn())?; + + let timeline_rc = Arc::new(timeline); + timelines.insert(timelineid, timeline_rc.clone()); + Ok(timeline_rc) + } + } + } + + pub fn new( + conf: &'static PageServerConf, + walredo_mgr: Arc, + ) -> LayeredRepository { + LayeredRepository { + conf: conf, + timelines: Mutex::new(HashMap::new()), + walredo_mgr, + } + } + + /// Save metadata to file + fn save_metadata( + conf: &'static PageServerConf, + timelineid: ZTimelineId, + data: &TimelineMetadata, + ) -> Result<()> { + let path = conf.timeline_path(timelineid).join("metadata"); + let mut file = File::create(&path)?; + + file.write_all(&TimelineMetadata::ser(data)?)?; + + Ok(()) + } + + fn load_metadata( + conf: &'static PageServerConf, + timelineid: ZTimelineId, + ) -> Result { + let path = conf.timeline_path(timelineid).join("metadata"); + let data = std::fs::read(&path)?; + + Ok(TimelineMetadata::des(&data)?) + } +} + +/// LayerMap is a BTreeMap keyed by RelTag and the snapshot file's start LSN. +/// It provides a couple of convenience functions over a plain BTreeMap +struct LayerMap(BTreeMap<(RelTag, Lsn), Arc>); + +impl LayerMap { + /// + /// Look up using the given rel tag and LSN. This differs from a plain + /// key-value lookup in that if there is any layer that covers the + /// given LSN, or precedes the given LSN, it is returned. In other words, + /// you don't need to know the exact start LSN of the layer. + /// + fn get(&self, tag: RelTag, lsn: Lsn) -> Option> { + let startkey = (tag, Lsn(0)); + let endkey = (tag, lsn); + + if let Some((_k, v)) = self + .0 + .range((Included(startkey), Included(endkey))) + .next_back() + { + Some(Arc::clone(v)) + } else { + None + } + } + + fn insert(&mut self, layer: Arc) { + let tag = layer.get_tag(); + let start_lsn = layer.get_start_lsn(); + + self.0.insert((tag, start_lsn), Arc::clone(&layer)); + } +} + +impl Default for LayerMap { + fn default() -> Self { + LayerMap(BTreeMap::new()) + } +} + +/// Metadata stored on disk for each timeline +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct TimelineMetadata { + last_valid_lsn: Lsn, + last_record_lsn: Lsn, + ancestor_timeline: Option, + ancestor_lsn: Lsn, +} + +pub struct LayeredTimeline { + conf: &'static PageServerConf, + + timelineid: ZTimelineId, + + layers: Mutex, + + // WAL redo manager + walredo_mgr: Arc, + + // What page versions do we hold in the repository? If we get a + // request > last_valid_lsn, we need to wait until we receive all + // the WAL up to the request. The SeqWait provides functions for + // that. TODO: If we get a request for an old LSN, such that the + // versions have already been garbage collected away, we should + // throw an error, but we don't track that currently. + // + // last_record_lsn points to the end of last processed WAL record. + // It can lag behind last_valid_lsn, if the WAL receiver has + // received some WAL after the end of last record, but not the + // whole next record yet. In the page cache, we care about + // last_valid_lsn, but if the WAL receiver needs to restart the + // streaming, it needs to restart at the end of last record, so we + // track them separately. last_record_lsn should perhaps be in + // walreceiver.rs instead of here, but it seems convenient to keep + // all three values together. + // + last_valid_lsn: SeqWait, + last_record_lsn: AtomicLsn, + + // Parent timeline that this timeline was branched from, and the LSN + // of the branch point. + ancestor_timeline: Option>, + ancestor_lsn: Lsn, +} + +/// Public interface functions +impl Timeline for LayeredTimeline { + /// Look up given page in the cache. + fn get_page_at_lsn(&self, tag: BufferTag, lsn: Lsn) -> Result { + trace!("get_page_at_lsn: {:?} at {}", tag, lsn); + let lsn = self.wait_lsn(lsn)?; + + if let Some((snapfile, lsn)) = self.get_snapshot_file_for_read(tag.rel, lsn)? { + snapfile.get_page_at_lsn(&*self.walredo_mgr, tag.blknum, lsn) + } else { + bail!("relation {} not found at {}", tag.rel, lsn); + } + } + + fn get_rel_size(&self, rel: RelTag, lsn: Lsn) -> Result { + let lsn = self.wait_lsn(lsn)?; + + if let Some((snapfile, lsn)) = self.get_snapshot_file_for_read(rel, lsn)? { + let result = snapfile.get_rel_size(lsn); + trace!( + "get_relsize: rel {} at {}/{} -> {:?}", + rel, + self.timelineid, + lsn, + result + ); + result + } else { + warn!( + "get_relsize: rel {} at {}/{} -> not found", + rel, self.timelineid, lsn + ); + bail!("relation {} not found at {}", rel, lsn); + } + } + + fn get_rel_exists(&self, rel: RelTag, lsn: Lsn) -> Result { + let lsn = self.wait_lsn(lsn)?; + + let result; + if let Some((snapfile, lsn)) = self.get_snapshot_file_for_read(rel, lsn)? { + result = snapfile.get_rel_exists(lsn)?; + } else { + result = false; + } + + trace!("get_relsize_exists: {:?} at {} -> {:?}", rel, lsn, result); + Ok(result) + } + + fn list_rels(&self, spcnode: u32, dbnode: u32, _lsn: Lsn) -> Result> { + // SnapshotFile::list_rels works by scanning the directory on disk. Make sure + // we have a file on disk for each relation. + self.checkpoint()?; + + // List all rels in this timeline, and all its ancestors. + let mut all_rels = HashSet::new(); + let mut timeline = self; + loop { + let rels = SnapshotLayer::list_rels(self.conf, timeline.timelineid, spcnode, dbnode)?; + + // FIXME: We should filter out relations that don't exist at the given LSN. + all_rels.extend(rels.iter()); + + if let Some(ancestor) = timeline.ancestor_timeline.as_ref() { + timeline = ancestor; + continue; + } else { + break; + } + } + + Ok(all_rels) + } + + fn history<'a>(&'a self) -> Result> { + // TODO + todo!(); + } + + fn gc_iteration(&self, _horizon: u64) -> Result { + //TODO + Ok(Default::default()) + } + + fn put_wal_record(&self, tag: BufferTag, rec: WALRecord) -> Result<()> { + debug!("put_wal_record: {:?} at {}", tag, rec.lsn); + + let snapfile = self.get_snapshot_file_for_write(tag.rel, rec.lsn)?; + snapfile.put_wal_record(tag.blknum, rec) + } + + fn put_truncation(&self, rel: RelTag, lsn: Lsn, relsize: u32) -> anyhow::Result<()> { + debug!("put_truncation: {:?} at {}", relsize, lsn); + + let snapfile = self.get_snapshot_file_for_write(rel, lsn)?; + snapfile.put_truncation(lsn, relsize) + } + + fn put_page_image(&self, tag: BufferTag, lsn: Lsn, img: Bytes) -> Result<()> { + debug!("put_page_image: {:?} at {}", tag, lsn); + + let snapfile = self.get_snapshot_file_for_write(tag.rel, lsn)?; + snapfile.put_page_image(tag.blknum, lsn, img) + } + + fn put_unlink(&self, _tag: RelTag, _lsn: Lsn) -> Result<()> { + // TODO + Ok(()) + } + + /// + /// Flush to disk all data that was written with the put_* functions + /// + /// NOTE: This has nothing to do with checkpoint in PostgreSQL. We don't + /// know anything about them here in the repository. + fn checkpoint(&self) -> Result<()> { + let last_valid_lsn = self.last_valid_lsn.load(); + + let mut layers = self.layers.lock().unwrap(); + + // Walk through each SnapshotFile in memory, and write any + // dirty ones to disk. + // + // Note: We release all the in-memory SnapshotFile entries, and + // start fresh with an empty map. This keeps memory usage in check, + // but is perhaps too aggressive. + // + let snapfiles = std::mem::take(&mut *layers); + for snapfile in snapfiles.0.values() { + if !snapfile.is_frozen() { + snapfile.freeze(last_valid_lsn + 1)?; + } + } + + // Also save the metadata, with updated last_valid_lsn and last_record_lsn, to a + // file in the timeline dir + let ancestor_timelineid = if let Some(x) = &self.ancestor_timeline { + Some(x.timelineid) + } else { + None + }; + + let metadata = TimelineMetadata { + last_valid_lsn: self.last_valid_lsn.load(), + last_record_lsn: self.last_record_lsn.load(), + ancestor_timeline: ancestor_timelineid, + ancestor_lsn: self.ancestor_lsn, + }; + LayeredRepository::save_metadata(self.conf, self.timelineid, &metadata)?; + + // If there were any concurrent updates on the timeline, we would have to work + // harder to make sure we don't lose the new updates. Currently, that shouldn't + // happen, because the WAL receiver process is responsible for both updating + // the timeline and calling checkpoint() + assert!(self.last_valid_lsn.load() == last_valid_lsn); + + Ok(()) + } + + /// Remember that WAL has been received and added to the page cache up to the given LSN + fn advance_last_valid_lsn(&self, lsn: Lsn) { + let old = self.last_valid_lsn.advance(lsn); + + // Can't move backwards. + if lsn < old { + warn!( + "attempted to move last valid LSN backwards (was {}, new {})", + old, lsn + ); + } + } + + fn init_valid_lsn(&self, lsn: Lsn) { + let old = self.last_valid_lsn.advance(lsn); + assert!(old == Lsn(0)); + let old = self.last_record_lsn.fetch_max(lsn); + assert!(old == Lsn(0)); + } + + fn get_last_valid_lsn(&self) -> Lsn { + self.last_valid_lsn.load() + } + + /// + /// Remember the (end of) last valid WAL record remembered in the page cache. + /// + /// NOTE: this updates last_valid_lsn as well. + /// + fn advance_last_record_lsn(&self, lsn: Lsn) { + // Can't move backwards. + let old = self.last_record_lsn.fetch_max(lsn); + assert!(old <= lsn); + + // Also advance last_valid_lsn + let old = self.last_valid_lsn.advance(lsn); + // Can't move backwards. + if lsn < old { + warn!( + "attempted to move last record LSN backwards (was {}, new {})", + old, lsn + ); + } + } + + fn get_last_record_lsn(&self) -> Lsn { + self.last_record_lsn.load() + } +} + +impl LayeredTimeline { + /// Open a Timeline handle. + /// + /// Loads the metadata for the timeline into memory. + fn new( + conf: &'static PageServerConf, + metadata: TimelineMetadata, + ancestor: Option>, + timelineid: ZTimelineId, + walredo_mgr: Arc, + ) -> Result { + let timeline = LayeredTimeline { + conf, + timelineid, + layers: Mutex::new(LayerMap::default()), + + walredo_mgr, + + last_valid_lsn: SeqWait::new(metadata.last_valid_lsn), + last_record_lsn: AtomicLsn::new(metadata.last_record_lsn.0), + + ancestor_timeline: ancestor, + ancestor_lsn: metadata.ancestor_lsn, + }; + Ok(timeline) + } + + /// + /// Get a handle to a SnapshotFile for reading. + /// + /// The returned SnapshotFile might be from an ancestor timeline, if the + /// relation hasn't been updated on this timeline yet. + /// + fn get_snapshot_file_for_read( + &self, + tag: RelTag, + lsn: Lsn, + ) -> Result, Lsn)>> { + // First dig the right ancestor timeline + let mut timeline = self; + let mut lsn = lsn; + trace!( + "get_snapshot_file_for_read called for {} at {}/{}", + tag, + self.timelineid, + lsn + ); + + // If you requested a page at an older LSN, before the branch point, dig into + // the right ancestor timeline. This can only happen if you launch a read-only + // node with an old LSN. A primary always uses a recent LSN in its requests. + while lsn < timeline.ancestor_lsn { + trace!("going into ancestor {} ", timeline.ancestor_lsn); + timeline = &timeline.ancestor_timeline.as_ref().unwrap(); + } + + loop { + // Then look up the snapshot file + let mut layers = timeline.layers.lock().unwrap(); + + // FIXME: If there is an entry in memory for an older snapshot file, + // but there is a newere snapshot file on disk, this will incorrectly + // return the older entry from memory. + if let Some(layer) = layers.get(tag, lsn) { + trace!("found snapshot file in memory: {}", layer.get_start_lsn()); + return Ok(Some((layer.clone(), lsn))); + } else { + // No layer in memory for this relation yet. Read it from disk. + if let Some(layer) = + SnapshotLayer::load(timeline.conf, timeline.timelineid, tag, lsn)? + { + trace!( + "found snapshot file on disk: {}-{}", + layer.get_start_lsn(), + layer.get_end_lsn() + ); + let layer_rc: Arc = Arc::new(layer); + layers.insert(Arc::clone(&layer_rc)); + + return Ok(Some((layer_rc, lsn))); + } else { + // No snapshot files for this relation on this timeline. But there might still + // be one on the ancestor timeline + if let Some(ancestor) = &timeline.ancestor_timeline { + lsn = timeline.ancestor_lsn; + timeline = &ancestor.as_ref(); + trace!("recursing into ancestor at {}/{}", timeline.timelineid, lsn); + continue; + } + return Ok(None); + } + } + } + } + + /// + /// Get a handle to the latest SnapshotFile for appending. + /// + fn get_snapshot_file_for_write(&self, tag: RelTag, lsn: Lsn) -> Result> { + if lsn < self.last_valid_lsn.load() { + bail!("cannot modify relation after advancing last_valid_lsn"); + } + + // Look up the snapshot file + let layers = self.layers.lock().unwrap(); + if let Some(layer) = layers.get(tag, lsn) { + if !layer.is_frozen() { + return Ok(Arc::clone(&layer)); + } + } + + // No SnapshotFile for this relation yet. Create one. + // + // Is this a completely new relation? Or the first modification after branching? + // + + // FIXME: race condition, if another thread creates the SnapshotFile while + // we're busy looking up the previous one. We should hold the mutex throughout + // this operation, but for that we'll need a versio of get_snapshot_file_for_read() + // that doesn't try to also grab the mutex. + drop(layers); + + let layer; + if let Some((prev_snapfile, _prev_lsn)) = self.get_snapshot_file_for_read(tag, lsn)? { + // Create new entry after the previous one. + let lsn; + if prev_snapfile.get_timeline_id() != self.timelineid { + // First modification on this timeline + lsn = self.ancestor_lsn; + trace!( + "creating file for write for {} at branch point {}/{}", + tag, + self.timelineid, + lsn + ); + } else { + lsn = prev_snapfile.get_end_lsn(); + trace!( + "creating file for write for {} after previous snapfile {}/{}", + tag, + self.timelineid, + lsn + ); + } + trace!( + "prev snapfile is at {}/{} - {}", + prev_snapfile.get_timeline_id(), + prev_snapfile.get_start_lsn(), + prev_snapfile.get_end_lsn() + ); + layer = InMemoryLayer::copy_snapshot( + self.conf, + &*self.walredo_mgr, + &*prev_snapfile, + self.timelineid, + lsn, + )?; + } else { + // New relation. + trace!( + "creating file for write for new rel {} at {}/{}", + tag, + self.timelineid, + lsn + ); + + // Scan the directory for latest existing file. + // FIXME: if this is truly a new rel, none should exist right? + let start_lsn; + if let Some((_start, end)) = SnapshotLayer::find_latest_snapshot_file( + self.conf, + self.timelineid, + tag, + Lsn(u64::MAX), + )? { + start_lsn = end; + } else { + start_lsn = lsn; + } + layer = InMemoryLayer::create(self.conf, self.timelineid, tag, start_lsn)?; + } + + let mut layers = self.layers.lock().unwrap(); + let layer_rc: Arc = Arc::new(layer); + layers.insert(Arc::clone(&layer_rc)); + + Ok(layer_rc) + } + + /// + /// Wait until WAL has been received up to the given LSN. + /// + fn wait_lsn(&self, mut lsn: Lsn) -> anyhow::Result { + // When invalid LSN is requested, it means "don't wait, return latest version of the page" + // This is necessary for bootstrap. + if lsn == Lsn(0) { + let last_valid_lsn = self.last_valid_lsn.load(); + trace!( + "walreceiver doesn't work yet last_valid_lsn {}, requested {}", + last_valid_lsn, + lsn + ); + lsn = last_valid_lsn; + } + + self.last_valid_lsn + .wait_for_timeout(lsn, TIMEOUT) + .with_context(|| { + format!( + "Timed out while waiting for WAL record at LSN {} to arrive", + lsn + ) + })?; + + Ok(lsn) + } +} diff --git a/pageserver/src/layered_repository/inmemory_layer.rs b/pageserver/src/layered_repository/inmemory_layer.rs new file mode 100644 index 0000000000..21693a7585 --- /dev/null +++ b/pageserver/src/layered_repository/inmemory_layer.rs @@ -0,0 +1,371 @@ +//! +//! 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::storage_layer::Layer; +use crate::layered_repository::storage_layer::PageVersion; +use crate::layered_repository::SnapshotLayer; +use crate::repository::{RelTag, WALRecord}; +use crate::walredo::WalRedoManager; +use crate::PageServerConf; +use crate::ZTimelineId; +use anyhow::{bail, Result}; +use bytes::Bytes; +use log::*; +use std::collections::BTreeMap; +use std::ops::Bound::Included; +use std::sync::Mutex; + +use zenith_utils::lsn::Lsn; + +static ZERO_PAGE: Bytes = Bytes::from_static(&[0u8; 8192]); + +pub struct InMemoryLayer { + conf: &'static PageServerConf, + timelineid: ZTimelineId, + tag: RelTag, + + /// + /// This layer contains all the changes from 'start_lsn'. The + /// start is inclusive. There is no end LSN; we only use in-memory + /// layer at the end of a timeline. + /// + start_lsn: Lsn, + + /// + /// All versions of all pages in the layer are are kept here. + /// Indexed by block number and LSN. + /// + page_versions: Mutex>, + + /// + /// `relsizes` tracks the size of the relation at different points in time. + /// + relsizes: Mutex>, +} + +impl Layer for InMemoryLayer { + fn is_frozen(&self) -> bool { + return false; + } + + fn get_timeline_id(&self) -> ZTimelineId { + return self.timelineid; + } + + fn get_tag(&self) -> RelTag { + return self.tag; + } + + fn get_start_lsn(&self) -> Lsn { + return self.start_lsn; + } + + fn get_end_lsn(&self) -> Lsn { + return Lsn(u64::MAX); + } + + /// Look up given page in the cache. + fn get_page_at_lsn( + &self, + walredo_mgr: &dyn WalRedoManager, + blknum: u32, + lsn: Lsn, + ) -> Result { + // Scan the BTreeMap backwards, starting from the given entry. + let mut records: Vec = Vec::new(); + let mut page_img: Option = None; + let mut need_base_image_lsn: Option = Some(lsn); + { + let page_versions = self.page_versions.lock().unwrap(); + let minkey = (blknum, Lsn(0)); + let maxkey = (blknum, lsn); + let mut iter = page_versions.range((Included(&minkey), Included(&maxkey))); + while let Some(((_blknum, entry_lsn), entry)) = iter.next_back() { + if let Some(img) = &entry.page_image { + page_img = Some(img.clone()); + need_base_image_lsn = None; + break; + } else if let Some(rec) = &entry.record { + records.push(rec.clone()); + if rec.will_init { + // This WAL record initializes the page, so no need to go further back + need_base_image_lsn = None; + break; + } else { + need_base_image_lsn = Some(*entry_lsn); + } + } else { + // No base image, and no WAL record. Huh? + bail!("no page image or WAL record for requested page"); + } + } + + // release lock on 'page_versions' + } + records.reverse(); + + // If we needed a base image to apply the WAL records against, we should have found it in memory. + if let Some(lsn) = need_base_image_lsn { + if records.is_empty() { + // no records, and no base image. This can happen if PostgreSQL extends a relation + // but never writes the page. + // + // Would be nice to detect that situation better. + warn!("Page {:?}/{} at {} not found", self.tag, blknum, lsn); + return Ok(ZERO_PAGE.clone()); + } + bail!( + "No base image found for page {} blk {} at {}/{}", + self.tag, + blknum, + self.timelineid, + lsn + ); + } + + // If we have a page image, and no WAL, we're all set + if records.is_empty() { + if let Some(img) = page_img { + trace!( + "found page image for blk {} in {} at {}/{}, no WAL redo required", + blknum, + self.tag, + self.timelineid, + lsn + ); + Ok(img) + } else { + // FIXME: this ought to be an error? + warn!("Page {:?}/{} at {} not found", self.tag, blknum, lsn); + Ok(ZERO_PAGE.clone()) + } + } else { + // We need to do WAL redo. + // + // If we don't have a base image, then the oldest WAL record better initialize + // the page + if page_img.is_none() && !records.first().unwrap().will_init { + // FIXME: this ought to be an error? + warn!( + "Base image for page {:?}/{} at {} not found, but got {} WAL records", + self.tag, + blknum, + lsn, + records.len() + ); + Ok(ZERO_PAGE.clone()) + } else { + if page_img.is_some() { + trace!("found {} WAL records and a base image for blk {} in {} at {}/{}, performing WAL redo", records.len(), blknum, self.tag, self.timelineid, lsn); + } else { + trace!("found {} WAL records that will init the page for blk {} in {} at {}/{}, performing WAL redo", records.len(), blknum, self.tag, self.timelineid, lsn); + } + let img = walredo_mgr.request_redo( + self.rel, + blknum, + lsn, + page_img, + records, + )?; + + self.put_page_image(blknum, lsn, img.clone())?; + + Ok(img) + } + } + } + + /// Get size of the relation at given LSN + fn get_rel_size(&self, lsn: Lsn) -> Result { + // Scan the BTreeMap backwards, starting from the given entry. + let relsizes = self.relsizes.lock().unwrap(); + let mut iter = relsizes.range((Included(&Lsn(0)), Included(&lsn))); + + if let Some((_entry_lsn, entry)) = iter.next_back() { + trace!("get_relsize: {} at {} -> {}", self.tag, lsn, *entry); + Ok(*entry) + } else { + bail!( + "No size found for relfile {:?} at {} in memory", + self.tag, + lsn + ); + } + } + + /// Does this relation exist at given LSN? + fn get_rel_exists(&self, lsn: Lsn) -> Result { + // Scan the BTreeMap backwards, starting from the given entry. + let relsizes = self.relsizes.lock().unwrap(); + + let mut iter = relsizes.range((Included(&Lsn(0)), Included(&lsn))); + + let result = if let Some((_entry_lsn, _entry)) = iter.next_back() { + true + } else { + false + }; + Ok(result) + } + + // Write operations + + /// Common subroutine of the public put_wal_record() and put_page_image() functions. + /// Adds the page version to the in-memory tree + fn put_page_version(&self, blknum: u32, lsn: Lsn, pv: PageVersion) -> Result<()> { + trace!( + "put_page_version blk {} of {} at {}/{}", + blknum, + self.tag, + self.timelineid, + lsn + ); + { + let mut page_versions = self.page_versions.lock().unwrap(); + let old = page_versions.insert((blknum, lsn), pv); + + if old.is_some() { + // We already had an entry for this LSN. That's odd.. + warn!( + "Page version of rel {:?} blk {} at {} already exists", + self.tag, blknum, lsn + ); + } + + // release lock on 'page_versions' + } + + // Also update the relation size, if this extended the relation. + { + let mut relsizes = self.relsizes.lock().unwrap(); + let mut iter = relsizes.range((Included(&Lsn(0)), Included(&lsn))); + + let oldsize; + if let Some((_entry_lsn, entry)) = iter.next_back() { + oldsize = *entry; + } else { + oldsize = 0; + //bail!("No old size found for {} at {}", self.tag, lsn); + } + if blknum >= oldsize { + trace!( + "enlarging relation {} from {} to {} blocks", + self.tag, + oldsize, + blknum + 1 + ); + relsizes.insert(lsn, blknum + 1); + } + } + + Ok(()) + } + + /// Remember that the relation was truncated at given LSN + fn put_truncation(&self, lsn: Lsn, relsize: u32) -> anyhow::Result<()> { + let mut relsizes = self.relsizes.lock().unwrap(); + let old = relsizes.insert(lsn, relsize); + + if old.is_some() { + // We already had an entry for this LSN. That's odd.. + warn!("Inserting truncation, but had an entry for the LSN already"); + } + + Ok(()) + } + + /// + /// Write the this in-memory layer to disk, as a snapshot layer. + /// + fn freeze(&self, end_lsn: Lsn) -> Result<()> { + let page_versions = self.page_versions.lock().unwrap(); + let relsizes = self.relsizes.lock().unwrap(); + + // FIXME: we assume there are no modification in-flight, and that there are no + // changes past 'lsn'. + + let page_versions = page_versions.clone(); + let relsizes = relsizes.clone(); + + let _snapfile = SnapshotLayer::create( + self.conf, + self.timelineid, + self.tag, + self.start_lsn, + end_lsn, + page_versions, + relsizes, + )?; + + Ok(()) + } +} + +impl InMemoryLayer { + /// + /// Create a new, empty, in-memory layer + /// + pub fn create( + conf: &'static PageServerConf, + timelineid: ZTimelineId, + tag: RelTag, + start_lsn: Lsn, + ) -> Result { + debug!( + "initializing new InMemoryLayer for writing {} on timeline {}", + tag, timelineid + ); + + Ok(InMemoryLayer { + conf, + timelineid, + tag, + start_lsn, + page_versions: Mutex::new(BTreeMap::new()), + relsizes: Mutex::new(BTreeMap::new()), + }) + } + + /// + /// Initialize a new InMemoryLayer for, by copying the state at the given + /// point in time from given existing layer. + /// + pub fn copy_snapshot( + conf: &'static PageServerConf, + walredo_mgr: &dyn WalRedoManager, + src: &dyn Layer, + timelineid: ZTimelineId, + lsn: Lsn, + ) -> Result { + debug!( + "initializing new InMemoryLayer for writing {} on timeline {}", + src.get_tag(), + timelineid + ); + let mut page_versions = BTreeMap::new(); + let mut relsizes = BTreeMap::new(); + + let size = src.get_rel_size(lsn)?; + relsizes.insert(lsn, size); + + for blknum in 0..size { + let img = src.get_page_at_lsn(walredo_mgr, blknum, lsn)?; + let pv = PageVersion { + page_image: Some(img), + record: None, + }; + page_versions.insert((blknum, lsn), pv); + } + + Ok(InMemoryLayer { + conf, + timelineid, + tag: src.get_tag(), + start_lsn: lsn, + page_versions: Mutex::new(page_versions), + relsizes: Mutex::new(relsizes), + }) + } +} diff --git a/pageserver/src/layered_repository/snapshot_layer.rs b/pageserver/src/layered_repository/snapshot_layer.rs new file mode 100644 index 0000000000..6a80da61b0 --- /dev/null +++ b/pageserver/src/layered_repository/snapshot_layer.rs @@ -0,0 +1,486 @@ +//! +//! A SnapshotLayer represents one snapshot file on disk. One file holds all page versions +//! and size information of one relation, in a range of LSN. +//! The name "snapshot file" is a bit of a misnomer because a snapshot file doesn't +//! contain a snapshot at a specific LSN, but rather all the page versions in a range +//! of LSNs. +//! +//! Currently, a snapshot file contains full information needed to reconstruct any +//! page version in the LSN range, without consulting any other snapshot files. When +//! a new snapshot file is created for writing, the full contents of relation are +//! materialized as it is at the beginning of the LSN range. That can be very expensive, +//! we should find a way to store differential files. But this keeps the read-side +//! of things simple. You can find the correct snapshot file based on RelTag and +//! timeline+LSN, and once you've located it, you have all the data you need to in that +//! file. +//! +//! When a snapshot file needs to be accessed, we slurp the whole file into memory, into +//! a SnapshotLayer struct. +//! +//! On disk, a snapshot file is actually two files: one containing all the page versions, +//! and another containing the relation size information. That's just for the convenience +//! of serializing the two objects. +//! +//! The files are stored in .zenith/timelines/ directory. +//! Currently, there are no subdirectories, and each snapshot file is named like this: +//! +//! _____ +//! +//! And the corresponding file containing the relation size information has _relsizes +//! suffix. For example: +//! +//! 1663_13990_2609_0_000000000169C348_000000000169C349 +//! 1663_13990_2609_0_000000000169C348_000000000169C349_relsizes +//! + +use crate::layered_repository::storage_layer::Layer; +use crate::layered_repository::storage_layer::PageVersion; +use crate::repository::{RelTag, WALRecord}; +use crate::walredo::WalRedoManager; +use crate::PageServerConf; +use crate::ZTimelineId; +use anyhow::{bail, Result}; +use bytes::Bytes; +use log::*; +use std::collections::{BTreeMap, HashSet}; +use std::fs; +use std::fs::File; +use std::io::Write; +use std::ops::Bound::Included; +use std::path::{Path, PathBuf}; +use std::sync::Mutex; + +use zenith_utils::bin_ser::BeSer; +use zenith_utils::lsn::Lsn; + +static ZERO_PAGE: Bytes = Bytes::from_static(&[0u8; 8192]); + +/// +/// SnapshotLayer is the in-memory data structure associated with an on-disk snapshot file. +/// It is also used to accumulate new changes at the tip of a branch; end_lsn is u64::MAX +/// in that case. +/// +pub struct SnapshotLayer { + conf: &'static PageServerConf, + pub timelineid: ZTimelineId, + pub tag: RelTag, + + // + // This entry contains all the changes from 'start_lsn' to 'end_lsn'. The + // start is inclusive, and end is exclusive. + pub start_lsn: Lsn, + pub end_lsn: Lsn, + + /// + /// All versions of all pages in the file are are kept here. + /// Indexed by block number and LSN. + /// + page_versions: Mutex>, + + /// + /// `relsizes` tracks the size of the relation at different points in time. + /// + relsizes: Mutex>, +} + +impl Layer for SnapshotLayer { + fn is_frozen(&self) -> bool { + return true; + } + + fn get_timeline_id(&self) -> ZTimelineId { + return self.timelineid; + } + + fn get_tag(&self) -> RelTag { + return self.tag; + } + + fn get_start_lsn(&self) -> Lsn { + return self.start_lsn; + } + + fn get_end_lsn(&self) -> Lsn { + return self.end_lsn; + } + + /// Look up given page in the cache. + fn get_page_at_lsn( + &self, + walredo_mgr: &dyn WalRedoManager, + blknum: u32, + lsn: Lsn, + ) -> Result { + // Scan the BTreeMap backwards, starting from the given entry. + let mut records: Vec = Vec::new(); + let mut page_img: Option = None; + let mut need_base_image_lsn: Option = Some(lsn); + { + let page_versions = self.page_versions.lock().unwrap(); + let minkey = (blknum, Lsn(0)); + let maxkey = (blknum, lsn); + let mut iter = page_versions.range((Included(&minkey), Included(&maxkey))); + while let Some(((_blknum, entry_lsn), entry)) = iter.next_back() { + if let Some(img) = &entry.page_image { + page_img = Some(img.clone()); + need_base_image_lsn = None; + break; + } else if let Some(rec) = &entry.record { + records.push(rec.clone()); + if rec.will_init { + // This WAL record initializes the page, so no need to go further back + need_base_image_lsn = None; + break; + } else { + need_base_image_lsn = Some(*entry_lsn); + } + } else { + // No base image, and no WAL record. Huh? + bail!("no page image or WAL record for requested page"); + } + } + + // release lock on 'page_versions' + } + records.reverse(); + + // If we needed a base image to apply the WAL records against, we should have found it in memory. + if let Some(lsn) = need_base_image_lsn { + if records.is_empty() { + // no records, and no base image. This can happen if PostgreSQL extends a relation + // but never writes the page. + // + // Would be nice to detect that situation better. + warn!("Page {:?}/{} at {} not found", self.tag, blknum, lsn); + return Ok(ZERO_PAGE.clone()); + } + bail!( + "No base image found for page {} blk {} at {}/{}", + self.tag, + blknum, + self.timelineid, + lsn + ); + } + + // If we have a page image, and no WAL, we're all set + if records.is_empty() { + if let Some(img) = page_img { + trace!( + "found page image for blk {} in {} at {}/{}, no WAL redo required", + blknum, + self.tag, + self.timelineid, + lsn + ); + Ok(img) + } else { + // FIXME: this ought to be an error? + warn!("Page {:?}/{} at {} not found", self.tag, blknum, lsn); + Ok(ZERO_PAGE.clone()) + } + } else { + // We need to do WAL redo. + // + // If we don't have a base image, then the oldest WAL record better initialize + // the page + if page_img.is_none() && !records.first().unwrap().will_init { + // FIXME: this ought to be an error? + warn!( + "Base image for page {:?}/{} at {} not found, but got {} WAL records", + self.tag, + blknum, + lsn, + records.len() + ); + Ok(ZERO_PAGE.clone()) + } else { + if page_img.is_some() { + trace!("found {} WAL records and a base image for blk {} in {} at {}/{}, performing WAL redo", records.len(), blknum, self.tag, self.timelineid, lsn); + } else { + trace!("found {} WAL records that will init the page for blk {} in {} at {}/{}, performing WAL redo", records.len(), blknum, self.tag, self.timelineid, lsn); + } + let img = walredo_mgr.request_redo( + self.tag, + blknum, + lsn, + page_img, + records, + )?; + + // FIXME: Should we memoize the page image in memory, so that + // we wouldn't need to reconstruct it again, if it's requested again? + //self.put_page_image(blknum, lsn, img.clone())?; + + Ok(img) + } + } + } + + /// Get size of the relation at given LSN + fn get_rel_size(&self, lsn: Lsn) -> Result { + // Scan the BTreeMap backwards, starting from the given entry. + let relsizes = self.relsizes.lock().unwrap(); + let mut iter = relsizes.range((Included(&Lsn(0)), Included(&lsn))); + + if let Some((_entry_lsn, entry)) = iter.next_back() { + trace!("get_relsize: {} at {} -> {}", self.tag, lsn, *entry); + Ok(*entry) + } else { + bail!( + "No size found for relfile {:?} at {} in memory", + self.tag, + lsn + ); + } + } + + /// Does this relation exist at given LSN? + fn get_rel_exists(&self, lsn: Lsn) -> Result { + // Scan the BTreeMap backwards, starting from the given entry. + let relsizes = self.relsizes.lock().unwrap(); + + let mut iter = relsizes.range((Included(&Lsn(0)), Included(&lsn))); + + let result = if let Some((_entry_lsn, _entry)) = iter.next_back() { + true + } else { + false + }; + Ok(result) + } + + // Unsupported write operations + fn put_page_version(&self, blknum: u32, lsn: Lsn, _pv: PageVersion) -> Result<()> { + panic!( + "cannot modify historical snapshot file, rel {} blk {} at {}/{}, {}-{}", + self.tag, blknum, self.timelineid, lsn, self.start_lsn, self.end_lsn + ); + } + fn put_truncation(&self, _lsn: Lsn, _relsize: u32) -> anyhow::Result<()> { + bail!("cannot modify historical snapshot file"); + } + + fn freeze(&self, _end_lsn: Lsn) -> Result<()> { + bail!("cannot freeze historical snapshot file"); + } +} + +impl SnapshotLayer { + fn path(&self) -> PathBuf { + Self::path_for( + self.conf, + self.timelineid, + self.tag, + self.start_lsn, + self.end_lsn, + ) + } + + fn path_for( + conf: &'static PageServerConf, + timelineid: ZTimelineId, + tag: RelTag, + start_lsn: Lsn, + end_lsn: Lsn, + ) -> PathBuf { + let fname = format!( + "{}_{}_{}_{}_{:016X}_{:016X}", + tag.spcnode, + tag.dbnode, + tag.relnode, + tag.forknum, + u64::from(start_lsn), + u64::from(end_lsn) + ); + + conf.timeline_path(timelineid).join(&fname) + } + + fn relsizes_path(path: &Path) -> PathBuf { + let mut fname = path.file_name().unwrap().to_os_string(); + fname.push("_relsizes"); + + path.with_file_name(fname) + } + + /// Create a new snapshot file, using the given btreemaps containing the page versions and + /// relsizes. + /// + /// 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. + pub fn create( + conf: &'static PageServerConf, + timelineid: ZTimelineId, + tag: RelTag, + start_lsn: Lsn, + end_lsn: Lsn, + page_versions: BTreeMap<(u32, Lsn), PageVersion>, + relsizes: BTreeMap, + ) -> Result { + let snapfile = SnapshotLayer { + conf: conf, + timelineid: timelineid, + tag: tag, + start_lsn: start_lsn, + end_lsn, + page_versions: Mutex::new(page_versions), + relsizes: Mutex::new(relsizes), + }; + + snapfile.save()?; + Ok(snapfile) + } + + /// Write the in-memory btreemaps into files + fn save(&self) -> Result<()> { + let path = self.path(); + + let page_versions = self.page_versions.lock().unwrap(); + let relsizes = self.relsizes.lock().unwrap(); + + // Note: This overwrites any existing file. There shouldn't be any. + // FIXME: throw an error instead? + + // Write out page versions + let mut file = File::create(&path)?; + let buf = BTreeMap::ser(&page_versions)?; + file.write_all(&buf)?; + + // and relsizes to separate file + let mut file = File::create(Self::relsizes_path(&path))?; + let buf = BTreeMap::ser(&relsizes)?; + file.write_all(&buf)?; + + debug!("saved {}", &path.display()); + + Ok(()) + } + + /// + /// Find the snapshot file with latest LSN that covers the given 'lsn', or is before it. + /// + pub fn find_latest_snapshot_file( + conf: &'static PageServerConf, + timelineid: ZTimelineId, + tag: RelTag, + lsn: Lsn, + ) -> Result> { + // Scan the timeline directory to get all rels in this timeline. + let path = conf.timeline_path(timelineid); + let mut result_start_lsn = Lsn(0); + let mut result_end_lsn = Lsn(0); + for direntry in fs::read_dir(path)? { + let direntry = direntry?; + + let fname = direntry.file_name(); + let fname = fname.to_str().unwrap(); + + if let Some((reltag, start_lsn, end_lsn)) = Self::fname_to_tag(fname) { + if reltag == tag && start_lsn <= lsn && start_lsn > result_start_lsn { + result_start_lsn = start_lsn; + result_end_lsn = end_lsn; + } + } + } + if result_start_lsn != Lsn(0) { + Ok(Some((result_start_lsn, result_end_lsn))) + } else { + Ok(None) + } + } + + /// + /// Load the state for one relation back into memory. + /// + /// Returns the latest snapshot file that before the given 'lsn'. + /// + pub fn load( + conf: &'static PageServerConf, + timelineid: ZTimelineId, + tag: RelTag, + lsn: Lsn, + ) -> Result> { + if let Some((start_lsn, end_lsn)) = + Self::find_latest_snapshot_file(conf, timelineid, tag, lsn)? + { + let snap = Self::load_path(conf, timelineid, tag, start_lsn, end_lsn)?; + Ok(Some(snap)) + } else { + Ok(None) + } + } + + fn load_path( + conf: &'static PageServerConf, + timelineid: ZTimelineId, + tag: RelTag, + start_lsn: Lsn, + end_lsn: Lsn, + ) -> Result { + let path = Self::path_for(conf, timelineid, tag, start_lsn, end_lsn); + + let content = std::fs::read(&path)?; + let page_versions = BTreeMap::des(&content)?; + debug!("loaded from {}", &path.display()); + + let content = std::fs::read(Self::relsizes_path(&path))?; + let relsizes = BTreeMap::des(&content)?; + Ok(SnapshotLayer { + conf, + timelineid, + tag, + start_lsn, + end_lsn, + page_versions: Mutex::new(page_versions), + relsizes: Mutex::new(relsizes), + }) + } + + pub fn list_rels( + conf: &'static PageServerConf, + timelineid: ZTimelineId, + spcnode: u32, + dbnode: u32, + ) -> Result> { + let mut rels: HashSet = HashSet::new(); + + // Scan the timeline directory to get all rels in this timeline. + let path = conf.timeline_path(timelineid); + for direntry in fs::read_dir(path)? { + let direntry = direntry?; + + let fname = direntry.file_name(); + let fname = fname.to_str().unwrap(); + + if let Some((reltag, _start_lsn, _end_lsn)) = Self::fname_to_tag(fname) { + if (spcnode == 0 || reltag.spcnode == spcnode) + && (dbnode == 0 || reltag.dbnode == dbnode) + { + rels.insert(reltag); + } + } + } + Ok(rels) + } + + fn fname_to_tag(fname: &str) -> Option<(RelTag, Lsn, Lsn)> { + // Split the filename into parts + // + // _____ + // + let mut parts = fname.split('_'); + + let reltag = RelTag { + spcnode: parts.next()?.parse::().ok()?, + dbnode: parts.next()?.parse::().ok()?, + relnode: parts.next()?.parse::().ok()?, + forknum: parts.next()?.parse::().ok()?, + }; + let start_lsn = Lsn::from_hex(parts.next()?).ok()?; + let end_lsn = Lsn::from_hex(parts.next()?).ok()?; + + Some((reltag, start_lsn, end_lsn)) + } +} diff --git a/pageserver/src/layered_repository/storage_layer.rs b/pageserver/src/layered_repository/storage_layer.rs new file mode 100644 index 0000000000..aab2c0652a --- /dev/null +++ b/pageserver/src/layered_repository/storage_layer.rs @@ -0,0 +1,75 @@ +use crate::repository::RelTag; +use crate::repository::WALRecord; +use crate::walredo::WalRedoManager; +use crate::ZTimelineId; +use anyhow::Result; +use bytes::Bytes; +use serde::{Deserialize, Serialize}; + +use zenith_utils::lsn::Lsn; + +/// +/// Represents a version of a page at a specific LSN. The LSN is the key of the +/// entry in the 'page_versions' hash, it is not duplicated here. +/// +/// A page version can be stored as a full page image, or as WAL record that needs +/// to be applied over the previous page version to reconstruct this version. +/// +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct PageVersion { + /// an 8kb page image + pub page_image: Option, + /// WAL record to get from previous page version to this one. + pub record: Option, +} + +pub trait Layer: Send + Sync { + fn is_frozen(&self) -> bool; + + fn get_timeline_id(&self) -> ZTimelineId; + fn get_tag(&self) -> RelTag; + fn get_start_lsn(&self) -> Lsn; + fn get_end_lsn(&self) -> Lsn; + + fn get_page_at_lsn( + &self, + walredo_mgr: &dyn WalRedoManager, + blknum: u32, + lsn: Lsn, + ) -> Result; + + fn get_rel_size(&self, lsn: Lsn) -> Result; + + fn get_rel_exists(&self, lsn: Lsn) -> Result; + + fn put_page_version(&self, blknum: u32, lsn: Lsn, pv: PageVersion) -> Result<()>; + + fn put_truncation(&self, lsn: Lsn, relsize: u32) -> anyhow::Result<()>; + + /// Remember new page version, as a WAL record over previous version + fn put_wal_record(&self, blknum: u32, rec: WALRecord) -> Result<()> { + // FIXME: If this is the first version of this page, reconstruct the image + self.put_page_version( + blknum, + rec.lsn, + PageVersion { + page_image: None, + record: Some(rec), + }, + ) + } + + /// Remember new page version, as a full page image + fn put_page_image(&self, blknum: u32, lsn: Lsn, img: Bytes) -> Result<()> { + self.put_page_version( + blknum, + lsn, + PageVersion { + page_image: Some(img), + record: None, + }, + ) + } + + fn freeze(&self, end_lsn: Lsn) -> Result<()>; +} diff --git a/pageserver/src/lib.rs b/pageserver/src/lib.rs index 0f8be62b4b..7dc0d1926b 100644 --- a/pageserver/src/lib.rs +++ b/pageserver/src/lib.rs @@ -9,6 +9,7 @@ use std::time::Duration; pub mod basebackup; pub mod branches; +pub mod layered_repository; pub mod logger; pub mod object_key; pub mod object_repository; @@ -40,6 +41,14 @@ pub struct PageServerConf { pub workdir: PathBuf, pub pg_distrib_dir: PathBuf, + + pub repository_format: RepositoryFormat, +} + +#[derive(Debug, Clone, PartialEq)] +pub enum RepositoryFormat { + Layered, + RocksDb, } impl PageServerConf { diff --git a/pageserver/src/page_cache.rs b/pageserver/src/page_cache.rs index 1467afeefc..84c0b40b52 100644 --- a/pageserver/src/page_cache.rs +++ b/pageserver/src/page_cache.rs @@ -2,11 +2,12 @@ //! page server. use crate::branches; +use crate::layered_repository::LayeredRepository; use crate::object_repository::ObjectRepository; use crate::repository::Repository; use crate::rocksdb_storage::RocksObjectStore; use crate::walredo::PostgresRedoManager; -use crate::{PageServerConf, ZTenantId}; +use crate::{PageServerConf, RepositoryFormat, ZTenantId}; use anyhow::{anyhow, bail, Result}; use lazy_static::lazy_static; use log::info; @@ -23,6 +24,7 @@ lazy_static! { pub fn init(conf: &'static PageServerConf) { let mut m = REPOSITORY.lock().unwrap(); + for dir_entry in fs::read_dir(conf.tenants_path()).unwrap() { let tenantid = ZTenantId::from_str(dir_entry.unwrap().file_name().to_str().unwrap()).unwrap(); @@ -32,8 +34,20 @@ pub fn init(conf: &'static PageServerConf) { let walredo_mgr = PostgresRedoManager::new(conf, tenantid); // Set up an object repository, for actual data storage. - let repo = - ObjectRepository::new(conf, Arc::new(obj_store), Arc::new(walredo_mgr), tenantid); + let repo: Arc = match conf.repository_format { + RepositoryFormat::Layered => Arc::new(LayeredRepository::new(conf, Arc::new(walredo_mgr))), + RepositoryFormat::RocksDb => { + let obj_store = RocksObjectStore::open(conf).unwrap(); + + Arc::new(ObjectRepository::new( + conf, + Arc::new(obj_store), + Arc::new(walredo_mgr), + tenantid + )) + } + }; + info!("initialized storage for tenant: {}", &tenantid); m.insert(tenantid, Arc::new(repo)); } diff --git a/pageserver/src/repository.rs b/pageserver/src/repository.rs index 80dd2d5d8a..6b684c6b18 100644 --- a/pageserver/src/repository.rs +++ b/pageserver/src/repository.rs @@ -249,11 +249,12 @@ impl WALRecord { #[cfg(test)] mod tests { use super::*; + use crate::layered_repository::LayeredRepository; use crate::object_repository::ObjectRepository; use crate::object_repository::{ObjectValue, PageEntry, RelationSizeEntry}; use crate::rocksdb_storage::RocksObjectStore; use crate::walredo::{WalRedoError, WalRedoManager}; - use crate::{PageServerConf, ZTenantId}; + use crate::{PageServerConf, RepositoryFormat, ZTenantId}; use postgres_ffi::pg_constants; use std::fs; use std::path::PathBuf; @@ -285,10 +286,14 @@ mod tests { buf.freeze() } - fn get_test_repo(test_name: &str) -> Result> { + fn get_test_repo( + test_name: &str, + repository_format: RepositoryFormat, + ) -> Result> { let repo_dir = PathBuf::from(format!("../tmp_check/test_{}", test_name)); let _ = fs::remove_dir_all(&repo_dir); - fs::create_dir_all(&repo_dir).unwrap(); + fs::create_dir_all(&repo_dir)?; + fs::create_dir_all(&repo_dir.join("timelines"))?; let conf = PageServerConf { daemonize: false, @@ -298,6 +303,7 @@ mod tests { superuser: "zenith_admin".to_string(), workdir: repo_dir, pg_distrib_dir: "".into(), + repository_format, }; // Make a static copy of the config. This can never be free'd, but that's // OK in a test. @@ -305,24 +311,45 @@ mod tests { let tenantid = ZTenantId::generate(); fs::create_dir_all(conf.tenant_path(&tenantid)).unwrap(); - let obj_store = RocksObjectStore::create(conf, &tenantid)?; - let walredo_mgr = TestRedoManager {}; - let repo = - ObjectRepository::new(conf, Arc::new(obj_store), Arc::new(walredo_mgr), tenantid); + let repo: Box = match conf.repository_format { + RepositoryFormat::Layered => { + Box::new(LayeredRepository::new(conf, Arc::new(walredo_mgr))) + } + RepositoryFormat::RocksDb => { + let obj_store = RocksObjectStore::create(conf, &tenantid)?; - Ok(Box::new(repo)) + Box::new(ObjectRepository::new( + conf, + Arc::new(obj_store), + Arc::new(walredo_mgr), + tenantid + )) + } + }; + + Ok(repo) } /// Test get_relsize() and truncation. #[test] - fn test_relsize() -> Result<()> { + fn test_relsize_rocksdb() -> Result<()> { + let repo = get_test_repo("test_relsize_rocksdb", RepositoryFormat::RocksDb)?; + test_relsize(&*repo) + } + + #[test] + fn test_relsize_layered() -> Result<()> { + let repo = get_test_repo("test_relsize_layered", RepositoryFormat::Layered)?; + test_relsize(&*repo) + } + + fn test_relsize(repo: &dyn Repository) -> Result<()> { // get_timeline() with non-existent timeline id should fail //repo.get_timeline("11223344556677881122334455667788"); // Create timeline to work on - let repo = get_test_repo("test_relsize")?; let timelineid = ZTimelineId::from_str("11223344556677881122334455667788").unwrap(); let tline = repo.create_empty_timeline(timelineid, Lsn(0))?; @@ -407,14 +434,24 @@ mod tests { /// This isn't very interesting with the RocksDb implementation, as we don't pay /// any attention to Postgres segment boundaries there. #[test] - fn test_large_rel() -> Result<()> { - let repo = get_test_repo("test_large_rel")?; + fn test_large_rel_rocksdb() -> Result<()> { + let repo = get_test_repo("test_large_rel_rocksdb", RepositoryFormat::RocksDb)?; + test_large_rel(&*repo) + } + + #[test] + fn test_large_rel_layered() -> Result<()> { + let repo = get_test_repo("test_large_rel_layered", RepositoryFormat::Layered)?; + test_large_rel(&*repo) + } + + fn test_large_rel(repo: &dyn Repository) -> Result<()> { let timelineid = ZTimelineId::from_str("11223344556677881122334455667788").unwrap(); let tline = repo.create_empty_timeline(timelineid, Lsn(0))?; tline.init_valid_lsn(Lsn(1)); - let mut lsn = 0; + let mut lsn = 1; for blknum in 0..pg_constants::RELSEG_SIZE + 1 { let img = TEST_IMG(&format!("foo blk {} at {}", blknum, Lsn(lsn))); lsn += 1; @@ -460,12 +497,22 @@ mod tests { })) } + #[test] + fn test_branch_rocksdb() -> Result<()> { + let repo = get_test_repo("test_branch_rocksdb", RepositoryFormat::RocksDb)?; + test_branch(&*repo) + } + + #[test] + fn test_branch_layered() -> Result<()> { + let repo = get_test_repo("test_branch_layered", RepositoryFormat::Layered)?; + test_branch(&*repo) + } + /// /// Test branch creation /// - #[test] - fn test_branch() -> Result<()> { - let repo = get_test_repo("test_branch")?; + fn test_branch(repo: &dyn Repository) -> Result<()> { let timelineid = ZTimelineId::from_str("11223344556677881122334455667788").unwrap(); let tline = repo.create_empty_timeline(timelineid, Lsn(0))?; @@ -510,8 +557,16 @@ mod tests { } #[test] - fn test_history() -> Result<()> { - let repo = get_test_repo("test_snapshot")?; + fn test_history_rocksdb() -> Result<()> { + let repo = get_test_repo("test_history_rocksdb", RepositoryFormat::RocksDb)?; + test_history(&*repo) + } + #[test] + fn test_history_layered() -> Result<()> { + let repo = get_test_repo("test_history_layered", RepositoryFormat::Layered)?; + test_history(&*repo) + } + fn test_history(repo: &dyn Repository) -> Result<()> { let timelineid = ZTimelineId::from_str("11223344556677881122334455667788").unwrap(); let tline = repo.create_empty_timeline(timelineid, Lsn(0))?;