TLA Line data Source code
1 : use anyhow::Context;
2 :
3 : use once_cell::sync::OnceCell;
4 : use tokio_util::sync::CancellationToken;
5 : use utils::lsn::Lsn;
6 :
7 : use std::sync::atomic::{AtomicBool, AtomicI64, Ordering as AtomicOrdering};
8 :
9 : /// Internal structure to hold all data needed for logical size calculation.
10 : ///
11 : /// Calculation consists of two stages:
12 : ///
13 : /// 1. Initial size calculation. That might take a long time, because it requires
14 : /// reading all layers containing relation sizes at `initial_part_end`.
15 : ///
16 : /// 2. Collecting an incremental part and adding that to the initial size.
17 : /// Increments are appended on walreceiver writing new timeline data,
18 : /// which result in increase or decrease of the logical size.
19 : pub(super) struct LogicalSize {
20 : /// Size, potentially slow to compute. Calculating this might require reading multiple
21 : /// layers, and even ancestor's layers.
22 : ///
23 : /// NOTE: size at a given LSN is constant, but after a restart we will calculate
24 : /// the initial size at a different LSN.
25 : pub initial_logical_size: OnceCell<(
26 : u64,
27 : crate::metrics::initial_logical_size::FinishedCalculationGuard,
28 : )>,
29 :
30 : /// Cancellation for the best-effort logical size calculation.
31 : ///
32 : /// The token is kept in a once-cell so that we can error out if a higher priority
33 : /// request comes in *before* we have started the normal logical size calculation.
34 : pub(crate) cancel_wait_for_background_loop_concurrency_limit_semaphore:
35 : OnceCell<CancellationToken>,
36 :
37 : /// Once the initial logical size is initialized, this is notified.
38 : pub(crate) initialized: tokio::sync::Semaphore,
39 :
40 : /// Latest Lsn that has its size uncalculated, could be absent for freshly created timelines.
41 : pub initial_part_end: Option<Lsn>,
42 :
43 : /// All other size changes after startup, combined together.
44 : ///
45 : /// Size shouldn't ever be negative, but this is signed for two reasons:
46 : ///
47 : /// 1. If we initialized the "baseline" size lazily, while we already
48 : /// process incoming WAL, the incoming WAL records could decrement the
49 : /// variable and temporarily make it negative. (This is just future-proofing;
50 : /// the initialization is currently not done lazily.)
51 : ///
52 : /// 2. If there is a bug and we e.g. forget to increment it in some cases
53 : /// when size grows, but remember to decrement it when it shrinks again, the
54 : /// variable could go negative. In that case, it seems better to at least
55 : /// try to keep tracking it, rather than clamp or overflow it. Note that
56 : /// get_current_logical_size() will clamp the returned value to zero if it's
57 : /// negative, and log an error. Could set it permanently to zero or some
58 : /// special value to indicate "broken" instead, but this will do for now.
59 : ///
60 : /// Note that we also expose a copy of this value as a prometheus metric,
61 : /// see `current_logical_size_gauge`. Use the `update_current_logical_size`
62 : /// to modify this, it will also keep the prometheus metric in sync.
63 : pub size_added_after_initial: AtomicI64,
64 :
65 : /// For [`crate::metrics::initial_logical_size::TIMELINES_WHERE_WALRECEIVER_GOT_APPROXIMATE_SIZE`].
66 : pub(super) did_return_approximate_to_walreceiver: AtomicBool,
67 : }
68 :
69 : /// Normalized current size, that the data in pageserver occupies.
70 UBC 0 : #[derive(Debug, Clone, Copy)]
71 : pub(crate) enum CurrentLogicalSize {
72 : /// The size is not yet calculated to the end, this is an intermediate result,
73 : /// constructed from walreceiver increments and normalized: logical data could delete some objects, hence be negative,
74 : /// yet total logical size cannot be below 0.
75 : Approximate(Approximate),
76 : // Fully calculated logical size, only other future walreceiver increments are changing it, and those changes are
77 : // available for observation without any calculations.
78 : Exact(Exact),
79 : }
80 :
81 CBC 526 : #[derive(Debug, Copy, Clone, PartialEq, Eq)]
82 : pub(crate) enum Accuracy {
83 : Approximate,
84 : Exact,
85 : }
86 :
87 UBC 0 : #[derive(Debug, Clone, Copy)]
88 : pub(crate) struct Approximate(u64);
89 0 : #[derive(Debug, Clone, Copy)]
90 : pub(crate) struct Exact(u64);
91 :
92 : impl From<&Approximate> for u64 {
93 CBC 866 : fn from(value: &Approximate) -> Self {
94 866 : value.0
95 866 : }
96 : }
97 :
98 : impl From<&Exact> for u64 {
99 995760 : fn from(val: &Exact) -> Self {
100 995760 : val.0
101 995760 : }
102 : }
103 :
104 : impl CurrentLogicalSize {
105 568864 : pub(crate) fn size_dont_care_about_accuracy(&self) -> u64 {
106 568864 : match self {
107 866 : Self::Approximate(size) => size.into(),
108 567998 : Self::Exact(size) => size.into(),
109 : }
110 568864 : }
111 572400 : pub(crate) fn accuracy(&self) -> Accuracy {
112 572400 : match self {
113 1293 : Self::Approximate(_) => Accuracy::Approximate,
114 571107 : Self::Exact(_) => Accuracy::Exact,
115 : }
116 572400 : }
117 : }
118 :
119 : impl LogicalSize {
120 572 : pub(super) fn empty_initial() -> Self {
121 572 : Self {
122 572 : initial_logical_size: OnceCell::with_value((0, {
123 572 : crate::metrics::initial_logical_size::START_CALCULATION
124 572 : .first(crate::metrics::initial_logical_size::StartCircumstances::EmptyInitial)
125 572 : .calculation_result_saved()
126 572 : })),
127 572 : cancel_wait_for_background_loop_concurrency_limit_semaphore: OnceCell::new(),
128 572 : initial_part_end: None,
129 572 : size_added_after_initial: AtomicI64::new(0),
130 572 : did_return_approximate_to_walreceiver: AtomicBool::new(false),
131 572 : initialized: tokio::sync::Semaphore::new(0),
132 572 : }
133 572 : }
134 :
135 718 : pub(super) fn deferred_initial(compute_to: Lsn) -> Self {
136 718 : Self {
137 718 : initial_logical_size: OnceCell::new(),
138 718 : cancel_wait_for_background_loop_concurrency_limit_semaphore: OnceCell::new(),
139 718 : initial_part_end: Some(compute_to),
140 718 : size_added_after_initial: AtomicI64::new(0),
141 718 : did_return_approximate_to_walreceiver: AtomicBool::new(false),
142 718 : initialized: tokio::sync::Semaphore::new(0),
143 718 : }
144 718 : }
145 :
146 997301 : pub(super) fn current_size(&self) -> CurrentLogicalSize {
147 997301 : let size_increment: i64 = self.size_added_after_initial.load(AtomicOrdering::Acquire);
148 997301 : // ^^^ keep this type explicit so that the casts in this function break if
149 997301 : // we change the type.
150 997301 : match self.initial_logical_size.get() {
151 996286 : Some((initial_size, _)) => {
152 996286 : CurrentLogicalSize::Exact(Exact(initial_size.checked_add_signed(size_increment)
153 996286 : .with_context(|| format!("Overflow during logical size calculation, initial_size: {initial_size}, size_increment: {size_increment}"))
154 996286 : .unwrap()))
155 : }
156 : None => {
157 :
158 1015 : let non_negative_size_increment = u64::try_from(size_increment).unwrap_or(0);
159 1015 : CurrentLogicalSize::Approximate(Approximate(non_negative_size_increment))
160 : }
161 : }
162 997301 : }
163 :
164 427894 : pub(super) fn increment_size(&self, delta: i64) {
165 427894 : self.size_added_after_initial
166 427894 : .fetch_add(delta, AtomicOrdering::SeqCst);
167 427894 : }
168 :
169 : /// Make the value computed by initial logical size computation
170 : /// available for re-use. This doesn't contain the incremental part.
171 621 : pub(super) fn initialized_size(&self, lsn: Lsn) -> Option<u64> {
172 607 : match self.initial_part_end {
173 607 : Some(v) if v == lsn => self.initial_logical_size.get().map(|(s, _)| *s),
174 41 : _ => None,
175 : }
176 621 : }
177 : }
|
