implement proper error handling

Author: Bobonium

CHANGELOG.md

@@ -1,3 +1,15 @@
+# 0.6.0
+
+ - add error handling
+   - `ActorBuilder.spawn()` now returns a Result, containing either the `ActorWrapper<A>` or an `ActorError`
+   - added `ActorResult` that is returned by all relevant `Actor` functions
+     - configures how the Actor should proceed
+   - Proper panic handling now in all parts of the `Actor` and `Handler<M>` and `ActorFactory<A>`
+     - panic now triggers `Actor.on_panic` providing the source of panic and allows User to determine how to proceed
+       - `Actor.on_panic` is allowed to panic once and will be re-triggered in that case. If another panic happens in the retry, the Actor will be stopped
+     - handling a panic within `ActorFactory<A>.new_actor()` by returning `ActorResult::Restart()` in `Actor.on_panic` can trigger a restart loop that will block the thread until `ActorFactory<A>.new_actor()` was successful
+ - replaced `RestartPolicy` with `ActorResult`
+
 # 0.5.0
 
 - added `send_after` to `ActorWrapper<A>` to allow sending of delayed messages

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "tyra"
-version = "0.5.0"
+version = "0.6.0"
 authors = ["sers.dev <[email protected]>"]
 edition = "2018"
 license = "MIT OR Apache-2.0"
@@ -26,6 +26,7 @@ crossbeam-channel = "^0.5"
 crossbeam-utils = "^0.8"
 dashmap = "^4.0"
 serde = { version = "^1.0", features = ["derive"] }
+thiserror = "1.0.31"
 
 [dev-dependencies]
 bincode = "1.3.3"

examples/actor.rs

@@ -1,8 +1,6 @@
 use std::process::exit;
 use std::time::Duration;
