blocking workers

Cargo.toml

@@ -27,7 +27,7 @@ oneshot = { version = "0.1.6", default-features = false, features = ["std"] }
 parking_lot = "0.12.1"
 pin-project = "1.1.5"
 rmodbus = { version = "0.9.3" }
-roboplc-derive = { version = "0.1.2" }
+roboplc-derive = { version = "0.1.3" }
 serde = { version = "1.0.197", features = ["derive", "rc"] }
 serial = "0.4.0"
 sysinfo = "0.30.6"

examples/shutdown.rs

@@ -17,6 +17,22 @@ enum Message {
     Terminate,
 }
 
+/// A worker which has got a blocking loop (e.g. listening to a socket, having a long cycle etc.)
+/// and it is not possible to terminate it immediately. In this case the worker is not joined on
+/// shutdown
+#[derive(WorkerOpts)]
+#[worker_opts(name = "veryblocking", blocking = true)]
+struct VeryBlocking {}
+
+impl Worker<Message, ()> for VeryBlocking {
+    fn run(&mut self, _context: &Context<Message, ()>) -> WResult {
+        for _ in interval(Duration::from_secs(120)) {
+            info!(worker = self.worker_name(), "I am still running");
+        }
+        Ok(())
+    }
+}
+
 #[derive(WorkerOpts)]
 #[worker_opts(name = "parser")]
 struct DataParser {}
@@ -97,6 +113,7 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
     controller.spawn_worker(DataGenerator {})?;
     controller.spawn_worker(DataParser {})?;
     controller.spawn_worker(SignalHandler {})?;
+    controller.spawn_worker(VeryBlocking {})?;
     info!("controller started");
     controller.block();
     info!("controller terminated");

src/controller.rs

@@ -145,7 +145,8 @@ where
         }
         let mut builder = Builder::new()
             .name(worker.worker_name())
-            .rt_params(rt_params);
+            .rt_params(rt_params)
+            .blocking(worker.worker_is_blocking());
         if let Some(stack_size) = worker.worker_stack_size() {
             builder = builder.stack_size(stack_size);
         }
@@ -156,7 +157,7 @@ where
         })?;
         Ok(())
     }
-    /// Spawns a task thread (non-real-time)
+    /// Spawns a task thread (non-real-time) with the default options
     pub fn spawn_task<F>(&mut self, name: &str, f: F) -> Result<(), Error>
     where
         F: FnOnce() + Send + 'static,
@@ -290,4 +291,10 @@ pub trait WorkerOptions {
     fn worker_cpu_ids(&self) -> Option<&[usize]> {
         None
     }
+    /// A hint for task supervisors that the worker blocks the thread (e.g. listens to a socket or
+    /// has got a big interval in the main loop, does not return any useful result and should not
+    /// be joined)
+    fn worker_is_blocking(&self) -> bool {
+        false
+    }
 }

src/supervisor.rs

