实战：构建分布式文件共享系统

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

use std::collections::{HashMap, HashSet};
use std::sync::Arc;
use tokio::sync::RwLock;

/// Piece 下载状态
#[derive(Debug, Clone, PartialEq)]
enum PieceStatus {
    Missing,     // 尚未开始下载
    Downloading, // 正在从某个节点下载
    Complete,    // 下载完成且校验通过
}

/// 线程安全的下载调度器
/// Arc<RwLock<>> 允许多个异步任务安全地并发读写
struct DownloadScheduler {
    metadata: FileMetadata,
    piece_status: Arc<RwLock<Vec<PieceStatus>>>,
    peer_pieces: Arc<RwLock<HashMap<PeerId, HashSet<usize>>>>,
    output_path: String,
}

 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

use sha2::{Sha256, Digest};
use std::fs::File;
use std::io::{Read, Write, Seek, SeekFrom};

const PIECE_SIZE: usize = 256 * 1024; // 256KB

#[derive(Debug, Clone, Serialize, Deserialize)]
struct FileMetadata {
    filename: String,
    total_size: u64,
    piece_size: usize,
    piece_hashes: Vec<String>,
}

impl FileMetadata {
    fn from_file(path: &str) -> anyhow::Result<Self> {
        let mut file = File::open(path)?;
        let total_size = file.metadata()?.len();
        let mut piece_hashes = Vec::new();
        let mut buffer = vec![0u8; PIECE_SIZE];

        loop {
            let n = file.read(&mut buffer)?;
            if n == 0 { break; }
            let mut hasher = Sha256::new();
            hasher.update(&buffer[..n]);
            piece_hashes.push(hex::encode(hasher.finalize()));
        }

        Ok(Self {
            filename: path.to_string(),
            total_size,
            piece_size: PIECE_SIZE,
            piece_hashes,
        })
    }

    fn verify_piece(&self, index: usize, data: &[u8]) -> bool {
        if index >= self.piece_hashes.len() { return false; }
        let mut hasher = Sha256::new();
        hasher.update(data);
        hex::encode(hasher.finalize()) == self.piece_hashes[index]
    }
}

 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

impl DownloadScheduler {
    fn new(metadata: FileMetadata, output_path: String) -> Self {
        let piece_count = metadata.piece_hashes.len();
        Self {
            metadata,
            piece_status: Arc::new(RwLock::new(vec![PieceStatus::Missing; piece_count])),
            peer_pieces: Arc::new(RwLock::new(HashMap::new())),
            output_path,
        }
    }

    /// 稀有优先选择算法
    async fn select_next_piece(&self, peer: PeerId) -> Option<usize> {
        let status = self.piece_status.read().await;
        let peer_map = self.peer_pieces.read().await;
        let peer_has = peer_map.get(&peer)?;

        let mut rarest: Vec<(usize, usize)> = status
            .iter().enumerate()
            .filter(|(i, s)| *s == PieceStatus::Missing && peer_has.contains(i))
            .map(|(i, _)| {
                let count = peer_map.values()
                    .filter(|pieces| pieces.contains(&i)).count();
                (i, count)
            }).collect();
        rarest.sort_by_key(|&(_, count)| count);
        rarest.first().map(|&(i, _)| i)
    }

    /// 安全地写入 Piece（使用写锁保护文件操作）
    async fn write_piece(&self, index: usize, data: &[u8]) -> anyhow::Result<()> {
        // 先校验哈希
        if !self.metadata.verify_piece(index, data) {
            anyhow::bail!("Piece {} hash verification failed", index);
        }

        // 写入文件（文件操作本身是同步的，但放在 async 上下文中）
        let mut file = OpenOptions::new()
            .write(true).create(true).open(&self.output_path)?;
        let offset = index as u64 * self.metadata.piece_size as u64;
        file.seek(SeekFrom::Start(offset))?;
        file.write_all(data)?;

        // 更新状态为 Complete
        let mut status = self.piece_status.write().await;
        status[index] = PieceStatus::Complete;
        Ok(())
    }

    async fn is_complete(&self) -> bool {
        let status = self.piece_status.read().await;
        status.iter().all(|s| *s == PieceStatus::Complete)
    }
}

 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

/// Seeder 端的 Piece 服务处理器
async fn handle_piece_request(
    metadata: &FileMetadata,
    file_path: &str,
    piece_index: usize,
) -> anyhow::Result<Vec<u8>> {
    if piece_index >= metadata.piece_hashes.len() {
        anyhow::bail!("Invalid piece index");
    }

    let mut file = File::open(file_path)?;
    let offset = piece_index as u64 * metadata.piece_size as u64;
    file.seek(SeekFrom::Start(offset))?;

    // 最后一个 Piece 可能不满 piece_size
    let remaining = metadata.total_size - offset;
    let read_size = std::cmp::min(metadata.piece_size as u64, remaining) as usize;
    let mut data = vec![0u8; read_size];
    file.read_exact(&mut data)?;

    // 校验后返回
    if !metadata.verify_piece(piece_index, &data) {
        anyhow::bail!("Local file corrupted at piece {}", piece_index);
    }
    Ok(data)
}

1
2
3
4
5

# Seeder：分块文件，计算哈希，将元数据存入 DHT
cargo run -- serve --file ./ubuntu-22.04.iso
# 输出：File ID: QmYyQSo1c1Ym7orWxLYvCrM2EmxFTANf8wXmmE7DWjhx5N
# 输出：Metadata published to DHT
# 输出：Listening for piece requests...

1
2
3
4
5
6
7
8

cargo run -- download \
    --peer /ip4/127.0.0.1/tcp/4001/p2p/Qm... \
    --file-id QmYyQSo1c1Ym7orWxLYvCrM2EmxFTANf8wXmmE7DWjhx5N
# 输出：Fetching metadata from DHT...
# 输出：Got metadata: 1024 pieces, 256KB each
# 输出：Downloading piece 0/1024 (rarest first)...
# 输出：Verifying piece 0... OK
# 输出：All pieces downloaded. Reassembling...

1
2
3
4
5
6
7
8
9

// lib/protocol/vector.go
type Vector struct {
    Counters []Counter // {ID: device_short_id, Value: counter} 有序数组
}

type Counter struct {
    ID    uint64 // 设备短 ID
    Value uint64 // 单调递增计数器
}

1
2
3
4
5
6
7

const (
    Equal             // 完全相同
    Greater           // 严格更新
    Lesser            // 严格更旧
    ConcurrentLesser  // 并发修改但排序上更旧
    ConcurrentGreater // 并发修改但排序上更新
)

1
2
3
4
5
6
7
8

// 伪代码：增量索引交换逻辑
if local.indexID == remote.indexID {
    // 只发送 Sequence 之后的变更
    sendDelta(local.sequence, remote.sequence)
} else {
    // 全量索引
    sendFullIndex()
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

type parallelHasher struct {
    workChan chan *hasherWork
    outChan  chan *hashResult
    workers  int // 默认 2 个并行 worker
}

func (ph *parallelHasher) HashFile(ctx context.Context, path string) {
    // 1. 初始 Stat
    // 2. 分块读取并发送到 workChan
    // 3. 并行 Worker 计算 SHA-256
    // 4. 最终 Stat 确认文件未变更
    // 5. 所有 Worker 完成后汇总结果
}