kqueue

    pub fn new() -> io::Result<Selector> {
        // SAFETY: `kqueue(2)` ensures the fd is valid.
        let kq = unsafe { OwnedFd::from_raw_fd(syscall!(kqueue())?) };
        syscall!(fcntl(kq.as_raw_fd(), libc::F_SETFD, libc::FD_CLOEXEC))?;
        Ok(Selector {
            #[cfg(debug_assertions)]
            id: NEXT_ID.fetch_add(1, Ordering::Relaxed),
            kq,
        })
    }

    pub fn try_clone(&self) -> io::Result<Selector> {
        self.kq.try_clone().map(|kq| Selector {
            // It's the same selector, so we use the same id.
            #[cfg(debug_assertions)]
            id: self.id,
            kq,
        })
    }

    pub fn register(&self, fd: RawFd, token: Token, interests: Interest) -> io::Result<()> {
        let flags = libc::EV_CLEAR | libc::EV_RECEIPT | libc::EV_ADD;
        // At most we need two changes, but maybe we only need 1.
        let mut changes: [MaybeUninit<libc::kevent>; 2] =
            [MaybeUninit::uninit(), MaybeUninit::uninit()];
        let mut n_changes = 0;

        if interests.is_writable() {
            let kevent = kevent!(fd, libc::EVFILT_WRITE, flags, token.0);
            changes[n_changes] = MaybeUninit::new(kevent);
            n_changes += 1;
        }

        if interests.is_readable() {
            let kevent = kevent!(fd, libc::EVFILT_READ, flags, token.0);
            changes[n_changes] = MaybeUninit::new(kevent);
            n_changes += 1;
        }

        // ... 关于 EPIPE 的注释，见下文
        let changes = unsafe {
            // This is safe because we ensure that at least `n_changes` are in
            // the array.
            slice::from_raw_parts_mut(changes[0].as_mut_ptr(), n_changes)
        };
        kevent_register(self.kq.as_raw_fd(), changes, &[libc::EPIPE as i64])
    }

fn kevent_register(
    kq: RawFd,
    changes: &mut [libc::kevent],
    ignored_errors: &[i64],
) -> io::Result<()> {
    syscall!(kevent(
        kq,
        changes.as_ptr(),
        changes.len() as Count,
        changes.as_mut_ptr(),
        changes.len() as Count,
        ptr::null(),
    ))
    .map(|_| ())
    .or_else(|err| {
        // According to the manual page of FreeBSD: "When kevent() call fails
        // with EINTR error, all changes in the changelist have been applied",
        // so we can safely ignore it.
        if err.raw_os_error() == Some(libc::EINTR) {
            Ok(())
        } else {
            Err(err)
        }
    })
    .and_then(|()| check_errors(changes, ignored_errors))
}

/// Check all events for possible errors, it returns the first error found.
fn check_errors(events: &[libc::kevent], ignored_errors: &[i64]) -> io::Result<()> {
    for event in events {
        // We can't use references to packed structures (in checking the ignored
        // errors), so we need copy the data out before use.
        let data = event.data as _;
        // Check for the error flag, the actual error will be in the `data`
        // field.
        if (event.flags & libc::EV_ERROR != 0) && data != 0 && !ignored_errors.contains(&data) {
            return Err(io::Error::from_raw_os_error(data as i32));
        }
    }
    Ok(())
}

// Raw kevent(2) calls, the way kqueue.rs makes them, with and without EV_RECEIPT.
use std::io::Write;
use std::net::{TcpListener, TcpStream};
use std::os::fd::AsRawFd;
use std::{mem, ptr};

use exp_mio_kqueue::fmt_kevent;

fn kev(fd: i32, filter: i16, flags: u16, udata: usize) -> libc::kevent {
  libc::kevent {
    ident: fd as libc::uintptr_t,
    filter,
    flags,
    udata: udata as *mut libc::c_void,
    ..unsafe { mem::zeroed() }
  }
}

// changelist and eventlist are the same buffer, like kevent_register
fn submit(kq: i32, changes: &mut [libc::kevent], nevents: usize) {
  let zero = libc::timespec { tv_sec: 0, tv_nsec: 0 };
  let n = unsafe {
    libc::kevent(kq, changes.as_ptr(), changes.len() as i32, changes.as_mut_ptr(), nevents as i32, &zero)
  };
  if n < 0 {
    println!("  kevent -> -1, errno={}", std::io::Error::last_os_error());
    return;
  }
  println!("  kevent -> {n}");
  for c in &changes[..n as usize] {
    println!("    {}", fmt_kevent(c));
  }
}

