diff --git a/clover-hub/src/server/modman/busses/proxies/group/can_2/bus_manager.rs b/clover-hub/src/server/modman/busses/proxies/group/can_2/bus_manager.rs index 97febf9..cb661a8 100644 --- a/clover-hub/src/server/modman/busses/proxies/group/can_2/bus_manager.rs +++ b/clover-hub/src/server/modman/busses/proxies/group/can_2/bus_manager.rs @@ -1,88 +1,199 @@ -use std::sync::Arc; - -use can_iso_tp::{ - self, - IsoTpNode, +use std::{ + collections::HashMap, + sync::Arc, }; + +use anyhow::anyhow; use embedded_can::Id; use linux_socketcan_iso_tp::{ self, IsoTpKernelOptions, TokioSocketCanIsoTp, }; -use tokio::sync::{ - broadcast::{ - channel as broadcast_channel, - Receiver as BroadcastReceiver, - Sender as BroadcastSender, - }, - oneshot::{ - channel as oneshot_channel, - Receiver as OneShotReceiver, - Sender as OneShotSender, - }, -}; +use regex::Regex; +use tokio::sync::Mutex; use tokio_util::sync::CancellationToken; -use tracing::instrument; +use tracing::{ + debug, + error, + info, + instrument, +}; -use crate::server::modman::busses::{ - models::BusMessage, - proxies::group::can_2::CAN2Bus, +use crate::server::modman::{ + busses::proxies::group::can_2::{ + module_listener::can_bus_listener, + CAN2Bus, + }, + models::PortStatus, }; +#[instrument] +pub fn parse_id_str(id_str: &str) -> Result { + todo!() +} + +pub fn match_to_str(match_struct: regex::Match<'_>) -> &str { + as Into<&str>>::into(match_struct) +} + #[instrument(skip(ctx, cancellation_token))] pub async fn can_bus_manager( ctx: Arc, cancellation_token: CancellationToken, - iface_details: (String, u32), + iface_name: String, ) { - let (iface_name, iface_index) = iface_details; - - // Modules shouldn't be sending events over CAN before we introduce ourselves, but we have a buffer just in case. - let (to_bus_tx, to_bus) = broadcast_channel(16); - let (from_bus, from_bus_rx) = broadcast_channel(16); - - // I'd rather return the JoinHandle from the function, but due to using libc, we should avoid doing that. - let (status_channel, status_channel_rx) = oneshot_channel(); - - tokio::task::spawn(async move { - can_bus_listener( - to_bus, - from_bus, - cancellation_token.clone(), - status_channel, - // We recreate the iface_details tuple due to an error created by the instrument macro. - (iface_name.clone(), iface_index), + let can_2_port_status_mutex = ctx.store.port_statuses.can_2.clone(); + let listener_registry: Arc>> = + Arc::new(Mutex::new(HashMap::new())); + + while !cancellation_token.is_cancelled() { + let port_statuses = can_2_port_status_mutex.lock().await; + let mut port_statuses_snapshot = Vec::new(); + + // Matches against strings like: `"can0/0xFFF:0xFFF"`. + // This regex does not validate if the specified IDs are within the CAN range though. + // That's done later by `embedded_can`. + let port_specifier_re = Regex::new( + r"^(?(?:\\w|[0-9])+)\\/(?0[xX][0-9a-fA-F]{3}):(?0[xX][0-9a-fA-F]{3})$", ) - .await - }); + .unwrap(); + + // We need to make sure that we're not leaving that mutex locked for too long. + for port_status in port_statuses.iter() { + port_statuses_snapshot.push((port_status.0.clone(), port_status.1.clone())); + } + + drop(port_statuses); + + for port_status_tuple in port_statuses_snapshot { + let (port_path, port_status) = port_status_tuple; + + match port_specifier_re.captures(&port_path) { + Some(re_captures) => { + // Known good value since the regex is static, and the haystack matched. + let requested_iface = re_captures.name("iface").unwrap(); - match status_channel_rx.await { - Ok(_) => while !ctx.cancellation_token.is_cancelled() {}, - Err(err) => todo!(), + if match_to_str(requested_iface) == &iface_name { + match port_status { + PortStatus::Requested(module_id) => { + setup_listener( + ctx.clone(), + port_path.clone(), + module_id, + ( + match_to_str(re_captures.name("rx_id").unwrap()), + match_to_str(re_captures.name("tx_id").unwrap()), + ), + listener_registry.clone(), + ) + .await; + } + PortStatus::Unavailable(module_id) => { + setup_listener( + ctx.clone(), + port_path.clone(), + module_id, + ( + match_to_str(re_captures.name("tx_id").unwrap()), + match_to_str(re_captures.name("tx_id").unwrap()), + ), + listener_registry.clone(), + ) + .await; + } + PortStatus::Unrequested(module_id) => { + match listener_registry.lock().await.get(&module_id) { + Some(listener_token) => { + info!("Shutting down CAN 2 listener for Module: {module_id}..."); + listener_token.cancel(); + } + None => { + error!("Listener for Module: {module_id}, was asked to be unbound, but there's no listener for that module; this is a bug and should be reported!"); + } + } + } + _ => {} + } + } + } + None => {} + } + } + } + + for (module_id, listener_token) in listener_registry.lock().await.iter() { + debug!("Shutting down CAN 2 listener for module: {module_id}..."); + listener_token.cancel(); } } -#[instrument(skip(to_bus, from_bus, cancellation_token))] -pub async fn can_bus_listener( - to_bus: BroadcastReceiver, - from_bus: BroadcastSender, - cancellation_token: CancellationToken, - status_channel: OneShotSender>, - iface_details: (String, u32), +#[instrument(skip(ctx, listener_registry))] +pub async fn setup_listener( + ctx: Arc, + iface_name: String, + module_id: String, + id_tuple: (&str, &str), + listener_registry: Arc>>, ) { - let options = IsoTpKernelOptions::default(); - let (iface_name, _iface_index) = iface_details; - - match TokioSocketCanIsoTp::open( - &iface_name, - Id::Standard(embedded_can::StandardId::new(0x101).expect("0x101 is a valid standard CAN ID")), - Id::Standard(embedded_can::StandardId::new(0x201).expect("0x201 is a valid standard CAN ID")), - &options, - ) { - Ok(socket) => { - + let mut ret: Option = None; + + match parse_id_str(id_tuple.0) { + Ok(raw_rx_id) => match parse_id_str(id_tuple.1) { + Ok(raw_tx_id) => { + let options = IsoTpKernelOptions::default(); + + match embedded_can::StandardId::new(raw_rx_id).ok_or(anyhow!( + "We expect that an RX ID of {}, is valid. Check your config or there's a bug in manifest validation!", + raw_rx_id + )) { + Ok(rx_id) => { + match embedded_can::StandardId::new(raw_tx_id).ok_or(anyhow!( + "We expect that a TX ID of {}, is valid. Check your config or there's a bug in manifest validation!", + raw_tx_id + )) { + Ok(tx_id) => { + match TokioSocketCanIsoTp::open( + &iface_name, + Id::Standard(rx_id), + Id::Standard(tx_id), + &options, + ) { + Ok(socket) => { + let listener_token = CancellationToken::new(); + + listener_registry.lock().await.insert(module_id.clone(), listener_token.clone()); + + tokio::task::spawn(async move { + can_bus_listener(ctx.clone(), listener_token.clone(), module_id, socket).await + }); + }, + Err(err) => { + ret = Some(err.into()); + }, + } + }, + Err(err) => { + ret = Some(err.into()); + }, + } + }, + Err(err) => { + ret = Some(err.into()); + }, + } + } + Err(err) => { + ret = Some(err.into()); + } + }, + Err(err) => { + ret = Some(err.into()); } - Err(err) => todo!(), + } + + match ret { + Some(err) => todo!(), + None => {} } } diff --git a/clover-hub/src/server/modman/busses/proxies/group/can_2/lookout.rs b/clover-hub/src/server/modman/busses/proxies/group/can_2/interface_lookout.rs similarity index 78% rename from clover-hub/src/server/modman/busses/proxies/group/can_2/lookout.rs rename to clover-hub/src/server/modman/busses/proxies/group/can_2/interface_lookout.rs index 59d132f..47fa085 100644 --- a/clover-hub/src/server/modman/busses/proxies/group/can_2/lookout.rs +++ b/clover-hub/src/server/modman/busses/proxies/group/can_2/interface_lookout.rs @@ -5,6 +5,7 @@ use std::{ thread::sleep as std_sleep, }; +use crate::server::modman::busses::proxies::group::can_2::bus_manager::can_bus_manager; use crate::server::modman::busses::proxies::group::can_2::CAN2Bus; use nix::net::if_::if_nameindex; @@ -36,7 +37,8 @@ pub async fn can_lookout_thread(ctx: Arc) { // Something something tokio task pool is limited. std::thread::spawn(|| can_interface_lookout(lookout_ctx, lookout_tx)); - tokio::task::spawn(async move { can_bus_registrar(registrar_ctx, lookout_rx).await }).await; + let _ = + tokio::task::spawn(async move { can_bus_registrar(registrar_ctx, lookout_rx).await }).await; } /// Detects network interfaces to try and bind. @@ -135,23 +137,43 @@ pub async fn can_bus_registrar(ctx: Arc, mut channel: UnboundedReceiver if let Some(lookout_event) = channel.recv().await { match lookout_event { CanLookoutEvent::IFaceCreate(iface_details) => { - let (iface_name, iface_index) = iface_details; + let (iface_name, _iface_index) = iface_details; for permitted_iface in can2_config.permitted_interfaces.clone() { - if permitted_iface == iface_name { + if permitted_iface == iface_name.clone() { match bus_registry.get(&iface_name) { Some(_) => {} None => { info!( - "Found configured interface: {}, starting up a CAN bus listener...", + "Found configured interface: {}, starting up a CAN bus manager...", iface_name.clone() ); + + let manager_token = CancellationToken::new(); + let manager_ctx = ctx.clone(); + let manager_iface_name = iface_name.clone(); + + bus_registry.insert(iface_name.clone(), manager_token.clone()); + + tokio::task::spawn(async move { + can_bus_manager(manager_ctx, manager_token.clone(), manager_iface_name).await; + }); } } } } } - CanLookoutEvent::IFaceDestroy(iface_name) => todo!(), + CanLookoutEvent::IFaceDestroy(iface_name) => match bus_registry.get(&iface_name) { + Some(manager_token) => { + info!("Shutting down CAN 2 bus manager for interface: {iface_name}..."); + manager_token.cancel(); + } + // We assume that the other branch has loaded values in for the allowed interfaces. + // We could do that check again, and error out, but we is lazy. + // Also that check would be subject to a race condition if the interface no longer existed while being created, + // so we just do nothing. + None => {} + }, } } } diff --git a/clover-hub/src/server/modman/busses/proxies/group/can_2/mod.rs b/clover-hub/src/server/modman/busses/proxies/group/can_2/mod.rs index a2718b3..b406bec 100644 --- a/clover-hub/src/server/modman/busses/proxies/group/can_2/mod.rs +++ b/clover-hub/src/server/modman/busses/proxies/group/can_2/mod.rs @@ -4,7 +4,8 @@ //! pub mod bus_manager; -pub mod lookout; +pub mod interface_lookout; +pub mod module_listener; use std::{ sync::Arc, @@ -12,7 +13,7 @@ use std::{ }; use crate::server::modman::{ - busses::proxies::group::can_2::lookout::can_lookout_thread, + busses::proxies::group::can_2::interface_lookout::can_lookout_thread, models::store::ModManStore, }; diff --git a/clover-hub/src/server/modman/busses/proxies/group/can_2/module_listener.rs b/clover-hub/src/server/modman/busses/proxies/group/can_2/module_listener.rs new file mode 100644 index 0000000..c9fc410 --- /dev/null +++ b/clover-hub/src/server/modman/busses/proxies/group/can_2/module_listener.rs @@ -0,0 +1,52 @@ +use std::sync::Arc; + +use linux_socketcan_iso_tp::TokioSocketCanIsoTp; +use tokio_util::sync::CancellationToken; +use tracing::instrument; + +use crate::server::modman::busses::proxies::group::can_2::CAN2Bus; + +// JK You thought this function actually did something lmaoooooo +#[instrument(skip(cancellation_token, raw_socket))] +pub async fn can_bus_listener( + ctx: Arc, + cancellation_token: CancellationToken, + module_id: String, + raw_socket: TokioSocketCanIsoTp, +) { + let socket = Arc::new(raw_socket); + + let rx_session = ctx.session.clone(); + let rx_token = cancellation_token.clone(); + let rx_socket = socket.clone(); + let rx_id = module_id.clone(); + tokio::task::spawn(async move { + can_module_rx(rx_session, rx_token, rx_socket, rx_id).await; + }); + + let tx_session = ctx.session.clone(); + let tx_token = cancellation_token.clone(); + let tx_socket = socket.clone(); + let tx_id = module_id.clone(); + tokio::task::spawn(async move { + can_module_tx(tx_session, tx_token, tx_socket, tx_id).await; + }); +} + +#[instrument(skip(session, cancellation_token, socket))] +pub async fn can_module_rx( + session: Arc, + cancellation_token: CancellationToken, + socket: Arc, + module_id: String, +) { +} + +#[instrument(skip(session, cancellation_token, socket))] +pub async fn can_module_tx( + session: Arc, + cancellation_token: CancellationToken, + socket: Arc, + module_id: String, +) { +} diff --git a/clover-hub/src/server/modman/models/mod.rs b/clover-hub/src/server/modman/models/mod.rs index 4d1b7e5..14ab9bd 100644 --- a/clover-hub/src/server/modman/models/mod.rs +++ b/clover-hub/src/server/modman/models/mod.rs @@ -16,18 +16,27 @@ pub mod store; // TODO: Define defaults via `Default` trait impl. +/// Enum used to track the status of ports that clover knows about and can use. #[derive(Debug, Clone, Serialize, Deserialize)] pub enum PortStatus { /// Available but unused. + /// + /// Ports get set to this state if they're allowed in config and then aren't bound to by a proxy bus, + /// or are unrequested due to a module being disconnected. #[serde(rename = "available")] Available, - /// Requested by $MODULE_ID, but the UART bus isn't initialized yet + /// Requested by $MODULE_ID, but the bus isn't initialized yet. + /// + /// Usually occurs during startup. #[serde(rename = "requested")] Requested(String), - /// Currently being used by $MODULE_ID + /// Currently being used by $MODULE_ID. #[serde(rename = "bound")] Bound(String), - /// Unavailable, but still requested by $MODULE_ID + /// Couldn't be bound, but still requested by $MODULE_ID #[serde(rename = "unavailable")] Unavailable(String), + /// Is currently bound by $MODULE_ID, but the module is being deinitialized. + #[serde(rename = "unrequested")] + Unrequested(String), } diff --git a/clover-hub/src/server/modman/models/store.rs b/clover-hub/src/server/modman/models/store.rs index 8ce3d2e..3363480 100644 --- a/clover-hub/src/server/modman/models/store.rs +++ b/clover-hub/src/server/modman/models/store.rs @@ -17,14 +17,24 @@ pub struct PortStatuses { /// Used by the [UART Bus](super::busses::proxies::uart::UARTBus). pub uart: Arc>>, /// Used by the [CAN2 Bus](super::busses::proxies::can_2::CAN2Bus). + /// + /// Identifier string format is `$IFACE/$RX:$TX` where (e.g. `can0/0x101:0x201`): + /// - `$IFACE` is the network interface for the needed CAN bus + /// - `$RX` is the ***HEX*** representation of the CAN ID used for recieving messages from the module on. (Linux kernel will automatically filter replies for us.) + /// - `$TX` is the ***HEX*** representation of the CAN ID of the module that we're trying to communicate with. + /// + /// Matches regex: `r"^(?(?:\\w|[0-9])+)\\/(?0[xX][0-9a-fA-F]{3}):(?0[xX][0-9a-fA-F]{3})$"` pub can_2: Arc>>, } /// In memory data-store for components, modules, and any needed configuration. #[derive(Debug, Clone)] pub struct ModManStore { + /// All currently known modules and their states. pub modules: Arc>>, + /// All currently known components and their states. pub components: Arc>>>, + /// Global access to the current configuration. pub config: Arc>, pub gesture_states: Arc>>, pub foreground_gesture_priority: Arc>>, diff --git a/core/modules/blinkie/module.clover.jsonc b/core/modules/blinkie/module.clover.jsonc index 02423d1..6410452 100644 --- a/core/modules/blinkie/module.clover.jsonc +++ b/core/modules/blinkie/module.clover.jsonc @@ -2,6 +2,13 @@ "name": "C.O.R.E. Blinkie Example", "location": "@core.humanoid.chest", "internal": false, + "connection": { + "@default": { + "type": "can_2", + "tx_id": 201, + "rx_id": 101 + } + }, "components": { "light": { "type": "indicator",