Skip to main content

Source

mpi::point_to_point

Trait Source 

Source 
pub trait Source {

Show 20 methods    // Required method
    fn source_rank(&self) -> Rank;

    // Provided methods
    fn receive<Msg: Equivalence>(&self) -> (Msg, Status) { ... }
    fn receive_with_tag<Msg: Equivalence>(&self, tag: Tag) -> (Msg, Status) { ... }
    fn receive_vec<Msg: Equivalence>(&self) -> (Vec<Msg>, Status) { ... }
    fn receive_vec_with_tag<Msg: Equivalence>(
        &self,
        tag: Tag,
    ) -> (Vec<Msg>, Status) { ... }
    fn receive_into<Buf: BufferMut + ?Sized>(&self, buf: &mut Buf) -> Status { ... }
    fn receive_into_with_tag<Buf: BufferMut + ?Sized>(
        &self,
        buf: &mut Buf,
        tag: Tag,
    ) -> Status { ... }
    fn probe(&self) -> Status { ... }
    fn probe_with_tag(&self, tag: Tag) -> Status { ... }
    fn immediate_probe(&self) -> Option<Status> { ... }
    fn immediate_probe_with_tag(&self, tag: Tag) -> Option<Status> { ... }
    fn matched_probe(&self) -> (Message, Status) { ... }
    fn matched_probe_with_tag(&self, tag: Tag) -> (Message, Status) { ... }
    fn immediate_matched_probe(&self) -> Option<(Message, Status)> { ... }
    fn immediate_matched_probe_with_tag(
        &self,
        tag: Tag,
    ) -> Option<(Message, Status)> { ... }
    fn immediate_receive<Msg: Equivalence>(&self) -> ReceiveFuture<Msg> { ... }
    fn immediate_receive_with_tag<Msg: Equivalence>(
        &self,
        tag: Tag,
    ) -> ReceiveFuture<Msg> { ... }
    fn immediate_receive_into<'a, Sc, Buf>(
        &self,
        scope: Sc,
        buf: &'a mut Buf,
    ) -> Request<'a, Buf, Sc>
       where Buf: 'a + BufferMut + ?Sized,
             Sc: Scope<'a> { ... }
    fn immediate_receive_into_with_tag<'a, Sc, Buf>(
        &self,
        scope: Sc,
        buf: &'a mut Buf,
        tag: Tag,
    ) -> Request<'a, Buf, Sc>
       where Buf: 'a + BufferMut + ?Sized,
             Sc: Scope<'a> { ... }
    fn receive_init<'a, Buf: BufferMut + ?Sized>(
        &self,
        buf: &'a mut Buf,
    ) -> PersistentRequest<'a> { ... }

}
Expand description

The receive side of point-to-point communication (MPI_ANY_SOURCE or a specific rank).

Required Methods§

Source

fn source_rank(&self) -> Rank

The rank being received from (or MPI_ANY_SOURCE).

Provided Methods§

Source

fn receive<Msg: Equivalence>(&self) -> (Msg, Status)

Receive a single value (MPI_Recv).

