996 Advanced

996 Timeout Pattern

Functional Programming

Tutorial

The Problem

Implement timeout patterns in Rust using mpsc::recv_timeout. Run an operation in a background thread and wait for its result with a deadline; return an error if the deadline expires. Also implement a "race" pattern where N tasks run concurrently and the first result wins — the async equivalent of Lwt.pick.

🎯 Learning Outcomes

• Use rx.recv_timeout(Duration) to wait for a channel message with a deadline

• Implement with_timeout<T, F>(timeout, f) -> Option<T> — run f in a thread, return None on timeout

• Implement race<T>(tasks, timeout) -> Option<T> — spawn N tasks sharing one sender, return first result

• Handle RecvTimeoutError::Timeout vs RecvTimeoutError::Disconnected in match arms

• Understand why leftover threads continue running after timeout — graceful cancellation requires AtomicBool

Code Example

#![allow(clippy::all)]
// 996: Timeout Pattern
// Rust: mpsc::recv_timeout — like OCaml's Lwt.pick with sleep

use std::sync::mpsc;
use std::thread;
use std::time::Duration;

// --- Approach 1: recv_timeout on a channel ---
fn channel_with_timeout(delay_ms: u64, timeout_ms: u64) -> Result<i32, &'static str> {
    let (tx, rx) = mpsc::channel::<i32>();

    thread::spawn(move || {
        thread::sleep(Duration::from_millis(delay_ms));
        tx.send(42).ok(); // may fail if receiver timed out and was dropped
    });

    match rx.recv_timeout(Duration::from_millis(timeout_ms)) {
        Ok(v) => Ok(v),
        Err(mpsc::RecvTimeoutError::Timeout) => Err("timeout"),
        Err(mpsc::RecvTimeoutError::Disconnected) => Err("disconnected"),
    }
}

// --- Approach 2: Run any function with a timeout via thread ---
fn with_timeout<T, F>(timeout: Duration, f: F) -> Option<T>
where
    T: Send + 'static,
    F: FnOnce() -> T + Send + 'static,
{
    let (tx, rx) = mpsc::channel::<T>();
    thread::spawn(move || {
        let result = f();
        tx.send(result).ok();
    });
    rx.recv_timeout(timeout).ok()
}

// --- Approach 3: First-of-N wins (Lwt.pick analogue) ---
fn race<T: Send + 'static>(
    tasks: Vec<Box<dyn FnOnce() -> T + Send + 'static>>,
    timeout: Duration,
) -> Option<T> {
    let (tx, rx) = mpsc::channel::<T>();

    for task in tasks {
        let tx = tx.clone();
        thread::spawn(move || {
            let result = task();
            tx.send(result).ok(); // first to arrive wins
        });
    }
    drop(tx); // close original sender

    rx.recv_timeout(timeout).ok()
}

// --- Approach 4: Retry with overall deadline ---
fn retry_with_deadline<T, E, F>(
    max_attempts: usize,
    timeout_per_attempt: Duration,
    f: F,
) -> Result<T, &'static str>
where
    T: Send + 'static,
    E: Send + 'static,
    F: Fn() -> Result<T, E> + Send + Sync + Clone + 'static,
{
    for attempt in 0..max_attempts {
        let f = f.clone();
        let result = with_timeout(timeout_per_attempt, f);
        match result {
            Some(Ok(v)) => return Ok(v),
            Some(Err(_)) | None => {
                if attempt + 1 < max_attempts {
                    thread::sleep(Duration::from_millis(1 << attempt));
                }
            }
        }
    }
    Err("max attempts exceeded")
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_recv_before_timeout() {
        let result = channel_with_timeout(10, 500);
        assert_eq!(result, Ok(42));
    }

    #[test]
    fn test_recv_after_timeout() {
        let result = channel_with_timeout(200, 20);
        assert_eq!(result, Err("timeout"));
    }

    #[test]
    fn test_with_timeout_succeeds() {
        let result = with_timeout(Duration::from_millis(500), || {
            thread::sleep(Duration::from_millis(5));
            99i32
        });
        assert_eq!(result, Some(99));
    }

    #[test]
    fn test_with_timeout_expires() {
        let result = with_timeout(Duration::from_millis(5), || {
            thread::sleep(Duration::from_millis(100));
            99i32
        });
        assert_eq!(result, None);
    }

    #[test]
    fn test_race_fastest_wins() {
        let tasks: Vec<Box<dyn FnOnce() -> u32 + Send + 'static>> = vec![
            Box::new(|| {
                thread::sleep(Duration::from_millis(50));
                1
            }),
            Box::new(|| {
                thread::sleep(Duration::from_millis(5));
                2
            }),
            Box::new(|| {
                thread::sleep(Duration::from_millis(30));
                3
            }),
        ];
        let winner = race(tasks, Duration::from_millis(200));
        assert_eq!(winner, Some(2)); // fastest thread wins
    }

    #[test]
    fn test_recv_timeout_error_types() {
        let (_, rx) = mpsc::channel::<i32>();
        // Disconnected immediately (no sender)
        let err = rx.recv_timeout(Duration::from_millis(1));
        assert!(err.is_err());
    }
}

