Add monitor loop to supervisor logic

[roman]

Context

This pull request modifies the supervisor logic in a way that:

• It runs on a new task when start is called
• It supports running on a new task or in the same thread (run method)
• It keeps the start and termination ordering guarantees that were implemented before

[codecov]

Codecov Report

Merging #9 into master will decrease coverage by 0.19%.
The diff coverage is 71.22%.

@@            Coverage Diff             @@
##           master       #9      +/-   ##
==========================================
- Coverage   64.69%   64.50%   -0.20%     
==========================================
  Files           6        7       +1     
  Lines         966     1034      +68     
  Branches      184      195      +11     
==========================================
+ Hits          625      667      +42     
- Misses        109      120      +11     
- Partials      232      247      +15     

Impacted Files	Coverage Δ
src/events.rs	62.94% <0.00%> (-1.63%)	⬇️
src/supervisor/tests.rs	71.32% <66.66%> (ø)
src/supervisor/mod.rs	65.82% <67.54%> (-0.29%)	⬇️
src/notifier.rs	100.00% <100.00%> (ø)
src/worker/mod.rs	69.84% <100.00%> (-0.93%)	⬇️
src/worker/tests.rs	54.85% <0.00%> (-1.15%)	⬇️

---

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update 963a23f...83d292d. Read the comment docs.

[RadicalZephyr]

Quick review of some more superficial stuff, I'm going to take another look later this weekend when I'm feeling less distracted.

[roman]

I added this file by accident, currently is not being included in the lib.rs file

[RadicalZephyr]

Looks good to me now, other than that one new Sync bound which looks a bit fishy

[RadicalZephyr]

I think this Sync bound is again not necessary.

[roman]

This change

