feat(plugin): simple timers implemented via the `set_timeout()` call

2 files changed, 57 insertions, 31 deletions

diff --git a/src/common/mod.rs b/src/common/mod.rs
index 2388a162..27e94f5c 100644
--- a/src/common/mod.rs
+++ b/src/common/mod.rs
@@ -40,7 +40,7 @@ use pty_bus::{PtyBus, PtyInstruction};
 use screen::{Screen, ScreenInstruction};
 use serde::{Deserialize, Serialize};
 use utils::consts::ZELLIJ_IPC_PIPE;
-use wasm_vm::{wasi_stdout, wasi_write_json, zellij_exports, PluginEnv, PluginInstruction};
+use wasm_vm::{wasi_read_string, wasi_write_object, zellij_exports, PluginEnv, PluginInstruction};
 use wasmer::{ChainableNamedResolver, Instance, Module, Store, Value};
 use wasmer_wasi::{Pipe, WasiState};
 use zellij_tile::data::{EventType, ModeInfo};
@@ -457,23 +457,12 @@ pub fn start(mut os_input: Box<dyn OsApi>, opts: CliArgs) {
             let send_pty_instructions = send_pty_instructions.clone();
             let send_screen_instructions = send_screen_instructions.clone();
             let send_app_instructions = send_app_instructions.clone();
+            let send_plugin_instructions = send_plugin_instructions.clone();
 
             let store = Store::default();
             let mut plugin_id = 0;
             let mut plugin_map = HashMap::new();
             move || loop {
-                // BLUEPRINT:
-                /* next = binheap.get_timer(); // peek
-                let event = if let Some(time) = next {
-                    // match for the right timeout error here...
-                    if let Some(event) = chan.recv_timeout(time) {
-                        event
-                    } else {
-                        continue
-                    }
-                } else {
-                    ch.recv().unwrap()
-                } */
                 let (event, mut err_ctx) = receive_plugin_instructions
                     .recv()
                     .expect("failed to receive event on channel");
@@ -513,6 +502,7 @@ pub fn start(mut os_input: Box<dyn OsApi>, opts: CliArgs) {
                             send_pty_instructions: send_pty_instructions.clone(),
                             send_screen_instructions: send_screen_instructions.clone(),
                             send_app_instructions: send_app_instructions.clone(),
+                            send_plugin_instructions: send_plugin_instructions.clone(),
                             wasi_env,
                             subscriptions: Arc::new(Mutex::new(HashSet::new())),
                         };
@@ -536,7 +526,7 @@ pub fn start(mut os_input: Box<dyn OsApi>, opts: CliArgs) {
                             let event_type = EventType::from_str(&event.to_string()).unwrap();
                             if (pid.is_none() || pid == Some(i)) && subs.contains(&event_type) {
                                 let update = instance.exports.get_function("update").unwrap();
-                                wasi_write_json(&plugin_env.wasi_env, &event);
+                                wasi_write_object(&plugin_env.wasi_env, &event);
                                 update.call(&[]).unwrap();
                             }
                         }
@@ -551,7 +541,7 @@ pub fn start(mut os_input: Box<dyn OsApi>, opts: CliArgs) {
                             .call(&[Value::I32(rows as i32), Value::I32(cols as i32)])
                             .unwrap();
 
-                        buf_tx.send(wasi_stdout(&plugin_env.wasi_env)).unwrap();
+                        buf_tx.send(wasi_read_string(&plugin_env.wasi_env)).unwrap();
                     }
                     PluginInstruction::Unload(pid) => drop(plugin_map.remove(&pid)),
                     PluginInstruction::Quit => break,
diff --git a/src/common/wasm_vm.rs b/src/common/wasm_vm.rs
index 6d31a6bb..bd98a85b 100644
--- a/src/common/wasm_vm.rs
+++ b/src/common/wasm_vm.rs
@@ -1,5 +1,12 @@
-use serde::Serialize;
-use std::{collections::HashSet, path::PathBuf, process, sync::{mpsc::Sender, Arc, Mutex}, time::{Duration, Instant}};
+use serde::{de::DeserializeOwned, Serialize};
+use std::{
+    collections::HashSet,
+    path::PathBuf,
+    process,
+    sync::{mpsc::Sender, Arc, Mutex},
+    thread,
+    time::{Duration, Instant},
+};
 use wasmer::{imports, Function, ImportObject, Store, WasmerEnv};
 use wasmer_wasi::WasiEnv;
 use zellij_tile::data::{Event, EventType, PluginIds};
@@ -20,18 +27,13 @@ pub enum PluginInstruction {
 #[derive(WasmerEnv, Clone)]
 pub struct PluginEnv {
     pub plugin_id: u32,
+    // FIXME: This should be a big bundle of all of the channels
     pub send_screen_instructions: SenderWithContext<ScreenInstruction>,
     pub send_app_instructions: SenderWithContext<AppInstruction>,
-    pub send_pty_instructions: SenderWithContext<PtyInstruction>, // FIXME: This should be a big bundle of all of the channels
+    pub send_pty_instructions: SenderWithContext<PtyInstruction>,
+    pub send_plugin_instructions: SenderWithContext<PluginInstruction>,
     pub wasi_env: WasiEnv,
     pub subscriptions: Arc<Mutex<HashSet<EventType>>>,
-    // FIXME: Add timers struct here! BinaryHeap?
-}
-
-struct CallbackTimer {
-    wake_time: Instant,
-    plugin_id: u32,
-    recurring: bool,
 }
 
 // Plugin API ---------------------------------------------------------------------------------------------------------
@@ -56,18 +58,19 @@ pub fn zellij_exports(store: &Store, plugin_env: &PluginEnv) -> ImportObject {
         host_set_selectable,
         host_get_plugin_ids,
         host_open_file,
+        host_set_timeout,
     }
 }
 
 fn host_subscribe(plugin_env: &PluginEnv) {
     let mut subscriptions = plugin_env.subscriptions.lock().unwrap();
-    let new: HashSet<EventType> = serde_json::from_str(&wasi_stdout(&plugin_env.wasi_env)).unwrap();
+    let new: HashSet<EventType> = wasi_read_object(&plugin_env.wasi_env);
     subscriptions.extend(new);
 }
 
 fn host_unsubscribe(plugin_env: &PluginEnv) {
     let mut subscriptions = plugin_env.subscriptions.lock().unwrap();
-    let old: HashSet<EventType> = serde_json::from_str(&wasi_stdout(&plugin_env.wasi_env)).unwrap();
+    let old: HashSet<EventType> = wasi_read_object(&plugin_env.wasi_env);
     subscriptions.retain(|k| !old.contains(k));
 }
 
@@ -109,21 +112,49 @@ fn host_get_plugin_ids(plugin_env: &PluginEnv) {
         plugin_id: plugin_env.plugin_id,
         zellij_pid: process::id(),
     };
-    wasi_write_json(&plugin_env.wasi_env, &ids);
+    wasi_write_object(&plugin_env.wasi_env, &ids);
 }
 
 fn host_open_file(plugin_env: &PluginEnv) {
-    let path = PathBuf::from(wasi_stdout(&plugin_env.wasi_env).lines().next().unwrap());
+    let path: PathBuf = wasi_read_object(&plugin_env.wasi_env);
     plugin_env
         .send_pty_instructions
         .send(PtyInstruction::SpawnTerminal(Some(path)))
         .unwrap();
 }
 
+fn host_set_timeout(plugin_env: &PluginEnv, secs: f64) {
+    // There is a fancy, high-performance way to do this with zero additional threads:
+    // If the plugin thread keeps a BinaryHeap of timer structs, it can manage multiple and easily `.peek()` at the
+    // next time to trigger in O(1) time. Once the wake-up time is known, the `wasm` thread can use `recv_timeout()`
+    // to wait for an event with the timeout set to be the time of the next wake up. If events come in in the meantime,
+    // they are handled, but if the timeout triggers, we replace the event from `recv()` with an
+    // `Update(pid, TimerEvent)` and pop the timer from the Heap (or reschedule it). No additional threads for as many
+    // timers as we'd like.
+    //
+    // But that's a lot of code, and this is a few lines:
+    let send_plugin_instructions = plugin_env.send_plugin_instructions.clone();
+    let update_target = Some(plugin_env.plugin_id);
+    thread::spawn(move || {
+        let start_time = Instant::now();
+        thread::sleep(Duration::from_secs_f64(secs));
+        // FIXME: The way that elapsed time is being calculated here is not exact; it doesn't take into account the
+        // time it takes an event to actually reach the plugin after it's sent to the `wasm` thread.
+        let elapsed_time = Instant::now().duration_since(start_time).as_secs_f64();
+
+        send_plugin_instructions
+            .send(PluginInstruction::Update(
+                update_target,
+                Event::Timer(elapsed_time),
+            ))
+            .unwrap();
+    });
+}
+
 // Helper Functions ---------------------------------------------------------------------------------------------------
 
 // FIXME: Unwrap city
-pub fn wasi_stdout(wasi_env: &WasiEnv) -> String {
+pub fn wasi_read_string(wasi_env: &WasiEnv) -> String {
     let mut state = wasi_env.state();
     let wasi_file = state.fs.stdout_mut().unwrap().as_mut().unwrap();
     let mut buf = String::new();
@@ -137,6 +168,11 @@ pub fn wasi_write_string(wasi_env: &WasiEnv, buf: &str) {
     writeln!(wasi_file, "{}\r", buf).unwrap();
 }
 
-pub fn wasi_write_json(wasi_env: &WasiEnv, object: &impl Serialize) {
+pub fn wasi_write_object(wasi_env: &WasiEnv, object: &impl Serialize) {
     wasi_write_string(wasi_env, &serde_json::to_string(&object).unwrap());
 }
+
+pub fn wasi_read_object<T: DeserializeOwned>(wasi_env: &WasiEnv) -> T {
+    let json = wasi_read_string(wasi_env);
+    serde_json::from_str(&json).unwrap()
+}