Repository: actix/actix (language: rust)

Issue:
Handling panics in Actors run by SyncArbiters
Hi there!

First off, I've been a huge fan of Actix (and Actix Web) so far. Thank you for all of your hard work! :)

Like the issue title says, I'm curious about how to _correctly_ handle panics of Actors running in separate threads.

For some context, I'm writing an IRC relay bot using the irc crate and actix. I have an actor wrapping the future::Stream of events coming from the IRC client. So far so good. The IRC client is running on the main thread sending these events to another actor running on a different thread (via a SyncArbiter):

```rust
let sys = actix::System::new("ircrelay");
let message_handler = Arbiter::start(|_| MessageHandler::with_scrollback_capacity(100));

let message_handler_clone = message_handler.clone();
let cfg_clone = config.clone();
let _irc_actor: Addr<Unsync, IrcActor> =
        actix::Supervisor::start(move |_| IrcActor::new(message_handler_clone, &cfg_clone));
```

From my IrcActor, I send the message to the `MessageHandler`

```rust
if self.should_relay(&irc_message) {
    if let Some(message) = StructuredIrcMessage::from_message(irc_message) {
        self.handler_addr.try_send(message).expect("Whoops, failed to send!");
    }
}
```

At one point I had a panic occur in the recipient as a result of the message, which meant the thread died. It keeps running along despite the panic until another message comes through at which point:

```
thread 'main' panicked at 'Failed to send!: SendError::Closed(..)', libcore/result.rs:945:5
```

Which is expected, but untimely. In terms of wanting to write reliable software, I'd ideally like to recover from that failure as quickly as possible (possibly by killing the process and letting systemd restart the process) rather than having a ticking timebomb waiting for the next message to cause the main thread panic. I'd originally used `do_send` and the process silently failed until I noticed it was failing.

