fn receiving_part(
    config: &Config,
    queue: &Arc<Mutex<BinaryHeap<PacketWrapper>>>,
    condvar: &Arc<Condvar>,
    socket: &Arc<UdpSocket>,
    thread_name: &str,
    brk: Arc<AtomicBool>,
) -> JoinHandle<()> {
    let config = config.clone();
    let queue = queue.clone();
    let condvar = condvar.clone();
    let socket = socket.clone();

    thread::Builder::new()
        .name(format!("{}_receive", thread_name))
        .spawn(move || {
            // create variables
            let mut buff = vec![0; BUFFER_SIZE];
            let mut rand_gen = thread_rng();
            let probability_dist = Uniform::new(0.0, 1.0);
            let byte_dist = Uniform::new(0, 255);

            while !brk.load(Ordering::SeqCst) {
                // set socket timeout
                socket.set_read_timeout(Some(Duration::from_millis(1000)))
                      .expect("Can't change read timeout of the packet");
                // receive packet
                let recv = recv_with_timeout(&socket, &mut buff, Box::new(&config));
                if let Err(_) = recv {
                    continue;
                };
                let (size, sender) = recv.unwrap();
                config.vlog(&format!("Received {}b of data from {}.", size, sender));

                // drop packet if dropout
                if rand_gen.sample(probability_dist) < config.drop_rate {
                    config.vlog("Packet drop");
                    continue;
                }

                // modify packet and shorten it if necessary
                let content_length = min(size, config.packet_size as usize);
                if config.modify_prob > 0.0 {
                    for i in 0..content_length {
                        if rand_gen.sample(probability_dist) < config.modify_prob {
                            buff[i] = rand_gen.sample(byte_dist);
                        }
                    }
                }
                let content = Vec::from(&buff[..content_length]);

                // get delay and create wrapper
                let delay: f32 = f32::max(0.0, config.delay_std * rand_gen.gen::<f32>() + config.delay_mean);
                let wrapper = PacketWrapper::new(content, delay as u32);

                // add packet to the queue
                {
                    let mut queue = queue.lock().expect("Can't lock mutex from receiving part");
                    queue.push(wrapper);
                    condvar.notify_one();
                }
                config.vlog(&format!("Packet add to the queue"));
            }
        }).expect(&format!("Can't create receiving part of the {}", thread_name))
}
/// Handles receiving part of the communication.
/// It receives packets from `socket` and add them to the `queue`.
/// After adding content to the `queue` it notifies other thread (one) using `condvar` variable.
/// It decides about the delay, modification, and whether the packet should be dropped.
Handles receiving part of the communication.
It receives packets from `socket` and add them to the `queue`.
After adding content to the `queue` it notifies other thread (one) using `condvar` variable.
It decides about the delay, modification, and whether the packet should be dropped.
Handles receiving part of the communication . It receives packets from ` socket ` and add them to the ` queue ` . After adding content to the ` queue ` it notifies other thread ( one ) using ` condvar ` variable . It decides about the delay modification and whether the packet should be dropped .
fn receiving_part(
    config: &Config,
    queue: &Arc<Mutex<BinaryHeap<PacketWrapper>>>,
    condvar: &Arc<Condvar>,
    socket: &Arc<UdpSocket>,
    thread_name: &str,
    brk: Arc<AtomicBool>,
) -> JoinHandle<()> {
    let config = config.clone();
    let queue = queue.clone();
    let condvar = condvar.clone();
    let socket = socket.clone();
    thread::Builder::new()
        .name(format!("{}_receive", thread_name))
        .spawn(move || {
            // create variables
            let mut buff = vec![0; BUFFER_SIZE];
            let mut rand_gen = thread_rng();
            let probability_dist = Uniform::new(0.0, 1.0);
            let byte_dist = Uniform::new(0, 255);
            while !brk.load(Ordering::SeqCst) {
                // set socket timeout
                socket.set_read_timeout(Some(Duration::from_millis(1000)))
                      .expect("Can't change read timeout of the packet");
                // receive packet
                let recv = recv_with_timeout(&socket, &mut buff, Box::new(&config));
                if let Err(_) = recv {
                    continue;
                };
                let (size, sender) = recv.unwrap();
                config.vlog(&format!("Received {}b of data from {}.", size, sender));
                // drop packet if dropout
                if rand_gen.sample(probability_dist) < config.drop_rate {
                    config.vlog("Packet drop");
                    continue;
                }
                // modify packet and shorten it if necessary
                let content_length = min(size, config.packet_size as usize);
                if config.modify_prob > 0.0 {
                    for i in 0..content_length {
                        if rand_gen.sample(probability_dist) < config.modify_prob {
                            buff[i] = rand_gen.sample(byte_dist);
                        }
                    }
                }
                let content = Vec::from(&buff[..content_length]);
                // get delay and create wrapper
                let delay: f32 = f32::max(0.0, config.delay_std * rand_gen.gen::<f32>() + config.delay_mean);
                let wrapper = PacketWrapper::new(content, delay as u32);
                // add packet to the queue
                {
                    let mut queue = queue.lock().expect("Can't lock mutex from receiving part");
                    queue.push(wrapper);
                    condvar.notify_one();
                }
                config.vlog(&format!("Packet add to the queue"));
            }
        }).expect(&format!("Can't create receiving part of the {}", thread_name))
}
fn receiving_part ( config : & Config , queue : & Arc < Mutex < BinaryHeap < PacketWrapper > > > , condvar : & Arc < Condvar > , socket : & Arc < UdpSocket > , thread_name : & str , brk : Arc < AtomicBool > , ) -> JoinHandle < ( ) > { let config = config . clone ( ) ; let queue = queue . clone ( ) ; let condvar = condvar . clone ( ) ; let socket = socket . clone ( ) ; thread :: Builder :: new ( ) . name ( format ! ( " " thread_name ) ) . spawn ( move | | { // create variables let mut buff = vec ! [ 0 BUFFER_SIZE ] ; let mut rand_gen = thread_rng ( ) ; let probability_dist = Uniform :: new ( 0.0 , 1.0 ) ; let byte_dist = Uniform :: new ( 0 , 255 ) ; while ! brk . load ( Ordering :: SeqCst ) { // set socket timeout socket . set_read_timeout ( Some ( Duration :: from_millis ( 1000 ) ) ) . expect ( " " ) ; // receive packet let recv = recv_with_timeout ( & socket , & mut buff , Box :: new ( & config ) ) ; if let Err ( _ ) = recv { continue ; } ; let ( size , sender ) = recv . unwrap ( ) ; config . vlog ( & format ! ( " " size sender ) ) ; // drop packet if dropout if rand_gen . sample ( probability_dist ) < config . drop_rate { config . vlog ( " " ) ; continue ; } // modify packet and shorten it if necessary let content_length = min ( size , config . packet_size as usize ) ; if config . modify_prob > 0.0 { for i in 0 .. content_length { if rand_gen . sample ( probability_dist ) < config . modify_prob { buff [ i ] = rand_gen . sample ( byte_dist ) ; } } } let content = Vec :: from ( & buff [ .. content_length ] ) ; // get delay and create wrapper let delay : f32 = f32 :: max ( 0.0 , config . delay_std * rand_gen . gen :: < f32 > ( ) + config . delay_mean ) ; let wrapper = PacketWrapper :: new ( content , delay as u32 ) ; // add packet to the queue { let mut queue = queue . lock ( ) . expect ( " " ) ; queue . push ( wrapper ) ; condvar . notify_one ( ) ; } config . vlog ( & format ! ( " " ) ) ; } } ) . expect ( & format ! ( " " thread_name ) ) }
Handles receiving part of the communication.
Handles receiving part of the communication .
config
&Config
queue
&Arc<Mutex<BinaryHeap<PacketWrapper>>>
condvar
&Arc<Condvar>
socket
&Arc<UdpSocket>
thread_name
&str
brk
Arc<AtomicBool>
config
&Config
queue
&Arc<Mutex<BinaryHeap<PacketWrapper>>>
condvar
&Arc<Condvar>
socket
&Arc<UdpSocket>
thread_name
&str
brk
Arc<AtomicBool>