// kevent(kq, NULL, 0, buf, 8, 0): what select would get now
fn pending(kq: i32) {
  let mut buf: [libc::kevent; 8] = unsafe { mem::zeroed() };
  let zero = libc::timespec { tv_sec: 0, tv_nsec: 0 };
  let n = unsafe { libc::kevent(kq, ptr::null(), 0, buf.as_mut_ptr(), 8, &zero) };
  println!("  then select -> {n}");
  for c in &buf[..n.max(0) as usize] {
    println!("    {}", fmt_kevent(c));
  }
}

fn main() {
  let l = TcpListener::bind("127.0.0.1:0").unwrap();
  let mut c = TcpStream::connect(l.local_addr().unwrap()).unwrap();
  let (s, _) = l.accept().unwrap();
  c.write_all(b"hi").unwrap();
  std::thread::sleep(std::time::Duration::from_millis(50));
  let fd = s.as_raw_fd();
  let base = libc::EV_ADD | libc::EV_CLEAR;
  println!("socket fd={fd}, readable (2 bytes) and writable");

  println!("1. EV_ADD|EV_CLEAR, no EV_RECEIPT, eventlist = changelist");
  let kq = unsafe { libc::kqueue() };
  submit(kq, &mut [kev(fd, libc::EVFILT_WRITE, base, 7), kev(fd, libc::EVFILT_READ, base, 7)], 2);
  pending(kq);

  println!("2. EV_ADD|EV_CLEAR|EV_RECEIPT, eventlist = changelist");
  let kq = unsafe { libc::kqueue() };
  let r = base | libc::EV_RECEIPT;
  submit(kq, &mut [kev(fd, libc::EVFILT_WRITE, r, 7), kev(fd, libc::EVFILT_READ, r, 7)], 2);
  pending(kq);

  println!("3. EV_RECEIPT: bad fd 9999 (EV_ADD) + EV_DELETE of a filter never added");
  let kq = unsafe { libc::kqueue() };
  let d = libc::EV_DELETE | libc::EV_RECEIPT;
  submit(kq, &mut [kev(9999, libc::EVFILT_READ, r, 7), kev(fd, libc::EVFILT_WRITE, d, 0)], 2);
  // and EV_DELETE on the bad fd
  submit(kq, &mut [kev(9999, libc::EVFILT_READ, d, 0)], 1);

  println!("4. no EV_RECEIPT, nevents=0: bad fd 9999 first, good fd second");
  let kq = unsafe { libc::kqueue() };
  submit(kq, &mut [kev(9999, libc::EVFILT_READ, base, 7), kev(fd, libc::EVFILT_READ, base, 8)], 0);
  pending(kq);

  println!("5. EV_RECEIPT: pipe write end (read end closed), pipe read end (write end closed)");
  let mut p1 = [0; 2];
  let mut p2 = [0; 2];
  unsafe { libc::pipe(p1.as_mut_ptr()); libc::close(p1[0]); libc::pipe(p2.as_mut_ptr()); libc::close(p2[1]); }
  let kq = unsafe { libc::kqueue() };
  submit(kq, &mut [kev(p1[1], libc::EVFILT_WRITE, r, 3), kev(p2[0], libc::EVFILT_READ, r, 2)], 2);

  println!("6. mio: SourceFd(&9999) register / deregister");
  let poll = mio::Poll::new().unwrap();
  let bad = 9999;
  let res = poll.registry().register(&mut mio::unix::SourceFd(&bad), mio::Token(7), mio::Interest::READABLE);
  println!("  register   {res:?}");
  let res = poll.registry().deregister(&mut mio::unix::SourceFd(&bad));
  println!("  deregister {res:?}");
}

socket fd=5, readable (2 bytes) and writable
1. EV_ADD|EV_CLEAR, no EV_RECEIPT, eventlist = changelist
  kevent -> 2
    ident=5 filter=WRITE flags=ADD|CLEAR fflags=0 data=146988 udata=7
    ident=5 filter=READ flags=ADD|CLEAR fflags=0 data=2 udata=7
  then select -> 0
