cat_dev/fsemul/sdio/server/

mod.rs

//! "Server" implementations for SDIO protocols for handling PCFS for cat-dev.
//!
//! Reminder that SDIO Server, actually _connects_ to an open TCP port (like a
//! client normally would). It is called SDIO Server as it still
//! processes packets as a server.

mod message;
mod read;

use crate::{
	errors::CatBridgeError,
	fsemul::{
		filesystem::host::HostFilesystem,
		sdio::{
			DEFAULT_SDIO_BLOCK_PORT, DEFAULT_SDIO_CONTROL_PORT, SDIO_DATA_STREAMS,
			data_stream::DataStream,
			proto::{SdioControlPacketType, message::SdioControlTelnetChannel},
		},
	},
	mion::proto::control::MionBootType,
	net::{
		DEFAULT_CAT_DEV_CHUNK_SIZE, DEFAULT_CAT_DEV_SLOWDOWN,
		additions::{RequestIDLayer, StreamIDLayer},
		models::FromRef,
		server::{Router, TCPServer, models::ResponseStreamEvent},
	},
};
use scc::HashMap as ConcurrentHashMap;
use std::{
	net::{Ipv4Addr, SocketAddr, SocketAddrV4},
	sync::{Arc, LazyLock},
	time::Duration,
};
use tokio::sync::{Mutex, mpsc::Sender as BoundedSender, oneshot::Sender as OneshotSender};
use tower::ServiceBuilder;
use valuable::{Fields, NamedField, NamedValues, StructDef, Structable, Valuable, Value, Visit};

/// Underlying 'printf' buffers used to buffer messages that are incomplete.
///
/// The SDIO stream also supports a serial style logging. However, again as a
/// TCP Stream packets may be split even across the same `write` call. This
/// buffer ensures we only flush when we have a complete line.
///
/// Similar to SDIO Data Streams, this is created on SDIO Control stream start,
/// and gets claned up on SDIO Control stream end.
static SDIO_PRINTF_BUFFS: LazyLock<ConcurrentHashMap<(u64, SdioControlTelnetChannel), String>> =
	LazyLock::new(|| ConcurrentHashMap::with_capacity(1));

/// A builder for getting a TCP Server that is capable of serving SDIO to a
/// cat-dev console.
///
/// Unlike most "TCP Servers" though this is actually created by connecting to
/// the console directly. We don't actually bind a port on the host-pc side
/// that listens that the console connects to. Instead we're a server that
/// _Connects_ to our client.
#[derive(Clone, Debug)]
pub struct SdioServerBuilder {
	/// A hook that gets notified when the SDIO server has hit 'active' point, and all shells are definetly active.
	active_hook: Option<Arc<Mutex<Option<OneshotSender<()>>>>>,
	/// The boot type the underlying device is using.
	boot_type: MionBootType,
	/// Allow overriding the amount of time that we wait to make sure everything
	/// is safe for a cat-dev.
	///
	/// *note: `None` does not mean cat-dev sleep is disabled, that is controleld
	/// through [`Self::fully_disable_cat_dev_sleep`]*.
	cat_dev_sleep_override: Option<Duration>,
	/// Override the chunking value, and how many bytes we'll purposefully chunk on
	/// when sending out TCP data.
	///
	/// *note: `None` does not mean cat-dev sleep is disabled, that is controleld
	/// through [`Self::fully_disable_chunk_override`]*.
	chunk_override: Option<usize>,
	/// The port for SDIO Printf/Control to communicate on if not the default.
	control_port: Option<u16>,
	/// The port for SDIO Block Data to communicate on if not the default.
	data_port: Option<u16>,
	/// An override to fully disable any cat-dev sleeping to prevent issues.
	fully_disable_cat_dev_sleep: bool,
	/// An override to fully disable any TCP level chunking.
	fully_disable_chunk_override: bool,
	/// The underlying host filesystem to read from SDIO.
	host_filesystem: Option<HostFilesystem>,
	/// The IP to actually connect to.
	mion_ip: Ipv4Addr,
	/// A hook to modify
	sdio_telnet_message_hook: Option<BoundedSender<(SdioControlTelnetChannel, String)>>,
	/// If we should trace all I/O when built with debug assertions.
	trace_during_debug: bool,
}

impl SdioServerBuilder {
	#[must_use]
	pub const fn new(
		boot_type: MionBootType,
		host_filesystem: Option<HostFilesystem>,
		mion_ip: Ipv4Addr,
	) -> Self {
		Self {
			active_hook: None,
			boot_type,
			cat_dev_sleep_override: None,
			chunk_override: None,
			control_port: None,
			data_port: None,
			host_filesystem,
			fully_disable_cat_dev_sleep: false,
			fully_disable_chunk_override: false,
			mion_ip,
			sdio_telnet_message_hook: None,
			trace_during_debug: false,
		}
	}

