diff --git a/crates/event-notifier/examples/webhook.rs b/crates/event-notifier/examples/webhook.rs index c754f8223..4cdf02c66 100644 --- a/crates/event-notifier/examples/webhook.rs +++ b/crates/event-notifier/examples/webhook.rs @@ -37,7 +37,7 @@ async fn receive_webhook(Json(payload): Json) -> StatusCode { println!("current time:{:04}-{:02}-{:02} {:02}:{:02}:{:02}", year, month, day, hour, minute, second); println!( "received a webhook request time:{} content:\n {}", - seconds.to_string(), + seconds, serde_json::to_string_pretty(&payload).unwrap() ); StatusCode::OK @@ -66,10 +66,10 @@ fn convert_seconds_to_date(seconds: u64) -> (u32, u32, u32, u32, u32, u32) { // calculate month let days_in_month = [31, 28, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31]; - for m in 0..12 { - if total_seconds >= days_in_month[m] * seconds_per_day { + for m in &days_in_month { + if total_seconds >= m * seconds_per_day { month += 1; - total_seconds -= days_in_month[m] * seconds_per_day; + total_seconds -= m * seconds_per_day; } else { break; } diff --git a/crates/event-notifier/src/global.rs b/crates/event-notifier/src/global.rs index 25fadeeb4..19693532d 100644 --- a/crates/event-notifier/src/global.rs +++ b/crates/event-notifier/src/global.rs @@ -189,7 +189,7 @@ mod tests { let config = NotifierConfig::default(); let _ = initialize(config.clone()).await; // first initialization let result = initialize(config).await; // second initialization - assert!(!result.is_ok(), "Initialization should succeed"); + assert!(result.is_err(), "Initialization should succeed"); assert!(result.is_err(), "Re-initialization should fail"); } @@ -211,7 +211,7 @@ mod tests { ..Default::default() }; let result = initialize(config).await; - assert!(!result.is_err(), "Initialization with invalid config should fail"); + assert!(result.is_ok(), "Initialization with invalid config should fail"); assert!(is_initialized(), "System should not be marked as initialized after failure"); assert!(is_ready(), "System should not be marked as ready after failure"); } diff --git a/crates/obs/src/telemetry.rs b/crates/obs/src/telemetry.rs index 95bedfc4a..c73d35c0d 100644 --- a/crates/obs/src/telemetry.rs +++ b/crates/obs/src/telemetry.rs @@ -218,7 +218,7 @@ pub fn init_telemetry(config: &OtelConfig) -> OtelGuard { let tracer = tracer_provider.tracer(Cow::Borrowed(service_name).to_string()); // Configure registry to avoid repeated calls to filter methods - let _registry = tracing_subscriber::registry() + tracing_subscriber::registry() .with(filter) .with(ErrorLayer::default()) .with(if config.local_logging_enabled.unwrap_or(false) { diff --git a/ecstore/src/bitrot.rs b/ecstore/src/bitrot.rs index b64c84b3b..05e55a430 100644 --- a/ecstore/src/bitrot.rs +++ b/ecstore/src/bitrot.rs @@ -534,6 +534,7 @@ impl Writer for BitrotFileWriter { self } + #[tracing::instrument(level = "info", skip_all)] async fn write(&mut self, buf: Bytes) -> Result<()> { if buf.is_empty() { return Ok(()); diff --git a/ecstore/src/erasure.rs b/ecstore/src/erasure.rs index 590889e78..942f461bd 100644 --- a/ecstore/src/erasure.rs +++ b/ecstore/src/erasure.rs @@ -1,5 +1,6 @@ use crate::bitrot::{BitrotReader, BitrotWriter}; use crate::error::clone_err; +use crate::io::Etag; use crate::quorum::{object_op_ignored_errs, reduce_write_quorum_errs}; use bytes::{Bytes, BytesMut}; use common::error::{Error, Result}; @@ -8,8 +9,10 @@ use reed_solomon_erasure::galois_8::ReedSolomon; use smallvec::SmallVec; use std::any::Any; use std::io::ErrorKind; +use std::sync::Arc; use tokio::io::{AsyncRead, AsyncWrite}; use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::sync::mpsc; use tracing::warn; use tracing::{error, info}; // use tracing::debug; @@ -26,7 +29,7 @@ pub struct Erasure { encoder: Option, pub block_size: usize, _id: Uuid, - buf: Vec, + _buf: Vec, } impl Erasure { @@ -46,61 +49,62 @@ impl Erasure { block_size, encoder, _id: Uuid::new_v4(), - buf: vec![0u8; block_size], + _buf: vec![0u8; block_size], } } - #[tracing::instrument(level = "debug", skip(self, reader, writers))] + #[tracing::instrument(level = "info", skip(self, reader, writers))] pub async fn encode( - &mut self, - reader: &mut S, + self: Arc, + mut reader: S, writers: &mut [Option], // block_size: usize, total_size: usize, write_quorum: usize, - ) -> Result + ) -> Result<(usize, String)> where - S: AsyncRead + Unpin + Send + 'static, + S: AsyncRead + Etag + Unpin + Send + 'static, { - // pin_mut!(body); - // let mut reader = tokio_util::io::StreamReader::new( - // body.map(|f| f.map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e.to_string()))), - // ); - - let mut total: usize = 0; - let mut blocks = >::new(); - - loop { - if total_size > 0 { - let new_len = { - let remain = total_size - total; - if remain > self.block_size { - self.block_size - } else { - remain - } - }; - - if new_len == 0 && total > 0 { - break; - } - - self.buf.resize(new_len, 0u8); - match reader.read_exact(&mut self.buf).await { - Ok(res) => res, - Err(e) => { - if let ErrorKind::UnexpectedEof = e.kind() { - break; + let (tx, mut rx) = mpsc::channel(5); + let task = tokio::spawn(async move { + let mut buf = Vec::new(); + let mut total: usize = 0; + loop { + if total_size > 0 { + let new_len = { + let remain = total_size - total; + if remain > self.block_size { + self.block_size } else { - return Err(Error::new(e)); + remain } + }; + + if new_len == 0 && total > 0 { + break; } - }; - total += self.buf.len(); + + buf.resize(new_len, 0u8); + match reader.read_exact(&mut buf).await { + Ok(res) => res, + Err(e) => { + if let ErrorKind::UnexpectedEof = e.kind() { + break; + } else { + return Err(Error::new(e)); + } + } + }; + total += buf.len(); + } + let blocks = Arc::new(Box::pin(self.clone().encode_data(&buf)?)); + let _ = tx.send(blocks).await; } + let etag = reader.etag().await; + Ok((total, etag)) + }); - self.encode_data(&self.buf, &mut blocks)?; - + while let Some(blocks) = rx.recv().await { let write_futures = writers.iter_mut().enumerate().map(|(i, w_op)| { let i_inner = i; let blocks_inner = blocks.clone(); @@ -130,8 +134,7 @@ impl Erasure { break; } } - - Ok(total) + task.await? // // let stream = ChunkedStream::new(body, self.block_size); // let stream = ChunkedStream::new(body, total_size, self.block_size, false); @@ -356,8 +359,8 @@ impl Erasure { self.data_shards + self.parity_shards } - #[tracing::instrument(level = "debug", skip_all, fields(data_len=data.len()))] - pub fn encode_data(&self, data: &[u8], shards: &mut SmallVec<[Bytes; 16]>) -> Result<()> { + #[tracing::instrument(level = "info", skip_all, fields(data_len=data.len()))] + pub fn encode_data(self: Arc, data: &[u8]) -> Result> { let (shard_size, total_size) = self.need_size(data.len()); // 生成一个新的 所需的所有分片数据长度 @@ -379,14 +382,13 @@ impl Erasure { // 零拷贝分片，所有 shard 引用 data_buffer let mut data_buffer = data_buffer.freeze(); - shards.clear(); - shards.reserve(self.total_shard_count()); + let mut shards = Vec::with_capacity(self.total_shard_count()); for _ in 0..self.total_shard_count() { let shard = data_buffer.split_to(shard_size); shards.push(shard); } - Ok(()) + Ok(shards) } pub fn decode_data(&self, shards: &mut [Option>]) -> Result<()> { @@ -617,7 +619,6 @@ impl ShardReader { #[cfg(test)] mod test { - use super::*; #[test] @@ -626,8 +627,7 @@ mod test { let parity_shards = 2; let data: &[u8] = &[0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11]; let ec = Erasure::new(data_shards, parity_shards, 1); - let mut shards = SmallVec::new(); - ec.encode_data(data, &mut shards).unwrap(); + let shards = Arc::new(ec).encode_data(data).unwrap(); println!("shards:{:?}", shards); let mut s: Vec<_> = shards @@ -643,6 +643,7 @@ mod test { println!("sss:{:?}", &s); + let ec = Erasure::new(data_shards, parity_shards, 1); ec.decode_data(&mut s).unwrap(); // ec.encoder.reconstruct(&mut s).unwrap(); diff --git a/ecstore/src/io.rs b/ecstore/src/io.rs index f2affe8cb..2bd02c117 100644 --- a/ecstore/src/io.rs +++ b/ecstore/src/io.rs @@ -1,8 +1,12 @@ +use async_trait::async_trait; use bytes::Bytes; use futures::TryStreamExt; use md5::Digest; use md5::Md5; +use pin_project_lite::pin_project; +use std::io; use std::pin::Pin; +use std::task::ready; use std::task::Context; use std::task::Poll; use tokio::io::AsyncRead; @@ -125,10 +129,17 @@ impl AsyncRead for HttpFileReader { } } -pub struct EtagReader { - inner: R, - bytes_tx: mpsc::Sender, - md5_rx: oneshot::Receiver, +#[async_trait] +pub trait Etag { + async fn etag(self) -> String; +} + +pin_project! { + pub struct EtagReader { + inner: R, + bytes_tx: mpsc::Sender, + md5_rx: oneshot::Receiver, + } } impl EtagReader { @@ -148,8 +159,11 @@ impl EtagReader { EtagReader { inner, bytes_tx, md5_rx } } +} - pub async fn etag(self) -> String { +#[async_trait] +impl Etag for EtagReader { + async fn etag(self) -> String { drop(self.inner); drop(self.bytes_tx); self.md5_rx.await.unwrap() @@ -157,21 +171,28 @@ impl EtagReader { } impl AsyncRead for EtagReader { - #[tracing::instrument(level = "debug", skip_all)] - fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll> { - let poll = Pin::new(&mut self.inner).poll_read(cx, buf); - if let Poll::Ready(Ok(())) = &poll { - if buf.remaining() == 0 { + #[tracing::instrument(level = "info", skip_all)] + fn poll_read(self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll> { + let me = self.project(); + + loop { + let rem = buf.remaining(); + if rem != 0 { + ready!(Pin::new(&mut *me.inner).poll_read(cx, buf))?; + if buf.remaining() == rem { + return Err(io::Error::new(io::ErrorKind::UnexpectedEof, "early eof")).into(); + } + } else { let bytes = buf.filled(); let bytes = Bytes::copy_from_slice(bytes); - let tx = self.bytes_tx.clone(); + let tx = me.bytes_tx.clone(); tokio::spawn(async move { if let Err(e) = tx.send(bytes).await { warn!("EtagReader send error: {:?}", e); } }); + return Poll::Ready(Ok(())); } } - poll } } diff --git a/ecstore/src/set_disk.rs b/ecstore/src/set_disk.rs index bd55ab39f..12491b105 100644 --- a/ecstore/src/set_disk.rs +++ b/ecstore/src/set_disk.rs @@ -3759,7 +3759,7 @@ impl ObjectIO for SetDisks { let tmp_object = format!("{}/{}/part.1", tmp_dir, fi.data_dir.unwrap()); - let mut erasure = Erasure::new(fi.erasure.data_blocks, fi.erasure.parity_blocks, fi.erasure.block_size); + let erasure = Erasure::new(fi.erasure.data_blocks, fi.erasure.parity_blocks, fi.erasure.block_size); let is_inline_buffer = { if let Some(sc) = GLOBAL_StorageClass.get() { @@ -3798,19 +3798,18 @@ impl ObjectIO for SetDisks { } let stream = replace(&mut data.stream, Box::new(empty())); - let mut etag_stream = EtagReader::new(stream); + let etag_stream = EtagReader::new(stream); // TODO: etag from header - let w_size = erasure - .encode(&mut etag_stream, &mut writers, data.content_length, write_quorum) + let (w_size, etag) = Arc::new(erasure) + .encode(etag_stream, &mut writers, data.content_length, write_quorum) .await?; // TODO: 出错，删除临时目录 if let Err(err) = close_bitrot_writers(&mut writers).await { error!("close_bitrot_writers err {:?}", err); } - let etag = etag_stream.etag().await; //TODO: userDefined user_defined.insert("etag".to_owned(), etag.clone()); @@ -4408,21 +4407,19 @@ impl StorageAPI for SetDisks { } } - let mut erasure = Erasure::new(fi.erasure.data_blocks, fi.erasure.parity_blocks, fi.erasure.block_size); + let erasure = Erasure::new(fi.erasure.data_blocks, fi.erasure.parity_blocks, fi.erasure.block_size); let stream = replace(&mut data.stream, Box::new(empty())); - let mut etag_stream = EtagReader::new(stream); + let etag_stream = EtagReader::new(stream); - let w_size = erasure - .encode(&mut etag_stream, &mut writers, data.content_length, write_quorum) + let (w_size, mut etag) = Arc::new(erasure) + .encode(etag_stream, &mut writers, data.content_length, write_quorum) .await?; if let Err(err) = close_bitrot_writers(&mut writers).await { error!("close_bitrot_writers err {:?}", err); } - let mut etag = etag_stream.etag().await; - if let Some(ref tag) = opts.preserve_etag { etag = tag.clone(); }