mito2/compaction/
task.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, Formatter};
16use std::sync::Arc;
17use std::time::Instant;
18
19use common_telemetry::{error, info};
20use snafu::ResultExt;
21use tokio::sync::mpsc;
22
23use crate::compaction::compactor::{CompactionRegion, Compactor};
24use crate::compaction::picker::{CompactionTask, PickerOutput};
25use crate::error;
26use crate::error::CompactRegionSnafu;
27use crate::manifest::action::RegionEdit;
28use crate::metrics::{COMPACTION_FAILURE_COUNT, COMPACTION_STAGE_ELAPSED};
29use crate::request::{
30    BackgroundNotify, CompactionFailed, CompactionFinished, OutputTx, WorkerRequest,
31};
32use crate::worker::WorkerListener;
33
34/// Maximum number of compaction tasks in parallel.
35pub const MAX_PARALLEL_COMPACTION: usize = 1;
36
37pub(crate) struct CompactionTaskImpl {
38    pub compaction_region: CompactionRegion,
39    /// Request sender to notify the worker.
40    pub(crate) request_sender: mpsc::Sender<WorkerRequest>,
41    /// Senders that are used to notify waiters waiting for pending compaction tasks.
42    pub waiters: Vec<OutputTx>,
43    /// Start time of compaction task
44    pub start_time: Instant,
45    /// Event listener.
46    pub(crate) listener: WorkerListener,
47    /// Compactor to handle compaction.
48    pub(crate) compactor: Arc<dyn Compactor>,
49    /// Output of the picker.
50    pub(crate) picker_output: PickerOutput,
51}
52
53impl Debug for CompactionTaskImpl {
54    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
55        f.debug_struct("TwcsCompactionTask")
56            .field("region_id", &self.compaction_region.region_id)
57            .field("picker_output", &self.picker_output)
58            .field(
59                "append_mode",
60                &self.compaction_region.region_options.append_mode,
61            )
62            .finish()
63    }
64}
65
66impl Drop for CompactionTaskImpl {
67    fn drop(&mut self) {
68        self.mark_files_compacting(false)
69    }
70}
71
72impl CompactionTaskImpl {
73    fn mark_files_compacting(&self, compacting: bool) {
74        self.picker_output
75            .outputs
76            .iter()
77            .for_each(|o| o.inputs.iter().for_each(|f| f.set_compacting(compacting)));
78    }
79
80    async fn handle_compaction(&mut self) -> error::Result<RegionEdit> {
81        self.mark_files_compacting(true);
82
83        let merge_timer = COMPACTION_STAGE_ELAPSED
84            .with_label_values(&["merge"])
85            .start_timer();
86
87        let compaction_result = match self
88            .compactor
89            .merge_ssts(&self.compaction_region, self.picker_output.clone())
90            .await
91        {
92            Ok(v) => v,
93            Err(e) => {
94                error!(e; "Failed to compact region: {}", self.compaction_region.region_id);
95                merge_timer.stop_and_discard();
96                return Err(e);
97            }
98        };
99        let merge_time = merge_timer.stop_and_record();
100
101        info!(
102            "Compacted SST files, region_id: {}, input: {:?}, output: {:?}, window: {:?}, waiter_num: {}, merge_time: {}s",
103            self.compaction_region.region_id,
104            compaction_result.files_to_remove,
105            compaction_result.files_to_add,
106            compaction_result.compaction_time_window,
107            self.waiters.len(),
108            merge_time,
109        );
110
111        self.listener
112            .on_merge_ssts_finished(self.compaction_region.region_id)
113            .await;
114
115        let _manifest_timer = COMPACTION_STAGE_ELAPSED
116            .with_label_values(&["write_manifest"])
117            .start_timer();
118
119        self.compactor
120            .update_manifest(&self.compaction_region, compaction_result)
121            .await
122    }
123
124    /// Handles compaction failure, notifies all waiters.
125    fn on_failure(&mut self, err: Arc<error::Error>) {
126        COMPACTION_FAILURE_COUNT.inc();
127        for waiter in self.waiters.drain(..) {
128            waiter.send(Err(err.clone()).context(CompactRegionSnafu {
129                region_id: self.compaction_region.region_id,
130            }));
131        }
132    }
133
134    /// Notifies region worker to handle post-compaction tasks.
135    async fn send_to_worker(&self, request: WorkerRequest) {
136        if let Err(e) = self.request_sender.send(request).await {
137            error!(
138                "Failed to notify compaction job status for region {}, request: {:?}",
139                self.compaction_region.region_id, e.0
140            );
141        }
142    }
143}
144
145#[async_trait::async_trait]
146impl CompactionTask for CompactionTaskImpl {
147    async fn run(&mut self) {
148        let notify = match self.handle_compaction().await {
149            Ok(edit) => BackgroundNotify::CompactionFinished(CompactionFinished {
150                region_id: self.compaction_region.region_id,
151                senders: std::mem::take(&mut self.waiters),
152                start_time: self.start_time,
153                edit,
154            }),
155            Err(e) => {
156                error!(e; "Failed to compact region, region id: {}", self.compaction_region.region_id);
157                let err = Arc::new(e);
158                // notify compaction waiters
159                self.on_failure(err.clone());
160                BackgroundNotify::CompactionFailed(CompactionFailed {
161                    region_id: self.compaction_region.region_id,
162                    err,
163                })
164            }
165        };
166
167        self.send_to_worker(WorkerRequest::Background {
168            region_id: self.compaction_region.region_id,
169            notify,
170        })
171        .await;
172    }
173}
mito2/compaction/task.rs

mito2/compaction/
task.rs
