Clean up event_loop module

The main loop body was originally all written inline. There's now
separate functions for each of the actions the loop handles including
channel events, pty reading, and pty writing. There's also helper
functions on State for managing the write list.

The `EventLoop` and its `State` are returned when joining with the
thread it spawns. This will potentially be helpful once config reloading
is introduced.

src/event_loop.rs

@@ -1,5 +1,6 @@
 //! The main event loop which performs I/O on the pseudoterminal
 use std::borrow::Cow;
+use std::collections::VecDeque;
 use std::io::{self, ErrorKind};
 use std::os::unix::io::AsRawFd;
 use std::sync::Arc;
@@ -14,6 +15,17 @@ use sync::FairMutex;
 
 use super::Flag;
 
+/// Messages that may be sent to the `EventLoop`
+#[derive(Debug)]
+pub enum Msg {
+    /// Data that should be written to the pty
+    Input(Cow<'static, [u8]>),
+}
+
+/// The main event!.. loop.
+///
+/// Handles all the pty I/O and runs the pty parser which updates terminal
+/// state.
 pub struct EventLoop<Io> {
     poll: mio::Poll,
     pty: Io,
@@ -24,47 +36,63 @@ pub struct EventLoop<Io> {
     signal_flag: Flag,
 }
 
-
+/// Helper type which tracks how much of a buffer has been written.
-#[derive(Debug)]
-pub enum Msg {
-    Input(Cow<'static, [u8]>),
-}
-
-const CHANNEL: mio::Token = mio::Token(0);
-const PTY: mio::Token = mio::Token(1);
-
-impl<Io> EventLoop<Io>
-    where Io: io::Read + io::Write + Send + AsRawFd + 'static
-{
-    pub fn new(
-        terminal: Arc<FairMutex<Term>>,
-        proxy: ::glutin::WindowProxy,
-        signal_flag: Flag,
-        pty: Io,
-    ) -> EventLoop<Io> {
-        let (tx, rx) = ::mio::channel::channel();
-        EventLoop {
-            poll: mio::Poll::new().expect("create mio Poll"),
-            pty: pty,
-            tx: tx,
-            rx: rx,
-            terminal: terminal,
-            proxy: proxy,
-            signal_flag: signal_flag
-        }
-    }
-
-    pub fn channel(&self) -> mio::channel::Sender<Msg> {
-        self.tx.clone()
-    }
-
-    pub fn spawn(self) -> thread::JoinHandle<()> {
-
 struct Writing {
     source: Cow<'static, [u8]>,
     written: usize,
 }
 
+/// 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.
+pub struct State {
+    write_list: VecDeque<Cow<'static, [u8]>>,
+    writing: Option<Writing>,
+    parser: ansi::Processor,
+}
+
+impl Default for State {
+    fn default() -> State {
+        State {
+            write_list: VecDeque::new(),
+            parser: ansi::Processor::new(),
+            writing: None,
+        }
+    }
+}
+
+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(|c| Writing::new(c));
+    }
+
+    #[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 {
@@ -87,76 +115,84 @@ impl<Io> EventLoop<Io>
     }
 }
 
-        thread::spawn_named("pty reader", move || {
+/// mio::Token for the event loop channel
+const CHANNEL: mio::Token = mio::Token(0);
 
-            let EventLoop { poll, mut pty, rx, terminal, proxy, signal_flag, .. } = self;
+/// mio::Token for the pty file descriptor
+const PTY: mio::Token = mio::Token(1);
 
+impl<Io> EventLoop<Io>
+    where Io: io::Read + io::Write + Send + AsRawFd + 'static
+{
+    /// Create a new event loop
+    pub fn new(
+        terminal: Arc<FairMutex<Term>>,
+        proxy: ::glutin::WindowProxy,
+        signal_flag: Flag,
+        pty: Io,
+    ) -> EventLoop<Io> {
+        let (tx, rx) = ::mio::channel::channel();
+        EventLoop {
+            poll: mio::Poll::new().expect("create mio Poll"),
+            pty: pty,
+            tx: tx,
+            rx: rx,
+            terminal: terminal,
+            proxy: proxy,
+            signal_flag: signal_flag
+        }
+    }
 
-            let mut buf = [0u8; 4096];
+    pub fn channel(&self) -> mio::channel::Sender<Msg> {
-            let mut pty_parser = ansi::Processor::new();
+        self.tx.clone()
-            let fd = pty.as_raw_fd();
+    }
-            let fd = EventedFd(&fd);
 
-            poll.register(&rx, CHANNEL, Ready::readable(), PollOpt::edge() | PollOpt::oneshot())
+    #[inline]
-                .unwrap();
+    fn channel_event(&mut self, state: &mut State) {
-            poll.register(&fd, PTY, Ready::readable(), PollOpt::edge() | PollOpt::oneshot())
+        while let Ok(msg) = self.rx.try_recv() {
-                .unwrap();
-
-            let mut events = Events::with_capacity(1024);
-            let mut write_list = ::std::collections::VecDeque::new();
-            let mut writing = None;
-
-            'event_loop: loop {
-                poll.poll(&mut events, None).expect("poll ok");
-
-                for event in events.iter() {
-                    match event.token() {
-                        CHANNEL => {
-                            while let Ok(msg) = rx.try_recv() {
             match msg {
                 Msg::Input(input) => {
-                                        write_list.push_back(input);
+                    state.write_list.push_back(input);
                 }
             }
         }
 
-                            poll.reregister(
+        self.poll.reregister(
-                                &rx, CHANNEL,
+            &self.rx, CHANNEL,
             Ready::readable(),
             PollOpt::edge() | PollOpt::oneshot()
         ).expect("reregister channel");
 
-                            if writing.is_some() || !write_list.is_empty() {
+        if state.needs_write() {
-                                poll.reregister(
+            self.poll.reregister(
-                                    &fd,
+                &EventedFd(&self.pty.as_raw_fd()),
                 PTY,
                 Ready::readable() | Ready::writable(),
                 PollOpt::edge() | PollOpt::oneshot()
             ).expect("reregister fd after channel recv");
         }
-                        },
+    }
-                        PTY => {
-                            let kind = event.kind();
 
-                            if kind.is_readable() {
+    #[inline]
+    fn pty_read(&mut self, state: &mut State, buf: &mut [u8]) {
         loop {
-                                    match pty.read(&mut buf[..]) {
+            match self.pty.read(&mut buf[..]) {
                 Ok(0) => break,
                 Ok(got) => {
-                                            let mut terminal = terminal.lock();
+                    let mut terminal = self.terminal.lock();
                     for byte in &buf[..got] {
-                                                pty_parser.advance(&mut *terminal, *byte);
+                        state.parser.advance(&mut *terminal, *byte);
                     }
 
                     terminal.dirty = true;
 
                     // Only wake up the event loop if it hasn't already been
-                                            // signaled. This is a really important optimization
+                    // signaled. This is a really important optimization because
-                                            // because waking up the event loop redundantly burns *a
+                    // waking up the event loop redundantly burns *a lot* of
-                                            // lot* of cycles.
+                    // cycles.
-                                            if !signal_flag.get() {
+                    if !self.signal_flag.get() {
-                                                proxy.wakeup_event_loop();
+                        self.proxy.wakeup_event_loop();
-                                                signal_flag.set(true);
+                        self.signal_flag.set(true);
                     }
                 },
                 Err(err) => {
@@ -169,35 +205,28 @@ impl<Io> EventLoop<Io>
         }
     }
 
-                            if kind.is_writable() {
+    #[inline]
-                                if writing.is_none() {
+    fn pty_write(&mut self, state: &mut State) {
-                                    writing = write_list
+        state.ensure_next();
-                                        .pop_front()
-                                        .map(|c| Writing::new(c));
-                                }
 
-                                'write_list_loop: while let Some(mut write_now) = writing.take() {
+        'write_many: while let Some(mut current) = state.take_current() {
-                                    loop {
+            'write_one: loop {
-                                        match pty.write(write_now.remaining_bytes()) {
+                match self.pty.write(current.remaining_bytes()) {
                     Ok(0) => {
-                                                writing = Some(write_now);
+                        state.set_current(Some(current));
-                                                break 'write_list_loop;
+                        break 'write_many;
                     },
                     Ok(n) => {
-                                                write_now.advance(n);
+                        current.advance(n);
-                                                if write_now.finished() {
+                        if current.finished() {
-                                                    writing = write_list
+                            state.goto_next();
-                                                        .pop_front()
+                            break 'write_one;
-                                                        .map(|next| Writing::new(next));
-
-                                                    break;
-                                                } else {
                         }
                     },
                     Err(err) => {
-                                                writing = Some(write_now);
+                        state.set_current(Some(current));
                         match err.kind() {
-                                                    ErrorKind::WouldBlock => break 'write_list_loop,
+                            ErrorKind::WouldBlock => break 'write_many,
                             // TODO
                             _ => panic!("unexpected err: {:?}", err),
                         }
@@ -208,16 +237,51 @@ impl<Io> EventLoop<Io>
         }
     }
 
+    pub fn spawn(mut self, state: Option<State>) -> thread::JoinHandle<(EventLoop<Io>, State)> {
+        thread::spawn_named("pty reader", move || {
+            let mut state = state.unwrap_or_else(Default::default);
+            let mut buf = [0u8; 4096];
+
+            let fd = self.pty.as_raw_fd();
+            let fd = EventedFd(&fd);
+
+            let poll_opts = PollOpt::edge() | PollOpt::oneshot();
+
+            self.poll.register(&self.rx, CHANNEL, Ready::readable(), poll_opts).unwrap();
+            self.poll.register(&fd, PTY, Ready::readable(), poll_opts).unwrap();
+
+            let mut events = Events::with_capacity(1024);
+
+            'event_loop: loop {
+                self.poll.poll(&mut events, None).expect("poll ok");
+
+                for event in events.iter() {
+                    match event.token() {
+                        CHANNEL => self.channel_event(&mut state),
+                        PTY => {
+                            let kind = event.kind();
+
+                            if kind.is_readable() {
+                                self.pty_read(&mut state, &mut buf);
+                            }
+
+                            if kind.is_writable() {
+                                self.pty_write(&mut state);
+                            }
+
                             if kind.is_hup() {
                                 break 'event_loop;
                             }
 
+                            // Figure out pty interest
                             let mut interest = Ready::readable();
-                            if writing.is_some() || !write_list.is_empty() {
+                            if state.needs_write() {
                                 interest.insert(Ready::writable());
                             }
 
-                            poll.reregister(&fd, PTY, interest, PollOpt::edge() | PollOpt::oneshot())
+                            // Reregister pty
+                            self.poll
+                                .reregister(&fd, PTY, interest, poll_opts)
                                 .expect("register fd after read/write");
                         },
                         _ => (),
@@ -225,7 +289,10 @@ impl<Io> EventLoop<Io>
                 }
             }
 
-            println!("pty reader stopped");
+            self.poll.deregister(&self.rx).expect("deregister channel");
+            self.poll.deregister(&fd).expect("deregister pty");
+
+            (self, state)
         })
     }
 }

src/main.rs

@@ -196,7 +196,7 @@ fn main() {
     );
 
     let loop_tx = event_loop.channel();
-    let event_loop_handle = event_loop.spawn();
+    let event_loop_handle = event_loop.spawn(None);
 
     // Wraps a renderer and gives simple draw() api.
     let mut display = Display::new(