Struct Semaphore

Source

pub struct Semaphore { /* private fields */ }

Expand description

An async weighted semaphore. See crate documentation for usage.

Implementations§

Source §

impl Semaphore

Source

pub const MAX_AVAILABLE: usize = 2_305_843_009_213_693_951usize

The maximum number of permits that can be made available. This is slightly smaller than usize::MAX. If the number of available permits exceeds this number, it may poison the semaphore.

§Examples

struct ReadWriteLock(Semaphore);
impl ReadWriteLock {
    fn new() -> Self {
        ReadWriteLock(Semaphore::new(Semaphore::MAX_AVAILABLE))
    }
    // Acquire one permit, allowing up to MAX_AVAILABLE concurrent readers.
    async fn read(&self) -> SemaphoreGuard<'_> {
        self.0.acquire(1).await.unwrap()
    }
    // The writer acquires all the permits, prevent any concurrent writers or readers. The
    // first-in-first-out priority policy prevents writer starvation.
    async fn write(&self) -> SemaphoreGuard<'_> {
        self.0.acquire(Semaphore::MAX_AVAILABLE).await.unwrap()
    }
}

Source

pub fn new(initial: usize) -> Self

Create a new semaphore with an initial number of permits.

§Examples

use async_weighted_semaphore::Semaphore;
let semaphore = Semaphore::new(1024);

Source

pub fn acquire(&self, amount: usize) -> AcquireFuture<'_> ⓘ

Wait until there are no older pending calls to acquire and at least amount permits available. Then consume the requested permits and return a SemaphoreGuard.

§Errors

Returns PoisonError is the semaphore is poisoned.

§Examples

use async_weighted_semaphore::Semaphore;
async fn limit_concurrency(semaphore: &Semaphore, future: impl Future<Output=()>) {
    let guard = semaphore.acquire(1).await.unwrap();
    future.await
}

Source

pub fn try_acquire( &self, amount: usize, ) -> Result<SemaphoreGuard<'_>, TryAcquireError>

Like acquire, but fails if the call would block.

§Errors

Returns TryAcquireError::Poisoned is the semaphore is poisoned.
Returns TryAcquireError::WouldBlock if a call to acquire would have blocked. This can occur if there are insufficient available permits or if there is another pending call to acquire.

§Examples

use async_weighted_semaphore::Semaphore;
async fn run_if_safe(semaphore: &Semaphore, future: impl Future<Output=()>) {
    if semaphore.try_acquire(1).is_ok() {
        future.await
    }
}

Source

pub fn acquire_arc(self: &Arc<Self>, amount: usize) -> AcquireFutureArc ⓘ

Like acquire, but takes an Arc <Semaphore> and returns a guard that is 'static, Send and Sync.

§Examples

use async_channel::{Sender, SendError};
// Limit size of a producer-consumer queue
async fn send<T>(semaphore: &Arc<Semaphore>,
                 sender: &Sender<(SemaphoreGuardArc, T)>,
                 message: T
        ) -> Result<(), SendError<T>>{
    match semaphore.acquire_arc(1).await {
        // A semaphore can be poisoned to prevent deadlock when a channel closes.
        Err(PoisonError) => Err(SendError(message)),
        Ok(guard) => match sender.send((guard, message)).await{
            Err(SendError((guard, message))) => Err(SendError(message)),
            Ok(()) => Ok(())
        }
    }
}

Source

pub fn try_acquire_arc( self: &Arc<Self>, amount: usize, ) -> Result<SemaphoreGuardArc, TryAcquireError>

Like try_acquire, but takes an Arc <Semaphore>, and returns a guard that is 'static, Send and Sync.

§Examples

use async_channel::{Sender, TrySendError};
// Limit size of a producer-consumer queue
async fn try_send<T>(semaphore: &Arc<Semaphore>,
                 sender: &Sender<(SemaphoreGuardArc, T)>,
                 message: T
        ) -> Result<(), TrySendError<T>>{
    match semaphore.try_acquire_arc(1) {
        Err(TryAcquireError::WouldBlock) => Err(TrySendError::Full(message)),
        // A semaphore can be poisoned to prevent deadlock when a channel closes.
        Err(TryAcquireError::Poisoned) => Err(TrySendError::Closed(message)),
        Ok(guard) => match sender.try_send((guard, message)) {
            Err(TrySendError::Closed((guard, message))) => Err(TrySendError::Closed(message)),
            Err(TrySendError::Full((guard, message))) => Err(TrySendError::Full(message)),
            Ok(()) => Ok(())
        }
    }
}

Source

pub fn release(&self, amount: usize)

Return amount permits to the semaphore. This will eventually wake any calls to acquire that can succeed with the additional permits. Calling release often makes sense after calling SemaphoreGuard::forget or when using the semaphore to signal the number of elements that are available for processing.

§Examples

use async_channel::{Receiver, RecvError};
// Limit size of a producer-consumer queue
async fn recv<T>(semaphore: &Semaphore, recv: &Receiver<T>) -> Result<T, RecvError>{
    let result = recv.recv().await?;
    // Note that this only guards elements in the queue, not those being processed after the
    // queue.
    semaphore.release(1);
    Ok(result)
}

Source

pub fn poison(&self)

Poison the semaphore, causing all pending and future calls to acquire to fail immediately. This can be used to unblock pending acquires when the guarded operation would fail anyway.

§Examples

use async_channel::{Receiver, RecvError};
async fn consume(semaphore: &Semaphore, receiver: Receiver<usize>){
    while let Ok(x) = receiver.recv().await {
        println!("{:?}", x);
        semaphore.release(1);
    }
    // There will be no more calls to recv, so unblock all senders.
    semaphore.poison();
}