Semaphore, Notify, Barrier, and OnceCell

A mutex protects data. The types in this lesson coordinate tasks: how many may run at once, when one may proceed, when a group has assembled, and who initializes a shared value. Each example uses the paused clock, so timing results are exact and repeatable.

Type	Purpose
`Semaphore`	at most N tasks in a section: `acquire`, `acquire_owned`, `try_acquire`, `close`, `add_permits`
`Notify`	wake a waiting task: `notify_one`, `notify_waiters`, the stored-permit rule
`Barrier`	make N tasks wait for each other
`OnceCell`	a value initialized once by an async function
`SetOnce`	a cell written once, with waiters that resume when it is set

Examples with start_paused = true use #[tokio::main(flavor = "current_thread", start_paused = true)]: one thread and a virtual clock that jumps forward when every task waits on a timer. It needs the test-util feature and is explained in Time: sleep, timeout, and interval.

Example 1: Permits: acquire and release

A semaphore holds a number of permits. A task takes a permit with acquire and gives it back when the permit value is dropped. When no permit is left, acquire waits. The Tokio documentation describes the type as one that "can allow more than one concurrent caller to access the shared resource at a time", in contrast with a mutex.

Rust

use tokio::sync::Semaphore;

#[tokio::main]
async fn main() {
    let semaphore = Semaphore::new(2);
    println!("available: {}", semaphore.available_permits());

    let first = semaphore.acquire().await.unwrap();
    let second = semaphore.acquire().await.unwrap();
    println!("after two acquires: {}", semaphore.available_permits());

    drop(first);
    println!("after dropping one permit: {}", semaphore.available_permits());
    drop(second);
    println!("after dropping both: {}", semaphore.available_permits());
}

Output

available: 2
after two acquires: 0
after dropping one permit: 1
after dropping both: 2

acquire returns Result<SemaphorePermit, AcquireError>. The error appears only when the semaphore has been closed (Example 6), so code that never closes the semaphore can call unwrap.

Example 2: try_acquire never waits

try_acquire returns at once. Its error type, TryAcquireError, has two cases: NoPermits when none is free and Closed when the semaphore is closed.

Rust

use tokio::sync::{Semaphore, TryAcquireError};

#[tokio::main]
async fn main() {
    let semaphore = Semaphore::new(1);

    let permit = semaphore.try_acquire().unwrap();
    match semaphore.try_acquire() {
        Ok(_) => println!("second permit granted"),
        Err(TryAcquireError::NoPermits) => println!("no permits available"),
        Err(TryAcquireError::Closed) => println!("closed"),
    }
    println!("error text: {}", semaphore.try_acquire().unwrap_err());

    drop(permit);
    println!("after release: {}", semaphore.try_acquire().is_ok());
}

Output

no permits available
error text: no permits available
after release: true

try_acquire suits code that prefers to reject work over waiting, such as shedding load when a service is busy. Backpressure and Load Shedding builds on it.

Example 3: Limit how many tasks run at once

The main use of a semaphore is a concurrency limit. Each task takes a permit before its work and holds it until the work ends. The example starts six tasks, allows two at a time, and records the highest number of tasks inside the section.

Rust

use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::Semaphore;
use tokio::time::{sleep, Instant};

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let semaphore = Arc::new(Semaphore::new(2));
    let running = Arc::new(AtomicUsize::new(0));
    let peak = Arc::new(AtomicUsize::new(0));
    let start = Instant::now();

    let mut handles = Vec::new();
    for _ in 0..6 {
        let semaphore = Arc::clone(&semaphore);
        let running = Arc::clone(&running);
        let peak = Arc::clone(&peak);
        handles.push(tokio::spawn(async move {
            let _permit = semaphore.acquire().await.unwrap();
            let now = running.fetch_add(1, Ordering::SeqCst) + 1;
            peak.fetch_max(now, Ordering::SeqCst);
            sleep(Duration::from_millis(10)).await; // the limited work
            running.fetch_sub(1, Ordering::SeqCst);
        }));
    }
    for handle in handles {
        handle.await.unwrap();
    }

    println!("peak concurrency = {}", peak.load(Ordering::SeqCst));
    println!("virtual elapsed = {} ms", start.elapsed().as_millis());
}

