Improve pty reading and renderer synchronization

The pty read thread now runs the parser and directly updates the
terminal in the same thread. This obviates the need for a channel which
sends every char read from the pty; this is a huge performance boon.

Synchronization between the updater and the renderer is now achieved
with a PriorityMutex. Previously, an atomic bool was (poorly) used to
request the lock on terminal. The PriorityMutex is dead simple to use,
and it _Just Works_.

3 files changed, 160 insertions, 68 deletions

diff --git a/src/io.rs b/src/io.rs
index 688e72a4..5801efaf 100644
--- a/src/io.rs
+++ b/src/io.rs
@@ -69,7 +69,7 @@ pub enum Utf8CharsError {
     /// of a byte sequence well-formed in UTF-8, but ends prematurely.
     ///
     /// Contains number of unused bytes
-    IncompleteUtf8(u8),
+    IncompleteUtf8(usize),
 
     /// Variant representing that an I/O error occurred.
     Io(Error),
diff --git a/src/main.rs b/src/main.rs
index cbfa7425..7bf4d7d3 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -47,12 +47,12 @@ pub mod ansi;
 mod term;
 mod util;
 mod io;
+mod sync;
 
-use std::io::{Write, BufWriter, BufReader};
-use std::sync::atomic::{AtomicBool, Ordering};
+use std::io::{Write, BufWriter, Read};
 use std::sync::{mpsc, Arc};
 
-use parking_lot::Mutex;
+use sync::PriorityMutex;
 
 use config::Config;
 use font::FontDesc;
@@ -62,14 +62,7 @@ use term::Term;
 use tty::process_should_exit;
 use util::thread;
 
-use io::Utf8Chars;
-
-/// Things that the render/update thread needs to respond to
-#[derive(Debug)]
-enum Event {
-    PtyChar(char),
-    Glutin(glutin::Event),
-}
+use io::{Utf8Chars, Utf8CharsError};
 
 #[derive(Debug, Copy, Clone, Eq, PartialEq)]
 enum ShouldExit {
@@ -85,44 +78,39 @@ impl<'a, W: Write> input::Notify for WriteNotifier<'a, W> {
 }
 
 /// Channel used by resize handling on mac
-static mut resize_sender: Option<mpsc::Sender<Event>> = None;
+static mut resize_sender: Option<mpsc::Sender<glutin::Event>> = None;
 
 /// Resize handling for Mac
 fn window_resize_handler(width: u32, height: u32) {
     unsafe {
         if let Some(ref tx) = resize_sender {
-            let _ = tx.send(Event::Glutin(glutin::Event::Resized(width, height)));
+            let _ = tx.send(glutin::Event::Resized(width, height));
         }
     }
 }
 
-fn handle_event<W>(event: Event,
+fn handle_event<W>(event: glutin::Event,
                    writer: &mut W,
                    terminal: &mut Term,
-                   pty_parser: &mut ansi::Parser,
                    render_tx: &mpsc::Sender<(u32, u32)>,
                    input_processor: &mut input::Processor) -> ShouldExit
     where W: Write
 {
+    // Handle keyboard/mouse input and other window events
     match event {
-        // Handle char from pty
-        Event::PtyChar(c) => pty_parser.advance(terminal, c),
-        // Handle keyboard/mouse input and other window events
-        Event::Glutin(gevent) => match gevent {
-            glutin::Event::Closed => return ShouldExit::Yes,
-            glutin::Event::ReceivedCharacter(c) => {
-                let encoded = c.encode_utf8();
-                writer.write(encoded.as_slice()).unwrap();
-            },
-            glutin::Event::Resized(w, h) => {
-                terminal.resize(w as f32, h as f32);
-                render_tx.send((w, h)).expect("render thread active");
-            },
-            glutin::Event::KeyboardInput(state, _code, key) => {
-                input_processor.process(state, key, &mut WriteNotifier(writer), *terminal.mode())
-            },
-            _ => ()
-        }
+        glutin::Event::Closed => return ShouldExit::Yes,
+        glutin::Event::ReceivedCharacter(c) => {
+            let encoded = c.encode_utf8();
+            writer.write(encoded.as_slice()).unwrap();
+        },
+        glutin::Event::Resized(w, h) => {
+            terminal.resize(w as f32, h as f32);
+            render_tx.send((w, h)).expect("render thread active");
+        },
+        glutin::Event::KeyboardInput(state, _code, key) => {
+            input_processor.process(state, key, &mut WriteNotifier(writer), *terminal.mode())
+        },
+        _ => ()
     }
 
     ShouldExit::No
@@ -178,47 +166,68 @@ fn main() {
 
     let terminal = Term::new(width as f32, height as f32, cell_width as f32, cell_height as f32);
 
-    let reader = terminal.tty().reader();
+    let mut reader = terminal.tty().reader();
     let writer = terminal.tty().writer();
 
     let mut glyph_cache = GlyphCache::new(rasterizer, desc, font.size());
-    let needs_render = Arc::new(AtomicBool::new(true));
-    let needs_render2 = needs_render.clone();
 
     let (tx, rx) = mpsc::channel();
-    let reader_tx = tx.clone();
     unsafe {
         resize_sender = Some(tx.clone());
     }
+
+    let terminal = Arc::new(PriorityMutex::new(terminal));
+    let term_ref = terminal.clone();
+    let term_updater = terminal.clone();
+
+    // Rendering thread
+    //
+    // Holds lock only when updating terminal
     let reader_thread = thread::spawn_named("TTY Reader", move || {
-        let chars = Utf8Chars::new(BufReader::new(reader));
-        for c in chars {
-            let c = c.unwrap();
-            reader_tx.send(Event::PtyChar(c)).unwrap();
+        let mut buf = [0u8; 4096];
+        let mut start = 0;
+        let mut pty_parser = ansi::Parser::new();
+
+        'reader: loop {
+            let got = reader.read(&mut buf[start..]).expect("pty fd active");
+            let mut remain = 0;
+
+            // if `start` is nonzero, then actual bytes in buffer is > `got` by `start` bytes.
+            let end = start + got;
+            let mut terminal = term_updater.lock_low();
+            for c in Utf8Chars::new(&buf[..end]) {
+                match c {
+                    Ok(c) => pty_parser.advance(&mut *terminal, c),
+                    Err(err) => match err {
+                        Utf8CharsError::IncompleteUtf8(unused) => {
+                            remain = unused;
+                            break;
+                        },
+                        _ => panic!("{}", err),
+                    }
+                }
+            }
+
+            // Move any leftover bytes to front of buffer
+            for i in 0..remain {
+                buf[i] = buf[end - (remain - i)];
+            }
+            start = remain;
         }
     });
 
-    let terminal = Arc::new(Mutex::new(terminal));
-    let term_ref = terminal.clone();
     let mut meter = Meter::new();
 
-    let mut pty_parser = ansi::Parser::new();
-
     let window = Arc::new(window);
     let window_ref = window.clone();
 
     let (render_tx, render_rx) = mpsc::channel::<(u32, u32)>();
 
     let update_thread = thread::spawn_named("Update", move || {
+        let mut input_processor = input::Processor::new();
+
         'main_loop: loop {
             let mut writer = BufWriter::new(&writer);
-            let mut input_processor = input::Processor::new();
-
-            // Handle case where renderer didn't acquire lock yet
-            if needs_render.load(Ordering::Acquire) {
-                ::std::thread::yield_now();
-                continue;
-            }
 
             if process_should_exit() {
                 break;
@@ -231,11 +240,10 @@ fn main() {
             };
 
             // Need mutable terminal for updates; lock it.
-            let mut terminal = terminal.lock();
+            let mut terminal = terminal.lock_low();
             let res = handle_event(event,
                                    &mut writer,
                                    &mut *terminal,
-                                   &mut pty_parser,
                                    &render_tx,
                                    &mut input_processor);
             if res == ShouldExit::Yes {
@@ -249,7 +257,6 @@ fn main() {
                         let res = handle_event(e,
                                                &mut writer,
                                                &mut *terminal,
-                                               &mut pty_parser,
                                                &render_tx,
                                                &mut input_processor);
 
@@ -260,11 +267,6 @@ fn main() {
                     Err(mpsc::TryRecvError::Disconnected) => break 'main_loop,
                     Err(mpsc::TryRecvError::Empty) => break,
                 }
-
-                // Release the lock if a render is needed
-                if needs_render.load(Ordering::Acquire) {
-                    break;
-                }
             }
         }
     });
@@ -283,7 +285,7 @@ fn main() {
 
         // Initialize glyph cache
         {
-            let terminal = term_ref.lock();
+            let terminal = term_ref.lock_high();
             renderer.with_api(terminal.size_info(), |mut api| {
                 glyph_cache.init(&mut api);
             });
@@ -306,12 +308,8 @@ fn main() {
 
             // Need scope so lock is released when swap_buffers is called
             {
-                // Flag that it's time for render
-                needs_render2.store(true, Ordering::Release);
                 // Acquire term lock
-                let terminal = term_ref.lock();
-                // Have the lock, ok to lower flag
-                needs_render2.store(false, Ordering::Relaxed);
+                let terminal = term_ref.lock_high();
 
                 // Draw grid + cursor
                 {
@@ -348,7 +346,7 @@ fn main() {
 
     'event_processing: loop {
         for event in window_ref.wait_events() {
-            tx.send(Event::Glutin(event)).unwrap();
+            tx.send(event).unwrap();
             if process_should_exit() {
                 break 'event_processing;
             }
diff --git a/src/sync.rs b/src/sync.rs
new file mode 100644
index 00000000..e341d78e
--- /dev/null
+++ b/src/sync.rs
@@ -0,0 +1,94 @@
+// Copyright 2016 Joe Wilm, The Alacritty Project Contributors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Synchronization types
+//!
+//! Most importantly, a priority mutex is included
+use std::ops::{Deref, DerefMut};
+
+use parking_lot::{Mutex, MutexGuard};
+
+/// A priority mutex
+///
+/// A triple locking strategy is used where low priority locks must go through an additional mutex
+/// to access the data. The gist is
+///
+/// Low priority: lock low, lock next, lock data, unlock next, {do work}, unlock data, unlock low
+/// High priority: lock next, lock data, unlock next, {do work}, unlock data
+///
+/// By keeping the low lock active while working on data, a high priority consumer has immediate
+/// access to the next mutex.
+pub struct PriorityMutex<T> {
+    /// Data
+    data: Mutex<T>,
+    /// Next-to-access
+    next: Mutex<()>,
+    /// Low-priority access
+    low: Mutex<()>,
+}
+
+/// Mutex guard for low priority locks
+pub struct LowPriorityMutexGuard<'a, T: 'a> {
+    data: MutexGuard<'a, T>,
+    _low: MutexGuard<'a, ()>,
+}
+
+impl<'a, T> Deref for LowPriorityMutexGuard<'a, T> {
+    type Target = T;
+
+    #[inline]
+    fn deref(&self) -> &T {
+        self.data.deref()
+    }
+}
+
+impl<'a, T> DerefMut for LowPriorityMutexGuard<'a, T> {
+    #[inline]
+    fn deref_mut(&mut self) -> &mut T {
+        self.data.deref_mut()
+    }
+}
+
+impl<T> PriorityMutex<T> {
+    /// Create a new priority mutex
+    pub fn new(data: T) -> PriorityMutex<T> {
+        PriorityMutex {
+            data: Mutex::new(data),
+            next: Mutex::new(()),
+            low: Mutex::new(()),
+        }
+    }
+
+    /// Lock the mutex with high priority
+    pub fn lock_high(&self) -> MutexGuard<T> {
+        // Must bind to a temporary or the lock will be freed before going
+        // into data.lock()
+        let _next = self.next.lock();
+        self.data.lock()
+    }
+
+    /// Lock the mutex with low priority
+    pub fn lock_low(&self) -> LowPriorityMutexGuard<T> {
+        let low = self.low.lock();
+        // Must bind to a temporary or the lock will be freed before going
+        // into data.lock()
+        let _next = self.next.lock();
+        let data = self.data.lock();
+
+        LowPriorityMutexGuard {
+            data: data,
+            _low: low,
+        }
+    }
+}