Struct Ring 

pub struct Ring { /* private fields */ }

Sparse directed message rings packed into collective RMA windows.

Safe to share between threads on the same process. Lifetime is the underlying MPI window To drop the ring use Self::close (collective) or let the destructor run symmetrically on every rank.

Implementations

impl Ring

pub fn safe<C: Communicator + ?Sized>( comm: &C, rings: &[(Rank, Rank, usize, usize)], ) -> Result<Self, Error>

Construct a safe (overwrite-gated) ring. Collective over comm.

rings names the active directed lanes, one (source, destination, depth, capacity) tuple per lane: depth is the number of fixed slots the lane holds, capacity the payload bytes one slot fits. Every rank passes the same list.

Each source lane gets a cumulative-acknowledgement counter at the source The senders spin on yield_now until the receiver has acked enough earlier messages to make room in the slot ring. Requires a unified window memory model otherwise contruction fails

Errors

• Error::Intercommunicator if comm is an intercommunicator.
• Error::Ring for configuration disagreement, invalid or repeated lanes, or zero depth or capacity.
• Error::Window if the window uses a separate memory model, which cannot support the local polling the ring relies on.
• Plus whatever CommunicatorRmaExt::allocate_window returns for the slot and counter windows.

Examples found in repository

examples/test_layout.rs (line 23)

fn run(world: &SimpleCommunicator, safe: bool) {
    let rank = world.rank();
    let ring = if safe {
        Ring::safe(world, &lanes(2, 8)).unwrap()
    } else {
        Ring::raw(world, &lanes(2, 8)).unwrap()
    };

    if rank < 2 {
        ring.send(2, &[rank as u8, 1]).unwrap();
    }
    world.barrier();

    if rank == 2 {
        let messages = ring.poll().unwrap();
        assert_eq!(messages.len(), 2);
        assert_eq!(messages[0].origin, 0);
        assert_eq!(messages[0].data, vec![0, 1]);
        assert_eq!(messages[1].origin, 1);
        assert_eq!(messages[1].data, vec![1, 1]);
        ring.ack(0, 1).unwrap();
        ring.ack(1, 1).unwrap();
        ring.send(0, &[2, 1]).unwrap();
        ring.send(1, &[2, 1]).unwrap();
    }
    world.barrier();

    if rank < 2 {
        let messages = ring.poll().unwrap();
        assert_eq!(
            messages,
            vec![Message {
                origin: 2,
                sequence: 1,
                data: vec![2, 1]
            }]
        );
        ring.ack(2, 1).unwrap();
    }
    world.barrier();

    ring.close().unwrap();
}

examples/test_ring.rs (line 25)

fn safe_basics(world: &SimpleCommunicator, rank: Rank, next: Rank, prev: Rank) {
    // Depth 4 against 3 messages: this covers ordering and acknowledge
    // validation with the overwrite gate deliberately out of the way.
    // Backpressure is its own test, on a one-way lane.
    let ring = Ring::safe(world, &pair(4, 8)).unwrap();
    assert!(ring.is_safe());
    assert_eq!(ring.depth(next), Some(4));
    assert_eq!(ring.capacity(next), Some(8));

    assert!(matches!(
        ring.send(next, &[0; 9]),
        Err(Error::Payload {
            len: 9,
            capacity: 8
        })
    ));
    assert!(matches!(
        ring.ack(prev, 1),
        Err(Error::Ack {
            sequence: 1,
            received: 0,
            ..
        })
    ));

    for s in 1..=3u8 {
        assert_eq!(ring.send(next, &[rank as u8, s]).unwrap(), u64::from(s));
    }
    world.barrier();

    let messages = ring.poll().unwrap();
    assert_eq!(
        messages,
        (1..=3)
            .map(|s| Message {
                origin: prev,
                sequence: s,
                data: vec![prev as u8, s as u8],
            })
            .collect::<Vec<_>>()
    );
    assert!(ring.poll().unwrap().is_empty());
    assert_eq!(ring.lost(), 0);
    assert_eq!(ring.max_lag(), 3);

    assert!(matches!(
        ring.ack(prev, 4),
        Err(Error::Ack {
            sequence: 4,
            received: 3,
            ..
        })
    ));
    ring.ack(prev, 3).unwrap();
    ring.ack(prev, 3).unwrap();
    ring.ack(prev, 2).unwrap();

    world.barrier();
    ring.close().unwrap();
}

/// A safe sender with no free slot blocks until the receiver acknowledges.
///
/// One-way, so the ack timing belongs to the test. On a symmetric pair each
/// rank's ack races the other's gate check: whichever receiver drains first
/// opens its peer's gate before that peer ever evaluates it, and neither side
/// can be relied on to block.
fn backpressure(world: &SimpleCommunicator, rank: Rank) {
    let ring = Ring::safe(world, &oneway(2, 8)).unwrap();
    if rank == 0 {
        assert_eq!(ring.send(1, &[1]).unwrap(), 1);
        assert_eq!(ring.send(1, &[2]).unwrap(), 2);
    }
    world.barrier();

    if rank == 0 {
        let sent = thread::scope(|scope| {
            let sender = &ring;
            let handle = scope.spawn(move || sender.send(1, &[3]));
            // Depth is 2 and nothing has been acknowledged, so there is no slot
            // for the third message and the send has to wait.
            let deadline = Instant::now() + Duration::from_secs(5);
            while ring.waits() == 0 {
                assert!(Instant::now() < deadline, "safe sender never blocked");
                thread::yield_now();
            }
            // The receiver is parked here, so it cannot have drained early.
            world.barrier();
            handle.join().expect("sender thread panicked").unwrap()
        });
        assert_eq!(sent, 3);
        assert!(ring.waits() > 0);
        assert!(ring.wait_ns() > 0);
    } else {
        world.barrier();
        let messages = ring.poll().unwrap();
        assert_eq!(messages.len(), 2);
        assert_eq!(messages[0].data, vec![1]);
        assert_eq!(messages[1].data, vec![2]);
        assert!(matches!(
            ring.ack(0, 3),
            Err(Error::Ack {
                sequence: 3,
                received: 2,
                ..
            })
        ));
        ring.ack(0, 2).unwrap();
        ring.ack(0, 2).unwrap();
    }

    world.barrier();
    if rank == 1 {
        let messages = ring.poll().unwrap();
        assert_eq!(messages.len(), 1);
        assert_eq!(messages[0].sequence, 3);
        assert_eq!(messages[0].data, vec![3]);
        ring.ack(0, 3).unwrap();
    }
    world.barrier();
    ring.close().unwrap();
}

fn oneway_basics(world: &SimpleCommunicator, rank: Rank) {
    let ring = Ring::safe(world, &oneway(2, 8)).unwrap();
    if rank == 0 {
        assert_eq!(ring.send(1, &[7]).unwrap(), 1);
    }
    world.barrier();
    if rank == 1 {
        let messages = ring.poll().unwrap();
        assert_eq!(messages.len(), 1);
        assert_eq!(messages[0].origin, 0);
        assert_eq!(messages[0].data, vec![7]);
        ring.ack(0, 1).unwrap();
    }
    world.barrier();
    ring.close().unwrap();
}

fn raw_basics(world: &SimpleCommunicator, rank: Rank, next: Rank, prev: Rank) {
    let ring = Ring::raw(world, &pair(2, 8)).unwrap();
    assert!(!ring.is_safe());

    for s in 1..=3u8 {
        assert_eq!(ring.send(next, &[rank as u8, s]).unwrap(), u64::from(s));
    }
    world.barrier();

    let messages = ring.poll().unwrap();
    assert_eq!(
        messages,
        vec![
            Message {
                origin: prev,
                sequence: 2,
                data: vec![prev as u8, 2]
            },
            Message {
                origin: prev,
                sequence: 3,
                data: vec![prev as u8, 3]
            },
        ]
    );
    assert_eq!(ring.lost(), 1);

    // Raw acks are no-ops.
    ring.ack(prev, u64::MAX).unwrap();
    world.barrier();
    ring.close().unwrap();
}

fn main() {
    let (universe, provided) =
        mpi::initialize_with_threading(Threading::Multiple).expect("MPI must initialize once");
    assert_eq!(provided, Threading::Multiple);
    let world = universe.world();
    let rank = world.rank();
    let size = world.size();
    assert_eq!(size, 2);
    let next = (rank + 1) % size;
    let prev = (rank + size - 1) % size;

    // Configuration must agree across ranks.
    if rank == 0 {
        assert!(matches!(
            Ring::safe(&world, &pair(1, 8)),
            Err(Error::Ring("configuration differs between ranks"))
        ));
    } else {
        assert!(matches!(
            Ring::safe(&world, &pair(2, 8)),
            Err(Error::Ring("configuration differs between ranks"))
        ));
    }
    assert!(matches!(
        Ring::raw(&world, &pair(0, 8)),
        Err(Error::Ring("depth must be positive"))
    ));

    safe_basics(&world, rank, next, prev);
    oneway_basics(&world, rank);
    backpressure(&world, rank);
    raw_basics(&world, rank, next, prev);

    if rank == 0 {
        println!("test_ring: ok");
    }
}

examples/compare.rs (line 410)

fn main() {
    let mut args = std::env::args().skip(1);
    let messages: u64 = args.next().and_then(|s| s.parse().ok()).unwrap_or(20_000);
    let depth: usize = args.next().and_then(|s| s.parse().ok()).unwrap_or(32);
    let noise_kib: usize = args.next().and_then(|s| s.parse().ok()).unwrap_or(0);
    let repeats: u32 = args.next().and_then(|s| s.parse().ok()).unwrap_or(5);

    let (mut universe, provided) =
        mpi::initialize_with_threading(Threading::Multiple).expect("MPI must initialize once");
    assert_eq!(provided, Threading::Multiple);
    let world = universe.world();
    let rank = world.rank();
    let size = world.size();
    assert!(size >= 2, "compare needs at least 2 ranks");
    assert!(noise_kib == 0 || size >= 4, "noise needs at least 4 ranks");
    // `p2p-bsend` needs somewhere to copy to. Only rank 0 sends buffered, so
    // only it attaches the buffer; every rank allocating 512 MiB would need
    // tens of GiB at scale. Sized past the largest amount a stream can leave
    // in flight, so the comparison measures the transport and not a buffer
    // running out.
    if rank == 0 {
        universe.set_buffer_size(BSEND_BUF);
    }

    let stop = Arc::new(AtomicBool::new(false));
    let noisy = noise_kib > 0;

    std::thread::scope(|scope| {
        if noisy && rank >= 2 {
            let stop = Arc::clone(&stop);
            let shared = Shared(&world);
            scope.spawn(move || {
                let shared = shared;
                noise(shared.0, noise_kib, &stop);
            });
        }
        if rank > 1 {
            // Noise ranks take no part in the measurement. They wait here for
            // rank 0 to call time, then let their noise thread finish.
            world
                .process_at_rank(0)
                .receive_vec_with_tag::<u8>(TAG_QUIT);
            stop.store(true, Ordering::Relaxed);
            return;
        }

        if rank == 0 {
            println!("{HEADER}");
        }

        // The measured pair only. A world barrier here would deadlock against
        // ranks sitting in the noise loop.
        let pair: Vec<Rank> = vec![0, 1];
        let group = world.group().include(&pair);
        let duo = world
            .split_by_subgroup(&group)
            .expect("ranks 0 and 1 form the measured pair");

        for rep in 0..repeats {
            for payload in SIZES {
                // Both ring modes share the lane shape, so the only difference
                // between the two rows is the acknowledgement gate.
                let lanes = vec![
                    (0 as Rank, 1 as Rank, depth, payload),
                    (1, 0, depth, payload),
                ];
                // Cap the stream by bytes as well as count: 256 KiB x 20k would be
                // 5 GiB through the wire for one row.
                let n = messages.min((STREAM_BYTES / payload as u64).max(64));
                let iters = if payload >= 32768 { 500 } else { 2000 };

                for (name, safe) in [("ring-safe", true), ("ring-raw", false)] {
                    let ring = if safe {
                        Ring::safe(&duo, &lanes).unwrap()
                    } else {
                        Ring::raw(&duo, &lanes).unwrap()
                    };
                    let (sent, got, inject, total, wait) =
                        ring_stream(&duo, &ring, rank, rep, payload, n);
                    if rank == 0 {
                        Row {
                            transport: name,
                            measure: "stream",
                            payload,
                            sent,
                            delivered: got,
                            inject,
                            total,
                            wait,
                            max_us: 0.0,
                        }
                        .print(size, depth, noise_kib, rep);
                    }
                    duo.barrier();
                    let stalled = ring.wait_ns();
                    let (dt, max_us) = ring_rtt(&ring, rank, rep, payload, iters);
                    if rank == 0 {
                        Row {
                            transport: name,
                            measure: "rtt",
                            payload,
                            sent: iters * 2,
                            delivered: iters * 2,
                            inject: dt,
                            total: dt,
                            wait: (ring.wait_ns() - stalled) as f64 * 1e-9,
                            max_us,
                        }
                        .print(size, depth, noise_kib, rep);
                    }
                    duo.barrier();
                    ring.close().unwrap();
                }

                for (name, buffered) in [("p2p-send", false), ("p2p-bsend", true)] {
                    let (sent, got, inject, total) =
                        p2p_stream(&duo, rank, rep, payload, n, buffered);
                    if rank == 0 {
                        Row {
                            transport: name,
                            measure: "stream",
                            payload,
                            sent,
                            delivered: got,
                            inject,
                            total,
                            wait: 0.0,
                            max_us: 0.0,
                        }
                        .print(size, depth, noise_kib, rep);
                    }
                    duo.barrier();
                }
                let (dt, max_us) = p2p_rtt(&duo, rank, rep, payload, iters);
                if rank == 0 {
                    Row {
                        transport: "p2p-send",
                        measure: "rtt",
                        payload,
                        sent: iters * 2,
                        delivered: iters * 2,
                        inject: dt,
                        total: dt,
                        wait: 0.0,
                        max_us,
                    }
                    .print(size, depth, noise_kib, rep);
                }
                duo.barrier();
            }
        }

        if rank == 0 {
            for r in 2..size {
                world.process_at_rank(r).send_with_tag(&[0u8], TAG_QUIT);
            }
        }
    });

    // Buffered sends must be drained before MPI tears the buffer down.
    world.barrier();
}

examples/soak.rs (line 159)

fn main() {
    let mut args = std::env::args().skip(1);
    let mode = args.next().unwrap_or_else(|| "safe".into());
    let messages: u64 = args.next().and_then(|s| s.parse().ok()).unwrap_or(20_000);
    let payload: usize = args.next().and_then(|s| s.parse().ok()).unwrap_or(256);
    let depth: usize = args.next().and_then(|s| s.parse().ok()).unwrap_or(8);
    let noise_kib: usize = args.next().and_then(|s| s.parse().ok()).unwrap_or(0);
    let pace_ns: u64 = args.next().and_then(|s| s.parse().ok()).unwrap_or(0);
    // Echoed straight into the output so repeats of one point stay distinguishable.
    let rep: u32 = args.next().and_then(|s| s.parse().ok()).unwrap_or(0);
    assert!(
        payload >= STAMP,
        "payload must hold the origin and sequence"
    );
    let safe = match mode.as_str() {
        "safe" => true,
        "raw" => false,
        other => panic!("mode must be safe or raw, got {other:?}"),
    };

    let (universe, provided) =
        mpi::initialize_with_threading(Threading::Multiple).expect("MPI must initialize once");
    assert_eq!(provided, Threading::Multiple);
    let world = universe.world();
    let rank = world.rank();
    let size = world.size();
    assert!(size >= 2, "soak needs at least 2 ranks");

    // All-to-all: every ordered pair of distinct ranks gets a lane.
    let mut lanes = Vec::new();
    for source in 0..size {
        for destination in 0..size {
            if source != destination {
                lanes.push((source, destination, depth, payload));
            }
        }
    }
    let ring = Arc::new(if safe {
        Ring::safe(&world, &lanes).unwrap()
    } else {
        Ring::raw(&world, &lanes).unwrap()
    });

    let stop = Arc::new(AtomicBool::new(false));
    let done = Arc::new(AtomicBool::new(false));
    let count = Arc::new(AtomicU64::new(0));
    let peers = (size - 1) as u64;
    let sent = messages * peers;

    world.barrier();
    let started = std::time::Instant::now();

    let outcome = std::thread::scope(|scope| {
        if noise_kib > 0 {
            let stop = Arc::clone(&stop);
            let shared = Shared(&world);
            scope.spawn(move || {
                // Name the wrapper as a whole first. Closures capture the
                // narrowest place they use, and `shared.0` alone would be
                // captured as a bare reference, losing the `Send` impl.
                let shared = shared;
                noise(shared.0, noise_kib, &stop);
            });
        }

        let poller = {
            let (ring, done, count) = (Arc::clone(&ring), Arc::clone(&done), Arc::clone(&count));
            scope.spawn(move || -> Result<Vec<u64>, String> {
                let mut seen = vec![0u64; size as usize];
                let mut total = 0u64;
                loop {
                    // If true, this poll follows the global send barrier.
                    let finished = done.load(Ordering::Acquire);
                    let batch = ring.poll().map_err(|e| format!("poll: {e}"))?;
                    let empty = batch.is_empty();
                    let mut furthest: HashMap<Rank, u64> = HashMap::new();
                    for m in &batch {
                        verify(m)?;
                        let previous = seen[m.origin as usize];
                        if m.sequence <= previous {
                            return Err(format!(
                                "sequence from {} went backwards: {} after {previous}",
                                m.origin, m.sequence
                            ));
                        }
                        if safe && m.sequence != previous + 1 {
                            return Err(format!(
                                "safe lane from {} skipped {} to {}",
                                m.origin,
                                previous + 1,
                                m.sequence
                            ));
                        }
                        seen[m.origin as usize] = m.sequence;
                        let f = furthest.entry(m.origin).or_insert(0);
                        *f = (*f).max(m.sequence);
                        total += 1;
                    }
                    for (origin, sequence) in furthest {
                        ring.ack(origin, sequence)
                            .map_err(|e| format!("ack to {origin}: {e}"))?;
                    }
                    count.store(total, Ordering::Relaxed);
                    if finished && empty {
                        return Ok(seen);
                    }
                    if empty {
                        std::thread::yield_now();
                    }
                }
            })
        };

        // Round-robin over peers so every outgoing lane stays hot at once.
        let mut buf = vec![0u8; payload];
        let pace = std::time::Duration::from_nanos(pace_ns);
        let opened = std::time::Instant::now();
        for sequence in 1..=messages {
            for destination in (0..size).filter(|&r| r != rank) {
                paint(rank, sequence, &mut buf);
                ring.send(destination, &buf)
                    .unwrap_or_else(|e| panic!("rank {rank}: send to {destination}: {e}"));
            }
            // Absolute deadlines, so a slow round is not paid for twice and the
            // offered rate stays the one that was asked for.
            if pace_ns > 0 {
                let due = opened + pace * sequence as u32;
                while std::time::Instant::now() < due {
                    std::hint::spin_loop();
                }
            }
        }
        // Every rank's sends have completed at their targets before this
        // returns, so after the barrier nothing new can appear in a slot.
        world.barrier();
        done.store(true, Ordering::Release);

        let seen = poller.join().expect("poller thread panicked");
        stop.store(true, Ordering::Relaxed);
        seen
    });

    let elapsed = started.elapsed().as_secs_f64();
    let fail = |why: String| -> ! {
        eprintln!("[soak] FAIL rank {rank}: {why}");
        world.abort(1)
    };
    let seen = outcome.unwrap_or_else(|why| fail(why));

    let got = count.load(Ordering::Relaxed);
    let lost = ring.lost();
    let corrupt = ring.corrupt();

    // The last message on a lane is never overwritten, so every lane must end
    // exactly at the sender's high-water mark in both modes.
    for origin in (0..size).filter(|&r| r != rank) {
        if seen[origin as usize] != messages {
            fail(format!(
                "lane from {origin} ended at {} not {messages}",
                seen[origin as usize]
            ));
        }
    }
    if safe && lost != 0 {
        fail(format!("safe ring lost {lost}"));
    }
    if got + lost != sent {
        fail(format!("received {got} + lost {lost} != sent {sent}"));
    }

    // Gather the counters and let rank 0 write every row. Interleaving the
    // ranks' own stdout would scatter the header into the middle of the table:
    // mpirun orders a rank's output against itself and nothing else.
    let mine = [
        got,
        lost,
        corrupt,
        ring.max_lag(),
        ring.waits(),
        ring.wait_ns(),
    ];
    let mut all = vec![0u64; mine.len() * size as usize];
    world.all_gather_into(&mine[..], &mut all[..]);
    if rank == 0 {
        println!(
            "mode\tranks\tdepth\tpayload\tnoise_kib\tpace_ns\trep\trank\tsent\treceived\tlost\tcorrupt\tmax_lag\twaits\twait_s\tseconds"
        );
        for (r, row) in all.chunks_exact(mine.len()).enumerate() {
            println!(
                "{mode}\t{size}\t{depth}\t{payload}\t{noise_kib}\t{pace_ns}\t{rep}\t{r}\t{sent}\t{}\t{}\t{}\t{}\t{}\t{:.6}\t{elapsed:.6}",
                row[0],
                row[1],
                row[2],
                row[3],
                row[4],
                row[5] as f64 * 1e-9,
            );
        }
    }

    world.barrier();
    Arc::try_unwrap(ring)
        .unwrap_or_else(|_| panic!("ring still shared at close"))
        .close()
        .unwrap();
    if rank == 0 {
        eprintln!("[soak] ok: {mode}, {size} ranks, {sent} messages/rank in {elapsed:.2}s");
    }
}

pub fn raw<C: Communicator + ?Sized>( comm: &C, rings: &[(Rank, Rank, usize, usize)], ) -> Result<Self, Error>

Construct a raw (overwrite-capable) ring. Collective over comm.

rings has the same shape as for Self::safe. No acknowledgement counter: unread slots are silently overwritten when the depth is exhausted, and the receiver reports the gap via Self::lost. The sender never blocks on a slow peer. Requires a unified window memory model.

Errors

Same conditions as Self::safe.

Examples found in repository

examples/test_ring.rs (line 162)

fn raw_basics(world: &SimpleCommunicator, rank: Rank, next: Rank, prev: Rank) {
    let ring = Ring::raw(world, &pair(2, 8)).unwrap();
    assert!(!ring.is_safe());

    for s in 1..=3u8 {
        assert_eq!(ring.send(next, &[rank as u8, s]).unwrap(), u64::from(s));
    }
    world.barrier();

    let messages = ring.poll().unwrap();
    assert_eq!(
        messages,
        vec![
            Message {
                origin: prev,
                sequence: 2,
                data: vec![prev as u8, 2]
            },
            Message {
                origin: prev,
                sequence: 3,
                data: vec![prev as u8, 3]
            },
        ]
    );
    assert_eq!(ring.lost(), 1);

    // Raw acks are no-ops.
    ring.ack(prev, u64::MAX).unwrap();
    world.barrier();
    ring.close().unwrap();
}

