epoll

#[derive(Debug)]
pub struct Selector {
    #[cfg(debug_assertions)]
    id: usize,
    ep: OwnedFd,
}

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

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

        // Avoid using file descriptors below 3 as they are used for stdio
        let fd = cvt(unsafe { libc::fcntl(self.as_raw_fd(), cmd, 3) })?;

fn main() {
  println!("epoll_pwait = {}", libc::SYS_epoll_pwait);
  println!("epoll_wait  = {}", libc::SYS_epoll_wait);
}

error[E0425]: cannot find value `SYS_epoll_wait` in crate `libc`
   --> src/bin/no_epoll_wait.rs:3:38
    |
  3 |   println!("epoll_wait  = {}", libc::SYS_epoll_wait);
    |                                      ^^^^^^^^^^^^^^
    |
   ::: ~/.cargo/registry/src/index.crates.io-1949cf8c6b5b557f/libc-0.2.191/src/unix/linux_like/linux/musl/b64/aarch64/mod.rs:323:1
    |
323 | pub const SYS_epoll_pwait: c_long = 22;
    | --------------------------------- similarly named constant `SYS_epoll_pwait` defined here
    |
help: a constant with a similar name exists
    |
  3 |   println!("epoll_wait  = {}", libc::SYS_epoll_pwait);
    |                                                +

For more information about this error, try `rustc --explain E0425`.
error: could not compile `exp-mio-epoll` (bin "no_epoll_wait") due to 1 previous error

    pub fn register(&self, fd: RawFd, token: Token, interests: Interest) -> io::Result<()> {
        let mut event = libc::epoll_event {
            events: interests_to_epoll(interests),
            u64: usize::from(token) as u64,
            #[cfg(target_os = "redox")]
            _pad: 0,
        };

        let ep = self.ep.as_raw_fd();
        syscall!(epoll_ctl(ep, libc::EPOLL_CTL_ADD, fd, &mut event)).map(|_| ())
    }

    cfg_any_os_ext! {
    pub fn reregister(&self, fd: RawFd, token: Token, interests: Interest) -> io::Result<()> {
        // ... 和 register 逐字相同，只是 EPOLL_CTL_MOD
    }

    pub fn deregister(&self, fd: RawFd) -> io::Result<()> {
        let ep = self.ep.as_raw_fd();
        syscall!(epoll_ctl(ep, libc::EPOLL_CTL_DEL, fd, ptr::null_mut())).map(|_| ())
    }
    }

use std::io::Write;
use std::net::Shutdown;
use std::os::fd::AsRawFd;
use std::time::Duration;

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

// The same server socket sits in two epoll instances: mio's
// (EPOLLIN|EPOLLRDHUP|EPOLLET) and a raw one without EPOLLRDHUP.
fn raw_wait(ep: i32) -> String {
  let mut ev = [libc::epoll_event { events: 0, u64: 0 }; 4];
  let n = unsafe { libc::epoll_wait(ep, ev.as_mut_ptr(), 4, 100) };
  let names = [(libc::EPOLLIN, "IN"), (libc::EPOLLOUT, "OUT"), (libc::EPOLLRDHUP, "RDHUP"), (libc::EPOLLHUP, "HUP"), (libc::EPOLLERR, "ERR")];
  (0..n as usize)
    .map(|i| {
      let bits = ev[i].events as i32;
      names.iter().filter(|(b, _)| bits & b != 0).map(|(_, s)| *s).collect::<Vec<_>>().join("|")
    })
    .collect::<Vec<_>>()
    .join(", ")
}

fn mio_wait(poll: &mut Poll) -> String {
  let mut events = Events::with_capacity(4);
  poll.poll(&mut events, Some(Duration::from_millis(100))).unwrap();
  events
    .iter()
    .map(|e| format!("readable={} read_closed={}", e.is_readable(), e.is_read_closed()))
    .collect::<Vec<_>>()
    .join(", ")
}

fn main() -> std::io::Result<()> {
  for case in ["peer writes 1 byte", "peer shutdown(Write)"] {
    let l = std::net::TcpListener::bind("127.0.0.1:0")?;
    let mut c = std::net::TcpStream::connect(l.local_addr()?)?;
    let (s, _) = l.accept()?;
    s.set_nonblocking(true)?;
    let mut s = TcpStream::from_std(s);
    let mut poll = Poll::new()?;
    poll.registry().register(&mut s, Token(1), Interest::READABLE)?;
    let ep = unsafe { libc::epoll_create1(libc::EPOLL_CLOEXEC) };
    let mut ev = libc::epoll_event { events: (libc::EPOLLIN | libc::EPOLLET) as u32, u64: 1 };
    assert_eq!(unsafe { libc::epoll_ctl(ep, libc::EPOLL_CTL_ADD, s.as_raw_fd(), &mut ev) }, 0);

    if case.starts_with("peer writes") {
      c.write_all(b"x")?;
    } else {
      c.shutdown(Shutdown::Write)?;
    }
    eprintln!("{case}:");
    eprintln!("  raw epoll (IN|ET)    -> {}", raw_wait(ep));
    eprintln!("  mio (IN|RDHUP|ET)    -> {}", mio_wait(&mut poll));
    unsafe { libc::close(ep) };
  }
  Ok(())
}