Output

peak concurrency = 2
virtual elapsed = 30 ms

The peak is 2. Six tasks in batches of two take three rounds of 10 ms each: 30 ms. The permit is a value bound to _permit, so it is released at the end of the task even if the task returns early or panics.

Example 4: acquire_owned: take the permit before spawning

acquire returns a permit that borrows the semaphore, so it cannot move into a spawned task. acquire_owned takes an Arc<Semaphore> and returns an OwnedSemaphorePermit that can move anywhere.

Acquiring the permit in the spawning loop, before spawn, also limits how many tasks exist at once. The loop waits for a permit, so it never creates more than N unfinished tasks.

Rust

use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::Semaphore;
use tokio::time::sleep;

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let semaphore = Arc::new(Semaphore::new(3));
    let alive = Arc::new(AtomicUsize::new(0));
    let peak_alive = Arc::new(AtomicUsize::new(0));

    let mut handles = Vec::new();
    for _ in 0..10 {
        // The loop waits here when 3 tasks are already running.
        let permit = Arc::clone(&semaphore).acquire_owned().await.unwrap();
        let now = alive.fetch_add(1, Ordering::SeqCst) + 1;
        peak_alive.fetch_max(now, Ordering::SeqCst);

        let alive = Arc::clone(&alive);
        handles.push(tokio::spawn(async move {
            sleep(Duration::from_millis(10)).await;
            alive.fetch_sub(1, Ordering::SeqCst);
            drop(permit); // released when the task finishes
        }));
    }
    for handle in handles {
        handle.await.unwrap();
    }

    println!("spawned 10 tasks, at most {} alive at once", peak_alive.load(Ordering::SeqCst));
    println!("permits at the end = {}", semaphore.available_permits());
}

Output

spawned 10 tasks, at most 3 alive at once
permits at the end = 3

Ten tasks were spawned, and no more than three existed at any moment. This is the shape of a bounded fan-out. JoinSet and Structured Concurrency uses it with JoinSet.

Example 5: Fairness: permits are given out in request order

The Semaphore documentation states: "This Semaphore is fair, which means that permits are given out in the order they were requested. This fairness is also applied when acquire_many gets involved, so if a call to acquire_many at the front of the queue requests more permits than currently available, this can prevent a call to acquire from completing, even if the semaphore has enough permits complete the call to acquire."

acquire_many(n) asks for n permits in one call. In the example, one permit is held, so one of two is free. A task asks for two and waits at the front of the queue. A later request for one permit also waits, although one permit exists.

Rust

use std::sync::Arc;
use std::time::Duration;
use tokio::sync::Semaphore;
use tokio::time::{sleep, timeout};

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let semaphore = Arc::new(Semaphore::new(2));
    let held = semaphore.acquire().await.unwrap();

    let big = {
        let semaphore = Arc::clone(&semaphore);
        tokio::spawn(async move {
            let _permits = semaphore.acquire_many(2).await.unwrap();
            println!("big request got 2 permits");
        })
    };
    sleep(Duration::from_millis(1)).await; // big is now waiting at the front

    let small = timeout(Duration::from_millis(10), semaphore.acquire()).await;
    println!("small request timed out: {}", small.is_err());

    drop(held);
    big.await.unwrap();
    println!("permits at the end = {}", semaphore.available_permits());
}

Output

small request timed out: true
big request got 2 permits
permits at the end = 2

The small request waits behind the large one and gives up after 10 ms. Releasing the held permit lets the large request finish. A queue in request order prevents a stream of small requests from starving a large one.

