fix: use bounded queue between pty and screen

Pty reads a command's output and feeds it to Screen using an unbounded
queue.

However, if the command produces output faster than what `Screen` can
render, `Pty` still pushes it on the queue, causing it to grow
indefinitely, resulting in high memory usage and latency.

This patch fixes this by using a bounded queue between Pty and Screen,
so if Screen can't keep up with Pty, the queue will fill up, exerting
back pressure on Pty, making it read the command's output only as fast
as Screen renders it.

The unbounded queue is kept between Screen and producers other than Pty
to avoid a deadlock such as this scenario:
* pty thread filling up screen queue as soon as screen thread pops
  something from it
* wasm thread is processing a Render instruction, blocking on the screen
  queue
* screen thread is trying to render a plugin pane. It attempts to send a
  Render insturction to the blocked wasm thread running the plugin.

This patch also adds a generous amount of sleeps to the integration
tests as having backpressure changes the timing of how instructions are
processed.

Fixes #525.

1 files changed, 15 insertions, 11 deletions

diff --git a/zellij-server/src/pty.rs b/zellij-server/src/pty.rs
index 803c04248..41f9d1a9b 100644
--- a/zellij-server/src/pty.rs
+++ b/zellij-server/src/pty.rs
@@ -148,6 +148,12 @@ async fn deadline_read(
     }
 }
 
+async fn async_send_to_screen(senders: ThreadSenders, screen_instruction: ScreenInstruction) {
+    task::spawn_blocking(move || senders.send_to_screen(screen_instruction))
+        .await
+        .unwrap()
+}
+
 fn stream_terminal_bytes(
     pid: RawFd,
     senders: ThreadSenders,
@@ -171,36 +177,34 @@ fn stream_terminal_bytes(
                 match deadline_read(async_reader.as_mut(), render_deadline, &mut buf).await {
                     ReadResult::Ok(0) | ReadResult::Err(_) => break, // EOF or error
                     ReadResult::Timeout => {
-                        let _ = senders.send_to_screen(ScreenInstruction::Render);
+                        async_send_to_screen(senders.clone(), ScreenInstruction::Render).await;
                         // next read does not need a deadline as we just rendered everything
                         render_deadline = None;
-
-                        // yield so Screen thread has some time to render before send additional
-                        // PtyBytes.
-                        task::sleep(Duration::from_millis(10)).await;
                     }
                     ReadResult::Ok(n_bytes) => {
                         let bytes = &buf[..n_bytes];
                         if debug {
                             let _ = debug_to_file(bytes, pid);
                         }
-                        let _ = senders
-                            .send_to_screen(ScreenInstruction::PtyBytes(pid, bytes.to_vec()));
+                        async_send_to_screen(
+                            senders.clone(),
+                            ScreenInstruction::PtyBytes(pid, bytes.to_vec()),
+                        )
+                        .await;
                         // if we already have a render_deadline we keep it, otherwise we set it
                         // to the duration of `render_pause`.
                         render_deadline.get_or_insert(Instant::now() + render_pause);
                     }
                 }
             }
-            let _ = senders.send_to_screen(ScreenInstruction::Render);
+            async_send_to_screen(senders.clone(), ScreenInstruction::Render).await;
 
             #[cfg(not(any(feature = "test", test)))]
             // this is a little hacky, and is because the tests end the file as soon as
             // we read everything, rather than hanging until there is new data
             // a better solution would be to fix the test fakes, but this will do for now
-            senders
-                .send_to_screen(ScreenInstruction::ClosePane(PaneId::Terminal(pid)))
-                .unwrap();
+            async_send_to_screen(senders, ScreenInstruction::ClosePane(PaneId::Terminal(pid)))
+                .await;
         }
     })
 }