Examples found in repository?
examples/abort.rs (line 14)
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/spawn.rs (line 22)
14fn main() {
15    let universe = mpi::initialize().unwrap();
16    let world = universe.world();
17
18    if let Some(parent) = world.parent() {
19        // ---- child side ----
20        let rank = world.rank();
21        // Receive a value from parent rank 0, send it back incremented.
22        let (val, status) = parent.process_at_rank(0).receive::<i32>();
23        assert_eq!(status.source_rank(), 0);
24        parent.process_at_rank(0).send(&(val + 1));
25        println!(
26            "child {rank}/{} (parent group size {}) got {val}",
27            world.size(),
28            parent.remote_size()
29        );
30    } else {
31        // ---- parent side ----
32        let exe = std::env::current_exe().unwrap();
33        let inter = world
34            .process_at_rank(0)
35            .spawn(exe.to_str().unwrap(), &[], N_CHILDREN)
36            .expect("spawn failed");
37
38        assert_eq!(inter.remote_size(), N_CHILDREN, "wrong child count");
39
40        if world.rank() == 0 {
41            for c in 0..inter.remote_size() {
42                inter.process_at_rank(c).send(&(100 + c));
43                let (reply, _) = inter.process_at_rank(c).receive::<i32>();
44                assert_eq!(reply, 100 + c + 1, "child reply mismatch");
45            }
46            println!(
47                "SPAWN PASS: parent (size {}) round-tripped with {} spawned children.",
48                world.size(),
49                inter.remote_size()
50            );
51        }
52        world.barrier();
53    }
54}
examples/topo_inter.rs (line 62)
5fn main() {
6    let universe = mpi::initialize().unwrap();
7    let world = universe.world();
8    let rank = world.rank();
9    let size = world.size();
10
11    // ---- graph topology: a ring, node i adjacent to i-1 and i+1 ----
12    if size >= 2 {
13        let mut index = Vec::new();
14        let mut edges = Vec::new();
15        for i in 0..size {
16            edges.push((i - 1 + size) % size);
17            edges.push((i + 1) % size);
18            index.push(2 * (i + 1));
19        }
20        let g = world
21            .create_graph_communicator(&index, &edges)
22            .expect("graph create returned None");
23        let expected = vec![(rank - 1 + size) % size, (rank + 1) % size];
24        assert_eq!(g.my_neighbors(), expected, "graph neighbours mismatch");
25        assert_eq!(g.neighbor_count(rank), 2);
26        // Neighbourhood all-gather: each neighbour contributes its own rank.
27        let mut ng = vec![-1i32; 2];
28        g.neighbor_all_gather_into(&rank, &mut ng[..]);
29        assert_eq!(ng, expected, "neighbor_all_gather mismatch");
30        g.barrier();
31    }
32
33    // ---- distributed-graph adjacent: directed ring (recv prev, send next) ----
34    if size >= 2 {
35        let prev = (rank - 1 + size) % size;
36        let next = (rank + 1) % size;
37        let dg = world.create_dist_graph_adjacent(&[prev], &[next]);
38        assert_eq!(dg.in_degree(), 1, "dist-graph in-degree");
39        assert_eq!(dg.out_degree(), 1, "dist-graph out-degree");
40        let mut r = vec![-1i32; 1];
41        dg.neighbor_all_gather_into(&rank, &mut r[..]);
42        assert_eq!(r, vec![prev], "dist-graph neighbor mismatch");
43        dg.barrier();
44    }
45
46    // ---- inter-communicator: split world into two halves ----
47    if size >= 2 {
48        let half = size / 2;
49        let in_a = rank < half;
50        let inter = world.split_intercommunicator(in_a);
51        assert_eq!(
52            inter.local_size() + inter.remote_size(),
53            size,
54            "intercomm size mismatch"
55        );
56
57        // Leaders of the two groups exchange their world ranks across the
58        // inter-communicator.
59        if inter.rank() == 0 {
60            if in_a {
61                inter.process_at_rank(0).send(&rank);
62                let (peer, _): (i32, _) = inter.process_at_rank(0).receive();
63                assert_eq!(peer, half, "intercomm exchange mismatch (A)");
64            } else {
65                let (peer, _): (i32, _) = inter.process_at_rank(0).receive();
66                assert_eq!(peer, 0, "intercomm exchange mismatch (B)");
67                inter.process_at_rank(0).send(&rank);
68            }
69        }
70
71        // Merge back into one intra-communicator spanning everyone.
72        let merged = inter.merge();
73        assert_eq!(merged.size(), size, "merge size mismatch");
74        merged.barrier();
75    }
76
77    world.barrier();
78    if rank == 0 {
79        println!("TOPO/INTER PASS: graph + inter-communicator verified on {size} ranks.");
80    }
81}
examples/selftest.rs (line 22)
11fn main() {
12    let universe = mpi::initialize().unwrap();
13    let world = universe.world();
14    let rank = world.rank();
15    let size = world.size();
16
17    // ---- point-to-point: typed scalar ----
18    if size >= 2 {
19        if rank == 0 {
20            world.process_at_rank(1).send(&42u64);
21        } else if rank == 1 {
22            let (v, status) = world.process_at_rank(0).receive::<u64>();
23            assert_eq!(v, 42);
24            assert_eq!(status.source_rank(), 0);
25        }
26    }
27
28    // ---- point-to-point: vector + probe ----
29    if size >= 2 {
30        if rank == 0 {
31            let data: Vec<i32> = vec![10, 20, 30, 40, 50];
32            world.process_at_rank(1).send(&data[..]);
33        } else if rank == 1 {
34            let status = world.process_at_rank(0).probe();
35            let count = status.count(i32::equivalent_datatype());
36            assert_eq!(count, 5, "probe reported wrong count");
37            let (data, _s) = world.process_at_rank(0).receive_vec::<i32>();
38            assert_eq!(data, vec![10, 20, 30, 40, 50]);
39        }
40    }
41
42    // ---- send_receive ring ----
43    {
44        let next = (rank + 1) % size;
45        let prev = (rank - 1 + size) % size;
46        let (got, _status): (i32, _) = mpi::point_to_point::send_receive(
47            &rank,
48            &world.process_at_rank(next),
49            &world.process_at_rank(prev),
50        );
51        assert_eq!(got, prev, "send_receive ring mismatch");
52    }
53
54    // ---- broadcast ----
55    {
56        let root = world.process_at_rank(0);
57        let mut buf = if rank == 0 {
58            vec![2, 4, 8, 16]
59        } else {
60            vec![0; 4]
61        };
62        root.broadcast_into(&mut buf[..]);
63        assert_eq!(buf, vec![2, 4, 8, 16], "broadcast mismatch");
64    }
65
66    // ---- scatter ----
67    {
68        let root = world.process_at_rank(0);
69        let mut mine = 0i32;
70        if rank == 0 {
71            let send: Vec<i32> = (0..size).collect();
72            root.scatter_into_root(&send[..], &mut mine);
73        } else {
74            root.scatter_into(&mut mine);
75        }
76        assert_eq!(mine, rank, "scatter mismatch");
77    }
78
79    // ---- gather ----
80    {
81        let root = world.process_at_rank(0);
82        if rank == 0 {
83            let mut buf = vec![-1i32; size as usize];
84            root.gather_into_root(&rank, &mut buf[..]);
85            let expected: Vec<i32> = (0..size).collect();
86            assert_eq!(buf, expected, "gather mismatch");
87        } else {
88            root.gather_into(&rank);
89        }
90    }
91
92    // ---- all_gather ----
93    {
94        let mut buf = vec![-1i32; size as usize];
95        world.all_gather_into(&rank, &mut buf[..]);
96        let expected: Vec<i32> = (0..size).collect();
97        assert_eq!(buf, expected, "all_gather mismatch");
98    }
99
100    // ---- all_to_all ----
101    {
102        let send: Vec<i32> = (0..size).map(|j| rank * 100 + j).collect();
103        let mut recv = vec![-1i32; size as usize];
104        world.all_to_all_into(&send[..], &mut recv[..]);
105        for i in 0..size {
106            assert_eq!(recv[i as usize], i * 100 + rank, "all_to_all mismatch");
107        }
108    }
109
110    // ---- reduce to root (sum) ----
111    {
112        let root = world.process_at_rank(0);
113        if rank == 0 {
114            let mut sum = 0i32;
115            root.reduce_into_root(&rank, &mut sum, SystemOperation::sum());
116            assert_eq!(sum, (0..size).sum::<i32>(), "reduce sum mismatch");
117        } else {
118            root.reduce_into(&rank, SystemOperation::sum());
119        }
120    }
121
122    // ---- all_reduce (max) ----
123    {
124        let mut m = 0i32;
125        world.all_reduce_into(&rank, &mut m, SystemOperation::max());
126        assert_eq!(m, size - 1, "all_reduce max mismatch");
127    }
128
129    // ---- inclusive scan ----
130    {
131        let mut acc = 0i32;
132        world.scan_into(&rank, &mut acc, SystemOperation::sum());
133        let expected: i32 = (0..=rank).sum();
134        assert_eq!(acc, expected, "scan mismatch");
135    }
136
137    // ---- exclusive scan ----
138    {
139        let mut acc = -1i32;
140        world.exclusive_scan_into(&rank, &mut acc, SystemOperation::sum());
141        if rank > 0 {
142            let expected: i32 = (0..rank).sum();
143            assert_eq!(acc, expected, "exclusive_scan mismatch");
144        }
145    }
146
147    // ---- reduce_scatter_block ----
148    {
149        let send: Vec<i32> = vec![rank + 1; size as usize];
150        let mut recv = [0i32; 1];
151        world.reduce_scatter_block_into(&send[..], &mut recv[..], SystemOperation::sum());
152        let expected: i32 = (0..size).map(|k| k + 1).sum();
153        assert_eq!(recv[0], expected, "reduce_scatter_block mismatch");
154    }
155
156    // ---- user-defined operation (sum) matches built-in ----
157    {
158        let op = UserOperation::commutative(|inv: &[i32], inout: &mut [i32]| {
159            for (i, o) in inv.iter().zip(inout.iter_mut()) {
160                *o += *i;
161            }
162        });
163        let mut u = 0i32;
164        world.all_reduce_into(&rank, &mut u, op);
165        assert_eq!(u, (0..size).sum::<i32>(), "user op mismatch");
166    }
167
168    // ---- communicator split by parity ----
169    {
170        let color = mpi::topology::Color::with_value(rank % 2);
171        let sub = world.split_by_color(color).expect("split produced None");
172        let expected_size = (0..size).filter(|r| r % 2 == rank % 2).count() as i32;
173        assert_eq!(sub.size(), expected_size, "split size mismatch");
174        // Ranks in the sub-communicator sum to a known value.
175        let mut s = 0i32;
176        sub.all_reduce_into(&sub.rank(), &mut s, SystemOperation::sum());
177        let expected_sum: i32 = (0..sub.size()).sum();
178        assert_eq!(s, expected_sum, "split all_reduce mismatch");
179    }
180
181    // ---- communicator duplicate ----
182    {
183        let dup = world.duplicate();
184        assert_eq!(dup.size(), size);
185        assert_eq!(dup.rank(), rank);
186        dup.barrier();
187    }
188
189    // ---- group ----
190    {
191        let g = world.group();
192        assert_eq!(g.size(), size, "group size mismatch");
193        assert_eq!(g.rank(), Some(rank), "group rank mismatch");
194    }
195
196    // ---- group set algebra ----
197    {
198        let g = world.group();
199        let evens: Vec<i32> = (0..size).filter(|r| r % 2 == 0).collect();
200        let odds: Vec<i32> = (0..size).filter(|r| r % 2 == 1).collect();
201        let ge = g.include(&evens);
202        let go = g.include(&odds);
203        assert_eq!(ge.size() + go.size(), size, "group split size mismatch");
204        assert_eq!(ge.union(&go).size(), size, "group union mismatch");
205        assert_eq!(
206            ge.intersection(&go).size(),
207            0,
208            "group intersection mismatch"
209        );
210    }
211
212    // ---- varying-count all-gather (allgatherv) ----
213    {
214        let counts: Vec<i32> = (0..size).map(|i| i + 1).collect();
215        let mut displs = vec![0i32; size as usize];
216        for i in 1..size as usize {
217            displs[i] = displs[i - 1] + counts[i - 1];
218        }
219        let total: i32 = counts.iter().sum();
220        let send = vec![rank; (rank + 1) as usize];
221        let mut recvbuf = vec![-1i32; total as usize];
222        {
223            let mut part = PartitionMut::new(&mut recvbuf[..], counts.clone(), displs.clone());
224            world.all_gather_varcount_into(&send[..], &mut part);
225        }
226        for i in 0..size as usize {
227            let start = displs[i] as usize;
228            for k in 0..counts[i] as usize {
229                assert_eq!(recvbuf[start + k], i as i32, "allgatherv mismatch");
230            }
231        }
232    }
233
234    // ---- varying-count all-to-all (alltoallv) ----
235    {
236        let counts: Vec<i32> = vec![1; size as usize];
237        let displs: Vec<i32> = (0..size).collect();
238        let send: Vec<i32> = (0..size).map(|j| rank * 10 + j).collect();
239        let mut recv = vec![-1i32; size as usize];
240        {
241            let sp = Partition::new(&send[..], counts.clone(), displs.clone());
242            let mut rp = PartitionMut::new(&mut recv[..], counts.clone(), displs.clone());
243            world.all_to_all_varcount_into(&sp, &mut rp);
244        }
245        for i in 0..size {
246            assert_eq!(recv[i as usize], i * 10 + rank, "alltoallv mismatch");
247        }
248    }
249
250    // ---- Cartesian topology (1-D periodic ring) ----
251    {
252        let cart = world
253            .create_cartesian_communicator(&[size], &[true], false)
254            .expect("cartesian create returned None");
255        assert_eq!(cart.my_coordinates(), vec![rank], "cart coords mismatch");
256        let (src, dst) = cart.shift(0, 1);
257        assert_eq!(dst, Some((rank + 1) % size), "cart shift dest mismatch");
258        assert_eq!(
259            src,
260            Some((rank - 1 + size) % size),
261            "cart shift source mismatch"
262        );
263        assert_eq!(cart.rank_from_coordinates(&[0]), Some(0));
264        cart.barrier();
265    }
266
267    // ---- error handlers ----
268    {
269        use mpi::error::{class, error_string, ErrorHandler};
270        world.set_error_handler(ErrorHandler::Return);
271        assert_eq!(
272            world.error_handler(),
273            ErrorHandler::Return,
274            "errhandler mismatch"
275        );
276        world.set_error_handler(ErrorHandler::Fatal);
277        assert_eq!(world.error_handler(), ErrorHandler::Fatal);
278        assert!(!error_string(class::TRUNCATE).is_empty());
279    }
280
281    // ---- generalized request (completed from another thread) ----
282    {
283        use mpi::request::GeneralizedRequest;
284        let (req, completer) = GeneralizedRequest::start();
285        let handle = std::thread::spawn(move || completer.complete());
286        let _status = req.wait();
287        handle.join().unwrap();
288    }
289
290    // ---- pack / unpack ----
291    {
292        let val = vec![rank, rank * 2, rank * 3];
293        let packed = world.pack(&val[..]);
294        let mut out = vec![0i32; 3];
295        let pos = unsafe { world.unpack_into(&packed, &mut out[..], 0) };
296        assert_eq!(out, val, "pack/unpack mismatch");
297        assert_eq!(pos as usize, packed.len(), "unpack position mismatch");
298    }
299
300    // ---- communicator naming ----
301    {
302        assert_eq!(world.get_name(), "MPI_COMM_WORLD", "comm name mismatch");
303        let d = world.duplicate();
304        d.set_name("dup");
305        assert_eq!(d.get_name(), "dup", "set/get name mismatch");
306    }
307
308    // ---- non-blocking all-reduce + barrier ----
309    {
310        let send = rank;
311        let mut recv = 0i32;
312        mpi::request::scope(|s| {
313            world
314                .immediate_all_reduce_into(s, &send, &mut recv, SystemOperation::sum())
315                .wait();
316        });
317        assert_eq!(
318            recv,
319            (0..size).sum::<i32>(),
320            "immediate all_reduce mismatch"
321        );
322        world.immediate_barrier().wait();
323    }
324
325    // ---- non-blocking broadcast (Root) ----
326    {
327        let root = world.process_at_rank(0);
328        let mut buf = if rank == 0 {
329            vec![7, 7, 7]
330        } else {
331            vec![0, 0, 0]
332        };
333        mpi::request::scope(|s| {
334            root.immediate_broadcast_into(s, &mut buf[..]).wait();
335        });
336        assert_eq!(buf, vec![7, 7, 7], "immediate broadcast mismatch");
337    }
338
339    // ---- in-place all-reduce (MPI_IN_PLACE) ----
340    {
341        let mut b = vec![rank, rank + 1];
342        world.all_reduce_into_in_place(&mut b[..], SystemOperation::sum());
343        let s: i32 = (0..size).sum();
344        assert_eq!(b, vec![s, s + size], "in-place all_reduce mismatch");
345    }
346
347    // ---- persistent send/receive, reused across iterations ----
348    {
349        let next = (rank + 1) % size;
350        let prev = (rank - 1 + size) % size;
351        let sendbuf = [rank, rank * 2];
352        let mut recvbuf = vec![-1i32; 2];
353        {
354            let mut sreq = world.process_at_rank(next).send_init(&sendbuf[..]);
355            let mut rreq = world.process_at_rank(prev).receive_init(&mut recvbuf[..]);
356            for _ in 0..3 {
357                rreq.start();
358                sreq.start();
359                sreq.wait();
360                rreq.wait();
361            }
362        } // requests dropped -> buffer borrows released
363        assert_eq!(recvbuf, vec![prev, prev * 2], "persistent request mismatch");
364    }
365
366    world.barrier();
367    if rank == 0 {
368        println!("SELFTEST PASS: all checks passed on {size} ranks.");
369    }
370}
Source

