Poll 与 Registry

pub struct Poll {
    registry: Registry,
}

/// Registers I/O resources.
pub struct Registry {
    selector: sys::Selector,
    /// Whether this selector currently has an associated waker.
    #[cfg(all(debug_assertions, not(any(target_os = "wasi", target_os = "horizon"))))]
    has_waker: Arc<AtomicBool>,
}

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

macos: Poll, Registry, Events, Event, Token, Interest are Send + Sync
linux: Poll, Registry, Events, Event, Token, Interest are Send + Sync

    pub fn poll(&mut self, events: &mut Events, timeout: Option<Duration>) -> io::Result<()> {
        self.registry.selector.select(events.sys(), timeout)
    }

    pub fn select(&self, events: &mut Events, timeout: Option<Duration>) -> io::Result<()> {
        let timeout = timeout
            .map(|to| {
                // `Duration::as_millis` truncates, so round up. This avoids
                // turning sub-millisecond timeouts into a zero timeout, unless
                // the caller explicitly requests that by specifying a zero
                // timeout.
                libc::c_int::try_from(
                    to.checked_add(Duration::from_nanos(999_999))
                        .unwrap_or(to)
                        .as_millis(),
                )
                .unwrap_or(libc::c_int::MAX)
            })
            .unwrap_or(-1);

        events.clear();
        syscall!(epoll_wait(
            self.ep.as_raw_fd(),
            events.as_mut_ptr(),
            events.capacity() as i32,
            timeout,
        ))
        .map(|n_events| {
            // This is safe because `epoll_wait` ensures that `n_events` are
            // assigned.
            unsafe { events.set_len(n_events as usize) };
        })
    }

        let timeout = timeout.map(|to| libc::timespec {
            tv_sec: cmp::min(to.as_secs(), libc::time_t::MAX as u64) as libc::time_t,
            // `Duration::subsec_nanos` is guaranteed to be less than one
            // billion (the number of nanoseconds in a second), making the
            // cast to i32 safe. The cast itself is needed for platforms
            // where C's long is only 32 bits.
            tv_nsec: libc::c_long::from(to.subsec_nanos() as i32),
        });
        let timeout = timeout
            .as_ref()
            .map(|s| s as *const _)
            .unwrap_or(ptr::null_mut());

use std::sync::Arc;
use std::thread;
use std::time::{Duration, Instant};

use mio::{Events, Poll, Token, Waker};

// mio 1.0.0 ..= 1.2.0: `as` cast, keeps the low 32 bits
fn old_ms(to: Duration) -> i32 {
  to.checked_add(Duration::from_nanos(999_999)).unwrap_or(to).as_millis() as i32
}

// mio 1.2.1+: try_from, clamp to c_int::MAX
fn new_ms(to: Duration) -> i32 {
  i32::try_from(to.checked_add(Duration::from_nanos(999_999)).unwrap_or(to).as_millis()).unwrap_or(i32::MAX)
}

fn main() -> std::io::Result<()> {
  let day = 86_400;
  let cases = [
    ("ZERO", Duration::ZERO),
    ("1ns", Duration::from_nanos(1)),
    ("1.5ms", Duration::from_micros(1500)),
    ("2^31-1 ms", Duration::from_millis(i32::MAX as u64)),
    ("2^31 ms", Duration::from_millis(1 << 31)),
    ("30 days", Duration::from_secs(30 * day)),
    ("2^32+5 ms", Duration::from_millis((1 << 32) + 5)),
    ("Duration::MAX", Duration::MAX),
  ];
  println!("{:<14} {:>12} {:>12}", "timeout", "old(<=1.2.0)", "new(1.2.1+)");
  for (name, d) in cases {
    println!("{:<14} {:>12} {:>12}", name, old_ms(d), new_ms(d));
  }

  let mut poll = Poll::new()?;
  let waker = Arc::new(Waker::new(poll.registry(), Token(0))?);
  let mut events = Events::with_capacity(8);
  for (name, to) in [
    ("Some(ZERO)", Some(Duration::ZERO)),
    ("Some(1ns)", Some(Duration::from_nanos(1))),
    ("Some(1.5ms)", Some(Duration::from_micros(1500))),
    ("Some(30 days)", Some(Duration::from_secs(30 * day))),
    ("None", None),
  ] {
    let long = to.map_or(true, |t| t > Duration::from_secs(1));
    if long {
      let w = waker.clone();
      thread::spawn(move || {
        thread::sleep(Duration::from_millis(100));
        w.wake().unwrap();
      });
    }
    let start = Instant::now();
    poll.poll(&mut events, to)?;
    println!("poll({name}) -> {} events after {:?}", events.iter().count(), start.elapsed().as_millis());
  }
  Ok(())
}

