Trait CommunicatorCollectives

pub trait CommunicatorCollectives: Communicator {
    // Provided methods
    fn barrier(&self) { ... }
    fn all_gather_into<S, R>(&self, sendbuf: &S, recvbuf: &mut R)
       where S: Buffer + ?Sized,
             R: BufferMut + ?Sized { ... }
    fn all_gather_varcount_into<S, R>(&self, sendbuf: &S, recvbuf: &mut R)
       where S: Buffer + ?Sized,
             R: PartitionedBufferMut + ?Sized { ... }
    fn all_to_all_into<S, R>(&self, sendbuf: &S, recvbuf: &mut R)
       where S: Buffer + ?Sized,
             R: BufferMut + ?Sized { ... }
    fn all_to_all_varcount_into<S, R>(&self, sendbuf: &S, recvbuf: &mut R)
       where S: PartitionedBuffer + ?Sized,
             R: PartitionedBufferMut + ?Sized { ... }
    fn all_reduce_into<S, R, O>(&self, sendbuf: &S, recvbuf: &mut R, op: O)
       where S: Buffer + ?Sized,
             R: BufferMut + ?Sized,
             O: Operation { ... }
    fn reduce_scatter_block_into<S, R, O>(
        &self,
        sendbuf: &S,
        recvbuf: &mut R,
        op: O,
    )
       where S: Buffer + ?Sized,
             R: BufferMut + ?Sized,
             O: Operation { ... }
    fn scan_into<S, R, O>(&self, sendbuf: &S, recvbuf: &mut R, op: O)
       where S: Buffer + ?Sized,
             R: BufferMut + ?Sized,
             O: Operation { ... }
    fn exclusive_scan_into<S, R, O>(&self, sendbuf: &S, recvbuf: &mut R, op: O)
       where S: Buffer + ?Sized,
             R: BufferMut + ?Sized,
             O: Operation { ... }
    fn immediate_barrier(&self) -> Request<'static, ()> { ... }
    fn immediate_all_gather_into<'a, S, R, Sc>(
        &self,
        scope: Sc,
        sendbuf: &'a S,
        recvbuf: &'a mut R,
    ) -> Request<'a, R, Sc>
       where S: 'a + Buffer + ?Sized,
             R: 'a + BufferMut + ?Sized,
             Sc: Scope<'a> { ... }
    fn immediate_all_gather_varcount_into<'a, S, R, Sc>(
        &self,
        scope: Sc,
        sendbuf: &'a S,
        recvbuf: &'a mut R,
    ) -> Request<'a, R, Sc>
       where S: 'a + Buffer + ?Sized,
             R: 'a + PartitionedBufferMut + ?Sized,
             Sc: Scope<'a> { ... }
    fn immediate_all_to_all_into<'a, S, R, Sc>(
        &self,
        scope: Sc,
        sendbuf: &'a S,
        recvbuf: &'a mut R,
    ) -> Request<'a, R, Sc>
       where S: 'a + Buffer + ?Sized,
             R: 'a + BufferMut + ?Sized,
             Sc: Scope<'a> { ... }
    fn immediate_all_to_all_varcount_into<'a, S, R, Sc>(
        &self,
        scope: Sc,
        sendbuf: &'a S,
        recvbuf: &'a mut R,
    ) -> Request<'a, R, Sc>
       where S: 'a + PartitionedBuffer + ?Sized,
             R: 'a + PartitionedBufferMut + ?Sized,
             Sc: Scope<'a> { ... }
    fn immediate_all_reduce_into<'a, S, R, O, Sc>(
        &self,
        scope: Sc,
        sendbuf: &'a S,
        recvbuf: &'a mut R,
        op: O,
    ) -> Request<'a, R, Sc>
       where S: 'a + Buffer + ?Sized,
             R: 'a + BufferMut + ?Sized,
             O: 'a + Operation,
             Sc: Scope<'a> { ... }
    fn immediate_reduce_scatter_block_into<'a, S, R, O, Sc>(
        &self,
        scope: Sc,
        sendbuf: &'a S,
        recvbuf: &'a mut R,
        op: O,
    ) -> Request<'a, R, Sc>
       where S: 'a + Buffer + ?Sized,
             R: 'a + BufferMut + ?Sized,
             O: 'a + Operation,
             Sc: Scope<'a> { ... }
    fn immediate_scan_into<'a, S, R, O, Sc>(
        &self,
        scope: Sc,
        sendbuf: &'a S,
        recvbuf: &'a mut R,
        op: O,
    ) -> Request<'a, R, Sc>
       where S: 'a + Buffer + ?Sized,
             R: 'a + BufferMut + ?Sized,
             O: 'a + Operation,
             Sc: Scope<'a> { ... }
    fn immediate_exclusive_scan_into<'a, S, R, O, Sc>(
        &self,
        scope: Sc,
        sendbuf: &'a S,
        recvbuf: &'a mut R,
        op: O,
    ) -> Request<'a, R, Sc>
       where S: 'a + Buffer + ?Sized,
             R: 'a + BufferMut + ?Sized,
             O: 'a + Operation,
             Sc: Scope<'a> { ... }
}

Collective communication patterns defined on Communicators

Provided Methods

fn barrier(&self)

Barrier synchronization among all processes in a Communicator

Partake in a barrier synchronization across all processes in the Communicator &self.

Calling processes (or threads within the calling processes) will enter the barrier and block execution until all processes in the Communicator &self have entered the barrier.

Examples

See examples/barrier.rs

Standard section(s)

5.3

Examples found in repository

examples/duplicate.rs (line 11)

