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
|
use std::path::PathBuf;
use bytes::Bytes;
use serde::{Deserialize, Serialize};
use zstd::stream::Encoder;
use garage_util::data::*;
use garage_util::error::*;
use garage_net::stream::ByteStream;
#[derive(Debug, Serialize, Deserialize, Copy, Clone)]
pub enum DataBlockHeader {
Plain,
Compressed,
}
#[derive(Debug)]
pub struct DataBlockElem<T> {
header: DataBlockHeader,
elem: T,
}
/// A possibly compressed block of data
pub type DataBlock = DataBlockElem<Bytes>;
/// A path to a possibly compressed block of data
pub type DataBlockPath = DataBlockElem<PathBuf>;
/// A stream of possibly compressed block data
pub type DataBlockStream = DataBlockElem<ByteStream>;
impl DataBlockHeader {
pub fn is_compressed(&self) -> bool {
matches!(self, DataBlockHeader::Compressed)
}
}
impl<T> DataBlockElem<T> {
pub fn from_parts(header: DataBlockHeader, elem: T) -> Self {
Self { header, elem }
}
pub fn plain(elem: T) -> Self {
Self {
header: DataBlockHeader::Plain,
elem,
}
}
pub fn compressed(elem: T) -> Self {
Self {
header: DataBlockHeader::Compressed,
elem,
}
}
pub fn into_parts(self) -> (DataBlockHeader, T) {
(self.header, self.elem)
}
pub fn as_parts_ref(&self) -> (DataBlockHeader, &T) {
(self.header, &self.elem)
}
}
impl DataBlock {
/// Verify data integrity. Does not return the buffer content.
pub fn verify(&self, hash: Hash) -> Result<(), Error> {
match self.header {
DataBlockHeader::Plain => {
if blake2sum(&self.elem) == hash {
Ok(())
} else {
Err(Error::CorruptData(hash))
}
}
DataBlockHeader::Compressed => {
zstd::stream::copy_decode(&self.elem[..], std::io::sink())
.map_err(|_| Error::CorruptData(hash))
}
}
}
pub async fn from_buffer(data: Bytes, level: Option<i32>) -> DataBlock {
tokio::task::spawn_blocking(move || {
if let Some(level) = level {
if let Ok(data_compressed) = zstd_encode(&data[..], level) {
return DataBlock::compressed(data_compressed.into());
}
}
DataBlock::plain(data.into())
})
.await
.unwrap()
}
}
fn zstd_encode<R: std::io::Read>(mut source: R, level: i32) -> std::io::Result<Vec<u8>> {
let mut result = Vec::<u8>::new();
let mut encoder = Encoder::new(&mut result, level)?;
encoder.include_checksum(true)?;
std::io::copy(&mut source, &mut encoder)?;
encoder.finish()?;
Ok(result)
}
|