timeout        old(<=1.2.0)  new(1.2.1+)
ZERO                      0            0
1ns                       1            1
1.5ms                     2            2
2^31-1 ms        2147483647   2147483647
2^31 ms         -2147483648   2147483647
30 days         -1702967296   2147483647
2^32+5 ms                 5   2147483647
Duration::MAX            -1   2147483647
epoll_pwait(3, [], 8, 0, NULL, 8)       = 0
poll(Some(ZERO)) -> 0 events after 0
epoll_pwait(3, [], 8, 1, NULL, 8)       = 0
poll(Some(1ns)) -> 0 events after 1
epoll_pwait(3, [], 8, 2, NULL, 8)       = 0
poll(Some(1.5ms)) -> 0 events after 2
strace: Process 4 attached
[pid     3] epoll_pwait(3, [{events=EPOLLIN, data=0}], 8, 2147483647, NULL, 8) = 1
poll(Some(30 days)) -> 1 events after 110
[pid     4] +++ exited with 0 +++
strace: Process 5 attached
[pid     3] epoll_pwait(3, [{events=EPOLLIN, data=0}], 8, -1, NULL, 8) = 1
poll(None) -> 1 events after 110
[pid     5] +++ exited with 0 +++
+++ exited with 0 +++

macro_rules! syscall {
    ($fn: ident ( $($arg: expr),* $(,)* ) ) => {{
        #[allow(unused_unsafe)]
        let res = unsafe { libc::$fn($($arg, )*) };
        if res < 0 {
            Err(std::io::Error::last_os_error())
        } else {
            Ok(res)
        }
    }};
}

        match self.poll.poll(events, max_wait) {
            Ok(()) => {}
            Err(ref e) if e.kind() == io::ErrorKind::Interrupted => {}

error[E0502]: cannot borrow `poll` as mutable because it is also borrowed as immutable
  --> src/bin/borrow.rs:10:17
   |
 9 |   let registry = poll.registry(); // &Registry borrowed from poll
   |                  ---- immutable borrow occurs here
10 |   thread::scope(|s| {
   |                 ^^^ mutable borrow occurs here
...
13 |       registry.register(&mut l, Token(1), Interest::READABLE).unwrap();
   |       -------- immutable borrow later captured here by closure
14 |     });
15 |     poll.poll(&mut events, None).unwrap();
   |     ---- second borrow occurs due to use of `poll` in closure

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

    pub fn try_clone(&self) -> io::Result<Registry> {
        self.selector.try_clone().map(|selector| Registry {
            selector,
            #[cfg(all(debug_assertions, not(any(target_os = "wasi", target_os = "horizon"))))]
            has_waker: Arc::clone(&self.has_waker),
        })
    }

    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,
        })
    }

use std::os::fd::AsRawFd;
use std::thread;
use std::time::{Duration, Instant};

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

fn main() -> std::io::Result<()> {
  let mut poll = Poll::new()?;
  let registry = poll.registry().try_clone()?;
  println!("poll fd = {}, cloned registry fd = {}", poll.as_raw_fd(), registry.as_raw_fd());

  let start = Instant::now();
  let helper = thread::spawn(move || -> std::io::Result<()> {
    thread::sleep(Duration::from_millis(200));
    let mut listener = TcpListener::bind("127.0.0.1:0".parse().unwrap())?;
    // register while the main thread is blocked in poll(None)
    registry.register(&mut listener, Token(7), Interest::READABLE)?;
    println!("[{:>3}ms] helper: registered listener fd {} as Token(7)", start.elapsed().as_millis(), listener.as_raw_fd());
    thread::sleep(Duration::from_millis(200));
    let _client = std::net::TcpStream::connect(listener.local_addr()?)?;
    println!("[{:>3}ms] helper: connected", start.elapsed().as_millis());
    thread::sleep(Duration::from_millis(200));
    Ok(())
  });

  let mut events = Events::with_capacity(8);
  poll.poll(&mut events, None)?;
  for e in &events {
    println!("[{:>3}ms] main: {:?} readable={}", start.elapsed().as_millis(), e.token(), e.is_readable());
  }
  helper.join().unwrap()
}