	#[must_use]
	pub fn set_active_hook(mut self, hook: Option<OneshotSender<()>>) -> Self {
		self.active_hook = hook.map(|h| Arc::new(Mutex::new(Some(h))));
		self
	}
	#[must_use]
	pub fn raw_set_active_hook(
		mut self,
		hook: Option<Arc<Mutex<Option<OneshotSender<()>>>>>,
	) -> Self {
		self.active_hook = hook;
		self
	}

	#[must_use]
	pub const fn boot_type(&self) -> MionBootType {
		self.boot_type
	}
	#[must_use]
	pub const fn set_boot_type(mut self, new_type: MionBootType) -> Self {
		self.boot_type = new_type;
		self
	}

	#[must_use]
	pub const fn cat_dev_sleep_override(&self) -> Option<Duration> {
		self.cat_dev_sleep_override
	}
	#[must_use]
	pub const fn set_cat_dev_sleep_override(mut self, duration: Option<Duration>) -> Self {
		self.cat_dev_sleep_override = duration;
		self
	}

	#[must_use]
	pub const fn chunk_override(&self) -> Option<usize> {
		self.chunk_override
	}
	#[must_use]
	pub const fn set_chunk_override(mut self, new: Option<usize>) -> Self {
		self.chunk_override = new;
		self
	}

	#[must_use]
	pub const fn control_port(&self) -> Option<u16> {
		self.control_port
	}
	#[must_use]
	pub const fn set_control_port(mut self, new: Option<u16>) -> Self {
		self.control_port = new;
		self
	}

	#[must_use]
	pub const fn data_port(&self) -> Option<u16> {
		self.data_port
	}
	#[must_use]
	pub const fn set_data_port(mut self, new: Option<u16>) -> Self {
		self.data_port = new;
		self
	}

	#[must_use]
	pub const fn fully_disable_cat_dev_sleep(&self) -> bool {
		self.fully_disable_cat_dev_sleep
	}
	#[must_use]
	pub const fn set_fully_disable_cat_dev_sleep(mut self, new: bool) -> Self {
		self.fully_disable_cat_dev_sleep = new;
		self
	}

	#[must_use]
	pub const fn fully_disable_chunk_override(&self) -> bool {
		self.fully_disable_chunk_override
	}
	#[must_use]
	pub const fn set_fully_disable_chunk_override(mut self, new: bool) -> Self {
		self.fully_disable_chunk_override = new;
		self
	}

	#[must_use]
	pub const fn host_filesystem(&self) -> Option<&HostFilesystem> {
		self.host_filesystem.as_ref()
	}
	#[must_use]
	pub fn set_host_filesystem(mut self, new: Option<HostFilesystem>) -> Self {
		self.host_filesystem = new;
		self
	}

	#[must_use]
	pub const fn mion_ip(&self) -> Ipv4Addr {
		self.mion_ip
	}
	#[must_use]
	pub const fn set_mion_ip(mut self, new: Ipv4Addr) -> Self {
		self.mion_ip = new;
		self
	}

	#[must_use]
	pub const fn sdio_telnet_message_hook(
		&self,
	) -> Option<&BoundedSender<(SdioControlTelnetChannel, String)>> {
		self.sdio_telnet_message_hook.as_ref()
	}
	#[must_use]
	pub fn set_sdio_telnet_message_hook(
		mut self,
		new: Option<BoundedSender<(SdioControlTelnetChannel, String)>>,
	) -> Self {
		self.sdio_telnet_message_hook = new;
		self
	}

	#[must_use]
	pub const fn trace_during_debug(&self) -> bool {
		self.trace_during_debug
	}
	#[must_use]
	pub const fn set_trace_during_debug(mut self, new: bool) -> Self {
		self.trace_during_debug = new;
		self
	}