Example 6: close wakes every waiter with an error

close prevents the semaphore from issuing more permits and wakes every waiting task. Their acquire calls return Err(AcquireError). New try_acquire calls return TryAcquireError::Closed. A permit that was already granted stays valid.

Rust

use std::sync::Arc;
use std::time::Duration;
use tokio::sync::Semaphore;
use tokio::time::sleep;

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let semaphore = Arc::new(Semaphore::new(1));
    let held = semaphore.acquire().await.unwrap();

    let waiter = {
        let semaphore = Arc::clone(&semaphore);
        tokio::spawn(async move {
            match semaphore.acquire().await {
                Ok(_) => "waiter got a permit".to_string(),
                Err(e) => format!("waiter failed: {e}"),
            }
        })
    };
    sleep(Duration::from_millis(1)).await;

    println!("closed before: {}", semaphore.is_closed());
    semaphore.close();
    println!("closed after: {}", semaphore.is_closed());
    println!("{}", waiter.await.unwrap());
    println!("try_acquire: {}", semaphore.try_acquire().unwrap_err());

    drop(held); // the permit granted before close was still valid
    println!("available after drop: {}", semaphore.available_permits());
}

Output

closed before: false
closed after: true
waiter failed: semaphore closed
try_acquire: semaphore closed
available after drop: 1

Closing is a shutdown signal: tasks blocked on acquire learn that no permit will arrive and can stop.

Example 7: add_permits, forget, and forget_permits

A semaphore can also count events. add_permits(n) adds permits, and a waiting task receives one. SemaphorePermit::forget consumes a permit without returning it, and forget_permits(n) removes up to n free permits. The total is limited to Semaphore::MAX_PERMITS.

Rust

use std::sync::Arc;
use std::time::Duration;
use tokio::sync::Semaphore;
use tokio::time::sleep;

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    // A semaphore that starts empty works as a signal.
    let gate = Arc::new(Semaphore::new(0));
    let waiter = {
        let gate = Arc::clone(&gate);
        tokio::spawn(async move {
            let permit = gate.acquire().await.unwrap();
            permit.forget(); // consume the permit for good
            "waiter passed the gate"
        })
    };
    sleep(Duration::from_millis(1)).await;
    gate.add_permits(1);
    println!("{}", waiter.await.unwrap());
    println!("permits after forget: {}", gate.available_permits());

    // Resize a pool.
    let pool = Semaphore::new(4);
    let removed = pool.forget_permits(3);
    println!("removed {removed}, left {}", pool.available_permits());
    pool.add_permits(2);
    println!("after add_permits(2): {}", pool.available_permits());
    println!("MAX_PERMITS is usize::MAX >> 3: {}", Semaphore::MAX_PERMITS == usize::MAX >> 3);
}

Output

waiter passed the gate
permits after forget: 0
removed 3, left 1
after add_permits(2): 3
MAX_PERMITS is usize::MAX >> 3: true

add_permits panics when the total would exceed MAX_PERMITS. Forgotten permits leave the pool until add_permits puts them back, which makes a semaphore usable as a resizable limit.

Example 8: Notify: wake a waiting task

Notify signals an event without carrying data. One task waits with notified().await, another calls notify_one. The Tokio documentation describes it as a semaphore that starts with zero permits, where notify_one sets a permit.

Rust

use std::sync::Arc;
use std::time::Duration;
use tokio::sync::Notify;
use tokio::time::sleep;

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let notify = Arc::new(Notify::new());

    let waiter = {
        let notify = Arc::clone(&notify);
        tokio::spawn(async move {
            println!("waiter: waiting");
            notify.notified().await;
            println!("waiter: notified");
        })
    };

    sleep(Duration::from_millis(5)).await;
    println!("main: notifying");
    notify.notify_one();
    waiter.await.unwrap();
}

Output

waiter: waiting
main: notifying
waiter: notified

Example 9: The stored-permit rule

