/// Creates a separate to concurrently receive ICMP packets that are sent, and returns a JoinHandle that
/// allows the received packets count and lost packets count to be accessed.
///
/// # Arguments
/// * `not_done` - Whether this process is exiting.
/// * `receiver` - TransportReceiver for receiving packets.
/// * `total_packets` - Number of packets sent so far.
/// * `iterations` - Maximum number of packets to send.
/// * `identifier` - A number uniquely identifying this process.
/// * `timeout` - How long to wait for an echo reply when an echo request is sent.
/// * `addr` - Address of where to send packets. This is mainly used for printing to stdout.
Creates a separate to concurrently receive ICMP packets that are sent, and returns a JoinHandle that
allows the received packets count and lost packets count to be accessed.
Creates a separate to concurrently receive ICMP packets that are sent and returns a JoinHandle that allows the received packets count and lost packets count to be accessed .
fn make_icmp_receiver_thread(
    not_done: sync::Arc<atomic::AtomicBool>,
    mut receiver: TransportReceiver,
    total_packets: sync::Arc<sync::Mutex<usize>>,
    iterations: usize,
    identifier: u16,
    timeout: time::Duration,
    addr: String,
) -> thread::JoinHandle<(usize, usize)> {
    thread::spawn(move || {
        let mut receiver_iter = icmp_packet_iter(&mut receiver);
        let mut total_rtt = 0;
        let mut received_packets = 0;
        let mut lost_packets = 0;
        let mut timed_out_packets = 0;
        // HashSet to keep track of duplicate packets
        let mut received = collections::HashSet::new();
        let receiver_delay = time::Duration::from_millis(100);
        while not_done.load(atomic::Ordering::SeqCst) {
            // receiver_delay should be low to keep this thread responsive to not_done changes
            let next_res = receiver_iter
                .next_with_timeout(receiver_delay)
                .expect("Error receiving packet! 9S unhappy :( ");
            let (res_packet, res_ip) = match next_res {
                Some(res) => res,
                None => continue,
            };
            let curr_time = time::SystemTime::now()
                .duration_since(time::UNIX_EPOCH)
                .unwrap()
                .as_millis();
            // we don't care if this is changed by the main thread (we want consistency in the
            // printed results), so just read it once
            let total_packets_sent = { *total_packets.lock().unwrap() };
            match res_packet.get_icmp_type() {
                icmp::IcmpTypes::EchoReply => {
                    let (res_identifier, res_seq_num, res_send_time) =
                        read_payload(res_packet.payload());
                    if res_identifier == identifier && (res_seq_num as usize) <= total_packets_sent
                    {
                        // at this point we double checked that the received packet is definitely a packet
                        // sent by this specific process (and not some other process doing a ping)
                        if received.contains(&res_seq_num) {
                            println!(
                                "9S received a duplicate packet (seq num: {}) from {}!",
                                res_seq_num, addr
                            );
                        } else {
                            let elapsed_ms = curr_time - res_send_time;
                            // a packet is timed out even if we receive it after the timeout
                            if elapsed_ms > timeout.as_millis() {
                                timed_out_packets += 1;
                                println!("9S received timed out packet (seq num: {}) from {} in {} ms!\n\tSent {}, with {} ({:.1}%) lost so far.",
                                res_seq_num, addr, elapsed_ms, total_packets_sent, lost_packets, lost_packets as f64 / total_packets_sent as f64 * 100f64);
                            } else {
                                total_rtt += elapsed_ms;
                                received_packets += 1;
                                println!("9S received packet (seq num: {}) from {} in {} ms (avg rtt: {:.1} ms)!\n\tSent {}, with {} ({:.1}%) lost so far.",
                                    res_seq_num, addr, elapsed_ms, total_rtt as f64 / received_packets as f64, total_packets_sent, lost_packets, lost_packets as f64 / total_packets_sent as f64 * 100f64);
                            }
                            received.insert(res_seq_num);
                        }
                    }
                    // not our packet, not our problem
                }
                icmp::IcmpTypes::DestinationUnreachable => {
                    // only part of the original packet is returned
                    // payload bytes 0..4 is unused, 4..24 is the IPv4 header,
                    // 24..28 is the ICMP header, and 28..32 is the identifier and sequence number
                    let (res_identifier, res_seq_num) =
                        read_payload_id(&res_packet.payload()[28..32]);
                    if res_identifier == identifier && (res_seq_num as usize) <= total_packets_sent
                    {
                        if received.contains(&res_seq_num) {
                            println!(
                                "9S received a duplicate packet (seq num: {}) from {}!",
                                res_seq_num, addr
                            );
                        } else {
                            lost_packets += 1;
                            println!("9S received a destination unreachable packet (seq num: {}) from {} (code: {})!\n\tSent {}, with {} ({:.1}%) lost so far.",
                                res_seq_num, addr, res_packet.get_icmp_code().0, total_packets_sent, lost_packets, lost_packets as f64 / total_packets_sent as f64 * 100f64);
                            received.insert(res_seq_num);
                        }
                    }
                    // not our packet, not our problem
                }
                icmp::IcmpTypes::TimeExceeded => {
                    // only part of the original packet is returned
                    // payload bytes 0..4 is unused, 4..24 is the IPv4 header,
                    // 24..28 is the ICMP header, and 28..32 is the identifier and sequence number
                    let (res_identifier, res_seq_num) =
                        read_payload_id(&res_packet.payload()[28..32]);
                    if res_identifier == identifier && (res_seq_num as usize) <= total_packets_sent
                    {
                        if received.contains(&res_seq_num) {
                            println!(
                                "9S received a duplicate packet (seq num: {}) from {}!",
                                res_seq_num, addr
                            );
                        } else {
                            lost_packets += 1;
                            println!("9S received a time exceeded packet (seq num: {}) before reaching {} (last host: {}, code: {})!\n\tSent {}, with {} ({:.1}%) lost so far.",
                                res_seq_num, addr, res_ip, res_packet.get_icmp_code().0, total_packets_sent, lost_packets, lost_packets as f64 / total_packets_sent as f64 * 100f64);
                            received.insert(res_seq_num);
                        }
                    }
                    // not our packet, not our problem
                }
                _ => (), // quietly skip this received packet if it is not what we were expecting
            }
            // future work: also count timed out packets that were never received
            if received_packets + lost_packets + timed_out_packets >= iterations {
                not_done.store(false, atomic::Ordering::SeqCst);
                break;
            }
        }
        (received_packets, lost_packets)
    })
}
fn make_icmp_receiver_thread ( not_done : sync :: Arc < atomic :: AtomicBool > , mut receiver : TransportReceiver , total_packets : sync :: Arc < sync :: Mutex < usize > > , iterations : usize , identifier : u16 , timeout : time :: Duration , addr : String , ) -> thread :: JoinHandle < ( usize , usize ) > { thread :: spawn ( move | | { let mut receiver_iter = icmp_packet_iter ( & mut receiver ) ; let mut total_rtt = 0 ; let mut received_packets = 0 ; let mut lost_packets = 0 ; let mut timed_out_packets = 0 ; // HashSet to keep track of duplicate packets let mut received = collections :: HashSet :: new ( ) ; let receiver_delay = time :: Duration :: from_millis ( 100 ) ; while not_done . load ( atomic :: Ordering :: SeqCst ) { // receiver_delay should be low to keep this thread responsive to not_done changes let next_res = receiver_iter . next_with_timeout ( receiver_delay ) . expect ( " " ) ; let ( res_packet , res_ip ) = match next_res { Some ( res ) => res , None => continue , } ; let curr_time = time :: SystemTime :: now ( ) . duration_since ( time :: UNIX_EPOCH ) . unwrap ( ) . as_millis ( ) ; // we don't care if this is changed by the main thread (we want consistency in the // printed results), so just read it once let total_packets_sent = { * total_packets . lock ( ) . unwrap ( ) } ; match res_packet . get_icmp_type ( ) { icmp :: IcmpTypes :: EchoReply => { let ( res_identifier , res_seq_num , res_send_time ) = read_payload ( res_packet . payload ( ) ) ; if res_identifier == identifier && ( res_seq_num as usize ) <= total_packets_sent { // at this point we double checked that the received packet is definitely a packet // sent by this specific process (and not some other process doing a ping) if received . contains ( & res_seq_num ) { println ! ( " " res_seq_num addr ) ; } else { let elapsed_ms = curr_time - res_send_time ; // a packet is timed out even if we receive it after the timeout if elapsed_ms > timeout . as_millis ( ) { timed_out_packets += 1 ; println ! ( " \n \t " res_seq_num addr elapsed_ms total_packets_sent lost_packets lost_packets as f64 total_packets_sent as f64 100f64 ) ; } else { total_rtt += elapsed_ms ; received_packets += 1 ; println ! ( " \n \t " res_seq_num addr elapsed_ms total_rtt as f64 received_packets as f64 total_packets_sent lost_packets lost_packets as f64 total_packets_sent as f64 100f64 ) ; } received . insert ( res_seq_num ) ; } } // not our packet, not our problem } icmp :: IcmpTypes :: DestinationUnreachable => { // only part of the original packet is returned // payload bytes 0..4 is unused, 4..24 is the IPv4 header, // 24..28 is the ICMP header, and 28..32 is the identifier and sequence number let ( res_identifier , res_seq_num ) = read_payload_id ( & res_packet . payload ( ) [ 28 .. 32 ] ) ; if res_identifier == identifier && ( res_seq_num as usize ) <= total_packets_sent { if received . contains ( & res_seq_num ) { println ! ( " " res_seq_num addr ) ; } else { lost_packets += 1 ; println ! ( " \n \t " res_seq_num addr res_packet get_icmp_code ( ) 0 total_packets_sent lost_packets lost_packets as f64 total_packets_sent as f64 100f64 ) ; received . insert ( res_seq_num ) ; } } // not our packet, not our problem } icmp :: IcmpTypes :: TimeExceeded => { // only part of the original packet is returned // payload bytes 0..4 is unused, 4..24 is the IPv4 header, // 24..28 is the ICMP header, and 28..32 is the identifier and sequence number let ( res_identifier , res_seq_num ) = read_payload_id ( & res_packet . payload ( ) [ 28 .. 32 ] ) ; if res_identifier == identifier && ( res_seq_num as usize ) <= total_packets_sent { if received . contains ( & res_seq_num ) { println ! ( " " res_seq_num addr ) ; } else { lost_packets += 1 ; println ! ( " \n \t " res_seq_num addr res_ip res_packet get_icmp_code ( ) 0 total_packets_sent lost_packets lost_packets as f64 total_packets_sent as f64 100f64 ) ; received . insert ( res_seq_num ) ; } } // not our packet, not our problem } _ => ( ) , // quietly skip this received packet if it is not what we were expecting } // future work: also count timed out packets that were never received if received_packets + lost_packets + timed_out_packets >= iterations { not_done . store ( false , atomic :: Ordering :: SeqCst ) ; break ; } } ( received_packets , lost_packets ) } ) }
Creates a separate to concurrently receive ICMP packets that are sent, and returns a JoinHandle that
allows the received packets count and lost packets count to be accessed.
Creates a separate to concurrently receive ICMP packets that are sent and returns a JoinHandle that allows the received packets count and lost packets count to be accessed .
not_done
sync::Arc<atomic::AtomicBool>
receiver
TransportReceiver
total_packets
sync::Arc<sync::Mutex<usize>>
iterations
usize
identifier
u16
timeout
time::Duration
addr
String