Networking

CHANGELOG.md

@@ -1,3 +1,32 @@
+# 1.1.0
+
+ - upgrade dependencies
+ - `LeastMessageRouter.min_mailbox_size` is now working correctly
+ - Routers now come with 2 new configurations to configure behavior
+   - `stop_on_system_stop` => if false, the user needs to manually stop the router for a clean and quick shutdown of the system
+   - `stop_on_empty_targets` => automatically stops the router if there are no more targets to receive a message to be routed. This does not apply to manually removed targets
+ - Routers now automatically remove stopped actors from their target pool if they have been stopped
+ - `Sharded_Router` now makes use of `HashRing`
+ - `ActorMessage` now needs to implement `Hash` and no longer need to explicitly implement `ActorMessage.get_id()` to be able to properly make use of the `ShardedRouter`
+   - renamed `ActorMessage.get_id()` to `ActorMessage.get_hash()` and changed return type to u64
+   - changed default implementation to `ActorMessage.get_hash()` to make use of `Hash` implementation
+ - `ActorMessage` now needs to implement `Serialize` 
+   - this requirement comes from the fact, that all messages need to be serializable in theory to be able to be sent to other actor systems
+   - if a message is really intended to be sent it should obviously also implement `Deserialize`
+ - Added `.is_mailbox_stopped()`, `is_stopped()` and `wait_for_stop()` to `ActorWrapper<A>`
+ - Added ability to re-initialize `ActorWrapper<A>` after deserializing
+ - Added `general.graceful_timeout_in_seconds` to config
+   - reworked `system.stop()` to make use of configured value
+     - added `system.stop_override_graceful_termination_timeout(Duration)` to allow override of configured default
+   - value is also used for signal handling graceful timeout and timeout for network manager
+ - Added `ActorBuilder<A>.spawn_multiple()`
+ - All routers now support a `SendToAllTargetsMessage<M>` that will forward `M` to all active targets
+ - Users can now use `ActorBuilder.get_existing(address: ActorAddress)` to get an `ActorWrapper<A>` for an already existing `Actor`
+   - `ActorBuilder.spawn()` will continue to return the `ActorWrapper<A>` for already existing actors
+ - renamed config `global.name` to `global.hostname`
+     - also renamed `ActorAddress.remote` to `ActorAddress.hostname`
+     - added new config `global.name` that defaults to the cargo package name or `tyra` if the application is not built using cargo
+
 # 1.0.0
 
  - added `LeastMessageRouter`

Cargo.toml

@@ -18,17 +18,27 @@ name = "tyra"
 path = "src/lib.rs"
 
 [dependencies]
-config = "0.13.4"
-hostname = "0.3.1"
-num_cpus = "1.16.0"
+config = "0.15.18"
+hostname = "0.4.1"
+num_cpus = "1.17.0"
 threadpool = "1.8.1"
-crossbeam-channel = "0.5.13"
-flume = "0.10.14"
-dashmap = "5.5.3"
-serde = { version = "1.0", features = ["derive"] }
-thiserror = "1.0"
-log = "0.4"
+crossbeam-channel = "0.5.15"
+flume = "0.11.1"
+dashmap = "6.1.0"
+serde = { version = "1.0.228", features = ["derive"] }
+serde-encrypt = "0.7.0"
+thiserror = "2.0.17"
+log = "0.4.28"
+ctrlc = { version = "3.5.0", features = ["termination"] }
+mio = {version = "1.0.4", features = ["os-poll", "os-ext", "net"]}
+quiche = "0.24.6"
+io-arc = "1.0.0"
+hashring = "0.3.6"
+bincode = "1.3.3"
+regex = "1.11.3"
+trust-dns-resolver = "0.23.2"
+
 
 [dev-dependencies]
-bincode = "1.3.3"
-ntest = "0.8.1"
+ntest = "0.9.3"
+simple_logger = "5.0.0"

README.md