The documentation states: "If notify_one() is called before notified().await, then the next call to notified().await will complete immediately, consuming the permit." Many calls before the wait store only one permit: "only a single permit is stored".

Rust

use std::time::Duration;
use tokio::sync::Notify;
use tokio::time::timeout;

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let notify = Notify::new();

    notify.notify_one();
    notify.notify_one();
    notify.notify_one(); // three calls, one stored permit

    notify.notified().await;
    println!("first notified(): completed at once");

    let second = timeout(Duration::from_millis(10), notify.notified()).await;
    println!("second notified() timed out: {}", second.is_err());
}

Output

first notified(): completed at once
second notified() timed out: true

The stored permit prevents a lost wakeup: a notification sent before the receiver started waiting. The rule also means that three notifications can collapse into one wakeup. A receiver must therefore drain all pending work after a wakeup, not assume one wakeup per unit of work.

Example 10: notify_waiters stores nothing

notify_waiters wakes every task that is currently waiting and does not store a permit. The documentation states: "Unlike with notify_one(), no permit is stored to be used by the next call to notified().await." A Notified future is registered for notify_waiters as soon as notified() creates it, even before it is polled.

Rust

use std::sync::Arc;
use std::time::Duration;
use tokio::sync::Notify;
use tokio::time::{sleep, timeout};

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let notify = Arc::new(Notify::new());

    // No one is waiting: the call is lost.
    notify.notify_waiters();
    let late = timeout(Duration::from_millis(10), notify.notified()).await;
    println!("waiter after notify_waiters timed out: {}", late.is_err());

    // Three tasks are waiting: all wake.
    let mut handles = Vec::new();
    for i in 0..3 {
        let notify = Arc::clone(&notify);
        handles.push(tokio::spawn(async move {
            notify.notified().await;
            i
        }));
    }
    sleep(Duration::from_millis(1)).await;
    notify.notify_waiters();

    let mut woken = Vec::new();
    for handle in handles {
        woken.push(handle.await.unwrap());
    }
    println!("woken tasks: {woken:?}");
}

Output

waiter after notify_waiters timed out: true
woken tasks: [0, 1, 2]

Call	Waiters present	No waiters present
`notify_one`	wakes the oldest waiter	stores one permit
`notify_last`	wakes the newest waiter	stores one permit
`notify_waiters`	wakes all waiters	does nothing

Example 11: notify_one is FIFO, notify_last is LIFO

notify_one wakes the waiter that has waited longest. notify_last wakes the most recent waiter. Three tasks start waiting at 1, 2, and 3 ms. One notify_one and one notify_last wake two of them.

Rust

use std::sync::Arc;
use std::time::Duration;
use tokio::sync::Notify;
use tokio::time::sleep;

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let notify = Arc::new(Notify::new());
    let woken = Arc::new(std::sync::Mutex::new(Vec::new()));

    let mut handles = Vec::new();
    for i in 1..=3u64 {
        let notify = Arc::clone(&notify);
        let woken = Arc::clone(&woken);
        handles.push(tokio::spawn(async move {
            sleep(Duration::from_millis(i)).await; // task i starts waiting at i ms
            notify.notified().await;
            woken.lock().unwrap().push(i);
        }));
    }

    sleep(Duration::from_millis(10)).await;
    notify.notify_one();
    sleep(Duration::from_millis(1)).await;
    println!("after notify_one: {:?}", woken.lock().unwrap());

    notify.notify_last();
    sleep(Duration::from_millis(1)).await;
    println!("after notify_last: {:?}", woken.lock().unwrap());

    notify.notify_one(); // wake the last remaining task
    for handle in handles {
        handle.await.unwrap();
    }
    println!("all woken: {:?}", woken.lock().unwrap());
}

Output

after notify_one: [1]
after notify_last: [1, 3]
all woken: [1, 3, 2]