fn main() {
    let (universe, provided) =
        mpi::initialize_with_threading(Threading::Multiple).expect("MPI must initialize once");
    assert_eq!(provided, Threading::Multiple);
    let world = universe.world();
    let rank = world.rank();
    let size = world.size();
    assert_eq!(size, 2);
    let next = (rank + 1) % size;
    let prev = (rank + size - 1) % size;

    // Configuration must agree across ranks.
    if rank == 0 {
        assert!(matches!(
            Ring::safe(&world, &pair(1, 8)),
            Err(Error::Ring("configuration differs between ranks"))
        ));
    } else {
        assert!(matches!(
            Ring::safe(&world, &pair(2, 8)),
            Err(Error::Ring("configuration differs between ranks"))
        ));
    }
    assert!(matches!(
        Ring::raw(&world, &pair(0, 8)),
        Err(Error::Ring("depth must be positive"))
    ));

    safe_basics(&world, rank, next, prev);
    oneway_basics(&world, rank);
    backpressure(&world, rank);
    raw_basics(&world, rank, next, prev);

    if rank == 0 {
        println!("test_ring: ok");
    }
}

examples/test_layout.rs (line 25)

fn run(world: &SimpleCommunicator, safe: bool) {
    let rank = world.rank();
    let ring = if safe {
        Ring::safe(world, &lanes(2, 8)).unwrap()
    } else {
        Ring::raw(world, &lanes(2, 8)).unwrap()
    };

    if rank < 2 {
        ring.send(2, &[rank as u8, 1]).unwrap();
    }
    world.barrier();

    if rank == 2 {
        let messages = ring.poll().unwrap();
        assert_eq!(messages.len(), 2);
        assert_eq!(messages[0].origin, 0);
        assert_eq!(messages[0].data, vec![0, 1]);
        assert_eq!(messages[1].origin, 1);
        assert_eq!(messages[1].data, vec![1, 1]);
        ring.ack(0, 1).unwrap();
        ring.ack(1, 1).unwrap();
        ring.send(0, &[2, 1]).unwrap();
        ring.send(1, &[2, 1]).unwrap();
    }
    world.barrier();

    if rank < 2 {
        let messages = ring.poll().unwrap();
        assert_eq!(
            messages,
            vec![Message {
                origin: 2,
                sequence: 1,
                data: vec![2, 1]
            }]
        );
        ring.ack(2, 1).unwrap();
    }
    world.barrier();

    ring.close().unwrap();
}

examples/compare.rs (line 412)

fn main() {
    let mut args = std::env::args().skip(1);
    let messages: u64 = args.next().and_then(|s| s.parse().ok()).unwrap_or(20_000);
    let depth: usize = args.next().and_then(|s| s.parse().ok()).unwrap_or(32);
    let noise_kib: usize = args.next().and_then(|s| s.parse().ok()).unwrap_or(0);
    let repeats: u32 = args.next().and_then(|s| s.parse().ok()).unwrap_or(5);

    let (mut universe, provided) =
        mpi::initialize_with_threading(Threading::Multiple).expect("MPI must initialize once");
    assert_eq!(provided, Threading::Multiple);
    let world = universe.world();
    let rank = world.rank();
    let size = world.size();
    assert!(size >= 2, "compare needs at least 2 ranks");
    assert!(noise_kib == 0 || size >= 4, "noise needs at least 4 ranks");
    // `p2p-bsend` needs somewhere to copy to. Only rank 0 sends buffered, so
    // only it attaches the buffer; every rank allocating 512 MiB would need
    // tens of GiB at scale. Sized past the largest amount a stream can leave
    // in flight, so the comparison measures the transport and not a buffer
    // running out.
    if rank == 0 {
        universe.set_buffer_size(BSEND_BUF);
    }

    let stop = Arc::new(AtomicBool::new(false));
    let noisy = noise_kib > 0;

    std::thread::scope(|scope| {
        if noisy && rank >= 2 {
            let stop = Arc::clone(&stop);
            let shared = Shared(&world);
            scope.spawn(move || {
                let shared = shared;
                noise(shared.0, noise_kib, &stop);
            });
        }
        if rank > 1 {
            // Noise ranks take no part in the measurement. They wait here for
            // rank 0 to call time, then let their noise thread finish.
            world
                .process_at_rank(0)
                .receive_vec_with_tag::<u8>(TAG_QUIT);
            stop.store(true, Ordering::Relaxed);
            return;
        }

        if rank == 0 {
            println!("{HEADER}");
        }

        // The measured pair only. A world barrier here would deadlock against
        // ranks sitting in the noise loop.
        let pair: Vec<Rank> = vec![0, 1];
        let group = world.group().include(&pair);
        let duo = world
            .split_by_subgroup(&group)
            .expect("ranks 0 and 1 form the measured pair");

        for rep in 0..repeats {
            for payload in SIZES {
                // Both ring modes share the lane shape, so the only difference
                // between the two rows is the acknowledgement gate.
                let lanes = vec![
                    (0 as Rank, 1 as Rank, depth, payload),
                    (1, 0, depth, payload),
                ];
                // Cap the stream by bytes as well as count: 256 KiB x 20k would be
                // 5 GiB through the wire for one row.
                let n = messages.min((STREAM_BYTES / payload as u64).max(64));
                let iters = if payload >= 32768 { 500 } else { 2000 };

                for (name, safe) in [("ring-safe", true), ("ring-raw", false)] {
                    let ring = if safe {
                        Ring::safe(&duo, &lanes).unwrap()
                    } else {
                        Ring::raw(&duo, &lanes).unwrap()
                    };
                    let (sent, got, inject, total, wait) =
                        ring_stream(&duo, &ring, rank, rep, payload, n);
                    if rank == 0 {
                        Row {
                            transport: name,
                            measure: "stream",
                            payload,
                            sent,
                            delivered: got,
                            inject,
                            total,
                            wait,
                            max_us: 0.0,
                        }
                        .print(size, depth, noise_kib, rep);
                    }
                    duo.barrier();
                    let stalled = ring.wait_ns();
                    let (dt, max_us) = ring_rtt(&ring, rank, rep, payload, iters);
                    if rank == 0 {
                        Row {
                            transport: name,
                            measure: "rtt",
                            payload,
                            sent: iters * 2,
                            delivered: iters * 2,
                            inject: dt,
                            total: dt,
                            wait: (ring.wait_ns() - stalled) as f64 * 1e-9,
                            max_us,
                        }
                        .print(size, depth, noise_kib, rep);
                    }
                    duo.barrier();
                    ring.close().unwrap();
                }

                for (name, buffered) in [("p2p-send", false), ("p2p-bsend", true)] {
                    let (sent, got, inject, total) =
                        p2p_stream(&duo, rank, rep, payload, n, buffered);
                    if rank == 0 {
                        Row {
                            transport: name,
                            measure: "stream",
                            payload,
                            sent,
                            delivered: got,
                            inject,
                            total,
                            wait: 0.0,
                            max_us: 0.0,
                        }
                        .print(size, depth, noise_kib, rep);
                    }
                    duo.barrier();
                }
                let (dt, max_us) = p2p_rtt(&duo, rank, rep, payload, iters);
                if rank == 0 {
                    Row {
                        transport: "p2p-send",
                        measure: "rtt",
                        payload,
                        sent: iters * 2,
                        delivered: iters * 2,
                        inject: dt,
                        total: dt,
                        wait: 0.0,
                        max_us,
                    }
                    .print(size, depth, noise_kib, rep);
                }
                duo.barrier();
            }
        }

        if rank == 0 {
            for r in 2..size {
                world.process_at_rank(r).send_with_tag(&[0u8], TAG_QUIT);
            }
        }
    });

    // Buffered sends must be drained before MPI tears the buffer down.
    world.barrier();
}

examples/soak.rs (line 161)

fn main() {
    let mut args = std::env::args().skip(1);
    let mode = args.next().unwrap_or_else(|| "safe".into());
    let messages: u64 = args.next().and_then(|s| s.parse().ok()).unwrap_or(20_000);
    let payload: usize = args.next().and_then(|s| s.parse().ok()).unwrap_or(256);
    let depth: usize = args.next().and_then(|s| s.parse().ok()).unwrap_or(8);
    let noise_kib: usize = args.next().and_then(|s| s.parse().ok()).unwrap_or(0);
    let pace_ns: u64 = args.next().and_then(|s| s.parse().ok()).unwrap_or(0);
    // Echoed straight into the output so repeats of one point stay distinguishable.
    let rep: u32 = args.next().and_then(|s| s.parse().ok()).unwrap_or(0);
    assert!(
        payload >= STAMP,
        "payload must hold the origin and sequence"
    );
    let safe = match mode.as_str() {
        "safe" => true,
        "raw" => false,
        other => panic!("mode must be safe or raw, got {other:?}"),
    };

    let (universe, provided) =
        mpi::initialize_with_threading(Threading::Multiple).expect("MPI must initialize once");
    assert_eq!(provided, Threading::Multiple);
    let world = universe.world();
    let rank = world.rank();
    let size = world.size();
    assert!(size >= 2, "soak needs at least 2 ranks");

    // All-to-all: every ordered pair of distinct ranks gets a lane.
    let mut lanes = Vec::new();
    for source in 0..size {
        for destination in 0..size {
            if source != destination {
                lanes.push((source, destination, depth, payload));
            }
        }
    }
    let ring = Arc::new(if safe {
        Ring::safe(&world, &lanes).unwrap()
    } else {
        Ring::raw(&world, &lanes).unwrap()
    });

    let stop = Arc::new(AtomicBool::new(false));
    let done = Arc::new(AtomicBool::new(false));
    let count = Arc::new(AtomicU64::new(0));
    let peers = (size - 1) as u64;
    let sent = messages * peers;

    world.barrier();
    let started = std::time::Instant::now();

    let outcome = std::thread::scope(|scope| {
        if noise_kib > 0 {
            let stop = Arc::clone(&stop);
            let shared = Shared(&world);
            scope.spawn(move || {
                // Name the wrapper as a whole first. Closures capture the
                // narrowest place they use, and `shared.0` alone would be
                // captured as a bare reference, losing the `Send` impl.
                let shared = shared;
                noise(shared.0, noise_kib, &stop);
            });
        }

        let poller = {
            let (ring, done, count) = (Arc::clone(&ring), Arc::clone(&done), Arc::clone(&count));
            scope.spawn(move || -> Result<Vec<u64>, String> {
                let mut seen = vec![0u64; size as usize];
                let mut total = 0u64;
                loop {
                    // If true, this poll follows the global send barrier.
                    let finished = done.load(Ordering::Acquire);
                    let batch = ring.poll().map_err(|e| format!("poll: {e}"))?;
                    let empty = batch.is_empty();
                    let mut furthest: HashMap<Rank, u64> = HashMap::new();
                    for m in &batch {
                        verify(m)?;
                        let previous = seen[m.origin as usize];
                        if m.sequence <= previous {
                            return Err(format!(
                                "sequence from {} went backwards: {} after {previous}",
                                m.origin, m.sequence
                            ));
                        }
                        if safe && m.sequence != previous + 1 {
                            return Err(format!(
                                "safe lane from {} skipped {} to {}",
                                m.origin,
                                previous + 1,
                                m.sequence
                            ));
                        }
                        seen[m.origin as usize] = m.sequence;
                        let f = furthest.entry(m.origin).or_insert(0);
                        *f = (*f).max(m.sequence);
                        total += 1;
                    }
                    for (origin, sequence) in furthest {
                        ring.ack(origin, sequence)
                            .map_err(|e| format!("ack to {origin}: {e}"))?;
                    }
                    count.store(total, Ordering::Relaxed);
                    if finished && empty {
                        return Ok(seen);
                    }
                    if empty {
                        std::thread::yield_now();
                    }
                }
            })
        };

        // Round-robin over peers so every outgoing lane stays hot at once.
        let mut buf = vec![0u8; payload];
        let pace = std::time::Duration::from_nanos(pace_ns);
        let opened = std::time::Instant::now();
        for sequence in 1..=messages {
            for destination in (0..size).filter(|&r| r != rank) {
                paint(rank, sequence, &mut buf);
                ring.send(destination, &buf)
                    .unwrap_or_else(|e| panic!("rank {rank}: send to {destination}: {e}"));
            }
            // Absolute deadlines, so a slow round is not paid for twice and the
            // offered rate stays the one that was asked for.
            if pace_ns > 0 {
                let due = opened + pace * sequence as u32;
                while std::time::Instant::now() < due {
                    std::hint::spin_loop();
                }
            }
        }
        // Every rank's sends have completed at their targets before this
        // returns, so after the barrier nothing new can appear in a slot.
        world.barrier();
        done.store(true, Ordering::Release);

        let seen = poller.join().expect("poller thread panicked");
        stop.store(true, Ordering::Relaxed);
        seen
    });

    let elapsed = started.elapsed().as_secs_f64();
    let fail = |why: String| -> ! {
        eprintln!("[soak] FAIL rank {rank}: {why}");
        world.abort(1)
    };
    let seen = outcome.unwrap_or_else(|why| fail(why));

    let got = count.load(Ordering::Relaxed);
    let lost = ring.lost();
    let corrupt = ring.corrupt();

    // The last message on a lane is never overwritten, so every lane must end
    // exactly at the sender's high-water mark in both modes.
    for origin in (0..size).filter(|&r| r != rank) {
        if seen[origin as usize] != messages {
            fail(format!(
                "lane from {origin} ended at {} not {messages}",
                seen[origin as usize]
            ));
        }
    }
    if safe && lost != 0 {
        fail(format!("safe ring lost {lost}"));
    }
    if got + lost != sent {
        fail(format!("received {got} + lost {lost} != sent {sent}"));
    }

    // Gather the counters and let rank 0 write every row. Interleaving the
    // ranks' own stdout would scatter the header into the middle of the table:
    // mpirun orders a rank's output against itself and nothing else.
    let mine = [
        got,
        lost,
        corrupt,
        ring.max_lag(),
        ring.waits(),
        ring.wait_ns(),
    ];
    let mut all = vec![0u64; mine.len() * size as usize];
    world.all_gather_into(&mine[..], &mut all[..]);
    if rank == 0 {
        println!(
            "mode\tranks\tdepth\tpayload\tnoise_kib\tpace_ns\trep\trank\tsent\treceived\tlost\tcorrupt\tmax_lag\twaits\twait_s\tseconds"
        );
        for (r, row) in all.chunks_exact(mine.len()).enumerate() {
            println!(
                "{mode}\t{size}\t{depth}\t{payload}\t{noise_kib}\t{pace_ns}\t{rep}\t{r}\t{sent}\t{}\t{}\t{}\t{}\t{}\t{:.6}\t{elapsed:.6}",
                row[0],
                row[1],
                row[2],
                row[3],
                row[4],
                row[5] as f64 * 1e-9,
            );
        }
    }

    world.barrier();
    Arc::try_unwrap(ring)
        .unwrap_or_else(|_| panic!("ring still shared at close"))
        .close()
        .unwrap();
    if rank == 0 {
        eprintln!("[soak] ok: {mode}, {size} ranks, {sent} messages/rank in {elapsed:.2}s");
    }
}

pub fn is_safe(&self) -> bool

Whether this ring gates overwrites with cumulative acknowledgements.

Examples found in repository

examples/test_ring.rs (line 26)

fn safe_basics(world: &SimpleCommunicator, rank: Rank, next: Rank, prev: Rank) {
    // Depth 4 against 3 messages: this covers ordering and acknowledge
    // validation with the overwrite gate deliberately out of the way.
    // Backpressure is its own test, on a one-way lane.
    let ring = Ring::safe(world, &pair(4, 8)).unwrap();
    assert!(ring.is_safe());
    assert_eq!(ring.depth(next), Some(4));
    assert_eq!(ring.capacity(next), Some(8));

    assert!(matches!(
        ring.send(next, &[0; 9]),
        Err(Error::Payload {
            len: 9,
            capacity: 8
        })
    ));
    assert!(matches!(
        ring.ack(prev, 1),
        Err(Error::Ack {
            sequence: 1,
            received: 0,
            ..
        })
    ));

    for s in 1..=3u8 {
        assert_eq!(ring.send(next, &[rank as u8, s]).unwrap(), u64::from(s));
    }
    world.barrier();

    let messages = ring.poll().unwrap();
    assert_eq!(
        messages,
        (1..=3)
            .map(|s| Message {
                origin: prev,
                sequence: s,
                data: vec![prev as u8, s as u8],
            })
            .collect::<Vec<_>>()
    );
    assert!(ring.poll().unwrap().is_empty());
    assert_eq!(ring.lost(), 0);
    assert_eq!(ring.max_lag(), 3);

    assert!(matches!(
        ring.ack(prev, 4),
        Err(Error::Ack {
            sequence: 4,
            received: 3,
            ..
        })
    ));
    ring.ack(prev, 3).unwrap();
    ring.ack(prev, 3).unwrap();
    ring.ack(prev, 2).unwrap();

    world.barrier();
    ring.close().unwrap();
}

/// A safe sender with no free slot blocks until the receiver acknowledges.
///
/// One-way, so the ack timing belongs to the test. On a symmetric pair each
/// rank's ack races the other's gate check: whichever receiver drains first
/// opens its peer's gate before that peer ever evaluates it, and neither side
/// can be relied on to block.
fn backpressure(world: &SimpleCommunicator, rank: Rank) {
    let ring = Ring::safe(world, &oneway(2, 8)).unwrap();
    if rank == 0 {
        assert_eq!(ring.send(1, &[1]).unwrap(), 1);
        assert_eq!(ring.send(1, &[2]).unwrap(), 2);
    }
    world.barrier();

    if rank == 0 {
        let sent = thread::scope(|scope| {
            let sender = &ring;
            let handle = scope.spawn(move || sender.send(1, &[3]));
            // Depth is 2 and nothing has been acknowledged, so there is no slot
            // for the third message and the send has to wait.
            let deadline = Instant::now() + Duration::from_secs(5);
            while ring.waits() == 0 {
                assert!(Instant::now() < deadline, "safe sender never blocked");
                thread::yield_now();
            }
            // The receiver is parked here, so it cannot have drained early.
            world.barrier();
            handle.join().expect("sender thread panicked").unwrap()
        });
        assert_eq!(sent, 3);
        assert!(ring.waits() > 0);
        assert!(ring.wait_ns() > 0);
    } else {
        world.barrier();
        let messages = ring.poll().unwrap();
        assert_eq!(messages.len(), 2);
        assert_eq!(messages[0].data, vec![1]);
        assert_eq!(messages[1].data, vec![2]);
        assert!(matches!(
            ring.ack(0, 3),
            Err(Error::Ack {
                sequence: 3,
                received: 2,
                ..
            })
        ));
        ring.ack(0, 2).unwrap();
        ring.ack(0, 2).unwrap();
    }

    world.barrier();
    if rank == 1 {
        let messages = ring.poll().unwrap();
        assert_eq!(messages.len(), 1);
        assert_eq!(messages[0].sequence, 3);
        assert_eq!(messages[0].data, vec![3]);
        ring.ack(0, 3).unwrap();
    }
    world.barrier();
    ring.close().unwrap();
}

fn oneway_basics(world: &SimpleCommunicator, rank: Rank) {
    let ring = Ring::safe(world, &oneway(2, 8)).unwrap();
    if rank == 0 {
        assert_eq!(ring.send(1, &[7]).unwrap(), 1);
    }
    world.barrier();
    if rank == 1 {
        let messages = ring.poll().unwrap();
        assert_eq!(messages.len(), 1);
        assert_eq!(messages[0].origin, 0);
        assert_eq!(messages[0].data, vec![7]);
        ring.ack(0, 1).unwrap();
    }
    world.barrier();
    ring.close().unwrap();
}

fn raw_basics(world: &SimpleCommunicator, rank: Rank, next: Rank, prev: Rank) {
    let ring = Ring::raw(world, &pair(2, 8)).unwrap();
    assert!(!ring.is_safe());

    for s in 1..=3u8 {
        assert_eq!(ring.send(next, &[rank as u8, s]).unwrap(), u64::from(s));
    }
    world.barrier();

    let messages = ring.poll().unwrap();
    assert_eq!(
        messages,
        vec![
            Message {
                origin: prev,
                sequence: 2,
                data: vec![prev as u8, 2]
            },
            Message {
                origin: prev,
                sequence: 3,
                data: vec![prev as u8, 3]
            },
        ]
    );
    assert_eq!(ring.lost(), 1);

    // Raw acks are no-ops.
    ring.ack(prev, u64::MAX).unwrap();
    world.barrier();
    ring.close().unwrap();
}

pub fn depth(&self, destination: Rank) -> Option<usize>

Slot count configured for the outgoing lane to destination, if any.

Examples found in repository

examples/test_ring.rs (line 27)

fn safe_basics(world: &SimpleCommunicator, rank: Rank, next: Rank, prev: Rank) {
    // Depth 4 against 3 messages: this covers ordering and acknowledge
    // validation with the overwrite gate deliberately out of the way.
    // Backpressure is its own test, on a one-way lane.
    let ring = Ring::safe(world, &pair(4, 8)).unwrap();
    assert!(ring.is_safe());
    assert_eq!(ring.depth(next), Some(4));
    assert_eq!(ring.capacity(next), Some(8));

    assert!(matches!(
        ring.send(next, &[0; 9]),
        Err(Error::Payload {
            len: 9,
            capacity: 8
        })
    ));
    assert!(matches!(
        ring.ack(prev, 1),
        Err(Error::Ack {
            sequence: 1,
            received: 0,
            ..
        })
    ));

    for s in 1..=3u8 {
        assert_eq!(ring.send(next, &[rank as u8, s]).unwrap(), u64::from(s));
    }
    world.barrier();

    let messages = ring.poll().unwrap();
    assert_eq!(
        messages,
        (1..=3)
            .map(|s| Message {
                origin: prev,
                sequence: s,
                data: vec![prev as u8, s as u8],
            })
            .collect::<Vec<_>>()
    );
    assert!(ring.poll().unwrap().is_empty());
    assert_eq!(ring.lost(), 0);
    assert_eq!(ring.max_lag(), 3);

    assert!(matches!(
        ring.ack(prev, 4),
        Err(Error::Ack {
            sequence: 4,
            received: 3,
            ..
        })
    ));
    ring.ack(prev, 3).unwrap();
    ring.ack(prev, 3).unwrap();
    ring.ack(prev, 2).unwrap();

    world.barrier();
    ring.close().unwrap();
}

