Skip to content

Commit c530f04

Browse files
bors[bot]matklad
bors[bot]
and
matklad
committed
Merge #833
833: automatically wait for worker threads r=matklad a=matklad closes #817 Co-authored-by: Aleksey Kladov <[email protected]>
2 parents 10bf61b + e0b8942 commit c530f04

File tree

10 files changed

+119
-153
lines changed

10 files changed

+119
-153
lines changed

‎Cargo.lock

-1
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎crates/ra_batch/src/lib.rs

-1
Original file line numberDiff line numberDiff line change
@@ -121,7 +121,6 @@ impl BatchDatabase {
121121
.collect();
122122

123123
let db = BatchDatabase::load(crate_graph, &mut vfs);
124-
let _ = vfs.shutdown();
125124
Ok((db, local_roots))
126125
}
127126
}

‎crates/ra_lsp_server/src/main_loop.rs

+12-14
Original file line numberDiff line numberDiff line change
@@ -54,19 +54,20 @@ pub fn main_loop(
5454
) -> Result<()> {
5555
let pool = ThreadPool::new(THREADPOOL_SIZE);
5656
let (task_sender, task_receiver) = unbounded::<Task>();
57-
let (ws_worker, ws_watcher) = workspace_loader();
5857

59-
ws_worker.send(ws_root.clone()).unwrap();
6058
// FIXME: support dynamic workspace loading.
61-
let workspaces = match ws_worker.recv().unwrap() {
62-
Ok(ws) => vec![ws],
63-
Err(e) => {
64-
log::error!("loading workspace failed: {}", e);
65-
Vec::new()
59+
let workspaces = {
60+
let ws_worker = workspace_loader();
61+
ws_worker.sender().send(ws_root.clone()).unwrap();
62+
match ws_worker.receiver().recv().unwrap() {
63+
Ok(ws) => vec![ws],
64+
Err(e) => {
65+
log::error!("loading workspace failed: {}", e);
66+
Vec::new()
67+
}
6668
}
6769
};
68-
ws_worker.shutdown();
69-
ws_watcher.shutdown().map_err(|_| format_err!("ws watcher died"))?;
70+
7071
let mut state = ServerWorldState::new(ws_root.clone(), workspaces);
7172

7273
log::info!("server initialized, serving requests");
@@ -94,12 +95,9 @@ pub fn main_loop(
9495
log::info!("...threadpool has finished");
9596

9697
let vfs = Arc::try_unwrap(state.vfs).expect("all snapshots should be dead");
97-
let vfs_res = vfs.into_inner().shutdown();
98+
drop(vfs);
9899

99-
main_res?;
100-
vfs_res.map_err(|_| format_err!("fs watcher died"))?;
101-
102-
Ok(())
100+
main_res
103101
}
104102

105103
enum Event {

‎crates/ra_lsp_server/src/project_model.rs

+3-3
Original file line numberDiff line numberDiff line change
@@ -1,15 +1,15 @@
11
use std::path::PathBuf;
22

3-
use thread_worker::{WorkerHandle, Worker};
3+
use thread_worker::Worker;
44

55
use crate::Result;
66

77
pub use ra_project_model::{
88
ProjectWorkspace, CargoWorkspace, Package, Target, TargetKind, Sysroot,
99
};
1010

11-
pub fn workspace_loader() -> (Worker<PathBuf, Result<ProjectWorkspace>>, WorkerHandle) {
12-
thread_worker::spawn::<PathBuf, Result<ProjectWorkspace>, _>(
11+
pub fn workspace_loader() -> Worker<PathBuf, Result<ProjectWorkspace>> {
12+
Worker::<PathBuf, Result<ProjectWorkspace>>::spawn(
1313
"workspace loader",
1414
1,
1515
|input_receiver, output_sender| {

‎crates/ra_lsp_server/tests/heavy_tests/support.rs

+5-12
Original file line numberDiff line numberDiff line change
@@ -17,7 +17,7 @@ use lsp_types::{
1717
use serde::Serialize;
1818
use serde_json::{to_string_pretty, Value};
1919
use tempfile::TempDir;
20-
use thread_worker::{WorkerHandle, Worker};
20+
use thread_worker::Worker;
2121
use test_utils::{parse_fixture, find_mismatch};
2222

2323
use ra_lsp_server::{
@@ -45,13 +45,12 @@ pub struct Server {
4545
messages: RefCell<Vec<RawMessage>>,
4646
dir: TempDir,
4747
worker: Option<Worker<RawMessage, RawMessage>>,
48-
watcher: Option<WorkerHandle>,
4948
}
5049

5150
impl Server {
5251
fn new(dir: TempDir, files: Vec<(PathBuf, String)>) -> Server {
5352
let path = dir.path().to_path_buf();
54-
let (worker, watcher) = thread_worker::spawn::<RawMessage, RawMessage, _>(
53+
let worker = Worker::<RawMessage, RawMessage>::spawn(
5554
"test server",
5655
128,
5756
move |mut msg_receiver, mut msg_sender| {
@@ -63,7 +62,6 @@ impl Server {
6362
dir,
6463
messages: Default::default(),
6564
worker: Some(worker),
66-
watcher: Some(watcher),
6765
};
6866

6967
for (path, text) in files {
@@ -117,7 +115,7 @@ impl Server {
117115
}
118116
fn send_request_(&self, r: RawRequest) -> Value {
119117
let id = r.id;
120-
self.worker.as_ref().unwrap().send(RawMessage::Request(r)).unwrap();
118+
self.worker.as_ref().unwrap().sender().send(RawMessage::Request(r)).unwrap();
121119
while let Some(msg) = self.recv() {
122120
match msg {
123121
RawMessage::Request(req) => panic!("unexpected request: {:?}", req),
@@ -157,24 +155,19 @@ impl Server {
157155
}
158156
}
159157
fn recv(&self) -> Option<RawMessage> {
160-
recv_timeout(&self.worker.as_ref().unwrap().out).map(|msg| {
158+
recv_timeout(&self.worker.as_ref().unwrap().receiver()).map(|msg| {
161159
self.messages.borrow_mut().push(msg.clone());
162160
msg
163161
})
164162
}
165163
fn send_notification(&self, not: RawNotification) {
166-
self.worker.as_ref().unwrap().send(RawMessage::Notification(not)).unwrap();
164+
self.worker.as_ref().unwrap().sender().send(RawMessage::Notification(not)).unwrap();
167165
}
168166
}
169167

170168
impl Drop for Server {
171169
fn drop(&mut self) {
172170
self.send_request::<Shutdown>(());
173-
let receiver = self.worker.take().unwrap().shutdown();
174-
while let Some(msg) = recv_timeout(&receiver) {
175-
drop(msg);
176-
}
177-
self.watcher.take().unwrap().shutdown().unwrap();
178171
}
179172
}
180173

‎crates/ra_vfs/src/io.rs

+35-56
Original file line numberDiff line numberDiff line change
@@ -1,13 +1,11 @@
11
use std::{
22
fs,
3-
thread,
43
path::{Path, PathBuf},
54
sync::{mpsc, Arc},
65
time::Duration,
76
};
8-
use crossbeam_channel::{Receiver, Sender, unbounded, RecvError, select};
7+
use crossbeam_channel::{Sender, unbounded, RecvError, select};
98
use relative_path::RelativePathBuf;
10-
use thread_worker::WorkerHandle;
119
use walkdir::WalkDir;
1210
use notify::{DebouncedEvent, RecommendedWatcher, RecursiveMode, Watcher as _Watcher};
1311

@@ -48,37 +46,42 @@ enum ChangeKind {
4846

4947
const WATCHER_DELAY: Duration = Duration::from_millis(250);
5048

51-
pub(crate) struct Worker {
52-
worker: thread_worker::Worker<Task, TaskResult>,
53-
worker_handle: WorkerHandle,
54-
}
55-
56-
impl Worker {
57-
pub(crate) fn start(roots: Arc<Roots>) -> Worker {
58-
// This is a pretty elaborate setup of threads & channels! It is
59-
// explained by the following concerns:
60-
// * we need to burn a thread translating from notify's mpsc to
61-
// crossbeam_channel.
62-
// * we want to read all files from a single thread, to guarantee that
63-
// we always get fresher versions and never go back in time.
64-
// * we want to tear down everything neatly during shutdown.
65-
let (worker, worker_handle) = thread_worker::spawn(
66-
"vfs",
67-
128,
68-
// This are the channels we use to communicate with outside world.
69-
// If `input_receiver` is closed we need to tear ourselves down.
70-
// `output_sender` should not be closed unless the parent died.
71-
move |input_receiver, output_sender| {
49+
pub(crate) type Worker = thread_worker::Worker<Task, TaskResult>;
50+
pub(crate) fn start(roots: Arc<Roots>) -> Worker {
51+
// This is a pretty elaborate setup of threads & channels! It is
52+
// explained by the following concerns:
53+
// * we need to burn a thread translating from notify's mpsc to
54+
// crossbeam_channel.
55+
// * we want to read all files from a single thread, to guarantee that
56+
// we always get fresher versions and never go back in time.
57+
// * we want to tear down everything neatly during shutdown.
58+
Worker::spawn(
59+
"vfs",
60+
128,
61+
// This are the channels we use to communicate with outside world.
62+
// If `input_receiver` is closed we need to tear ourselves down.
63+
// `output_sender` should not be closed unless the parent died.
64+
move |input_receiver, output_sender| {
65+
// Make sure that the destruction order is
66+
//
67+
// * notify_sender
68+
// * _thread
69+
// * watcher_sender
70+
//
71+
// this is required to avoid deadlocks.
72+
73+
// These are the corresponding crossbeam channels
74+
let (watcher_sender, watcher_receiver) = unbounded();
75+
let _thread;
76+
{
7277
// These are `std` channels notify will send events to
7378
let (notify_sender, notify_receiver) = mpsc::channel();
74-
// These are the corresponding crossbeam channels
75-
let (watcher_sender, watcher_receiver) = unbounded();
7679

7780
let mut watcher = notify::watcher(notify_sender, WATCHER_DELAY)
7881
.map_err(|e| log::error!("failed to spawn notify {}", e))
7982
.ok();
8083
// Start a silly thread to transform between two channels
81-
let thread = thread::spawn(move || {
84+
_thread = thread_worker::ScopedThread::spawn("notify-convertor", move || {
8285
notify_receiver
8386
.into_iter()
8487
.for_each(|event| convert_notify_event(event, &watcher_sender))
@@ -110,35 +113,11 @@ impl Worker {
110113
},
111114
}
112115
}
113-
// Stopped the watcher
114-
drop(watcher.take());
115-
// Drain pending events: we are not interested in them anyways!
116-
watcher_receiver.into_iter().for_each(|_| ());
117-
118-
let res = thread.join();
119-
match &res {
120-
Ok(()) => log::info!("... Watcher terminated with ok"),
121-
Err(_) => log::error!("... Watcher terminated with err"),
122-
}
123-
res.unwrap();
124-
},
125-
);
126-
127-
Worker { worker, worker_handle }
128-
}
129-
130-
pub(crate) fn sender(&self) -> &Sender<Task> {
131-
&self.worker.inp
132-
}
133-
134-
pub(crate) fn receiver(&self) -> &Receiver<TaskResult> {
135-
&self.worker.out
136-
}
137-
138-
pub(crate) fn shutdown(self) -> thread::Result<()> {
139-
let _ = self.worker.shutdown();
140-
self.worker_handle.shutdown()
141-
}
116+
}
117+
// Drain pending events: we are not interested in them anyways!
118+
watcher_receiver.into_iter().for_each(|_| ());
119+
},
120+
)
142121
}
143122

144123
fn watch_root(

‎crates/ra_vfs/src/lib.rs

+1-7
Original file line numberDiff line numberDiff line change
@@ -22,7 +22,6 @@ use std::{
2222
fmt, fs, mem,
2323
path::{Path, PathBuf},
2424
sync::Arc,
25-
thread,
2625
};
2726

2827
use crossbeam_channel::Receiver;
@@ -160,7 +159,7 @@ impl fmt::Debug for Vfs {
160159
impl Vfs {
161160
pub fn new(roots: Vec<PathBuf>) -> (Vfs, Vec<VfsRoot>) {
162161
let roots = Arc::new(Roots::new(roots));
163-
let worker = io::Worker::start(Arc::clone(&roots));
162+
let worker = io::start(Arc::clone(&roots));
164163
let mut root2files = ArenaMap::default();
165164

166165
for (root, config) in roots.iter() {
@@ -337,11 +336,6 @@ impl Vfs {
337336
mem::replace(&mut self.pending_changes, Vec::new())
338337
}
339338

340-
/// Shutdown the VFS and terminate the background watching thread.
341-
pub fn shutdown(self) -> thread::Result<()> {
342-
self.worker.shutdown()
343-
}
344-
345339
fn add_file(
346340
&mut self,
347341
root: VfsRoot,

‎crates/ra_vfs/tests/vfs.rs

-1
Original file line numberDiff line numberDiff line change
@@ -158,6 +158,5 @@ fn test_vfs_works() -> std::io::Result<()> {
158158
Err(RecvTimeoutError::Timeout)
159159
);
160160

161-
vfs.shutdown().unwrap();
162161
Ok(())
163162
}

‎crates/thread_worker/Cargo.toml

-1
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,6 @@ version = "0.1.0"
55
authors = ["rust-analyzer developers"]
66

77
[dependencies]
8-
drop_bomb = "0.1.0"
98
crossbeam-channel = "0.3.5"
109
log = "0.4.3"
1110

0 commit comments

Comments
 (0)