Take async channel in process_notifications()

So that we only process notifications when a message is waiting

Author: lionel-

crates/ark/src/interface.rs

@@ -88,6 +88,7 @@ use regex::Regex;
 use serde_json::json;
 use stdext::result::ResultOrLog;
 use stdext::*;
+use tokio::sync::mpsc::UnboundedReceiver as AsyncUnboundedReceiver;
 use uuid::Uuid;
 
 use crate::dap::dap::DapBackendEvent;
@@ -323,7 +324,7 @@ impl RMain {
         dap: Arc<Mutex<Dap>>,
         session_mode: SessionMode,
         default_repos: DefaultRepos,
-        graphics_device_rx: Receiver<GraphicsDeviceNotification>,
+        graphics_device_rx: AsyncUnboundedReceiver<GraphicsDeviceNotification>,
     ) {
         // Set the main thread ID.
         // Must happen before doing anything that checks `RMain::on_main_thread()`,

crates/ark/src/plots/graphics_device.rs

@@ -32,7 +32,6 @@ use amalthea::wire::update_display_data::UpdateDisplayData;
 use anyhow::anyhow;
 use base64::engine::general_purpose;
 use base64::Engine;
-use crossbeam::channel::Receiver;
 use crossbeam::channel::Select;
 use crossbeam::channel::Sender;
 use harp::exec::RFunction;
@@ -45,7 +44,7 @@ use libr::SEXP;
 use serde_json::json;
 use stdext::result::ResultOrLog;
 use stdext::unwrap;
-use tokio::task::yield_now;
+use tokio::sync::mpsc::UnboundedReceiver as AsyncUnboundedReceiver;
 use uuid::Uuid;
 
 use crate::interface::RMain;
@@ -69,40 +68,35 @@ const POSITRON_PLOT_CHANNEL_ID: &str = "positron.plot";
 pub(crate) fn init_graphics_device(
     comm_manager_tx: Sender<CommManagerEvent>,
     iopub_tx: Sender<IOPubMessage>,
-    graphics_device_rx: Receiver<GraphicsDeviceNotification>,
+    graphics_device_rx: AsyncUnboundedReceiver<GraphicsDeviceNotification>,
 ) {
     DEVICE_CONTEXT.set(DeviceContext::new(comm_manager_tx, iopub_tx));
 
     // Launch an R thread task to process messages from the frontend
     r_task::spawn_interrupt(|| async move { process_notifications(graphics_device_rx).await });
 }
 
-async fn process_notifications(graphics_device_rx: Receiver<GraphicsDeviceNotification>) {
+async fn process_notifications(
+    mut graphics_device_rx: AsyncUnboundedReceiver<GraphicsDeviceNotification>,
+) {
     log::trace!("Now listening for graphics device notifications");
 
     loop {
-        let mut i = 0;
-
-        while let Ok(notification) = graphics_device_rx.try_recv() {
+        while let Some(notification) = graphics_device_rx.recv().await {
             log::trace!("Got graphics device notification: {notification:#?}");
-            i = i + 1;
 
             match notification {
                 GraphicsDeviceNotification::DidChangePlotRenderSettings(plot_render_settings) => {
+                    // Safety: Note that `DEVICE_CONTEXT` is accessed at
+                    // interrupt time. Other methods in this file should be
+                    // written in accordance and avoid causing R interrupt
+                    // checks while they themselves access the device.
                     DEVICE_CONTEXT.with_borrow(|ctx| {
                         ctx.current_render_settings.replace(plot_render_settings)
                     });
                 },
             }
-
-            // Yield regularly to the R thread when something went wrong on the
-            // frontend side and is spamming messages
-            if i >= 5 {
-                break;
-            }
         }
-
-        yield_now().await;
     }
 }
 

crates/ark/src/shell.rs

@@ -37,6 +37,7 @@ use harp::ParseResult;
 use log::*;
 use serde_json::json;
 use stdext::unwrap;
+use tokio::sync::mpsc::UnboundedSender as AsyncUnboundedSender;
 
 use crate::help::r_help::RHelp;
 use crate::help_proxy;
@@ -56,7 +57,7 @@ pub struct Shell {
     kernel_request_tx: Sender<KernelRequest>,
     kernel_init_rx: BusReader<KernelInfo>,
     kernel_info: Option<KernelInfo>,
-    graphics_device_tx: Sender<GraphicsDeviceNotification>,
+    graphics_device_tx: AsyncUnboundedSender<GraphicsDeviceNotification>,
 }
 
 #[derive(Debug)]
@@ -72,7 +73,7 @@ impl Shell {
         stdin_request_tx: Sender<StdInRequest>,
         kernel_init_rx: BusReader<KernelInfo>,
         kernel_request_tx: Sender<KernelRequest>,
-        graphics_device_tx: Sender<GraphicsDeviceNotification>,
+        graphics_device_tx: AsyncUnboundedSender<GraphicsDeviceNotification>,
     ) -> Self {
         Self {
             comm_manager_tx,
@@ -254,7 +255,7 @@ fn handle_comm_open_ui(
     comm: CommSocket,
     stdin_request_tx: Sender<StdInRequest>,
     kernel_request_tx: Sender<KernelRequest>,
-    graphics_device_tx: Sender<GraphicsDeviceNotification>,
+    graphics_device_tx: AsyncUnboundedSender<GraphicsDeviceNotification>,
 ) -> amalthea::Result<bool> {
     // Create a frontend to wrap the comm channel we were just given. This starts
     // a thread that proxies messages to the frontend.

crates/ark/src/start.rs

@@ -72,7 +72,7 @@ pub fn start_kernel(
     // Communication channel between the graphics device (running on the R
     // thread) and the shell thread
     let (graphics_device_tx, graphics_device_rx) =
-        crossbeam::channel::unbounded::<GraphicsDeviceNotification>();
+        tokio::sync::mpsc::unbounded_channel::<GraphicsDeviceNotification>();
 
     // Create the shell.
     let kernel_init_rx = kernel_init_tx.add_rx();

crates/ark/src/ui/ui.rs

@@ -23,6 +23,7 @@ use harp::object::RObject;
 use serde_json::Value;
 use stdext::spawn;
 use stdext::unwrap;
+use tokio::sync::mpsc::UnboundedSender as AsyncUnboundedSender;
 
 use crate::plots::graphics_device::GraphicsDeviceNotification;
 use crate::r_task;
@@ -40,7 +41,7 @@ pub struct UiComm {
     comm: CommSocket,
     ui_comm_rx: Receiver<UiCommMessage>,
     stdin_request_tx: Sender<StdInRequest>,
-    graphics_device_tx: Sender<GraphicsDeviceNotification>,
+    graphics_device_tx: AsyncUnboundedSender<GraphicsDeviceNotification>,
 }
 
 // Pass a channel to the
@@ -49,7 +50,7 @@ impl UiComm {
     pub(crate) fn start(
         comm: CommSocket,
         stdin_request_tx: Sender<StdInRequest>,
-        graphics_device_tx: Sender<GraphicsDeviceNotification>,
+        graphics_device_tx: AsyncUnboundedSender<GraphicsDeviceNotification>,
     ) -> Sender<UiCommMessage> {
         // Create a sender-receiver pair for Positron global events
         let (ui_comm_tx, ui_comm_rx) = crossbeam::channel::unbounded::<UiCommMessage>();
@@ -266,7 +267,7 @@ mod tests {
         let (stdin_request_tx, _stdin_request_rx) = bounded::<StdInRequest>(1);
 
         let (graphics_device_tx, _graphics_device_rx) =
-            crossbeam::channel::unbounded::<GraphicsDeviceNotification>();
+            tokio::sync::mpsc::unbounded_channel::<GraphicsDeviceNotification>();
 
         // Create a frontend instance, get access to the sender channel
         let ui_comm_tx = UiComm::start(comm_socket.clone(), stdin_request_tx, graphics_device_tx);