(* 996: Timeout Pattern *)
(* OCaml: Lwt.pick [operation; Lwt_unix.sleep timeout] concept *)
(* Pure Thread version: run operation in separate thread, wait with timeout *)

(* --- Approach 1: Thread + timed wait via Condition --- *)

type 'a timed_result = Ok of 'a | Timeout | Error of string

let with_timeout_secs timeout_s f =
  let result = ref None in
  let m = Mutex.create () in
  let cond = Condition.create () in

  let worker = Thread.create (fun () ->
    let v = (try Some (f ()) with e -> Some (Error (Printexc.to_string e))) in
    Mutex.lock m;
    result := Some v;
    Condition.signal cond;
    Mutex.unlock m
  ) () in

  Mutex.lock m;
  let deadline = Unix.gettimeofday () +. timeout_s in
  while !result = None do
    let remaining = deadline -. Unix.gettimeofday () in
    if remaining <= 0.0 then (
      result := Some (Some (Error "forced timeout"));
      (* Note: OCaml has no thread kill — worker will finish eventually *)
    ) else
      Condition.wait cond m
      (* In real Lwt: Lwt.pick cancels the losing promise *)
  done;
  Mutex.unlock m;
  Thread.join worker;

  match !result with
  | None | Some None -> Timeout
  | Some (Some (Error msg)) when msg = "forced timeout" -> Timeout
  | Some (Some (Error msg)) -> Error msg
  | Some (Some v) -> Ok v

(* --- Approach 1: fast operation completes in time --- *)

let () =
  let r = with_timeout_secs 1.0 (fun () ->
    Unix.sleepf 0.01;
    42
  ) in
  (match r with
  | Ok v -> assert (v = 42); Printf.printf "Approach 1 (ok): %d\n" v
  | Timeout -> assert false
  | Error e -> Printf.printf "Error: %s\n" e)

(* --- Approach 2: Simulated recv_timeout (channel with deadline) --- *)

type 'a chan = { q: 'a Queue.t; m: Mutex.t; cond: Condition.t }

let make_chan () = { q = Queue.create (); m = Mutex.create (); cond = Condition.create () }

let send c v =
  Mutex.lock c.m; Queue.push v c.q;
  Condition.signal c.cond; Mutex.unlock c.m

let recv_timeout c timeout_s =
  let deadline = Unix.gettimeofday () +. timeout_s in
  Mutex.lock c.m;
  while Queue.is_empty c.q && Unix.gettimeofday () < deadline do
    let remaining = deadline -. Unix.gettimeofday () in
    if remaining > 0.0 then
      Condition.wait c.m c.cond  (* simplified: real code uses timed wait *)
  done;
  let v = if Queue.is_empty c.q then None else Some (Queue.pop c.q) in
  Mutex.unlock c.m;
  v

let () =
  let c = make_chan () in
  let _ = Thread.create (fun () ->
    Unix.sleepf 0.02;
    send c 99
  ) () in
  (* Very short timeout — will miss the send *)
  match recv_timeout c 0.001 with
  | None -> Printf.printf "Approach 2 (timeout): timed out as expected\n"
  | Some v -> Printf.printf "Approach 2 (got): %d\n" v

let () = Printf.printf "✓ All tests passed\n"

Key Differences

Aspect	Rust	OCaml
Timeout primitive	`recv_timeout(Duration)`	`Lwt.pick` + `Lwt_unix.sleep`
Cancellation	Thread leaks (manual `AtomicBool` needed)	`Lwt.pick` cancels losers automatically
Non-blocking sleep	`thread::sleep` blocks the OS thread	`Lwt_unix.sleep` is a non-blocking yield
Race result	First to send wins; rest silently fail	First resolved; rest are cancelled

