diff --git a/mujina-miner/Cargo.toml b/mujina-miner/Cargo.toml index 9e3939c3..5c5cbdc7 100644 --- a/mujina-miner/Cargo.toml +++ b/mujina-miner/Cargo.toml @@ -37,11 +37,9 @@ utoipa-axum = { workspace = true } utoipa-swagger-ui = { workspace = true } tracing = { workspace = true } tracing-subscriber = { workspace = true } -nix = { workspace = true } parking_lot = { workspace = true } regex = { workspace = true } reqwest = { workspace = true } -rustix = { workspace = true } slotmap = { workspace = true } num-traits = { workspace = true } ruint = "1.17.0" @@ -54,6 +52,20 @@ io-kit-sys = { workspace = true } tracing-journald = { workspace = true } udev = { workspace = true } +[target.'cfg(unix)'.dependencies] +nix = { workspace = true } +rustix = { workspace = true } + +[target.'cfg(windows)'.dependencies] +# Enable UsbPortInfo.interface so Windows serial discovery can order COM +# ports by USB interface number (interface 0 = control). See +# transport/usb/windows.rs. +serialport = { version = "4", features = ["usbportinfo-interface"] } +# nusb's Windows backend offloads blocking USB syscalls to the async runtime, +# so it needs the `tokio` feature to be awaited (Linux uses native async and +# does not require this). +nusb = { workspace = true, features = ["tokio"] } + [[bin]] name = "mujina-minerd" path = "src/bin/minerd.rs" diff --git a/mujina-miner/src/api_client/types.rs b/mujina-miner/src/api_client/types.rs index 6042ba0b..b149678e 100644 --- a/mujina-miner/src/api_client/types.rs +++ b/mujina-miner/src/api_client/types.rs @@ -29,6 +29,9 @@ pub struct BoardTelemetry { pub name: String, pub model: String, pub serial: Option, + /// ASIC hash clock in MHz, or null if the board does not report one. + #[serde(skip_serializing_if = "Option::is_none")] + pub frequency_mhz: Option, pub fans: Vec, pub temperatures: Vec, pub powers: Vec, diff --git a/mujina-miner/src/asic/bm13xx/thread.rs b/mujina-miner/src/asic/bm13xx/thread.rs index 6e874bdd..d0a03deb 100644 --- a/mujina-miner/src/asic/bm13xx/thread.rs +++ b/mujina-miner/src/asic/bm13xx/thread.rs @@ -27,6 +27,10 @@ use crate::{ types::{Difficulty, HashRate, ShareRate}, }; +/// Target hash clock the chip is ramped to during initialization, in MHz. +/// Boards report this as their operating frequency in telemetry. +pub const TARGET_FREQUENCY_MHZ: f32 = 525.0; + /// Tracks tasks sent to chip hardware, indexed by chip_job_id. /// /// BM13xx chips use 4-bit job IDs. This tracker maintains snapshots of @@ -435,9 +439,9 @@ where ) .await?; - // Frequency ramping (56.25 MHz -> 525 MHz) - debug!("Ramping frequency from 56.25 MHz to 525 MHz"); - let frequency_steps = generate_frequency_ramp_steps(56.25, 525.0, 6.25); + // Frequency ramping (56.25 MHz -> target) + debug!("Ramping frequency from 56.25 MHz to {TARGET_FREQUENCY_MHZ} MHz"); + let frequency_steps = generate_frequency_ramp_steps(56.25, TARGET_FREQUENCY_MHZ, 6.25); for (i, pll_config) in frequency_steps.iter().enumerate() { send_reg(chip_commands, true, Register::PllDivider(*pll_config)) diff --git a/mujina-miner/src/board/bitaxe.rs b/mujina-miner/src/board/bitaxe.rs index 2108e665..cee3a2e6 100644 --- a/mujina-miner/src/board/bitaxe.rs +++ b/mujina-miner/src/board/bitaxe.rs @@ -50,20 +50,21 @@ use crate::{ types::Temperature, }; -use super::{ - BackplaneConnector, BoardInfo, - pattern::{Match, StringMatch}, -}; +use super::{BackplaneConnector, BoardInfo, pattern::Match}; // Register this board type with the inventory system inventory::submit! { crate::board::BoardDescriptor { pattern: crate::board::pattern::BoardPattern { - vid: Match::Any, - pid: Match::Any, + // Match by VID:PID (c0de:cafe for bitaxe-raw firmware). Windows + // reports a generic "Microsoft" manufacturer for CDC ACM devices + // instead of the real "OSMU"/"Bitaxe" strings, so string matching + // fails there; VID:PID is stable across platforms. + vid: Match::Specific(0xc0de), + pid: Match::Specific(0xcafe), bcd_device: Match::Any, - manufacturer: Match::Specific(StringMatch::Exact("OSMU")), - product: Match::Specific(StringMatch::Exact("Bitaxe")), + manufacturer: Match::Any, + product: Match::Any, serial_pattern: Match::Any, }, name: "Bitaxe Gamma", @@ -111,33 +112,54 @@ async fn create_from_usb(device: UsbDeviceInfo) -> Result { time::sleep(Duration::from_millis(500)).await; - // Release ASIC from reset for discovery + // Release ASIC from reset for discovery. The BM1370 needs time to boot + // its UART before it will answer; give it the same ~500ms the reference + // Bitaxe firmware allows rather than a marginal 200ms. debug!("De-asserting ASIC nRST"); reset_pin.write(PinValue::High).await?; - time::sleep(Duration::from_millis(200)).await; - - // Version mask and chip discovery - debug!("Sending version mask configuration (3 times)"); - for i in 1..=3 { - trace!("Version mask send {}/3", i); - let version_cmd = Command::WriteRegister { - broadcast: true, - chip_address: 0x00, - register: bm13xx::protocol::Register::VersionMask( - bm13xx::protocol::VersionMask::full_rolling(), - ), - }; - data_writer - .send(version_cmd) - .await - .context("failed to send config command")?; - time::sleep(Duration::from_millis(5)).await; - } + time::sleep(Duration::from_millis(500)).await; - time::sleep(Duration::from_millis(10)).await; + // Version mask + chip discovery, retried a few times. A cold ASIC can + // miss the first round if its UART is not fully up, so re-send the + // version mask and re-run discovery until chips answer rather than + // failing the whole board on a single silent window. + const DISCOVERY_ATTEMPTS: usize = 5; + let mut chip_infos = Vec::new(); + for attempt in 1..=DISCOVERY_ATTEMPTS { + debug!("Sending version mask configuration (3 times)"); + for i in 1..=3 { + trace!("Version mask send {}/3", i); + let version_cmd = Command::WriteRegister { + broadcast: true, + chip_address: 0x00, + register: bm13xx::protocol::Register::VersionMask( + bm13xx::protocol::VersionMask::full_rolling(), + ), + }; + data_writer + .send(version_cmd) + .await + .context("failed to send config command")?; + time::sleep(Duration::from_millis(5)).await; + } - let chip_infos = discover_chips(&mut data_reader, &mut data_writer).await?; + time::sleep(Duration::from_millis(10)).await; + + match discover_chips(&mut data_reader, &mut data_writer).await { + Ok(chips) => { + chip_infos = chips; + break; + } + Err(e) if attempt < DISCOVERY_ATTEMPTS => { + warn!( + "Chip discovery attempt {attempt}/{DISCOVERY_ATTEMPTS} failed: {e}; retrying" + ); + time::sleep(Duration::from_millis(200)).await; + } + Err(e) => return Err(e), + } + } debug!(count = chip_infos.len(), "Discovered chips"); @@ -194,6 +216,7 @@ async fn create_from_usb(device: UsbDeviceInfo) -> Result { name: board_name.clone(), model: "Bitaxe Gamma".into(), serial: serial.clone(), + frequency_mhz: Some(bm13xx::thread::TARGET_FREQUENCY_MHZ), ..Default::default() }; let (telemetry_tx, telemetry_rx) = watch::channel(initial_state); @@ -212,7 +235,8 @@ async fn create_from_usb(device: UsbDeviceInfo) -> Result { board_name, board_model: "Bitaxe Gamma", board_serial: serial, - bad_thermal_count: 0, + over_temp_count: 0, + sensor_fault_count: 0, asic_enable: asic_enable_monitor, }; @@ -242,9 +266,14 @@ struct Bitaxe { board_name: String, board_model: &'static str, board_serial: Option, - /// Consecutive bad thermal readings (I2C error, out-of-range, or - /// above emergency threshold). Triggers emergency shutdown. - bad_thermal_count: u32, + /// Consecutive readings at or above the emergency temperature. + /// Triggers a fast thermal shutdown. + over_temp_count: u32, + /// Consecutive cycles with no usable temperature reading (I2C + /// error or out-of-range value). Tolerated longer than a genuine + /// over-temp because it is usually a transient control-bus desync, + /// but still shuts the board down if the sensor stays unreadable. + sensor_fault_count: u32, asic_enable: BitaxeAsicEnable, } @@ -284,15 +313,19 @@ impl Bitaxe { /// telemetry, and logs a periodic summary. /// /// Temperature readings fall into four categories: - /// - I2C errors or diode faults: no temperature available. + /// - I2C errors: no temperature available (usually a transient + /// control-bus desync), increments the sensor-fault counter. /// - Out of plausible range (outside 0..120 C): implausible, - /// treated as a bad reading. - /// - Above the emergency threshold: valid but dangerous. - /// - Normal: valid and safe, resets the bad-reading counter. + /// also counted as a sensor fault. + /// - At or above the emergency threshold: valid but dangerous, + /// increments the over-temperature counter. + /// - Normal: valid and safe, resets both counters. /// - /// Any category except normal increments a consecutive - /// bad-reading counter. After BAD_READING_LIMIT consecutive - /// bad readings, the board shuts down. + /// A genuine over-temperature trips a fast shutdown after + /// OVER_TEMP_LIMIT consecutive readings. An unreadable sensor is + /// tolerated for longer (SENSOR_FAULT_LIMIT) so a transient bus + /// glitch can re-sync, but still shuts down if it persists — we + /// cannot run the ASIC blind to its temperature. async fn monitor_tick( &mut self, tx: &watch::Sender, @@ -325,7 +358,12 @@ impl Bitaxe { const EXPECTED_MIN_C: f32 = 0.0; const EXPECTED_MAX_C: f32 = 120.0; const EMERGENCY_TEMP_C: f32 = 80.0; - const BAD_READING_LIMIT: u32 = 3; + // Consecutive genuinely-dangerous readings before shutdown. + const OVER_TEMP_LIMIT: u32 = 3; + // Consecutive unreadable-sensor cycles before shutdown. At the + // ~2s monitor cadence this is ~20s, long enough for a transient + // I2C/control-bus desync to re-sync without nuking the board. + const SENSOR_FAULT_LIMIT: u32 = 10; // The EMC2101 measures temperature via a diode on the ASIC // die. When the ASIC comes out of reset, the resulting // electrical transient corrupts the first few ADC conversions. @@ -339,47 +377,71 @@ impl Bitaxe { let asic_temp = if diode_ready { match raw_temp { Ok(t) if !(EXPECTED_MIN_C..=EXPECTED_MAX_C).contains(&t) => { - self.bad_thermal_count += 1; + self.sensor_fault_count += 1; trace!(temp_c = t, "Discarding out-of-range temperature reading"); None } Ok(t) if t >= EMERGENCY_TEMP_C => { - self.bad_thermal_count += 1; + self.over_temp_count += 1; + self.sensor_fault_count = 0; warn!( temp_c = t, - consecutive = self.bad_thermal_count, + consecutive = self.over_temp_count, "Temperature above emergency threshold" ); Some(t) } Ok(t) => { - self.bad_thermal_count = 0; + self.over_temp_count = 0; + self.sensor_fault_count = 0; Some(t) } Err(e) => { - self.bad_thermal_count += 1; - warn!("Temperature read failed: {}", e); + self.sensor_fault_count += 1; + warn!( + consecutive = self.sensor_fault_count, + "Temperature read failed: {}", e + ); None } } } else { - self.bad_thermal_count = 0; + self.over_temp_count = 0; + self.sensor_fault_count = 0; None }; - // Without reliable temperature readings we cannot operate - // safely. Shut down the board. - if self.bad_thermal_count >= BAD_READING_LIMIT { + // A sustained run of genuinely dangerous temperatures is a real + // thermal emergency: shut down fast. + if self.over_temp_count >= OVER_TEMP_LIMIT { error!( - consecutive = self.bad_thermal_count, + consecutive = self.over_temp_count, "THERMAL EMERGENCY: shutting down board" ); if let Err(e) = self.emc2101.set_fan_speed(Percent::FULL).await { error!("Failed to set fan speed: {}", e); } bail!( - "thermal emergency after {} consecutive bad readings", - self.bad_thermal_count + "thermal emergency after {} consecutive over-temperature readings", + self.over_temp_count + ); + } + + // A missing temperature reading is usually a transient control-bus + // desync that re-syncs on its own. Tolerate a longer window, but + // still shut down if the sensor stays unreadable — we cannot run + // the ASIC blind to its temperature. + if self.sensor_fault_count >= SENSOR_FAULT_LIMIT { + error!( + consecutive = self.sensor_fault_count, + "Temperature sensor unreadable: shutting down board" + ); + if let Err(e) = self.emc2101.set_fan_speed(Percent::FULL).await { + error!("Failed to set fan speed: {}", e); + } + bail!( + "temperature sensor unreadable after {} consecutive failed readings", + self.sensor_fault_count ); } @@ -388,6 +450,7 @@ impl Bitaxe { name: self.board_name.clone(), model: self.board_model.into(), serial: self.board_serial.clone(), + frequency_mhz: Some(bm13xx::thread::TARGET_FREQUENCY_MHZ), fans: vec![Fan { name: "fan".into(), rpm: fan_rpm, diff --git a/mujina-miner/src/daemon.rs b/mujina-miner/src/daemon.rs index ea916aa0..02d20fa8 100644 --- a/mujina-miner/src/daemon.rs +++ b/mujina-miner/src/daemon.rs @@ -5,6 +5,7 @@ use std::env; +#[cfg(unix)] use tokio::signal::unix::{self, SignalKind}; use tokio::sync::{mpsc, watch}; use tokio_util::{sync::CancellationToken, task::TaskTracker}; @@ -281,18 +282,28 @@ impl Daemon { info!("Started."); info!("For debugging, set RUST_LOG=mujina_miner=debug or trace."); - // Install signal handlers - let mut sigint = unix::signal(SignalKind::interrupt())?; - let mut sigterm = unix::signal(SignalKind::terminate())?; - - // Wait for shutdown signal - tokio::select! { - _ = sigint.recv() => { - info!("Received SIGINT."); - }, - _ = sigterm.recv() => { - info!("Received SIGTERM."); - }, + // Wait for shutdown signal (platform-specific). + #[cfg(unix)] + { + // Install Unix signal handlers. + let mut sigint = unix::signal(SignalKind::interrupt())?; + let mut sigterm = unix::signal(SignalKind::terminate())?; + + tokio::select! { + _ = sigint.recv() => { + info!("Received SIGINT."); + }, + _ = sigterm.recv() => { + info!("Received SIGTERM."); + }, + } + } + + // Windows has no SIGTERM; Ctrl-C is the graceful-shutdown signal. + #[cfg(windows)] + { + tokio::signal::ctrl_c().await?; + info!("Received Ctrl-C."); } // Initiate shutdown diff --git a/mujina-miner/src/transport/mod.rs b/mujina-miner/src/transport/mod.rs index 47bcf16f..1fae3f55 100644 --- a/mujina-miner/src/transport/mod.rs +++ b/mujina-miner/src/transport/mod.rs @@ -8,6 +8,10 @@ use anyhow::Result; pub mod cpu; +#[cfg(not(windows))] +pub mod serial; +#[cfg(windows)] +#[path = "serial_windows.rs"] pub mod serial; pub mod usb; diff --git a/mujina-miner/src/transport/serial_windows.rs b/mujina-miner/src/transport/serial_windows.rs new file mode 100644 index 00000000..ffc0870f --- /dev/null +++ b/mujina-miner/src/transport/serial_windows.rs @@ -0,0 +1,375 @@ +//! Windows serial stream implementation using tokio-serial. +//! +//! ## Why this exists +//! +//! `transport/serial.rs` is a Unix-only implementation built directly on raw +//! file descriptors and termios (via `rustix`) so that a `SerialControl` +//! handle can reconfigure the port (in particular, change the baud rate) even +//! after the stream has been split into concurrent read/write halves. +//! +//! Windows has no raw-fd/termios model, so this module reimplements the same +//! public API on top of `tokio-serial`, which itself wraps `serialport-rs`'s +//! Windows backend (`SetCommState`/`SetCommTimeouts` via the Win32 comm API). +//! Shared ownership + a `parking_lot::RwLock` around the underlying +//! `tokio_serial::SerialStream` gives `SerialControl` mutable access to +//! reconfigure the port while `SerialReader`/`SerialWriter` are in use +//! elsewhere, mirroring the atomic/lock design of the Unix implementation. +//! +//! ## Known behavioral differences vs. the Unix implementation +//! +//! - `SerialControl::set_baud_rate` on Unix drains the output buffer before +//! applying the new baud rate. `serialport-rs`'s Windows backend +//! reconfigures the port's DCB directly and does not itself guarantee +//! queued writes have drained first. Because `set_baud_rate` is a +//! synchronous, `&self` method (to match the cross-platform API), it cannot +//! `.await` a flush here. Callers relying on drain-before-switch semantics +//! should ensure the writer has finished before calling it. +//! - I/O errors are surfaced as the underlying `io::Error` from `tokio-serial` +//! rather than being classified into `SerialError::Disconnected` / +//! `SerialError::HardwareError`, since POSIX errno values have no reliable +//! Windows equivalent. +//! - `SerialStream::from_fd` has no Windows analog and is intentionally not +//! provided. On Unix it is `#[cfg(test)] pub(crate)` and only used by that +//! module's own PTY-based unit tests; no code outside `serial.rs` calls it. + +use std::io; +use std::pin::Pin; +use std::sync::Arc; +use std::sync::atomic::{AtomicU8, AtomicU32, AtomicU64, Ordering}; +use std::task::{Context, Poll}; +use std::time::Duration; + +use parking_lot::RwLock; +use tokio::io::{AsyncRead, AsyncWrite, ReadBuf}; +use tokio_serial::{SerialPort, SerialPortBuilderExt, SerialStream as TokioSerialStream}; + +/// Parity configuration. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Parity { + None = 0, + Odd = 1, + Even = 2, +} + +impl From for tokio_serial::Parity { + fn from(p: Parity) -> Self { + match p { + Parity::None => tokio_serial::Parity::None, + Parity::Odd => tokio_serial::Parity::Odd, + Parity::Even => tokio_serial::Parity::Even, + } + } +} + +/// Serial port configuration. +#[derive(Debug, Clone, Copy)] +pub struct SerialConfig { + pub baud_rate: u32, + pub data_bits: u8, + pub stop_bits: u8, + pub parity: Parity, +} + +impl Default for SerialConfig { + fn default() -> Self { + Self { + baud_rate: 115200, + data_bits: 8, + stop_bits: 1, + parity: Parity::None, + } + } +} + +/// Serial port error types. +#[derive(Debug, thiserror::Error)] +pub enum SerialError { + #[error("Failed to open serial port: {0}")] + OpenError(#[source] io::Error), + + #[error("Unsupported baud rate: {0}")] + UnsupportedBaudRate(u32), + + #[error("Configuration failed: {0}")] + ConfigError(String), + + #[error("I/O error: {0}")] + IoError(#[from] io::Error), + + #[error("Serial port disconnected")] + Disconnected, + + #[error("Hardware error on serial port")] + HardwareError, + + #[error("Operation timed out")] + Timeout, +} + +/// Statistics for a serial port. +#[derive(Debug, Clone, Copy)] +pub struct SerialStats { + /// Total bytes read from the port. + pub bytes_read: u64, + /// Total bytes written to the port. + pub bytes_written: u64, + /// Current baud rate. + pub baud_rate: u32, +} + +/// A serial stream implementation that supports runtime reconfiguration. +pub struct SerialStream { + inner: Arc, +} + +struct SerialInner { + /// The underlying tokio-serial stream. Guarded by a lock so + /// `SerialControl` can reconfigure it (e.g. change baud rate) while + /// `SerialReader`/`SerialWriter` are concurrently performing I/O. + /// + /// `tokio_serial::SerialStream` is `Send`, so `RwLock` + /// is `Send + Sync` without needing any `unsafe impl`. + stream: RwLock, + + /// Current configuration - atomic for lock-free reads. + baud_rate: AtomicU32, + data_bits: AtomicU8, + stop_bits: AtomicU8, + parity: AtomicU8, // 0 = None, 1 = Odd, 2 = Even + + /// Lock held only for the duration of an actual reconfiguration. + reconfig_lock: RwLock<()>, + + /// Statistics (lock-free). + bytes_read: AtomicU64, + bytes_written: AtomicU64, +} + +/// Reader half of a split serial stream. +pub struct SerialReader { + inner: Arc, +} + +/// Writer half of a split serial stream. +pub struct SerialWriter { + inner: Arc, +} + +/// Control handle for a split serial stream. +pub struct SerialControl { + inner: Arc, +} + +/// Build a `tokio_serial::SerialStream` from a `SerialConfig`. +fn build_stream(path: &str, config: &SerialConfig) -> Result { + let data_bits = match config.data_bits { + 5 => tokio_serial::DataBits::Five, + 6 => tokio_serial::DataBits::Six, + 7 => tokio_serial::DataBits::Seven, + 8 => tokio_serial::DataBits::Eight, + _ => { + return Err(SerialError::ConfigError(format!( + "Invalid data bits: {}", + config.data_bits + ))); + } + }; + + let stop_bits = match config.stop_bits { + 1 => tokio_serial::StopBits::One, + 2 => tokio_serial::StopBits::Two, + _ => { + return Err(SerialError::ConfigError(format!( + "Invalid stop bits: {}", + config.stop_bits + ))); + } + }; + + tokio_serial::new(path, config.baud_rate) + .data_bits(data_bits) + .stop_bits(stop_bits) + .parity(config.parity.into()) + .timeout(Duration::from_millis(100)) + .open_native_async() + .map_err(|e| SerialError::OpenError(io::Error::other(e))) +} + +impl SerialStream { + /// Open a new serial port with the specified baud rate. + /// + /// Uses default configuration of 8N1 (8 data bits, no parity, 1 stop bit). + pub fn new(path: &str, baud_rate: u32) -> Result { + let config = SerialConfig { + baud_rate, + ..Default::default() + }; + Self::with_config(path, config) + } + + /// Open a new serial port with the specified configuration. + /// + /// This is synchronous to match the Unix implementation's API. + /// `open_native_async` does not block on I/O; it constructs the OS handle + /// and binds it to the current Tokio reactor (a `Handle` must already be + /// current, i.e. this must be called from within a Tokio runtime context, + /// exactly as on the Unix path via `AsyncFd::new`). + pub fn with_config(path: &str, config: SerialConfig) -> Result { + let stream = build_stream(path, &config)?; + + Ok(Self { + inner: Arc::new(SerialInner { + stream: RwLock::new(stream), + baud_rate: AtomicU32::new(config.baud_rate), + data_bits: AtomicU8::new(config.data_bits), + stop_bits: AtomicU8::new(config.stop_bits), + parity: AtomicU8::new(config.parity as u8), + reconfig_lock: RwLock::new(()), + bytes_read: AtomicU64::new(0), + bytes_written: AtomicU64::new(0), + }), + }) + } + + /// Split the stream into reader, writer, and control handles. + /// + /// This allows concurrent reading and writing while maintaining the + /// ability to reconfigure the port. + pub fn split(self) -> (SerialReader, SerialWriter, SerialControl) { + ( + SerialReader { + inner: self.inner.clone(), + }, + SerialWriter { + inner: self.inner.clone(), + }, + SerialControl { inner: self.inner }, + ) + } +} + +impl AsyncRead for SerialReader { + fn poll_read( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &mut ReadBuf<'_>, + ) -> Poll> { + let before = buf.filled().len(); + let mut stream = self.inner.stream.write(); + let result = Pin::new(&mut *stream).poll_read(cx, buf); + if result.is_ready() { + let n = buf.filled().len() - before; + if n > 0 { + self.inner.bytes_read.fetch_add(n as u64, Ordering::Relaxed); + } + } + result + } +} + +impl AsyncWrite for SerialWriter { + fn poll_write( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &[u8], + ) -> Poll> { + let mut stream = self.inner.stream.write(); + let result = Pin::new(&mut *stream).poll_write(cx, buf); + if let Poll::Ready(Ok(n)) = &result + && *n > 0 + { + self.inner + .bytes_written + .fetch_add(*n as u64, Ordering::Relaxed); + } + result + } + + fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + let mut stream = self.inner.stream.write(); + Pin::new(&mut *stream).poll_flush(cx) + } + + fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + let mut stream = self.inner.stream.write(); + Pin::new(&mut *stream).poll_shutdown(cx) + } +} + +impl SerialControl { + /// Change the baud rate of the serial port. + /// + /// Thread-safe and callable while I/O is in progress. See the module-level + /// docs for a note on drain semantics differing from the Unix version. + pub fn set_baud_rate(&self, baud_rate: u32) -> Result<(), SerialError> { + const TIMEOUT: Duration = Duration::from_secs(5); + + let _lock = match self.inner.reconfig_lock.try_write_for(TIMEOUT) { + Some(guard) => guard, + None => { + return Err(SerialError::ConfigError( + "Failed to acquire configuration lock - possible deadlock".to_string(), + )); + } + }; + + let mut stream = self.inner.stream.write(); + stream + .set_baud_rate(baud_rate) + .map_err(|e| SerialError::ConfigError(format!("Failed to set baud rate: {}", e)))?; + drop(stream); + + self.inner.baud_rate.store(baud_rate, Ordering::Release); + + Ok(()) + } + + /// Get the current baud rate. + pub fn current_baud_rate(&self) -> u32 { + self.inner.baud_rate.load(Ordering::Acquire) + } + + /// Get the current data bits configuration. + pub fn current_data_bits(&self) -> u8 { + self.inner.data_bits.load(Ordering::Acquire) + } + + /// Get the current stop bits configuration. + pub fn current_stop_bits(&self) -> u8 { + self.inner.stop_bits.load(Ordering::Acquire) + } + + /// Get the current parity configuration. + pub fn current_parity(&self) -> Parity { + match self.inner.parity.load(Ordering::Acquire) { + 1 => Parity::Odd, + 2 => Parity::Even, + _ => Parity::None, + } + } + + /// Get the current serial port configuration. + pub fn current_config(&self) -> SerialConfig { + SerialConfig { + baud_rate: self.current_baud_rate(), + data_bits: self.current_data_bits(), + stop_bits: self.current_stop_bits(), + parity: self.current_parity(), + } + } + + /// Get statistics about the serial port. + pub fn stats(&self) -> SerialStats { + SerialStats { + bytes_read: self.inner.bytes_read.load(Ordering::Relaxed), + bytes_written: self.inner.bytes_written.load(Ordering::Relaxed), + baud_rate: self.current_baud_rate(), + } + } + + /// Reset the statistics counters to zero. + pub fn reset_stats(&self) { + self.inner.bytes_read.store(0, Ordering::Relaxed); + self.inner.bytes_written.store(0, Ordering::Relaxed); + } +} diff --git a/mujina-miner/src/transport/usb.rs b/mujina-miner/src/transport/usb.rs index 1111d778..2b4565c3 100644 --- a/mujina-miner/src/transport/usb.rs +++ b/mujina-miner/src/transport/usb.rs @@ -35,11 +35,16 @@ mod macos; #[cfg(target_os = "macos")] use macos as platform; +#[cfg(target_os = "windows")] +mod windows; +#[cfg(target_os = "windows")] +use windows as platform; + // On unsupported platforms, serial port discovery is a stub so the // miner still compiles (e.g., for CPU mining). If this is ever // called, something has gone wrong because a board matched a USB // device on a platform where we can't find its serial ports. -#[cfg(not(any(target_os = "linux", target_os = "macos")))] +#[cfg(not(any(target_os = "linux", target_os = "macos", target_os = "windows")))] mod platform { use anyhow::{Result, bail}; use nusb::DeviceInfo; diff --git a/mujina-miner/src/transport/usb/windows.rs b/mujina-miner/src/transport/usb/windows.rs new file mode 100644 index 00000000..d15b13d4 --- /dev/null +++ b/mujina-miner/src/transport/usb/windows.rs @@ -0,0 +1,91 @@ +//! Windows serial-port discovery for USB devices. +//! +//! `main` discovers USB devices with nusb (cross-platform), but mapping a +//! device to its COM ports is platform-specific. On Windows we enumerate COM +//! ports via `serialport`/`tokio-serial` and match them to the nusb device by +//! VID:PID:serial. +//! +//! Windows assigns COM numbers from the registry with no relation to USB +//! interface order, so we order the returned ports by USB interface number +//! (interface 0 first) to keep control/data assignment stable across machines. +//! This requires the `usbportinfo-interface` feature on `serialport`. + +use std::time::Duration; + +use anyhow::{Result, bail}; +use nusb::DeviceInfo; +use tokio_serial::{SerialPortType, available_ports}; + +/// Build the platform device path. On Windows we key devices by +/// `vid:pid:serial` (lowercased) so the value can be matched against the +/// COM-port enumeration, which is the only identifier the two APIs share. +pub fn device_path(device: &DeviceInfo) -> String { + make_key( + device.vendor_id(), + device.product_id(), + device.serial_number(), + ) +} + +/// Normalized `vid:pid:serial` key. Serial is lowercased because nusb and +/// `serialport` can report it in different cases on Windows. +fn make_key(vid: u16, pid: u16, serial: Option<&str>) -> String { + format!( + "{:04x}:{:04x}:{}", + vid, + pid, + serial.unwrap_or("no-serial").to_lowercase() + ) +} + +/// Find the COM ports for the USB device identified by `device_path` (a +/// `vid:pid:serial` key from [`device_path`]), ordered by USB interface +/// number. Retries until at least `expected` ports appear or the timeout +/// elapses, since COM ports can lag slightly behind USB enumeration. +pub async fn get_serial_ports(device_path: &str, expected: usize) -> Result> { + const RETRY_INTERVAL: Duration = Duration::from_millis(250); + const MAX_ATTEMPTS: usize = 24; // ~6 seconds + // After a hotplug, a COM port can appear in the enumeration slightly before + // its device node is openable (CreateFile returns ERROR_FILE_NOT_FOUND). + // Once all expected ports are present, wait this long and re-check so the + // caller's subsequent open() succeeds. + const SETTLE: Duration = Duration::from_millis(1000); + + let mut last_found = 0; + for _ in 0..MAX_ATTEMPTS { + let ports = matching_ports(device_path)?; + if ports.len() >= expected { + tokio::time::sleep(SETTLE).await; + let settled = matching_ports(device_path)?; + if settled.len() >= expected { + return Ok(settled); + } + } + last_found = ports.len(); + tokio::time::sleep(RETRY_INTERVAL).await; + } + + bail!( + "expected {expected} serial ports for USB device {device_path}, \ + found {last_found} after retries" + ); +} + +/// Enumerate COM ports belonging to the given `vid:pid:serial` device, +/// ordered by USB interface number (falling back to port name). +fn matching_ports(device_path: &str) -> Result> { + let mut matched: Vec<(Option, String)> = Vec::new(); + + for port in available_ports()? { + if let SerialPortType::UsbPort(usb) = &port.port_type + && make_key(usb.vid, usb.pid, usb.serial_number.as_deref()) == device_path + { + matched.push((usb.interface, port.port_name.clone())); + } + } + + // Windows COM numbers don't follow interface order; sort by interface + // (interface 0 = control) so control/data mapping is deterministic. + matched.sort_by(|a, b| a.0.cmp(&b.0).then_with(|| a.1.cmp(&b.1))); + Ok(matched.into_iter().map(|(_, name)| name).collect()) +}