Line data Source code
1 : use super::storage_layer::LayerName;
2 : use super::storage_layer::ResidentLayer;
3 : use crate::tenant::metadata::TimelineMetadata;
4 : use crate::tenant::remote_timeline_client::index::IndexPart;
5 : use crate::tenant::remote_timeline_client::index::LayerFileMetadata;
6 : use std::collections::{HashMap, VecDeque};
7 : use std::fmt::Debug;
8 :
9 : use chrono::NaiveDateTime;
10 : use std::sync::Arc;
11 : use tracing::info;
12 : use utils::lsn::AtomicLsn;
13 :
14 : use std::sync::atomic::AtomicU32;
15 : use utils::lsn::Lsn;
16 :
17 : #[cfg(feature = "testing")]
18 : use utils::generation::Generation;
19 :
20 : // clippy warns that Uninitialized is much smaller than Initialized, which wastes
21 : // memory for Uninitialized variants. Doesn't matter in practice, there are not
22 : // that many upload queues in a running pageserver, and most of them are initialized
23 : // anyway.
24 : #[allow(clippy::large_enum_variant)]
25 : pub(super) enum UploadQueue {
26 : Uninitialized,
27 : Initialized(UploadQueueInitialized),
28 : Stopped(UploadQueueStopped),
29 : }
30 :
31 : impl UploadQueue {
32 0 : pub fn as_str(&self) -> &'static str {
33 0 : match self {
34 0 : UploadQueue::Uninitialized => "Uninitialized",
35 0 : UploadQueue::Initialized(_) => "Initialized",
36 0 : UploadQueue::Stopped(_) => "Stopped",
37 : }
38 0 : }
39 : }
40 :
41 : /// This keeps track of queued and in-progress tasks.
42 : pub(crate) struct UploadQueueInitialized {
43 : /// Counter to assign task IDs
44 : pub(crate) task_counter: u64,
45 :
46 : /// The next uploaded index_part.json; assumed to be dirty.
47 : ///
48 : /// Should not be read, directly except for layer file updates. Instead you should add a
49 : /// projected field.
50 : pub(crate) dirty: IndexPart,
51 :
52 : /// The latest remote persisted IndexPart.
53 : ///
54 : /// Each completed metadata upload will update this. The second item is the task_id which last
55 : /// updated the value, used to ensure we never store an older value over a newer one.
56 : pub(crate) clean: (IndexPart, Option<u64>),
57 :
58 : /// How many file uploads or deletions been scheduled, since the
59 : /// last (scheduling of) metadata index upload?
60 : pub(crate) latest_files_changes_since_metadata_upload_scheduled: u64,
61 :
62 : /// The Lsn is only updated after our generation has been validated with
63 : /// the control plane (unlesss a timeline's generation is None, in which case
64 : /// we skip validation)
65 : pub(crate) visible_remote_consistent_lsn: Arc<AtomicLsn>,
66 :
67 : // Breakdown of different kinds of tasks currently in-progress
68 : pub(crate) num_inprogress_layer_uploads: usize,
69 : pub(crate) num_inprogress_metadata_uploads: usize,
70 : pub(crate) num_inprogress_deletions: usize,
71 :
72 : /// Tasks that are currently in-progress. In-progress means that a tokio Task
73 : /// has been launched for it. An in-progress task can be busy uploading, but it can
74 : /// also be waiting on the `concurrency_limiter` Semaphore in S3Bucket, or it can
75 : /// be waiting for retry in `exponential_backoff`.
76 : pub(crate) inprogress_tasks: HashMap<u64, Arc<UploadTask>>,
77 :
78 : /// Queued operations that have not been launched yet. They might depend on previous
79 : /// tasks to finish. For example, metadata upload cannot be performed before all
80 : /// preceding layer file uploads have completed.
81 : pub(crate) queued_operations: VecDeque<UploadOp>,
82 :
83 : /// Files which have been unlinked but not yet had scheduled a deletion for. Only kept around
84 : /// for error logging.
85 : ///
86 : /// Putting this behind a testing feature to catch problems in tests, but assuming we could have a
87 : /// bug causing leaks, then it's better to not leave this enabled for production builds.
88 : #[cfg(feature = "testing")]
89 : pub(crate) dangling_files: HashMap<LayerName, Generation>,
90 :
91 : /// Set to true when we have inserted the `UploadOp::Shutdown` into the `inprogress_tasks`.
92 : pub(crate) shutting_down: bool,
93 :
94 : /// Permitless semaphore on which any number of `RemoteTimelineClient::shutdown` futures can
95 : /// wait on until one of them stops the queue. The semaphore is closed when
96 : /// `RemoteTimelineClient::launch_queued_tasks` encounters `UploadOp::Shutdown`.
97 : pub(crate) shutdown_ready: Arc<tokio::sync::Semaphore>,
98 : }
99 :
100 : impl UploadQueueInitialized {
101 0 : pub(super) fn no_pending_work(&self) -> bool {
102 0 : self.inprogress_tasks.is_empty() && self.queued_operations.is_empty()
103 0 : }
104 :
105 0 : pub(super) fn get_last_remote_consistent_lsn_visible(&self) -> Lsn {
106 0 : self.visible_remote_consistent_lsn.load()
107 0 : }
108 :
109 0 : pub(super) fn get_last_remote_consistent_lsn_projected(&self) -> Option<Lsn> {
110 0 : let lsn = self.clean.0.metadata.disk_consistent_lsn();
111 0 : self.clean.1.map(|_| lsn)
112 0 : }
113 : }
114 :
115 : #[derive(Clone, Copy)]
116 : pub(super) enum SetDeletedFlagProgress {
117 : NotRunning,
118 : InProgress(NaiveDateTime),
119 : Successful(NaiveDateTime),
120 : }
121 :
122 : pub(super) struct UploadQueueStoppedDeletable {
123 : pub(super) upload_queue_for_deletion: UploadQueueInitialized,
124 : pub(super) deleted_at: SetDeletedFlagProgress,
125 : }
126 :
127 : pub(super) enum UploadQueueStopped {
128 : Deletable(UploadQueueStoppedDeletable),
129 : Uninitialized,
130 : }
131 :
132 0 : #[derive(thiserror::Error, Debug)]
133 : pub(crate) enum NotInitialized {
134 : #[error("queue is in state Uninitialized")]
135 : Uninitialized,
136 : #[error("queue is in state Stopped")]
137 : Stopped,
138 : #[error("queue is shutting down")]
139 : ShuttingDown,
140 : }
141 :
142 : impl NotInitialized {
143 0 : pub(crate) fn is_stopping(&self) -> bool {
144 0 : use NotInitialized::*;
145 0 : match self {
146 0 : Uninitialized => false,
147 0 : Stopped => true,
148 0 : ShuttingDown => true,
149 : }
150 0 : }
151 : }
152 :
153 : impl UploadQueue {
154 377 : pub(crate) fn initialize_empty_remote(
155 377 : &mut self,
156 377 : metadata: &TimelineMetadata,
157 377 : ) -> anyhow::Result<&mut UploadQueueInitialized> {
158 377 : match self {
159 377 : UploadQueue::Uninitialized => (),
160 : UploadQueue::Initialized(_) | UploadQueue::Stopped(_) => {
161 0 : anyhow::bail!("already initialized, state {}", self.as_str())
162 : }
163 : }
164 :
165 377 : info!("initializing upload queue for empty remote");
166 :
167 377 : let index_part = IndexPart::empty(metadata.clone());
168 377 :
169 377 : let state = UploadQueueInitialized {
170 377 : dirty: index_part.clone(),
171 377 : clean: (index_part, None),
172 377 : latest_files_changes_since_metadata_upload_scheduled: 0,
173 377 : visible_remote_consistent_lsn: Arc::new(AtomicLsn::new(0)),
174 377 : // what follows are boring default initializations
175 377 : task_counter: 0,
176 377 : num_inprogress_layer_uploads: 0,
177 377 : num_inprogress_metadata_uploads: 0,
178 377 : num_inprogress_deletions: 0,
179 377 : inprogress_tasks: HashMap::new(),
180 377 : queued_operations: VecDeque::new(),
181 377 : #[cfg(feature = "testing")]
182 377 : dangling_files: HashMap::new(),
183 377 : shutting_down: false,
184 377 : shutdown_ready: Arc::new(tokio::sync::Semaphore::new(0)),
185 377 : };
186 377 :
187 377 : *self = UploadQueue::Initialized(state);
188 377 : Ok(self.initialized_mut().expect("we just set it"))
189 377 : }
190 :
191 6 : pub(crate) fn initialize_with_current_remote_index_part(
192 6 : &mut self,
193 6 : index_part: &IndexPart,
194 6 : ) -> anyhow::Result<&mut UploadQueueInitialized> {
195 6 : match self {
196 6 : UploadQueue::Uninitialized => (),
197 : UploadQueue::Initialized(_) | UploadQueue::Stopped(_) => {
198 0 : anyhow::bail!("already initialized, state {}", self.as_str())
199 : }
200 : }
201 :
202 6 : info!(
203 0 : "initializing upload queue with remote index_part.disk_consistent_lsn: {}",
204 0 : index_part.metadata.disk_consistent_lsn()
205 : );
206 :
207 6 : let state = UploadQueueInitialized {
208 6 : dirty: index_part.clone(),
209 6 : clean: (index_part.clone(), None),
210 6 : latest_files_changes_since_metadata_upload_scheduled: 0,
211 6 : visible_remote_consistent_lsn: Arc::new(
212 6 : index_part.metadata.disk_consistent_lsn().into(),
213 6 : ),
214 6 : // what follows are boring default initializations
215 6 : task_counter: 0,
216 6 : num_inprogress_layer_uploads: 0,
217 6 : num_inprogress_metadata_uploads: 0,
218 6 : num_inprogress_deletions: 0,
219 6 : inprogress_tasks: HashMap::new(),
220 6 : queued_operations: VecDeque::new(),
221 6 : #[cfg(feature = "testing")]
222 6 : dangling_files: HashMap::new(),
223 6 : shutting_down: false,
224 6 : shutdown_ready: Arc::new(tokio::sync::Semaphore::new(0)),
225 6 : };
226 6 :
227 6 : *self = UploadQueue::Initialized(state);
228 6 : Ok(self.initialized_mut().expect("we just set it"))
229 6 : }
230 :
231 3920 : pub(crate) fn initialized_mut(&mut self) -> anyhow::Result<&mut UploadQueueInitialized> {
232 3920 : use UploadQueue::*;
233 3920 : match self {
234 0 : Uninitialized => Err(NotInitialized::Uninitialized.into()),
235 3920 : Initialized(x) => {
236 3920 : if x.shutting_down {
237 0 : Err(NotInitialized::ShuttingDown.into())
238 : } else {
239 3920 : Ok(x)
240 : }
241 : }
242 0 : Stopped(_) => Err(NotInitialized::Stopped.into()),
243 : }
244 3920 : }
245 :
246 0 : pub(crate) fn stopped_mut(&mut self) -> anyhow::Result<&mut UploadQueueStoppedDeletable> {
247 0 : match self {
248 : UploadQueue::Initialized(_) | UploadQueue::Uninitialized => {
249 0 : anyhow::bail!("queue is in state {}", self.as_str())
250 : }
251 : UploadQueue::Stopped(UploadQueueStopped::Uninitialized) => {
252 0 : anyhow::bail!("queue is in state Stopped(Uninitialized)")
253 : }
254 0 : UploadQueue::Stopped(UploadQueueStopped::Deletable(deletable)) => Ok(deletable),
255 : }
256 0 : }
257 : }
258 :
259 : /// An in-progress upload or delete task.
260 : #[derive(Debug)]
261 : pub(crate) struct UploadTask {
262 : /// Unique ID of this task. Used as the key in `inprogress_tasks` above.
263 : pub(crate) task_id: u64,
264 : pub(crate) retries: AtomicU32,
265 :
266 : pub(crate) op: UploadOp,
267 : }
268 :
269 : /// A deletion of some layers within the lifetime of a timeline. This is not used
270 : /// for timeline deletion, which skips this queue and goes directly to DeletionQueue.
271 : #[derive(Debug)]
272 : pub(crate) struct Delete {
273 : pub(crate) layers: Vec<(LayerName, LayerFileMetadata)>,
274 : }
275 :
276 : #[derive(Debug)]
277 : pub(crate) enum UploadOp {
278 : /// Upload a layer file
279 : UploadLayer(ResidentLayer, LayerFileMetadata),
280 :
281 : /// Upload a index_part.json file
282 : UploadMetadata {
283 : /// The next [`UploadQueueInitialized::clean`] after this upload succeeds.
284 : uploaded: Box<IndexPart>,
285 : },
286 :
287 : /// Delete layer files
288 : Delete(Delete),
289 :
290 : /// Barrier. When the barrier operation is reached, the channel is closed.
291 : Barrier(tokio::sync::watch::Sender<()>),
292 :
293 : /// Shutdown; upon encountering this operation no new operations will be spawned, otherwise
294 : /// this is the same as a Barrier.
295 : Shutdown,
296 : }
297 :
298 : impl std::fmt::Display for UploadOp {
299 0 : fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
300 0 : match self {
301 0 : UploadOp::UploadLayer(layer, metadata) => {
302 0 : write!(
303 0 : f,
304 0 : "UploadLayer({}, size={:?}, gen={:?})",
305 0 : layer, metadata.file_size, metadata.generation
306 0 : )
307 : }
308 0 : UploadOp::UploadMetadata { uploaded, .. } => {
309 0 : write!(
310 0 : f,
311 0 : "UploadMetadata(lsn: {})",
312 0 : uploaded.metadata.disk_consistent_lsn()
313 0 : )
314 : }
315 0 : UploadOp::Delete(delete) => {
316 0 : write!(f, "Delete({} layers)", delete.layers.len())
317 : }
318 0 : UploadOp::Barrier(_) => write!(f, "Barrier"),
319 0 : UploadOp::Shutdown => write!(f, "Shutdown"),
320 : }
321 0 : }
322 : }
|