Task 1 waited longest, so notify_one woke it. Task 3 waited least, so notify_last woke it. Task 2 was left for the final notify_one.

Example 12: enable: register before checking a condition

Checking a condition and then waiting has a race window: the event can happen between the check and the wait. A stored permit covers notify_one but only one permit exists. The Notified::enable method registers the future in the waiter list without polling it. Once enabled, a notify_one wakes that future and does not store a permit.

The example creates two Notified futures and sends two notify_one calls before either is awaited. Without enable, the second call finds the permit already stored and is lost. With enable, each call finds a registered waiter.

Rust

use std::time::Duration;
use tokio::sync::Notify;
use tokio::time::timeout;

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    // Without enable: the two notifications collapse into one permit.
    let notify = Notify::new();
    let first = notify.notified();
    let second = notify.notified();
    tokio::pin!(first, second);
    notify.notify_one();
    notify.notify_one();
    first.as_mut().await;
    let lost = timeout(Duration::from_millis(10), second.as_mut()).await;
    println!("without enable, second waiter timed out: {}", lost.is_err());

    // With enable: each notification reaches a registered waiter.
    let notify = Notify::new();
    let first = notify.notified();
    let second = notify.notified();
    tokio::pin!(first, second);
    first.as_mut().enable();
    second.as_mut().enable();
    notify.notify_one();
    notify.notify_one();
    first.as_mut().await;
    let kept = timeout(Duration::from_millis(10), second.as_mut()).await;
    println!("with enable, second waiter completed: {}", kept.is_ok());
}

Output

without enable, second waiter timed out: true
with enable, second waiter completed: true

The Tokio documentation uses enable in its example of a channel built from a queue and a Notify: enable the Notified future, then check the queue, then await. Because the future is registered before the check, an item that arrives after the check still wakes it.

Example 13: A wait_until pattern without a lost wakeup

Notify carries no data, so the shared state lives elsewhere, here in an atomic flag. The waiting loop creates and enables the Notified future before it reads the flag. A set that lands after the read still wakes the waiter.

Rust

use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::Notify;
use tokio::time::sleep;

struct Flag {
    value: AtomicBool,
    notify: Notify,
}

impl Flag {
    fn set(&self) {
        self.value.store(true, Ordering::SeqCst);
        self.notify.notify_waiters();
    }

    async fn wait(&self) {
        loop {
            let notified = self.notify.notified();
            tokio::pin!(notified);
            notified.as_mut().enable(); // register before checking the flag
            if self.value.load(Ordering::SeqCst) {
                return;
            }
            notified.await;
        }
    }
}

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let flag = Arc::new(Flag { value: AtomicBool::new(false), notify: Notify::new() });

    let mut handles = Vec::new();
    for i in 0..3 {
        let flag = Arc::clone(&flag);
        handles.push(tokio::spawn(async move {
            flag.wait().await;
            println!("waiter {i} saw the flag");
        }));
    }

    sleep(Duration::from_millis(5)).await;
    flag.set();
    for handle in handles {
        handle.await.unwrap();
    }

    flag.wait().await; // already set: returns without waiting
    println!("late waiter returned at once");
}

Output

waiter 0 saw the flag
waiter 1 saw the flag
waiter 2 saw the flag
late waiter returned at once

notify_waiters wakes all three waiters. A waiter that arrives after set sees the flag on its first check. The enable call closes the gap between the check and the wait.

Example 14: Barrier: wait for a group

A Barrier::new(n) makes tasks that call wait block until n tasks have called it. Then all are released together. Exactly one of them gets a BarrierWaitResult for which is_leader() is true.

The example uses three tasks that arrive at different virtual times. After the barrier each task reads the arrival counter, and all read 3.

Rust

