Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

doc: Add examples to Semaphore #3808

Merged
merged 10 commits into from Jun 11, 2021
Merged

doc: Add examples to Semaphore #3808

Changes from 7 commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
67b810c
doc: Add examples to
oguzbilgener May 25, 2021
fcd40df
Apply suggestions from code review
oguzbilgener May 26, 2021
d22c439
Simplify per the code review
oguzbilgener May 26, 2021
f2f7f06
Cleanup
oguzbilgener May 26, 2021
5b0a3fa
Simplify more
oguzbilgener May 26, 2021
5573454
Apply suggestions from code review
oguzbilgener Jun 1, 2021
be537be
Fix a few things
oguzbilgener Jun 1, 2021
26d17a6
Update tokio/src/sync/semaphore.rs
oguzbilgener Jun 2, 2021
759b003
simplify try owned examples
oguzbilgener Jun 6, 2021
90f27e7
remove async main from examples that don't need it
oguzbilgener Jun 6, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
232 changes: 232 additions & 0 deletions tokio/src/sync/semaphore.rs
View file
Open in desktop
Expand Up @@ -24,7 +24,55 @@ use std::sync::Arc;
/// To use the `Semaphore` in a poll function, you can use the [`PollSemaphore`]
/// utility.
///
/// # Examples
///
/// Basic usage:
///
/// ```
/// use tokio::sync::{Semaphore, TryAcquireError};
///
/// #[tokio::main]
/// async fn main() {
/// let semaphore = Semaphore::new(3);
///
/// let a_permit = semaphore.acquire().await.unwrap();
/// let two_permits = semaphore.acquire_many(2).await.unwrap();
///
/// assert_eq!(semaphore.available_permits(), 0);
///
/// let permit_attempt = semaphore.try_acquire();
/// assert_eq!(permit_attempt.err(), Some(TryAcquireError::NoPermits));
/// }
/// ```
///
/// Use [`Semaphore::acquire_owned`] to move permits across tasks:
///
/// ```
/// use std::sync::Arc;
/// use tokio::sync::Semaphore;
///
/// #[tokio::main]
/// async fn main() {
/// let semaphore = Arc::new(Semaphore::new(3));
/// let mut join_handles = Vec::new();
///
/// for _ in 1..=5 {
/// let permit = semaphore.clone().acquire_owned().await.unwrap();
/// join_handles.push(tokio::spawn(async move {
/// // perform task...
/// // explicitly own `permit` in the task
/// drop(permit);
/// }));
/// }
///
/// for handle in join_handles {
/// handle.await.unwrap();
/// }
/// }
/// ```
///
/// [`PollSemaphore`]: https://docs.rs/tokio-util/0.6/tokio_util/sync/struct.PollSemaphore.html
/// [`Semaphore::acquire_owned`]: crate::sync::Semaphore::acquire_owned
#[derive(Debug)]
pub struct Semaphore {
/// The low level semaphore
Expand Down Expand Up @@ -79,6 +127,15 @@ impl Semaphore {
}

/// Creates a new semaphore with the initial number of permits.
///
/// # Examples
///
/// ```
/// use tokio::sync::Semaphore;
///
/// static SEM: Semaphore = Semaphore::const_new(10);
/// ```
///
#[cfg(all(feature = "parking_lot", not(all(loom, test))))]
#[cfg_attr(docsrs, doc(cfg(feature = "parking_lot")))]
pub const fn const_new(permits: usize) -> Self {
Expand All @@ -105,6 +162,26 @@ impl Semaphore {
/// Otherwise, this returns a [`SemaphorePermit`] representing the
/// acquired permit.
///
/// # Examples
///
/// ```
/// use tokio::sync::Semaphore;
///
/// #[tokio::main]
/// async fn main() {
/// let semaphore = Semaphore::new(2);
///
/// let permit_1 = semaphore.acquire().await.unwrap();
/// assert_eq!(semaphore.available_permits(), 1);
///
/// let permit_2 = semaphore.acquire().await.unwrap();
/// assert_eq!(semaphore.available_permits(), 0);
///
/// drop(permit_1);
/// assert_eq!(semaphore.available_permits(), 1);
/// }
/// ```
///
/// [`AcquireError`]: crate::sync::AcquireError
/// [`SemaphorePermit`]: crate::sync::SemaphorePermit
pub async fn acquire(&self) -> Result<SemaphorePermit<'_>, AcquireError> {
Expand All @@ -121,6 +198,20 @@ impl Semaphore {
/// Otherwise, this returns a [`SemaphorePermit`] representing the
/// acquired permits.
///
/// # Examples
///
/// ```
/// use tokio::sync::Semaphore;
///
/// #[tokio::main]
/// async fn main() {
/// let semaphore = Semaphore::new(5);
///
/// let p1 = semaphore.acquire_many(3).await.unwrap();
/// assert_eq!(semaphore.available_permits(), 2);
/// }
/// ```
///
/// [`AcquireError`]: crate::sync::AcquireError
/// [`SemaphorePermit`]: crate::sync::SemaphorePermit
pub async fn acquire_many(&self, n: u32) -> Result<SemaphorePermit<'_>, AcquireError> {
Expand All @@ -137,6 +228,26 @@ impl Semaphore {
/// and a [`TryAcquireError::NoPermits`] if there are no permits left. Otherwise,
/// this returns a [`SemaphorePermit`] representing the acquired permits.
///
/// # Examples
///
/// ```
/// use tokio::sync::{Semaphore, TryAcquireError};
///
/// #[tokio::main]
/// async fn main() {
/// let semaphore = Semaphore::new(2);
///
/// let permit_1 = semaphore.try_acquire().unwrap();
/// assert_eq!(semaphore.available_permits(), 1);
///
/// let permit_2 = semaphore.try_acquire().unwrap();
/// assert_eq!(semaphore.available_permits(), 0);
///
/// let permit_3 = semaphore.try_acquire();
/// assert_eq!(permit_3.err(), Some(TryAcquireError::NoPermits));
/// }
/// ```
///
/// [`TryAcquireError::Closed`]: crate::sync::TryAcquireError::Closed
/// [`TryAcquireError::NoPermits`]: crate::sync::TryAcquireError::NoPermits
/// [`SemaphorePermit`]: crate::sync::SemaphorePermit
Expand All @@ -156,6 +267,23 @@ impl Semaphore {
/// and a [`TryAcquireError::NoPermits`] if there are no permits left. Otherwise,
/// this returns a [`SemaphorePermit`] representing the acquired permits.
oguzbilgener marked this conversation as resolved.
Show resolved Hide resolved
///
/// # Examples
///
/// ```
/// use tokio::sync::{Semaphore, TryAcquireError};
///
/// #[tokio::main]
/// async fn main() {
/// let semaphore = Semaphore::new(4);
///
/// let permit_1 = semaphore.try_acquire_many(3).unwrap();
/// assert_eq!(semaphore.available_permits(), 1);
///
/// let permit_2 = semaphore.try_acquire_many(2);
/// assert_eq!(permit_2.err(), Some(TryAcquireError::NoPermits));
/// }
/// ```
///
/// [`TryAcquireError::Closed`]: crate::sync::TryAcquireError::Closed
/// [`TryAcquireError::NoPermits`]: crate::sync::TryAcquireError::NoPermits
/// [`SemaphorePermit`]: crate::sync::SemaphorePermit
Expand All @@ -176,6 +304,32 @@ impl Semaphore {
/// Otherwise, this returns a [`OwnedSemaphorePermit`] representing the
/// acquired permit.
///
/// # Examples
///
/// ```
/// use std::sync::Arc;
/// use tokio::sync::Semaphore;
///
/// #[tokio::main]
/// async fn main() {
/// let semaphore = Arc::new(Semaphore::new(3));
/// let mut join_handles = Vec::new();
///
/// for _ in 1..=5 {
/// let permit = semaphore.clone().acquire_owned().await.unwrap();
/// join_handles.push(tokio::spawn(async move {
/// // perform task...
/// // explicitly own `permit` in the task
/// drop(permit);
/// }));
/// }
///
/// for handle in join_handles {
/// handle.await.unwrap();
/// }
/// }
/// ```
///
/// [`Arc`]: std::sync::Arc
/// [`AcquireError`]: crate::sync::AcquireError
/// [`OwnedSemaphorePermit`]: crate::sync::OwnedSemaphorePermit
Expand All @@ -194,6 +348,32 @@ impl Semaphore {
/// Otherwise, this returns a [`OwnedSemaphorePermit`] representing the
/// acquired permit.
///
/// # Examples
///
/// ```
/// use std::sync::Arc;
/// use tokio::sync::Semaphore;
///
/// #[tokio::main]
/// async fn main() {
/// let semaphore = Arc::new(Semaphore::new(10));
/// let mut join_handles = Vec::new();
///
/// for _ in 1..=5 {
/// let permit = semaphore.clone().acquire_many_owned(2).await.unwrap();
/// join_handles.push(tokio::spawn(async move {
/// // perform task...
/// // explicitly own `permit` in the task
/// drop(permit);
/// }));
/// }
///
/// for handle in join_handles {
/// handle.await.unwrap();
/// }
/// }
/// ```
///
/// [`Arc`]: std::sync::Arc
/// [`AcquireError`]: crate::sync::AcquireError
/// [`OwnedSemaphorePermit`]: crate::sync::OwnedSemaphorePermit
Expand All @@ -216,6 +396,32 @@ impl Semaphore {
/// Otherwise, this returns a [`OwnedSemaphorePermit`] representing the
/// acquired permit.
///
/// # Examples
///
/// ```
/// use std::sync::Arc;
/// use tokio::sync::Semaphore;
///
/// #[tokio::main]
/// async fn main() {
/// let semaphore = Arc::new(Semaphore::new(5));
/// let mut join_handles = Vec::new();
///
/// for _ in 1..=5 {
/// let permit = semaphore.clone().try_acquire_owned().unwrap();
/// join_handles.push(tokio::spawn(async move {
/// // perform task...
/// // explicitly own `permit` in the task
/// drop(permit);
/// }));
/// }
///
/// for handle in join_handles {
/// handle.await.unwrap();
/// }
/// }
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This example isn't great. You wouldn't use try_acquire_owned like this.

Copy link
Sponsor Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The only realistic example I could think of that doesn't involve writing a poll method is something like this. Do you have any feedback?

use std::sync::Arc;
use tokio::sync::Semaphore;
use tokio_stream::{self as stream, StreamExt};

#[tokio::main]
async fn main() {
    // Simulate a stream of incoming requests from some I/O resource
    let mut requests = stream::iter(1u32..=20);
    // Allow up to 4 concurrent operations
    let semaphore = Arc::new(Semaphore::new(4));
    let mut join_handles = Vec::new();

    while let Some(id) = requests.next().await {
        match Arc::clone(&semaphore).try_acquire_owned() {
            Ok(permit) => {
                join_handles.push(tokio::spawn(async move {
                    perform_operation(id).await;
                    // Hold the permit until our operation is over.
                    drop(permit);
                }));
            }
            Err(TryAcquireError::NoPermits) => {
                // We're busy, we can respond immediately to tell the client to try again later.
            }
            Err(TryAcquireError::Closed) => {
                unreachable!() // no one calls `semaphore.close()`
            }
        }
    }

    for handle in join_handles {
        let _ = handle.await;
    }
}

async fn perform_operation(_id: u32) {
    tokio::time::sleep(Duration::from_nanos(1)).await
}

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Maybe just do the same thing as the example on try_acquire. It's a lot simpler even if it doesn't show off the move-to-new-task feature.

Copy link
Sponsor Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done, thank you for the review.

/// ```
///
/// [`Arc`]: std::sync::Arc
/// [`TryAcquireError::Closed`]: crate::sync::TryAcquireError::Closed
/// [`TryAcquireError::NoPermits`]: crate::sync::TryAcquireError::NoPermits
Expand All @@ -238,6 +444,32 @@ impl Semaphore {
/// Otherwise, this returns a [`OwnedSemaphorePermit`] representing the
/// acquired permit.
///
/// # Examples
///
/// ```
/// use std::sync::Arc;
/// use tokio::sync::Semaphore;
///
/// #[tokio::main]
/// async fn main() {
/// let semaphore = Arc::new(Semaphore::new(10));
/// let mut join_handles = Vec::new();
///
/// for _ in 1..=5 {
/// let permit = semaphore.clone().try_acquire_many_owned(2).unwrap();
/// join_handles.push(tokio::spawn(async move {
/// // perform task...
/// // explicitly own `permit` in the task
/// drop(permit);
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Same here.

/// }));
/// }
///
/// for handle in join_handles {
/// handle.await.unwrap();
/// }
/// }
/// ```
///
/// [`Arc`]: std::sync::Arc
/// [`TryAcquireError::Closed`]: crate::sync::TryAcquireError::Closed
/// [`TryAcquireError::NoPermits`]: crate::sync::TryAcquireError::NoPermits
Expand Down