raft_engine/file_pipe_log/
format.rs

1// Copyright (c) 2017-present, PingCAP, Inc. Licensed under Apache-2.0.
2
3//! Representations of objects in filesystem.
4
5use std::io::BufRead;
6use std::path::{Path, PathBuf};
7
8use num_traits::{FromPrimitive, ToPrimitive};
9
10use crate::codec::{self, NumberEncoder};
11use crate::pipe_log::{FileId, FileSeq, LogQueue, Version};
12use crate::{Error, Result};
13
14/// Width to format log sequence number.
15const LOG_SEQ_WIDTH: usize = 16;
16/// Name suffix for Append queue files.
17const LOG_APPEND_SUFFIX: &str = ".raftlog";
18/// Name suffix for Rewrite queue files.
19const LOG_REWRITE_SUFFIX: &str = ".rewrite";
20/// Name suffix for reserved log files that contain only zeros.
21const LOG_APPEND_RESERVED_SUFFIX: &str = ".raftlog.reserved";
22/// File header.
23const LOG_FILE_MAGIC_HEADER: &[u8] = b"RAFT-LOG-FILE-HEADER-9986AB3E47F320B394C8E84916EB0ED5";
24
25/// Checks whether the given `buf` is padded with zeros.
26///
27/// To simplify the checking strategy, we just check the first
28/// and last byte in the `buf`.
29///
30/// In most common cases, the paddings will be filled with `0`,
31/// and several corner cases, where there exists corrupted blocks
32/// in the disk, might pass through this rule, but will failed in
33/// followed processing. So, we can just keep it simplistic.
34#[inline]
35pub(crate) fn is_zero_padded(buf: &[u8]) -> bool {
36    buf.is_empty() || (buf[0] == 0 && buf[buf.len() - 1] == 0)
37}
38
39/// `FileNameExt` offers file name formatting extensions to [`FileId`].
40pub trait FileNameExt: Sized {
41    fn parse_file_name(file_name: &str) -> Option<Self>;
42
43    fn build_file_name(&self) -> String;
44
45    fn build_file_path<P: AsRef<Path>>(&self, dir: P) -> PathBuf {
46        let mut path = PathBuf::from(dir.as_ref());
47        path.push(self.build_file_name());
48        path
49    }
50}
51
52impl FileNameExt for FileId {
53    fn parse_file_name(file_name: &str) -> Option<FileId> {
54        if file_name.len() > LOG_SEQ_WIDTH {
55            if let Ok(seq) = file_name[..LOG_SEQ_WIDTH].parse::<u64>() {
56                if file_name.ends_with(LOG_APPEND_SUFFIX) {
57                    return Some(FileId {
58                        queue: LogQueue::Append,
59                        seq,
60                    });
61                } else if file_name.ends_with(LOG_REWRITE_SUFFIX) {
62                    return Some(FileId {
63                        queue: LogQueue::Rewrite,
64                        seq,
65                    });
66                }
67            }
68        }
69        None
70    }
71
72    fn build_file_name(&self) -> String {
73        let width = LOG_SEQ_WIDTH;
74        match self.queue {
75            LogQueue::Append => format!("{:0width$}{LOG_APPEND_SUFFIX}", self.seq,),
76            LogQueue::Rewrite => format!("{:0width$}{LOG_REWRITE_SUFFIX}", self.seq,),
77        }
78    }
79}
80
81pub fn parse_reserved_file_name(file_name: &str) -> Option<FileSeq> {
82    if file_name.len() > LOG_SEQ_WIDTH {
83        if let Ok(seq) = file_name[..LOG_SEQ_WIDTH].parse::<u64>() {
84            if file_name.ends_with(LOG_APPEND_RESERVED_SUFFIX) {
85                // As reserved files are only used for LogQueue::Append,
86                // we just return the related FileSeq of it.
87                return Some(seq);
88            }
89        }
90    }
91    None
92}
93
94pub fn build_reserved_file_name(seq: FileSeq) -> String {
95    let width = LOG_SEQ_WIDTH;
96    format!("{seq:0width$}{LOG_APPEND_RESERVED_SUFFIX}",)
97}
98
99/// Path to the lock file under `dir`.
100pub(super) fn lock_file_path<P: AsRef<Path>>(dir: P) -> PathBuf {
101    let mut path = PathBuf::from(dir.as_ref());
102    path.push("LOCK");
103    path
104}
105
106/// Log file format. It will be encoded to file header.
107#[derive(Copy, Clone, Debug, Eq, PartialEq, Default)]
108pub struct LogFileFormat {
109    pub version: Version,
110    /// 0 stands for no alignment.
111    pub alignment: u64,
112}
113
114impl LogFileFormat {
115    pub fn new(version: Version, alignment: u64) -> Self {
116        Self { version, alignment }
117    }
118
119    /// Length of header written on storage.
120    const fn header_len() -> usize {
121        LOG_FILE_MAGIC_HEADER.len() + std::mem::size_of::<Version>()
122    }
123
124    const fn payload_len(version: Version) -> usize {
125        match version {
126            Version::V1 => 0,
127            Version::V2 => std::mem::size_of::<u64>(),
128        }
129    }
130
131    pub const fn max_encoded_len() -> usize {
132        Self::header_len() + Self::payload_len(Version::V2)
133    }
134
135    /// Length of whole `LogFileFormat` written on storage.
136    pub fn encoded_len(version: Version) -> usize {
137        Self::header_len() + Self::payload_len(version)
138    }
139
140    /// Decodes a slice of bytes into a `LogFileFormat`.
141    pub fn decode(buf: &mut &[u8]) -> Result<LogFileFormat> {
142        let mut format = LogFileFormat::default();
143        if !buf.starts_with(LOG_FILE_MAGIC_HEADER) {
144            return Err(Error::Corruption(
145                "log file magic header mismatch".to_owned(),
146            ));
147        }
148        buf.consume(LOG_FILE_MAGIC_HEADER.len());
149
150        let version_u64 = codec::decode_u64(buf)?;
151        if let Some(version) = Version::from_u64(version_u64) {
152            format.version = version;
153        } else {
154            return Err(Error::Corruption(format!(
155                "unrecognized log file version: {version_u64}",
156            )));
157        }
158
159        let payload_len = Self::payload_len(format.version);
160        if buf.len() < payload_len {
161            return Err(Error::Corruption("missing header payload".to_owned()));
162        } else if payload_len > 0 {
163            format.alignment = codec::decode_u64(buf)?;
164        }
165
166        Ok(format)
167    }
168
169    /// Encodes this header and appends the bytes to the provided buffer.
170    pub fn encode(&self, buf: &mut Vec<u8>) -> Result<()> {
171        buf.extend_from_slice(LOG_FILE_MAGIC_HEADER);
172        buf.encode_u64(self.version.to_u64().unwrap())?;
173        if Self::payload_len(self.version) > 0 {
174            buf.encode_u64(self.alignment)?;
175        } else {
176            assert_eq!(self.alignment, 0);
177        }
178        #[cfg(feature = "failpoints")]
179        {
180            // Set header corrupted.
181            let corrupted = || {
182                fail::fail_point!("log_file_header::corrupted", |_| true);
183                false
184            };
185            // Set abnormal DataLayout.
186            let too_large = || {
187                fail::fail_point!("log_file_header::too_large", |_| true);
188                false
189            };
190            // Set corrupted DataLayout for `payload`.
191            let too_small = || {
192                fail::fail_point!("log_file_header::too_small", |_| true);
193                false
194            };
195            if corrupted() {
196                buf[0] += 1;
197            }
198            assert!(!(too_large() && too_small()));
199            if too_large() {
200                buf.encode_u64(0_u64)?;
201            }
202            if too_small() {
203                buf.pop();
204            }
205        }
206        Ok(())
207    }
208}
209
210#[cfg(test)]
211mod tests {
212    use super::*;
213    use crate::pipe_log::LogFileContext;
214    use crate::test_util::catch_unwind_silent;
215
216    #[test]
217    fn test_check_paddings_is_valid() {
218        // normal buffer
219        let mut buf = vec![0; 128];
220        // len < 8
221        assert!(is_zero_padded(&buf[0..6]));
222        // len == 8
223        assert!(is_zero_padded(&buf[120..]));
224        // len > 8
225        assert!(is_zero_padded(&buf));
226
227        // abnormal buffer
228        buf[127] = 3_u8;
229        assert!(is_zero_padded(&buf[0..110]));
230        assert!(is_zero_padded(&buf[120..125]));
231        assert!(!is_zero_padded(&buf[124..128]));
232        assert!(!is_zero_padded(&buf[120..]));
233        assert!(!is_zero_padded(&buf));
234    }
235
236    #[test]
237    fn test_file_name() {
238        let file_name: &str = "0000000000000123.raftlog";
239        let file_id = FileId {
240            queue: LogQueue::Append,
241            seq: 123,
242        };
243        assert_eq!(FileId::parse_file_name(file_name).unwrap(), file_id,);
244        assert_eq!(file_id.build_file_name(), file_name);
245
246        let file_name: &str = "0000000000000123.rewrite";
247        let file_id = FileId {
248            queue: LogQueue::Rewrite,
249            seq: 123,
250        };
251        assert_eq!(FileId::parse_file_name(file_name).unwrap(), file_id,);
252        assert_eq!(file_id.build_file_name(), file_name);
253
254        let invalid_cases = vec!["0000000000000123.log", "123.rewrite"];
255        for case in invalid_cases {
256            assert!(FileId::parse_file_name(case).is_none());
257        }
258    }
259
260    #[test]
261    fn test_version() {
262        let version = Version::default();
263        assert_eq!(Version::V1.to_u64().unwrap(), version.to_u64().unwrap());
264        let version2 = Version::from_u64(1).unwrap();
265        assert_eq!(version, version2);
266    }
267
268    #[test]
269    fn test_encoding_decoding_file_format() {
270        fn enc_dec_file_format(file_format: LogFileFormat) -> Result<LogFileFormat> {
271            let mut buf = Vec::with_capacity(
272                LogFileFormat::header_len() + LogFileFormat::payload_len(file_format.version),
273            );
274            file_format.encode(&mut buf).unwrap();
275            LogFileFormat::decode(&mut &buf[..])
276        }
277        // header with aligned-sized data_layout
278        {
279            let mut buf = Vec::with_capacity(LogFileFormat::header_len());
280            let version = Version::V2;
281            let alignment = 4096;
282            buf.extend_from_slice(LOG_FILE_MAGIC_HEADER);
283            buf.encode_u64(version.to_u64().unwrap()).unwrap();
284            buf.encode_u64(alignment).unwrap();
285            assert_eq!(
286                LogFileFormat::decode(&mut &buf[..]).unwrap(),
287                LogFileFormat::new(version, alignment)
288            );
289        }
290        // header with abnormal version
291        {
292            let mut buf = Vec::with_capacity(LogFileFormat::header_len());
293            let abnormal_version = 4_u64; /* abnormal version */
294            buf.extend_from_slice(LOG_FILE_MAGIC_HEADER);
295            buf.encode_u64(abnormal_version).unwrap();
296            buf.encode_u64(16).unwrap();
297            assert!(LogFileFormat::decode(&mut &buf[..]).is_err());
298        }
299        {
300            let file_format = LogFileFormat::new(Version::default(), 0);
301            assert_eq!(
302                LogFileFormat::new(Version::default(), 0),
303                enc_dec_file_format(file_format).unwrap()
304            );
305            let file_format = LogFileFormat::new(Version::default(), 4096);
306            assert!(catch_unwind_silent(|| enc_dec_file_format(file_format)).is_err());
307        }
308    }
309
310    #[test]
311    fn test_file_context() {
312        let mut file_context =
313            LogFileContext::new(FileId::dummy(LogQueue::Append), Version::default());
314        assert_eq!(file_context.get_signature(), None);
315        file_context.id.seq = 10;
316        file_context.version = Version::V2;
317        assert_eq!(file_context.get_signature().unwrap(), 10);
318        let abnormal_seq = (file_context.id.seq << 32) + 100_u64;
319        file_context.id.seq = abnormal_seq;
320        assert_ne!(file_context.get_signature().unwrap() as u64, abnormal_seq);
321        assert_eq!(file_context.get_signature().unwrap(), 100);
322    }
323}
raft_engine/file_pipe_log/format.rs

raft_engine/file_pipe_log/
format.rs