use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::Barrier;
use tokio::time::sleep;

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let barrier = Arc::new(Barrier::new(3));
    let arrived = Arc::new(AtomicUsize::new(0));

    let mut handles = Vec::new();
    for i in 0..3u64 {
        let barrier = Arc::clone(&barrier);
        let arrived = Arc::clone(&arrived);
        handles.push(tokio::spawn(async move {
            sleep(Duration::from_millis(i * 5)).await; // arrive at 0, 5, 10 ms
            arrived.fetch_add(1, Ordering::SeqCst);
            let result = barrier.wait().await;
            (arrived.load(Ordering::SeqCst), result.is_leader())
        }));
    }

    let mut leaders = 0;
    for handle in handles {
        let (seen, is_leader) = handle.await.unwrap();
        println!("after barrier: arrivals seen = {seen}");
        if is_leader {
            leaders += 1;
        }
    }
    println!("leaders = {leaders}");
}

Output

after barrier: arrivals seen = 3
after barrier: arrivals seen = 3
after barrier: arrivals seen = 3
leaders = 1

Every task sees 3 arrivals: none passes the barrier before the last task arrives. The leader is one arbitrary task among the n.

Example 15: Barriers are reusable, and a missing task blocks the rest

After n tasks pass, the barrier resets for the next round. The documentation says "Barriers are re-usable after all tasks have rendezvoused once". If fewer than n tasks call wait, the callers wait indefinitely. The documentation also states that Barrier::wait "is not cancel safe". Barrier::new(0) behaves like Barrier::new(1), the same as std::sync::Barrier.

Rust

use std::sync::Arc;
use std::time::Duration;
use tokio::sync::Barrier;
use tokio::time::timeout;

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    // Two rounds with the same barrier.
    let barrier = Arc::new(Barrier::new(2));
    let mut handles = Vec::new();
    for id in 0..2 {
        let barrier = Arc::clone(&barrier);
        handles.push(tokio::spawn(async move {
            for round in 1..=2 {
                barrier.wait().await;
                if id == 0 {
                    println!("round {round} complete");
                }
            }
        }));
    }
    for handle in handles {
        handle.await.unwrap();
    }

    // Too few tasks: the only caller waits until the timeout ends the wait.
    let short = Barrier::new(3);
    let result = timeout(Duration::from_millis(50), short.wait()).await;
    println!("wait with 1 of 3 tasks timed out: {}", result.is_err());

    // A zero-size barrier behaves like a size-one barrier.
    let zero = Barrier::new(0);
    println!("Barrier::new(0) wait is leader: {}", zero.wait().await.is_leader());
}

Output

round 1 complete
round 2 complete
wait with 1 of 3 tasks timed out: true
Barrier::new(0) wait is leader: true

The documentation marks wait as not cancel safe, so the timeout here only makes the demonstration finite. It is not a recovery strategy. Code that must survive a missing participant needs another design, for example a Semaphore or a channel with a deadline.

Example 16: OnceCell: initialize once, asynchronously

OnceCell<T> holds a value that is set at most once. get_or_init takes an async function. If several tasks call it at the same time, one runs the function and the others wait for its result. The documentation states: "If some other task is currently working on initializing the OnceCell, this call will wait for that other task to finish, then return the value that the other task produced."

Rust

use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::OnceCell;
use tokio::time::sleep;

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let cell = Arc::new(OnceCell::new());
    let init_calls = Arc::new(AtomicUsize::new(0));

    let mut handles = Vec::new();
    for _ in 0..5 {
        let cell = Arc::clone(&cell);
        let init_calls = Arc::clone(&init_calls);
        handles.push(tokio::spawn(async move {
            *cell
                .get_or_init(|| async {
                    init_calls.fetch_add(1, Ordering::SeqCst);
                    sleep(Duration::from_millis(20)).await; // slow setup
                    42
                })
                .await
        }));
    }

    let mut values = Vec::new();
    for handle in handles {
        values.push(handle.await.unwrap());
    }
    println!("values = {values:?}");
    println!("initializer ran {} time(s)", init_calls.load(Ordering::SeqCst));
    println!("get() now = {:?}", cell.get());
}

