Token、Interest 与 Event

#[derive(Copy, Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub struct Token(pub usize);

impl From<Token> for usize {
    fn from(val: Token) -> usize {
        val.0
    }
}

        let mut event = libc::epoll_event {
            events: interests_to_epoll(interests),
            u64: usize::from(token) as u64,
            #[cfg(target_os = "redox")]
            _pad: 0,
        };

macro_rules! kevent {
    ($id: expr, $filter: expr, $flags: expr, $data: expr) => {
        libc::kevent {
            ident: $id as libc::uintptr_t,
            filter: $filter as Filter,
            flags: $flags,
            udata: $data as UData,
            ..unsafe { mem::zeroed() }
        }
    };
}

use std::time::Duration;

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

fn main() -> std::io::Result<()> {
  let mut poll = Poll::new()?;
  let mut listener = TcpListener::bind("127.0.0.1:0".parse().unwrap())?;
  poll.registry().register(&mut listener, Token(42), Interest::READABLE)?;
  let mut stream = TcpStream::connect(listener.local_addr()?)?;
  let big = Token(0x1234_5678_9abc_def0);
  poll.registry().register(&mut stream, big, Interest::READABLE | Interest::WRITABLE)?;

  let mut events = Events::with_capacity(8);
  for round in 1..=2 {
    poll.poll(&mut events, Some(Duration::from_millis(100)))?;
    for e in &events {
      println!("poll #{round}: Token({:#x})  {}", e.token().0, raw(e));
    }
  }
  Ok(())
}

epoll_ctl(3, EPOLL_CTL_ADD, 4, {events=EPOLLIN|EPOLLRDHUP|EPOLLET, data=0x2a}) = 0
epoll_ctl(3, EPOLL_CTL_ADD, 5, {events=EPOLLIN|EPOLLOUT|EPOLLRDHUP|EPOLLET, data=0x123456789abcdef0}) = 0
epoll_pwait(3, [{events=EPOLLIN, data=0x2a}, {events=EPOLLOUT, data=0x123456789abcdef0}], 8, 100, NULL, 8) = 2
poll #1: Token(0x2a)  events=IN data=0x2a
poll #1: Token(0x123456789abcdef0)  events=OUT data=0x123456789abcdef0
epoll_pwait(3, [], 8, 100, NULL, 8)     = 0
+++ exited with 0 +++

poll #1: Token(0x123456789abcdef0)  filter=WRITE flags=CLEAR fflags=0 data=146988 udata=0x123456789abcdef0
poll #2: Token(0x2a)  filter=READ flags=CLEAR fflags=0 data=1 udata=0x2a

#[derive(Copy, PartialEq, Eq, Clone, PartialOrd, Ord)]
pub struct Interest(NonZeroU8);

// These must be unique.
const READABLE: u8 = 0b0001;
const WRITABLE: u8 = 0b0010;
// The following are not available on all platforms.
const AIO: u8 = 0b0100;
const LIO: u8 = 0b1000;
const PRIORITY: u8 = 0b10000;

    pub fn remove(self, other: Interest) -> Option<Interest> {
        NonZeroU8::new(self.0.get() & !other.0.get()).map(Interest)
    }

size_of::<Interest>()         = 1
size_of::<Option<Interest>>() = 1
READABLE | WRITABLE           = READABLE | WRITABLE
rw.remove(READABLE)           = Some(WRITABLE)
rw.remove(rw)                 = None
READABLE < WRITABLE           = true

error[E0599]: no associated function or constant named `PRIORITY` found for struct `Interest` in the current scope
 --> src/bin/platform_interest.rs:4:42
  |
4 |   let _ = Interest::READABLE | Interest::PRIORITY;
  |                                          ^^^^^^^^ associated function or constant not found in `Interest`

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

error[E0599]: no associated function or constant named `AIO` found for struct `Interest` in the current scope
 --> src/bin/platform_interest.rs:5:42
  |
5 |   let _ = Interest::READABLE | Interest::AIO;
  |                                          ^^^ associated function or constant not found in `Interest`

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

fn interests_to_epoll(interests: Interest) -> u32 {
    let mut kind = EPOLLET;

    if interests.is_readable() {
        kind = kind | EPOLLIN | EPOLLRDHUP;
    }

    if interests.is_writable() {
        kind |= EPOLLOUT;
    }

    if interests.is_priority() {
        kind |= EPOLLPRI;
    }

    kind as u32
}