pub fn capacity(&self, destination: Rank) -> Option<usize>

Per-slot payload capacity configured for the outgoing lane to destination, if any.

Examples found in repository

examples/test_ring.rs (line 28)

fn safe_basics(world: &SimpleCommunicator, rank: Rank, next: Rank, prev: Rank) {
    // Depth 4 against 3 messages: this covers ordering and acknowledge
    // validation with the overwrite gate deliberately out of the way.
    // Backpressure is its own test, on a one-way lane.
    let ring = Ring::safe(world, &pair(4, 8)).unwrap();
    assert!(ring.is_safe());
    assert_eq!(ring.depth(next), Some(4));
    assert_eq!(ring.capacity(next), Some(8));

    assert!(matches!(
        ring.send(next, &[0; 9]),
        Err(Error::Payload {
            len: 9,
            capacity: 8
        })
    ));
    assert!(matches!(
        ring.ack(prev, 1),
        Err(Error::Ack {
            sequence: 1,
            received: 0,
            ..
        })
    ));

    for s in 1..=3u8 {
        assert_eq!(ring.send(next, &[rank as u8, s]).unwrap(), u64::from(s));
    }
    world.barrier();

    let messages = ring.poll().unwrap();
    assert_eq!(
        messages,
        (1..=3)
            .map(|s| Message {
                origin: prev,
                sequence: s,
                data: vec![prev as u8, s as u8],
            })
            .collect::<Vec<_>>()
    );
    assert!(ring.poll().unwrap().is_empty());
    assert_eq!(ring.lost(), 0);
    assert_eq!(ring.max_lag(), 3);

    assert!(matches!(
        ring.ack(prev, 4),
        Err(Error::Ack {
            sequence: 4,
            received: 3,
            ..
        })
    ));
    ring.ack(prev, 3).unwrap();
    ring.ack(prev, 3).unwrap();
    ring.ack(prev, 2).unwrap();

    world.barrier();
    ring.close().unwrap();
}

pub fn lost(&self) -> u64

Total messages observed as lost (raw mode only).

Examples found in repository

examples/test_ring.rs (line 63)

fn safe_basics(world: &SimpleCommunicator, rank: Rank, next: Rank, prev: Rank) {
    // Depth 4 against 3 messages: this covers ordering and acknowledge
    // validation with the overwrite gate deliberately out of the way.
    // Backpressure is its own test, on a one-way lane.
    let ring = Ring::safe(world, &pair(4, 8)).unwrap();
    assert!(ring.is_safe());
    assert_eq!(ring.depth(next), Some(4));
    assert_eq!(ring.capacity(next), Some(8));

    assert!(matches!(
        ring.send(next, &[0; 9]),
        Err(Error::Payload {
            len: 9,
            capacity: 8
        })
    ));
    assert!(matches!(
        ring.ack(prev, 1),
        Err(Error::Ack {
            sequence: 1,
            received: 0,
            ..
        })
    ));

    for s in 1..=3u8 {
        assert_eq!(ring.send(next, &[rank as u8, s]).unwrap(), u64::from(s));
    }
    world.barrier();

    let messages = ring.poll().unwrap();
    assert_eq!(
        messages,
        (1..=3)
            .map(|s| Message {
                origin: prev,
                sequence: s,
                data: vec![prev as u8, s as u8],
            })
            .collect::<Vec<_>>()
    );
    assert!(ring.poll().unwrap().is_empty());
    assert_eq!(ring.lost(), 0);
    assert_eq!(ring.max_lag(), 3);

    assert!(matches!(
        ring.ack(prev, 4),
        Err(Error::Ack {
            sequence: 4,
            received: 3,
            ..
        })
    ));
    ring.ack(prev, 3).unwrap();
    ring.ack(prev, 3).unwrap();
    ring.ack(prev, 2).unwrap();

    world.barrier();
    ring.close().unwrap();
}

/// A safe sender with no free slot blocks until the receiver acknowledges.
///
/// One-way, so the ack timing belongs to the test. On a symmetric pair each
/// rank's ack races the other's gate check: whichever receiver drains first
/// opens its peer's gate before that peer ever evaluates it, and neither side
/// can be relied on to block.
fn backpressure(world: &SimpleCommunicator, rank: Rank) {
    let ring = Ring::safe(world, &oneway(2, 8)).unwrap();
    if rank == 0 {
        assert_eq!(ring.send(1, &[1]).unwrap(), 1);
        assert_eq!(ring.send(1, &[2]).unwrap(), 2);
    }
    world.barrier();

    if rank == 0 {
        let sent = thread::scope(|scope| {
            let sender = &ring;
            let handle = scope.spawn(move || sender.send(1, &[3]));
            // Depth is 2 and nothing has been acknowledged, so there is no slot
            // for the third message and the send has to wait.
            let deadline = Instant::now() + Duration::from_secs(5);
            while ring.waits() == 0 {
                assert!(Instant::now() < deadline, "safe sender never blocked");
                thread::yield_now();
            }
            // The receiver is parked here, so it cannot have drained early.
            world.barrier();
            handle.join().expect("sender thread panicked").unwrap()
        });
        assert_eq!(sent, 3);
        assert!(ring.waits() > 0);
        assert!(ring.wait_ns() > 0);
    } else {
        world.barrier();
        let messages = ring.poll().unwrap();
        assert_eq!(messages.len(), 2);
        assert_eq!(messages[0].data, vec![1]);
        assert_eq!(messages[1].data, vec![2]);
        assert!(matches!(
            ring.ack(0, 3),
            Err(Error::Ack {
                sequence: 3,
                received: 2,
                ..
            })
        ));
        ring.ack(0, 2).unwrap();
        ring.ack(0, 2).unwrap();
    }

    world.barrier();
    if rank == 1 {
        let messages = ring.poll().unwrap();
        assert_eq!(messages.len(), 1);
        assert_eq!(messages[0].sequence, 3);
        assert_eq!(messages[0].data, vec![3]);
        ring.ack(0, 3).unwrap();
    }
    world.barrier();
    ring.close().unwrap();
}

fn oneway_basics(world: &SimpleCommunicator, rank: Rank) {
    let ring = Ring::safe(world, &oneway(2, 8)).unwrap();
    if rank == 0 {
        assert_eq!(ring.send(1, &[7]).unwrap(), 1);
    }
    world.barrier();
    if rank == 1 {
        let messages = ring.poll().unwrap();
        assert_eq!(messages.len(), 1);
        assert_eq!(messages[0].origin, 0);
        assert_eq!(messages[0].data, vec![7]);
        ring.ack(0, 1).unwrap();
    }
    world.barrier();
    ring.close().unwrap();
}

fn raw_basics(world: &SimpleCommunicator, rank: Rank, next: Rank, prev: Rank) {
    let ring = Ring::raw(world, &pair(2, 8)).unwrap();
    assert!(!ring.is_safe());

    for s in 1..=3u8 {
        assert_eq!(ring.send(next, &[rank as u8, s]).unwrap(), u64::from(s));
    }
    world.barrier();

    let messages = ring.poll().unwrap();
    assert_eq!(
        messages,
        vec![
            Message {
                origin: prev,
                sequence: 2,
                data: vec![prev as u8, 2]
            },
            Message {
                origin: prev,
                sequence: 3,
                data: vec![prev as u8, 3]
            },
        ]
    );
    assert_eq!(ring.lost(), 1);

    // Raw acks are no-ops.
    ring.ack(prev, u64::MAX).unwrap();
    world.barrier();
    ring.close().unwrap();
}

examples/soak.rs (line 271)

fn main() {
    let mut args = std::env::args().skip(1);
    let mode = args.next().unwrap_or_else(|| "safe".into());
    let messages: u64 = args.next().and_then(|s| s.parse().ok()).unwrap_or(20_000);
    let payload: usize = args.next().and_then(|s| s.parse().ok()).unwrap_or(256);
    let depth: usize = args.next().and_then(|s| s.parse().ok()).unwrap_or(8);
    let noise_kib: usize = args.next().and_then(|s| s.parse().ok()).unwrap_or(0);
    let pace_ns: u64 = args.next().and_then(|s| s.parse().ok()).unwrap_or(0);
    // Echoed straight into the output so repeats of one point stay distinguishable.
    let rep: u32 = args.next().and_then(|s| s.parse().ok()).unwrap_or(0);
    assert!(
        payload >= STAMP,
        "payload must hold the origin and sequence"
    );
    let safe = match mode.as_str() {
        "safe" => true,
        "raw" => false,
        other => panic!("mode must be safe or raw, got {other:?}"),
    };

    let (universe, provided) =
        mpi::initialize_with_threading(Threading::Multiple).expect("MPI must initialize once");
    assert_eq!(provided, Threading::Multiple);
    let world = universe.world();
    let rank = world.rank();
    let size = world.size();
    assert!(size >= 2, "soak needs at least 2 ranks");

    // All-to-all: every ordered pair of distinct ranks gets a lane.
    let mut lanes = Vec::new();
    for source in 0..size {
        for destination in 0..size {
            if source != destination {
                lanes.push((source, destination, depth, payload));
            }
        }
    }
    let ring = Arc::new(if safe {
        Ring::safe(&world, &lanes).unwrap()
    } else {
        Ring::raw(&world, &lanes).unwrap()
    });

    let stop = Arc::new(AtomicBool::new(false));
    let done = Arc::new(AtomicBool::new(false));
    let count = Arc::new(AtomicU64::new(0));
    let peers = (size - 1) as u64;
    let sent = messages * peers;

    world.barrier();
    let started = std::time::Instant::now();

    let outcome = std::thread::scope(|scope| {
        if noise_kib > 0 {
            let stop = Arc::clone(&stop);
            let shared = Shared(&world);
            scope.spawn(move || {
                // Name the wrapper as a whole first. Closures capture the
                // narrowest place they use, and `shared.0` alone would be
                // captured as a bare reference, losing the `Send` impl.
                let shared = shared;
                noise(shared.0, noise_kib, &stop);
            });
        }

        let poller = {
            let (ring, done, count) = (Arc::clone(&ring), Arc::clone(&done), Arc::clone(&count));
            scope.spawn(move || -> Result<Vec<u64>, String> {
                let mut seen = vec![0u64; size as usize];
                let mut total = 0u64;
                loop {
                    // If true, this poll follows the global send barrier.
                    let finished = done.load(Ordering::Acquire);
                    let batch = ring.poll().map_err(|e| format!("poll: {e}"))?;
                    let empty = batch.is_empty();
                    let mut furthest: HashMap<Rank, u64> = HashMap::new();
                    for m in &batch {
                        verify(m)?;
                        let previous = seen[m.origin as usize];
                        if m.sequence <= previous {
                            return Err(format!(
                                "sequence from {} went backwards: {} after {previous}",
                                m.origin, m.sequence
                            ));
                        }
                        if safe && m.sequence != previous + 1 {
                            return Err(format!(
                                "safe lane from {} skipped {} to {}",
                                m.origin,
                                previous + 1,
                                m.sequence
                            ));
                        }
                        seen[m.origin as usize] = m.sequence;
                        let f = furthest.entry(m.origin).or_insert(0);
                        *f = (*f).max(m.sequence);
                        total += 1;
                    }
                    for (origin, sequence) in furthest {
                        ring.ack(origin, sequence)
                            .map_err(|e| format!("ack to {origin}: {e}"))?;
                    }
                    count.store(total, Ordering::Relaxed);
                    if finished && empty {
                        return Ok(seen);
                    }
                    if empty {
                        std::thread::yield_now();
                    }
                }
            })
        };

        // Round-robin over peers so every outgoing lane stays hot at once.
        let mut buf = vec![0u8; payload];
        let pace = std::time::Duration::from_nanos(pace_ns);
        let opened = std::time::Instant::now();
        for sequence in 1..=messages {
            for destination in (0..size).filter(|&r| r != rank) {
                paint(rank, sequence, &mut buf);
                ring.send(destination, &buf)
                    .unwrap_or_else(|e| panic!("rank {rank}: send to {destination}: {e}"));
            }
            // Absolute deadlines, so a slow round is not paid for twice and the
            // offered rate stays the one that was asked for.
            if pace_ns > 0 {
                let due = opened + pace * sequence as u32;
                while std::time::Instant::now() < due {
                    std::hint::spin_loop();
                }
            }
        }
        // Every rank's sends have completed at their targets before this
        // returns, so after the barrier nothing new can appear in a slot.
        world.barrier();
        done.store(true, Ordering::Release);

        let seen = poller.join().expect("poller thread panicked");
        stop.store(true, Ordering::Relaxed);
        seen
    });

    let elapsed = started.elapsed().as_secs_f64();
    let fail = |why: String| -> ! {
        eprintln!("[soak] FAIL rank {rank}: {why}");
        world.abort(1)
    };
    let seen = outcome.unwrap_or_else(|why| fail(why));

    let got = count.load(Ordering::Relaxed);
    let lost = ring.lost();
    let corrupt = ring.corrupt();

    // The last message on a lane is never overwritten, so every lane must end
    // exactly at the sender's high-water mark in both modes.
    for origin in (0..size).filter(|&r| r != rank) {
        if seen[origin as usize] != messages {
            fail(format!(
                "lane from {origin} ended at {} not {messages}",
                seen[origin as usize]
            ));
        }
    }
    if safe && lost != 0 {
        fail(format!("safe ring lost {lost}"));
    }
    if got + lost != sent {
        fail(format!("received {got} + lost {lost} != sent {sent}"));
    }

    // Gather the counters and let rank 0 write every row. Interleaving the
    // ranks' own stdout would scatter the header into the middle of the table:
    // mpirun orders a rank's output against itself and nothing else.
    let mine = [
        got,
        lost,
        corrupt,
        ring.max_lag(),
        ring.waits(),
        ring.wait_ns(),
    ];
    let mut all = vec![0u64; mine.len() * size as usize];
    world.all_gather_into(&mine[..], &mut all[..]);
    if rank == 0 {
        println!(
            "mode\tranks\tdepth\tpayload\tnoise_kib\tpace_ns\trep\trank\tsent\treceived\tlost\tcorrupt\tmax_lag\twaits\twait_s\tseconds"
        );
        for (r, row) in all.chunks_exact(mine.len()).enumerate() {
            println!(
                "{mode}\t{size}\t{depth}\t{payload}\t{noise_kib}\t{pace_ns}\t{rep}\t{r}\t{sent}\t{}\t{}\t{}\t{}\t{}\t{:.6}\t{elapsed:.6}",
                row[0],
                row[1],
                row[2],
                row[3],
                row[4],
                row[5] as f64 * 1e-9,
            );
        }
    }

    world.barrier();
    Arc::try_unwrap(ring)
        .unwrap_or_else(|_| panic!("ring still shared at close"))
        .close()
        .unwrap();
    if rank == 0 {
        eprintln!("[soak] ok: {mode}, {size} ranks, {sent} messages/rank in {elapsed:.2}s");
    }
}

pub fn corrupt(&self) -> u64

Total corrupt slot reads: torn header/footer, bad length, or CRC mismatch.

Examples found in repository

examples/soak.rs (line 272)

fn main() {
    let mut args = std::env::args().skip(1);
    let mode = args.next().unwrap_or_else(|| "safe".into());
    let messages: u64 = args.next().and_then(|s| s.parse().ok()).unwrap_or(20_000);
    let payload: usize = args.next().and_then(|s| s.parse().ok()).unwrap_or(256);
    let depth: usize = args.next().and_then(|s| s.parse().ok()).unwrap_or(8);
    let noise_kib: usize = args.next().and_then(|s| s.parse().ok()).unwrap_or(0);
    let pace_ns: u64 = args.next().and_then(|s| s.parse().ok()).unwrap_or(0);
    // Echoed straight into the output so repeats of one point stay distinguishable.
    let rep: u32 = args.next().and_then(|s| s.parse().ok()).unwrap_or(0);
    assert!(
        payload >= STAMP,
        "payload must hold the origin and sequence"
    );
    let safe = match mode.as_str() {
        "safe" => true,
        "raw" => false,
        other => panic!("mode must be safe or raw, got {other:?}"),
    };

    let (universe, provided) =
        mpi::initialize_with_threading(Threading::Multiple).expect("MPI must initialize once");
    assert_eq!(provided, Threading::Multiple);
    let world = universe.world();
    let rank = world.rank();
    let size = world.size();
    assert!(size >= 2, "soak needs at least 2 ranks");

    // All-to-all: every ordered pair of distinct ranks gets a lane.
    let mut lanes = Vec::new();
    for source in 0..size {
        for destination in 0..size {
            if source != destination {
                lanes.push((source, destination, depth, payload));
            }
        }
    }
    let ring = Arc::new(if safe {
        Ring::safe(&world, &lanes).unwrap()
    } else {
        Ring::raw(&world, &lanes).unwrap()
    });

    let stop = Arc::new(AtomicBool::new(false));
    let done = Arc::new(AtomicBool::new(false));
    let count = Arc::new(AtomicU64::new(0));
    let peers = (size - 1) as u64;
    let sent = messages * peers;

    world.barrier();
    let started = std::time::Instant::now();

    let outcome = std::thread::scope(|scope| {
        if noise_kib > 0 {
            let stop = Arc::clone(&stop);
            let shared = Shared(&world);
            scope.spawn(move || {
                // Name the wrapper as a whole first. Closures capture the
                // narrowest place they use, and `shared.0` alone would be
                // captured as a bare reference, losing the `Send` impl.
                let shared = shared;
                noise(shared.0, noise_kib, &stop);
            });
        }

        let poller = {
            let (ring, done, count) = (Arc::clone(&ring), Arc::clone(&done), Arc::clone(&count));
            scope.spawn(move || -> Result<Vec<u64>, String> {
                let mut seen = vec![0u64; size as usize];
                let mut total = 0u64;
                loop {
                    // If true, this poll follows the global send barrier.
                    let finished = done.load(Ordering::Acquire);
                    let batch = ring.poll().map_err(|e| format!("poll: {e}"))?;
                    let empty = batch.is_empty();
                    let mut furthest: HashMap<Rank, u64> = HashMap::new();
                    for m in &batch {
                        verify(m)?;
                        let previous = seen[m.origin as usize];
                        if m.sequence <= previous {
                            return Err(format!(
                                "sequence from {} went backwards: {} after {previous}",
                                m.origin, m.sequence
                            ));
                        }
                        if safe && m.sequence != previous + 1 {
                            return Err(format!(
                                "safe lane from {} skipped {} to {}",
                                m.origin,
                                previous + 1,
                                m.sequence
                            ));
                        }
                        seen[m.origin as usize] = m.sequence;
                        let f = furthest.entry(m.origin).or_insert(0);
                        *f = (*f).max(m.sequence);
                        total += 1;
                    }
                    for (origin, sequence) in furthest {
                        ring.ack(origin, sequence)
                            .map_err(|e| format!("ack to {origin}: {e}"))?;
                    }
                    count.store(total, Ordering::Relaxed);
                    if finished && empty {
                        return Ok(seen);
                    }
                    if empty {
                        std::thread::yield_now();
                    }
                }
            })
        };

        // Round-robin over peers so every outgoing lane stays hot at once.
        let mut buf = vec![0u8; payload];
        let pace = std::time::Duration::from_nanos(pace_ns);
        let opened = std::time::Instant::now();
        for sequence in 1..=messages {
            for destination in (0..size).filter(|&r| r != rank) {
                paint(rank, sequence, &mut buf);
                ring.send(destination, &buf)
                    .unwrap_or_else(|e| panic!("rank {rank}: send to {destination}: {e}"));
            }
            // Absolute deadlines, so a slow round is not paid for twice and the
            // offered rate stays the one that was asked for.
            if pace_ns > 0 {
                let due = opened + pace * sequence as u32;
                while std::time::Instant::now() < due {
                    std::hint::spin_loop();
                }
            }
        }
        // Every rank's sends have completed at their targets before this
        // returns, so after the barrier nothing new can appear in a slot.
        world.barrier();
        done.store(true, Ordering::Release);

        let seen = poller.join().expect("poller thread panicked");
        stop.store(true, Ordering::Relaxed);
        seen
    });

    let elapsed = started.elapsed().as_secs_f64();
    let fail = |why: String| -> ! {
        eprintln!("[soak] FAIL rank {rank}: {why}");
        world.abort(1)
    };
    let seen = outcome.unwrap_or_else(|why| fail(why));

    let got = count.load(Ordering::Relaxed);
    let lost = ring.lost();
    let corrupt = ring.corrupt();

    // The last message on a lane is never overwritten, so every lane must end
    // exactly at the sender's high-water mark in both modes.
    for origin in (0..size).filter(|&r| r != rank) {
        if seen[origin as usize] != messages {
            fail(format!(
                "lane from {origin} ended at {} not {messages}",
                seen[origin as usize]
            ));
        }
    }
    if safe && lost != 0 {
        fail(format!("safe ring lost {lost}"));
    }
    if got + lost != sent {
        fail(format!("received {got} + lost {lost} != sent {sent}"));
    }

    // Gather the counters and let rank 0 write every row. Interleaving the
    // ranks' own stdout would scatter the header into the middle of the table:
    // mpirun orders a rank's output against itself and nothing else.
    let mine = [
        got,
        lost,
        corrupt,
        ring.max_lag(),
        ring.waits(),
        ring.wait_ns(),
    ];
    let mut all = vec![0u64; mine.len() * size as usize];
    world.all_gather_into(&mine[..], &mut all[..]);
    if rank == 0 {
        println!(
            "mode\tranks\tdepth\tpayload\tnoise_kib\tpace_ns\trep\trank\tsent\treceived\tlost\tcorrupt\tmax_lag\twaits\twait_s\tseconds"
        );
        for (r, row) in all.chunks_exact(mine.len()).enumerate() {
            println!(
                "{mode}\t{size}\t{depth}\t{payload}\t{noise_kib}\t{pace_ns}\t{rep}\t{r}\t{sent}\t{}\t{}\t{}\t{}\t{}\t{:.6}\t{elapsed:.6}",
                row[0],
                row[1],
                row[2],
                row[3],
                row[4],
                row[5] as f64 * 1e-9,
            );
        }
    }

    world.barrier();
    Arc::try_unwrap(ring)
        .unwrap_or_else(|_| panic!("ring still shared at close"))
        .close()
        .unwrap();
    if rank == 0 {
        eprintln!("[soak] ok: {mode}, {size} ranks, {sent} messages/rank in {elapsed:.2}s");
    }
}

