Port from mio to polling

This patch replaces the mio crate with the polling. Now that
smol-rs/polling#96 has been merged, we should be at full feature parity
with mio v0.6 now.

Fixes #7104.
Fixes #6486.

1 files changed, 172 insertions, 151 deletions

diff --git a/alacritty_terminal/src/event_loop.rs b/alacritty_terminal/src/event_loop.rs
index 61dc69bc..de22e49d 100644
--- a/alacritty_terminal/src/event_loop.rs
+++ b/alacritty_terminal/src/event_loop.rs
@@ -5,15 +5,14 @@ use std::collections::VecDeque;
 use std::fs::File;
 use std::io::{self, ErrorKind, Read, Write};
 use std::marker::Send;
+use std::num::NonZeroUsize;
+use std::sync::mpsc::{self, Receiver, Sender, TryRecvError};
 use std::sync::Arc;
 use std::thread::JoinHandle;
 use std::time::Instant;
 
 use log::error;
-#[cfg(not(windows))]
-use mio::unix::UnixReady;
-use mio::{self, Events, PollOpt, Ready};
-use mio_extras::channel::{self, Receiver, Sender};
+use polling::{Event as PollingEvent, Events, PollMode};
 
 use crate::event::{self, Event, EventListener, WindowSize};
 use crate::sync::FairMutex;
@@ -21,7 +20,7 @@ use crate::term::Term;
 use crate::{ansi, thread, tty};
 
 /// Max bytes to read from the PTY before forced terminal synchronization.
-const READ_BUFFER_SIZE: usize = 0x10_0000;
+pub(crate) const READ_BUFFER_SIZE: usize = 0x10_0000;
 
 /// Max bytes to read from the PTY while the terminal is locked.
 const MAX_LOCKED_READ: usize = u16::MAX as usize;
@@ -39,14 +38,14 @@ pub enum Msg {
     Resize(WindowSize),
 }
 
-/// The main event!.. loop.
+/// The main event loop.
 ///
 /// Handles all the PTY I/O and runs the PTY parser which updates terminal
 /// state.
 pub struct EventLoop<T: tty::EventedPty, U: EventListener> {
-    poll: mio::Poll,
+    poll: Arc<polling::Poller>,
     pty: T,
-    rx: Receiver<Msg>,
+    rx: PeekableReceiver<Msg>,
     tx: Sender<Msg>,
     terminal: Arc<FairMutex<Term<U>>>,
     event_proxy: U,
@@ -54,97 +53,6 @@ pub struct EventLoop<T: tty::EventedPty, U: EventListener> {
     ref_test: bool,
 }
 
-/// Helper type which tracks how much of a buffer has been written.
-struct Writing {
-    source: Cow<'static, [u8]>,
-    written: usize,
-}
-
-pub struct Notifier(pub Sender<Msg>);
-
-impl event::Notify for Notifier {
-    fn notify<B>(&self, bytes: B)
-    where
-        B: Into<Cow<'static, [u8]>>,
-    {
-        let bytes = bytes.into();
-        // terminal hangs if we send 0 bytes through.
-        if bytes.len() == 0 {
-            return;
-        }
-
-        let _ = self.0.send(Msg::Input(bytes));
-    }
-}
-
-impl event::OnResize for Notifier {
-    fn on_resize(&mut self, window_size: WindowSize) {
-        let _ = self.0.send(Msg::Resize(window_size));
-    }
-}
-
-/// All of the mutable state needed to run the event loop.
-///
-/// Contains list of items to write, current write state, etc. Anything that
-/// would otherwise be mutated on the `EventLoop` goes here.
-#[derive(Default)]
-pub struct State {
-    write_list: VecDeque<Cow<'static, [u8]>>,
-    writing: Option<Writing>,
-    parser: ansi::Processor,
-}
-
-impl State {
-    #[inline]
-    fn ensure_next(&mut self) {
-        if self.writing.is_none() {
-            self.goto_next();
-        }
-    }
-
-    #[inline]
-    fn goto_next(&mut self) {
-        self.writing = self.write_list.pop_front().map(Writing::new);
-    }
-
-    #[inline]
-    fn take_current(&mut self) -> Option<Writing> {
-        self.writing.take()
-    }
-
-    #[inline]
-    fn needs_write(&self) -> bool {
-        self.writing.is_some() || !self.write_list.is_empty()
-    }
-
-    #[inline]
-    fn set_current(&mut self, new: Option<Writing>) {
-        self.writing = new;
-    }
-}
-
-impl Writing {
-    #[inline]
-    fn new(c: Cow<'static, [u8]>) -> Writing {
-        Writing { source: c, written: 0 }
-    }
-
-    #[inline]
-    fn advance(&mut self, n: usize) {
-        self.written += n;
-    }
-
-    #[inline]
-    fn remaining_bytes(&self) -> &[u8] {
-        &self.source[self.written..]
-    }
-
-    #[inline]
-    fn finished(&self) -> bool {
-        self.written >= self.source.len()
-    }
-}
-
 impl<T, U> EventLoop<T, U>
 where
     T: tty::EventedPty + event::OnResize + Send + 'static,
@@ -158,12 +66,12 @@ where
         hold: bool,
         ref_test: bool,
     ) -> EventLoop<T, U> {
-        let (tx, rx) = channel::channel();
+        let (tx, rx) = mpsc::channel();
         EventLoop {
-            poll: mio::Poll::new().expect("create mio Poll"),
+            poll: polling::Poller::new().expect("create Poll").into(),
             pty,
             tx,
-            rx,
+            rx: PeekableReceiver::new(rx),
             terminal,
             event_proxy,
             hold,
@@ -171,15 +79,15 @@ where
         }
     }
 