fn main() {
    let universe = mpi::initialize().unwrap();
    let world = universe.world();
    let moon = world.duplicate();

    world.barrier();
    moon.barrier();

    assert_eq!(CommunicatorRelation::Congruent, world.compare(&moon));
}

examples/time.rs (line 10)

fn main() {
    let universe = mpi::initialize().unwrap();
    let world = universe.world();

    let t_start = mpi::time();
    world.barrier();
    let t_end = mpi::time();

    println!("barrier took: {} s", t_end - t_start);
    println!(
        "the clock has a resoltion of {} seconds",
        mpi::time_resolution()
    );
}

examples/barrier.rs (line 16)

fn main() {
    let universe = mpi::initialize().unwrap();
    let world = universe.world();
    let receiver_rank = 0;

    if world.rank() == receiver_rank {
        let n = (world.size() - 1) as usize;
        let mut buf = vec![0u64; 2 * n];
        for x in buf[0..n].iter_mut() {
            world.any_process().receive_into(x);
        }
        world.barrier();
        for x in buf[n..2 * n].iter_mut() {
            world.any_process().receive_into(x);
        }
        println!("{:?}", buf);
        assert!(buf[0..n].iter().all(|&x| x == 1));
        assert!(buf[n..2 * n].iter().all(|&x| x == 2));
    } else {
        world.process_at_rank(0).send(&1u64);
        world.barrier();
        world.process_at_rank(0).send(&2u64);
    }
}

examples/ready_send.rs (line 14)

fn main() {
    let universe = mpi::initialize().unwrap();
    let world = universe.world();
    let size = world.size();
    let rank = world.rank();

    if rank > 0 {
        let msg = rank as u8;
        world.barrier();
        unsafe { world.process_at_rank(0).ready_send(&msg) };
    } else {
        let mut v = vec![0u8; (size - 1) as usize];
        mpi::request::scope(|scope| {
            let reqs = v
                .iter_mut()
                .zip(1..)
                .map(|(x, i)| {
                    world
                        .process_at_rank(i as Rank)
                        .immediate_receive_into(scope, x)
                })
                .collect::<Vec<_>>();
            world.barrier();
            for req in reqs {
                req.wait();
            }
        });
        println!("Got message: {:?}", v);
        assert!(v.iter().zip(1..).all(|(x, i)| i == *x as usize));
    }
}

examples/contiguous.rs (line 22)

fn main() {
    let universe = mpi::initialize().unwrap();
    let world = universe.world();
    let rank = world.rank();
    let size = world.size();

    let next_rank = (rank + 1) % size;
    let next_process = world.process_at_rank(next_rank);
    let previous_rank = (rank - 1 + size) % size;
    let previous_process = world.process_at_rank(previous_rank);

    let b1 = (1..).map(|x| rank * x).take(3).collect::<Vec<_>>();
    let mut b2 = std::iter::repeat(-1).take(3).collect::<Vec<_>>();
    println!("Rank {} sending message: {:?}.", rank, b1);
    world.barrier();

    let t = UserDatatype::contiguous(3, &Rank::equivalent_datatype());
    let status;
    {
        let v1 = unsafe { View::with_count_and_datatype(&b1[..], 1, &t) };
        let mut v2 = unsafe { MutView::with_count_and_datatype(&mut b2[..], 1, &t) };
        status = p2p::send_receive_into(&v1, &next_process, &mut v2, &previous_process);
    }

    println!(
        "Rank {} received message: {:?}, status: {:?}.",
        rank, b2, status
    );
    world.barrier();

    let b3 = (1..).map(|x| previous_rank * x).take(3).collect::<Vec<_>>();
    assert_eq!(b3, b2);
}

examples/vector.rs (line 22)

fn main() {
    let universe = mpi::initialize().unwrap();
    let world = universe.world();
    let rank = world.rank();
    let size = world.size();

    let next_rank = (rank + 1) % size;
    let next_process = world.process_at_rank(next_rank);
    let previous_rank = (rank - 1 + size) % size;
    let previous_process = world.process_at_rank(previous_rank);

    let b1 = (1..).map(|x| rank * x).take(6).collect::<Vec<_>>();
    let mut b2 = std::iter::repeat(-1).take(6).collect::<Vec<_>>();
    println!("Rank {} sending message: {:?}.", rank, b1);
    world.barrier();

    let t = UserDatatype::vector(2, 2, 3, &Rank::equivalent_datatype());
    let status;
    {
        let v1 = unsafe { View::with_count_and_datatype(&b1[..], 1, &t) };
        let mut v2 = unsafe { MutView::with_count_and_datatype(&mut b2[..], 1, &t) };
        status = p2p::send_receive_into(&v1, &next_process, &mut v2, &previous_process);
    }

    println!(
        "Rank {} received message: {:?}, status: {:?}.",
        rank, b2, status
    );
    world.barrier();

    let b3 = (1..)
        .map(|x| if x % 3 == 0 { -1 } else { previous_rank * x })
        .take(6)
        .collect::<Vec<_>>();
    assert_eq!(b3, b2);
}

Additional examples can be found in:

• examples/send_receive.rs

fn all_gather_into<S, R>(&self, sendbuf: &S, recvbuf: &mut R)

where S: Buffer + ?Sized, R: BufferMut + ?Sized,

Gather contents of buffers on all participating processes.

After the call completes, the contents of the send Buffers on all processes will be concatenated into the receive Buffers on all ranks.

All send Buffers must contain the same count of elements.

Examples

See examples/all_gather.rs

Standard section(s)

5.7