pub fn max_lag(&self) -> u64

Greatest sequence advance one Self::poll made on one lane.

In raw mode this includes gaps counted as lost. In safe mode it is the number of messages drained.

Examples found in repository

examples/test_ring.rs (line 64)

fn safe_basics(world: &SimpleCommunicator, rank: Rank, next: Rank, prev: Rank) {
    // Depth 4 against 3 messages: this covers ordering and acknowledge
    // validation with the overwrite gate deliberately out of the way.
    // Backpressure is its own test, on a one-way lane.
    let ring = Ring::safe(world, &pair(4, 8)).unwrap();
    assert!(ring.is_safe());
    assert_eq!(ring.depth(next), Some(4));
    assert_eq!(ring.capacity(next), Some(8));

    assert!(matches!(
        ring.send(next, &[0; 9]),
        Err(Error::Payload {
            len: 9,
            capacity: 8
        })
    ));
    assert!(matches!(
        ring.ack(prev, 1),
        Err(Error::Ack {
            sequence: 1,
            received: 0,
            ..
        })
    ));

    for s in 1..=3u8 {
        assert_eq!(ring.send(next, &[rank as u8, s]).unwrap(), u64::from(s));
    }
    world.barrier();

    let messages = ring.poll().unwrap();
    assert_eq!(
        messages,
        (1..=3)
            .map(|s| Message {
                origin: prev,
                sequence: s,
                data: vec![prev as u8, s as u8],
            })
            .collect::<Vec<_>>()
    );
    assert!(ring.poll().unwrap().is_empty());
    assert_eq!(ring.lost(), 0);
    assert_eq!(ring.max_lag(), 3);

    assert!(matches!(
        ring.ack(prev, 4),
        Err(Error::Ack {
            sequence: 4,
            received: 3,
            ..
        })
    ));
    ring.ack(prev, 3).unwrap();
    ring.ack(prev, 3).unwrap();
    ring.ack(prev, 2).unwrap();

    world.barrier();
    ring.close().unwrap();
}

examples/soak.rs (line 298)

fn main() {
    let mut args = std::env::args().skip(1);
    let mode = args.next().unwrap_or_else(|| "safe".into());
    let messages: u64 = args.next().and_then(|s| s.parse().ok()).unwrap_or(20_000);
    let payload: usize = args.next().and_then(|s| s.parse().ok()).unwrap_or(256);
    let depth: usize = args.next().and_then(|s| s.parse().ok()).unwrap_or(8);
    let noise_kib: usize = args.next().and_then(|s| s.parse().ok()).unwrap_or(0);
    let pace_ns: u64 = args.next().and_then(|s| s.parse().ok()).unwrap_or(0);
    // Echoed straight into the output so repeats of one point stay distinguishable.
    let rep: u32 = args.next().and_then(|s| s.parse().ok()).unwrap_or(0);
    assert!(
        payload >= STAMP,
        "payload must hold the origin and sequence"
    );
    let safe = match mode.as_str() {
        "safe" => true,
        "raw" => false,
        other => panic!("mode must be safe or raw, got {other:?}"),
    };

    let (universe, provided) =
        mpi::initialize_with_threading(Threading::Multiple).expect("MPI must initialize once");
    assert_eq!(provided, Threading::Multiple);
    let world = universe.world();
    let rank = world.rank();
    let size = world.size();
    assert!(size >= 2, "soak needs at least 2 ranks");

    // All-to-all: every ordered pair of distinct ranks gets a lane.
    let mut lanes = Vec::new();
    for source in 0..size {
        for destination in 0..size {
            if source != destination {
                lanes.push((source, destination, depth, payload));
            }
        }
    }
    let ring = Arc::new(if safe {
        Ring::safe(&world, &lanes).unwrap()
    } else {
        Ring::raw(&world, &lanes).unwrap()
    });

    let stop = Arc::new(AtomicBool::new(false));
    let done = Arc::new(AtomicBool::new(false));
    let count = Arc::new(AtomicU64::new(0));
    let peers = (size - 1) as u64;
    let sent = messages * peers;

    world.barrier();
    let started = std::time::Instant::now();

    let outcome = std::thread::scope(|scope| {
        if noise_kib > 0 {
            let stop = Arc::clone(&stop);
            let shared = Shared(&world);
            scope.spawn(move || {
                // Name the wrapper as a whole first. Closures capture the
                // narrowest place they use, and `shared.0` alone would be
                // captured as a bare reference, losing the `Send` impl.
                let shared = shared;
                noise(shared.0, noise_kib, &stop);
            });
        }

        let poller = {
            let (ring, done, count) = (Arc::clone(&ring), Arc::clone(&done), Arc::clone(&count));
            scope.spawn(move || -> Result<Vec<u64>, String> {
                let mut seen = vec![0u64; size as usize];
                let mut total = 0u64;
                loop {
                    // If true, this poll follows the global send barrier.
                    let finished = done.load(Ordering::Acquire);
                    let batch = ring.poll().map_err(|e| format!("poll: {e}"))?;
                    let empty = batch.is_empty();
                    let mut furthest: HashMap<Rank, u64> = HashMap::new();
                    for m in &batch {
                        verify(m)?;
                        let previous = seen[m.origin as usize];
                        if m.sequence <= previous {
                            return Err(format!(
                                "sequence from {} went backwards: {} after {previous}",
                                m.origin, m.sequence
                            ));
                        }
                        if safe && m.sequence != previous + 1 {
                            return Err(format!(
                                "safe lane from {} skipped {} to {}",
                                m.origin,
                                previous + 1,
                                m.sequence
                            ));
                        }
                        seen[m.origin as usize] = m.sequence;
                        let f = furthest.entry(m.origin).or_insert(0);
                        *f = (*f).max(m.sequence);
                        total += 1;
                    }
                    for (origin, sequence) in furthest {
                        ring.ack(origin, sequence)
                            .map_err(|e| format!("ack to {origin}: {e}"))?;
                    }
                    count.store(total, Ordering::Relaxed);
                    if finished && empty {
                        return Ok(seen);
                    }
                    if empty {
                        std::thread::yield_now();
                    }
                }
            })
        };

        // Round-robin over peers so every outgoing lane stays hot at once.
        let mut buf = vec![0u8; payload];
        let pace = std::time::Duration::from_nanos(pace_ns);
        let opened = std::time::Instant::now();
        for sequence in 1..=messages {
            for destination in (0..size).filter(|&r| r != rank) {
                paint(rank, sequence, &mut buf);
                ring.send(destination, &buf)
                    .unwrap_or_else(|e| panic!("rank {rank}: send to {destination}: {e}"));
            }
            // Absolute deadlines, so a slow round is not paid for twice and the
            // offered rate stays the one that was asked for.
            if pace_ns > 0 {
                let due = opened + pace * sequence as u32;
                while std::time::Instant::now() < due {
                    std::hint::spin_loop();
                }
            }
        }
        // Every rank's sends have completed at their targets before this
        // returns, so after the barrier nothing new can appear in a slot.
        world.barrier();
        done.store(true, Ordering::Release);

        let seen = poller.join().expect("poller thread panicked");
        stop.store(true, Ordering::Relaxed);
        seen
    });

    let elapsed = started.elapsed().as_secs_f64();
    let fail = |why: String| -> ! {
        eprintln!("[soak] FAIL rank {rank}: {why}");
        world.abort(1)
    };
    let seen = outcome.unwrap_or_else(|why| fail(why));

    let got = count.load(Ordering::Relaxed);
    let lost = ring.lost();
    let corrupt = ring.corrupt();

    // The last message on a lane is never overwritten, so every lane must end
    // exactly at the sender's high-water mark in both modes.
    for origin in (0..size).filter(|&r| r != rank) {
        if seen[origin as usize] != messages {
            fail(format!(
                "lane from {origin} ended at {} not {messages}",
                seen[origin as usize]
            ));
        }
    }
    if safe && lost != 0 {
        fail(format!("safe ring lost {lost}"));
    }
    if got + lost != sent {
        fail(format!("received {got} + lost {lost} != sent {sent}"));
    }

    // Gather the counters and let rank 0 write every row. Interleaving the
    // ranks' own stdout would scatter the header into the middle of the table:
    // mpirun orders a rank's output against itself and nothing else.
    let mine = [
        got,
        lost,
        corrupt,
        ring.max_lag(),
        ring.waits(),
        ring.wait_ns(),
    ];
    let mut all = vec![0u64; mine.len() * size as usize];
    world.all_gather_into(&mine[..], &mut all[..]);
    if rank == 0 {
        println!(
            "mode\tranks\tdepth\tpayload\tnoise_kib\tpace_ns\trep\trank\tsent\treceived\tlost\tcorrupt\tmax_lag\twaits\twait_s\tseconds"
        );
        for (r, row) in all.chunks_exact(mine.len()).enumerate() {
            println!(
                "{mode}\t{size}\t{depth}\t{payload}\t{noise_kib}\t{pace_ns}\t{rep}\t{r}\t{sent}\t{}\t{}\t{}\t{}\t{}\t{:.6}\t{elapsed:.6}",
                row[0],
                row[1],
                row[2],
                row[3],
                row[4],
                row[5] as f64 * 1e-9,
            );
        }
    }

    world.barrier();
    Arc::try_unwrap(ring)
        .unwrap_or_else(|_| panic!("ring still shared at close"))
        .close()
        .unwrap();
    if rank == 0 {
        eprintln!("[soak] ok: {mode}, {size} ranks, {sent} messages/rank in {elapsed:.2}s");
    }
}

pub fn waits(&self) -> u64

Number of times a safe-mode sender actually blocked on acknowledgements.

Refreshing the cached counter does not count; only a sender that found no headroom after refreshing and had to spin does.

Examples found in repository

examples/test_ring.rs (line 103)

fn backpressure(world: &SimpleCommunicator, rank: Rank) {
    let ring = Ring::safe(world, &oneway(2, 8)).unwrap();
    if rank == 0 {
        assert_eq!(ring.send(1, &[1]).unwrap(), 1);
        assert_eq!(ring.send(1, &[2]).unwrap(), 2);
    }
    world.barrier();

    if rank == 0 {
        let sent = thread::scope(|scope| {
            let sender = ˚
            let handle = scope.spawn(move || sender.send(1, &[3]));
            // Depth is 2 and nothing has been acknowledged, so there is no slot
            // for the third message and the send has to wait.
            let deadline = Instant::now() + Duration::from_secs(5);
            while ring.waits() == 0 {
                assert!(Instant::now() < deadline, "safe sender never blocked");
                thread::yield_now();
            }
            // The receiver is parked here, so it cannot have drained early.
            world.barrier();
            handle.join().expect("sender thread panicked").unwrap()
        });
        assert_eq!(sent, 3);
        assert!(ring.waits() > 0);
        assert!(ring.wait_ns() > 0);
    } else {
        world.barrier();
        let messages = ring.poll().unwrap();
        assert_eq!(messages.len(), 2);
        assert_eq!(messages[0].data, vec![1]);
        assert_eq!(messages[1].data, vec![2]);
        assert!(matches!(
            ring.ack(0, 3),
            Err(Error::Ack {
                sequence: 3,
                received: 2,
                ..
            })
        ));
        ring.ack(0, 2).unwrap();
        ring.ack(0, 2).unwrap();
    }

    world.barrier();
    if rank == 1 {
        let messages = ring.poll().unwrap();
        assert_eq!(messages.len(), 1);
        assert_eq!(messages[0].sequence, 3);
        assert_eq!(messages[0].data, vec![3]);
        ring.ack(0, 3).unwrap();
    }
    world.barrier();
    ring.close().unwrap();
}

examples/soak.rs (line 299)

fn main() {
    let mut args = std::env::args().skip(1);
    let mode = args.next().unwrap_or_else(|| "safe".into());
    let messages: u64 = args.next().and_then(|s| s.parse().ok()).unwrap_or(20_000);
    let payload: usize = args.next().and_then(|s| s.parse().ok()).unwrap_or(256);
    let depth: usize = args.next().and_then(|s| s.parse().ok()).unwrap_or(8);
    let noise_kib: usize = args.next().and_then(|s| s.parse().ok()).unwrap_or(0);
    let pace_ns: u64 = args.next().and_then(|s| s.parse().ok()).unwrap_or(0);
    // Echoed straight into the output so repeats of one point stay distinguishable.
    let rep: u32 = args.next().and_then(|s| s.parse().ok()).unwrap_or(0);
    assert!(
        payload >= STAMP,
        "payload must hold the origin and sequence"
    );
    let safe = match mode.as_str() {
        "safe" => true,
        "raw" => false,
        other => panic!("mode must be safe or raw, got {other:?}"),
    };

    let (universe, provided) =
        mpi::initialize_with_threading(Threading::Multiple).expect("MPI must initialize once");
    assert_eq!(provided, Threading::Multiple);
    let world = universe.world();
    let rank = world.rank();
    let size = world.size();
    assert!(size >= 2, "soak needs at least 2 ranks");

    // All-to-all: every ordered pair of distinct ranks gets a lane.
    let mut lanes = Vec::new();
    for source in 0..size {
        for destination in 0..size {
            if source != destination {
                lanes.push((source, destination, depth, payload));
            }
        }
    }
    let ring = Arc::new(if safe {
        Ring::safe(&world, &lanes).unwrap()
    } else {
        Ring::raw(&world, &lanes).unwrap()
    });

    let stop = Arc::new(AtomicBool::new(false));
    let done = Arc::new(AtomicBool::new(false));
    let count = Arc::new(AtomicU64::new(0));
    let peers = (size - 1) as u64;
    let sent = messages * peers;

    world.barrier();
    let started = std::time::Instant::now();

    let outcome = std::thread::scope(|scope| {
        if noise_kib > 0 {
            let stop = Arc::clone(&stop);
            let shared = Shared(&world);
            scope.spawn(move || {
                // Name the wrapper as a whole first. Closures capture the
                // narrowest place they use, and `shared.0` alone would be
                // captured as a bare reference, losing the `Send` impl.
                let shared = shared;
                noise(shared.0, noise_kib, &stop);
            });
        }

        let poller = {
            let (ring, done, count) = (Arc::clone(&ring), Arc::clone(&done), Arc::clone(&count));
            scope.spawn(move || -> Result<Vec<u64>, String> {
                let mut seen = vec![0u64; size as usize];
                let mut total = 0u64;
                loop {
                    // If true, this poll follows the global send barrier.
                    let finished = done.load(Ordering::Acquire);
                    let batch = ring.poll().map_err(|e| format!("poll: {e}"))?;
                    let empty = batch.is_empty();
                    let mut furthest: HashMap<Rank, u64> = HashMap::new();
                    for m in &batch {
                        verify(m)?;
                        let previous = seen[m.origin as usize];
                        if m.sequence <= previous {
                            return Err(format!(
                                "sequence from {} went backwards: {} after {previous}",
                                m.origin, m.sequence
                            ));
                        }
                        if safe && m.sequence != previous + 1 {
                            return Err(format!(
                                "safe lane from {} skipped {} to {}",
                                m.origin,
                                previous + 1,
                                m.sequence
                            ));
                        }
                        seen[m.origin as usize] = m.sequence;
                        let f = furthest.entry(m.origin).or_insert(0);
                        *f = (*f).max(m.sequence);
                        total += 1;
                    }
                    for (origin, sequence) in furthest {
                        ring.ack(origin, sequence)
                            .map_err(|e| format!("ack to {origin}: {e}"))?;
                    }
                    count.store(total, Ordering::Relaxed);
                    if finished && empty {
                        return Ok(seen);
                    }
                    if empty {
                        std::thread::yield_now();
                    }
                }
            })
        };

        // Round-robin over peers so every outgoing lane stays hot at once.
        let mut buf = vec![0u8; payload];
        let pace = std::time::Duration::from_nanos(pace_ns);
        let opened = std::time::Instant::now();
        for sequence in 1..=messages {
            for destination in (0..size).filter(|&r| r != rank) {
                paint(rank, sequence, &mut buf);
                ring.send(destination, &buf)
                    .unwrap_or_else(|e| panic!("rank {rank}: send to {destination}: {e}"));
            }
            // Absolute deadlines, so a slow round is not paid for twice and the
            // offered rate stays the one that was asked for.
            if pace_ns > 0 {
                let due = opened + pace * sequence as u32;
                while std::time::Instant::now() < due {
                    std::hint::spin_loop();
                }
            }
        }
        // Every rank's sends have completed at their targets before this
        // returns, so after the barrier nothing new can appear in a slot.
        world.barrier();
        done.store(true, Ordering::Release);

        let seen = poller.join().expect("poller thread panicked");
        stop.store(true, Ordering::Relaxed);
        seen
    });

    let elapsed = started.elapsed().as_secs_f64();
    let fail = |why: String| -> ! {
        eprintln!("[soak] FAIL rank {rank}: {why}");
        world.abort(1)
    };
    let seen = outcome.unwrap_or_else(|why| fail(why));

    let got = count.load(Ordering::Relaxed);
    let lost = ring.lost();
    let corrupt = ring.corrupt();

    // The last message on a lane is never overwritten, so every lane must end
    // exactly at the sender's high-water mark in both modes.
    for origin in (0..size).filter(|&r| r != rank) {
        if seen[origin as usize] != messages {
            fail(format!(
                "lane from {origin} ended at {} not {messages}",
                seen[origin as usize]
            ));
        }
    }
    if safe && lost != 0 {
        fail(format!("safe ring lost {lost}"));
    }
    if got + lost != sent {
        fail(format!("received {got} + lost {lost} != sent {sent}"));
    }

    // Gather the counters and let rank 0 write every row. Interleaving the
    // ranks' own stdout would scatter the header into the middle of the table:
    // mpirun orders a rank's output against itself and nothing else.
    let mine = [
        got,
        lost,
        corrupt,
        ring.max_lag(),
        ring.waits(),
        ring.wait_ns(),
    ];
    let mut all = vec![0u64; mine.len() * size as usize];
    world.all_gather_into(&mine[..], &mut all[..]);
    if rank == 0 {
        println!(
            "mode\tranks\tdepth\tpayload\tnoise_kib\tpace_ns\trep\trank\tsent\treceived\tlost\tcorrupt\tmax_lag\twaits\twait_s\tseconds"
        );
        for (r, row) in all.chunks_exact(mine.len()).enumerate() {
            println!(
                "{mode}\t{size}\t{depth}\t{payload}\t{noise_kib}\t{pace_ns}\t{rep}\t{r}\t{sent}\t{}\t{}\t{}\t{}\t{}\t{:.6}\t{elapsed:.6}",
                row[0],
                row[1],
                row[2],
                row[3],
                row[4],
                row[5] as f64 * 1e-9,
            );
        }
    }

    world.barrier();
    Arc::try_unwrap(ring)
        .unwrap_or_else(|_| panic!("ring still shared at close"))
        .close()
        .unwrap();
    if rank == 0 {
        eprintln!("[soak] ok: {mode}, {size} ranks, {sent} messages/rank in {elapsed:.2}s");
    }
}

pub fn wait_ns(&self) -> u64

Cumulative nanoseconds spent in safe-mode wait spins.

Examples found in repository

examples/test_ring.rs (line 113)

fn backpressure(world: &SimpleCommunicator, rank: Rank) {
    let ring = Ring::safe(world, &oneway(2, 8)).unwrap();
    if rank == 0 {
        assert_eq!(ring.send(1, &[1]).unwrap(), 1);
        assert_eq!(ring.send(1, &[2]).unwrap(), 2);
    }
    world.barrier();

    if rank == 0 {
        let sent = thread::scope(|scope| {
            let sender = ˚
            let handle = scope.spawn(move || sender.send(1, &[3]));
            // Depth is 2 and nothing has been acknowledged, so there is no slot
            // for the third message and the send has to wait.
            let deadline = Instant::now() + Duration::from_secs(5);
            while ring.waits() == 0 {
                assert!(Instant::now() < deadline, "safe sender never blocked");
                thread::yield_now();
            }
            // The receiver is parked here, so it cannot have drained early.
            world.barrier();
            handle.join().expect("sender thread panicked").unwrap()
        });
        assert_eq!(sent, 3);
        assert!(ring.waits() > 0);
        assert!(ring.wait_ns() > 0);
    } else {
        world.barrier();
        let messages = ring.poll().unwrap();
        assert_eq!(messages.len(), 2);
        assert_eq!(messages[0].data, vec![1]);
        assert_eq!(messages[1].data, vec![2]);
        assert!(matches!(
            ring.ack(0, 3),
            Err(Error::Ack {
                sequence: 3,
                received: 2,
                ..
            })
        ));
        ring.ack(0, 2).unwrap();
        ring.ack(0, 2).unwrap();
    }

    world.barrier();
    if rank == 1 {
        let messages = ring.poll().unwrap();
        assert_eq!(messages.len(), 1);
        assert_eq!(messages[0].sequence, 3);
        assert_eq!(messages[0].data, vec![3]);
        ring.ack(0, 3).unwrap();
    }
    world.barrier();
    ring.close().unwrap();
}

examples/compare.rs (line 146)

fn ring_stream(
    comm: &SimpleCommunicator,
    ring: &Ring,
    rank: Rank,
    rep: u32,
    payload: usize,
    n: u64,
) -> (u64, u64, f64, f64, f64) {
    comm.barrier();
    let stalled = ring.wait_ns();
    let t0 = Instant::now();
    if rank == 0 {
        let buf = shuffled(rep, payload);
        let mut sent = 0u64;
        while sent < n {
            ring.send(1, &buf).unwrap();
            sent += 1;
        }
        let inject = t0.elapsed().as_secs_f64();
        let wait = (ring.wait_ns() - stalled) as f64 * 1e-9;
        // Tell the drain side to stop, then collect what it actually saw.
        comm.process_at_rank(1).send_with_tag(&[0u8], TAG_STOP);
        let (reply, _) = comm
            .process_at_rank(1)
            .receive_vec_with_tag::<u64>(TAG_STOP);
        (sent, reply[0], inject, t0.elapsed().as_secs_f64(), wait)
    } else {
        let mut got = 0u64;
        let mut stopping = false;
        loop {
            let batch = ring.poll().unwrap();
            let empty = batch.is_empty();
            let mut furthest = 0u64;
            for m in &batch {
                got += 1;
                furthest = furthest.max(m.sequence);
            }
            if furthest > 0 {
                ring.ack(0, furthest).unwrap();
            }
            if stopping && empty {
                break;
            }
            if !stopping
                && comm
                    .any_process()
                    .immediate_probe_with_tag(TAG_STOP)
                    .is_some()
            {
                comm.any_process().receive_vec_with_tag::<u8>(TAG_STOP);
                stopping = true;
            }
            if empty {
                std::thread::yield_now();
            }
        }
        comm.process_at_rank(0).send_with_tag(&[got][..], TAG_STOP);
        let dt = t0.elapsed().as_secs_f64();
        (0, got, dt, dt, 0.0)
    }
}

