diff --git a/ecstore/src/erasure.rs b/ecstore/src/erasure.rs index 8f871d1dc..bac408e6d 100644 --- a/ecstore/src/erasure.rs +++ b/ecstore/src/erasure.rs @@ -50,25 +50,35 @@ impl Erasure { S: Stream> + Send + Sync + 'static, W: AsyncWrite + Unpin, { - let mut stream = ChunkedStream::new(body, total_size, self.block_size, true); + let mut stream = ChunkedStream::new(body, total_size, self.block_size, false); let mut total: usize = 0; let mut idx = 0; while let Some(result) = stream.next().await { match result { Ok(data) => { total += data.len(); + + // EOF + if data.is_empty() { + break; + } + + idx += 1; + // warn!("encode {} get data {}", idx, data.len()); + let blocks = self.encode_data(data.as_ref())?; - warn!( - "encode shard size: {}/{} from block_size {}, total_size {} ", - blocks[0].len(), - blocks.len(), - data.len(), - total_size - ); + // warn!( + // "encode shard {} size: {}/{} from block_size {}, total_size {} ", + // idx, + // blocks[0].len(), + // blocks.len(), + // data.len(), + // total_size + // ); let mut errs = Vec::new(); - idx += 1; + for (i, w) in writers.iter_mut().enumerate() { // debug!( // "{} {}-{} encode write {} , total:{}, readed:{}", @@ -98,7 +108,7 @@ impl Erasure { } } - warn!(" encode_data done shard block num {}", idx); + // warn!(" encode_data done shard block num {}", idx); Ok(total) @@ -124,15 +134,15 @@ impl Erasure { let mut reader = ShardReader::new(readers, self, offset, total_length); - warn!("ShardReader {:?}", &reader); + // warn!("ShardReader {:?}", &reader); let start_block = offset / self.block_size; let end_block = (offset + length) / self.block_size; - warn!("decode block from {} to {}", start_block, end_block); + // warn!("decode block from {} to {}", start_block, end_block); let mut bytes_writed = 0; - let mut idx = 0; + for block_idx in start_block..=end_block { let mut block_offset = 0; let mut block_length = 0; @@ -151,10 +161,11 @@ impl Erasure { } if block_length == 0 { + // warn!("block_length == 0 break"); break; } - warn!("decode {} block_offset {},block_length {} ", idx, block_offset, block_length); + // warn!("decode {} block_offset {},block_length {} ", block_idx, block_offset, block_length); let mut bufs = reader.read().await?; @@ -165,11 +176,12 @@ impl Erasure { .await?; bytes_writed += writed_n; - idx += 1; + + // warn!("decode {} writed_n {}, total_writed: {} ", block_idx, writed_n, bytes_writed); } if bytes_writed != length { - warn!("bytes_writed != length: {} != {} ", bytes_writed, length); + // warn!("bytes_writed != length: {} != {} ", bytes_writed, length); return Err(Error::msg("erasure decode less data")); } @@ -200,7 +212,7 @@ impl Erasure { let mut offset = offset; - warn!("write_data_blocks offset {}, length {}", offset, length); + // warn!("write_data_blocks offset {}, length {}", offset, length); let mut write = length; let mut total_writed = 0; @@ -217,10 +229,22 @@ impl Erasure { offset = 0; - // if write < buf.len() {} + // warn!("write_data_blocks write buf len {}", buf.len()); - let n = writer.write(buf).await?; + if write < buf.len() { + let buf = &buf[..write]; + // warn!("write_data_blocks write buf less len {}", buf.len()); + writer.write_all(buf).await?; + // warn!("write_data_blocks write done len {}", buf.len()); + total_writed += buf.len(); + break; + } + + writer.write_all(buf).await?; + let n = buf.len(); + + // warn!("write_data_blocks write done len {}", n); write -= n; total_writed += n; } @@ -279,16 +303,16 @@ impl Erasure { return 0; } - let mut num_shards = total_size / self.block_size; + let num_shards = total_size / self.block_size; let last_block_size = total_size % self.block_size; - // let last_shard_size = (last_block_size + self.data_shards - 1) / self.data_shards; - // num_shards * self.shard_size(self.block_size) + last_shard_size + let last_shard_size = (last_block_size + self.data_shards - 1) / self.data_shards; + num_shards * self.shard_size(self.block_size) + last_shard_size - // 因为写入的时候ec需要补全，所以最后一个长度应该也是一样的 - if last_block_size != 0 { - num_shards += 1 - } - num_shards * self.shard_size(self.block_size) + // // 因为写入的时候ec需要补全，所以最后一个长度应该也是一样的 + // if last_block_size != 0 { + // num_shards += 1 + // } + // num_shards * self.shard_size(self.block_size) } } @@ -318,10 +342,17 @@ impl ShardReader { pub async fn read(&mut self) -> Result>>> { // let mut disks = self.readers; - - warn!("shard reader read offset {}, shard_size {}", self.offset, self.shard_size); - let reader_length = self.readers.len(); + let mut read_length = self.shard_size; + if self.offset + read_length > self.shard_file_size { + read_length = self.shard_file_size - self.offset + } + + if read_length == 0 { + return Ok(vec![None; reader_length]); + } + + // warn!("shard reader read offset {}, shard_size {}", self.offset, read_length); let mut futures = Vec::with_capacity(reader_length); let mut errors = Vec::with_capacity(reader_length); @@ -336,7 +367,7 @@ impl ShardReader { } let disk: &mut FileReader = disk.as_mut().unwrap(); - futures.push(disk.read_at(self.offset, self.shard_size)); + futures.push(disk.read_at(self.offset, read_length)); } let results = join_all(futures).await; diff --git a/ecstore/src/store_api.rs b/ecstore/src/store_api.rs index 56ee977bf..c2c46f3c8 100644 --- a/ecstore/src/store_api.rs +++ b/ecstore/src/store_api.rs @@ -6,6 +6,7 @@ use s3s::dto::StreamingBlob; use serde::{Deserialize, Serialize}; use time::OffsetDateTime; use tokio::io::{AsyncRead, DuplexStream}; +use tracing::warn; use uuid::Uuid; pub const ERASURE_ALGORITHM: &str = "rs-vandermonde"; diff --git a/rustfs/src/storage/ecfs.rs b/rustfs/src/storage/ecfs.rs index a760827c1..780b3cf24 100644 --- a/rustfs/src/storage/ecfs.rs +++ b/rustfs/src/storage/ecfs.rs @@ -155,11 +155,16 @@ impl S3 for FS { .await ); - // let body = bytes_stream(reader.stream, 100); + let info = reader.object_info; + + let content_type = try_!(ContentType::from_str("application/x-msdownload")); + let last_modified = Timestamp::from(info.mod_time); let output = GetObjectOutput { body: Some(reader.stream), - content_length: Some(reader.object_info.size as i64), + content_length: Some(info.size as i64), + last_modified: Some(last_modified), + content_type: Some(content_type), ..Default::default() }; @@ -208,11 +213,12 @@ impl S3 for FS { debug!("info {:?}", info); let content_type = try_!(ContentType::from_str("application/x-msdownload")); + let last_modified = Timestamp::from(info.mod_time); let output = HeadObjectOutput { content_length: Some(try_!(i64::try_from(info.size))), content_type: Some(content_type), - last_modified: Some(info.mod_time.into()), + last_modified: Some(last_modified), // metadata: object_metadata, ..Default::default() }; @@ -268,6 +274,8 @@ impl S3 for FS { .. } = input; + debug!("put_object metadata {:?}", metadata); + let Some(body) = body else { return Err(s3_error!(IncompleteBody)) }; let Some(content_length) = content_length else { return Err(s3_error!(IncompleteBody)) };