2. EV_ADD|EV_CLEAR|EV_RECEIPT, eventlist = changelist
  kevent -> 2
    ident=5 filter=WRITE flags=ADD|CLEAR|RECEIPT|ERROR fflags=0 data=0 udata=7
    ident=5 filter=READ flags=ADD|CLEAR|RECEIPT|ERROR fflags=0 data=0 udata=7
  then select -> 2
    ident=5 filter=WRITE flags=ADD|CLEAR|RECEIPT fflags=0 data=146988 udata=7
    ident=5 filter=READ flags=ADD|CLEAR|RECEIPT fflags=0 data=2 udata=7
3. EV_RECEIPT: bad fd 9999 (EV_ADD) + EV_DELETE of a filter never added
  kevent -> 2
    ident=9999 filter=READ flags=ADD|CLEAR|RECEIPT|ERROR fflags=0 data=9 udata=7
    ident=5 filter=WRITE flags=DELETE|RECEIPT|ERROR fflags=0 data=2 udata=0
  kevent -> 1
    ident=9999 filter=READ flags=DELETE|RECEIPT|ERROR fflags=0 data=2 udata=0
4. no EV_RECEIPT, nevents=0: bad fd 9999 first, good fd second
  kevent -> -1, errno=Bad file descriptor (os error 9)
  then select -> 0
5. EV_RECEIPT: pipe write end (read end closed), pipe read end (write end closed)
  kevent -> 2
    ident=11 filter=WRITE flags=ADD|CLEAR|RECEIPT|ERROR fflags=0 data=0 udata=3
    ident=10 filter=READ flags=ADD|CLEAR|RECEIPT|ERROR fflags=0 data=0 udata=2
6. mio: SourceFd(&9999) register / deregister
  register   Err(Os { code: 9, kind: Uncategorized, message: "Bad file descriptor" })
  deregister Ok(())

// Register a kqueue fd in another Poll, like the does_not_register_rw test
use std::os::fd::AsRawFd;
use std::time::Duration;

use exp_mio_kqueue::raw;
use mio::unix::SourceFd;
use mio::{Events, Interest, Poll, Token};

fn main() {
  for (name, interest) in [
    ("READABLE", Interest::READABLE),
    ("WRITABLE", Interest::WRITABLE),
    ("READABLE|WRITABLE", Interest::READABLE.add(Interest::WRITABLE)),
  ] {
    let mut poll = Poll::new().unwrap();
    let inner = Poll::new().unwrap();
    let r = poll.registry().register(&mut SourceFd(&inner.as_raw_fd()), Token(1), interest);
    println!("{name:<18} register {r:?}");
    // make the inner kqueue readable: it gets a pending event
    let waker = mio::Waker::new(inner.registry(), Token(5)).unwrap();
    waker.wake().unwrap();
    let mut events = Events::with_capacity(8);
    poll.poll(&mut events, Some(Duration::from_millis(100))).unwrap();
    println!("{:<18} then poll: {} event(s)", "", events.iter().count());
    for e in &events {
      println!("{:<18}   token={} {}", "", e.token().0, raw(e));
    }
  }
}

    pub fn reregister(&self, fd: RawFd, token: Token, interests: Interest) -> io::Result<()> {
        let flags = libc::EV_CLEAR | libc::EV_RECEIPT;
        let write_flags = if interests.is_writable() {
            flags | libc::EV_ADD
        } else {
            flags | libc::EV_DELETE
        };
        let read_flags = if interests.is_readable() {
            flags | libc::EV_ADD
        } else {
            flags | libc::EV_DELETE
        };

        let mut changes: [libc::kevent; 2] = [
            kevent!(fd, libc::EVFILT_WRITE, write_flags, token.0),
            kevent!(fd, libc::EVFILT_READ, read_flags, token.0),
        ];

        // Since there is no way to check with which interests the fd was
        // registered we modify both readable and write, adding it when required
        // and removing it otherwise, ignoring the ENOENT error when it comes
        // up. The ENOENT error informs us that a filter we're trying to remove
        // wasn't there in first place, but we don't really care since our goal
        // is accomplished.
        //
        // For the explanation of ignoring `EPIPE` see `register`.
        kevent_register(
            self.kq.as_raw_fd(),
            &mut changes,
            &[libc::ENOENT as i64, libc::EPIPE as i64],
        )
    }

