27694abd94
138: Fix some clippy lints r=matklad a=alanhdu
I went ahead and fixed all the clippy lints (there were a couple I thought would be better unfixed and added `cfg` statements to allow them) and also re-enabled clippy and rustfmt in CI.
They were disabled with `no time to explain, disable clippy checks`, so hopefully this won't go against whatever the reason at the time was 😆.
One question about the CI though: right now, it's an allowed failure that runs against the latest nightly each time. Would it be better to pin it to a specific nightly (or use the `beta` versions) to lower the churn?
Co-authored-by: Alan Du <alanhdu@gmail.com>
81 lines
2.1 KiB
Rust
81 lines
2.1 KiB
Rust
use std::thread;
|
|
|
|
use crossbeam_channel::{bounded, unbounded, Receiver, Sender};
|
|
use drop_bomb::DropBomb;
|
|
|
|
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 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<()>,
|
|
bomb: DropBomb,
|
|
}
|
|
|
|
impl ThreadWatcher {
|
|
fn spawn(name: &'static str, f: impl FnOnce() + Send + 'static) -> ThreadWatcher {
|
|
let thread = thread::spawn(f);
|
|
ThreadWatcher {
|
|
name,
|
|
thread,
|
|
bomb: DropBomb::new(format!("ThreadWatcher {} was not stopped", name)),
|
|
}
|
|
}
|
|
|
|
pub fn stop(mut self) -> Result<()> {
|
|
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));
|
|
match &res {
|
|
Ok(()) => info!("... {} terminated with ok", name),
|
|
Err(_) => error!("... {} terminated with err", name),
|
|
}
|
|
res
|
|
}
|
|
}
|
|
|
|
/// 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.
|
|
fn worker_chan<I, O>(buf: usize) -> (Worker<I, O>, Receiver<I>, Sender<O>) {
|
|
let (input_sender, input_receiver) = bounded::<I>(buf);
|
|
let (output_sender, output_receiver) = unbounded::<O>();
|
|
(
|
|
Worker {
|
|
inp: input_sender,
|
|
out: output_receiver,
|
|
},
|
|
input_receiver,
|
|
output_sender,
|
|
)
|
|
}
|