parallel_processor/buckets/writers/
mod.rs

1use crate::memory_fs::file::writer::FileWriter;
2use desse::{Desse, DesseSized};
3use mt_debug_counters::counter::{AtomicCounter, SumMode};
4use serde::{Deserialize, Serialize};
5use std::io::Write;
6
7use super::CheckpointData;
8
9pub mod compressed_binary_writer;
10pub mod lock_free_binary_writer;
11
12pub(crate) static THREADS_BUSY_WRITING: AtomicCounter<SumMode> =
13    declare_counter_i64!("threads_busy_writing", SumMode, false);
14
15#[derive(Debug, Desse, DesseSized, Default)]
16pub(crate) struct BucketHeader {
17    pub magic: [u8; 16],
18    pub index_offset: u64,
19    pub data_format_info: [u8; Self::MAX_DATA_FORMAT_INFO_SIZE],
20}
21
22impl BucketHeader {
23    pub const MAX_DATA_FORMAT_INFO_SIZE: usize = 32;
24}
25
26#[derive(Serialize, Deserialize)]
27pub(crate) struct BucketCheckpoints {
28    pub index: Vec<CheckpointData>,
29}
30
31pub(crate) fn initialize_bucket_file(file: &mut FileWriter) -> u64 {
32    // Write empty header
33    file.write_all(&BucketHeader::default().serialize()[..])
34        .unwrap();
35
36    file.len() as u64
37}
38
39pub(crate) fn finalize_bucket_file(
40    mut file: FileWriter,
41    magic: &[u8; 16],
42    checkpoints: Vec<CheckpointData>,
43    format_info: &[u8],
44) {
45    file.flush().unwrap();
46    let index_position = file.len() as u64;
47    bincode::serialize_into(&mut file, &BucketCheckpoints { index: checkpoints }).unwrap();
48
49    let data_format_info = {
50        let mut array = [0; 32];
51        array[0..format_info.len()].copy_from_slice(format_info);
52        array
53    };
54
55    file.write_at_start(
56        &BucketHeader {
57            magic: *magic,
58            index_offset: index_position,
59            data_format_info,
60        }
61        .serialize()[..],
62    )
63    .unwrap();
64    file.flush_async();
65}
parallel_processor/buckets/writers/mod.rs

parallel_processor/buckets/writers/
mod.rs