	/// Build a new TCP server that is capable of connecting to and handling SDIO
	/// streams.
	///
	/// ## Errors
	///
	/// If we cannot lookup the address to bind too, or there's been some
	/// programming error, and we run into an API error calling an internal
	/// api wrong.
	pub async fn build(self) -> Result<TCPServer<SdioStreamState>, CatBridgeError> {
		let control_port = self.control_port.unwrap_or(DEFAULT_SDIO_CONTROL_PORT);
		let data_port = self.data_port.unwrap_or(DEFAULT_SDIO_BLOCK_PORT);

		let mut router = Router::<SdioStreamState>::new();
		router.add_route(
			&[
				u8::try_from(u16::from(SdioControlPacketType::Read)).unwrap_or(u8::MAX),
				0,
			],
			read::handle_read_request,
		)?;
		router.add_route(
			&[
				u8::try_from(u16::from(SdioControlPacketType::TelnetMessage)).unwrap_or(u8::MAX),
				0,
			],
			message::handle_telnet_message,
		)?;

		let mut control_server = TCPServer::new_with_state(
			"sdio",
			SocketAddr::V4(SocketAddrV4::new(self.mion_ip, control_port)),
			router,
			(None, None),
			512_usize,
			SdioStreamState::new(
				self.active_hook,
				self.boot_type,
				self.chunk_override,
				data_port,
				self.host_filesystem,
				if self.fully_disable_cat_dev_sleep {
					None
				} else if let Some(over_ride) = self.cat_dev_sleep_override {
					Some(over_ride)
				} else {
					Some(DEFAULT_CAT_DEV_SLOWDOWN)
				},
				self.sdio_telnet_message_hook,
				#[cfg(debug_assertions)]
				self.trace_during_debug,
			),
			self.trace_during_debug,
		)
		.await?;
		control_server.set_on_stream_begin(on_sdio_stream_begin)?;
		control_server.set_on_stream_end(on_sdio_stream_end)?;
		if self.trace_during_debug {
			control_server.layer_initial_service(
				ServiceBuilder::new()
					.layer(RequestIDLayer::new("sdio".to_owned()))
					.layer(StreamIDLayer),
			);
		} else {
			control_server.layer_initial_service(
				ServiceBuilder::new().layer(RequestIDLayer::new("sdio".to_owned())),
			);
		}
		// We are still communicating with a slowdown...
		control_server.set_cat_dev_slowdown(control_server.state().cat_dev_slowdown);
		control_server.set_chunk_output_at_size(if self.fully_disable_chunk_override {
			None
		} else if let Some(over_ride) = self.chunk_override {
			Some(over_ride)
		} else {
			Some(DEFAULT_CAT_DEV_CHUNK_SIZE)
		});

		Ok(control_server)
	}
}

const SDIO_SERVER_BUILDER_FIELDS: &[NamedField<'static>] = &[
	NamedField::new("boot_type"),
	NamedField::new("cat_dev_sleep_override"),
	NamedField::new("chunk_override"),
	NamedField::new("control_port"),
	NamedField::new("data_port"),
	NamedField::new("fully_disable_cat_dev_sleep"),
	NamedField::new("fully_disable_chunk_override"),
	NamedField::new("host_filesystem"),
	NamedField::new("mion_ip"),
	NamedField::new("trace_during_debug"),
];

impl Structable for SdioServerBuilder {
	fn definition(&self) -> StructDef<'_> {
		StructDef::new_static(
			"SdioServerBuilder",
			Fields::Named(SDIO_SERVER_BUILDER_FIELDS),
		)
	}
}

impl Valuable for SdioServerBuilder {
	fn as_value(&self) -> Value<'_> {
		Value::Structable(self)
	}

	fn visit(&self, visitor: &mut dyn Visit) {
		visitor.visit_named_fields(&NamedValues::new(
			SDIO_SERVER_BUILDER_FIELDS,
			&[
				Valuable::as_value(&self.boot_type),
				Valuable::as_value(
					&self
						.cat_dev_sleep_override
						.map_or_else(|| "<none>".to_owned(), |dur| format!("{}s", dur.as_secs())),
				),
				Valuable::as_value(&self.chunk_override),
				Valuable::as_value(&self.control_port),
				Valuable::as_value(&self.data_port),
				Valuable::as_value(&self.fully_disable_cat_dev_sleep),
				Valuable::as_value(&self.fully_disable_chunk_override),
				Valuable::as_value(&self.host_filesystem),
				Valuable::as_value(&format!("{}", self.mion_ip)),
				Valuable::as_value(&self.trace_during_debug),
			],
		));
	}
}

/// Gets called when an SDIO Control stream begins.
///
/// This is where we actually connect to data-streams, and initialize our
/// PRINTF buffs. Right now this is fairly 'overkill' as TCP Servers using
/// the connection model (which SDIO uses) only actually have one underlying
/// stream, and can't possibly have multiple.
///
/// This means multiple SDIO servers will have to be created in order to
/// balance multiple connections. This mirrors the official SDK behavior.
///
/// ## Errors
///
/// If we cannot connect to the other SDIO data stream.
async fn on_sdio_stream_begin(
	event: ResponseStreamEvent<SdioStreamState>,
) -> Result<bool, CatBridgeError> {
	let mut addr = *event.source();
	addr.set_port(event.state().data_port);
	let stream = DataStream::connect(
		addr,
		event.state().cat_dev_slowdown,
		event.state().chunk_size,
		#[cfg(debug_assertions)]
		event.state().trace_during_debug,
	)
	.await?;
	let sid = event.stream_id();

	_ = SDIO_DATA_STREAMS.insert_async(sid, stream).await;
	_ = SDIO_PRINTF_BUFFS
		.insert_async(
			(sid, SdioControlTelnetChannel::SysConfigTool),
			String::with_capacity(0),
		)
		.await;

	Ok(true)
}

