handling startup timeout (#21)

* handling startup timeout

* allow Scheduler::new function to take more than element

Author: MohamedRady-99

Cargo.lock

@@ -195,6 +195,7 @@ mod direct_mpsc {
             recorder_ids: vec![],
             worker_assignments: app_config.worker_assignments().remove(&agent_id).unwrap(),
             timeout: Duration::from_secs(10),
+            startup_timeout: Duration::from_secs(10),
         }
     }
 
@@ -245,6 +246,7 @@ mod direct_sockets {
             worker_assignments: app_config.worker_assignments().remove(&agent_id).unwrap(),
             timeout: Duration::from_secs(10),
             connection_timeout: Duration::from_secs(10),
+            startup_timeout: Duration::from_secs(10),
             endpoint: endpoint(&app_config, signalling),
             activity_agent_map: app_config
                 .activity_worker_map()
@@ -337,6 +339,7 @@ mod relayed_sockets {
             worker_assignments: app_config.worker_assignments().remove(&agent_id).unwrap(),
             timeout: Duration::from_secs(10),
             connection_timeout: Duration::from_secs(10),
+            startup_timeout: Duration::from_secs(10),
             bind_address_senders: endpoints.0,
             bind_address_receivers: endpoints.1,
             id: agent_id,

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

@@ -106,6 +106,7 @@ mod cfg {
             recorder_ids: vec![],
             worker_assignments: agent_assignments().remove(&AGENT_ID).unwrap(),
             timeout: Duration::from_secs(10),
+            startup_timeout: Duration::from_secs(10),
         }
     }
 }
@@ -144,6 +145,7 @@ mod cfg {
             worker_assignments: agent_assignments().remove(&AGENT_ID).unwrap(),
             timeout: Duration::from_secs(10),
             connection_timeout: Duration::from_secs(10),
+            startup_timeout: Duration::from_secs(10),
             endpoint: NodeAddress::Tcp(BIND_ADDR),
             activity_agent_map: activity_worker_map
                 .iter()
@@ -190,6 +192,7 @@ mod cfg {
             worker_assignments: agent_assignments().remove(&AGENT_ID).unwrap(),
             timeout: Duration::from_secs(10),
             connection_timeout: Duration::from_secs(10),
+            startup_timeout: Duration::from_secs(10),
             endpoint: NodeAddress::UnixSocket(socket_paths().0),
             activity_agent_map: activity_worker_map
                 .iter()
@@ -233,6 +236,7 @@ mod cfg {
             worker_assignments: agent_assignments().remove(&AGENT_ID).unwrap(),
             timeout: Duration::from_secs(10),
             connection_timeout: Duration::from_secs(10),
+            startup_timeout: Duration::from_secs(10),
             bind_address_senders: NodeAddress::Tcp(BIND_ADDR),
             bind_address_receivers: NodeAddress::Tcp(BIND_ADDR2),
             id: AGENT_ID,
@@ -273,6 +277,7 @@ mod cfg {
             worker_assignments: agent_assignments().remove(&AGENT_ID).unwrap(),
             timeout: Duration::from_secs(10),
             connection_timeout: Duration::from_secs(10),
+            startup_timeout: Duration::from_secs(10),
             bind_address_senders: NodeAddress::UnixSocket(socket_paths().0),
             bind_address_receivers: NodeAddress::UnixSocket(socket_paths().1),
             id: AGENT_ID,

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

@@ -48,6 +48,8 @@ pub struct PrimaryConfig {
     pub timeout: Duration,
     /// Timeout for waiting on initial connections from workers/recorders.
     pub connection_timeout: Duration,
+    /// Timeout for waiting on activities to become ready during startup.
+    pub startup_timeout: Duration,
     /// Endpoint on which the connector of the scheduler waits for connections
     pub endpoint: NodeAddress,
     /// Map of all activities to agent ids
@@ -72,6 +74,7 @@ impl Primary {
             endpoint,
             timeout,
             connection_timeout,
+            startup_timeout,
             activity_agent_map,
             ..
         } = config;
@@ -145,6 +148,7 @@ impl Primary {
             config.id,
             cycle_time,
             timeout,
+            startup_timeout,
             activity_dependencies,
             connector,
             recorder_ids,

feo/src/agent/direct/primary.rs

@@ -44,6 +44,8 @@ pub struct PrimaryConfig {
     pub worker_assignments: Vec<(WorkerId, Vec<ActivityIdAndBuilder>)>,
     /// Receive timeout of the scheduler's connector
     pub timeout: Duration,
+    /// Timeout for waiting on activities to become ready during startup.
+    pub startup_timeout: Duration,
 }
 
 /// Primary agent
@@ -62,6 +64,7 @@ impl Primary {
             activity_dependencies,
             recorder_ids,
             timeout,
+            startup_timeout,
             ..
         } = config;
 
@@ -112,6 +115,7 @@ impl Primary {
             config.id,
             cycle_time,
             timeout,
+            startup_timeout,
             activity_dependencies,
             connector,
             recorder_ids,

feo/src/agent/direct/primary_mpsc.rs

@@ -47,6 +47,8 @@ pub struct PrimaryConfig {
     pub timeout: Duration,
     /// Timeout for waiting on initial connections from workers/recorders.
     pub connection_timeout: Duration,
+    /// Timeout for waiting on activities to become ready during startup.
+    pub startup_timeout: Duration,
     /// The socket address to which secondary agents' senders shall connect
     pub bind_address_senders: NodeAddress,
     /// The socket address to which secondary agents' receivers shall connect
@@ -80,6 +82,7 @@ impl Primary {
             worker_assignments,
             timeout,
             connection_timeout,
+            startup_timeout,
             worker_agent_map,
             activity_worker_map,
         } = config;
@@ -152,6 +155,7 @@ impl Primary {
             id,
             cycle_time,
             timeout,
+            startup_timeout,
             activity_dependencies,
             connector,
             recorder_ids,

feo/src/agent/relayed/primary.rs

@@ -37,7 +37,8 @@ pub(crate) struct Scheduler {
     cycle_time: feo_time::Duration,
     /// Timeout of receive function
     receive_timeout: core::time::Duration,
-
+    /// Timeout for waiting on activities to become ready during startup.
+    startup_timeout: core::time::Duration,
     /// For each activity: list of activities it depends on
     activity_depends: HashMap<ActivityId, Vec<ActivityId>>,
     /// Map keeping track of activity states
@@ -55,10 +56,12 @@ pub(crate) struct Scheduler {
 }
 
 impl Scheduler {
+    #[allow(clippy::too_many_arguments)]
     pub(crate) fn new(
         agent_id: AgentId,
         feo_cycle_time: feo_time::Duration,
         receive_timeout: core::time::Duration,
+        startup_timeout: core::time::Duration,
         activity_depends: HashMap<ActivityId, Vec<ActivityId>>,
         connector: Box<dyn ConnectScheduler>,
         recorder_ids: Vec<AgentId>,
@@ -84,6 +87,7 @@ impl Scheduler {
             agent_id,
             cycle_time: feo_cycle_time,
             receive_timeout,
+            startup_timeout,
             activity_depends,
             connector,
             activity_states,
@@ -117,10 +121,28 @@ impl Scheduler {
             Self::startup_activity(activity_id, &self.recorder_ids, &mut self.connector).unwrap();
         }
 
-        // Wait until all activities have returned their ready signal
+        // Wait until all activities have returned their ready signal, with a timeout.
+        let startup_start = Instant::now();
         while !self.all_ready() {
-            self.wait_next_ready()
-                .expect("failed while waiting for ready signal");
+            if startup_start.elapsed() > self.startup_timeout {
+                let reason = alloc::format!(
+                    "Startup timeout of {:?} exceeded. Not all activities became ready.",
+                    self.startup_timeout
+                );
+                error!("{}", reason);
+                self.shutdown_gracefully(&reason);
+                return;
+            }
+            if self.wait_next_ready().is_err() {
+                // An error here (like a timeout on receive) can also be a startup failure.
+                let reason = alloc::format!(
+                    "Failed to receive ready signal from all activities within startup timeout {:?}.",
+                    self.startup_timeout
+                );
+                error!("{}", reason);
+                self.shutdown_gracefully(&reason);
+                return;
+            }
         }
 
         // Loop the FEO task chain