mito2/series_index/
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
15//! Worker-owned background maintenance for series indexes.
16
17use std::sync::Arc;
18use std::sync::atomic::{AtomicBool, Ordering};
19use std::time::Duration;
20
21use common_telemetry::info;
22use object_store::ObjectStore;
23use tokio::sync::Notify;
24use tokio::sync::mpsc::UnboundedReceiver;
25use tokio::task::JoinHandle;
26use tokio::time::{Instant, MissedTickBehavior};
27
28use crate::series_index::purger::{PurgeRequest, run_index_purge_task};
29
30/// Shared lifecycle state for a worker's series-index task.
31#[derive(Debug)]
32pub(crate) struct SeriesIndexTaskState {
33    running: AtomicBool,
34    notify: Notify,
35}
36
37impl SeriesIndexTaskState {
38    pub(crate) fn new() -> Self {
39        Self {
40            running: AtomicBool::new(true),
41            notify: Notify::new(),
42        }
43    }
44
45    pub(crate) fn is_running(&self) -> bool {
46        self.running.load(Ordering::Acquire)
47    }
48
49    pub(crate) fn stop(&self) {
50        self.running.store(false, Ordering::Release);
51        // Retain a permit if maintenance has not started waiting yet.
52        self.notify.notify_one();
53    }
54
55    pub(crate) async fn notified(&self) {
56        self.notify.notified().await;
57    }
58}
59
60/// Starts both tasks, detaching purge and returning the maintenance handle.
61pub(crate) fn spawn_series_index_tasks(
62    worker_id: u32,
63    store: ObjectStore,
64    state: Arc<SeriesIndexTaskState>,
65    purge_receiver: UnboundedReceiver<PurgeRequest>,
66    interval: Duration,
67) -> JoinHandle<()> {
68    // Snapshots may retain senders after the worker stops; purge until all senders drop.
69    common_runtime::spawn_global(run_index_purge_task(worker_id, store, purge_receiver));
70    common_runtime::spawn_global(async move {
71        SeriesIndexTask {
72            worker_id,
73            state,
74            interval,
75        }
76        .run()
77        .await;
78    })
79}
80
81/// Periodic series-index maintenance for one region worker.
82struct SeriesIndexTask {
83    worker_id: u32,
84    state: Arc<SeriesIndexTaskState>,
85    interval: Duration,
86}
87
88impl SeriesIndexTask {
89    /// Runs periodic maintenance until the worker stops.
90    async fn run(mut self) {
91        let worker_id = self.worker_id;
92        info!("Start series-index background task, worker: {worker_id}");
93        let interval = self.interval;
94        let mut timer = tokio::time::interval_at(Instant::now() + interval, interval);
95        timer.set_missed_tick_behavior(MissedTickBehavior::Skip);
96        while self.state.is_running() {
97            tokio::select! {
98                _ = self.state.notified() => {}
99                _ = timer.tick() => {
100                    if self.state.is_running() {
101                        self.maintain().await;
102                    }
103                }
104            }
105        }
106        info!("Stop series-index background task, worker: {worker_id}");
107    }
108
109    /// Runs periodic maintenance independently of incoming deletion requests.
110    async fn maintain(&mut self) {
111        // TODO: Reconcile indexes and perform other periodic maintenance here.
112    }
113}
mito2/series_index/task.rs

mito2/series_index/
task.rs