epoll_create1(EPOLL_CLOEXEC)            = 6
epoll_ctl(6, EPOLL_CTL_ADD, 5, {events=EPOLLIN|EPOLLRDHUP|EPOLLET, data=0x1}) = 0
epoll_create1(EPOLL_CLOEXEC)            = 7
epoll_ctl(7, EPOLL_CTL_ADD, 5, {events=EPOLLIN|EPOLLET, data=0x1}) = 0
sendto(4, "x", 1, MSG_NOSIGNAL, NULL, 0) = 1
peer writes 1 byte:
epoll_pwait(7, [{events=EPOLLIN, data=0x1}], 4, 100, NULL, 8) = 1
  raw epoll (IN|ET)    -> IN
epoll_pwait(6, [{events=EPOLLIN, data=0x1}], 4, 100, NULL, 8) = 1
  mio (IN|RDHUP|ET)    -> readable=true read_closed=false
epoll_create1(EPOLL_CLOEXEC)            = 6
epoll_ctl(6, EPOLL_CTL_ADD, 5, {events=EPOLLIN|EPOLLRDHUP|EPOLLET, data=0x1}) = 0
epoll_create1(EPOLL_CLOEXEC)            = 7
epoll_ctl(7, EPOLL_CTL_ADD, 5, {events=EPOLLIN|EPOLLET, data=0x1}) = 0
shutdown(4, SHUT_WR)                    = 0
peer shutdown(Write):
epoll_pwait(7, [{events=EPOLLIN, data=0x1}], 4, 100, NULL, 8) = 1
  raw epoll (IN|ET)    -> IN
epoll_pwait(6, [{events=EPOLLIN|EPOLLRDHUP, data=0x1}], 4, 100, NULL, 8) = 1
  mio (IN|RDHUP|ET)    -> readable=true read_closed=true

use std::io::Read;
use std::net::Shutdown;
use std::os::fd::AsRawFd;
use std::time::Duration;

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

// stderr is unbuffered, so these lines interleave with strace in order
fn step(poll: &mut Poll, title: &str) {
  let mut events = Events::with_capacity(8);
  poll.poll(&mut events, Some(Duration::from_millis(100))).unwrap();
  eprintln!("== {title}: {} event(s)", events.iter().count());
  for e in &events {
    eprintln!("   {e:?}");
  }
}

fn main() -> std::io::Result<()> {
  let l = std::net::TcpListener::bind("127.0.0.1:0")?;
  let c = std::net::TcpStream::connect(l.local_addr()?)?;
  let (s, _) = l.accept()?;
  s.set_nonblocking(true)?;
  let mut s = TcpStream::from_std(s);
  let mut poll = Poll::new()?;
  eprintln!("server fd = {}", s.as_raw_fd());

  poll.registry().register(&mut s, Token(1), Interest::READABLE)?;
  step(&mut poll, "1. register READABLE");

  poll.registry().reregister(&mut s, Token(2), Interest::READABLE | Interest::WRITABLE)?;
  step(&mut poll, "2. reregister READABLE|WRITABLE as Token(2)");
  step(&mut poll, "2a. poll again, nothing changed");
  poll.registry().reregister(&mut s, Token(2), Interest::READABLE | Interest::WRITABLE)?;
  step(&mut poll, "2b. reregister the same READABLE|WRITABLE again");

  poll.registry().reregister(&mut s, Token(3), Interest::READABLE | Interest::PRIORITY)?;
  let n = unsafe { libc::send(c.as_raw_fd(), b"!".as_ptr().cast(), 1, libc::MSG_OOB) };
  assert_eq!(n, 1);
  step(&mut poll, "3. reregister READABLE|PRIORITY as Token(3), peer sends 1 byte MSG_OOB");

  c.shutdown(Shutdown::Write)?;
  step(&mut poll, "4. peer shutdown(Write)");
  let mut buf = [0u8; 16];
  eprintln!("   read -> {:?}", s.read(&mut buf).map_err(|e| e.kind()));

  poll.registry().deregister(&mut s)?;
  drop(c); // full close from the peer: would be HUP if still registered
  step(&mut poll, "5. deregister, peer closes");
  Ok(())
}

epoll_create1(EPOLL_CLOEXEC)            = 6
server fd = 5
epoll_ctl(6, EPOLL_CTL_ADD, 5, {events=EPOLLIN|EPOLLRDHUP|EPOLLET, data=0x1}) = 0
epoll_pwait(6, [], 8, 100, NULL, 8)     = 0
== 1. register READABLE: 0 event(s)
epoll_ctl(6, EPOLL_CTL_MOD, 5, {events=EPOLLIN|EPOLLOUT|EPOLLRDHUP|EPOLLET, data=0x2}) = 0
epoll_pwait(6, [{events=EPOLLOUT, data=0x2}], 8, 100, NULL, 8) = 1
== 2. reregister READABLE|WRITABLE as Token(2): 1 event(s)
   Event { token: Token(2), readable: false, writable: true, error: false, read_closed: false, write_closed: false, priority: false, aio: false, lio: false }
