risingwave_storage/hummock/local_version/
pinned_version.rs

1// Copyright 2022 RisingWave Labs
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::collections::BTreeMap;
16use std::iter::empty;
17use std::ops::Deref;
18use std::sync::{Arc, LazyLock};
19use std::time::{Duration, Instant};
20
21use auto_enums::auto_enum;
22use risingwave_common::catalog::TableId;
23use risingwave_common::log::LogSuppressor;
24use risingwave_common::util::retry::exponential_backoff;
25use risingwave_hummock_sdk::level::{Level, Levels};
26use risingwave_hummock_sdk::version::HummockVersion;
27use risingwave_hummock_sdk::{CompactionGroupId, HummockVersionId, INVALID_VERSION_ID};
28use risingwave_rpc_client::HummockMetaClient;
29use thiserror_ext::AsReport;
30use tokio::sync::mpsc::error::TryRecvError;
31use tokio::sync::mpsc::{UnboundedReceiver, UnboundedSender};
32use tokio_retry::strategy::jitter;
33
34#[derive(Debug, Clone)]
35pub enum PinVersionAction {
36    Pin(HummockVersionId),
37    Unpin(HummockVersionId),
38}
39
40struct PinnedVersionGuard {
41    version_id: HummockVersionId,
42    pinned_version_manager_tx: UnboundedSender<PinVersionAction>,
43}
44
45impl PinnedVersionGuard {
46    /// Creates a new `PinnedVersionGuard` and send a pin request to `pinned_version_worker`.
47    fn new(
48        version_id: HummockVersionId,
49        pinned_version_manager_tx: UnboundedSender<PinVersionAction>,
50    ) -> Self {
51        if pinned_version_manager_tx
52            .send(PinVersionAction::Pin(version_id))
53            .is_err()
54        {
55            tracing::warn!("failed to send req pin version id{}", version_id);
56        }
57
58        Self {
59            version_id,
60            pinned_version_manager_tx,
61        }
62    }
63}
64
65impl Drop for PinnedVersionGuard {
66    fn drop(&mut self) {
67        if self
68            .pinned_version_manager_tx
69            .send(PinVersionAction::Unpin(self.version_id))
70            .is_err()
71        {
72            static LOG_SUPPRESSOR: LazyLock<LogSuppressor> =
73                LazyLock::new(|| LogSuppressor::per_second(1));
74            if let Ok(suppressed_count) = LOG_SUPPRESSOR.check() {
75                tracing::warn!(
76                    suppressed_count,
77                    version_id = %self.version_id,
78                    "failed to send req unpin"
79                );
80            }
81        }
82    }
83}
84
85#[derive(Clone)]
86pub struct PinnedVersion {
87    version: Arc<HummockVersion>,
88    guard: Arc<PinnedVersionGuard>,
89}
90
91impl Deref for PinnedVersion {
92    type Target = HummockVersion;
93
94    fn deref(&self) -> &Self::Target {
95        &self.version
96    }
97}
98
99impl PinnedVersion {
100    pub fn new(
101        version: HummockVersion,
102        pinned_version_manager_tx: UnboundedSender<PinVersionAction>,
103    ) -> Self {
104        let version_id = version.id;
105        PinnedVersion {
106            guard: Arc::new(PinnedVersionGuard::new(
107                version_id,
108                pinned_version_manager_tx,
109            )),
110            version: Arc::new(version),
111        }
112    }
113
114    pub fn new_pin_version(&self, version: HummockVersion) -> Option<Self> {
115        assert!(
116            version.id >= self.version.id,
117            "pinning a older version {}. Current is {}",
118            version.id,
119            self.version.id
120        );
121        if version.id == self.version.id {
122            return None;
123        }
124        let version_id = version.id;
125        Some(PinnedVersion {
126            guard: Arc::new(PinnedVersionGuard::new(
127                version_id,
128                self.guard.pinned_version_manager_tx.clone(),
129            )),
130            version: Arc::new(version),
131        })
132    }
133
134    /// Create a new `PinnedVersion` with the given version. Referring to the usage in the
135    /// `hummock_event_handler`.
136    pub fn new_with_local_version(&self, version: HummockVersion) -> Option<Self> {
137        assert!(
138            version.id >= self.version.id,
139            "pinning a older version {}. Current is {}",
140            version.id,
141            self.version.id
142        );
143        if version.id == self.version.id {
144            return None;
145        }
146
147        let version_id = version.id;
148
149        Some(PinnedVersion {
150            guard: Arc::new(PinnedVersionGuard::new(
151                version_id,
152                self.guard.pinned_version_manager_tx.clone(),
153            )),
154            version: Arc::new(version),
155        })
156    }
157
158    pub fn id(&self) -> HummockVersionId {
159        self.version.id
160    }
161
162    pub fn is_valid(&self) -> bool {
163        self.version.id != INVALID_VERSION_ID
164    }
165
166    fn levels_by_compaction_groups_id(&self, compaction_group_id: CompactionGroupId) -> &Levels {
167        self.version
168            .levels
169            .get(&compaction_group_id)
170            .unwrap_or_else(|| {
171                panic!(
172                    "levels for compaction group {} not found in version {}",
173                    compaction_group_id,
174                    self.id()
175                )
176            })
177    }
178
179    pub fn levels(&self, table_id: TableId) -> impl Iterator<Item = &Level> {
180        #[auto_enum(Iterator)]
181        match self.version.state_table_info.info().get(&table_id) {
182            Some(info) => {
183                let compaction_group_id = info.compaction_group_id;
184                let levels = self.levels_by_compaction_groups_id(compaction_group_id);
185                levels
186                    .l0
187                    .sub_levels
188                    .iter()
189                    .rev()
190                    .chain(levels.levels.iter())
191            }
192            None => empty(),
193        }
194    }
195}
196
197pub(crate) async fn start_pinned_version_worker(
198    mut rx: UnboundedReceiver<PinVersionAction>,
199    hummock_meta_client: Arc<dyn HummockMetaClient>,
200    max_version_pinning_duration_sec: u64,
201) {
202    let min_execute_interval = Duration::from_millis(1000);
203    let max_retry_interval = Duration::from_secs(10);
204    let get_backoff_strategy =
205        || exponential_backoff(Duration::from_millis(10), 10, max_retry_interval).map(jitter);
206    let mut retry_backoff = get_backoff_strategy();
207    let mut min_execute_interval_tick = tokio::time::interval(min_execute_interval);
208    min_execute_interval_tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
209    let mut need_unpin = false;
210
211    let mut version_ids_in_use: BTreeMap<HummockVersionId, (usize, Instant)> = BTreeMap::new();
212    let max_version_pinning_duration_sec = Duration::from_secs(max_version_pinning_duration_sec);
213    // For each run in the loop, accumulate versions to unpin and call unpin RPC once.
214    loop {
215        min_execute_interval_tick.tick().await;
216        // 0. Expire versions.
217        while version_ids_in_use.len() > 1
218            && let Some(e) = version_ids_in_use.first_entry()
219        {
220            if e.get().1.elapsed() < max_version_pinning_duration_sec {
221                break;
222            }
223            need_unpin = true;
224            e.remove();
225        }
226
227        // 1. Collect new versions to unpin.
228        let mut versions_to_unpin = vec![];
229        let inst = Instant::now();
230        'collect: loop {
231            match rx.try_recv() {
232                Ok(version_action) => match version_action {
233                    PinVersionAction::Pin(version_id) => {
234                        version_ids_in_use
235                            .entry(version_id)
236                            .and_modify(|e| {
237                                e.0 += 1;
238                                e.1 = inst;
239                            })
240                            .or_insert((1, inst));
241                    }
242                    PinVersionAction::Unpin(version_id) => {
243                        versions_to_unpin.push(version_id);
244                    }
245                },
246                Err(err) => match err {
247                    TryRecvError::Empty => {
248                        break 'collect;
249                    }
250                    TryRecvError::Disconnected => {
251                        tracing::info!("Shutdown hummock unpin worker");
252                        return;
253                    }
254                },
255            }
256        }
257        if !versions_to_unpin.is_empty() {
258            need_unpin = true;
259        }
260        if !need_unpin {
261            continue;
262        }
263
264        for version in &versions_to_unpin {
265            match version_ids_in_use.get_mut(version) {
266                Some((counter, _)) => {
267                    *counter -= 1;
268                    if *counter == 0 {
269                        version_ids_in_use.remove(version);
270                    }
271                }
272                None => tracing::warn!(
273                    "version {} to unpin does not exist, may already be unpinned due to expiration",
274                    version
275                ),
276            }
277        }
278
279        match version_ids_in_use.first_entry() {
280            Some(unpin_before) => {
281                // 2. Call unpin RPC, including versions failed to unpin in previous RPC calls.
282                match hummock_meta_client
283                    .unpin_version_before(*unpin_before.key())
284                    .await
285                {
286                    Ok(_) => {
287                        versions_to_unpin.clear();
288                        need_unpin = false;
289                        retry_backoff = get_backoff_strategy();
290                    }
291                    Err(err) => {
292                        let retry_after = retry_backoff.next().unwrap_or(max_retry_interval);
293                        tracing::warn!(
294                            error = %err.as_report(),
295                            "Failed to unpin version. Will retry after about {} milliseconds",
296                            retry_after.as_millis()
297                        );
298                        tokio::time::sleep(retry_after).await;
299                    }
300                }
301            }
302            None => tracing::warn!("version_ids_in_use is empty!"),
303        }
304    }
305}
risingwave_storage/hummock/local_version/pinned_version.rs

risingwave_storage/hummock/local_version/
pinned_version.rs
