286: move thread worker to a separate crate r=matklad a=matklad

Going to tackle https://github.com/rust-analyzer/rust-analyzer/issues/243, this is a refactoring towards that goal

Co-authored-by: Aleksey Kladov <aleksey.kladov@gmail.com>
This commit is contained in:
bors[bot] 2018-12-18 09:54:56 +00:00
commit f3ea21ac16
9 changed files with 62 additions and 46 deletions

10
Cargo.lock generated
View file

@ -700,6 +700,7 @@ dependencies = [
"tempdir 0.3.7 (registry+https://github.com/rust-lang/crates.io-index)",
"test_utils 0.1.0",
"text_unit 0.1.5 (registry+https://github.com/rust-lang/crates.io-index)",
"thread_worker 0.1.0",
"threadpool 1.7.1 (registry+https://github.com/rust-lang/crates.io-index)",
"url_serde 0.2.0 (registry+https://github.com/rust-lang/crates.io-index)",
"walkdir 2.2.6 (registry+https://github.com/rust-lang/crates.io-index)",
@ -1116,6 +1117,15 @@ dependencies = [
"lazy_static 1.2.0 (registry+https://github.com/rust-lang/crates.io-index)",
]
[[package]]
name = "thread_worker"
version = "0.1.0"
dependencies = [
"crossbeam-channel 0.2.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)",
]
[[package]]
name = "threadpool"
version = "1.7.1"

View file

@ -26,6 +26,7 @@ text_unit = { version = "0.1.2", features = ["serde"] }
smol_str = { version = "0.1.5", features = ["serde"] }
rustc-hash = "1.0"
thread_worker = { path = "../thread_worker" }
ra_syntax = { path = "../ra_syntax" }
ra_editor = { path = "../ra_editor" }
ra_text_edit = { path = "../ra_text_edit" }

View file

@ -5,7 +5,6 @@ mod path_map;
mod project_model;
pub mod req;
mod server_world;
pub mod thread_watcher;
mod vfs;
pub type Result<T> = ::std::result::Result<T, ::failure::Error>;

View file

@ -10,6 +10,7 @@ use gen_lsp_server::{
use languageserver_types::NumberOrString;
use ra_analysis::{Canceled, FileId, LibraryData};
use rayon;
use thread_worker::Worker;
use threadpool::ThreadPool;
use rustc_hash::FxHashSet;
use serde::{de::DeserializeOwned, Serialize};
@ -21,7 +22,6 @@ use crate::{
project_model::{workspace_loader, CargoWorkspace},
req,
server_world::{ServerWorld, ServerWorldState},
thread_watcher::Worker,
vfs::{self, FileEvent},
Result,
};
@ -92,8 +92,8 @@ pub fn main_loop(
let ws_res = ws_watcher.stop();
main_res?;
fs_res?;
ws_res?;
fs_res.map_err(|_| format_err!("fs watcher died"))?;
ws_res.map_err(|_| format_err!("ws watcher died"))?;
Ok(())
}

View file

@ -4,11 +4,9 @@ use cargo_metadata::{metadata_run, CargoOpt};
use ra_syntax::SmolStr;
use rustc_hash::{FxHashMap, FxHashSet};
use failure::{format_err, bail};
use thread_worker::{WorkerHandle, Worker};
use crate::{
Result,
thread_watcher::{ThreadWatcher, Worker},
};
use crate::Result;
/// `CargoWorksapce` represents the logical structure of, well, a Cargo
/// workspace. It pretty closely mirrors `cargo metadata` output.
@ -199,8 +197,8 @@ impl TargetKind {
}
}
pub fn workspace_loader() -> (Worker<PathBuf, Result<CargoWorkspace>>, ThreadWatcher) {
Worker::<PathBuf, Result<CargoWorkspace>>::spawn(
pub fn workspace_loader() -> (Worker<PathBuf, Result<CargoWorkspace>>, WorkerHandle) {
thread_worker::spawn::<PathBuf, Result<CargoWorkspace>, _>(
"workspace loader",
1,
|input_receiver, output_sender| {

View file

@ -4,8 +4,7 @@ use std::{
};
use walkdir::WalkDir;
use crate::thread_watcher::{ThreadWatcher, Worker};
use thread_worker::{WorkerHandle, Worker};
#[derive(Debug)]
pub struct FileEvent {
@ -18,8 +17,8 @@ pub enum FileEventKind {
Add(String),
}
pub fn roots_loader() -> (Worker<PathBuf, (PathBuf, Vec<FileEvent>)>, ThreadWatcher) {
Worker::<PathBuf, (PathBuf, Vec<FileEvent>)>::spawn(
pub fn roots_loader() -> (Worker<PathBuf, (PathBuf, Vec<FileEvent>)>, WorkerHandle) {
thread_worker::spawn::<PathBuf, (PathBuf, Vec<FileEvent>), _>(
"roots loader",
128,
|input_receiver, output_sender| {

View file

@ -17,11 +17,11 @@ use languageserver_types::{
use serde::Serialize;
use serde_json::{to_string_pretty, Value};
use tempdir::TempDir;
use thread_worker::{WorkerHandle, Worker};
use test_utils::{parse_fixture, find_mismatch};
use ra_lsp_server::{
main_loop, req,
thread_watcher::{ThreadWatcher, Worker},
};
pub fn project(fixture: &str) -> Server {
@ -45,13 +45,13 @@ pub struct Server {
messages: RefCell<Vec<RawMessage>>,
dir: TempDir,
worker: Option<Worker<RawMessage, RawMessage>>,
watcher: Option<ThreadWatcher>,
watcher: Option<WorkerHandle>,
}
impl Server {
fn new(dir: TempDir, files: Vec<(PathBuf, String)>) -> Server {
let path = dir.path().to_path_buf();
let (worker, watcher) = Worker::<RawMessage, RawMessage>::spawn(
let (worker, watcher) = thread_worker::spawn::<RawMessage, RawMessage, _>(
"test server",
128,
move |mut msg_receiver, mut msg_sender| {

View file

@ -0,0 +1,11 @@
[package]
edition = "2018"
name = "thread_worker"
version = "0.1.0"
authors = ["Aleksey Kladov <aleksey.kladov@gmail.com>"]
[dependencies]
drop_bomb = "0.1.0"
crossbeam-channel = "0.2.4"
log = "0.4.3"

View file

@ -1,28 +1,35 @@
//! Small utility to correctly spawn crossbeam-channel based worker threads.
use std::thread;
use crossbeam_channel::{bounded, unbounded, Receiver, Sender};
use drop_bomb::DropBomb;
use failure::format_err;
use crate::Result;
pub struct Worker<I, O> {
pub inp: Sender<I>,
pub out: Receiver<O>,
}
impl<I, O> Worker<I, O> {
pub fn spawn<F>(name: &'static str, buf: usize, f: F) -> (Self, ThreadWatcher)
where
F: FnOnce(Receiver<I>, Sender<O>) + Send + 'static,
I: Send + 'static,
O: Send + 'static,
{
let (worker, inp_r, out_s) = worker_chan(buf);
let watcher = ThreadWatcher::spawn(name, move || f(inp_r, out_s));
(worker, watcher)
}
pub struct WorkerHandle {
name: &'static str,
thread: thread::JoinHandle<()>,
bomb: DropBomb,
}
pub fn spawn<I, O, F>(name: &'static str, buf: usize, f: F) -> (Worker<I, O>, WorkerHandle)
where
F: FnOnce(Receiver<I>, Sender<O>) + Send + 'static,
I: Send + 'static,
O: Send + 'static,
{
let (worker, inp_r, out_s) = worker_chan(buf);
let watcher = WorkerHandle::spawn(name, move || f(inp_r, out_s));
(worker, watcher)
}
impl<I, O> Worker<I, O> {
/// Stops the worker. Returns the message receiver to fetch results which
/// have become ready before the worker is stopped.
pub fn stop(self) -> Receiver<O> {
self.out
}
@ -32,30 +39,21 @@ impl<I, O> Worker<I, O> {
}
}
pub struct ThreadWatcher {
name: &'static str,
thread: thread::JoinHandle<()>,
bomb: DropBomb,
}
impl ThreadWatcher {
fn spawn(name: &'static str, f: impl FnOnce() + Send + 'static) -> ThreadWatcher {
impl WorkerHandle {
fn spawn(name: &'static str, f: impl FnOnce() + Send + 'static) -> WorkerHandle {
let thread = thread::spawn(f);
ThreadWatcher {
WorkerHandle {
name,
thread,
bomb: DropBomb::new(format!("ThreadWatcher {} was not stopped", name)),
bomb: DropBomb::new(format!("WorkerHandle {} was not stopped", name)),
}
}
pub fn stop(mut self) -> Result<()> {
pub fn stop(mut self) -> thread::Result<()> {
log::info!("waiting for {} to finish ...", self.name);
let name = self.name;
self.bomb.defuse();
let res = self
.thread
.join()
.map_err(|_| format_err!("ThreadWatcher {} died", name));
let res = self.thread.join();
match &res {
Ok(()) => log::info!("... {} terminated with ok", name),
Err(_) => log::error!("... {} terminated with err", name),