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))?;