#[derive(Debug, StructOpt)]
#[structopt(
    name = "mping",
    version = "0.1.4",
    about = "A multi-targets ping tool, which supports 10,000 packets/second."
)]
struct Opt {
    #[structopt(
        short = "w",
        long = "timeout",
        default_value = "1",
        help = "timeout in seconds"
    )]
    timeout: u64,
    #[structopt(short = "t", long = "ttl", default_value = "64", help = "time to live")]
    ttl: u32,
    #[structopt(short = "z", long = "tos", help = "type of service")]
    tos: Option<u32>,
    ....
}

fn parse_ips(input: &str) -> Vec<IpAddr> {
    let mut ips = Vec::new();
    for s in input.split(',') {
        match s.parse::<IpNetwork>() {
            Ok(network) => {
                for ip in network.iter() {
                    ips.push(ip);
                }
            }
            Err(_) => {
                if let Ok(ip) = s.parse::<IpAddr>() {
                    ips.push(ip);
                } else if let Ok(addrs) = (s, 0).to_socket_addrs() {
                    for addr in addrs {
                        if let IpAddr::V4(ipv4) = addr.ip() {
                            ips.push(IpAddr::V4(ipv4));
                            break;
                        }
                    }
                }
            }
        }
    }
    ips
}

let popt = mping::ping::PingOption {
    timeout,
    ttl: opt.ttl,
    tos: opt.tos,
    ident: pid,
    len: opt.size,
    rate: opt.rate,
    delay: opt.delay,
    count: opt.count,
};
mping::ping::ping(ip_addrs, popt)?;

