Concurrency Coordination Reports

MPSC Try-Recv Coordination Report

A drain report records nonblocking receives from a channel. This program fills a channel, then uses try_recv a selected number of times and reports drained versus empty receives.

Program

Play the program to choose how many receive attempts run and watch the queue drain and then report empty.

attempts
mpsc_try_recv_coordination_report.rs
Replay: real traced execution (multi-file project)
use std::sync::mpsc::channel;

fn main() {
    let attempts = 3;
    let (tx, rx) = channel();
    for value in 0..3 {
        let _ = tx.send(value);
    }
    let mut drained = 0;
    let mut empty_hits = 0;
    for _ in 0..attempts {
        match rx.try_recv() {
            Ok(_) => drained += 1,
            Err(_) => empty_hits += 1,
        }
    }
    let status = if drained == 0 {
        "idle"
    } else if empty_hits > 0 {
        "drained"
    } else {
        "draining"
    };
    println!("attempts={attempts} drained={drained} empty={empty_hits} {status}");
}

use std::sync::mpsc::channel;

fn main() {
    let attempts = 0;
    let (tx, rx) = channel();
    for value in 0..3 {
        let _ = tx.send(value);
    }
    let mut drained = 0;
    let mut empty_hits = 0;
    for _ in 0..attempts {
        match rx.try_recv() {
            Ok(_) => drained += 1,
            Err(_) => empty_hits += 1,
        }
    }
    let status = if drained == 0 {
        "idle"
    } else if empty_hits > 0 {
        "drained"
    } else {
        "draining"
    };
    println!("attempts={attempts} drained={drained} empty={empty_hits} {status}");
}

use std::sync::mpsc::channel;

fn main() {
    let attempts = 5;
    let (tx, rx) = channel();
    for value in 0..3 {
        let _ = tx.send(value);
    }
    let mut drained = 0;
    let mut empty_hits = 0;
    for _ in 0..attempts {
        match rx.try_recv() {
            Ok(_) => drained += 1,
            Err(_) => empty_hits += 1,
        }
    }
    let status = if drained == 0 {
        "idle"
    } else if empty_hits > 0 {
        "drained"
    } else {
        "draining"
    };
    println!("attempts={attempts} drained={drained} empty={empty_hits} {status}");
}

  1. attempts ← 3

    3fn main() {4    let attempt→ 3s = 3; //@attempts=3, 0, 55    let (tx, rx) = channel();6    for value in 0..3 {

  2. for value in 0..3

    pass 1 of 3
    5let (tx, rx) = channel();6for valu0e in 0..3 {7    let _ = tx.send(value);8}
    All 3 passes — pass 1 is the card above
    passvalue
    10
    21
    32

  3. drained ← 0, empty_hits ← 0

    8}9let mut draine→ 0d = 0;10let mut empty_hit→ 0s = 0;11for _ in 0..attempts {

  4. for _ in 0..attempts

    pass 1 of 3
    10let mut empty_hits = 0;11for _ in 0..attempt3s {12    match rx.try_recv() {

  5. status ← "draining"

    16    }17    let statu→ "draining"s = if draine3d == 0 {18        "idle"19    } else if empty_hit0s > 0 {20        "drained"21    } else {22        "draining"23    };24    println!("attempts={attempts} drained={drained} empty={empty_hits} {status}");25}
    outputattempts=3 drained=3 empty=0 draining

  1. attempts ← 0

    3fn main() {4    let attempt→ 0s = 0;5    let (tx, rx) = channel();6    for value in 0..3 {

  2. for value in 0..3

    pass 1 of 3
    5let (tx, rx) = channel();6for valu0e in 0..3 {7    let _ = tx.send(value);8}
    All 3 passes — pass 1 is the card above
    passvalue
    10
    21
    32

  3. drained ← 0, empty_hits ← 0, status ← "idle"

    8    }9    let mut draine→ 0d = 0;10    let mut empty_hit→ 0s = 0;11    for _ in 0..attempts {12        match rx.try_recv() {13            Ok(_) => drained += 1,14            Err(_) => empty_hits += 1,15        }16    }17    let statu→ "idle"s = if draine0d == 0 {18        "idle"19    } else if empty_hit0s > 0 {20        "drained"21    } else {22        "draining"23    };24    println!("attempts={attempts} drained={drained} empty={empty_hits} {status}");25}
    outputattempts=0 drained=0 empty=0 idle

  1. attempts ← 5

    3fn main() {4    let attempt→ 5s = 5;5    let (tx, rx) = channel();6    for value in 0..3 {

  2. for value in 0..3

    pass 1 of 3
    5let (tx, rx) = channel();6for valu0e in 0..3 {7    let _ = tx.send(value);8}
    All 3 passes — pass 1 is the card above
    passvalue
    10
    21
    32

  3. drained ← 0, empty_hits ← 0

    8}9let mut draine→ 0d = 0;10let mut empty_hit→ 0s = 0;11for _ in 0..attempts {

  4. for _ in 0..attempts

    pass 1 of 5
    10let mut empty_hits = 0;11for _ in 0..attempt5s {12    match rx.try_recv() {

  5. status ← "drained"

    16    }17    let statu→ "drained"s = if draine3d == 0 {18        "idle"19    } else if empty_hit2s > 0 {20        "drained"21    } else {22        "draining"23    };24    println!("attempts={attempts} drained={drained} empty={empty_hits} {status}");25}
    outputattempts=5 drained=3 empty=2 drained
nonblocking receive `try_recv` returns `Ok` while messages remain and `Err` once the queue is empty, so it never blocks the caller.
drain report Counting `Ok` results reports how many buffered messages were drained by the selected number of attempts.
empty signal Extra attempts past the buffered messages report empty receives instead of waiting for new data.