5a558e0 · helix · git.iko.soy

git.iko.soy/helix

on master help sign in

lsp: Delay requests & notifications until initialization is complete

5a558e0Blaž Hrastnik5 years ago

helix-lsp/src/client.rs+10−5

compare with master

unifiedsplit

@@ -9,13 +9,16 @@ use lsp_types as lsp;

9use serde_json::Value;

10use std::future::Future;

11use std::process::Stdio;

12+use std::sync::{

12−use std::sync::atomic::{AtomicU64, Ordering};

13+ atomic::{AtomicU64, Ordering},

14+ Arc,

15+};

13use tokio::{

16use tokio::{

14 io::{BufReader, BufWriter},

17 io::{BufReader, BufWriter},

15 process::{Child, Command},

18 process::{Child, Command},

16 sync::{

19 sync::{

17 mpsc::{channel, UnboundedReceiver, UnboundedSender},

20 mpsc::{channel, UnboundedReceiver, UnboundedSender},

18− OnceCell,

21+ Notify, OnceCell,

19 },

22 },

20};

23};

@@ -31,12 +34,13 @@ pub struct Client {

31}

34}

33impl Client {

36impl Client {

37+ #[allow(clippy::type_complexity)]

34 pub fn start(

38 pub fn start(

35 cmd: &str,

39 cmd: &str,

36 args: &[String],

40 args: &[String],

37 config: Option<Value>,

41 config: Option<Value>,

38 id: usize,

42 id: usize,

39− ) -> Result<(Self, UnboundedReceiver<(usize, Call)>)> {

43+ ) -> Result<(Self, UnboundedReceiver<(usize, Call)>, Arc<Notify>)> {

40 let process = Command::new(cmd)

44 let process = Command::new(cmd)

41 .args(args)

45 .args(args)

42 .stdin(Stdio::piped())

46 .stdin(Stdio::piped())

@@ -53,7 +57,8 @@ impl Client {

53 let reader = BufReader::new(process.stdout.take().expect("Failed to open stdout"));

57 let reader = BufReader::new(process.stdout.take().expect("Failed to open stdout"));

54 let stderr = BufReader::new(process.stderr.take().expect("Failed to open stderr"));

58 let stderr = BufReader::new(process.stderr.take().expect("Failed to open stderr"));

60+ let (server_rx, server_tx, initialize_notify) =

56− let (server_rx, server_tx) = Transport::start(reader, writer, stderr, id);

61+ Transport::start(reader, writer, stderr, id);

58 let client = Self {

63 let client = Self {

59 id,

64 id,

@@ -65,7 +70,7 @@ impl Client {

65 config,

70 config,

66 };

71 };

68− Ok((client, server_rx))

73+ Ok((client, server_rx, initialize_notify))

69 }

74 }

71 pub fn id(&self) -> usize {

76 pub fn id(&self) -> usize {

helix-lsp/src/lib.rs+5−6

@@ -312,7 +312,7 @@ impl Registry {

312 Entry::Vacant(entry) => {

313 // initialize a new client

314 let id = self.counter.fetch_add(1, Ordering::Relaxed);

315− let (client, incoming) = Client::start(

315+ let (client, incoming, initialize_notify) = Client::start(

316 &config.command,

317 &config.args,

318 serde_json::from_str(language_config.config.as_deref().unwrap_or("")).ok(),

@@ -322,9 +322,9 @@ impl Registry {

322 let client = Arc::new(client);

323

324 let _client = client.clone();

325+ // Initialize the client asynchronously

325− let initialize = tokio::spawn(async move {

326+ tokio::spawn(async move {

326 use futures_util::TryFutureExt;

327 use futures_util::TryFutureExt;

327−

328 let value = _client

329 .capabilities

330 .get_or_try_init(|| {

@@ -341,10 +341,9 @@ impl Registry {

341 .notify::<lsp::notification::Initialized>(lsp::InitializedParams {})

342 .await

343 .unwrap();

344− });

345

344

346− // TODO: remove this block

345+ initialize_notify.notify_one();

347− futures_executor::block_on(initialize).map_err(|_| anyhow::anyhow!("bail"))?;

346+ });

348

347

349 entry.insert((id, client.clone()));

348 entry.insert((id, client.clone()));

350 Ok(client)

349 Ok(client)

helix-lsp/src/transport.rs+76−15

@@ -1,7 +1,7 @@

1use crate::{Error, Result};

2use anyhow::Context;

3use jsonrpc_core as jsonrpc;

4−use log::{debug, error, info, warn};

4+use log::{error, info};

5use serde::{Deserialize, Serialize};

6use serde_json::Value;

7use std::collections::HashMap;

@@ -11,7 +11,7 @@ use tokio::{

11 process::{ChildStderr, ChildStdin, ChildStdout},

12 sync::{

13 mpsc::{unbounded_channel, Sender, UnboundedReceiver, UnboundedSender},

14− Mutex,

14+ Mutex, Notify,

15 },

16};

@@ -51,9 +51,11 @@ impl Transport {

51 ) -> (

52 UnboundedReceiver<(usize, jsonrpc::Call)>,

53 UnboundedSender<Payload>,

54+ Arc<Notify>,

54 ) {

55 ) {

55 let (client_tx, rx) = unbounded_channel();

56 let (client_tx, rx) = unbounded_channel();

56 let (tx, client_rx) = unbounded_channel();

57 let (tx, client_rx) = unbounded_channel();

58+ let notify = Arc::new(Notify::new());

58 let transport = Self {

60 let transport = Self {

59 id,

61 id,

@@ -64,9 +66,14 @@ impl Transport {

65 tokio::spawn(Self::recv(transport.clone(), server_stdout, client_tx));

67 tokio::spawn(Self::recv(transport.clone(), server_stdout, client_tx));

66 tokio::spawn(Self::err(transport.clone(), server_stderr));

68 tokio::spawn(Self::err(transport.clone(), server_stderr));

67− tokio::spawn(Self::send(transport, server_stdin, client_rx));

69+ tokio::spawn(Self::send(

68−

70+ transport,

71+ server_stdin,

72+ client_rx,

73+ notify.clone(),

74+ ));

75+

69− (rx, tx)

76+ (rx, tx, notify)

70 }

77 }

72 async fn recv_server_message(

79 async fn recv_server_message(

@@ -82,7 +89,8 @@ impl Transport {

83 // debug!("<- header {:?}", buffer);

90 // debug!("<- header {:?}", buffer);

85− if header.is_empty() {

92+ if buffer == "\r\n" {

93+ // look for an empty CRLF line

86 break;

94 break;

87 }

95 }

@@ -99,7 +107,8 @@ impl Transport {

99 // Workaround: Some non-conformant language servers will output logging and other garbage

107 // Workaround: Some non-conformant language servers will output logging and other garbage

100 // into the same stream as JSON-RPC messages. This can also happen from shell scripts that spawn

108 // into the same stream as JSON-RPC messages. This can also happen from shell scripts that spawn

101 // the server. Skip such lines and log a warning.

109 // the server. Skip such lines and log a warning.

110+

102− warn!("Failed to parse header: {:?}", header);

111+ // warn!("Failed to parse header: {:?}", header);

103 }

112 }

104 }

113 }

105 }

114 }

@@ -261,15 +270,67 @@ impl Transport {

261 transport: Arc<Self>,

270 transport: Arc<Self>,

262 mut server_stdin: BufWriter<ChildStdin>,

271 mut server_stdin: BufWriter<ChildStdin>,

263 mut client_rx: UnboundedReceiver<Payload>,

272 mut client_rx: UnboundedReceiver<Payload>,

273+ initialize_notify: Arc<Notify>,

264 ) {

274 ) {

275+ let mut pending_messages: Vec<Payload> = Vec::new();

276+ let mut is_pending = true;

277+

278+ // Determine if a message is allowed to be sent early

279+ fn is_initialize(payload: &Payload) -> bool {

280+ use lsp_types::{

281+ notification::{Initialized, Notification},

282+ request::{Initialize, Request},

283+ };

284+ match payload {

285+ Payload::Request {

286+ value: jsonrpc::MethodCall { method, .. },

287+ ..

288+ } if method == Initialize::METHOD => true,

289+ Payload::Notification(jsonrpc::Notification { method, .. })

290+ if method == Initialized::METHOD =>

291+ {

292+ true

293+ }

294+ _ => false,

295+ }

296+ }

297+

298+ // TODO: events that use capabilities need to do the right thing

299+

300+ loop {

301+ tokio::select! {

302+ biased;

303+ _ = initialize_notify.notified() => { // TODO: notified is technically not cancellation safe

304+ // server successfully initialized

305+ is_pending = false;

306+ // drain the pending queue and send payloads to server

307+ for msg in pending_messages.drain(..) {

308+ log::info!("Draining pending message {:?}", msg);

309+ match transport.send_payload_to_server(&mut server_stdin, msg).await {

310+ Ok(_) => {}

311+ Err(err) => {

312+ error!("err: <- {:?}", err);

313+ }

314+ }

315+ }

316+ }

265− while let Some(msg) = client_rx.recv().await {

317+ msg = client_rx.recv() => {

266− match transport

318+ if let Some(msg) = msg {

319+ if is_pending && !is_initialize(&msg) {

320+ log::info!("Language server not initialized, delaying request");

321+ pending_messages.push(msg);

322+ } else {

267− .send_payload_to_server(&mut server_stdin, msg)

323+ match transport.send_payload_to_server(&mut server_stdin, msg).await {

268− .await

269− {

270− Ok(_) => {}

324+ Ok(_) => {}

271− Err(err) => {

325+ Err(err) => {

272− error!("err: <- {:?}", err);

326+ error!("err: <- {:?}", err);

327+ }

328+ }

329+ }

330+ } else {

331+ // channel closed

332+ break;

333+ }

273 }

334 }

274 }

335 }

275 }

336 }