epoll_pwait(6, [], 8, 100, NULL, 8)     = 0
== 2a. poll again, nothing changed: 0 event(s)
epoll_ctl(6, EPOLL_CTL_MOD, 5, {events=EPOLLIN|EPOLLOUT|EPOLLRDHUP|EPOLLET, data=0x2}) = 0
epoll_pwait(6, [{events=EPOLLOUT, data=0x2}], 8, 100, NULL, 8) = 1
== 2b. reregister the same READABLE|WRITABLE again: 1 event(s)
   Event { token: Token(2), readable: false, writable: true, error: false, read_closed: false, write_closed: false, priority: false, aio: false, lio: false }
epoll_ctl(6, EPOLL_CTL_MOD, 5, {events=EPOLLIN|EPOLLPRI|EPOLLRDHUP|EPOLLET, data=0x3}) = 0
sendto(4, "!", 1, MSG_OOB, NULL, 0)     = 1
epoll_pwait(6, [{events=EPOLLPRI, data=0x3}], 8, 100, NULL, 8) = 1
== 3. reregister READABLE|PRIORITY as Token(3), peer sends 1 byte MSG_OOB: 1 event(s)
   Event { token: Token(3), readable: true, writable: false, error: false, read_closed: false, write_closed: false, priority: true, aio: false, lio: false }
shutdown(4, SHUT_WR)                    = 0
epoll_pwait(6, [{events=EPOLLIN|EPOLLPRI|EPOLLRDHUP, data=0x3}], 8, 100, NULL, 8) = 1
== 4. peer shutdown(Write): 1 event(s)
   Event { token: Token(3), readable: true, writable: false, error: false, read_closed: true, write_closed: false, priority: true, aio: false, lio: false }
recvfrom(5, "", 16, 0, NULL, NULL)      = 0
   read -> Ok(0)
epoll_ctl(6, EPOLL_CTL_DEL, 5, NULL)    = 0
epoll_pwait(6, [], 8, 100, NULL, 8)     = 0
== 5. deregister, peer closes: 0 event(s)
+++ exited with 0 +++

    pub fn debug_details(f: &mut fmt::Formatter<'_>, event: &Event) -> fmt::Result {
        #[allow(clippy::trivially_copy_pass_by_ref)]
        fn check_events(got: &u32, want: &libc::c_int) -> bool {
            (*got as libc::c_int & want) != 0
        }
        debug_detail!(
            EventsDetails(u32),
            check_events,
            libc::EPOLLIN,
            libc::EPOLLPRI,
            // ... EPOLLOUT、EPOLLRDNORM、EPOLLRDBAND、EPOLLWRNORM、EPOLLWRBAND、
            //     EPOLLMSG、EPOLLERR、EPOLLHUP、EPOLLET、EPOLLRDHUP、
            //     EPOLLONESHOT、EPOLLEXCLUSIVE、EPOLLWAKEUP
            libc::EPOLL_CLOEXEC,
        );

        // Can't reference fields in packed structures.
        let e_u64 = event.u64;
        f.debug_struct("epoll_event")
            .field("events", &EventsDetails(event.events))
            .field("u64", &e_u64)
            .finish()
    }

use std::net::Shutdown;
use std::time::Duration;

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

fn main() -> std::io::Result<()> {
  let l = std::net::TcpListener::bind("127.0.0.1:0")?;
  let c = std::net::TcpStream::connect(l.local_addr()?)?;
  let (s, _) = l.accept()?;
  s.set_nonblocking(true)?;
  let mut s = TcpStream::from_std(s);
  let mut poll = Poll::new()?;
  poll.registry().register(&mut s, Token(7), Interest::READABLE | Interest::WRITABLE)?;
  c.shutdown(Shutdown::Write)?;
  let mut events = Events::with_capacity(8);
  poll.poll(&mut events, Some(Duration::from_millis(100)))?;
  for e in &events {
    println!("{e:#?}");
  }
  Ok(())
}

Event {
    token: Token(
        7,
    ),
    readable: true,
    writable: true,
    error: false,
    read_closed: true,
    write_closed: false,
    priority: false,
    aio: false,
    lio: false,
    details: epoll_event {
        events: EPOLLIN|EPOLLOUT|EPOLLRDHUP,
        u64: 7,
    },
}

linux initial: 1 event(s)
linux after peer shutdown(Write): 1 event(s)
   Event { token: Token(1), readable: false, writable: true, error: false, read_closed: false, write_closed: false, priority: false, aio: false, lio: false }

macos initial: 1 event(s)
macos after peer shutdown(Write): 0 event(s)