Skip to content

Commit

Permalink
util: add pollable Semaphore (#3444)
Browse files Browse the repository at this point in the history
  • Loading branch information
Darksonn authored Jan 20, 2021
1 parent c1cf6b7 commit 27b2d68
Show file tree
Hide file tree
Showing 3 changed files with 89 additions and 2 deletions.
3 changes: 1 addition & 2 deletions tokio-util/Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -47,6 +47,7 @@ futures-util = { version = "0.3.0", optional = true }
log = "0.4"
pin-project-lite = "0.2.0"
slab = { version = "0.4.1", optional = true } # Backs `DelayQueue`
async-stream = "0.3.0"

[dev-dependencies]
tokio = { version = "1.0.0", features = ["full"] }
Expand All @@ -55,8 +56,6 @@ tokio-test = { version = "0.4.0" }
futures = "0.3.0"
futures-test = "0.3.5"

async-stream = "0.3.0"

[package.metadata.docs.rs]
all-features = true
rustdoc-args = ["--cfg", "docsrs"]
3 changes: 3 additions & 0 deletions tokio-util/src/sync/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -4,3 +4,6 @@ mod cancellation_token;
pub use cancellation_token::{CancellationToken, WaitForCancellationFuture};

mod intrusive_double_linked_list;

mod poll_semaphore;
pub use poll_semaphore::PollSemaphore;
85 changes: 85 additions & 0 deletions tokio-util/src/sync/poll_semaphore.rs
Original file line number Diff line number Diff line change
@@ -0,0 +1,85 @@
use futures_core::Stream;
use std::fmt;
use std::pin::Pin;
use std::sync::Arc;
use std::task::{Context, Poll};
use tokio::sync::{OwnedSemaphorePermit, Semaphore};

/// A wrapper around [`Semaphore`] that provides a `poll_acquire` method.
///
/// [`Semaphore`]: tokio::sync::Semaphore
pub struct PollSemaphore {
semaphore: Arc<Semaphore>,
inner: Pin<Box<dyn Stream<Item = OwnedSemaphorePermit> + Send + Sync>>,
}

impl PollSemaphore {
/// Create a new `PollSemaphore`.
pub fn new(semaphore: Arc<Semaphore>) -> Self {
Self {
semaphore: semaphore.clone(),
inner: Box::pin(async_stream::stream! {
loop {
match semaphore.clone().acquire_owned().await {
Ok(permit) => yield permit,
Err(_closed) => break,
}
}
}),
}
}

/// Closes the semaphore.
pub fn close(&self) {
self.semaphore.close()
}

/// Obtain a clone of the inner semaphore.
pub fn clone_inner(&self) -> Arc<Semaphore> {
self.semaphore.clone()
}

/// Get back the inner semaphore.
pub fn into_inner(self) -> Arc<Semaphore> {
self.semaphore
}

/// Poll to acquire a permit from the semaphore.
///
/// This can return the following values:
///
/// - `Poll::Pending` if a permit is not currently available.
/// - `Poll::Ready(Some(permit))` if a permit was acquired.
/// - `Poll::Ready(None)` if the semaphore has been closed.
///
/// When this method returns `Poll::Pending`, the current task is scheduled
/// to receive a wakeup when a permit becomes available, or when the
/// semaphore is closed. Note that on multiple calls to `poll_acquire`, only
/// the `Waker` from the `Context` passed to the most recent call is
/// scheduled to receive a wakeup.
pub fn poll_acquire(&mut self, cx: &mut Context<'_>) -> Poll<Option<OwnedSemaphorePermit>> {
self.inner.as_mut().poll_next(cx)
}
}

impl Stream for PollSemaphore {
type Item = OwnedSemaphorePermit;

fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<OwnedSemaphorePermit>> {
Pin::into_inner(self).poll_acquire(cx)
}
}

impl Clone for PollSemaphore {
fn clone(&self) -> PollSemaphore {
PollSemaphore::new(self.clone_inner())
}
}

impl fmt::Debug for PollSemaphore {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("PollSemaphore")
.field("semaphore", &self.semaphore)
.finish()
}
}

0 comments on commit 27b2d68

Please sign in to comment.