Examples found in repository

examples/all_gather_bool.rs (line 13)

fn main() {
    let universe = mpi::initialize().unwrap();
    let world = universe.world();

    let rank = world.rank();
    let count = world.size() as usize;

    let mut a = vec![false; count];
    world.all_gather_into(&(rank % 2 == 0), &mut a[..]);

    let answer: Vec<_> = (0..count).map(|i| i % 2 == 0).collect();

    assert_eq!(answer, a);
}

examples/all_gather.rs (line 16)

fn main() {
    let universe = mpi::initialize().unwrap();
    let world = universe.world();
    let root_rank = 0;

    let count = world.size() as usize;
    let i = 2_u64.pow(world.rank() as u32 + 1);
    let mut a = vec![0u64; count];

    world.all_gather_into(&i, &mut a[..]);

    if world.rank() == root_rank {
        println!("Root gathered sequence: {:?}.", a);
    }
    assert!(a
        .iter()
        .enumerate()
        .all(|(a, &b)| b == 2u64.pow(a as u32 + 1)));

    let factor = world.rank() as u64 + 1;
    let a = (1_u64..)
        .take(count)
        .map(|x| x * factor)
        .collect::<Vec<_>>();
    let mut t = vec![0u64; count * count];

    world.all_gather_into(&a[..], &mut t[..]);

    if world.rank() == root_rank {
        println!("Root gathered table:");
        for r in t.chunks(count) {
            println!("{:?}", r);
        }
    }
    assert!((0_u64..)
        .zip(t.iter())
        .all(|(a, &b)| b == (a / count as u64 + 1) * (a % count as u64 + 1)));

    let d = UserDatatype::contiguous(count as Count, &u64::equivalent_datatype());
    t = vec![0u64; count * count];

    {
        let sv = unsafe { View::with_count_and_datatype(&a[..], 1, &d) };
        let mut rv = unsafe { MutView::with_count_and_datatype(&mut t[..], count as Count, &d) };

        world.all_gather_into(&sv, &mut rv);
    }

    if world.rank() == root_rank {
        println!("Root gathered table:");
        for r in t.chunks(count) {
            println!("{:?}", r);
        }
    }
    assert!((0_u64..)
        .zip(t.iter())
        .all(|(a, &b)| b == (a / count as u64 + 1) * (a % count as u64 + 1)));
}

fn all_gather_varcount_into<S, R>(&self, sendbuf: &S, recvbuf: &mut R)

where S: Buffer + ?Sized, R: PartitionedBufferMut + ?Sized,

Gather contents of buffers on all participating processes.

After the call completes, the contents of the send Buffers on all processes will be concatenated into the receive Buffers on all ranks.

The send Buffers may contain different counts of elements on different processes. The distribution of elements in the receive Buffers is specified via Partitioned.

Examples

See examples/all_gather_varcount.rs

Standard section(s)

5.7

Examples found in repository

examples/all_gather_varcount.rs (line 29)

fn main() {
    let universe = mpi::initialize().unwrap();
    let world = universe.world();

    let rank = world.rank();
    let size = world.size();

    let msg: Vec<_> = (0..rank).collect();

    let counts: Vec<Count> = (0..size).collect();
    let displs: Vec<Count> = counts
        .iter()
        .scan(0, |acc, &x| {
            let tmp = *acc;
            *acc += x;
            Some(tmp)
        })
        .collect();

    let mut buf = vec![0; (size * (size - 1) / 2) as usize];
    {
        let mut partition = PartitionMut::new(&mut buf[..], counts, &displs[..]);
        world.all_gather_varcount_into(&msg[..], &mut partition);
    }

    assert!(buf
        .iter()
        .zip((0..size).flat_map(|r| (0..r)))
        .all(|(&i, j)| i == j));
    println!("Process {} got message {:?}", rank, buf);
}

fn all_to_all_into<S, R>(&self, sendbuf: &S, recvbuf: &mut R)

where S: Buffer + ?Sized, R: BufferMut + ?Sized,

Distribute the send Buffers from all processes to the receive Buffers on all processes.

Each process sends and receives the same count of elements to and from each process.

Examples

See examples/all_to_all.rs

Standard section(s)

5.8

Examples found in repository

examples/all_to_all.rs (line 14)

fn main() {
    let universe = mpi::initialize().unwrap();
    let world = universe.world();
    let rank = world.rank();
    let size = world.size();

    let u = vec![rank; size as usize];
    let mut v = vec![0; size as usize];

    world.all_to_all_into(&u[..], &mut v[..]);

    println!("u: {:?}", u);
    println!("v: {:?}", v);

    assert!(v.into_iter().zip(0..size).all(|(i, j)| i == j));
}

fn all_to_all_varcount_into<S, R>(&self, sendbuf: &S, recvbuf: &mut R)

where S: PartitionedBuffer + ?Sized, R: PartitionedBufferMut + ?Sized,

Distribute the send Buffers from all processes to the receive Buffers on all processes.

The count of elements to send and receive to and from each process can vary and is specified using Partitioned.

Standard section(s)

5.8

fn all_reduce_into<S, R, O>(&self, sendbuf: &S, recvbuf: &mut R, op: O)

where S: Buffer + ?Sized, R: BufferMut + ?Sized, O: Operation,

Performs a global reduction under the operation op of the input data in sendbuf and stores the result in recvbuf on all processes.

Examples

See examples/reduce.rs

Standard section(s)

5.9.6

Examples found in repository

examples/reduce.rs (lines 19-29)

