risingwave_common/util/
epoch.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::sync::LazyLock;
16use std::time::{Duration, SystemTime};
17
18use easy_ext::ext;
19use parse_display::Display;
20
21use crate::types::{ScalarImpl, Timestamptz};
22
23static UNIX_RISINGWAVE_DATE_SEC: u64 = 1_617_235_200;
24
25/// [`UNIX_RISINGWAVE_DATE_EPOCH`] represents the risingwave date of the UNIX epoch:
26/// 2021-04-01T00:00:00Z.
27pub static UNIX_RISINGWAVE_DATE_EPOCH: LazyLock<SystemTime> =
28    LazyLock::new(|| SystemTime::UNIX_EPOCH + Duration::from_secs(UNIX_RISINGWAVE_DATE_SEC));
29
30#[derive(Clone, Copy, Debug, Display, PartialEq, Eq, PartialOrd, Ord, Hash)]
31pub struct Epoch(pub u64);
32
33/// `INVALID_EPOCH` defines the invalid epoch value.
34pub const INVALID_EPOCH: u64 = 0;
35
36const EPOCH_PHYSICAL_SHIFT_BITS: u8 = 16;
37
38/// Minimum physical-time delta between normally generated epochs.
39///
40/// A delta of 2 leaves one complete physical epoch for state machines that need to split an
41/// existing epoch interval later. With the default 1-second barrier interval, normally generated
42/// epochs are about 1000 physical ticks apart and therefore leave about 999 intermediate epochs.
43const MIN_EPOCH_PHYSICAL_TIME_DELTA: u64 = 2;
44
45impl Epoch {
46    pub fn now() -> Self {
47        Self(Self::physical_now() << EPOCH_PHYSICAL_SHIFT_BITS)
48    }
49
50    #[must_use]
51    pub fn next(self) -> Self {
52        let mut physical_now = Epoch::physical_now();
53        let prev_physical_time = self.physical_time();
54        let min_physical_time = prev_physical_time
55            .checked_add(MIN_EPOCH_PHYSICAL_TIME_DELTA)
56            .expect("next epoch physical time should not overflow");
57
58        loop {
59            if physical_now >= min_physical_time {
60                break;
61            }
62            physical_now = Epoch::physical_now();
63
64            #[cfg(madsim)]
65            tokio::time::advance(std::time::Duration::from_micros(10));
66            #[cfg(not(madsim))]
67            std::hint::spin_loop();
68        }
69        // The last 16 bits of the previous epoch ((prev_epoch + 1, prev_epoch + 65536)) will be
70        // used as the gap epoch when the mem table spill occurs.
71        let next_epoch = Self::from_physical_time(physical_now);
72
73        assert!(next_epoch.0 > self.0);
74        next_epoch
75    }
76
77    /// milliseconds since the RisingWave epoch
78    pub fn physical_time(&self) -> u64 {
79        self.0 >> EPOCH_PHYSICAL_SHIFT_BITS
80    }
81
82    pub fn from_physical_time(time: u64) -> Self {
83        Epoch(time << EPOCH_PHYSICAL_SHIFT_BITS)
84    }
85
86    pub fn from_unix_millis(mi: u64) -> Self {
87        Epoch((mi - UNIX_RISINGWAVE_DATE_SEC * 1000) << EPOCH_PHYSICAL_SHIFT_BITS)
88    }
89
90    pub fn from_unix_millis_or_earliest(mi: u64) -> Self {
91        Epoch((mi.saturating_sub(UNIX_RISINGWAVE_DATE_SEC * 1000)) << EPOCH_PHYSICAL_SHIFT_BITS)
92    }
93
94    pub fn physical_now() -> u64 {
95        UNIX_RISINGWAVE_DATE_EPOCH
96            .elapsed()
97            .expect("system clock set earlier than risingwave date!")
98            .as_millis() as u64
99    }
100
101    pub fn as_unix_millis(&self) -> u64 {
102        UNIX_RISINGWAVE_DATE_SEC * 1000 + self.physical_time()
103    }
104
105    pub fn as_unix_secs(&self) -> u64 {
106        UNIX_RISINGWAVE_DATE_SEC + self.physical_time() / 1000
107    }
108
109    /// Returns the epoch in a Timestamptz.
110    pub fn as_timestamptz(&self) -> Timestamptz {
111        Timestamptz::from_millis(self.as_unix_millis() as i64).expect("epoch is out of range")
112    }
113
114    /// Returns the epoch in a Timestamptz scalar.
115    pub fn as_scalar(&self) -> ScalarImpl {
116        self.as_timestamptz().into()
117    }
118
119    /// Returns the epoch in real system time.
120    pub fn as_system_time(&self) -> SystemTime {
121        *UNIX_RISINGWAVE_DATE_EPOCH + Duration::from_millis(self.physical_time())
122    }
123
124    /// Returns the epoch subtract `relative_time_ms`, which used for ttl to get epoch corresponding
125    /// to the lowerbound timepoint (`src/storage/src/hummock/iterator/forward_user.rs`)
126    pub fn subtract_ms(&self, relative_time_ms: u64) -> Self {
127        let physical_time = self.physical_time();
128
129        if physical_time < relative_time_ms {
130            Epoch(INVALID_EPOCH)
131        } else {
132            Epoch((physical_time - relative_time_ms) << EPOCH_PHYSICAL_SHIFT_BITS)
133        }
134    }
135}
136
137pub const EPOCH_AVAILABLE_BITS: u64 = 16;
138pub const MAX_SPILL_TIMES: u16 = ((1 << EPOCH_AVAILABLE_BITS) - 1) as u16;
139// Low EPOCH_AVAILABLE_BITS bits set to 1
140pub const EPOCH_SPILL_TIME_MASK: u64 = (1 << EPOCH_AVAILABLE_BITS) - 1;
141// High (64-EPOCH_AVAILABLE_BITS) bits set to 1
142pub const MAX_EPOCH: u64 = !((1 << EPOCH_AVAILABLE_BITS) - 1);
143
144// EPOCH_INC_MIN_STEP_FOR_TEST is the minimum increment step for epoch in unit tests.
145// We need to keep the lower 16 bits of the epoch unchanged during each increment,
146// and only increase the upper 48 bits.
147const EPOCH_INC_MIN_STEP_FOR_TEST: u64 = test_epoch(1);
148
149pub fn is_max_epoch(epoch: u64) -> bool {
150    // Since we have write `MAX_EPOCH` as max epoch to sstable in some previous version,
151    // it means that there may be two value in our system which represent infinite. We must check
152    // both of them for compatibility. See bug description in https://github.com/risingwavelabs/risingwave/issues/13717
153    epoch >= MAX_EPOCH
154}
155pub fn is_compatibility_max_epoch(epoch: u64) -> bool {
156    // See bug description in https://github.com/risingwavelabs/risingwave/issues/13717
157    epoch == MAX_EPOCH
158}
159impl From<u64> for Epoch {
160    fn from(epoch: u64) -> Self {
161        Self(epoch)
162    }
163}
164
165#[derive(Debug, Clone, Copy, PartialEq)]
166pub struct EpochPair {
167    pub curr: u64,
168    pub prev: u64,
169}
170
171impl EpochPair {
172    pub fn new(curr: u64, prev: u64) -> Self {
173        assert!(curr > prev);
174        Self { curr, prev }
175    }
176
177    pub fn inc_for_test(&mut self) {
178        self.prev = self.curr;
179        self.curr += EPOCH_INC_MIN_STEP_FOR_TEST;
180    }
181
182    pub fn new_test_epoch(curr: u64) -> Self {
183        if !is_max_epoch(curr) {
184            assert!(curr >= EPOCH_INC_MIN_STEP_FOR_TEST);
185            assert!((curr & EPOCH_SPILL_TIME_MASK) == 0);
186        }
187        Self::new(curr, curr - EPOCH_INC_MIN_STEP_FOR_TEST)
188    }
189}
190
191/// As most unit tests initialize a new epoch from a random value (e.g. 1, 2, 233 etc.), but the correct epoch in the system is a u64 with the last `EPOCH_AVAILABLE_BITS` bits set to 0.
192/// This method is to turn a a random epoch into a well shifted value.
193pub const fn test_epoch(value_millis: u64) -> u64 {
194    value_millis << EPOCH_AVAILABLE_BITS
195}
196
197/// There are numerous operations in our system's unit tests that involve incrementing or decrementing the epoch.
198/// These extensions for u64 type are specifically used within the unit tests.
199#[ext(EpochExt)]
200pub impl u64 {
201    fn inc_epoch(&mut self) {
202        *self += EPOCH_INC_MIN_STEP_FOR_TEST;
203    }
204
205    fn dec_epoch(&mut self) {
206        *self -= EPOCH_INC_MIN_STEP_FOR_TEST;
207    }
208
209    fn next_epoch(self) -> u64 {
210        self + EPOCH_INC_MIN_STEP_FOR_TEST
211    }
212
213    fn prev_epoch(self) -> u64 {
214        self - EPOCH_INC_MIN_STEP_FOR_TEST
215    }
216}
217
218/// Task-local storage for the epoch pair.
219pub mod task_local {
220    use futures::Future;
221    use tokio::task_local;
222
223    use super::{Epoch, EpochPair};
224
225    task_local! {
226        static TASK_LOCAL_EPOCH_PAIR: EpochPair;
227    }
228
229    /// Retrieve the current epoch from the task local storage.
230    ///
231    /// This value is updated after every yield of the barrier message. Returns `None` if the first
232    /// barrier message is not yielded.
233    pub fn curr_epoch() -> Option<Epoch> {
234        TASK_LOCAL_EPOCH_PAIR.try_with(|e| Epoch(e.curr)).ok()
235    }
236
237    /// Retrieve the previous epoch from the task local storage.
238    ///
239    /// This value is updated after every yield of the barrier message. Returns `None` if the first
240    /// barrier message is not yielded.
241    pub fn prev_epoch() -> Option<Epoch> {
242        TASK_LOCAL_EPOCH_PAIR.try_with(|e| Epoch(e.prev)).ok()
243    }
244
245    /// Retrieve the epoch pair from the task local storage.
246    ///
247    /// This value is updated after every yield of the barrier message. Returns `None` if the first
248    /// barrier message is not yielded.
249    pub fn epoch() -> Option<EpochPair> {
250        TASK_LOCAL_EPOCH_PAIR.try_with(|e| *e).ok()
251    }
252
253    /// Provides the given epoch pair in the task local storage for the scope of the given future.
254    pub async fn scope<F>(epoch: EpochPair, f: F) -> F::Output
255    where
256        F: Future,
257    {
258        TASK_LOCAL_EPOCH_PAIR.scope(epoch, f).await
259    }
260}
261
262#[cfg(test)]
263mod tests {
264    use chrono::{Local, TimeZone, Utc};
265
266    use super::*;
267
268    #[test]
269    fn test_risingwave_system_time() {
270        let utc = Utc.with_ymd_and_hms(2021, 4, 1, 0, 0, 0).unwrap();
271        let risingwave_dt = Local.from_utc_datetime(&utc.naive_utc());
272        let risingwave_st = SystemTime::from(risingwave_dt);
273        assert_eq!(risingwave_st, *UNIX_RISINGWAVE_DATE_EPOCH);
274    }
275
276    #[tokio::test]
277    async fn test_epoch_generate() {
278        let mut prev_epoch = Epoch::now();
279        for _ in 0..1000 {
280            let epoch = prev_epoch.next();
281            assert!(epoch > prev_epoch);
282            assert!(
283                epoch.physical_time() >= prev_epoch.physical_time() + MIN_EPOCH_PHYSICAL_TIME_DELTA
284            );
285            prev_epoch = epoch;
286        }
287    }
288
289    #[test]
290    fn test_subtract_ms() {
291        {
292            let epoch = Epoch(10);
293            assert_eq!(0, epoch.physical_time());
294            assert_eq!(0, epoch.subtract_ms(20).0);
295        }
296
297        {
298            let epoch = Epoch::now();
299            let physical_time = epoch.physical_time();
300            let interval = 10;
301
302            assert_ne!(0, physical_time);
303            assert_eq!(
304                physical_time - interval,
305                epoch.subtract_ms(interval).physical_time()
306            );
307        }
308    }
309}
risingwave_common/util/epoch.rs

risingwave_common/util/
epoch.rs
