fix a starvation issue on linux/x11 systems

The `SpawnQueue::run_impl` would loop until it had exhausted
all queued items.  This prevents returning to the main loop
and resulted in the UI hanging while eg: `yes` was running,
and could also block accepting keyboard input, which is
pretty bad.

In addition, the queue implementation could fill up a pipe
and block the write side while it held a lock, which in
turn would prevent the read side from making room for the
write to succeed!

This commit changes the behavior on linux to change the wakeup
behavior of the queue from having a 1:1 relationship between
enqueue:wakeup to n:m where n and m are both >= 1.  This is
sufficient to wake a sleeping gui thread.  The gui thread
can then pop and process a single item at a time, interleaved
with dispatching the gui events.

The result is a bit more responsive, however, there is no
backpressure from the gui to the read side, so if the read
side is eating 2MB/s of data and the GUI side is processing
less than this, then an interrupt signal may still take a
few seconds to take effect.

I have mixed feelings about adding backpressure, because
I'm not sure that it is worth actually rendering all of
the parsed output text when there is a lot of it.

I need to follow up and verify these changes on macOS
and Windows too.

Refs: https://github.com/wez/wezterm/issues/65

window/src/os/x11/connection.rs

@@ -264,12 +264,21 @@ impl ConnectionOps for Connection {
             // relied solely on that.
             self.process_queued_xcb()?;
 
-            let period = self
-                .timers
-                .borrow()
-                .time_until_due(Instant::now())
-                .map(|duration| duration.min(period))
-                .unwrap_or(period);
+            // Check the spawn queue before we try to sleep; there may
+            // be work pending and we don't guarantee that there is a
+            // 1:1 wakeup to queued function, so we need to be assertive
+            // in order to avoid missing wakeups
+            let period = if SPAWN_QUEUE.run() {
+                // if we processed one, we don't want to sleep because
+                // there may be others to deal with
+                Duration::new(0, 0)
+            } else {
+                self.timers
+                    .borrow()
+                    .time_until_due(Instant::now())
+                    .map(|duration| duration.min(period))
+                    .unwrap_or(period)
+            };
 
             match poll.poll(&mut events, Some(period)) {
                 Ok(_) => {
@@ -282,7 +291,6 @@ impl ConnectionOps for Connection {
                         } else {
                         }
                     }
-                    // self.process_sigchld();
                 }
 
                 Err(err) => {

window/src/spawn.rs

@@ -39,7 +39,7 @@ impl SpawnQueue {
         self.spawn_impl(f)
     }
 
-    pub fn run(&self) {
+    pub fn run(&self) -> bool {
         self.run_impl()
     }
 
@@ -67,18 +67,36 @@ impl SpawnQueue {
         self.event_handle.set_event();
     }
 
-    fn run_impl(&self) {
+    fn run_impl(&self) -> bool {
         self.event_handle.reset_event();
+        let mut did_any = false;
         while let Some(func) = self.pop_func() {
             func();
+            did_any = true;
         }
+        did_any
     }
 }
 
 #[cfg(all(unix, not(target_os = "macos")))]
 impl SpawnQueue {
     fn new_impl() -> Fallible<Self> {
+        // On linux we have a slightly sloppy wakeup mechanism;
+        // we have a non-blocking pipe that we can use to get
+        // woken up after some number of enqueues.  We don't
+        // guarantee a 1:1 enqueue to wakeup with this mechanism
+        // but in practical terms it does guarantee a wakeup
+        // if the main thread is asleep and we enqueue some
+        // number of items.
+        // We can't affort to use a blocking pipe for the wakeup
+        // because the write needs to hold a mutex and that
+        // can block reads as well as other writers.
         let pipe = Pipe::new()?;
+        let on = 1;
+        unsafe {
+            libc::ioctl(pipe.write.as_raw_fd(), libc::FIONBIO, &on);
+            libc::ioctl(pipe.read.as_raw_fd(), libc::FIONBIO, &on);
+        }
         Ok(Self {
             spawned_funcs: Mutex::new(VecDeque::new()),
             write: Mutex::new(pipe.write),
@@ -93,13 +111,19 @@ impl SpawnQueue {
         self.write.lock().unwrap().write(b"x").ok();
     }
 
-    fn run_impl(&self) {
+    fn run_impl(&self) -> bool {
+        // On linux we only ever process one at at time, so that
+        // we can return to the main loop and process messages
+        // from the X server
         use std::io::Read;
-        while let Some(func) = self.pop_func() {
+        if let Some(func) = self.pop_func() {
             func();
 
             let mut byte = [0u8];
             self.read.lock().unwrap().read(&mut byte).ok();
+            true
+        } else {
+            false
         }
     }
 }
@@ -168,10 +192,13 @@ impl SpawnQueue {
         }
     }
 
-    fn run_impl(&self) {
+    fn run_impl(&self) -> bool {
+        let mut did_any = false;
         while let Some(func) = self.pop_func() {
             func();
+            did_any = true;
         }
+        did_any
     }
 }