@@ -99,7 +99,7 @@ impl ActorFactory<TestActor> for TestActorFactory {
 impl Handler<TestMessage> for TestActor {
     fn handle(&mut self, _msg: TestMessage, context: &ActorContext<Self>) -> Result<ActorResult, Box<dyn Error>> {
         println!("HELLO WORLD!");
-        context.system.stop(Duration::from_millis(1000));
+        context.system.stop();
         Ok(ActorResult::Ok)
     }
 }

examples/benchmark_bulk_router.rs

@@ -1,18 +1,22 @@
+use serde::Serialize;
 use std::error::Error;
 use std::process::exit;
 use std::thread::sleep;
 use std::time::{Duration, Instant};
 use tyra::prelude::*;
 use tyra::router::{AddActorMessage, BulkRouterMessage, RoundRobinRouterFactory};
 
+#[derive(Hash, Serialize)]
 struct MessageA {}
 
 impl ActorMessage for MessageA {}
 
+#[derive(Hash, Serialize)]
 struct Finish {}
 
 impl ActorMessage for Finish {}
 
+#[derive(Hash, Serialize)]
 struct Start {}
 
 impl ActorMessage for Start {}
@@ -139,7 +143,7 @@ impl Handler<Finish> for Aggregator {
                 "{} It took {:?} to finish {} actors",
                 self.name, duration, self.total_actors
             );
-            self.ctx.system.stop(Duration::from_secs(60));
+            self.ctx.system.stop();
         }
         Ok(ActorResult::Ok)
     }
@@ -165,7 +169,7 @@ fn main() {
     // ideal number is "amount of threads - 3"
     let actor_count = 7;
 
-    let router_factory = RoundRobinRouterFactory::new();
+    let router_factory = RoundRobinRouterFactory::new(true, true);
     let router = actor_system
         .builder()
         .spawn("benchmark-router", router_factory)

examples/benchmark_router_round_robin.rs

@@ -1,17 +1,21 @@
+use serde::Serialize;
 use std::error::Error;
 use std::process::exit;
-use std::time::{Duration, Instant};
+use std::time::Instant;
 use tyra::prelude::*;
 use tyra::router::{AddActorMessage, RoundRobinRouterFactory};
 
+#[derive(Hash, Serialize)]
 struct MessageA {}
 
 impl ActorMessage for MessageA {}
 
+#[derive(Hash, Serialize)]
 struct Finish {}
 
 impl ActorMessage for Finish {}
 
+#[derive(Hash, Serialize)]
 struct Start {}
 
 impl ActorMessage for Start {}
@@ -134,7 +138,7 @@ impl Handler<Finish> for Aggregator {
                 "{} It took {:?} to finish {} actors",
                 self.name, duration, self.total_actors
             );
-            self.ctx.system.stop(Duration::from_secs(60));
+            self.ctx.system.stop();
         }
         Ok(ActorResult::Ok)
     }
@@ -158,7 +162,7 @@ fn main() {
     let message_count = 10000000;
     let actor_count = 10;
 
-    let router_factory = RoundRobinRouterFactory::new();
+    let router_factory = RoundRobinRouterFactory::new(true, true);
     let router = actor_system
         .builder()
         .spawn("benchmark-router", router_factory)

examples/benchmark_single_actor.rs

@@ -1,8 +1,10 @@
+use serde::Serialize;
 use std::error::Error;
 use std::process::exit;
-use std::time::{Duration, Instant};
+use std::time::Instant;
 use tyra::prelude::*;
 
+#[derive(Hash, Serialize)]
 struct MessageA {}
 
 impl ActorMessage for MessageA {}
@@ -61,7 +63,7 @@ impl Handler<MessageA> for Benchmark {
             );
         }
         if self.count == self.total_msgs {
-            context.system.stop(Duration::from_secs(60));
+            context.system.stop();
         }
         Ok(ActorResult::Ok)
     }

examples/benchmark_single_actor_process_after_send.rs

@@ -1,8 +1,10 @@
+use serde::Serialize;
 use std::error::Error;
 use std::process::exit;
 use std::time::{Duration, Instant};
 use tyra::prelude::*;
 
+#[derive(Hash, Serialize)]
 struct MessageA {}
 
 impl ActorMessage for MessageA {}
@@ -61,7 +63,7 @@ impl Handler<MessageA> for Benchmark {
             );
         }
         if self.count == self.total_msgs {
-            context.system.stop(Duration::from_secs(60));
+            context.system.stop();
         }
         Ok(ActorResult::Ok)
     }

examples/benchmark_single_actor_process_after_send_single_thread.rs

@@ -1,8 +1,10 @@
+use serde::Serialize;
 use std::error::Error;
 use std::process::exit;
 use std::time::{Duration, Instant};
 use tyra::prelude::*;
 
+#[derive(Hash, Serialize)]
 struct MessageA {}
 
 impl ActorMessage for MessageA {}
@@ -61,7 +63,7 @@ impl Handler<MessageA> for Benchmark {
             );
         }
         if self.count == self.total_msgs {
-            context.system.stop(Duration::from_secs(60));
+            context.system.stop();
         }
         Ok(ActorResult::Ok)
     }

examples/benchmark_single_actor_single_thread.rs

@@ -1,8 +1,10 @@
+use serde::Serialize;
 use std::error::Error;
 use std::process::exit;
