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

refactor(app): inline object request context types (#2462)

Browse Source
This commit is contained in:
安正超
2026-04-10 10:27:08 +08:00
committed by GitHub
parent 80cb6b3939
commit 9ec6465b26
5 changed files with 84 additions and 153 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rustfs/src/app/object_usecase.rs
+82 -56
View File
@@ -18,11 +18,9 @@ mod get_object_flow;
mod get_object_zero_copy;
mod put_object_extract;
mod put_object_flow;
mod types;
#[cfg(test)]
mod zero_copy_tests;
use self::get_object_flow::GetObjectBootstrap;
use self::types::*;
use crate::app::context::{AppContext, default_notify_interface, get_global_app_context};
use crate::capacity::record_capacity_write;
@@ -141,6 +139,34 @@ struct DeadlockRequestGuard {
request_id: String,
}
#[derive(Clone)]
pub(super) struct GetObjectRequestContext {
pub(super) bucket: String,
pub(super) key: String,
pub(super) part_number: Option<usize>,
pub(super) rs: Option<HTTPRangeSpec>,
pub(super) opts: ObjectOptions,
pub(super) headers: HeaderMap,
pub(super) sse_customer_key: Option<String>,
pub(super) sse_customer_key_md5: Option<String>,
}
pub(super) type PutObjectChecksums = rustfs_object_io::put::PutObjectChecksums;
#[derive(Clone)]
pub(super) struct PutObjectRequestContext {
pub(super) headers: HeaderMap,
pub(super) trailing_headers: Option<s3s::TrailingHeaders>,
pub(super) uri_query: Option<String>,
pub(super) is_post_object: bool,
pub(super) method: hyper::Method,
pub(super) uri: hyper::Uri,
pub(super) extensions: http::Extensions,
pub(super) credentials: Option<s3s::auth::Credentials>,
pub(super) region: Option<s3s::region::Region>,
pub(super) service: Option<String>,
}
impl DeadlockRequestGuard {
fn new(deadlock_detector: Arc<deadlock_detector::DeadlockDetector>, request_id: String) -> Self {
Self {
@@ -315,6 +341,58 @@ fn build_put_object_expiration_header(event: &lifecycle::Event) -> Option<String
Some(format!("expiry-date=\"{}\", rule-id=\"{}\"", expiry_date, event.rule_id))
}
async fn prepare_get_object_request_context(req: &S3Request<GetObjectInput>) -> S3Result<GetObjectRequestContext> {
let GetObjectInput {
bucket,
key,
version_id,
part_number,
range,
..
} = req.input.clone();
validate_object_key(&key, "GET")?;
let part_number = part_number.map(|value| value as usize);
if let Some(part_number) = part_number
&& part_number == 0
{
return Err(s3_error!(InvalidArgument, "Invalid part number: part number must be greater than 0"));
}
let rs = range.map(|value| match value {
Range::Int { first, last } => HTTPRangeSpec {
is_suffix_length: false,
start: first as i64,
end: last.map_or(-1, |last| last as i64),
},
Range::Suffix { length } => HTTPRangeSpec {
is_suffix_length: true,
start: length as i64,
end: -1,
},
});
if rs.is_some() && part_number.is_some() {
return Err(s3_error!(InvalidArgument, "range and part_number invalid"));
}
let opts: ObjectOptions = get_opts(&bucket, &key, version_id, part_number, &req.headers)
.await
.map_err(ApiError::from)?;
Ok(GetObjectRequestContext {
bucket,
key,
part_number,
rs,
opts,
headers: req.headers.clone(),
sse_customer_key: req.input.sse_customer_key.clone(),
sse_customer_key_md5: req.input.sse_customer_key_md5.clone(),
})
}
#[allow(clippy::too_many_arguments)]
fn apply_put_request_metadata(
metadata: &mut HashMap<String, String>,
@@ -1065,60 +1143,8 @@ impl DefaultObjectUsecase {
_deadlock_request_guard: deadlock_request_guard,
}
};
let (request_context, version_id_for_event) = {
let GetObjectInput {
bucket,
key,
version_id,
part_number,
range,
..
} = req.input.clone();
validate_object_key(&key, "GET")?;
let part_number = part_number.map(|value| value as usize);
if let Some(part_number) = part_number
&& part_number == 0
{
return Err(s3_error!(InvalidArgument, "Invalid part number: part number must be greater than 0"));
}
let rs = range.map(|value| match value {
Range::Int { first, last } => HTTPRangeSpec {
is_suffix_length: false,
start: first as i64,
end: last.map_or(-1, |last| last as i64),
},
Range::Suffix { length } => HTTPRangeSpec {
is_suffix_length: true,
start: length as i64,
end: -1,
},
});
if rs.is_some() && part_number.is_some() {
return Err(s3_error!(InvalidArgument, "range and part_number invalid"));
}
let opts: ObjectOptions = get_opts(&bucket, &key, version_id.clone(), part_number, &req.headers)
.await
.map_err(ApiError::from)?;
(
GetObjectRequestContext {
bucket,
key,
part_number,
rs,
opts,
headers: req.headers.clone(),
sse_customer_key: req.input.sse_customer_key.clone(),
sse_customer_key_md5: req.input.sse_customer_key_md5.clone(),
},
version_id.unwrap_or_default(),
)
};
let version_id_for_event = req.input.version_id.clone().unwrap_or_default();
let request_context = prepare_get_object_request_context(&req).await?;
let base_buffer_size = self.base_buffer_size();
let manager = get_concurrency_manager();
let cors_bucket = request_context.bucket.clone();
rustfs/src/app/object_usecase/get_object_flow.rs
+1 -1
View File
@@ -13,8 +13,8 @@
// limitations under the License.
use super::DeadlockRequestGuard;
use super::GetObjectRequestContext;
use super::get_object_zero_copy::{GetObjectIoPlanning, GetObjectPreparedRead, prepare_get_object_read_execution};
use super::types::GetObjectRequestContext;
use crate::error::ApiError;
use crate::storage::concurrency::{ConcurrencyManager, GetObjectGuard, get_buffer_size_opt_in};
use crate::storage::options::filter_object_metadata;
rustfs/src/app/object_usecase/get_object_zero_copy.rs
+1 -1
View File
@@ -12,7 +12,7 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use super::types::GetObjectRequestContext;
use super::GetObjectRequestContext;
use crate::error::ApiError;
use crate::storage::concurrency::{self, ConcurrencyManager};
use crate::storage::timeout_wrapper::{RequestTimeoutWrapper, TimeoutConfig};
rustfs/src/app/object_usecase/types.rs
-43
View File
@@ -1,43 +0,0 @@
// Copyright 2024 RustFS Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
use super::*;
#[derive(Clone)]
pub(super) struct GetObjectRequestContext {
pub(super) bucket: String,
pub(super) key: String,
pub(super) part_number: Option<usize>,
pub(super) rs: Option<HTTPRangeSpec>,
pub(super) opts: ObjectOptions,
pub(super) headers: HeaderMap,
pub(super) sse_customer_key: Option<String>,
pub(super) sse_customer_key_md5: Option<String>,
}
pub(super) type PutObjectChecksums = rustfs_object_io::put::PutObjectChecksums;
#[derive(Clone)]
pub(super) struct PutObjectRequestContext {
pub(super) headers: HeaderMap,
pub(super) trailing_headers: Option<s3s::TrailingHeaders>,
pub(super) uri_query: Option<String>,
pub(super) is_post_object: bool,
pub(super) method: hyper::Method,
pub(super) uri: hyper::Uri,
pub(super) extensions: http::Extensions,
pub(super) credentials: Option<s3s::auth::Credentials>,
pub(super) region: Option<s3s::region::Region>,
pub(super) service: Option<String>,
}
rustfs/src/app/object_usecase/zero_copy_tests.rs
-52
View File
@@ -40,58 +40,6 @@ static DIRECT_CHUNK_TEST_ENV: OnceLock<(Vec<PathBuf>, Arc<ECStore>)> = OnceLock:
static DIRECT_CHUNK_MULTI_DISK_TEST_ENV: OnceLock<(Vec<PathBuf>, Arc<ECStore>)> = OnceLock::new();
static DIRECT_CHUNK_TEST_INIT: Once = Once::new();
async fn prepare_get_object_request_context(req: &S3Request<GetObjectInput>) -> S3Result<GetObjectRequestContext> {
let GetObjectInput {
bucket,
key,
version_id,
part_number,
range,
..
} = req.input.clone();
validate_object_key(&key, "GET")?;
let part_number = part_number.map(|value| value as usize);
if let Some(part_number) = part_number
&& part_number == 0
{
return Err(s3_error!(InvalidArgument, "Invalid part number: part number must be greater than 0"));
}
let rs = range.map(|value| match value {
Range::Int { first, last } => HTTPRangeSpec {
is_suffix_length: false,
start: first as i64,
end: last.map_or(-1, |last| last as i64),
},
Range::Suffix { length } => HTTPRangeSpec {
is_suffix_length: true,
start: length as i64,
end: -1,
},
});
if rs.is_some() && part_number.is_some() {
return Err(s3_error!(InvalidArgument, "range and part_number invalid"));
}
let opts: ObjectOptions = get_opts(&bucket, &key, version_id.clone(), part_number, &req.headers)
.await
.map_err(ApiError::from)?;
Ok(GetObjectRequestContext {
bucket,
key,
part_number,
rs,
opts,
headers: req.headers.clone(),
sse_customer_key: req.input.sse_customer_key.clone(),
sse_customer_key_md5: req.input.sse_customer_key_md5.clone(),
})
}
fn init_direct_chunk_test_tracing() {
DIRECT_CHUNK_TEST_INIT.call_once(|| {});
}