Struct Universe

Source

pub struct Universe { /* private fields */ }

Expand description

Represents the initialized MPI environment.

The environment is finalized when the Universe is dropped, mirroring rsmpi where Universe owns the lifetime of MPI_Init / MPI_Finalize.

Implementations§

Source §

impl Universe

Source

pub fn world(&self) -> SystemCommunicator

The MPI_COMM_WORLD communicator containing every process in the job.

Examples found in repository ?

examples/abort.rs (line 9)

7fn main() {
8    let universe = mpi::initialize().unwrap();
9    let world = universe.world();
10
11    if world.rank() == 1 {
12        // Would block forever — but the abort from rank 0 unblocks us by
13        // terminating the process.
14        let _ = world.process_at_rank(0).receive::<i32>();
15        eprintln!("rank 1 should never reach here");
16    } else if world.rank() == 0 {
17        eprintln!("rank 0 calling abort");
18        world.abort(7);
19    }
20}

More examples

Hide additional examples

examples/hello.rs (line 8)

6fn main() {
7    let universe = mpi::initialize().unwrap();
8    let world = universe.world();
9    let size = world.size();
10    let rank = world.rank();
11    let processor = mpi::processor_name().unwrap_or_else(|_| "unknown".into());
12
13    println!("Hello from rank {rank} of {size} on {processor}");
14
15    // Make sure the output is interleaved cleanly before exit.
16    world.barrier();
17    if rank == 0 {
18        println!("All {size} processes checked in.");
19    }
20}

examples/ring.rs (line 9)

7fn main() {
8    let universe = mpi::initialize().unwrap();
9    let world = universe.world();
10    let size = world.size();
11    let rank = world.rank();
12
13    let next_rank = (rank + 1) % size;
14    let previous_rank = (rank - 1 + size) % size;
15
16    let msg = [rank, 2 * rank, 4 * rank];
17    mpi::request::scope(|scope| {
18        let _sreq = WaitGuard::from(
19            world
20                .process_at_rank(next_rank)
21                .immediate_send(scope, &msg[..]),
22        );
23
24        let (msg, status) = world.any_process().receive_vec::<i32>();
25
26        println!(
27            "Rank {rank} received {msg:?} from rank {} (tag {}).",
28            status.source_rank(),
29            status.tag()
30        );
31        assert_eq!(status.source_rank(), previous_rank);
32        let x = previous_rank;
33        assert_eq!(vec![x, 2 * x, 4 * x], msg);
34    });
35
36    world.barrier();
37    if rank == 0 {
38        println!("Ring exchange completed successfully.");
39    }
40}

examples/parallel_io.rs (line 9)

7fn main() {
8    let universe = mpi::initialize().unwrap();
9    let world = universe.world();
10    let rank = world.rank();
11    let size = world.size();
12
13    // Path depends on world size so concurrent jobs of different sizes don't
14    // collide; identical across the ranks of a single job.
15    let path = std::env::temp_dir().join(format!("mpi_no_c_io_{size}.bin"));
16    let mut f = File::open(
17        &world,
18        &path,
19        MODE_CREATE | MODE_RDWR | MODE_DELETE_ON_CLOSE,
20    )
21    .expect("File::open failed");
22
23    let block = [rank * 100, rank * 100 + 1, rank * 100 + 2, rank * 100 + 3];
24    f.write_at_all(rank as u64 * 16, &block)
25        .expect("write failed");
26
27    if rank == 0 {
28        for r in 0..size {
29            let mut buf = [0i32; 4];
30            f.read_at(r as u64 * 16, &mut buf).expect("read failed");
31            assert_eq!(
32                buf,
33                [r * 100, r * 100 + 1, r * 100 + 2, r * 100 + 3],
34                "I/O block mismatch"
35            );
36        }
37        println!("IO PASS: parallel write/read verified on {size} ranks.");
38    }
39
40    world.barrier();
41    // File is deleted on close (drop).
42}

examples/thread_multiple.rs (line 12)

10fn main() {
11    let universe = mpi::initialize().unwrap();
12    let world = universe.world();
13    let rank = world.rank();
14    let size = world.size();
15
16    let next = (rank + 1) % size;
17    let prev = (rank - 1 + size) % size;
18
19    let handles: Vec<_> = (0..THREADS)
20        .map(|t| {
21            let w = world.clone(); // SimpleCommunicator is Send + Sync (Arc)
22            thread::spawn(move || {
23                let tag = 1000 + t;
24                let msg = [rank * 100 + t, rank, t];
25                // Concurrent send + receive on this thread's own tag.
26                w.process_at_rank(next).send_with_tag(&msg[..], tag);
27                let (got, status) = w.process_at_rank(prev).receive_vec_with_tag::<i32>(tag);
28                assert_eq!(status.tag(), tag, "tag mismatch");
29                assert_eq!(
30                    got,
31                    vec![prev * 100 + t, prev, t],
32                    "thread {t} payload mismatch"
33                );
34            })
35        })
36        .collect();
37
38    for h in handles {
39        h.join().expect("thread panicked");
40    }
41
42    world.barrier();
43    if rank == 0 {
44        println!(
45            "THREAD PASS: {THREADS} threads/rank did concurrent tagged exchanges on {size} ranks."
46        );
47    }
48}

examples/cluster.rs (line 14)

12fn main() {
13    let universe = mpi::initialize().unwrap();
14    let world = universe.world();
15    let rank = world.rank();
16    let size = world.size();
17
18    println!(
19        "rank {rank}/{size}: os={} arch={}",
20        std::env::consts::OS,
21        std::env::consts::ARCH
22    );
23
24    // Ring send/receive across nodes.
25    let next = (rank + 1) % size;
26    let prev = (rank - 1 + size) % size;
27    let msg = rank * 1000;
28    let (got, status): (i32, _) = mpi::point_to_point::send_receive(
29        &msg,
30        &world.process_at_rank(next),
31        &world.process_at_rank(prev),
32    );
33    assert_eq!(got, prev * 1000, "ring mismatch");
34    assert_eq!(status.source_rank(), prev);
35
36    // All-reduce sum of ranks.
37    let mut sum = 0i32;
38    world.all_reduce_into(&rank, &mut sum, SystemOperation::sum());
39    assert_eq!(sum, (0..size).sum::<i32>(), "allreduce mismatch");
40
41    // Broadcast from rank 0.
42    let mut buf = if rank == 0 {
43        vec![42i32; 3]
44    } else {
45        vec![0; 3]
46    };
47    world.process_at_rank(0).broadcast_into(&mut buf[..]);
48    assert_eq!(buf, vec![42, 42, 42], "broadcast mismatch");
49
50    world.barrier();
51    println!("rank {rank}: cross-node comm OK (sum of ranks = {sum})");
52    if rank == 0 {
53        println!("CLUSTER PASS: {size} ranks communicated across hosts.");
54    }
55}