diff --git a/crates/ecstore/src/disk/fs.rs b/crates/ecstore/src/disk/fs.rs index cdd70aef7..63d3216c2 100644 --- a/crates/ecstore/src/disk/fs.rs +++ b/crates/ecstore/src/disk/fs.rs @@ -161,33 +161,39 @@ pub async fn make_dir_all(path: impl AsRef) -> io::Result<()> { fs::create_dir_all(path.as_ref()).await } +fn is_dir_error(e: &io::Error) -> bool { + e.raw_os_error() == Some(libc::EISDIR) + || e.kind() == io::ErrorKind::IsADirectory + // macOS: remove_file on a directory returns EPERM + || (cfg!(target_os = "macos") && e.raw_os_error() == Some(libc::EPERM)) +} + #[tracing::instrument(level = "debug", skip_all)] pub async fn remove(path: impl AsRef) -> io::Result<()> { - let meta = fs::metadata(path.as_ref()).await?; - if meta.is_dir() { - fs::remove_dir(path.as_ref()).await - } else { - fs::remove_file(path.as_ref()).await + // Try remove_file first; fall back to remove_dir if it's a directory + match fs::remove_file(path.as_ref()).await { + Ok(()) => Ok(()), + Err(e) if is_dir_error(&e) => fs::remove_dir(path.as_ref()).await, + Err(e) => Err(e), } } pub async fn remove_all(path: impl AsRef) -> io::Result<()> { - let meta = fs::metadata(path.as_ref()).await?; - if meta.is_dir() { - fs::remove_dir_all(path.as_ref()).await - } else { - fs::remove_file(path.as_ref()).await + // Try remove_file first; fall back to remove_dir_all if it's a directory + match fs::remove_file(path.as_ref()).await { + Ok(()) => Ok(()), + Err(e) if is_dir_error(&e) => fs::remove_dir_all(path.as_ref()).await, + Err(e) => Err(e), } } #[tracing::instrument(level = "debug", skip_all)] pub fn remove_std(path: impl AsRef) -> io::Result<()> { - let path = path.as_ref(); - let meta = std::fs::metadata(path)?; - if meta.is_dir() { - std::fs::remove_dir(path) - } else { - std::fs::remove_file(path) + // Try remove_file first; fall back to remove_dir if it's a directory + match std::fs::remove_file(path.as_ref()) { + Ok(()) => Ok(()), + Err(e) if is_dir_error(&e) => std::fs::remove_dir(path.as_ref()), + Err(e) => Err(e), } } diff --git a/crates/ecstore/src/disk/local.rs b/crates/ecstore/src/disk/local.rs index 0890cfdc5..5983cebea 100644 --- a/crates/ecstore/src/disk/local.rs +++ b/crates/ecstore/src/disk/local.rs @@ -2557,75 +2557,57 @@ impl DiskAPI for LocalDisk { check_path_length(src_file_path.to_string_lossy().to_string().as_str())?; check_path_length(dst_file_path.to_string_lossy().to_string().as_str())?; - // Read the previous xl.meta - - let has_dst_buf = match super::fs::read_file(&dst_file_path).await { - Ok(res) => Some(res), - Err(e) => { - let e: DiskError = to_file_error(e).into(); - - if e != DiskError::FileNotFound { - return Err(e); - } - - None - } - }; - - let mut xlmeta = FileMeta::new(); - - if let Some(dst_buf) = has_dst_buf.as_ref() - && FileMeta::is_xl2_v1_format(dst_buf) - && let Ok(nmeta) = FileMeta::load(dst_buf) - { - xlmeta = nmeta - } - - let mut skip_parent = dst_volume_dir.clone(); - if has_dst_buf.as_ref().is_some() - && let Some(parent) = dst_file_path.parent() - { - skip_parent = parent.to_path_buf(); - } - - // TODO: Healing - - let version_id = fi.version_id.unwrap_or_default(); - let search_version_id = Some(version_id); let no_inline = fi.data.is_none() && fi.size > 0; - // Check if there's an existing version with the same version_id that has a data_dir to clean up - // Reuse one metadata scan to find the version data_dir and determine whether it is shared. - let has_old_data_dir = xlmeta.find_unshared_data_dir_for_version(search_version_id); - if let Some(old_data_dir) = has_old_data_dir.as_ref() { - let _ = xlmeta.data.remove_two(version_id, *old_data_dir); - } - - xlmeta.add_version(fi)?; - - if xlmeta.versions.len() <= 10 { - // TODO: Sign - } - - if let Some((src_data_path, dst_data_path)) = has_data_dir_path.as_ref() { - let src_file_parent = src_file_path.parent().unwrap_or(src_volume_dir.as_path()); - let meta_skip_parent = if no_inline { - src_file_parent - } else { - src_volume_dir.as_path() + if no_inline { + // Non-inline: read xl.meta, parse, write, rename data dir, rename xl.meta + let has_dst_buf = match super::fs::read_file(&dst_file_path).await { + Ok(res) => Some(res), + Err(e) => { + let e: DiskError = to_file_error(e).into(); + if e != DiskError::FileNotFound { + return Err(e); + } + None + } }; + + let mut xlmeta = FileMeta::new(); + if let Some(dst_buf) = has_dst_buf.as_ref() + && FileMeta::is_xl2_v1_format(dst_buf) + && let Ok(nmeta) = FileMeta::load(dst_buf) + { + xlmeta = nmeta + } + + let mut skip_parent = dst_volume_dir.clone(); + if has_dst_buf.as_ref().is_some() + && let Some(parent) = dst_file_path.parent() + { + skip_parent = parent.to_path_buf(); + } + + let version_id = fi.version_id.unwrap_or_default(); + let has_old_data_dir = xlmeta.find_unshared_data_dir_for_version(Some(version_id)); + if let Some(old_data_dir) = has_old_data_dir.as_ref() { + let _ = xlmeta.data.remove_two(version_id, *old_data_dir); + } + xlmeta.add_version(fi)?; let new_dst_buf = xlmeta.marshal_msg()?; + let src_file_parent = src_file_path.parent().unwrap_or(src_volume_dir.as_path()); self.write_all_private( src_volume, - format!("{}/{}", &src_path, STORAGE_FORMAT_FILE).as_str(), + &format!("{}/{}", &src_path, STORAGE_FORMAT_FILE), new_dst_buf.into(), true, - meta_skip_parent, + src_file_parent, ) .await?; - if no_inline && let Err(err) = rename_all(&src_data_path, &dst_data_path, &skip_parent).await { + if let Some((src_data_path, dst_data_path)) = has_data_dir_path.as_ref() + && let Err(err) = rename_all(src_data_path, dst_data_path, &skip_parent).await + { let _ = self.delete_file(&dst_volume_dir, dst_data_path, false, false).await; info!( "rename all failed src_data_path: {:?}, dst_data_path: {:?}, err: {:?}", @@ -2633,19 +2615,21 @@ impl DiskAPI for LocalDisk { ); return Err(err); } - } else { - let new_dst_buf = xlmeta.marshal_msg()?; - self.write_all(src_volume, format!("{}/{}", &src_path, STORAGE_FORMAT_FILE).as_str(), new_dst_buf.into()) - .await?; - } - if let Some(old_data_dir) = has_old_data_dir { - // preserve current xl.meta inside the oldDataDir. - if let Some(dst_buf) = has_dst_buf + if let Err(err) = rename_all(&src_file_path, &dst_file_path, &skip_parent).await { + if let Some((_, dst_data_path)) = has_data_dir_path.as_ref() { + let _ = self.delete_file(&dst_volume_dir, dst_data_path, false, false).await; + } + info!("rename all failed err: {:?}", err); + return Err(err); + } + + if let Some(old_data_dir) = has_old_data_dir + && let Some(dst_buf) = has_dst_buf && let Err(err) = self .write_all_private( dst_volume, - format!("{}/{}/{}", &dst_path, &old_data_dir.to_string(), STORAGE_FORMAT_FILE).as_str(), + &format!("{}/{}/{}", &dst_path, &old_data_dir.to_string(), STORAGE_FORMAT_FILE), dst_buf.into(), true, &skip_parent, @@ -2655,30 +2639,106 @@ impl DiskAPI for LocalDisk { info!("write_all_private failed err: {:?}", err); return Err(err); } - } - if let Err(err) = rename_all(&src_file_path, &dst_file_path, &skip_parent).await { - if let Some((_, dst_data_path)) = has_data_dir_path.as_ref() { - let _ = self.delete_file(&dst_volume_dir, dst_data_path, false, false).await; + if let Some(src_file_path_parent) = src_file_path.parent() { + if src_volume != super::RUSTFS_META_MULTIPART_BUCKET { + let _ = remove_std(src_file_path_parent); + } else { + let _ = self + .delete_file(&dst_volume_dir, &src_file_path_parent.to_path_buf(), true, false) + .await; + } } - info!("rename all failed err: {:?}", err); - return Err(err); - } - if let Some(src_file_path_parent) = src_file_path.parent() { - if src_volume != super::RUSTFS_META_MULTIPART_BUCKET { - let _ = remove_std(src_file_path_parent); + Ok(RenameDataResp { + old_data_dir: has_old_data_dir, + sign: None, + }) + } else { + // Inline: merge read + parse + write + rename into single spawn_blocking + let src = src_file_path.clone(); + let dst = dst_file_path.clone(); + let cleanup_path = if src_volume == super::RUSTFS_META_MULTIPART_BUCKET { + src_file_path.parent().map(|p| p.to_path_buf()) } else { - let _ = self - .delete_file(&dst_volume_dir, &src_file_path_parent.to_path_buf(), true, false) - .await; - } - } + None + }; - Ok(RenameDataResp { - old_data_dir: has_old_data_dir, - sign: None, // TODO: - }) + let (old_data_dir, _dst_buf) = tokio::task::spawn_blocking(move || { + // Read existing xl.meta + let has_dst_buf = match std::fs::read(&dst) { + Ok(buf) => Some(Bytes::from(buf)), + Err(e) if e.kind() == std::io::ErrorKind::NotFound => None, + Err(e) => return Err(to_file_error(e)), + }; + + let mut xlmeta = FileMeta::new(); + if let Some(ref buf) = has_dst_buf + && FileMeta::is_xl2_v1_format(buf) + && let Ok(nmeta) = FileMeta::load(buf) + { + xlmeta = nmeta + } + + let version_id = fi.version_id.unwrap_or_default(); + let old_data_dir = xlmeta.find_unshared_data_dir_for_version(Some(version_id)); + if let Some(d) = old_data_dir.as_ref() { + let _ = xlmeta.data.remove_two(version_id, *d); + } + xlmeta.add_version(fi)?; + let new_buf = xlmeta.marshal_msg()?; + + // Write new xl.meta + rename + if let Some(parent) = src.parent() { + std::fs::create_dir_all(parent)?; + } + let mut f = std::fs::OpenOptions::new() + .create(true) + .write(true) + .truncate(true) + .open(&src)?; + std::io::Write::write_all(&mut f, &new_buf)?; + match std::fs::rename(&src, &dst) { + Ok(()) => Ok(()), + Err(err) if err.kind() == std::io::ErrorKind::NotFound && !src.exists() => Ok(()), + Err(err) if err.kind() == std::io::ErrorKind::NotFound => { + if let Some(parent) = dst.parent() { + std::fs::create_dir_all(parent)?; + } + std::fs::rename(&src, &dst).map_err(to_file_error)?; + Ok(()) + } + Err(err) => Err(to_file_error(err)), + }?; + + if let Some(old_dir) = old_data_dir.as_ref() + && let Some(ref buf) = has_dst_buf + && let Some(dst_parent) = dst.parent() + { + let old_path = dst_parent.join(old_dir.to_string()).join(STORAGE_FORMAT_FILE); + if let Some(old_parent) = old_path.parent() { + std::fs::create_dir_all(old_parent)?; + } + std::fs::write(&old_path, buf).map_err(to_file_error)?; + } + + Ok::<(Option, Option), std::io::Error>((old_data_dir, has_dst_buf)) + }) + .await + .map_err(DiskError::from)??; + + // Cleanup + if let Some(ref cleanup) = cleanup_path { + let _ = self.delete_file(&dst_volume_dir, cleanup, true, false).await; + } else if let Some(parent) = src_file_path.parent() { + let _ = remove_std(parent); + } + + Ok(RenameDataResp { + old_data_dir, + sign: None, + }) + } } #[tracing::instrument(skip(self))] diff --git a/crates/ecstore/src/disk/os.rs b/crates/ecstore/src/disk/os.rs index ac6751429..c4172d8c9 100644 --- a/crates/ecstore/src/disk/os.rs +++ b/crates/ecstore/src/disk/os.rs @@ -148,7 +148,6 @@ async fn reliable_rename( if let Some(parent) = dst_file_path.as_ref().parent() && !file_exists(parent) { - // info!("reliable_rename reliable_mkdir_all parent: {:?}", parent); reliable_mkdir_all(parent, base_dir.as_ref()).await?; } diff --git a/crates/ecstore/src/erasure_coding/encode.rs b/crates/ecstore/src/erasure_coding/encode.rs index d4781af0b..386ecff28 100644 --- a/crates/ecstore/src/erasure_coding/encode.rs +++ b/crates/ecstore/src/erasure_coding/encode.rs @@ -294,6 +294,33 @@ impl Erasure { writers.shutdown().await?; Ok((reader, total)) } + + /// Fast path for small inline objects: skip tokio::spawn + mpsc channel. + /// Reads all data, encodes directly, writes shards sequentially. + pub async fn encode_inline_small( + self: Arc, + mut reader: R, + writers: &mut [Option], + quorum: usize, + ) -> std::io::Result<(R, usize)> + where + R: AsyncRead + Send + Sync + Unpin, + { + use tokio::io::AsyncReadExt; + + let mut buf = Vec::with_capacity(self.block_size); + let total = reader.read_to_end(&mut buf).await?; + + if total == 0 { + return Ok((reader, 0)); + } + + let shards = self.encode_data(&buf)?; + let mut mw = MultiWriter::new(writers, quorum); + mw.write(shards).await?; + mw.shutdown().await?; + Ok((reader, total)) + } } #[cfg(test)] @@ -357,6 +384,61 @@ mod tests { assert!(!committed.lock().unwrap().is_empty()); } + /// encode_inline_small: empty reader returns (reader, 0) without writing to any shard. + #[tokio::test] + async fn encode_inline_small_empty_stream_returns_zero() { + let committed = Arc::new(Mutex::new(Vec::new())); + let writer = DeferredCommitWriter::new(committed.clone()); + // 1 data shard, 0 parity shards, block_size = 16 + let mut writers = vec![Some(BitrotWriterWrapper::new( + CustomWriter::new_tokio_writer(writer), + 16, + HashAlgorithm::HighwayHash256S, + ))]; + + let erasure = Arc::new(Erasure::new(1, 0, 16)); + let reader = tokio::io::BufReader::new(std::io::Cursor::new(Vec::::new())); + let (_reader, total) = erasure.encode_inline_small(reader, &mut writers, 1).await.unwrap(); + + assert_eq!(total, 0); + // No shutdown was called, so nothing should be committed + assert!(committed.lock().unwrap().is_empty()); + } + + /// encode_inline_small: small payload is encoded into the correct number of shards + /// and each writer receives data after shutdown. + #[tokio::test] + async fn encode_inline_small_payload_writes_all_shards() { + const DATA_SHARDS: usize = 2; + const PARITY_SHARDS: usize = 2; + const TOTAL_SHARDS: usize = DATA_SHARDS + PARITY_SHARDS; + const BLOCK_SIZE: usize = 64; + + let committed: Vec>>> = (0..TOTAL_SHARDS).map(|_| Arc::new(Mutex::new(Vec::new()))).collect(); + + let mut writers: Vec> = committed + .iter() + .map(|c| { + Some(BitrotWriterWrapper::new( + CustomWriter::new_tokio_writer(DeferredCommitWriter::new(c.clone())), + BLOCK_SIZE / DATA_SHARDS, + HashAlgorithm::HighwayHash256S, + )) + }) + .collect(); + + let payload = b"hello inline small"; + let erasure = Arc::new(Erasure::new(DATA_SHARDS, PARITY_SHARDS, BLOCK_SIZE)); + let reader = tokio::io::BufReader::new(std::io::Cursor::new(payload.to_vec())); + let (_reader, total) = erasure.encode_inline_small(reader, &mut writers, DATA_SHARDS).await.unwrap(); + + assert_eq!(total, payload.len()); + // All shards must have received data (shutdown flushed the bitrot header + shard bytes) + for (i, c) in committed.iter().enumerate() { + assert!(!c.lock().unwrap().is_empty(), "shard {i} should have received data"); + } + } + #[test] fn encode_channel_capacity_never_returns_zero() { assert_eq!(encode_channel_capacity(0, 1024), 1); diff --git a/crates/ecstore/src/set_disk.rs b/crates/ecstore/src/set_disk.rs index 425f35d83..3ea052286 100644 --- a/crates/ecstore/src/set_disk.rs +++ b/crates/ecstore/src/set_disk.rs @@ -844,38 +844,45 @@ impl ObjectIO for SetDisks { } }; - let mut writers = Vec::with_capacity(shuffle_disks.len()); - let mut errors = Vec::with_capacity(shuffle_disks.len()); - for disk_op in shuffle_disks.iter() { - if let Some(disk) = disk_op - && disk.is_online().await - { - let writer = match create_bitrot_writer( - is_inline_buffer, - Some(disk), - RUSTFS_META_TMP_BUCKET, - &tmp_object, - erasure.shard_file_size(data.size()), - erasure.shard_size(), - HashAlgorithm::HighwayHash256S, - ) - .await - { - Ok(writer) => writer, - Err(err) => { - warn!("create_bitrot_writer disk {}, err {:?}, skipping operation", disk.to_string(), err); - errors.push(Some(err)); - writers.push(None); - continue; + let shard_file_size = erasure.shard_file_size(data.size()); + let shard_size = erasure.shard_size(); + let writer_futs: Vec<_> = shuffle_disks + .iter() + .map(|disk_op| { + let tmp_obj = tmp_object.clone(); + async move { + if let Some(disk) = disk_op + && disk.is_online().await + { + match create_bitrot_writer( + is_inline_buffer, + Some(disk), + RUSTFS_META_TMP_BUCKET, + &tmp_obj, + shard_file_size, + shard_size, + HashAlgorithm::HighwayHash256S, + ) + .await + { + Ok(writer) => (Some(writer), None), + Err(err) => { + warn!("create_bitrot_writer disk {}, err {:?}, skipping operation", disk.to_string(), err); + (None, Some(err)) + } + } + } else { + (None, Some(DiskError::DiskNotFound)) } - }; - - writers.push(Some(writer)); - errors.push(None); - } else { - errors.push(Some(DiskError::DiskNotFound)); - writers.push(None); - } + } + }) + .collect(); + let writer_results = join_all(writer_futs).await; + let mut writers = Vec::with_capacity(writer_results.len()); + let mut errors = Vec::with_capacity(writer_results.len()); + for (w, e) in writer_results { + writers.push(w); + errors.push(e); } let nil_count = errors.iter().filter(|&e| e.is_none()).count(); @@ -893,13 +900,28 @@ impl ObjectIO for SetDisks { HashReader::from_stream(Cursor::new(Vec::new()), 0, 0, None, None, false)?, ); - let (reader, w_size) = match Arc::new(erasure).encode(stream, &mut writers, write_quorum).await { - Ok((r, w)) => (r, w), - Err(e) => { - error!("encode err {:?}", e); - return Err(e.into()); + let use_fast_path = is_inline_buffer && data.size() <= fi.erasure.block_size as i64; + + let (reader, w_size) = if use_fast_path { + match Arc::new(erasure) + .encode_inline_small(stream, &mut writers, write_quorum) + .await + { + Ok((r, w)) => (r, w), + Err(e) => { + error!("encode_inline_small err {:?}", e); + return Err(e.into()); + } } - }; // TODO: delete temporary directory on error + } else { + match Arc::new(erasure).encode(stream, &mut writers, write_quorum).await { + Ok((r, w)) => (r, w), + Err(e) => { + error!("encode err {:?}", e); + return Err(e.into()); + } + } + }; let _ = mem::replace(&mut data.stream, reader); // if let Err(err) = close_bitrot_writers(&mut writers).await {