poll fd = 3, cloned registry fd = 4
[210ms] helper: registered listener fd 5 as Token(7)
[418ms] helper: connected
[418ms] main: Token(7) readable=true

epoll_create1(EPOLL_CLOEXEC)            = 3
fcntl(3, F_DUPFD_CLOEXEC, 3)            = 4
fcntl(4, F_SETFD, FD_CLOEXEC)           = 0
poll fd = 3, cloned registry fd = 4
strace: Process 4 attached
[pid     3] epoll_pwait(3 <unfinished ...>
[pid     4] epoll_ctl(4, EPOLL_CTL_ADD, 5, {events=EPOLLIN|EPOLLRDHUP|EPOLLET, data=[209ms] helper: registered listener fd 5 as Token(7)
0x7}) = 0
[pid     4] connect(6, {sa_family=AF_INET, sin_port=htons(43701), sin_addr=inet_addr("127.0.0.1")}, 16 <unfinished ...>
[pid     3] <... epoll_pwait resumed>, [{events=EPOLLIN, data=0x7}], 8, -1, NULL, 8) = 1
[pid     4] <... connect resumed>)      = 0
[414ms] main: Token(7) readable=true
[414ms] helper: connected
[pid     4] +++ exited with 0 +++
+++ exited with 0 +++

    pub fn register<S>(&self, source: &mut S, token: Token, interests: Interest) -> io::Result<()>
    where
        S: event::Source + ?Sized,
    {
        trace!(
            "registering event source with poller: token={:?}, interests={:?}",
            token,
            interests
        );
        source.register(self, token, interests)
    }

use std::io::{self, Write};
use std::os::fd::AsRawFd;
use std::time::Duration;

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

const R: Interest = Interest::READABLE;

fn show(name: &str, r: io::Result<()>) {
  match r {
    Ok(()) => println!("  {name:<32} Ok(())"),
    Err(e) => println!("  {name:<32} Err({e:?})"),
  }
}

fn listener() -> TcpListener {
  TcpListener::bind("127.0.0.1:0".parse().unwrap()).unwrap()
}

fn tokens(poll: &mut Poll) -> Vec<usize> {
  let mut events = Events::with_capacity(8);
  poll.poll(&mut events, Some(Duration::from_millis(100))).unwrap();
  events.iter().map(|e| e.token().0).collect()
}

fn contract_cases(label: &str, mk: &dyn Fn() -> Box<dyn FnMut(&mio::Registry, u8) -> io::Result<()>>) {
  // op: 0 = register(Token(1)), 1 = reregister(Token(1)), 2 = deregister
  println!("{label}:");
  let run = |ops: &[u8]| -> io::Result<()> {
    let poll = Poll::new()?;
    let mut src = mk();
    let n = ops.len();
    for &op in &ops[..n - 1] {
      src(poll.registry(), op)?;
    }
    src(poll.registry(), ops[n - 1])
  };
  show("register, register", run(&[0, 0]));
  show("reregister (never registered)", run(&[1]));
  show("register, deregister, deregister", run(&[0, 2, 2]));
  show("deregister (never registered)", run(&[2]));
  let two = || -> io::Result<()> {
    let (a, b) = (Poll::new()?, Poll::new()?);
    let mut src = mk();
    src(a.registry(), 0)?;
    src(b.registry(), 0)
  };
  show("register on Poll A, then Poll B", two());
}

fn op<S: mio::event::Source + ?Sized>(src: &mut S, reg: &mio::Registry, op: u8) -> io::Result<()> {
  match op {
    0 => reg.register(src, Token(1), R),
    1 => reg.reregister(src, Token(1), R),
    _ => reg.deregister(src),
  }
}

