Merge pull request #473 from zellij-org/bus-business-with-extra-merge-fun

Refactoring sender/receiver stuff - post-merge

16 files changed, 984 insertions, 905 deletions

diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md
index a54585ae4..9b2ee7565 100644
--- a/docs/ARCHITECTURE.md
+++ b/docs/ARCHITECTURE.md
@@ -37,5 +37,5 @@ The Boundaries refer to those lines that are drawn between terminal panes. A few
   * The Rect trait is here so that different panes can implement it, giving boundaries a generic way to calculate the size of the pane and draw boundaries around it.
   * Here we use the [unicode box drawing characters](https://en.wikipedia.org/wiki/Box-drawing_character) in order to draw the borders. There's some logic here about combining them together for all possible combinations of pane locations.
 
-## PTY Bus (src/pty_bus.rs)
+## PTY Bus (src/pty.rs)
 The PtyBus keeps track of several asynchronous streams that read from pty sockets (eg. /dev/pts/999), parse those bytes into ANSI/VT events and send them on to the Screen so that they can be received in the relevant TerminalPane.
diff --git a/src/client/mod.rs b/src/client/mod.rs
index 8a5834f06..ba27bc3e7 100644
--- a/src/client/mod.rs
+++ b/src/client/mod.rs
@@ -16,7 +16,7 @@ use crate::common::{
     input::config::Config,
     input::handler::input_loop,
     os_input_output::ClientOsApi,
-    SenderType, SenderWithContext, SyncChannelWithContext,
+    thread_bus::{SenderType, SenderWithContext, SyncChannelWithContext},
 };
 use crate::server::ServerInstruction;
 
diff --git a/src/client/panes/plugin_pane.rs b/src/client/panes/plugin_pane.rs
index c3b633ecc..dde3cd9b2 100644
--- a/src/client/panes/plugin_pane.rs
+++ b/src/client/panes/plugin_pane.rs
@@ -1,9 +1,12 @@
-use crate::{common::SenderWithContext, pty_bus::VteBytes, tab::Pane, wasm_vm::PluginInstruction};
+use std::sync::mpsc::channel;
+use std::time::Instant;
+use std::unimplemented;
 
+use crate::common::thread_bus::SenderWithContext;
 use crate::panes::{PaneId, PositionAndSize};
-
-use std::time::Instant;
-use std::{sync::mpsc::channel, unimplemented};
+use crate::pty::VteBytes;
+use crate::tab::Pane;
+use crate::wasm_vm::PluginInstruction;
 
 pub struct PluginPane {
     pub pid: u32,
diff --git a/src/client/panes/terminal_pane.rs b/src/client/panes/terminal_pane.rs
index f58c3a273..5792bba93 100644
--- a/src/client/panes/terminal_pane.rs
+++ b/src/client/panes/terminal_pane.rs
@@ -1,15 +1,16 @@
-use crate::tab::Pane;
-use ::nix::pty::Winsize;
-use ::std::os::unix::io::RawFd;
-use serde::{Deserialize, Serialize};
 use std::fmt::Debug;
+use std::os::unix::io::RawFd;
 use std::time::Instant;
 
+use nix::pty::Winsize;
+use serde::{Deserialize, Serialize};
+
 use crate::panes::grid::Grid;
 use crate::panes::terminal_character::{
     CharacterStyles, TerminalCharacter, EMPTY_TERMINAL_CHARACTER,
 };
-use crate::pty_bus::VteBytes;
+use crate::pty::VteBytes;
+use crate::tab::Pane;
 
 #[derive(PartialEq, Eq, Ord, PartialOrd, Hash, Clone, Copy, Debug, Serialize, Deserialize)]
 pub enum PaneId {
diff --git a/src/client/tab.rs b/src/client/tab.rs
index c4dc34743..bbd68a4d5 100644
--- a/src/client/tab.rs
+++ b/src/client/tab.rs
@@ -2,11 +2,12 @@
 //! as well as how they should be resized
 
 use crate::client::pane_resizer::PaneResizer;
-use crate::common::{input::handler::parse_keys, SenderWithContext};
+use crate::common::input::handler::parse_keys;
+use crate::common::thread_bus::ThreadSenders;
 use crate::layout::Layout;
 use crate::os_input_output::ServerOsApi;
 use crate::panes::{PaneId, PositionAndSize, TerminalPane};
-use crate::pty_bus::{PtyInstruction, VteBytes};
+use crate::pty::{PtyInstruction, VteBytes};
 use crate::server::ServerInstruction;
 use crate::utils::shared::adjust_to_size;
 use crate::wasm_vm::PluginInstruction;
@@ -69,9 +70,7 @@ pub struct Tab {
     full_screen_ws: PositionAndSize,
     fullscreen_is_active: bool,
     os_api: Box<dyn ServerOsApi>,
-    send_plugin_instructions: SenderWithContext<PluginInstruction>,
-    send_pty_instructions: SenderWithContext<PtyInstruction>,
-    send_server_instructions: SenderWithContext<ServerInstruction>,
+    pub senders: ThreadSenders,
     synchronize_is_active: bool,
     should_clear_display_before_rendering: bool,
     pub mode_info: ModeInfo,
@@ -219,7 +218,7 @@ pub trait Pane {
 }
 
 impl Tab {
-    // FIXME: Too many arguments here! Maybe bundle all of the senders for the whole program in a struct?
+    // FIXME: Still too many arguments for clippy to be happy...
     #[allow(clippy::too_many_arguments)]
     pub fn new(
         index: usize,
@@ -227,9 +226,7 @@ impl Tab {
         name: String,
         full_screen_ws: &PositionAndSize,
         mut os_api: Box<dyn ServerOsApi>,
-        send_plugin_instructions: SenderWithContext<PluginInstruction>,
-        send_pty_instructions: SenderWithContext<PtyInstruction>,
-        send_server_instructions: SenderWithContext<ServerInstruction>,
+        senders: ThreadSenders,
         max_panes: Option<usize>,
         pane_id: Option<PaneId>,
         mode_info: ModeInfo,
@@ -261,9 +258,7 @@ impl Tab {
             fullscreen_is_active: false,
             synchronize_is_active: false,
             os_api,
-            send_plugin_instructions,
-            send_pty_instructions,
-            send_server_instructions,
+            senders,
             should_clear_display_before_rendering: false,
             mode_info,
             input_mode,
@@ -315,14 +310,14 @@ impl Tab {
             // Just a regular terminal
             if let Some(plugin) = &layout.plugin {
                 let (pid_tx, pid_rx) = channel();
-                self.send_plugin_instructions
-                    .send(PluginInstruction::Load(pid_tx, plugin.clone()))
+                self.senders
+                    .send_to_plugin(PluginInstruction::Load(pid_tx, plugin.clone()))
                     .unwrap();
                 let pid = pid_rx.recv().unwrap();
                 let mut new_plugin = PluginPane::new(
                     pid,
                     *position_and_size,
-                    self.send_plugin_instructions.clone(),
+                    self.senders.to_plugin.as_ref().unwrap().clone(),
                 );
                 if let Some(max_rows) = position_and_size.max_rows {
                     new_plugin.set_max_height(max_rows);
@@ -332,8 +327,8 @@ impl Tab {
                 }
                 self.panes.insert(PaneId::Plugin(pid), Box::new(new_plugin));
                 // Send an initial mode update to the newly loaded plugin only!
-                self.send_plugin_instructions
-                    .send(PluginInstruction::Update(
+                self.senders
+                    .send_to_plugin(PluginInstruction::Update(
                         Some(pid),
                         Event::ModeUpdate(self.mode_info.clone()),
                     ))
@@ -355,8 +350,8 @@ impl Tab {
             // this is a bit of a hack and happens because we don't have any central location that
             // can query the screen as to how many panes it needs to create a layout
             // fixing this will require a bit of an architecture change
-            self.send_pty_instructions
-                .send(PtyInstruction::ClosePane(PaneId::Terminal(*unused_pid)))
+            self.senders
+                .send_to_pty(PtyInstruction::ClosePane(PaneId::Terminal(*unused_pid)))
                 .unwrap();
         }
         self.active_terminal = self.panes.iter().map(|(id, _)| id.to_owned()).next();
@@ -400,9 +395,9 @@ impl Tab {
                 },
             );
             if terminal_id_to_split.is_none() {
-                self.send_pty_instructions
-                    .send(PtyInstruction::ClosePane(pid))
-                    .unwrap(); // we can't open this pane, close the pty
+                self.senders
+                    .send_to_pty(PtyInstruction::ClosePane(pid)) // we can't open this pane, close the pty
+                    .unwrap();
                 return; // likely no terminal large enough to split
             }
             let terminal_id_to_split = terminal_id_to_split.unwrap();
@@ -481,9 +476,9 @@ impl Tab {
             let active_pane_id = &self.get_active_pane_id().unwrap();
             let active_pane = self.panes.get_mut(active_pane_id).unwrap();
             if active_pane.rows() < MIN_TERMINAL_HEIGHT * 2 + 1 {
-                self.send_pty_instructions
-                    .send(PtyInstruction::ClosePane(pid))
-                    .unwrap(); // we can't open this pane, close the pty
+                self.senders
+                    .send_to_pty(PtyInstruction::ClosePane(pid)) // we can't open this pane, close the pty
+                    .unwrap();
                 return;
             }
             let terminal_ws = PositionAndSize {
@@ -538,9 +533,9 @@ impl Tab {
             let active_pane_id = &self.get_active_pane_id().unwrap();
             let active_pane = self.panes.get_mut(active_pane_id).unwrap();
             if active_pane.columns() < MIN_TERMINAL_WIDTH * 2 + 1 {
-                self.send_pty_instructions
-                    .send(PtyInstruction::ClosePane(pid))
-                    .unwrap(); // we can't open this pane, close the pty
+                self.senders
+                    .send_to_pty(PtyInstruction::ClosePane(pid)) // we can't open this pane, close the pty
+                    .unwrap();
                 return;
             }
             let terminal_ws = PositionAndSize {
@@ -597,7 +592,7 @@ impl Tab {
     }
     pub fn handle_pty_bytes(&mut self, pid: RawFd, bytes: VteBytes) {
         // if we don't have the terminal in self.terminals it's probably because
-        // of a race condition where the terminal was created in pty_bus but has not
+        // of a race condition where the terminal was created in pty but has not
         // yet been created in Screen. These events are currently not buffered, so
         // if you're debugging seemingly randomly missing stdout data, this is
         // the reason
@@ -628,8 +623,8 @@ impl Tab {
             }
             PaneId::Plugin(pid) => {
                 for key in parse_keys(&input_bytes) {
-                    self.send_plugin_instructions
-                        .send(PluginInstruction::Update(Some(pid), Event::KeyPress(key)))
+                    self.senders
+                        .send_to_plugin(PluginInstruction::Update(Some(pid), Event::KeyPress(key)))
                         .unwrap()
                 }
             }
@@ -706,7 +701,7 @@ impl Tab {
     pub fn is_sync_panes_active(&self) -> bool {
         self.synchronize_is_active
     }
-    pub fn toggle_sync_tab_is_active(&mut self) {
+    pub fn toggle_sync_panes_is_active(&mut self) {
         self.synchronize_is_active = !self.synchronize_is_active;
     }
     pub fn panes_contain_widechar(&self) -> bool {
@@ -781,8 +776,8 @@ impl Tab {
             }
         }
 
-        self.send_server_instructions
-            .send(ServerInstruction::Render(Some(output)))
+        self.senders
+            .send_to_server(ServerInstruction::Render(Some(output)))
             .unwrap();
     }
     fn get_panes(&self) -> impl Iterator<Item = (&PaneId, &Box<dyn Pane>)> {
@@ -2086,8 +2081,8 @@ impl Tab {
         if let Some(max_panes) = self.max_panes {
             let terminals = self.get_pane_ids();
             for &pid in terminals.iter().skip(max_panes - 1) {
-                self.send_pty_instructions
-                    .send(PtyInstruction::ClosePane(pid))
+                self.senders
+                    .send_to_pty(PtyInstruction::ClosePane(pid))
                     .unwrap();
                 self.close_pane_without_rerender(pid);
             }
@@ -2198,8 +2193,8 @@ impl Tab {
     pub fn close_focused_pane(&mut self) {
         if let Some(active_pane_id) = self.get_active_pane_id() {
             self.close_pane(active_pane_id);
-            self.send_pty_instructions
-                .send(PtyInstruction::ClosePane(active_pane_id))
+            self.senders
+                .send_to_pty(PtyInstruction::ClosePane(active_pane_id))
                 .unwrap();
         }
     }
diff --git a/src/common/errors.rs b/src/common/errors.rs
index d8327462f..148a3a9e8 100644
--- a/src/common/errors.rs
+++ b/src/common/errors.rs
@@ -1,20 +1,22 @@
 //! Error context system based on a thread-local representation of the call stack, itself based on
 //! the instructions that are sent between threads.
 
-use super::{ServerInstruction, ASYNCOPENCALLS, OPENCALLS};
-use crate::client::ClientInstruction;
-use crate::pty_bus::PtyInstruction;
-use crate::screen::ScreenInstruction;
 use serde::{Deserialize, Serialize};
 
 use std::fmt::{Display, Error, Formatter};
 
+use crate::client::ClientInstruction;
+use crate::common::thread_bus::{ASYNCOPENCALLS, OPENCALLS};
+use crate::pty::PtyInstruction;
+use crate::screen::ScreenInstruction;
+use crate::server::ServerInstruction;
+
 /// The maximum amount of calls an [`ErrorContext`] will keep track
 /// of in its stack representation. This is a per-thread maximum.
 const MAX_THREAD_CALL_STACK: usize = 6;
 
 #[cfg(not(test))]
-use super::SenderWithContext;
+use super::thread_bus::SenderWithContext;
 #[cfg(not(test))]
 use std::panic::PanicInfo;
 /// Custom panic handler/hook. Prints the [`ErrorContext`].
diff --git a/src/common/input/handler.rs b/src/common/input/handler.rs
index a42932200..786352372 100644
--- a/src/common/input/handler.rs
+++ b/src/common/input/handler.rs
@@ -4,7 +4,7 @@ use super::actions::Action;
 use super::keybinds::Keybinds;
 use crate::client::ClientInstruction;
 use crate::common::input::config::Config;
-use crate::common::{SenderWithContext, OPENCALLS};
+use crate::common::thread_bus::{SenderWithContext, OPENCALLS};
 use crate::errors::ContextType;
 use crate::os_input_output::ClientOsApi;
 use crate::server::ServerInstruction;
diff --git a/src/common/mod.rs b/src/common/mod.rs
index 6061f1a62..c91e1e05b 100644
--- a/src/common/mod.rs
+++ b/src/common/mod.rs
@@ -3,73 +3,12 @@ pub mod errors;
 pub mod input;
 pub mod ipc;
 pub mod os_input_output;
-pub mod pty_bus;
+pub mod pty;
 pub mod screen;
 pub mod setup;
+pub mod thread_bus;
 pub mod utils;
 pub mod wasm_vm;
 
 use crate::panes::PaneId;
 use crate::server::ServerInstruction;
-use async_std::task_local;
-use errors::{get_current_ctx, ErrorContext};
-use std::cell::RefCell;
-use std::sync::mpsc;
-
-/// An [MPSC](mpsc) asynchronous channel with added error context.
-pub type ChannelWithContext<T> = (
-    mpsc::Sender<(T, ErrorContext)>,
-    mpsc::Receiver<(T, ErrorContext)>,
-);
-/// An [MPSC](mpsc) synchronous channel with added error context.
-pub type SyncChannelWithContext<T> = (
-    mpsc::SyncSender<(T, ErrorContext)>,
-    mpsc::Receiver<(T, ErrorContext)>,
-);
-
-/// Wrappers around the two standard [MPSC](mpsc) sender types, [`mpsc::Sender`] and [`mpsc::SyncSender`], with an additional [`ErrorContext`].
-#[derive(Clone)]
-pub enum SenderType<T: Clone> {
-    /// A wrapper around an [`mpsc::Sender`], adding an [`ErrorContext`].
-    Sender(mpsc::Sender<(T, ErrorContext)>),
-    /// A wrapper around an [`mpsc::SyncSender`], adding an [`ErrorContext`].
-    SyncSender(mpsc::SyncSender<(T, ErrorContext)>),
-}
-
-/// Sends messages on an [MPSC](std::sync::mpsc) channel, along with an [`ErrorContext`],
-/// synchronously or asynchronously depending on the underlying [`SenderType`].
-#[derive(Clone)]
-pub struct SenderWithContext<T: Clone> {
-    sender: SenderType<T>,
-}
-
-impl<T: Clone> SenderWithContext<T> {
-    pub fn new(sender: SenderType<T>) -> Self {
-        Self { sender }
-    }
-
-    /// Sends an event, along with the current [`ErrorContext`], on this
-    /// [`SenderWithContext`]'s channel.
-    pub fn send(&self, event: T) -> Result<(), mpsc::SendError<(T, ErrorContext)>> {
-        let err_ctx = get_current_ctx();
-        match self.sender {
-            SenderType::Sender(ref s) => s.send((event, err_ctx)),
-            SenderType::SyncSender(ref s) => s.send((event, err_ctx)),
-        }
-    }
-}
-
-unsafe impl<T: Clone> Send for SenderWithContext<T> {}
-unsafe impl<T: Clone> Sync for SenderWithContext<T> {}
-
-thread_local!(
-    /// A key to some thread local storage (TLS) that holds a representation of the thread's call
-    /// stack in the form of an [`ErrorContext`].
-    pub static OPENCALLS: RefCell<ErrorContext> = RefCell::default()
-);
-
-task_local! {
-    /// A key to some task local storage that holds a representation of the task's call
-    /// stack in the form of an [`ErrorContext`].
-    static ASYNCOPENCALLS: RefCell<ErrorContext> = RefCell::default()
-}
diff --git a/src/common/pty_bus.rs b/src/common/pty.rs
index 316961868..3c471d6ae 100644
--- a/src/common/pty_bus.rs
+++ b/src/common/pty.rs
@@ -1,23 +1,21 @@
-use ::async_std::stream::*;
-use ::async_std::task;
-use ::async_std::task::*;
-use ::std::collections::HashMap;
-use ::std::os::unix::io::RawFd;
-use ::std::pin::*;
-use ::std::sync::mpsc::Receiver;
-use ::std::time::{Duration, Instant};
+use async_std::stream::*;
+use async_std::task;
+use async_std::task::*;
+use std::collections::HashMap;
+use std::os::unix::io::RawFd;
 use std::path::PathBuf;
+use std::pin::*;
+use std::time::{Duration, Instant};
 
-use super::SenderWithContext;
+use crate::client::panes::PaneId;
+use crate::common::errors::{get_current_ctx, ContextType, PtyContext};
+use crate::common::screen::ScreenInstruction;
+use crate::common::thread_bus::{Bus, ThreadSenders};
 use crate::layout::Layout;
 use crate::os_input_output::ServerOsApi;
+use crate::server::ServerInstruction;
 use crate::utils::logging::debug_to_file;
-use crate::{
-    errors::{get_current_ctx, ContextType, ErrorContext},
-    panes::PaneId,
-    screen::ScreenInstruction,
-    wasm_vm::PluginInstruction,
-};
+use crate::wasm_vm::PluginInstruction;
 
 pub struct ReadFromPid {
     pid: RawFd,
@@ -79,19 +77,72 @@ pub enum PtyInstruction {
     Exit,
 }
 
-pub struct PtyBus {
-    pub receive_pty_instructions: Receiver<(PtyInstruction, ErrorContext)>,
+pub struct Pty {
+    pub bus: Bus<PtyInstruction>,
     pub id_to_child_pid: HashMap<RawFd, RawFd>,
-    pub send_screen_instructions: SenderWithContext<ScreenInstruction>,
-    send_plugin_instructions: SenderWithContext<PluginInstruction>,
-    os_input: Box<dyn ServerOsApi>,
     debug_to_file: bool,
     task_handles: HashMap<RawFd, JoinHandle<()>>,
 }
 
+pub fn pty_thread_main(mut pty: Pty, maybe_layout: Option<Layout>) {
+    loop {
+        let (event, mut err_ctx) = pty.bus.recv().expect("failed to receive event on channel");
+        err_ctx.add_call(ContextType::Pty(PtyContext::from(&event)));
+        match event {
+            PtyInstruction::SpawnTerminal(file_to_open) => {
+                let pid = pty.spawn_terminal(file_to_open);
+                pty.bus
+                    .senders
+                    .send_to_screen(ScreenInstruction::NewPane(PaneId::Terminal(pid)))
+                    .unwrap();
+            }
+            PtyInstruction::SpawnTerminalVertically(file_to_open) => {
+                let pid = pty.spawn_terminal(file_to_open);
+                pty.bus
+                    .senders
+                    .send_to_screen(ScreenInstruction::VerticalSplit(PaneId::Terminal(pid)))
+                    .unwrap();
+            }
+            PtyInstruction::SpawnTerminalHorizontally(file_to_open) => {
+                let pid = pty.spawn_terminal(file_to_open);
+                pty.bus
+                    .senders
+                    .send_to_screen(ScreenInstruction::HorizontalSplit(PaneId::Terminal(pid)))
+                    .unwrap();
+            }
+            PtyInstruction::NewTab => {
+                if let Some(layout) = maybe_layout.clone() {
+                    pty.spawn_terminals_for_layout(layout);
+                } else {
+                    let pid = pty.spawn_terminal(None);
+                    pty.bus
+                        .senders
+                        .send_to_screen(ScreenInstruction::NewTab(pid))
+                        .unwrap();
+                }
+            }
+            PtyInstruction::ClosePane(id) => {
+                pty.close_pane(id);
+                pty.bus
+                    .senders
+                    .send_to_server(ServerInstruction::UnblockInputThread)
+                    .unwrap();
+            }
+            PtyInstruction::CloseTab(ids) => {
+                pty.close_tab(ids);
+                pty.bus
+                    .senders
+                    .send_to_server(ServerInstruction::UnblockInputThread)
+                    .unwrap();
+            }
+            PtyInstruction::Exit => break,
+        }
+    }
+}
+
 fn stream_terminal_bytes(
     pid: RawFd,
-    send_screen_instructions: SenderWithContext<ScreenInstruction>,
+    senders: ThreadSenders,
     os_input: Box<dyn ServerOsApi>,
     debug: bool,
 ) -> JoinHandle<()> {
@@ -113,7 +164,7 @@ fn stream_terminal_bytes(
                     }
                 }
                 if !bytes_is_empty {
-                    let _ = send_screen_instructions.send(ScreenInstruction::PtyBytes(pid, bytes));
+                    let _ = senders.send_to_screen(ScreenInstruction::PtyBytes(pid, bytes));
                     // for UX reasons, if we got something on the wire, we only send the render notice if:
                     // 1. there aren't any more bytes on the wire afterwards
                     // 2. a certain period (currently 30ms) has elapsed since the last render
@@ -124,9 +175,7 @@ fn stream_terminal_bytes(
                         Some(receive_time) => {
                             if receive_time.elapsed() > max_render_pause {
                                 pending_render = false;
-                                send_screen_instructions
-                                    .send(ScreenInstruction::Render)
-                                    .unwrap();
+                                let _ = senders.send_to_screen(ScreenInstruction::Render);
                                 last_byte_receive_time = Some(Instant::now());
                             } else {
                                 pending_render = true;
@@ -140,53 +189,44 @@ fn stream_terminal_bytes(
                 } else {
                     if pending_render {
                         pending_render = false;
-                        send_screen_instructions
-                            .send(ScreenInstruction::Render)
-                            .unwrap();
+                        let _ = senders.send_to_screen(ScreenInstruction::Render);
                     }
                     last_byte_receive_time = None;
                     task::sleep(::std::time::Duration::from_millis(10)).await;
                 }
             }
-            send_screen_instructions
-                .send(ScreenInstruction::Render)
-                .unwrap();
+            senders.send_to_screen(ScreenInstruction::Render).unwrap();
             #[cfg(not(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
-            send_screen_instructions
-                .send(ScreenInstruction::ClosePane(PaneId::Terminal(pid)))
+            senders
+                .send_to_screen(ScreenInstruction::ClosePane(PaneId::Terminal(pid)))
                 .unwrap();
         }
     })
 }
 
-impl PtyBus {
-    pub fn new(
-        receive_pty_instructions: Receiver<(PtyInstruction, ErrorContext)>,
-        send_screen_instructions: SenderWithContext<ScreenInstruction>,
-        send_plugin_instructions: SenderWithContext<PluginInstruction>,
-        os_input: Box<dyn ServerOsApi>,
-