-    pub fn channel(&self) -> Sender<Msg> {
-        self.tx.clone()
+    pub fn channel(&self) -> EventLoopSender {
+        EventLoopSender { sender: self.tx.clone(), poller: self.poll.clone() }
     }
 
     /// Drain the channel.
     ///
     /// Returns `false` when a shutdown message was received.
     fn drain_recv_channel(&mut self, state: &mut State) -> bool {
-        while let Ok(msg) = self.rx.try_recv() {
+        while let Some(msg) = self.rx.recv() {
             match msg {
                 Msg::Input(input) => state.write_list.push_back(input),
                 Msg::Resize(window_size) => self.pty.on_resize(window_size),
@@ -190,20 +98,6 @@ where
         true
     }
 
-    /// Returns a `bool` indicating whether or not the event loop should continue running.
-    #[inline]
-    fn channel_event(&mut self, token: mio::Token, state: &mut State) -> bool {
-        if !self.drain_recv_channel(state) {
-            return false;
-        }
-
-        self.poll
-            .reregister(&self.rx, token, Ready::readable(), PollOpt::edge() | PollOpt::oneshot())
-            .unwrap();
-
-        true
-    }
-
     #[inline]
     fn pty_read<X>(
         &mut self,
@@ -313,17 +207,15 @@ where
             let mut state = State::default();
             let mut buf = [0u8; READ_BUFFER_SIZE];
 
-            let mut tokens = (0..).map(Into::into);
-
-            let poll_opts = PollOpt::edge() | PollOpt::oneshot();
-
-            let channel_token = tokens.next().unwrap();
-            self.poll.register(&self.rx, channel_token, Ready::readable(), poll_opts).unwrap();
+            let poll_opts = PollMode::Level;
+            let mut interest = PollingEvent::readable(0);
 
             // Register TTY through EventedRW interface.
-            self.pty.register(&self.poll, &mut tokens, Ready::readable(), poll_opts).unwrap();
+            unsafe {
+                self.pty.register(&self.poll, interest, poll_opts).unwrap();
+            }
 
-            let mut events = Events::with_capacity(1024);
+            let mut events = Events::with_capacity(NonZeroUsize::new(1024).unwrap());
 
             let mut pipe = if self.ref_test {
                 Some(File::create("./alacritty.recording").expect("create alacritty recording"))
@@ -337,7 +229,8 @@ where
                 let timeout =
                     handler.sync_timeout().map(|st| st.saturating_duration_since(Instant::now()));
 
-                if let Err(err) = self.poll.poll(&mut events, timeout) {
+                events.clear();
+                if let Err(err) = self.poll.wait(&mut events, timeout) {
                     match err.kind() {
                         ErrorKind::Interrupted => continue,
                         _ => panic!("EventLoop polling error: {err:?}"),
@@ -345,21 +238,20 @@ where
                 }
 
                 // Handle synchronized update timeout.
-                if events.is_empty() {
+                if events.is_empty() && self.rx.peek().is_none() {
                     state.parser.stop_sync(&mut *self.terminal.lock());
                     self.event_proxy.send_event(Event::Wakeup);
                     continue;
                 }
 
-                for event in events.iter() {
-                    match event.token() {
-                        token if token == channel_token => {
-                            if !self.channel_event(channel_token, &mut state) {
-                                break 'event_loop;
-                            }
-                        },
+                // Handle channel events, if there are any.
+                if !self.drain_recv_channel(&mut state) {
+                    break;
+                }
 
-                        token if token == self.pty.child_event_token() => {
+                for event in events.iter() {
+                    match event.key {
+                        tty::PTY_CHILD_EVENT_TOKEN => {
                             if let Some(tty::ChildEvent::Exited) = self.pty.next_child_event() {
                                 if self.hold {
                                     // With hold enabled, make sure the PTY is drained.
@@ -374,17 +266,13 @@ where
                             }
                         },
 
-                        token
-                            if token == self.pty.read_token()
-                                || token == self.pty.write_token() =>
-                        {
-                            #[cfg(unix)]
-                            if UnixReady::from(event.readiness()).is_hup() {
+                        tty::PTY_READ_WRITE_TOKEN => {
+                            if event.is_interrupt() {
                                 // Don't try to do I/O on a dead PTY.
                                 continue;
                             }
 
-                            if event.readiness().is_readable() {
+                            if event.readable {
                                 if let Err(err) = self.pty_read(&mut state, &mut buf, pipe.as_mut())
                                 {
                                     // On Linux, a `read` on the master side of a PTY can fail
@@ -402,7 +290,7 @@ where
                                 }
                             }
 
-                            if event.readiness().is_writable() {
+                            if event.writable {
                                 if let Err(err) = self.pty_write(&mut state) {
                                     error!("Error writing to PTY in event loop: {}", err);
                                     break 'event_loop;
@@ -414,19 +302,152 @@ where
                 }
 
                 // Register write interest if necessary.
-                let mut interest = Ready::readable();
-                if state.needs_write() {
-                    interest.insert(Ready::writable());
+                let needs_write = state.needs_write();
+                if needs_write != interest.writable {
+                    interest.writable = needs_write;
+
+                    // Re-register with new interest.
+                    self.pty.reregister(&self.poll, interest, poll_opts).unwrap();
                 }
-                // Reregister with new interest.
-                self.pty.reregister(&self.poll, interest, poll_opts).unwrap();
             }
 
             // The evented instances are not dropped here so deregister them explicitly.
-            let _ = self.poll.deregister(&self.rx);
             let _ = self.pty.deregister(&self.poll);
 
             (self, state)
         })
     }
 }
+
+/// Helper type which tracks how much of a buffer has been written.
+struct Writing {
+    source: Cow<'static, [u8]>,
+    written: usize,
+}
+
+pub struct Notifier(pub EventLoopSender);
+
+impl event::Notify for Notifier {
+    fn notify<B>(&self, bytes: B)
+    where
+        B: Into<Cow<'static, [u8]>>,
+    {
+        let bytes = bytes.into();
+        // Terminal hangs if we send 0 bytes through.
+        if bytes.len() == 0 {
+            return;
+        }
+
+        self.0.send(Msg::Input(bytes));
+    }
+}
+
+impl event::OnResize for Notifier {
+    fn on_resize(&mut self, window_size: WindowSize) {
+        self.0.send(Msg::Resize(window_size));
+    }
+}
+
+pub struct EventLoopSender {
+    sender: Sender<Msg>,
+    poller: Arc<polling::Poller>,
+}
+
+impl EventLoopSender {
+    pub fn send(&self, msg: Msg) {
+        let _ = self.sender.send(msg);
+        let _ = self.poller.notify();
+    }
+}
+
+/// All of the mutable state needed to run the event loop.
+///
+/// Contains list of items to write, current write state, etc. Anything that
+/// would otherwise be mutated on the `EventLoop` goes here.
+#[derive(Default)]
+pub struct State {
+    write_list: VecDeque<Cow<'static, [u8]>>,
+    writing: Option<Writing>,
+    parser: ansi::Processor,
+}
+
+impl State {
+    #[inline]
+    fn ensure_next(&mut self) {
+        if self.writing.is_none() {
+            self.goto_next();
+        }
+    }
+
+    #[inline]
+    fn goto_next(&mut self) {
+        self.writing = self.write_list.pop_front().map(Writing::new);
+    }
+
+    #[inline]
+    fn take_current(&mut self) -> Option<Writing> {
+        self.writing.take()
+    }
+
+    #[inline]
+    fn needs_write(&self) -> bool {
+        self.writing.is_some() || !self.write_list.is_empty()
+    }
+
+    #[inline]
+    fn set_current(&mut self, new: Option<Writing>) {
+        self.writing = new;
+    }
+}
+
+impl Writing {
+    #[inline]
+    fn new(c: Cow<'static, [u8]>) -> Writing {
+        Writing { source: c, written: 0 }
+    }
+
+    #[inline]
+    fn advance(&mut self, n: usize) {
+        self.written += n;
+    }
+
+    #[inline]
+    fn remaining_bytes(&self) -> &[u8] {
+        &self.source[self.written..]
+    }
+
+    #[inline]
+    fn finished(&self) -> bool {
+        self.written >= self.source.len()
+    }
+}
+
+struct PeekableReceiver<T> {
+    rx: Receiver<T>,
+    peeked: Option<T>,
+}
+
+impl<T> PeekableReceiver<T> {
+    fn new(rx: Receiver<T>) -> Self {
+        Self { rx, peeked: None }
+    }
+
+    fn peek(&mut self) -> Option<&T> {
+        if self.peeked.is_none() {
+            self.peeked = self.rx.try_recv().ok();
+        }
+
+        self.peeked.as_ref()
+    }
+
+    fn recv(&mut self) -> Option<T> {
+        if self.peeked.is_some() {
+            self.peeked.take()
+        } else {
+            match self.rx.try_recv() {
+                Err(TryRecvError::Disconnected) => panic!("event loop channel closed"),
+                res => res.ok(),
+            }
+        }
+    }
+}