diff --git a/control-core/src/machines/manager.rs b/control-core/src/machines/manager.rs index b39d5764..254cf23c 100644 --- a/control-core/src/machines/manager.rs +++ b/control-core/src/machines/manager.rs @@ -1,4 +1,8 @@ -use smol::lock::{Mutex, RwLock}; +use smol::{ + channel::Sender, + lock::{Mutex, RwLock}, +}; +use socketioxide::extract::SocketRef; use crate::{ machines::{ @@ -6,6 +10,7 @@ use crate::{ new::{MachineNewHardware, MachineNewHardwareEthercat, MachineNewParams}, }, serial::SerialDevice, + socketio::event::GenericEvent, }; use std::{collections::HashMap, sync::Arc}; @@ -36,6 +41,7 @@ impl MachineManager { device_identifications: &Vec, machine_registry: &MachineRegistry, hardware: &MachineNewHardwareEthercat, + socket_queue_tx: Sender<(SocketRef, Arc)>, ) { // empty ethercat machines self.ethercat_machines.clear(); @@ -62,6 +68,7 @@ impl MachineManager { let new_machine = machine_registry.new_machine(&MachineNewParams { device_group, hardware: &machine_new_hardware, + socket_queue_tx: socket_queue_tx.clone(), }); // insert the machine into the ethercat machines map @@ -86,6 +93,7 @@ impl MachineManager { device_identification: &DeviceIdentification, device: Arc>, machine_registry: &MachineRegistry, + socket_queue_tx: Sender<(SocketRef, Arc)>, ) { let hardware = MachineNewHardwareSerial { device }; @@ -98,6 +106,7 @@ impl MachineManager { let new_machine = machine_registry.new_machine(&MachineNewParams { device_group: &vec![device_identification_identified.clone()], hardware: &MachineNewHardware::Serial(&hardware), + socket_queue_tx, }); tracing::info!("Adding serial machine {:?}", new_machine); diff --git a/control-core/src/machines/new.rs b/control-core/src/machines/new.rs index f972387a..a4443cb7 100644 --- a/control-core/src/machines/new.rs +++ b/control-core/src/machines/new.rs @@ -6,9 +6,12 @@ use ethercat_hal::{ devices::EthercatDevice, helpers::ethercrab_types::EthercrabSubDevicePreoperational, }; use ethercrab::{SubDevice, SubDeviceRef}; -use smol::lock::RwLock; +use smol::{channel::Sender, lock::RwLock}; +use socketioxide::extract::SocketRef; use std::sync::Arc; +use crate::socketio::event::GenericEvent; + pub trait MachineNewTrait { fn new<'maindevice, 'subdevices>( params: &MachineNewParams<'maindevice, 'subdevices, '_, '_, '_, '_, '_>, @@ -39,6 +42,7 @@ pub struct MachineNewParams< 'machine_new_hardware_etehrcat, 'machine_new_hardware_serial, >, + pub socket_queue_tx: Sender<(SocketRef, Arc)>, } pub enum MachineNewHardware< diff --git a/control-core/src/socketio/mod.rs b/control-core/src/socketio/mod.rs index b440d7d9..d8cdda2e 100644 --- a/control-core/src/socketio/mod.rs +++ b/control-core/src/socketio/mod.rs @@ -1,4 +1,3 @@ pub mod event; pub mod namespace; pub mod namespace_id; -pub mod socket_queue; diff --git a/control-core/src/socketio/namespace.rs b/control-core/src/socketio/namespace.rs index 5560727c..32edeb39 100644 --- a/control-core/src/socketio/namespace.rs +++ b/control-core/src/socketio/namespace.rs @@ -1,23 +1,22 @@ use crate::socketio::event::GenericEvent; -use socketioxide::{extract::SocketRef, socket::Sid}; +use smol::channel::Sender; +use socketioxide::extract::SocketRef; use std::{collections::HashMap, sync::Arc, time::Duration}; use tracing::instrument; -use super::socket_queue::SocketQueue; - #[derive(Debug)] pub struct Namespace { pub sockets: Vec, pub events: HashMap>>, - pub socket_queues: HashMap, + pub socket_queue_tx: Sender<(SocketRef, Arc)>, } impl Namespace { - pub fn new() -> Self { + pub fn new(socket_queue_tx: Sender<(SocketRef, Arc)>) -> Self { Self { sockets: vec![], events: HashMap::new(), - socket_queues: HashMap::new(), + socket_queue_tx, } } } @@ -32,8 +31,6 @@ impl Namespace { pub fn subscribe(&mut self, socket: SocketRef) { // add the socket to the list self.sockets.push(socket.clone()); - // create a new queue for this socket - self.socket_queues.insert(socket.id, SocketQueue::new()); } /// Removes a socket from the namespace. @@ -45,8 +42,6 @@ impl Namespace { pub fn unsubscribe(&mut self, socket: SocketRef) { // remove the socket from the list self.sockets.retain(|s| s.id != socket.id); - // remove the socket's queue - self.socket_queues.remove(&socket.id); } /// Re-emits cached events to a specific socket. @@ -59,37 +54,18 @@ impl Namespace { /// * `socket` - A reference to the socket that will receive the cached events #[instrument(skip_all)] pub fn reemit(&mut self, socket: SocketRef) { - if let Some(queue) = self.socket_queues.get(&socket.id) { - // Collect events grouped by name/kind with their counts for sorting - let mut event_groups: Vec<(&String, &Vec>)> = - self.events.iter().collect(); + // Collect events grouped by name/kind with their counts for sorting + let mut event_groups: Vec<(&String, &Vec>)> = + self.events.iter().collect(); - // Sort by event count (ascending - lowest count first) - event_groups.sort_by(|a, b| a.1.len().cmp(&b.1.len())); + // Sort by event count (ascending - lowest count first) + event_groups.sort_by(|a, b| a.1.len().cmp(&b.1.len())); - // Emit events in order of lowest count first - for (_event_name, events) in event_groups { - for event in events { - queue.push(event.clone()); - } - } - - queue.flush(socket.clone()); - } - } - - /// Emits an event to all sockets in the namespace. - /// - /// # Arguments - /// - /// * `event` - The event to be emitted - #[instrument(skip_all)] - pub fn emit(&mut self, event: Arc) { - // Use the new emit function which combines push and flush - for socket in self.sockets.clone() { - if let Some(queue) = self.socket_queues.get(&socket.id) { - queue.push(event.clone()); - queue.flush(socket.clone()); + // Emit events in order of lowest count first + for (_event_name, events) in event_groups { + for event in events { + // Send to global queue instead of per-socket queue + self.send_to_queue(&socket, &event, "reemit"); } } } @@ -102,7 +78,7 @@ impl Namespace { /// * `event` - The event to be cached /// * `buffer_fn` - A function that defines how the event should be added to the cache buffer #[instrument(skip_all)] - pub fn cache( + fn cache( &mut self, event: Arc, buffer_fn: &Box>, &Arc) -> ()>, @@ -116,16 +92,13 @@ impl Namespace { /// Emits an event to all sockets in the namespace and caches it. /// - /// This is a convenience method that combines the functionality of - /// [`Namespace::emit`] and [`Namespace::cache`]. /// /// # Arguments /// /// * `event` - The event to be emitted and cached - /// * `cache_key` - A string key to identify the cache /// * `buffer_fn` - A function that defines how the event should be added to the cache buffer #[instrument(skip_all)] - pub fn emit_cached( + pub fn emit( &mut self, event: Arc, buffer_fn: &Box>, &Arc) -> ()>, @@ -133,8 +106,50 @@ impl Namespace { // cache the event self.cache(event.clone(), &buffer_fn); - // emit the event - self.emit(event); + // emit the event - inlined from emit function + // Send to global queue for each socket in the namespace + for socket in self.sockets.clone() { + self.send_to_queue(&socket, &event, "emit"); + } + } + + /// Sends an event to the global queue for a specific socket. + /// + /// # Arguments + /// + /// * `socket` - The socket to send the event to + /// * `event` - The event to be sent + /// * `operation` - A description of the operation for logging purposes (e.g., "reemit", "emit") + #[instrument(skip_all)] + fn send_to_queue(&self, socket: &SocketRef, event: &Arc, operation: &str) { + tracing::trace!( + socket_id = ?socket.id, + event = %event.name, + operation = %operation, + "Sending event to global queue" + ); + match self + .socket_queue_tx + .try_send((socket.clone(), event.clone())) + { + Ok(_) => { + tracing::trace!( + socket_id = ?socket.id, + event = %event.name, + operation = %operation, + "Successfully sent event to global queue" + ); + } + Err(e) => { + tracing::error!( + socket_id = ?socket.id, + event = %event.name, + operation = %operation, + error = %e, + "Failed to send event to global queue" + ); + } + } } } @@ -146,7 +161,7 @@ pub trait NamespaceCacheingLogic where V: CacheableEvents, { - fn emit_cached(&mut self, event: V); + fn emit(&mut self, event: V); } pub trait CacheableEvents { @@ -218,8 +233,9 @@ mod tests { // Create a cache function that stores the last 2 events let cache_fn = cache_n_events(2); - // Create namespace - let mut namespace = Namespace::new(); + // Create namespace with dummy queue_tx + let (queue_tx, _queue_rx) = smol::channel::unbounded(); + let mut namespace = Namespace::new(queue_tx); assert!(namespace.events.is_empty()); // Add event @@ -276,9 +292,10 @@ mod tests { let duration = Duration::new(10, 0); let bucket_size = Duration::new(1, 0); let cache_fn = cache_duration(duration, bucket_size); + let (queue_tx, _queue_rx) = smol::channel::unbounded(); // Create namespace - let mut namespace = Namespace::new(); + let mut namespace = Namespace::new(queue_tx); assert!(namespace.events.is_empty()); // Add events every 100ms for 20 seconds diff --git a/control-core/src/socketio/socket_queue.rs b/control-core/src/socketio/socket_queue.rs deleted file mode 100644 index bc19f039..00000000 --- a/control-core/src/socketio/socket_queue.rs +++ /dev/null @@ -1,123 +0,0 @@ -use std::{ - collections::VecDeque, - sync::{ - Arc, Mutex, - atomic::{AtomicBool, Ordering}, - }, - time::Instant, -}; - -use socketioxide::extract::SocketRef; -use tracing::{debug_span, instrument}; - -use super::event::GenericEvent; - -/// A queue for managing events for a specific socket -#[derive(Debug)] -pub struct SocketQueue { - pub queue: Arc>>>, - pub is_flushing: Arc, -} - -impl SocketQueue { - /// Create a new socket queue for the given socket ID - pub fn new() -> Self { - Self { - queue: Arc::new(Mutex::new(VecDeque::new())), - is_flushing: Arc::new(AtomicBool::new(false)), - } - } - - /// Add an event to the queue - #[instrument(skip_all)] - pub fn push(&self, event: Arc) { - if let Ok(mut queue) = self.queue.lock() { - queue.push_back(event); - } - } - - /// Send a single event with retry logic - async fn send_event(socket: &SocketRef, event: Arc) { - // retry loop for each event - loop { - // check if socket is still connected - if !socket.connected() { - break; // Exit the loop if the socket is not connected - } - - match socket.emit("event", event.as_ref()) { - Ok(_) => break, // Successfully emitted, exit loop - Err(e) => match e { - socketioxide::SendError::Serialize(_) => { - // no reason in retrying serialization errors - break; - } - socketioxide::SendError::Socket(socket_error) => match socket_error { - socketioxide::SocketError::InternalChannelFull => { - continue; // Retry sending the event - } - socketioxide::SocketError::Closed => { - // Socket is closed, no point in retrying - break; - } - }, - }, - } - } - } - - /// Force flush events asynchronously (assumes flushing flag is already set) - #[instrument(skip_all)] - pub fn flush(&self, socket: SocketRef) { - // Check if we're already flushing - if self - .is_flushing - .compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst) - .is_err() - { - // Already flushing, return early - return; - } - - let queue = Arc::clone(&self.queue); - let is_flushing = Arc::clone(&self.is_flushing); - - // Spawn a smol task to flush the queue - smol::spawn(async move { - let span = debug_span!("socket_queue_flush", socket_id = ?socket.id); - let _enter = span.enter(); - - // Process events directly from the queue without collecting first - let mut event_cnt = 0; - let start_t = Instant::now(); - loop { - let event = { - if let Ok(mut queue_guard) = queue.lock() { - queue_guard.pop_front() - } else { - break; // Exit if we can't lock the queue - } - }; - - let Some(event) = event else { - break; // No more events in queue - }; - - Self::send_event(&socket, event).await; - event_cnt += 1; - } - let elapsed = start_t.elapsed(); - - log::info!( - "[SocketQueue::flush] Flushed {} events for socket {} in {:?}us", - event_cnt, - socket.id, - elapsed.as_micros() - ); - - // Reset the flushing flag - is_flushing.store(false, Ordering::SeqCst); - }) - .detach(); - } -} diff --git a/server/src/app_state.rs b/server/src/app_state.rs index 9df8900b..ed20b618 100644 --- a/server/src/app_state.rs +++ b/server/src/app_state.rs @@ -5,16 +5,21 @@ use control_core::machines::Machine; use control_core::machines::identification::{DeviceIdentification, MachineIdentificationUnique}; use control_core::machines::manager::MachineManager; use control_core::serial::serial_detection::SerialDetection; +use control_core::socketio::event::GenericEvent; use ethercat_hal::devices::EthercatDevice; use ethercrab::{MainDevice, SubDeviceGroup, subdevice_group::Op}; +use smol::channel::{Receiver, Sender}; use smol::lock::RwLock; use socketioxide::SocketIo; +use socketioxide::extract::SocketRef; use std::collections::HashMap; use std::sync::Arc; pub struct SocketioSetup { pub socketio: RwLock>, pub namespaces: RwLock, + pub socket_queue_tx: Sender<(SocketRef, Arc)>, + pub socket_queue_rx: Receiver<(SocketRef, Arc)>, } pub struct SerialSetup { @@ -61,10 +66,13 @@ impl EthercatSetup { impl AppState { pub fn new() -> Self { + let (socket_queue_tx, socket_queue_rx) = smol::channel::unbounded(); Self { socketio_setup: SocketioSetup { socketio: RwLock::new(None), - namespaces: RwLock::new(Namespaces::new()), + namespaces: RwLock::new(Namespaces::new(socket_queue_tx.clone())), + socket_queue_tx, + socket_queue_rx, }, ethercat_setup: Arc::new(RwLock::new(None)), serial_setup: Arc::new(RwLock::new(SerialSetup { diff --git a/server/src/ethercat/init.rs b/server/src/ethercat/init.rs index 65b287d8..d246758c 100644 --- a/server/src/ethercat/init.rs +++ b/server/src/ethercat/init.rs @@ -27,7 +27,7 @@ pub fn init_ethercat( .await .main_namespace; let event = EthercatInterfaceDiscoveryEvent::Discovering(true).build(); - main_namespace.emit_cached(MainNamespaceEvents::EthercatInterfaceDiscoveryEvent(event)); + main_namespace.emit(MainNamespaceEvents::EthercatInterfaceDiscoveryEvent(event)); }); // tries to find a suitable interface in a loop @@ -58,7 +58,7 @@ pub fn init_ethercat( .await .main_namespace; let event = EthercatInterfaceDiscoveryEvent::Done(interface_clone).build(); - main_namespace.emit_cached(MainNamespaceEvents::EthercatInterfaceDiscoveryEvent(event)); + main_namespace.emit(MainNamespaceEvents::EthercatInterfaceDiscoveryEvent(event)); }); smol::block_on(setup_loop(thread_panic_tx, &interface, app_state.clone()))?; diff --git a/server/src/ethercat/setup.rs b/server/src/ethercat/setup.rs index 108715fe..925262e5 100644 --- a/server/src/ethercat/setup.rs +++ b/server/src/ethercat/setup.rs @@ -87,7 +87,7 @@ pub async fn setup_loop( .await .main_namespace; let event = EthercatDevicesEventBuilder().initializing(); - main_namespace.emit_cached(MainNamespaceEvents::EthercatDevicesEvent(event)); + main_namespace.emit(MainNamespaceEvents::EthercatDevicesEvent(event)); } }); @@ -180,6 +180,7 @@ pub async fn setup_loop( ethercat_devices: &identified_devices, subdevices: &identified_subdevices, }, + app_state.socketio_setup.socket_queue_tx.clone(), ); } @@ -199,7 +200,7 @@ pub async fn setup_loop( .await .main_namespace; let event = MachinesEventBuilder().build(app_state_clone.clone()).await; - main_namespace.emit_cached(MainNamespaceEvents::MachinesEvent(event)); + main_namespace.emit(MainNamespaceEvents::MachinesEvent(event)); }); // Put group in operational state @@ -237,7 +238,7 @@ pub async fn setup_loop( let event = EthercatDevicesEventBuilder() .build(app_state_clone.clone()) .await; - main_namespace.emit_cached(MainNamespaceEvents::EthercatDevicesEvent(event)); + main_namespace.emit(MainNamespaceEvents::EthercatDevicesEvent(event)); }); Ok(()) diff --git a/server/src/machines/extruder1/api.rs b/server/src/machines/extruder1/api.rs index e41ad484..e38713aa 100644 --- a/server/src/machines/extruder1/api.rs +++ b/server/src/machines/extruder1/api.rs @@ -13,6 +13,8 @@ use control_core::{ }; use serde::{Deserialize, Serialize}; use serde_json::Value; +use smol::channel::Sender; +use socketioxide::extract::SocketRef; use tracing::instrument; #[derive(Serialize, Debug, Clone)] @@ -210,11 +212,13 @@ enum Mutation { } #[derive(Debug)] -pub struct ExtruderV2Namespace(Namespace); +pub struct ExtruderV2Namespace { + pub namespace: Namespace, +} impl NamespaceCacheingLogic for ExtruderV2Namespace { #[instrument(skip_all)] - fn emit_cached(&mut self, events: ExtruderV2Events) { + fn emit(&mut self, events: ExtruderV2Events) { let event = match events.event_value() { Ok(event) => event, Err(err) => { @@ -224,13 +228,15 @@ impl NamespaceCacheingLogic for ExtruderV2Namespace { }; let event = Arc::new(event); let buffer_fn = events.event_cache_fn(); - self.0.emit_cached(event, &buffer_fn); + self.namespace.emit(event, &buffer_fn); } } impl ExtruderV2Namespace { - pub fn new() -> Self { - Self(Namespace::new()) + pub fn new(socket_queue_tx: Sender<(SocketRef, Arc)>) -> Self { + Self { + namespace: Namespace::new(socket_queue_tx), + } } } @@ -300,6 +306,6 @@ impl MachineApi for ExtruderV2 { } fn api_event_namespace(&mut self) -> &mut Namespace { - &mut self.namespace.0 + &mut self.namespace.namespace } } diff --git a/server/src/machines/extruder1/mod.rs b/server/src/machines/extruder1/mod.rs index 7dea0c84..671c55dd 100644 --- a/server/src/machines/extruder1/mod.rs +++ b/server/src/machines/extruder1/mod.rs @@ -105,7 +105,7 @@ impl ExtruderV2 { .build(); self.namespace - .emit_cached(ExtruderV2Events::ExtruderSettingsStateEvent(event)); + .emit(ExtruderV2Events::ExtruderSettingsStateEvent(event)); } } @@ -193,7 +193,7 @@ impl ExtruderV2 { } .build(); self.namespace - .emit_cached(ExtruderV2Events::RotationStateEvent(event)) + .emit(ExtruderV2Events::RotationStateEvent(event)) } fn set_mode_state(&mut self, mode: ExtruderV2Mode) { @@ -207,8 +207,7 @@ impl ExtruderV2 { mode: self.mode.clone(), } .build(); - self.namespace - .emit_cached(ExtruderV2Events::ModeEvent(event)); + self.namespace.emit(ExtruderV2Events::ModeEvent(event)); } } @@ -225,7 +224,7 @@ impl ExtruderV2 { } .build(); self.namespace - .emit_cached(ExtruderV2Events::RegulationStateEvent(event)); + .emit(ExtruderV2Events::RegulationStateEvent(event)); } fn set_target_pressure(&mut self, pressure: f64) { @@ -251,7 +250,7 @@ impl ExtruderV2 { .build(heating_type); self.namespace - .emit_cached(ExtruderV2Events::HeatingStateEvent(event)); + .emit(ExtruderV2Events::HeatingStateEvent(event)); } fn emit_heating_element_power(&mut self, heating_type: HeatingType) { @@ -272,7 +271,7 @@ impl ExtruderV2 { let event = api::HeatingPowerEvent { wattage }.build(heating_type); self.namespace - .emit_cached(ExtruderV2Events::HeatingPowerEvent(event)); + .emit(ExtruderV2Events::HeatingPowerEvent(event)); } fn set_target_temperature(&mut self, target_temperature: f64, heating_type: HeatingType) { @@ -329,7 +328,7 @@ impl ExtruderV2 { } .build(); self.namespace - .emit_cached(ExtruderV2Events::ScrewStateEvent(event)); + .emit(ExtruderV2Events::ScrewStateEvent(event)); } fn emit_bar(&mut self) { @@ -341,6 +340,6 @@ impl ExtruderV2 { } .build(); self.namespace - .emit_cached(ExtruderV2Events::PressureStateEvent(event)); + .emit(ExtruderV2Events::PressureStateEvent(event)); } } diff --git a/server/src/machines/extruder1/new.rs b/server/src/machines/extruder1/new.rs index f5f4bc5b..d96d1445 100644 --- a/server/src/machines/extruder1/new.rs +++ b/server/src/machines/extruder1/new.rs @@ -396,7 +396,7 @@ impl MachineNewTrait for ExtruderV2 { ScrewSpeedController::new(inverter, target_pressure, target_rpm, pressure_sensor); let extruder: ExtruderV2 = Self { - namespace: ExtruderV2Namespace::new(), + namespace: ExtruderV2Namespace::new(params.socket_queue_tx.clone()), last_measurement_emit: Instant::now(), mode: ExtruderV2Mode::Standby, temperature_controller_front: temperature_controller_front, diff --git a/server/src/machines/laser/api.rs b/server/src/machines/laser/api.rs index 5cbaa0f0..b31e0c0b 100644 --- a/server/src/machines/laser/api.rs +++ b/server/src/machines/laser/api.rs @@ -11,6 +11,8 @@ use control_core::{ }; use serde::{Deserialize, Serialize}; use serde_json::Value; +use smol::channel::Sender; +use socketioxide::extract::SocketRef; use std::{sync::Arc, time::Duration}; use tracing::instrument; @@ -43,11 +45,15 @@ pub enum LaserEvents { } #[derive(Debug)] -pub struct LaserMachineNamespace(Namespace); +pub struct LaserMachineNamespace { + pub namespace: Namespace, +} impl LaserMachineNamespace { - pub fn new() -> Self { - Self(Namespace::new()) + pub fn new(socket_queue_tx: Sender<(SocketRef, Arc)>) -> Self { + Self { + namespace: Namespace::new(socket_queue_tx), + } } } @@ -82,7 +88,7 @@ enum Mutation { impl NamespaceCacheingLogic for LaserMachineNamespace { #[instrument(skip_all)] - fn emit_cached(&mut self, events: LaserEvents) { + fn emit(&mut self, events: LaserEvents) { let event = match events.event_value() { Ok(event) => event, Err(err) => { @@ -92,7 +98,7 @@ impl NamespaceCacheingLogic for LaserMachineNamespace { }; let event = Arc::new(event); let buffer_fn = events.event_cache_fn(); - self.0.emit_cached(event, &buffer_fn); + self.namespace.emit(event, &buffer_fn); } } @@ -114,6 +120,6 @@ impl MachineApi for LaserMachine { } fn api_event_namespace(&mut self) -> &mut Namespace { - &mut self.namespace.0 + &mut self.namespace.namespace } } diff --git a/server/src/machines/laser/mod.rs b/server/src/machines/laser/mod.rs index 08d74036..1d9886b8 100644 --- a/server/src/machines/laser/mod.rs +++ b/server/src/machines/laser/mod.rs @@ -39,7 +39,7 @@ impl LaserMachine { diameter: diameter.unwrap_or(0.0), }; self.namespace - .emit_cached(LaserEvents::Diameter(diameter_event.build())); + .emit(LaserEvents::Diameter(diameter_event.build())); } pub fn emit_laser_state(&mut self) { @@ -50,7 +50,7 @@ impl LaserMachine { }; self.namespace - .emit_cached(LaserEvents::LaserState(laser_state_event.build())); + .emit(LaserEvents::LaserState(laser_state_event.build())); } pub fn target_set_higher_tolerance(&mut self, higher_tolerance: f64) { diff --git a/server/src/machines/laser/new.rs b/server/src/machines/laser/new.rs index da95f6a5..c3b03f54 100644 --- a/server/src/machines/laser/new.rs +++ b/server/src/machines/laser/new.rs @@ -43,7 +43,7 @@ impl MachineNewTrait for LaserMachine { }; Ok(Self { laser, - namespace: LaserMachineNamespace::new(), + namespace: LaserMachineNamespace::new(params.socket_queue_tx.clone()), last_measurement_emit: Instant::now(), laser_target, }) diff --git a/server/src/machines/mock/api.rs b/server/src/machines/mock/api.rs index 3eebfe52..0f4a0d06 100644 --- a/server/src/machines/mock/api.rs +++ b/server/src/machines/mock/api.rs @@ -1,4 +1,3 @@ -use std::sync::Arc; use super::MockMachine; use control_core::{ machines::api::MachineApi, @@ -12,7 +11,9 @@ use control_core::{ }; use serde::{Deserialize, Serialize}; use serde_json::Value; -use std::sync::Arc; +use smol::channel::Sender; +use socketioxide::extract::SocketRef; +use std::{sync::Arc, time::Duration}; use tracing::instrument; #[derive(Serialize, Deserialize, Debug, Clone, PartialEq)] @@ -61,11 +62,15 @@ pub enum MockEvents { } #[derive(Debug)] -pub struct MockMachineNamespace(Namespace); +pub struct MockMachineNamespace { + pub namespace: Namespace, +} impl MockMachineNamespace { - pub fn new() -> Self { - Self(Namespace::new()) + pub fn new(socket_queue_tx: Sender<(SocketRef, Arc)>) -> Self { + Self { + namespace: Namespace::new(socket_queue_tx), + } } } @@ -100,7 +105,7 @@ enum Mutation { impl NamespaceCacheingLogic for MockMachineNamespace { #[instrument(skip_all)] - fn emit_cached(&mut self, events: MockEvents) { + fn emit(&mut self, events: MockEvents) { let event = match events.event_value() { Ok(event) => event, Err(err) => { @@ -110,7 +115,7 @@ impl NamespaceCacheingLogic for MockMachineNamespace { }; let event = Arc::new(event); let buffer_fn = events.event_cache_fn(); - self.0.emit_cached(event, &buffer_fn); + self.namespace.emit(event, &buffer_fn); } } @@ -129,6 +134,6 @@ impl MachineApi for MockMachine { } fn api_event_namespace(&mut self) -> &mut Namespace { - &mut self.namespace.0 + &mut self.namespace.namespace } } diff --git a/server/src/machines/mock/mod.rs b/server/src/machines/mock/mod.rs index 39f0ac8e..380658d0 100644 --- a/server/src/machines/mock/mod.rs +++ b/server/src/machines/mock/mod.rs @@ -46,7 +46,7 @@ impl MockMachine { let sine_wave_event = SineWaveEvent { amplitude: y }; self.namespace - .emit_cached(MockEvents::SineWave(sine_wave_event.build())); + .emit(MockEvents::SineWave(sine_wave_event.build())); } /// Emit the current state of the mock machine only if values have changed @@ -62,7 +62,7 @@ impl MockMachine { }; self.namespace - .emit_cached(MockEvents::SineWaveState(mock_state_event.build())); + .emit(MockEvents::SineWaveState(mock_state_event.build())); // Update last emitted values self.last_emitted_frequency = Some(current_frequency_mhz); @@ -82,7 +82,7 @@ impl MockMachine { }; self.namespace - .emit_cached(MockEvents::ModeState(mode_state_event.build())); + .emit(MockEvents::ModeState(mode_state_event.build())); // Update last emitted values self.last_emitted_mode = Some(current_mode); diff --git a/server/src/machines/mock/new.rs b/server/src/machines/mock/new.rs index 0ff5b3eb..23ae154e 100644 --- a/server/src/machines/mock/new.rs +++ b/server/src/machines/mock/new.rs @@ -38,8 +38,9 @@ impl MachineNewTrait for MockMachine { } let now = Instant::now(); + Ok(Self { - namespace: MockMachineNamespace::new(), + namespace: MockMachineNamespace::new(params.socket_queue_tx.clone()), last_measurement_emit: now, t_0: now, // Initialize start time to current time frequency: Frequency::new::(0.5), // Default frequency of 500 mHz diff --git a/server/src/machines/winder2/api.rs b/server/src/machines/winder2/api.rs index aa7aa5b3..f9861e07 100644 --- a/server/src/machines/winder2/api.rs +++ b/server/src/machines/winder2/api.rs @@ -13,6 +13,8 @@ use control_core::{ }; use serde::{Deserialize, Serialize}; use serde_json::Value; +use smol::channel::Sender; +use socketioxide::extract::SocketRef; use tracing::instrument; #[derive(Serialize, Deserialize, Debug, Clone)] @@ -264,11 +266,13 @@ pub enum Winder2Events { } #[derive(Debug)] -pub struct Winder2Namespace(Namespace); +pub struct Winder2Namespace { + pub namespace: Namespace, +} impl NamespaceCacheingLogic for Winder2Namespace { #[instrument(skip_all)] - fn emit_cached(&mut self, events: Winder2Events) { + fn emit(&mut self, events: Winder2Events) { let event = match events.event_value() { Ok(event) => event, Err(err) => { @@ -278,13 +282,15 @@ impl NamespaceCacheingLogic for Winder2Namespace { }; let event = Arc::new(event); let buffer_fn = events.event_cache_fn(); - self.0.emit_cached(event, &buffer_fn); + self.namespace.emit(event, &buffer_fn); } } impl Winder2Namespace { - pub fn new() -> Self { - Self(Namespace::new()) + pub fn new(socket_queue_tx: Sender<(SocketRef, Arc)>) -> Self { + Self { + namespace: Namespace::new(socket_queue_tx), + } } } @@ -351,6 +357,6 @@ impl MachineApi for Winder2 { } fn api_event_namespace(&mut self) -> &mut Namespace { - &mut self.namespace.0 + &mut self.namespace.namespace } } diff --git a/server/src/machines/winder2/mod.rs b/server/src/machines/winder2/mod.rs index fd969861..2c4a0e51 100644 --- a/server/src/machines/winder2/mod.rs +++ b/server/src/machines/winder2/mod.rs @@ -149,7 +149,7 @@ impl Winder2 { .map(|x| x.get::()); let event = TraversePositionEvent { position }.build(); self.namespace - .emit_cached(Winder2Events::TraversePosition(event)) + .emit(Winder2Events::TraversePosition(event)) } fn emit_traverse_state(&mut self) { @@ -184,7 +184,7 @@ impl Winder2 { } .build(); self.namespace - .emit_cached(Winder2Events::TraverseState(event)) + .emit(Winder2Events::TraverseState(event)) } pub fn sync_traverse_speed(&mut self) { @@ -436,7 +436,7 @@ impl Winder2 { can_wind: self.can_wind(), } .build(); - self.namespace.emit_cached(Winder2Events::Mode(event)) + self.namespace.emit(Winder2Events::Mode(event)) } } @@ -463,7 +463,7 @@ impl Winder2 { let event = TensionArmAngleEvent { degree: angle_deg }.build(); self.namespace - .emit_cached(Winder2Events::TensionArmAngleEvent(event)) + .emit(Winder2Events::TensionArmAngleEvent(event)) } fn emit_tension_arm_state(&mut self) { @@ -472,7 +472,7 @@ impl Winder2 { } .build(); self.namespace - .emit_cached(Winder2Events::TensionArmStateEvent(event)) + .emit(Winder2Events::TensionArmStateEvent(event)) } } @@ -497,7 +497,7 @@ impl Winder2 { .steps_to_angular_velocity(self.spool.get_speed() as f64) .get::(); let event = api::SpoolRpmEvent { rpm }.build(); - self.namespace.emit_cached(Winder2Events::SpoolRpm(event)) + self.namespace.emit(Winder2Events::SpoolRpm(event)) } } @@ -557,7 +557,7 @@ impl Winder2 { } .build(), ); - self.namespace.emit_cached(event); + self.namespace.emit(event); } pub fn emit_puller_state(&mut self) { @@ -576,7 +576,7 @@ impl Winder2 { } .build(), ); - self.namespace.emit_cached(event); + self.namespace.emit(event); } } diff --git a/server/src/machines/winder2/new.rs b/server/src/machines/winder2/new.rs index 03454e31..caddd9f9 100644 --- a/server/src/machines/winder2/new.rs +++ b/server/src/machines/winder2/new.rs @@ -359,7 +359,7 @@ impl MachineNewTrait for Winder2 { EL7031_0030AnalogInputPort::AI1, ))), laser: DigitalOutputSetter::new(DigitalOutput::new(el2002, EL2002Port::DO1)), - namespace: Winder2Namespace::new(), + namespace: Winder2Namespace::new(params.socket_queue_tx.clone()), mode: mode.clone(), spool_step_converter: AngularStepConverter::new(200), spool_speed_controller: SpoolSpeedController::new(), diff --git a/server/src/main.rs b/server/src/main.rs index 5da18a0f..440b1ee1 100644 --- a/server/src/main.rs +++ b/server/src/main.rs @@ -25,6 +25,7 @@ use serial::init::init_serial; use jemalloc_stats::init_jemalloc_stats; use crate::panic::init_panic; +use crate::socketio::queue::init_socketio_queue; pub mod app_state; pub mod ethercat; @@ -60,6 +61,7 @@ fn main() { let thread_panic_tx = thread_panic_tx.clone(); let app_state = app_state.clone(); move || { + init_socketio_queue(thread_panic_tx.clone(), app_state.clone()); init_api(thread_panic_tx.clone(), app_state.clone()) .expect("Failed to initialize API"); #[cfg(not(feature = "mock-machine"))] diff --git a/server/src/mock/init.rs b/server/src/mock/init.rs index a7f3888d..bfe35f6f 100644 --- a/server/src/mock/init.rs +++ b/server/src/mock/init.rs @@ -33,6 +33,7 @@ pub fn init_mock(app_state: Arc) -> Result<(), anyhow::Error> { &device_identification, mock_serial_device, &MACHINE_REGISTRY, + app_state.socketio_setup.socket_queue_tx.clone(), ); } @@ -45,7 +46,7 @@ pub fn init_mock(app_state: Arc) -> Result<(), anyhow::Error> { .await .main_namespace; let event = MachinesEventBuilder().build(app_state_event.clone()).await; - main_namespace.emit_cached(MainNamespaceEvents::MachinesEvent(event)); + main_namespace.emit(MainNamespaceEvents::MachinesEvent(event)); Ok(()) } Err(e) => { diff --git a/server/src/serial/init.rs b/server/src/serial/init.rs index 951a5c47..079a3e5b 100644 --- a/server/src/serial/init.rs +++ b/server/src/serial/init.rs @@ -50,6 +50,7 @@ pub fn init_serial( &device_identifiaction, device.clone(), &MACHINE_REGISTRY, + app_state_clone.socketio_setup.socket_queue_tx.clone(), ) } for device_identification in result.removed { @@ -68,8 +69,7 @@ pub fn init_serial( .main_namespace; let event = MachinesEventBuilder().build(app_state_event.clone()).await; - main_namespace - .emit_cached(MainNamespaceEvents::MachinesEvent(event)); + main_namespace.emit(MainNamespaceEvents::MachinesEvent(event)); }); } smol::Timer::after(Duration::from_millis(300)).await; diff --git a/server/src/socketio/main_namespace/mod.rs b/server/src/socketio/main_namespace/mod.rs index 5223062b..0233e8db 100644 --- a/server/src/socketio/main_namespace/mod.rs +++ b/server/src/socketio/main_namespace/mod.rs @@ -7,17 +7,23 @@ use control_core::socketio::{ use ethercat_devices_event::EthercatDevicesEvent; use ethercat_interface_discovery_event::EthercatInterfaceDiscoveryEvent; use machines_event::MachinesEvent; +use smol::channel::Sender; +use socketioxide::extract::SocketRef; use tracing::instrument; pub mod ethercat_devices_event; pub mod ethercat_interface_discovery_event; pub mod machines_event; -pub struct MainRoom(pub Namespace); +pub struct MainRoom { + pub namespace: Namespace, +} impl MainRoom { - pub fn new() -> Self { - Self(Namespace::new()) + pub fn new(socket_queue_tx: Sender<(SocketRef, Arc)>) -> Self { + Self { + namespace: Namespace::new(socket_queue_tx), + } } } @@ -26,7 +32,7 @@ where MainNamespaceEvents: CacheableEvents, { #[instrument(skip_all)] - fn emit_cached(&mut self, event: MainNamespaceEvents) { + fn emit(&mut self, event: MainNamespaceEvents) { let buffer_fn = event.event_cache_fn(); let generic_event = match event.event_value() { Ok(event) => event, @@ -36,7 +42,7 @@ where } }; let generic_event = Arc::new(generic_event); - self.0.emit_cached(generic_event, &buffer_fn); + self.namespace.emit(generic_event, &buffer_fn); } } diff --git a/server/src/socketio/mod.rs b/server/src/socketio/mod.rs index 29b642d0..95c08749 100644 --- a/server/src/socketio/mod.rs +++ b/server/src/socketio/mod.rs @@ -1,3 +1,4 @@ pub mod init; pub mod main_namespace; pub mod namespaces; +pub mod queue; diff --git a/server/src/socketio/namespaces.rs b/server/src/socketio/namespaces.rs index 215afc77..221a6eca 100644 --- a/server/src/socketio/namespaces.rs +++ b/server/src/socketio/namespaces.rs @@ -1,6 +1,10 @@ use std::sync::Arc; -use control_core::socketio::{namespace::Namespace, namespace_id::NamespaceId}; +use control_core::socketio::{ + event::GenericEvent, namespace::Namespace, namespace_id::NamespaceId, +}; +use smol::channel::Sender; +use socketioxide::extract::SocketRef; use crate::app_state; @@ -11,9 +15,9 @@ pub struct Namespaces { } impl Namespaces { - pub fn new() -> Self { + pub fn new(socket_queue_tx: Sender<(SocketRef, Arc)>) -> Self { Self { - main_namespace: MainRoom::new(), + main_namespace: MainRoom::new(socket_queue_tx), } } @@ -24,7 +28,7 @@ impl Namespaces { callback: impl FnOnce(Result<&mut Namespace, anyhow::Error>), ) { match namespace_id { - NamespaceId::Main => callback(Ok(&mut self.main_namespace.0)), + NamespaceId::Main => callback(Ok(&mut self.main_namespace.namespace)), NamespaceId::Machine(machine_identification_unique) => { // Lock machines and work directly with the reference to avoid cloning issues let machines_guard = app_state.machines.read().await; diff --git a/server/src/socketio/queue.rs b/server/src/socketio/queue.rs new file mode 100644 index 00000000..46d8e782 --- /dev/null +++ b/server/src/socketio/queue.rs @@ -0,0 +1,115 @@ +use crate::{ + app_state::AppState, + panic::{PanicDetails, send_panic}, +}; +use smol::channel::Sender; +use std::{sync::Arc, time::Instant}; +use tracing::{debug, error, info, instrument, trace}; + +/// Send a single event with retry logic +#[instrument(skip_all)] +async fn send_event_with_retry( + socket: &socketioxide::extract::SocketRef, + event: &Arc, +) { + // retry loop for each event + loop { + // check if socket is still connected + if !socket.connected() { + trace!( + socket_id = ?socket.id, + event = %event.name, + "Socket disconnected, skipping event" + ); + break; // Exit the loop if the socket is not connected + } + + match socket.emit("event", event.as_ref()) { + Ok(_) => { + trace!( + socket_id = ?socket.id, + event = %event.name, + timestamp = event.ts, + "Successfully emitted event" + ); + break; // Successfully emitted, exit loop + } + Err(e) => match e { + socketioxide::SendError::Serialize(serialize_error) => { + trace!( + socket_id = ?socket.id, + event = %event.name, + error = %serialize_error, + "Serialization error, skipping event" + ); + break; // no reason in retrying serialization errors + } + socketioxide::SendError::Socket(socket_error) => match socket_error { + socketioxide::SocketError::InternalChannelFull => { + continue; // Retry sending the event + } + socketioxide::SocketError::Closed => { + trace!( + socket_id = ?socket.id, + event = %event.name, + "Socket closed, skipping event" + ); + break; // Socket is closed, no point in retrying + } + }, + }, + } + } +} + +pub fn init_socketio_queue(thread_panic_tx: Sender, app_state: Arc) { + std::thread::Builder::new() + .name("socketio-queue".to_string()) + .spawn(move || { + send_panic(thread_panic_tx); + + let rt = tokio::runtime::Builder::new_current_thread() + .enable_io() + .enable_time() + .build() + .expect("Failed to create runtime"); + + rt.block_on(async { + info!("SocketIO global queue listener started"); + + let mut event_count = 0; + let mut batch_start = Instant::now(); + + loop { + match app_state.socketio_setup.socket_queue_rx.recv().await { + Ok((socket, event)) => { + event_count += 1; + + // Handle the received message with retry logic + send_event_with_retry(&socket, &event).await; + + // Log batch statistics every 5 seconds + if batch_start.elapsed().as_secs() >= 5 { + let elapsed = batch_start.elapsed(); + if event_count > 0 { + debug!( + events_per_second = + (event_count as f64 / elapsed.as_secs_f64()) as u64, + "SocketIO Stats" + ); + } + event_count = 0; + batch_start = Instant::now(); + } + } + Err(e) => { + error!(error = %e, "Error receiving from global socketio queue"); + info!("SocketIO global queue listener stopping"); + break; + } + } + } + }); + }) + .expect("Failed to spawn socketio queue thread"); +}