You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@opendal.apache.org by xu...@apache.org on 2025/03/20 15:27:17 UTC
(opendal) branch refactor-timeout-layer created (now c8751f38a)
This is an automated email from the ASF dual-hosted git repository.
xuanwo pushed a change to branch refactor-timeout-layer
in repository https://gitbox.apache.org/repos/asf/opendal.git
at c8751f38a refactor(core)!: Migrate to context based executor
This branch includes the following new commits:
new c8751f38a refactor(core)!: Migrate to context based executor
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(opendal) 01/01: refactor(core)!: Migrate to context based executor
Posted by xu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xuanwo pushed a commit to branch refactor-timeout-layer
in repository https://gitbox.apache.org/repos/asf/opendal.git
commit c8751f38ad285197322961274d1a81adabb05ac5
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Thu Mar 20 23:26:35 2025 +0800
refactor(core)!: Migrate to context based executor
Signed-off-by: Xuanwo <gi...@xuanwo.io>
---
core/src/layers/timeout.rs | 23 ++++---------
core/src/raw/ops.rs | 53 -----------------------------
core/src/types/operator/operator.rs | 21 +++---------
core/src/types/operator/operator_futures.rs | 15 --------
core/src/types/read/buffer_stream.rs | 2 +-
5 files changed, 12 insertions(+), 102 deletions(-)
diff --git a/core/src/layers/timeout.rs b/core/src/layers/timeout.rs
index 0a1dfaac8..e322671db 100644
--- a/core/src/layers/timeout.rs
+++ b/core/src/layers/timeout.rs
@@ -169,6 +169,11 @@ impl<A: Access> Layer<A> for TimeoutLayer {
type LayeredAccess = TimeoutAccessor<A>;
fn layer(&self, inner: A) -> Self::LayeredAccess {
+ let info = inner.info();
+ info.update_executor(|exec| {
+ Executor::with(TimeoutExecutor::new(exec.into_inner(), self.io_timeout))
+ });
+
TimeoutAccessor {
inner,
@@ -232,27 +237,13 @@ impl<A: Access> LayeredAccess for TimeoutAccessor<A> {
.await
}
- async fn read(&self, path: &str, mut args: OpRead) -> Result<(RpRead, Self::Reader)> {
- if let Some(exec) = args.executor().cloned() {
- args = args.with_executor(Executor::with(TimeoutExecutor::new(
- exec.into_inner(),
- self.io_timeout,
- )));
- }
-
+ async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
self.io_timeout(Operation::ReaderStart, self.inner.read(path, args))
.await
.map(|(rp, r)| (rp, TimeoutWrapper::new(r, self.io_timeout)))
}
- async fn write(&self, path: &str, mut args: OpWrite) -> Result<(RpWrite, Self::Writer)> {
- if let Some(exec) = args.executor().cloned() {
- args = args.with_executor(Executor::with(TimeoutExecutor::new(
- exec.into_inner(),
- self.io_timeout,
- )));
- }
-
+ async fn write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::Writer)> {
self.io_timeout(Operation::WriterStart, self.inner.write(path, args))
.await
.map(|(rp, r)| (rp, TimeoutWrapper::new(r, self.io_timeout)))
diff --git a/core/src/raw/ops.rs b/core/src/raw/ops.rs
index b76dbaff1..4c07d9e92 100644
--- a/core/src/raw/ops.rs
+++ b/core/src/raw/ops.rs
@@ -20,7 +20,6 @@
//! By using ops, users can add more context for operation.
use crate::raw::*;
-use crate::*;
use chrono::{DateTime, Utc};
use std::collections::HashMap;
use std::time::Duration;
@@ -313,7 +312,6 @@ pub struct OpRead {
override_cache_control: Option<String>,
override_content_disposition: Option<String>,
version: Option<String>,
- executor: Option<Executor>,
}
impl OpRead {
@@ -426,31 +424,6 @@ impl OpRead {
pub fn version(&self) -> Option<&str> {
self.version.as_deref()
}
-
- /// Set the executor of the option
- pub fn with_executor(mut self, executor: Executor) -> Self {
- self.executor = Some(executor);
- self
- }
-
- /// Merge given executor into option.
- ///
- /// If executor has already been set, this will do nothing.
- /// Otherwise, this will set the given executor.
- pub(crate) fn merge_executor(self, executor: Option<Executor>) -> Self {
- if self.executor.is_some() {
- return self;
- }
- if let Some(exec) = executor {
- return self.with_executor(exec);
- }
- self
- }
-
- /// Get executor from option
- pub fn executor(&self) -> Option<&Executor> {
- self.executor.as_ref()
- }
}
/// Args for reader operation.
@@ -632,7 +605,6 @@ pub struct OpWrite {
content_disposition: Option<String>,
content_encoding: Option<String>,
cache_control: Option<String>,
- executor: Option<Executor>,
if_match: Option<String>,
if_none_match: Option<String>,
if_not_exists: bool,
@@ -721,17 +693,6 @@ impl OpWrite {
self
}
- /// Get the executor from option
- pub fn executor(&self) -> Option<&Executor> {
- self.executor.as_ref()
- }
-
- /// Set the executor of the option
- pub fn with_executor(mut self, executor: Executor) -> Self {
- self.executor = Some(executor);
- self
- }
-
/// Set the If-Match of the option
pub fn with_if_match(mut self, s: &str) -> Self {
self.if_match = Some(s.to_string());
@@ -765,20 +726,6 @@ impl OpWrite {
self.if_not_exists
}
- /// Merge given executor into option.
- ///
- /// If executor has already been set, this will do nothing.
- /// Otherwise, this will set the given executor.
- pub(crate) fn merge_executor(self, executor: Option<Executor>) -> Self {
- if self.executor.is_some() {
- return self;
- }
- if let Some(exec) = executor {
- return self.with_executor(exec);
- }
- self
- }
-
/// Set the user defined metadata of the op
pub fn with_user_metadata(mut self, metadata: HashMap<String, String>) -> Self {
self.user_metadata = Some(metadata);
diff --git a/core/src/types/operator/operator.rs b/core/src/types/operator/operator.rs
index c79a9f6fa..3efcb2e9a 100644
--- a/core/src/types/operator/operator.rs
+++ b/core/src/types/operator/operator.rs
@@ -586,10 +586,7 @@ impl Operator {
OperatorFuture::new(
self.inner().clone(),
path,
- (
- OpRead::default().merge_executor(self.default_executor.clone()),
- OpReader::default(),
- ),
+ (OpRead::default(), OpReader::default()),
|inner, path, (args, options)| async move {
if !validate_path(&path, EntryMode::FILE) {
return Err(
@@ -673,10 +670,7 @@ impl Operator {
OperatorFuture::new(
self.inner().clone(),
path,
- (
- OpRead::default().merge_executor(self.default_executor.clone()),
- OpReader::default(),
- ),
+ (OpRead::default(), OpReader::default()),
|inner, path, (args, options)| async move {
if !validate_path(&path, EntryMode::FILE) {
return Err(
@@ -937,10 +931,7 @@ impl Operator {
OperatorFuture::new(
self.inner().clone(),
path,
- (
- OpWrite::default().merge_executor(self.default_executor.clone()),
- OpWriter::default(),
- ),
+ (OpWrite::default(), OpWriter::default()),
|inner, path, (args, options)| async move {
if !validate_path(&path, EntryMode::FILE) {
return Err(
@@ -1014,11 +1005,7 @@ impl Operator {
OperatorFuture::new(
self.inner().clone(),
path,
- (
- OpWrite::default().merge_executor(self.default_executor.clone()),
- OpWriter::default(),
- bs,
- ),
+ (OpWrite::default(), OpWriter::default(), bs),
|inner, path, (args, options, bs)| async move {
if !validate_path(&path, EntryMode::FILE) {
return Err(
diff --git a/core/src/types/operator/operator_futures.rs b/core/src/types/operator/operator_futures.rs
index c70481720..72ef6bd62 100644
--- a/core/src/types/operator/operator_futures.rs
+++ b/core/src/types/operator/operator_futures.rs
@@ -224,11 +224,6 @@ impl<F: Future<Output = Result<PresignedRequest>>> FuturePresignWrite<F> {
pub type FutureRead<F> = OperatorFuture<(OpRead, OpReader), Buffer, F>;
impl<F: Future<Output = Result<Buffer>>> FutureRead<F> {
- /// Set the executor for this operation.
- pub fn executor(self, executor: Executor) -> Self {
- self.map(|(args, op_reader)| (args.with_executor(executor), op_reader))
- }
-
/// Set `range` for this `read` request.
///
/// If we have a file with size `n`.
@@ -580,11 +575,6 @@ impl<F: Future<Output = Result<Reader>>> FutureReader<F> {
pub type FutureWrite<F> = OperatorFuture<(OpWrite, OpWriter, Buffer), Metadata, F>;
impl<F: Future<Output = Result<Metadata>>> FutureWrite<F> {
- /// Set the executor for this operation.
- pub fn executor(self, executor: Executor) -> Self {
- self.map(|(args, options, bs)| (args.with_executor(executor), options, bs))
- }
-
/// Sets append mode for this write request.
///
/// ### Capability
@@ -1049,11 +1039,6 @@ impl<F: Future<Output = Result<Metadata>>> FutureWrite<F> {
pub type FutureWriter<F> = OperatorFuture<(OpWrite, OpWriter), Writer, F>;
impl<F: Future<Output = Result<Writer>>> FutureWriter<F> {
- /// Set the executor for this operation.
- pub fn executor(self, executor: Executor) -> Self {
- self.map(|(args, options)| (args.with_executor(executor), options))
- }
-
/// Sets append mode for this write request.
///
/// ### Capability
diff --git a/core/src/types/read/buffer_stream.rs b/core/src/types/read/buffer_stream.rs
index b9bde6f9c..5073426bb 100644
--- a/core/src/types/read/buffer_stream.rs
+++ b/core/src/types/read/buffer_stream.rs
@@ -88,7 +88,7 @@ impl ChunkedReader {
/// We don't need to handle `Executor::timeout` since we are outside the layer.
fn new(ctx: Arc<ReadContext>, range: BytesRange) -> Self {
let tasks = ConcurrentTasks::new(
- ctx.args().executor().cloned().unwrap_or_default(),
+ ctx.accessor().info().executor(),
ctx.options().concurrent(),
|mut r: oio::Reader| {
Box::pin(async {