Merge #833

833: automatically wait for worker threads r=matklad a=matklad

closes #817 



Co-authored-by: Aleksey Kladov <[email protected]>

3 files changed, 20 insertions, 29 deletions

diff --git a/crates/ra_lsp_server/src/main_loop.rs b/crates/ra_lsp_server/src/main_loop.rs
index a51299851..06443bb76 100644
--- a/crates/ra_lsp_server/src/main_loop.rs
+++ b/crates/ra_lsp_server/src/main_loop.rs
@@ -54,19 +54,20 @@ pub fn main_loop(
 ) -> Result<()> {
     let pool = ThreadPool::new(THREADPOOL_SIZE);
     let (task_sender, task_receiver) = unbounded::<Task>();
-    let (ws_worker, ws_watcher) = workspace_loader();
 
-    ws_worker.send(ws_root.clone()).unwrap();
     // FIXME: support dynamic workspace loading.
-    let workspaces = match ws_worker.recv().unwrap() {
-        Ok(ws) => vec![ws],
-        Err(e) => {
-            log::error!("loading workspace failed: {}", e);
-            Vec::new()
+    let workspaces = {
+        let ws_worker = workspace_loader();
+        ws_worker.sender().send(ws_root.clone()).unwrap();
+        match ws_worker.receiver().recv().unwrap() {
+            Ok(ws) => vec![ws],
+            Err(e) => {
+                log::error!("loading workspace failed: {}", e);
+                Vec::new()
+            }
         }
     };
-    ws_worker.shutdown();
-    ws_watcher.shutdown().map_err(|_| format_err!("ws watcher died"))?;
+
     let mut state = ServerWorldState::new(ws_root.clone(), workspaces);
 
     log::info!("server initialized, serving requests");
@@ -94,12 +95,9 @@ pub fn main_loop(
     log::info!("...threadpool has finished");
 
     let vfs = Arc::try_unwrap(state.vfs).expect("all snapshots should be dead");
-    let vfs_res = vfs.into_inner().shutdown();
+    drop(vfs);
 
-    main_res?;
-    vfs_res.map_err(|_| format_err!("fs watcher died"))?;
-
-    Ok(())
+    main_res
 }
 
 enum Event {
diff --git a/crates/ra_lsp_server/src/project_model.rs b/crates/ra_lsp_server/src/project_model.rs
index 6800eb138..7d6440fad 100644
--- a/crates/ra_lsp_server/src/project_model.rs
+++ b/crates/ra_lsp_server/src/project_model.rs
@@ -1,6 +1,6 @@
 use std::path::PathBuf;
 
-use thread_worker::{WorkerHandle, Worker};
+use thread_worker::Worker;
 
 use crate::Result;
 
@@ -8,8 +8,8 @@ pub use ra_project_model::{
     ProjectWorkspace, CargoWorkspace, Package, Target, TargetKind, Sysroot,
 };
 
-pub fn workspace_loader() -> (Worker<PathBuf, Result<ProjectWorkspace>>, WorkerHandle) {
-    thread_worker::spawn::<PathBuf, Result<ProjectWorkspace>, _>(
+pub fn workspace_loader() -> Worker<PathBuf, Result<ProjectWorkspace>> {
+    Worker::<PathBuf, Result<ProjectWorkspace>>::spawn(
         "workspace loader",
         1,
         |input_receiver, output_sender| {
diff --git a/crates/ra_lsp_server/tests/heavy_tests/support.rs b/crates/ra_lsp_server/tests/heavy_tests/support.rs
index eee85f8c8..11f94b4ab 100644
--- a/crates/ra_lsp_server/tests/heavy_tests/support.rs
+++ b/crates/ra_lsp_server/tests/heavy_tests/support.rs
@@ -17,7 +17,7 @@ use lsp_types::{
 use serde::Serialize;
 use serde_json::{to_string_pretty, Value};
 use tempfile::TempDir;
-use thread_worker::{WorkerHandle, Worker};
+use thread_worker::Worker;
 use test_utils::{parse_fixture, find_mismatch};
 
 use ra_lsp_server::{
@@ -45,13 +45,12 @@ pub struct Server {
     messages: RefCell<Vec<RawMessage>>,
     dir: TempDir,
     worker: Option<Worker<RawMessage, RawMessage>>,
-    watcher: Option<WorkerHandle>,
 }
 
 impl Server {
     fn new(dir: TempDir, files: Vec<(PathBuf, String)>) -> Server {
         let path = dir.path().to_path_buf();
-        let (worker, watcher) = thread_worker::spawn::<RawMessage, RawMessage, _>(
+        let worker = Worker::<RawMessage, RawMessage>::spawn(
             "test server",
             128,
             move |mut msg_receiver, mut msg_sender| {
@@ -63,7 +62,6 @@ impl Server {
             dir,
             messages: Default::default(),
             worker: Some(worker),
-            watcher: Some(watcher),
         };
 
         for (path, text) in files {
@@ -117,7 +115,7 @@ impl Server {
     }
     fn send_request_(&self, r: RawRequest) -> Value {
         let id = r.id;
-        self.worker.as_ref().unwrap().send(RawMessage::Request(r)).unwrap();
+        self.worker.as_ref().unwrap().sender().send(RawMessage::Request(r)).unwrap();
         while let Some(msg) = self.recv() {
             match msg {
                 RawMessage::Request(req) => panic!("unexpected request: {:?}", req),
@@ -157,24 +155,19 @@ impl Server {
         }
     }
     fn recv(&self) -> Option<RawMessage> {
-        recv_timeout(&self.worker.as_ref().unwrap().out).map(|msg| {
+        recv_timeout(&self.worker.as_ref().unwrap().receiver()).map(|msg| {
             self.messages.borrow_mut().push(msg.clone());
             msg
         })
     }
     fn send_notification(&self, not: RawNotification) {
-        self.worker.as_ref().unwrap().send(RawMessage::Notification(not)).unwrap();
+        self.worker.as_ref().unwrap().sender().send(RawMessage::Notification(not)).unwrap();
     }
 }
 
 impl Drop for Server {
     fn drop(&mut self) {
         self.send_request::<Shutdown>(());
-        let receiver = self.worker.take().unwrap().shutdown();
-        while let Some(msg) = recv_timeout(&receiver) {
-            drop(msg);
-        }
-        self.watcher.take().unwrap().shutdown().unwrap();
     }
 }