mito2/manifest/
checkpointer.rs

1// Copyright 2023 Greptime Team
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use std::fmt::Debug;
16use std::sync::Arc;
17use std::sync::atomic::{AtomicU64, Ordering};
18
19use common_runtime::JoinHandle;
20use common_telemetry::{error, info, warn};
21use store_api::storage::RegionId;
22use store_api::{MIN_VERSION, ManifestVersion};
23#[cfg(test)]
24use tokio::sync::Notify;
25
26use crate::error::Result;
27use crate::manifest::action::{RegionCheckpoint, RegionManifest};
28use crate::manifest::manager::RegionManifestOptions;
29use crate::manifest::storage::ManifestObjectStore;
30use crate::metrics::MANIFEST_OP_ELAPSED;
31
32/// [`Checkpointer`] is responsible for doing checkpoint for a region, in an asynchronous way.
33#[derive(Debug)]
34pub(crate) struct Checkpointer {
35    manifest_options: RegionManifestOptions,
36    inner: Arc<Inner>,
37    checkpoint_task: Option<JoinHandle<()>>,
38    #[cfg(test)]
39    pending_checkpoint_wait_started: Arc<Notify>,
40}
41
42#[derive(Debug)]
43struct Inner {
44    region_id: RegionId,
45    manifest_store: ManifestObjectStore,
46    last_checkpoint_version: AtomicU64,
47}
48
49impl Inner {
50    async fn do_checkpoint(&self, checkpoint: RegionCheckpoint) {
51        let _t = MANIFEST_OP_ELAPSED
52            .with_label_values(&["checkpoint"])
53            .start_timer();
54
55        let region_id = self.region_id();
56        let version = checkpoint.last_version();
57        let checkpoint = match checkpoint.encode() {
58            Ok(checkpoint) => checkpoint,
59            Err(e) => {
60                error!(e; "Failed to encode checkpoint {:?}", checkpoint);
61                return;
62            }
63        };
64        if let Err(e) = self
65            .manifest_store
66            .save_checkpoint(version, &checkpoint)
67            .await
68        {
69            error!(e; "Failed to save checkpoint for region {}", region_id);
70            return;
71        }
72
73        // Advance the in-memory checkpoint version as soon as the checkpoint file
74        // is durable. If the subsequent delta cleanup fails, the on-disk state is
75        // still consistent (the `_last_checkpoint` metadata points at the new
76        // checkpoint) and `maybe_do_checkpoint` must not re-checkpoint the same
77        // range.
78        self.last_checkpoint_version
79            .store(version, Ordering::Relaxed);
80
81        // Logical retention changes when the checkpoint is published, even if cleanup fails.
82        self.manifest_store.retain_manifest_size_after(version);
83
84        if let Err(e) = self.manifest_store.delete_until(version, true).await {
85            warn!(e; "Failed to delete manifest actions until version {} for region {}, leftover files will be ignored on recovery", version, region_id);
86        }
87
88        info!(
89            "Checkpoint for region {} success, version: {}",
90            region_id, version
91        );
92    }
93
94    fn region_id(&self) -> RegionId {
95        self.region_id
96    }
97}
98
99impl Checkpointer {
100    pub(crate) fn new(
101        region_id: RegionId,
102        manifest_options: RegionManifestOptions,
103        manifest_store: ManifestObjectStore,
104        last_checkpoint_version: ManifestVersion,
105    ) -> Self {
106        Self {
107            manifest_options,
108            inner: Arc::new(Inner {
109                region_id,
110                manifest_store,
111                last_checkpoint_version: AtomicU64::new(last_checkpoint_version),
112            }),
113            checkpoint_task: None,
114            #[cfg(test)]
115            pending_checkpoint_wait_started: Arc::new(Notify::new()),
116        }
117    }
118
119    pub(crate) fn last_checkpoint_version(&self) -> ManifestVersion {
120        self.inner.last_checkpoint_version.load(Ordering::Relaxed)
121    }
122
123    /// Update the `removed_files` field in the manifest by the options in `manifest_options`.
124    /// This should be called before maybe do checkpoint to update the manifest.
125    pub(crate) fn update_manifest_removed_files(
126        &self,
127        mut manifest: RegionManifest,
128    ) -> Result<RegionManifest> {
129        let opt = &self.manifest_options.remove_file_options;
130
131        manifest.removed_files.evict_old_removed_files(opt)?;
132
133        // TODO(discord9): consider also check object store to clear removed files that are already deleted? How costly it is?
134
135        Ok(manifest)
136    }
137
138    /// Check if it's needed to do checkpoint for the region by the checkpoint distance.
139    /// If needed, and there's no currently running checkpoint task, it will start a new checkpoint
140    /// task running in the background.
141    pub(crate) async fn maybe_do_checkpoint(&mut self, manifest: &RegionManifest) {
142        if self
143            .checkpoint_task
144            .as_ref()
145            .is_some_and(|handle| !handle.is_finished())
146        {
147            return;
148        }
149
150        // Reap a completed task before checking whether to start the next one.
151        // This keeps the handle as the single source of truth for task state.
152        self.wait_for_pending_checkpoint().await;
153
154        if self.manifest_options.checkpoint_distance == 0 {
155            return;
156        }
157
158        let last_checkpoint_version = self.last_checkpoint_version();
159        if manifest.manifest_version - last_checkpoint_version
160            < self.manifest_options.checkpoint_distance
161        {
162            return;
163        }
164
165        let start_version = if last_checkpoint_version == 0 {
166            // Checkpoint version can't be zero by implementation.
167            // So last checkpoint version is zero means no last checkpoint.
168            MIN_VERSION
169        } else {
170            last_checkpoint_version + 1
171        };
172        let end_version = manifest.manifest_version;
173        info!(
174            "Start doing checkpoint for region {}, compacted version: [{}, {}]",
175            self.inner.region_id(),
176            start_version,
177            end_version,
178        );
179
180        let checkpoint = RegionCheckpoint {
181            last_version: end_version,
182            compacted_actions: (end_version - start_version + 1) as usize,
183            checkpoint: Some(manifest.clone()),
184        };
185        self.do_checkpoint(checkpoint);
186    }
187
188    fn do_checkpoint(&mut self, checkpoint: RegionCheckpoint) {
189        let inner = self.inner.clone();
190        self.checkpoint_task = Some(common_runtime::spawn_global(async move {
191            inner.do_checkpoint(checkpoint).await;
192        }));
193    }
194
195    /// Waits for the current checkpoint task without removing its handle first.
196    ///
197    /// Keeping the handle in `self` while awaiting is important. If the caller is
198    /// cancelled, another lifecycle transition can still wait for the same task.
199    pub(crate) async fn wait_for_pending_checkpoint(&mut self) {
200        let Some(handle) = self.checkpoint_task.as_mut() else {
201            return;
202        };
203
204        #[cfg(test)]
205        self.pending_checkpoint_wait_started.notify_one();
206
207        let result = (&mut *handle).await;
208        // There is no cancellation point between observing completion and
209        // clearing the handle.
210        self.checkpoint_task = None;
211
212        if let Err(e) = result {
213            warn!(e; "Failed to join checkpoint task for region {}", self.inner.region_id());
214        }
215    }
216
217    #[cfg(test)]
218    pub(crate) fn is_doing_checkpoint(&self) -> bool {
219        self.checkpoint_task
220            .as_ref()
221            .is_some_and(|handle| !handle.is_finished())
222    }
223
224    #[cfg(test)]
225    pub(crate) fn pending_checkpoint_wait_started(&self) -> Arc<Notify> {
226        self.pending_checkpoint_wait_started.clone()
227    }
228}
mito2/manifest/checkpointer.rs

mito2/manifest/
checkpointer.rs