fn receive_with_tag<Msg: Equivalence>(&self, tag: Tag) -> (Msg, Status)

Receive a single value with a specific tag.

Source

fn receive_vec<Msg: Equivalence>(&self) -> (Vec<Msg>, Status)

Receive a variable number of values into a new Vec.

Examples found in repository?
examples/ring.rs (line 24)
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}
More examples
Hide additional examples
examples/bigmsg.rs (line 19)
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    // ---- point-to-point: 4 MB round trip (well above the 64 KiB threshold) ----
14    let n: i32 = 1_000_000;
15    if size >= 2 {
16        if rank == 0 {
17            let data: Vec<i32> = (0..n).collect();
18            world.process_at_rank(1).send(&data[..]);
19            let (back, _) = world.process_at_rank(1).receive_vec::<i32>();
20            assert_eq!(back.len(), n as usize, "echo length");
21            assert_eq!(back[n as usize - 1], n - 1, "echo tail");
22        } else if rank == 1 {
23            let (data, _) = world.process_at_rank(0).receive_vec::<i32>();
24            assert_eq!(data.len(), n as usize, "bigmsg length");
25            assert_eq!(data[0], 0);
26            assert_eq!(data[n as usize - 1], n - 1, "bigmsg content");
27            world.process_at_rank(0).send(&data[..]);
28        }
29    }
30
31    // ---- large all-reduce (400 KB per rank -> rendezvous inside the tree) ----
32    {
33        let local = vec![rank; 100_000];
34        let mut sum = vec![0i32; 100_000];
35        world.all_reduce_into(&local[..], &mut sum[..], SystemOperation::sum());
36        let expected: i32 = (0..size).sum();
37        assert!(
38            sum.iter().all(|&x| x == expected),
39            "big all_reduce mismatch"
40        );
41    }
42
43    world.barrier();
44    if rank == 0 {
45        println!(
46            "BIGMSG PASS: {} MB round-trip + large all-reduce via rendezvous on {size} ranks.",
47            n * 4 / 1_000_000
48        );
49    }
50}
examples/pingpong.rs (line 25)
8fn main() {
9    let universe = mpi::initialize().unwrap();
10    let world = universe.world();
11    let rank = world.rank();
12    let size = world.size();
13
14    if size >= 2 {
15        let latency_iters = 2000;
16        let bw_iters = 100;
17        let bw_bytes = 1 << 20; // 1 MiB
18
19        if rank == 0 {
20            // ---- latency: 8-byte round trip ----
21            let small = [0u8; 8];
22            // warm up
23            for _ in 0..100 {
24                world.process_at_rank(1).send(&small[..]);
25                let _ = world.process_at_rank(1).receive_vec::<u8>();
26            }
27            let t0 = mpi::time();
28            for _ in 0..latency_iters {
29                world.process_at_rank(1).send(&small[..]);
30                let _ = world.process_at_rank(1).receive_vec::<u8>();
31            }
32            let dt = mpi::time() - t0;
33            let lat_us = dt / latency_iters as f64 / 2.0 * 1e6;
34            println!("latency:   {lat_us:.2} µs (8 B, half round-trip, {latency_iters} iters)");
35
36            // ---- bandwidth: 1 MiB round trip ----
37            let big = vec![0u8; bw_bytes];
38            let t0 = mpi::time();
39            for _ in 0..bw_iters {
40                world.process_at_rank(1).send(&big[..]);
41                let _ = world.process_at_rank(1).receive_vec::<u8>();
42            }
43            let dt = mpi::time() - t0;
44            let mb = bw_bytes as f64 * bw_iters as f64 * 2.0 / 1e6;
45            println!("bandwidth: {:.1} MB/s (1 MiB, {bw_iters} iters)", mb / dt);
46        } else if rank == 1 {
47            let small = [0u8; 8];
48            for _ in 0..latency_iters + 100 {
49                let _ = world.process_at_rank(0).receive_vec::<u8>();
50                world.process_at_rank(0).send(&small[..]);
51            }
52            for _ in 0..bw_iters {
53                let (v, _) = world.process_at_rank(0).receive_vec::<u8>();
54                world.process_at_rank(0).send(&v[..]);
55            }
56        }
57    }
58
59    // ---- collective timing: all-reduce over all ranks ----
60    world.barrier();
61    let coll_iters = 1000;
62    let x = rank;
63    let mut acc = 0i32;
64    let t0 = mpi::time();
65    for _ in 0..coll_iters {
66        world.all_reduce_into(&x, &mut acc, SystemOperation::sum());
67    }
68    let dt = mpi::time() - t0;
69    if rank == 0 {
70        println!(
71            "all_reduce: {:.2} µs/op over {size} ranks ({coll_iters} iters), result {acc}",
72            dt / coll_iters as f64 * 1e6
73        );
74    }
75    world.barrier();
76}
examples/selftest.rs (line 37)
11fn main() {
12    let universe = mpi::initialize().unwrap();
13    let world = universe.world();
14    let rank = world.rank();
15    let size = world.size();
16
17    // ---- point-to-point: typed scalar ----
18    if size >= 2 {
19        if rank == 0 {
20            world.process_at_rank(1).send(&42u64);
21        } else if rank == 1 {
22            let (v, status) = world.process_at_rank(0).receive::<u64>();
23            assert_eq!(v, 42);
24            assert_eq!(status.source_rank(), 0);
25        }
26    }
27
28    // ---- point-to-point: vector + probe ----
29    if size >= 2 {
30        if rank == 0 {
31            let data: Vec<i32> = vec![10, 20, 30, 40, 50];
32            world.process_at_rank(1).send(&data[..]);
33        } else if rank == 1 {
34            let status = world.process_at_rank(0).probe();
35            let count = status.count(i32::equivalent_datatype());
36            assert_eq!(count, 5, "probe reported wrong count");
37            let (data, _s) = world.process_at_rank(0).receive_vec::<i32>();
38            assert_eq!(data, vec![10, 20, 30, 40, 50]);
39        }
40    }
41
42    // ---- send_receive ring ----
43    {
44        let next = (rank + 1) % size;
45        let prev = (rank - 1 + size) % size;
46        let (got, _status): (i32, _) = mpi::point_to_point::send_receive(
47            &rank,
48            &world.process_at_rank(next),
49            &world.process_at_rank(prev),
50        );
51        assert_eq!(got, prev, "send_receive ring mismatch");
52    }
53
54    // ---- broadcast ----
55    {
56        let root = world.process_at_rank(0);
57        let mut buf = if rank == 0 {
58            vec![2, 4, 8, 16]
59        } else {
60            vec![0; 4]
61        };
62        root.broadcast_into(&mut buf[..]);
63        assert_eq!(buf, vec![2, 4, 8, 16], "broadcast mismatch");
64    }
65
66    // ---- scatter ----
67    {
68        let root = world.process_at_rank(0);
69        let mut mine = 0i32;
70        if rank == 0 {
71            let send: Vec<i32> = (0..size).collect();
72            root.scatter_into_root(&send[..], &mut mine);
73        } else {
74            root.scatter_into(&mut mine);
75        }
76        assert_eq!(mine, rank, "scatter mismatch");
77    }
78
79    // ---- gather ----
80    {
81        let root = world.process_at_rank(0);
82        if rank == 0 {
83            let mut buf = vec![-1i32; size as usize];
84            root.gather_into_root(&rank, &mut buf[..]);
85            let expected: Vec<i32> = (0..size).collect();
86            assert_eq!(buf, expected, "gather mismatch");
87        } else {
88            root.gather_into(&rank);
89        }
90    }
91
92    // ---- all_gather ----
93    {
94        let mut buf = vec![-1i32; size as usize];
95        world.all_gather_into(&rank, &mut buf[..]);
96        let expected: Vec<i32> = (0..size).collect();
97        assert_eq!(buf, expected, "all_gather mismatch");
98    }
99
100    // ---- all_to_all ----
101    {
102        let send: Vec<i32> = (0..size).map(|j| rank * 100 + j).collect();
103        let mut recv = vec![-1i32; size as usize];
104        world.all_to_all_into(&send[..], &mut recv[..]);
105        for i in 0..size {
106            assert_eq!(recv[i as usize], i * 100 + rank, "all_to_all mismatch");
107        }
108    }
109
110    // ---- reduce to root (sum) ----
111    {
112        let root = world.process_at_rank(0);
113        if rank == 0 {
114            let mut sum = 0i32;
115            root.reduce_into_root(&rank, &mut sum, SystemOperation::sum());
116            assert_eq!(sum, (0..size).sum::<i32>(), "reduce sum mismatch");
117        } else {
118            root.reduce_into(&rank, SystemOperation::sum());
119        }
120    }
121
122    // ---- all_reduce (max) ----
123    {
124        let mut m = 0i32;
125        world.all_reduce_into(&rank, &mut m, SystemOperation::max());
126        assert_eq!(m, size - 1, "all_reduce max mismatch");
127    }
128
129    // ---- inclusive scan ----
130    {
131        let mut acc = 0i32;
132        world.scan_into(&rank, &mut acc, SystemOperation::sum());
133        let expected: i32 = (0..=rank).sum();
134        assert_eq!(acc, expected, "scan mismatch");
135    }
136
137    // ---- exclusive scan ----
138    {
139        let mut acc = -1i32;
140        world.exclusive_scan_into(&rank, &mut acc, SystemOperation::sum());
141        if rank > 0 {
142            let expected: i32 = (0..rank).sum();
143            assert_eq!(acc, expected, "exclusive_scan mismatch");
144        }
145    }
146
147    // ---- reduce_scatter_block ----
148    {
149        let send: Vec<i32> = vec![rank + 1; size as usize];
150        let mut recv = [0i32; 1];
151        world.reduce_scatter_block_into(&send[..], &mut recv[..], SystemOperation::sum());
152        let expected: i32 = (0..size).map(|k| k + 1).sum();
153        assert_eq!(recv[0], expected, "reduce_scatter_block mismatch");
154    }
155
156    // ---- user-defined operation (sum) matches built-in ----
157    {
158        let op = UserOperation::commutative(|inv: &[i32], inout: &mut [i32]| {
159            for (i, o) in inv.iter().zip(inout.iter_mut()) {
160                *o += *i;
161            }
162        });
163        let mut u = 0i32;
164        world.all_reduce_into(&rank, &mut u, op);
165        assert_eq!(u, (0..size).sum::<i32>(), "user op mismatch");
166    }
167
168    // ---- communicator split by parity ----
169    {
170        let color = mpi::topology::Color::with_value(rank % 2);
171        let sub = world.split_by_color(color).expect("split produced None");
172        let expected_size = (0..size).filter(|r| r % 2 == rank % 2).count() as i32;
173        assert_eq!(sub.size(), expected_size, "split size mismatch");
174        // Ranks in the sub-communicator sum to a known value.
175        let mut s = 0i32;
176        sub.all_reduce_into(&sub.rank(), &mut s, SystemOperation::sum());
177        let expected_sum: i32 = (0..sub.size()).sum();
178        assert_eq!(s, expected_sum, "split all_reduce mismatch");
179    }
180
181    // ---- communicator duplicate ----
182    {
183        let dup = world.duplicate();
184        assert_eq!(dup.size(), size);
185        assert_eq!(dup.rank(), rank);
186        dup.barrier();
187    }
188
189    // ---- group ----
190    {
191        let g = world.group();
192        assert_eq!(g.size(), size, "group size mismatch");
193        assert_eq!(g.rank(), Some(rank), "group rank mismatch");
194    }
195
196    // ---- group set algebra ----
197    {
198        let g = world.group();
199        let evens: Vec<i32> = (0..size).filter(|r| r % 2 == 0).collect();
200        let odds: Vec<i32> = (0..size).filter(|r| r % 2 == 1).collect();
201        let ge = g.include(&evens);
202        let go = g.include(&odds);
203        assert_eq!(ge.size() + go.size(), size, "group split size mismatch");
204        assert_eq!(ge.union(&go).size(), size, "group union mismatch");
205        assert_eq!(
206            ge.intersection(&go).size(),
207            0,
208            "group intersection mismatch"
209        );
210    }
211
212    // ---- varying-count all-gather (allgatherv) ----
213    {
214        let counts: Vec<i32> = (0..size).map(|i| i + 1).collect();
215        let mut displs = vec![0i32; size as usize];
216        for i in 1..size as usize {
217            displs[i] = displs[i - 1] + counts[i - 1];
218        }
219        let total: i32 = counts.iter().sum();
220        let send = vec![rank; (rank + 1) as usize];
221        let mut recvbuf = vec![-1i32; total as usize];
222        {
223            let mut part = PartitionMut::new(&mut recvbuf[..], counts.clone(), displs.clone());
224            world.all_gather_varcount_into(&send[..], &mut part);
225        }
226        for i in 0..size as usize {
227            let start = displs[i] as usize;
228            for k in 0..counts[i] as usize {
229                assert_eq!(recvbuf[start + k], i as i32, "allgatherv mismatch");
230            }
231        }
232    }
233
234    // ---- varying-count all-to-all (alltoallv) ----
235    {
236        let counts: Vec<i32> = vec![1; size as usize];
237        let displs: Vec<i32> = (0..size).collect();
238        let send: Vec<i32> = (0..size).map(|j| rank * 10 + j).collect();
239        let mut recv = vec![-1i32; size as usize];
240        {
241            let sp = Partition::new(&send[..], counts.clone(), displs.clone());
242            let mut rp = PartitionMut::new(&mut recv[..], counts.clone(), displs.clone());
243            world.all_to_all_varcount_into(&sp, &mut rp);
244        }
245        for i in 0..size {
246            assert_eq!(recv[i as usize], i * 10 + rank, "alltoallv mismatch");
247        }
248    }
249
250    // ---- Cartesian topology (1-D periodic ring) ----
251    {
252        let cart = world
253            .create_cartesian_communicator(&[size], &[true], false)
254            .expect("cartesian create returned None");
255        assert_eq!(cart.my_coordinates(), vec![rank], "cart coords mismatch");
256        let (src, dst) = cart.shift(0, 1);
257        assert_eq!(dst, Some((rank + 1) % size), "cart shift dest mismatch");
258        assert_eq!(
259            src,
260            Some((rank - 1 + size) % size),
261            "cart shift source mismatch"
262        );
263        assert_eq!(cart.rank_from_coordinates(&[0]), Some(0));
264        cart.barrier();
265    }
266
267    // ---- error handlers ----
268    {
269        use mpi::error::{class, error_string, ErrorHandler};
270        world.set_error_handler(ErrorHandler::Return);
271        assert_eq!(
272            world.error_handler(),
273            ErrorHandler::Return,
274            "errhandler mismatch"
275        );
276        world.set_error_handler(ErrorHandler::Fatal);
277        assert_eq!(world.error_handler(), ErrorHandler::Fatal);
278        assert!(!error_string(class::TRUNCATE).is_empty());
279    }
280
281    // ---- generalized request (completed from another thread) ----
282    {
283        use mpi::request::GeneralizedRequest;
284        let (req, completer) = GeneralizedRequest::start();
285        let handle = std::thread::spawn(move || completer.complete());
286        let _status = req.wait();
287        handle.join().unwrap();
288    }
289
290    // ---- pack / unpack ----
291    {
292        let val = vec![rank, rank * 2, rank * 3];
293        let packed = world.pack(&val[..]);
294        let mut out = vec![0i32; 3];
295        let pos = unsafe { world.unpack_into(&packed, &mut out[..], 0) };
296        assert_eq!(out, val, "pack/unpack mismatch");
297        assert_eq!(pos as usize, packed.len(), "unpack position mismatch");
298    }
299
300    // ---- communicator naming ----
301    {
302        assert_eq!(world.get_name(), "MPI_COMM_WORLD", "comm name mismatch");
303        let d = world.duplicate();
304        d.set_name("dup");
305        assert_eq!(d.get_name(), "dup", "set/get name mismatch");
306    }
307
308    // ---- non-blocking all-reduce + barrier ----
309    {
310        let send = rank;
311        let mut recv = 0i32;
312        mpi::request::scope(|s| {
313            world
314                .immediate_all_reduce_into(s, &send, &mut recv, SystemOperation::sum())
315                .wait();
316        });
317        assert_eq!(
318            recv,
319            (0..size).sum::<i32>(),
320            "immediate all_reduce mismatch"
321        );
322        world.immediate_barrier().wait();
323    }
324
325    // ---- non-blocking broadcast (Root) ----
326    {
327        let root = world.process_at_rank(0);
328        let mut buf = if rank == 0 {
329            vec![7, 7, 7]
330        } else {
331            vec![0, 0, 0]
332        };
333        mpi::request::scope(|s| {
334            root.immediate_broadcast_into(s, &mut buf[..]).wait();
335        });
336        assert_eq!(buf, vec![7, 7, 7], "immediate broadcast mismatch");
337    }
338
339    // ---- in-place all-reduce (MPI_IN_PLACE) ----
340    {
341        let mut b = vec![rank, rank + 1];
342        world.all_reduce_into_in_place(&mut b[..], SystemOperation::sum());
343        let s: i32 = (0..size).sum();
344        assert_eq!(b, vec![s, s + size], "in-place all_reduce mismatch");
345    }
346
347    // ---- persistent send/receive, reused across iterations ----
348    {
349        let next = (rank + 1) % size;
350        let prev = (rank - 1 + size) % size;
351        let sendbuf = [rank, rank * 2];
352        let mut recvbuf = vec![-1i32; 2];
353        {
354            let mut sreq = world.process_at_rank(next).send_init(&sendbuf[..]);
355            let mut rreq = world.process_at_rank(prev).receive_init(&mut recvbuf[..]);
356            for _ in 0..3 {
357                rreq.start();
358                sreq.start();
359                sreq.wait();
360                rreq.wait();
361            }
362        } // requests dropped -> buffer borrows released
363        assert_eq!(recvbuf, vec![prev, prev * 2], "persistent request mismatch");
364    }
365
366    world.barrier();
367    if rank == 0 {
368        println!("SELFTEST PASS: all checks passed on {size} ranks.");
369    }
370}
Source