#[derive(Clone)]
#[repr(transparent)]
pub struct Event {
    inner: sys::Event,
}

    pub(crate) fn from_sys_event_ref(sys_event: &sys::Event) -> &Event {
        unsafe {
            // This is safe because the memory layout of `Event` is
            // the same as `sys::Event` due to the `repr(transparent)` attribute.
            &*(sys_event as *const sys::Event as *const Event)
        }
    }

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

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

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

fn drain(poll: &mut Poll, title: &str) {
  let mut events = Events::with_capacity(16);
  poll.poll(&mut events, Some(Duration::from_millis(100))).unwrap();
  println!("{title}");
  if events.is_empty() {
    println!("    (no events)");
  }
  for e in &events {
    println!("    {}  => {}", raw(e), bools(e));
  }
}

// registered server-side stream + std client, after the initial writable event
fn setup() -> (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);
  let mut poll = Poll::new().unwrap();
  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();
  (poll, s, c)
}

fn main() -> std::io::Result<()> {
  println!("os={}", std::env::consts::OS);
  let (mut poll, _s, mut c) = setup();
  c.write_all(b"hi")?;
  drain(&mut poll, "1. peer writes 2 bytes");

  let (mut poll, _s, c) = setup();
  c.shutdown(Shutdown::Write)?;
  drain(&mut poll, "2. peer shutdown(Write) (FIN)");

  let (mut poll, _s, c) = setup();
  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 libc::socklen_t);
  }
  drop(c);
  drain(&mut poll, "3. peer closes with SO_LINGER=0 (RST)");

  let (mut poll, s, _c) = setup();
  s.shutdown(Shutdown::Write)?;
  drain(&mut poll, "4. local shutdown(Write)");

  let (mut poll, s, _c) = setup();
  s.shutdown(Shutdown::Both)?;
  drain(&mut poll, "5. local shutdown(Both)");

  // 6. connect to a port nobody listens on
  let port = {
    let l = std::net::TcpListener::bind("127.0.0.1:0")?;
    l.local_addr()?
  };
  let mut poll = Poll::new()?;
  let mut s = TcpStream::connect(port)?;
  poll.registry().register(&mut s, Token(1), RW)?;
  drain(&mut poll, "6. connect to a closed port");

  // 7/8. pipes
  let mut poll = Poll::new()?;
  let (tx, mut rx) = pipe::new()?;
  poll.registry().register(&mut rx, Token(1), Interest::READABLE)?;
  drop(tx);
  drain(&mut poll, "7. pipe: write end closed, reader registered READABLE");

  let mut poll = Poll::new()?;
  let (mut tx, rx) = pipe::new()?;
  poll.registry().register(&mut tx, Token(1), Interest::WRITABLE)?;
  let mut events = Events::with_capacity(16);
  poll.poll(&mut events, Some(Duration::from_millis(100)))?;
  drop(rx);
  drain(&mut poll, "8. pipe: read end closed, writer registered WRITABLE");

  #[cfg(target_os = "linux")]
  {
    let (mut poll, mut s, c) = setup();
    poll.registry().reregister(&mut s, Token(1), 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);
    drain(&mut poll, "9. peer sends 1 byte MSG_OOB, READABLE|PRIORITY");
    let mut buf = [0u8; 8];
    let _ = std::io::Read::read(&mut s, &mut buf);
  }
  Ok(())
}

os=linux
1. peer writes 2 bytes
    events=IN|OUT data=0x1  => readable writable
2. peer shutdown(Write) (FIN)
    events=IN|OUT|RDHUP data=0x1  => readable writable read_closed
3. peer closes with SO_LINGER=0 (RST)
    events=IN|OUT|ERR|HUP|RDHUP data=0x1  => readable writable error read_closed write_closed
4. local shutdown(Write)
    events=OUT data=0x1  => writable
5. local shutdown(Both)
    events=IN|OUT|HUP|RDHUP data=0x1  => readable writable read_closed write_closed
6. connect to a closed port
    events=IN|OUT|ERR|HUP|RDHUP data=0x1  => readable writable error read_closed write_closed
7. pipe: write end closed, reader registered READABLE
    events=HUP data=0x1  => read_closed write_closed
8. pipe: read end closed, writer registered WRITABLE
    events=OUT|ERR data=0x1  => writable error write_closed
9. peer sends 1 byte MSG_OOB, READABLE|PRIORITY
    events=PRI data=0x1  => readable priority

os=macos
1. peer writes 2 bytes
    filter=READ flags=CLEAR fflags=0 data=2 udata=0x1  => readable
2. peer shutdown(Write) (FIN)
    filter=READ flags=EOF|CLEAR fflags=0 data=0 udata=0x1  => readable read_closed