-use tyra::prelude::{
-    Actor, ActorContext, ActorFactory, ActorMessage, ActorSystem, Handler, TyraConfig,
-};
+use tyra::prelude::{Actor, ActorContext, ActorFactory, ActorMessage, ActorResult, ActorSystem, Handler, TyraConfig};
 
 struct MessageA {
     text: String,
@@ -41,18 +39,20 @@ impl ActorFactory<HelloWorld> for HelloWorldFactory {
     }
 }
 impl Handler<MessageA> for HelloWorld {
-    fn handle(&mut self, msg: MessageA, _context: &ActorContext<Self>) {
+    fn handle(&mut self, msg: MessageA, _context: &ActorContext<Self>) -> ActorResult {
         let text: String = [self.text.clone(), String::from(msg.text)].join(" -> ");
         self.count += 1;
-        println!("AAAA: {} Count: {}", text, self.count)
+        println!("AAAA: {} Count: {}", text, self.count);
+        ActorResult::Ok
     }
 }
 
 impl Handler<MessageB> for HelloWorld {
-    fn handle(&mut self, msg: MessageB, _context: &ActorContext<Self>) {
+    fn handle(&mut self, msg: MessageB, _context: &ActorContext<Self>) -> ActorResult {
         let text: String = [self.text.clone(), String::from(msg.text)].join(" -> ");
         self.count -= 1;
-        println!("BBBB: {} Count: {}", text, self.count)
+        println!("BBBB: {} Count: {}", text, self.count);
+        ActorResult::Ok
     }
 }
 

examples/benchmark.rs

@@ -1,9 +1,7 @@
 use std::process::exit;
 use std::thread::sleep;
 use std::time::{Duration, Instant};
-use tyra::prelude::{
-    Actor, ActorContext, ActorFactory, ActorMessage, ActorSystem, Handler, TyraConfig,
-};
+use tyra::prelude::{Actor, ActorContext, ActorFactory, ActorMessage, ActorResult, ActorSystem, Handler, TyraConfig};
 
 struct MessageA {}
 
@@ -41,7 +39,7 @@ impl Benchmark {
 impl Actor for Benchmark {}
 
 impl Handler<MessageA> for Benchmark {
-    fn handle(&mut self, _msg: MessageA, context: &ActorContext<Self>) {
+    fn handle(&mut self, _msg: MessageA, context: &ActorContext<Self>) -> ActorResult {
         if self.count == 0 {
             println!("Sleep 3 now");
             sleep(Duration::from_secs((3) as u64));
@@ -63,6 +61,7 @@ impl Handler<MessageA> for Benchmark {
         if self.count == self.total_msgs {
             context.system.stop(Duration::from_secs(60));
         }
+        ActorResult::Ok
     }
 }
 

examples/bulk_router_benchmark.rs

@@ -1,9 +1,7 @@
 use std::process::exit;
 use std::thread::sleep;
 use std::time::{Duration, Instant};
-use tyra::prelude::{
-    Actor, ActorContext, ActorFactory, ActorMessage, ActorSystem, ActorWrapper, Handler, TyraConfig,
-};
+use tyra::prelude::{Actor, ActorContext, ActorFactory, ActorMessage, ActorResult, ActorSystem, ActorWrapper, Handler, TyraConfig};
 use tyra::router::{AddActorMessage, BulkRouterMessage, RoundRobinRouterFactory};
 
 struct MessageA {}
@@ -63,7 +61,7 @@ impl Benchmark {
 impl Actor for Benchmark {}
 
 impl Handler<MessageA> for Benchmark {
-    fn handle(&mut self, _msg: MessageA, _context: &ActorContext<Self>) {
+    fn handle(&mut self, _msg: MessageA, _context: &ActorContext<Self>) -> ActorResult {
         if self.count == 0 {
             sleep(Duration::from_secs((3) as u64));
             self.start = Instant::now();
@@ -79,6 +77,7 @@ impl Handler<MessageA> for Benchmark {
         if self.count == self.total_msgs {
             self.aggregator.send(Finish {});
         }
+        ActorResult::Ok
     }
 }
 
@@ -116,7 +115,7 @@ impl ActorFactory<Aggregator> for AggregatorFactory {
 }
 
 impl Handler<Finish> for Aggregator {
-    fn handle(&mut self, _msg: Finish, _context: &ActorContext<Self>) {
+    fn handle(&mut self, _msg: Finish, _context: &ActorContext<Self>) -> ActorResult {
         self.actors_finished += 1;
         if self.actors_finished == self.total_actors {
             let duration = self.start.elapsed();
@@ -126,13 +125,15 @@ impl Handler<Finish> for Aggregator {
             );
             self.ctx.system.stop(Duration::from_secs(60));
         }
+        ActorResult::Ok
     }
 }
 
 impl Handler<Start> for Aggregator {
-    fn handle(&mut self, _msg: Start, _context: &ActorContext<Self>) {
+    fn handle(&mut self, _msg: Start, _context: &ActorContext<Self>) -> ActorResult {
         sleep(Duration::from_secs((3) as u64));
         self.start = Instant::now();
+        ActorResult::Ok
     }
 }
 

examples/delayed.rs

@@ -54,8 +54,9 @@ impl ActorFactory<TemplateActor> for TemplateActorFactory {
 //Handler//
 ///////////
 impl Handler<TemplateMessage> for TemplateActor {
-    fn handle(&mut self, _msg: TemplateMessage, _context: &ActorContext<Self>) {
-        println!("{}", _msg.id)
+    fn handle(&mut self, _msg: TemplateMessage, _context: &ActorContext<Self>) -> ActorResult {
+        println!("{}", _msg.id);
+        ActorResult::Ok
     }
 }
 

examples/error.rs

@@ -1,8 +1,6 @@
 use std::process::exit;
 use std::time::Duration;
-use tyra::prelude::{
-    Actor, ActorContext, ActorFactory, ActorMessage, ActorSystem, Handler, TyraConfig,
-};
+use tyra::prelude::{Actor, ActorContext, ActorFactory, ActorMessage, ActorResult, ActorSystem, Handler, TyraConfig};
 
 #[derive(Clone)]
 struct ErrMsg {
@@ -19,12 +17,13 @@ struct ErrActor {
 impl Actor for ErrActor {}
 
 impl Handler<ErrMsg> for ErrActor {
-    fn handle(&mut self, msg: ErrMsg, _context: &ActorContext<Self>) {
+    fn handle(&mut self, msg: ErrMsg, _context: &ActorContext<Self>) -> ActorResult {
         self.counter += 1;
         if msg.text == "sers+1" {
             panic!("ficl");
         }
         println!("Received SERS: {}", self.counter);
+        ActorResult::Ok
     }
 }
 

examples/getting_started.rs

@@ -53,8 +53,9 @@ impl ActorFactory<TemplateActor> for TemplateActorFactory {
 //Handler//
 ///////////
 impl Handler<TemplateMessage> for TemplateActor {
-    fn handle(&mut self, _msg: TemplateMessage, _context: &ActorContext<Self>) {
-        println!("SERS: {}", _msg.id)
+    fn handle(&mut self, _msg: TemplateMessage, _context: &ActorContext<Self>) -> ActorResult {
+        println!("SERS: {}", _msg.id);
+        ActorResult::Ok
     }
 }
 

examples/router.rs

@@ -1,9 +1,7 @@
 use std::process::exit;
 use std::thread::sleep;
 use std::time::Duration;
-use tyra::prelude::{
-    Actor, ActorContext, ActorFactory, ActorMessage, ActorSystem, Handler, TyraConfig,
-};
+use tyra::prelude::{Actor, ActorContext, ActorFactory, ActorMessage, ActorResult, ActorSystem, Handler, TyraConfig};
 use tyra::router::{AddActorMessage, RemoveActorMessage, RoundRobinRouterFactory, RouterMessage};
 
 struct MessageA {}
@@ -23,9 +21,10 @@ impl ActorFactory<HelloWorld> for HelloWorldFactory {
     }
 }
 impl Handler<MessageA> for HelloWorld {
-    fn handle(&mut self, _msg: MessageA, _context: &ActorContext<Self>) {
+    fn handle(&mut self, _msg: MessageA, _context: &ActorContext<Self>) -> ActorResult {
         self.counter += 1;
         println!("Received MSG {}", self.counter);
+        ActorResult::Ok
     }
 }
 

examples/router_benchmark.rs

@@ -1,9 +1,6 @@
 use std::process::exit;
-use std::thread::sleep;
 use std::time::{Duration, Instant};
-use tyra::prelude::{
-    Actor, ActorContext, ActorFactory, ActorMessage, ActorSystem, ActorWrapper, Handler, TyraConfig,
-};
+use tyra::prelude::{Actor, ActorContext, ActorFactory, ActorMessage, ActorResult, ActorSystem, ActorWrapper, Handler, TyraConfig};
 use tyra::router::{AddActorMessage, RoundRobinRouterFactory, RouterMessage};
 
 struct MessageA {}
@@ -53,9 +50,9 @@ impl Benchmark {
 impl Actor for Benchmark {}
 
 impl Handler<MessageA> for Benchmark {
-    fn handle(&mut self, _msg: MessageA, _context: &ActorContext<Self>) {
+    fn handle(&mut self, _msg: MessageA, _context: &ActorContext<Self>) -> ActorResult {
         if self.count == 0 {
-            sleep(Duration::from_secs((3) as u64));
+            //sleep(Duration::from_secs((3) as u64));
             self.start = Instant::now();
         }
         self.count += 1;
@@ -69,6 +66,7 @@ impl Handler<MessageA> for Benchmark {
         if self.count == self.total_msgs {
             self.aggregator.send(Finish {});
         }
+        ActorResult::Ok
     }
 }
 
@@ -106,7 +104,7 @@ impl ActorFactory<Aggregator> for AggregatorFactory {
 }
 
 impl Handler<Finish> for Aggregator {
-    fn handle(&mut self, _msg: Finish, _context: &ActorContext<Self>) {
+    fn handle(&mut self, _msg: Finish, _context: &ActorContext<Self>) -> ActorResult {
         self.actors_finished += 1;
         if self.actors_finished == self.total_actors {
             let duration = self.start.elapsed();
@@ -116,13 +114,15 @@ impl Handler<Finish> for Aggregator {
             );
             self.ctx.system.stop(Duration::from_secs(60));
         }
+        ActorResult::Ok
     }
 }
 
 impl Handler<Start> for Aggregator {
-    fn handle(&mut self, _msg: Start, _context: &ActorContext<Self>) {
-        sleep(Duration::from_secs((3) as u64));
+    fn handle(&mut self, _msg: Start, _context: &ActorContext<Self>) -> ActorResult {
+        //sleep(Duration::from_secs((3) as u64));
         self.start = Instant::now();
+        ActorResult::Ok
     }
 }
 
@@ -132,7 +132,7 @@ fn main() {
 
     let message_count = 10000000;
     // ideal number is "amount of threads - 3"
-    let actor_count = 7;
+    let actor_count = 10;
 
     let router_factory = RoundRobinRouterFactory::new();
     let router = actor_system