/// One message each way, `iters` times, on the ring.
///
/// Returns `(total, max_us)`: the run wall time and the worst single
/// round trip, halved to a one-way latency so it shares the `us_per_msg`
/// unit of the mean.
fn ring_rtt(ring: &Ring, rank: Rank, rep: u32, payload: usize, iters: u64) -> (f64, f64) {
    let buf = shuffled(rep, payload);
    let other: Rank = if rank == 0 { 1 } else { 0 };
    let mut seen = 0u64;
    let await_one = |ring: &Ring, seen: &mut u64| {
        loop {
            let batch = ring.poll().unwrap();
            if let Some(last) = batch.last() {
                *seen = last.sequence;
                ring.ack(other, *seen).unwrap();
                return;
            }
            std::hint::spin_loop();
        }
    };
    let mut max_us = 0.0f64;
    let t0 = Instant::now();
    for _ in 0..iters {
        let start = Instant::now();
        if rank == 0 {
            ring.send(1, &buf).unwrap();
            await_one(ring, &mut seen);
        } else {
            await_one(ring, &mut seen);
            ring.send(0, &buf).unwrap();
        }
        max_us = max_us.max(start.elapsed().as_secs_f64() * 5e5);
    }
    (t0.elapsed().as_secs_f64(), max_us)
}

// ── Point to point ──────────────────────────────────────────────────────────

fn p2p_stream(
    comm: &SimpleCommunicator,
    rank: Rank,
    rep: u32,
    payload: usize,
    n: u64,
    buffered: bool,
) -> (u64, u64, f64, f64) {
    comm.barrier();
    let t0 = Instant::now();
    if rank == 0 {
        let buf = shuffled(rep, payload);
        let peer = comm.process_at_rank(1);
        let mut sent = 0u64;
        while sent < n {
            if buffered {
                peer.buffered_send_with_tag(&buf[..], TAG_DATA);
            } else {
                peer.send_with_tag(&buf[..], TAG_DATA);
            }
            sent += 1;
        }
        let inject = t0.elapsed().as_secs_f64();
        peer.send_with_tag(&[0u8], TAG_STOP);
        // The receiver's count is authoritative and equals the sender's here:
        // p2p does not drop. Waiting for it also drains the queue before the
        // next measurement starts.
        let (reply, _) = comm
            .process_at_rank(1)
            .receive_vec_with_tag::<u64>(TAG_STOP);
        (sent, reply[0], inject, t0.elapsed().as_secs_f64())
    } else {
        let mut got = 0u64;
        let source = comm.process_at_rank(0);
        loop {
            let (_, status) = source.receive_vec::<u8>();
            if status.tag() == TAG_STOP {
                break;
            }
            got += 1;
        }
        comm.process_at_rank(0).send_with_tag(&[got][..], TAG_STOP);
        let dt = t0.elapsed().as_secs_f64();
        (got, got, dt, dt)
    }
}

fn p2p_rtt(
    comm: &SimpleCommunicator,
    rank: Rank,
    rep: u32,
    payload: usize,
    iters: u64,
) -> (f64, f64) {
    let buf = shuffled(rep, payload);
    let peer = comm.process_at_rank(if rank == 0 { 1 } else { 0 });
    let mut max_us = 0.0f64;
    let t0 = Instant::now();
    for _ in 0..iters {
        let start = Instant::now();
        if rank == 0 {
            peer.send_with_tag(&buf[..], TAG_DATA);
            peer.receive_vec_with_tag::<u8>(TAG_DATA);
        } else {
            peer.receive_vec_with_tag::<u8>(TAG_DATA);
            peer.send_with_tag(&buf[..], TAG_DATA);
        }
        max_us = max_us.max(start.elapsed().as_secs_f64() * 5e5);
    }
    (t0.elapsed().as_secs_f64(), max_us)
}

/// Background ping-pong among ranks 2+, in pairs, until `stop`.
fn noise(world: &SimpleCommunicator, kib: usize, stop: &AtomicBool) {
    let rank = world.rank();
    let base = rank - 2;
    let partner = if base % 2 == 0 { rank + 1 } else { rank - 1 };
    if partner >= world.size() {
        return;
    }
    let buf = vec![0xA5u8; kib * 1024];
    let peer = world.process_at_rank(partner);
    loop {
        if rank < partner {
            if stop.load(Ordering::Relaxed) {
                peer.send_with_tag(&[0u8], TAG_QUIT);
                return;
            }
            peer.send_with_tag(&buf[..], TAG_NOISE);
            peer.receive_vec_with_tag::<u8>(TAG_NOISE);
        } else {
            let (_, status) = peer.receive_vec::<u8>();
            if status.tag() == TAG_QUIT {
                return;
            }
            peer.send_with_tag(&buf[..], TAG_NOISE);
        }
    }
}

fn main() {
    let mut args = std::env::args().skip(1);
    let messages: u64 = args.next().and_then(|s| s.parse().ok()).unwrap_or(20_000);
    let depth: usize = args.next().and_then(|s| s.parse().ok()).unwrap_or(32);
    let noise_kib: usize = args.next().and_then(|s| s.parse().ok()).unwrap_or(0);
    let repeats: u32 = args.next().and_then(|s| s.parse().ok()).unwrap_or(5);

    let (mut universe, provided) =
        mpi::initialize_with_threading(Threading::Multiple).expect("MPI must initialize once");
    assert_eq!(provided, Threading::Multiple);
    let world = universe.world();
    let rank = world.rank();
    let size = world.size();
    assert!(size >= 2, "compare needs at least 2 ranks");
    assert!(noise_kib == 0 || size >= 4, "noise needs at least 4 ranks");
    // `p2p-bsend` needs somewhere to copy to. Only rank 0 sends buffered, so
    // only it attaches the buffer; every rank allocating 512 MiB would need
    // tens of GiB at scale. Sized past the largest amount a stream can leave
    // in flight, so the comparison measures the transport and not a buffer
    // running out.
    if rank == 0 {
        universe.set_buffer_size(BSEND_BUF);
    }

    let stop = Arc::new(AtomicBool::new(false));
    let noisy = noise_kib > 0;

    std::thread::scope(|scope| {
        if noisy && rank >= 2 {
            let stop = Arc::clone(&stop);
            let shared = Shared(&world);
            scope.spawn(move || {
                let shared = shared;
                noise(shared.0, noise_kib, &stop);
            });
        }
        if rank > 1 {
            // Noise ranks take no part in the measurement. They wait here for
            // rank 0 to call time, then let their noise thread finish.
            world
                .process_at_rank(0)
                .receive_vec_with_tag::<u8>(TAG_QUIT);
            stop.store(true, Ordering::Relaxed);
            return;
        }

        if rank == 0 {
            println!("{HEADER}");
        }

        // The measured pair only. A world barrier here would deadlock against
        // ranks sitting in the noise loop.
        let pair: Vec<Rank> = vec![0, 1];
        let group = world.group().include(&pair);
        let duo = world
            .split_by_subgroup(&group)
            .expect("ranks 0 and 1 form the measured pair");

        for rep in 0..repeats {
            for payload in SIZES {
                // Both ring modes share the lane shape, so the only difference
                // between the two rows is the acknowledgement gate.
                let lanes = vec![
                    (0 as Rank, 1 as Rank, depth, payload),
                    (1, 0, depth, payload),
                ];
                // Cap the stream by bytes as well as count: 256 KiB x 20k would be
                // 5 GiB through the wire for one row.
                let n = messages.min((STREAM_BYTES / payload as u64).max(64));
                let iters = if payload >= 32768 { 500 } else { 2000 };

                for (name, safe) in [("ring-safe", true), ("ring-raw", false)] {
                    let ring = if safe {
                        Ring::safe(&duo, &lanes).unwrap()
                    } else {
                        Ring::raw(&duo, &lanes).unwrap()
                    };
                    let (sent, got, inject, total, wait) =
                        ring_stream(&duo, &ring, rank, rep, payload, n);
                    if rank == 0 {
                        Row {
                            transport: name,
                            measure: "stream",
                            payload,
                            sent,
                            delivered: got,
                            inject,
                            total,
                            wait,
                            max_us: 0.0,
                        }
                        .print(size, depth, noise_kib, rep);
                    }
                    duo.barrier();
                    let stalled = ring.wait_ns();
                    let (dt, max_us) = ring_rtt(&ring, rank, rep, payload, iters);
                    if rank == 0 {
                        Row {
                            transport: name,
                            measure: "rtt",
                            payload,
                            sent: iters * 2,
                            delivered: iters * 2,
                            inject: dt,
                            total: dt,
                            wait: (ring.wait_ns() - stalled) as f64 * 1e-9,
                            max_us,
                        }
                        .print(size, depth, noise_kib, rep);
                    }
                    duo.barrier();
                    ring.close().unwrap();
                }

                for (name, buffered) in [("p2p-send", false), ("p2p-bsend", true)] {
                    let (sent, got, inject, total) =
                        p2p_stream(&duo, rank, rep, payload, n, buffered);
                    if rank == 0 {
                        Row {
                            transport: name,
                            measure: "stream",
                            payload,
                            sent,
                            delivered: got,
                            inject,
                            total,
                            wait: 0.0,
                            max_us: 0.0,
                        }
                        .print(size, depth, noise_kib, rep);
                    }
                    duo.barrier();
                }
                let (dt, max_us) = p2p_rtt(&duo, rank, rep, payload, iters);
                if rank == 0 {
                    Row {
                        transport: "p2p-send",
                        measure: "rtt",
                        payload,
                        sent: iters * 2,
                        delivered: iters * 2,
                        inject: dt,
                        total: dt,
                        wait: 0.0,
                        max_us,
                    }
                    .print(size, depth, noise_kib, rep);
                }
                duo.barrier();
            }
        }

        if rank == 0 {
            for r in 2..size {
                world.process_at_rank(r).send_with_tag(&[0u8], TAG_QUIT);
            }
        }
    });

    // Buffered sends must be drained before MPI tears the buffer down.
    world.barrier();
}

examples/soak.rs (line 300)

fn main() {
    let mut args = std::env::args().skip(1);
    let mode = args.next().unwrap_or_else(|| "safe".into());
    let messages: u64 = args.next().and_then(|s| s.parse().ok()).unwrap_or(20_000);
    let payload: usize = args.next().and_then(|s| s.parse().ok()).unwrap_or(256);
    let depth: usize = args.next().and_then(|s| s.parse().ok()).unwrap_or(8);
    let noise_kib: usize = args.next().and_then(|s| s.parse().ok()).unwrap_or(0);
    let pace_ns: u64 = args.next().and_then(|s| s.parse().ok()).unwrap_or(0);
    // Echoed straight into the output so repeats of one point stay distinguishable.
    let rep: u32 = args.next().and_then(|s| s.parse().ok()).unwrap_or(0);
    assert!(
        payload >= STAMP,
        "payload must hold the origin and sequence"
    );
    let safe = match mode.as_str() {
        "safe" => true,
        "raw" => false,
        other => panic!("mode must be safe or raw, got {other:?}"),
    };

    let (universe, provided) =
        mpi::initialize_with_threading(Threading::Multiple).expect("MPI must initialize once");
    assert_eq!(provided, Threading::Multiple);
    let world = universe.world();
    let rank = world.rank();
    let size = world.size();
    assert!(size >= 2, "soak needs at least 2 ranks");

    // All-to-all: every ordered pair of distinct ranks gets a lane.
    let mut lanes = Vec::new();
    for source in 0..size {
        for destination in 0..size {
            if source != destination {
                lanes.push((source, destination, depth, payload));
            }
        }
    }
    let ring = Arc::new(if safe {
        Ring::safe(&world, &lanes).unwrap()
    } else {
        Ring::raw(&world, &lanes).unwrap()
    });

    let stop = Arc::new(AtomicBool::new(false));
    let done = Arc::new(AtomicBool::new(false));
    let count = Arc::new(AtomicU64::new(0));
    let peers = (size - 1) as u64;
    let sent = messages * peers;

    world.barrier();
    let started = std::time::Instant::now();

    let outcome = std::thread::scope(|scope| {
        if noise_kib > 0 {
            let stop = Arc::clone(&stop);
            let shared = Shared(&world);
            scope.spawn(move || {
                // Name the wrapper as a whole first. Closures capture the
                // narrowest place they use, and `shared.0` alone would be
                // captured as a bare reference, losing the `Send` impl.
                let shared = shared;
                noise(shared.0, noise_kib, &stop);
            });
        }

        let poller = {
            let (ring, done, count) = (Arc::clone(&ring), Arc::clone(&done), Arc::clone(&count));
            scope.spawn(move || -> Result<Vec<u64>, String> {
                let mut seen = vec![0u64; size as usize];
                let mut total = 0u64;
                loop {
                    // If true, this poll follows the global send barrier.
                    let finished = done.load(Ordering::Acquire);
                    let batch = ring.poll().map_err(|e| format!("poll: {e}"))?;
                    let empty = batch.is_empty();
                    let mut furthest: HashMap<Rank, u64> = HashMap::new();
                    for m in &batch {
                        verify(m)?;
                        let previous = seen[m.origin as usize];
                        if m.sequence <= previous {
                            return Err(format!(
                                "sequence from {} went backwards: {} after {previous}",
                                m.origin, m.sequence
                            ));
                        }
                        if safe && m.sequence != previous + 1 {
                            return Err(format!(
                                "safe lane from {} skipped {} to {}",
                                m.origin,
                                previous + 1,
                                m.sequence
                            ));
                        }
                        seen[m.origin as usize] = m.sequence;
                        let f = furthest.entry(m.origin).or_insert(0);
                        *f = (*f).max(m.sequence);
                        total += 1;
                    }
                    for (origin, sequence) in furthest {
                        ring.ack(origin, sequence)
                            .map_err(|e| format!("ack to {origin}: {e}"))?;
                    }
                    count.store(total, Ordering::Relaxed);
                    if finished && empty {
                        return Ok(seen);
                    }
                    if empty {
                        std::thread::yield_now();
                    }
                }
            })
        };

        // Round-robin over peers so every outgoing lane stays hot at once.
        let mut buf = vec![0u8; payload];
        let pace = std::time::Duration::from_nanos(pace_ns);
        let opened = std::time::Instant::now();
        for sequence in 1..=messages {
            for destination in (0..size).filter(|&r| r != rank) {
                paint(rank, sequence, &mut buf);
                ring.send(destination, &buf)
                    .unwrap_or_else(|e| panic!("rank {rank}: send to {destination}: {e}"));
            }
            // Absolute deadlines, so a slow round is not paid for twice and the
            // offered rate stays the one that was asked for.
            if pace_ns > 0 {
                let due = opened + pace * sequence as u32;
                while std::time::Instant::now() < due {
                    std::hint::spin_loop();
                }
            }
        }
        // Every rank's sends have completed at their targets before this
        // returns, so after the barrier nothing new can appear in a slot.
        world.barrier();
        done.store(true, Ordering::Release);

        let seen = poller.join().expect("poller thread panicked");
        stop.store(true, Ordering::Relaxed);
        seen
    });

    let elapsed = started.elapsed().as_secs_f64();
    let fail = |why: String| -> ! {
        eprintln!("[soak] FAIL rank {rank}: {why}");
        world.abort(1)
    };
    let seen = outcome.unwrap_or_else(|why| fail(why));

    let got = count.load(Ordering::Relaxed);
    let lost = ring.lost();
    let corrupt = ring.corrupt();

    // The last message on a lane is never overwritten, so every lane must end
    // exactly at the sender's high-water mark in both modes.
    for origin in (0..size).filter(|&r| r != rank) {
        if seen[origin as usize] != messages {
            fail(format!(
                "lane from {origin} ended at {} not {messages}",
                seen[origin as usize]
            ));
        }
    }
    if safe && lost != 0 {
        fail(format!("safe ring lost {lost}"));
    }
    if got + lost != sent {
        fail(format!("received {got} + lost {lost} != sent {sent}"));
    }

    // Gather the counters and let rank 0 write every row. Interleaving the
    // ranks' own stdout would scatter the header into the middle of the table:
    // mpirun orders a rank's output against itself and nothing else.
    let mine = [
        got,
        lost,
        corrupt,
        ring.max_lag(),
        ring.waits(),
        ring.wait_ns(),
    ];
    let mut all = vec![0u64; mine.len() * size as usize];
    world.all_gather_into(&mine[..], &mut all[..]);
    if rank == 0 {
        println!(
            "mode\tranks\tdepth\tpayload\tnoise_kib\tpace_ns\trep\trank\tsent\treceived\tlost\tcorrupt\tmax_lag\twaits\twait_s\tseconds"
        );
        for (r, row) in all.chunks_exact(mine.len()).enumerate() {
            println!(
                "{mode}\t{size}\t{depth}\t{payload}\t{noise_kib}\t{pace_ns}\t{rep}\t{r}\t{sent}\t{}\t{}\t{}\t{}\t{}\t{:.6}\t{elapsed:.6}",
                row[0],
                row[1],
                row[2],
                row[3],
                row[4],
                row[5] as f64 * 1e-9,
            );
        }
    }

    world.barrier();
    Arc::try_unwrap(ring)
        .unwrap_or_else(|_| panic!("ring still shared at close"))
        .close()
        .unwrap();
    if rank == 0 {
        eprintln!("[soak] ok: {mode}, {size} ranks, {sent} messages/rank in {elapsed:.2}s");
    }
}

pub fn send(&self, destination: Rank, data: &[u8]) -> Result<u64, Error>

Send one message and return its sequence number.

Completes the put at the target before returning. In safe mode, spins on yield_now if the destination has not yet acknowledged enough earlier messages to make room in the slot ring; the wait is recorded in Self::waits and Self::wait_ns.

Errors

• Error::Rank if destination is outside the communicator.
• Error::Ring if no lane to destination is configured, the sequence number would overflow, the acknowledgement counter regresses or exceeds what was sent, or the send state is poisoned.
• Error::Payload if data does not fit the lane’s capacity.
• Whatever the underlying MPI_Put fails with (wrapped), e.g. Error::Mpi or Error::Range.

Examples found in repository

examples/test_layout.rs (line 29)

fn run(world: &SimpleCommunicator, safe: bool) {
    let rank = world.rank();
    let ring = if safe {
        Ring::safe(world, &lanes(2, 8)).unwrap()
    } else {
        Ring::raw(world, &lanes(2, 8)).unwrap()
    };

    if rank < 2 {
        ring.send(2, &[rank as u8, 1]).unwrap();
    }
    world.barrier();

    if rank == 2 {
        let messages = ring.poll().unwrap();
        assert_eq!(messages.len(), 2);
        assert_eq!(messages[0].origin, 0);
        assert_eq!(messages[0].data, vec![0, 1]);
        assert_eq!(messages[1].origin, 1);
        assert_eq!(messages[1].data, vec![1, 1]);
        ring.ack(0, 1).unwrap();
        ring.ack(1, 1).unwrap();
        ring.send(0, &[2, 1]).unwrap();
        ring.send(1, &[2, 1]).unwrap();
    }
    world.barrier();

    if rank < 2 {
        let messages = ring.poll().unwrap();
        assert_eq!(
            messages,
            vec![Message {
                origin: 2,
                sequence: 1,
                data: vec![2, 1]
            }]
        );
        ring.ack(2, 1).unwrap();
    }
    world.barrier();

    ring.close().unwrap();
}

examples/test_ring.rs (line 31)

fn safe_basics(world: &SimpleCommunicator, rank: Rank, next: Rank, prev: Rank) {
    // Depth 4 against 3 messages: this covers ordering and acknowledge
    // validation with the overwrite gate deliberately out of the way.
    // Backpressure is its own test, on a one-way lane.
    let ring = Ring::safe(world, &pair(4, 8)).unwrap();
    assert!(ring.is_safe());
    assert_eq!(ring.depth(next), Some(4));
    assert_eq!(ring.capacity(next), Some(8));

    assert!(matches!(
        ring.send(next, &[0; 9]),
        Err(Error::Payload {
            len: 9,
            capacity: 8
        })
    ));
    assert!(matches!(
        ring.ack(prev, 1),
        Err(Error::Ack {
            sequence: 1,
            received: 0,
            ..
        })
    ));

    for s in 1..=3u8 {
        assert_eq!(ring.send(next, &[rank as u8, s]).unwrap(), u64::from(s));
    }
    world.barrier();

    let messages = ring.poll().unwrap();
    assert_eq!(
        messages,
        (1..=3)
            .map(|s| Message {
                origin: prev,
                sequence: s,
                data: vec![prev as u8, s as u8],
            })
            .collect::<Vec<_>>()
    );
    assert!(ring.poll().unwrap().is_empty());
    assert_eq!(ring.lost(), 0);
    assert_eq!(ring.max_lag(), 3);

    assert!(matches!(
        ring.ack(prev, 4),
        Err(Error::Ack {
            sequence: 4,
            received: 3,
            ..
        })
    ));
    ring.ack(prev, 3).unwrap();
    ring.ack(prev, 3).unwrap();
    ring.ack(prev, 2).unwrap();

    world.barrier();
    ring.close().unwrap();
}

/// A safe sender with no free slot blocks until the receiver acknowledges.
///
/// One-way, so the ack timing belongs to the test. On a symmetric pair each
/// rank's ack races the other's gate check: whichever receiver drains first
/// opens its peer's gate before that peer ever evaluates it, and neither side
/// can be relied on to block.
fn backpressure(world: &SimpleCommunicator, rank: Rank) {
    let ring = Ring::safe(world, &oneway(2, 8)).unwrap();
    if rank == 0 {
        assert_eq!(ring.send(1, &[1]).unwrap(), 1);
        assert_eq!(ring.send(1, &[2]).unwrap(), 2);
    }
    world.barrier();

    if rank == 0 {
        let sent = thread::scope(|scope| {
            let sender = &ring;
            let handle = scope.spawn(move || sender.send(1, &[3]));
            // Depth is 2 and nothing has been acknowledged, so there is no slot
            // for the third message and the send has to wait.
            let deadline = Instant::now() + Duration::from_secs(5);
            while ring.waits() == 0 {
                assert!(Instant::now() < deadline, "safe sender never blocked");
                thread::yield_now();
            }
            // The receiver is parked here, so it cannot have drained early.
            world.barrier();
            handle.join().expect("sender thread panicked").unwrap()
        });
        assert_eq!(sent, 3);
        assert!(ring.waits() > 0);
        assert!(ring.wait_ns() > 0);
    } else {
        world.barrier();
        let messages = ring.poll().unwrap();
        assert_eq!(messages.len(), 2);
        assert_eq!(messages[0].data, vec![1]);
        assert_eq!(messages[1].data, vec![2]);
        assert!(matches!(
            ring.ack(0, 3),
            Err(Error::Ack {
                sequence: 3,
                received: 2,
                ..
            })
        ));
        ring.ack(0, 2).unwrap();
        ring.ack(0, 2).unwrap();
    }

    world.barrier();
    if rank == 1 {
        let messages = ring.poll().unwrap();
        assert_eq!(messages.len(), 1);
        assert_eq!(messages[0].sequence, 3);
        assert_eq!(messages[0].data, vec![3]);
        ring.ack(0, 3).unwrap();
    }
    world.barrier();
    ring.close().unwrap();
}