// register R|W -> reregister R -> reregister R|W -> deregister, on one socket
use std::io::{Read, Write};
use std::thread::sleep;
use std::time::Duration;

use exp_mio_kqueue::{bools, raw};
use mio::net::TcpStream;
use mio::{Events, Interest, Poll, Token};

const RW: Interest = Interest::READABLE.add(Interest::WRITABLE);

fn show(poll: &mut Poll, title: &str) {
  let mut events = Events::with_capacity(16);
  poll.poll(&mut events, Some(Duration::from_millis(100))).unwrap();
  println!("{title}: {} event(s)", events.iter().count());
  for e in &events {
    println!("    token={} {}  => {}", e.token().0, raw(e), bools(e));
  }
}

fn main() {
  let l = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
  let mut c = std::net::TcpStream::connect(l.local_addr().unwrap()).unwrap();
  let (s, _) = l.accept().unwrap();
  s.set_nonblocking(true).unwrap();
  let mut s = TcpStream::from_std(s);
  let mut poll = Poll::new().unwrap();
  let mut buf = [0u8; 64];
  let pause = || sleep(Duration::from_millis(50));

  c.write_all(b"hi").unwrap();
  pause();
  poll.registry().register(&mut s, Token(7), RW).unwrap();
  show(&mut poll, "1. register R|W (2 bytes waiting)");
  println!("    read {} bytes", s.read(&mut buf).unwrap());

  s.write_all(b"x").unwrap();
  pause();
  show(&mut poll, "2. we write 1 byte, peer acks");

  let r = poll.registry().reregister(&mut s, Token(8), Interest::READABLE);
  println!("3. reregister READABLE, Token(8): {r:?}");
  s.write_all(b"y").unwrap();
  pause();
  show(&mut poll, "   we write 1 byte, peer acks");
  c.write_all(b"abc").unwrap();
  pause();
  show(&mut poll, "   peer writes 3 bytes");
  println!("    read {} bytes", s.read(&mut buf).unwrap());

  let r = poll.registry().reregister(&mut s, Token(8), Interest::READABLE);
  println!("4. reregister READABLE again: {r:?}");
  show(&mut poll, "   (no I/O)");

  let r = poll.registry().reregister(&mut s, Token(9), RW);
  println!("5. reregister R|W, Token(9): {r:?}");
  show(&mut poll, "   (no I/O)");

  let r = poll.registry().deregister(&mut s);
  println!("6. deregister: {r:?}");
  c.write_all(b"defg").unwrap();
  s.write_all(b"z").unwrap();
  pause();
  show(&mut poll, "   both sides write");
  let r = poll.registry().deregister(&mut s);
  println!("7. deregister again: {r:?}");
}

1. register R|W (2 bytes waiting): 2 event(s)
    token=7 ident=5 filter=WRITE flags=ADD|CLEAR|RECEIPT fflags=0 data=146988 udata=7  => writable
    token=7 ident=5 filter=READ flags=ADD|CLEAR|RECEIPT fflags=0 data=2 udata=7  => readable
    read 2 bytes
2. we write 1 byte, peer acks: 1 event(s)
    token=7 ident=5 filter=WRITE flags=ADD|CLEAR|RECEIPT fflags=0 data=146988 udata=7  => writable
3. reregister READABLE, Token(8): Ok(())
   we write 1 byte, peer acks: 0 event(s)
   peer writes 3 bytes: 1 event(s)
    token=8 ident=5 filter=READ flags=ADD|CLEAR|RECEIPT fflags=0 data=3 udata=8  => readable
    read 3 bytes
4. reregister READABLE again: Ok(())
   (no I/O): 0 event(s)
5. reregister R|W, Token(9): Ok(())
   (no I/O): 1 event(s)
    token=9 ident=5 filter=WRITE flags=ADD|CLEAR|RECEIPT fflags=0 data=146988 udata=9  => writable
6. deregister: Ok(())
   both sides write: 0 event(s)
7. deregister again: Ok(())

        let kq = self.kq.as_raw_fd();
        syscall!(kevent(kq, &kevent, 1, ptr::null_mut(), 0, ptr::null()))?;
        Ok(())

// EV_EOF and fflags: peer FIN vs peer RST; then pipes with the other end closed
use std::os::fd::AsRawFd;
use std::thread::sleep;
use std::time::Duration;

