Skip to content

Move LSP's tokio::Runtime and Client to RMain #193

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
DavisVaughan merged 8 commits into from
Jan 8, 2024
Merged

Move LSP's tokio::Runtime and Client to RMain #193

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
eb510bc
Hack away enough to get a runtime global
DavisVaughan Jan 2, 2024
20210cc
Recognize that `enqueue_diagnostics()` isn't async
DavisVaughan Jan 3, 2024
1e92bf7
Use non-blocking `spawn()` to avoid deadlocking R
DavisVaughan Jan 3, 2024
20c9529
Let `start_kernel()` own the `runtime`
DavisVaughan Jan 3, 2024
0307a7a
Let `start_kernel()` build the LSP service
DavisVaughan Jan 3, 2024
f57aa38
Another bit of rationale
DavisVaughan Jan 3, 2024
f467711
Switch to `trace!()` and use `log::` everywhere
DavisVaughan Jan 8, 2024
cf346b5
Separate constructor and runner
DavisVaughan Jan 8, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
26 changes: 25 additions & 1 deletion crates/ark/src/interface.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
//
// interface.rs
//
// Copyright (C) 2023 Posit Software, PBC. All rights reserved.
// Copyright (C) 2023-2024 Posit Software, PBC. All rights reserved.
//
//

Expand Down Expand Up @@ -79,6 +79,8 @@ use regex::Regex;
use serde_json::json;
use stdext::result::ResultOrLog;
use stdext::*;
use tokio::runtime::Runtime;
use tower_lsp::Client;

