From 898937d5003776c43f716ba1d4078c9f41103dc2 Mon Sep 17 00:00:00 2001 From: Konstantin Knizhnik Date: Tue, 4 Oct 2022 20:42:34 +0300 Subject: [PATCH] Store partria image layers instead of compaction --- pageserver/src/http/routes.rs | 2 +- pageserver/src/tenant.rs | 2 +- pageserver/src/tenant/delta_layer.rs | 61 +++-- pageserver/src/tenant/inmemory_layer.rs | 4 +- pageserver/src/tenant/layer_map.rs | 26 +- pageserver/src/tenant/storage_layer.rs | 6 +- pageserver/src/tenant/timeline.rs | 310 ++++-------------------- 7 files changed, 105 insertions(+), 306 deletions(-) diff --git a/pageserver/src/http/routes.rs b/pageserver/src/http/routes.rs index 55429420a8..aa98f09748 100644 --- a/pageserver/src/http/routes.rs +++ b/pageserver/src/http/routes.rs @@ -810,7 +810,7 @@ async fn timeline_compact_handler(request: Request) -> Result for Summary { // Flag indicating that this version initialize the page const WILL_INIT: u64 = 1; +// Flag indicating page image +const IS_IMAGE: u64 = 2; + /// /// Struct representing reference to BLOB in layers. Reference contains BLOB /// offset, and for WAL records it also contains `will_init` flag. The flag @@ -109,15 +112,22 @@ impl BlobRef { (self.0 & WILL_INIT) != 0 } - pub fn pos(&self) -> u64 { - self.0 >> 1 + pub fn is_image(&self) -> bool { + (self.0 & IS_IMAGE) != 0 } - pub fn new(pos: u64, will_init: bool) -> BlobRef { - let mut blob_ref = pos << 1; + pub fn pos(&self) -> u64 { + self.0 >> 2 + } + + pub fn new(pos: u64, will_init: bool, is_image: bool) -> BlobRef { + let mut blob_ref = pos << 2; if will_init { blob_ref |= WILL_INIT; } + if is_image { + blob_ref |= IS_IMAGE; + } BlobRef(blob_ref) } } @@ -314,13 +324,13 @@ impl Layer for DeltaLayer { } } - fn key_iter<'a>(&'a self) -> Box + 'a> { + fn key_iter<'a>(&'a self, skip_images: bool) -> Box + 'a> { let inner = match self.load() { Ok(inner) => inner, Err(e) => panic!("Failed to load a delta layer: {e:?}"), }; - match DeltaKeyIter::new(inner) { + match DeltaKeyIter::new(inner, skip_images) { Ok(iter) => Box::new(iter), Err(e) => panic!("Layer index is corrupted: {e:?}"), } @@ -671,7 +681,13 @@ impl DeltaLayerWriter { /// The values must be appended in key, lsn order. /// pub fn put_value(&mut self, key: Key, lsn: Lsn, val: Value) -> Result<()> { - self.put_value_bytes(key, lsn, &Value::ser(&val)?, val.will_init()) + self.put_value_bytes( + key, + lsn, + &Value::ser(&val)?, + val.will_init(), + val.is_image(), + ) } pub fn put_value_bytes( @@ -680,12 +696,12 @@ impl DeltaLayerWriter { lsn: Lsn, val: &[u8], will_init: bool, + is_image: bool, ) -> Result<()> { assert!(self.lsn_range.start <= lsn); let off = self.blob_writer.write_blob(val)?; - - let blob_ref = BlobRef::new(off, will_init); + let blob_ref = BlobRef::new(off, will_init, is_image); let delta_key = DeltaKey::from_key_lsn(&key, lsn); self.tree.append(&delta_key.0, blob_ref.0)?; @@ -874,7 +890,7 @@ impl Iterator for DeltaKeyIter { } impl<'a> DeltaKeyIter { - fn new(inner: RwLockReadGuard<'a, DeltaLayerInner>) -> Result { + fn new(inner: RwLockReadGuard<'a, DeltaLayerInner>, skip_images: bool) -> Result { let file = inner.file.as_ref().unwrap(); let tree_reader = DiskBtreeReader::<_, DELTA_KEY_SIZE>::new( inner.index_start_blk, @@ -887,19 +903,22 @@ impl<'a> DeltaKeyIter { &[0u8; DELTA_KEY_SIZE], VisitDirection::Forwards, |key, value| { - let delta_key = DeltaKey::from_slice(key); - let pos = BlobRef(value).pos(); - if let Some(last) = all_keys.last_mut() { - if last.0.key() == delta_key.key() { - return true; - } else { - // subtract offset of new key BLOB and first blob of this key - // to get total size if values associated with this key - let first_pos = last.1; - last.1 = pos - first_pos; + let blob_ref = BlobRef(value); + if !blob_ref.is_image() || !skip_images { + let delta_key = DeltaKey::from_slice(key); + let pos = blob_ref.pos(); + if let Some(last) = all_keys.last_mut() { + if last.0.key() == delta_key.key() { + return true; + } else { + // subtract offset of new key BLOB and first blob of this key + // to get total size if values associated with this key + let first_pos = last.1; + last.1 = pos - first_pos; + } } + all_keys.push((delta_key, pos)); } - all_keys.push((delta_key, pos)); true }, )?; diff --git a/pageserver/src/tenant/inmemory_layer.rs b/pageserver/src/tenant/inmemory_layer.rs index 9aa33a72ca..6774351358 100644 --- a/pageserver/src/tenant/inmemory_layer.rs +++ b/pageserver/src/tenant/inmemory_layer.rs @@ -358,8 +358,8 @@ impl InMemoryLayer { // Write all page versions for (lsn, pos) in vec_map.as_slice() { cursor.read_blob_into_buf(*pos, &mut buf)?; - let will_init = Value::des(&buf)?.will_init(); - delta_layer_writer.put_value_bytes(key, *lsn, &buf, will_init)?; + let value = Value::des(&buf)?; + delta_layer_writer.put_value_bytes(key, *lsn, &buf, value.will_init(), value.is_image())?; } } diff --git a/pageserver/src/tenant/layer_map.rs b/pageserver/src/tenant/layer_map.rs index 495833e3ae..0f82a0b165 100644 --- a/pageserver/src/tenant/layer_map.rs +++ b/pageserver/src/tenant/layer_map.rs @@ -55,9 +55,8 @@ pub struct LayerMap { /// All the historic layers are kept here historic_layers: RTree, - /// L0 layers have key range Key::MIN..Key::MAX, and locating them using R-Tree search is very inefficient. - /// So L0 layers are held in l0_delta_layers vector, in addition to the R-tree. - l0_delta_layers: Vec>, + /// Latest stored delta layer + latest_delta_layer: Option>, } struct LayerRTreeObject { @@ -336,7 +335,7 @@ impl LayerMap { /// pub fn insert_historic(&mut self, layer: Arc) { if layer.get_key_range() == (Key::MIN..Key::MAX) { - self.l0_delta_layers.push(layer.clone()); + self.latest_delta_layer = Some(layer.clone()); } self.historic_layers.insert(LayerRTreeObject { layer }); NUM_ONDISK_LAYERS.inc(); @@ -349,16 +348,11 @@ impl LayerMap { /// pub fn remove_historic(&mut self, layer: Arc) { if layer.get_key_range() == (Key::MIN..Key::MAX) { - let len_before = self.l0_delta_layers.len(); - - // FIXME: ptr_eq might fail to return true for 'dyn' - // references. Clippy complains about this. In practice it - // seems to work, the assertion below would be triggered - // otherwise but this ought to be fixed. - #[allow(clippy::vtable_address_comparisons)] - self.l0_delta_layers - .retain(|other| !Arc::ptr_eq(other, &layer)); - assert_eq!(self.l0_delta_layers.len(), len_before - 1); + if let Some(latest_layer) = &self.latest_delta_layer { + if Arc::ptr_eq(&layer, latest_layer) { + self.latest_delta_layer = None; + } + } } assert!(self .historic_layers @@ -549,8 +543,8 @@ impl LayerMap { } /// Return all L0 delta layers - pub fn get_level0_deltas(&self) -> Result>> { - Ok(self.l0_delta_layers.clone()) + pub fn get_latest_delta_layer(&mut self) -> Option> { + self.latest_delta_layer.take() } /// debugging function to print out the contents of the layer map diff --git a/pageserver/src/tenant/storage_layer.rs b/pageserver/src/tenant/storage_layer.rs index 8dafcab124..7f025a5aba 100644 --- a/pageserver/src/tenant/storage_layer.rs +++ b/pageserver/src/tenant/storage_layer.rs @@ -139,9 +139,9 @@ pub trait Layer: Send + Sync { /// Iterate through all keys and values stored in the layer fn iter(&self) -> Box> + '_>; - /// Iterate through all keys stored in the layer. Returns key, lsn and value size - /// It is used only for compaction and so is currently implemented only for DeltaLayer - fn key_iter(&self) -> Box + '_> { + /// Iterate through all keys stored in the layer. Returns key, lsn and value size. + /// It is used only for reconstruction and so is currently implemented only for DeltaLayer + fn key_iter(&self, _skip_images: bool) -> Box + '_> { panic!("Not implemented") } diff --git a/pageserver/src/tenant/timeline.rs b/pageserver/src/tenant/timeline.rs index 247e076230..149432ca97 100644 --- a/pageserver/src/tenant/timeline.rs +++ b/pageserver/src/tenant/timeline.rs @@ -3,7 +3,6 @@ use anyhow::{anyhow, bail, ensure, Context, Result}; use bytes::Bytes; use fail::fail_point; -use itertools::Itertools; use once_cell::sync::OnceCell; use tokio::task::spawn_blocking; use tracing::*; @@ -119,7 +118,7 @@ pub struct Timeline { /// Layer removal lock. /// A lock to ensure that no layer of the timeline is removed concurrently by other tasks. - /// This lock is acquired in [`Timeline::gc`], [`Timeline::compact`], + /// This lock is acquired in [`Timeline::gc`], [`Timeline::reconstruct`], /// and [`Tenant::delete_timeline`]. layer_removal_cs: Mutex<()>, @@ -469,7 +468,7 @@ impl Timeline { CheckpointConfig::Forced => { self.freeze_inmem_layer(false); self.flush_frozen_layers(true)?; - self.compact() + self.reconstruct() } } } @@ -510,13 +509,6 @@ impl Timeline { .unwrap_or(self.conf.default_tenant_conf.compaction_target_size) } - fn get_compaction_threshold(&self) -> usize { - let tenant_conf = self.tenant_conf.read().unwrap(); - tenant_conf - .compaction_threshold - .unwrap_or(self.conf.default_tenant_conf.compaction_threshold) - } - fn get_image_creation_threshold(&self) -> usize { let tenant_conf = self.tenant_conf.read().unwrap(); tenant_conf @@ -731,7 +723,7 @@ impl Timeline { pub fn layer_removal_guard(&self) -> anyhow::Result> { self.layer_removal_cs .try_lock() - .map_err(|e| anyhow!("cannot lock compaction critical section {e}")) + .map_err(|e| anyhow!("cannot lock reconstruction critical section {e}")) } /// Retrieve current logical size of the timeline. @@ -1333,17 +1325,17 @@ impl Timeline { Ok(new_delta_path) } - pub fn compact(&self) -> anyhow::Result<()> { + pub fn reconstruct(&self) -> anyhow::Result<()> { let last_record_lsn = self.get_last_record_lsn(); // Last record Lsn could be zero in case the timelie was just created if !last_record_lsn.is_valid() { - warn!("Skipping compaction for potentially just initialized timeline, it has invalid last record lsn: {last_record_lsn}"); + warn!("Skipping reconstruction for potentially just initialized timeline, it has invalid last record lsn: {last_record_lsn}"); return Ok(()); } // - // High level strategy for compaction / image creation: + // High level strategy for reconstruction / image creation: // // 1. First, calculate the desired "partitioning" of the // currently in-use key space. The goal is to partition the @@ -1367,13 +1359,13 @@ impl Timeline { // total in the delta file. Or perhaps: if creating an image // file would allow to delete some older files. // - // 3. After that, we compact all level0 delta files if there - // are too many of them. While compacting, we also garbage + // 3. After that, we reconstruct all level0 delta files if there + // are too many of them. While reconstructing, we also garbage // collect any page versions that are no longer needed because // of the new image layers we created in step 2. // // TODO: This high level strategy hasn't been implemented yet. - // Below are functions compact_level0() and create_image_layers() + // Below are functions reconstruct_level0() and create_image_layers() // but they are a bit ad hoc and don't quite work like it's explained // above. Rewrite it. let _layer_removal_cs = self.layer_removal_cs.lock().unwrap(); @@ -1401,9 +1393,9 @@ impl Timeline { ); } - // 3. Compact - let timer = self.metrics.compact_time_histo.start_timer(); - self.compact_level0(target_file_size)?; + // 3. Reconstruct + let timer = self.metrics.reconstruct_time_histo.start_timer(); + self.reconstruct_level0(target_file_size)?; timer.stop_and_record(); } Err(err) => { @@ -1411,7 +1403,7 @@ impl Timeline { // as an empty timeline. Also in unit tests, when we use the timeline // as a simple key-value store, ignoring the datadir layout. Log the // error but continue. - error!("could not compact, repartitioning keyspace failed: {err:?}"); + error!("could not reconstruct, repartitioning keyspace failed: {err:?}"); } }; @@ -1514,7 +1506,7 @@ impl Timeline { // We must also fsync the timeline dir to ensure the directory entries for // new layer files are durable // - // Compaction creates multiple image layers. It would be better to create them all + // Reconstruction creates multiple image layers. It would be better to create them all // and fsync them all in parallel. let mut all_paths = Vec::from_iter(layer_paths_to_upload.clone()); all_paths.push(self.conf.timeline_path(&self.timeline_id, &self.tenant_id)); @@ -1534,230 +1526,46 @@ impl Timeline { } /// - /// Collect a bunch of Level 0 layer files, and compact and reshuffle them as + /// Collect a bunch of Level 0 layer files, and reconstruct and reshuffle them as /// as Level 1 files. /// - fn compact_level0(&self, target_file_size: u64) -> Result<()> { - let layers = self.layers.read().unwrap(); - let mut level0_deltas = layers.get_level0_deltas()?; + fn reconstruct_level0(&self, target_file_size: u64) -> Result<()> { + let mut layers = self.layers.write().unwrap(); + let latest_delta_layer = layers.get_latest_delta_layer(); drop(layers); - - // Only compact if enough layers have accumulated. - if level0_deltas.is_empty() || level0_deltas.len() < self.get_compaction_threshold() { - return Ok(()); - } - - // Gather the files to compact in this iteration. - // - // Start with the oldest Level 0 delta file, and collect any other - // level 0 files that form a contiguous sequence, such that the end - // LSN of previous file matches the start LSN of the next file. - // - // Note that if the files don't form such a sequence, we might - // "compact" just a single file. That's a bit pointless, but it allows - // us to get rid of the level 0 file, and compact the other files on - // the next iteration. This could probably made smarter, but such - // "gaps" in the sequence of level 0 files should only happen in case - // of a crash, partial download from cloud storage, or something like - // that, so it's not a big deal in practice. - level0_deltas.sort_by_key(|l| l.get_lsn_range().start); - let mut level0_deltas_iter = level0_deltas.iter(); - - let first_level0_delta = level0_deltas_iter.next().unwrap(); - let mut prev_lsn_end = first_level0_delta.get_lsn_range().end; - let mut deltas_to_compact = vec![Arc::clone(first_level0_delta)]; - for l in level0_deltas_iter { - let lsn_range = l.get_lsn_range(); - - if lsn_range.start != prev_lsn_end { - break; - } - deltas_to_compact.push(Arc::clone(l)); - prev_lsn_end = lsn_range.end; - } - let lsn_range = Range { - start: deltas_to_compact.first().unwrap().get_lsn_range().start, - end: deltas_to_compact.last().unwrap().get_lsn_range().end, - }; - - info!( - "Starting Level0 compaction in LSN range {}-{} for {} layers ({} deltas in total)", - lsn_range.start, - lsn_range.end, - deltas_to_compact.len(), - level0_deltas.len() - ); - for l in deltas_to_compact.iter() { - info!("compact includes {}", l.filename().display()); - } - // We don't need the original list of layers anymore. Drop it so that - // we don't accidentally use it later in the function. - drop(level0_deltas); - - // This iterator walks through all key-value pairs from all the layers - // we're compacting, in key, LSN order. - let all_values_iter = deltas_to_compact - .iter() - .map(|l| l.iter()) - .kmerge_by(|a, b| { - if let Ok((a_key, a_lsn, _)) = a { - if let Ok((b_key, b_lsn, _)) = b { - match a_key.cmp(b_key) { - Ordering::Less => true, - Ordering::Equal => a_lsn <= b_lsn, - Ordering::Greater => false, - } - } else { - false - } - } else { - true - } - }); - - // This iterator walks through all keys and is needed to calculate size used by each key - let mut all_keys_iter = deltas_to_compact - .iter() - .map(|l| l.key_iter()) - .kmerge_by(|a, b| { - let (a_key, a_lsn, _) = a; - let (b_key, b_lsn, _) = b; - match a_key.cmp(b_key) { - Ordering::Less => true, - Ordering::Equal => a_lsn <= b_lsn, - Ordering::Greater => false, - } - }); - - // Merge the contents of all the input delta layers into a new set - // of delta layers, based on the current partitioning. - // - // We split the new delta layers on the key dimension. We iterate through the key space, and for each key, check if including the next key to the current output layer we're building would cause the layer to become too large. If so, dump the current output layer and start new one. - // It's possible that there is a single key with so many page versions that storing all of them in a single layer file - // would be too large. In that case, we also split on the LSN dimension. - // - // LSN - // ^ - // | - // | +-----------+ +--+--+--+--+ - // | | | | | | | | - // | +-----------+ | | | | | - // | | | | | | | | - // | +-----------+ ==> | | | | | - // | | | | | | | | - // | +-----------+ | | | | | - // | | | | | | | | - // | +-----------+ +--+--+--+--+ - // | - // +--------------> key - // - // - // If one key (X) has a lot of page versions: - // - // LSN - // ^ - // | (X) - // | +-----------+ +--+--+--+--+ - // | | | | | | | | - // | +-----------+ | | +--+ | - // | | | | | | | | - // | +-----------+ ==> | | | | | - // | | | | | +--+ | - // | +-----------+ | | | | | - // | | | | | | | | - // | +-----------+ +--+--+--+--+ - // | - // +--------------> key - // TODO: this actually divides the layers into fixed-size chunks, not - // based on the partitioning. - // - // TODO: we should also opportunistically materialize and - // garbage collect what we can. + let mut writer: Option = None; let mut new_layers = Vec::new(); - let mut prev_key: Option = None; - let mut writer: Option = None; - let mut key_values_total_size = 0u64; - let mut dup_start_lsn: Lsn = Lsn::INVALID; // start LSN of layer containing values of the single key - let mut dup_end_lsn: Lsn = Lsn::INVALID; // end LSN of layer containing values of the single key - for x in all_values_iter { - let (key, lsn, value) = x?; - let same_key = prev_key.map_or(false, |prev_key| prev_key == key); - // We need to check key boundaries once we reach next key or end of layer with the same key - if !same_key || lsn == dup_end_lsn { - let mut next_key_size = 0u64; - let is_dup_layer = dup_end_lsn.is_valid(); - dup_start_lsn = Lsn::INVALID; - if !same_key { - dup_end_lsn = Lsn::INVALID; + let mut last_key: Option = None; + if let Some(last_delta_layer) = latest_delta_layer { + for (key, lsn, _) in last_delta_layer.key_iter(true) { + match &writer { + Some(curr_writer) => { + if curr_writer.size() > target_file_size { + new_layers.push(writer.take().unwrap().finish(key)?); + writer = None; + } + }, + _ => { + // Create writer if not initiaized yet + writer = Some(DeltaLayerWriter::new( + self.conf, + self.timeline_id, + self.tenant_id, + key, + last_delta_layer.get_lsn_range().clone(), + )?); + } } - // Determine size occupied by this key. We stop at next key or when size becomes larger than target_file_size - for (next_key, next_lsn, next_size) in all_keys_iter.by_ref() { - next_key_size = next_size; - if key != next_key { - if dup_end_lsn.is_valid() { - // We are writting segment with duplicates: - // place all remaining values of this key in separate segment - dup_start_lsn = dup_end_lsn; // new segments starts where old stops - dup_end_lsn = lsn_range.end; // there are no more values of this key till end of LSN range - } - break; - } - key_values_total_size += next_size; - // Check if it is time to split segment: if total keys size is larger than target file size. - // We need to avoid generation of empty segments if next_size > target_file_size. - if key_values_total_size > target_file_size && lsn != next_lsn { - // Split key between multiple layers: such layer can contain only single key - dup_start_lsn = if dup_end_lsn.is_valid() { - dup_end_lsn // new segment with duplicates starts where old one stops - } else { - lsn // start with the first LSN for this key - }; - dup_end_lsn = next_lsn; // upper LSN boundary is exclusive - break; - } - } - // handle case when loop reaches last key: in this case dup_end is non-zero but dup_start is not set. - if dup_end_lsn.is_valid() && !dup_start_lsn.is_valid() { - dup_start_lsn = dup_end_lsn; - dup_end_lsn = lsn_range.end; - } - if writer.is_some() { - let written_size = writer.as_mut().unwrap().size(); - // check if key cause layer overflow... - if is_dup_layer - || dup_end_lsn.is_valid() - || written_size + key_values_total_size > target_file_size - { - // ... if so, flush previous layer and prepare to write new one - new_layers.push(writer.take().unwrap().finish(prev_key.unwrap().next())?); - writer = None; - } - } - // Remember size of key value because at next iteration we will access next item - key_values_total_size = next_key_size; + let value = self.get(key, lsn)?; + writer + .as_mut() + .unwrap() + .put_value(key, lsn, Value::Image(value))?; + last_key = Some(key); } - if writer.is_none() { - // Create writer if not initiaized yet - writer = Some(DeltaLayerWriter::new( - self.conf, - self.timeline_id, - self.tenant_id, - key, - if dup_end_lsn.is_valid() { - // this is a layer containing slice of values of the same key - debug!("Create new dup layer {}..{}", dup_start_lsn, dup_end_lsn); - dup_start_lsn..dup_end_lsn - } else { - debug!("Create new layer {}..{}", lsn_range.start, lsn_range.end); - lsn_range.clone() - }, - )?); - } - writer.as_mut().unwrap().put_value(key, lsn, value)?; - prev_key = Some(key); } if let Some(writer) = writer { - new_layers.push(writer.finish(prev_key.unwrap().next())?); + new_layers.push(writer.finish(last_key.unwrap().next())?); } // Sync layers @@ -1787,23 +1595,6 @@ impl Timeline { new_layer_paths.insert(new_delta_path); layers.insert_historic(Arc::new(l)); } - - // Now that we have reshuffled the data to set of new delta layers, we can - // delete the old ones - let mut layer_paths_do_delete = HashSet::with_capacity(deltas_to_compact.len()); - drop(all_keys_iter); - for l in deltas_to_compact { - if let Some(path) = l.local_path() { - self.metrics - .current_physical_size_gauge - .sub(path.metadata()?.len()); - layer_paths_do_delete.insert(path); - } - l.delete()?; - layers.remove_historic(l); - } - drop(layers); - if self.upload_layers.load(atomic::Ordering::Relaxed) { storage_sync::schedule_layer_upload( self.tenant_id, @@ -1811,11 +1602,6 @@ impl Timeline { new_layer_paths, None, ); - storage_sync::schedule_layer_delete( - self.tenant_id, - self.timeline_id, - layer_paths_do_delete, - ); } Ok(()) @@ -1823,10 +1609,10 @@ impl Timeline { /// Update information about which layer files need to be retained on /// garbage collection. This is separate from actually performing the GC, - /// and is updated more frequently, so that compaction can remove obsolete + /// and is updated more frequently, so that reconstruction can remove obsolete /// page versions more aggressively. /// - /// TODO: that's wishful thinking, compaction doesn't actually do that + /// TODO: that's wishful thinking, reconstruction doesn't actually do that /// currently. /// /// The caller specifies how much history is needed with the 3 arguments: