Shutdown and termination sig handling (#15)

* Implement graceful shutdown and termination signalling

* ctrlc is not built after rebasing

* fixed ctrlc(score_crates) issue

* use alloc and core libs instead of std whenever possible

Author: ShoroukRamzy

Cargo.lock

@@ -32,6 +32,7 @@ feo-log = { path = "feo-log" }
 feo-logger = { path = "feo-logger" }
 feo-time = { path = "feo-time" }
 feo-tracing = { path = "feo-tracing" }
+ctrlc = { version = "3.4.4", features = ["termination"] }
 futures = "0.3.31"
 human_bytes = "0.4.3"
 iceoryx2 = "0.5.0"
@@ -64,7 +65,6 @@ tracing = { version = "0.1.41", features = [
     "attributes",
 ], default-features = false }
 tracing-subscriber = { version = "0.3.19", default-features = false }
-
 [profile.profiling]
 inherits = "release"
 debug = true

Cargo.toml

@@ -33,7 +33,7 @@ git_override(
 bazel_dep(name = "score_crates", version = "1.0.0")
 git_override(
     module_name = "score_crates",
-    commit = "2f0616c5af797f05788e4e5cad16f37699ea7a17",
+    commit = "0748fc81379faf33dcb5616e7f0f4c17bce79372",
     remote = "https://github.com/eclipse-score/score-crates.git",
 )
 

MODULE.bazel

@@ -89,7 +89,7 @@ Install *bytehound* with the following commands:
 
 ```sh
 wget https://github.com/koute/bytehound/releases/download/0.11.0/bytehound-x86_64-unknown-linux-gnu.tgz
-tar xzf bytehound-x86_64-unknown-linux-gnu.tgz bytehound libbytehound.so 
+tar xzf bytehound-x86_64-unknown-linux-gnu.tgz bytehound libbytehound.so
 mv bytehound libbytehound.so $HOME/.cargo/bin
 ```
 
@@ -113,5 +113,3 @@ bytehound server memory-profiling_*.dat
 
 Click on the [link](http://127.0.0.1:8080) in the output to open the browser and
 see the results. Setup ssh port forwarding if needed when working remote (`ssh -L 8080:localhost:8080 host`).
-
-

README.md

@@ -188,6 +188,7 @@ mod direct_mpsc {
 
         let agent_id = params.agent_id;
         PrimaryConfig {
+            id: agent_id,
             cycle_time: params.feo_cycle_time,
             activity_dependencies: app_config.activity_dependencies(),
             // With only one agent, we cannot attach a recorder
@@ -237,13 +238,24 @@ mod direct_sockets {
     ) -> PrimaryConfig {
         let agent_id = params.agent_id;
         PrimaryConfig {
+            id: agent_id,
             cycle_time: params.feo_cycle_time,
             activity_dependencies: app_config.activity_dependencies(),
             recorder_ids: app_config.recorders(),
             worker_assignments: app_config.worker_assignments().remove(&agent_id).unwrap(),
             timeout: Duration::from_secs(10),
             connection_timeout: Duration::from_secs(10),
             endpoint: endpoint(&app_config, signalling),
+            activity_agent_map: app_config
+                .activity_worker_map()
+                .iter()
+                .map(|(act_id, w_id)| {
+                    (
+                        *act_id,
+                        app_config.worker_agent_map().get(w_id).copied().unwrap(),
+                    )
+                })
+                .collect(),
         }
     }
 

examples/rust/cycle-benchmark/src/bin/cycle_bench.rs

@@ -58,7 +58,7 @@ rust_library(
     crate_features = [
         "com_linux_shm",
         "recording",
-        "signalling_relayed_tcp",
+        "signalling_direct_tcp",
     ],
     crate_name = "mini_adas",
     proc_macro_deps = [
@@ -131,7 +131,7 @@ rust_binary(
     ],
     crate_features = [
         "recording",
-        "signalling_relayed_tcp",
+        "signalling_direct_tcp",
     ],
     visibility = ["//visibility:public"],
     deps = [

examples/rust/mini-adas/BUILD.bazel

@@ -93,5 +93,19 @@ cargo run --no-default-features --features signalling_direct_tcp --bin adas_seco
 cargo run --no-default-features --features signalling_direct_tcp --bin adas_secondary 2
 ```
 
+### Direct Mode with a Recorder
+
+To run with a recorder in direct mode.
+
+```sh
+# Use 400ms cycle time and wait for recorder 900
+cargo run --no-default-features --features "signalling_direct_tcp,com_iox2" --bin adas_primary -- 400 900
+```
+
+```sh
+# Start recorder with ID 900 in direct mode
+cargo run --no-default-features --features "signalling_direct_tcp,recording,com_iox2" --bin adas_recorder -- 900
+```
+
 Note that for mpsc-only signalling, there can be only a primary process without
-any secondaries or recorders, because mpsc does not support inter-process signalling.
+any secondaries or recorders, because mpsc does not support inter-process signalling.

examples/rust/mini-adas/README.md

@@ -98,7 +98,9 @@ impl Activity for Camera {
     }
 
     #[instrument(name = "Camera shutdown")]
-    fn shutdown(&mut self) {}
+    fn shutdown(&mut self) {
+        debug!("Shutting down Camera activity {}", self.activity_id);
+    }
 }
 
 /// Radar activity
@@ -161,7 +163,9 @@ impl Activity for Radar {
     }
 
     #[instrument(name = "Radar shutdown")]
-    fn shutdown(&mut self) {}
+    fn shutdown(&mut self) {
+        debug!("Shutting down Radar activity {}", self.activity_id);
+    }
 }
 
 /// Neural network activity
@@ -251,7 +255,9 @@ impl Activity for NeuralNet {
     }
 
     #[instrument(name = "NeuralNet shutdown")]
-    fn shutdown(&mut self) {}
+    fn shutdown(&mut self) {
+        debug!("Shutting down NeuralNet activity {}", self.activity_id);
+    }
 }
 
 /// Emergency braking activity
@@ -328,7 +334,12 @@ impl Activity for EmergencyBraking {
     }
 
     #[instrument(name = "EmergencyBraking shutdown")]
-    fn shutdown(&mut self) {}
+    fn shutdown(&mut self) {
+        debug!(
+            "Shutting down EmergencyBraking activity {}",
+            self.activity_id
+        );
+    }
 }
 
 /// Brake controller activity
@@ -378,7 +389,12 @@ impl Activity for BrakeController {
     }
 
     #[instrument(name = "BrakeController shutdown")]
-    fn shutdown(&mut self) {}
+    fn shutdown(&mut self) {
+        debug!(
+            "Shutting down BrakeController activity {}",
+            self.activity_id
+        );
+    }
 }
 
 /// Environment renderer activity
@@ -422,7 +438,12 @@ impl Activity for EnvironmentRenderer {
     }
 
     #[instrument(name = "EnvironmentRenderer shutdown")]
-    fn shutdown(&mut self) {}
+    fn shutdown(&mut self) {
+        debug!(
+            "Shutting down EnvironmentRenderer activity {}",
+            self.activity_id
+        );
+    }
 }
 
 /// Steering controller activity
@@ -470,7 +491,12 @@ impl Activity for SteeringController {
     }
 
     #[instrument(name = "SteeringController shutdown")]
-    fn shutdown(&mut self) {}
+    fn shutdown(&mut self) {
+        debug!(
+            "Shutting down SteeringController activity {}",
+            self.activity_id
+        );
+    }
 }
 
 /// Create an activity input.

examples/rust/mini-adas/src/activities/components.rs

@@ -99,6 +99,7 @@ mod cfg {
 
     pub(super) fn make_config(params: Params) -> PrimaryConfig {
         PrimaryConfig {
+            id: AGENT_ID,
             cycle_time: params.feo_cycle_time,
             activity_dependencies: activity_dependencies(),
             // With only one agent, we cannot attach a recorder
@@ -113,51 +114,91 @@ mod cfg {
 #[cfg(feature = "signalling_direct_tcp")]
 mod cfg {
     use super::{check_ids, Duration, Params, AGENT_ID};
-    use feo::agent::NodeAddress;
-    use feo::ids::AgentId;
-    use mini_adas::config::{activity_dependencies, agent_assignments, BIND_ADDR};
-    use std::collections::HashSet;
+    use feo::{
+        agent::NodeAddress,
+        ids::{ActivityId, AgentId, WorkerId},
+    };
+    use mini_adas::config::{
+        activity_dependencies, agent_assignments, worker_agent_map, BIND_ADDR,
+    };
+    use std::collections::{HashMap, HashSet};
 
     pub(super) use feo::agent::direct::primary::{Primary, PrimaryConfig};
 
     pub(super) fn make_config(params: Params) -> PrimaryConfig {
         let agent_ids: HashSet<AgentId> = agent_assignments().keys().copied().collect();
         check_ids(&params.recorder_ids, &agent_ids);
 
+        let activity_worker_map: HashMap<ActivityId, WorkerId> = agent_assignments()
+            .values()
+            .flat_map(|vec| {
+                vec.iter()
+                    .flat_map(move |(wid, aid_b)| aid_b.iter().map(|v| (v.0, *wid)))
+            })
+            .collect();
+
         PrimaryConfig {
+            id: AGENT_ID,
             cycle_time: params.feo_cycle_time,
             activity_dependencies: activity_dependencies(),
             recorder_ids: params.recorder_ids,
             worker_assignments: agent_assignments().remove(&AGENT_ID).unwrap(),
             timeout: Duration::from_secs(10),
             connection_timeout: Duration::from_secs(10),
             endpoint: NodeAddress::Tcp(BIND_ADDR),
+            activity_agent_map: activity_worker_map
+                .iter()
+                .map(|(activity_id, worker_id)| {
+                    let agent_id = worker_agent_map().get(worker_id).copied().unwrap();
+                    (*activity_id, agent_id)
+                })
+                .collect(),
         }
     }
 }
 
 #[cfg(feature = "signalling_direct_unix")]
 mod cfg {
     use super::{check_ids, Duration, Params, AGENT_ID};
-    use feo::agent::NodeAddress;
-    use feo::ids::AgentId;
-    use mini_adas::config::{activity_dependencies, agent_assignments, socket_paths};
-    use std::collections::HashSet;
+    use feo::{
+        agent::NodeAddress,
+        ids::{ActivityId, AgentId, WorkerId},
+    };
+    use mini_adas::config::{
+        activity_dependencies, agent_assignments, socket_paths, worker_agent_map,
+    };
+    use std::collections::{HashMap, HashSet};
 
     pub(super) use feo::agent::direct::primary::{Primary, PrimaryConfig};
 
     pub(super) fn make_config(params: Params) -> PrimaryConfig {
         let agent_ids: HashSet<AgentId> = agent_assignments().keys().copied().collect();
         check_ids(&params.recorder_ids, &agent_ids);
 
+        let activity_worker_map: HashMap<ActivityId, WorkerId> = agent_assignments()
+            .values()
+            .flat_map(|vec| {
+                vec.iter()
+                    .flat_map(move |(wid, aid_b)| aid_b.iter().map(|v| (v.0, *wid)))
+            })
+            .collect();
+
         PrimaryConfig {
+            id: AGENT_ID,
             cycle_time: params.feo_cycle_time,
             activity_dependencies: activity_dependencies(),
             recorder_ids: params.recorder_ids,
             worker_assignments: agent_assignments().remove(&AGENT_ID).unwrap(),
             timeout: Duration::from_secs(10),
             connection_timeout: Duration::from_secs(10),
             endpoint: NodeAddress::UnixSocket(socket_paths().0),
+            activity_agent_map: activity_worker_map
+                .iter()
+                .map(|(activity_id, worker_id)| {
+                    let agent_id = worker_agent_map().get(worker_id).copied().unwrap();
+                    (*activity_id, agent_id)
+                })
+                .collect(),
         }
     }
 }

examples/rust/mini-adas/src/bin/adas_primary.rs

@@ -78,6 +78,7 @@ rust_library(
         "//feo-log:libfeo_log_rust",
         "//feo-time:libfeo_time_rust",
         "//feo-tracing:libfeo_tracing_rust",
+        "@score_crates//:ctrlc",
         "@score_crates//:libc",
         "@score_crates//:mio",
     ],
@@ -149,6 +150,7 @@ rust_library(
         "//feo-log:libfeo_log_rust",
         "//feo-time:libfeo_time_rust",
         "//feo-tracing:libfeo_tracing_rust",
+        "@score_crates//:ctrlc",
         "@score_crates//:libc",
         "@score_crates//:mio",
         "@score_crates//:postcard",