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

fix(ecstore): offload erasure encoding from async workers (#3099)

Browse Source 
* fix(ecstore): offload erasure encoding from async workers

* fix(ecstore): reuse encode buffers in blocking tasks

---------

Co-authored-by: Henry Guo <marshawcoco@users.noreply.github.com>
Co-authored-by: houseme <housemecn@gmail.com>
This commit is contained in:
Henry Guo
2026-05-28 01:54:05 +08:00
committed by GitHub
parent 5dfc1b5f07
commit f8e6fc1f10
2 changed files with 25 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/erasure_coding/encode.rs
+10 -1
View File
@@ -226,7 +226,16 @@ impl Erasure {
match rustfs_utils::read_full(&mut reader, &mut buf).await { match rustfs_utils::read_full(&mut reader, &mut buf).await {
Ok(n) if n > 0 => { Ok(n) if n > 0 => {
total += n; total += n;
let res = self.encode_data(&buf[..n])?; let erasure = self.clone();
let encode_buf = std::mem::take(&mut buf);
let (res, returned_buf) = tokio::task::spawn_blocking(move || {
let res = erasure.encode_data(&encode_buf[..n]);
(res, encode_buf)
})
.await
.map_err(|err| std::io::Error::other(format!("EC encode task failed: {err}")))?;
buf = returned_buf;
let res = res?;
let queued_bytes = queued_block_bytes(&res); let queued_bytes = queued_block_bytes(&res);
rustfs_io_metrics::add_ec_encode_inflight_bytes(queued_bytes); rustfs_io_metrics::add_ec_encode_inflight_bytes(queued_bytes);
if let Err(err) = tx.send(res).await { if let Err(err) = tx.send(res).await {
crates/ecstore/src/erasure_coding/erasure.rs
+15 -1
View File
@@ -587,7 +587,21 @@ impl Erasure {
Ok(n) if n > 0 => { Ok(n) if n > 0 => {
warn!("encode_stream_callback_async read n={}", n); warn!("encode_stream_callback_async read n={}", n);
total += n; total += n;
let res = self.encode_data(&buf[..n]); let erasure = self.clone();
let encode_buf = std::mem::take(&mut buf);
let (res, returned_buf) = match tokio::task::spawn_blocking(move || {
let res = erasure.encode_data(&encode_buf[..n]);
(res, encode_buf)
})
.await
{
Ok(result) => result,
Err(err) => {
on_block(Err(std::io::Error::other(format!("EC encode task failed: {err}")))).await?;
break;
}
};
buf = returned_buf;
on_block(res).await? on_block(res).await?
} }
Ok(_) => { Ok(_) => {