diff --git a/src/events.rs b/src/events.rs
index 74f1b49..a5a65a0 100644
--- a/src/events.rs
+++ b/src/events.rs
@@ -31,7 +31,7 @@ pub struct NodeData {
 
 /// NotifyFn is used by the supervision API to send events to an interested
 /// listener.
-type NotifyFn = Box<dyn (Fn(Event) -> BoxFuture<'static, ()>) + Send + Sync>;
+type NotifyFn = Box<dyn (Fn(Event) -> BoxFuture<'static, ()>) + Send>;
 
 /// EventNotifier is used by the internal supervision API to send events about a
 /// running supervision tree
@@ -41,7 +41,7 @@ pub struct EventNotifier(Arc<NotifyFn>);
 impl EventNotifier {
     pub fn new<F, O>(notify0: F) -> Self
     where
-        F: (Fn(Event) -> O) + Send + Sync + 'static,
+        F: (Fn(Event) -> O) + Send + 'static,
         O: Future<Output = ()> + FutureExt + Send + 'static,
     {
         let notify = move |ev| {

causes this compilation error:

❯ cargo test
   Compiling capataz v0.0.0 (/home/roman/tmp/rust-capataz)
error[E0277]: `(dyn std::ops::Fn(events::Event) -> std::pin::Pin<std::boxed::Box<(dyn futures::Future<Output = ()> + std::marker::Send + 'static)>> + std::marker::Send + 'static)` cannot be shared between threads safely
   --> src/supervisor/mod.rs:480:30
    |
480 |           let start_notifier = StartNotifier::from_oneshot(started_send);
    |                                ^^^^^^^^^^^^^^^^^^^^^^^^^^^ `(dyn std::ops::Fn(events::Event) -> std::pin::Pin<std::boxed::Box<(dyn futures::Future<Output = ()> + std::marker::Send + 'static)>> + std::marker::Send + 'static)` cannot be shared between threads safely
    | 
   ::: src/notifier.rs:8:5
    |
8   | /     pub fn from_oneshot(sender: oneshot::Sender<Result<T, E>>) -> Self
9   | |     where
10  | |         T: Send + 'static,
11  | |         E: Send + 'static,
...   |
15  | |         }))
16  | |     }
    | |_____- required by `notifier::StartNotifier::<T, E>::from_oneshot`
    |
    = help: the trait `std::marker::Sync` is not implemented for `(dyn std::ops::Fn(events::Event) -> std::pin::Pin<std::boxed::Box<(dyn futures::Future<Output = ()> + std::marker::Send + 'static)>> + std::marker::Send + 'static)`
    = note: required because of the requirements on the impl of `std::marker::Sync` for `std::ptr::Unique<(dyn std::ops::Fn(events::Event) -> std::pin::Pin<std::boxed::Box<(dyn futures::Future<Output = ()> + std::marker::Send + 'static)>> + std::marker::Send + 'static)>`
    = note: required because it appears within the type `std::boxed::Box<(dyn std::ops::Fn(events::Event) -> std::pin::Pin<std::boxed::Box<(dyn futures::Future<Output = ()> + std::marker::Send + 'static)>> + std::marker::Send + 'static)>`
    = note: required because of the requirements on the impl of `std::marker::Send` for `std::sync::Arc<std::boxed::Box<(dyn std::ops::Fn(events::Event) -> std::pin::Pin<std::boxed::Box<(dyn futures::Future<Output = ()> + std::marker::Send + 'static)>> + std::marker::Send + 'static)>>`
    = note: required because it appears within the type `events::EventNotifier`
    = note: required because it appears within the type `supervisor::SpecMeta`
    = note: required because it appears within the type `supervisor::Spec`
    = note: required because it appears within the type `(supervisor::Spec, std::sync::Arc<supervisor::StartError>)`

error[E0277]: `(dyn std::ops::Fn(events::Event) -> std::pin::Pin<std::boxed::Box<(dyn futures::Future<Output = ()> + std::marker::Send + 'static)>> + std::marker::Send + 'static)` cannot be shared between threads safely
   --> src/supervisor/mod.rs:488:27
    |
488 |         let join_handle = task::spawn(run_supervisor_monitor(
    |                           ^^^^^^^^^^^ `(dyn std::ops::Fn(events::Event) -> std::pin::Pin<std::boxed::Box<(dyn futures::Future<Output = ()> + std::marker::Send + 'static)>> + std::marker::Send + 'static)` cannot be shared between threads safely
    | 
   ::: /home/roman/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-0.2.22/src/task/spawn.rs:128:20
    |
128 |         T::Output: Send + 'static,
    |                    ---- required by this bound in `tokio::spawn`
    |
    = help: the trait `std::marker::Sync` is not implemented for `(dyn std::ops::Fn(events::Event) -> std::pin::Pin<std::boxed::Box<(dyn futures::Future<Output = ()> + std::marker::Send + 'static)>> + std::marker::Send + 'static)`
    = note: required because of the requirements on the impl of `std::marker::Sync` for `std::ptr::Unique<(dyn std::ops::Fn(events::Event) -> std::pin::Pin<std::boxed::Box<(dyn futures::Future<Output = ()> + std::marker::Send + 'static)>> + std::marker::Send + 'static)>`
    = note: required because it appears within the type `std::boxed::Box<(dyn std::ops::Fn(events::Event) -> std::pin::Pin<std::boxed::Box<(dyn futures::Future<Output = ()> + std::marker::Send + 'static)>> + std::marker::Send + 'static)>`
    = note: required because of the requirements on the impl of `std::marker::Send` for `std::sync::Arc<std::boxed::Box<(dyn std::ops::Fn(events::Event) -> std::pin::Pin<std::boxed::Box<(dyn futures::Future<Output = ()> + std::marker::Send + 'static)>> + std::marker::Send + 'static)>>`
    = note: required because it appears within the type `events::EventNotifier`
    = note: required because it appears within the type `supervisor::SpecMeta`
    = note: required because it appears within the type `supervisor::Spec`
    = note: required because it appears within the type `supervisor::SupervisorResult`

error: aborting due to 2 previous errors

For more information about this error, try `rustc --explain E0277`.
error: could not compile `capataz`.