Wait until 0MQ sockets are created before starting R (#43)

Author: lionel-

crates/amalthea/src/kernel.rs

@@ -105,6 +105,7 @@ impl Kernel {
         control_handler: Arc<Mutex<dyn ControlHandler>>,
         lsp_handler: Option<Arc<Mutex<dyn LspHandler>>>,
         stream_behavior: StreamBehavior,
+        conn_init_tx: Option<Sender<bool>>,
     ) -> Result<(), Error> {
         let ctx = zmq::Context::new();
 
@@ -205,6 +206,14 @@ impl Kernel {
             self.connection.endpoint(self.connection.control_port),
         )?;
 
+        // 0MQ sockets are now initialised. We can start the kernel runtime
+        // with relative multithreading safety. See
+        // https://github.com/rstudio/positron/issues/720
+        if let Some(tx) = conn_init_tx {
+            tx.send(true).unwrap();
+            drop(tx);
+        }
+
         // TODO: thread/join thread? Exiting this thread will cause the whole
         // kernel to exit.
         Self::control_thread(control_socket, control_handler);

crates/amalthea/tests/client.rs

@@ -54,7 +54,7 @@ fn test_kernel() {
 
     // Create the thread that will run the Amalthea kernel
     thread::spawn(
-        move || match kernel.connect(shell, control, None, StreamBehavior::None) {
+        move || match kernel.connect(shell, control, None, StreamBehavior::None, None) {
             Ok(_) => {
                 info!("Kernel connection initiated");
             },

crates/ark/src/main.rs

@@ -60,6 +60,8 @@ fn start_kernel(connection_file: ConnectionFile, capture_streams: bool) {
         kernel_init_tx.add_rx(),
     )));
 
+    let (conn_init_tx, conn_init_rx) = bounded::<bool>(0);
+
     // Create the shell.
     let kernel_init_rx = kernel_init_tx.add_rx();
     let shell = Shell::new(
@@ -68,6 +70,7 @@ fn start_kernel(connection_file: ConnectionFile, capture_streams: bool) {
         shell_request_rx,
         kernel_init_tx,
         kernel_init_rx,
+        conn_init_rx,
     );
 
     // Create the control handler; this is used to handle shutdown/interrupt and
@@ -83,7 +86,13 @@ fn start_kernel(connection_file: ConnectionFile, capture_streams: bool) {
 
     // Create the kernel
     let shell = Arc::new(Mutex::new(shell));
-    match kernel.connect(shell, control, Some(lsp), stream_behavior) {
+    match kernel.connect(
+        shell,
+        control,
+        Some(lsp),
+        stream_behavior,
+        Some(conn_init_tx),
+    ) {
         Ok(()) => {
             println!("R Kernel exiting.");
         },

crates/ark/src/shell.rs

@@ -67,9 +67,20 @@ impl Shell {
         shell_request_rx: Receiver<Request>,
         kernel_init_tx: Bus<KernelInfo>,
         kernel_init_rx: BusReader<KernelInfo>,
+        conn_init_rx: Receiver<bool>,
     ) -> Self {
         let iopub_tx = iopub_tx.clone();
         spawn!("ark-r-main-thread", move || {
+            // Block until 0MQ is initialised before starting R to avoid
+            // thread-safety issues. See https://github.com/rstudio/positron/issues/720
+            if let Err(err) = conn_init_rx.recv_timeout(std::time::Duration::from_secs(3)) {
+                warn!(
+                    "Failed to get init notification from main thread: {:?}",
+                    err
+                );
+            }
+            drop(conn_init_rx);
+
             Self::execution_thread(iopub_tx, kernel_init_tx, shell_request_rx);
         });
 

crates/echo/src/main.rs

@@ -37,7 +37,7 @@ fn start_kernel(connection_file: ConnectionFile) {
     let shell = Arc::new(Mutex::new(Shell::new(shell_tx)));
     let control = Arc::new(Mutex::new(Control {}));
 
-    match kernel.connect(shell, control, None, StreamBehavior::None) {
+    match kernel.connect(shell, control, None, StreamBehavior::None, None) {
         Ok(()) => {
             let mut s = String::new();
             println!("Kernel activated, press Ctrl+C to end ");