Merge #377

377: update crossbeam r=matklad a=matklad



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

13 files changed, 70 insertions, 74 deletions

diff --git a/Cargo.lock b/Cargo.lock
index 8e962d352..0d63ec639 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -166,13 +166,12 @@ dependencies = [
 
 [[package]]
 name = "crossbeam-channel"
-version = "0.2.6"
+version = "0.3.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 dependencies = [
- "crossbeam-epoch 0.6.1 (registry+https://github.com/rust-lang/crates.io-index)",
- "crossbeam-utils 0.5.0 (registry+https://github.com/rust-lang/crates.io-index)",
- "parking_lot 0.6.4 (registry+https://github.com/rust-lang/crates.io-index)",
- "rand 0.5.5 (registry+https://github.com/rust-lang/crates.io-index)",
+ "crossbeam-utils 0.6.3 (registry+https://github.com/rust-lang/crates.io-index)",
+ "parking_lot 0.7.0 (registry+https://github.com/rust-lang/crates.io-index)",
+ "rand 0.6.1 (registry+https://github.com/rust-lang/crates.io-index)",
  "smallvec 0.6.7 (registry+https://github.com/rust-lang/crates.io-index)",
 ]
 
@@ -200,19 +199,6 @@ dependencies = [
 ]
 
 [[package]]
-name = "crossbeam-epoch"
-version = "0.6.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-dependencies = [
- "arrayvec 0.4.10 (registry+https://github.com/rust-lang/crates.io-index)",
- "cfg-if 0.1.6 (registry+https://github.com/rust-lang/crates.io-index)",
- "crossbeam-utils 0.6.3 (registry+https://github.com/rust-lang/crates.io-index)",
- "lazy_static 1.2.0 (registry+https://github.com/rust-lang/crates.io-index)",
- "memoffset 0.2.1 (registry+https://github.com/rust-lang/crates.io-index)",
- "scopeguard 0.3.3 (registry+https://github.com/rust-lang/crates.io-index)",
-]
-
-[[package]]
 name = "crossbeam-utils"
 version = "0.2.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -222,11 +208,6 @@ dependencies = [
 
 [[package]]
 name = "crossbeam-utils"
-version = "0.5.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-
-[[package]]
-name = "crossbeam-utils"
 version = "0.6.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 dependencies = [