Output

values = [42, 42, 42, 42, 42]
initializer ran 1 time(s)
get() now = Some(42)

Five tasks asked for the value, and the initializer ran once. The other four waited for it, so none of them repeated the slow setup.

Example 17: get_or_try_init retries after an error

get_or_try_init accepts a function that returns Result. On Err the cell stays empty and the error goes to the caller. The next call tries again. The same holds when the function panics or its future is dropped: the documentation says another waiting task then starts a new attempt.

Rust

use std::sync::atomic::{AtomicUsize, Ordering};
use tokio::sync::OnceCell;

async fn connect(attempt: usize) -> Result<String, String> {
    if attempt < 3 {
        Err(format!("attempt {attempt} failed"))
    } else {
        Ok(format!("connected on attempt {attempt}"))
    }
}

#[tokio::main]
async fn main() {
    let cell: OnceCell<String> = OnceCell::new();
    let attempts = AtomicUsize::new(0);

    for _ in 0..4 {
        let result = cell
            .get_or_try_init(|| {
                let attempt = attempts.fetch_add(1, Ordering::SeqCst) + 1;
                connect(attempt)
            })
            .await;
        println!("{result:?}");
    }
    println!("attempts made = {}", attempts.load(Ordering::SeqCst));
}

Output

Err("attempt 1 failed")
Err("attempt 2 failed")
Ok("connected on attempt 3")
Ok("connected on attempt 3")
attempts made = 3

Attempts 1 and 2 fail and leave the cell empty. Attempt 3 succeeds and fills it. The fourth call finds the value and does not run the function.

Example 18: A global OnceCell with const_new

OnceCell::const_new is a const fn, so a OnceCell can be a static. This form fits lazily created global state such as a client or a configuration value.

Rust

use tokio::sync::OnceCell;

static SETTINGS: OnceCell<Vec<&'static str>> = OnceCell::const_new();

async fn settings() -> &'static Vec<&'static str> {
    SETTINGS
        .get_or_init(|| async {
            println!("loading settings");
            vec!["verbose", "retry=3"]
        })
        .await
}

#[tokio::main]
async fn main() {
    println!("before: initialized = {}", SETTINGS.initialized());
    println!("{:?}", settings().await);
    println!("{:?}", settings().await);
    println!("after: initialized = {}", SETTINGS.initialized());
}

Output

before: initialized = false
loading settings
["verbose", "retry=3"]
["verbose", "retry=3"]
after: initialized = true

"loading settings" prints once. The second call returns the stored reference.

Example 19: set and its errors

set stores a value without an initializer function. It fails with SetError when the cell already holds a value, or when another task is in the middle of get_or_init. SetError::is_already_init_err and is_initializing_err tell the two apart.

Rust

use std::sync::Arc;
use std::time::Duration;
use tokio::sync::OnceCell;
use tokio::time::sleep;

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let cell = OnceCell::new();
    println!("first set: {:?}", cell.set(1));
    let err = cell.set(2).unwrap_err();
    println!("second set error: {err}");
    println!("already initialized: {}", err.is_already_init_err());
    println!("value = {:?}", cell.get());

    let busy = Arc::new(OnceCell::new());
    let init = {
        let busy = Arc::clone(&busy);
        tokio::spawn(async move {
            busy.get_or_init(|| async {
                sleep(Duration::from_millis(20)).await;
                10
            })
            .await;
        })
    };
    sleep(Duration::from_millis(1)).await; // initialization is running
    let err = busy.set(99).unwrap_err();
    println!("set during init error: {err}");
    println!("initializing: {}", err.is_initializing_err());
    init.await.unwrap();
    println!("busy = {:?}", busy.get());
}

Output

first set: Ok(())
second set error: AlreadyInitializedError
already initialized: true
value = Some(1)
set during init error: InitializingError
initializing: true
busy = Some(10)