fn receive_vec_with_tag<Msg: Equivalence>(&self, tag: Tag) -> (Vec<Msg>, Status)

Receive a variable number of values into a new Vec, with a tag.

Examples found in repository?
examples/thread_multiple.rs (line 27)
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}
Source

fn receive_into<Buf: BufferMut + ?Sized>(&self, buf: &mut Buf) -> Status

Receive into an existing buffer (MPI_Recv).

Source

fn receive_into_with_tag<Buf: BufferMut + ?Sized>( &self, buf: &mut Buf, tag: Tag, ) -> Status

Receive into an existing buffer, with a tag.

Source

fn probe(&self) -> Status

Blocking probe (MPI_Probe): wait for a matching message, describe it without receiving.

Examples found in repository?
examples/selftest.rs (line 34)
11fn main() {
12    let universe = mpi::initialize().unwrap();
13    let world = universe.world();
14    let rank = world.rank();
15    let size = world.size();
16
17    // ---- point-to-point: typed scalar ----
18    if size >= 2 {
19        if rank == 0 {
20            world.process_at_rank(1).send(&42u64);
21        } else if rank == 1 {
22            let (v, status) = world.process_at_rank(0).receive::<u64>();
23            assert_eq!(v, 42);
24            assert_eq!(status.source_rank(), 0);
25        }
26    }
27
28    // ---- point-to-point: vector + probe ----
29    if size >= 2 {
30        if rank == 0 {
31            let data: Vec<i32> = vec![10, 20, 30, 40, 50];
32            world.process_at_rank(1).send(&data[..]);
33        } else if rank == 1 {
34            let status = world.process_at_rank(0).probe();
35            let count = status.count(i32::equivalent_datatype());
36            assert_eq!(count, 5, "probe reported wrong count");
37            let (data, _s) = world.process_at_rank(0).receive_vec::<i32>();
38            assert_eq!(data, vec![10, 20, 30, 40, 50]);
39        }
40    }
41
42    // ---- send_receive ring ----
43    {
44        let next = (rank + 1) % size;
45        let prev = (rank - 1 + size) % size;
46        let (got, _status): (i32, _) = mpi::point_to_point::send_receive(
47            &rank,
48            &world.process_at_rank(next),
49            &world.process_at_rank(prev),
50        );
51        assert_eq!(got, prev, "send_receive ring mismatch");
52    }
53
54    // ---- broadcast ----
55    {
56        let root = world.process_at_rank(0);
57        let mut buf = if rank == 0 {
58            vec![2, 4, 8, 16]
59        } else {
60            vec![0; 4]
61        };
62        root.broadcast_into(&mut buf[..]);
63        assert_eq!(buf, vec![2, 4, 8, 16], "broadcast mismatch");
64    }
65
66    // ---- scatter ----
67    {
68        let root = world.process_at_rank(0);
69        let mut mine = 0i32;
70        if rank == 0 {
71            let send: Vec<i32> = (0..size).collect();
72            root.scatter_into_root(&send[..], &mut mine);
73        } else {
74            root.scatter_into(&mut mine);
75        }
76        assert_eq!(mine, rank, "scatter mismatch");
77    }
78
79    // ---- gather ----
80    {
81        let root = world.process_at_rank(0);
82        if rank == 0 {
83            let mut buf = vec![-1i32; size as usize];
84            root.gather_into_root(&rank, &mut buf[..]);
85            let expected: Vec<i32> = (0..size).collect();
86            assert_eq!(buf, expected, "gather mismatch");
87        } else {
88            root.gather_into(&rank);
89        }
90    }
91
92    // ---- all_gather ----
93    {
94        let mut buf = vec![-1i32; size as usize];
95        world.all_gather_into(&rank, &mut buf[..]);
96        let expected: Vec<i32> = (0..size).collect();
97        assert_eq!(buf, expected, "all_gather mismatch");
98    }
99
100    // ---- all_to_all ----
101    {
102        let send: Vec<i32> = (0..size).map(|j| rank * 100 + j).collect();
103        let mut recv = vec![-1i32; size as usize];
104        world.all_to_all_into(&send[..], &mut recv[..]);
105        for i in 0..size {
106            assert_eq!(recv[i as usize], i * 100 + rank, "all_to_all mismatch");
107        }
108    }
109
110    // ---- reduce to root (sum) ----
111    {
112        let root = world.process_at_rank(0);
113        if rank == 0 {
114            let mut sum = 0i32;
115            root.reduce_into_root(&rank, &mut sum, SystemOperation::sum());
116            assert_eq!(sum, (0..size).sum::<i32>(), "reduce sum mismatch");
117        } else {
118            root.reduce_into(&rank, SystemOperation::sum());
119        }
120    }
121
122    // ---- all_reduce (max) ----
123    {
124        let mut m = 0i32;
125        world.all_reduce_into(&rank, &mut m, SystemOperation::max());
126        assert_eq!(m, size - 1, "all_reduce max mismatch");
127    }
128
129    // ---- inclusive scan ----
130    {
131        let mut acc = 0i32;
132        world.scan_into(&rank, &mut acc, SystemOperation::sum());
133        let expected: i32 = (0..=rank).sum();
134        assert_eq!(acc, expected, "scan mismatch");
135    }
136
137    // ---- exclusive scan ----
138    {
139        let mut acc = -1i32;
140        world.exclusive_scan_into(&rank, &mut acc, SystemOperation::sum());
141        if rank > 0 {
142            let expected: i32 = (0..rank).sum();
143            assert_eq!(acc, expected, "exclusive_scan mismatch");
144        }
145    }
146
147    // ---- reduce_scatter_block ----
148    {
149        let send: Vec<i32> = vec![rank + 1; size as usize];
150        let mut recv = [0i32; 1];
151        world.reduce_scatter_block_into(&send[..], &mut recv[..], SystemOperation::sum());
152        let expected: i32 = (0..size).map(|k| k + 1).sum();
153        assert_eq!(recv[0], expected, "reduce_scatter_block mismatch");
154    }
155
156    // ---- user-defined operation (sum) matches built-in ----
157    {
158        let op = UserOperation::commutative(|inv: &[i32], inout: &mut [i32]| {
159            for (i, o) in inv.iter().zip(inout.iter_mut()) {
160                *o += *i;
161            }
162        });
163        let mut u = 0i32;
164        world.all_reduce_into(&rank, &mut u, op);
165        assert_eq!(u, (0..size).sum::<i32>(), "user op mismatch");
166    }
167
168    // ---- communicator split by parity ----
169    {
170        let color = mpi::topology::Color::with_value(rank % 2);
171        let sub = world.split_by_color(color).expect("split produced None");
172        let expected_size = (0..size).filter(|r| r % 2 == rank % 2).count() as i32;
173        assert_eq!(sub.size(), expected_size, "split size mismatch");
174        // Ranks in the sub-communicator sum to a known value.
175        let mut s = 0i32;
176        sub.all_reduce_into(&sub.rank(), &mut s, SystemOperation::sum());
177        let expected_sum: i32 = (0..sub.size()).sum();
178        assert_eq!(s, expected_sum, "split all_reduce mismatch");
179    }
180
181    // ---- communicator duplicate ----
182    {
183        let dup = world.duplicate();
184        assert_eq!(dup.size(), size);
185        assert_eq!(dup.rank(), rank);
186        dup.barrier();
187    }
188
189    // ---- group ----
190    {
191        let g = world.group();
192        assert_eq!(g.size(), size, "group size mismatch");
193        assert_eq!(g.rank(), Some(rank), "group rank mismatch");
194    }
195
196    // ---- group set algebra ----
197    {
198        let g = world.group();
199        let evens: Vec<i32> = (0..size).filter(|r| r % 2 == 0).collect();
200        let odds: Vec<i32> = (0..size).filter(|r| r % 2 == 1).collect();
201        let ge = g.include(&evens);
202        let go = g.include(&odds);
203        assert_eq!(ge.size() + go.size(), size, "group split size mismatch");
204        assert_eq!(ge.union(&go).size(), size, "group union mismatch");
205        assert_eq!(
206            ge.intersection(&go).size(),
207            0,
208            "group intersection mismatch"
209        );
210    }
211
212    // ---- varying-count all-gather (allgatherv) ----
213    {
214        let counts: Vec<i32> = (0..size).map(|i| i + 1).collect();
215        let mut displs = vec![0i32; size as usize];
216        for i in 1..size as usize {
217            displs[i] = displs[i - 1] + counts[i - 1];
218        }
219        let total: i32 = counts.iter().sum();
220        let send = vec![rank; (rank + 1) as usize];
221        let mut recvbuf = vec![-1i32; total as usize];
222        {
223            let mut part = PartitionMut::new(&mut recvbuf[..], counts.clone(), displs.clone());
224            world.all_gather_varcount_into(&send[..], &mut part);
225        }
226        for i in 0..size as usize {
227            let start = displs[i] as usize;
228            for k in 0..counts[i] as usize {
229                assert_eq!(recvbuf[start + k], i as i32, "allgatherv mismatch");
230            }
231        }
232    }
233
234    // ---- varying-count all-to-all (alltoallv) ----
235    {
236        let counts: Vec<i32> = vec![1; size as usize];
237        let displs: Vec<i32> = (0..size).collect();
238        let send: Vec<i32> = (0..size).map(|j| rank * 10 + j).collect();
239        let mut recv = vec![-1i32; size as usize];
240        {
241            let sp = Partition::new(&send[..], counts.clone(), displs.clone());
242            let mut rp = PartitionMut::new(&mut recv[..], counts.clone(), displs.clone());
243            world.all_to_all_varcount_into(&sp, &mut rp);
244        }
245        for i in 0..size {
246            assert_eq!(recv[i as usize], i * 10 + rank, "alltoallv mismatch");
247        }
248    }
249
250    // ---- Cartesian topology (1-D periodic ring) ----
251    {
252        let cart = world
253            .create_cartesian_communicator(&[size], &[true], false)
254            .expect("cartesian create returned None");
255        assert_eq!(cart.my_coordinates(), vec![rank], "cart coords mismatch");
256        let (src, dst) = cart.shift(0, 1);
257        assert_eq!(dst, Some((rank + 1) % size), "cart shift dest mismatch");
258        assert_eq!(
259            src,
260            Some((rank - 1 + size) % size),
261            "cart shift source mismatch"
262        );
263        assert_eq!(cart.rank_from_coordinates(&[0]), Some(0));
264        cart.barrier();
265    }
266
267    // ---- error handlers ----
268    {
269        use mpi::error::{class, error_string, ErrorHandler};
270        world.set_error_handler(ErrorHandler::Return);
271        assert_eq!(
272            world.error_handler(),
273            ErrorHandler::Return,
274            "errhandler mismatch"
275        );
276        world.set_error_handler(ErrorHandler::Fatal);
277        assert_eq!(world.error_handler(), ErrorHandler::Fatal);
278        assert!(!error_string(class::TRUNCATE).is_empty());
279    }
280
281    // ---- generalized request (completed from another thread) ----
282    {
283        use mpi::request::GeneralizedRequest;
284        let (req, completer) = GeneralizedRequest::start();
285        let handle = std::thread::spawn(move || completer.complete());
286        let _status = req.wait();
287        handle.join().unwrap();
288    }
289
290    // ---- pack / unpack ----
291    {
292        let val = vec![rank, rank * 2, rank * 3];
293        let packed = world.pack(&val[..]);
294        let mut out = vec![0i32; 3];
295        let pos = unsafe { world.unpack_into(&packed, &mut out[..], 0) };
296        assert_eq!(out, val, "pack/unpack mismatch");
297        assert_eq!(pos as usize, packed.len(), "unpack position mismatch");
298    }
299
300    // ---- communicator naming ----
301    {
302        assert_eq!(world.get_name(), "MPI_COMM_WORLD", "comm name mismatch");
303        let d = world.duplicate();
304        d.set_name("dup");
305        assert_eq!(d.get_name(), "dup", "set/get name mismatch");
306    }
307
308    // ---- non-blocking all-reduce + barrier ----
309    {
310        let send = rank;
311        let mut recv = 0i32;
312        mpi::request::scope(|s| {
313            world
314                .immediate_all_reduce_into(s, &send, &mut recv, SystemOperation::sum())
315                .wait();
316        });
317        assert_eq!(
318            recv,
319            (0..size).sum::<i32>(),
320            "immediate all_reduce mismatch"
321        );
322        world.immediate_barrier().wait();
323    }
324
325    // ---- non-blocking broadcast (Root) ----
326    {
327        let root = world.process_at_rank(0);
328        let mut buf = if rank == 0 {
329            vec![7, 7, 7]
330        } else {
331            vec![0, 0, 0]
332        };
333        mpi::request::scope(|s| {
334            root.immediate_broadcast_into(s, &mut buf[..]).wait();
335        });
336        assert_eq!(buf, vec![7, 7, 7], "immediate broadcast mismatch");
337    }
338
339    // ---- in-place all-reduce (MPI_IN_PLACE) ----
340    {
341        let mut b = vec![rank, rank + 1];
342        world.all_reduce_into_in_place(&mut b[..], SystemOperation::sum());
343        let s: i32 = (0..size).sum();
344        assert_eq!(b, vec![s, s + size], "in-place all_reduce mismatch");
345    }
346
347    // ---- persistent send/receive, reused across iterations ----
348    {
349        let next = (rank + 1) % size;
350        let prev = (rank - 1 + size) % size;
351        let sendbuf = [rank, rank * 2];
352        let mut recvbuf = vec![-1i32; 2];
353        {
354            let mut sreq = world.process_at_rank(next).send_init(&sendbuf[..]);
355            let mut rreq = world.process_at_rank(prev).receive_init(&mut recvbuf[..]);
356            for _ in 0..3 {
357                rreq.start();
358                sreq.start();
359                sreq.wait();
360                rreq.wait();
361            }
362        } // requests dropped -> buffer borrows released
363        assert_eq!(recvbuf, vec![prev, prev * 2], "persistent request mismatch");
364    }
365
366    world.barrier();
367    if rank == 0 {
368        println!("SELFTEST PASS: all checks passed on {size} ranks.");
369    }
370}
Source

