ffsync 0.5.0

Chunk-level file synchronization for big files and fast links
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158

use crossbeam::channel;
use memmap2::Mmap;
use std::fs::File;
use std::io;
use std::path::Path;
use std::sync::Arc;
use std::thread;

use super::chunker::{Chunker, FastCDCChunker};
use super::hasher::{Blake3Hasher, Hasher};
use super::types::{Chunk, ChunkEvent};

pub struct PipelineConfig {
    pub read_buffer_size: usize,
    pub chunk_queue_size: usize,
    pub hash_workers: usize,
}

impl Default for PipelineConfig {
    fn default() -> Self {
        let cores = num_cpus::get();
        Self {
            read_buffer_size: 256 * 1024 * 1024,
            chunk_queue_size: 2000,
            hash_workers: cores,
        }
    }
}

struct ChunkWork {
    offset: u64,
    length: usize,
}

/// Scans a file through the pipeline and returns all chunks in canonical
/// file order (sorted by offset).
///
/// Hash workers complete chunks concurrently, so the channel's arrival
/// order is worker-completion order, not file order. Every consumer that
/// needs a deterministic chunk list (chunk-tree roots, delta diffing,
/// reconstruction plans) must use this ordering. Offsets are unique because
/// chunks partition the file.
pub fn scan_file_ordered(
    path: &Path,
    config: &PipelineConfig,
    cdc: (usize, usize, usize),
) -> io::Result<(u64, Vec<Chunk>)> {
    let (min, avg, max) = cdc;
    let rx = process_file(
        path,
        move || FastCDCChunker::new(min, avg, max),
        Blake3Hasher,
        config,
    )?;
    let mut hashed: u64 = 0;
    let mut chunks = Vec::new();
    loop {
        match rx.recv() {
            Ok(ChunkEvent::Chunk(chunk)) => {
                hashed += chunk.length as u64;
                chunks.push(chunk);
            }
            Ok(ChunkEvent::Done) => {}
            Err(_) => break,
        }
    }
    chunks.sort_by_key(|c| c.offset);
    Ok((hashed, chunks))
}

pub fn process_file<C, H>(
    path: &Path,
    chunker_factory: impl Fn() -> C + Send + Sync + 'static,
    hasher: H,
    config: &PipelineConfig,
) -> io::Result<channel::Receiver<ChunkEvent>>
where
    C: Chunker + 'static,
    H: Hasher + 'static,
{
    let file = File::open(path)?;
    let mmap = unsafe { Mmap::map(&file)? };
    let mmap = Arc::new(mmap);

    let (chunk_tx, chunk_rx) = channel::bounded(config.chunk_queue_size);
    let (result_tx, result_rx) = channel::bounded(config.chunk_queue_size);

    let total_size = mmap.len();
    let chunk_threads = config.hash_workers;
    let region_size = config.read_buffer_size;

    for thread_id in 0..chunk_threads {
        let mmap_clone = Arc::clone(&mmap);
        let chunk_tx = chunk_tx.clone();
        let mut chunker = chunker_factory();

        thread::spawn(move || {
            let start_offset = thread_id * region_size;

            for buffer_start in (start_offset..total_size).step_by(region_size * chunk_threads) {
                let buffer_end = (buffer_start + region_size).min(total_size);
                let buffer = &mmap_clone[buffer_start..buffer_end];

                for boundary in chunker.chunk(buffer) {
                    let chunk_offset = buffer_start as u64 + boundary.offset as u64;

                    if chunk_tx
                        .send(ChunkWork {
                            offset: chunk_offset,
                            length: boundary.length,
                        })
                        .is_err()
                    {
                        return;
                    }
                }
            }
        });
    }

    drop(chunk_tx);

    let hasher = Arc::new(hasher);
    let hash_workers = config.hash_workers;

    for _ in 0..hash_workers {
        let chunk_rx = chunk_rx.clone();
        let result_tx = result_tx.clone();
        let hasher = Arc::clone(&hasher);
        let mmap_clone = Arc::clone(&mmap);

        thread::spawn(move || {
            while let Ok(work) = chunk_rx.recv() {
                let data =
                    &mmap_clone[work.offset as usize..(work.offset + work.length as u64) as usize];
                let hash = hasher.hash(data);

                let chunk = Chunk {
                    offset: work.offset,
                    length: work.length,
                    hash,
                };

                if result_tx.send(ChunkEvent::Chunk(chunk)).is_err() {
                    return;
                }
            }
        });
    }

    drop(chunk_rx);

    thread::spawn(move || {
        drop(result_tx);
    });

    Ok(result_rx)
}