You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@opendal.apache.org by "PsiACE (via GitHub)" <gi...@apache.org> on 2025/10/02 05:54:41 UTC
[PR] 壮举：与operation_semaphore共享semaphore [opendal]
PsiACE opened a new pull request, #6618:
URL: https://github.com/apache/opendal/pull/6618
# Which issue does this PR close?
<!--
We generally require a GitHub issue to be filed for all bug fixes and enhancements. You can link an issue to this PR using the GitHub syntax. For example `Closes #123` indicates that this PR will close issue #123.
-->
Closes #6591 . also cc @bonsairobo for review
# Rationale for this change
<!--
Why are you proposing this change? If this is already explained clearly in the issue then this section is not needed.
Explaining clearly why changes are proposed helps reviewers understand your changes and offer better suggestions for fixes.
-->
Expose the internal semaphore used by `ConcurrentLimitLayer` so external components can coordinate concurrency
with OpenDAL without reinventing their own limiters.
# What changes are included in this PR?
<!--
There is no need to duplicate the description in the issue here but it is sometimes worth providing a summary of the individual changes in this PR.
-->
- Added operation_semaphore() to return the layer’s shared Arc<Semaphore>.
- Expanded the method’s docs to warn about modifying permits.
- Added a unit test demonstrating that holding the shared semaphore blocks layer operations until released.
# Are there any user-facing changes?
Yes. Applications can now reuse ConcurrentLimitLayer’s semaphore directly to align their own concurrency
controls with OpenDAL.
<!--
If there are user-facing changes then we may require documentation to be updated before approving the PR.
-->
<!---
If there are any breaking changes to public APIs, please add the `breaking-changes` label.
-->
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: make ConcurrentLimitLayer accept Semaphore [opendal]
Posted by "bonsairobo (via GitHub)" <gi...@apache.org>.
bonsairobo commented on PR #6618:
URL: https://github.com/apache/opendal/pull/6618#issuecomment-3367012988
@tisonkun I don't
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: make ConcurrentLimitLayer accept Semaphore [opendal]
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on code in PR #6618:
URL: https://github.com/apache/opendal/pull/6618#discussion_r2451257495
##########
core/src/layers/concurrent_limit.rs:
##########
@@ -90,24 +90,61 @@ use crate::*;
/// Ok(())
/// # }
/// ```
+/// ConcurrencySemaphore abstracts a semaphore-like concurrency primitive
+/// that yields an owned permit released on drop. It mirrors RetryLayer's
+/// interceptor pattern by serving as a generic extension point for the layer.
+pub trait ConcurrencySemaphore: Send + Sync + Any + Clone + 'static {
+ /// The owned permit type associated with the semaphore. Dropping it
+ /// must release the permit back to the semaphore.
+ type Permit: Send + Sync + 'static;
+
+ /// Acquire an owned permit asynchronously.
+ fn acquire_owned(&self) -> BoxFuture<'static, Self::Permit>;
Review Comment:
Maybe we can just use `async fn acquire_owned(&self)` here? And since we don't have other APIs, is it a good idea to just use `acquire`?
##########
core/src/layers/concurrent_limit.rs:
##########
@@ -90,24 +90,61 @@ use crate::*;
/// Ok(())
/// # }
/// ```
+/// ConcurrencySemaphore abstracts a semaphore-like concurrency primitive
Review Comment:
Hi, please review your code before submitting.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: make ConcurrentLimitLayer accept Semaphore [opendal]
Posted by "tisonkun (via GitHub)" <gi...@apache.org>.
tisonkun commented on PR #6618:
URL: https://github.com/apache/opendal/pull/6618#issuecomment-3380248044
@bonsairobo Thanks for your clarification. Could you elaborate a bit how the semaphore share between different resources and how it benefits your use case?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: make ConcurrentLimitLayer accept Semaphore [opendal]
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on PR #6618:
URL: https://github.com/apache/opendal/pull/6618#issuecomment-3414319906
I'm thinking of provide a trait like we do for other layers like `RetryLayer`. We can allow users to pass any `Semaphore` they want in the given layer. Do you want to work on this way? @PsiACE
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: make ConcurrentLimitLayer accept Semaphore [opendal]
Posted by "bonsairobo (via GitHub)" <gi...@apache.org>.
bonsairobo commented on PR #6618:
URL: https://github.com/apache/opendal/pull/6618#issuecomment-3380229506
@tisonkun You asked if I have a *public* example. I don't, but my company has a private repository that would make use of this feature.
> it does no benefits to tight our interfaces with a certain sync utils impls.
That's not true. While I agree that there is tight coupling, that doesn't imply that no one uses this type of semaphore.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: make ConcurrentLimitLayer accept Semaphore [opendal]
Posted by "bonsairobo (via GitHub)" <gi...@apache.org>.
bonsairobo commented on PR #6618:
URL: https://github.com/apache/opendal/pull/6618#issuecomment-3382469400
> Could you elaborate a bit how the semaphore share between different resources and how it benefits your use case?
@tisonkun OpenDAL is not the only code in our system that opens files. Processes have an open file limit, which is especially low on MacOS (256). We use a global semaphore to limit open file handles. It only works if all code shares that semaphore, so we need to create it ourselves and feed it into the concurrent limit layer.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: make ConcurrentLimitLayer accept Semaphore [opendal]
Posted by "Xuanwo (via GitHub)" <gi...@apache.org>.
Xuanwo commented on code in PR #6618:
URL: https://github.com/apache/opendal/pull/6618#discussion_r2567276751
##########
bindings/java/src/layer.rs:
##########
@@ -59,6 +59,6 @@ pub extern "system" fn Java_org_apache_opendal_layer_ConcurrentLimitLayer_doLaye
permits: jlong,
) -> jlong {
let op = unsafe { &*op };
- let concurrent_limit = ConcurrentLimitLayer::new(permits as usize);
+ let concurrent_limit = ConcurrentLimitLayer::with_permits(permits as usize);
Review Comment:
I don't think we need to this change this API.
##########
core/src/layers/concurrent_limit.rs:
##########
@@ -15,20 +15,52 @@
// specific language governing permissions and limitations
// under the License.
-use std::fmt::Debug;
use std::pin::Pin;
use std::sync::Arc;
use std::task::Context;
use std::task::Poll;
use futures::Stream;
use futures::StreamExt;
-use tokio::sync::OwnedSemaphorePermit;
+use std::any::Any;
+use std::future::Future;
use tokio::sync::Semaphore;
use crate::raw::*;
use crate::*;
+/// ConcurrencySemaphore abstracts a semaphore-like concurrency primitive
+/// that yields an owned permit released on drop. It mirrors RetryLayer's
+/// interceptor pattern by serving as a generic extension point for the layer.
+pub trait ConcurrencySemaphore: Send + Sync + Any + Clone + 'static {
Review Comment:
How about naming it after the layer like `ConcurrentLimitSemaphore`?
##########
core/src/layers/concurrent_limit.rs:
##########
@@ -81,41 +121,78 @@ use crate::*;
/// # }
/// ```
#[derive(Clone)]
-pub struct ConcurrentLimitLayer {
- operation_semaphore: Arc<Semaphore>,
- http_semaphore: Option<Arc<Semaphore>>,
+pub struct ConcurrentLimitLayer<S: ConcurrencySemaphore = TokioSemaphore> {
+ operation_semaphore: Arc<S>,
+ http_semaphore: Option<Arc<S>>,
}
-impl ConcurrentLimitLayer {
- /// Create a new ConcurrentLimitLayer will specify permits.
+impl ConcurrentLimitLayer<TokioSemaphore> {
+ /// Create a new `ConcurrentLimitLayer` with a custom semaphore.
///
- /// This permits will applied to all operations.
- pub fn new(permits: usize) -> Self {
+ /// The provided semaphore will be shared by every operator wrapped by this
+ /// layer, giving callers full control over its configuration.
+ pub fn new(operation_semaphore: Arc<Semaphore>) -> Self {
+ Self::with_semaphore(Arc::new(TokioSemaphore::new(operation_semaphore)))
+ }
+
+ // with_semaphore is provided by the generic impl below for the default Tokio adapter.
+
+ /// Create a new `ConcurrentLimitLayer` with the specified number of
+ /// permits.
+ pub fn with_permits(permits: usize) -> Self {
+ Self::new(Arc::new(Semaphore::new(permits)))
+ }
Review Comment:
We can remove this API.
##########
core/src/layers/concurrent_limit.rs:
##########
@@ -81,41 +121,78 @@ use crate::*;
/// # }
/// ```
#[derive(Clone)]
-pub struct ConcurrentLimitLayer {
- operation_semaphore: Arc<Semaphore>,
- http_semaphore: Option<Arc<Semaphore>>,
+pub struct ConcurrentLimitLayer<S: ConcurrencySemaphore = TokioSemaphore> {
+ operation_semaphore: Arc<S>,
+ http_semaphore: Option<Arc<S>>,
}
-impl ConcurrentLimitLayer {
- /// Create a new ConcurrentLimitLayer will specify permits.
+impl ConcurrentLimitLayer<TokioSemaphore> {
+ /// Create a new `ConcurrentLimitLayer` with a custom semaphore.
///
- /// This permits will applied to all operations.
- pub fn new(permits: usize) -> Self {
+ /// The provided semaphore will be shared by every operator wrapped by this
+ /// layer, giving callers full control over its configuration.
+ pub fn new(operation_semaphore: Arc<Semaphore>) -> Self {
Review Comment:
Please don't change public API unless we have to.
##########
core/src/layers/concurrent_limit.rs:
##########
@@ -81,41 +121,78 @@ use crate::*;
/// # }
/// ```
#[derive(Clone)]
-pub struct ConcurrentLimitLayer {
- operation_semaphore: Arc<Semaphore>,
- http_semaphore: Option<Arc<Semaphore>>,
+pub struct ConcurrentLimitLayer<S: ConcurrencySemaphore = TokioSemaphore> {
+ operation_semaphore: Arc<S>,
+ http_semaphore: Option<Arc<S>>,
}
-impl ConcurrentLimitLayer {
- /// Create a new ConcurrentLimitLayer will specify permits.
+impl ConcurrentLimitLayer<TokioSemaphore> {
+ /// Create a new `ConcurrentLimitLayer` with a custom semaphore.
///
- /// This permits will applied to all operations.
- pub fn new(permits: usize) -> Self {
+ /// The provided semaphore will be shared by every operator wrapped by this
+ /// layer, giving callers full control over its configuration.
+ pub fn new(operation_semaphore: Arc<Semaphore>) -> Self {
+ Self::with_semaphore(Arc::new(TokioSemaphore::new(operation_semaphore)))
+ }
+
+ // with_semaphore is provided by the generic impl below for the default Tokio adapter.
+
+ /// Create a new `ConcurrentLimitLayer` with the specified number of
+ /// permits.
+ pub fn with_permits(permits: usize) -> Self {
+ Self::new(Arc::new(Semaphore::new(permits)))
+ }
+
+ /// Set a concurrent limit for HTTP requests.
+ ///
+ /// This convenience helper constructs a new semaphore with the specified
+ /// number of permits and calls [`ConcurrentLimitLayer::with_http_semaphore`].
+ /// Use [`ConcurrentLimitLayer::with_http_semaphore`] directly when reusing
+ /// a shared semaphore.
+ pub fn with_http_concurrent_limit(self, permits: usize) -> Self {
+ self.with_http_semaphore(Arc::new(Semaphore::new(permits)))
+ }
+
+ /// Provide a custom semaphore to limit HTTP request concurrency.
+ ///
+ /// Sharing the same semaphore across layers allows coordinating HTTP
+ /// usage with other parts of the application.
+ pub fn with_http_semaphore(mut self, semaphore: Arc<Semaphore>) -> Self {
+ self.http_semaphore = Some(Arc::new(TokioSemaphore::new(semaphore)));
+ self
+ }
+
+ // with_http_concurrency is provided by the generic impl below.
+}
+
+impl<S: ConcurrencySemaphore> ConcurrentLimitLayer<S> {
+ /// Create a layer with any ConcurrencySemaphore implementation.
+ pub fn with_semaphore(operation_semaphore: Arc<S>) -> Self {
Self {
- operation_semaphore: Arc::new(Semaphore::new(permits)),
+ operation_semaphore,
http_semaphore: None,
}
}
- /// Set a concurrent limit for HTTP requests.
- ///
- /// This will limit the number of concurrent HTTP requests made by the
- /// operator.
- pub fn with_http_concurrent_limit(mut self, permits: usize) -> Self {
- self.http_semaphore = Some(Arc::new(Semaphore::new(permits)));
+ /// Provide a custom HTTP concurrency semaphore instance.
+ pub fn with_http_concurrency(mut self, semaphore: Arc<S>) -> Self {
+ self.http_semaphore = Some(semaphore);
Review Comment:
It's better to always follow the same API naming pattern.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: make ConcurrentLimitLayer accept Semaphore [opendal]
Posted by "PsiACE (via GitHub)" <gi...@apache.org>.
PsiACE commented on PR #6618:
URL: https://github.com/apache/opendal/pull/6618#issuecomment-3680907482
Sorry for the delay — this PR has been open for a while and I lost track. I’ve ported the changes to #7082 for a fresh update.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: make ConcurrentLimitLayer accept Semaphore [opendal]
Posted by "PsiACE (via GitHub)" <gi...@apache.org>.
PsiACE closed pull request #6618: feat: make ConcurrentLimitLayer accept Semaphore
URL: https://github.com/apache/opendal/pull/6618
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: make ConcurrentLimitLayer accept Semaphore [opendal]
Posted by "PsiACE (via GitHub)" <gi...@apache.org>.
PsiACE commented on code in PR #6618:
URL: https://github.com/apache/opendal/pull/6618#discussion_r2542957074
##########
core/src/layers/concurrent_limit.rs:
##########
@@ -90,24 +90,61 @@ use crate::*;
/// Ok(())
/// # }
/// ```
+/// ConcurrencySemaphore abstracts a semaphore-like concurrency primitive
+/// that yields an owned permit released on drop. It mirrors RetryLayer's
+/// interceptor pattern by serving as a generic extension point for the layer.
+pub trait ConcurrencySemaphore: Send + Sync + Any + Clone + 'static {
+ /// The owned permit type associated with the semaphore. Dropping it
+ /// must release the permit back to the semaphore.
+ type Permit: Send + Sync + 'static;
+
+ /// Acquire an owned permit asynchronously.
+ fn acquire_owned(&self) -> BoxFuture<'static, Self::Permit>;
Review Comment:
Switched the trait to a single `acquire` method that returns `impl Future + MaybeSend`. Implementations still use `async fn acquire(&self)`, but the trait now enforces the `Send` requirement our layer methods need.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: make ConcurrentLimitLayer accept Semaphore [opendal]
Posted by "tisonkun (via GitHub)" <gi...@apache.org>.
tisonkun commented on PR #6618:
URL: https://github.com/apache/opendal/pull/6618#issuecomment-3366414914
Ummm I can see the original issue is about using user's tokio Semaphore now. Let me think a bit ...
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: make ConcurrentLimitLayer accept Semaphore [opendal]
Posted by "tisonkun (via GitHub)" <gi...@apache.org>.
tisonkun commented on PR #6618:
URL: https://github.com/apache/opendal/pull/6618#issuecomment-3366418167
@bonsairobo do you have a public example how a Semaphore would be shared in this case?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@opendal.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org