fn probe_with_tag(&self, tag: Tag) -> Status

Blocking probe with a tag.

Source

fn immediate_probe(&self) -> Option<Status>

Non-blocking probe (MPI_Iprobe).

Source

fn immediate_probe_with_tag(&self, tag: Tag) -> Option<Status>

Non-blocking probe with a tag.

Source

fn matched_probe(&self) -> (Message, Status)

Matched probe (MPI_Mprobe): remove a matching message from the queue and return a Message to receive it.

Source

fn matched_probe_with_tag(&self, tag: Tag) -> (Message, Status)

Matched probe with a tag.

Source

fn immediate_matched_probe(&self) -> Option<(Message, Status)>

Non-blocking matched probe (MPI_Improbe).

Source

fn immediate_matched_probe_with_tag( &self, tag: Tag, ) -> Option<(Message, Status)>

Non-blocking matched probe with a tag.

Source

fn immediate_receive<Msg: Equivalence>(&self) -> ReceiveFuture<Msg>

Non-blocking receive of a single value, returning a ReceiveFuture.

Source

fn immediate_receive_with_tag<Msg: Equivalence>( &self, tag: Tag, ) -> ReceiveFuture<Msg>

Non-blocking receive of a single value with a tag.

Source

fn immediate_receive_into<'a, Sc, Buf>( &self, scope: Sc, buf: &'a mut Buf, ) -> Request<'a, Buf, Sc>
where Buf: 'a + BufferMut + ?Sized, Sc: Scope<'a>,