use exp_mio_kqueue::{bools, raw};
use mio::net::TcpStream;
use mio::unix::SourceFd;
use mio::{Events, Interest, Poll, Token};

const RW: Interest = Interest::READABLE.add(Interest::WRITABLE);

fn show(poll: &mut Poll, title: &str) {
  let mut events = Events::with_capacity(16);
  poll.poll(&mut events, Some(Duration::from_millis(100))).unwrap();
  println!("{title}: {} event(s)", events.iter().count());
  for e in &events {
    println!("    {}  => {}", raw(e), bools(e));
  }
}

fn pair(poll: &mut Poll) -> (TcpStream, std::net::TcpStream) {
  let l = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
  let c = std::net::TcpStream::connect(l.local_addr().unwrap()).unwrap();
  let (s, _) = l.accept().unwrap();
  s.set_nonblocking(true).unwrap();
  let mut s = TcpStream::from_std(s);
  poll.registry().register(&mut s, Token(1), RW).unwrap();
  let mut events = Events::with_capacity(16);
  poll.poll(&mut events, Some(Duration::from_millis(100))).unwrap(); // initial WRITE
  (s, c)
}

fn main() {
  let mut poll = Poll::new().unwrap();

  let (_s, c) = pair(&mut poll);
  drop(c);
  sleep(Duration::from_millis(50));
  show(&mut poll, "1. peer close() (FIN)");

  let mut poll = Poll::new().unwrap();
  let (_s, c) = pair(&mut poll);
  let linger = libc::linger { l_onoff: 1, l_linger: 0 };
  unsafe {
    libc::setsockopt(c.as_raw_fd(), libc::SOL_SOCKET, libc::SO_LINGER,
      &linger as *const _ as *const libc::c_void, size_of::<libc::linger>() as u32);
  }
  drop(c);
  sleep(Duration::from_millis(50));
  show(&mut poll, "2. peer close() with SO_LINGER 0 (RST)");

  let mut fds = [0; 2];
  unsafe { libc::pipe(fds.as_mut_ptr()) };
  unsafe { libc::close(fds[1]) };
  let mut poll = Poll::new().unwrap();
  let r = poll.registry().register(&mut SourceFd(&fds[0]), Token(2), Interest::READABLE);
  println!("3. pipe read end, write end closed: register {r:?}");
  show(&mut poll, "   poll");

  unsafe { libc::pipe(fds.as_mut_ptr()) };
  unsafe { libc::close(fds[0]) };
  let mut poll = Poll::new().unwrap();
  let r = poll.registry().register(&mut SourceFd(&fds[1]), Token(3), Interest::WRITABLE);
  println!("4. pipe write end, read end closed: register {r:?}");
  show(&mut poll, "   poll");
}

1. peer close() (FIN): 1 event(s)
    ident=6 filter=READ flags=ADD|CLEAR|RECEIPT|EOF fflags=0 data=0 udata=1  => readable read_closed
2. peer close() with SO_LINGER 0 (RST): 2 event(s)
    ident=8 filter=WRITE flags=ADD|CLEAR|RECEIPT|EOF fflags=54 data=146988 udata=1  => writable error write_closed
    ident=8 filter=READ flags=ADD|CLEAR|RECEIPT|EOF fflags=54 data=0 udata=1  => readable error read_closed
3. pipe read end, write end closed: register Ok(())
   poll: 1 event(s)
    ident=5 filter=READ flags=ADD|CLEAR|RECEIPT|EOF fflags=0 data=0 udata=2  => readable read_closed
4. pipe write end, read end closed: register Ok(())
   poll: 1 event(s)
    ident=10 filter=WRITE flags=ADD|CLEAR|RECEIPT|EOF fflags=0 data=0 udata=3  => writable write_closed

READABLE           register Ok(())
                   then poll: 1 event(s)
                     token=1 ident=4 filter=READ flags=ADD|CLEAR|RECEIPT fflags=0 data=1 udata=1
WRITABLE           register Err(Os { code: 22, kind: InvalidInput, message: "Invalid argument" })
                   then poll: 0 event(s)
READABLE|WRITABLE  register Err(Os { code: 22, kind: InvalidInput, message: "Invalid argument" })
                   then poll: 1 event(s)
                     token=1 ident=4 filter=READ flags=ADD|CLEAR|RECEIPT fflags=0 data=1 udata=1