Add LimitStore (#2175) #2242
Add LimitStore (#2175) #2242
Conversation
Codecov Report
@@ Coverage Diff @@
## master #2242 +/- ##
=======================================
Coverage 82.30% 82.30%
=======================================
Files 241 242 +1
Lines 62437 62517 +80
=======================================
+ Hits 51389 51457 +68
- Misses 11048 11060 +12
Help us with your feedback. Take ten seconds to tell us how you rate us. |
| @@ -684,6 +684,7 @@ impl AmazonS3Builder { | |||
|
|
|||
| /// Sets the maximum number of concurrent outstanding | |||
| /// connectons. Default is `16`. | |||
| #[deprecated(note = "use LimitStore instead")] | |||
There was a problem hiding this comment.
This is a nice side-effect of #2204, we can deprecate individual things now
|
|
||
| /// Store wrapper that wraps an inner store and limits the maximum number of concurrent | ||
| /// object store operations. Where each call to an [`ObjectStore`] member function is | ||
| /// considered a single operation, even if it may result in more than one network call |
There was a problem hiding this comment.
It is worth highlighting the implementation of this functionality for S3 has the same property, despite calling it a connection limit
There was a problem hiding this comment.
An example in this doc of how to construct a LimitStore might be helpful
|
|
||
| /// Store wrapper that wraps an inner store and limits the maximum number of concurrent | ||
| /// object store operations. Where each call to an [`ObjectStore`] member function is | ||
| /// considered a single operation, even if it may result in more than one network call |
There was a problem hiding this comment.
An example in this doc of how to construct a LimitStore might be helpful
object_store/src/limit.rs
Outdated
| } | ||
|
|
||
| impl<T: ObjectStore> LimitStore<T> { | ||
| /// Create new limit store |
There was a problem hiding this comment.
| /// Create new limit store | |
| /// Create new limit store that will limit the maximum | |
| /// number of outstanding concurrent requests to | |
| /// `max_requests` |
object_store/src/limit.rs
Outdated
|
|
||
| impl<T: ObjectStore> std::fmt::Display for LimitStore<T> { | ||
| fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { | ||
| write!(f, "LimitStore({})", self.inner) |
There was a problem hiding this comment.
including max_requests would also be helpful here
| #[async_trait] | ||
| impl<T: ObjectStore> ObjectStore for LimitStore<T> { | ||
| async fn put(&self, location: &Path, bytes: Bytes) -> Result<()> { | ||
| let _permit = self.semaphore.acquire().await.unwrap(); |
There was a problem hiding this comment.
I double checked if we needed to check the error here -- https://docs.rs/tokio/1.20.1/tokio/sync/struct.Semaphore.html#method.acquire says that errors are only returned if the Semaphore is closed, which this code does not do 👍
|
|
||
| let t = Duration::from_millis(1); | ||
|
|
||
| // Expect to not be able to make another request |
There was a problem hiding this comment.
👍 I am always a little worried when I see a sleep style thing in a test as I worry about intermittent failures due to timing differences, but this use seems sound 👍
Maybe we can bump the timeout up to 20ms or something to be safe on slow machines
There was a problem hiding this comment.
FWIW as this is a timeout all a slow machine could potentially mean is that the test erroneous passes, when it should have failed. I'll bump the timeout to be on the safe side
|
|
|
Benchmark runs are scheduled for baseline = 281cd79 and contender = 6c3f9a2. 6c3f9a2 is a master commit associated with this PR. Results will be available as each benchmark for each run completes. |
Which issue does this PR close?
Part of #2175.
Rationale for this change
What changes are included in this PR?
Adds LimitStore
Are there any user-facing changes?
No, a future PR will remove this functionality from the AWS implementation