fn test_user_operations<C: Communicator>(comm: C) {
    let rank = comm.rank();
    let size = comm.size();
    let mut h = 0;
    comm.all_reduce_into(
        &(rank + 1),
        &mut h,
        &UserOperation::commutative(|x, y| {
            let x: &[Rank] = x.downcast().unwrap();
            let y: &mut [Rank] = y.downcast().unwrap();
            for (&x_i, y_i) in x.iter().zip(y) {
                *y_i += x_i;
            }
        }),
    );
    assert_eq!(h, size * (size + 1) / 2);
}

#[cfg(not(feature = "user-operations"))]
fn test_user_operations<C: Communicator>(_: C) {}

#[cfg(not(all(msmpi, target_arch = "x86")))]
unsafe extern "C" fn unsafe_add(
    invec: *mut c_void,
    inoutvec: *mut c_void,
    len: *mut c_int,
    _datatype: *mut MPI_Datatype,
) {
    use std::slice;

    let x: &[Rank] = slice::from_raw_parts(invec as *const Rank, *len as usize);
    let y: &mut [Rank] = slice::from_raw_parts_mut(inoutvec as *mut Rank, *len as usize);
    for (&x_i, y_i) in x.iter().zip(y) {
        *y_i += x_i;
    }
}

#[cfg(all(msmpi, target_arch = "x86"))]
unsafe extern "stdcall" fn unsafe_add(
    invec: *mut c_void,
    inoutvec: *mut c_void,
    len: *mut c_int,
    _datatype: *mut MPI_Datatype,
) {
    use std::slice;

    let x: &[Rank] = slice::from_raw_parts(invec as *const Rank, *len as usize);
    let y: &mut [Rank] = slice::from_raw_parts_mut(inoutvec as *mut Rank, *len as usize);
    for (&x_i, y_i) in x.iter().zip(y) {
        *y_i += x_i;
    }
}

fn main() {
    let universe = mpi::initialize().unwrap();
    let world = universe.world();
    let rank = world.rank();
    let size = world.size();
    let root_rank = 0;

    if rank == root_rank {
        let mut sum: Rank = 0;
        world
            .process_at_rank(root_rank)
            .reduce_into_root(&rank, &mut sum, SystemOperation::sum());
        assert_eq!(sum, size * (size - 1) / 2);
    } else {
        world
            .process_at_rank(root_rank)
            .reduce_into(&rank, SystemOperation::sum());
    }

    let mut max: Rank = -1;

    world.all_reduce_into(&rank, &mut max, SystemOperation::max());
    assert_eq!(max, size - 1);

    let a: u16 = 0b0000_1111_1111_0000;
    let b: u16 = 0b0011_1100_0011_1100;

    let mut c = b;
    collective::reduce_local_into(&a, &mut c, SystemOperation::bitwise_and());
    assert_eq!(c, 0b0000_1100_0011_0000);

    let mut d = b;
    collective::reduce_local_into(&a, &mut d, SystemOperation::bitwise_or());
    assert_eq!(d, 0b0011_1111_1111_1100);

    let mut e = b;
    collective::reduce_local_into(&a, &mut e, SystemOperation::bitwise_xor());
    assert_eq!(e, 0b0011_0011_1100_1100);

    let f = (0..size).collect::<Vec<_>>();
    let mut g: Rank = 0;

    world.reduce_scatter_block_into(&f[..], &mut g, SystemOperation::product());
    assert_eq!(g, rank.wrapping_pow(size as u32));

    test_user_operations(universe.world());

    let mut i = 0;
    let op = unsafe { UnsafeUserOperation::commutative(unsafe_add) };
    world.all_reduce_into(&(rank + 1), &mut i, &op);
    assert_eq!(i, size * (size + 1) / 2);
}

fn reduce_scatter_block_into<S, R, O>( &self, sendbuf: &S, recvbuf: &mut R, op: O, )

where S: Buffer + ?Sized, R: BufferMut + ?Sized, O: Operation,

Performs an element-wise global reduction under the operation op of the input data in sendbuf and scatters the result into equal sized blocks in the receive buffers on all processes.

Examples

See examples/reduce.rs

Standard section(s)

5.10.1

Examples found in repository

examples/reduce.rs (line 110)

fn main() {
    let universe = mpi::initialize().unwrap();
    let world = universe.world();
    let rank = world.rank();
    let size = world.size();
    let root_rank = 0;

    if rank == root_rank {
        let mut sum: Rank = 0;
        world
            .process_at_rank(root_rank)
            .reduce_into_root(&rank, &mut sum, SystemOperation::sum());
        assert_eq!(sum, size * (size - 1) / 2);
    } else {
        world
            .process_at_rank(root_rank)
            .reduce_into(&rank, SystemOperation::sum());
    }

    let mut max: Rank = -1;

    world.all_reduce_into(&rank, &mut max, SystemOperation::max());
    assert_eq!(max, size - 1);

    let a: u16 = 0b0000_1111_1111_0000;
    let b: u16 = 0b0011_1100_0011_1100;

    let mut c = b;
    collective::reduce_local_into(&a, &mut c, SystemOperation::bitwise_and());
    assert_eq!(c, 0b0000_1100_0011_0000);

    let mut d = b;
    collective::reduce_local_into(&a, &mut d, SystemOperation::bitwise_or());
    assert_eq!(d, 0b0011_1111_1111_1100);

    let mut e = b;
    collective::reduce_local_into(&a, &mut e, SystemOperation::bitwise_xor());
    assert_eq!(e, 0b0011_0011_1100_1100);

    let f = (0..size).collect::<Vec<_>>();
    let mut g: Rank = 0;

    world.reduce_scatter_block_into(&f[..], &mut g, SystemOperation::product());
    assert_eq!(g, rank.wrapping_pow(size as u32));

    test_user_operations(universe.world());

    let mut i = 0;
    let op = unsafe { UnsafeUserOperation::commutative(unsafe_add) };
    world.all_reduce_into(&(rank + 1), &mut i, &op);
    assert_eq!(i, size * (size + 1) / 2);
}