thread::sleep blocks the OS thread — spawning many concurrent timeouts wastes one thread each. Use tokio::time::timeout for non-blocking async timeouts in production.

OCaml Approach

open Lwt

(* Lwt.pick cancels all losers; Lwt.choose does not *)
let with_timeout duration f =
  Lwt.pick [
    (let* () = Lwt_unix.sleep duration in Lwt.return None);
    (let* v = f () in Lwt.return (Some v));
  ]

let race tasks timeout =
  Lwt.pick (
    (let* () = Lwt_unix.sleep timeout in Lwt.return None) ::
    List.map (fun f ->
      let* v = f () in Lwt.return (Some v)
    ) tasks
  )

Lwt.pick runs all promises concurrently and returns the first result, cancelling all others. Lwt_unix.sleep integrates with the Lwt scheduler for non-blocking sleep — unlike Thread.sleep which blocks the OS thread.

Full Source

#![allow(clippy::all)]
// 996: Timeout Pattern
// Rust: mpsc::recv_timeout — like OCaml's Lwt.pick with sleep

use std::sync::mpsc;
use std::thread;
use std::time::Duration;

// --- Approach 1: recv_timeout on a channel ---
fn channel_with_timeout(delay_ms: u64, timeout_ms: u64) -> Result<i32, &'static str> {
    let (tx, rx) = mpsc::channel::<i32>();

    thread::spawn(move || {
        thread::sleep(Duration::from_millis(delay_ms));
        tx.send(42).ok(); // may fail if receiver timed out and was dropped
    });

    match rx.recv_timeout(Duration::from_millis(timeout_ms)) {
        Ok(v) => Ok(v),
        Err(mpsc::RecvTimeoutError::Timeout) => Err("timeout"),
        Err(mpsc::RecvTimeoutError::Disconnected) => Err("disconnected"),
    }
}

// --- Approach 2: Run any function with a timeout via thread ---
fn with_timeout<T, F>(timeout: Duration, f: F) -> Option<T>
where
    T: Send + 'static,
    F: FnOnce() -> T + Send + 'static,
{
    let (tx, rx) = mpsc::channel::<T>();
    thread::spawn(move || {
        let result = f();
        tx.send(result).ok();
    });
    rx.recv_timeout(timeout).ok()
}

// --- Approach 3: First-of-N wins (Lwt.pick analogue) ---
fn race<T: Send + 'static>(
    tasks: Vec<Box<dyn FnOnce() -> T + Send + 'static>>,
    timeout: Duration,
) -> Option<T> {
    let (tx, rx) = mpsc::channel::<T>();

    for task in tasks {
        let tx = tx.clone();
        thread::spawn(move || {
            let result = task();
            tx.send(result).ok(); // first to arrive wins
        });
    }
    drop(tx); // close original sender

    rx.recv_timeout(timeout).ok()
}

// --- Approach 4: Retry with overall deadline ---
fn retry_with_deadline<T, E, F>(
    max_attempts: usize,
    timeout_per_attempt: Duration,
    f: F,
) -> Result<T, &'static str>
where
    T: Send + 'static,
    E: Send + 'static,
    F: Fn() -> Result<T, E> + Send + Sync + Clone + 'static,
{
    for attempt in 0..max_attempts {
        let f = f.clone();
        let result = with_timeout(timeout_per_attempt, f);
        match result {
            Some(Ok(v)) => return Ok(v),
            Some(Err(_)) | None => {
                if attempt + 1 < max_attempts {
                    thread::sleep(Duration::from_millis(1 << attempt));
                }
            }
        }
    }
    Err("max attempts exceeded")
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_recv_before_timeout() {
        let result = channel_with_timeout(10, 500);
        assert_eq!(result, Ok(42));
    }

    #[test]
    fn test_recv_after_timeout() {
        let result = channel_with_timeout(200, 20);
        assert_eq!(result, Err("timeout"));
    }

    #[test]
    fn test_with_timeout_succeeds() {
        let result = with_timeout(Duration::from_millis(500), || {
            thread::sleep(Duration::from_millis(5));
            99i32
        });
        assert_eq!(result, Some(99));
    }

    #[test]
    fn test_with_timeout_expires() {
        let result = with_timeout(Duration::from_millis(5), || {
            thread::sleep(Duration::from_millis(100));
            99i32
        });
        assert_eq!(result, None);
    }

    #[test]
    fn test_race_fastest_wins() {
        let tasks: Vec<Box<dyn FnOnce() -> u32 + Send + 'static>> = vec![
            Box::new(|| {
                thread::sleep(Duration::from_millis(50));
                1
            }),
            Box::new(|| {
                thread::sleep(Duration::from_millis(5));
                2
            }),
            Box::new(|| {
                thread::sleep(Duration::from_millis(30));
                3
            }),
        ];
        let winner = race(tasks, Duration::from_millis(200));
        assert_eq!(winner, Some(2)); // fastest thread wins
    }

    #[test]
    fn test_recv_timeout_error_types() {
        let (_, rx) = mpsc::channel::<i32>();
        // Disconnected immediately (no sender)
        let err = rx.recv_timeout(Duration::from_millis(1));
        assert!(err.is_err());
    }
}

