LCOV - code coverage report
Current view: top level - pageserver/src/tenant - storage_layer.rs (source / functions) Coverage Total Hit
Test: 07bee600374ccd486c69370d0972d9035964fe68.info Lines: 83.5 % 552 461
Test Date: 2025-02-20 13:11:02 Functions: 84.1 % 82 69

            Line data    Source code
       1              : //! Common traits and structs for layers
       2              : 
       3              : pub mod batch_split_writer;
       4              : pub mod delta_layer;
       5              : pub mod filter_iterator;
       6              : pub mod image_layer;
       7              : pub mod inmemory_layer;
       8              : pub(crate) mod layer;
       9              : mod layer_desc;
      10              : mod layer_name;
      11              : pub mod merge_iterator;
      12              : 
      13              : use crate::config::PageServerConf;
      14              : use crate::context::{AccessStatsBehavior, RequestContext};
      15              : use bytes::Bytes;
      16              : use futures::stream::FuturesUnordered;
      17              : use futures::StreamExt;
      18              : use pageserver_api::key::Key;
      19              : use pageserver_api::keyspace::{KeySpace, KeySpaceRandomAccum};
      20              : use pageserver_api::record::NeonWalRecord;
      21              : use pageserver_api::value::Value;
      22              : use std::cmp::Ordering;
      23              : use std::collections::hash_map::Entry;
      24              : use std::collections::{BinaryHeap, HashMap};
      25              : use std::future::Future;
      26              : use std::ops::Range;
      27              : use std::pin::Pin;
      28              : use std::sync::atomic::AtomicUsize;
      29              : use std::sync::Arc;
      30              : use std::time::{Duration, SystemTime, UNIX_EPOCH};
      31              : use tracing::{trace, Instrument};
      32              : use utils::sync::gate::GateGuard;
      33              : 
      34              : use utils::lsn::Lsn;
      35              : 
      36              : pub use batch_split_writer::{BatchLayerWriter, SplitDeltaLayerWriter, SplitImageLayerWriter};
      37              : pub use delta_layer::{DeltaLayer, DeltaLayerWriter, ValueRef};
      38              : pub use image_layer::{ImageLayer, ImageLayerWriter};
      39              : pub use inmemory_layer::InMemoryLayer;
      40              : pub use layer_desc::{PersistentLayerDesc, PersistentLayerKey};
      41              : pub use layer_name::{DeltaLayerName, ImageLayerName, LayerName};
      42              : 
      43              : pub(crate) use layer::{EvictionError, Layer, ResidentLayer};
      44              : 
      45              : use self::inmemory_layer::InMemoryLayerFileId;
      46              : 
      47              : use super::timeline::{GetVectoredError, ReadPath};
      48              : use super::PageReconstructError;
      49              : 
      50            0 : pub fn range_overlaps<T>(a: &Range<T>, b: &Range<T>) -> bool
      51            0 : where
      52            0 :     T: PartialOrd<T>,
      53            0 : {
      54            0 :     if a.start < b.start {
      55            0 :         a.end > b.start
      56              :     } else {
      57            0 :         b.end > a.start
      58              :     }
      59            0 : }
      60              : 
      61              : /// Struct used to communicate across calls to 'get_value_reconstruct_data'.
      62              : ///
      63              : /// Before first call, you can fill in 'page_img' if you have an older cached
      64              : /// version of the page available. That can save work in
      65              : /// 'get_value_reconstruct_data', as it can stop searching for page versions
      66              : /// when all the WAL records going back to the cached image have been collected.
      67              : ///
      68              : /// When get_value_reconstruct_data returns Complete, 'img' is set to an image
      69              : /// of the page, or the oldest WAL record in 'records' is a will_init-type
      70              : /// record that initializes the page without requiring a previous image.
      71              : ///
      72              : /// If 'get_page_reconstruct_data' returns Continue, some 'records' may have
      73              : /// been collected, but there are more records outside the current layer. Pass
      74              : /// the same ValueReconstructState struct in the next 'get_value_reconstruct_data'
      75              : /// call, to collect more records.
      76              : ///
      77              : #[derive(Debug, Default)]
      78              : pub(crate) struct ValueReconstructState {
      79              :     pub(crate) records: Vec<(Lsn, NeonWalRecord)>,
      80              :     pub(crate) img: Option<(Lsn, Bytes)>,
      81              : }
      82              : 
      83              : impl ValueReconstructState {
      84              :     /// Returns the number of page deltas applied to the page image.
      85      1335951 :     pub fn num_deltas(&self) -> usize {
      86      1335951 :         match self.img {
      87      1335863 :             Some(_) => self.records.len(),
      88           88 :             None => self.records.len() - 1, // omit will_init record
      89              :         }
      90      1335951 :     }
      91              : }
      92              : 
      93              : #[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
      94              : pub(crate) enum ValueReconstructSituation {
      95              :     Complete,
      96              :     #[default]
      97              :     Continue,
      98              : }
      99              : 
     100              : /// On disk representation of a value loaded in a buffer
     101              : #[derive(Debug)]
     102              : pub(crate) enum OnDiskValue {
     103              :     /// Unencoded [`Value::Image`]
     104              :     RawImage(Bytes),
     105              :     /// Encoded [`Value`]. Can deserialize into an image or a WAL record
     106              :     WalRecordOrImage(Bytes),
     107              : }
     108              : 
     109              : /// Reconstruct data accumulated for a single key during a vectored get
     110              : #[derive(Debug, Default)]
     111              : pub(crate) struct VectoredValueReconstructState {
     112              :     pub(crate) on_disk_values: Vec<(Lsn, OnDiskValueIoWaiter)>,
     113              : 
     114              :     pub(crate) situation: ValueReconstructSituation,
     115              : }
     116              : 
     117              : #[derive(Debug)]
     118              : pub(crate) struct OnDiskValueIoWaiter {
     119              :     rx: tokio::sync::oneshot::Receiver<OnDiskValueIoResult>,
     120              : }
     121              : 
     122              : #[derive(Debug)]
     123              : #[must_use]
     124              : pub(crate) enum OnDiskValueIo {
     125              :     /// Traversal identified this IO as required to complete the vectored get.
     126              :     Required {
     127              :         num_active_ios: Arc<AtomicUsize>,
     128              :         tx: tokio::sync::oneshot::Sender<OnDiskValueIoResult>,
     129              :     },
     130              :     /// Sparse keyspace reads always read all the values for a given key,
     131              :     /// even though only the first value is needed.
     132              :     ///
     133              :     /// This variant represents the unnecessary IOs for those values at lower LSNs
     134              :     /// that aren't needed, but are currently still being done.
     135              :     ///
     136              :     /// The execution of unnecessary IOs was a pre-existing behavior before concurrent IO.
     137              :     /// We added this explicit representation here so that we can drop
     138              :     /// unnecessary IO results immediately, instead of buffering them in
     139              :     /// `oneshot` channels inside [`VectoredValueReconstructState`] until
     140              :     /// [`VectoredValueReconstructState::collect_pending_ios`] gets called.
     141              :     Unnecessary,
     142              : }
     143              : 
     144              : type OnDiskValueIoResult = Result<OnDiskValue, std::io::Error>;
     145              : 
     146              : impl OnDiskValueIo {
     147      1482275 :     pub(crate) fn complete(self, res: OnDiskValueIoResult) {
     148      1482275 :         match self {
     149      1337405 :             OnDiskValueIo::Required { num_active_ios, tx } => {
     150      1337405 :                 num_active_ios.fetch_sub(1, std::sync::atomic::Ordering::Release);
     151      1337405 :                 let _ = tx.send(res);
     152      1337405 :             }
     153       144870 :             OnDiskValueIo::Unnecessary => {
     154       144870 :                 // Nobody cared, see variant doc comment.
     155       144870 :             }
     156              :         }
     157      1482275 :     }
     158              : }
     159              : 
     160              : #[derive(Debug, thiserror::Error)]
     161              : pub(crate) enum WaitCompletionError {
     162              :     #[error("OnDiskValueIo was dropped without completing, likely the sidecar task panicked")]
     163              :     IoDropped,
     164              : }
     165              : 
     166              : impl OnDiskValueIoWaiter {
     167      1337403 :     pub(crate) async fn wait_completion(self) -> Result<OnDiskValueIoResult, WaitCompletionError> {
     168      1337403 :         // NB: for Unnecessary IOs, this method never gets called because we don't add them to `on_disk_values`.
     169      1337403 :         self.rx.await.map_err(|_| WaitCompletionError::IoDropped)
     170      1337403 :     }
     171              : }
     172              : 
     173              : impl VectoredValueReconstructState {
     174              :     /// # Cancel-Safety
     175              :     ///
     176              :     /// Technically fine to stop polling this future, but, the IOs will still
     177              :     /// be executed to completion by the sidecar task and hold on to / consume resources.
     178              :     /// Better not do it to make reasonsing about the system easier.
     179      1336091 :     pub(crate) async fn collect_pending_ios(
     180      1336091 :         self,
     181      1336091 :     ) -> Result<ValueReconstructState, PageReconstructError> {
     182              :         use utils::bin_ser::BeSer;
     183              : 
     184      1336091 :         let mut res = Ok(ValueReconstructState::default());
     185              : 
     186              :         // We should try hard not to bail early, so that by the time we return from this
     187              :         // function, all IO for this value is done. It's not required -- we could totally
     188              :         // stop polling the IO futures in the sidecar task, they need to support that,
     189              :         // but just stopping to poll doesn't reduce the IO load on the disk. It's easier
     190              :         // to reason about the system if we just wait for all IO to complete, even if
     191              :         // we're no longer interested in the result.
     192              :         //
     193              :         // Revisit this when IO futures are replaced with a more sophisticated IO system
     194              :         // and an IO scheduler, where we know which IOs were submitted and which ones
     195              :         // just queued. Cf the comment on IoConcurrency::spawn_io.
     196      2673494 :         for (lsn, waiter) in self.on_disk_values {
     197      1337403 :             let value_recv_res = waiter
     198      1337403 :                 .wait_completion()
     199      1337403 :                 // we rely on the caller to poll us to completion, so this is not a bail point
     200      1337403 :                 .await;
     201              :             // Force not bailing early by wrapping the code into a closure.
     202              :             #[allow(clippy::redundant_closure_call)]
     203      1337403 :             let _: () = (|| {
     204      1337403 :                 match (&mut res, value_recv_res) {
     205            0 :                     (Err(_), _) => {
     206            0 :                         // We've already failed, no need to process more.
     207            0 :                     }
     208            0 :                     (Ok(_), Err(wait_err)) => {
     209            0 :                         // This shouldn't happen - likely the sidecar task panicked.
     210            0 :                         res = Err(PageReconstructError::Other(wait_err.into()));
     211            0 :                     }
     212            0 :                     (Ok(_), Ok(Err(err))) => {
     213            0 :                         let err: std::io::Error = err;
     214            0 :                         // TODO: returning IO error here will fail a compute query.
     215            0 :                         // Probably not what we want, we're not doing `maybe_fatal_err`
     216            0 :                         // in the IO futures.
     217            0 :                         // But it's been like that for a long time, not changing it
     218            0 :                         // as part of concurrent IO.
     219            0 :                         // => https://github.com/neondatabase/neon/issues/10454
     220            0 :                         res = Err(PageReconstructError::Other(err.into()));
     221            0 :                     }
     222        38497 :                     (Ok(ok), Ok(Ok(OnDiskValue::RawImage(img)))) => {
     223        38497 :                         assert!(ok.img.is_none());
     224        38497 :                         ok.img = Some((lsn, img));
     225              :                     }
     226      1298906 :                     (Ok(ok), Ok(Ok(OnDiskValue::WalRecordOrImage(buf)))) => {
     227      1298906 :                         match Value::des(&buf) {
     228         1400 :                             Ok(Value::WalRecord(rec)) => {
     229         1400 :                                 ok.records.push((lsn, rec));
     230         1400 :                             }
     231      1297506 :                             Ok(Value::Image(img)) => {
     232      1297506 :                                 assert!(ok.img.is_none());
     233      1297506 :                                 ok.img = Some((lsn, img));
     234              :                             }
     235            0 :                             Err(err) => {
     236            0 :                                 res = Err(PageReconstructError::Other(err.into()));
     237            0 :                             }
     238              :                         }
     239              :                     }
     240              :                 }
     241      1337403 :             })();
     242      1337403 :         }
     243              : 
     244      1336091 :         res
     245      1336091 :     }
     246              : }
     247              : 
     248              : /// Bag of data accumulated during a vectored get..
     249              : pub(crate) struct ValuesReconstructState {
     250              :     /// The keys will be removed after `get_vectored` completes. The caller outside `Timeline`
     251              :     /// should not expect to get anything from this hashmap.
     252              :     pub(crate) keys: HashMap<Key, VectoredValueReconstructState>,
     253              :     /// The keys which are already retrieved
     254              :     keys_done: KeySpaceRandomAccum,
     255              : 
     256              :     /// The keys covered by the image layers
     257              :     keys_with_image_coverage: Option<Range<Key>>,
     258              : 
     259              :     // Statistics that are still accessible as a caller of `get_vectored_impl`.
     260              :     layers_visited: u32,
     261              :     delta_layers_visited: u32,
     262              : 
     263              :     pub(crate) io_concurrency: IoConcurrency,
     264              :     num_active_ios: Arc<AtomicUsize>,
     265              : 
     266              :     pub(crate) read_path: Option<ReadPath>,
     267              : }
     268              : 
     269              : /// The level of IO concurrency to be used on the read path
     270              : ///
     271              : /// The desired end state is that we always do parallel IO.
     272              : /// This struct and the dispatching in the impl will be removed once
     273              : /// we've built enough confidence.
     274              : pub(crate) enum IoConcurrency {
     275              :     Sequential,
     276              :     SidecarTask {
     277              :         task_id: usize,
     278              :         ios_tx: tokio::sync::mpsc::UnboundedSender<IoFuture>,
     279              :     },
     280              : }
     281              : 
     282              : type IoFuture = Pin<Box<dyn Send + Future<Output = ()>>>;
     283              : 
     284              : pub(crate) enum SelectedIoConcurrency {
     285              :     Sequential,
     286              :     SidecarTask(GateGuard),
     287              : }
     288              : 
     289              : impl std::fmt::Debug for IoConcurrency {
     290            0 :     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
     291            0 :         match self {
     292            0 :             IoConcurrency::Sequential => write!(f, "Sequential"),
     293            0 :             IoConcurrency::SidecarTask { .. } => write!(f, "SidecarTask"),
     294              :         }
     295            0 :     }
     296              : }
     297              : 
     298              : impl std::fmt::Debug for SelectedIoConcurrency {
     299           64 :     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
     300           64 :         match self {
     301           32 :             SelectedIoConcurrency::Sequential => write!(f, "Sequential"),
     302           32 :             SelectedIoConcurrency::SidecarTask(_) => write!(f, "SidecarTask"),
     303              :         }
     304           64 :     }
     305              : }
     306              : 
     307              : impl IoConcurrency {
     308              :     /// Force sequential IO. This is a temporary workaround until we have
     309              :     /// moved plumbing-through-the-call-stack
     310              :     /// of IoConcurrency into `RequestContextq.
     311              :     ///
     312              :     /// DO NOT USE for new code.
     313              :     ///
     314              :     /// Tracking issue: <https://github.com/neondatabase/neon/issues/10460>.
     315      1215203 :     pub(crate) fn sequential() -> Self {
     316      1215203 :         Self::spawn(SelectedIoConcurrency::Sequential)
     317      1215203 :     }
     318              : 
     319         1020 :     pub(crate) fn spawn_from_conf(
     320         1020 :         conf: &'static PageServerConf,
     321         1020 :         gate_guard: GateGuard,
     322         1020 :     ) -> IoConcurrency {
     323              :         use pageserver_api::config::GetVectoredConcurrentIo;
     324         1020 :         let selected = match conf.get_vectored_concurrent_io {
     325         1020 :             GetVectoredConcurrentIo::Sequential => SelectedIoConcurrency::Sequential,
     326            0 :             GetVectoredConcurrentIo::SidecarTask => SelectedIoConcurrency::SidecarTask(gate_guard),
     327              :         };
     328         1020 :         Self::spawn(selected)
     329         1020 :     }
     330              : 
     331      1216287 :     pub(crate) fn spawn(io_concurrency: SelectedIoConcurrency) -> Self {
     332      1216287 :         match io_concurrency {
     333      1216255 :             SelectedIoConcurrency::Sequential => IoConcurrency::Sequential,
     334           32 :             SelectedIoConcurrency::SidecarTask(gate_guard) => {
     335           32 :                 let (ios_tx, ios_rx) = tokio::sync::mpsc::unbounded_channel();
     336              :                 static TASK_ID: AtomicUsize = AtomicUsize::new(0);
     337           32 :                 let task_id = TASK_ID.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
     338              :                 // TODO: enrich the span with more context (tenant,shard,timeline) + (basebackup|pagestream|...)
     339           32 :                 let span =
     340           32 :                     tracing::info_span!(parent: None, "IoConcurrency_sidecar", task_id = task_id);
     341           32 :                 trace!(task_id, "spawning sidecar task");
     342           32 :                 tokio::spawn(async move {
     343           32 :                     trace!("start");
     344           32 :                     scopeguard::defer!{ trace!("end") };
     345              :                     type IosRx = tokio::sync::mpsc::UnboundedReceiver<IoFuture>;
     346              :                     enum State {
     347              :                         Waiting {
     348              :                             // invariant: is_empty(), but we recycle the allocation
     349              :                             empty_futures: FuturesUnordered<IoFuture>,
     350              :                             ios_rx: IosRx,
     351              :                         },
     352              :                         Executing {
     353              :                             futures: FuturesUnordered<IoFuture>,
     354              :                             ios_rx: IosRx,
     355              :                         },
     356              :                         ShuttingDown {
     357              :                             futures: FuturesUnordered<IoFuture>,
     358              :                         },
     359              :                     }
     360           32 :                     let mut state = State::Waiting {
     361           32 :                         empty_futures: FuturesUnordered::new(),
     362           32 :                         ios_rx,
     363           32 :                     };
     364              :                     loop {
     365        39092 :                         match state {
     366              :                             State::Waiting {
     367        18551 :                                 empty_futures,
     368        18551 :                                 mut ios_rx,
     369        18551 :                             } => {
     370        18551 :                                 assert!(empty_futures.is_empty());
     371        18551 :                                 tokio::select! {
     372        18551 :                                     fut = ios_rx.recv() => {
     373        18519 :                                         if let Some(fut) = fut {
     374        18519 :                                             trace!("received new io future");
     375        18519 :                                             empty_futures.push(fut);
     376        18519 :                                             state = State::Executing { futures: empty_futures, ios_rx };
     377              :                                         } else {
     378            0 :                                             state = State::ShuttingDown { futures: empty_futures }
     379              :                                         }
     380              :                                     }
     381              :                                 }
     382              :                             }
     383              :                             State::Executing {
     384        20541 :                                 mut futures,
     385        20541 :                                 mut ios_rx,
     386        20541 :                             } => {
     387        20541 :                                 tokio::select! {
     388        20541 :                                     res = futures.next() => {
     389        19530 :                                         trace!("io future completed");
     390        19530 :                                         assert!(res.is_some());
     391        19530 :                                         if futures.is_empty() {
     392        18519 :                                             state = State::Waiting { empty_futures: futures, ios_rx};
     393        18519 :                                         } else {
     394         1011 :                                             state = State::Executing { futures, ios_rx };
     395         1011 :                                         }
     396              :                                     }
     397        20541 :                                     fut = ios_rx.recv() => {
     398         1011 :                                         if let Some(fut) = fut {
     399         1011 :                                             trace!("received new io future");
     400         1011 :                                             futures.push(fut);
     401         1011 :                                             state =  State::Executing { futures, ios_rx};
     402            0 :                                         } else {
     403            0 :                                             state = State::ShuttingDown { futures };
     404            0 :                                         }
     405              :                                     }
     406              :                                 }
     407              :                             }
     408              :                             State::ShuttingDown {
     409            0 :                                 mut futures,
     410            0 :                             } => {
     411            0 :                                 trace!("shutting down");
     412            0 :                                 while let Some(()) = futures.next().await {
     413            0 :                                     trace!("io future completed (shutdown)");
     414              :                                     // drain
     415              :                                 }
     416            0 :                                 trace!("shutdown complete");
     417            0 :                                 break;
     418            0 :                             }
     419            0 :                         }
     420            0 :                     }
     421            0 :                     drop(gate_guard); // drop it right before we exit
     422           32 :                 }.instrument(span));
     423           32 :                 IoConcurrency::SidecarTask { task_id, ios_tx }
     424              :             }
     425              :         }
     426      1216287 :     }
     427              : 
     428        76362 :     pub(crate) fn clone(&self) -> Self {
     429        76362 :         match self {
     430        39484 :             IoConcurrency::Sequential => IoConcurrency::Sequential,
     431        36878 :             IoConcurrency::SidecarTask { task_id, ios_tx } => IoConcurrency::SidecarTask {
     432        36878 :                 task_id: *task_id,
     433        36878 :                 ios_tx: ios_tx.clone(),
     434        36878 :             },
     435              :         }
     436        76362 :     }
     437              : 
     438              :     /// Submit an IO to be executed in the background. DEADLOCK RISK, read the full doc string.
     439              :     ///
     440              :     /// The IO is represented as an opaque future.
     441              :     /// IO completion must be handled inside the future, e.g., through a oneshot channel.
     442              :     ///
     443              :     /// The API seems simple but there are multiple **pitfalls** involving
     444              :     /// DEADLOCK RISK.
     445              :     ///
     446              :     /// First, there are no guarantees about the exexecution of the IO.
     447              :     /// It may be `await`ed in-place before this function returns.
     448              :     /// It may be polled partially by this task and handed off to another task to be finished.
     449              :     /// It may be polled and then dropped before returning ready.
     450              :     ///
     451              :     /// This means that submitted IOs must not be interedependent.
     452              :     /// Interdependence may be through shared limited resources, e.g.,
     453              :     /// - VirtualFile file descriptor cache slot acquisition
     454              :     /// - tokio-epoll-uring slot
     455              :     ///
     456              :     /// # Why current usage is safe from deadlocks
     457              :     ///
     458              :     /// Textbook condition for a deadlock is that _all_ of the following be given
     459              :     /// - Mutual exclusion
     460              :     /// - Hold and wait
     461              :     /// - No preemption
     462              :     /// - Circular wait
     463              :     ///
     464              :     /// The current usage is safe because:
     465              :     /// - Mutual exclusion: IO futures definitely use mutexes, no way around that for now
     466              :     /// - Hold and wait: IO futures currently hold two kinds of locks/resources while waiting
     467              :     ///   for acquisition of other resources:
     468              :     ///    - VirtualFile file descriptor cache slot tokio mutex
     469              :     ///    - tokio-epoll-uring slot (uses tokio notify => wait queue, much like mutex)
     470              :     /// - No preemption: there's no taking-away of acquired locks/resources => given
     471              :     /// - Circular wait: this is the part of the condition that isn't met: all IO futures
     472              :     ///   first acquire VirtualFile mutex, then tokio-epoll-uring slot.
     473              :     ///   There is no IO future that acquires slot before VirtualFile.
     474              :     ///   Hence there can be no circular waiting.
     475              :     ///   Hence there cannot be a deadlock.
     476              :     ///
     477              :     /// This is a very fragile situation and must be revisited whenver any code called from
     478              :     /// inside the IO futures is changed.
     479              :     ///
     480              :     /// We will move away from opaque IO futures towards well-defined IOs at some point in
     481              :     /// the future when we have shipped this first version of concurrent IO to production
     482              :     /// and are ready to retire the Sequential mode which runs the futures in place.
     483              :     /// Right now, while brittle, the opaque IO approach allows us to ship the feature
     484              :     /// with minimal changes to the code and minimal changes to existing behavior in Sequential mode.
     485              :     ///
     486              :     /// Also read the comment in `collect_pending_ios`.
     487      1525814 :     pub(crate) async fn spawn_io<F>(&mut self, fut: F)
     488      1525814 :     where
     489      1525814 :         F: std::future::Future<Output = ()> + Send + 'static,
     490      1525814 :     {
     491      1525814 :         match self {
     492      1506284 :             IoConcurrency::Sequential => fut.await,
     493        19530 :             IoConcurrency::SidecarTask { ios_tx, .. } => {
     494        19530 :                 let fut = Box::pin(fut);
     495        19530 :                 // NB: experiments showed that doing an opportunistic poll of `fut` here was bad for throughput
     496        19530 :                 // while insignificant for latency.
     497        19530 :                 // It would make sense to revisit the tokio-epoll-uring API in the future such that we can try
     498        19530 :                 // a submission here, but never poll the future. That way, io_uring can make proccess while
     499        19530 :                 // the future sits in the ios_tx queue.
     500        19530 :                 match ios_tx.send(fut) {
     501        19530 :                     Ok(()) => {}
     502              :                     Err(_) => {
     503            0 :                         unreachable!("the io task must have exited, likely it panicked")
     504              :                     }
     505              :                 }
     506              :             }
     507              :         }
     508      1525814 :     }
     509              : 
     510              :     #[cfg(test)]
     511           64 :     pub(crate) fn spawn_for_test() -> impl std::ops::DerefMut<Target = Self> {
     512              :         use std::ops::{Deref, DerefMut};
     513              :         use tracing::info;
     514              :         use utils::sync::gate::Gate;
     515              : 
     516              :         // Spawn needs a Gate, give it one.
     517              :         struct Wrapper {
     518              :             inner: IoConcurrency,
     519              :             #[allow(dead_code)]
     520              :             gate: Box<Gate>,
     521              :         }
     522              :         impl Deref for Wrapper {
     523              :             type Target = IoConcurrency;
     524              : 
     525        36974 :             fn deref(&self) -> &Self::Target {
     526        36974 :                 &self.inner
     527        36974 :             }
     528              :         }
     529              :         impl DerefMut for Wrapper {
     530            0 :             fn deref_mut(&mut self) -> &mut Self::Target {
     531            0 :                 &mut self.inner
     532            0 :             }
     533              :         }
     534           64 :         let gate = Box::new(Gate::default());
     535              : 
     536              :         // The default behavior when running Rust unit tests without any further
     537              :         // flags is to use the new behavior.
     538              :         // The CI uses the following environment variable to unit test both old
     539              :         // and new behavior.
     540              :         // NB: the Python regression & perf tests take the `else` branch
     541              :         // below and have their own defaults management.
     542           64 :         let selected = {
     543              :             // The pageserver_api::config type is unsuitable because it's internally tagged.
     544           64 :             #[derive(serde::Deserialize)]
     545              :             #[serde(rename_all = "kebab-case")]
     546              :             enum TestOverride {
     547              :                 Sequential,
     548              :                 SidecarTask,
     549              :             }
     550              :             use once_cell::sync::Lazy;
     551           64 :             static TEST_OVERRIDE: Lazy<TestOverride> = Lazy::new(|| {
     552           64 :                 utils::env::var_serde_json_string(
     553           64 :                     "NEON_PAGESERVER_UNIT_TEST_GET_VECTORED_CONCURRENT_IO",
     554           64 :                 )
     555           64 :                 .unwrap_or(TestOverride::SidecarTask)
     556           64 :             });
     557              : 
     558           64 :             match *TEST_OVERRIDE {
     559           32 :                 TestOverride::Sequential => SelectedIoConcurrency::Sequential,
     560              :                 TestOverride::SidecarTask => {
     561           32 :                     SelectedIoConcurrency::SidecarTask(gate.enter().expect("just created it"))
     562              :                 }
     563              :             }
     564              :         };
     565              : 
     566           64 :         info!(?selected, "get_vectored_concurrent_io test");
     567              : 
     568           64 :         Wrapper {
     569           64 :             inner: Self::spawn(selected),
     570           64 :             gate,
     571           64 :         }
     572           64 :     }
     573              : }
     574              : 
     575              : /// Make noise in case the [`ValuesReconstructState`] gets dropped while
     576              : /// there are still IOs in flight.
     577              : /// Refer to `collect_pending_ios` for why we prefer not to do that.
     578              : //
     579              : /// We log from here instead of from the sidecar task because the [`ValuesReconstructState`]
     580              : /// gets dropped in a tracing span with more context.
     581              : /// We repeat the sidecar tasks's `task_id` so we can correlate what we emit here with
     582              : /// the logs / panic handler logs from the sidecar task, which also logs the `task_id`.
     583              : impl Drop for ValuesReconstructState {
     584      1255341 :     fn drop(&mut self) {
     585      1255341 :         let num_active_ios = self
     586      1255341 :             .num_active_ios
     587      1255341 :             .load(std::sync::atomic::Ordering::Acquire);
     588      1255341 :         if num_active_ios == 0 {
     589      1255339 :             return;
     590            2 :         }
     591            2 :         let sidecar_task_id = match &self.io_concurrency {
     592            0 :             IoConcurrency::Sequential => None,
     593            2 :             IoConcurrency::SidecarTask { task_id, .. } => Some(*task_id),
     594              :         };
     595            2 :         tracing::warn!(
     596              :             num_active_ios,
     597              :             ?sidecar_task_id,
     598            0 :             backtrace=%std::backtrace::Backtrace::force_capture(),
     599            0 :             "dropping ValuesReconstructState while some IOs have not been completed",
     600              :         );
     601      1255341 :     }
     602              : }
     603              : 
     604              : impl ValuesReconstructState {
     605      1255341 :     pub(crate) fn new(io_concurrency: IoConcurrency) -> Self {
     606      1255341 :         Self {
     607      1255341 :             keys: HashMap::new(),
     608      1255341 :             keys_done: KeySpaceRandomAccum::new(),
     609      1255341 :             keys_with_image_coverage: None,
     610      1255341 :             layers_visited: 0,
     611      1255341 :             delta_layers_visited: 0,
     612      1255341 :             io_concurrency,
     613      1255341 :             num_active_ios: Arc::new(AtomicUsize::new(0)),
     614      1255341 :             read_path: None,
     615      1255341 :         }
     616      1255341 :     }
     617              : 
     618              :     /// Absolutely read [`IoConcurrency::spawn_io`] to learn about assumptions & pitfalls.
     619      1525814 :     pub(crate) async fn spawn_io<F>(&mut self, fut: F)
     620      1525814 :     where
     621      1525814 :         F: std::future::Future<Output = ()> + Send + 'static,
     622      1525814 :     {
     623      1525814 :         self.io_concurrency.spawn_io(fut).await;
     624      1525814 :     }
     625              : 
     626      1692763 :     pub(crate) fn on_layer_visited(&mut self, layer: &ReadableLayer) {
     627      1692763 :         self.layers_visited += 1;
     628      1692763 :         if let ReadableLayer::PersistentLayer(layer) = layer {
     629       479416 :             if layer.layer_desc().is_delta() {
     630       434324 :                 self.delta_layers_visited += 1;
     631       434324 :             }
     632      1213347 :         }
     633      1692763 :     }
     634              : 
     635          460 :     pub(crate) fn get_delta_layers_visited(&self) -> u32 {
     636          460 :         self.delta_layers_visited
     637          460 :     }
     638              : 
     639      1255283 :     pub(crate) fn get_layers_visited(&self) -> u32 {
     640      1255283 :         self.layers_visited
     641      1255283 :     }
     642              : 
     643              :     /// On hitting image layer, we can mark all keys in this range as done, because
     644              :     /// if the image layer does not contain a key, it is deleted/never added.
     645        45116 :     pub(crate) fn on_image_layer_visited(&mut self, key_range: &Range<Key>) {
     646        45116 :         let prev_val = self.keys_with_image_coverage.replace(key_range.clone());
     647        45116 :         assert_eq!(
     648              :             prev_val, None,
     649            0 :             "should consume the keyspace before the next iteration"
     650              :         );
     651        45116 :     }
     652              : 
     653              :     /// Update the state collected for a given key.
     654              :     /// Returns true if this was the last value needed for the key and false otherwise.
     655              :     ///
     656              :     /// If the key is done after the update, mark it as such.
     657              :     ///
     658              :     /// If the key is in the sparse keyspace (i.e., aux files), we do not track them in
     659              :     /// `key_done`.
     660              :     // TODO: rename this method & update description.
     661      1482275 :     pub(crate) fn update_key(&mut self, key: &Key, lsn: Lsn, completes: bool) -> OnDiskValueIo {
     662      1482275 :         let state = self.keys.entry(*key).or_default();
     663      1482275 : 
     664      1482275 :         let is_sparse_key = key.is_sparse();
     665              : 
     666      1482275 :         let required_io = match state.situation {
     667              :             ValueReconstructSituation::Complete => {
     668       144870 :                 if is_sparse_key {
     669              :                     // Sparse keyspace might be visited multiple times because
     670              :                     // we don't track unmapped keyspaces.
     671       144870 :                     return OnDiskValueIo::Unnecessary;
     672              :                 } else {
     673            0 :                     unreachable!()
     674              :                 }
     675              :             }
     676              :             ValueReconstructSituation::Continue => {
     677      1337405 :                 self.num_active_ios
     678      1337405 :                     .fetch_add(1, std::sync::atomic::Ordering::Release);
     679      1337405 :                 let (tx, rx) = tokio::sync::oneshot::channel();
     680      1337405 :                 state.on_disk_values.push((lsn, OnDiskValueIoWaiter { rx }));
     681      1337405 :                 OnDiskValueIo::Required {
     682      1337405 :                     tx,
     683      1337405 :                     num_active_ios: Arc::clone(&self.num_active_ios),
     684      1337405 :                 }
     685      1337405 :             }
     686      1337405 :         };
     687      1337405 : 
     688      1337405 :         if completes && state.situation == ValueReconstructSituation::Continue {
     689      1336093 :             state.situation = ValueReconstructSituation::Complete;
     690      1336093 :             if !is_sparse_key {
     691      1208577 :                 self.keys_done.add_key(*key);
     692      1208577 :             }
     693         1312 :         }
     694              : 
     695      1337405 :         required_io
     696      1482275 :     }
     697              : 
     698              :     /// Returns the key space describing the keys that have
     699              :     /// been marked as completed since the last call to this function.
     700              :     /// Returns individual keys done, and the image layer coverage.
     701      3400578 :     pub(crate) fn consume_done_keys(&mut self) -> (KeySpace, Option<Range<Key>>) {
     702      3400578 :         (
     703      3400578 :             self.keys_done.consume_keyspace(),
     704      3400578 :             self.keys_with_image_coverage.take(),
     705      3400578 :         )
     706      3400578 :     }
     707              : }
     708              : 
     709              : /// A key that uniquely identifies a layer in a timeline
     710              : #[derive(Debug, PartialEq, Eq, Clone, Hash)]
     711              : pub(crate) enum LayerId {
     712              :     PersitentLayerId(PersistentLayerKey),
     713              :     InMemoryLayerId(InMemoryLayerFileId),
     714              : }
     715              : 
     716              : /// Uniquely identify a layer visit by the layer
     717              : /// and LSN floor (or start LSN) of the reads.
     718              : /// The layer itself is not enough since we may
     719              : /// have different LSN lower bounds for delta layer reads.
     720              : #[derive(Debug, PartialEq, Eq, Clone, Hash)]
     721              : struct LayerToVisitId {
     722              :     layer_id: LayerId,
     723              :     lsn_floor: Lsn,
     724              : }
     725              : 
     726              : /// Layer wrapper for the read path. Note that it is valid
     727              : /// to use these layers even after external operations have
     728              : /// been performed on them (compaction, freeze, etc.).
     729              : #[derive(Debug)]
     730              : pub(crate) enum ReadableLayer {
     731              :     PersistentLayer(Layer),
     732              :     InMemoryLayer(Arc<InMemoryLayer>),
     733              : }
     734              : 
     735              : /// A partial description of a read to be done.
     736              : #[derive(Debug, Clone)]
     737              : struct LayerVisit {
     738              :     /// An id used to resolve the readable layer within the fringe
     739              :     layer_to_visit_id: LayerToVisitId,
     740              :     /// Lsn range for the read, used for selecting the next read
     741              :     lsn_range: Range<Lsn>,
     742              : }
     743              : 
     744              : /// Data structure which maintains a fringe of layers for the
     745              : /// read path. The fringe is the set of layers which intersects
     746              : /// the current keyspace that the search is descending on.
     747              : /// Each layer tracks the keyspace that intersects it.
     748              : ///
     749              : /// The fringe must appear sorted by Lsn. Hence, it uses
     750              : /// a two layer indexing scheme.
     751              : #[derive(Debug)]
     752              : pub(crate) struct LayerFringe {
     753              :     planned_visits_by_lsn: BinaryHeap<LayerVisit>,
     754              :     visit_reads: HashMap<LayerToVisitId, LayerVisitReads>,
     755              : }
     756              : 
     757              : #[derive(Debug)]
     758              : struct LayerVisitReads {
     759              :     layer: ReadableLayer,
     760              :     target_keyspace: KeySpaceRandomAccum,
     761              : }
     762              : 
     763              : impl LayerFringe {
     764      1707815 :     pub(crate) fn new() -> Self {
     765      1707815 :         LayerFringe {
     766      1707815 :             planned_visits_by_lsn: BinaryHeap::new(),
     767      1707815 :             visit_reads: HashMap::new(),
     768      1707815 :         }
     769      1707815 :     }
     770              : 
     771      3400578 :     pub(crate) fn next_layer(&mut self) -> Option<(ReadableLayer, KeySpace, Range<Lsn>)> {
     772      3400578 :         let read_desc = self.planned_visits_by_lsn.pop()?;
     773              : 
     774      1692763 :         let removed = self.visit_reads.remove_entry(&read_desc.layer_to_visit_id);
     775      1692763 : 
     776      1692763 :         match removed {
     777              :             Some((
     778              :                 _,
     779              :                 LayerVisitReads {
     780      1692763 :                     layer,
     781      1692763 :                     mut target_keyspace,
     782      1692763 :                 },
     783      1692763 :             )) => Some((
     784      1692763 :                 layer,
     785      1692763 :                 target_keyspace.consume_keyspace(),
     786      1692763 :                 read_desc.lsn_range,
     787      1692763 :             )),
     788            0 :             None => unreachable!("fringe internals are always consistent"),
     789              :         }
     790      3400578 :     }
     791              : 
     792      1692851 :     pub(crate) fn update(
     793      1692851 :         &mut self,
     794      1692851 :         layer: ReadableLayer,
     795      1692851 :         keyspace: KeySpace,
     796      1692851 :         lsn_range: Range<Lsn>,
     797      1692851 :     ) {
     798      1692851 :         let layer_to_visit_id = LayerToVisitId {
     799      1692851 :             layer_id: layer.id(),
     800      1692851 :             lsn_floor: lsn_range.start,
     801      1692851 :         };
     802      1692851 : 
     803      1692851 :         let entry = self.visit_reads.entry(layer_to_visit_id.clone());
     804      1692851 :         match entry {
     805           88 :             Entry::Occupied(mut entry) => {
     806           88 :                 entry.get_mut().target_keyspace.add_keyspace(keyspace);
     807           88 :             }
     808      1692763 :             Entry::Vacant(entry) => {
     809      1692763 :                 self.planned_visits_by_lsn.push(LayerVisit {
     810      1692763 :                     lsn_range,
     811      1692763 :                     layer_to_visit_id: layer_to_visit_id.clone(),
     812      1692763 :                 });
     813      1692763 :                 let mut accum = KeySpaceRandomAccum::new();
     814      1692763 :                 accum.add_keyspace(keyspace);
     815      1692763 :                 entry.insert(LayerVisitReads {
     816      1692763 :                     layer,
     817      1692763 :                     target_keyspace: accum,
     818      1692763 :                 });
     819      1692763 :             }
     820              :         }
     821      1692851 :     }
     822              : }
     823              : 
     824              : impl Default for LayerFringe {
     825            0 :     fn default() -> Self {
     826            0 :         Self::new()
     827            0 :     }
     828              : }
     829              : 
     830              : impl Ord for LayerVisit {
     831           60 :     fn cmp(&self, other: &Self) -> Ordering {
     832           60 :         let ord = self.lsn_range.end.cmp(&other.lsn_range.end);
     833           60 :         if ord == std::cmp::Ordering::Equal {
     834           44 :             self.lsn_range.start.cmp(&other.lsn_range.start).reverse()
     835              :         } else {
     836           16 :             ord
     837              :         }
     838           60 :     }
     839              : }
     840              : 
     841              : impl PartialOrd for LayerVisit {
     842           60 :     fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
     843           60 :         Some(self.cmp(other))
     844           60 :     }
     845              : }
     846              : 
     847              : impl PartialEq for LayerVisit {
     848            0 :     fn eq(&self, other: &Self) -> bool {
     849            0 :         self.lsn_range == other.lsn_range
     850            0 :     }
     851              : }
     852              : 
     853              : impl Eq for LayerVisit {}
     854              : 
     855              : impl ReadableLayer {
     856      1692851 :     pub(crate) fn id(&self) -> LayerId {
     857      1692851 :         match self {
     858       479504 :             Self::PersistentLayer(layer) => LayerId::PersitentLayerId(layer.layer_desc().key()),
     859      1213347 :             Self::InMemoryLayer(layer) => LayerId::InMemoryLayerId(layer.file_id()),
     860              :         }
     861      1692851 :     }
     862              : 
     863      1692763 :     pub(crate) async fn get_values_reconstruct_data(
     864      1692763 :         &self,
     865      1692763 :         keyspace: KeySpace,
     866      1692763 :         lsn_range: Range<Lsn>,
     867      1692763 :         reconstruct_state: &mut ValuesReconstructState,
     868      1692763 :         ctx: &RequestContext,
     869      1692763 :     ) -> Result<(), GetVectoredError> {
     870      1692763 :         match self {
     871       479416 :             ReadableLayer::PersistentLayer(layer) => {
     872       479416 :                 layer
     873       479416 :                     .get_values_reconstruct_data(keyspace, lsn_range, reconstruct_state, ctx)
     874       479416 :                     .await
     875              :             }
     876      1213347 :             ReadableLayer::InMemoryLayer(layer) => {
     877      1213347 :                 layer
     878      1213347 :                     .get_values_reconstruct_data(keyspace, lsn_range.end, reconstruct_state, ctx)
     879      1213347 :                     .await
     880              :             }
     881              :         }
     882      1692763 :     }
     883              : }
     884              : 
     885              : /// Layers contain a hint indicating whether they are likely to be used for reads.
     886              : ///
     887              : /// This is a hint rather than an authoritative value, so that we do not have to update it synchronously
     888              : /// when changing the visibility of layers (for example when creating a branch that makes some previously
     889              : /// covered layers visible).  It should be used for cache management but not for correctness-critical checks.
     890              : #[derive(Debug, Clone, PartialEq, Eq)]
     891              : pub enum LayerVisibilityHint {
     892              :     /// A Visible layer might be read while serving a read, because there is not an image layer between it
     893              :     /// and a readable LSN (the tip of the branch or a child's branch point)
     894              :     Visible,
     895              :     /// A Covered layer probably won't be read right now, but _can_ be read in future if someone creates
     896              :     /// a branch or ephemeral endpoint at an LSN below the layer that covers this.
     897              :     Covered,
     898              : }
     899              : 
     900              : pub(crate) struct LayerAccessStats(std::sync::atomic::AtomicU64);
     901              : 
     902            0 : #[derive(Clone, Copy, strum_macros::EnumString)]
     903              : pub(crate) enum LayerAccessStatsReset {
     904              :     NoReset,
     905              :     AllStats,
     906              : }
     907              : 
     908              : impl Default for LayerAccessStats {
     909         3848 :     fn default() -> Self {
     910         3848 :         // Default value is to assume resident since creation time, and visible.
     911         3848 :         let (_mask, mut value) = Self::to_low_res_timestamp(Self::RTIME_SHIFT, SystemTime::now());
     912         3848 :         value |= 0x1 << Self::VISIBILITY_SHIFT;
     913         3848 : 
     914         3848 :         Self(std::sync::atomic::AtomicU64::new(value))
     915         3848 :     }
     916              : }
     917              : 
     918              : // Efficient store of two very-low-resolution timestamps and some bits.  Used for storing last access time and
     919              : // last residence change time.
     920              : impl LayerAccessStats {
     921              :     // How many high bits to drop from a u32 timestamp?
     922              :     // - Only storing up to a u32 timestamp will work fine until 2038 (if this code is still in use
     923              :     //   after that, this software has been very successful!)
     924              :     // - Dropping the top bit is implicitly safe because unix timestamps are meant to be
     925              :     // stored in an i32, so they never used it.
     926              :     // - Dropping the next two bits is safe because this code is only running on systems in
     927              :     // years >= 2024, and these bits have been 1 since 2021
     928              :     //
     929              :     // Therefore we may store only 28 bits for a timestamp with one second resolution.  We do
     930              :     // this truncation to make space for some flags in the high bits of our u64.
     931              :     const TS_DROP_HIGH_BITS: u32 = u32::count_ones(Self::TS_ONES) + 1;
     932              :     const TS_MASK: u32 = 0x1f_ff_ff_ff;
     933              :     const TS_ONES: u32 = 0x60_00_00_00;
     934              : 
     935              :     const ATIME_SHIFT: u32 = 0;
     936              :     const RTIME_SHIFT: u32 = 32 - Self::TS_DROP_HIGH_BITS;
     937              :     const VISIBILITY_SHIFT: u32 = 64 - 2 * Self::TS_DROP_HIGH_BITS;
     938              : 
     939       479716 :     fn write_bits(&self, mask: u64, value: u64) -> u64 {
     940       479716 :         self.0
     941       479716 :             .fetch_update(
     942       479716 :                 // TODO: decide what orderings are correct
     943       479716 :                 std::sync::atomic::Ordering::Relaxed,
     944       479716 :                 std::sync::atomic::Ordering::Relaxed,
     945       479716 :                 |v| Some((v & !mask) | (value & mask)),
     946       479716 :             )
     947       479716 :             .expect("Inner function is infallible")
     948       479716 :     }
     949              : 
     950       482828 :     fn to_low_res_timestamp(shift: u32, time: SystemTime) -> (u64, u64) {
     951       482828 :         // Drop the low three bits of the timestamp, for an ~8s accuracy
     952       482828 :         let timestamp = time.duration_since(UNIX_EPOCH).unwrap().as_secs() & (Self::TS_MASK as u64);
     953       482828 : 
     954       482828 :         ((Self::TS_MASK as u64) << shift, timestamp << shift)
     955       482828 :     }
     956              : 
     957          292 :     fn read_low_res_timestamp(&self, shift: u32) -> Option<SystemTime> {
     958          292 :         let read = self.0.load(std::sync::atomic::Ordering::Relaxed);
     959          292 : 
     960          292 :         let ts_bits = (read & ((Self::TS_MASK as u64) << shift)) >> shift;
     961          292 :         if ts_bits == 0 {
     962          132 :             None
     963              :         } else {
     964          160 :             Some(UNIX_EPOCH + Duration::from_secs(ts_bits | (Self::TS_ONES as u64)))
     965              :         }
     966          292 :     }
     967              : 
     968              :     /// Record a change in layer residency.
     969              :     ///
     970              :     /// Recording the event must happen while holding the layer map lock to
     971              :     /// ensure that latest-activity-threshold-based layer eviction (eviction_task.rs)
     972              :     /// can do an "imitate access" to this layer, before it observes `now-latest_activity() > threshold`.
     973              :     ///
     974              :     /// If we instead recorded the residence event with a timestamp from before grabbing the layer map lock,
     975              :     /// the following race could happen:
     976              :     ///
     977              :     /// - Compact: Write out an L1 layer from several L0 layers. This records residence event LayerCreate with the current timestamp.
     978              :     /// - Eviction: imitate access logical size calculation. This accesses the L0 layers because the L1 layer is not yet in the layer map.
     979              :     /// - Compact: Grab layer map lock, add the new L1 to layer map and remove the L0s, release layer map lock.
     980              :     /// - Eviction: observes the new L1 layer whose only activity timestamp is the LayerCreate event.
     981          100 :     pub(crate) fn record_residence_event_at(&self, now: SystemTime) {
     982          100 :         let (mask, value) = Self::to_low_res_timestamp(Self::RTIME_SHIFT, now);
     983          100 :         self.write_bits(mask, value);
     984          100 :     }
     985              : 
     986           96 :     pub(crate) fn record_residence_event(&self) {
     987           96 :         self.record_residence_event_at(SystemTime::now())
     988           96 :     }
     989              : 
     990       478880 :     fn record_access_at(&self, now: SystemTime) -> bool {
     991       478880 :         let (mut mask, mut value) = Self::to_low_res_timestamp(Self::ATIME_SHIFT, now);
     992       478880 : 
     993       478880 :         // A layer which is accessed must be visible.
     994       478880 :         mask |= 0x1 << Self::VISIBILITY_SHIFT;
     995       478880 :         value |= 0x1 << Self::VISIBILITY_SHIFT;
     996       478880 : 
     997       478880 :         let old_bits = self.write_bits(mask, value);
     998            4 :         !matches!(
     999       478880 :             self.decode_visibility(old_bits),
    1000              :             LayerVisibilityHint::Visible
    1001              :         )
    1002       478880 :     }
    1003              : 
    1004              :     /// Returns true if we modified the layer's visibility to set it to Visible implicitly
    1005              :     /// as a result of this access
    1006       479440 :     pub(crate) fn record_access(&self, ctx: &RequestContext) -> bool {
    1007       479440 :         if ctx.access_stats_behavior() == AccessStatsBehavior::Skip {
    1008          572 :             return false;
    1009       478868 :         }
    1010       478868 : 
    1011       478868 :         self.record_access_at(SystemTime::now())
    1012       479440 :     }
    1013              : 
    1014            0 :     fn as_api_model(
    1015            0 :         &self,
    1016            0 :         reset: LayerAccessStatsReset,
    1017            0 :     ) -> pageserver_api::models::LayerAccessStats {
    1018            0 :         let ret = pageserver_api::models::LayerAccessStats {
    1019            0 :             access_time: self
    1020            0 :                 .read_low_res_timestamp(Self::ATIME_SHIFT)
    1021            0 :                 .unwrap_or(UNIX_EPOCH),
    1022            0 :             residence_time: self
    1023            0 :                 .read_low_res_timestamp(Self::RTIME_SHIFT)
    1024            0 :                 .unwrap_or(UNIX_EPOCH),
    1025            0 :             visible: matches!(self.visibility(), LayerVisibilityHint::Visible),
    1026              :         };
    1027            0 :         match reset {
    1028            0 :             LayerAccessStatsReset::NoReset => {}
    1029            0 :             LayerAccessStatsReset::AllStats => {
    1030            0 :                 self.write_bits((Self::TS_MASK as u64) << Self::ATIME_SHIFT, 0x0);
    1031            0 :                 self.write_bits((Self::TS_MASK as u64) << Self::RTIME_SHIFT, 0x0);
    1032            0 :             }
    1033              :         }
    1034            0 :         ret
    1035            0 :     }
    1036              : 
    1037              :     /// Get the latest access timestamp, falling back to latest residence event.  The latest residence event
    1038              :     /// will be this Layer's construction time, if its residence hasn't changed since then.
    1039           84 :     pub(crate) fn latest_activity(&self) -> SystemTime {
    1040           84 :         if let Some(t) = self.read_low_res_timestamp(Self::ATIME_SHIFT) {
    1041           12 :             t
    1042              :         } else {
    1043           72 :             self.read_low_res_timestamp(Self::RTIME_SHIFT)
    1044           72 :                 .expect("Residence time is set on construction")
    1045              :         }
    1046           84 :     }
    1047              : 
    1048              :     /// Whether this layer has been accessed (excluding in [`AccessStatsBehavior::Skip`]).
    1049              :     ///
    1050              :     /// This indicates whether the layer has been used for some purpose that would motivate
    1051              :     /// us to keep it on disk, such as for serving a getpage request.
    1052           68 :     fn accessed(&self) -> bool {
    1053           68 :         // Consider it accessed if the most recent access is more recent than
    1054           68 :         // the most recent change in residence status.
    1055           68 :         match (
    1056           68 :             self.read_low_res_timestamp(Self::ATIME_SHIFT),
    1057           68 :             self.read_low_res_timestamp(Self::RTIME_SHIFT),
    1058              :         ) {
    1059           60 :             (None, _) => false,
    1060            0 :             (Some(_), None) => true,
    1061            8 :             (Some(a), Some(r)) => a >= r,
    1062              :         }
    1063           68 :     }
    1064              : 
    1065              :     /// Helper for extracting the visibility hint from the literal value of our inner u64
    1066       481206 :     fn decode_visibility(&self, bits: u64) -> LayerVisibilityHint {
    1067       481206 :         match (bits >> Self::VISIBILITY_SHIFT) & 0x1 {
    1068       481158 :             1 => LayerVisibilityHint::Visible,
    1069           48 :             0 => LayerVisibilityHint::Covered,
    1070            0 :             _ => unreachable!(),
    1071              :         }
    1072       481206 :     }
    1073              : 
    1074              :     /// Returns the old value which has been replaced
    1075          736 :     pub(crate) fn set_visibility(&self, visibility: LayerVisibilityHint) -> LayerVisibilityHint {
    1076          736 :         let value = match visibility {
    1077          632 :             LayerVisibilityHint::Visible => 0x1 << Self::VISIBILITY_SHIFT,
    1078          104 :             LayerVisibilityHint::Covered => 0x0,
    1079              :         };
    1080              : 
    1081          736 :         let old_bits = self.write_bits(0x1 << Self::VISIBILITY_SHIFT, value);
    1082          736 :         self.decode_visibility(old_bits)
    1083          736 :     }
    1084              : 
    1085         1590 :     pub(crate) fn visibility(&self) -> LayerVisibilityHint {
    1086         1590 :         let read = self.0.load(std::sync::atomic::Ordering::Relaxed);
    1087         1590 :         self.decode_visibility(read)
    1088         1590 :     }
    1089              : }
    1090              : 
    1091              : /// Get a layer descriptor from a layer.
    1092              : pub(crate) trait AsLayerDesc {
    1093              :     /// Get the layer descriptor.
    1094              :     fn layer_desc(&self) -> &PersistentLayerDesc;
    1095              : }
    1096              : 
    1097              : pub mod tests {
    1098              :     use pageserver_api::shard::TenantShardId;
    1099              :     use utils::id::TimelineId;
    1100              : 
    1101              :     use super::*;
    1102              : 
    1103              :     impl From<DeltaLayerName> for PersistentLayerDesc {
    1104           44 :         fn from(value: DeltaLayerName) -> Self {
    1105           44 :             PersistentLayerDesc::new_delta(
    1106           44 :                 TenantShardId::from([0; 18]),
    1107           44 :                 TimelineId::from_array([0; 16]),
    1108           44 :                 value.key_range,
    1109           44 :                 value.lsn_range,
    1110           44 :                 233,
    1111           44 :             )
    1112           44 :         }
    1113              :     }
    1114              : 
    1115              :     impl From<ImageLayerName> for PersistentLayerDesc {
    1116           48 :         fn from(value: ImageLayerName) -> Self {
    1117           48 :             PersistentLayerDesc::new_img(
    1118           48 :                 TenantShardId::from([0; 18]),
    1119           48 :                 TimelineId::from_array([0; 16]),
    1120           48 :                 value.key_range,
    1121           48 :                 value.lsn,
    1122           48 :                 233,
    1123           48 :             )
    1124           48 :         }
    1125              :     }
    1126              : 
    1127              :     impl From<LayerName> for PersistentLayerDesc {
    1128           92 :         fn from(value: LayerName) -> Self {
    1129           92 :             match value {
    1130           44 :                 LayerName::Delta(d) => Self::from(d),
    1131           48 :                 LayerName::Image(i) => Self::from(i),
    1132              :             }
    1133           92 :         }
    1134              :     }
    1135              : }
    1136              : 
    1137              : /// Range wrapping newtype, which uses display to render Debug.
    1138              : ///
    1139              : /// Useful with `Key`, which has too verbose `{:?}` for printing multiple layers.
    1140              : struct RangeDisplayDebug<'a, T: std::fmt::Display>(&'a Range<T>);
    1141              : 
    1142              : impl<T: std::fmt::Display> std::fmt::Debug for RangeDisplayDebug<'_, T> {
    1143            0 :     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
    1144            0 :         write!(f, "{}..{}", self.0.start, self.0.end)
    1145            0 :     }
    1146              : }
    1147              : 
    1148              : #[cfg(test)]
    1149              : mod tests2 {
    1150              :     use pageserver_api::key::DBDIR_KEY;
    1151              :     use tracing::info;
    1152              : 
    1153              :     use super::*;
    1154              :     use crate::tenant::storage_layer::IoConcurrency;
    1155              : 
    1156              :     /// TODO: currently this test relies on manual visual inspection of the --no-capture output.
    1157              :     /// Should look like so:
    1158              :     /// ```text
    1159              :     /// RUST_LOG=trace cargo nextest run  --features testing  --no-capture test_io_concurrency_noise
    1160              :     /// running 1 test
    1161              :     /// 2025-01-21T17:42:01.335679Z  INFO get_vectored_concurrent_io test selected=SidecarTask
    1162              :     /// 2025-01-21T17:42:01.335680Z TRACE spawning sidecar task task_id=0
    1163              :     /// 2025-01-21T17:42:01.335937Z TRACE IoConcurrency_sidecar{task_id=0}: start
    1164              :     /// 2025-01-21T17:42:01.335972Z TRACE IoConcurrency_sidecar{task_id=0}: received new io future
    1165              :     /// 2025-01-21T17:42:01.335999Z  INFO IoConcurrency_sidecar{task_id=0}: waiting for signal to complete IO
    1166              :     /// 2025-01-21T17:42:01.336229Z  WARN dropping ValuesReconstructState while some IOs have not been completed num_active_ios=1 sidecar_task_id=Some(0) backtrace=   0: <pageserver::tenant::storage_layer::ValuesReconstructState as core::ops::drop::Drop>::drop
    1167              :     ///              at ./src/tenant/storage_layer.rs:553:24
    1168              :     ///    1: core::ptr::drop_in_place<pageserver::tenant::storage_layer::ValuesReconstructState>
    1169              :     ///              at /home/christian/.rustup/toolchains/1.84.0-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/core/src/ptr/mod.rs:521:1
    1170              :     ///    2: core::mem::drop
    1171              :     ///              at /home/christian/.rustup/toolchains/1.84.0-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/core/src/mem/mod.rs:942:24
    1172              :     ///    3: pageserver::tenant::storage_layer::tests2::test_io_concurrency_noise::{{closure}}
    1173              :     ///              at ./src/tenant/storage_layer.rs:1159:9
    1174              :     ///   ...
    1175              :     ///   49: <unknown>
    1176              :     /// 2025-01-21T17:42:01.452293Z  INFO IoConcurrency_sidecar{task_id=0}: completing IO
    1177              :     /// 2025-01-21T17:42:01.452357Z TRACE IoConcurrency_sidecar{task_id=0}: io future completed
    1178              :     /// 2025-01-21T17:42:01.452473Z TRACE IoConcurrency_sidecar{task_id=0}: end
    1179              :     /// test tenant::storage_layer::tests2::test_io_concurrency_noise ... ok
    1180              :     ///
    1181              :     /// ```
    1182              :     #[tokio::test]
    1183            4 :     async fn test_io_concurrency_noise() {
    1184            4 :         crate::tenant::harness::setup_logging();
    1185            4 : 
    1186            4 :         let io_concurrency = IoConcurrency::spawn_for_test();
    1187            4 :         match *io_concurrency {
    1188            4 :             IoConcurrency::Sequential => {
    1189            4 :                 // This test asserts behavior in sidecar mode, doesn't make sense in sequential mode.
    1190            4 :                 return;
    1191            4 :             }
    1192            4 :             IoConcurrency::SidecarTask { .. } => {}
    1193            2 :         }
    1194            2 :         let mut reconstruct_state = ValuesReconstructState::new(io_concurrency.clone());
    1195            2 : 
    1196            2 :         let (io_fut_is_waiting_tx, io_fut_is_waiting) = tokio::sync::oneshot::channel();
    1197            2 :         let (do_complete_io, should_complete_io) = tokio::sync::oneshot::channel();
    1198            2 :         let (io_fut_exiting_tx, io_fut_exiting) = tokio::sync::oneshot::channel();
    1199            2 : 
    1200            2 :         let io = reconstruct_state.update_key(&DBDIR_KEY, Lsn(8), true);
    1201            2 :         reconstruct_state
    1202            2 :             .spawn_io(async move {
    1203            2 :                 info!("waiting for signal to complete IO");
    1204            4 :                 io_fut_is_waiting_tx.send(()).unwrap();
    1205            2 :                 should_complete_io.await.unwrap();
    1206            2 :                 info!("completing IO");
    1207            4 :                 io.complete(Ok(OnDiskValue::RawImage(Bytes::new())));
    1208            2 :                 io_fut_exiting_tx.send(()).unwrap();
    1209            2 :             })
    1210            2 :             .await;
    1211            4 : 
    1212            4 :         io_fut_is_waiting.await.unwrap();
    1213            2 : 
    1214            2 :         // this is what makes the noise
    1215            2 :         drop(reconstruct_state);
    1216            2 : 
    1217            2 :         do_complete_io.send(()).unwrap();
    1218            2 : 
    1219            2 :         io_fut_exiting.await.unwrap();
    1220            4 :     }
    1221              : }
        

Generated by: LCOV version 2.1-beta