fn main() -> io::Result<()> {
  println!("os={} debug_assertions={}", std::env::consts::OS, cfg!(debug_assertions));
  contract_cases("IoSource (mio::net::TcpListener)", &|| {
    let mut l = listener();
    Box::new(move |reg, o| op(&mut l, reg, o))
  });
  contract_cases("SourceFd (raw fd of a std TcpListener)", &|| {
    let l = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
    Box::new(move |reg, o| op(&mut SourceFd(&l.as_raw_fd()), reg, o))
  });

  println!("kernel view, via SourceFd (no debug check):");
  for (name, second) in [("register T1, register T2", 0u8), ("reregister T1 only", 1)] {
    let poll_l = std::net::TcpListener::bind("127.0.0.1:0")?;
    let mut poll = Poll::new()?;
    let fd = poll_l.as_raw_fd();
    let r = if second == 0 {
      poll.registry().register(&mut SourceFd(&fd), Token(1), R)?;
      poll.registry().register(&mut SourceFd(&fd), Token(2), R)
    } else {
      poll.registry().reregister(&mut SourceFd(&fd), Token(1), R)
    };
    let _c = std::net::TcpStream::connect(poll_l.local_addr()?)?;
    println!("  {name:<32} {:?}, then connect -> tokens {:?}", r.map_err(|e| e.raw_os_error()), tokens(&mut poll));
  }

  println!("drop without deregister:");
  let l = std::net::TcpListener::bind("127.0.0.1:0")?;
  let accept = || -> io::Result<TcpStream> {
    let (s, _) = l.accept()?;
    s.set_nonblocking(true)?;
    Ok(TcpStream::from_std(s))
  };
  // 1. plain drop: the fd is closed
  let mut poll = Poll::new()?;
  let mut c = std::net::TcpStream::connect(l.local_addr()?)?;
  let mut s = accept()?;
  poll.registry().register(&mut s, Token(1), R)?;
  drop(s);
  let _ = c.write_all(b"x");
  println!("  1. dropped, peer writes            -> tokens {:?}", tokens(&mut poll));

  // 2. drop while a dup of the fd is still open
  let mut poll = Poll::new()?;
  let mut c = std::net::TcpStream::connect(l.local_addr()?)?;
  let std_s = std::net::TcpStream::from(accept()?);
  let dup = std_s.try_clone()?; // same open file description, another fd
  let mut s = TcpStream::from_std(std_s);
  let old_fd = s.as_raw_fd();
  poll.registry().register(&mut s, Token(1), R)?;
  let mut c2 = std::net::TcpStream::connect(l.local_addr()?)?; // take its fd now
  drop(s);
  c.write_all(b"x")?;
  println!("  2. dropped fd {old_fd}, dup fd {} alive, peer writes -> tokens {:?}", dup.as_raw_fd(), tokens(&mut poll));
  show("   deregister via the dup fd", poll.registry().deregister(&mut SourceFd(&dup.as_raw_fd())));

  // 3. a new socket reuses the old fd number
  let mut s2 = accept()?;
  println!("  3. new socket got fd {} (old fd was {old_fd})", s2.as_raw_fd());
  show("   register it as Token(2)", poll.registry().register(&mut s2, Token(2), R));
  c2.write_all(b"y")?;
  c.write_all(b"x")?;
  println!("     both peers write              -> tokens {:?}", tokens(&mut poll));

  // 4. close the dup as well (drain it first so close sends FIN, not RST)
  let mut buf = [0u8; 16];
  let n = io::Read::read(&mut &dup, &mut buf)?;
  drop(dup);
  c.write_all(b"x")?;
  println!("  4. read {n} bytes, closed the dup, peer writes -> tokens {:?}", tokens(&mut poll));
  Ok(())
}