@@ -355,7 +336,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 name = "gen_lsp_server"
 version = "0.1.0"
 dependencies = [
- "crossbeam-channel 0.2.6 (registry+https://github.com/rust-lang/crates.io-index)",
+ "crossbeam-channel 0.3.6 (registry+https://github.com/rust-lang/crates.io-index)",
  "failure 0.1.3 (registry+https://github.com/rust-lang/crates.io-index)",
  "languageserver-types 0.53.1 (registry+https://github.com/rust-lang/crates.io-index)",
  "log 0.4.6 (registry+https://github.com/rust-lang/crates.io-index)",
@@ -764,7 +745,7 @@ name = "ra_lsp_server"
 version = "0.1.0"
 dependencies = [
  "cargo_metadata 0.6.4 (registry+https://github.com/rust-lang/crates.io-index)",
- "crossbeam-channel 0.2.6 (registry+https://github.com/rust-lang/crates.io-index)",
+ "crossbeam-channel 0.3.6 (registry+https://github.com/rust-lang/crates.io-index)",
  "drop_bomb 0.1.4 (registry+https://github.com/rust-lang/crates.io-index)",
  "failure 0.1.3 (registry+https://github.com/rust-lang/crates.io-index)",
  "failure_derive 0.1.3 (registry+https://github.com/rust-lang/crates.io-index)",
@@ -824,7 +805,7 @@ dependencies = [
 name = "ra_vfs"
 version = "0.1.0"
 dependencies = [
- "crossbeam-channel 0.2.6 (registry+https://github.com/rust-lang/crates.io-index)",
+ "crossbeam-channel 0.3.6 (registry+https://github.com/rust-lang/crates.io-index)",
  "log 0.4.6 (registry+https://github.com/rust-lang/crates.io-index)",
  "relative-path 0.4.0 (registry+https://github.com/rust-lang/crates.io-index)",
  "rustc-hash 1.0.1 (registry+https://github.com/rust-lang/crates.io-index)",
@@ -1308,7 +1289,7 @@ dependencies = [
 name = "thread_worker"
 version = "0.1.0"
 dependencies = [
- "crossbeam-channel 0.2.6 (registry+https://github.com/rust-lang/crates.io-index)",
+ "crossbeam-channel 0.3.6 (registry+https://github.com/rust-lang/crates.io-index)",
  "drop_bomb 0.1.4 (registry+https://github.com/rust-lang/crates.io-index)",
  "log 0.4.6 (registry+https://github.com/rust-lang/crates.io-index)",
 ]
@@ -1543,12 +1524,10 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 "checksum chrono 0.4.6 (registry+https://github.com/rust-lang/crates.io-index)" = "45912881121cb26fad7c38c17ba7daa18764771836b34fab7d3fbd93ed633878"
 "checksum clap 2.32.0 (registry+https://github.com/rust-lang/crates.io-index)" = "b957d88f4b6a63b9d70d5f454ac8011819c6efa7727858f458ab71c756ce2d3e"
 "checksum cloudabi 0.0.3 (registry+https://github.com/rust-lang/crates.io-index)" = "ddfc5b9aa5d4507acaf872de71051dfd0e309860e88966e1051e462a077aac4f"
-"checksum crossbeam-channel 0.2.6 (registry+https://github.com/rust-lang/crates.io-index)" = "7b85741761b7f160bc5e7e0c14986ef685b7f8bf9b7ad081c60c604bb4649827"
+"checksum crossbeam-channel 0.3.6 (registry+https://github.com/rust-lang/crates.io-index)" = "137bc235f622ffaa0428e3854e24acb53291fc0b3ff6fb2cb75a8be6fb02f06b"
 "checksum crossbeam-deque 0.2.0 (registry+https://github.com/rust-lang/crates.io-index)" = "f739f8c5363aca78cfb059edf753d8f0d36908c348f3d8d1503f03d8b75d9cf3"
 "checksum crossbeam-epoch 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)" = "927121f5407de9956180ff5e936fe3cf4324279280001cd56b669d28ee7e9150"
-"checksum crossbeam-epoch 0.6.1 (registry+https://github.com/rust-lang/crates.io-index)" = "2449aaa4ec7ef96e5fb24db16024b935df718e9ae1cec0a1e68feeca2efca7b8"
 "checksum crossbeam-utils 0.2.2 (registry+https://github.com/rust-lang/crates.io-index)" = "2760899e32a1d58d5abb31129f8fae5de75220bc2176e77ff7c627ae45c918d9"
-"checksum crossbeam-utils 0.5.0 (registry+https://github.com/rust-lang/crates.io-index)" = "677d453a17e8bd2b913fa38e8b9cf04bcdbb5be790aa294f2389661d72036015"
 "checksum crossbeam-utils 0.6.3 (registry+https://github.com/rust-lang/crates.io-index)" = "41ee4864f4797060e52044376f7d107429ce1fb43460021b126424b7180ee21a"
 "checksum derive-new 0.5.6 (registry+https://github.com/rust-lang/crates.io-index)" = "6ca414e896ae072546f4d789f452daaecf60ddee4c9df5dc6d5936d769e3d87c"
 "checksum deunicode 0.4.3 (registry+https://github.com/rust-lang/crates.io-index)" = "850878694b7933ca4c9569d30a34b55031b9b139ee1fc7b94a527c4ef960d690"
diff --git a/crates/gen_lsp_server/Cargo.toml b/crates/gen_lsp_server/Cargo.toml
index afeeb1513..2aee4ea16 100644
--- a/crates/gen_lsp_server/Cargo.toml
+++ b/crates/gen_lsp_server/Cargo.toml
@@ -13,4 +13,4 @@ log = "0.4.3"
 failure = "0.1.2"
 serde_json = "1.0.24"
 serde = { version = "1.0.83", features = ["derive"] }
-crossbeam-channel = "0.2.4"
+crossbeam-channel = "0.3.5"
diff --git a/crates/gen_lsp_server/src/lib.rs b/crates/gen_lsp_server/src/lib.rs
index 8779fbf0f..b20652928 100644
--- a/crates/gen_lsp_server/src/lib.rs
+++ b/crates/gen_lsp_server/src/lib.rs
@@ -95,7 +95,7 @@ pub fn run_server(
     server(params, &receiver, &sender)?;
     log::info!("lsp server waiting for exit notification");
     match receiver.recv() {
-        Some(RawMessage::Notification(n)) => n
+        Ok(RawMessage::Notification(n)) => n
             .cast::<Exit>()
             .map_err(|n| format_err!("unexpected notification during shutdown: {:?}", n))?,
         m => bail!("unexpected message during shutdown: {:?}", m),
@@ -109,7 +109,7 @@ pub fn handle_shutdown(req: RawRequest, sender: &Sender<RawMessage>) -> Option<R
     match req.cast::<Shutdown>() {
         Ok((id, ())) => {
             let resp = RawResponse::ok::<Shutdown>(id, &());
-            sender.send(RawMessage::Response(resp));
+            let _ = sender.send(RawMessage::Response(resp));
             None
         }
         Err(req) => Some(req),
@@ -122,16 +122,16 @@ fn initialize(
     caps: ServerCapabilities,
 ) -> Result<InitializeParams> {
     let (id, params) = match receiver.recv() {
-        Some(RawMessage::Request(req)) => match req.cast::<Initialize>() {
+        Ok(RawMessage::Request(req)) => match req.cast::<Initialize>() {
             Err(req) => bail!("expected initialize request, got {:?}", req),
             Ok(req) => req,
         },
         msg => bail!("expected initialize request, got {:?}", msg),
     };
     let resp = RawResponse::ok::<Initialize>(id, &InitializeResult { capabilities: caps });
-    sender.send(RawMessage::Response(resp));
+    sender.send(RawMessage::Response(resp)).unwrap();
     match receiver.recv() {
-        Some(RawMessage::Notification(n)) => {
+        Ok(RawMessage::Notification(n)) => {
             n.cast::<Initialized>()
                 .map_err(|_| format_err!("expected initialized notification"))?;
         }
diff --git a/crates/gen_lsp_server/src/stdio.rs b/crates/gen_lsp_server/src/stdio.rs
index 35d8e46d0..5c8e33854 100644
--- a/crates/gen_lsp_server/src/stdio.rs
+++ b/crates/gen_lsp_server/src/stdio.rs
@@ -9,11 +9,13 @@ use failure::bail;
 use crate::{RawMessage, Result};
 
 pub fn stdio_transport() -> (Receiver<RawMessage>, Sender<RawMessage>, Threads) {
-    let (writer_sender, mut writer_receiver) = bounded::<RawMessage>(16);
+    let (writer_sender, writer_receiver) = bounded::<RawMessage>(16);
     let writer = thread::spawn(move || {
         let stdout = stdout();
         let mut stdout = stdout.lock();
-        writer_receiver.try_for_each(|it| it.write(&mut stdout))?;
+        writer_receiver
+            .into_iter()
+            .try_for_each(|it| it.write(&mut stdout))?;
         Ok(())
     });
     let (reader_sender, reader_receiver) = bounded::<RawMessage>(16);
@@ -21,7 +23,9 @@ pub fn stdio_transport() -> (Receiver<RawMessage>, Sender<RawMessage>, Threads)
         let stdin = stdin();
         let mut stdin = stdin.lock();
         while let Some(msg) = RawMessage::read(&mut stdin)? {
-            reader_sender.send(msg);
+            if let Err(_) = reader_sender.send(msg) {
+                break;
+            }
         }
         Ok(())
     });
diff --git a/crates/ra_lsp_server/Cargo.toml b/crates/ra_lsp_server/Cargo.toml
index 646df2497..b1e8987fe 100644
--- a/crates/ra_lsp_server/Cargo.toml
+++ b/crates/ra_lsp_server/Cargo.toml
@@ -13,7 +13,7 @@ failure_derive = "0.1.2"
 serde_json = "1.0.24"
 serde = "1.0.83"
 drop_bomb = "0.1.0"
-crossbeam-channel = "0.2.4"
+crossbeam-channel = "0.3.5"
 flexi_logger = "0.10.0"
 log = "0.4.3"
 url_serde = "0.2.0"
diff --git a/crates/ra_lsp_server/src/main_loop.rs b/crates/ra_lsp_server/src/main_loop.rs
index 97c1be778..3ebae4ecd 100644
--- a/crates/ra_lsp_server/src/main_loop.rs
+++ b/crates/ra_lsp_server/src/main_loop.rs
@@ -7,7 +7,7 @@ use std::{
     sync::Arc,
 };
 
-use crossbeam_channel::{unbounded, select, Receiver, Sender};
+use crossbeam_channel::{unbounded, select, Receiver, Sender, RecvError};
 use gen_lsp_server::{
     handle_shutdown, ErrorCode, RawMessage, RawNotification, RawRequest, RawResponse,
 };
@@ -62,7 +62,7 @@ pub fn main_loop(
     let (task_sender, task_receiver) = unbounded::<Task>();
     let (ws_worker, ws_watcher) = workspace_loader();
 
-    ws_worker.send(ws_root.clone());
+    ws_worker.send(ws_root.clone()).unwrap();
     // FIXME: support dynamic workspace loading.
     let workspaces = match ws_worker.recv().unwrap() {
         Ok(ws) => vec![ws],
@@ -95,7 +95,9 @@ pub fn main_loop(
     );
 
     log::info!("waiting for tasks to finish...");
-    task_receiver.for_each(|task| on_task(task, msg_sender, &mut pending_requests));
+    task_receiver
+        .into_iter()
+        .for_each(|task| on_task(task, msg_sender, &mut pending_requests));
     log::info!("...tasks have finished");
     log::info!("joining threadpool...");
     drop(pool);
@@ -170,16 +172,16 @@ fn main_loop_inner(
     loop {
         log::trace!("selecting");
         let event = select! {
-            recv(msg_receiver, msg) => match msg {
-                Some(msg) => Event::Msg(msg),
-                None => bail!("client exited without shutdown"),
+            recv(msg_receiver) -> msg => match msg {
+                Ok(msg) => Event::Msg(msg),
+                Err(RecvError) => bail!("client exited without shutdown"),
             },
-            recv(task_receiver, task) => Event::Task(task.unwrap()),
-            recv(state.vfs.read().task_receiver(), task) => match task {
-                None => bail!("vfs died"),
-                Some(task) => Event::Vfs(task),
-            }
-            recv(libdata_receiver, data) => Event::Lib(data.unwrap())
+            recv(task_receiver) -> task => Event::Task(task.unwrap()),
+            recv(state.vfs.read().task_receiver()) -> task => match task {
+                Ok(task) => Event::Vfs(task),
+                Err(RecvError) => bail!("vfs died"),
+            },
+            recv(libdata_receiver) -> data => Event::Lib(data.unwrap())
         };
         log::info!("loop_turn = {:?}", event);
         let start = std::time::Instant::now();
@@ -209,7 +211,7 @@ fn main_loop_inner(
                                 ErrorCode::MethodNotFound as i32,
                                 "unknown request".to_string(),
                             );
-                            msg_sender.send(RawMessage::Response(resp))
+                            msg_sender.send(RawMessage::Response(resp)).unwrap()
                         }
                     }
                 }
@@ -229,7 +231,7 @@ fn main_loop_inner(
                 log::info!("indexing {:?} ... ", root);
                 let data = LibraryData::prepare(root, files);
                 log::info!("indexed {:?} {:?}", start.elapsed(), root);
-                sender.send(data);
+                sender.send(data).unwrap();
             });
         }
         if state.roots_to_scan == 0 {
@@ -253,10 +255,12 @@ fn on_task(task: Task, msg_sender: &Sender<RawMessage>, pending_requests: &mut F
     match task {
         Task::Respond(response) => {
             if pending_requests.remove(&response.id) {
-                msg_sender.send(RawMessage::Response(response))
+                msg_sender.send(RawMessage::Response(response)).unwrap();
             }
         }
-        Task::Notify(n) => msg_sender.send(RawMessage::Notification(n)),
+        Task::Notify(n) => {
+            msg_sender.send(RawMessage::Notification(n)).unwrap();
+        }
     }
 }
 
@@ -328,7 +332,7 @@ fn on_notification(
                     ErrorCode::RequestCancelled as i32,
                     "canceled by client".to_string(),
                 );
-                msg_sender.send(RawMessage::Response(response))
+                msg_sender.send(RawMessage::Response(response)).unwrap()
             }
             return Ok(());
         }
@@ -381,7 +385,7 @@ fn on_notification(
                 diagnostics: Vec::new(),
             };
             let not = RawNotification::new::<req::PublishDiagnostics>(&params);
-            msg_sender.send(RawMessage::Notification(not));
+            msg_sender.send(RawMessage::Notification(not)).unwrap();
             return Ok(());
         }
         Err(not) => not,
@@ -441,7 +445,7 @@ impl<'a> PoolDispatcher<'a> {
                         },
                     };
                     let task = Task::Respond(resp);
-                    sender.send(task);
+                    sender.send(task).unwrap();
                 });
                 self.res = Some(id);
             }
@@ -476,7 +480,7 @@ fn update_file_notifications_on_threadpool(
                 }
                 Ok(params) => {
                     let not = RawNotification::new::<req::PublishDiagnostics>(&params);
-                    sender.send(Task::Notify(not));
+                    sender.send(Task::Notify(not)).unwrap();
                 }
             }
             if publish_decorations {
@@ -488,7 +492,7 @@ fn update_file_notifications_on_threadpool(
                     }
                     Ok(params) => {
                         let not = RawNotification::new::<req::PublishDecorations>(&params);
-                        sender.send(Task::Notify(not))
+                        sender.send(Task::Notify(not)).unwrap();
                     }
                 }
             }
@@ -501,7 +505,7 @@ fn feedback(intrnal_mode: bool, msg: &str, sender: &Sender<RawMessage>) {
         return;
     }
     let not = RawNotification::new::<req::InternalFeedback>(&msg.to_string());
-    sender.send(RawMessage::Notification(not));
+    sender.send(RawMessage::Notification(not)).unwrap();
 }
 
 fn is_canceled(e: &failure::Error) -> bool {
diff --git a/crates/ra_lsp_server/src/project_model.rs b/crates/ra_lsp_server/src/project_model.rs
index 5852a157d..ae2149463 100644
--- a/crates/ra_lsp_server/src/project_model.rs
+++ b/crates/ra_lsp_server/src/project_model.rs
@@ -204,8 +204,10 @@ pub fn workspace_loader() -> (Worker<PathBuf, Result<CargoWorkspace>>, WorkerHan
         1,
         |input_receiver, output_sender| {
             input_receiver
+                .into_iter()
                 .map(|path| CargoWorkspace::from_cargo_metadata(path.as_path()))
-                .for_each(|it| output_sender.send(it))
+                .try_for_each(|it| output_sender.send(it))
+                .unwrap()
         },
     )
 }
diff --git a/crates/ra_lsp_server/tests/heavy_tests/support.rs b/crates/ra_lsp_server/tests/heavy_tests/support.rs
index c14d287ca..82ba12f87 100644
--- a/crates/ra_lsp_server/tests/heavy_tests/support.rs
+++ b/crates/ra_lsp_server/tests/heavy_tests/support.rs
@@ -118,7 +118,11 @@ impl Server {
     }
     fn send_request_(&self, r: RawRequest) -> Value {
         let id = r.id;
-        self.worker.as_ref().unwrap().send(RawMessage::Request(r));
+        self.worker
+            .as_ref()
+            .unwrap()
+            .send(RawMessage::Request(r))
+            .unwrap();
         while let Some(msg) = self.recv() {
             match msg {
                 RawMessage::Request(req) => panic!("unexpected request: {:?}", req),
@@ -167,7 +171,8 @@ impl Server {
         self.worker
             .as_ref()
             .unwrap()
-            .send(RawMessage::Notification(not));
+            .send(RawMessage::Notification(not))
+            .unwrap();
     }
 }
 
@@ -185,7 +190,7 @@ impl Drop for Server {
 fn recv_timeout(receiver: &Receiver<RawMessage>) -> Option<RawMessage> {
     let timeout = Duration::from_secs(5);
     select! {
-        recv(receiver, msg) => msg,
-        recv(after(timeout)) => panic!("timed out"),
+        recv(receiver) -> msg => msg.ok(),
+        recv(after(timeout)) -> _ => panic!("timed out"),
     }
 }
diff --git a/crates/ra_vfs/Cargo.toml b/crates/ra_vfs/Cargo.toml
index ccea8a866..7c170cdfc 100644
--- a/crates/ra_vfs/Cargo.toml
+++ b/crates/ra_vfs/Cargo.toml
@@ -8,7 +8,7 @@ authors = ["Aleksey Kladov <[email protected]>"]
 walkdir = "2.2.7"
 relative-path = "0.4.0"
 rustc-hash = "1.0"
-crossbeam-channel = "0.2.4"
+crossbeam-channel = "0.3.5"
 log = "0.4.6"
 
 thread_worker = { path = "../thread_worker" }
diff --git a/crates/ra_vfs/src/io.rs b/crates/ra_vfs/src/io.rs
index 4cfdb83da..80328ad18 100644
--- a/crates/ra_vfs/src/io.rs
+++ b/crates/ra_vfs/src/io.rs
@@ -32,8 +32,10 @@ pub(crate) type Worker = thread_worker::Worker<Task, TaskResult>;
 pub(crate) fn start() -> (Worker, WorkerHandle) {
     thread_worker::spawn("vfs", 128, |input_receiver, output_sender| {
         input_receiver
+            .into_iter()
             .map(handle_task)
-            .for_each(|it| output_sender.send(it))
+            .try_for_each(|it| output_sender.send(it))
+            .unwrap()
     })
 }
 
diff --git a/crates/ra_vfs/src/lib.rs b/crates/ra_vfs/src/lib.rs
index 90d5e21f4..757eac95b 100644
--- a/crates/ra_vfs/src/lib.rs
+++ b/crates/ra_vfs/src/lib.rs
@@ -148,7 +148,7 @@ impl Vfs {
                 path: path.clone(),
                 filter: Box::new(filter),
             };
-            res.worker.inp.send(task);
+            res.worker.inp.send(task).unwrap();
         }
         let roots = res.roots.iter().map(|(id, _)| id).collect();
         (res, roots)
diff --git a/crates/thread_worker/Cargo.toml b/crates/thread_worker/Cargo.toml
index 62d66a1a3..c74b376e2 100644
--- a/crates/thread_worker/Cargo.toml
+++ b/crates/thread_worker/Cargo.toml
@@ -6,6 +6,6 @@ authors = ["Aleksey Kladov <[email protected]>"]
 
 [dependencies]
 drop_bomb = "0.1.0"
-crossbeam-channel = "0.2.4"
+crossbeam-channel = "0.3.5"
 log = "0.4.3"
 
diff --git a/crates/thread_worker/src/lib.rs b/crates/thread_worker/src/lib.rs
index 12e8bf17e..5e46f62fe 100644
--- a/crates/thread_worker/src/lib.rs
+++ b/crates/thread_worker/src/lib.rs
@@ -2,7 +2,7 @@
 
 use std::thread;
 
-use crossbeam_channel::{bounded, unbounded, Receiver, Sender};
+use crossbeam_channel::{bounded, unbounded, Receiver, Sender, RecvError, SendError};
 use drop_bomb::DropBomb;
 
 pub struct Worker<I, O> {
@@ -34,10 +34,10 @@ impl<I, O> Worker<I, O> {
         self.out
     }
 
-    pub fn send(&self, item: I) {
+    pub fn send(&self, item: I) -> Result<(), SendError<I>> {
         self.inp.send(item)
     }
-    pub fn recv(&self) -> Option<O> {
+    pub fn recv(&self) -> Result<O, RecvError> {
         self.out.recv()
     }
 }