fn scan_into<S, R, O>(&self, sendbuf: &S, recvbuf: &mut R, op: O)

where S: Buffer + ?Sized, R: BufferMut + ?Sized, O: Operation,

Performs a global inclusive prefix reduction of the data in sendbuf into recvbuf under operation op.

Examples

See examples/scan.rs

Standard section(s)

5.11.1

Examples found in repository

examples/dyn_communicator.rs (line 32)

fn prefix_sum(comm: &dyn Communicator, output_prefix: &str) {
    let rank = comm.rank() as usize;
    let mut target = 0;

    comm.scan_into(&rank, &mut target, SystemOperation::sum());

    println!("{} rank {}: {:?}", output_prefix, rank, target);
}

examples/scan.rs (line 17)

fn main() {
    let universe = mpi::initialize().unwrap();
    let world = universe.world();
    let rank = world.rank();

    let mut x = 0;
    world.scan_into(&rank, &mut x, &SystemOperation::sum());
    assert_eq!(x, (rank * (rank + 1)) / 2);

    let y = rank + 1;
    let mut z = 0;
    world.exclusive_scan_into(&y, &mut z, &SystemOperation::product());
    if rank > 0 {
        assert_eq!(z, fac(y - 1));
    }
}

fn exclusive_scan_into<S, R, O>(&self, sendbuf: &S, recvbuf: &mut R, op: O)

where S: Buffer + ?Sized, R: BufferMut + ?Sized, O: Operation,

Performs a global exclusive prefix reduction of the data in sendbuf into recvbuf under operation op.

Examples

See examples/scan.rs

Standard section(s)

5.11.2

Examples found in repository

examples/scan.rs (line 22)

fn main() {
    let universe = mpi::initialize().unwrap();
    let world = universe.world();
    let rank = world.rank();

    let mut x = 0;
    world.scan_into(&rank, &mut x, &SystemOperation::sum());
    assert_eq!(x, (rank * (rank + 1)) / 2);

    let y = rank + 1;
    let mut z = 0;
    world.exclusive_scan_into(&y, &mut z, &SystemOperation::product());
    if rank > 0 {
        assert_eq!(z, fac(y - 1));
    }
}

fn immediate_barrier(&self) -> Request<'static, ()>

Non-blocking barrier synchronization among all processes in a Communicator

Calling processes (or threads within the calling processes) enter the barrier. Completion methods on the associated request object will block until all processes have entered.

Examples

See examples/immediate_barrier.rs

Standard section(s)

5.12.1

Examples found in repository

examples/immediate_barrier.rs (line 26)

fn main() {
    use mpi::traits::*;
    let universe = mpi::initialize().unwrap();
    let world = universe.world();
    let size = world.size();
    let receiver_rank = 0;

    if world.rank() == receiver_rank {
        // receiver process
        let n = (size - 1) as usize;
        let mut buf = vec![0u64; 3 * n];
        // receive first 2 * n messages
        for x in buf[0..2 * n].iter_mut() {
            world.any_process().receive_into(x);
        }
        // signal the waiting senders that 2 * n messages have been received
        let breq = world.immediate_barrier();
        // receive remaining n messages
        for x in buf[2 * n..3 * n].iter_mut() {
            world.any_process().receive_into(x);
        }
        println!("{:?}", buf);
        // messages "1" and "2" may be interleaved, but all have to be contained within the first
        // 2 * n slots of the buffer
        assert_eq!(buf[0..2 * n].iter().filter(|&&x| x == 1).count(), n);
        assert_eq!(buf[0..2 * n].iter().filter(|&&x| x == 2).count(), n);
        // the last n slots in the buffer may only contain message "3"
        assert!(buf[2 * n..3 * n].iter().all(|&x| x == 3));
        // clean up the barrier request
        breq.wait();
    } else {
        // sender processes
        // send message "1"
        world.process_at_rank(0).send(&1u64);
        // join barrier, but do not block
        let breq = world.immediate_barrier();
        // send message "2"
        world.process_at_rank(0).send(&2u64);
        // wait for receiver process to receive the first 2 * n messages
        breq.wait();
        // send message "3"
        world.process_at_rank(0).send(&3u64);
    }
}

examples/immediate_multiple_barrier.rs (line 18)

