nordic-dev.net/rust
nordic-dev.net/rust
2
0
Fork 0
mirror of https://github.com/rust-lang/rust.git synced 2024-11-30 02:33:55 +00:00
Code Issues Packages Projects Releases Wiki Activity

Some abstraction around workers

Browse Source
This commit is contained in:
Aleksey Kladov 2018-09-08 13:15:01 +03:00
parent 326ffcefe0
commit 7daaddb2ac
5 changed files with 92 additions and 74 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

28
crates/server/src/main_loop/mod.rs
View File

@ -23,6 +23,7 @@ use {
server_world::{ServerWorldState, ServerWorld},
main_loop::subscriptions::{Subscriptions},
project_model::{CargoWorkspace, workspace_loader},
thread_watcher::Worker,
};
#[derive(Debug)]
@ -43,8 +44,8 @@ pub fn main_loop(
.build()
.unwrap();
let (task_sender, task_receiver) = unbounded::<Task>();
let ((fs_sender, fs_receiver), fs_watcher) = vfs::roots_loader();
let ((ws_sender, ws_receiver), ws_watcher) = workspace_loader();
let (fs_worker, fs_watcher) = vfs::roots_loader();
let (ws_worker, ws_watcher) = workspace_loader();
info!("server initialized, serving requests");
let mut state = ServerWorldState::new();
@ -59,10 +60,8 @@ pub fn main_loop(
msg_receriver,
task_sender,
task_receiver.clone(),
fs_sender,
fs_receiver,
ws_sender,
ws_receiver,
fs_worker,
ws_worker,
&mut state,
&mut pending_requests,
&mut subs,
@ -93,17 +92,15 @@ fn main_loop_inner(
msg_receiver: &mut Receiver<RawMessage>,
task_sender: Sender<Task>,
task_receiver: Receiver<Task>,
fs_sender: Sender<PathBuf>,
fs_receiver: Receiver<(PathBuf, Vec<FileEvent>)>,
ws_sender: Sender<PathBuf>,
ws_receiver: Receiver<Result<CargoWorkspace>>,
fs_worker: Worker<PathBuf, (PathBuf, Vec<FileEvent>)>,
ws_worker: Worker<PathBuf, Result<CargoWorkspace>>,
state: &mut ServerWorldState,
pending_requests: &mut HashMap<u64, JobHandle>,
subs: &mut Subscriptions,
) -> Result<()> {
let (libdata_sender, libdata_receiver) = unbounded();
ws_sender.send(ws_root.clone());
fs_sender.send(ws_root.clone());
ws_worker.send(ws_root.clone());
fs_worker.send(ws_root.clone());
loop {
#[derive(Debug)]
enum Event {
@ -120,11 +117,11 @@ fn main_loop_inner(
None => bail!("client exited without shutdown"),
},
recv(task_receiver, task) => Event::Task(task.unwrap()),
recv(fs_receiver, events) => match events {
recv(fs_worker.out, events) => match events {
None => bail!("roots watcher died"),
Some((pb, events)) => Event::Fs(pb, events),
}
recv(ws_receiver, ws) => match ws {
recv(ws_worker.out, ws) => match ws {
None => bail!("workspace watcher died"),
Some(ws) => Event::Ws(ws),
}
@ -158,8 +155,7 @@ fn main_loop_inner(
for ws in workspaces.iter() {
for pkg in ws.packages().filter(|pkg| !pkg.is_member(ws)) {
debug!("sending root, {}", pkg.root(ws).to_path_buf().display());
// deadlocky :-(
fs_sender.send(pkg.root(ws).to_path_buf());
fs_worker.send(pkg.root(ws).to_path_buf());
}
}
state.set_workspaces(workspaces);

24
crates/server/src/project_model.rs
View File

@ -3,12 +3,11 @@ use std::{
path::{Path, PathBuf},
};
use cargo_metadata::{metadata_run, CargoOpt};
use crossbeam_channel::{Sender, Receiver};
use libsyntax2::SmolStr;
use {
Result,
thread_watcher::{ThreadWatcher, worker_chan},
thread_watcher::{Worker, ThreadWatcher},
};
#[derive(Debug, Clone)]
@ -162,14 +161,15 @@ impl TargetKind {
}
}
pub fn workspace_loader() -> ((Sender<PathBuf>, Receiver<Result<CargoWorkspace>>), ThreadWatcher) {
let (interface, input_receiver, output_sender) = worker_chan::<PathBuf, Result<CargoWorkspace>>(1);
let thread = ThreadWatcher::spawn("workspace loader", move || {
input_receiver
.into_iter()
.map(|path| CargoWorkspace::from_cargo_metadata(path.as_path()))
.for_each(|it| output_sender.send(it))
});
(interface, thread)
pub fn workspace_loader() -> (Worker<PathBuf, Result<CargoWorkspace>>, ThreadWatcher) {
Worker::<PathBuf, Result<CargoWorkspace>>::spawn(
"workspace loader",
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))
}
)
}

31
crates/server/src/thread_watcher.rs
View File

@ -3,6 +3,33 @@ use crossbeam_channel::{bounded, unbounded, Sender, Receiver};
use drop_bomb::DropBomb;
use 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 ((inp, out), inp_r, out_s) = worker_chan(buf);
let worker = Worker { inp, out };
let watcher = ThreadWatcher::spawn(name, move || f(inp_r, out_s));
(worker, watcher)
}
pub fn stop(self) -> Receiver<O> {
self.out
}
pub fn send(&self, item: I) {
self.inp.send(item)
}
}
pub struct ThreadWatcher {
name: &'static str,
thread: thread::JoinHandle<()>,
@ -10,7 +37,7 @@ pub struct ThreadWatcher {
}
impl ThreadWatcher {
pub fn spawn(name: &'static str, f: impl FnOnce() + Send + 'static) -> ThreadWatcher {
fn spawn(name: &'static str, f: impl FnOnce() + Send + 'static) -> ThreadWatcher {
let thread = thread::spawn(f);
ThreadWatcher {
name,
@ -36,7 +63,7 @@ impl ThreadWatcher {
/// Sets up worker channels in a deadlock-avoind way.
/// If one sets both input and output buffers to a fixed size,
/// a worker might get stuck.
pub fn worker_chan<I, O>(buf: usize) -> ((Sender<I>, Receiver<O>), Receiver<I>, Sender<O>) {
fn worker_chan<I, O>(buf: usize) -> ((Sender<I>, Receiver<O>), Receiver<I>, Sender<O>) {
let (input_sender, input_receiver) = bounded::<I>(buf);
let (output_sender, output_receiver) = unbounded::<O>();
((input_sender, output_receiver), input_receiver, output_sender)

34
crates/server/src/vfs.rs
View File

@ -3,11 +3,10 @@ use std::{
fs,
};
use crossbeam_channel::{Sender, Receiver};
use walkdir::WalkDir;
use {
thread_watcher::{ThreadWatcher, worker_chan},
thread_watcher::{Worker, ThreadWatcher},
};
@ -22,22 +21,21 @@ pub enum FileEventKind {
Add(String),
}
pub fn roots_loader() -> ((Sender<PathBuf>, Receiver<(PathBuf, Vec<FileEvent>)>), ThreadWatcher) {
let (interface, input_receiver, output_sender) =
worker_chan::<PathBuf, (PathBuf, Vec<FileEvent>)>(128);
let thread = ThreadWatcher::spawn("roots loader", move || {
input_receiver
.into_iter()
.map(|path| {
debug!("loading {} ...", path.as_path().display());
let events = load_root(path.as_path());
debug!("... loaded {}", path.as_path().display());
(path, events)
})
.for_each(|it| output_sender.send(it))
});
(interface, thread)
pub fn roots_loader() -> (Worker<PathBuf, (PathBuf, Vec<FileEvent>)>, ThreadWatcher) {
Worker::<PathBuf, (PathBuf, Vec<FileEvent>)>::spawn(
"roots loader",
128, |input_receiver, output_sender| {
input_receiver
.into_iter()
.map(|path| {
debug!("loading {} ...", path.as_path().display());
let events = load_root(path.as_path());
debug!("... loaded {}", path.as_path().display());
(path, events)
})
.for_each(|it| output_sender.send(it))
}
)
}
fn load_root(path: &Path) -> Vec<FileEvent> {

49
crates/server/tests/heavy_tests/support.rs
View File

@ -1,6 +1,5 @@
use std::{
fs,
thread,
cell::{Cell, RefCell},
path::PathBuf,
time::Duration,
@ -8,7 +7,7 @@ use std::{
};
use tempdir::TempDir;
use crossbeam_channel::{after, Sender, Receiver};
use crossbeam_channel::{after, Receiver};
use flexi_logger::Logger;
use languageserver_types::{
Url,
@ -22,7 +21,7 @@ use serde::Serialize;
use serde_json::{Value, from_str, to_string_pretty};
use gen_lsp_server::{RawMessage, RawRequest, RawNotification};
use m::{Result, main_loop, req, thread_watcher::worker_chan};
use m::{main_loop, req, thread_watcher::{ThreadWatcher, Worker}};
pub fn project(fixture: &str) -> Server {
static INIT: Once = Once::new();
@ -61,28 +60,27 @@ pub struct Server {
req_id: Cell<u64>,
messages: RefCell<Vec<RawMessage>>,
dir: TempDir,
sender: Option<Sender<RawMessage>>,
receiver: Receiver<RawMessage>,
server: Option<thread::JoinHandle<Result<()>>>,
worker: Option<Worker<RawMessage, RawMessage>>,
watcher: Option<ThreadWatcher>,
}
impl Server {
fn new(dir: TempDir, files: Vec<(PathBuf, String)>) -> Server {
let path = dir.path().to_path_buf();
let ((msg_sender, msg_receiver), server) = {
let (api, mut msg_receiver, mut msg_sender) = worker_chan::<RawMessage, RawMessage>(128);
let server = thread::spawn(move || {
let (worker, watcher) = Worker::<RawMessage, RawMessage>::spawn(
"test server",
128,
move |mut msg_receiver, mut msg_sender| {
main_loop(true, path, &mut msg_receiver, &mut msg_sender)
});
(api, server)
};
.unwrap()
}
);
let res = Server {
req_id: Cell::new(1),
dir,
messages: Default::default(),
sender: Some(msg_sender),
receiver: msg_receiver,
server: Some(server),
worker: Some(worker),
watcher: Some(watcher),
};
for (path, text) in files {
@ -140,7 +138,7 @@ impl Server {
fn send_request_(&self, r: RawRequest) -> Value
{
let id = r.id;
self.sender.as_ref()
self.worker.as_ref()
.unwrap()
.send(RawMessage::Request(r));
while let Some(msg) = self.recv() {
@ -183,14 +181,14 @@ impl Server {
}
}
fn recv(&self) -> Option<RawMessage> {
recv_timeout(&self.receiver)
recv_timeout(&self.worker.as_ref().unwrap().out)
.map(|msg| {
self.messages.borrow_mut().push(msg.clone());
msg
})
}
fn send_notification(&self, not: RawNotification) {
self.sender.as_ref()
self.worker.as_ref()
.unwrap()
.send(RawMessage::Notification(not));
}
@ -198,16 +196,15 @@ impl Server {
impl Drop for Server {
fn drop(&mut self) {
{
self.send_request::<Shutdown>(666, ());
drop(self.sender.take().unwrap());
while let Some(msg) = recv_timeout(&self.receiver) {
drop(msg);
}
self.send_request::<Shutdown>(666, ());
let receiver = self.worker.take().unwrap().stop();
while let Some(msg) = recv_timeout(&receiver) {
drop(msg);
}
self.server.take()
self.watcher.take()
.unwrap()
.join().unwrap().unwrap();
.stop()
.unwrap();
}
}