/// Gets called when an SDIO Control stream ends.
///
/// This is where we actually 'cleanup' all the data related to the SDIO data
/// stream. This includes both the printf buff, but also dropping the
/// tcp stream.
///
/// ## Errors
///
/// This shouldn't ever error, but needs to make the signature pass to auto
/// turn into a tower service.
async fn on_sdio_stream_end(
	event: ResponseStreamEvent<SdioStreamState>,
) -> Result<(), CatBridgeError> {
	let sid = event.stream_id();

	_ = SDIO_DATA_STREAMS.remove_async(&sid).await;
	_ = SDIO_PRINTF_BUFFS
		.remove_async(&(sid, SdioControlTelnetChannel::SysConfigTool))
		.await;

	Ok(())
}

/// The state of the SDIO Control server that every client ends up interacting
/// with.
///
/// Most folks will only actually use the host filesystem from this state, and
/// not interact with it directly.
#[derive(Clone, Debug)]
pub struct SdioStreamState {
	/// A hook to call when this SDIO Stream is active.
	active_hook: Option<Arc<Mutex<Option<OneshotSender<()>>>>>,
	/// The configure boot type of the Ccat-dev we're pointed at.
	boot_type: MionBootType,
	/// The slowdown to apply to data stream connections.
	cat_dev_slowdown: Option<Duration>,
	/// How much we should chunk packets coming out of the data stream.
	chunk_size: Option<usize>,
	/// The data port to connect to data streams on.
	data_port: u16,
	/// The host filesystem that actually contains pointers to the filesystem.
	host_fs: Option<HostFilesystem>,
	/// A hook that can 'watch' for any telnet messages, and do processing on
	/// them.
	sdio_telnet_message_hook: Option<BoundedSender<(SdioControlTelnetChannel, String)>>,
	/// Trace when debug mode is active.
	#[cfg(debug_assertions)]
	trace_during_debug: bool,
}

impl SdioStreamState {
	#[allow(clippy::too_many_arguments)]
	#[must_use]
	pub fn new(
		active_hook: Option<Arc<Mutex<Option<OneshotSender<()>>>>>,
		boot_type: MionBootType,
		chunk_size: Option<usize>,
		data_port: u16,
		host_fs: Option<HostFilesystem>,
		cat_dev_sleep: Option<Duration>,
		sdio_telnet_message_hook: Option<BoundedSender<(SdioControlTelnetChannel, String)>>,
		#[cfg(debug_assertions)] trace_during_debug: bool,
	) -> Self {
		Self {
			active_hook,
			boot_type,
			chunk_size,
			cat_dev_slowdown: cat_dev_sleep,
			data_port,
			host_fs,
			sdio_telnet_message_hook,
			#[cfg(debug_assertions)]
			trace_during_debug,
		}
	}
}

impl FromRef<SdioStreamState> for Option<HostFilesystem> {
	fn from_ref(input: &SdioStreamState) -> Self {
		input.host_fs.clone()
	}
}

impl FromRef<SdioStreamState> for MionBootType {
	fn from_ref(input: &SdioStreamState) -> Self {
		input.boot_type
	}
}

const SDIO_STREAM_STATE_FIELDS: &[NamedField<'static>] = &[
	NamedField::new("active_hook"),
	NamedField::new("boot_type"),
	NamedField::new("cat_dev_slowdown"),
	NamedField::new("chunk_size"),
	NamedField::new("data_port"),
	NamedField::new("host_fs"),
	NamedField::new("sdio_telnet_message_hook"),
	#[cfg(debug_assertions)]
	NamedField::new("trace_during_debug"),
];

impl Structable for SdioStreamState {
	fn definition(&self) -> StructDef<'_> {
		StructDef::new_static("SdioStreamState", Fields::Named(SDIO_STREAM_STATE_FIELDS))
	}
}

impl Valuable for SdioStreamState {
	fn as_value(&self) -> Value<'_> {
		Value::Structable(self)
	}

	fn visit(&self, visitor: &mut dyn Visit) {
		visitor.visit_named_fields(&NamedValues::new(
			SDIO_STREAM_STATE_FIELDS,
			&[
				Valuable::as_value(&self.active_hook.is_some()),
				Valuable::as_value(&self.boot_type),
				Valuable::as_value(&if let Some(slowdown) = self.cat_dev_slowdown {
					slowdown.as_secs()
				} else {
					0_u64
				}),
				Valuable::as_value(&self.chunk_size),
				Valuable::as_value(&self.data_port),
				Valuable::as_value(&self.host_fs),
				Valuable::as_value(&self.sdio_telnet_message_hook.is_some()),
				#[cfg(debug_assertions)]
				Valuable::as_value(&self.trace_during_debug),
			],
		));
	}
}