fn main() {
    use mpi::traits::*;

    const COUNT: usize = 128;

    let universe = mpi::initialize().unwrap();
    let world = universe.world();

    // Try wait_any()
    mpi::request::multiple_scope(COUNT, |_, coll| {
        for _ in 0..COUNT {
            let req = world.immediate_barrier();
            coll.add(req);
        }
        let mut finished = 0;
        while coll.incomplete() > 0 {
            coll.wait_any().unwrap();
            finished += 1;
            assert_eq!(coll.incomplete(), COUNT - finished);
        }
    });

    // Try wait_some()
    mpi::request::multiple_scope(COUNT, |_, coll| {
        for _ in 0..COUNT {
            let req = world.immediate_barrier();
            coll.add(req);
        }
        let mut result = vec![];
        let mut finished = 0;
        while coll.incomplete() > 0 {
            coll.wait_some(&mut result);
            finished += result.len();
            assert_eq!(coll.incomplete(), COUNT - finished);
        }
    });

    // Try wait_all()
    mpi::request::multiple_scope(COUNT, |_, coll| {
        for _ in 0..COUNT {
            let req = world.immediate_barrier();
            coll.add(req);
        }
        let mut result = vec![];
        coll.wait_all(&mut result);
        assert_eq!(result.len(), COUNT);
    });

    // Try test_any()
    mpi::request::multiple_scope(COUNT, |_, coll| {
        for _ in 0..COUNT {
            let req = world.immediate_barrier();
            coll.add(req);
        }
        let mut finished = 0;
        while coll.incomplete() > 0 {
            if coll.test_any().is_some() {
                finished += 1;
            }
            assert_eq!(coll.incomplete(), COUNT - finished);
        }
        assert_eq!(finished, COUNT);
    });

    // Try test_some()
    mpi::request::multiple_scope(COUNT, |_, coll| {
        for _ in 0..COUNT {
            let req = world.immediate_barrier();
            coll.add(req);
        }
        let mut result = vec![];
        let mut finished = 0;
        while coll.incomplete() > 0 {
            coll.test_some(&mut result);
            finished += result.len();
            assert_eq!(coll.incomplete(), COUNT - finished);
        }
        assert_eq!(finished, COUNT);
    });

    // Try test_all()
    mpi::request::multiple_scope(COUNT, |_, coll| {
        for _ in 0..COUNT {
            let req = world.immediate_barrier();
            coll.add(req);
        }
        let mut result = vec![];
        while coll.incomplete() > 0 {
            if coll.test_all(&mut result) {
                assert_eq!(coll.incomplete(), 0);
            }
        }
        assert_eq!(result.len(), COUNT);
    });
}

fn immediate_all_gather_into<'a, S, R, Sc>( &self, scope: Sc, sendbuf: &'a S, recvbuf: &'a mut R, ) -> Request<'a, R, Sc>

where S: 'a + Buffer + ?Sized, R: 'a + BufferMut + ?Sized, Sc: Scope<'a>,

Initiate non-blocking gather of the contents of all sendbufs into all rcevbufs on all processes in the communicator.

Examples

See examples/immediate_all_gather.rs

Standard section(s)

5.12.5

Examples found in repository

examples/immediate_all_gather.rs (line 18)

fn main() {
    let universe = mpi::initialize().unwrap();
    let world = universe.world();
    let root_rank = 0;

    let count = world.size() as usize;
    let i = 2_u64.pow(world.rank() as u32 + 1);
    let mut a = vec![0u64; count];

    mpi::request::scope(|scope| {
        world
            .immediate_all_gather_into(scope, &i, &mut a[..])
            .wait();
    });

    if world.rank() == root_rank {
        println!("Root gathered sequence: {:?}.", a);
    }
    assert!(a
        .iter()
        .enumerate()
        .all(|(a, &b)| b == 2u64.pow(a as u32 + 1)));

    let factor = world.rank() as u64 + 1;
    let a = (1_u64..)
        .take(count)
        .map(|x| x * factor)
        .collect::<Vec<_>>();
    let mut t = vec![0u64; count * count];

    mpi::request::scope(|scope| {
        world
            .immediate_all_gather_into(scope, &a[..], &mut t[..])
            .wait();
    });

    if world.rank() == root_rank {
        println!("Root gathered table:");
        for r in t.chunks(count) {
            println!("{:?}", r);
        }
    }
    assert!((0_u64..)
        .zip(t.iter())
        .all(|(a, &b)| b == (a / count as u64 + 1) * (a % count as u64 + 1)));

    let d = UserDatatype::contiguous(count as Count, &u64::equivalent_datatype());
    t = vec![0u64; count * count];

    {
        let sv = unsafe { View::with_count_and_datatype(&a[..], 1, &d) };
        let mut rv = unsafe { MutView::with_count_and_datatype(&mut t[..], count as Count, &d) };
        mpi::request::scope(|scope| {
            world.immediate_all_gather_into(scope, &sv, &mut rv).wait();
        });
    }

    if world.rank() == root_rank {
        println!("Root gathered table:");
        for r in t.chunks(count) {
            println!("{:?}", r);
        }
    }
    assert!((0_u64..)
        .zip(t.iter())
        .all(|(a, &b)| b == (a / count as u64 + 1) * (a % count as u64 + 1)));
}

fn immediate_all_gather_varcount_into<'a, S, R, Sc>( &self, scope: Sc, sendbuf: &'a S, recvbuf: &'a mut R, ) -> Request<'a, R, Sc>

where S: 'a + Buffer + ?Sized, R: 'a + PartitionedBufferMut + ?Sized, Sc: Scope<'a>,

Initiate non-blocking gather of the contents of all sendbufs into all rcevbufs on all processes in the communicator.

Examples

See examples/immediate_all_gather_varcount.rs

Standard section(s)

5.12.5

Examples found in repository

examples/immediate_all_gather_varcount.rs (line 30)

fn main() {
    let universe = mpi::initialize().unwrap();
    let world = universe.world();

    let rank = world.rank();
    let size = world.size();

    let msg: Vec<_> = (0..rank).collect();

    let counts: Vec<Count> = (0..size).collect();
    let displs: Vec<Count> = counts
        .iter()
        .scan(0, |acc, &x| {
            let tmp = *acc;
            *acc += x;
            Some(tmp)
        })
        .collect();

    let mut buf = vec![0; (size * (size - 1) / 2) as usize];
    {
        let mut partition = PartitionMut::new(&mut buf[..], counts, &displs[..]);
        mpi::request::scope(|scope| {
            let req = world.immediate_all_gather_varcount_into(scope, &msg[..], &mut partition);
            req.wait();
        });
    }

    assert!(buf
        .iter()
        .zip((0..size).flat_map(|r| (0..r)))
        .all(|(&i, j)| i == j));
    println!("Process {} got message {:?}", rank, buf);
}

