gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-02 11:29:17 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #369 from rustfs/pref

Browse Source 
Pref
This commit is contained in:
guojidan
2025-04-29 11:42:22 +08:00
committed by GitHub
parent 5a363c5523 1dde7015de
commit bd9509e030
7 changed files with 101 additions and 81 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/event-notifier/examples/webhook.rs
+4 -4
View File
@@ -37,7 +37,7 @@ async fn receive_webhook(Json(payload): Json<Value>) -> 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;
}
crates/event-notifier/src/global.rs
+2 -2
View File
@@ -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");
}
crates/obs/src/telemetry.rs
+1 -1
View File
@@ -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) {
ecstore/src/bitrot.rs
+1
View File
@@ -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(());
ecstore/src/erasure.rs
+52 -51
View File
@@ -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<ReedSolomon>,
pub block_size: usize,
_id: Uuid,
buf: Vec<u8>,
_buf: Vec<u8>,
}
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<S>(
&mut self,
reader: &mut S,
self: Arc<Self>,
mut reader: S,
writers: &mut [Option<BitrotWriter>],
// block_size: usize,
total_size: usize,
write_quorum: usize,
) -> Result<usize>
) -> 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 = <SmallVec<[Bytes; 16]>>::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<Self>, data: &[u8]) -> Result<Vec<Bytes>> {
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<Vec<u8>>]) -> 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();
ecstore/src/io.rs
+33 -12
View File
@@ -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<R> {
inner: R,
bytes_tx: mpsc::Sender<Bytes>,
md5_rx: oneshot::Receiver<String>,
#[async_trait]
pub trait Etag {
async fn etag(self) -> String;
}
pin_project! {
pub struct EtagReader<R> {
inner: R,
bytes_tx: mpsc::Sender<Bytes>,
md5_rx: oneshot::Receiver<String>,
}
}
impl<R> EtagReader<R> {
@@ -148,8 +159,11 @@ impl<R> EtagReader<R> {
EtagReader { inner, bytes_tx, md5_rx }
}
}
pub async fn etag(self) -> String {
#[async_trait]
impl<R: Send> Etag for EtagReader<R> {
async fn etag(self) -> String {
drop(self.inner);
drop(self.bytes_tx);
self.md5_rx.await.unwrap()
@@ -157,21 +171,28 @@ impl<R> EtagReader<R> {
}
impl<R: AsyncRead + Unpin> AsyncRead for EtagReader<R> {
#[tracing::instrument(level = "debug", skip_all)]
fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<tokio::io::Result<()>> {
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<tokio::io::Result<()>> {
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
}
}
ecstore/src/set_disk.rs
+8 -11
View File
@@ -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();
}