fn oneway_basics(world: &SimpleCommunicator, rank: Rank) {
    let ring = Ring::safe(world, &oneway(2, 8)).unwrap();
    if rank == 0 {
        assert_eq!(ring.send(1, &[7]).unwrap(), 1);
    }
    world.barrier();
    if rank == 1 {
        let messages = ring.poll().unwrap();
        assert_eq!(messages.len(), 1);
        assert_eq!(messages[0].origin, 0);
        assert_eq!(messages[0].data, vec![7]);
        ring.ack(0, 1).unwrap();
    }
    world.barrier();
    ring.close().unwrap();
}

fn raw_basics(world: &SimpleCommunicator, rank: Rank, next: Rank, prev: Rank) {
    let ring = Ring::raw(world, &pair(2, 8)).unwrap();
    assert!(!ring.is_safe());

    for s in 1..=3u8 {
        assert_eq!(ring.send(next, &[rank as u8, s]).unwrap(), u64::from(s));
    }
    world.barrier();

    let messages = ring.poll().unwrap();
    assert_eq!(
        messages,
        vec![
            Message {
                origin: prev,
                sequence: 2,
                data: vec![prev as u8, 2]
            },
            Message {
                origin: prev,
                sequence: 3,
                data: vec![prev as u8, 3]
            },
        ]
    );
    assert_eq!(ring.lost(), 1);

    // Raw acks are no-ops.
    ring.ack(prev, u64::MAX).unwrap();
    world.barrier();
    ring.close().unwrap();
}

examples/compare.rs (line 152)

fn ring_stream(
    comm: &SimpleCommunicator,
    ring: &Ring,
    rank: Rank,
    rep: u32,
    payload: usize,
    n: u64,
) -> (u64, u64, f64, f64, f64) {
    comm.barrier();
    let stalled = ring.wait_ns();
    let t0 = Instant::now();
    if rank == 0 {
        let buf = shuffled(rep, payload);
        let mut sent = 0u64;
        while sent < n {
            ring.send(1, &buf).unwrap();
            sent += 1;
        }
        let inject = t0.elapsed().as_secs_f64();
        let wait = (ring.wait_ns() - stalled) as f64 * 1e-9;
        // Tell the drain side to stop, then collect what it actually saw.
        comm.process_at_rank(1).send_with_tag(&[0u8], TAG_STOP);
        let (reply, _) = comm
            .process_at_rank(1)
            .receive_vec_with_tag::<u64>(TAG_STOP);
        (sent, reply[0], inject, t0.elapsed().as_secs_f64(), wait)
    } else {
        let mut got = 0u64;
        let mut stopping = false;
        loop {
            let batch = ring.poll().unwrap();
            let empty = batch.is_empty();
            let mut furthest = 0u64;
            for m in &batch {
                got += 1;
                furthest = furthest.max(m.sequence);
            }
            if furthest > 0 {
                ring.ack(0, furthest).unwrap();
            }
            if stopping && empty {
                break;
            }
            if !stopping
                && comm
                    .any_process()
                    .immediate_probe_with_tag(TAG_STOP)
                    .is_some()
            {
                comm.any_process().receive_vec_with_tag::<u8>(TAG_STOP);
                stopping = true;
            }
            if empty {
                std::thread::yield_now();
            }
        }
        comm.process_at_rank(0).send_with_tag(&[got][..], TAG_STOP);
        let dt = t0.elapsed().as_secs_f64();
        (0, got, dt, dt, 0.0)
    }
}

/// One message each way, `iters` times, on the ring.
///
/// Returns `(total, max_us)`: the run wall time and the worst single
/// round trip, halved to a one-way latency so it shares the `us_per_msg`
/// unit of the mean.
fn ring_rtt(ring: &Ring, rank: Rank, rep: u32, payload: usize, iters: u64) -> (f64, f64) {
    let buf = shuffled(rep, payload);
    let other: Rank = if rank == 0 { 1 } else { 0 };
    let mut seen = 0u64;
    let await_one = |ring: &Ring, seen: &mut u64| {
        loop {
            let batch = ring.poll().unwrap();
            if let Some(last) = batch.last() {
                *seen = last.sequence;
                ring.ack(other, *seen).unwrap();
                return;
            }
            std::hint::spin_loop();
        }
    };
    let mut max_us = 0.0f64;
    let t0 = Instant::now();
    for _ in 0..iters {
        let start = Instant::now();
        if rank == 0 {
            ring.send(1, &buf).unwrap();
            await_one(ring, &mut seen);
        } else {
            await_one(ring, &mut seen);
            ring.send(0, &buf).unwrap();
        }
        max_us = max_us.max(start.elapsed().as_secs_f64() * 5e5);
    }
    (t0.elapsed().as_secs_f64(), max_us)
}

examples/soak.rs (line 241)

fn main() {
    let mut args = std::env::args().skip(1);
    let mode = args.next().unwrap_or_else(|| "safe".into());
    let messages: u64 = args.next().and_then(|s| s.parse().ok()).unwrap_or(20_000);
    let payload: usize = args.next().and_then(|s| s.parse().ok()).unwrap_or(256);
    let depth: usize = args.next().and_then(|s| s.parse().ok()).unwrap_or(8);
    let noise_kib: usize = args.next().and_then(|s| s.parse().ok()).unwrap_or(0);
    let pace_ns: u64 = args.next().and_then(|s| s.parse().ok()).unwrap_or(0);
    // Echoed straight into the output so repeats of one point stay distinguishable.
    let rep: u32 = args.next().and_then(|s| s.parse().ok()).unwrap_or(0);
    assert!(
        payload >= STAMP,
        "payload must hold the origin and sequence"
    );
    let safe = match mode.as_str() {
        "safe" => true,
        "raw" => false,
        other => panic!("mode must be safe or raw, got {other:?}"),
    };

    let (universe, provided) =
        mpi::initialize_with_threading(Threading::Multiple).expect("MPI must initialize once");
    assert_eq!(provided, Threading::Multiple);
    let world = universe.world();
    let rank = world.rank();
    let size = world.size();
    assert!(size >= 2, "soak needs at least 2 ranks");

    // All-to-all: every ordered pair of distinct ranks gets a lane.
    let mut lanes = Vec::new();
    for source in 0..size {
        for destination in 0..size {
            if source != destination {
                lanes.push((source, destination, depth, payload));
            }
        }
    }
    let ring = Arc::new(if safe {
        Ring::safe(&world, &lanes).unwrap()
    } else {
        Ring::raw(&world, &lanes).unwrap()
    });

    let stop = Arc::new(AtomicBool::new(false));
    let done = Arc::new(AtomicBool::new(false));
    let count = Arc::new(AtomicU64::new(0));
    let peers = (size - 1) as u64;
    let sent = messages * peers;

    world.barrier();
    let started = std::time::Instant::now();

    let outcome = std::thread::scope(|scope| {
        if noise_kib > 0 {
            let stop = Arc::clone(&stop);
            let shared = Shared(&world);
            scope.spawn(move || {
                // Name the wrapper as a whole first. Closures capture the
                // narrowest place they use, and `shared.0` alone would be
                // captured as a bare reference, losing the `Send` impl.
                let shared = shared;
                noise(shared.0, noise_kib, &stop);
            });
        }

        let poller = {
            let (ring, done, count) = (Arc::clone(&ring), Arc::clone(&done), Arc::clone(&count));
            scope.spawn(move || -> Result<Vec<u64>, String> {
                let mut seen = vec![0u64; size as usize];
                let mut total = 0u64;
                loop {
                    // If true, this poll follows the global send barrier.
                    let finished = done.load(Ordering::Acquire);
                    let batch = ring.poll().map_err(|e| format!("poll: {e}"))?;
                    let empty = batch.is_empty();
                    let mut furthest: HashMap<Rank, u64> = HashMap::new();
                    for m in &batch {
                        verify(m)?;
                        let previous = seen[m.origin as usize];
                        if m.sequence <= previous {
                            return Err(format!(
                                "sequence from {} went backwards: {} after {previous}",
                                m.origin, m.sequence
                            ));
                        }
                        if safe && m.sequence != previous + 1 {
                            return Err(format!(
                                "safe lane from {} skipped {} to {}",
                                m.origin,
                                previous + 1,
                                m.sequence
                            ));
                        }
                        seen[m.origin as usize] = m.sequence;
                        let f = furthest.entry(m.origin).or_insert(0);
                        *f = (*f).max(m.sequence);
                        total += 1;
                    }
                    for (origin, sequence) in furthest {
                        ring.ack(origin, sequence)
                            .map_err(|e| format!("ack to {origin}: {e}"))?;
                    }
                    count.store(total, Ordering::Relaxed);
                    if finished && empty {
                        return Ok(seen);
                    }
                    if empty {
                        std::thread::yield_now();
                    }
                }
            })
        };

        // Round-robin over peers so every outgoing lane stays hot at once.
        let mut buf = vec![0u8; payload];
        let pace = std::time::Duration::from_nanos(pace_ns);
        let opened = std::time::Instant::now();
        for sequence in 1..=messages {
            for destination in (0..size).filter(|&r| r != rank) {
                paint(rank, sequence, &mut buf);
                ring.send(destination, &buf)
                    .unwrap_or_else(|e| panic!("rank {rank}: send to {destination}: {e}"));
            }
            // Absolute deadlines, so a slow round is not paid for twice and the
            // offered rate stays the one that was asked for.
            if pace_ns > 0 {
                let due = opened + pace * sequence as u32;
                while std::time::Instant::now() < due {
                    std::hint::spin_loop();
                }
            }
        }
        // Every rank's sends have completed at their targets before this
        // returns, so after the barrier nothing new can appear in a slot.
        world.barrier();
        done.store(true, Ordering::Release);

        let seen = poller.join().expect("poller thread panicked");
        stop.store(true, Ordering::Relaxed);
        seen
    });

    let elapsed = started.elapsed().as_secs_f64();
    let fail = |why: String| -> ! {
        eprintln!("[soak] FAIL rank {rank}: {why}");
        world.abort(1)
    };
    let seen = outcome.unwrap_or_else(|why| fail(why));

    let got = count.load(Ordering::Relaxed);
    let lost = ring.lost();
    let corrupt = ring.corrupt();

    // The last message on a lane is never overwritten, so every lane must end
    // exactly at the sender's high-water mark in both modes.
    for origin in (0..size).filter(|&r| r != rank) {
        if seen[origin as usize] != messages {
            fail(format!(
                "lane from {origin} ended at {} not {messages}",
                seen[origin as usize]
            ));
        }
    }
    if safe && lost != 0 {
        fail(format!("safe ring lost {lost}"));
    }
    if got + lost != sent {
        fail(format!("received {got} + lost {lost} != sent {sent}"));
    }

    // Gather the counters and let rank 0 write every row. Interleaving the
    // ranks' own stdout would scatter the header into the middle of the table:
    // mpirun orders a rank's output against itself and nothing else.
    let mine = [
        got,
        lost,
        corrupt,
        ring.max_lag(),
        ring.waits(),
        ring.wait_ns(),
    ];
    let mut all = vec![0u64; mine.len() * size as usize];
    world.all_gather_into(&mine[..], &mut all[..]);
    if rank == 0 {
        println!(
            "mode\tranks\tdepth\tpayload\tnoise_kib\tpace_ns\trep\trank\tsent\treceived\tlost\tcorrupt\tmax_lag\twaits\twait_s\tseconds"
        );
        for (r, row) in all.chunks_exact(mine.len()).enumerate() {
            println!(
                "{mode}\t{size}\t{depth}\t{payload}\t{noise_kib}\t{pace_ns}\t{rep}\t{r}\t{sent}\t{}\t{}\t{}\t{}\t{}\t{:.6}\t{elapsed:.6}",
                row[0],
                row[1],
                row[2],
                row[3],
                row[4],
                row[5] as f64 * 1e-9,
            );
        }
    }

    world.barrier();
    Arc::try_unwrap(ring)
        .unwrap_or_else(|_| panic!("ring still shared at close"))
        .close()
        .unwrap();
    if rank == 0 {
        eprintln!("[soak] ok: {mode}, {size} ranks, {sent} messages/rank in {elapsed:.2}s");
    }
}

pub fn poll(&self) -> Result<Vec<Message>, Error>

Drain every incoming lane and return newly observed messages.

Does not call MPI. Safe mode delivers in-order; raw mode scans all slots when one is lapped, sorts the survivors, and counts the gaps into Self::lost. The per-origin seen cursor only advances.

Errors

• Error::Lapped if a safe lane was overwritten before consumption, which the acknowledge gate makes impossible and so indicates a corrupted counter or window rather than backpressure.
• Any window read error from the local slot memory.

Examples found in repository

examples/test_layout.rs (line 34)

fn run(world: &SimpleCommunicator, safe: bool) {
    let rank = world.rank();
    let ring = if safe {
        Ring::safe(world, &lanes(2, 8)).unwrap()
    } else {
        Ring::raw(world, &lanes(2, 8)).unwrap()
    };

    if rank < 2 {
        ring.send(2, &[rank as u8, 1]).unwrap();
    }
    world.barrier();

    if rank == 2 {
        let messages = ring.poll().unwrap();
        assert_eq!(messages.len(), 2);
        assert_eq!(messages[0].origin, 0);
        assert_eq!(messages[0].data, vec![0, 1]);
        assert_eq!(messages[1].origin, 1);
        assert_eq!(messages[1].data, vec![1, 1]);
        ring.ack(0, 1).unwrap();
        ring.ack(1, 1).unwrap();
        ring.send(0, &[2, 1]).unwrap();
        ring.send(1, &[2, 1]).unwrap();
    }
    world.barrier();

    if rank < 2 {
        let messages = ring.poll().unwrap();
        assert_eq!(
            messages,
            vec![Message {
                origin: 2,
                sequence: 1,
                data: vec![2, 1]
            }]
        );
        ring.ack(2, 1).unwrap();
    }
    world.barrier();

    ring.close().unwrap();
}

examples/test_ring.rs (line 51)

fn safe_basics(world: &SimpleCommunicator, rank: Rank, next: Rank, prev: Rank) {
    // Depth 4 against 3 messages: this covers ordering and acknowledge
    // validation with the overwrite gate deliberately out of the way.
    // Backpressure is its own test, on a one-way lane.
    let ring = Ring::safe(world, &pair(4, 8)).unwrap();
    assert!(ring.is_safe());
    assert_eq!(ring.depth(next), Some(4));
    assert_eq!(ring.capacity(next), Some(8));

    assert!(matches!(
        ring.send(next, &[0; 9]),
        Err(Error::Payload {
            len: 9,
            capacity: 8
        })
    ));
    assert!(matches!(
        ring.ack(prev, 1),
        Err(Error::Ack {
            sequence: 1,
            received: 0,
            ..
        })
    ));

    for s in 1..=3u8 {
        assert_eq!(ring.send(next, &[rank as u8, s]).unwrap(), u64::from(s));
    }
    world.barrier();

    let messages = ring.poll().unwrap();
    assert_eq!(
        messages,
        (1..=3)
            .map(|s| Message {
                origin: prev,
                sequence: s,
                data: vec![prev as u8, s as u8],
            })
            .collect::<Vec<_>>()
    );
    assert!(ring.poll().unwrap().is_empty());
    assert_eq!(ring.lost(), 0);
    assert_eq!(ring.max_lag(), 3);

    assert!(matches!(
        ring.ack(prev, 4),
        Err(Error::Ack {
            sequence: 4,
            received: 3,
            ..
        })
    ));
    ring.ack(prev, 3).unwrap();
    ring.ack(prev, 3).unwrap();
    ring.ack(prev, 2).unwrap();

    world.barrier();
    ring.close().unwrap();
}

/// A safe sender with no free slot blocks until the receiver acknowledges.
///
/// One-way, so the ack timing belongs to the test. On a symmetric pair each
/// rank's ack races the other's gate check: whichever receiver drains first
/// opens its peer's gate before that peer ever evaluates it, and neither side
/// can be relied on to block.
fn backpressure(world: &SimpleCommunicator, rank: Rank) {
    let ring = Ring::safe(world, &oneway(2, 8)).unwrap();
    if rank == 0 {
        assert_eq!(ring.send(1, &[1]).unwrap(), 1);
        assert_eq!(ring.send(1, &[2]).unwrap(), 2);
    }
    world.barrier();

    if rank == 0 {
        let sent = thread::scope(|scope| {
            let sender = &ring;
            let handle = scope.spawn(move || sender.send(1, &[3]));
            // Depth is 2 and nothing has been acknowledged, so there is no slot
            // for the third message and the send has to wait.
            let deadline = Instant::now() + Duration::from_secs(5);
            while ring.waits() == 0 {
                assert!(Instant::now() < deadline, "safe sender never blocked");
                thread::yield_now();
            }
            // The receiver is parked here, so it cannot have drained early.
            world.barrier();
            handle.join().expect("sender thread panicked").unwrap()
        });
        assert_eq!(sent, 3);
        assert!(ring.waits() > 0);
        assert!(ring.wait_ns() > 0);
    } else {
        world.barrier();
        let messages = ring.poll().unwrap();
        assert_eq!(messages.len(), 2);
        assert_eq!(messages[0].data, vec![1]);
        assert_eq!(messages[1].data, vec![2]);
        assert!(matches!(
            ring.ack(0, 3),
            Err(Error::Ack {
                sequence: 3,
                received: 2,
                ..
            })
        ));
        ring.ack(0, 2).unwrap();
        ring.ack(0, 2).unwrap();
    }

    world.barrier();
    if rank == 1 {
        let messages = ring.poll().unwrap();
        assert_eq!(messages.len(), 1);
        assert_eq!(messages[0].sequence, 3);
        assert_eq!(messages[0].data, vec![3]);
        ring.ack(0, 3).unwrap();
    }
    world.barrier();
    ring.close().unwrap();
}

fn oneway_basics(world: &SimpleCommunicator, rank: Rank) {
    let ring = Ring::safe(world, &oneway(2, 8)).unwrap();
    if rank == 0 {
        assert_eq!(ring.send(1, &[7]).unwrap(), 1);
    }
    world.barrier();
    if rank == 1 {
        let messages = ring.poll().unwrap();
        assert_eq!(messages.len(), 1);
        assert_eq!(messages[0].origin, 0);
        assert_eq!(messages[0].data, vec![7]);
        ring.ack(0, 1).unwrap();
    }
    world.barrier();
    ring.close().unwrap();
}

fn raw_basics(world: &SimpleCommunicator, rank: Rank, next: Rank, prev: Rank) {
    let ring = Ring::raw(world, &pair(2, 8)).unwrap();
    assert!(!ring.is_safe());

    for s in 1..=3u8 {
        assert_eq!(ring.send(next, &[rank as u8, s]).unwrap(), u64::from(s));
    }
    world.barrier();

    let messages = ring.poll().unwrap();
    assert_eq!(
        messages,
        vec![
            Message {
                origin: prev,
                sequence: 2,
                data: vec![prev as u8, 2]
            },
            Message {
                origin: prev,
                sequence: 3,
                data: vec![prev as u8, 3]
            },
        ]
    );
    assert_eq!(ring.lost(), 1);

    // Raw acks are no-ops.
    ring.ack(prev, u64::MAX).unwrap();
    world.barrier();
    ring.close().unwrap();
}

examples/compare.rs (line 167)

fn ring_stream(
    comm: &SimpleCommunicator,
    ring: &Ring,
    rank: Rank,
    rep: u32,
    payload: usize,
    n: u64,
) -> (u64, u64, f64, f64, f64) {
    comm.barrier();
    let stalled = ring.wait_ns();
    let t0 = Instant::now();
    if rank == 0 {
        let buf = shuffled(rep, payload);
        let mut sent = 0u64;
        while sent < n {
            ring.send(1, &buf).unwrap();
            sent += 1;
        }
        let inject = t0.elapsed().as_secs_f64();
        let wait = (ring.wait_ns() - stalled) as f64 * 1e-9;
        // Tell the drain side to stop, then collect what it actually saw.
        comm.process_at_rank(1).send_with_tag(&[0u8], TAG_STOP);
        let (reply, _) = comm
            .process_at_rank(1)
            .receive_vec_with_tag::<u64>(TAG_STOP);
        (sent, reply[0], inject, t0.elapsed().as_secs_f64(), wait)
    } else {
        let mut got = 0u64;
        let mut stopping = false;
        loop {
            let batch = ring.poll().unwrap();
            let empty = batch.is_empty();
            let mut furthest = 0u64;
            for m in &batch {
                got += 1;
                furthest = furthest.max(m.sequence);
            }
            if furthest > 0 {
                ring.ack(0, furthest).unwrap();
            }
            if stopping && empty {
                break;
            }
            if !stopping
                && comm
                    .any_process()
                    .immediate_probe_with_tag(TAG_STOP)
                    .is_some()
            {
                comm.any_process().receive_vec_with_tag::<u8>(TAG_STOP);
                stopping = true;
            }
            if empty {
                std::thread::yield_now();
            }
        }
        comm.process_at_rank(0).send_with_tag(&[got][..], TAG_STOP);
        let dt = t0.elapsed().as_secs_f64();
        (0, got, dt, dt, 0.0)
    }
}

/// One message each way, `iters` times, on the ring.
///
/// Returns `(total, max_us)`: the run wall time and the worst single
/// round trip, halved to a one-way latency so it shares the `us_per_msg`
/// unit of the mean.
fn ring_rtt(ring: &Ring, rank: Rank, rep: u32, payload: usize, iters: u64) -> (f64, f64) {
    let buf = shuffled(rep, payload);
    let other: Rank = if rank == 0 { 1 } else { 0 };
    let mut seen = 0u64;
    let await_one = |ring: &Ring, seen: &mut u64| {
        loop {
            let batch = ring.poll().unwrap();
            if let Some(last) = batch.last() {
                *seen = last.sequence;
                ring.ack(other, *seen).unwrap();
                return;
            }
            std::hint::spin_loop();
        }
    };
    let mut max_us = 0.0f64;
    let t0 = Instant::now();
    for _ in 0..iters {
        let start = Instant::now();
        if rank == 0 {
            ring.send(1, &buf).unwrap();
            await_one(ring, &mut seen);
        } else {
            await_one(ring, &mut seen);
            ring.send(0, &buf).unwrap();
        }
        max_us = max_us.max(start.elapsed().as_secs_f64() * 5e5);
    }
    (t0.elapsed().as_secs_f64(), max_us)
}