-use std::time::{Duration, Instant};
+use std::time::Instant;
 use tyra::prelude::*;
 
+#[derive(Hash, Serialize)]
 struct MessageA {}
 
 impl ActorMessage for MessageA {}
@@ -61,7 +63,7 @@ impl Handler<MessageA> for Benchmark {
             );
         }
         if self.count == self.total_msgs {
-            context.system.stop(Duration::from_secs(60));
+            context.system.stop();
         }
         Ok(ActorResult::Ok)
     }

examples/net.rs

@@ -0,0 +1,20 @@
+use simple_logger::SimpleLogger;
+use tyra::prelude::*;
+
+fn main() {
+    SimpleLogger::new().init().unwrap();
+
+    // generate config
+    let mut actor_config = TyraConfig::new().unwrap();
+    actor_config.cluster.enabled = true;
+    let cluster = ThreadPoolConfig::new(22, 4, 4, 1.00);
+    actor_config
+        .thread_pool
+        .config
+        .insert(String::from("mio"), cluster);
+    // start system with config
+    let actor_system = ActorSystem::new(actor_config);
+
+
+    std::process::exit(actor_system.await_shutdown());
+}

examples/quickstart.rs

@@ -1,8 +1,10 @@
+use serde::Serialize;
 use std::error::Error;
-use std::time::Duration;
 use tyra::prelude::*;
 
 // define an `ActorMessage` that can be sent to `Actors` that implement the corresponding `Handler<T>`
+
+#[derive(Hash, Serialize)]
 struct TestMessage {}
 impl TestMessage {
     pub fn new() -> Self {
@@ -44,7 +46,7 @@ impl Handler<TestMessage> for TestActor {
         context: &ActorContext<Self>,
     ) -> Result<ActorResult, Box<dyn Error>> {
         println!("HELLO WORLD!");
-        context.system.stop(Duration::from_millis(1000));
+        context.system.stop();
         Ok(ActorResult::Ok)
     }
 }

examples/serialize.rs

@@ -1,17 +1,17 @@
 use serde::{Deserialize, Serialize};
 use std::error::Error;
 use std::process::exit;
-use std::time::{Duration, Instant};
+use std::time::Duration;
 use tyra::prelude::*;
 
-#[derive(Serialize, Deserialize, PartialEq, Debug, Clone)]
+#[derive(Serialize, Deserialize, Hash, Clone)]
 struct TestMsg {
     content: String,
+    actor_wrapper: ActorWrapper<RemoteActor>,
 }
 
 impl ActorMessage for TestMsg {}
 
-#[derive(Clone)]
 struct RemoteActor {}
 
 impl Actor for RemoteActor {
@@ -24,10 +24,9 @@ impl Actor for RemoteActor {
         if result.is_err() {
             return Ok(ActorResult::Ok);
         }
-        let decoded: TestMsg = result.unwrap();
-        context
-            .actor_ref
-            .send_after(decoded, Duration::from_millis(50))?;
+        let mut deserialized: TestMsg = result.unwrap();
+        deserialized.actor_wrapper.init_after_deserialize(&context.system);
+        deserialized.actor_wrapper.send_after(deserialized.clone(), Duration::from_millis(50))?;
         Ok(ActorResult::Ok)
     }
 }
@@ -36,10 +35,11 @@ impl Handler<TestMsg> for RemoteActor {
     fn handle(
         &mut self,
         msg: TestMsg,
-        _context: &ActorContext<Self>,
+        context: &ActorContext<Self>,
     ) -> Result<ActorResult, Box<dyn Error>> {
         println!("{}", msg.content);
-        Ok(ActorResult::Ok)
+        context.system.stop();
+        Ok(ActorResult::Stop)
     }
 }
 
@@ -59,19 +59,15 @@ fn main() {
     let actor_system = ActorSystem::new(actor_config);
 
     let hw = RemoteActorFactory {};
-    let x = actor_system.builder().spawn("hello-world", hw).unwrap();
+    let remote_actor = actor_system.builder().spawn("hello-world", hw).unwrap();
     let msg = TestMsg {
         content: String::from("Hello World!"),
+        actor_wrapper: remote_actor.clone(),
     };
     let serialized = bincode::serialize(&msg).unwrap();
-    actor_system.send_to_address(x.get_address(), SerializedMessage::new(serialized));
-    let start = Instant::now();
+    actor_system.send_to_address(remote_actor.get_address(), serialized);
 
-    actor_system.stop(Duration::from_secs(10));
     let result = actor_system.await_shutdown();
 
-    let duration = start.elapsed();
-    println!("It took {:?} to send stop", duration);
-
     exit(result);
 }