extending shutdown to handle single agent setup in relayed mode (#19)

* extending shutdown to handle single agent setup in relayed mode

* extending shutdown to handle single agent setup in relayed mode

---------

Signed-off-by: ShoroukRamzy <82869891+ShoroukRamzy@users.noreply.github.com>

Author: ShoroukRamzy

examples/rust/mini-adas/src/config.rs

@@ -133,7 +133,6 @@ pub fn agent_assignments() -> HashMap<AgentId, Vec<(WorkerId, Vec<ActivityIdAndB
     ]
     .into_iter()
     .collect();
-
     #[cfg(feature = "signalling_direct_mpsc")]
     let assignment = [(100.into(), vec![w40, w41, w42, w43, w44])]
         .into_iter()

feo/src/agent/relayed/primary.rs

@@ -184,7 +184,7 @@ impl Primary {
             th.join().unwrap();
         }
         // Wait for the communication relay threads to complete their shutdown.
-        for th in self.relay_threads.drain(..) {
+        for th in core::mem::take(&mut self.relay_threads) {
             th.join().unwrap();
         }
         debug!("Primary finished!!");

feo/src/signalling/relayed/connectors/scheduler.rs

@@ -30,8 +30,8 @@ use std::thread::JoinHandle;
 pub(crate) struct SchedulerConnector<Inter: IsChannel, Intra: IsChannel> {
     local_workers: HashSet<WorkerId>,
     intra_receiver: Intra::MultiReceiver,
-    ipc_receive_relay: PrimaryReceiveRelay<Inter, Intra>,
-    ipc_send_relay: PrimarySendRelay<Inter>,
+    ipc_receive_relay: Option<PrimaryReceiveRelay<Inter, Intra>>,
+    ipc_send_relay: Option<PrimarySendRelay<Inter>>,
     worker_sender: Intra::MultiSender,
     timeout: Duration,
     worker_connector_builders: Option<HashMap<WorkerId, Builder<WorkerConnector>>>,
@@ -45,8 +45,8 @@ impl<Inter: IsChannel, Intra: IsChannel> SchedulerConnector<Inter, Intra> {
     pub fn with_fields(
         local_workers: HashSet<WorkerId>,
         intra_receiver: Intra::MultiReceiver,
-        ipc_receive_relay: PrimaryReceiveRelay<Inter, Intra>,
-        ipc_send_relay: PrimarySendRelay<Inter>,
+        ipc_receive_relay: Option<PrimaryReceiveRelay<Inter, Intra>>,
+        ipc_send_relay: Option<PrimarySendRelay<Inter>>,
         worker_sender: Intra::MultiSender,
         timeout: Duration,
         worker_connector_builders: Option<HashMap<WorkerId, Builder<WorkerConnector>>>,
@@ -72,7 +72,11 @@ impl<Inter: IsChannel, Intra: IsChannel> SchedulerConnector<Inter, Intra> {
         agent_id: AgentId,
         signal: Inter::ProtocolSignal,
     ) -> Result<(), Error> {
-        self.ipc_send_relay.send_to_agent(agent_id, signal)
+        if let Some(relay) = self.ipc_send_relay.as_mut() {
+            relay.send_to_agent(agent_id, signal)
+        } else {
+            Err(Error::ChannelNotFound(ChannelId::Agent(agent_id)))
+        }
     }
 
     // Relay signal onto inter-process connector
@@ -103,9 +107,13 @@ impl<Inter: IsChannel, Intra: IsChannel> SchedulerConnector<Inter, Intra> {
 
     pub fn run_and_connect(&mut self) -> Result<(), Error> {
         debug!("Starting MixedSchedulerConnector");
-        let receive_relay_handle = self.ipc_receive_relay.run_and_connect();
-        self.relay_threads.push(receive_relay_handle);
-        self.ipc_send_relay.connect()?;
+        if let Some(relay) = self.ipc_receive_relay.as_mut() {
+            let receive_relay_handle = relay.run_and_connect();
+            self.relay_threads.push(receive_relay_handle);
+        }
+        if let Some(relay) = self.ipc_send_relay.as_mut() {
+            relay.connect()?;
+        }
         self.intra_receiver.connect_senders(self.timeout)?;
         self.worker_sender.connect_receivers(self.timeout)
     }
@@ -115,7 +123,11 @@ impl<Inter: IsChannel, Intra: IsChannel> SchedulerConnector<Inter, Intra> {
     }
 
     fn sync_time(&mut self) -> Result<(), Error> {
-        self.ipc_send_relay.sync_time()
+        if let Some(relay) = self.ipc_send_relay.as_mut() {
+            relay.sync_time()
+        } else {
+            Ok(())
+        }
     }
 }
 
@@ -131,7 +143,9 @@ impl<Inter: IsChannel, Intra: IsChannel> ConnectScheduler for SchedulerConnector
     fn get_connected_agent_ids(&self) -> Vec<AgentId> {
         let mut agent_ids: BTreeSet<_> = self.worker_agent_map.values().copied().collect();
         // In relayed mode, recorders are also agents we talk to.
-        agent_ids.extend(self.ipc_send_relay.get_remote_agents());
+        if let Some(relay) = self.ipc_send_relay.as_ref() {
+            agent_ids.extend(relay.get_remote_agents());
+        }
         agent_ids.into_iter().collect()
     }
 
@@ -157,7 +171,9 @@ impl<Inter: IsChannel, Intra: IsChannel> ConnectScheduler for SchedulerConnector
 
     fn broadcast_terminate(&mut self, signal: &Signal) -> Result<(), Error> {
         // Broadcast to remote agents via the IPC relay.
-        self.ipc_send_relay.broadcast((*signal).into())?;
+        if let Some(relay) = self.ipc_send_relay.as_mut() {
+            relay.broadcast((*signal).into())?;
+        }
 
         // Also broadcast to local workers via the MPSC sender.
         self.worker_sender.broadcast((*signal).into())

feo/src/signalling/relayed/sockets_mpsc.rs

@@ -217,22 +217,27 @@ impl<Inter: SocketChannel> scheduler::SchedulerConnector<Inter, IntraChannel> {
             intra_builders::multi_sender_builder(&local_worker_channels);
         let worker_sender = worker_sender_builder();
 
-        // Create an IPC relay for receiving from remote processes
         let channel_ids: Vec<ChannelId> = remote_agents
             .iter()
             .copied()
             .map(ChannelId::Agent)
             .collect();
 
-        let relay_sender_builder = local_sender_builders.remove(&relay_channel).unwrap();
-        let relay_receiver_builder =
-            Inter::multi_receiver_builder(channel_ids.clone(), bind_address_senders);
-        let ipc_receive_relay =
-            PrimaryReceiveRelay::new(relay_sender_builder, relay_receiver_builder, timeout);
-
-        let ipc_sender = Inter::new_multi_sender(&channel_ids, bind_address_receivers);
-
-        let ipc_send_relay = PrimarySendRelay::new(remote_agents, ipc_sender, timeout);
+        // Create IPC relays only if there are remote agents to communicate with.
+        let (ipc_receive_relay, ipc_send_relay) = if !remote_agents.is_empty() {
+            let relay_sender_builder = local_sender_builders.remove(&relay_channel).unwrap();
+            let relay_receiver_builder =
+                Inter::multi_receiver_builder(channel_ids.clone(), bind_address_senders);
+            let receive_relay =
+                PrimaryReceiveRelay::new(relay_sender_builder, relay_receiver_builder, timeout);
+
+            let ipc_sender = Inter::new_multi_sender(&channel_ids, bind_address_receivers);
+            let send_relay = PrimarySendRelay::new(remote_agents, ipc_sender, timeout);
+
+            (Some(receive_relay), Some(send_relay))
+        } else {
+            (None, None)
+        };
 
         // Create worker connector builders for local workers
         let worker_connector_builders: HashMap<WorkerId, Builder<WorkerConnector>> =