mirror of
https://github.com/rustfs/rustfs.git
synced 2026-08-05 21:07:43 +00:00
Fix large file upload freeze with adaptive buffer sizing (#869)
* Initial plan * Fix large file upload freeze by increasing StreamReader buffer size Co-authored-by: houseme <4829346+houseme@users.noreply.github.com> * Add comprehensive documentation for large file upload freeze fix Co-authored-by: houseme <4829346+houseme@users.noreply.github.com> * upgrade s3s version * Fix compilation error: use BufReader instead of non-existent StreamReader::with_capacity Co-authored-by: houseme <4829346+houseme@users.noreply.github.com> * Update documentation with correct BufReader implementation Co-authored-by: houseme <4829346+houseme@users.noreply.github.com> * add tokio feature `io-util` * Implement adaptive buffer sizing based on file size Co-authored-by: houseme <4829346+houseme@users.noreply.github.com> * Constants are managed uniformly and fmt code * fix * Fix: Trigger self-heal on read when shards missing from rejoined nodes (#871) * Initial plan * Fix: Trigger self-heal when missing shards detected during read - Added proactive heal detection in get_object_with_fileinfo - When reading an object, now checks if any shards are missing even if read succeeds - Sends low-priority heal request to reconstruct missing shards on rejoined nodes - This fixes the issue where data written during node outage is not healed when node rejoins Co-authored-by: houseme <4829346+houseme@users.noreply.github.com> * fix * Unify CRC implementations to crc-fast (#873) * Initial plan * Replace CRC libraries with unified crc-fast implementation Co-authored-by: houseme <4829346+houseme@users.noreply.github.com> * fix * fix: replace low to Normal --------- Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com> Co-authored-by: houseme <4829346+houseme@users.noreply.github.com> Co-authored-by: houseme <housemecn@gmail.com> --------- Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com> Co-authored-by: houseme <housemecn@gmail.com> Co-authored-by: houseme <4829346+houseme@users.noreply.github.com> --------- Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com> Co-authored-by: houseme <4829346+houseme@users.noreply.github.com> Co-authored-by: houseme <housemecn@gmail.com>
This commit is contained in:
Copilot
@@ -68,6 +68,7 @@ use md5::{Digest as Md5Digest, Md5};
|
||||
use rand::{Rng, seq::SliceRandom};
|
||||
use regex::Regex;
|
||||
use rustfs_common::heal_channel::{DriveState, HealChannelPriority, HealItemType, HealOpts, HealScanMode, send_heal_disk};
|
||||
use rustfs_config::MI_B;
|
||||
use rustfs_filemeta::{
|
||||
FileInfo, FileMeta, FileMetaShallowVersion, MetaCacheEntries, MetaCacheEntry, MetadataResolutionParams, ObjectPartInfo,
|
||||
RawFileInfo, ReplicationStatusType, VersionPurgeStatusType, file_info_from_raw, merge_file_meta_versions,
|
||||
@@ -111,7 +112,7 @@ use tracing::error;
|
||||
use tracing::{debug, info, warn};
|
||||
use uuid::Uuid;
|
||||
|
||||
pub const DEFAULT_READ_BUFFER_SIZE: usize = 1024 * 1024;
|
||||
pub const DEFAULT_READ_BUFFER_SIZE: usize = MI_B; // 1 MiB = 1024 * 1024;
|
||||
pub const MAX_PARTS_COUNT: usize = 10000;
|
||||
const DISK_ONLINE_TIMEOUT: Duration = Duration::from_secs(1);
|
||||
const DISK_HEALTH_CACHE_TTL: Duration = Duration::from_millis(750);
|
||||
@@ -2212,7 +2213,7 @@ impl SetDisks {
|
||||
where
|
||||
W: AsyncWrite + Send + Sync + Unpin + 'static,
|
||||
{
|
||||
tracing::debug!(bucket, object, requested_length = length, offset, "get_object_with_fileinfo start");
|
||||
debug!(bucket, object, requested_length = length, offset, "get_object_with_fileinfo start");
|
||||
let (disks, files) = Self::shuffle_disks_and_parts_metadata_by_index(disks, &files, &fi);
|
||||
|
||||
let total_size = fi.size as usize;
|
||||
@@ -2237,27 +2238,20 @@ impl SetDisks {
|
||||
|
||||
let (last_part_index, last_part_relative_offset) = fi.to_part_offset(end_offset)?;
|
||||
|
||||
tracing::debug!(
|
||||
debug!(
|
||||
bucket,
|
||||
object,
|
||||
offset,
|
||||
length,
|
||||
end_offset,
|
||||
part_index,
|
||||
last_part_index,
|
||||
last_part_relative_offset,
|
||||
"Multipart read bounds"
|
||||
object, offset, length, end_offset, part_index, last_part_index, last_part_relative_offset, "Multipart read bounds"
|
||||
);
|
||||
|
||||
let erasure = erasure_coding::Erasure::new(fi.erasure.data_blocks, fi.erasure.parity_blocks, fi.erasure.block_size);
|
||||
|
||||
let part_indices: Vec<usize> = (part_index..=last_part_index).collect();
|
||||
tracing::debug!(bucket, object, ?part_indices, "Multipart part indices to stream");
|
||||
debug!(bucket, object, ?part_indices, "Multipart part indices to stream");
|
||||
|
||||
let mut total_read = 0;
|
||||
for current_part in part_indices {
|
||||
if total_read == length {
|
||||
tracing::debug!(
|
||||
debug!(
|
||||
bucket,
|
||||
object,
|
||||
total_read,
|
||||
@@ -2279,7 +2273,7 @@ impl SetDisks {
|
||||
|
||||
let read_offset = (part_offset / erasure.block_size) * erasure.shard_size();
|
||||
|
||||
tracing::debug!(
|
||||
debug!(
|
||||
bucket,
|
||||
object,
|
||||
part_index = current_part,
|
||||
@@ -2334,12 +2328,39 @@ impl SetDisks {
|
||||
return Err(Error::other(format!("not enough disks to read: {errors:?}")));
|
||||
}
|
||||
|
||||
// Check if we have missing shards even though we can read successfully
|
||||
// This happens when a node was offline during write and comes back online
|
||||
let total_shards = erasure.data_shards + erasure.parity_shards;
|
||||
let missing_shards = total_shards - nil_count;
|
||||
if missing_shards > 0 && nil_count >= erasure.data_shards {
|
||||
// We have missing shards but enough to read - trigger background heal
|
||||
info!(
|
||||
bucket,
|
||||
object,
|
||||
part_number,
|
||||
missing_shards,
|
||||
available_shards = nil_count,
|
||||
"Detected missing shards during read, triggering background heal"
|
||||
);
|
||||
let _ = rustfs_common::heal_channel::send_heal_request(
|
||||
rustfs_common::heal_channel::create_heal_request_with_options(
|
||||
bucket.to_string(),
|
||||
Some(object.to_string()),
|
||||
false,
|
||||
Some(HealChannelPriority::Normal), // Use low priority for proactive healing
|
||||
Some(pool_index),
|
||||
Some(set_index),
|
||||
),
|
||||
)
|
||||
.await;
|
||||
}
|
||||
|
||||
// debug!(
|
||||
// "read part {} part_offset {},part_length {},part_size {} ",
|
||||
// part_number, part_offset, part_length, part_size
|
||||
// );
|
||||
let (written, err) = erasure.decode(writer, readers, part_offset, part_length, part_size).await;
|
||||
tracing::debug!(
|
||||
debug!(
|
||||
bucket,
|
||||
object,
|
||||
part_index = current_part,
|
||||
@@ -2386,7 +2407,7 @@ impl SetDisks {
|
||||
|
||||
// debug!("read end");
|
||||
|
||||
tracing::debug!(bucket, object, total_read, expected_length = length, "Multipart read finished");
|
||||
debug!(bucket, object, total_read, expected_length = length, "Multipart read finished");
|
||||
|
||||
Ok(())
|
||||
}
|
||||
@@ -5663,7 +5684,7 @@ impl StorageAPI for SetDisks {
|
||||
}
|
||||
|
||||
let ext_part = &curr_fi.parts[i];
|
||||
tracing::info!(target:"rustfs_ecstore::set_disk", part_number = p.part_num, part_size = ext_part.size, part_actual_size = ext_part.actual_size, "Completing multipart part");
|
||||
info!(target:"rustfs_ecstore::set_disk", part_number = p.part_num, part_size = ext_part.size, part_actual_size = ext_part.actual_size, "Completing multipart part");
|
||||
|
||||
// Normalize ETags by removing quotes before comparison (PR #592 compatibility)
|
||||
let client_etag = p.etag.as_ref().map(|e| rustfs_utils::path::trim_etag(e));
|
||||
|
||||
@@ -26,7 +26,7 @@ categories = ["web-programming", "development-tools", "filesystem"]
|
||||
documentation = "https://docs.rs/rustfs-filemeta/latest/rustfs_filemeta/"
|
||||
|
||||
[dependencies]
|
||||
crc32fast = { workspace = true }
|
||||
crc-fast = { workspace = true }
|
||||
rmp.workspace = true
|
||||
rmp-serde.workspace = true
|
||||
serde.workspace = true
|
||||
|
||||
@@ -220,7 +220,11 @@ impl FileInfo {
|
||||
let indices = {
|
||||
let cardinality = data_blocks + parity_blocks;
|
||||
let mut nums = vec![0; cardinality];
|
||||
let key_crc = crc32fast::hash(object.as_bytes());
|
||||
let key_crc = {
|
||||
let mut hasher = crc_fast::Digest::new(crc_fast::CrcAlgorithm::Crc32IsoHdlc);
|
||||
hasher.update(object.as_bytes());
|
||||
hasher.finalize() as u32
|
||||
};
|
||||
|
||||
let start = key_crc as usize % cardinality;
|
||||
for i in 1..=cardinality {
|
||||
|
||||
@@ -33,7 +33,7 @@ tokio = { workspace = true, features = ["full"] }
|
||||
rand = { workspace = true }
|
||||
http.workspace = true
|
||||
aes-gcm = { workspace = true }
|
||||
crc32fast = { workspace = true }
|
||||
crc-fast = { workspace = true }
|
||||
pin-project-lite.workspace = true
|
||||
serde = { workspace = true }
|
||||
bytes.workspace = true
|
||||
@@ -49,10 +49,8 @@ thiserror.workspace = true
|
||||
base64.workspace = true
|
||||
sha1.workspace = true
|
||||
sha2.workspace = true
|
||||
crc64fast-nvme.workspace = true
|
||||
s3s.workspace = true
|
||||
hex-simd.workspace = true
|
||||
crc32c.workspace = true
|
||||
|
||||
[dev-dependencies]
|
||||
tokio-test = { workspace = true }
|
||||
|
||||
+30
-17
@@ -15,7 +15,6 @@
|
||||
use crate::errors::ChecksumMismatch;
|
||||
use base64::{Engine as _, engine::general_purpose};
|
||||
use bytes::Bytes;
|
||||
use crc32fast::Hasher as Crc32Hasher;
|
||||
use http::HeaderMap;
|
||||
use sha1::Sha1;
|
||||
use sha2::{Digest, Sha256};
|
||||
@@ -612,7 +611,7 @@ pub trait ChecksumHasher: Write + Send + Sync {
|
||||
|
||||
/// CRC32 IEEE hasher
|
||||
pub struct Crc32IeeeHasher {
|
||||
hasher: Crc32Hasher,
|
||||
hasher: crc_fast::Digest,
|
||||
}
|
||||
|
||||
impl Default for Crc32IeeeHasher {
|
||||
@@ -624,7 +623,7 @@ impl Default for Crc32IeeeHasher {
|
||||
impl Crc32IeeeHasher {
|
||||
pub fn new() -> Self {
|
||||
Self {
|
||||
hasher: Crc32Hasher::new(),
|
||||
hasher: crc_fast::Digest::new(crc_fast::CrcAlgorithm::Crc32IsoHdlc),
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -642,27 +641,36 @@ impl Write for Crc32IeeeHasher {
|
||||
|
||||
impl ChecksumHasher for Crc32IeeeHasher {
|
||||
fn finalize(&mut self) -> Vec<u8> {
|
||||
self.hasher.clone().finalize().to_be_bytes().to_vec()
|
||||
(self.hasher.clone().finalize() as u32).to_be_bytes().to_vec()
|
||||
}
|
||||
|
||||
fn reset(&mut self) {
|
||||
self.hasher = Crc32Hasher::new();
|
||||
self.hasher = crc_fast::Digest::new(crc_fast::CrcAlgorithm::Crc32IsoHdlc);
|
||||
}
|
||||
}
|
||||
|
||||
/// CRC32 Castagnoli hasher
|
||||
#[derive(Default)]
|
||||
pub struct Crc32CastagnoliHasher(u32);
|
||||
pub struct Crc32CastagnoliHasher {
|
||||
hasher: crc_fast::Digest,
|
||||
}
|
||||
|
||||
impl Default for Crc32CastagnoliHasher {
|
||||
fn default() -> Self {
|
||||
Self::new()
|
||||
}
|
||||
}
|
||||
|
||||
impl Crc32CastagnoliHasher {
|
||||
pub fn new() -> Self {
|
||||
Self::default()
|
||||
Self {
|
||||
hasher: crc_fast::Digest::new(crc_fast::CrcAlgorithm::Crc32Iscsi),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl Write for Crc32CastagnoliHasher {
|
||||
fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
|
||||
self.0 = crc32c::crc32c_append(self.0, buf);
|
||||
self.hasher.update(buf);
|
||||
Ok(buf.len())
|
||||
}
|
||||
|
||||
@@ -673,11 +681,11 @@ impl Write for Crc32CastagnoliHasher {
|
||||
|
||||
impl ChecksumHasher for Crc32CastagnoliHasher {
|
||||
fn finalize(&mut self) -> Vec<u8> {
|
||||
self.0.to_be_bytes().to_vec()
|
||||
(self.hasher.clone().finalize() as u32).to_be_bytes().to_vec()
|
||||
}
|
||||
|
||||
fn reset(&mut self) {
|
||||
self.0 = 0;
|
||||
self.hasher = crc_fast::Digest::new(crc_fast::CrcAlgorithm::Crc32Iscsi);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -758,22 +766,27 @@ impl ChecksumHasher for Sha256Hasher {
|
||||
}
|
||||
|
||||
/// CRC64 NVME hasher
|
||||
#[derive(Default)]
|
||||
pub struct Crc64NvmeHasher {
|
||||
hasher: crc64fast_nvme::Digest,
|
||||
hasher: crc_fast::Digest,
|
||||
}
|
||||
|
||||
impl Default for Crc64NvmeHasher {
|
||||
fn default() -> Self {
|
||||
Self::new()
|
||||
}
|
||||
}
|
||||
|
||||
impl Crc64NvmeHasher {
|
||||
pub fn new() -> Self {
|
||||
Self {
|
||||
hasher: Default::default(),
|
||||
hasher: crc_fast::Digest::new(crc_fast::CrcAlgorithm::Crc64Nvme),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl Write for Crc64NvmeHasher {
|
||||
fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
|
||||
self.hasher.write(buf);
|
||||
self.hasher.update(buf);
|
||||
Ok(buf.len())
|
||||
}
|
||||
|
||||
@@ -784,11 +797,11 @@ impl Write for Crc64NvmeHasher {
|
||||
|
||||
impl ChecksumHasher for Crc64NvmeHasher {
|
||||
fn finalize(&mut self) -> Vec<u8> {
|
||||
self.hasher.sum64().to_be_bytes().to_vec()
|
||||
self.hasher.clone().finalize().to_be_bytes().to_vec()
|
||||
}
|
||||
|
||||
fn reset(&mut self) {
|
||||
self.hasher = Default::default();
|
||||
self.hasher = crc_fast::Digest::new(crc_fast::CrcAlgorithm::Crc64Nvme);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -356,7 +356,11 @@ where
|
||||
*this.compressed_len = 0;
|
||||
return Poll::Ready(Err(io::Error::new(io::ErrorKind::InvalidData, "Decompressed length mismatch")));
|
||||
}
|
||||
let actual_crc = crc32fast::hash(&decompressed);
|
||||
let actual_crc = {
|
||||
let mut hasher = crc_fast::Digest::new(crc_fast::CrcAlgorithm::Crc32IsoHdlc);
|
||||
hasher.update(&decompressed);
|
||||
hasher.finalize() as u32
|
||||
};
|
||||
if actual_crc != crc {
|
||||
// error!("DecompressReader CRC32 mismatch: actual {actual_crc} != expected {crc}");
|
||||
this.compressed_buf.take();
|
||||
@@ -404,7 +408,11 @@ where
|
||||
|
||||
/// Build compressed block with header + uvarint + compressed data
|
||||
fn build_compressed_block(uncompressed_data: &[u8], compression_algorithm: CompressionAlgorithm) -> Vec<u8> {
|
||||
let crc = crc32fast::hash(uncompressed_data);
|
||||
let crc = {
|
||||
let mut hasher = crc_fast::Digest::new(crc_fast::CrcAlgorithm::Crc32IsoHdlc);
|
||||
hasher.update(uncompressed_data);
|
||||
hasher.finalize() as u32
|
||||
};
|
||||
let compressed_data = compress_block(uncompressed_data, compression_algorithm);
|
||||
let uncompressed_len = uncompressed_data.len();
|
||||
let mut uncompressed_len_buf = [0u8; 10];
|
||||
|
||||
@@ -102,7 +102,11 @@ where
|
||||
let nonce = Nonce::try_from(this.nonce.as_slice()).map_err(|_| Error::other("invalid nonce length"))?;
|
||||
let plaintext = &temp_buf.filled()[..n];
|
||||
let plaintext_len = plaintext.len();
|
||||
let crc = crc32fast::hash(plaintext);
|
||||
let crc = {
|
||||
let mut hasher = crc_fast::Digest::new(crc_fast::CrcAlgorithm::Crc32IsoHdlc);
|
||||
hasher.update(plaintext);
|
||||
hasher.finalize() as u32
|
||||
};
|
||||
let ciphertext = cipher
|
||||
.encrypt(&nonce, plaintext)
|
||||
.map_err(|e| Error::other(format!("encrypt error: {e}")))?;
|
||||
@@ -409,7 +413,11 @@ where
|
||||
return Poll::Ready(Err(Error::other("Plaintext length mismatch")));
|
||||
}
|
||||
|
||||
let actual_crc = crc32fast::hash(&plaintext);
|
||||
let actual_crc = {
|
||||
let mut hasher = crc_fast::Digest::new(crc_fast::CrcAlgorithm::Crc32IsoHdlc);
|
||||
hasher.update(&plaintext);
|
||||
hasher.finalize() as u32
|
||||
};
|
||||
if actual_crc != crc {
|
||||
this.ciphertext_buf.take();
|
||||
*this.ciphertext_read = 0;
|
||||
|
||||
@@ -29,7 +29,7 @@ base64-simd = { workspace = true, optional = true }
|
||||
blake3 = { workspace = true, optional = true }
|
||||
brotli = { workspace = true, optional = true }
|
||||
bytes = { workspace = true, optional = true }
|
||||
crc32fast = { workspace = true, optional = true }
|
||||
crc-fast = { workspace = true, optional = true }
|
||||
flate2 = { workspace = true, optional = true }
|
||||
futures = { workspace = true, optional = true }
|
||||
hashbrown = { workspace = true, optional = true }
|
||||
@@ -88,7 +88,7 @@ notify = ["dep:hyper", "dep:s3s", "dep:hashbrown", "dep:thiserror", "dep:serde",
|
||||
compress = ["dep:flate2", "dep:brotli", "dep:snap", "dep:lz4", "dep:zstd"]
|
||||
string = ["dep:regex", "dep:rand"]
|
||||
crypto = ["dep:base64-simd", "dep:hex-simd", "dep:hmac", "dep:hyper", "dep:sha1"]
|
||||
hash = ["dep:highway", "dep:md-5", "dep:sha2", "dep:blake3", "dep:serde", "dep:siphasher", "dep:hex-simd", "dep:base64-simd", "dep:crc32fast"]
|
||||
hash = ["dep:highway", "dep:md-5", "dep:sha2", "dep:blake3", "dep:serde", "dep:siphasher", "dep:hex-simd", "dep:base64-simd", "dep:crc-fast"]
|
||||
os = ["dep:nix", "dep:tempfile", "winapi"] # operating system utilities
|
||||
integration = [] # integration test features
|
||||
sys = ["dep:sysinfo"] # system information features
|
||||
|
||||
@@ -115,7 +115,6 @@ impl HashAlgorithm {
|
||||
}
|
||||
}
|
||||
|
||||
use crc32fast::Hasher;
|
||||
use siphasher::sip::SipHasher;
|
||||
|
||||
pub const EMPTY_STRING_SHA256_HASH: &str = "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855";
|
||||
@@ -151,11 +150,9 @@ pub fn sip_hash(key: &str, cardinality: usize, id: &[u8; 16]) -> usize {
|
||||
/// A usize representing the bucket index
|
||||
///
|
||||
pub fn crc_hash(key: &str, cardinality: usize) -> usize {
|
||||
let mut hasher = Hasher::new(); // Create a new hasher
|
||||
|
||||
hasher.update(key.as_bytes()); // Update hash state, add data
|
||||
|
||||
let checksum = hasher.finalize();
|
||||
let mut hasher = crc_fast::Digest::new(crc_fast::CrcAlgorithm::Crc32IsoHdlc);
|
||||
hasher.update(key.as_bytes());
|
||||
let checksum = hasher.finalize() as u32;
|
||||
|
||||
checksum as usize % cardinality
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user