move thread worker to a separate crate

2 files changed, 90 insertions, 0 deletions

diff --git a/crates/thread_worker/Cargo.toml b/crates/thread_worker/Cargo.toml
new file mode 100644
index 000000000..62d66a1a3
--- /dev/null
+++ b/crates/thread_worker/Cargo.toml
@@ -0,0 +1,11 @@
+[package]
+edition = "2018"
+name = "thread_worker"
+version = "0.1.0"
+authors = ["Aleksey Kladov <[email protected]>"]
+
+[dependencies]
+drop_bomb = "0.1.0"
+crossbeam-channel = "0.2.4"
+log = "0.4.3"
+
diff --git a/crates/thread_worker/src/lib.rs b/crates/thread_worker/src/lib.rs
new file mode 100644
index 000000000..e558559ef
--- /dev/null
+++ b/crates/thread_worker/src/lib.rs
@@ -0,0 +1,79 @@
+//! Small utility to correctly spawn crossbeam-channel based worker threads.
+
+use std::thread;
+
+use crossbeam_channel::{bounded, unbounded, Receiver, Sender};
+use drop_bomb::DropBomb;
+
+pub struct Worker<I, O> {
+    pub inp: Sender<I>,
+    pub out: Receiver<O>,
+}
+
+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
+    }
+
+    pub fn send(&self, item: I) {
+        self.inp.send(item)
+    }
+}
+
+impl WorkerHandle {
+    fn spawn(name: &'static str, f: impl FnOnce() + Send + 'static) -> WorkerHandle {
+        let thread = thread::spawn(f);
+        WorkerHandle {
+            name,
+            thread,
+            bomb: DropBomb::new(format!("WorkerHandle {} was not stopped", name)),
+        }
+    }
+
+    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();
+        match &res {
+            Ok(()) => log::info!("... {} terminated with ok", name),
+            Err(_) => log::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,
+    )
+}