fn immediate_all_to_all_into<'a, S, R, Sc>( &self, scope: Sc, sendbuf: &'a S, recvbuf: &'a mut R, ) -> Request<'a, R, Sc>

where S: 'a + Buffer + ?Sized, R: 'a + BufferMut + ?Sized, Sc: Scope<'a>,

Initiate non-blocking all-to-all communication.

Examples

See examples/immediate_all_to_all.rs

Standard section(s)

5.12.6

Examples found in repository

examples/immediate_all_to_all.rs (line 16)

fn main() {
    let universe = mpi::initialize().unwrap();
    let world = universe.world();
    let rank = world.rank();
    let size = world.size();

    let u = vec![rank; size as usize];
    let mut v = vec![0; size as usize];

    mpi::request::scope(|scope| {
        world
            .immediate_all_to_all_into(scope, &u[..], &mut v[..])
            .wait();
    });

    println!("u: {:?}", u);
    println!("v: {:?}", v);

    assert!(v.into_iter().zip(0..size).all(|(i, j)| i == j));
}

fn immediate_all_to_all_varcount_into<'a, S, R, Sc>( &self, scope: Sc, sendbuf: &'a S, recvbuf: &'a mut R, ) -> Request<'a, R, Sc>

where S: 'a + PartitionedBuffer + ?Sized, R: 'a + PartitionedBufferMut + ?Sized, Sc: Scope<'a>,

Initiate non-blocking all-to-all communication.

Standard section(s)

5.12.6

fn immediate_all_reduce_into<'a, S, R, O, Sc>( &self, scope: Sc, sendbuf: &'a S, recvbuf: &'a mut R, op: O, ) -> Request<'a, R, Sc>

where S: 'a + Buffer + ?Sized, R: 'a + BufferMut + ?Sized, O: 'a + Operation, Sc: Scope<'a>,

Initiates a non-blocking global reduction under the operation op of the input data in sendbuf and stores the result in recvbuf on all processes.

Examples

See examples/immediate_reduce.rs

Standard section(s)

5.12.8

Examples found in repository

examples/immediate_reduce.rs (line 26)

fn test_user_operations<C: Communicator>(comm: C) {
    let op = UserOperation::commutative(|x, y| {
        let x: &[Rank] = x.downcast().unwrap();
        let y: &mut [Rank] = y.downcast().unwrap();
        for (&x_i, y_i) in x.iter().zip(y) {
            *y_i += x_i;
        }
    });
    let rank = comm.rank();
    let size = comm.size();
    let mut c = 0;
    mpi::request::scope(|scope| {
        comm.immediate_all_reduce_into(scope, &rank, &mut c, &op)
            .wait();
    });
    assert_eq!(c, size * (size - 1) / 2);
}

#[cfg(not(feature = "user-operations"))]
fn test_user_operations<C: Communicator>(_: C) {}

#[cfg(not(all(msmpi, target_arch = "x86")))]
unsafe extern "C" fn unsafe_add(
    invec: *mut c_void,
    inoutvec: *mut c_void,
    len: *mut c_int,
    _datatype: *mut MPI_Datatype,
) {
    use std::slice;

    let x: &[Rank] = slice::from_raw_parts(invec as *const Rank, *len as usize);
    let y: &mut [Rank] = slice::from_raw_parts_mut(inoutvec as *mut Rank, *len as usize);
    for (&x_i, y_i) in x.iter().zip(y) {
        *y_i += x_i;
    }
}

#[cfg(all(msmpi, target_arch = "x86"))]
unsafe extern "stdcall" fn unsafe_add(
    invec: *mut c_void,
    inoutvec: *mut c_void,
    len: *mut c_int,
    _datatype: *mut MPI_Datatype,
) {
    use std::slice;

    let x: &[Rank] = slice::from_raw_parts(invec as *const Rank, *len as usize);
    let y: &mut [Rank] = slice::from_raw_parts_mut(inoutvec as *mut Rank, *len as usize);
    for (&x_i, y_i) in x.iter().zip(y) {
        *y_i += x_i;
    }
}

fn main() {
    let universe = mpi::initialize().unwrap();
    let world = universe.world();
    let rank = world.rank();
    let size = world.size();
    let root_rank = 0;

    if rank == root_rank {
        let mut sum: Rank = 0;
        mpi::request::scope(|scope| {
            world
                .process_at_rank(root_rank)
                .immediate_reduce_into_root(scope, &rank, &mut sum, SystemOperation::sum())
                .wait();
        });
        assert_eq!(sum, size * (size - 1) / 2);
    } else {
        mpi::request::scope(|scope| {
            world
                .process_at_rank(root_rank)
                .immediate_reduce_into(scope, &rank, SystemOperation::sum())
                .wait();
        });
    }

    let mut max: Rank = -1;

    mpi::request::scope(|scope| {
        world
            .immediate_all_reduce_into(scope, &rank, &mut max, SystemOperation::max())
            .wait();
    });
    assert_eq!(max, size - 1);

    let a = (0..size).collect::<Vec<_>>();
    let mut b: Rank = 0;

    mpi::request::scope(|scope| {
        world
            .immediate_reduce_scatter_block_into(scope, &a[..], &mut b, SystemOperation::product())
            .wait();
    });
    assert_eq!(b, rank.wrapping_pow(size as u32));

    test_user_operations(universe.world());

    let mut d = 0;
    let op = unsafe { UnsafeUserOperation::commutative(unsafe_add) };
    mpi::request::scope(|scope| {
        world
            .immediate_all_reduce_into(scope, &rank, &mut d, &op)
            .wait();
    });
    assert_eq!(d, size * (size - 1) / 2);
}

