diff --git a/pageserver/benches/bench_layer_map.rs b/pageserver/benches/bench_layer_map.rs index e18c00da96..5edfa84d8a 100644 --- a/pageserver/benches/bench_layer_map.rs +++ b/pageserver/benches/bench_layer_map.rs @@ -1,8 +1,7 @@ use pageserver::keyspace::{KeyPartitioning, KeySpace}; use pageserver::repository::Key; use pageserver::tenant::layer_map::LayerMap; -use pageserver::tenant::storage_layer::Layer; -use pageserver::tenant::storage_layer::{DeltaFileName, ImageFileName, LayerDescriptor}; +use pageserver::tenant::storage_layer::{Layer, LayerDescriptor, LayerFileName}; use rand::prelude::{SeedableRng, SliceRandom, StdRng}; use std::cmp::{max, min}; use std::fs::File; @@ -26,30 +25,15 @@ fn build_layer_map(filename_dump: PathBuf) -> LayerMap { let mut updates = layer_map.batch_update(); for fname in filenames { - let fname = &fname.unwrap(); - if let Some(imgfilename) = ImageFileName::parse_str(fname) { - let layer = LayerDescriptor { - key: imgfilename.key_range, - lsn: imgfilename.lsn..(imgfilename.lsn + 1), - is_incremental: false, - short_id: fname.to_string(), - }; - updates.insert_historic(Arc::new(layer)); - min_lsn = min(min_lsn, imgfilename.lsn); - max_lsn = max(max_lsn, imgfilename.lsn); - } else if let Some(deltafilename) = DeltaFileName::parse_str(fname) { - let layer = LayerDescriptor { - key: deltafilename.key_range.clone(), - lsn: deltafilename.lsn_range.clone(), - is_incremental: true, - short_id: fname.to_string(), - }; - updates.insert_historic(Arc::new(layer)); - min_lsn = min(min_lsn, deltafilename.lsn_range.start); - max_lsn = max(max_lsn, deltafilename.lsn_range.end); - } else { - panic!("unexpected filename {fname}"); - } + let fname = fname.unwrap(); + let fname = LayerFileName::from_str(&fname).unwrap(); + let layer = LayerDescriptor::from(fname); + + let lsn_range = layer.get_lsn_range(); + min_lsn = min(min_lsn, lsn_range.start); + max_lsn = max(max_lsn, Lsn(lsn_range.end.0 - 1)); + + updates.insert_historic(Arc::new(layer)); } println!("min: {min_lsn}, max: {max_lsn}"); diff --git a/pageserver/src/tenant/layer_map.rs b/pageserver/src/tenant/layer_map.rs index bb386b436b..59a358a355 100644 --- a/pageserver/src/tenant/layer_map.rs +++ b/pageserver/src/tenant/layer_map.rs @@ -59,6 +59,7 @@ use std::sync::Arc; use utils::lsn::Lsn; use historic_layer_coverage::BufferedHistoricLayerCoverage; +pub use historic_layer_coverage::Replacement; use super::storage_layer::range_eq; @@ -138,6 +139,24 @@ where self.layer_map.remove_historic_noflush(layer) } + /// Replaces existing layer iff it is the `expected`. + /// + /// If the expected layer has been removed it will not be inserted by this function. + /// + /// Returned `Replacement` describes succeeding in replacement or the reason why it could not + /// be done. + /// + /// TODO replacement can be done without buffering and rebuilding layer map updates. + /// One way to do that is to add a layer of indirection for returned values, so + /// that we can replace values only by updating a hashmap. + pub fn replace_historic( + &mut self, + expected: &Arc, + new: Arc, + ) -> anyhow::Result>> { + self.layer_map.replace_historic_noflush(expected, new) + } + // We will flush on drop anyway, but this method makes it // more explicit that there is some work being done. /// Apply all updates @@ -254,14 +273,8 @@ where /// Helper function for BatchedUpdates::insert_historic /// pub(self) fn insert_historic_noflush(&mut self, layer: Arc) { - let kr = layer.get_key_range(); - let lr = layer.get_lsn_range(); self.historic.insert( - historic_layer_coverage::LayerKey { - key: kr.start.to_i128()..kr.end.to_i128(), - lsn: lr.start.0..lr.end.0, - is_image: !layer.is_incremental(), - }, + historic_layer_coverage::LayerKey::from(&*layer), Arc::clone(&layer), ); @@ -278,30 +291,72 @@ where /// Helper function for BatchedUpdates::remove_historic /// pub fn remove_historic_noflush(&mut self, layer: Arc) { - let kr = layer.get_key_range(); - let lr = layer.get_lsn_range(); - self.historic.remove(historic_layer_coverage::LayerKey { - key: kr.start.to_i128()..kr.end.to_i128(), - lsn: lr.start.0..lr.end.0, - is_image: !layer.is_incremental(), - }); + self.historic + .remove(historic_layer_coverage::LayerKey::from(&*layer)); if Self::is_l0(&layer) { 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); + .retain(|other| !Self::compare_arced_layers(other, &layer)); + // this assertion is related to use of Arc::ptr_eq in Self::compare_arced_layers, + // there's a chance that the comparison fails at runtime due to it comparing (pointer, + // vtable) pairs. + assert_eq!( + self.l0_delta_layers.len(), + len_before - 1, + "failed to locate removed historic layer from l0_delta_layers" + ); } NUM_ONDISK_LAYERS.dec(); } + pub(self) fn replace_historic_noflush( + &mut self, + expected: &Arc, + new: Arc, + ) -> anyhow::Result>> { + let key = historic_layer_coverage::LayerKey::from(&**expected); + let other = historic_layer_coverage::LayerKey::from(&*new); + + let expected_l0 = Self::is_l0(expected); + let new_l0 = Self::is_l0(&new); + + anyhow::ensure!( + key == other, + "expected and new must have equal LayerKeys: {key:?} != {other:?}" + ); + + anyhow::ensure!( + expected_l0 == new_l0, + "expected and new must both be l0 deltas or neither should be: {expected_l0} != {new_l0}" + ); + + let l0_index = if expected_l0 { + // find the index in case replace worked, we need to replace that as well + Some( + self.l0_delta_layers + .iter() + .position(|slot| Self::compare_arced_layers(slot, expected)) + .ok_or_else(|| anyhow::anyhow!("existing l0 delta layer was not found"))?, + ) + } else { + None + }; + + let replaced = self.historic.replace(&key, new.clone(), |existing| { + Self::compare_arced_layers(existing, expected) + }); + + if let Replacement::Replaced { .. } = &replaced { + if let Some(index) = l0_index { + self.l0_delta_layers[index] = new; + } + } + + Ok(replaced) + } + /// Helper function for BatchedUpdates::drop. pub(self) fn flush_updates(&mut self) { self.historic.rebuild(); @@ -675,4 +730,105 @@ where println!("End dump LayerMap"); Ok(()) } + + #[inline(always)] + fn compare_arced_layers(left: &Arc, right: &Arc) -> bool { + // FIXME: ptr_eq might fail to return true for 'dyn' references because of multiple vtables + // can be created in compilation. Clippy complains about this. In practice it seems to + // work. + // + // In future rust versions this might become Arc::as_ptr(left) as *const () == + // Arc::as_ptr(right) as *const (), we could change to that before. + #[allow(clippy::vtable_address_comparisons)] + Arc::ptr_eq(left, right) + } +} + +#[cfg(test)] +mod tests { + use super::{LayerMap, Replacement}; + use crate::tenant::storage_layer::{Layer, LayerDescriptor, LayerFileName}; + use std::str::FromStr; + use std::sync::Arc; + + mod l0_delta_layers_updated { + + use super::*; + + #[test] + fn for_full_range_delta() { + // l0_delta_layers are used by compaction, and should observe all buffered updates + l0_delta_layers_updated_scenario( + "000000000000000000000000000000000000-FFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF__0000000053423C21-0000000053424D69", + true + ) + } + + #[test] + fn for_non_full_range_delta() { + // has minimal uncovered areas compared to l0_delta_layers_updated_on_insert_replace_remove_for_full_range_delta + l0_delta_layers_updated_scenario( + "000000000000000000000000000000000001-FFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFE__0000000053423C21-0000000053424D69", + // because not full range + false + ) + } + + #[test] + fn for_image() { + l0_delta_layers_updated_scenario( + "000000000000000000000000000000000000-000000000000000000000000000000010000__0000000053424D69", + // code only checks if it is a full range layer, doesn't care about images, which must + // mean we should in practice never have full range images + false + ) + } + + fn l0_delta_layers_updated_scenario(layer_name: &str, expected_l0: bool) { + let name = LayerFileName::from_str(layer_name).unwrap(); + let skeleton = LayerDescriptor::from(name); + + let remote: Arc = Arc::new(skeleton.clone()); + let downloaded: Arc = Arc::new(skeleton); + + let mut map = LayerMap::default(); + + // two disjoint Arcs in different lifecycle phases. + assert!(!LayerMap::compare_arced_layers(&remote, &downloaded)); + + let expected_in_counts = (1, usize::from(expected_l0)); + + map.batch_update().insert_historic(remote.clone()); + assert_eq!(count_layer_in(&map, &remote), expected_in_counts); + + let replaced = map + .batch_update() + .replace_historic(&remote, downloaded.clone()) + .expect("name derived attributes are the same"); + assert!( + matches!(replaced, Replacement::Replaced { .. }), + "{replaced:?}" + ); + assert_eq!(count_layer_in(&map, &downloaded), expected_in_counts); + + map.batch_update().remove_historic(downloaded.clone()); + assert_eq!(count_layer_in(&map, &downloaded), (0, 0)); + } + + fn count_layer_in(map: &LayerMap, layer: &Arc) -> (usize, usize) { + let historic = map + .iter_historic_layers() + .filter(|x| LayerMap::compare_arced_layers(x, layer)) + .count(); + let l0s = map + .get_level0_deltas() + .expect("why does this return a result"); + let l0 = l0s + .iter() + .filter(|x| LayerMap::compare_arced_layers(x, layer)) + .count(); + + (historic, l0) + } + } } diff --git a/pageserver/src/tenant/layer_map/historic_layer_coverage.rs b/pageserver/src/tenant/layer_map/historic_layer_coverage.rs index 46821aef15..b63c361314 100644 --- a/pageserver/src/tenant/layer_map/historic_layer_coverage.rs +++ b/pageserver/src/tenant/layer_map/historic_layer_coverage.rs @@ -41,6 +41,18 @@ impl Ord for LayerKey { } } +impl<'a, L: crate::tenant::storage_layer::Layer + ?Sized> From<&'a L> for LayerKey { + fn from(layer: &'a L) -> Self { + let kr = layer.get_key_range(); + let lr = layer.get_lsn_range(); + LayerKey { + key: kr.start.to_i128()..kr.end.to_i128(), + lsn: lr.start.0..lr.end.0, + is_image: !layer.is_incremental(), + } + } +} + /// Efficiently queryable layer coverage for each LSN. /// /// Allows answering layer map queries very efficiently, @@ -82,15 +94,13 @@ impl HistoricLayerCoverage { } // Insert into data structure - if layer_key.is_image { - self.head - .image_coverage - .insert(layer_key.key, layer_key.lsn.clone(), value); + let target = if layer_key.is_image { + &mut self.head.image_coverage } else { - self.head - .delta_coverage - .insert(layer_key.key, layer_key.lsn.clone(), value); - } + &mut self.head.delta_coverage + }; + + target.insert(layer_key.key, layer_key.lsn.clone(), value); // Remember history. Clone is O(1) self.historic.insert(layer_key.lsn.start, self.head.clone()); @@ -415,6 +425,59 @@ impl BufferedHistoricLayerCoverage { self.buffer.insert(layer_key, None); } + /// Replaces a previous layer with a new layer value. + /// + /// The replacement is conditional on: + /// - there is an existing `LayerKey` record + /// - there is no buffered removal for the given `LayerKey` + /// - the given closure returns true for the current `Value` + /// + /// The closure is used to compare the latest value (buffered insert, or existing layer) + /// against some expectation. This allows to use `Arc::ptr_eq` or similar which would be + /// inaccessible via `PartialEq` trait. + /// + /// Returns a `Replacement` value describing the outcome; only the case of + /// `Replacement::Replaced` modifies the map and requires a rebuild. + pub fn replace( + &mut self, + layer_key: &LayerKey, + new: Value, + check_expected: F, + ) -> Replacement + where + F: FnOnce(&Value) -> bool, + { + let (slot, in_buffered) = match self.buffer.get(layer_key) { + Some(inner @ Some(_)) => { + // we compare against the buffered version, because there will be a later + // rebuild before querying + (inner.as_ref(), true) + } + Some(None) => { + // buffer has removal for this key; it will not be equivalent by any check_expected. + return Replacement::RemovalBuffered; + } + None => { + // no pending modification for the key, check layers + (self.layers.get(layer_key), false) + } + }; + + match slot { + Some(existing) if !check_expected(existing) => { + // unfortunate clone here, but otherwise the nll borrowck grows the region of + // 'a to cover the whole function, and we could not mutate in the other + // Some(existing) branch + Replacement::Unexpected(existing.clone()) + } + None => Replacement::NotFound, + Some(_existing) => { + self.insert(layer_key.to_owned(), new); + Replacement::Replaced { in_buffered } + } + } + } + pub fn rebuild(&mut self) { // Find the first LSN that needs to be rebuilt let rebuild_since: u64 = match self.buffer.iter().next() { @@ -483,6 +546,22 @@ impl BufferedHistoricLayerCoverage { } } +/// Outcome of the replace operation. +#[derive(Debug)] +pub enum Replacement { + /// Previous value was replaced with the new value. + Replaced { + /// Replacement happened for a scheduled insert. + in_buffered: bool, + }, + /// Key was not found buffered updates or existing layers. + NotFound, + /// Key has been scheduled for removal, it was not replaced. + RemovalBuffered, + /// Previous value was rejected by the closure. + Unexpected(Value), +} + #[test] fn test_retroactive_regression_1() { let mut map = BufferedHistoricLayerCoverage::new(); @@ -548,7 +627,7 @@ fn test_retroactive_simple() { LayerKey { key: 2..5, lsn: 105..106, - is_image: true, + is_image: false, }, "Delta 1".to_string(), ); @@ -556,17 +635,24 @@ fn test_retroactive_simple() { // Rebuild so we can start querying map.rebuild(); - // Query key 4 - let version = map.get().unwrap().get_version(90); - assert!(version.is_none()); - let version = map.get().unwrap().get_version(102).unwrap(); - assert_eq!(version.image_coverage.query(4), Some("Image 1".to_string())); - let version = map.get().unwrap().get_version(107).unwrap(); - assert_eq!(version.image_coverage.query(4), Some("Delta 1".to_string())); - let version = map.get().unwrap().get_version(115).unwrap(); - assert_eq!(version.image_coverage.query(4), Some("Image 2".to_string())); - let version = map.get().unwrap().get_version(125).unwrap(); - assert_eq!(version.image_coverage.query(4), Some("Image 3".to_string())); + { + let map = map.get().expect("rebuilt"); + + let version = map.get_version(90); + assert!(version.is_none()); + let version = map.get_version(102).unwrap(); + assert_eq!(version.image_coverage.query(4), Some("Image 1".to_string())); + + let version = map.get_version(107).unwrap(); + assert_eq!(version.image_coverage.query(4), Some("Image 1".to_string())); + assert_eq!(version.delta_coverage.query(4), Some("Delta 1".to_string())); + + let version = map.get_version(115).unwrap(); + assert_eq!(version.image_coverage.query(4), Some("Image 2".to_string())); + + let version = map.get_version(125).unwrap(); + assert_eq!(version.image_coverage.query(4), Some("Image 3".to_string())); + } // Remove Image 3 map.remove(LayerKey { @@ -576,8 +662,147 @@ fn test_retroactive_simple() { }); map.rebuild(); - // Check deletion worked - let version = map.get().unwrap().get_version(125).unwrap(); - assert_eq!(version.image_coverage.query(4), Some("Image 2".to_string())); - assert_eq!(version.image_coverage.query(8), Some("Image 4".to_string())); + { + // Check deletion worked + let map = map.get().expect("rebuilt"); + let version = map.get_version(125).unwrap(); + assert_eq!(version.image_coverage.query(4), Some("Image 2".to_string())); + assert_eq!(version.image_coverage.query(8), Some("Image 4".to_string())); + } +} + +#[test] +fn test_retroactive_replacement() { + let mut map = BufferedHistoricLayerCoverage::new(); + + let keys = [ + LayerKey { + key: 0..5, + lsn: 100..101, + is_image: true, + }, + LayerKey { + key: 3..9, + lsn: 110..111, + is_image: true, + }, + LayerKey { + key: 4..6, + lsn: 120..121, + is_image: true, + }, + ]; + + let layers = [ + "Image 1".to_string(), + "Image 2".to_string(), + "Image 3".to_string(), + ]; + + for (key, layer) in keys.iter().zip(layers.iter()) { + map.insert(key.to_owned(), layer.to_owned()); + } + + // rebuild is not necessary here, because replace works for both buffered updates and existing + // layers. + + for (key, orig_layer) in keys.iter().zip(layers.iter()) { + let replacement = format!("Remote {orig_layer}"); + + // evict + let ret = map.replace(key, replacement.clone(), |l| l == orig_layer); + assert!( + matches!(ret, Replacement::Replaced { .. }), + "replace {orig_layer}: {ret:?}" + ); + map.rebuild(); + + let at = key.lsn.end + 1; + + let version = map.get().expect("rebuilt").get_version(at).unwrap(); + assert_eq!( + version.image_coverage.query(4).as_deref(), + Some(replacement.as_str()), + "query for 4 at version {at} after eviction", + ); + + // download + let ret = map.replace(key, orig_layer.clone(), |l| l == &replacement); + assert!( + matches!(ret, Replacement::Replaced { .. }), + "replace {orig_layer} back: {ret:?}" + ); + map.rebuild(); + let version = map.get().expect("rebuilt").get_version(at).unwrap(); + assert_eq!( + version.image_coverage.query(4).as_deref(), + Some(orig_layer.as_str()), + "query for 4 at version {at} after download", + ); + } +} + +#[test] +fn missing_key_is_not_inserted_with_replace() { + let mut map = BufferedHistoricLayerCoverage::new(); + let key = LayerKey { + key: 0..5, + lsn: 100..101, + is_image: true, + }; + + let ret = map.replace(&key, "should not replace", |_| true); + assert!(matches!(ret, Replacement::NotFound), "{ret:?}"); + map.rebuild(); + assert!(map + .get() + .expect("no changes to rebuild") + .get_version(102) + .is_none()); +} + +#[test] +fn replacing_buffered_insert_and_remove() { + let mut map = BufferedHistoricLayerCoverage::new(); + let key = LayerKey { + key: 0..5, + lsn: 100..101, + is_image: true, + }; + + map.insert(key.clone(), "Image 1"); + let ret = map.replace(&key, "Remote Image 1", |&l| l == "Image 1"); + assert!( + matches!(ret, Replacement::Replaced { in_buffered: true }), + "{ret:?}" + ); + map.rebuild(); + + assert_eq!( + map.get() + .expect("rebuilt") + .get_version(102) + .unwrap() + .image_coverage + .query(4), + Some("Remote Image 1") + ); + + map.remove(key.clone()); + let ret = map.replace(&key, "should not replace", |_| true); + assert!( + matches!(ret, Replacement::RemovalBuffered), + "cannot replace after scheduled remove: {ret:?}" + ); + + map.rebuild(); + + let ret = map.replace(&key, "should not replace", |_| true); + assert!( + matches!(ret, Replacement::NotFound), + "cannot replace after remove + rebuild: {ret:?}" + ); + + let at_version = map.get().expect("rebuilt").get_version(102); + assert!(at_version.is_none()); } diff --git a/pageserver/src/tenant/storage_layer.rs b/pageserver/src/tenant/storage_layer.rs index 48db364b8c..d719e19bf9 100644 --- a/pageserver/src/tenant/storage_layer.rs +++ b/pageserver/src/tenant/storage_layer.rs @@ -212,6 +212,10 @@ impl std::fmt::Debug for dyn Layer { } /// Holds metadata about a layer without any content. Used mostly for testing. +/// +/// To use filenames as fixtures, parse them as [`LayerFileName`] then convert from that to a +/// LayerDescriptor. +#[derive(Clone)] pub struct LayerDescriptor { pub key: Range, pub lsn: Range, @@ -251,6 +255,40 @@ impl Layer for LayerDescriptor { } } +impl From for LayerDescriptor { + fn from(value: DeltaFileName) -> Self { + let short_id = value.to_string(); + LayerDescriptor { + key: value.key_range, + lsn: value.lsn_range, + is_incremental: true, + short_id, + } + } +} + +impl From for LayerDescriptor { + fn from(value: ImageFileName) -> Self { + let short_id = value.to_string(); + let lsn = value.lsn_as_range(); + LayerDescriptor { + key: value.key_range, + lsn, + is_incremental: false, + short_id, + } + } +} + +impl From for LayerDescriptor { + fn from(value: LayerFileName) -> Self { + match value { + LayerFileName::Delta(d) => Self::from(d), + LayerFileName::Image(i) => Self::from(i), + } + } +} + /// Helper enum to hold a PageServerConf, or a path /// /// This is used by DeltaLayer and ImageLayer. Normally, this holds a reference to the diff --git a/pageserver/src/tenant/storage_layer/filename.rs b/pageserver/src/tenant/storage_layer/filename.rs index e77f2b25c1..bd3d2c42c1 100644 --- a/pageserver/src/tenant/storage_layer/filename.rs +++ b/pageserver/src/tenant/storage_layer/filename.rs @@ -128,6 +128,13 @@ impl Ord for ImageFileName { } } +impl ImageFileName { + pub fn lsn_as_range(&self) -> Range { + // Saves from having to copypaste this all over + self.lsn..(self.lsn + 1) + } +} + /// /// Represents the filename of an ImageLayer /// diff --git a/pageserver/src/tenant/storage_layer/remote_layer.rs b/pageserver/src/tenant/storage_layer/remote_layer.rs index 29cb4af0fa..8656ad0289 100644 --- a/pageserver/src/tenant/storage_layer/remote_layer.rs +++ b/pageserver/src/tenant/storage_layer/remote_layer.rs @@ -172,7 +172,7 @@ impl RemoteLayer { tenantid, timelineid, key_range: fname.key_range.clone(), - lsn_range: fname.lsn..(fname.lsn + 1), + lsn_range: fname.lsn_as_range(), is_delta: false, is_incremental: false, file_name: fname.to_owned().into(), diff --git a/pageserver/src/tenant/timeline.rs b/pageserver/src/tenant/timeline.rs index e80da302b7..fc6f7ad96c 100644 --- a/pageserver/src/tenant/timeline.rs +++ b/pageserver/src/tenant/timeline.rs @@ -3386,10 +3386,42 @@ impl Timeline { let mut layers = self_clone.layers.write().unwrap(); let mut updates = layers.batch_update(); { + use crate::tenant::layer_map::Replacement; let l: Arc = remote_layer.clone(); - updates.remove_historic(l); + match updates.replace_historic(&l, new_layer) { + Ok(Replacement::Replaced { .. }) => { /* expected */ } + Ok(Replacement::NotFound) => { + // TODO: the downloaded file should probably be removed, otherwise + // it will be added to the layermap on next load? we should + // probably restart any get_reconstruct_data search as well. + // + // See: https://github.com/neondatabase/neon/issues/3533 + error!("replacing downloaded layer into layermap failed because layer was not found"); + } + Ok(Replacement::RemovalBuffered) => { + unreachable!("current implementation does not remove anything") + } + Ok(Replacement::Unexpected(other)) => { + // if the other layer would have the same pointer value as + // expected, it means they differ only on vtables. + // + // otherwise there's no known reason for this to happen as + // compacted layers should have different covering rectangle + // leading to produce Replacement::NotFound. + + error!( + expected.ptr = ?Arc::as_ptr(&l), + other.ptr = ?Arc::as_ptr(&other), + "replacing downloaded layer into layermap failed because another layer was found instead of expected" + ); + } + Err(e) => { + // this is a precondition failure, the layer filename derived + // attributes didn't match up, which doesn't seem likely. + error!("replacing downloaded layer into layermap failed: {e:#?}") + } + } } - updates.insert_historic(new_layer); updates.flush(); drop(layers);