(* 996: Timeout Pattern *)
(* OCaml: Lwt.pick [operation; Lwt_unix.sleep timeout] concept *)
(* Pure Thread version: run operation in separate thread, wait with timeout *)

(* --- Approach 1: Thread + timed wait via Condition --- *)

type 'a timed_result = Ok of 'a | Timeout | Error of string

let with_timeout_secs timeout_s f =
  let result = ref None in
  let m = Mutex.create () in
  let cond = Condition.create () in

  let worker = Thread.create (fun () ->
    let v = (try Some (f ()) with e -> Some (Error (Printexc.to_string e))) in
    Mutex.lock m;
    result := Some v;
    Condition.signal cond;
    Mutex.unlock m
  ) () in

  Mutex.lock m;
  let deadline = Unix.gettimeofday () +. timeout_s in
  while !result = None do
    let remaining = deadline -. Unix.gettimeofday () in
    if remaining <= 0.0 then (
      result := Some (Some (Error "forced timeout"));
      (* Note: OCaml has no thread kill — worker will finish eventually *)
    ) else
      Condition.wait cond m
      (* In real Lwt: Lwt.pick cancels the losing promise *)
  done;
  Mutex.unlock m;
  Thread.join worker;

  match !result with
  | None | Some None -> Timeout
  | Some (Some (Error msg)) when msg = "forced timeout" -> Timeout
  | Some (Some (Error msg)) -> Error msg
  | Some (Some v) -> Ok v

(* --- Approach 1: fast operation completes in time --- *)

let () =
  let r = with_timeout_secs 1.0 (fun () ->
    Unix.sleepf 0.01;
    42
  ) in
  (match r with
  | Ok v -> assert (v = 42); Printf.printf "Approach 1 (ok): %d\n" v
  | Timeout -> assert false
  | Error e -> Printf.printf "Error: %s\n" e)

(* --- Approach 2: Simulated recv_timeout (channel with deadline) --- *)

type 'a chan = { q: 'a Queue.t; m: Mutex.t; cond: Condition.t }

let make_chan () = { q = Queue.create (); m = Mutex.create (); cond = Condition.create () }

let send c v =
  Mutex.lock c.m; Queue.push v c.q;
  Condition.signal c.cond; Mutex.unlock c.m

let recv_timeout c timeout_s =
  let deadline = Unix.gettimeofday () +. timeout_s in
  Mutex.lock c.m;
  while Queue.is_empty c.q && Unix.gettimeofday () < deadline do
    let remaining = deadline -. Unix.gettimeofday () in
    if remaining > 0.0 then
      Condition.wait c.m c.cond  (* simplified: real code uses timed wait *)
  done;
  let v = if Queue.is_empty c.q then None else Some (Queue.pop c.q) in
  Mutex.unlock c.m;
  v

let () =
  let c = make_chan () in
  let _ = Thread.create (fun () ->
    Unix.sleepf 0.02;
    send c 99
  ) () in
  (* Very short timeout — will miss the send *)
  match recv_timeout c 0.001 with
  | None -> Printf.printf "Approach 2 (timeout): timed out as expected\n"
  | Some v -> Printf.printf "Approach 2 (got): %d\n" v

let () = Printf.printf "✓ All tests passed\n"