@@ -97,7 +97,9 @@ impl<T> Supervisor<T> {
     pub fn join_all(&mut self) -> BTreeMap<String, thread::Result<T>> {
         let mut result = BTreeMap::new();
         for (name, task) in mem::take(&mut self.tasks) {
-            result.insert(name, task.join());
+            if !task.is_blocking() {
+                result.insert(name, task.join());
+            }
         }
         result
     }
@@ -178,7 +180,9 @@ impl<'a, 'env, T> ScopedSupervisor<'a, 'env, T> {
     pub fn join_all(&mut self) -> BTreeMap<String, thread::Result<T>> {
         let mut result = BTreeMap::new();
         for (name, task) in mem::take(&mut self.tasks) {
-            result.insert(name, task.join());
+            if !task.is_blocking() {
+                result.insert(name, task.join());
+            }
         }
         result
     }

src/thread_rt.rs

@@ -29,6 +29,7 @@ pub fn set_simulated() {
 pub struct Builder {
     pub(crate) name: Option<String>,
     stack_size: Option<usize>,
+    blocking: bool,
     rt_params: RTParams,
     // an internal parameter to suspend (park) failed threads instead of panic
     pub(crate) park_on_errors: bool,
@@ -89,6 +90,15 @@ impl Builder {
         self.stack_size = Some(size);
         self
     }
+    /// A hint for task supervisors that the task blocks the thread (e.g. listens to a socket or
+    /// has got a big interval in the main loop, does not return any useful result and should not
+    /// be joined)
+    ///
+    /// For scoped tasks: the task may be still forcibly joined at the end of the scope
+    pub fn blocking(mut self, blocking: bool) -> Self {
+        self.blocking = blocking;
+        self
+    }
     /// Applies real-time parameters to the task
     ///
     /// See [`RTParams`]
@@ -98,7 +108,7 @@ impl Builder {
     }
     fn try_into_thread_builder_name_and_params(
         self,
-    ) -> Result<(thread::Builder, String, RTParams, bool)> {
+    ) -> Result<(thread::Builder, String, bool, RTParams, bool)> {
         let mut builder = thread::Builder::new();
         if let Some(ref name) = self.name {
             if name.len() > 15 {
@@ -115,6 +125,7 @@ impl Builder {
         Ok((
             builder,
             self.name.unwrap_or_default(),
+            self.blocking,
             self.rt_params,
             self.park_on_errors,
         ))
@@ -130,7 +141,7 @@ impl Builder {
         F: FnOnce() -> T + Send + 'static,
         T: Send + 'static,
     {
-        let (builder, name, rt_params, park_on_errors) =
+        let (builder, name, blocking, rt_params, park_on_errors) =
             self.try_into_thread_builder_name_and_params()?;
         let (tx, rx) = oneshot::channel();
         let handle = builder.spawn(move || {
@@ -141,6 +152,7 @@ impl Builder {
         Ok(Task {
             name,
             handle,
+            blocking,
             tid,
             rt_params,
             info: <_>::default(),
@@ -180,7 +192,7 @@ impl Builder {
         F: FnOnce() -> T + Send + 'scope,
         T: Send + 'scope,
     {
-        let (builder, name, rt_params, park_on_errors) =
+        let (builder, name, blocking, rt_params, park_on_errors) =
             self.try_into_thread_builder_name_and_params()?;
         let (tx, rx) = oneshot::channel();
         let handle = builder.spawn_scoped(scope, move || {
@@ -191,6 +203,7 @@ impl Builder {
         Ok(ScopedTask {
             name,
             handle,
+            blocking,
             tid,
             rt_params,
             info: <_>::default(),
@@ -239,6 +252,7 @@ pub struct Task<T> {
         serialize_with = "serialize_join_handle_active"
     )]
     handle: JoinHandle<T>,
+    blocking: bool,
     tid: libc::c_int,
     rt_params: RTParams,
     info: TaskInfo,
@@ -277,6 +291,9 @@ impl<T> Task<T> {
     pub fn elapsed(&self) -> Duration {
         self.info.started_mt.elapsed()
     }
+    pub fn is_blocking(&self) -> bool {
+        self.blocking
+    }
 }
 
 impl<T> From<Task<T>> for JoinHandle<T> {
@@ -296,6 +313,7 @@ pub struct ScopedTask<'scope, T> {
         serialize_with = "serialize_scoped_join_handle_active"
     )]
     handle: ScopedJoinHandle<'scope, T>,
+    blocking: bool,
     tid: libc::c_int,
     rt_params: RTParams,
     info: TaskInfo,
@@ -334,6 +352,9 @@ impl<'scope, T> ScopedTask<'scope, T> {
     pub fn elapsed(&self) -> Duration {
         self.info.started_mt.elapsed()
     }
+    pub fn is_blocking(&self) -> bool {
+        self.blocking
+    }
 }
 
 impl<'scope, T> From<ScopedTask<'scope, T>> for ScopedJoinHandle<'scope, T> {