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 : /// All layer files stored in the remote storage, taking into account all
47 : /// in-progress and queued operations
48 : pub(crate) latest_files: HashMap<LayerName, LayerFileMetadata>,
49 :
50 : /// How many file uploads or deletions been scheduled, since the
51 : /// last (scheduling of) metadata index upload?
52 : pub(crate) latest_files_changes_since_metadata_upload_scheduled: u64,
53 :
54 : /// Metadata stored in the remote storage, taking into account all
55 : /// in-progress and queued operations.
56 : /// DANGER: do not return to outside world, e.g., safekeepers.
57 : pub(crate) latest_metadata: TimelineMetadata,
58 :
59 : /// `disk_consistent_lsn` from the last metadata file that was successfully
60 : /// uploaded. `Lsn(0)` if nothing was uploaded yet.
61 : /// Unlike `latest_files` or `latest_metadata`, this value is never ahead.
62 : /// Safekeeper can rely on it to make decisions for WAL storage.
63 : ///
64 : /// visible_remote_consistent_lsn is only updated after our generation has been validated with
65 : /// the control plane (unlesss a timeline's generation is None, in which case
66 : /// we skip validation)
67 : pub(crate) projected_remote_consistent_lsn: Option<Lsn>,
68 : pub(crate) visible_remote_consistent_lsn: Arc<AtomicLsn>,
69 :
70 : // Breakdown of different kinds of tasks currently in-progress
71 : pub(crate) num_inprogress_layer_uploads: usize,
72 : pub(crate) num_inprogress_metadata_uploads: usize,
73 : pub(crate) num_inprogress_deletions: usize,
74 :
75 : /// Tasks that are currently in-progress. In-progress means that a tokio Task
76 : /// has been launched for it. An in-progress task can be busy uploading, but it can
77 : /// also be waiting on the `concurrency_limiter` Semaphore in S3Bucket, or it can
78 : /// be waiting for retry in `exponential_backoff`.
79 : pub(crate) inprogress_tasks: HashMap<u64, Arc<UploadTask>>,
80 :
81 : /// Queued operations that have not been launched yet. They might depend on previous
82 : /// tasks to finish. For example, metadata upload cannot be performed before all
83 : /// preceding layer file uploads have completed.
84 : pub(crate) queued_operations: VecDeque<UploadOp>,
85 :
86 : /// Files which have been unlinked but not yet had scheduled a deletion for. Only kept around
87 : /// for error logging.
88 : ///
89 : /// Putting this behind a testing feature to catch problems in tests, but assuming we could have a
90 : /// bug causing leaks, then it's better to not leave this enabled for production builds.
91 : #[cfg(feature = "testing")]
92 : pub(crate) dangling_files: HashMap<LayerName, Generation>,
93 :
94 : /// Set to true when we have inserted the `UploadOp::Shutdown` into the `inprogress_tasks`.
95 : pub(crate) shutting_down: bool,
96 :
97 : /// Permitless semaphore on which any number of `RemoteTimelineClient::shutdown` futures can
98 : /// wait on until one of them stops the queue. The semaphore is closed when
99 : /// `RemoteTimelineClient::launch_queued_tasks` encounters `UploadOp::Shutdown`.
100 : pub(crate) shutdown_ready: Arc<tokio::sync::Semaphore>,
101 : }
102 :
103 : impl UploadQueueInitialized {
104 0 : pub(super) fn no_pending_work(&self) -> bool {
105 0 : self.inprogress_tasks.is_empty() && self.queued_operations.is_empty()
106 0 : }
107 :
108 0 : pub(super) fn get_last_remote_consistent_lsn_visible(&self) -> Lsn {
109 0 : self.visible_remote_consistent_lsn.load()
110 0 : }
111 :
112 0 : pub(super) fn get_last_remote_consistent_lsn_projected(&self) -> Option<Lsn> {
113 0 : self.projected_remote_consistent_lsn
114 0 : }
115 : }
116 :
117 : #[derive(Clone, Copy)]
118 : pub(super) enum SetDeletedFlagProgress {
119 : NotRunning,
120 : InProgress(NaiveDateTime),
121 : Successful(NaiveDateTime),
122 : }
123 :
124 : pub(super) struct UploadQueueStoppedDeletable {
125 : pub(super) upload_queue_for_deletion: UploadQueueInitialized,
126 : pub(super) deleted_at: SetDeletedFlagProgress,
127 : }
128 :
129 : pub(super) enum UploadQueueStopped {
130 : Deletable(UploadQueueStoppedDeletable),
131 : Uninitialized,
132 : }
133 :
134 0 : #[derive(thiserror::Error, Debug)]
135 : pub(crate) enum NotInitialized {
136 : #[error("queue is in state Uninitialized")]
137 : Uninitialized,
138 : #[error("queue is in state Stopped")]
139 : Stopped,
140 : #[error("queue is shutting down")]
141 : ShuttingDown,
142 : }
143 :
144 : impl NotInitialized {
145 0 : pub(crate) fn is_stopping(&self) -> bool {
146 0 : use NotInitialized::*;
147 0 : match self {
148 0 : Uninitialized => false,
149 0 : Stopped => true,
150 0 : ShuttingDown => true,
151 : }
152 0 : }
153 : }
154 :
155 : impl UploadQueue {
156 328 : pub(crate) fn initialize_empty_remote(
157 328 : &mut self,
158 328 : metadata: &TimelineMetadata,
159 328 : ) -> anyhow::Result<&mut UploadQueueInitialized> {
160 328 : match self {
161 328 : UploadQueue::Uninitialized => (),
162 : UploadQueue::Initialized(_) | UploadQueue::Stopped(_) => {
163 0 : anyhow::bail!("already initialized, state {}", self.as_str())
164 : }
165 : }
166 :
167 328 : info!("initializing upload queue for empty remote");
168 :
169 328 : let state = UploadQueueInitialized {
170 328 : // As described in the doc comment, it's ok for `latest_files` and `latest_metadata` to be ahead.
171 328 : latest_files: HashMap::new(),
172 328 : latest_files_changes_since_metadata_upload_scheduled: 0,
173 328 : latest_metadata: metadata.clone(),
174 328 : projected_remote_consistent_lsn: None,
175 328 : visible_remote_consistent_lsn: Arc::new(AtomicLsn::new(0)),
176 328 : // what follows are boring default initializations
177 328 : task_counter: 0,
178 328 : num_inprogress_layer_uploads: 0,
179 328 : num_inprogress_metadata_uploads: 0,
180 328 : num_inprogress_deletions: 0,
181 328 : inprogress_tasks: HashMap::new(),
182 328 : queued_operations: VecDeque::new(),
183 328 : #[cfg(feature = "testing")]
184 328 : dangling_files: HashMap::new(),
185 328 : shutting_down: false,
186 328 : shutdown_ready: Arc::new(tokio::sync::Semaphore::new(0)),
187 328 : };
188 328 :
189 328 : *self = UploadQueue::Initialized(state);
190 328 : Ok(self.initialized_mut().expect("we just set it"))
191 328 : }
192 :
193 6 : pub(crate) fn initialize_with_current_remote_index_part(
194 6 : &mut self,
195 6 : index_part: &IndexPart,
196 6 : ) -> anyhow::Result<&mut UploadQueueInitialized> {
197 6 : match self {
198 6 : UploadQueue::Uninitialized => (),
199 : UploadQueue::Initialized(_) | UploadQueue::Stopped(_) => {
200 0 : anyhow::bail!("already initialized, state {}", self.as_str())
201 : }
202 : }
203 :
204 6 : let mut files = HashMap::with_capacity(index_part.layer_metadata.len());
205 22 : for (layer_name, layer_metadata) in &index_part.layer_metadata {
206 16 : files.insert(
207 16 : layer_name.to_owned(),
208 16 : LayerFileMetadata::from(layer_metadata),
209 16 : );
210 16 : }
211 :
212 6 : info!(
213 0 : "initializing upload queue with remote index_part.disk_consistent_lsn: {}",
214 0 : index_part.metadata.disk_consistent_lsn()
215 : );
216 :
217 6 : let state = UploadQueueInitialized {
218 6 : latest_files: files,
219 6 : latest_files_changes_since_metadata_upload_scheduled: 0,
220 6 : latest_metadata: index_part.metadata.clone(),
221 6 : projected_remote_consistent_lsn: Some(index_part.metadata.disk_consistent_lsn()),
222 6 : visible_remote_consistent_lsn: Arc::new(
223 6 : index_part.metadata.disk_consistent_lsn().into(),
224 6 : ),
225 6 : // what follows are boring default initializations
226 6 : task_counter: 0,
227 6 : num_inprogress_layer_uploads: 0,
228 6 : num_inprogress_metadata_uploads: 0,
229 6 : num_inprogress_deletions: 0,
230 6 : inprogress_tasks: HashMap::new(),
231 6 : queued_operations: VecDeque::new(),
232 6 : #[cfg(feature = "testing")]
233 6 : dangling_files: HashMap::new(),
234 6 : shutting_down: false,
235 6 : shutdown_ready: Arc::new(tokio::sync::Semaphore::new(0)),
236 6 : };
237 6 :
238 6 : *self = UploadQueue::Initialized(state);
239 6 : Ok(self.initialized_mut().expect("we just set it"))
240 6 : }
241 :
242 3196 : pub(crate) fn initialized_mut(&mut self) -> anyhow::Result<&mut UploadQueueInitialized> {
243 3196 : use UploadQueue::*;
244 3196 : match self {
245 0 : Uninitialized => Err(NotInitialized::Uninitialized.into()),
246 3196 : Initialized(x) => {
247 3196 : if x.shutting_down {
248 0 : Err(NotInitialized::ShuttingDown.into())
249 : } else {
250 3196 : Ok(x)
251 : }
252 : }
253 0 : Stopped(_) => Err(NotInitialized::Stopped.into()),
254 : }
255 3196 : }
256 :
257 0 : pub(crate) fn stopped_mut(&mut self) -> anyhow::Result<&mut UploadQueueStoppedDeletable> {
258 0 : match self {
259 : UploadQueue::Initialized(_) | UploadQueue::Uninitialized => {
260 0 : anyhow::bail!("queue is in state {}", self.as_str())
261 : }
262 : UploadQueue::Stopped(UploadQueueStopped::Uninitialized) => {
263 0 : anyhow::bail!("queue is in state Stopped(Uninitialized)")
264 : }
265 0 : UploadQueue::Stopped(UploadQueueStopped::Deletable(deletable)) => Ok(deletable),
266 : }
267 0 : }
268 : }
269 :
270 : /// An in-progress upload or delete task.
271 : #[derive(Debug)]
272 : pub(crate) struct UploadTask {
273 : /// Unique ID of this task. Used as the key in `inprogress_tasks` above.
274 : pub(crate) task_id: u64,
275 : pub(crate) retries: AtomicU32,
276 :
277 : pub(crate) op: UploadOp,
278 : }
279 :
280 : /// A deletion of some layers within the lifetime of a timeline. This is not used
281 : /// for timeline deletion, which skips this queue and goes directly to DeletionQueue.
282 : #[derive(Debug)]
283 : pub(crate) struct Delete {
284 : pub(crate) layers: Vec<(LayerName, LayerFileMetadata)>,
285 : }
286 :
287 : #[derive(Debug)]
288 : pub(crate) enum UploadOp {
289 : /// Upload a layer file
290 : UploadLayer(ResidentLayer, LayerFileMetadata),
291 :
292 : /// Upload the metadata file
293 : UploadMetadata(IndexPart, Lsn),
294 :
295 : /// Delete layer files
296 : Delete(Delete),
297 :
298 : /// Barrier. When the barrier operation is reached,
299 : Barrier(tokio::sync::watch::Sender<()>),
300 :
301 : /// Shutdown; upon encountering this operation no new operations will be spawned, otherwise
302 : /// this is the same as a Barrier.
303 : Shutdown,
304 : }
305 :
306 : impl std::fmt::Display for UploadOp {
307 0 : fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
308 0 : match self {
309 0 : UploadOp::UploadLayer(layer, metadata) => {
310 0 : write!(
311 0 : f,
312 0 : "UploadLayer({}, size={:?}, gen={:?})",
313 0 : layer,
314 0 : metadata.file_size(),
315 0 : metadata.generation
316 0 : )
317 : }
318 0 : UploadOp::UploadMetadata(_, lsn) => {
319 0 : write!(f, "UploadMetadata(lsn: {})", lsn)
320 : }
321 0 : UploadOp::Delete(delete) => {
322 0 : write!(f, "Delete({} layers)", delete.layers.len())
323 : }
324 0 : UploadOp::Barrier(_) => write!(f, "Barrier"),
325 0 : UploadOp::Shutdown => write!(f, "Shutdown"),
326 : }
327 0 : }
328 : }
|