I could wrap my handler bodies in [catch_unwind](https://doc.rust-lang.org/std/panic/fn.catch_unwind.html) but that doesn't really appeal to me and if anything that should happen at the framework level as a way of cleanly aborting the process.

What's the right way of dealing with panics in actors in other threads?


Discussion / hints:
i am not sure how to handle panics in this case, we may start sync actor in catch_unwind.

if you want to restart process immediately you can use "panic=abort" in your cargo.toml

```
[profile.release]
panic = "abort"
```
Well, to be honest, in first place panic is pretty abnormal condition and I would consider it actually a bug for your code.

So why wouldn't you prevent panic in first place?
> i am not sure how to handle panics in this case, we may start sync actor in catch_unwind.

It seems worthwhile -- if an Arbiter catches a panic, shutdown the actor system on a best-effort basis (probably by sending SystemExit to System), since now it's unlikely that the program is operating correctly. Some panics are worse than others, so it's possible that after a panic invariants have been invalidated.

> if you want to restart process immediately you can use "panic=abort" in your cargo.toml

I thought about that. I'm not expecting panics during the normal course of operation, but when it does a backtrace is really helpful for diagnosing the issue. `abort` is a really heavy hammer.

> So why wouldn't you prevent panic in first place?

I'm a human that occasionally writes buggy code. In the case that I do, I want my software to handle it as gracefully as possible: log the panic, restart. That way I can investigate the panic and remediate it, but restore service. Arguably, the same message/query/whatever will do it again, and that's fine, I'd rather my processes crashloop than appear to be working but either silently fail or are waiting to fail.

For instance, I had an off-by-one error when slicing up a message, leading to a panic when Rust tried to index into a middle of a grapheme. I fixed it by using an Iterator, but still -- in the hours I was asleep my program responded by silently failing until I restarted it and fixed the problem. Had it restarted, I'd have an error log to investigate, but it would have kept working.

> in first place panic is pretty abnormal condition

I agree, which is why the Actor system probably shouldn't continue operating if there's a panic in a separate arbiter thread. After a panic, you can't really guarantee what state the program is in.

If there's interest, I'll send a PR that allows optional use of `std::panic::catch_unwind` to exit.
PR sounds good. but i'd like default behavior is to restart sync arbiter maybe with some counter before exit.
Considering that panic implies abnormal state, I believe it would best to adapt approach similar to `Supervisor`.
I.e. So if actor is stopped(especially abnormally) there should be also way for user to clean-up state in one place so we should rely here on [Supervised](https://actix.rs/actix/actix/trait.Supervised.html) to provide state clean-up
I don't it is possible in general case. we would need to add `UnwindSafe` constraint, which is problematic especially for async code. it is easier to restart whole thread.
There's some state that's hard to clean up from a position of something like `Supervised`, i.e. a poisoned mutex (if you're lucky -- I can imagine other more subtle issues that could occur). I'm not sure restarting the thread would always be the right thing to do.
Ok, lets add optional exit on panic for sync arbiter.


Fix patch:
diff --git a/CHANGES.md b/CHANGES.md
index 313c4f519..5137e97f6 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -1,5 +1,12 @@
 # CHANGES
 
+## [0.7.1] (2018-07-xx)
+
+### Added
+
+* Arbiter now has `Arbiter::builder()` which allows opt-in of behavior to stop
+  the actor system on uncaught panic in any arbiter thread. See #111 for examples.
+
 ## [0.7.0] (2018-07-05)
 
 ### Changed
diff --git a/Cargo.toml b/Cargo.toml
index 49c951a32..0a6b0ef89 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "actix"
-version = "0.7.0"
+version = "0.7.1"
 authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
 description = "Actor framework for Rust"
 readme = "README.md"
diff --git a/src/arbiter.rs b/src/arbiter.rs
index 52e6ac4fa..0b3cc528c 100644
--- a/src/arbiter.rs
+++ b/src/arbiter.rs
@@ -30,22 +30,55 @@ thread_local!(
 /// Arbiter controls event loop in it's thread. Each arbiter runs in separate
 /// thread. Arbiter provides several api for event loop access. Each arbiter
 /// can belongs to specific `System` actor.
+///
+/// By default, a panic in an Arbiter does _not_ stop the rest of the System,
+/// unless the panic is in the System actor. Users of Arbiter can opt into
+/// shutting down the system on panic by using `Arbiter::builder()` and enabling
+/// `stop_system_on_panic`.
 pub struct Arbiter {
     stop: Option<Sender<i32>>,
+    stop_system_on_panic: bool,
 }
 
 impl Actor for Arbiter {
     type Context = Context<Self>;
 }
 
+impl Drop for Arbiter {
+    fn drop(&mut self) {
+        if self.stop_system_on_panic && thread::panicking() {
+            eprintln!("Panic in Arbiter thread, shutting down system.");
+            System::current().stop_with_code(1)
+        }
+    }
+}
+
 impl Arbiter {
     /// Spawn new thread and run event loop in spawned thread.
     /// Returns address of newly created arbiter.
+    /// Does not stop the system on panic.
+    pub fn builder() -> Builder {
+        Builder::new()
+    }
+
+    /// Spawn new thread and run event loop in spawned thread.
+    /// Returns address of newly created arbiter.
+    /// Does not stop the system on panic.
     pub fn new<T: Into<String>>(name: T) -> Addr<Arbiter> {
+        Arbiter::new_with_builder(Arbiter::builder().name(name))
+    }
+
+    /// Spawn new thread and run event loop in spawned thread.
+    /// Returns address of newly created arbiter.
+    fn new_with_builder(builder: Builder) -> Addr<Arbiter> {
         let (tx, rx) = std::sync::mpsc::channel();
 
         let id = Uuid::new_v4();
-        let name = format!("arbiter:{}:{}", id.hyphenated().to_string(), name.into());
+        let name = format!(
+            "arbiter:{}:{}",
+            id.hyphenated().to_string(),
+            builder.name.as_ref().unwrap_or(&"actor".into())
+        );
         let sys = System::current();
 
         let _ = thread::Builder::new().name(name.clone()).spawn(move || {
@@ -61,7 +94,10 @@ impl Arbiter {
             // start arbiter
             let addr =
                 rt.block_on(future::lazy(move || {
-                    let addr = Actor::start(Arbiter { stop: Some(stop) });
+                    let addr = Actor::start(Arbiter {
+                        stop: Some(stop),
+                        stop_system_on_panic: builder.stop_system_on_panic,
+                    });
                     Ok::<_, ()>(addr)
                 })).unwrap();
             ADDR.with(|cell| *cell.borrow_mut() = Some(addr.clone()));
@@ -97,7 +133,10 @@ impl Arbiter {
 
         // start arbiter
         let ctx = Context::with_receiver(rx);
-        let fut = ctx.into_future(Arbiter { stop: None });
+        let fut = ctx.into_future(Arbiter {
+            stop: None,
+            stop_system_on_panic: true, // If the system Arbiter panics, stop the system.
+        });
         let addr = fut.address();
         Arbiter::spawn(fut);
         ADDR.with(|cell| *cell.borrow_mut() = Some(addr.clone()));
@@ -167,14 +206,22 @@ impl Arbiter {
     /// Start new arbiter and then start actor in created arbiter.
     /// Returns `Addr<Syn, A>` of created actor.
     pub fn start<A, F>(f: F) -> Addr<A>
-    where
-        A: Actor<Context = Context<A>>,
-        F: FnOnce(&mut A::Context) -> A + Send + 'static,
+        where
+            A: Actor<Context = Context<A>>,
+            F: FnOnce(&mut A::Context) -> A + Send + 'static,
+    {
+        Arbiter::builder().start(f)
+    }
+
+    fn start_with_builder<A, F>(builder: Builder, f: F) -> Addr<A>
+        where
+            A: Actor<Context = Context<A>>,
+            F: FnOnce(&mut A::Context) -> A + Send + 'static,
     {
         let (stx, srx) = channel::channel(DEFAULT_CAPACITY);
 
         // new arbiter
-        let addr = Arbiter::new("actor");
+        let addr = Arbiter::new_with_builder(builder);
 
         // create actor
         addr.do_send::<Execute>(Execute::new(move || {
@@ -217,3 +264,52 @@ impl<I: Send, E: Send> Handler<Execute<I, E>> for Arbiter {
         msg.exec()
     }
 }
+
+/// Builder struct for an Arbiter.
+pub struct Builder {
+    /// Name of the Arbiter. Defaults to "actor" if unset.
+    name: Option<String>,
+
+    /// Whether the Arbiter will stop the whole System on uncaught panic. Defaults to false.
+    stop_system_on_panic: bool,
+}
+
+impl Builder {
+    fn new() -> Self {
+        Builder {
+            name: None,
+            stop_system_on_panic: false,
+        }
+    }
+
+    /// Sets the name of the Arbiter.
+    pub fn name<T: Into<String>>(mut self, name: T) -> Self {
+        self.name = Some(name.into());
+        self
+    }
+
+    /// Sets the option 'stop_system_on_panic' which controls whether the System is stopped when an
+    /// uncaught panic is thrown from an Actor in the Arbiter.
+    ///
+    /// Defaults to false.
+    pub fn stop_system_on_panic(mut self, stop_system_on_panic: bool) -> Self {
+        self.stop_system_on_panic = stop_system_on_panic;
+        self
+    }
+
+    /// Spawn new thread and run event loop in spawned thread.
+    /// Returns address of newly created arbiter.
+    pub fn build(self) -> Addr<Arbiter> {
+        Arbiter::new_with_builder(self)
+    }
+
+    /// Start new arbiter and then start actor in created arbiter.
+    /// Returns `Addr<Syn, A>` of created actor.
+    pub fn start<A, F>(self, f: F) -> Addr<A>
+        where
+            A: Actor<Context = Context<A>>,
+            F: FnOnce(&mut A::Context) -> A + Send + 'static,
+    {
+        Arbiter::start_with_builder(self, f)
+    }
+}
diff --git a/src/system.rs b/src/system.rs
index ea598ae2b..7b85ee35e 100644
--- a/src/system.rs
+++ b/src/system.rs
@@ -109,7 +109,12 @@ impl System {
 
     /// Stop the system
     pub fn stop(&self) {
-        self.sys.do_send(SystemExit(0));
+        self.stop_with_code(0)
+    }
+
+    /// Stop the system with a particular exit code.
+    pub(crate) fn stop_with_code(&self, code: i32) {
+        self.sys.do_send(SystemExit(code));
     }
 
     pub(crate) fn sys(&self) -> &Addr<SystemArbiter> {
@@ -183,7 +188,7 @@ pub struct SystemRunner {
 
 impl SystemRunner {
     /// This function will start event loop and will finish once the
-    /// `System::stop()` function get called.
+    /// `System::stop()` function is called.
     pub fn run(self) -> i32 {
         let SystemRunner { mut rt, stop, .. } = self;
 


Test patch:
diff --git a/tests/test_arbiter.rs b/tests/test_arbiter.rs
index 38131a12e..7a2b1a188 100644
--- a/tests/test_arbiter.rs
+++ b/tests/test_arbiter.rs
@@ -8,6 +8,9 @@ use futures::Future;
 use std::sync::atomic::{AtomicUsize, Ordering};
 use std::sync::Arc;
 
+#[derive(Debug, Message)]
+struct Panic();
+
 #[derive(Debug, Message)]
 struct Ping(usize);
 
@@ -27,6 +30,14 @@ impl Handler<Ping> for MyActor {
     }
 }
 
+impl Handler<Panic> for MyActor {
+    type Result = ();
+
+    fn handle(&mut self, _: Panic, _: &mut actix::Context<MyActor>) {
+        panic!("Whoops!");
+    }
+}
+
 #[test]
 fn test_start_actor() {
     let count = Arc::new(AtomicUsize::new(0));
@@ -41,6 +52,36 @@ fn test_start_actor() {
     assert_eq!(count.load(Ordering::Relaxed), 1);
 }
 
+#[test]
+fn test_start_actor_builder() {
+    let count = Arc::new(AtomicUsize::new(0));
+    let act_count = Arc::clone(&count);
+
+    System::run(move || {
+        let addr = Arbiter::builder().start(move |_| MyActor(act_count));
+
+        addr.do_send(Ping(1));
+    });
+
+    assert_eq!(count.load(Ordering::Relaxed), 1);
+}
+
+#[test]
+fn test_panic_stops_system() {
+    let count = Arc::new(AtomicUsize::new(0));
+    let act_count = Arc::clone(&count);
+
+    let return_code = System::run(move || {
+        let addr = Arbiter::builder()
+            .stop_system_on_panic(true)
+            .start(move |_| MyActor(act_count));
+
+        addr.do_send(Panic());
+    });
+
+    assert_eq!(return_code, 1);
+}
+
 #[test]
 fn test_start_actor_message() {
     let count = Arc::new(AtomicUsize::new(0));