examples/soak.rs (line 194)

fn main() {
    let mut args = std::env::args().skip(1);
    let mode = args.next().unwrap_or_else(|| "safe".into());
    let messages: u64 = args.next().and_then(|s| s.parse().ok()).unwrap_or(20_000);
    let payload: usize = args.next().and_then(|s| s.parse().ok()).unwrap_or(256);
    let depth: usize = args.next().and_then(|s| s.parse().ok()).unwrap_or(8);
    let noise_kib: usize = args.next().and_then(|s| s.parse().ok()).unwrap_or(0);
    let pace_ns: u64 = args.next().and_then(|s| s.parse().ok()).unwrap_or(0);
    // Echoed straight into the output so repeats of one point stay distinguishable.
    let rep: u32 = args.next().and_then(|s| s.parse().ok()).unwrap_or(0);
    assert!(
        payload >= STAMP,
        "payload must hold the origin and sequence"
    );
    let safe = match mode.as_str() {
        "safe" => true,
        "raw" => false,
        other => panic!("mode must be safe or raw, got {other:?}"),
    };

    let (universe, provided) =
        mpi::initialize_with_threading(Threading::Multiple).expect("MPI must initialize once");
    assert_eq!(provided, Threading::Multiple);
    let world = universe.world();
    let rank = world.rank();
    let size = world.size();
    assert!(size >= 2, "soak needs at least 2 ranks");

    // All-to-all: every ordered pair of distinct ranks gets a lane.
    let mut lanes = Vec::new();
    for source in 0..size {
        for destination in 0..size {
            if source != destination {
                lanes.push((source, destination, depth, payload));
            }
        }
    }
    let ring = Arc::new(if safe {
        Ring::safe(&world, &lanes).unwrap()
    } else {
        Ring::raw(&world, &lanes).unwrap()
    });

    let stop = Arc::new(AtomicBool::new(false));
    let done = Arc::new(AtomicBool::new(false));
    let count = Arc::new(AtomicU64::new(0));
    let peers = (size - 1) as u64;
    let sent = messages * peers;

    world.barrier();
    let started = std::time::Instant::now();

    let outcome = std::thread::scope(|scope| {
        if noise_kib > 0 {
            let stop = Arc::clone(&stop);
            let shared = Shared(&world);
            scope.spawn(move || {
                // Name the wrapper as a whole first. Closures capture the
                // narrowest place they use, and `shared.0` alone would be
                // captured as a bare reference, losing the `Send` impl.
                let shared = shared;
                noise(shared.0, noise_kib, &stop);
            });
        }

        let poller = {
            let (ring, done, count) = (Arc::clone(&ring), Arc::clone(&done), Arc::clone(&count));
            scope.spawn(move || -> Result<Vec<u64>, String> {
                let mut seen = vec![0u64; size as usize];
                let mut total = 0u64;
                loop {
                    // If true, this poll follows the global send barrier.
                    let finished = done.load(Ordering::Acquire);
                    let batch = ring.poll().map_err(|e| format!("poll: {e}"))?;
                    let empty = batch.is_empty();
                    let mut furthest: HashMap<Rank, u64> = HashMap::new();
                    for m in &batch {
                        verify(m)?;
                        let previous = seen[m.origin as usize];
                        if m.sequence <= previous {
                            return Err(format!(
                                "sequence from {} went backwards: {} after {previous}",
                                m.origin, m.sequence
                            ));
                        }
                        if safe && m.sequence != previous + 1 {
                            return Err(format!(
                                "safe lane from {} skipped {} to {}",
                                m.origin,
                                previous + 1,
                                m.sequence
                            ));
                        }
                        seen[m.origin as usize] = m.sequence;
                        let f = furthest.entry(m.origin).or_insert(0);
                        *f = (*f).max(m.sequence);
                        total += 1;
                    }
                    for (origin, sequence) in furthest {
                        ring.ack(origin, sequence)
                            .map_err(|e| format!("ack to {origin}: {e}"))?;
                    }
                    count.store(total, Ordering::Relaxed);
                    if finished && empty {
                        return Ok(seen);
                    }
                    if empty {
                        std::thread::yield_now();
                    }
                }
            })
        };

        // Round-robin over peers so every outgoing lane stays hot at once.
        let mut buf = vec![0u8; payload];
        let pace = std::time::Duration::from_nanos(pace_ns);
        let opened = std::time::Instant::now();
        for sequence in 1..=messages {
            for destination in (0..size).filter(|&r| r != rank) {
                paint(rank, sequence, &mut buf);
                ring.send(destination, &buf)
                    .unwrap_or_else(|e| panic!("rank {rank}: send to {destination}: {e}"));
            }
            // Absolute deadlines, so a slow round is not paid for twice and the
            // offered rate stays the one that was asked for.
            if pace_ns > 0 {
                let due = opened + pace * sequence as u32;
                while std::time::Instant::now() < due {
                    std::hint::spin_loop();
                }
            }
        }
        // Every rank's sends have completed at their targets before this
        // returns, so after the barrier nothing new can appear in a slot.
        world.barrier();
        done.store(true, Ordering::Release);

        let seen = poller.join().expect("poller thread panicked");
        stop.store(true, Ordering::Relaxed);
        seen
    });

    let elapsed = started.elapsed().as_secs_f64();
    let fail = |why: String| -> ! {
        eprintln!("[soak] FAIL rank {rank}: {why}");
        world.abort(1)
    };
    let seen = outcome.unwrap_or_else(|why| fail(why));

    let got = count.load(Ordering::Relaxed);
    let lost = ring.lost();
    let corrupt = ring.corrupt();

    // The last message on a lane is never overwritten, so every lane must end
    // exactly at the sender's high-water mark in both modes.
    for origin in (0..size).filter(|&r| r != rank) {
        if seen[origin as usize] != messages {
            fail(format!(
                "lane from {origin} ended at {} not {messages}",
                seen[origin as usize]
            ));
        }
    }
    if safe && lost != 0 {
        fail(format!("safe ring lost {lost}"));
    }
    if got + lost != sent {
        fail(format!("received {got} + lost {lost} != sent {sent}"));
    }

    // Gather the counters and let rank 0 write every row. Interleaving the
    // ranks' own stdout would scatter the header into the middle of the table:
    // mpirun orders a rank's output against itself and nothing else.
    let mine = [
        got,
        lost,
        corrupt,
        ring.max_lag(),
        ring.waits(),
        ring.wait_ns(),
    ];
    let mut all = vec![0u64; mine.len() * size as usize];
    world.all_gather_into(&mine[..], &mut all[..]);
    if rank == 0 {
        println!(
            "mode\tranks\tdepth\tpayload\tnoise_kib\tpace_ns\trep\trank\tsent\treceived\tlost\tcorrupt\tmax_lag\twaits\twait_s\tseconds"
        );
        for (r, row) in all.chunks_exact(mine.len()).enumerate() {
            println!(
                "{mode}\t{size}\t{depth}\t{payload}\t{noise_kib}\t{pace_ns}\t{rep}\t{r}\t{sent}\t{}\t{}\t{}\t{}\t{}\t{:.6}\t{elapsed:.6}",
                row[0],
                row[1],
                row[2],
                row[3],
                row[4],
                row[5] as f64 * 1e-9,
            );
        }
    }

    world.barrier();
    Arc::try_unwrap(ring)
        .unwrap_or_else(|_| panic!("ring still shared at close"))
        .close()
        .unwrap();
    if rank == 0 {
        eprintln!("[soak] ok: {mode}, {size} ranks, {sent} messages/rank in {elapsed:.2}s");
    }
}

pub fn ack(&self, origin: Rank, sequence: u64) -> Result<(), Error>

Acknowledge all messages from origin through sequence.

Cumulative and idempotent. In safe mode performs one atomic MPI_Fetch_and_op against origin’s counter window. In raw mode this is a no-op (returns Ok).

Errors

• Error::Rank if origin is outside the communicator.
• Error::Ring if no incoming lane from origin is configured, or the counter diverges from the locally tracked value.
• Error::Ack if sequence is past the last received message.

Examples found in repository

examples/test_layout.rs (line 40)

fn run(world: &SimpleCommunicator, safe: bool) {
    let rank = world.rank();
    let ring = if safe {
        Ring::safe(world, &lanes(2, 8)).unwrap()
    } else {
        Ring::raw(world, &lanes(2, 8)).unwrap()
    };

    if rank < 2 {
        ring.send(2, &[rank as u8, 1]).unwrap();
    }
    world.barrier();

    if rank == 2 {
        let messages = ring.poll().unwrap();
        assert_eq!(messages.len(), 2);
        assert_eq!(messages[0].origin, 0);
        assert_eq!(messages[0].data, vec![0, 1]);
        assert_eq!(messages[1].origin, 1);
        assert_eq!(messages[1].data, vec![1, 1]);
        ring.ack(0, 1).unwrap();
        ring.ack(1, 1).unwrap();
        ring.send(0, &[2, 1]).unwrap();
        ring.send(1, &[2, 1]).unwrap();
    }
    world.barrier();

    if rank < 2 {
        let messages = ring.poll().unwrap();
        assert_eq!(
            messages,
            vec![Message {
                origin: 2,
                sequence: 1,
                data: vec![2, 1]
            }]
        );
        ring.ack(2, 1).unwrap();
    }
    world.barrier();

    ring.close().unwrap();
}

examples/test_ring.rs (line 38)

fn safe_basics(world: &SimpleCommunicator, rank: Rank, next: Rank, prev: Rank) {
    // Depth 4 against 3 messages: this covers ordering and acknowledge
    // validation with the overwrite gate deliberately out of the way.
    // Backpressure is its own test, on a one-way lane.
    let ring = Ring::safe(world, &pair(4, 8)).unwrap();
    assert!(ring.is_safe());
    assert_eq!(ring.depth(next), Some(4));
    assert_eq!(ring.capacity(next), Some(8));

    assert!(matches!(
        ring.send(next, &[0; 9]),
        Err(Error::Payload {
            len: 9,
            capacity: 8
        })
    ));
    assert!(matches!(
        ring.ack(prev, 1),
        Err(Error::Ack {
            sequence: 1,
            received: 0,
            ..
        })
    ));

    for s in 1..=3u8 {
        assert_eq!(ring.send(next, &[rank as u8, s]).unwrap(), u64::from(s));
    }
    world.barrier();

    let messages = ring.poll().unwrap();
    assert_eq!(
        messages,
        (1..=3)
            .map(|s| Message {
                origin: prev,
                sequence: s,
                data: vec![prev as u8, s as u8],
            })
            .collect::<Vec<_>>()
    );
    assert!(ring.poll().unwrap().is_empty());
    assert_eq!(ring.lost(), 0);
    assert_eq!(ring.max_lag(), 3);

    assert!(matches!(
        ring.ack(prev, 4),
        Err(Error::Ack {
            sequence: 4,
            received: 3,
            ..
        })
    ));
    ring.ack(prev, 3).unwrap();
    ring.ack(prev, 3).unwrap();
    ring.ack(prev, 2).unwrap();

    world.barrier();
    ring.close().unwrap();
}

/// A safe sender with no free slot blocks until the receiver acknowledges.
///
/// One-way, so the ack timing belongs to the test. On a symmetric pair each
/// rank's ack races the other's gate check: whichever receiver drains first
/// opens its peer's gate before that peer ever evaluates it, and neither side
/// can be relied on to block.
fn backpressure(world: &SimpleCommunicator, rank: Rank) {
    let ring = Ring::safe(world, &oneway(2, 8)).unwrap();
    if rank == 0 {
        assert_eq!(ring.send(1, &[1]).unwrap(), 1);
        assert_eq!(ring.send(1, &[2]).unwrap(), 2);
    }
    world.barrier();

    if rank == 0 {
        let sent = thread::scope(|scope| {
            let sender = &ring;
            let handle = scope.spawn(move || sender.send(1, &[3]));
            // Depth is 2 and nothing has been acknowledged, so there is no slot
            // for the third message and the send has to wait.
            let deadline = Instant::now() + Duration::from_secs(5);
            while ring.waits() == 0 {
                assert!(Instant::now() < deadline, "safe sender never blocked");
                thread::yield_now();
            }
            // The receiver is parked here, so it cannot have drained early.
            world.barrier();
            handle.join().expect("sender thread panicked").unwrap()
        });
        assert_eq!(sent, 3);
        assert!(ring.waits() > 0);
        assert!(ring.wait_ns() > 0);
    } else {
        world.barrier();
        let messages = ring.poll().unwrap();
        assert_eq!(messages.len(), 2);
        assert_eq!(messages[0].data, vec![1]);
        assert_eq!(messages[1].data, vec![2]);
        assert!(matches!(
            ring.ack(0, 3),
            Err(Error::Ack {
                sequence: 3,
                received: 2,
                ..
            })
        ));
        ring.ack(0, 2).unwrap();
        ring.ack(0, 2).unwrap();
    }

    world.barrier();
    if rank == 1 {
        let messages = ring.poll().unwrap();
        assert_eq!(messages.len(), 1);
        assert_eq!(messages[0].sequence, 3);
        assert_eq!(messages[0].data, vec![3]);
        ring.ack(0, 3).unwrap();
    }
    world.barrier();
    ring.close().unwrap();
}

fn oneway_basics(world: &SimpleCommunicator, rank: Rank) {
    let ring = Ring::safe(world, &oneway(2, 8)).unwrap();
    if rank == 0 {
        assert_eq!(ring.send(1, &[7]).unwrap(), 1);
    }
    world.barrier();
    if rank == 1 {
        let messages = ring.poll().unwrap();
        assert_eq!(messages.len(), 1);
        assert_eq!(messages[0].origin, 0);
        assert_eq!(messages[0].data, vec![7]);
        ring.ack(0, 1).unwrap();
    }
    world.barrier();
    ring.close().unwrap();
}

fn raw_basics(world: &SimpleCommunicator, rank: Rank, next: Rank, prev: Rank) {
    let ring = Ring::raw(world, &pair(2, 8)).unwrap();
    assert!(!ring.is_safe());

    for s in 1..=3u8 {
        assert_eq!(ring.send(next, &[rank as u8, s]).unwrap(), u64::from(s));
    }
    world.barrier();

    let messages = ring.poll().unwrap();
    assert_eq!(
        messages,
        vec![
            Message {
                origin: prev,
                sequence: 2,
                data: vec![prev as u8, 2]
            },
            Message {
                origin: prev,
                sequence: 3,
                data: vec![prev as u8, 3]
            },
        ]
    );
    assert_eq!(ring.lost(), 1);

    // Raw acks are no-ops.
    ring.ack(prev, u64::MAX).unwrap();
    world.barrier();
    ring.close().unwrap();
}

examples/compare.rs (line 175)

fn ring_stream(
    comm: &SimpleCommunicator,
    ring: &Ring,
    rank: Rank,
    rep: u32,
    payload: usize,
    n: u64,
) -> (u64, u64, f64, f64, f64) {
    comm.barrier();
    let stalled = ring.wait_ns();
    let t0 = Instant::now();
    if rank == 0 {
        let buf = shuffled(rep, payload);
        let mut sent = 0u64;
        while sent < n {
            ring.send(1, &buf).unwrap();
            sent += 1;
        }
        let inject = t0.elapsed().as_secs_f64();
        let wait = (ring.wait_ns() - stalled) as f64 * 1e-9;
        // Tell the drain side to stop, then collect what it actually saw.
        comm.process_at_rank(1).send_with_tag(&[0u8], TAG_STOP);
        let (reply, _) = comm
            .process_at_rank(1)
            .receive_vec_with_tag::<u64>(TAG_STOP);
        (sent, reply[0], inject, t0.elapsed().as_secs_f64(), wait)
    } else {
        let mut got = 0u64;
        let mut stopping = false;
        loop {
            let batch = ring.poll().unwrap();
            let empty = batch.is_empty();
            let mut furthest = 0u64;
            for m in &batch {
                got += 1;
                furthest = furthest.max(m.sequence);
            }
            if furthest > 0 {
                ring.ack(0, furthest).unwrap();
            }
            if stopping && empty {
                break;
            }
            if !stopping
                && comm
                    .any_process()
                    .immediate_probe_with_tag(TAG_STOP)
                    .is_some()
            {
                comm.any_process().receive_vec_with_tag::<u8>(TAG_STOP);
                stopping = true;
            }
            if empty {
                std::thread::yield_now();
            }
        }
        comm.process_at_rank(0).send_with_tag(&[got][..], TAG_STOP);
        let dt = t0.elapsed().as_secs_f64();
        (0, got, dt, dt, 0.0)
    }
}

/// One message each way, `iters` times, on the ring.
///
/// Returns `(total, max_us)`: the run wall time and the worst single
/// round trip, halved to a one-way latency so it shares the `us_per_msg`
/// unit of the mean.
fn ring_rtt(ring: &Ring, rank: Rank, rep: u32, payload: usize, iters: u64) -> (f64, f64) {
    let buf = shuffled(rep, payload);
    let other: Rank = if rank == 0 { 1 } else { 0 };
    let mut seen = 0u64;
    let await_one = |ring: &Ring, seen: &mut u64| {
        loop {
            let batch = ring.poll().unwrap();
            if let Some(last) = batch.last() {
                *seen = last.sequence;
                ring.ack(other, *seen).unwrap();
                return;
            }
            std::hint::spin_loop();
        }
    };
    let mut max_us = 0.0f64;
    let t0 = Instant::now();
    for _ in 0..iters {
        let start = Instant::now();
        if rank == 0 {
            ring.send(1, &buf).unwrap();
            await_one(ring, &mut seen);
        } else {
            await_one(ring, &mut seen);
            ring.send(0, &buf).unwrap();
        }
        max_us = max_us.max(start.elapsed().as_secs_f64() * 5e5);
    }
    (t0.elapsed().as_secs_f64(), max_us)
}

examples/soak.rs (line 220)

fn main() {
    let mut args = std::env::args().skip(1);
    let mode = args.next().unwrap_or_else(|| "safe".into());
    let messages: u64 = args.next().and_then(|s| s.parse().ok()).unwrap_or(20_000);
    let payload: usize = args.next().and_then(|s| s.parse().ok()).unwrap_or(256);
    let depth: usize = args.next().and_then(|s| s.parse().ok()).unwrap_or(8);
    let noise_kib: usize = args.next().and_then(|s| s.parse().ok()).unwrap_or(0);
    let pace_ns: u64 = args.next().and_then(|s| s.parse().ok()).unwrap_or(0);
    // Echoed straight into the output so repeats of one point stay distinguishable.
    let rep: u32 = args.next().and_then(|s| s.parse().ok()).unwrap_or(0);
    assert!(
        payload >= STAMP,
        "payload must hold the origin and sequence"
    );
    let safe = match mode.as_str() {
        "safe" => true,
        "raw" => false,
        other => panic!("mode must be safe or raw, got {other:?}"),
    };

    let (universe, provided) =
        mpi::initialize_with_threading(Threading::Multiple).expect("MPI must initialize once");
    assert_eq!(provided, Threading::Multiple);
    let world = universe.world();
    let rank = world.rank();
    let size = world.size();
    assert!(size >= 2, "soak needs at least 2 ranks");

    // All-to-all: every ordered pair of distinct ranks gets a lane.
    let mut lanes = Vec::new();
    for source in 0..size {
        for destination in 0..size {
            if source != destination {
                lanes.push((source, destination, depth, payload));
            }
        }
    }
    let ring = Arc::new(if safe {
        Ring::safe(&world, &lanes).unwrap()
    } else {
        Ring::raw(&world, &lanes).unwrap()
    });

    let stop = Arc::new(AtomicBool::new(false));
    let done = Arc::new(AtomicBool::new(false));
    let count = Arc::new(AtomicU64::new(0));
    let peers = (size - 1) as u64;
    let sent = messages * peers;

    world.barrier();
    let started = std::time::Instant::now();

    let outcome = std::thread::scope(|scope| {
        if noise_kib > 0 {
            let stop = Arc::clone(&stop);
            let shared = Shared(&world);
            scope.spawn(move || {
                // Name the wrapper as a whole first. Closures capture the
                // narrowest place they use, and `shared.0` alone would be
                // captured as a bare reference, losing the `Send` impl.
                let shared = shared;
                noise(shared.0, noise_kib, &stop);
            });
        }

        let poller = {
            let (ring, done, count) = (Arc::clone(&ring), Arc::clone(&done), Arc::clone(&count));
            scope.spawn(move || -> Result<Vec<u64>, String> {
                let mut seen = vec![0u64; size as usize];
                let mut total = 0u64;
                loop {
                    // If true, this poll follows the global send barrier.
                    let finished = done.load(Ordering::Acquire);
                    let batch = ring.poll().map_err(|e| format!("poll: {e}"))?;
                    let empty = batch.is_empty();
                    let mut furthest: HashMap<Rank, u64> = HashMap::new();
                    for m in &batch {
                        verify(m)?;
                        let previous = seen[m.origin as usize];
                        if m.sequence <= previous {
                            return Err(format!(
                                "sequence from {} went backwards: {} after {previous}",
                                m.origin, m.sequence
                            ));
                        }
                        if safe && m.sequence != previous + 1 {
                            return Err(format!(
                                "safe lane from {} skipped {} to {}",
                                m.origin,
                                previous + 1,
                                m.sequence
                            ));
                        }
                        seen[m.origin as usize] = m.sequence;
                        let f = furthest.entry(m.origin).or_insert(0);
                        *f = (*f).max(m.sequence);
                        total += 1;
                    }
                    for (origin, sequence) in furthest {
                        ring.ack(origin, sequence)
                            .map_err(|e| format!("ack to {origin}: {e}"))?;
                    }
                    count.store(total, Ordering::Relaxed);
                    if finished && empty {
                        return Ok(seen);
                    }
                    if empty {
                        std::thread::yield_now();
                    }
                }
            })
        };

        // Round-robin over peers so every outgoing lane stays hot at once.
        let mut buf = vec![0u8; payload];
        let pace = std::time::Duration::from_nanos(pace_ns);
        let opened = std::time::Instant::now();
        for sequence in 1..=messages {
            for destination in (0..size).filter(|&r| r != rank) {
                paint(rank, sequence, &mut buf);
                ring.send(destination, &buf)
                    .unwrap_or_else(|e| panic!("rank {rank}: send to {destination}: {e}"));
            }
            // Absolute deadlines, so a slow round is not paid for twice and the
            // offered rate stays the one that was asked for.
            if pace_ns > 0 {
                let due = opened + pace * sequence as u32;
                while std::time::Instant::now() < due {
                    std::hint::spin_loop();
                }
            }
        }
        // Every rank's sends have completed at their targets before this
        // returns, so after the barrier nothing new can appear in a slot.
        world.barrier();
        done.store(true, Ordering::Release);

        let seen = poller.join().expect("poller thread panicked");
        stop.store(true, Ordering::Relaxed);
        seen
    });

    let elapsed = started.elapsed().as_secs_f64();
    let fail = |why: String| -> ! {
        eprintln!("[soak] FAIL rank {rank}: {why}");
        world.abort(1)
    };
    let seen = outcome.unwrap_or_else(|why| fail(why));

    let got = count.load(Ordering::Relaxed);
    let lost = ring.lost();
    let corrupt = ring.corrupt();

    // The last message on a lane is never overwritten, so every lane must end
    // exactly at the sender's high-water mark in both modes.
    for origin in (0..size).filter(|&r| r != rank) {
        if seen[origin as usize] != messages {
            fail(format!(
                "lane from {origin} ended at {} not {messages}",
                seen[origin as usize]
            ));
        }
    }
    if safe && lost != 0 {
        fail(format!("safe ring lost {lost}"));
    }
    if got + lost != sent {
        fail(format!("received {got} + lost {lost} != sent {sent}"));
    }

    // Gather the counters and let rank 0 write every row. Interleaving the
    // ranks' own stdout would scatter the header into the middle of the table:
    // mpirun orders a rank's output against itself and nothing else.
    let mine = [
        got,
        lost,
        corrupt,
        ring.max_lag(),
        ring.waits(),
        ring.wait_ns(),
    ];
    let mut all = vec![0u64; mine.len() * size as usize];
    world.all_gather_into(&mine[..], &mut all[..]);
    if rank == 0 {
        println!(
            "mode\tranks\tdepth\tpayload\tnoise_kib\tpace_ns\trep\trank\tsent\treceived\tlost\tcorrupt\tmax_lag\twaits\twait_s\tseconds"
        );
        for (r, row) in all.chunks_exact(mine.len()).enumerate() {
            println!(
                "{mode}\t{size}\t{depth}\t{payload}\t{noise_kib}\t{pace_ns}\t{rep}\t{r}\t{sent}\t{}\t{}\t{}\t{}\t{}\t{:.6}\t{elapsed:.6}",
                row[0],
                row[1],
                row[2],
                row[3],
                row[4],
                row[5] as f64 * 1e-9,
            );
        }
    }

    world.barrier();
    Arc::try_unwrap(ring)
        .unwrap_or_else(|_| panic!("ring still shared at close"))
        .close()
        .unwrap();
    if rank == 0 {
        eprintln!("[soak] ok: {mode}, {size} ranks, {sent} messages/rank in {elapsed:.2}s");
    }
}

