Use r_task() in more places

Author: lionel-

crates/ark/src/data_viewer/r_data_viewer.rs

@@ -14,7 +14,6 @@ use crossbeam::channel::Sender;
 use harp::exec::RFunction;
 use harp::exec::RFunctionExt;
 use harp::object::RObject;
-use harp::r_lock;
 use harp::utils::r_assert_length;
 use harp::utils::r_is_data_frame;
 use harp::utils::r_is_matrix;
@@ -44,6 +43,7 @@ use crate::data_viewer::message::DataViewerMessageRequest;
 use crate::data_viewer::message::DataViewerMessageResponse;
 use crate::data_viewer::message::DataViewerRowRequest;
 use crate::data_viewer::message::DataViewerRowResponse;
+use crate::r_task;
 
 pub struct RDataViewer {
     title: String,
@@ -195,7 +195,7 @@ impl DataSet {
     }
 
     pub fn from_object(id: String, title: String, object: RObject) -> Result<Self, anyhow::Error> {
-        r_lock! {
+        r_task(|| unsafe {
             let row_count = {
                 if r_is_data_frame(*object) {
                     let row_names = Rf_getAttrib(*object, R_RowNamesSymbol);
@@ -214,10 +214,10 @@ impl DataSet {
             Ok(Self {
                 id: id.clone(),
                 title: title.clone(),
-                columns: columns,
-                row_count: row_count
+                columns,
+                row_count,
             })
-        }
+        })
     }
 
     fn slice_data(&self, start: usize, size: usize) -> Result<Vec<DataColumn>, anyhow::Error> {

crates/ark/src/interface.rs

@@ -52,7 +52,6 @@ use harp::interrupts::RInterruptsSuspendedScope;
 use harp::lock::R_RUNTIME_LOCK;
 use harp::lock::R_RUNTIME_LOCK_COUNT;
 use harp::object::RObject;
-use harp::r_lock;
 use harp::r_safely;
 use harp::r_symbol;
 use harp::routines::r_register_routines;
@@ -75,6 +74,7 @@ use crate::kernel::Kernel;
 use crate::lsp::events::EVENTS;
 use crate::modules;
 use crate::plots::graphics_device;
+use crate::r_task;
 use crate::r_task::RTaskMain;
 use crate::request::debug_request_command;
 use crate::request::RRequest;
@@ -999,7 +999,7 @@ fn peek_execute_response(exec_count: u32) -> ExecuteResponse {
         data.insert("text/plain".to_string(), json!(""));
 
         // Include HTML representation of data.frame
-        r_lock! {
+        r_task(|| unsafe {
             let value = Rf_findVarInFrame(R_GlobalEnv, r_symbol!(".Last.value"));
             if r_is_data_frame(value) {
                 match to_html(value) {
@@ -1010,7 +1010,7 @@ fn peek_execute_response(exec_count: u32) -> ExecuteResponse {
                     },
                 };
             }
-        }
+        });
 
         main.iopub_tx
             .send(IOPubMessage::ExecuteResult(ExecuteResult {

crates/ark/src/lsp/backend.rs

@@ -47,6 +47,7 @@ use crate::lsp::indexer;
 use crate::lsp::signature_help::signature_help;
 use crate::lsp::statement_range;
 use crate::lsp::symbols;
+use crate::r_task;
 use crate::request::KernelRequest;
 
 #[macro_export]
@@ -372,7 +373,16 @@ impl LanguageServer for Backend {
 
         log::info!("Completion context: {:#?}", context);
 
-        // add session completions
+        // FIXME: Use `r_task()`
+        // rustc [E0277]: `*const c_void` cannot be shared between threads safely
+        // within `completions::CompletionContext<'_>`, the trait `Sync` is not implemented for `*const c_void`
+        // required for `&completions::CompletionContext<'_>` to implement `Send`
+        // rustc [E0277]: `*const tree_sitter::ffi::TSTree` cannot be shared between threads safely
+        // within `completions::CompletionContext<'_>`, the trait `Sync` is not implemented for `*const tree_sitter::ffi::TSTree`
+        // required for `&completions::CompletionContext<'_>` to implement `Send`
+        // rustc [E0277]: required because it's used within this closure
+
+        // Add session completions
         let result = r_lock! { append_session_completions(&context, &mut completions) };
         if let Err(error) = result {
             error!("{:?}", error);
@@ -444,9 +454,7 @@ impl LanguageServer for Backend {
         });
 
         // Try resolving the completion item
-        let result = r_lock! {
-            resolve_completion_item(&mut item, &data)
-        };
+        let result = r_task(|| unsafe { resolve_completion_item(&mut item, &data) });
 
         // Handle error case
         unwrap!(result, Err(error) => {
@@ -474,6 +482,16 @@ impl LanguageServer for Backend {
             return Ok(None);
         });
 
+        // FIXME: Use `r_task()`
+        // rustc [E0277]: required by a bound introduced by this call
+        // rustc [E0277]: `*const tree_sitter::ffi::TSTree` cannot be shared between threads safely
+        // within `completions::CompletionContext<'_>`, the trait `Sync` is not implemented for `*const tree_sitter::ffi::TSTree`
+        // required for `&completions::CompletionContext<'_>` to implement `Send`
+        // rustc [E0277]: `*const c_void` cannot be shared between threads safely
+        // within `completions::CompletionContext<'_>`, the trait `Sync` is not implemented for `*const c_void`
+        // required for `&completions::CompletionContext<'_>` to implement `Send`
+        // rustc [E0277]: required because it's used within this closure
+
         // request hover information
         let result = r_lock! {
             hover(&document, &context)
@@ -508,9 +526,7 @@ impl LanguageServer for Backend {
         let position = params.text_document_position_params.position;
 
         // request signature help
-        let result = r_lock! {
-            signature_help(document.value(), &position)
-        };
+        let result = r_task(|| unsafe { signature_help(document.value(), &position) });
 
         // unwrap errors
         let result = unwrap!(result, Err(error) => {

crates/ark/src/lsp/diagnostics.rs

@@ -34,6 +34,7 @@ use tree_sitter::Node;
 use crate::lsp::backend::Backend;
 use crate::lsp::documents::Document;
 use crate::lsp::indexer;
+use crate::r_task;
 use crate::Range;
 
 #[derive(Clone)]
@@ -148,11 +149,10 @@ fn generate_diagnostics(doc: &Document) -> Vec<Diagnostic> {
             _ => {},
         });
 
-        r_lock! {
+        r_task(|| unsafe {
             // Get the set of symbols currently in scope.
             let mut envir = R_GlobalEnv;
             while envir != R_EmptyEnv {
-
                 // List symbol names in this environment.
                 let mut protect = RProtect::new();
                 let objects = protect.add(R_lsInternal(envir, 1));
@@ -185,7 +185,7 @@ fn generate_diagnostics(doc: &Document) -> Vec<Diagnostic> {
                     context.installed_packages.insert(name);
                 }
             }
-        }
+        });
 
         // Start iterating through the nodes.
         let root = doc.ast.root_node();
@@ -638,6 +638,12 @@ impl<'a> TreeSitterCall<'a> {
     }
 }
 
+// FIXME: Can't directly switch to `r_task()`
+// rustc [E0277]: `*const tree_sitter::ffi::TSTree` cannot be shared between threads safely
+// within `tree_sitter::Node<'_>`, the trait `Sync` is not implemented for `*const tree_sitter::ffi::TSTree`
+// required for `&tree_sitter::Node<'_>` to implement `Send`
+// rustc [E0277]: `*const c_void` cannot be shared between threads safely
+
 fn recurse_call_arguments_custom(
     node: Node,
     context: &mut DiagnosticContext,

crates/ark/src/modules.rs

@@ -14,7 +14,6 @@ use std::time::Duration;
 use harp::exec::RFunction;
 use harp::exec::RFunctionExt;
 use harp::protect::RProtect;
-use harp::r_lock;
 use harp::r_string;
 use harp::r_symbol;
 use libR_sys::*;
@@ -23,6 +22,8 @@ use stdext::spawn;
 use stdext::unwrap;
 use stdext::unwrap::IntoResult;
 
+use crate::r_task;
+
 // We use a set of three environments for the functions exposed to
 // the R session. The environment chain is:
 //
@@ -92,11 +93,11 @@ impl RModuleWatcher {
                 for (path, oldmeta) in self.cache.iter_mut() {
                     let newmeta = path.metadata()?;
                     if oldmeta.modified()? != newmeta.modified()? {
-                        r_lock! {
-                            if let Err(error) = import(path) {
-                                log::error!("{}", error);
+                        r_task(|| {
+                            if let Err(error) = unsafe { import(path) } {
+                                log::error!("{error:?}");
                             }
-                        };
+                        });
                         *oldmeta = newmeta;
                     }
                 }

crates/ark/src/plots/graphics_device.rs

@@ -42,7 +42,6 @@ use crossbeam::channel::Select;
 use crossbeam::channel::Sender;
 use harp::exec::RFunction;
 use harp::exec::RFunctionExt;
-use harp::r_lock;
 use libR_sys::*;
 use once_cell::sync::Lazy;
 use serde_json::json;
@@ -53,6 +52,7 @@ use uuid::Uuid;
 use crate::plots::message::PlotMessageInput;
 use crate::plots::message::PlotMessageOutput;
 use crate::plots::message::PlotMessageOutputImage;
+use crate::r_task;
 
 const POSITRON_PLOT_CHANNEL_ID: &str = "positron.plot";
 
@@ -364,15 +364,15 @@ impl DeviceContext {
         // TODO: Is it possible to do this without writing to file; e.g. could
         // we instead write to a connection or something else?
         self._rendering = true;
-        let image_path = r_lock! {
+        let image_path = r_task(|| unsafe {
             RFunction::from(".ps.graphics.renderPlot")
                 .param("id", plot_id)
                 .param("width", width)
                 .param("height", height)
                 .param("dpr", pixel_ratio)
                 .call()?
                 .to::<String>()
-        };
+        });
         self._rendering = false;
 
         let image_path = unwrap!(image_path, Err(error) => {

crates/ark/src/shell.rs

@@ -42,7 +42,6 @@ use crossbeam::channel::Sender;
 use harp::exec::r_parse_vector;
 use harp::exec::ParseResult;
 use harp::object::RObject;
-use harp::r_lock;
 use libR_sys::*;
 use log::*;
 use serde_json::json;
@@ -54,6 +53,7 @@ use crate::frontend::frontend::PositronFrontend;
 use crate::interface::KernelInfo;
 use crate::kernel::Kernel;
 use crate::plots::graphics_device;
+use crate::r_task;
 use crate::request::KernelRequest;
 use crate::request::RRequest;
 
@@ -221,9 +221,7 @@ impl ShellHandler for Shell {
         &self,
         req: &IsCompleteRequest,
     ) -> Result<IsCompleteReply, Exception> {
-        r_lock! {
-            self.handle_is_complete_request_impl(req)
-        }
+        r_task(|| unsafe { self.handle_is_complete_request_impl(req) })
     }
 
     /// Handles an ExecuteRequest by sending the code to the R execution thread
@@ -294,13 +292,11 @@ impl ShellHandler for Shell {
     /// Handles a request to open a new comm channel
     async fn handle_comm_open(&self, target: Comm, comm: CommSocket) -> Result<bool, Exception> {
         match target {
-            Comm::Environment => {
-                r_lock! {
-                    let global_env = RObject::view(R_GlobalEnv);
-                    REnvironment::start(global_env, comm.clone(), self.comm_manager_tx.clone());
-                    Ok(true)
-                }
-            },
+            Comm::Environment => r_task(|| unsafe {
+                let global_env = RObject::view(R_GlobalEnv);
+                REnvironment::start(global_env, comm.clone(), self.comm_manager_tx.clone());
+                Ok(true)
+            }),
             Comm::FrontEnd => {
                 // Create a frontend to wrap the comm channel we were just given. This starts
                 // a thread that proxies messages to the frontend.

crates/ark/tests/environment.rs

@@ -42,6 +42,8 @@ fn test_environment_list() {
     // against.
     start_r();
 
+    // TODO: Switch to `r_task()`
+
     // Create a new environment for the test. We use a new, empty environment
     // (with the empty environment as its parent) so that each test in this
     // file can run independently.

crates/harp/src/exec.rs

@@ -530,6 +530,7 @@ mod tests {
 
     #[test]
     fn test_threads() {
+        // TODO: Switch to `r_task()`
         r_test_unlocked! {
 
             // Spawn a bunch of threads that try to interact with R.

crates/harp/src/object.rs

@@ -216,7 +216,7 @@ impl Drop for RObject {
 
 // RObjects are not inherently thread-safe since they wrap a raw pointer, but we
 // allow them to be sent across threads because we require the acquisition of a
-// lock on the outer R interpreter (see `r_lock!`) before using them.
+// lock on the outer R interpreter (see `r_task()`) before using them.
 unsafe impl Send for RObject {}
 unsafe impl Sync for RObject {}