forked from Deuxfleurs/garage
Compute hashes on dedicated threads
This commit is contained in:
parent
a184f0d0b5
commit
1b2e1296eb
GPG key ID:
0E496D15096376BE
11 changed files with 188 additions and 51 deletions
95
Cargo.lock
generated
95
Cargo.lock
generated
|
|
@ -343,7 +343,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
|
|||
checksum = "0a4e37d16930f5459780f5621038b6382b9bb37c19016f39fb6b5808d831f174"
|
||||
dependencies = [
|
||||
"crypto-mac 0.8.0",
|
||||
"digest",
|
||||
"digest 0.9.0",
|
||||
"opaque-debug",
|
||||
]
|
||||
|
||||
|
|
@ -356,6 +356,15 @@ dependencies = [
|
|||
"generic-array",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "block-buffer"
|
||||
version = "0.10.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "0bf7fe51849ea569fd452f37822f606a5cabb684dc918707a0193fd4664ff324"
|
||||
dependencies = [
|
||||
"generic-array",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "bumpalo"
|
||||
version = "3.9.1"
|
||||
|
|
@ -589,6 +598,16 @@ dependencies = [
|
|||
"lazy_static",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "crypto-common"
|
||||
version = "0.1.6"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "1bfb12502f3fc46cca1bb51ac28df9d618d813cdc3d2f25b9fe775a34af26bb3"
|
||||
dependencies = [
|
||||
"generic-array",
|
||||
"typenum",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "crypto-mac"
|
||||
version = "0.8.0"
|
||||
|
|
@ -693,6 +712,17 @@ dependencies = [
|
|||
"generic-array",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "digest"
|
||||
version = "0.10.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "f2fb860ca6fafa5552fb6d0e816a69c8e49f0908bf524e30a90d97c85892d506"
|
||||
dependencies = [
|
||||
"block-buffer 0.10.2",
|
||||
"crypto-common",
|
||||
"subtle",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "dirs-next"
|
||||
version = "2.0.0"
|
||||
|
|
@ -991,7 +1021,7 @@ dependencies = [
|
|||
"serde",
|
||||
"serde_bytes",
|
||||
"serde_json",
|
||||
"sha2",
|
||||
"sha2 0.10.2",
|
||||
"static_init",
|
||||
"structopt",
|
||||
"timeago",
|
||||
|
|
@ -1008,7 +1038,7 @@ dependencies = [
|
|||
"base64",
|
||||
"bytes 1.1.0",
|
||||
"chrono",
|
||||
"crypto-mac 0.10.1",
|
||||
"crypto-common",
|
||||
"err-derive 0.3.1",
|
||||
"form_urlencoded",
|
||||
"futures",
|
||||
|
|
@ -1019,13 +1049,13 @@ dependencies = [
|
|||
"garage_table 0.7.0",
|
||||
"garage_util 0.7.0",
|
||||
"hex",
|
||||
"hmac 0.10.1",
|
||||
"hmac 0.12.1",
|
||||
"http",
|
||||
"http-range",
|
||||
"httpdate 0.3.2",
|
||||
"hyper",
|
||||
"idna",
|
||||
"md-5",
|
||||
"md-5 0.10.1",
|
||||
"multer",
|
||||
"nom",
|
||||
"opentelemetry",
|
||||
|
|
@ -1039,7 +1069,7 @@ dependencies = [
|
|||
"serde",
|
||||
"serde_bytes",
|
||||
"serde_json",
|
||||
"sha2",
|
||||
"sha2 0.10.2",
|
||||
"tokio",
|
||||
"tracing",
|
||||
"url",
|
||||
|
|
@ -1259,7 +1289,7 @@ dependencies = [
|
|||
"rmp-serde 0.15.5",
|
||||
"serde",
|
||||
"serde_json",
|
||||
"sha2",
|
||||
"sha2 0.9.9",
|
||||
"sled",
|
||||
"tokio",
|
||||
"toml",
|
||||
|
|
@ -1272,7 +1302,9 @@ version = "0.7.0"
|
|||
dependencies = [
|
||||
"async-trait",
|
||||
"blake2",
|
||||
"bytes 1.1.0",
|
||||
"chrono",
|
||||
"digest 0.10.3",
|
||||
"err-derive 0.3.1",
|
||||
"futures",
|
||||
"garage_db",
|
||||
|
|
@ -1285,7 +1317,7 @@ dependencies = [
|
|||
"rmp-serde 0.15.5",
|
||||
"serde",
|
||||
"serde_json",
|
||||
"sha2",
|
||||
"sha2 0.10.2",
|
||||
"tokio",
|
||||
"toml",
|
||||
"tracing",
|
||||
|
|
@ -1485,7 +1517,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
|
|||
checksum = "c1441c6b1e930e2817404b5046f1f989899143a12bf92de603b69f4e0aee1e15"
|
||||
dependencies = [
|
||||
"crypto-mac 0.10.1",
|
||||
"digest",
|
||||
"digest 0.9.0",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
|
|
@ -1495,7 +1527,16 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
|
|||
checksum = "2a2a2320eb7ec0ebe8da8f744d7812d9fc4cb4d09344ac01898dbcb6a20ae69b"
|
||||
dependencies = [
|
||||
"crypto-mac 0.11.1",
|
||||
"digest",
|
||||
"digest 0.9.0",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "hmac"
|
||||
version = "0.12.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "6c49c37c09c17a53d937dfbb742eb3a961d65a994e6bcdcf37e7399d0cc8ab5e"
|
||||
dependencies = [
|
||||
"digest 0.10.3",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
|
|
@ -1975,11 +2016,20 @@ version = "0.9.1"
|
|||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "7b5a279bb9607f9f53c22d496eade00d138d1bdcccd07d74650387cf94942a15"
|
||||
dependencies = [
|
||||
"block-buffer",
|
||||
"digest",
|
||||
"block-buffer 0.9.0",
|
||||
"digest 0.9.0",
|
||||
"opaque-debug",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "md-5"
|
||||
version = "0.10.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "658646b21e0b72f7866c7038ab086d3d5e1cd6271f060fd37defb241949d0582"
|
||||
dependencies = [
|
||||
"digest 0.10.3",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "md5"
|
||||
version = "0.7.0"
|
||||
|
|
@ -3000,20 +3050,20 @@ dependencies = [
|
|||
"base64",
|
||||
"bytes 1.1.0",
|
||||
"chrono",
|
||||
"digest",
|
||||
"digest 0.9.0",
|
||||
"futures",
|
||||
"hex",
|
||||
"hmac 0.11.0",
|
||||
"http",
|
||||
"hyper",
|
||||
"log",
|
||||
"md-5",
|
||||
"md-5 0.9.1",
|
||||
"percent-encoding",
|
||||
"pin-project-lite",
|
||||
"rusoto_credential",
|
||||
"rustc_version",
|
||||
"serde",
|
||||
"sha2",
|
||||
"sha2 0.9.9",
|
||||
"tokio",
|
||||
]
|
||||
|
||||
|
|
@ -3246,13 +3296,24 @@ version = "0.9.9"
|
|||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "4d58a1e1bf39749807d89cf2d98ac2dfa0ff1cb3faa38fbb64dd88ac8013d800"
|
||||
dependencies = [
|
||||
"block-buffer",
|
||||
"block-buffer 0.9.0",
|
||||
"cfg-if 1.0.0",
|
||||
"cpufeatures",
|
||||
"digest",
|
||||
"digest 0.9.0",
|
||||
"opaque-debug",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "sha2"
|
||||
version = "0.10.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "55deaec60f81eefe3cce0dc50bda92d6d8e88f2a27df7c5033b42afeb1ed2676"
|
||||
dependencies = [
|
||||
"cfg-if 1.0.0",
|
||||
"cpufeatures",
|
||||
"digest 0.10.3",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "shlex"
|
||||
version = "1.1.0"
|
||||
|
|
|
|||
|
|
@ -24,15 +24,15 @@ async-trait = "0.1.7"
|
|||
base64 = "0.13"
|
||||
bytes = "1.0"
|
||||
chrono = "0.4"
|
||||
crypto-mac = "0.10"
|
||||
crypto-common = "0.1"
|
||||
err-derive = "0.3"
|
||||
hex = "0.4"
|
||||
hmac = "0.10"
|
||||
hmac = "0.12"
|
||||
idna = "0.2"
|
||||
tracing = "0.1.30"
|
||||
md-5 = "0.9"
|
||||
md-5 = "0.10"
|
||||
nom = "7.1"
|
||||
sha2 = "0.9"
|
||||
sha2 = "0.10"
|
||||
|
||||
futures = "0.3"
|
||||
futures-util = "0.3"
|
||||
|
|
|
|||
|
|
@ -365,7 +365,10 @@ pub async fn handle_upload_part_copy(
|
|||
// we need to insert that data as a new block.
|
||||
async move {
|
||||
if must_upload {
|
||||
garage2.block_manager.rpc_put_block(final_hash, data).await
|
||||
garage2
|
||||
.block_manager
|
||||
.rpc_put_block(final_hash, data.into())
|
||||
.await
|
||||
} else {
|
||||
Ok(())
|
||||
}
|
||||
|
|
|
|||
|
|
@ -9,6 +9,7 @@ use md5::{digest::generic_array::*, Digest as Md5Digest, Md5};
|
|||
use sha2::Sha256;
|
||||
|
||||
use garage_table::*;
|
||||
use garage_util::async_hash::*;
|
||||
use garage_util::data::*;
|
||||
use garage_util::error::Error as GarageError;
|
||||
use garage_util::time::*;
|
||||
|
|
@ -130,7 +131,8 @@ pub(crate) async fn save_stream<S: Stream<Item = Result<Bytes, Error>> + Unpin>(
|
|||
garage.version_table.insert(&version).await?;
|
||||
|
||||
// Transfer data and verify checksum
|
||||
let first_block_hash = blake2sum(&first_block[..]);
|
||||
let first_block = Bytes::from(first_block);
|
||||
let first_block_hash = async_blake2sum(first_block.clone()).await;
|
||||
|
||||
let tx_result = (|| async {
|
||||
let (total_size, data_md5sum, data_sha256sum) = read_and_put_blocks(
|
||||
|
|
@ -273,14 +275,16 @@ async fn read_and_put_blocks<S: Stream<Item = Result<Bytes, Error>> + Unpin>(
|
|||
garage: &Garage,
|
||||
version: &Version,
|
||||
part_number: u64,
|
||||
first_block: Vec<u8>,
|
||||
first_block: Bytes,
|
||||
first_block_hash: Hash,
|
||||
chunker: &mut StreamChunker<S>,
|
||||
) -> Result<(u64, GenericArray<u8, typenum::U16>, Hash), Error> {
|
||||
let mut md5hasher = Md5::new();
|
||||
let mut sha256hasher = Sha256::new();
|
||||
md5hasher.update(&first_block[..]);
|
||||
sha256hasher.update(&first_block[..]);
|
||||
let first_block = Bytes::from(first_block);
|
||||
|
||||
let md5hasher = AsyncHasher::<Md5>::new();
|
||||
let sha256hasher = AsyncHasher::<Sha256>::new();
|
||||
md5hasher.update(first_block.clone());
|
||||
sha256hasher.update(first_block.clone());
|
||||
|
||||
let mut next_offset = first_block.len();
|
||||
let mut put_curr_version_block = put_block_meta(
|
||||
|
|
@ -302,9 +306,10 @@ async fn read_and_put_blocks<S: Stream<Item = Result<Bytes, Error>> + Unpin>(
|
|||
chunker.next(),
|
||||
)?;
|
||||
if let Some(block) = next_block {
|
||||
md5hasher.update(&block[..]);
|
||||
sha256hasher.update(&block[..]);
|
||||
let block_hash = blake2sum(&block[..]);
|
||||
let block = Bytes::from(block);
|
||||
md5hasher.update(block.clone());
|
||||
sha256hasher.update(block.clone());
|
||||
let block_hash = async_blake2sum(block.clone()).await;
|
||||
let block_len = block.len();
|
||||
put_curr_version_block = put_block_meta(
|
||||
garage,
|
||||
|
|
@ -322,9 +327,9 @@ async fn read_and_put_blocks<S: Stream<Item = Result<Bytes, Error>> + Unpin>(
|
|||
}
|
||||
|
||||
let total_size = next_offset as u64;
|
||||
let data_md5sum = md5hasher.finalize();
|
||||
let data_md5sum = md5hasher.finalize().await;
|
||||
|
||||
let data_sha256sum = sha256hasher.finalize();
|
||||
let data_sha256sum = sha256hasher.finalize().await;
|
||||
let data_sha256sum = Hash::try_from(&data_sha256sum[..]).unwrap();
|
||||
|
||||
Ok((total_size, data_md5sum, data_sha256sum))
|
||||
|
|
@ -504,7 +509,10 @@ pub async fn handle_put_part(
|
|||
|
||||
// Copy block to store
|
||||
let version = Version::new(version_uuid, bucket_id, key, false);
|
||||
let first_block_hash = blake2sum(&first_block[..]);
|
||||
|
||||
let first_block = Bytes::from(first_block);
|
||||
let first_block_hash = async_blake2sum(first_block.clone()).await;
|
||||
|
||||
let (_, data_md5sum, data_sha256sum) = read_and_put_blocks(
|
||||
&garage,
|
||||
&version,
|
||||
|
|
|
|||
|
|
@ -1,5 +1,5 @@
|
|||
use chrono::{DateTime, Utc};
|
||||
use hmac::{Hmac, Mac, NewMac};
|
||||
use hmac::{Hmac, Mac};
|
||||
use sha2::Sha256;
|
||||
|
||||
use garage_util::data::{sha256sum, Hash};
|
||||
|
|
@ -29,17 +29,17 @@ pub fn signing_hmac(
|
|||
secret_key: &str,
|
||||
region: &str,
|
||||
service: &str,
|
||||
) -> Result<HmacSha256, crypto_mac::InvalidKeyLength> {
|
||||
) -> Result<HmacSha256, crypto_common::InvalidLength> {
|
||||
let secret = String::from("AWS4") + secret_key;
|
||||
let mut date_hmac = HmacSha256::new_varkey(secret.as_bytes())?;
|
||||
let mut date_hmac = HmacSha256::new_from_slice(secret.as_bytes())?;
|
||||
date_hmac.update(datetime.format(SHORT_DATE).to_string().as_bytes());
|
||||
let mut region_hmac = HmacSha256::new_varkey(&date_hmac.finalize().into_bytes())?;
|
||||
let mut region_hmac = HmacSha256::new_from_slice(&date_hmac.finalize().into_bytes())?;
|
||||
region_hmac.update(region.as_bytes());
|
||||
let mut service_hmac = HmacSha256::new_varkey(®ion_hmac.finalize().into_bytes())?;
|
||||
let mut service_hmac = HmacSha256::new_from_slice(®ion_hmac.finalize().into_bytes())?;
|
||||
service_hmac.update(service.as_bytes());
|
||||
let mut signing_hmac = HmacSha256::new_varkey(&service_hmac.finalize().into_bytes())?;
|
||||
let mut signing_hmac = HmacSha256::new_from_slice(&service_hmac.finalize().into_bytes())?;
|
||||
signing_hmac.update(b"aws4_request");
|
||||
let hmac = HmacSha256::new_varkey(&signing_hmac.finalize().into_bytes())?;
|
||||
let hmac = HmacSha256::new_from_slice(&signing_hmac.finalize().into_bytes())?;
|
||||
Ok(hmac)
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -1,3 +1,4 @@
|
|||
use bytes::Bytes;
|
||||
use serde::{Deserialize, Serialize};
|
||||
use zstd::stream::{decode_all as zstd_decode, Encoder};
|
||||
|
||||
|
|
@ -61,13 +62,17 @@ impl DataBlock {
|
|||
}
|
||||
}
|
||||
|
||||
pub fn from_buffer(data: Vec<u8>, level: Option<i32>) -> DataBlock {
|
||||
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) = zstd_encode(&data[..], level) {
|
||||
return DataBlock::Compressed(data);
|
||||
}
|
||||
}
|
||||
DataBlock::Plain(data)
|
||||
DataBlock::Plain(data.to_vec()) // TODO: remove to_vec here
|
||||
})
|
||||
.await
|
||||
.unwrap()
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -5,6 +5,7 @@ use std::time::Duration;
|
|||
|
||||
use arc_swap::ArcSwapOption;
|
||||
use async_trait::async_trait;
|
||||
use bytes::Bytes;
|
||||
use serde::{Deserialize, Serialize};
|
||||
|
||||
use futures::future::*;
|
||||
|
|
@ -211,14 +212,15 @@ impl BlockManager {
|
|||
}
|
||||
|
||||
/// Send block to nodes that should have it
|
||||
pub async fn rpc_put_block(&self, hash: Hash, data: Vec<u8>) -> Result<(), Error> {
|
||||
pub async fn rpc_put_block(&self, hash: Hash, data: Bytes) -> Result<(), Error> {
|
||||
let who = self.replication.write_nodes(&hash);
|
||||
let data = DataBlock::from_buffer(data, self.compression_level);
|
||||
let data = DataBlock::from_buffer(data, self.compression_level).await;
|
||||
self.system
|
||||
.rpc
|
||||
.try_call_many(
|
||||
&self.endpoint,
|
||||
&who[..],
|
||||
// TODO: remove to_vec() here
|
||||
BlockRpc::PutBlock { hash, data },
|
||||
RequestStrategy::with_priority(PRIO_NORMAL)
|
||||
.with_quorum(self.replication.write_quorum())
|
||||
|
|
|
|||
|
|
@ -65,7 +65,7 @@ chrono = "0.4"
|
|||
http = "0.2"
|
||||
hmac = "0.10"
|
||||
hyper = { version = "0.14", features = ["client", "http1", "runtime"] }
|
||||
sha2 = "0.9"
|
||||
sha2 = "0.10"
|
||||
|
||||
static_init = "1.0"
|
||||
assert-json-diff = "2.0"
|
||||
|
|
|
|||
|
|
@ -18,12 +18,14 @@ garage_db = { version = "0.8.0", path = "../db" }
|
|||
|
||||
async-trait = "0.1"
|
||||
blake2 = "0.9"
|
||||
bytes = "1.0"
|
||||
digest = "0.10"
|
||||
err-derive = "0.3"
|
||||
xxhash-rust = { version = "0.8", default-features = false, features = ["xxh3"] }
|
||||
hex = "0.4"
|
||||
tracing = "0.1.30"
|
||||
rand = "0.8"
|
||||
sha2 = "0.9"
|
||||
sha2 = "0.10"
|
||||
|
||||
chrono = "0.4"
|
||||
rmp-serde = "0.15"
|
||||
|
|
|
|||
55
src/util/async_hash.rs
Normal file
55
src/util/async_hash.rs
Normal file
|
|
@ -0,0 +1,55 @@
|
|||
use bytes::Bytes;
|
||||
use digest::Digest;
|
||||
|
||||
use tokio::sync::mpsc;
|
||||
use tokio::task::JoinHandle;
|
||||
|
||||
use crate::data::*;
|
||||
|
||||
/// Compute the sha256 of a slice,
|
||||
/// spawning on a tokio thread for CPU-intensive processing
|
||||
/// The argument has to be an owned Bytes, as it is moved out to a new thread.
|
||||
pub async fn async_sha256sum(data: Bytes) -> Hash {
|
||||
tokio::task::spawn_blocking(move || sha256sum(&data))
|
||||
.await
|
||||
.unwrap()
|
||||
}
|
||||
|
||||
/// Compute the blake2sum of a slice,
|
||||
/// spawning on a tokio thread for CPU-intensive processing.
|
||||
/// The argument has to be an owned Bytes, as it is moved out to a new thread.
|
||||
pub async fn async_blake2sum(data: Bytes) -> Hash {
|
||||
tokio::task::spawn_blocking(move || blake2sum(&data))
|
||||
.await
|
||||
.unwrap()
|
||||
}
|
||||
|
||||
// ----
|
||||
|
||||
pub struct AsyncHasher<D: Digest> {
|
||||
sendblk: mpsc::UnboundedSender<Bytes>,
|
||||
task: JoinHandle<digest::Output<D>>,
|
||||
}
|
||||
|
||||
impl<D: Digest> AsyncHasher<D> {
|
||||
pub fn new() -> Self {
|
||||
let (sendblk, mut recvblk) = mpsc::unbounded_channel::<Bytes>();
|
||||
let task = tokio::task::spawn_blocking(move || {
|
||||
let mut digest = D::new();
|
||||
while let Some(blk) = recvblk.blocking_recv() {
|
||||
digest.update(&blk[..]);
|
||||
}
|
||||
digest.finalize()
|
||||
});
|
||||
Self { sendblk, task }
|
||||
}
|
||||
|
||||
pub fn update(&self, b: Bytes) {
|
||||
self.sendblk.send(b).unwrap()
|
||||
}
|
||||
|
||||
pub async fn finalize(self) -> digest::Output<D> {
|
||||
drop(self.sendblk);
|
||||
self.task.await.unwrap()
|
||||
}
|
||||
}
|
||||
|
|
@ -3,6 +3,7 @@
|
|||
#[macro_use]
|
||||
extern crate tracing;
|
||||
|
||||
pub mod async_hash;
|
||||
pub mod background;
|
||||
pub mod config;
|
||||
pub mod crdt;
|
||||
|
|
|
|||
Loading…
Reference in a new issue