thread::spawn(move || {
    let socket = Socket::new(Domain::IPV4, Type::RAW, Some(Protocol::ICMPV4)).unwrap();
    socket.set_ttl(popt.ttl).unwrap();
    socket.set_write_timeout(Some(popt.timeout)).unwrap();
    if let Some(tos_value) = popt.tos  {
        socket.set_tos(tos_value).unwrap();
    }
    ...
}

let zero_payload = vec![0; popt.len];
let one_payload = vec![1; popt.len];
let fivea_payload = vec![0x5A; popt.len];
let payloads: [&[u8]; 4] = [&rand_payload, &zero_payload, &one_payload, &fivea_payload];
let limiter = SyncLimiter::full(popt.rate, Duration::from_millis(1000));
let mut seq = 1u16;
let mut sent_count = 0;

let mut buf = vec![0; 8 + payload.len()]; // 8 bytes of header, then payload
let mut packet = echo_request::MutableEchoRequestPacket::new(&mut buf[..]).unwrap();
packet.set_icmp_type(icmp::IcmpTypes::EchoRequest);
packet.set_identifier(pid);
packet.set_sequence_number(seq);
...

for ip in &addrs {
    let dest = SocketAddr::new(*ip, 0);
    let data = send_buckets.lock().unwrap();
    data.add(...);
    drop(data);
    match socket.send_to(&buf, &dest.into()) {
        Ok(_) => {}
        Err(e) => {
            error!("Error in send: {:?}", e);
            return;
        }
    }
}

let socket2 = socket.try_clone().expect("Failed to clone socket");

let mut buffer: [u8; 2048] = [0; 2048];
    loop {
        let size = match socket2.read(&mut buffer) {
            Ok(n) => n,
            Err(e) => {
                if let Some(err_code) = e.raw_os_error() {
                    if err_code == 11 {
                        // { code: 11, kind: WouldBlock, message: "Resource temporarily unavailable" }
                        continue;
                    }
                }
                error!("Error in read: {:?}", &e);
                break;
            }
        };
        let buf = &buffer[..size];
        ...
    }

let buf = &buffer[..size];
let ipv4_packet = Ipv4Packet::new(buf).unwrap();
let icmp_packet = pnet_packet::icmp::IcmpPacket::new(ipv4_packet.payload()).unwrap();
if icmp_packet.get_icmp_type() != IcmpTypes::EchoReply
    || icmp_packet.get_icmp_code() != echo_reply::IcmpCodes::NoCode
{
    continue;
}
// 获得ICMP echo reply对象
let echo_reply = match icmp::echo_reply::EchoReplyPacket::new(icmp_packet.packet()) {
    Some(echo_reply) => echo_reply,
    None => {
        continue;
    }
};
// 一般使用进程的id做identifier。我们检查回包是不是和pid一致
if echo_reply.get_identifier() != pid {
    continue;
}
// 检查有没有改包
if payloads[echo_reply.get_sequence_number() as usize % payloads.len()][16..]
    != echo_reply.payload()[16..]
{
    warn!(
        "bitflip detected! seq={:?},",
        echo_reply.get_sequence_number()
    );
}
// 准备统计数据
let payload = echo_reply.payload();
let ts_bytes = &payload[..16];
let txts = u128::from_be_bytes(ts_bytes.try_into().unwrap());
let dest_ip = ipv4_packet.get_source();
let now = SystemTime::now();
let since_the_epoch = now.duration_since(UNIX_EPOCH).unwrap();
let timestamp = since_the_epoch.as_nanos();
// 把返回结果加入到统计数据的bucket中
let buckets = read_buckets.lock().unwrap();
buckets.add_reply(
    txts / 1_000_000_000,
    Result {
        ...
    },
);

#[derive(Default)]
pub struct Buckets {
    pub buckets: Mutex<BinaryHeap<Bucket>>,
    pub map: Mutex<HashMap<u128, Bucket>>,
}

impl Buckets {
    pub fn new() -> Buckets {
        Buckets {
            buckets: Mutex::new(BinaryHeap::new()),
            map: Mutex::new(HashMap::new()),
        }
    }
    pub fn add(&self, key: u128, value: Result) {
        let mut map = self.map.lock().unwrap();
        map.entry(key).or_insert_with(|| {
            let bucket = Bucket {
                key,
                value: RwLock::new(HashMap::new()),
            };
            self.buckets.lock().unwrap().push(bucket.clone());
            bucket
        });
        let bucket = map.get(&key).unwrap();
        bucket.add(value);
    }
    pub fn add_reply(&self, key: u128, result: Result) {
        let mut map = self.map.lock().unwrap();
        map.entry(key).or_insert_with(|| {
            self.buckets.lock().unwrap().push(Bucket::new(key));
            Bucket::new(key)
        });
        let bucket = map.get(&key).unwrap();
        bucket.add_reply(result);
    }
    pub fn pop(&self) -> Option<Bucket> {
        let mut buckets = self.buckets.lock().unwrap();
        let bucket = buckets.pop()?;
        let bucket = self.map.lock().unwrap().remove(&bucket.key).unwrap();
        Some(bucket)
    }
    pub fn last(&self) -> Option<Bucket> {
        let buckets = self.buckets.lock().unwrap();
        buckets.peek().cloned()
    }
}

pub struct Bucket {
    pub key: u128,
    pub value: RwLock<HashMap<String, Result>>,
}
impl Clone for Bucket {
    fn clone(&self) -> Self {
        let value = self.value.read().unwrap().clone();
        Bucket {
            key: self.key,
            value: RwLock::new(value),
        }
    }
}

let ticker = Ticker::new(0.., Duration::from_secs(1));
for _ in ticker {
    ...
}

let buckets = buckets.lock().unwrap();
let bucket = buckets.last(); // 得到最久的篮子
if bucket.is_none() {
    continue;
}
// 检查篮子是否已经处理过了
let bucket = bucket.unwrap();
if bucket.key <= last_key {
    buckets.pop();
    continue;
}
if bucket.key
    <= SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap()
        .as_nanos()
        - delay
{
    if let Some(pop) = buckets.pop() {
        if pop.key < bucket.key {
            continue;
        }
        last_key = pop.key;
        // 统计结果
        let mut target_results = BTreeMap::new();
        ...
        // 打印统计结果
        
    }
}

let enable = SOF_TIMESTAMPING_SOFTWARE
    | SOF_TIMESTAMPING_TX_SOFTWARE
    | SOF_TIMESTAMPING_RX_SOFTWARE
    | SOF_TIMESTAMPING_SYS_HARDWARE
    | SOF_TIMESTAMPING_TX_HARDWARE
    | SOF_TIMESTAMPING_RX_HARDWARE
    | SOF_TIMESTAMPING_RAW_HARDWARE
    | SOF_TIMESTAMPING_OPT_CMSG
    | SOF_TIMESTAMPING_OPT_TSONLY;
let ret = unsafe {
    setsockopt(
        socket2.as_raw_fd(),
        SOL_SOCKET,
        SO_TIMESTAMPING,
        &enable as *const _ as *const c_void,
        mem::size_of_val(&enable) as u32,
    )
};
if ret == -1 {
    warn!("Failed to set read SO_TIMESTAMPING");
    let enable: c_int = 1;
    let ret = unsafe {
        setsockopt(
            socket2.as_raw_fd(),
            SOL_SOCKET,
            SO_TIMESTAMP,
            &enable as *const _ as *const c_void,
            std::mem::size_of_val(&enable) as u32,
        )
    };
    if ret == -1 {
        warn!("Failed to set SO_TIMESTAMP");
    }
}

let mut buffer: [u8; 2048] = [0; 2048];
let mut control_buf = [0; 1024];
let mut iovec = iovec {
    iov_base: buffer.as_mut_ptr() as *mut c_void,
    iov_len: buffer.len(),
};
let mut msghdr = msghdr {
    msg_name: std::ptr::null_mut(),
    msg_namelen: 0,
    msg_iov: &mut iovec,
    msg_iovlen: 1,
    msg_control: control_buf.as_mut_ptr() as *mut c_void,
    msg_controllen: control_buf.len(),
    msg_flags: 0,
};

loop {
        let nbytes = unsafe { recvmsg(socket2.as_raw_fd(), &mut msghdr, 0) };
        if nbytes == -1 {
            let err = Error::last_os_error();
            if err.kind() == ErrorKind::WouldBlock {
                continue;
            }
            error!("Failed torr receive message");
            return Err(Error::new(ErrorKind::Other, "Failed to receive message").into());
        }
        let buf = &buffer[..nbytes as usize];
        let ipv4_packet = Ipv4Packet::new(buf).unwrap();
        let icmp_packet = pnet_packet::icmp::IcmpPacket::new(ipv4_packet.payload()).unwrap();
        ...
}

fn get_timestamp(msghdr: &mut msghdr) -> Option<SystemTime> {
    let mut cmsg: *mut cmsghdr = unsafe { libc::CMSG_FIRSTHDR(msghdr) };
    while !cmsg.is_null() {
        if unsafe { (*cmsg).cmsg_level == SOL_SOCKET && (*cmsg).cmsg_type == SO_TIMESTAMP } {
            let tv: *mut timeval = unsafe { libc::CMSG_DATA(cmsg) } as *mut timeval;
            let timestamp = unsafe { *tv };
            return Some(
                SystemTime::UNIX_EPOCH
                    + Duration::new(timestamp.tv_sec as u64, timestamp.tv_usec as u32 * 1000),
            );
        }
        if unsafe { (*cmsg).cmsg_level == SOL_SOCKET && (*cmsg).cmsg_type == SCM_TIMESTAMPING } {
            let tv: *mut [timespec; 3] = unsafe { libc::CMSG_DATA(cmsg) } as *mut [timespec; 3];
            let timestamps = unsafe { *tv };
            for timestamp in ×tamps {
                if timestamp.tv_sec != 0 || timestamp.tv_nsec != 0 {
                    let seconds = Duration::from_secs(timestamp.tv_sec as u64);
                    let nanoseconds = Duration::from_nanos(timestamp.tv_nsec as u64);
                    if let Some(duration) = seconds.checked_add(nanoseconds) {
                        return Some(SystemTime::UNIX_EPOCH + duration);
                    }
                }
            }
        }
        cmsg = unsafe { libc::CMSG_NXTHDR(msghdr, cmsg) };
    }
    None
}