use crate::dap::dap::DapBackendEvent;
use crate::dap::Dap;
Expand Down Expand Up @@ -125,6 +127,8 @@ pub fn start_r(
input_reply_rx: Receiver<InputReply>,
iopub_tx: Sender<IOPubMessage>,
kernel_init_tx: Bus<KernelInfo>,
lsp_runtime: Arc<Runtime>,
lsp_client: Client,
dap: Arc<Mutex<Dap>>,
) {
// Initialize global state (ensure we only do this once!)
Expand All @@ -145,6 +149,8 @@ pub fn start_r(
input_reply_rx,
iopub_tx,
kernel_init_tx,
lsp_runtime,
lsp_client,
dap,
));
});
Expand Down Expand Up @@ -236,6 +242,12 @@ pub struct RMain {
pub help_tx: Option<Sender<HelpRequest>>,
pub help_rx: Option<Receiver<HelpReply>>,

// LSP tokio runtime used to spawn LSP tasks on the executor and the
// corresponding client used to send LSP requests to the frontend.
// Used by R callbacks, like `ps_editor()` for `utils::file.edit()`.
lsp_runtime: Arc<Runtime>,
Copy link
Contributor

@lionel- lionel- Jan 8, 2024

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Should this be called tokio_runtime? I assume we'll share with entities other than the LSP if tokio becomes needed in other parts of ark?

Copy link
Contributor Author

@DavisVaughan DavisVaughan Jan 8, 2024

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Happy to rename this in the future, but right now I'm not 100% sure if we will use the same tokio runtime everywhere or not. For example, in help_proxy.rs we have another separate runtime already

// The help proxy main entry point.
#[tokio::main]
async fn task(target_port: u16) -> anyhow::Result<()> {
    // Create the help proxy.
    let help_proxy = HelpProxy::new(target_port)?;

    // Set the help proxy port.
    unsafe { browser::PORT = help_proxy.source_port };

    // Run the help proxy.
    Ok(help_proxy.run().await?)
}

lsp_client: Client,

dap: Arc<Mutex<Dap>>,
is_debugging: bool,

Expand Down Expand Up @@ -317,6 +329,8 @@ impl RMain {
input_reply_rx: Receiver<InputReply>,
iopub_tx: Sender<IOPubMessage>,
kernel_init_tx: Bus<KernelInfo>,
lsp_runtime: Arc<Runtime>,
lsp_client: Client,
dap: Arc<Mutex<Dap>>,
) -> Self {
Self {
Expand All @@ -338,6 +352,8 @@ impl RMain {
error_traceback: Vec::new(),
help_tx: None,
help_rx: None,
lsp_runtime,
lsp_client,
dap,
is_debugging: false,
is_busy: false,
Expand Down Expand Up @@ -971,6 +987,14 @@ impl RMain {
pub fn get_kernel(&self) -> &Arc<Mutex<Kernel>> {
&self.kernel
}

pub fn get_lsp_runtime(&self) -> &Arc<Runtime> {
&self.lsp_runtime
}

pub fn get_lsp_client(&self) -> &Client {
&self.lsp_client
}
}

/// Report an incomplete request to the front end
Expand Down
102 changes: 59 additions & 43 deletions crates/ark/src/lsp/backend.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
//
// backend.rs
//
// Copyright (C) 2022 Posit Software, PBC. All rights reserved.
// Copyright (C) 2022-2024 Posit Software, PBC. All rights reserved.
//
//

Expand All @@ -12,17 +12,18 @@ use std::sync::Arc;

use crossbeam::channel::Sender;
use dashmap::DashMap;
use log::*;
use parking_lot::Mutex;
use serde_json::Value;
use stdext::result::ResultOrLog;
use stdext::*;
use tokio::net::TcpListener;
use tokio::runtime::Runtime;
use tower_lsp::jsonrpc::Result;
use tower_lsp::lsp_types::request::GotoImplementationParams;
use tower_lsp::lsp_types::request::GotoImplementationResponse;
use tower_lsp::lsp_types::*;
use tower_lsp::Client;
use tower_lsp::ClientSocket;
use tower_lsp::LanguageServer;
use tower_lsp::LspService;
use tower_lsp::Server;
Expand All @@ -34,7 +35,6 @@ use crate::lsp::diagnostics;
use crate::lsp::document_context::DocumentContext;
use crate::lsp::documents::Document;
use crate::lsp::documents::DOCUMENT_INDEX;
use crate::lsp::globals;
use crate::lsp::help_topic;
use crate::lsp::hover::hover;
use crate::lsp::indexer;
Expand Down Expand Up @@ -89,12 +89,12 @@ impl Backend {
// then use that; otherwise, try to read the document from the provided
// path and use that instead.
let uri = unwrap!(Url::from_file_path(path), Err(_) => {
info!("couldn't construct uri from {}; reading from disk instead", path.display());
log::info!("couldn't construct uri from {}; reading from disk instead", path.display());
return fallback();
});

let document = unwrap!(self.documents.get(&uri), None => {
info!("no document for uri {}; reading from disk instead", uri);
log::info!("no document for uri {}; reading from disk instead", uri);
return fallback();
});

Expand Down Expand Up @@ -214,7 +214,7 @@ impl LanguageServer for Backend {
backend_trace!(self, "symbol({:?})", params);

let response = unwrap!(symbols::symbols(self, &params), Err(error) => {
error!("{:?}", error);
log::error!("{:?}", error);
return Ok(None);
});

Expand All @@ -228,7 +228,7 @@ impl LanguageServer for Backend {
backend_trace!(self, "document_symbols({})", params.text_document.uri);

let response = unwrap!(symbols::document_symbols(self, &params), Err(error) => {
error!("{:?}", error);
log::error!("{:?}", error);
return Ok(None);
});

Expand Down Expand Up @@ -280,15 +280,15 @@ impl LanguageServer for Backend {
if let Ok(path) = uri.to_file_path() {
let path = Path::new(&path);
if let Err(error) = indexer::update(&doc, &path) {
error!("{:?}", error);
log::error!("{:?}", error);
}
}

// publish diagnostics - but only publish them if the version of
// the document now matches the version of the change after applying
// it in `on_did_change()`
if params.text_document.version == version {
diagnostics::enqueue_diagnostics(self.clone(), uri.clone(), version).await;
diagnostics::enqueue_diagnostics(self.clone(), uri.clone(), version);
}
}

Expand Down Expand Up @@ -382,7 +382,7 @@ impl LanguageServer for Backend {

// unwrap errors
let result = unwrap!(result, Err(error) => {
error!("{:?}", error);
log::error!("{:?}", error);
return Ok(None);
});

Expand Down Expand Up @@ -413,7 +413,7 @@ impl LanguageServer for Backend {

// unwrap errors
let result = unwrap!(result, Err(error) => {
error!("{:?}", error);
log::error!("{:?}", error);
return Ok(None);
});

Expand All @@ -440,7 +440,7 @@ impl LanguageServer for Backend {

// build goto definition context
let result = unwrap!(unsafe { goto_definition(&document, params) }, Err(error) => {
error!("{}", error);
log::error!("{}", error);
return Ok(None);
});

Expand All @@ -453,7 +453,7 @@ impl LanguageServer for Backend {
) -> Result<Option<GotoImplementationResponse>> {
backend_trace!(self, "goto_implementation({:?})", params);
let _ = params;
error!("Got a textDocument/implementation request, but it is not implemented");
log::error!("Got a textDocument/implementation request, but it is not implemented");
return Ok(None);
}

Expand Down Expand Up @@ -492,39 +492,57 @@ impl LanguageServer for Backend {
// https://github.com/Microsoft/vscode-languageserver-node/blob/18fad46b0e8085bb72e1b76f9ea23a379569231a/client/src/common/client.ts#L701-L752
impl Backend {
async fn notification(&self, params: Option<Value>) {
info!("Received Positron notification: {:?}", params);
log::info!("Received Positron notification: {:?}", params);
}
}

#[tokio::main]
Copy link
Contributor Author

@DavisVaughan DavisVaughan Jan 3, 2024

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

No more attribute because we manage this manually now by creating the Runtime elsewhere and calling block_on() ourselves

Copy link
Contributor

@lionel- lionel- Jan 8, 2024

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Cool, this attribute seems a bit too magic/mysterious when you're not familiar with tokio

pub async fn start_lsp(address: String, conn_init_tx: Sender<bool>) {
#[cfg(feature = "runtime-agnostic")]
use tokio_util::compat::TokioAsyncReadCompatExt;
#[cfg(feature = "runtime-agnostic")]
use tokio_util::compat::TokioAsyncWriteCompatExt;

debug!("Connecting to LSP at '{}'", &address);
let listener = TcpListener::bind(&address).await.unwrap();

// Notify frontend that we are ready to accept connections
conn_init_tx
.send(true)
.or_log_warning("Couldn't send LSP server init notification");

let (stream, _) = listener.accept().await.unwrap();
debug!("Connected to LSP at '{}'", address);
let (read, write) = tokio::io::split(stream);
pub fn start_lsp(
runtime: Arc<Runtime>,
service: LspService<Backend>,
socket: ClientSocket,
address: String,
conn_init_tx: Sender<bool>,
) {
runtime.block_on(async {
#[cfg(feature = "runtime-agnostic")]
use tokio_util::compat::TokioAsyncReadCompatExt;
#[cfg(feature = "runtime-agnostic")]
use tokio_util::compat::TokioAsyncWriteCompatExt;

log::trace!("Connecting to LSP at '{}'", &address);
let listener = TcpListener::bind(&address).await.unwrap();

// Notify frontend that we are ready to accept connections
conn_init_tx
.send(true)
.or_log_warning("Couldn't send LSP server init notification");

let (stream, _) = listener.accept().await.unwrap();
log::trace!("Connected to LSP at '{}'", address);
let (read, write) = tokio::io::split(stream);

#[cfg(feature = "runtime-agnostic")]
let (read, write) = (read.compat(), write.compat_write());

let server = Server::new(read, write, socket);
server.serve(service).await;

log::trace!(
"LSP thread exiting gracefully after connection closed ({:?}).",
address
);
})
}

#[cfg(feature = "runtime-agnostic")]
let (read, write) = (read.compat(), write.compat_write());
pub fn build_lsp_service() -> (LspService<Backend>, ClientSocket, Client) {
Copy link
Contributor Author

@DavisVaughan DavisVaughan Jan 3, 2024

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This encapsulates the "trick" for getting access to the Client in a way that can be sent along to start_r()

let mut client = None;

let init = |client: Client| {
// initialize shared globals (needed for R callbacks)
globals::initialize(client.clone());
let init = |client_callback: Client| {
client = Some(client_callback.clone());

// create backend
let backend = Backend {
client,
client: client_callback,
documents: DOCUMENT_INDEX.clone(),
workspace: Arc::new(Mutex::new(Workspace::default())),
};
Expand All @@ -541,9 +559,7 @@ pub async fn start_lsp(address: String, conn_init_tx: Sender<bool>) {
.custom_method("positron/notification", Backend::notification)
.finish();

Server::new(read, write, socket).serve(service).await;
debug!(
"LSP thread exiting gracefully after connection closed ({:?}).",
address
);
let client = client.expect("`Client` should be initialized by the `build()` callback.");

(service, socket, client)
}
2 changes: 1 addition & 1 deletion crates/ark/src/lsp/diagnostics.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -85,7 +85,7 @@ impl<'a> DiagnosticContext<'a> {
}
}

pub async fn enqueue_diagnostics(backend: Backend, uri: Url, version: i32) {
pub fn enqueue_diagnostics(backend: Backend, uri: Url, version: i32) {
Copy link
Contributor Author

@DavisVaughan DavisVaughan Jan 3, 2024

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Happened to realize along the way that this doesn't need to be async

Copy link
Contributor

@lionel- lionel- Jan 8, 2024

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I was also thinking we could remove the unnecessary block_on() at some point since they complicate call stacks with tokio stuff.

// log::trace!("[diagnostics({version}, {uri})] Spawning task to enqueue diagnostics.");

// Spawn a task to enqueue diagnostics.
Expand Down
Loading