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        if let Err(e) = self.manifest_store.delete_until(version, true).await {
82            warn!(e; "Failed to delete manifest actions until version {} for region {}, leftover files will be ignored on recovery", version, region_id);
83        }
84
85        info!(
86            "Checkpoint for region {} success, version: {}",
87            region_id, version
88        );
89    }
90
91    fn region_id(&self) -> RegionId {
92        self.region_id
93    }
94}
95
96impl Checkpointer {
97    pub(crate) fn new(
98        region_id: RegionId,
99        manifest_options: RegionManifestOptions,
100        manifest_store: ManifestObjectStore,
101        last_checkpoint_version: ManifestVersion,
102    ) -> Self {
103        Self {
104            manifest_options,
105            inner: Arc::new(Inner {
106                region_id,
107                manifest_store,
108                last_checkpoint_version: AtomicU64::new(last_checkpoint_version),
109            }),
110            checkpoint_task: None,
111            #[cfg(test)]
112            pending_checkpoint_wait_started: Arc::new(Notify::new()),
113        }
114    }
115
116    pub(crate) fn last_checkpoint_version(&self) -> ManifestVersion {
117        self.inner.last_checkpoint_version.load(Ordering::Relaxed)
118    }
119
120    /// Update the `removed_files` field in the manifest by the options in `manifest_options`.
121    /// This should be called before maybe do checkpoint to update the manifest.
122    pub(crate) fn update_manifest_removed_files(
123        &self,
124        mut manifest: RegionManifest,
125    ) -> Result<RegionManifest> {
126        let opt = &self.manifest_options.remove_file_options;
127
128        manifest.removed_files.evict_old_removed_files(opt)?;
129
130        // TODO(discord9): consider also check object store to clear removed files that are already deleted? How costly it is?
131
132        Ok(manifest)
133    }
134
135    /// Check if it's needed to do checkpoint for the region by the checkpoint distance.
136    /// If needed, and there's no currently running checkpoint task, it will start a new checkpoint
137    /// task running in the background.
138    pub(crate) async fn maybe_do_checkpoint(&mut self, manifest: &RegionManifest) {
139        if self
140            .checkpoint_task
141            .as_ref()
142            .is_some_and(|handle| !handle.is_finished())
143        {
144            return;
145        }
146
147        // Reap a completed task before checking whether to start the next one.
148        // This keeps the handle as the single source of truth for task state.
149        self.wait_for_pending_checkpoint().await;
150
151        if self.manifest_options.checkpoint_distance == 0 {
152            return;
153        }
154
155        let last_checkpoint_version = self.last_checkpoint_version();
156        if manifest.manifest_version - last_checkpoint_version
157            < self.manifest_options.checkpoint_distance
158        {
159            return;
160        }
161
162        let start_version = if last_checkpoint_version == 0 {
163            // Checkpoint version can't be zero by implementation.
164            // So last checkpoint version is zero means no last checkpoint.
165            MIN_VERSION
166        } else {
167            last_checkpoint_version + 1
168        };
169        let end_version = manifest.manifest_version;
170        info!(
171            "Start doing checkpoint for region {}, compacted version: [{}, {}]",
172            self.inner.region_id(),
173            start_version,
174            end_version,
175        );
176
177        let checkpoint = RegionCheckpoint {
178            last_version: end_version,
179            compacted_actions: (end_version - start_version + 1) as usize,
180            checkpoint: Some(manifest.clone()),
181        };
182        self.do_checkpoint(checkpoint);
183    }
184
185    fn do_checkpoint(&mut self, checkpoint: RegionCheckpoint) {
186        let inner = self.inner.clone();
187        self.checkpoint_task = Some(common_runtime::spawn_global(async move {
188            inner.do_checkpoint(checkpoint).await;
189        }));
190    }
191
192    /// Waits for the current checkpoint task without removing its handle first.
193    ///
194    /// Keeping the handle in `self` while awaiting is important. If the caller is
195    /// cancelled, another lifecycle transition can still wait for the same task.
196    pub(crate) async fn wait_for_pending_checkpoint(&mut self) {
197        let Some(handle) = self.checkpoint_task.as_mut() else {
198            return;
199        };
200
201        #[cfg(test)]
202        self.pending_checkpoint_wait_started.notify_one();
203
204        let result = (&mut *handle).await;
205        // There is no cancellation point between observing completion and
206        // clearing the handle.
207        self.checkpoint_task = None;
208
209        if let Err(e) = result {
210            warn!(e; "Failed to join checkpoint task for region {}", self.inner.region_id());
211        }
212    }
213
214    #[cfg(test)]
215    pub(crate) fn is_doing_checkpoint(&self) -> bool {
216        self.checkpoint_task
217            .as_ref()
218            .is_some_and(|handle| !handle.is_finished())
219    }
220
221    #[cfg(test)]
222    pub(crate) fn pending_checkpoint_wait_started(&self) -> Arc<Notify> {
223        self.pending_checkpoint_wait_started.clone()
224    }
225}
mito2/manifest/checkpointer.rs

mito2/manifest/
checkpointer.rs
