🎬 Fearless Concurrency Threads, Arc>, channels — safe parallelism enforced by the compiler.

📝 Text version (for readers / accessibility)

• std::thread::spawn creates OS threads — closures must be Send + 'static

• Arc> provides shared mutable state across threads safely

• Channels (mpsc) enable message passing — multiple producers, single consumer

• Send and Sync marker traits enforce thread safety at compile time

• Data races are impossible — the type system prevents them before your code runs

350: Oneshot Channel

Difficulty: 2 Level: Intermediate Send exactly one value from one task to another — a future you resolve from outside.

The Problem This Solves

You spawn a background task to do some work and you need to get its result back. `mpsc` is overkill — you only need to send one value. What you want is a single-use channel: one sender, one receiver, one message. After the message is sent, the channel is done. This is the request-response pattern in async systems: the caller sends a request along with a `OneshotSender`; the handler processes the request and calls `sender.send(response)`. The caller had been `await`ing the `OneshotReceiver`, and now it unblocks with the result. This avoids shared mutable state entirely — the response travels through the channel rather than being written to a shared variable. `tokio::sync::oneshot` is the production version. It integrates with async/await: the receiver is a `Future` that resolves when the sender fires.

The Intuition

Like a JavaScript `Promise` with an external `resolve`:

let resolve;
const promise = new Promise(r => resolve = r);
// later, from anywhere:
resolve(42);
// awaiting promise now gets 42

The `OneshotSender` is `resolve`. The `OneshotReceiver` is the `promise`. The key property: the sender is consumed on send — it can't be called twice.

How It Works in Rust

fn oneshot<T>() -> (OneshotSender<T>, OneshotReceiver<T>) {
 let state = Arc::new((Mutex::new(None), Condvar::new()));
 (OneshotSender { state: Arc::clone(&state) }, OneshotReceiver { state })
}

impl<T> OneshotSender<T> {
 fn send(self, value: T) {  // `self` is consumed — can only send once
     let (lock, cvar) = &*self.state;
     *lock.lock().unwrap() = Some(value);
     cvar.notify_one();  // wake the receiver
 }
}

impl<T> OneshotReceiver<T> {
 fn recv(self) -> T {  // blocks until sender fires
     let (lock, cvar) = &*self.state;
     let mut guard = lock.lock().unwrap();
     while guard.is_none() {
         guard = cvar.wait(guard).unwrap();
     }
     guard.take().unwrap()
 }
}

The sender takes `self` by value — consuming it prevents sending twice. This is the Rust type system enforcing "exactly one send" at compile time, not runtime. In async Rust:

let (tx, rx) = tokio::sync::oneshot::channel::<i32>();
tokio::spawn(async move { tx.send(42).unwrap(); });
let value = rx.await.unwrap();

What This Unlocks

Request-response actors — send a request with an embedded `OneshotSender`; the actor sends the reply through it.
Parallel task results — spawn N tasks each with their own oneshot; collect all results independently.
Async barrier — use a oneshot as a ready signal: "wait here until this initialization is complete."

Key Differences

Concept	OCaml	Rust
Oneshot channel	`Lwt.wait ()` → `(promise, resolver)` pair	`tokio::sync::oneshot::channel()`
Send once	Convention — resolver can be called multiple times	Type system: `send(self)` consumes the sender
Waiting	`Lwt.bind promise (fun v -> ...)`	`rx.await` or `rx.recv()`
Cross-thread	`Mutex` + `Condition`	Built into `tokio::sync::oneshot`

use std::sync::{Arc, Mutex, Condvar};
use std::thread;
use std::time::Duration;

// Oneshot channel: one value, one send, one receive
struct OneshotSender<T> {
    state: Arc<(Mutex<Option<T>>, Condvar)>,
}

struct OneshotReceiver<T> {
    state: Arc<(Mutex<Option<T>>, Condvar)>,
}

fn oneshot<T>() -> (OneshotSender<T>, OneshotReceiver<T>) {
    let state = Arc::new((Mutex::new(None), Condvar::new()));
    (
        OneshotSender { state: Arc::clone(&state) },
        OneshotReceiver { state },
    )
}

impl<T> OneshotSender<T> {
    fn send(self, value: T) {
        let (lock, cvar) = &*self.state;
        *lock.lock().unwrap() = Some(value);
        cvar.notify_one();
        // self is consumed — cannot send again
    }
}

impl<T> OneshotReceiver<T> {
    fn recv(self) -> T {
        let (lock, cvar) = &*self.state;
        let mut guard = lock.lock().unwrap();
        while guard.is_none() {
            guard = cvar.wait(guard).unwrap();
        }
        guard.take().unwrap()
    }

    fn try_recv(&self) -> Option<T> {
        self.state.0.lock().unwrap().take()
    }
}

// Use case: request-response pattern
fn handle_request(data: String) -> (String, OneshotReceiver<String>) {
    let (tx, rx) = oneshot();
    thread::spawn(move || {
        thread::sleep(Duration::from_millis(5));
        let response = format!("processed: {data}");
        tx.send(response);
    });
    (data.clone(), rx)
}

fn main() {
    // Basic oneshot
    let (tx, rx) = oneshot::<i32>();
    thread::spawn(move || {
        thread::sleep(Duration::from_millis(10));
        tx.send(42);
    });
    let value = rx.recv();
    println!("Got: {value}");

    // Request-response
    let requests: Vec<_> = vec!["alpha", "beta", "gamma"]
        .into_iter()
        .map(|s| handle_request(s.to_string()))
        .collect();

    for (req, rx) in requests {
        println!("{req} -> {}", rx.recv());
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    #[test]
    fn oneshot_delivers_value() {
        let (tx, rx) = oneshot::<i32>();
        tx.send(99);
        assert_eq!(rx.recv(), 99);
    }
    #[test]
    fn oneshot_across_threads() {
        let (tx, rx) = oneshot::<String>();
        thread::spawn(move || { tx.send("hello".into()); });
        assert_eq!(rx.recv(), "hello");
    }
    #[test]
    fn try_recv_before_send_is_none() {
        let (_tx, rx) = oneshot::<i32>();
        assert!(rx.try_recv().is_none());
    }
}

(* OCaml: oneshot via Mutex + flag *)

type 'a oneshot = {
  mutable value : 'a option;
  mutex : Mutex.t;
  cond  : Condition.t;
}

let make () =
  { value = None; mutex = Mutex.create (); cond = Condition.create () }

let send os v =
  Mutex.lock os.mutex;
  os.value <- Some v;
  Condition.signal os.cond;
  Mutex.unlock os.mutex

let recv os =
  Mutex.lock os.mutex;
  while os.value = None do
    Condition.wait os.cond os.mutex
  done;
  let v = Option.get os.value in
  Mutex.unlock os.mutex;
  v

let () =
  let os = make () in
  let _ = Thread.create (fun () ->
    Thread.delay 0.01;
    send os 42
  ) () in
  let v = recv os in
  Printf.printf "Received: %d\n" v