3. peer closes with SO_LINGER=0 (RST)
    filter=WRITE flags=EOF|CLEAR fflags=54 data=146988 udata=0x1  => writable error write_closed
    filter=READ flags=EOF|CLEAR fflags=54 data=0 udata=0x1  => readable error read_closed
4. local shutdown(Write)
    filter=WRITE flags=EOF|CLEAR fflags=0 data=146988 udata=0x1  => writable write_closed
5. local shutdown(Both)
    filter=READ flags=EOF|CLEAR fflags=0 data=0 udata=0x1  => readable read_closed
    filter=WRITE flags=EOF|CLEAR fflags=0 data=146988 udata=0x1  => writable write_closed
6. connect to a closed port
    filter=WRITE flags=EOF|CLEAR fflags=61 data=2048 udata=0x1  => writable error write_closed
    filter=READ flags=EOF|CLEAR fflags=61 data=0 udata=0x1  => readable error read_closed
7. pipe: write end closed, reader registered READABLE
    filter=READ flags=EOF|CLEAR fflags=0 data=0 udata=0x1  => readable read_closed
8. pipe: read end closed, writer registered WRITABLE
    filter=WRITE flags=EOF|CLEAR fflags=0 data=0 udata=0x1  => writable write_closed

use std::io::Write;
use std::time::Duration;

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

fn main() -> std::io::Result<()> {
  println!("os={}", std::env::consts::OS);
  let interests = [("READABLE", Interest::READABLE), ("READABLE|WRITABLE", Interest::READABLE | Interest::WRITABLE)];
  for (name, interest) in interests {
    let l = std::net::TcpListener::bind("127.0.0.1:0")?;
    let mut poll = Poll::new()?;
    let mut keep = vec![];
    for i in 0..5 {
      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);
      c.write_all(b"x")?; // every socket is readable (and writable) before the first poll
      poll.registry().register(&mut s, Token(i), interest)?;
      keep.push((s, c));
    }
    std::thread::sleep(Duration::from_millis(50));
    let mut events = Events::with_capacity(2);
    println!("{name}, Events::with_capacity(2), capacity()={}", events.capacity());
    for round in 1.. {
      poll.poll(&mut events, Some(Duration::from_millis(100)))?;
      let got: Vec<String> = events.iter().map(|e| {
        let rw = match (e.is_readable(), e.is_writable()) { (true, true) => "rw", (true, false) => "r", (false, true) => "w", _ => "-" };
        format!("{}{rw}", e.token().0)
      }).collect();
      println!("  poll #{round}: {} events {:?}", got.len(), got);
      if got.is_empty() { break; }
    }
  }
  Ok(())
}

os=linux
READABLE, Events::with_capacity(2), capacity()=2
  poll #1: 2 events ["0r", "1r"]
  poll #2: 2 events ["2r", "3r"]
  poll #3: 1 events ["4r"]
  poll #4: 0 events []
READABLE|WRITABLE, Events::with_capacity(2), capacity()=2
  poll #1: 2 events ["0rw", "1rw"]
  poll #2: 2 events ["2rw", "3rw"]
  poll #3: 1 events ["4rw"]
  poll #4: 0 events []

epoll_pwait(4, [{events=EPOLLIN, data=0}, {events=EPOLLIN, data=0x1}], 2, 100, NULL, 8) = 2
epoll_pwait(4, [{events=EPOLLIN, data=0x2}, {events=EPOLLIN, data=0x3}], 2, 100, NULL, 8) = 2
epoll_pwait(4, [{events=EPOLLIN, data=0x4}], 2, 100, NULL, 8) = 1
epoll_pwait(4, [], 2, 100, NULL, 8)     = 0

macos capacity=0 -> Ok(()), 0 events, 0ms

epoll_pwait(3, 0x8, 0, 500, NULL, 8)    = -1 EINVAL (Invalid argument)
linux capacity=0 -> Err(Os { code: 22, kind: InvalidInput, message: "Invalid argument" }), 0 events, 0ms

os=macos
READABLE, Events::with_capacity(2), capacity()=2
  poll #1: 2 events ["0r", "1r"]
  poll #2: 2 events ["2r", "3r"]
  poll #3: 1 events ["4r"]
  poll #4: 0 events []
READABLE|WRITABLE, Events::with_capacity(2), capacity()=2
  poll #1: 2 events ["0w", "0r"]
  poll #2: 2 events ["1w", "1r"]
  poll #3: 2 events ["2w", "2r"]
  poll #4: 2 events ["3w", "3r"]
  poll #5: 2 events ["4w", "4r"]
  poll #6: 0 events []