✓ Tests Rust test suite

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_recv_before_timeout() {
        let result = channel_with_timeout(10, 500);
        assert_eq!(result, Ok(42));
    }

    #[test]
    fn test_recv_after_timeout() {
        let result = channel_with_timeout(200, 20);
        assert_eq!(result, Err("timeout"));
    }

    #[test]
    fn test_with_timeout_succeeds() {
        let result = with_timeout(Duration::from_millis(500), || {
            thread::sleep(Duration::from_millis(5));
            99i32
        });
        assert_eq!(result, Some(99));
    }

    #[test]
    fn test_with_timeout_expires() {
        let result = with_timeout(Duration::from_millis(5), || {
            thread::sleep(Duration::from_millis(100));
            99i32
        });
        assert_eq!(result, None);
    }

    #[test]
    fn test_race_fastest_wins() {
        let tasks: Vec<Box<dyn FnOnce() -> u32 + Send + 'static>> = vec![
            Box::new(|| {
                thread::sleep(Duration::from_millis(50));
                1
            }),
            Box::new(|| {
                thread::sleep(Duration::from_millis(5));
                2
            }),
            Box::new(|| {
                thread::sleep(Duration::from_millis(30));
                3
            }),
        ];
        let winner = race(tasks, Duration::from_millis(200));
        assert_eq!(winner, Some(2)); // fastest thread wins
    }

    #[test]
    fn test_recv_timeout_error_types() {
        let (_, rx) = mpsc::channel::<i32>();
        // Disconnected immediately (no sender)
        let err = rx.recv_timeout(Duration::from_millis(1));
        assert!(err.is_err());
    }
}

Deep Comparison

Timeout Pattern — Comparison

Core Insight

Timeouts express "I'd rather fail fast than wait forever." In OCaml's Lwt: Lwt.pick [op; sleep] races two promises and takes the first. In Rust std: recv_timeout is the primitive, or wrap in a thread for arbitrary operations.

OCaml Approach

• Lwt.pick [operation; Lwt_unix.sleep timeout] — cancels the loser

• Thread-based: spawn worker, timed Condition.wait with deadline

• OCaml cannot kill threads — worker keeps running after "timeout"

• Unix.gettimeofday for wall-clock deadline tracking

Rust Approach

• rx.recv_timeout(Duration) → Result<T, RecvTimeoutError>

• RecvTimeoutError::Timeout vs RecvTimeoutError::Disconnected

• with_timeout(dur, f) pattern: spawn thread, recv_timeout, discard handle

• The "lost" thread keeps running but its channel is dropped — no cleanup needed

• race(tasks, timeout) for "first-of-N" / Lwt.pick over multiple computations

Comparison Table

Concept	OCaml (Lwt)	Rust
Timeout primitive	`Lwt_unix.sleep t`	`rx.recv_timeout(Duration::from_millis(t))`
Race two futures	`Lwt.pick [f; sleep t]`	`race([task], timeout)`
Timeout result	`exception` or `None`	`Err(RecvTimeoutError::Timeout)`
Cancellation	Lwt cancels the losing promise	Thread keeps running (can't kill)
Timed channel recv	Manual Condition.wait with deadline	`rx.recv_timeout(dur)`
Wrap arbitrary work	`Lwt.wrap (fun () -> ...)`	`with_timeout(dur, \|\| f())`

std vs tokio

Aspect	std version	tokio version
Runtime	OS threads via `std::thread`	Async tasks on tokio runtime
Synchronization	`std::sync::Mutex`, `Condvar`	`tokio::sync::Mutex`, channels
Channels	`std::sync::mpsc` (unbounded)	`tokio::sync::mpsc` (bounded, async)
Blocking	Thread blocks on lock/recv	Task yields, runtime switches tasks
Overhead	One OS thread per task	Many tasks per thread (M:N)
Best for	CPU-bound, simple concurrency	I/O-bound, high-concurrency servers

Exercises

Implement with_cancel<T, F>(cancel: Arc<AtomicBool>, f: F) -> Option<T> — f checks cancel periodically and stops early.

Implement retry_with_timeout(n, timeout, f) — retry up to n times, each attempt capped at timeout.

Implement first_k_of_n<T>(tasks, k, timeout) -> Vec<T> — collect the first k results.

Rewrite with_timeout using tokio::time::timeout and compare implementation complexity.

Benchmark: 100 concurrent with_timeout calls using threads vs using tokio::spawn + tokio::time::timeout.

Open Source Repos

functional-rust

View the source for this example on GitHub — OCaml and Rust side by side in the repo.

Rust