Non-blocking receive into an existing buffer (MPI_Irecv), returning a Request that must be completed via wait/test.

Source

fn immediate_receive_into_with_tag<'a, Sc, Buf>( &self, scope: Sc, buf: &'a mut Buf, tag: Tag, ) -> Request<'a, Buf, Sc>
where Buf: 'a + BufferMut + ?Sized, Sc: Scope<'a>,

Non-blocking receive into a buffer, with a tag.

Source

fn receive_init<'a, Buf: BufferMut + ?Sized>( &self, buf: &'a mut Buf, ) -> PersistentRequest<'a>

Create a persistent receive request bound to buf (MPI_Recv_init). start it (and wait) repeatedly to re-post the receive.

Examples found in repository?
examples/selftest.rs (line 355)
11fn main() {
12    let universe = mpi::initialize().unwrap();
13    let world = universe.world();
14    let rank = world.rank();
15    let size = world.size();
16
17    // ---- point-to-point: typed scalar ----
18    if size >= 2 {
19        if rank == 0 {
20            world.process_at_rank(1).send(&42u64);
21        } else if rank == 1 {
22            let (v, status) = world.process_at_rank(0).receive::<u64>();
23            assert_eq!(v, 42);
24            assert_eq!(status.source_rank(), 0);
25        }
26    }
27
28    // ---- point-to-point: vector + probe ----
29    if size >= 2 {
30        if rank == 0 {
31            let data: Vec<i32> = vec![10, 20, 30, 40, 50];
32            world.process_at_rank(1).send(&data[..]);
33        } else if rank == 1 {
34            let status = world.process_at_rank(0).probe();
35            let count = status.count(i32::equivalent_datatype());
36            assert_eq!(count, 5, "probe reported wrong count");
37            let (data, _s) = world.process_at_rank(0).receive_vec::<i32>();
38            assert_eq!(data, vec![10, 20, 30, 40, 50]);
39        }
40    }
41
42    // ---- send_receive ring ----
43    {
44        let next = (rank + 1) % size;
45        let prev = (rank - 1 + size) % size;
46        let (got, _status): (i32, _) = mpi::point_to_point::send_receive(
47            &rank,
48            &world.process_at_rank(next),
49            &world.process_at_rank(prev),
50        );
51        assert_eq!(got, prev, "send_receive ring mismatch");
52    }
53
54    // ---- broadcast ----
55    {
56        let root = world.process_at_rank(0);
57        let mut buf = if rank == 0 {
58            vec![2, 4, 8, 16]
59        } else {
60            vec![0; 4]
61        };
62        root.broadcast_into(&mut buf[..]);
63        assert_eq!(buf, vec![2, 4, 8, 16], "broadcast mismatch");
64    }
65
66    // ---- scatter ----
67    {
68        let root = world.process_at_rank(0);
69        let mut mine = 0i32;
70        if rank == 0 {
71            let send: Vec<i32> = (0..size).collect();
72            root.scatter_into_root(&send[..], &mut mine);
73        } else {
74            root.scatter_into(&mut mine);
75        }
76        assert_eq!(mine, rank, "scatter mismatch");
77    }
78
79    // ---- gather ----
80    {
81        let root = world.process_at_rank(0);
82        if rank == 0 {
83            let mut buf = vec![-1i32; size as usize];
84            root.gather_into_root(&rank, &mut buf[..]);
85            let expected: Vec<i32> = (0..size).collect();
86            assert_eq!(buf, expected, "gather mismatch");
87        } else {
88            root.gather_into(&rank);
89        }
90    }
91
92    // ---- all_gather ----
93    {
94        let mut buf = vec![-1i32; size as usize];
95        world.all_gather_into(&rank, &mut buf[..]);
96        let expected: Vec<i32> = (0..size).collect();
97        assert_eq!(buf, expected, "all_gather mismatch");
98    }
99
100    // ---- all_to_all ----
101    {
102        let send: Vec<i32> = (0..size).map(|j| rank * 100 + j).collect();
103        let mut recv = vec![-1i32; size as usize];
104        world.all_to_all_into(&send[..], &mut recv[..]);
105        for i in 0..size {
106            assert_eq!(recv[i as usize], i * 100 + rank, "all_to_all mismatch");
107        }
108    }
109
110    // ---- reduce to root (sum) ----
111    {
112        let root = world.process_at_rank(0);
113        if rank == 0 {
114            let mut sum = 0i32;
115            root.reduce_into_root(&rank, &mut sum, SystemOperation::sum());
116            assert_eq!(sum, (0..size).sum::<i32>(), "reduce sum mismatch");
117        } else {
118            root.reduce_into(&rank, SystemOperation::sum());
119        }
120    }
121
122    // ---- all_reduce (max) ----
123    {
124        let mut m = 0i32;
125        world.all_reduce_into(&rank, &mut m, SystemOperation::max());
126        assert_eq!(m, size - 1, "all_reduce max mismatch");
127    }
128
129    // ---- inclusive scan ----
130    {
131        let mut acc = 0i32;
132        world.scan_into(&rank, &mut acc, SystemOperation::sum());
133        let expected: i32 = (0..=rank).sum();
134        assert_eq!(acc, expected, "scan mismatch");
135    }
136
137    // ---- exclusive scan ----
138    {
139        let mut acc = -1i32;
140        world.exclusive_scan_into(&rank, &mut acc, SystemOperation::sum());
141        if rank > 0 {
142            let expected: i32 = (0..rank).sum();
143            assert_eq!(acc, expected, "exclusive_scan mismatch");
144        }
145    }
146
147    // ---- reduce_scatter_block ----
148    {
149        let send: Vec<i32> = vec![rank + 1; size as usize];
150        let mut recv = [0i32; 1];
151        world.reduce_scatter_block_into(&send[..], &mut recv[..], SystemOperation::sum());
152        let expected: i32 = (0..size).map(|k| k + 1).sum();
153        assert_eq!(recv[0], expected, "reduce_scatter_block mismatch");
154    }
155
156    // ---- user-defined operation (sum) matches built-in ----
157    {
158        let op = UserOperation::commutative(|inv: &[i32], inout: &mut [i32]| {
159            for (i, o) in inv.iter().zip(inout.iter_mut()) {
160                *o += *i;
161            }
162        });
163        let mut u = 0i32;
164        world.all_reduce_into(&rank, &mut u, op);
165        assert_eq!(u, (0..size).sum::<i32>(), "user op mismatch");
166    }
167
168    // ---- communicator split by parity ----
169    {
170        let color = mpi::topology::Color::with_value(rank % 2);
171        let sub = world.split_by_color(color).expect("split produced None");
172        let expected_size = (0..size).filter(|r| r % 2 == rank % 2).count() as i32;
173        assert_eq!(sub.size(), expected_size, "split size mismatch");
174        // Ranks in the sub-communicator sum to a known value.
175        let mut s = 0i32;
176        sub.all_reduce_into(&sub.rank(), &mut s, SystemOperation::sum());
177        let expected_sum: i32 = (0..sub.size()).sum();
178        assert_eq!(s, expected_sum, "split all_reduce mismatch");
179    }
180
181    // ---- communicator duplicate ----
182    {
183        let dup = world.duplicate();
184        assert_eq!(dup.size(), size);
185        assert_eq!(dup.rank(), rank);
186        dup.barrier();
187    }
188
189    // ---- group ----
190    {
191        let g = world.group();
192        assert_eq!(g.size(), size, "group size mismatch");
193        assert_eq!(g.rank(), Some(rank), "group rank mismatch");
194    }
195
196    // ---- group set algebra ----
197    {
198        let g = world.group();
199        let evens: Vec<i32> = (0..size).filter(|r| r % 2 == 0).collect();
200        let odds: Vec<i32> = (0..size).filter(|r| r % 2 == 1).collect();
201        let ge = g.include(&evens);
202        let go = g.include(&odds);
203        assert_eq!(ge.size() + go.size(), size, "group split size mismatch");
204        assert_eq!(ge.union(&go).size(), size, "group union mismatch");
205        assert_eq!(
206            ge.intersection(&go).size(),
207            0,
208            "group intersection mismatch"
209        );
210    }
211
212    // ---- varying-count all-gather (allgatherv) ----
213    {
214        let counts: Vec<i32> = (0..size).map(|i| i + 1).collect();
215        let mut displs = vec![0i32; size as usize];
216        for i in 1..size as usize {
217            displs[i] = displs[i - 1] + counts[i - 1];
218        }
219        let total: i32 = counts.iter().sum();
220        let send = vec![rank; (rank + 1) as usize];
221        let mut recvbuf = vec![-1i32; total as usize];
222        {
223            let mut part = PartitionMut::new(&mut recvbuf[..], counts.clone(), displs.clone());
224            world.all_gather_varcount_into(&send[..], &mut part);
225        }
226        for i in 0..size as usize {
227            let start = displs[i] as usize;
228            for k in 0..counts[i] as usize {
229                assert_eq!(recvbuf[start + k], i as i32, "allgatherv mismatch");
230            }
231        }
232    }
233
234    // ---- varying-count all-to-all (alltoallv) ----
235    {
236        let counts: Vec<i32> = vec![1; size as usize];
237        let displs: Vec<i32> = (0..size).collect();
238        let send: Vec<i32> = (0..size).map(|j| rank * 10 + j).collect();
239        let mut recv = vec![-1i32; size as usize];
240        {
241            let sp = Partition::new(&send[..], counts.clone(), displs.clone());
242            let mut rp = PartitionMut::new(&mut recv[..], counts.clone(), displs.clone());
243            world.all_to_all_varcount_into(&sp, &mut rp);
244        }
245        for i in 0..size {
246            assert_eq!(recv[i as usize], i * 10 + rank, "alltoallv mismatch");
247        }
248    }
249
250    // ---- Cartesian topology (1-D periodic ring) ----
251    {
252        let cart = world
253            .create_cartesian_communicator(&[size], &[true], false)
254            .expect("cartesian create returned None");
255        assert_eq!(cart.my_coordinates(), vec![rank], "cart coords mismatch");
256        let (src, dst) = cart.shift(0, 1);
257        assert_eq!(dst, Some((rank + 1) % size), "cart shift dest mismatch");
258        assert_eq!(
259            src,
260            Some((rank - 1 + size) % size),
261            "cart shift source mismatch"
262        );
263        assert_eq!(cart.rank_from_coordinates(&[0]), Some(0));
264        cart.barrier();
265    }
266
267    // ---- error handlers ----
268    {
269        use mpi::error::{class, error_string, ErrorHandler};
270        world.set_error_handler(ErrorHandler::Return);
271        assert_eq!(
272            world.error_handler(),
273            ErrorHandler::Return,
274            "errhandler mismatch"
275        );
276        world.set_error_handler(ErrorHandler::Fatal);
277        assert_eq!(world.error_handler(), ErrorHandler::Fatal);
278        assert!(!error_string(class::TRUNCATE).is_empty());
279    }
280
281    // ---- generalized request (completed from another thread) ----
282    {
283        use mpi::request::GeneralizedRequest;
284        let (req, completer) = GeneralizedRequest::start();
285        let handle = std::thread::spawn(move || completer.complete());
286        let _status = req.wait();
287        handle.join().unwrap();
288    }
289
290    // ---- pack / unpack ----
291    {
292        let val = vec![rank, rank * 2, rank * 3];
293        let packed = world.pack(&val[..]);
294        let mut out = vec![0i32; 3];
295        let pos = unsafe { world.unpack_into(&packed, &mut out[..], 0) };
296        assert_eq!(out, val, "pack/unpack mismatch");
297        assert_eq!(pos as usize, packed.len(), "unpack position mismatch");
298    }
299
300    // ---- communicator naming ----
301    {
302        assert_eq!(world.get_name(), "MPI_COMM_WORLD", "comm name mismatch");
303        let d = world.duplicate();
304        d.set_name("dup");
305        assert_eq!(d.get_name(), "dup", "set/get name mismatch");
306    }
307
308    // ---- non-blocking all-reduce + barrier ----
309    {
310        let send = rank;
311        let mut recv = 0i32;
312        mpi::request::scope(|s| {
313            world
314                .immediate_all_reduce_into(s, &send, &mut recv, SystemOperation::sum())
315                .wait();
316        });
317        assert_eq!(
318            recv,
319            (0..size).sum::<i32>(),
320            "immediate all_reduce mismatch"
321        );
322        world.immediate_barrier().wait();
323    }
324
325    // ---- non-blocking broadcast (Root) ----
326    {
327        let root = world.process_at_rank(0);
328        let mut buf = if rank == 0 {
329            vec![7, 7, 7]
330        } else {
331            vec![0, 0, 0]
332        };
333        mpi::request::scope(|s| {
334            root.immediate_broadcast_into(s, &mut buf[..]).wait();
335        });
336        assert_eq!(buf, vec![7, 7, 7], "immediate broadcast mismatch");
337    }
338
339    // ---- in-place all-reduce (MPI_IN_PLACE) ----
340    {
341        let mut b = vec![rank, rank + 1];
342        world.all_reduce_into_in_place(&mut b[..], SystemOperation::sum());
343        let s: i32 = (0..size).sum();
344        assert_eq!(b, vec![s, s + size], "in-place all_reduce mismatch");
345    }
346
347    // ---- persistent send/receive, reused across iterations ----
348    {
349        let next = (rank + 1) % size;
350        let prev = (rank - 1 + size) % size;
351        let sendbuf = [rank, rank * 2];
352        let mut recvbuf = vec![-1i32; 2];
353        {
354            let mut sreq = world.process_at_rank(next).send_init(&sendbuf[..]);
355            let mut rreq = world.process_at_rank(prev).receive_init(&mut recvbuf[..]);
356            for _ in 0..3 {
357                rreq.start();
358                sreq.start();
359                sreq.wait();
360                rreq.wait();
361            }
362        } // requests dropped -> buffer borrows released
363        assert_eq!(recvbuf, vec![prev, prev * 2], "persistent request mismatch");
364    }
365
366    world.barrier();
367    if rank == 0 {
368        println!("SELFTEST PASS: all checks passed on {size} ranks.");
369    }
370}

Dyn Compatibility§

This trait is not dyn compatible.

In older versions of Rust, dyn compatibility was called "object safety".

Implementors§