fn immediate_reduce_scatter_block_into<'a, S, R, O, Sc>( &self, scope: Sc, sendbuf: &'a S, recvbuf: &'a mut R, op: O, ) -> Request<'a, R, Sc>

where S: 'a + Buffer + ?Sized, R: 'a + BufferMut + ?Sized, O: 'a + Operation, Sc: Scope<'a>,

Initiates a non-blocking element-wise global reduction under the operation op of the input data in sendbuf and scatters the result into equal sized blocks in the receive buffers on all processes.

Examples

See examples/immediate_reduce.rs

Standard section(s)

5.12.9

Examples found in repository

examples/immediate_reduce.rs (line 106)

fn main() {
    let universe = mpi::initialize().unwrap();
    let world = universe.world();
    let rank = world.rank();
    let size = world.size();
    let root_rank = 0;

    if rank == root_rank {
        let mut sum: Rank = 0;
        mpi::request::scope(|scope| {
            world
                .process_at_rank(root_rank)
                .immediate_reduce_into_root(scope, &rank, &mut sum, SystemOperation::sum())
                .wait();
        });
        assert_eq!(sum, size * (size - 1) / 2);
    } else {
        mpi::request::scope(|scope| {
            world
                .process_at_rank(root_rank)
                .immediate_reduce_into(scope, &rank, SystemOperation::sum())
                .wait();
        });
    }

    let mut max: Rank = -1;

    mpi::request::scope(|scope| {
        world
            .immediate_all_reduce_into(scope, &rank, &mut max, SystemOperation::max())
            .wait();
    });
    assert_eq!(max, size - 1);

    let a = (0..size).collect::<Vec<_>>();
    let mut b: Rank = 0;

    mpi::request::scope(|scope| {
        world
            .immediate_reduce_scatter_block_into(scope, &a[..], &mut b, SystemOperation::product())
            .wait();
    });
    assert_eq!(b, rank.wrapping_pow(size as u32));

    test_user_operations(universe.world());

    let mut d = 0;
    let op = unsafe { UnsafeUserOperation::commutative(unsafe_add) };
    mpi::request::scope(|scope| {
        world
            .immediate_all_reduce_into(scope, &rank, &mut d, &op)
            .wait();
    });
    assert_eq!(d, size * (size - 1) / 2);
}

fn immediate_scan_into<'a, S, R, O, Sc>( &self, scope: Sc, sendbuf: &'a S, recvbuf: &'a mut R, op: O, ) -> Request<'a, R, Sc>

where S: 'a + Buffer + ?Sized, R: 'a + BufferMut + ?Sized, O: 'a + Operation, Sc: Scope<'a>,

Initiates a non-blocking global inclusive prefix reduction of the data in sendbuf into recvbuf under operation op.

Examples

See examples/immediate_scan.rs

Standard section(s)

5.12.11

Examples found in repository

examples/immediate_scan.rs (line 19)

fn main() {
    let universe = mpi::initialize().unwrap();
    let world = universe.world();
    let rank = world.rank();

    let mut x = 0;
    mpi::request::scope(|scope| {
        world
            .immediate_scan_into(scope, &rank, &mut x, SystemOperation::sum())
            .wait();
    });
    assert_eq!(x, (rank * (rank + 1)) / 2);

    let y = rank + 1;
    let mut z = 0;
    mpi::request::scope(|scope| {
        world
            .immediate_exclusive_scan_into(scope, &y, &mut z, SystemOperation::product())
            .wait();
    });
    if rank > 0 {
        assert_eq!(z, fac(y - 1));
    }
}

fn immediate_exclusive_scan_into<'a, S, R, O, Sc>( &self, scope: Sc, sendbuf: &'a S, recvbuf: &'a mut R, op: O, ) -> Request<'a, R, Sc>

where S: 'a + Buffer + ?Sized, R: 'a + BufferMut + ?Sized, O: 'a + Operation, Sc: Scope<'a>,

Initiates a non-blocking global exclusive prefix reduction of the data in sendbuf into recvbuf under operation op.

Examples

See examples/immediate_scan.rs

Standard section(s)

5.12.12

Examples found in repository

examples/immediate_scan.rs (line 28)

fn main() {
    let universe = mpi::initialize().unwrap();
    let world = universe.world();
    let rank = world.rank();

    let mut x = 0;
    mpi::request::scope(|scope| {
        world
            .immediate_scan_into(scope, &rank, &mut x, SystemOperation::sum())
            .wait();
    });
    assert_eq!(x, (rank * (rank + 1)) / 2);

    let y = rank + 1;
    let mut z = 0;
    mpi::request::scope(|scope| {
        world
            .immediate_exclusive_scan_into(scope, &y, &mut z, SystemOperation::product())
            .wait();
    });
    if rank > 0 {
        assert_eq!(z, fac(y - 1));
    }
}

Dyn Compatibility

This trait is not dyn compatible.

In older versions of Rust, dyn compatibility was called "object safety", so this trait is not object safe.

Implementors

impl<C: Communicator + ?Sized> CommunicatorCollectives for C