mirror of
https://github.com/rustfs/rustfs.git
synced 2026-08-16 01:48:21 +00:00
Zhengchao An
eb41f45175
Removing both blankets exposes 23 items, of which only four are deleted. The ratio is the point: close to the core data path the blankets were hiding test assertions and migration seams, not dead code. A cfg-split function is the reason two symbols in the internode transport look dead when neither is. build_internode_data_transport_from_env has two bodies, one under #[cfg(test)] that calls build_internode_data_transport directly and one under #[cfg(not(test))] that goes through the INTERNODE_DATA_TRANSPORT static so tests do not share process-global transport state. Each half's helper is live in exactly one build, and because cargo check --tests compiles both the lib target and the test harness, both symbols appear in one warning list. Deleting either one breaks the other lane. Both are kept with allows naming their half. Three deletion candidates were withdrawn after a per-name grep: ParallelReader::new, ErasureDecodeReader::new and SyncErasureDecodeReader::new all have test callers. The last two are exactly the shape of the dead wrapper deleted in #6084 — a thin forward to a new_with_metrics_path sibling — except that sibling is live in production (set_disk/read.rs) and the wrappers are used by tests. Deleted: - RemotePeerS3Client::get_addr and RemoteLocker::from_url, neither with a consumer in any lane. - RemotePeerS3Client's node field, which new writes after using it to derive addr and nothing ever reads. Its only other writer was a test helper that built a whole Node solely to fill the field; that block goes too. - ParallelReader::can_decode, superseded by an inlined copy. The copy's comment named the method it replaced, so deleting the method alone would have left a dangling reference; the comment now describes the check instead of pointing at a method that no longer exists. Kept with allows: the erasure items are decode/encode invariants asserted by their own files' tests (shard_read_launch_order, decode_with_read_costs, emit_data_shards, queued_block_bytes, the engine trait facets, the ParallelReader and decode-reader constructors, encode_stream_callback_async). On the cluster side, peer_replay_state, heal_bucket_local and clone_drives are test-only, InternodeDataTransportCapabilities and tcp_http are constructed only by transport test doubles, and the InternodeDataTransport trait's name/capabilities pair is an unused capability-negotiation facet kept for the transport split (backlog#1350) — six impls provide them and no caller negotiates on them yet. Verification, four lanes warning-free: default, --tests, --features rio-v2 --tests, --features test-util --tests. cargo nextest run -p rustfs-ecstore 4041 passed; clippy --lib --tests -D warnings clean; make pre-commit exit 0. Ref rustfs/backlog#1823 (step 2).
2718 lines
108 KiB
Rust
2718 lines
108 KiB
Rust
// Copyright 2024 RustFS Team
|
|
//
|
|
// Licensed under the Apache License, Version 2.0 (the "License");
|
|
// you may not use this file except in compliance with the License.
|
|
// You may obtain a copy of the License at
|
|
//
|
|
// http://www.apache.org/licenses/LICENSE-2.0
|
|
//
|
|
// Unless required by applicable law or agreed to in writing, software
|
|
// distributed under the License is distributed on an "AS IS" BASIS,
|
|
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
// See the License for the specific language governing permissions and
|
|
// limitations under the License.
|
|
|
|
//! Erasure coding implementation using reed-solomon-erasure (GF(2^8)).
|
|
//! Supports legacy (reed-solomon-simd) for reading/healing old-version files.
|
|
//!
|
|
|
|
use bytes::{Bytes, BytesMut};
|
|
use reed_solomon_erasure::galois_8::ReedSolomon;
|
|
use reed_solomon_simd;
|
|
use smallvec::SmallVec;
|
|
use std::{
|
|
collections::HashMap,
|
|
io,
|
|
sync::{Arc, OnceLock, RwLock},
|
|
};
|
|
use tokio::io::AsyncRead;
|
|
use tracing::warn;
|
|
use uuid::Uuid;
|
|
|
|
pub(crate) struct EncodedBlock {
|
|
data: Bytes,
|
|
shard_size: usize,
|
|
}
|
|
|
|
impl EncodedBlock {
|
|
fn empty() -> Self {
|
|
Self {
|
|
data: Bytes::new(),
|
|
shard_size: 0,
|
|
}
|
|
}
|
|
|
|
pub(crate) fn is_empty(&self) -> bool {
|
|
self.data.is_empty()
|
|
}
|
|
|
|
pub(crate) fn queued_bytes(&self) -> usize {
|
|
self.data.len()
|
|
}
|
|
|
|
pub(crate) fn shards(&self) -> impl ExactSizeIterator<Item = &[u8]> {
|
|
debug_assert!(self.shard_size > 0, "only non-empty encoded blocks reach shard writers");
|
|
debug_assert_eq!(self.data.len() % self.shard_size, 0);
|
|
self.data.chunks_exact(self.shard_size)
|
|
}
|
|
|
|
fn into_shards(mut self, shard_count: usize) -> Vec<Bytes> {
|
|
if self.shard_size == 0 {
|
|
return vec![Bytes::new(); shard_count];
|
|
}
|
|
|
|
let mut shards = Vec::with_capacity(shard_count);
|
|
for _ in 0..shard_count {
|
|
shards.push(self.data.split_to(self.shard_size));
|
|
}
|
|
shards
|
|
}
|
|
}
|
|
|
|
const MODERN_MAX_TOTAL_SHARDS: usize = <reed_solomon_erasure::galois_8::Field as reed_solomon_erasure::Field>::ORDER;
|
|
const MODERN_REED_SOLOMON_CACHE_MAX_ENTRIES: usize = 64;
|
|
const LEGACY_REED_SOLOMON_CACHE_MAX_ENTRIES: usize = 16;
|
|
// Vec growth may retain twice the requested logical length. Keeping the logical
|
|
// workspace at half the budget bounds each cached workspace's shard allocation to 1 MiB.
|
|
const LEGACY_REED_SOLOMON_CACHE_MAX_LOGICAL_SHARD_BYTES_PER_WORKSPACE: usize = 512 * 1024;
|
|
|
|
type ModernReedSolomonCache = RwLock<HashMap<(usize, usize), Arc<ReedSolomon>>>;
|
|
type LegacyReedSolomonCache = RwLock<HashMap<(usize, usize), Arc<LegacyReedSolomonEncoder>>>;
|
|
|
|
static MODERN_REED_SOLOMON_CACHE: OnceLock<ModernReedSolomonCache> = OnceLock::new();
|
|
static LEGACY_REED_SOLOMON_CACHE: OnceLock<LegacyReedSolomonCache> = OnceLock::new();
|
|
|
|
/// Errors returned when constructing an [`Erasure`] codec.
|
|
#[derive(Debug, thiserror::Error)]
|
|
pub enum ErasureConstructionError {
|
|
/// Erasure coding requires at least one data shard.
|
|
#[error("data_shards must be greater than zero")]
|
|
ZeroDataShards,
|
|
|
|
/// Erasure coding requires a non-zero logical block size.
|
|
#[error("block_size must be greater than zero")]
|
|
ZeroBlockSize,
|
|
|
|
/// The total shard count cannot be represented by `usize`.
|
|
#[error("data_shards ({data_shards}) + parity_shards ({parity_shards}) overflows usize")]
|
|
ShardCountOverflow { data_shards: usize, parity_shards: usize },
|
|
|
|
/// The modern GF(2^8) codec cannot represent this shard configuration.
|
|
#[error("modern codec does not support {data_shards} data shards and {parity_shards} parity shards")]
|
|
UnsupportedModernShardCount { data_shards: usize, parity_shards: usize },
|
|
|
|
/// The legacy SIMD codec cannot represent this shard configuration.
|
|
#[error("legacy codec does not support {data_shards} data shards and {parity_shards} parity shards")]
|
|
UnsupportedLegacyShardCount { data_shards: usize, parity_shards: usize },
|
|
|
|
/// The modern encoder rejected dimensions that passed the preflight checks.
|
|
#[error("failed to construct modern Reed-Solomon encoder")]
|
|
ModernEncoder {
|
|
#[source]
|
|
source: reed_solomon_erasure::Error,
|
|
},
|
|
|
|
/// The legacy encoder wrapper failed to initialize.
|
|
#[error("failed to construct legacy Reed-Solomon encoder")]
|
|
LegacyEncoder {
|
|
#[source]
|
|
source: io::Error,
|
|
},
|
|
}
|
|
|
|
#[derive(Debug, thiserror::Error)]
|
|
#[error("{source}")]
|
|
struct ErasureConstructionIoSource {
|
|
#[source]
|
|
source: ErasureConstructionError,
|
|
}
|
|
|
|
impl ErasureConstructionError {
|
|
pub(crate) fn into_io_error(self) -> io::Error {
|
|
// `io::Error::source` forwards to the wrapped error's source rather
|
|
// than exposing the wrapped error itself. This adapter keeps the
|
|
// construction error as an observable link in the source chain.
|
|
io::Error::other(ErasureConstructionIoSource { source: self })
|
|
}
|
|
}
|
|
|
|
/// Legacy calc_shard_size formula: (block_size.div_ceil(data_shards) + 1) & !1
|
|
/// Matches main branch and filemeta::ErasureInfo for old-version files.
|
|
pub fn calc_shard_size_legacy(block_size: usize, data_shards: usize) -> usize {
|
|
(block_size.div_ceil(data_shards) + 1) & !1
|
|
}
|
|
|
|
/// Reed-Solomon encoder for legacy (main branch) format using reed-solomon-simd.
|
|
/// Used when decoding/encoding files with uses_legacy_checksum == true.
|
|
struct LegacyReedSolomonEncoder {
|
|
data_shards: usize,
|
|
parity_shards: usize,
|
|
cache_workspaces: bool,
|
|
encoder_cache: RwLock<Option<reed_solomon_simd::ReedSolomonEncoder>>,
|
|
decoder_cache: RwLock<Option<reed_solomon_simd::ReedSolomonDecoder>>,
|
|
}
|
|
|
|
impl LegacyReedSolomonEncoder {
|
|
fn new(data_shards: usize, parity_shards: usize) -> io::Result<Self> {
|
|
Self::with_workspace_cache(data_shards, parity_shards, false)
|
|
}
|
|
|
|
fn with_workspace_cache(data_shards: usize, parity_shards: usize, cache_workspaces: bool) -> io::Result<Self> {
|
|
Ok(Self {
|
|
data_shards,
|
|
parity_shards,
|
|
cache_workspaces,
|
|
encoder_cache: RwLock::new(None),
|
|
decoder_cache: RwLock::new(None),
|
|
})
|
|
}
|
|
|
|
fn logical_shard_bytes_upper_bound(&self, shard_len: usize) -> Option<usize> {
|
|
let aligned_shard_len = shard_len.checked_add(63)?.checked_div(64)?.checked_mul(64)?;
|
|
let high_rate_decoder_work_count = self
|
|
.parity_shards
|
|
.checked_next_power_of_two()?
|
|
.checked_add(self.data_shards)?
|
|
.checked_next_power_of_two()?;
|
|
let low_rate_decoder_work_count = self
|
|
.data_shards
|
|
.checked_next_power_of_two()?
|
|
.checked_add(self.parity_shards)?
|
|
.checked_next_power_of_two()?;
|
|
aligned_shard_len.checked_mul(high_rate_decoder_work_count.max(low_rate_decoder_work_count))
|
|
}
|
|
|
|
fn should_cache_workspace(&self, shard_len: usize) -> bool {
|
|
self.cache_workspaces
|
|
&& self
|
|
.logical_shard_bytes_upper_bound(shard_len)
|
|
.is_some_and(|bytes| bytes <= LEGACY_REED_SOLOMON_CACHE_MAX_LOGICAL_SHARD_BYTES_PER_WORKSPACE)
|
|
}
|
|
|
|
fn encode(&self, shards: SmallVec<[&mut [u8]; 16]>) -> io::Result<()> {
|
|
let mut shards_vec: Vec<&mut [u8]> = shards.into_vec();
|
|
if shards_vec.is_empty() {
|
|
return Ok(());
|
|
}
|
|
let shard_len = shards_vec[0].len();
|
|
let cached_encoder = self
|
|
.encoder_cache
|
|
.write()
|
|
.map_err(|_| io::Error::other("Failed to acquire encoder cache lock"))?
|
|
.take();
|
|
let mut encoder = {
|
|
match cached_encoder {
|
|
Some(mut cached) => {
|
|
if cached.reset(self.data_shards, self.parity_shards, shard_len).is_err() {
|
|
reed_solomon_simd::ReedSolomonEncoder::new(self.data_shards, self.parity_shards, shard_len)
|
|
.map_err(|e| io::Error::other(format!("Failed to create SIMD encoder: {e:?}")))?
|
|
} else {
|
|
cached
|
|
}
|
|
}
|
|
None => reed_solomon_simd::ReedSolomonEncoder::new(self.data_shards, self.parity_shards, shard_len)
|
|
.map_err(|e| io::Error::other(format!("Failed to create SIMD encoder: {e:?}")))?,
|
|
}
|
|
};
|
|
for (i, shard) in shards_vec.iter().enumerate().take(self.data_shards) {
|
|
encoder
|
|
.add_original_shard(shard)
|
|
.map_err(|e| io::Error::other(format!("Failed to add shard {i}: {e:?}")))?;
|
|
}
|
|
let result = encoder
|
|
.encode()
|
|
.map_err(|e| io::Error::other(format!("SIMD encoding failed: {e:?}")))?;
|
|
for (i, recovery_shard) in result.recovery_iter().enumerate() {
|
|
if i + self.data_shards < shards_vec.len() {
|
|
shards_vec[i + self.data_shards].copy_from_slice(recovery_shard);
|
|
}
|
|
}
|
|
drop(result);
|
|
if self.should_cache_workspace(shard_len) {
|
|
let mut cache = self
|
|
.encoder_cache
|
|
.write()
|
|
.map_err(|_| io::Error::other("Failed to return encoder to cache"))?;
|
|
if cache.is_none() {
|
|
*cache = Some(encoder);
|
|
}
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
fn reconstruct_data(&self, shards: &mut [Option<Vec<u8>>]) -> io::Result<()> {
|
|
if recover_empty_payload_data_shards(shards, self.data_shards, self.parity_shards)? {
|
|
return Ok(());
|
|
}
|
|
|
|
let shard_len = shards
|
|
.iter()
|
|
.find_map(|s| s.as_ref().map(|v| v.len()))
|
|
.ok_or_else(|| io::Error::other("No valid shards found for reconstruction"))?;
|
|
|
|
let cached_decoder = self
|
|
.decoder_cache
|
|
.write()
|
|
.map_err(|_| io::Error::other("Failed to acquire decoder cache lock"))?
|
|
.take();
|
|
let mut decoder = {
|
|
match cached_decoder {
|
|
Some(mut cached_decoder) => {
|
|
if let Err(e) = cached_decoder.reset(self.data_shards, self.parity_shards, shard_len) {
|
|
warn!("Failed to reset SIMD decoder: {:?}, creating new one", e);
|
|
reed_solomon_simd::ReedSolomonDecoder::new(self.data_shards, self.parity_shards, shard_len)
|
|
.map_err(|e| io::Error::other(format!("Failed to create SIMD decoder: {e:?}")))?
|
|
} else {
|
|
cached_decoder
|
|
}
|
|
}
|
|
None => reed_solomon_simd::ReedSolomonDecoder::new(self.data_shards, self.parity_shards, shard_len)
|
|
.map_err(|e| io::Error::other(format!("Failed to create SIMD decoder: {e:?}")))?,
|
|
}
|
|
};
|
|
|
|
for (i, shard_opt) in shards.iter().enumerate() {
|
|
if let Some(shard) = shard_opt {
|
|
if i < self.data_shards {
|
|
decoder
|
|
.add_original_shard(i, shard)
|
|
.map_err(|e| io::Error::other(format!("Failed to add original shard for reconstruction: {e:?}")))?;
|
|
} else {
|
|
let recovery_idx = i - self.data_shards;
|
|
decoder
|
|
.add_recovery_shard(recovery_idx, shard)
|
|
.map_err(|e| io::Error::other(format!("Failed to add recovery shard for reconstruction: {e:?}")))?;
|
|
}
|
|
}
|
|
}
|
|
|
|
let result = decoder
|
|
.decode()
|
|
.map_err(|e| io::Error::other(format!("SIMD decode error: {e:?}")))?;
|
|
|
|
for (i, shard_opt) in shards.iter_mut().enumerate() {
|
|
if shard_opt.is_none() && i < self.data_shards {
|
|
for (restored_index, restored_data) in result.restored_original_iter() {
|
|
if restored_index == i {
|
|
*shard_opt = Some(restored_data.to_vec());
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
drop(result);
|
|
|
|
if self.should_cache_workspace(shard_len) {
|
|
let mut cache = self
|
|
.decoder_cache
|
|
.write()
|
|
.map_err(|_| io::Error::other("Failed to return decoder to cache"))?;
|
|
if cache.is_none() {
|
|
*cache = Some(decoder);
|
|
}
|
|
}
|
|
|
|
Ok(())
|
|
}
|
|
|
|
fn reconstruct(&self, shards: &mut [Option<Vec<u8>>]) -> io::Result<()> {
|
|
self.reconstruct_data(shards)?;
|
|
self.encode_parity(shards)
|
|
}
|
|
|
|
fn verify(&self, shards: &[&[u8]]) -> io::Result<bool> {
|
|
let expected_shards = self.data_shards + self.parity_shards;
|
|
if shards.len() != expected_shards {
|
|
return Err(io::Error::other(format!(
|
|
"invalid shard count: got {}, expected {}",
|
|
shards.len(),
|
|
expected_shards
|
|
)));
|
|
}
|
|
if shards.iter().all(|shard| shard.is_empty()) {
|
|
return Ok(true);
|
|
}
|
|
|
|
let mut expected = shards.iter().map(|shard| Some(shard.to_vec())).collect::<Vec<_>>();
|
|
self.encode_parity(&mut expected)?;
|
|
|
|
for index in self.data_shards..expected_shards {
|
|
let Some(expected_parity) = expected[index].as_ref() else {
|
|
return Err(io::Error::other(format!("missing parity shard {index} after verification encode")));
|
|
};
|
|
if expected_parity.as_slice() != shards[index] {
|
|
return Ok(false);
|
|
}
|
|
}
|
|
|
|
Ok(true)
|
|
}
|
|
|
|
fn encode_parity(&self, shards: &mut [Option<Vec<u8>>]) -> io::Result<()> {
|
|
encode_parity_shards(shards, self.data_shards, self.parity_shards, |shards| self.encode(shards))
|
|
}
|
|
}
|
|
|
|
/// Reed-Solomon encoder using reed-solomon-erasure
|
|
pub struct ReedSolomonEncoder {
|
|
data_shards: usize,
|
|
parity_shards: usize,
|
|
encoder: Option<Arc<ReedSolomon>>,
|
|
}
|
|
|
|
impl Clone for ReedSolomonEncoder {
|
|
fn clone(&self) -> Self {
|
|
Self {
|
|
data_shards: self.data_shards,
|
|
parity_shards: self.parity_shards,
|
|
encoder: self.encoder.clone(),
|
|
}
|
|
}
|
|
}
|
|
|
|
impl ReedSolomonEncoder {
|
|
fn try_new_typed(data_shards: usize, parity_shards: usize) -> Result<Self, reed_solomon_erasure::Error> {
|
|
let encoder = if parity_shards > 0 {
|
|
Some(cached_modern_reed_solomon(data_shards, parity_shards)?)
|
|
} else {
|
|
None
|
|
};
|
|
|
|
Ok(ReedSolomonEncoder {
|
|
data_shards,
|
|
parity_shards,
|
|
encoder,
|
|
})
|
|
}
|
|
|
|
/// Create a new Reed-Solomon encoder with specified data and parity shards.
|
|
pub fn new(data_shards: usize, parity_shards: usize) -> io::Result<Self> {
|
|
Self::try_new_typed(data_shards, parity_shards)
|
|
.map_err(|error| io::Error::other(format!("Failed to create Reed-Solomon encoder: {error:?}")))
|
|
}
|
|
|
|
/// Encode data shards with parity.
|
|
pub fn encode(&self, shards: SmallVec<[&mut [u8]; 16]>) -> io::Result<()> {
|
|
let mut shards_vec: Vec<&mut [u8]> = shards.into_vec();
|
|
if shards_vec.is_empty() {
|
|
return Ok(());
|
|
}
|
|
|
|
if let Some(ref rs) = self.encoder {
|
|
rs.encode(&mut shards_vec)
|
|
.map_err(|e| io::Error::other(format!("Reed-Solomon encode failed: {e:?}")))
|
|
} else {
|
|
Ok(())
|
|
}
|
|
}
|
|
|
|
/// Reconstruct missing data shards.
|
|
pub fn reconstruct_data(&self, shards: &mut [Option<Vec<u8>>]) -> io::Result<()> {
|
|
if recover_empty_payload_data_shards(shards, self.data_shards, self.parity_shards)? {
|
|
return Ok(());
|
|
}
|
|
|
|
if let Some(ref rs) = self.encoder {
|
|
rs.reconstruct_data(shards)
|
|
.map_err(|e| io::Error::other(format!("Reed-Solomon reconstruct failed: {e:?}")))
|
|
} else {
|
|
Ok(())
|
|
}
|
|
}
|
|
|
|
/// Reconstruct missing data shards and regenerate parity shards.
|
|
pub fn reconstruct(&self, shards: &mut [Option<Vec<u8>>]) -> io::Result<()> {
|
|
self.reconstruct_data(shards)?;
|
|
self.encode_parity(shards)
|
|
}
|
|
|
|
pub fn verify(&self, shards: &[&[u8]]) -> io::Result<bool> {
|
|
if shards.iter().all(|shard| shard.is_empty()) {
|
|
return Ok(true);
|
|
}
|
|
|
|
if let Some(ref rs) = self.encoder {
|
|
rs.verify(shards)
|
|
.map_err(|e| io::Error::other(format!("Reed-Solomon verify failed: {e:?}")))
|
|
} else {
|
|
Ok(true)
|
|
}
|
|
}
|
|
|
|
fn encode_parity(&self, shards: &mut [Option<Vec<u8>>]) -> io::Result<()> {
|
|
encode_parity_shards(shards, self.data_shards, self.parity_shards, |shards| self.encode(shards))
|
|
}
|
|
}
|
|
|
|
fn cached_modern_reed_solomon(data_shards: usize, parity_shards: usize) -> Result<Arc<ReedSolomon>, reed_solomon_erasure::Error> {
|
|
let key = (data_shards, parity_shards);
|
|
let cache = MODERN_REED_SOLOMON_CACHE.get_or_init(|| RwLock::new(HashMap::new()));
|
|
|
|
if let Some(encoder) = cache
|
|
.read()
|
|
.unwrap_or_else(|poisoned| poisoned.into_inner())
|
|
.get(&key)
|
|
.cloned()
|
|
{
|
|
return Ok(encoder);
|
|
}
|
|
|
|
let encoder = Arc::new(ReedSolomon::new(data_shards, parity_shards)?);
|
|
let mut cache = cache.write().unwrap_or_else(|poisoned| poisoned.into_inner());
|
|
if let Some(existing) = cache.get(&key) {
|
|
return Ok(Arc::clone(existing));
|
|
}
|
|
if cache.len() < MODERN_REED_SOLOMON_CACHE_MAX_ENTRIES {
|
|
cache.insert(key, Arc::clone(&encoder));
|
|
}
|
|
Ok(encoder)
|
|
}
|
|
|
|
fn cached_legacy_reed_solomon(data_shards: usize, parity_shards: usize) -> io::Result<Arc<LegacyReedSolomonEncoder>> {
|
|
let cache = LEGACY_REED_SOLOMON_CACHE.get_or_init(|| RwLock::new(HashMap::new()));
|
|
cached_legacy_reed_solomon_in(cache, data_shards, parity_shards)
|
|
}
|
|
|
|
fn cached_legacy_reed_solomon_in(
|
|
cache: &LegacyReedSolomonCache,
|
|
data_shards: usize,
|
|
parity_shards: usize,
|
|
) -> io::Result<Arc<LegacyReedSolomonEncoder>> {
|
|
let key = (data_shards, parity_shards);
|
|
if let Some(encoder) = cache
|
|
.read()
|
|
.unwrap_or_else(|poisoned| poisoned.into_inner())
|
|
.get(&key)
|
|
.cloned()
|
|
{
|
|
return Ok(encoder);
|
|
}
|
|
|
|
let mut cache = cache.write().unwrap_or_else(|poisoned| poisoned.into_inner());
|
|
if let Some(existing) = cache.get(&key) {
|
|
return Ok(Arc::clone(existing));
|
|
}
|
|
if cache.len() < LEGACY_REED_SOLOMON_CACHE_MAX_ENTRIES {
|
|
let encoder = Arc::new(LegacyReedSolomonEncoder::with_workspace_cache(data_shards, parity_shards, true)?);
|
|
cache.insert(key, Arc::clone(&encoder));
|
|
return Ok(encoder);
|
|
}
|
|
drop(cache);
|
|
Ok(Arc::new(LegacyReedSolomonEncoder::new(data_shards, parity_shards)?))
|
|
}
|
|
|
|
fn encode_parity_shards<F>(shards: &mut [Option<Vec<u8>>], data_shards: usize, parity_shards: usize, encode: F) -> io::Result<()>
|
|
where
|
|
F: FnOnce(SmallVec<[&mut [u8]; 16]>) -> io::Result<()>,
|
|
{
|
|
let expected_shards = data_shards + parity_shards;
|
|
if shards.len() != expected_shards {
|
|
return Err(io::Error::other(format!(
|
|
"invalid shard count: got {}, expected {}",
|
|
shards.len(),
|
|
expected_shards
|
|
)));
|
|
}
|
|
|
|
let shard_len = shards
|
|
.iter()
|
|
.find_map(|s| s.as_ref().map(Vec::len))
|
|
.ok_or_else(|| io::Error::other("No valid shards found for parity encoding"))?;
|
|
|
|
for shard in shards.iter_mut().skip(data_shards) {
|
|
if shard.is_none() {
|
|
*shard = Some(vec![0; shard_len]);
|
|
}
|
|
}
|
|
|
|
if shard_len == 0 {
|
|
for (index, shard) in shards.iter().enumerate() {
|
|
let shard = shard
|
|
.as_ref()
|
|
.ok_or_else(|| io::Error::other(format!("missing shard {index} after data reconstruction")))?;
|
|
if !shard.is_empty() {
|
|
return Err(io::Error::other(format!(
|
|
"inconsistent shard length at index {index}: got {}, expected {}",
|
|
shard.len(),
|
|
shard_len
|
|
)));
|
|
}
|
|
}
|
|
return Ok(());
|
|
}
|
|
|
|
let mut shard_refs: SmallVec<[&mut [u8]; 16]> = SmallVec::new();
|
|
for (index, shard) in shards.iter_mut().enumerate() {
|
|
let shard = shard
|
|
.as_mut()
|
|
.ok_or_else(|| io::Error::other(format!("missing shard {index} after data reconstruction")))?;
|
|
if shard.len() != shard_len {
|
|
return Err(io::Error::other(format!(
|
|
"inconsistent shard length at index {index}: got {}, expected {}",
|
|
shard.len(),
|
|
shard_len
|
|
)));
|
|
}
|
|
shard_refs.push(shard.as_mut_slice());
|
|
}
|
|
|
|
encode(shard_refs)
|
|
}
|
|
|
|
fn recover_empty_payload_data_shards(
|
|
shards: &mut [Option<Vec<u8>>],
|
|
data_shards: usize,
|
|
parity_shards: usize,
|
|
) -> io::Result<bool> {
|
|
let expected_shards = data_shards + parity_shards;
|
|
if shards.len() != expected_shards {
|
|
return Err(io::Error::other(format!(
|
|
"invalid shard count: got {}, expected {}",
|
|
shards.len(),
|
|
expected_shards
|
|
)));
|
|
}
|
|
|
|
let mut has_present_shard = false;
|
|
for shard in shards.iter().filter_map(|shard| shard.as_ref()) {
|
|
has_present_shard = true;
|
|
if !shard.is_empty() {
|
|
return Ok(false);
|
|
}
|
|
}
|
|
if !has_present_shard {
|
|
return Ok(false);
|
|
}
|
|
|
|
for shard in shards.iter_mut().take(data_shards) {
|
|
if shard.is_none() {
|
|
*shard = Some(Vec::new());
|
|
}
|
|
}
|
|
Ok(true)
|
|
}
|
|
|
|
/// Erasure coding utility for data reliability using Reed-Solomon codes.
|
|
///
|
|
/// This struct provides encoding and decoding of data into data and parity shards.
|
|
/// It supports splitting data into multiple shards, generating parity for fault tolerance,
|
|
/// and reconstructing lost shards.
|
|
///
|
|
/// # Fields
|
|
/// - `data_shards`: Number of data shards.
|
|
/// - `parity_shards`: Number of parity shards.
|
|
/// - `encoder`: Optional ReedSolomon encoder instance.
|
|
/// - `block_size`: Block size for each shard.
|
|
/// - `_id`: Unique identifier for the erasure instance.
|
|
///
|
|
/// # Example
|
|
/// ```ignore
|
|
/// use rustfs_ecstore::api::erasure::Erasure;
|
|
/// let erasure = Erasure::new(4, 2, 8);
|
|
/// let data = b"hello world";
|
|
/// let shards = erasure.encode_data(data).expect("operation should succeed");
|
|
/// // Simulate loss and recovery...
|
|
/// ```
|
|
pub struct Erasure {
|
|
pub data_shards: usize,
|
|
pub parity_shards: usize,
|
|
encoder: Option<ReedSolomonEncoder>,
|
|
legacy_encoder: Option<Arc<LegacyReedSolomonEncoder>>,
|
|
pub block_size: usize,
|
|
uses_legacy: bool,
|
|
_id: Uuid,
|
|
}
|
|
|
|
impl Default for Erasure {
|
|
fn default() -> Self {
|
|
Self {
|
|
data_shards: 0,
|
|
parity_shards: 0,
|
|
encoder: None,
|
|
legacy_encoder: None,
|
|
block_size: 0,
|
|
uses_legacy: false,
|
|
_id: Uuid::nil(),
|
|
}
|
|
}
|
|
}
|
|
|
|
impl Clone for Erasure {
|
|
fn clone(&self) -> Self {
|
|
Self {
|
|
data_shards: self.data_shards,
|
|
parity_shards: self.parity_shards,
|
|
encoder: self.encoder.clone(),
|
|
legacy_encoder: self.legacy_encoder.clone(),
|
|
block_size: self.block_size,
|
|
uses_legacy: self.uses_legacy,
|
|
_id: self._id, // Shared by clones; this field is unused in hot paths.
|
|
}
|
|
}
|
|
}
|
|
|
|
pub fn calc_shard_size(block_size: usize, data_shards: usize) -> usize {
|
|
block_size.div_ceil(data_shards)
|
|
}
|
|
|
|
impl Erasure {
|
|
/// Create a new Erasure instance
|
|
///
|
|
/// # Arguments
|
|
/// * `data_shards` - Number of data shards.
|
|
/// * `parity_shards` - Number of parity shards.
|
|
/// * `block_size` - Block size for each shard.
|
|
///
|
|
/// # Panics
|
|
/// Panics when the dimensions are invalid or unsupported. RustFS production
|
|
/// paths use [`Self::try_new`] instead.
|
|
pub fn new(data_shards: usize, parity_shards: usize, block_size: usize) -> Self {
|
|
match Self::try_new(data_shards, parity_shards, block_size) {
|
|
Ok(erasure) => erasure,
|
|
Err(error) => panic!("invalid erasure codec configuration: {error}"),
|
|
}
|
|
}
|
|
|
|
/// Create a new Erasure instance with legacy format support.
|
|
///
|
|
/// When `uses_legacy` is true, uses main-branch shard_size formula and reed-solomon-simd
|
|
/// for decode/reconstruct (for reading and healing old-version files).
|
|
///
|
|
/// # Panics
|
|
/// Panics when the dimensions are invalid or unsupported. RustFS production
|
|
/// paths use [`Self::try_new_with_options`] instead.
|
|
pub fn new_with_options(data_shards: usize, parity_shards: usize, block_size: usize, uses_legacy: bool) -> Self {
|
|
match Self::try_new_with_options(data_shards, parity_shards, block_size, uses_legacy) {
|
|
Ok(erasure) => erasure,
|
|
Err(error) => panic!("invalid erasure codec configuration: {error}"),
|
|
}
|
|
}
|
|
|
|
/// Tries to create a modern Erasure codec.
|
|
///
|
|
/// # Errors
|
|
/// Returns [`ErasureConstructionError`] when dimensions are zero,
|
|
/// overflow the shard count, exceed the modern codec's supported range, or
|
|
/// the underlying encoder rejects the configuration.
|
|
pub fn try_new(data_shards: usize, parity_shards: usize, block_size: usize) -> Result<Self, ErasureConstructionError> {
|
|
Self::try_new_with_options(data_shards, parity_shards, block_size, false)
|
|
}
|
|
|
|
/// Tries to create an Erasure codec using the selected format backend.
|
|
///
|
|
/// # Errors
|
|
/// Returns [`ErasureConstructionError`] when dimensions are zero,
|
|
/// overflow the shard count, are unsupported by the selected codec, or the
|
|
/// selected encoder fails to initialize.
|
|
pub fn try_new_with_options(
|
|
data_shards: usize,
|
|
parity_shards: usize,
|
|
block_size: usize,
|
|
uses_legacy: bool,
|
|
) -> Result<Self, ErasureConstructionError> {
|
|
if data_shards == 0 {
|
|
return Err(ErasureConstructionError::ZeroDataShards);
|
|
}
|
|
if block_size == 0 {
|
|
return Err(ErasureConstructionError::ZeroBlockSize);
|
|
}
|
|
|
|
let total_shards = data_shards
|
|
.checked_add(parity_shards)
|
|
.ok_or(ErasureConstructionError::ShardCountOverflow {
|
|
data_shards,
|
|
parity_shards,
|
|
})?;
|
|
|
|
if parity_shards > 0 {
|
|
if uses_legacy
|
|
&& (total_shards > reed_solomon_simd::engine::GF_ORDER
|
|
|| !reed_solomon_simd::ReedSolomonEncoder::supports(data_shards, parity_shards))
|
|
{
|
|
return Err(ErasureConstructionError::UnsupportedLegacyShardCount {
|
|
data_shards,
|
|
parity_shards,
|
|
});
|
|
}
|
|
if !uses_legacy && total_shards > MODERN_MAX_TOTAL_SHARDS {
|
|
return Err(ErasureConstructionError::UnsupportedModernShardCount {
|
|
data_shards,
|
|
parity_shards,
|
|
});
|
|
}
|
|
}
|
|
|
|
let encoder = if !uses_legacy && parity_shards > 0 {
|
|
Some(
|
|
ReedSolomonEncoder::try_new_typed(data_shards, parity_shards)
|
|
.map_err(|source| ErasureConstructionError::ModernEncoder { source })?,
|
|
)
|
|
} else {
|
|
None
|
|
};
|
|
|
|
let legacy_encoder = if uses_legacy && parity_shards > 0 {
|
|
Some(
|
|
cached_legacy_reed_solomon(data_shards, parity_shards)
|
|
.map_err(|source| ErasureConstructionError::LegacyEncoder { source })?,
|
|
)
|
|
} else {
|
|
None
|
|
};
|
|
|
|
Ok(Erasure {
|
|
data_shards,
|
|
parity_shards,
|
|
block_size,
|
|
encoder,
|
|
legacy_encoder,
|
|
uses_legacy,
|
|
_id: Uuid::nil(), // Unused in hot paths; avoid CSPRNG syscall
|
|
})
|
|
}
|
|
|
|
/// Encode data into data and parity shards.
|
|
///
|
|
/// # Arguments
|
|
/// * `data` - The input data to encode.
|
|
///
|
|
/// # Returns
|
|
/// A vector of encoded shards as `Bytes`.
|
|
#[tracing::instrument(level = "debug", skip_all, fields(data_len=data.len()))]
|
|
#[hotpath::measure(impl_type = "Erasure")]
|
|
pub fn encode_data(&self, data: &[u8]) -> io::Result<Vec<Bytes>> {
|
|
self.encode_data_block_inner(data)
|
|
.map(|block| block.into_shards(self.total_shard_count()))
|
|
}
|
|
|
|
#[tracing::instrument(level = "debug", skip_all, fields(data_len=data.len()))]
|
|
#[hotpath::measure(label = "Erasure::encode_data", impl_type = "Erasure")]
|
|
pub(crate) fn encode_data_block(&self, data: &[u8]) -> io::Result<EncodedBlock> {
|
|
self.encode_data_block_inner(data)
|
|
}
|
|
|
|
fn encode_data_block_inner(&self, data: &[u8]) -> io::Result<EncodedBlock> {
|
|
let mut data_buffer = BytesMut::with_capacity(self.encoded_capacity_for_data_len(data.len()));
|
|
data_buffer.extend_from_slice(data);
|
|
self.encode_buffer(data_buffer, data.len())
|
|
}
|
|
|
|
/// Encode owned data, avoiding a copy when the caller already has a heap buffer.
|
|
/// Falls back to copying into a new buffer if zero-copy conversion fails.
|
|
#[hotpath::measure(impl_type = "Erasure")]
|
|
pub fn encode_data_owned(&self, data: Vec<u8>) -> io::Result<Vec<Bytes>> {
|
|
self.encode_data_owned_block_inner(data)
|
|
.map(|block| block.into_shards(self.total_shard_count()))
|
|
}
|
|
|
|
#[hotpath::measure(label = "Erasure::encode_data_owned", impl_type = "Erasure")]
|
|
pub(crate) fn encode_data_owned_block(&self, data: Vec<u8>) -> io::Result<EncodedBlock> {
|
|
self.encode_data_owned_block_inner(data)
|
|
}
|
|
|
|
fn encode_data_owned_block_inner(&self, data: Vec<u8>) -> io::Result<EncodedBlock> {
|
|
let data_len = data.len();
|
|
// Try zero-copy: Vec<u8> -> Bytes -> BytesMut (succeeds when refcount == 1)
|
|
let data_buffer = match Bytes::from(data).try_into_mut() {
|
|
Ok(data_buffer) => data_buffer,
|
|
Err(b) => {
|
|
// Rare path: refcount != 1, fall back to copy
|
|
let mut data_buffer = BytesMut::with_capacity(self.encoded_capacity_for_data_len(data_len));
|
|
data_buffer.extend_from_slice(&b);
|
|
data_buffer
|
|
}
|
|
};
|
|
self.encode_buffer(data_buffer, data_len)
|
|
}
|
|
|
|
/// Encode data from an owned `BytesMut` buffer, avoiding the initial copy
|
|
/// from a borrowed slice into a fresh `BytesMut`.
|
|
///
|
|
/// Capacity contract: when the caller pre-reserves
|
|
/// `shard_size() * total_shard_count()` bytes (the EC-expanded size of a full
|
|
/// block), the `resize(need_total_size)` below stays within capacity for every
|
|
/// `data_len <= block_size` — both shard-size formulas are monotone in
|
|
/// `data_len` — so this function never reallocates the buffer.
|
|
#[hotpath::measure(impl_type = "Erasure")]
|
|
pub fn encode_data_bytes_mut(&self, data_buffer: BytesMut, data_len: usize) -> io::Result<Vec<Bytes>> {
|
|
self.encode_buffer(data_buffer, data_len)
|
|
.map(|block| block.into_shards(self.total_shard_count()))
|
|
}
|
|
|
|
#[hotpath::measure(label = "Erasure::encode_data_bytes_mut", impl_type = "Erasure")]
|
|
pub(crate) fn encode_data_bytes_mut_block(&self, data_buffer: BytesMut, data_len: usize) -> io::Result<EncodedBlock> {
|
|
self.encode_buffer(data_buffer, data_len)
|
|
}
|
|
|
|
fn encode_buffer(&self, mut data_buffer: BytesMut, data_len: usize) -> io::Result<EncodedBlock> {
|
|
let shard_size_fn = if self.uses_legacy {
|
|
calc_shard_size_legacy
|
|
} else {
|
|
calc_shard_size
|
|
};
|
|
let per_shard_size = shard_size_fn(data_len, self.data_shards);
|
|
if per_shard_size == 0 {
|
|
return Ok(EncodedBlock::empty());
|
|
}
|
|
let need_total_size = per_shard_size * self.total_shard_count();
|
|
|
|
if data_buffer.len() > data_len {
|
|
data_buffer.truncate(data_len);
|
|
}
|
|
data_buffer.resize(need_total_size, 0u8);
|
|
|
|
{
|
|
let data_slices: SmallVec<[&mut [u8]; 16]> = data_buffer.chunks_exact_mut(per_shard_size).collect();
|
|
|
|
if self.parity_shards > 0 {
|
|
if self.uses_legacy {
|
|
if let Some(encoder) = self.legacy_encoder.as_ref() {
|
|
encoder.encode(data_slices)?;
|
|
} else {
|
|
warn!("parity_shards > 0, uses_legacy but legacy_encoder is None");
|
|
}
|
|
} else if let Some(encoder) = self.encoder.as_ref() {
|
|
encoder.encode(data_slices)?;
|
|
} else {
|
|
warn!("parity_shards > 0, but encoder is None");
|
|
}
|
|
}
|
|
}
|
|
|
|
Ok(EncodedBlock {
|
|
data: data_buffer.freeze(),
|
|
shard_size: per_shard_size,
|
|
})
|
|
}
|
|
|
|
/// Decode and reconstruct missing data shards in-place.
|
|
///
|
|
/// # Arguments
|
|
/// * `shards` - Mutable slice of optional shard data. Missing shards should be `None`.
|
|
///
|
|
/// # Returns
|
|
/// Ok if reconstruction succeeds, error otherwise.
|
|
#[hotpath::measure(impl_type = "Erasure")]
|
|
pub fn decode_data(&self, shards: &mut [Option<Vec<u8>>]) -> io::Result<()> {
|
|
if self.parity_shards > 0 {
|
|
if self.uses_legacy {
|
|
if let Some(encoder) = self.legacy_encoder.as_ref() {
|
|
encoder.reconstruct_data(shards)?;
|
|
} else {
|
|
warn!("parity_shards > 0, uses_legacy but legacy_encoder is None");
|
|
}
|
|
} else if let Some(encoder) = self.encoder.as_ref() {
|
|
encoder.reconstruct_data(shards)?;
|
|
} else {
|
|
warn!("parity_shards > 0, but encoder is None");
|
|
}
|
|
}
|
|
|
|
Ok(())
|
|
}
|
|
|
|
/// Decode and reconstruct missing data shards, then regenerate parity shards.
|
|
#[hotpath::measure(impl_type = "Erasure")]
|
|
pub fn decode_data_and_parity(&self, shards: &mut [Option<Vec<u8>>]) -> io::Result<()> {
|
|
if self.parity_shards > 0 {
|
|
if self.uses_legacy {
|
|
if let Some(encoder) = self.legacy_encoder.as_ref() {
|
|
encoder.reconstruct(shards)?;
|
|
} else {
|
|
warn!("parity_shards > 0, uses_legacy but legacy_encoder is None");
|
|
}
|
|
} else if let Some(encoder) = self.encoder.as_ref() {
|
|
encoder.reconstruct(shards)?;
|
|
} else {
|
|
warn!("parity_shards > 0, but encoder is None");
|
|
}
|
|
}
|
|
|
|
Ok(())
|
|
}
|
|
|
|
pub(crate) fn decode_data_with_reconstruction_verification(&self, shards: &mut [Option<Vec<u8>>]) -> io::Result<()> {
|
|
let missing_data_source = shards.iter().take(self.data_shards).any(|shard| shard.is_none());
|
|
let available_shards = shards.iter().filter(|shard| shard.is_some()).count();
|
|
let source_parity = if missing_data_source && available_shards > self.data_shards {
|
|
shards
|
|
.iter()
|
|
.enumerate()
|
|
.skip(self.data_shards)
|
|
.filter_map(|(index, shard)| shard.as_ref().map(|shard| (index, shard.clone())))
|
|
.collect::<Vec<_>>()
|
|
} else {
|
|
Vec::new()
|
|
};
|
|
|
|
if source_parity.is_empty() {
|
|
return self.decode_data(shards);
|
|
}
|
|
|
|
self.decode_data_and_parity(shards)?;
|
|
for (index, source) in source_parity {
|
|
let Some(rebuilt) = shards[index].as_ref() else {
|
|
return Err(io::Error::new(
|
|
io::ErrorKind::InvalidData,
|
|
"missing rebuilt parity shard after read verification",
|
|
));
|
|
};
|
|
if rebuilt != &source {
|
|
warn!(
|
|
shard_index = index,
|
|
data_shards = self.data_shards,
|
|
parity_shards = self.parity_shards,
|
|
"erasure decode rejected inconsistent read source shards"
|
|
);
|
|
return Err(io::Error::new(io::ErrorKind::InvalidData, "inconsistent read source shards"));
|
|
}
|
|
}
|
|
|
|
Ok(())
|
|
}
|
|
|
|
pub(crate) fn verify_data_and_parity(&self, shards: &[Option<Vec<u8>>]) -> io::Result<bool> {
|
|
let expected_shards = self.total_shard_count();
|
|
if shards.len() != expected_shards {
|
|
return Err(io::Error::other(format!(
|
|
"invalid shard count: got {}, expected {}",
|
|
shards.len(),
|
|
expected_shards
|
|
)));
|
|
}
|
|
if self.parity_shards == 0 {
|
|
return Ok(true);
|
|
}
|
|
|
|
let mut shard_refs = Vec::with_capacity(expected_shards);
|
|
let mut shard_len = None;
|
|
for (index, shard) in shards.iter().enumerate() {
|
|
let shard = shard
|
|
.as_deref()
|
|
.ok_or_else(|| io::Error::other(format!("missing shard {index} for data/parity verification")))?;
|
|
if let Some(expected_len) = shard_len {
|
|
if shard.len() != expected_len {
|
|
return Err(io::Error::other(format!(
|
|
"inconsistent shard length at index {index}: got {}, expected {}",
|
|
shard.len(),
|
|
expected_len
|
|
)));
|
|
}
|
|
} else {
|
|
shard_len = Some(shard.len());
|
|
}
|
|
shard_refs.push(shard);
|
|
}
|
|
|
|
if self.uses_legacy {
|
|
if let Some(encoder) = self.legacy_encoder.as_ref() {
|
|
encoder.verify(&shard_refs)
|
|
} else {
|
|
Err(io::Error::other("parity_shards > 0, uses_legacy but legacy_encoder is None"))
|
|
}
|
|
} else if let Some(encoder) = self.encoder.as_ref() {
|
|
encoder.verify(&shard_refs)
|
|
} else {
|
|
Err(io::Error::other("parity_shards > 0, but encoder is None"))
|
|
}
|
|
}
|
|
|
|
/// Get the total number of shards (data + parity).
|
|
pub fn total_shard_count(&self) -> usize {
|
|
self.data_shards + self.parity_shards
|
|
}
|
|
|
|
pub(crate) fn encoded_capacity_for_data_len(&self, data_len: usize) -> usize {
|
|
let shard_size_fn = if self.uses_legacy {
|
|
calc_shard_size_legacy
|
|
} else {
|
|
calc_shard_size
|
|
};
|
|
shard_size_fn(data_len, self.data_shards).saturating_mul(self.total_shard_count())
|
|
}
|
|
|
|
/// Whether the erasure dimensions are safe for the shard/offset arithmetic.
|
|
///
|
|
/// `block_size` and `data_shards` come straight from on-disk metadata; a
|
|
/// corrupted or crafted `xl.meta` can carry zero values that would panic the
|
|
/// `block_size`/`data_shards` divisions in [`Self::shard_size`],
|
|
/// [`Self::shard_file_size`] and [`Self::shard_file_offset`]. Read paths must
|
|
/// reject such metadata before performing those divisions.
|
|
pub fn has_valid_dimensions(&self) -> bool {
|
|
self.block_size > 0 && self.data_shards > 0
|
|
}
|
|
// /// Calculate the shard size and total size for a given data size.
|
|
// // Returns (shard_size, total_size) for the given data size
|
|
// fn need_size(&self, data_size: usize) -> (usize, usize) {
|
|
// let shard_size = self.shard_size(data_size);
|
|
// (shard_size, shard_size * (self.total_shard_count()))
|
|
// }
|
|
|
|
/// Calculate the size of each shard.
|
|
pub fn shard_size(&self) -> usize {
|
|
if self.uses_legacy {
|
|
calc_shard_size_legacy(self.block_size, self.data_shards)
|
|
} else {
|
|
calc_shard_size(self.block_size, self.data_shards)
|
|
}
|
|
}
|
|
/// Calculate the total erasure file size for a given original size.
|
|
// Returns the final erasure size from the original size
|
|
pub fn shard_file_size(&self, total_length: i64) -> i64 {
|
|
if total_length == 0 {
|
|
return 0;
|
|
}
|
|
if total_length < 0 {
|
|
return total_length;
|
|
}
|
|
|
|
let total_length = total_length as usize;
|
|
let shard_size_fn = if self.uses_legacy {
|
|
calc_shard_size_legacy
|
|
} else {
|
|
calc_shard_size
|
|
};
|
|
|
|
let num_shards = total_length / self.block_size;
|
|
let last_block_size = total_length % self.block_size;
|
|
let last_shard_size = shard_size_fn(last_block_size, self.data_shards);
|
|
(num_shards * self.shard_size() + last_shard_size) as i64
|
|
}
|
|
|
|
/// Calculate the offset in the erasure file where reading begins.
|
|
// Returns the offset in the erasure file where reading begins
|
|
pub fn shard_file_offset(&self, start_offset: usize, length: usize, total_length: usize) -> usize {
|
|
let shard_size = self.shard_size();
|
|
let shard_file_size = self.shard_file_size(total_length as i64) as usize;
|
|
let end_shard = (start_offset + length) / self.block_size;
|
|
let mut till_offset = end_shard * shard_size + shard_size;
|
|
if till_offset > shard_file_size {
|
|
till_offset = shard_file_size;
|
|
}
|
|
|
|
till_offset
|
|
}
|
|
|
|
/// Encode all data from a reader in blocks, calling an async callback for each encoded block.
|
|
/// This method is async and returns the total bytes read after all blocks are processed.
|
|
///
|
|
/// # Arguments
|
|
/// * `reader` - An async reader implementing AsyncRead + Send + Sync + Unpin
|
|
/// * `mut on_block` - Async callback that receives encoded blocks and returns a Result
|
|
/// * `F` - Callback type: FnMut(Result<Vec<Bytes>, std::io::Error>) -> Future<Output=Result<(), E>> + Send
|
|
/// * `Fut` - Future type returned by the callback
|
|
/// * `E` - Error type returned by the callback
|
|
/// * `R` - Reader type implementing AsyncRead + Send + Sync + Unpin
|
|
///
|
|
/// # Returns
|
|
/// Result<usize, E> containing total bytes read, or error from callback
|
|
///
|
|
/// # Errors
|
|
/// Returns error if reading from reader fails or if callback returns error
|
|
#[allow(
|
|
dead_code,
|
|
reason = "callback encode path exercised only by this file's tests (backlog#1823)"
|
|
)]
|
|
pub(crate) async fn encode_stream_callback_async<F, Fut, E, R>(
|
|
self: std::sync::Arc<Self>,
|
|
reader: &mut R,
|
|
mut on_block: F,
|
|
) -> Result<usize, E>
|
|
where
|
|
R: AsyncRead + Send + Sync + Unpin,
|
|
F: FnMut(io::Result<Vec<Bytes>>) -> Fut + Send,
|
|
Fut: Future<Output = Result<(), E>> + Send,
|
|
{
|
|
if self.block_size == 0 {
|
|
on_block(Err(io::Error::new(io::ErrorKind::InvalidInput, "erasure block_size must be non-zero"))).await?;
|
|
return Ok(0);
|
|
}
|
|
|
|
let block_size = self.block_size;
|
|
let mut total = 0;
|
|
let mut buf = vec![0u8; block_size];
|
|
loop {
|
|
match rustfs_utils::read_full_or_eof(&mut *reader, &mut buf).await {
|
|
Ok(Some(n)) => {
|
|
debug_assert!(n > 0, "non-zero block_size prevents zero-length reads");
|
|
warn!("encode_stream_callback_async read n={}", n);
|
|
total += n;
|
|
let erasure = self.clone();
|
|
let encode_buf = std::mem::take(&mut buf);
|
|
let (res, returned_buf) = match tokio::task::spawn_blocking(move || {
|
|
let res = erasure.encode_data(&encode_buf[..n]);
|
|
(res, encode_buf)
|
|
})
|
|
.await
|
|
{
|
|
Ok(result) => result,
|
|
Err(err) => {
|
|
on_block(Err(io::Error::other(format!("EC encode task failed: {err}")))).await?;
|
|
break;
|
|
}
|
|
};
|
|
buf = returned_buf;
|
|
on_block(res).await?
|
|
}
|
|
Ok(None) => {
|
|
warn!("encode_stream_callback_async read unexpected ok");
|
|
break;
|
|
}
|
|
Err(e) if e.kind() == io::ErrorKind::UnexpectedEof => {
|
|
warn!("encode_stream_callback_async read unexpected eof");
|
|
break;
|
|
}
|
|
Err(e) => {
|
|
warn!("encode_stream_callback_async read error={:?}", e);
|
|
on_block(Err(e)).await?;
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
Ok(total)
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
use proptest::collection::{btree_set, vec};
|
|
use proptest::prelude::*;
|
|
use std::pin::Pin;
|
|
use std::task::{Context, Poll};
|
|
use tokio::io::ReadBuf;
|
|
|
|
fn erasure_with_invalid_dimensions(data_shards: usize, parity_shards: usize, block_size: usize) -> Erasure {
|
|
Erasure {
|
|
data_shards,
|
|
parity_shards,
|
|
block_size,
|
|
..Default::default()
|
|
}
|
|
}
|
|
|
|
fn optional_shards(shards: &[Bytes]) -> Vec<Option<Vec<u8>>> {
|
|
shards.iter().map(|shard| Some(shard.to_vec())).collect()
|
|
}
|
|
|
|
fn recover_data(shards: &[Option<Vec<u8>>], data_shards: usize, original_len: usize) -> Vec<u8> {
|
|
let mut recovered = Vec::new();
|
|
for shard in shards.iter().take(data_shards) {
|
|
recovered.extend_from_slice(
|
|
shard
|
|
.as_ref()
|
|
.expect("reconstructed data shard should be present after decode_data_and_parity"),
|
|
);
|
|
}
|
|
recovered.truncate(original_len);
|
|
recovered
|
|
}
|
|
|
|
fn erasure_recoverability_case_strategy()
|
|
-> impl Strategy<Value = (usize, usize, usize, bool, Vec<u8>, std::collections::BTreeSet<usize>)> {
|
|
(2usize..=8, 1usize..=4, prop::sample::select(vec![64usize, 256, 1024]), any::<bool>()).prop_flat_map(
|
|
|(data_shards, parity_shards, block_size, uses_legacy)| {
|
|
let total_shards = data_shards + parity_shards;
|
|
(
|
|
Just(data_shards),
|
|
Just(parity_shards),
|
|
Just(block_size),
|
|
Just(uses_legacy),
|
|
vec(any::<u8>(), 0..=4096),
|
|
btree_set(0usize..total_shards, 0..=parity_shards),
|
|
)
|
|
},
|
|
)
|
|
}
|
|
|
|
fn assert_owned_encode_matches_borrowed(erasure: &Erasure, data: Vec<u8>) {
|
|
let borrowed = erasure.encode_data(&data).expect("borrowed encode should succeed");
|
|
let owned = erasure.encode_data_owned(data).expect("owned encode should succeed");
|
|
|
|
assert_eq!(owned, borrowed);
|
|
}
|
|
|
|
struct ErrorAfterPartialReader {
|
|
emitted: bool,
|
|
}
|
|
|
|
impl AsyncRead for ErrorAfterPartialReader {
|
|
fn poll_read(mut self: Pin<&mut Self>, _cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<io::Result<()>> {
|
|
if !self.emitted {
|
|
self.emitted = true;
|
|
buf.put_slice(&[1]);
|
|
return Poll::Ready(Ok(()));
|
|
}
|
|
Poll::Ready(Err(io::Error::new(io::ErrorKind::BrokenPipe, "partial read failure")))
|
|
}
|
|
}
|
|
|
|
struct ImmediateErrorReader;
|
|
|
|
impl AsyncRead for ImmediateErrorReader {
|
|
fn poll_read(self: Pin<&mut Self>, _cx: &mut Context<'_>, _buf: &mut ReadBuf<'_>) -> Poll<io::Result<()>> {
|
|
Poll::Ready(Err(io::Error::other("immediate read failure")))
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn try_new_rejects_zero_and_overflowing_dimensions_with_typed_errors() {
|
|
let Err(zero_data) = Erasure::try_new(0, 2, 64) else {
|
|
panic!("zero data shards must be rejected");
|
|
};
|
|
assert!(matches!(zero_data, ErasureConstructionError::ZeroDataShards));
|
|
|
|
let Err(zero_block) = Erasure::try_new(2, 2, 0) else {
|
|
panic!("zero block size must be rejected");
|
|
};
|
|
assert!(matches!(zero_block, ErasureConstructionError::ZeroBlockSize));
|
|
|
|
let Err(overflow) = Erasure::try_new(usize::MAX, 1, 64) else {
|
|
panic!("overflowing shard counts must be rejected");
|
|
};
|
|
assert!(matches!(
|
|
overflow,
|
|
ErasureConstructionError::ShardCountOverflow {
|
|
data_shards: usize::MAX,
|
|
parity_shards: 1,
|
|
}
|
|
));
|
|
}
|
|
|
|
#[test]
|
|
fn try_new_rejects_backend_specific_unsupported_dimensions() {
|
|
let Err(modern) = Erasure::try_new(256, 1, 64) else {
|
|
panic!("modern dimensions beyond GF(2^8) must be rejected");
|
|
};
|
|
assert!(matches!(
|
|
modern,
|
|
ErasureConstructionError::UnsupportedModernShardCount {
|
|
data_shards: 256,
|
|
parity_shards: 1,
|
|
}
|
|
));
|
|
|
|
let Err(legacy) = Erasure::try_new_with_options(60_000, 5_000, 64, true) else {
|
|
panic!("unsupported legacy SIMD dimensions must be rejected");
|
|
};
|
|
assert!(matches!(
|
|
legacy,
|
|
ErasureConstructionError::UnsupportedLegacyShardCount {
|
|
data_shards: 60_000,
|
|
parity_shards: 5_000,
|
|
}
|
|
));
|
|
}
|
|
|
|
#[test]
|
|
fn try_new_accepts_exact_backend_shard_limits() {
|
|
let modern_data_shards = MODERN_MAX_TOTAL_SHARDS - 1;
|
|
let modern =
|
|
Erasure::try_new(modern_data_shards, 1, 64).expect("the modern codec's exact field-size boundary must remain valid");
|
|
assert_eq!(modern.total_shard_count(), MODERN_MAX_TOTAL_SHARDS);
|
|
assert!(modern.encoder.is_some());
|
|
assert!(modern.legacy_encoder.is_none());
|
|
|
|
let legacy_data_shards = reed_solomon_simd::engine::GF_ORDER / 2;
|
|
let legacy_parity_shards = reed_solomon_simd::engine::GF_ORDER - legacy_data_shards;
|
|
let legacy = Erasure::try_new_with_options(legacy_data_shards, legacy_parity_shards, 64, true)
|
|
.expect("the legacy codec's exact field-size boundary must remain valid");
|
|
assert_eq!(legacy.total_shard_count(), reed_solomon_simd::engine::GF_ORDER);
|
|
assert!(legacy.encoder.is_none());
|
|
assert!(legacy.legacy_encoder.is_some());
|
|
}
|
|
|
|
#[test]
|
|
fn try_new_accepts_zero_parity_without_backend_limits() {
|
|
let no_parity = Erasure::try_new(2, 0, 64).expect("zero parity is a valid generic codec configuration");
|
|
assert_eq!(no_parity.total_shard_count(), 2);
|
|
assert!(no_parity.encoder.is_none());
|
|
assert!(no_parity.legacy_encoder.is_none());
|
|
|
|
let large_modern = Erasure::try_new(257, 0, 64).expect("zero parity must not inherit the modern backend shard limit");
|
|
assert_eq!(large_modern.total_shard_count(), 257);
|
|
assert!(large_modern.encoder.is_none());
|
|
assert!(large_modern.legacy_encoder.is_none());
|
|
|
|
let legacy_data_shards = reed_solomon_simd::engine::GF_ORDER + 1;
|
|
let large_legacy = Erasure::try_new_with_options(legacy_data_shards, 0, 64, true)
|
|
.expect("zero parity must not inherit the legacy backend shard limit");
|
|
assert_eq!(large_legacy.total_shard_count(), legacy_data_shards);
|
|
assert!(large_legacy.encoder.is_none());
|
|
assert!(large_legacy.legacy_encoder.is_none());
|
|
}
|
|
|
|
#[test]
|
|
fn try_new_accepts_generic_layouts_above_storage_drive_limits() {
|
|
let erasure = Erasure::try_new(2, 3, 64).expect("generic 2+3 erasure coding must remain supported");
|
|
assert_eq!(erasure.total_shard_count(), 5);
|
|
assert!(erasure.encoder.is_some());
|
|
assert!(erasure.legacy_encoder.is_none());
|
|
|
|
let modern = Erasure::try_new(9, 8, 64).expect("the codec must not impose the storage layer's drive limit");
|
|
assert_eq!(modern.total_shard_count(), 17);
|
|
assert!(modern.encoder.is_some());
|
|
assert!(modern.legacy_encoder.is_none());
|
|
|
|
let legacy = Erasure::try_new_with_options(9, 8, 64, true)
|
|
.expect("the legacy codec must not impose the storage layer's drive limit");
|
|
assert_eq!(legacy.total_shard_count(), 17);
|
|
assert!(legacy.encoder.is_none());
|
|
assert!(legacy.legacy_encoder.is_some());
|
|
}
|
|
|
|
#[test]
|
|
fn try_new_with_options_initializes_only_the_selected_backend() {
|
|
let modern = Erasure::try_new_with_options(4, 2, 64, false).expect("modern codec should construct");
|
|
assert!(modern.encoder.is_some());
|
|
assert!(modern.legacy_encoder.is_none());
|
|
|
|
let legacy = Erasure::try_new_with_options(4, 2, 64, true).expect("legacy codec should construct");
|
|
assert!(legacy.encoder.is_none());
|
|
assert!(legacy.legacy_encoder.is_some());
|
|
}
|
|
|
|
#[test]
|
|
fn modern_encoder_construction_reuses_cached_codec() {
|
|
let first = ReedSolomonEncoder::try_new_typed(31, 7).expect("modern codec should construct");
|
|
let second = ReedSolomonEncoder::try_new_typed(31, 7).expect("modern codec should construct");
|
|
|
|
let first = first.encoder.as_ref().expect("modern codec should initialize an encoder");
|
|
let second = second.encoder.as_ref().expect("modern codec should initialize an encoder");
|
|
assert!(Arc::ptr_eq(first, second));
|
|
}
|
|
|
|
#[test]
|
|
fn construction_errors_preserve_encoder_sources() {
|
|
let modern = ErasureConstructionError::ModernEncoder {
|
|
source: reed_solomon_erasure::Error::TooManyShards,
|
|
};
|
|
assert!(
|
|
std::error::Error::source(&modern)
|
|
.and_then(|source| source.downcast_ref::<reed_solomon_erasure::Error>())
|
|
.is_some()
|
|
);
|
|
|
|
let legacy = ErasureConstructionError::LegacyEncoder {
|
|
source: io::Error::other("legacy encoder failure"),
|
|
};
|
|
assert!(
|
|
std::error::Error::source(&legacy)
|
|
.and_then(|source| source.downcast_ref::<io::Error>())
|
|
.is_some()
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
#[should_panic(expected = "invalid erasure codec configuration")]
|
|
fn new_panics_on_invalid_dimensions() {
|
|
let _ = Erasure::new(0, 1, 64);
|
|
}
|
|
|
|
#[test]
|
|
#[should_panic(expected = "invalid erasure codec configuration")]
|
|
fn new_with_options_panics_on_invalid_dimensions() {
|
|
let _ = Erasure::new_with_options(1, 1, 0, true);
|
|
}
|
|
|
|
#[test]
|
|
fn has_valid_dimensions_rejects_zero_block_size_or_data_shards() {
|
|
// Well-formed erasure metadata is accepted.
|
|
assert!(Erasure::new(4, 2, 64).has_valid_dimensions());
|
|
assert!(Erasure::new_with_options(6, 4, 1, true).has_valid_dimensions());
|
|
|
|
// Corrupted on-disk metadata with a zero block_size or zero data_shards
|
|
// must be rejected before it reaches the shard/offset divisions.
|
|
assert!(!erasure_with_invalid_dimensions(4, 2, 0).has_valid_dimensions());
|
|
assert!(!erasure_with_invalid_dimensions(0, 0, 64).has_valid_dimensions());
|
|
assert!(!erasure_with_invalid_dimensions(0, 0, 0).has_valid_dimensions());
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn encode_stream_callback_async_stops_on_reader_errors() {
|
|
let erasure = std::sync::Arc::new(Erasure::new(2, 1, 4));
|
|
let mut partial_error = ErrorAfterPartialReader { emitted: false };
|
|
let mut partial_callbacks = Vec::new();
|
|
let total = erasure
|
|
.clone()
|
|
.encode_stream_callback_async(&mut partial_error, |result| {
|
|
partial_callbacks.push(result.map(|blocks| blocks.len()).map_err(|err| err.kind()));
|
|
async { Ok::<(), io::Error>(()) }
|
|
})
|
|
.await
|
|
.expect("partial read error should not make callback fail");
|
|
assert_eq!(total, 0);
|
|
assert!(
|
|
partial_callbacks.is_empty(),
|
|
"unexpected EOF after a partial read should stop without emitting a block"
|
|
);
|
|
|
|
let mut immediate_error = ImmediateErrorReader;
|
|
let mut immediate_callbacks = Vec::new();
|
|
let total = erasure
|
|
.encode_stream_callback_async(&mut immediate_error, |result| {
|
|
immediate_callbacks.push(result.map(|blocks| blocks.len()).map_err(|err| err.kind()));
|
|
async { Ok::<(), io::Error>(()) }
|
|
})
|
|
.await
|
|
.expect("immediate read error should be delivered to callback");
|
|
assert_eq!(total, 0);
|
|
assert_eq!(immediate_callbacks, vec![Err(io::ErrorKind::Other)]);
|
|
}
|
|
|
|
#[test]
|
|
fn default_and_legacy_clone_preserve_safe_zero_state_and_restore_data() {
|
|
let default = Erasure::default();
|
|
assert_eq!(default.total_shard_count(), 0);
|
|
assert!(!default.has_valid_dimensions());
|
|
assert_eq!(default.shard_file_size(0), 0);
|
|
assert_eq!(default.shard_file_size(-7), -7);
|
|
|
|
let legacy = Erasure::new_with_options(2, 2, 64, true);
|
|
let cloned = legacy.clone();
|
|
assert_eq!(cloned.data_shards, legacy.data_shards);
|
|
assert_eq!(cloned.parity_shards, legacy.parity_shards);
|
|
assert_eq!(cloned.block_size, legacy.block_size);
|
|
assert!(cloned.uses_legacy);
|
|
|
|
let data = b"legacy clone should preserve SIMD codec behavior";
|
|
let encoded = cloned.encode_data(data).expect("legacy clone should encode");
|
|
let mut shards = optional_shards(&encoded);
|
|
shards[0] = None;
|
|
cloned.decode_data(&mut shards).expect("legacy clone should decode");
|
|
assert_eq!(recover_data(&shards, cloned.data_shards, data.len()), data);
|
|
}
|
|
|
|
#[test]
|
|
fn legacy_codecs_share_process_cache_across_erasure_instances() {
|
|
let first = Erasure::new_with_options(6, 3, 64, true)
|
|
.legacy_encoder
|
|
.expect("legacy codec should be initialized");
|
|
let second = Erasure::new_with_options(6, 3, 128, true)
|
|
.legacy_encoder
|
|
.expect("same legacy shard layout should be initialized");
|
|
|
|
assert!(Arc::ptr_eq(&first, &second));
|
|
}
|
|
|
|
#[test]
|
|
fn legacy_workspace_cache_rejects_oversize_buffers_and_isolates_layouts() {
|
|
let four_plus_two = Erasure::new_with_options(4, 2, 64, true)
|
|
.legacy_encoder
|
|
.expect("legacy codec should be initialized");
|
|
let four_plus_one = Erasure::new_with_options(4, 1, 64, true)
|
|
.legacy_encoder
|
|
.expect("distinct parity layout should be initialized");
|
|
let three_plus_two = Erasure::new_with_options(3, 2, 64, true)
|
|
.legacy_encoder
|
|
.expect("distinct data layout should be initialized");
|
|
|
|
assert!(!Arc::ptr_eq(&four_plus_two, &four_plus_one));
|
|
assert!(!Arc::ptr_eq(&four_plus_two, &three_plus_two));
|
|
assert_eq!(four_plus_two.logical_shard_bytes_upper_bound(64 * 1024), Some(512 * 1024));
|
|
assert!(four_plus_two.should_cache_workspace(64 * 1024));
|
|
assert!(!four_plus_two.should_cache_workspace(64 * 1024 + 1));
|
|
|
|
let nine_plus_seven =
|
|
LegacyReedSolomonEncoder::with_workspace_cache(9, 7, true).expect("9+7 legacy codec should construct");
|
|
assert_eq!(nine_plus_seven.logical_shard_bytes_upper_bound(16 * 1024), Some(512 * 1024));
|
|
assert!(nine_plus_seven.should_cache_workspace(16 * 1024));
|
|
assert!(!nine_plus_seven.should_cache_workspace(16 * 1024 + 1));
|
|
|
|
let uncached = LegacyReedSolomonEncoder::new(4, 2).expect("uncached legacy codec should construct");
|
|
assert!(!uncached.should_cache_workspace(64));
|
|
}
|
|
|
|
#[test]
|
|
fn saturated_legacy_codec_cache_does_not_retain_more_workspaces() {
|
|
let cache = RwLock::new(HashMap::new());
|
|
for parity_shards in 1..=LEGACY_REED_SOLOMON_CACHE_MAX_ENTRIES {
|
|
let cached =
|
|
cached_legacy_reed_solomon_in(&cache, 32, parity_shards).expect("cacheable legacy codec should construct");
|
|
assert!(cached.cache_workspaces);
|
|
}
|
|
|
|
let uncached =
|
|
cached_legacy_reed_solomon_in(&cache, 31, 1).expect("uncached legacy codec should construct after saturation");
|
|
assert!(!uncached.cache_workspaces);
|
|
assert_eq!(
|
|
cache.read().expect("cache lock should remain healthy").len(),
|
|
LEGACY_REED_SOLOMON_CACHE_MAX_ENTRIES
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn concurrent_legacy_codecs_preserve_byte_exact_results() {
|
|
let barrier = Arc::new(std::sync::Barrier::new(2));
|
|
let payloads = [vec![0x35; 257], vec![0xca; 1025]];
|
|
|
|
std::thread::scope(|scope| {
|
|
let handles = payloads.each_ref().map(|payload| {
|
|
let barrier = Arc::clone(&barrier);
|
|
scope.spawn(move || {
|
|
let erasure = Erasure::new_with_options(6, 3, 2048, true);
|
|
barrier.wait();
|
|
let encoded = erasure.encode_data(payload).expect("concurrent legacy encode should succeed");
|
|
barrier.wait();
|
|
|
|
let mut shards = optional_shards(&encoded);
|
|
shards[0] = None;
|
|
erasure
|
|
.decode_data(&mut shards)
|
|
.expect("concurrent legacy decode should reconstruct the missing shard");
|
|
recover_data(&shards, erasure.data_shards, payload.len())
|
|
})
|
|
});
|
|
|
|
for (handle, payload) in handles.into_iter().zip(payloads.iter()) {
|
|
assert_eq!(handle.join().expect("concurrent legacy codec worker should not panic"), *payload);
|
|
}
|
|
});
|
|
}
|
|
|
|
#[test]
|
|
fn legacy_verify_reports_invalid_empty_valid_and_corrupt_parity_sets() {
|
|
let legacy = LegacyReedSolomonEncoder::new(2, 2).expect("legacy encoder should construct");
|
|
let wrong_count: [&[u8]; 1] = [&[]];
|
|
let err = legacy.verify(&wrong_count).expect_err("wrong shard count must be rejected");
|
|
assert!(err.to_string().contains("invalid shard count"));
|
|
|
|
let empty: [&[u8]; 4] = [&[], &[], &[], &[]];
|
|
assert!(
|
|
legacy
|
|
.verify(&empty)
|
|
.expect("all-empty legacy shards are internally consistent")
|
|
);
|
|
|
|
let erasure = Erasure::new_with_options(2, 2, 64, true);
|
|
let encoded = erasure
|
|
.encode_data(b"legacy verify should compare regenerated parity")
|
|
.expect("legacy encode should succeed");
|
|
let refs = encoded.iter().map(Bytes::as_ref).collect::<Vec<_>>();
|
|
assert!(legacy.verify(&refs).expect("valid legacy shards should verify"));
|
|
|
|
let mut corrupt = encoded.iter().map(|shard| shard.to_vec()).collect::<Vec<_>>();
|
|
corrupt[erasure.data_shards][0] ^= 0x80;
|
|
let corrupt_refs = corrupt.iter().map(Vec::as_slice).collect::<Vec<_>>();
|
|
assert!(!legacy.verify(&corrupt_refs).expect("corrupt parity should be detected"));
|
|
}
|
|
|
|
#[test]
|
|
fn parity_helper_and_empty_payload_recovery_fail_closed_on_malformed_shards() {
|
|
let mut wrong_count = vec![Some(vec![1])];
|
|
let err = encode_parity_shards(&mut wrong_count, 2, 1, |_| panic!("wrong count must fail before encode"))
|
|
.expect_err("wrong shard count must be rejected");
|
|
assert!(err.to_string().contains("invalid shard count"));
|
|
|
|
let mut inconsistent_empty = vec![Some(Vec::new()), Some(vec![1])];
|
|
let err = encode_parity_shards(&mut inconsistent_empty, 1, 1, |_| panic!("zero-length mismatch must fail before encode"))
|
|
.expect_err("mixed empty and non-empty shards must be rejected");
|
|
assert!(err.to_string().contains("inconsistent shard length"));
|
|
|
|
let mut inconsistent_non_empty = vec![Some(vec![1]), Some(vec![2, 3])];
|
|
let err = encode_parity_shards(&mut inconsistent_non_empty, 1, 1, |_| {
|
|
panic!("non-empty length mismatch must fail before encode")
|
|
})
|
|
.expect_err("mismatched non-empty shards must be rejected");
|
|
assert!(err.to_string().contains("inconsistent shard length"));
|
|
|
|
let mut no_present_empty_payload = vec![None, None, None];
|
|
assert!(
|
|
!recover_empty_payload_data_shards(&mut no_present_empty_payload, 2, 1)
|
|
.expect("all-missing empty payload marker should not be synthesized")
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn verify_data_and_parity_handles_zero_parity_and_rejects_wrong_count() {
|
|
let no_parity = Erasure::new(2, 0, 64);
|
|
assert!(
|
|
no_parity
|
|
.verify_data_and_parity(&[Some(vec![1]), Some(vec![2, 3])])
|
|
.expect("zero parity requires no parity verification")
|
|
);
|
|
|
|
let erasure = Erasure::new(2, 2, 64);
|
|
let err = erasure
|
|
.verify_data_and_parity(&[Some(Vec::new())])
|
|
.expect_err("wrong shard count must fail verification");
|
|
assert!(err.to_string().contains("invalid shard count"));
|
|
}
|
|
|
|
#[test]
|
|
fn encode_data_owned_matches_borrowed_path() {
|
|
for uses_legacy in [false, true] {
|
|
let erasure = Erasure::new_with_options(4, 2, 64, uses_legacy);
|
|
for data in [
|
|
Vec::new(),
|
|
vec![0xA5; 1],
|
|
b"small payload".to_vec(),
|
|
(0_u8..37).collect(),
|
|
vec![0xA5; erasure.block_size - 1],
|
|
vec![0x5A; erasure.block_size],
|
|
] {
|
|
assert_owned_encode_matches_borrowed(&erasure, data);
|
|
}
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn encode_data_bytes_mut_matches_borrowed_path() {
|
|
for uses_legacy in [false, true] {
|
|
let erasure = Erasure::new_with_options(4, 2, 64, uses_legacy);
|
|
let block_size = erasure.block_size;
|
|
let expanded_block_bytes = erasure.shard_size() * erasure.total_shard_count();
|
|
let cases: Vec<Vec<u8>> = vec![
|
|
Vec::new(),
|
|
b"small payload".to_vec(),
|
|
(0_u8..37).collect(),
|
|
vec![0xA5; block_size - 1], // last block one byte short of full
|
|
vec![0x5A; block_size], // exactly one full block
|
|
];
|
|
for data in cases {
|
|
let borrowed = erasure.encode_data(&data).expect("borrowed encode should succeed");
|
|
|
|
let owned = erasure
|
|
.encode_data_bytes_mut(BytesMut::from(&data[..]), data.len())
|
|
.expect("bytesmut encode should succeed");
|
|
assert_eq!(owned, borrowed);
|
|
|
|
// Ingest-shaped buffer: spare capacity pre-reserved for the EC-expanded
|
|
// block, exactly what the HP-10 streaming ingest path hands over.
|
|
let mut ingest = BytesMut::with_capacity(expanded_block_bytes.max(block_size));
|
|
ingest.extend_from_slice(&data);
|
|
let preallocated = erasure
|
|
.encode_data_bytes_mut(ingest, data.len())
|
|
.expect("preallocated bytesmut encode should succeed");
|
|
assert_eq!(preallocated, borrowed);
|
|
|
|
// Buffer longer than data_len (stale bytes past the logical block)
|
|
// must be truncated before padding.
|
|
let mut oversized = BytesMut::from(&data[..]);
|
|
oversized.extend_from_slice(&[0xFF; 8]);
|
|
let truncated = erasure
|
|
.encode_data_bytes_mut(oversized, data.len())
|
|
.expect("oversized bytesmut encode should succeed");
|
|
assert_eq!(truncated, borrowed);
|
|
}
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn streaming_encoded_block_uses_one_contiguous_backing_buffer() {
|
|
for uses_legacy in [false, true] {
|
|
let erasure = Erasure::new_with_options(8, 8, 64, uses_legacy);
|
|
|
|
for data_len in [0, 1, 63, 64] {
|
|
let data = (0..data_len).map(|i| i as u8).collect::<Vec<_>>();
|
|
let expected = erasure.encode_data(&data).expect("public encode should succeed");
|
|
let borrowed = erasure
|
|
.encode_data_block(&data)
|
|
.expect("borrowed streaming encode should succeed");
|
|
let owned = erasure
|
|
.encode_data_owned_block(data.clone())
|
|
.expect("owned streaming encode should succeed");
|
|
let bytes_mut = erasure
|
|
.encode_data_bytes_mut_block(BytesMut::from(&data[..]), data.len())
|
|
.expect("BytesMut streaming encode should succeed");
|
|
|
|
assert_eq!(borrowed.queued_bytes(), owned.queued_bytes());
|
|
assert_eq!(borrowed.queued_bytes(), bytes_mut.queued_bytes());
|
|
|
|
if data_len == 0 {
|
|
assert!(expected.iter().all(Bytes::is_empty));
|
|
assert!(borrowed.is_empty());
|
|
assert!(owned.is_empty());
|
|
assert!(bytes_mut.is_empty());
|
|
continue;
|
|
}
|
|
|
|
assert!(borrowed.shards().eq(expected.iter().map(Bytes::as_ref)));
|
|
assert!(owned.shards().eq(expected.iter().map(Bytes::as_ref)));
|
|
assert!(bytes_mut.shards().eq(expected.iter().map(Bytes::as_ref)));
|
|
assert_eq!(borrowed.shards().len(), 16);
|
|
let first = borrowed.shards().next().expect("encoded block should have shards").as_ptr();
|
|
for (index, shard) in borrowed.shards().enumerate() {
|
|
assert_eq!(shard.as_ptr(), first.wrapping_add(index * shard.len()));
|
|
}
|
|
}
|
|
}
|
|
assert_eq!(
|
|
std::mem::size_of::<EncodedBlock>(),
|
|
std::mem::size_of::<Bytes>() + std::mem::size_of::<usize>(),
|
|
"queue entries must contain one backing buffer handle, not per-shard handles"
|
|
);
|
|
}
|
|
|
|
/// HP-10 capacity invariant: both shard-size formulas are monotone in `data_len`,
|
|
/// so pre-reserving `shard_size(block_size) * total_shard_count` covers the
|
|
/// `need_total_size` of every block-or-smaller payload and the ingest buffer
|
|
/// never reallocates inside `encode_data_bytes_mut`.
|
|
#[test]
|
|
fn shard_size_monotonicity_bounds_expanded_block_capacity() {
|
|
for shard_fn in [calc_shard_size, calc_shard_size_legacy] {
|
|
for data_shards in 1usize..=16 {
|
|
for block_size in [1usize, 2, 63, 64, 65, 1024, 4096] {
|
|
let full = shard_fn(block_size, data_shards);
|
|
let mut prev = shard_fn(0, data_shards);
|
|
for data_len in 1..=block_size {
|
|
let cur = shard_fn(data_len, data_shards);
|
|
assert!(cur >= prev, "shard size must be monotone (len {data_len}, shards {data_shards})");
|
|
assert!(cur <= full, "per-block shard size must not exceed the full-block bound");
|
|
prev = cur;
|
|
}
|
|
}
|
|
// Spot-check the production-scale block size at its boundaries.
|
|
let block_size = 1usize << 20;
|
|
let full = shard_fn(block_size, data_shards);
|
|
for data_len in [0usize, 1, block_size / 2, block_size - 1, block_size] {
|
|
assert!(shard_fn(data_len, data_shards) <= full);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn decode_data_keeps_missing_parity_shard_unreconstructed() {
|
|
let erasure = Erasure::new(2, 2, 64);
|
|
let data = b"read decode should not rebuild parity";
|
|
let encoded = erasure.encode_data(data).expect("encode should succeed");
|
|
let mut shards = optional_shards(&encoded);
|
|
let missing_parity = erasure.data_shards;
|
|
shards[missing_parity] = None;
|
|
|
|
erasure.decode_data(&mut shards).expect("decode should succeed");
|
|
|
|
assert!(shards[missing_parity].is_none(), "read decode should leave parity missing");
|
|
for index in 0..erasure.data_shards {
|
|
assert_eq!(
|
|
shards[index].as_deref(),
|
|
Some(encoded[index].as_ref()),
|
|
"data shard {index} should remain unchanged"
|
|
);
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn legacy_decode_data_keeps_missing_parity_shard_unreconstructed() {
|
|
let erasure = Erasure::new_with_options(2, 2, 64, true);
|
|
let data = b"legacy read decode should not rebuild parity";
|
|
let encoded = erasure.encode_data(data).expect("encode should succeed");
|
|
let mut shards = optional_shards(&encoded);
|
|
let missing_parity = erasure.data_shards + 1;
|
|
shards[missing_parity] = None;
|
|
|
|
erasure.decode_data(&mut shards).expect("decode should succeed");
|
|
|
|
assert!(shards[missing_parity].is_none(), "legacy read decode should leave parity missing");
|
|
for index in 0..erasure.data_shards {
|
|
assert_eq!(
|
|
shards[index].as_deref(),
|
|
Some(encoded[index].as_ref()),
|
|
"legacy data shard {index} should remain unchanged"
|
|
);
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn decode_data_rejects_missing_data_above_parity_budget() {
|
|
for uses_legacy in [false, true] {
|
|
let erasure = Erasure::new_with_options(3, 2, 64, uses_legacy);
|
|
let data = b"read restore must fail when too many data shards are gone";
|
|
let encoded = erasure.encode_data(data).expect("encode should succeed");
|
|
let mut shards = optional_shards(&encoded);
|
|
shards[0] = None;
|
|
shards[1] = None;
|
|
shards[2] = None;
|
|
|
|
let err = erasure
|
|
.decode_data(&mut shards)
|
|
.expect_err("decode_data must fail when missing data shards exceed parity");
|
|
|
|
assert!(!err.to_string().is_empty());
|
|
assert!(shards.iter().take(erasure.data_shards).any(Option::is_none));
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn decode_data_and_parity_rejects_wrong_shard_count() {
|
|
for uses_legacy in [false, true] {
|
|
let erasure = Erasure::new_with_options(4, 2, 128, uses_legacy);
|
|
let data = b"wrong shard count should not be accepted as a restored object";
|
|
let encoded = erasure.encode_data(data).expect("encode should succeed");
|
|
let mut shards = optional_shards(&encoded);
|
|
let _ = shards.pop();
|
|
|
|
let err = erasure
|
|
.decode_data_and_parity(&mut shards)
|
|
.expect_err("decode_data_and_parity must reject wrong shard count");
|
|
|
|
assert!(!err.to_string().is_empty());
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn decode_data_with_verification_rejects_corrupt_surplus_parity() {
|
|
let erasure = Erasure::new(3, 2, 128);
|
|
let data = b"verified reads must fail closed on stale surplus parity";
|
|
let encoded = erasure.encode_data(data).expect("encode should succeed");
|
|
let mut shards = optional_shards(&encoded);
|
|
shards[1] = None;
|
|
shards[erasure.data_shards].as_mut().expect("parity shard should be present")[0] ^= 0x7d;
|
|
|
|
let err = erasure
|
|
.decode_data_with_reconstruction_verification(&mut shards)
|
|
.expect_err("verified decode must reject inconsistent parity");
|
|
|
|
assert_eq!(err.kind(), io::ErrorKind::InvalidData);
|
|
}
|
|
|
|
#[test]
|
|
fn verify_data_and_parity_rejects_missing_and_mismatched_shards() {
|
|
let erasure = Erasure::new(4, 2, 128);
|
|
let data = b"verification must reject incomplete or malformed shard sets";
|
|
let encoded = erasure.encode_data(data).expect("encode should succeed");
|
|
let mut shards = optional_shards(&encoded);
|
|
|
|
shards[0] = None;
|
|
let missing = erasure
|
|
.verify_data_and_parity(&shards)
|
|
.expect_err("verification must reject missing shards");
|
|
assert!(missing.to_string().contains("missing shard"));
|
|
|
|
let mut mismatched = optional_shards(&encoded);
|
|
let _ = mismatched[0].as_mut().expect("data shard should be present").pop();
|
|
let mismatch = erasure
|
|
.verify_data_and_parity(&mismatched)
|
|
.expect_err("verification must reject mismatched shard lengths");
|
|
assert!(mismatch.to_string().contains("inconsistent shard length"));
|
|
}
|
|
|
|
#[test]
|
|
fn decode_data_and_parity_leaves_complete_shards_unchanged() {
|
|
let erasure = Erasure::new(4, 2, 128);
|
|
let data = b"complete shards should not be changed";
|
|
let encoded = erasure.encode_data(data).expect("encode should succeed");
|
|
let original = optional_shards(&encoded);
|
|
let mut shards = original.clone();
|
|
|
|
erasure
|
|
.decode_data_and_parity(&mut shards)
|
|
.expect("decode should succeed without missing shards");
|
|
|
|
assert_eq!(shards, original);
|
|
}
|
|
|
|
#[test]
|
|
fn decode_data_and_parity_reconstructs_missing_parity_shard() {
|
|
let erasure = Erasure::new(2, 2, 64);
|
|
let data = b"parity shard must be rebuilt";
|
|
let encoded = erasure.encode_data(data).expect("encode should succeed");
|
|
let mut shards = optional_shards(&encoded);
|
|
shards[2] = None;
|
|
|
|
erasure
|
|
.decode_data_and_parity(&mut shards)
|
|
.expect("decode should rebuild parity");
|
|
|
|
for (index, shard) in shards.iter().enumerate() {
|
|
assert_eq!(
|
|
shard.as_deref(),
|
|
Some(encoded[index].as_ref()),
|
|
"shard {index} should match encoded source"
|
|
);
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn decode_data_and_parity_reconstructs_missing_data_and_parity_shards() {
|
|
let erasure = Erasure::new(4, 2, 128);
|
|
let data = b"data and parity shards should both be reconstructed";
|
|
let encoded = erasure.encode_data(data).expect("encode should succeed");
|
|
let mut shards = optional_shards(&encoded);
|
|
shards[1] = None;
|
|
shards[4] = None;
|
|
|
|
erasure
|
|
.decode_data_and_parity(&mut shards)
|
|
.expect("decode should rebuild all missing shards");
|
|
|
|
for (index, shard) in shards.iter().enumerate() {
|
|
assert_eq!(
|
|
shard.as_deref(),
|
|
Some(encoded[index].as_ref()),
|
|
"shard {index} should match encoded source"
|
|
);
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn legacy_decode_data_and_parity_reconstructs_missing_parity_shard() {
|
|
let erasure = Erasure::new_with_options(2, 2, 64, true);
|
|
let data = b"legacy parity shard must be rebuilt";
|
|
let encoded = erasure.encode_data(data).expect("encode should succeed");
|
|
let mut shards = optional_shards(&encoded);
|
|
shards[3] = None;
|
|
|
|
erasure
|
|
.decode_data_and_parity(&mut shards)
|
|
.expect("decode should rebuild parity");
|
|
|
|
for (index, shard) in shards.iter().enumerate() {
|
|
assert_eq!(
|
|
shard.as_deref(),
|
|
Some(encoded[index].as_ref()),
|
|
"shard {index} should match encoded source"
|
|
);
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn legacy_decode_data_and_parity_reconstructs_empty_object_shards() {
|
|
let erasure = Erasure::new_with_options(3, 3, 64, true);
|
|
let encoded = erasure.encode_data(&[]).expect("empty encode should succeed");
|
|
let mut shards = optional_shards(&encoded);
|
|
shards[1] = None;
|
|
shards[4] = None;
|
|
|
|
erasure
|
|
.decode_data_and_parity(&mut shards)
|
|
.expect("empty decode should rebuild missing shards without SIMD");
|
|
|
|
for (index, shard) in shards.iter().enumerate() {
|
|
assert_eq!(
|
|
shard.as_deref(),
|
|
Some(encoded[index].as_ref()),
|
|
"empty shard {index} should match encoded source"
|
|
);
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn test_shard_file_size_cases2() {
|
|
let erasure = Erasure::new(12, 4, 1024 * 1024);
|
|
assert_eq!(erasure.shard_file_size(1572864), 131073);
|
|
}
|
|
|
|
proptest! {
|
|
#[test]
|
|
fn decode_data_and_parity_round_trips_bounded_recoverability(
|
|
(data_shards, parity_shards, block_size, uses_legacy, data, missing_indices) in
|
|
erasure_recoverability_case_strategy(),
|
|
) {
|
|
let erasure = Erasure::new_with_options(data_shards, parity_shards, block_size, uses_legacy);
|
|
let encoded = erasure.encode_data(&data)
|
|
.expect("encode_data should succeed for bounded recoverability property cases");
|
|
let mut shards = optional_shards(&encoded);
|
|
|
|
for index in &missing_indices {
|
|
shards[*index] = None;
|
|
}
|
|
|
|
erasure.decode_data_and_parity(&mut shards)
|
|
.expect("decode_data_and_parity should recover when missing shard count does not exceed parity");
|
|
|
|
let recovered = recover_data(&shards, data_shards, data.len());
|
|
prop_assert_eq!(recovered, data);
|
|
|
|
for (index, shard) in shards.iter().enumerate() {
|
|
prop_assert_eq!(
|
|
shard.as_deref(),
|
|
Some(encoded[index].as_ref()),
|
|
"reconstructed shard {} should match the original encoded shard",
|
|
index
|
|
);
|
|
}
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn test_shard_file_size_cases() {
|
|
let erasure = Erasure::new(4, 2, 8);
|
|
|
|
// Case 1: total_length == 0
|
|
assert_eq!(erasure.shard_file_size(0), 0);
|
|
|
|
// Case 2: total_length < block_size
|
|
assert_eq!(erasure.shard_file_size(5), 2); // 5 div_ceil 4 = 2
|
|
|
|
// Case 3: total_length == block_size
|
|
assert_eq!(erasure.shard_file_size(8), 2);
|
|
|
|
// Case 4: total_length > block_size, not aligned
|
|
assert_eq!(erasure.shard_file_size(13), 4); // 8/8=1, last=5, 5 div_ceil 4=2, 1*2+2=4
|
|
|
|
// Case 5: total_length > block_size, aligned
|
|
assert_eq!(erasure.shard_file_size(16), 4); // 16/8=2, last=0, 2*2+0=4
|
|
|
|
// MinIO-compatible: 1248739/8=156092, last=3, ceil(3/4)=1, 156092*2+1=312185
|
|
assert_eq!(erasure.shard_file_size(1248739), 312185);
|
|
|
|
// MinIO-compatible: 43/8=5, last=3, ceil(3/4)=1, 5*2+1=11
|
|
assert_eq!(erasure.shard_file_size(43), 11);
|
|
|
|
// 1572864 with block_size=8: 196608 full blocks, last=0, 196608*2+0=393216
|
|
assert_eq!(erasure.shard_file_size(1572864), 393216);
|
|
}
|
|
|
|
#[test]
|
|
fn test_encode_decode_roundtrip() {
|
|
let data_shards = 4;
|
|
let parity_shards = 2;
|
|
let block_size = 1024; // SIMD mode
|
|
let erasure = Erasure::new(data_shards, parity_shards, block_size);
|
|
|
|
// Use sufficient test data for SIMD optimization
|
|
let test_data = b"SIMD mode test data for encoding and decoding roundtrip verification with sufficient length to ensure shard size requirements are met for proper SIMD optimization.".repeat(20); // ~3KB for SIMD
|
|
|
|
let data = &test_data;
|
|
let encoded_shards = erasure.encode_data(data).expect("operation should succeed");
|
|
assert_eq!(encoded_shards.len(), data_shards + parity_shards);
|
|
|
|
// Create decode input with some shards missing, convert to the format expected by decode_data
|
|
let mut decode_input: Vec<Option<Vec<u8>>> = vec![None; data_shards + parity_shards];
|
|
for i in 0..data_shards {
|
|
decode_input[i] = Some(encoded_shards[i].to_vec());
|
|
}
|
|
|
|
erasure.decode_data(&mut decode_input).expect("operation should succeed");
|
|
|
|
// Recover original data
|
|
let mut recovered = Vec::new();
|
|
for shard in decode_input.iter().take(data_shards) {
|
|
recovered.extend_from_slice(shard.as_ref().expect("operation should succeed"));
|
|
}
|
|
recovered.truncate(data.len());
|
|
assert_eq!(&recovered, data);
|
|
}
|
|
|
|
#[test]
|
|
fn test_encode_decode_large_1m() {
|
|
let data_shards = 4;
|
|
let parity_shards = 2;
|
|
let block_size = 512 * 3; // SIMD mode
|
|
let erasure = Erasure::new(data_shards, parity_shards, block_size);
|
|
|
|
// Generate 1MB test data
|
|
let data: Vec<u8> = (0..1048576).map(|i| (i % 256) as u8).collect();
|
|
|
|
let encoded_shards = erasure.encode_data(&data).expect("operation should succeed");
|
|
assert_eq!(encoded_shards.len(), data_shards + parity_shards);
|
|
|
|
// Create decode input with some shards missing, convert to the format expected by decode_data
|
|
let mut decode_input: Vec<Option<Vec<u8>>> = vec![None; data_shards + parity_shards];
|
|
for i in 0..data_shards {
|
|
decode_input[i] = Some(encoded_shards[i].to_vec());
|
|
}
|
|
|
|
erasure.decode_data(&mut decode_input).expect("operation should succeed");
|
|
|
|
// Recover original data
|
|
let mut recovered = Vec::new();
|
|
for shard in decode_input.iter().take(data_shards) {
|
|
recovered.extend_from_slice(shard.as_ref().expect("operation should succeed"));
|
|
}
|
|
recovered.truncate(data.len());
|
|
assert_eq!(recovered, data);
|
|
}
|
|
|
|
#[test]
|
|
fn test_encode_all_zero_data() {
|
|
let data_shards = 3;
|
|
let parity_shards = 2;
|
|
let block_size = 6;
|
|
let erasure = Erasure::new(data_shards, parity_shards, block_size);
|
|
let data = vec![0u8; block_size];
|
|
let shards = erasure.encode_data(&data).expect("operation should succeed");
|
|
assert_eq!(shards.len(), data_shards + parity_shards);
|
|
let total_len: usize = shards.iter().map(|b| b.len()).sum();
|
|
assert_eq!(total_len, erasure.shard_size() * (data_shards + parity_shards));
|
|
}
|
|
|
|
#[test]
|
|
fn test_shard_size_and_file_size() {
|
|
let erasure = Erasure::new(4, 2, 8);
|
|
assert_eq!(erasure.shard_file_size(33), 9);
|
|
assert_eq!(erasure.shard_file_size(0), 0);
|
|
}
|
|
|
|
#[test]
|
|
fn test_legacy_shard_size_and_file_size() {
|
|
let erasure = Erasure::new_with_options(4, 2, 8, true);
|
|
assert_eq!(erasure.shard_size(), 2);
|
|
assert_eq!(calc_shard_size_legacy(8, 4), 2);
|
|
assert_eq!(calc_shard_size_legacy(1, 4), 2);
|
|
assert_eq!(erasure.shard_file_size(33), 10);
|
|
assert_eq!(erasure.shard_file_size(0), 0);
|
|
}
|
|
|
|
#[test]
|
|
fn test_legacy_encode_decode_roundtrip() {
|
|
let data_shards = 4;
|
|
let parity_shards = 2;
|
|
let block_size = 1024;
|
|
let erasure = Erasure::new_with_options(data_shards, parity_shards, block_size, true);
|
|
|
|
let data = b"Legacy encode/decode roundtrip test data with sufficient length.".repeat(20);
|
|
let encoded_shards = erasure.encode_data(&data).expect("operation should succeed");
|
|
assert_eq!(encoded_shards.len(), data_shards + parity_shards);
|
|
|
|
let mut decode_input: Vec<Option<Vec<u8>>> = vec![None; data_shards + parity_shards];
|
|
for i in 0..data_shards {
|
|
decode_input[i] = Some(encoded_shards[i].to_vec());
|
|
}
|
|
|
|
erasure.decode_data(&mut decode_input).expect("operation should succeed");
|
|
|
|
let mut recovered = Vec::new();
|
|
for shard in decode_input.iter().take(data_shards) {
|
|
recovered.extend_from_slice(shard.as_ref().expect("operation should succeed"));
|
|
}
|
|
recovered.truncate(data.len());
|
|
assert_eq!(&recovered, &data);
|
|
}
|
|
|
|
#[test]
|
|
fn test_legacy_decode_with_missing_shards() {
|
|
let data_shards = 4;
|
|
let parity_shards = 2;
|
|
let block_size = 256;
|
|
let erasure = Erasure::new_with_options(data_shards, parity_shards, block_size, true);
|
|
|
|
let data = b"Legacy decode with missing shards test.".repeat(10);
|
|
let encoded_shards = erasure.encode_data(&data).expect("operation should succeed");
|
|
|
|
let mut shards_opt: Vec<Option<Vec<u8>>> = encoded_shards.iter().map(|s| Some(s.to_vec())).collect();
|
|
shards_opt[1] = None;
|
|
shards_opt[5] = None;
|
|
|
|
erasure.decode_data(&mut shards_opt).expect("operation should succeed");
|
|
|
|
let mut recovered = Vec::new();
|
|
for shard in shards_opt.iter().take(data_shards) {
|
|
recovered.extend_from_slice(shard.as_ref().expect("operation should succeed"));
|
|
}
|
|
recovered.truncate(data.len());
|
|
assert_eq!(&recovered, &data);
|
|
}
|
|
|
|
#[test]
|
|
fn test_shard_file_offset() {
|
|
let erasure = Erasure::new(8, 8, 1024 * 1024);
|
|
let offset = erasure.shard_file_offset(0, 86, 86);
|
|
println!("offset={offset}");
|
|
assert!(offset > 0);
|
|
|
|
let total_length = erasure.shard_file_size(86);
|
|
println!("total_length={total_length}");
|
|
assert!(total_length > 0);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_encode_stream_callback_async_error_propagation() {
|
|
use std::io::Cursor;
|
|
use std::sync::Arc;
|
|
use tokio::sync::mpsc;
|
|
|
|
let data_shards = 4;
|
|
let parity_shards = 2;
|
|
let block_size = 1024; // SIMD mode
|
|
let erasure = Arc::new(Erasure::new(data_shards, parity_shards, block_size));
|
|
|
|
// Use test data suitable for SIMD mode
|
|
let data =
|
|
b"Async error test data with sufficient length to meet requirements for proper testing and validation.".repeat(20); // ~2KB
|
|
|
|
let mut reader = Cursor::new(data);
|
|
let (tx, mut rx) = mpsc::channel::<Vec<Bytes>>(8);
|
|
let erasure_clone = erasure.clone();
|
|
let handle = tokio::spawn(async move {
|
|
erasure_clone
|
|
.encode_stream_callback_async::<_, _, (), _>(&mut reader, move |res| {
|
|
let tx = tx.clone();
|
|
async move {
|
|
let shards = res.expect("operation should succeed");
|
|
tx.send(shards).await.expect("operation should succeed");
|
|
Ok(())
|
|
}
|
|
})
|
|
.await
|
|
.expect("operation should succeed");
|
|
});
|
|
let result = handle.await;
|
|
assert!(result.is_ok());
|
|
let collected_shards = rx.recv().await.expect("operation should succeed");
|
|
assert_eq!(collected_shards.len(), data_shards + parity_shards);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_encode_stream_callback_async_channel_decode() {
|
|
use std::io::Cursor;
|
|
use std::sync::Arc;
|
|
use tokio::sync::mpsc;
|
|
|
|
let data_shards = 4;
|
|
let parity_shards = 2;
|
|
let block_size = 1024; // SIMD mode
|
|
let erasure = Arc::new(Erasure::new(data_shards, parity_shards, block_size));
|
|
|
|
// Use test data that fits in exactly one block to avoid multi-block complexity
|
|
let data =
|
|
b"Channel async callback test data with sufficient length to ensure proper operation and validation requirements."
|
|
.repeat(8); // ~1KB
|
|
|
|
let data_clone = data.clone(); // Clone for later comparison
|
|
let mut reader = Cursor::new(data);
|
|
let (tx, mut rx) = mpsc::channel::<Vec<Bytes>>(8);
|
|
let erasure_clone = erasure.clone();
|
|
let handle = tokio::spawn(async move {
|
|
erasure_clone
|
|
.encode_stream_callback_async::<_, _, (), _>(&mut reader, move |res| {
|
|
let tx = tx.clone();
|
|
async move {
|
|
let shards = res.expect("operation should succeed");
|
|
tx.send(shards).await.expect("operation should succeed");
|
|
Ok(())
|
|
}
|
|
})
|
|
.await
|
|
.expect("operation should succeed");
|
|
});
|
|
let result = handle.await;
|
|
assert!(result.is_ok());
|
|
let shards = rx.recv().await.expect("operation should succeed");
|
|
assert_eq!(shards.len(), data_shards + parity_shards);
|
|
|
|
// Test decode using the old API that operates in-place
|
|
let mut decode_input: Vec<Option<Vec<u8>>> = vec![None; data_shards + parity_shards];
|
|
for i in 0..data_shards {
|
|
decode_input[i] = Some(shards[i].to_vec());
|
|
}
|
|
erasure.decode_data(&mut decode_input).expect("operation should succeed");
|
|
|
|
// Recover original data
|
|
let mut recovered = Vec::new();
|
|
for shard in decode_input.iter().take(data_shards) {
|
|
recovered.extend_from_slice(shard.as_ref().expect("operation should succeed"));
|
|
}
|
|
recovered.truncate(data_clone.len());
|
|
assert_eq!(&recovered, &data_clone);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_encode_stream_callback_async_reports_zero_block_size() {
|
|
use std::io::Cursor;
|
|
use std::sync::{Arc, Mutex};
|
|
|
|
let erasure = Arc::new(erasure_with_invalid_dimensions(1, 0, 0));
|
|
let mut reader = Cursor::new(b"payload".to_vec());
|
|
let observed = Arc::new(Mutex::new(None));
|
|
let observed_clone = observed.clone();
|
|
|
|
let total = erasure
|
|
.encode_stream_callback_async::<_, _, (), _>(&mut reader, move |res| {
|
|
let observed = observed_clone.clone();
|
|
async move {
|
|
let err = res.expect_err("zero block size should report an error");
|
|
*observed.lock().expect("operation should succeed") = Some((err.kind(), err.to_string()));
|
|
Ok(())
|
|
}
|
|
})
|
|
.await
|
|
.expect("callback should handle the zero block size error");
|
|
|
|
assert_eq!(total, 0);
|
|
let observed = observed.lock().expect("operation should succeed");
|
|
let (kind, message) = observed.as_ref().expect("callback should be invoked once");
|
|
assert_eq!(*kind, io::ErrorKind::InvalidInput);
|
|
assert!(message.contains("block_size"));
|
|
}
|
|
|
|
// SIMD mode specific tests
|
|
mod simd_tests {
|
|
use super::*;
|
|
|
|
#[test]
|
|
fn test_simd_encode_decode_roundtrip() {
|
|
let data_shards = 4;
|
|
let parity_shards = 2;
|
|
let block_size = 1024; // Use larger block size for SIMD mode
|
|
let erasure = Erasure::new(data_shards, parity_shards, block_size);
|
|
|
|
// Use data that will create shards >= 512 bytes for SIMD optimization
|
|
let test_data = b"SIMD mode test data for encoding and decoding roundtrip verification with sufficient length to ensure shard size requirements are met for proper SIMD optimization and validation.";
|
|
let data = test_data.repeat(25); // Create much larger data: ~5KB total, ~1.25KB per shard
|
|
|
|
let encoded_shards = erasure.encode_data(&data).expect("operation should succeed");
|
|
assert_eq!(encoded_shards.len(), data_shards + parity_shards);
|
|
|
|
// Create decode input with some shards missing
|
|
let mut shards_opt: Vec<Option<Vec<u8>>> = encoded_shards.iter().map(|shard| Some(shard.to_vec())).collect();
|
|
|
|
// Lose one data shard and one parity shard (should still be recoverable)
|
|
shards_opt[1] = None; // Lose second data shard
|
|
shards_opt[5] = None; // Lose second parity shard
|
|
|
|
erasure.decode_data(&mut shards_opt).expect("operation should succeed");
|
|
|
|
// Verify recovered data
|
|
let mut recovered = Vec::new();
|
|
for shard in shards_opt.iter().take(data_shards) {
|
|
recovered.extend_from_slice(shard.as_ref().expect("operation should succeed"));
|
|
}
|
|
recovered.truncate(data.len());
|
|
assert_eq!(&recovered, &data);
|
|
}
|
|
|
|
#[test]
|
|
fn test_simd_all_zero_data() {
|
|
let data_shards = 4;
|
|
let parity_shards = 2;
|
|
let block_size = 1024; // Use larger block size for SIMD mode
|
|
let erasure = Erasure::new(data_shards, parity_shards, block_size);
|
|
|
|
// Create all-zero data that ensures adequate shard size for SIMD optimization
|
|
let data = vec![0u8; 1024]; // 1KB of zeros, each shard will be 256 bytes
|
|
|
|
let encoded_shards = erasure.encode_data(&data).expect("operation should succeed");
|
|
assert_eq!(encoded_shards.len(), data_shards + parity_shards);
|
|
|
|
// Verify that all data shards are zeros
|
|
for (i, shard) in encoded_shards.iter().enumerate().take(data_shards) {
|
|
assert!(shard.iter().all(|&x| x == 0), "Data shard {i} should be all zeros");
|
|
}
|
|
|
|
// Test recovery with some shards missing
|
|
let mut shards_opt: Vec<Option<Vec<u8>>> = encoded_shards.iter().map(|shard| Some(shard.to_vec())).collect();
|
|
|
|
// Lose maximum recoverable shards (equal to parity_shards)
|
|
shards_opt[0] = None; // Lose first data shard
|
|
shards_opt[4] = None; // Lose first parity shard
|
|
|
|
erasure.decode_data(&mut shards_opt).expect("operation should succeed");
|
|
|
|
// Verify recovered data is still all zeros
|
|
let mut recovered = Vec::new();
|
|
for shard in shards_opt.iter().take(data_shards) {
|
|
recovered.extend_from_slice(shard.as_ref().expect("operation should succeed"));
|
|
}
|
|
recovered.truncate(data.len());
|
|
assert!(recovered.iter().all(|&x| x == 0), "Recovered data should be all zeros");
|
|
}
|
|
|
|
#[test]
|
|
fn test_simd_large_data_1kb() {
|
|
let data_shards = 8;
|
|
let parity_shards = 4;
|
|
let block_size = 1024; // 1KB block size optimal for SIMD
|
|
let erasure = Erasure::new(data_shards, parity_shards, block_size);
|
|
|
|
// Create 1KB of test data
|
|
let mut data = Vec::with_capacity(1024);
|
|
for i in 0..1024 {
|
|
data.push((i % 256) as u8);
|
|
}
|
|
|
|
let shards = erasure.encode_data(&data).expect("operation should succeed");
|
|
assert_eq!(shards.len(), data_shards + parity_shards);
|
|
|
|
// Simulate the loss of multiple shards
|
|
let mut shards_opt: Vec<Option<Vec<u8>>> = shards.iter().map(|b| Some(b.to_vec())).collect();
|
|
shards_opt[0] = None;
|
|
shards_opt[3] = None;
|
|
shards_opt[9] = None; // Parity shard
|
|
shards_opt[11] = None; // Parity shard
|
|
|
|
// Decode
|
|
erasure.decode_data(&mut shards_opt).expect("operation should succeed");
|
|
|
|
// Recover original data
|
|
let mut recovered = Vec::new();
|
|
for shard in shards_opt.iter().take(data_shards) {
|
|
recovered.extend_from_slice(shard.as_ref().expect("operation should succeed"));
|
|
}
|
|
recovered.truncate(data.len());
|
|
assert_eq!(&recovered, &data);
|
|
}
|
|
|
|
#[test]
|
|
fn test_simd_minimum_shard_size() {
|
|
let data_shards = 4;
|
|
let parity_shards = 2;
|
|
let block_size = 256; // Use 256 bytes to ensure sufficient shard size
|
|
let erasure = Erasure::new(data_shards, parity_shards, block_size);
|
|
|
|
// Create data that will result in 64+ byte shards
|
|
let data = vec![0x42u8; 200]; // 200 bytes, should create ~50 byte shards per data shard
|
|
|
|
let shards = erasure.encode_data(&data).expect("minimum shard size data should encode");
|
|
println!("SIMD encoding succeeded with shard size: {}", shards[0].len());
|
|
|
|
// Test decoding
|
|
let mut shards_opt: Vec<Option<Vec<u8>>> = shards.iter().map(|b| Some(b.to_vec())).collect();
|
|
shards_opt[1] = None;
|
|
|
|
erasure
|
|
.decode_data(&mut shards_opt)
|
|
.expect("minimum shard size data should decode");
|
|
let mut recovered = Vec::new();
|
|
for shard in shards_opt.iter().take(data_shards) {
|
|
recovered.extend_from_slice(shard.as_ref().expect("operation should succeed"));
|
|
}
|
|
recovered.truncate(data.len());
|
|
assert_eq!(&recovered, &data);
|
|
}
|
|
|
|
#[test]
|
|
fn test_simd_maximum_erasures() {
|
|
let data_shards = 5;
|
|
let parity_shards = 3;
|
|
let block_size = 512;
|
|
let erasure = Erasure::new(data_shards, parity_shards, block_size);
|
|
|
|
let data =
|
|
b"Testing maximum erasure capacity with SIMD Reed-Solomon implementation for robustness verification!".repeat(3);
|
|
|
|
let shards = erasure.encode_data(&data).expect("operation should succeed");
|
|
|
|
// Lose exactly the maximum number of shards (equal to parity_shards)
|
|
let mut shards_opt: Vec<Option<Vec<u8>>> = shards.iter().map(|b| Some(b.to_vec())).collect();
|
|
shards_opt[0] = None; // Data shard
|
|
shards_opt[2] = None; // Data shard
|
|
shards_opt[6] = None; // Parity shard
|
|
|
|
// Should succeed with maximum erasures
|
|
erasure.decode_data(&mut shards_opt).expect("operation should succeed");
|
|
|
|
let mut recovered = Vec::new();
|
|
for shard in shards_opt.iter().take(data_shards) {
|
|
recovered.extend_from_slice(shard.as_ref().expect("operation should succeed"));
|
|
}
|
|
recovered.truncate(data.len());
|
|
assert_eq!(&recovered, &data);
|
|
}
|
|
|
|
/// Generates 7557 bytes identical to MinIO generateCompatTestData.
|
|
fn generate_compat_test_data(size: usize) -> Vec<u8> {
|
|
(0..size).map(|i| ((i * 7 + 13) % 256) as u8).collect()
|
|
}
|
|
|
|
/// Verifies reed-solomon-simd produces same shards.
|
|
/// Data shards (0-3) must match for MinIO to read RustFS part files.
|
|
/// Parity shards (4-5) differ: reed-solomon-simd vs klauspost use different RS encoding.
|
|
/// Run: cargo test -p rustfs-ecstore test_reed_solomon_compat
|
|
#[test]
|
|
fn test_reed_solomon_compat() {
|
|
let data = generate_compat_test_data(7557);
|
|
let erasure = Erasure::new(4, 2, 7557);
|
|
let shards = erasure.encode_data(&data).expect("operation should succeed");
|
|
assert_eq!(shards.len(), 6, "expected 6 shards (4 data + 2 parity)");
|
|
|
|
// Per-shard HighwayHash
|
|
let expected_hashes: [&str; 6] = [
|
|
"fb3db9338e610cec541504ddae4b0bfd54445bcbd45318cf21f35f024240914d", // data 0
|
|
"a545269a3196e18e77ef9f5ec6e735a4f4ebe82d342db666b11a5256eb305720", // data 1
|
|
"2adbf0058f36c4cbcb5c9c16c38a6530c54198dfe504179a6f92d2349f245318", // data 2
|
|
"898e6d060b0cb4f0e830add7e1f936bc8b78442bf582283ee244a3a058602db8", // data 3
|
|
"4a20460bca044b3a777b26f2b0bcd371e3eab2f156f84778be3ccd8edd521ef2", // parity 4
|
|
"eb8ba4c0db15ca910d58d031f74e4601ba2fed62ad03ec29cadde3367ab0d415", // parity 5
|
|
];
|
|
|
|
let mut data_shards_match = true;
|
|
let mut parity_shards_match = true;
|
|
for (i, shard) in shards.iter().enumerate() {
|
|
let hash = rustfs_utils::HashAlgorithm::HighwayHash256S.hash_encode(shard);
|
|
let got = hex_simd::encode_to_string(hash.as_ref(), hex_simd::AsciiCase::Lower);
|
|
let matches = got == expected_hashes[i];
|
|
if i < 4 {
|
|
data_shards_match &= matches;
|
|
} else {
|
|
parity_shards_match &= matches;
|
|
}
|
|
if !matches {
|
|
eprintln!(
|
|
"Shard {} ({}): got {} want {}",
|
|
i,
|
|
if i < 4 { "data" } else { "parity" },
|
|
got,
|
|
expected_hashes[i]
|
|
);
|
|
}
|
|
}
|
|
assert!(data_shards_match, "Data shards (0-3) must match");
|
|
assert!(parity_shards_match, "Parity shards (4-5): reed-solomon-simd differs");
|
|
}
|
|
|
|
#[test]
|
|
fn test_simd_small_data_handling() {
|
|
let data_shards = 4;
|
|
let parity_shards = 2;
|
|
let block_size = 32; // Small block size for testing edge cases
|
|
let erasure = Erasure::new(data_shards, parity_shards, block_size);
|
|
|
|
// Use small data to test SIMD handling of small shards
|
|
let small_data = b"tiny!123".to_vec(); // 8 bytes data
|
|
|
|
// Test encoding with small data
|
|
let shards = erasure.encode_data(&small_data).expect("small data should encode");
|
|
println!("SIMD encoding succeeded: {} bytes into {} shards", small_data.len(), shards.len());
|
|
assert_eq!(shards.len(), data_shards + parity_shards);
|
|
|
|
// Test decoding
|
|
let mut shards_opt: Vec<Option<Vec<u8>>> = shards.iter().map(|shard| Some(shard.to_vec())).collect();
|
|
|
|
// Lose some shards to test recovery
|
|
shards_opt[1] = None; // Lose one data shard
|
|
shards_opt[4] = None; // Lose one parity shard
|
|
|
|
erasure.decode_data(&mut shards_opt).expect("small data should decode");
|
|
let mut recovered = Vec::new();
|
|
for shard in shards_opt.iter().take(data_shards) {
|
|
recovered.extend_from_slice(shard.as_ref().expect("operation should succeed"));
|
|
}
|
|
recovered.truncate(small_data.len());
|
|
println!("recovered: {recovered:?}");
|
|
println!("small_data: {small_data:?}");
|
|
assert_eq!(&recovered, &small_data);
|
|
}
|
|
|
|
#[test]
|
|
fn test_simd_large_block_1mb() {
|
|
let data_shards = 6;
|
|
let parity_shards = 3;
|
|
let block_size = 1024 * 1024; // 1MB block size
|
|
let erasure = Erasure::new(data_shards, parity_shards, block_size);
|
|
|
|
// Build 2 MB of test data so multiple 1 MB chunks are exercised
|
|
let mut data = Vec::with_capacity(2 * 1024 * 1024);
|
|
for i in 0..(2 * 1024 * 1024) {
|
|
data.push((i % 256) as u8);
|
|
}
|
|
|
|
println!("🚀 Testing SIMD with 1MB block size and 2MB data");
|
|
println!(
|
|
"📊 Data shards: {}, Parity shards: {}, Total data: {}KB",
|
|
data_shards,
|
|
parity_shards,
|
|
data.len() / 1024
|
|
);
|
|
|
|
// Encode the data
|
|
let start = std::time::Instant::now();
|
|
let shards = erasure.encode_data(&data).expect("operation should succeed");
|
|
let encode_duration = start.elapsed();
|
|
|
|
println!("⏱️ Encoding completed in: {encode_duration:?}");
|
|
println!("📦 Generated {} shards, each shard size: {}KB", shards.len(), shards[0].len() / 1024);
|
|
|
|
assert_eq!(shards.len(), data_shards + parity_shards);
|
|
|
|
// Verify that each shard is large enough for SIMD optimization
|
|
for (i, shard) in shards.iter().enumerate() {
|
|
println!("🔍 Shard {}: {} bytes ({}KB)", i, shard.len(), shard.len() / 1024);
|
|
assert!(shard.len() >= 512, "Shard {} is too small for SIMD: {} bytes", i, shard.len());
|
|
}
|
|
|
|
// Simulate data loss - lose maximum recoverable number of shards
|
|
let mut shards_opt: Vec<Option<Vec<u8>>> = shards.iter().map(|b| Some(b.to_vec())).collect();
|
|
shards_opt[0] = None; // Lose 1st data shard
|
|
shards_opt[2] = None; // Lose 3rd data shard
|
|
shards_opt[8] = None; // Lose 3rd parity shard (index 6+3-1=8)
|
|
|
|
println!("💥 Simulated loss of 3 shards (max recoverable with 3 parity shards)");
|
|
|
|
// Decode and recover data
|
|
let start = std::time::Instant::now();
|
|
erasure.decode_data(&mut shards_opt).expect("operation should succeed");
|
|
let decode_duration = start.elapsed();
|
|
|
|
println!("⏱️ Decoding completed in: {decode_duration:?}");
|
|
|
|
// Verify recovered data integrity
|
|
let mut recovered = Vec::new();
|
|
for shard in shards_opt.iter().take(data_shards) {
|
|
recovered.extend_from_slice(shard.as_ref().expect("operation should succeed"));
|
|
}
|
|
recovered.truncate(data.len());
|
|
|
|
assert_eq!(recovered.len(), data.len());
|
|
assert_eq!(&recovered, &data, "Data mismatch after recovery!");
|
|
|
|
println!("✅ Successfully verified data integrity after recovery");
|
|
println!("📈 Performance summary:");
|
|
println!(
|
|
" - Encode: {:?} ({:.2} MB/s)",
|
|
encode_duration,
|
|
(data.len() as f64 / (1024.0 * 1024.0)) / encode_duration.as_secs_f64()
|
|
);
|
|
println!(
|
|
" - Decode: {:?} ({:.2} MB/s)",
|
|
decode_duration,
|
|
(data.len() as f64 / (1024.0 * 1024.0)) / decode_duration.as_secs_f64()
|
|
);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_simd_stream_callback() {
|
|
use std::io::Cursor;
|
|
use std::sync::Arc;
|
|
use tokio::sync::mpsc;
|
|
|
|
let data_shards = 4;
|
|
let parity_shards = 2;
|
|
let block_size = 256; // Larger block for SIMD
|
|
let erasure = Arc::new(Erasure::new(data_shards, parity_shards, block_size));
|
|
|
|
let test_data = b"SIMD stream processing test with sufficient data length for multiple blocks and proper SIMD optimization verification!";
|
|
let data = test_data.repeat(5); // Create owned Vec<u8>
|
|
let data_clone = data.clone(); // Clone for later comparison
|
|
let mut reader = Cursor::new(data);
|
|
|
|
let (tx, mut rx) = mpsc::channel::<Vec<Bytes>>(16);
|
|
let erasure_clone = erasure.clone();
|
|
|
|
let handle = tokio::spawn(async move {
|
|
erasure_clone
|
|
.encode_stream_callback_async::<_, _, (), _>(&mut reader, move |res| {
|
|
let tx = tx.clone();
|
|
async move {
|
|
let shards = res.expect("operation should succeed");
|
|
tx.send(shards).await.expect("operation should succeed");
|
|
Ok(())
|
|
}
|
|
})
|
|
.await
|
|
.expect("operation should succeed");
|
|
});
|
|
|
|
let mut all_blocks = Vec::new();
|
|
while let Some(block) = rx.recv().await {
|
|
all_blocks.push(block);
|
|
}
|
|
handle.await.expect("operation should succeed");
|
|
|
|
// Verify we got multiple blocks
|
|
assert!(all_blocks.len() > 1, "Should have multiple blocks for stream test");
|
|
|
|
// Test recovery for each block
|
|
let mut recovered = Vec::new();
|
|
for block in &all_blocks {
|
|
let mut shards_opt: Vec<Option<Vec<u8>>> = block.iter().map(|b| Some(b.to_vec())).collect();
|
|
// Lose one data shard and one parity shard
|
|
shards_opt[1] = None;
|
|
shards_opt[5] = None;
|
|
|
|
erasure.decode_data(&mut shards_opt).expect("operation should succeed");
|
|
|
|
for shard in shards_opt.iter().take(data_shards) {
|
|
recovered.extend_from_slice(shard.as_ref().expect("operation should succeed"));
|
|
}
|
|
}
|
|
|
|
recovered.truncate(data_clone.len());
|
|
assert_eq!(&recovered, &data_clone);
|
|
}
|
|
}
|
|
}
|