pub fn close(self) -> Result<(), Error>

Close the underlying windows. Collective over the ring communicator.

Prefer this over relying on Drop: the destructor’s shutdown runs the same steps but the collective boundary is implicit.

Examples found in repository

examples/test_layout.rs (line 61)

fn run(world: &SimpleCommunicator, safe: bool) {
    let rank = world.rank();
    let ring = if safe {
        Ring::safe(world, &lanes(2, 8)).unwrap()
    } else {
        Ring::raw(world, &lanes(2, 8)).unwrap()
    };

    if rank < 2 {
        ring.send(2, &[rank as u8, 1]).unwrap();
    }
    world.barrier();

    if rank == 2 {
        let messages = ring.poll().unwrap();
        assert_eq!(messages.len(), 2);
        assert_eq!(messages[0].origin, 0);
        assert_eq!(messages[0].data, vec![0, 1]);
        assert_eq!(messages[1].origin, 1);
        assert_eq!(messages[1].data, vec![1, 1]);
        ring.ack(0, 1).unwrap();
        ring.ack(1, 1).unwrap();
        ring.send(0, &[2, 1]).unwrap();
        ring.send(1, &[2, 1]).unwrap();
    }
    world.barrier();

    if rank < 2 {
        let messages = ring.poll().unwrap();
        assert_eq!(
            messages,
            vec![Message {
                origin: 2,
                sequence: 1,
                data: vec![2, 1]
            }]
        );
        ring.ack(2, 1).unwrap();
    }
    world.barrier();

    ring.close().unwrap();
}

examples/test_ring.rs (line 79)

fn safe_basics(world: &SimpleCommunicator, rank: Rank, next: Rank, prev: Rank) {
    // Depth 4 against 3 messages: this covers ordering and acknowledge
    // validation with the overwrite gate deliberately out of the way.
    // Backpressure is its own test, on a one-way lane.
    let ring = Ring::safe(world, &pair(4, 8)).unwrap();
    assert!(ring.is_safe());
    assert_eq!(ring.depth(next), Some(4));
    assert_eq!(ring.capacity(next), Some(8));

    assert!(matches!(
        ring.send(next, &[0; 9]),
        Err(Error::Payload {
            len: 9,
            capacity: 8
        })
    ));
    assert!(matches!(
        ring.ack(prev, 1),
        Err(Error::Ack {
            sequence: 1,
            received: 0,
            ..
        })
    ));

    for s in 1..=3u8 {
        assert_eq!(ring.send(next, &[rank as u8, s]).unwrap(), u64::from(s));
    }
    world.barrier();

    let messages = ring.poll().unwrap();
    assert_eq!(
        messages,
        (1..=3)
            .map(|s| Message {
                origin: prev,
                sequence: s,
                data: vec![prev as u8, s as u8],
            })
            .collect::<Vec<_>>()
    );
    assert!(ring.poll().unwrap().is_empty());
    assert_eq!(ring.lost(), 0);
    assert_eq!(ring.max_lag(), 3);

    assert!(matches!(
        ring.ack(prev, 4),
        Err(Error::Ack {
            sequence: 4,
            received: 3,
            ..
        })
    ));
    ring.ack(prev, 3).unwrap();
    ring.ack(prev, 3).unwrap();
    ring.ack(prev, 2).unwrap();

    world.barrier();
    ring.close().unwrap();
}

/// A safe sender with no free slot blocks until the receiver acknowledges.
///
/// One-way, so the ack timing belongs to the test. On a symmetric pair each
/// rank's ack races the other's gate check: whichever receiver drains first
/// opens its peer's gate before that peer ever evaluates it, and neither side
/// can be relied on to block.
fn backpressure(world: &SimpleCommunicator, rank: Rank) {
    let ring = Ring::safe(world, &oneway(2, 8)).unwrap();
    if rank == 0 {
        assert_eq!(ring.send(1, &[1]).unwrap(), 1);
        assert_eq!(ring.send(1, &[2]).unwrap(), 2);
    }
    world.barrier();

    if rank == 0 {
        let sent = thread::scope(|scope| {
            let sender = &ring;
            let handle = scope.spawn(move || sender.send(1, &[3]));
            // Depth is 2 and nothing has been acknowledged, so there is no slot
            // for the third message and the send has to wait.
            let deadline = Instant::now() + Duration::from_secs(5);
            while ring.waits() == 0 {
                assert!(Instant::now() < deadline, "safe sender never blocked");
                thread::yield_now();
            }
            // The receiver is parked here, so it cannot have drained early.
            world.barrier();
            handle.join().expect("sender thread panicked").unwrap()
        });
        assert_eq!(sent, 3);
        assert!(ring.waits() > 0);
        assert!(ring.wait_ns() > 0);
    } else {
        world.barrier();
        let messages = ring.poll().unwrap();
        assert_eq!(messages.len(), 2);
        assert_eq!(messages[0].data, vec![1]);
        assert_eq!(messages[1].data, vec![2]);
        assert!(matches!(
            ring.ack(0, 3),
            Err(Error::Ack {
                sequence: 3,
                received: 2,
                ..
            })
        ));
        ring.ack(0, 2).unwrap();
        ring.ack(0, 2).unwrap();
    }

    world.barrier();
    if rank == 1 {
        let messages = ring.poll().unwrap();
        assert_eq!(messages.len(), 1);
        assert_eq!(messages[0].sequence, 3);
        assert_eq!(messages[0].data, vec![3]);
        ring.ack(0, 3).unwrap();
    }
    world.barrier();
    ring.close().unwrap();
}

fn oneway_basics(world: &SimpleCommunicator, rank: Rank) {
    let ring = Ring::safe(world, &oneway(2, 8)).unwrap();
    if rank == 0 {
        assert_eq!(ring.send(1, &[7]).unwrap(), 1);
    }
    world.barrier();
    if rank == 1 {
        let messages = ring.poll().unwrap();
        assert_eq!(messages.len(), 1);
        assert_eq!(messages[0].origin, 0);
        assert_eq!(messages[0].data, vec![7]);
        ring.ack(0, 1).unwrap();
    }
    world.barrier();
    ring.close().unwrap();
}

fn raw_basics(world: &SimpleCommunicator, rank: Rank, next: Rank, prev: Rank) {
    let ring = Ring::raw(world, &pair(2, 8)).unwrap();
    assert!(!ring.is_safe());

    for s in 1..=3u8 {
        assert_eq!(ring.send(next, &[rank as u8, s]).unwrap(), u64::from(s));
    }
    world.barrier();

    let messages = ring.poll().unwrap();
    assert_eq!(
        messages,
        vec![
            Message {
                origin: prev,
                sequence: 2,
                data: vec![prev as u8, 2]
            },
            Message {
                origin: prev,
                sequence: 3,
                data: vec![prev as u8, 3]
            },
        ]
    );
    assert_eq!(ring.lost(), 1);

    // Raw acks are no-ops.
    ring.ack(prev, u64::MAX).unwrap();
    world.barrier();
    ring.close().unwrap();
}

examples/compare.rs (line 448)

fn main() {
    let mut args = std::env::args().skip(1);
    let messages: u64 = args.next().and_then(|s| s.parse().ok()).unwrap_or(20_000);
    let depth: usize = args.next().and_then(|s| s.parse().ok()).unwrap_or(32);
    let noise_kib: usize = args.next().and_then(|s| s.parse().ok()).unwrap_or(0);
    let repeats: u32 = args.next().and_then(|s| s.parse().ok()).unwrap_or(5);

    let (mut universe, provided) =
        mpi::initialize_with_threading(Threading::Multiple).expect("MPI must initialize once");
    assert_eq!(provided, Threading::Multiple);
    let world = universe.world();
    let rank = world.rank();
    let size = world.size();
    assert!(size >= 2, "compare needs at least 2 ranks");
    assert!(noise_kib == 0 || size >= 4, "noise needs at least 4 ranks");
    // `p2p-bsend` needs somewhere to copy to. Only rank 0 sends buffered, so
    // only it attaches the buffer; every rank allocating 512 MiB would need
    // tens of GiB at scale. Sized past the largest amount a stream can leave
    // in flight, so the comparison measures the transport and not a buffer
    // running out.
    if rank == 0 {
        universe.set_buffer_size(BSEND_BUF);
    }

    let stop = Arc::new(AtomicBool::new(false));
    let noisy = noise_kib > 0;

    std::thread::scope(|scope| {
        if noisy && rank >= 2 {
            let stop = Arc::clone(&stop);
            let shared = Shared(&world);
            scope.spawn(move || {
                let shared = shared;
                noise(shared.0, noise_kib, &stop);
            });
        }
        if rank > 1 {
            // Noise ranks take no part in the measurement. They wait here for
            // rank 0 to call time, then let their noise thread finish.
            world
                .process_at_rank(0)
                .receive_vec_with_tag::<u8>(TAG_QUIT);
            stop.store(true, Ordering::Relaxed);
            return;
        }

        if rank == 0 {
            println!("{HEADER}");
        }

        // The measured pair only. A world barrier here would deadlock against
        // ranks sitting in the noise loop.
        let pair: Vec<Rank> = vec![0, 1];
        let group = world.group().include(&pair);
        let duo = world
            .split_by_subgroup(&group)
            .expect("ranks 0 and 1 form the measured pair");

        for rep in 0..repeats {
            for payload in SIZES {
                // Both ring modes share the lane shape, so the only difference
                // between the two rows is the acknowledgement gate.
                let lanes = vec![
                    (0 as Rank, 1 as Rank, depth, payload),
                    (1, 0, depth, payload),
                ];
                // Cap the stream by bytes as well as count: 256 KiB x 20k would be
                // 5 GiB through the wire for one row.
                let n = messages.min((STREAM_BYTES / payload as u64).max(64));
                let iters = if payload >= 32768 { 500 } else { 2000 };

                for (name, safe) in [("ring-safe", true), ("ring-raw", false)] {
                    let ring = if safe {
                        Ring::safe(&duo, &lanes).unwrap()
                    } else {
                        Ring::raw(&duo, &lanes).unwrap()
                    };
                    let (sent, got, inject, total, wait) =
                        ring_stream(&duo, &ring, rank, rep, payload, n);
                    if rank == 0 {
                        Row {
                            transport: name,
                            measure: "stream",
                            payload,
                            sent,
                            delivered: got,
                            inject,
                            total,
                            wait,
                            max_us: 0.0,
                        }
                        .print(size, depth, noise_kib, rep);
                    }
                    duo.barrier();
                    let stalled = ring.wait_ns();
                    let (dt, max_us) = ring_rtt(&ring, rank, rep, payload, iters);
                    if rank == 0 {
                        Row {
                            transport: name,
                            measure: "rtt",
                            payload,
                            sent: iters * 2,
                            delivered: iters * 2,
                            inject: dt,
                            total: dt,
                            wait: (ring.wait_ns() - stalled) as f64 * 1e-9,
                            max_us,
                        }
                        .print(size, depth, noise_kib, rep);
                    }
                    duo.barrier();
                    ring.close().unwrap();
                }

                for (name, buffered) in [("p2p-send", false), ("p2p-bsend", true)] {
                    let (sent, got, inject, total) =
                        p2p_stream(&duo, rank, rep, payload, n, buffered);
                    if rank == 0 {
                        Row {
                            transport: name,
                            measure: "stream",
                            payload,
                            sent,
                            delivered: got,
                            inject,
                            total,
                            wait: 0.0,
                            max_us: 0.0,
                        }
                        .print(size, depth, noise_kib, rep);
                    }
                    duo.barrier();
                }
                let (dt, max_us) = p2p_rtt(&duo, rank, rep, payload, iters);
                if rank == 0 {
                    Row {
                        transport: "p2p-send",
                        measure: "rtt",
                        payload,
                        sent: iters * 2,
                        delivered: iters * 2,
                        inject: dt,
                        total: dt,
                        wait: 0.0,
                        max_us,
                    }
                    .print(size, depth, noise_kib, rep);
                }
                duo.barrier();
            }
        }

        if rank == 0 {
            for r in 2..size {
                world.process_at_rank(r).send_with_tag(&[0u8], TAG_QUIT);
            }
        }
    });

    // Buffered sends must be drained before MPI tears the buffer down.
    world.barrier();
}

examples/soak.rs (line 324)

fn main() {
    let mut args = std::env::args().skip(1);
    let mode = args.next().unwrap_or_else(|| "safe".into());
    let messages: u64 = args.next().and_then(|s| s.parse().ok()).unwrap_or(20_000);
    let payload: usize = args.next().and_then(|s| s.parse().ok()).unwrap_or(256);
    let depth: usize = args.next().and_then(|s| s.parse().ok()).unwrap_or(8);
    let noise_kib: usize = args.next().and_then(|s| s.parse().ok()).unwrap_or(0);
    let pace_ns: u64 = args.next().and_then(|s| s.parse().ok()).unwrap_or(0);
    // Echoed straight into the output so repeats of one point stay distinguishable.
    let rep: u32 = args.next().and_then(|s| s.parse().ok()).unwrap_or(0);
    assert!(
        payload >= STAMP,
        "payload must hold the origin and sequence"
    );
    let safe = match mode.as_str() {
        "safe" => true,
        "raw" => false,
        other => panic!("mode must be safe or raw, got {other:?}"),
    };

    let (universe, provided) =
        mpi::initialize_with_threading(Threading::Multiple).expect("MPI must initialize once");
    assert_eq!(provided, Threading::Multiple);
    let world = universe.world();
    let rank = world.rank();
    let size = world.size();
    assert!(size >= 2, "soak needs at least 2 ranks");

    // All-to-all: every ordered pair of distinct ranks gets a lane.
    let mut lanes = Vec::new();
    for source in 0..size {
        for destination in 0..size {
            if source != destination {
                lanes.push((source, destination, depth, payload));
            }
        }
    }
    let ring = Arc::new(if safe {
        Ring::safe(&world, &lanes).unwrap()
    } else {
        Ring::raw(&world, &lanes).unwrap()
    });

    let stop = Arc::new(AtomicBool::new(false));
    let done = Arc::new(AtomicBool::new(false));
    let count = Arc::new(AtomicU64::new(0));
    let peers = (size - 1) as u64;
    let sent = messages * peers;

    world.barrier();
    let started = std::time::Instant::now();

    let outcome = std::thread::scope(|scope| {
        if noise_kib > 0 {
            let stop = Arc::clone(&stop);
            let shared = Shared(&world);
            scope.spawn(move || {
                // Name the wrapper as a whole first. Closures capture the
                // narrowest place they use, and `shared.0` alone would be
                // captured as a bare reference, losing the `Send` impl.
                let shared = shared;
                noise(shared.0, noise_kib, &stop);
            });
        }

        let poller = {
            let (ring, done, count) = (Arc::clone(&ring), Arc::clone(&done), Arc::clone(&count));
            scope.spawn(move || -> Result<Vec<u64>, String> {
                let mut seen = vec![0u64; size as usize];
                let mut total = 0u64;
                loop {
                    // If true, this poll follows the global send barrier.
                    let finished = done.load(Ordering::Acquire);
                    let batch = ring.poll().map_err(|e| format!("poll: {e}"))?;
                    let empty = batch.is_empty();
                    let mut furthest: HashMap<Rank, u64> = HashMap::new();
                    for m in &batch {
                        verify(m)?;
                        let previous = seen[m.origin as usize];
                        if m.sequence <= previous {
                            return Err(format!(
                                "sequence from {} went backwards: {} after {previous}",
                                m.origin, m.sequence
                            ));
                        }
                        if safe && m.sequence != previous + 1 {
                            return Err(format!(
                                "safe lane from {} skipped {} to {}",
                                m.origin,
                                previous + 1,
                                m.sequence
                            ));
                        }
                        seen[m.origin as usize] = m.sequence;
                        let f = furthest.entry(m.origin).or_insert(0);
                        *f = (*f).max(m.sequence);
                        total += 1;
                    }
                    for (origin, sequence) in furthest {
                        ring.ack(origin, sequence)
                            .map_err(|e| format!("ack to {origin}: {e}"))?;
                    }
                    count.store(total, Ordering::Relaxed);
                    if finished && empty {
                        return Ok(seen);
                    }
                    if empty {
                        std::thread::yield_now();
                    }
                }
            })
        };

        // Round-robin over peers so every outgoing lane stays hot at once.
        let mut buf = vec![0u8; payload];
        let pace = std::time::Duration::from_nanos(pace_ns);
        let opened = std::time::Instant::now();
        for sequence in 1..=messages {
            for destination in (0..size).filter(|&r| r != rank) {
                paint(rank, sequence, &mut buf);
                ring.send(destination, &buf)
                    .unwrap_or_else(|e| panic!("rank {rank}: send to {destination}: {e}"));
            }
            // Absolute deadlines, so a slow round is not paid for twice and the
            // offered rate stays the one that was asked for.
            if pace_ns > 0 {
                let due = opened + pace * sequence as u32;
                while std::time::Instant::now() < due {
                    std::hint::spin_loop();
                }
            }
        }
        // Every rank's sends have completed at their targets before this
        // returns, so after the barrier nothing new can appear in a slot.
        world.barrier();
        done.store(true, Ordering::Release);

        let seen = poller.join().expect("poller thread panicked");
        stop.store(true, Ordering::Relaxed);
        seen
    });

    let elapsed = started.elapsed().as_secs_f64();
    let fail = |why: String| -> ! {
        eprintln!("[soak] FAIL rank {rank}: {why}");
        world.abort(1)
    };
    let seen = outcome.unwrap_or_else(|why| fail(why));

    let got = count.load(Ordering::Relaxed);
    let lost = ring.lost();
    let corrupt = ring.corrupt();

    // The last message on a lane is never overwritten, so every lane must end
    // exactly at the sender's high-water mark in both modes.
    for origin in (0..size).filter(|&r| r != rank) {
        if seen[origin as usize] != messages {
            fail(format!(
                "lane from {origin} ended at {} not {messages}",
                seen[origin as usize]
            ));
        }
    }
    if safe && lost != 0 {
        fail(format!("safe ring lost {lost}"));
    }
    if got + lost != sent {
        fail(format!("received {got} + lost {lost} != sent {sent}"));
    }

    // Gather the counters and let rank 0 write every row. Interleaving the
    // ranks' own stdout would scatter the header into the middle of the table:
    // mpirun orders a rank's output against itself and nothing else.
    let mine = [
        got,
        lost,
        corrupt,
        ring.max_lag(),
        ring.waits(),
        ring.wait_ns(),
    ];
    let mut all = vec![0u64; mine.len() * size as usize];
    world.all_gather_into(&mine[..], &mut all[..]);
    if rank == 0 {
        println!(
            "mode\tranks\tdepth\tpayload\tnoise_kib\tpace_ns\trep\trank\tsent\treceived\tlost\tcorrupt\tmax_lag\twaits\twait_s\tseconds"
        );
        for (r, row) in all.chunks_exact(mine.len()).enumerate() {
            println!(
                "{mode}\t{size}\t{depth}\t{payload}\t{noise_kib}\t{pace_ns}\t{rep}\t{r}\t{sent}\t{}\t{}\t{}\t{}\t{}\t{:.6}\t{elapsed:.6}",
                row[0],
                row[1],
                row[2],
                row[3],
                row[4],
                row[5] as f64 * 1e-9,
            );
        }
    }

    world.barrier();
    Arc::try_unwrap(ring)
        .unwrap_or_else(|_| panic!("ring still shared at close"))
        .close()
        .unwrap();
    if rank == 0 {
        eprintln!("[soak] ok: {mode}, {size} ranks, {sent} messages/rank in {elapsed:.2}s");
    }
}