os=linux debug_assertions=true
IoSource (mio::net::TcpListener):
  register, register               Err(Custom { kind: AlreadyExists, error: "I/O source already registered with a `Registry`" })
  reregister (never registered)    Err(Custom { kind: NotFound, error: "I/O source not registered with `Registry`" })
  register, deregister, deregister Err(Custom { kind: NotFound, error: "I/O source not registered with `Registry`" })
  deregister (never registered)    Err(Custom { kind: NotFound, error: "I/O source not registered with `Registry`" })
  register on Poll A, then Poll B  Err(Custom { kind: AlreadyExists, error: "I/O source already registered with a `Registry`" })
SourceFd (raw fd of a std TcpListener):
  register, register               Err(Os { code: 17, kind: AlreadyExists, message: "File exists" })
  reregister (never registered)    Err(Os { code: 2, kind: NotFound, message: "No such file or directory" })
  register, deregister, deregister Err(Os { code: 2, kind: NotFound, message: "No such file or directory" })
  deregister (never registered)    Err(Os { code: 2, kind: NotFound, message: "No such file or directory" })
  register on Poll A, then Poll B  Ok(())
kernel view, via SourceFd (no debug check):
  register T1, register T2         Err(Some(17)), then connect -> tokens [1]
  reregister T1 only               Err(Some(2)), then connect -> tokens []
drop without deregister:
  1. dropped, peer writes            -> tokens []
  2. dropped fd 8, dup fd 9 alive, peer writes -> tokens [1]
     deregister via the dup fd     Err(Os { code: 2, kind: NotFound, message: "No such file or directory" })
  3. new socket got fd 8 (old fd was 8)
     register it as Token(2)       Ok(())
     both peers write              -> tokens [2, 1]
  4. read 2 bytes, closed the dup, peer writes -> tokens []

os=linux debug_assertions=false
IoSource (mio::net::TcpListener):
  register, register               Err(Os { code: 17, kind: AlreadyExists, message: "File exists" })
  reregister (never registered)    Err(Os { code: 2, kind: NotFound, message: "No such file or directory" })
  register, deregister, deregister Err(Os { code: 2, kind: NotFound, message: "No such file or directory" })
  deregister (never registered)    Err(Os { code: 2, kind: NotFound, message: "No such file or directory" })
  register on Poll A, then Poll B  Ok(())

os=macos debug_assertions=false
IoSource (mio::net::TcpListener):
  register, register               Ok(())
  reregister (never registered)    Ok(())
  register, deregister, deregister Ok(())
  deregister (never registered)    Ok(())
  register on Poll A, then Poll B  Ok(())
SourceFd (raw fd of a std TcpListener):
  register, register               Ok(())
  reregister (never registered)    Ok(())
  register, deregister, deregister Ok(())
  deregister (never registered)    Ok(())
  register on Poll A, then Poll B  Ok(())
kernel view, via SourceFd (no debug check):
  register T1, register T2         Ok(()), then connect -> tokens [2]
  reregister T1 only               Ok(()), then connect -> tokens [1]
drop without deregister:
  1. dropped, peer writes            -> tokens []
  2. dropped fd 8, dup fd 9 alive, peer writes -> tokens []
     deregister via the dup fd     Ok(())
  3. new socket got fd 8 (old fd was 8)
     register it as Token(2)       Ok(())
     both peers write              -> tokens [2]
  4. read 2 bytes, closed the dup, peer writes -> tokens []

    pub(crate) fn register_waker(&self) {
        assert!(
            !self.has_waker.swap(true, Ordering::AcqRel),
            "Only a single `Waker` can be active per `Poll` instance"
        );
    }

macos no SA_RESTART : Err kind=Interrupted raw=Some(4) "Interrupted system call (os error 4)" after 210ms
macos SA_RESTART    : Err kind=Interrupted raw=Some(4) "Interrupted system call (os error 4)" after 203ms
linux no SA_RESTART : Err kind=Interrupted raw=Some(4) "Interrupted system call (os error 4)" after 210ms
linux SA_RESTART    : Err kind=Interrupted raw=Some(4) "Interrupted system call (os error 4)" after 202ms

[pid     4] epoll_pwait(3, 0xffff98bbc630, 8, 2000, NULL, 8) = -1 EINTR (Interrupted system call)
[pid     4] --- SIGUSR1 {si_signo=SIGUSR1, si_code=SI_TKILL, si_pid=3, si_uid=0} ---