Immediate send/recvs hang when the packets being sent are large about rsmpi HOT 16 OPEN

  mpi::request::multiple_scope(nreqs as usize, |scope, coll| {

        for (i, packet) in packets.iter().enumerate() {
            let sreq = world
                .process_at_rank(packet_destinations[i])
                .immediate_send(scope, &packet[..]);
                coll.add(sreq);
        }

        for (i, buffer) in buffers.iter_mut().enumerate() {
            let rreq = world
                .process_at_rank(received_packet_sources[i])
                .immediate_receive_into(scope, &mut buffer[..]);
            coll.add(rreq);
        }
        let mut out = vec![];
        coll.wait_all(&mut out);
        assert_eq!(out.len(), nreqs as usize);
    });

use bincode;
#[allow(unused_imports)]
use crdts::{CmRDT, Dot, GCounter};
use mpi::request::WaitGuard;
use mpi::traits::*;
use rand::{thread_rng, Rng};
use std::{thread, time};

fn main() {
    #[allow(unused_variables)]
    // number of fake jobs to process (i.e: count up to this number)
    const NJOBS: u16 = 2000;

    // Represents the worst time job can take to finish
    let dur = time::Duration::from_millis(25);

    // Represents the probability of performing a large job:
    const P: f64 = 0.015; // 15% chance of a long running job (i.e: 2.5 seconds)
    let mut rng = thread_rng();

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

    // Get the rank and size of the communicator
    let size = world.size();
    let rank = world.rank();

    // Define the processor that sends to this rank and  that receives from this rank:
    let prev_rank = (rank + size - 1) % size;
    let next_rank = (rank + 1) % size;

    // Init local replica of the count
    let mut local_counter = GCounter::new();

    println!(
        "Prev rank: {}, This rank: {}, Next rank: {}",
        prev_rank, rank, next_rank
    );
    for _n in 0..NJOBS {
        // simulate a long or short process between incrementing the count:
        if rng.gen_bool(P) {
             thread::sleep(dur);
             //println!("Running a long job on processor {}", rank);
        }

        // Declare intent to increment by creating an `Op`. Use rank as the uid of the replica
        let inc_op = local_counter.inc(rank);

        // Apply the operation to our local counter first and then send the `Op` off to others
        local_counter.apply(inc_op.clone()); // Apply the increment:

        // Serialize the local operation for sending:
        let send_op: Vec<u8> = bincode::serialize(&inc_op).unwrap();

        // For getting the most up-to-date count of the remote Dot
        let mut remote_op: Vec<u8> = Vec::new();

        // Simple point-to-point messaging in a ring configuration:
        mpi::request::scope(|scope| {
            // Receive the operation to increment the counter from the previous rank:
            let _rreq = WaitGuard::from(
                world
                    .process_at_rank(prev_rank)
                    .immediate_receive_into(scope, &mut remote_op),
            );

            // Send the change that was applied by the local counter:
            let _sreq = WaitGuard::from(
                world
                    .process_at_rank(next_rank)
                    .immediate_ready_send(scope, &send_op),
            );
        });

        // Deserialize the remote_op:
        let recv_op: Dot<mpi::Rank> = bincode::deserialize(&remote_op).unwrap();

        // Apply the change to the local counter:
        local_counter.apply(recv_op);
    }
    println!(
        "Processor {} says the count = {:?}",
        rank,
        local_counter.read()
    );
}

use bincode;
#[allow(unused_imports)]
use crdts::{CmRDT, Dot, GCounter};
use mpi::request::WaitGuard;
use mpi::traits::*;
use rand::{thread_rng, Rng};
use std::{sync::WaitTimeoutResult, thread, time};

fn main() {
    #[allow(unused_variables)]
    // number of fake jobs to process (i.e: count up to this number)
    const NJOBS: u16 = 2000;

    // Represents the worst time job can take to finish
    let dur = time::Duration::from_millis(25);

    // Represents the probability of performing a large job:
    const P: f64 = 0.015; // 1.5% chance of a long running job (i.e: 25 milliseconds)
    let mut rng = thread_rng();

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

    // Get the rank and size of the communicator
    let size = world.size();
    let rank = world.rank();

    // Define the processor that sends to this rank and  that receives from this rank:
    let prev_rank = (rank + size - 1) % size;
    let next_rank = (rank + 1) % size;

    // Init local replica of the count
    let mut local_counter = GCounter::new();

    println!(
        "Prev rank: {}, This rank: {}, Next rank: {}",
        prev_rank, rank, next_rank
    );
    for _n in 0..NJOBS {
        // simulate a long or short process between incrementing the count:
        if rng.gen_bool(P) {
             thread::sleep(dur);
             // println!("Running a long job on processor {}", rank);
        }

        // Declare intent to increment by creating an `Op`. Use rank as the uid of the replica
        let inc_op = local_counter.inc(rank);

        // Apply the operation to our local counter first and then send the `Op` off to others
        local_counter.apply(inc_op.clone()); // Apply the increment:

        // Serialize the local operation for sending:
        let send_op: Vec<u8> = bincode::serialize(&inc_op).unwrap();

        // For getting the most up-to-date count of the remote Dot
        let mut remote_op = vec![0u8; 24]; // the buffer must provide a size

        // Simple point-to-point messaging in a ring configuration:
        mpi::request::scope(|scope| {
            let _rreq = WaitGuard::from(
                world
                    .process_at_rank(prev_rank)
                    .immediate_receive_into(scope, &mut remote_op),
            );
            let _sreq = WaitGuard::from(
                world
                    .process_at_rank(next_rank)
                    .immediate_send(scope, &send_op),
            );
        });

        // Deserialize the remote_op:
        let recv_op: Dot<mpi::Rank> = bincode::deserialize(&remote_op).unwrap();

        // Apply the change to the local counter:
        local_counter.apply(recv_op);
    }
    println!(
        "Processor {} says the count = {:?}",
        rank,
        local_counter.read()
    );
}