From c1590a054c2d2c3630014c442704de82fb021f27 Mon Sep 17 00:00:00 2001 From: junxiang Mu <1948535941@qq.com> Date: Mon, 28 Apr 2025 10:00:28 +0000 Subject: [PATCH 1/4] tmp Signed-off-by: junxiang Mu <1948535941@qq.com> --- ecstore/src/bitrot.rs | 1 + ecstore/src/erasure.rs | 126 +++++++++++++++++++++------------------- ecstore/src/io.rs | 54 +++++++++++++---- ecstore/src/set_disk.rs | 4 +- 4 files changed, 113 insertions(+), 72 deletions(-) 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..d5a08af4b 100644 --- a/ecstore/src/erasure.rs +++ b/ecstore/src/erasure.rs @@ -6,8 +6,10 @@ use common::error::{Error, Result}; use futures::future::join_all; use reed_solomon_erasure::galois_8::ReedSolomon; use smallvec::SmallVec; +use tokio::sync::mpsc; use std::any::Any; use std::io::ErrorKind; +use std::sync::Arc; use tokio::io::{AsyncRead, AsyncWrite}; use tokio::io::{AsyncReadExt, AsyncWriteExt}; use tracing::warn; @@ -52,8 +54,8 @@ impl Erasure { #[tracing::instrument(level = "debug", 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, @@ -67,40 +69,47 @@ impl Erasure { // 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(3); + let self_clone = self.clone(); + let task = tokio::spawn(async move { + let mut total: usize = 0; + let mut buf = Vec::new(); + loop { + let mut blocks = >::new(); + if total_size > 0 { + let new_len = { + let remain = total_size - total; + if remain > self_clone.block_size { + self_clone.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(); + } + self_clone.clone().encode_data(&buf, &mut blocks)?; + let _ = tx.send(blocks).await; } - - self.encode_data(&self.buf, &mut blocks)?; - + Ok(total) + }); + + 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 +139,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 +364,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], shards: &mut SmallVec<[Bytes; 16]>) -> Result<()> { let (shard_size, total_size) = self.need_size(data.len()); // 生成一个新的 所需的所有分片数据长度 @@ -618,34 +626,34 @@ impl ShardReader { #[cfg(test)] mod test { - use super::*; + // use super::*; - #[test] - fn test_erasure() { - let data_shards = 3; - 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(); - println!("shards:{:?}", shards); + // #[test] + // fn test_erasure() { + // let data_shards = 3; + // 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(); + // Arc::new(ec).encode_data(data, &mut shards).unwrap(); + // println!("shards:{:?}", shards); - let mut s: Vec<_> = shards - .iter() - .map(|d| if d.is_empty() { None } else { Some(d.to_vec()) }) - .collect(); + // let mut s: Vec<_> = shards + // .iter() + // .map(|d| if d.is_empty() { None } else { Some(d.to_vec()) }) + // .collect(); - // let mut s = shards_to_option_shards(&shards); + // // let mut s = shards_to_option_shards(&shards); - // s[0] = None; - s[4] = None; - s[3] = None; + // // s[0] = None; + // s[4] = None; + // s[3] = None; - println!("sss:{:?}", &s); + // println!("sss:{:?}", &s); - ec.decode_data(&mut s).unwrap(); - // ec.encoder.reconstruct(&mut s).unwrap(); + // ec.decode_data(&mut s).unwrap(); + // // ec.encoder.reconstruct(&mut s).unwrap(); - println!("sss:{:?}", &s); - } + // println!("sss:{:?}", &s); + // } } diff --git a/ecstore/src/io.rs b/ecstore/src/io.rs index f2affe8cb..6bdbd629b 100644 --- a/ecstore/src/io.rs +++ b/ecstore/src/io.rs @@ -2,7 +2,10 @@ 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,12 +128,19 @@ impl AsyncRead for HttpFileReader { } } -pub struct EtagReader { - inner: R, - bytes_tx: mpsc::Sender, - md5_rx: oneshot::Receiver, +pub trait { + } +pin_project! { + pub struct EtagReader { + inner: R, + bytes_tx: mpsc::Sender, + md5_rx: oneshot::Receiver, + } +} + + impl EtagReader { pub fn new(inner: R) -> Self { let (bytes_tx, mut bytes_rx) = mpsc::channel::(8); @@ -157,21 +167,43 @@ 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 + + // let poll = Pin::new(&mut self.inner).poll_read(cx, buf); + // if let Poll::Ready(Ok(())) = &poll { + // if buf.remaining() == 0 { + // let bytes = buf.filled(); + // let bytes = Bytes::copy_from_slice(bytes); + // let tx = self.bytes_tx.clone(); + // tokio::spawn(async move { + // if let Err(e) = tx.send(bytes).await { + // warn!("EtagReader send error: {:?}", e); + // } + // }); + // } + // } + // poll } } diff --git a/ecstore/src/set_disk.rs b/ecstore/src/set_disk.rs index bd55ab39f..3c2b033c8 100644 --- a/ecstore/src/set_disk.rs +++ b/ecstore/src/set_disk.rs @@ -3802,8 +3802,8 @@ impl ObjectIO for SetDisks { // TODO: etag from header - let w_size = erasure - .encode(&mut etag_stream, &mut writers, data.content_length, write_quorum) + let w_size = 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 { From 66f5bf1bbcfbc6ab65ee3ceb9d122a3e37641f79 Mon Sep 17 00:00:00 2001 From: junxiang Mu <1948535941@qq.com> Date: Mon, 28 Apr 2025 21:55:31 +0800 Subject: [PATCH 2/4] tmp1 Signed-off-by: junxiang Mu <1948535941@qq.com> --- ecstore/src/erasure.rs | 21 ++++++++++++--------- ecstore/src/io.rs | 12 ++++++++---- ecstore/src/set_disk.rs | 17 +++++++---------- 3 files changed, 27 insertions(+), 23 deletions(-) diff --git a/ecstore/src/erasure.rs b/ecstore/src/erasure.rs index d5a08af4b..56f5959c5 100644 --- a/ecstore/src/erasure.rs +++ b/ecstore/src/erasure.rs @@ -1,17 +1,18 @@ 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}; use futures::future::join_all; use reed_solomon_erasure::galois_8::ReedSolomon; use smallvec::SmallVec; -use tokio::sync::mpsc; 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; @@ -28,7 +29,7 @@ pub struct Erasure { encoder: Option, pub block_size: usize, _id: Uuid, - buf: Vec, + _buf: Vec, } impl Erasure { @@ -48,7 +49,7 @@ impl Erasure { block_size, encoder, _id: Uuid::new_v4(), - buf: vec![0u8; block_size], + _buf: vec![0u8; block_size], } } @@ -60,9 +61,9 @@ impl Erasure { // 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( @@ -85,11 +86,11 @@ impl Erasure { remain } }; - + if new_len == 0 && total > 0 { break; } - + buf.resize(new_len, 0u8); match reader.read_exact(&mut buf).await { Ok(res) => res, @@ -106,9 +107,11 @@ impl Erasure { self_clone.clone().encode_data(&buf, &mut blocks)?; let _ = tx.send(blocks).await; } - Ok(total) + // let etag = reader.etag().await; + let etag = String::new(); + Ok((total, etag)) }); - + while let Some(blocks) = rx.recv().await { let write_futures = writers.iter_mut().enumerate().map(|(i, w_op)| { let i_inner = i; diff --git a/ecstore/src/io.rs b/ecstore/src/io.rs index 6bdbd629b..056260fdf 100644 --- a/ecstore/src/io.rs +++ b/ecstore/src/io.rs @@ -1,3 +1,4 @@ +use async_trait::async_trait; use bytes::Bytes; use futures::TryStreamExt; use md5::Digest; @@ -128,8 +129,9 @@ impl AsyncRead for HttpFileReader { } } -pub trait { - +#[async_trait] +pub trait Etag { + async fn etag(self) -> String; } pin_project! { @@ -140,7 +142,6 @@ pin_project! { } } - impl EtagReader { pub fn new(inner: R) -> Self { let (bytes_tx, mut bytes_rx) = mpsc::channel::(8); @@ -158,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() diff --git a/ecstore/src/set_disk.rs b/ecstore/src/set_disk.rs index 3c2b033c8..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,11 +3798,11 @@ 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 = Arc::new(erasure) + let (w_size, etag) = Arc::new(erasure) .encode(etag_stream, &mut writers, data.content_length, write_quorum) .await?; // TODO: 出错，删除临时目录 @@ -3810,7 +3810,6 @@ impl ObjectIO for SetDisks { 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(); } From e346d202282454a854bed91adb89d4453576c949 Mon Sep 17 00:00:00 2001 From: junxiang Mu <1948535941@qq.com> Date: Tue, 29 Apr 2025 02:53:42 +0000 Subject: [PATCH 3/4] tmp1 Signed-off-by: junxiang Mu <1948535941@qq.com> --- crates/event-notifier/examples/webhook.rs | 8 +++---- crates/event-notifier/src/global.rs | 4 ++-- crates/obs/src/telemetry.rs | 2 +- ecstore/src/erasure.rs | 29 ++++++++--------------- 4 files changed, 17 insertions(+), 26 deletions(-) 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/erasure.rs b/ecstore/src/erasure.rs index 56f5959c5..247cadae3 100644 --- a/ecstore/src/erasure.rs +++ b/ecstore/src/erasure.rs @@ -53,7 +53,7 @@ impl Erasure { } } - #[tracing::instrument(level = "debug", skip(self, reader, writers))] + #[tracing::instrument(level = "info", skip(self, reader, writers))] pub async fn encode( self: Arc, mut reader: S, @@ -65,23 +65,16 @@ impl Erasure { where 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 (tx, mut rx) = mpsc::channel(3); - let self_clone = self.clone(); + let (tx, mut rx) = mpsc::channel(5); let task = tokio::spawn(async move { - let mut total: usize = 0; let mut buf = Vec::new(); + let mut total: usize = 0; loop { - let mut blocks = >::new(); if total_size > 0 { let new_len = { let remain = total_size - total; - if remain > self_clone.block_size { - self_clone.block_size + if remain > self.block_size { + self.block_size } else { remain } @@ -104,11 +97,10 @@ impl Erasure { }; total += buf.len(); } - self_clone.clone().encode_data(&buf, &mut blocks)?; + let blocks = Arc::new(Box::pin(self.clone().encode_data(&buf)?)); let _ = tx.send(blocks).await; } - // let etag = reader.etag().await; - let etag = String::new(); + let etag = reader.etag().await; Ok((total, etag)) }); @@ -368,7 +360,7 @@ impl Erasure { } #[tracing::instrument(level = "info", skip_all, fields(data_len=data.len()))] - pub fn encode_data(self: Arc, data: &[u8], shards: &mut SmallVec<[Bytes; 16]>) -> Result<()> { + pub fn encode_data(self: Arc, data: &[u8]) -> Result> { let (shard_size, total_size) = self.need_size(data.len()); // 生成一个新的 所需的所有分片数据长度 @@ -390,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<()> { From 1dde7015deead33fe931179a261661885ccc1552 Mon Sep 17 00:00:00 2001 From: junxiang Mu <1948535941@qq.com> Date: Tue, 29 Apr 2025 03:41:45 +0000 Subject: [PATCH 4/4] tmp2 Signed-off-by: junxiang Mu <1948535941@qq.com> --- ecstore/src/erasure.rs | 47 +++++++++++++++++++++--------------------- ecstore/src/io.rs | 15 -------------- 2 files changed, 23 insertions(+), 39 deletions(-) diff --git a/ecstore/src/erasure.rs b/ecstore/src/erasure.rs index 247cadae3..942f461bd 100644 --- a/ecstore/src/erasure.rs +++ b/ecstore/src/erasure.rs @@ -619,35 +619,34 @@ impl ShardReader { #[cfg(test)] mod test { + use super::*; - // use super::*; + #[test] + fn test_erasure() { + let data_shards = 3; + 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 shards = Arc::new(ec).encode_data(data).unwrap(); + println!("shards:{:?}", shards); - // #[test] - // fn test_erasure() { - // let data_shards = 3; - // 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(); - // Arc::new(ec).encode_data(data, &mut shards).unwrap(); - // println!("shards:{:?}", shards); + let mut s: Vec<_> = shards + .iter() + .map(|d| if d.is_empty() { None } else { Some(d.to_vec()) }) + .collect(); - // let mut s: Vec<_> = shards - // .iter() - // .map(|d| if d.is_empty() { None } else { Some(d.to_vec()) }) - // .collect(); + // let mut s = shards_to_option_shards(&shards); - // // let mut s = shards_to_option_shards(&shards); + // s[0] = None; + s[4] = None; + s[3] = None; - // // s[0] = None; - // s[4] = None; - // s[3] = None; + println!("sss:{:?}", &s); - // println!("sss:{:?}", &s); + let ec = Erasure::new(data_shards, parity_shards, 1); + ec.decode_data(&mut s).unwrap(); + // ec.encoder.reconstruct(&mut s).unwrap(); - // ec.decode_data(&mut s).unwrap(); - // // ec.encoder.reconstruct(&mut s).unwrap(); - - // println!("sss:{:?}", &s); - // } + println!("sss:{:?}", &s); + } } diff --git a/ecstore/src/io.rs b/ecstore/src/io.rs index 056260fdf..2bd02c117 100644 --- a/ecstore/src/io.rs +++ b/ecstore/src/io.rs @@ -194,20 +194,5 @@ impl AsyncRead for EtagReader { return Poll::Ready(Ok(())); } } - - // let poll = Pin::new(&mut self.inner).poll_read(cx, buf); - // if let Poll::Ready(Ok(())) = &poll { - // if buf.remaining() == 0 { - // let bytes = buf.filled(); - // let bytes = Bytes::copy_from_slice(bytes); - // let tx = self.bytes_tx.clone(); - // tokio::spawn(async move { - // if let Err(e) = tx.send(bytes).await { - // warn!("EtagReader send error: {:?}", e); - // } - // }); - // } - // } - // poll } }