Example 20: SetOnce: a cell that waiters can await

SetOnce<T> is written once with set. Unlike OnceCell, it has no initializer function. A task can call wait().await, which completes when another task sets the value. The documentation compares it to Python's asyncio.Event. A second set returns SetOnceError, which carries the rejected value.

Rust

use std::sync::Arc;
use std::time::Duration;
use tokio::sync::SetOnce;
use tokio::time::sleep;

#[tokio::main(flavor = "current_thread", start_paused = true)]
async fn main() {
    let once = Arc::new(SetOnce::new());

    let mut waiters = Vec::new();
    for i in 0..3 {
        let once = Arc::clone(&once);
        waiters.push(tokio::spawn(async move {
            let value: &String = once.wait().await;
            println!("waiter {i} got {value}");
        }));
    }

    sleep(Duration::from_millis(5)).await;
    println!("get before set: {:?}", once.get());
    once.set("ready".to_string()).unwrap();

    for waiter in waiters {
        waiter.await.unwrap();
    }

    let err = once.set("again".to_string()).unwrap_err();
    println!("second set: {err}, rejected value = {}", err.0);
    println!("get after set: {:?}", once.get());
}

Output

get before set: None
waiter 0 got ready
waiter 1 got ready
waiter 2 got ready
second set: SetOnceError, rejected value = again
get after set: Some("ready")

All three waiters resume when set runs. SetOnce fits a value that one task produces once and many tasks wait for, for example the address a server bound to.

Choosing between the types

Need	Type
At most N tasks in a section, or N resources	`Semaphore`
Wake a task when a flag or queue changes	`Notify` with shared state
N tasks must all reach a point before any continues	`Barrier`
Compute a value once on first use	`OnceCell::get_or_init`
Wait until another task provides a value once	`SetOnce::wait`
Pass a value, not just a signal	a channel (mpsc Channels)

Interview points

A semaphore counts permits. acquire waits when none is free, and dropping the permit returns it. It limits concurrency and works as a resource counter.
Tokio's semaphore is fair: permits are granted in request order. A large acquire_many at the front blocks later small requests even when enough permits are free for them.
acquire_owned returns an owned permit that can move into a spawned task. Acquiring before spawn limits the number of live tasks.
close wakes all waiters with AcquireError. try_acquire then returns TryAcquireError::Closed. Permits granted before the close stay valid.
add_permits adds capacity, forget and forget_permits remove it, and the total is capped at Semaphore::MAX_PERMITS.
Notify::notify_one stores at most one permit when no task waits, so a wakeup sent before the wait is not lost, and several sends before the wait collapse into one.
notify_waiters wakes only tasks registered at the time of the call and stores nothing.
notify_one wakes the oldest waiter, notify_last the newest.
A Notified future is registered by its first poll or by enable. Calling enable before checking a condition avoids the check-then-wait race.
Barrier releases n tasks together, elects one leader, and is reusable. Barrier::wait is not cancel safe. Barrier::new(0) acts as size one.
OnceCell::get_or_init runs one initializer for concurrent callers. A failed or cancelled initializer lets a waiter retry. It deadlocks if the initializer calls get_or_init on the same cell.
SetOnce adds wait, which lets many tasks await a value that one task sets once.

Summary

Semaphore limits concurrency with permits and hands them out in request order.
Permits can be borrowed (acquire) or owned (acquire_owned), taken without waiting (try_acquire), and taken in bulk (acquire_many).
close turns waiting into an error; add_permits and forget_permits resize the pool.
Notify signals events. notify_one keeps one permit for a later wait. notify_waiters keeps nothing.
enable registers a Notified future early, so a notification cannot slip between a check and a wait.
Barrier synchronizes N tasks and can be used for repeated rounds.
OnceCell initializes a value once, asynchronously, and retries after failure. SetOnce is set once and can be awaited.
These types build on the same fair queue as the mutex from Shared State: Mutex and RwLock.