Async functions to await a full or empty queue#4
Conversation
bikeshedder
left a comment
There was a problem hiding this comment.
This is a nice addition to the crate.
I just think the "simple" implementations of those methods using just Semaphore::acquire_many are wrong. I'd like to see a test case that actually checks if calling the full and empty method don't block concurrent push and pop calls. The tests would look like that:
- Full
- Create a queue with capacity of 2
- call
full(tokio::spawn) - call
push - call
popand make sure it doesn't block - call
pushtwice and check that thefullcall returned
- Empty
- Create a queue with capacity of 2
- call
pushtwice - call
empty(tokio::spawn) - call
push - call
popand make sure it doesn't block - call
poptwice and check thatemptycall returned
One more thing: I don't like the names empty and full. empty almost sounds like it is supposed to clear the queue. How about creating naming them wait_empty/wait_full or on_empty/on_full instead? That would make it very clear that the queues aren't modified but you're waiting for a specific event to happen.
|
If you were to change the Even then you might miss an |
|
Hi, read all these comments. Thanks for the feedback! Trying to find time to rework the PR accordingly. |
6464b89 to
5e15fd2
Compare
Allow awaiting a full or empty queue. - `Queue.empty()` is useful for chaining futures in a multi-threaded program when some subsequent action, like a verification step, needs to be taken once the queue is emptied. - `Queue.full()` can help manage backpressure so that an autoscaling system can monitor when the queue fills before it starts increasing the number of workers.
Same API as described in previous commit for limited::Queue, but the implementation is different, because resizable::Queue's capacity management is different. It doesn't use a pair of Semaphore structs that can be awaited to detect full and empty conditions; resizable::Queue only uses one Semaphore for push operations. This can be awaited to detect an empty queue. Detecting a full queue asynchronously is more complex: calls to `push()`, `try_push()`, and `resize()` call a private function to check if the queue has become full and notify any callers awaiting `full()`.
Also fix missing 0.2.3 and 0.2.4 version links.
Unlimited queues should be able to be awaited for an empty status. Required a separate implementation than limited and resizable queue due to different usage of internal semaphore.
Barrier ensures that the main test thread doesn't empty (or fill) the queue before the spawned task has a chance to start. The test still passes without the Barrier, but could be passing only due to the short-circuiting `self.if_empty()` or `self.capacity() == self.len()` logic in the body of `async fn empty()`/`full()`.
Eliminates deadlocks highlighted in code review caused by `Semaphore::acquire_many()` Moved common `Notifier` type alias to `src/lib.rs`
If `limited` feature isn't enabled, doctest in `src/lib.rs` fails to run, because it depends on `crate::limited::Queue`. I've wrapped it in a `cfg_attr` macro so that it is possible to run tests like: ```shell > cargo test --no-default-features --features unlimited ```
Discovered that `limited::Queue::available()` was returning incorrect values, because `self.available.add()` had been omitted from `try_push()`. Added unit tests for `available()` to all queue types. Also made a small change in statement order to `unlimited::Queue` to make the code identical to `limited` for easier comparison.
Per maintainer feedback, avoid using `Semaphore::available_permits` to detect queue full/empty state, rather use return values of atomic functions. That way notification is based on queue size at that moment in time, not a subsequent call into `Semaphore` that might race with other concurrent `push/pop` calls.
5e15fd2 to
a8d52fd
Compare
|
Finally circled back to this and updated the |
|
Sorry for taking so long. I completely forgot about that PR. 🙈 After reviewing it I only found two small details:
If you happen to find time to look into that I'll wait a bit. I'm just as happy to merge this PR without those changes. |
|
Oh, and please explain this comment to me... How is it borrowed mutably? 🤔 |
|
I should be able to look at these this weekend. Thanks!
|
|
@overhacked Any updates on this PR? |
Instead of having to write logic in consumers that adds or subtracts one from the returned `add/sub` value to get the new value for comparisons, have `Available` methods return the new value.
Expose the `subscribe()` method on the internal `tokio::watch::Sender<()>` so that API consumers can subscribe to repeated queue full/empty notifications without having to re-create and await a receiver every time, as with the `wait_full()/wait_empty()` methods.
Comment was regarding old `resize(&mut self, ...)` implementation, which was changed to `&self` in 39e5013.
|
Requested changes pushed. Sorry I lost track of wrapping this up in the last few months of the year. Nothing going on in November or December... nothing else at all. 😉 Thanks for your patience and support with this! |
|
Thanks a lot. Awesome work! 🎉 |
|
I just saw that I never actually released a new version including your changes. 🤦 I just releated 📦 https://crates.io/crates/deadqueue/0.2.5 Thanks again for this PR! ❤️ |
|
Amazing, thank you! |
Allow awaiting a full or empty queue.
Queue.empty()is useful for chaining futuresin a multi-threaded program when some subsequent
action, like a verification step, needs to be
taken once the queue is emptied.
Queue.full()can help manage backpressure so thatan autoscaling system can monitor when the queue fills
before it starts increasing the number of workers.