diff --git a/mujina-miner/src/api/server.rs b/mujina-miner/src/api/server.rs index 5a7a53d4..389c06b3 100644 --- a/mujina-miner/src/api/server.rs +++ b/mujina-miner/src/api/server.rs @@ -41,6 +41,12 @@ impl SharedState { .lock() .unwrap_or_else(|e| e.into_inner()) .boards(); + state.hashrate = state + .boards + .iter() + .flat_map(|board| board.threads.iter()) + .map(|thread| thread.hashrate) + .sum(); state } } @@ -136,7 +142,7 @@ mod tests { use super::*; use crate::api::commands::SchedulerCommand; - use crate::api_client::types::{BoardState, SourceState}; + use crate::api_client::types::{BoardState, SourceState, ThreadState}; use crate::board::BoardRegistration; /// Test fixtures returned by the router builder. @@ -205,6 +211,11 @@ mod tests { let board = BoardState { name: "test-board".into(), model: "TestModel".into(), + threads: vec![ThreadState { + name: "thread-0".into(), + hashrate: 1_000_000, + is_active: true, + }], ..Default::default() }; let fixtures = build_test_router(miner_state, vec![board]); diff --git a/mujina-miner/src/api_client/types.rs b/mujina-miner/src/api_client/types.rs index 3fdab742..926ad3fb 100644 --- a/mujina-miner/src/api_client/types.rs +++ b/mujina-miner/src/api_client/types.rs @@ -27,12 +27,33 @@ pub struct BoardState { pub name: String, pub model: String, pub serial: Option, + /// Configured board operating frequency in MHz, if known. + pub frequency_mhz: Option, + /// Number of hashboard channels exposed by this board, if known. + pub hashboard_count: Option, + /// Number of hashboard channels currently active in Mujina, if known. + pub active_hashboard_count: Option, pub fans: Vec, + pub hashboards: Vec, pub temperatures: Vec, pub powers: Vec, pub threads: Vec, } +/// Per-hashboard connectivity and activity state. +#[derive(Clone, Debug, Deserialize, Serialize, ToSchema)] +pub struct HashboardState { + pub index: u8, + /// Serial port assigned to this hashboard data channel, if known. + pub serial_port: Option, + /// Whether the bridge currently detects a hashboard on this channel. + pub is_present: bool, + /// Whether Mujina has started a worker thread for this channel. + pub is_active: bool, + /// Hashrate in hashes per second for this hashboard. + pub hashrate: u64, +} + /// Fan status. #[derive(Clone, Debug, Deserialize, Serialize, ToSchema)] pub struct Fan { diff --git a/mujina-miner/src/asic/bm13xx/chain_config.rs b/mujina-miner/src/asic/bm13xx/chain_config.rs index 40baf714..a5a0c93b 100644 --- a/mujina-miner/src/asic/bm13xx/chain_config.rs +++ b/mujina-miner/src/asic/bm13xx/chain_config.rs @@ -75,6 +75,10 @@ pub struct ChainPeripherals { /// Voltage regulator control (optional, may be shared across chains). pub voltage_regulator: Option>>, + + /// Serializes full cold-start initialization when multiple chains share + /// board-level control hardware such as a PSU or reset/control bridge. + pub initialization_lock: Arc>, } #[cfg(test)] @@ -111,6 +115,7 @@ mod tests { let peripherals = ChainPeripherals { asic_enable: enable, voltage_regulator: None, + initialization_lock: Arc::new(Mutex::new(())), }; // Hash thread enables diff --git a/mujina-miner/src/asic/bm13xx/thread_v2.rs b/mujina-miner/src/asic/bm13xx/thread_v2.rs index 4f4425f1..8d58df99 100644 --- a/mujina-miner/src/asic/bm13xx/thread_v2.rs +++ b/mujina-miner/src/asic/bm13xx/thread_v2.rs @@ -33,7 +33,7 @@ use crate::{ }, job_source::MerkleRootKind, tracing::prelude::*, - types::{Difficulty, HashRate}, + types::{Difficulty, HashRate, HashrateEstimator, Work}, }; /// Minimum chip count required for initialization to succeed. @@ -106,6 +106,7 @@ impl BM13xxThread { // share_target >= TicketMask difficulty. The HashrateEstimator will // refine this once shares start flowing. let initial_hashrate = HashRate::from_gigahashes(83.0 * chain.chip_count() as f64); + let nonce_work = nonce_work_for_chain(chain.chip_count()); // Spawn reader task - continuously reads from serial to prevent USB // CDC-ACM flow control from blocking TX. Runs until chip_rx closes. @@ -125,6 +126,8 @@ impl BM13xxThread { chip_state: ChipState::Disabled, current_task: None, chip_jobs: ChipJobs::new(), + hashrate_estimator: HashrateEstimator::new(Duration::from_secs(60)), + nonce_work, }; actor.run().await; }); @@ -139,6 +142,11 @@ impl BM13xxThread { event_rx: Some(evt_rx), }) } + + /// Shared cached runtime status for external telemetry publishers. + pub fn status_handle(&self) -> Arc> { + Arc::clone(&self.status) + } } /// Reader task that continuously reads from the serial port. @@ -175,30 +183,28 @@ impl HashThread for BM13xxThread { &mut self, new_task: HashTask, ) -> Result, HashThreadError> { - let (tx, rx) = oneshot::channel(); self.command_tx .send(ThreadCommand::UpdateTask { new_task, - response_tx: tx, + response_tx: oneshot::channel().0, }) .await .map_err(|_| HashThreadError::ThreadOffline)?; - rx.await.map_err(|_| HashThreadError::ThreadOffline)? + Ok(None) } async fn replace_task( &mut self, new_task: HashTask, ) -> Result, HashThreadError> { - let (tx, rx) = oneshot::channel(); self.command_tx .send(ThreadCommand::ReplaceTask { new_task, - response_tx: tx, + response_tx: oneshot::channel().0, }) .await .map_err(|_| HashThreadError::ThreadOffline)?; - rx.await.map_err(|_| HashThreadError::ThreadOffline)? + Ok(None) } async fn go_idle(&mut self) -> Result, HashThreadError> { @@ -263,6 +269,8 @@ struct BM13xxActor { current_task: Option, /// Maps chip job IDs to tasks for nonce correlation. chip_jobs: ChipJobs, + hashrate_estimator: HashrateEstimator, + nonce_work: Work, } /// Commands sent from the facade ([`BM13xxThread`]) to the actor. @@ -428,6 +436,8 @@ where // ntime rolling timer - sends new job every second with incremented timestamp let mut ntime_ticker = time::interval(Duration::from_secs(1)); ntime_ticker.set_missed_tick_behavior(time::MissedTickBehavior::Skip); + let mut status_ticker = time::interval(Duration::from_secs(2)); + status_ticker.set_missed_tick_behavior(time::MissedTickBehavior::Skip); loop { tokio::select! { @@ -444,6 +454,9 @@ where _ = ntime_ticker.tick(), if self.current_task.is_some() => { self.roll_ntime().await; } + _ = status_ticker.tick() => { + self.status.write().hashrate = self.hashrate_estimator.hashrate(); + } } } @@ -499,16 +512,22 @@ where &mut self, new_task: HashTask, ) -> Result, HashThreadError> { + let initialization_lock = self.peripherals.initialization_lock.clone(); + // Initialize chips if not already running if matches!(self.chip_state, ChipState::Disabled) { - self.initialize_chips().await?; + let _init_guard = initialization_lock.lock().await; + if matches!(self.chip_state, ChipState::Disabled) { + self.initialize_chips().await?; + } } // Send job to chips self.send_job(&new_task).await?; let old = self.current_task.replace(new_task); - self.status.write().is_active = true; + let mut status = self.status.write(); + status.is_active = true; Ok(old) } @@ -546,7 +565,9 @@ where self.disable_chips().await; let old = self.current_task.take(); - self.status.write().is_active = false; + let mut status = self.status.write(); + status.is_active = false; + status.hashrate = HashRate::from(0); Ok(old) } @@ -583,7 +604,10 @@ where drained += 1; } if drained > 0 { - debug!(count = drained, "Drained stale responses before enumeration"); + debug!( + count = drained, + "Drained stale responses before enumeration" + ); } // 2. Execute enumeration sequence (assigns addresses) @@ -775,13 +799,13 @@ where if responding < chip_count { warn!( expected = chip_count, - responding, - "Chips lost during frequency ramp" + responding, "Chips lost during frequency ramp" ); } if has_regulator { - let final_v = voltage_for_frequency_stacked(steps.last().unwrap().0, domain_count, max_v); + let final_v = + voltage_for_frequency_stacked(steps.last().unwrap().0, domain_count, max_v); info!( target_mhz = target.mhz(), voltage = format!("{:.2}V", final_v), @@ -892,6 +916,13 @@ where midstate_num, subcore_id, }) => { + self.hashrate_estimator.record(self.nonce_work); + { + let mut status = self.status.write(); + status.hashrate = self.hashrate_estimator.hashrate(); + status.chip_shares_found = status.chip_shares_found.saturating_add(1); + } + // HACK: BM1362 job_id fix - protocol.rs extracts job_id from bits 7-4, // but BM1362 returns it in bits 6-3. Reconstruct result_header and re-extract. // TODO: Move this to protocol.rs with chip-type-aware parsing. @@ -1017,8 +1048,24 @@ where } } +fn nonce_work_for_chain(chip_count: usize) -> Work { + let hashrate_gh = 83.0 * chip_count as f64; + let reporting_interval = protocol::ReportingInterval::from_rate( + protocol::Hashrate::gibihashes_per_sec(hashrate_gh), + protocol::ReportingRate::nonces_per_sec(1.0), + ); + work_from_hash_count(1u64 << reporting_interval.exponent()) +} + +fn work_from_hash_count(hash_count: u64) -> Work { + let mut bytes = [0u8; 32]; + bytes[..8].copy_from_slice(&hash_count.to_le_bytes()); + Work::from_le_bytes(bytes) +} + #[cfg(test)] mod tests { + use std::collections::VecDeque; use std::io; use std::pin::Pin; use std::sync::Arc; @@ -1134,6 +1181,71 @@ mod tests { } } + /// Sink that captures commands and injects scripted read responses on demand. + struct ScriptedResponseSink { + commands: Arc>>, + response_tx: mpsc::Sender>, + scripted_reads: VecDeque>>, + } + + impl ScriptedResponseSink { + fn new( + response_tx: mpsc::Sender>, + scripted_reads: Vec>>, + ) -> (Self, Arc>>) { + let commands = Arc::new(std::sync::Mutex::new(Vec::new())); + ( + Self { + commands: Arc::clone(&commands), + response_tx, + scripted_reads: scripted_reads.into(), + }, + commands, + ) + } + } + + impl Sink for ScriptedResponseSink { + type Error = io::Error; + + fn poll_ready(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(())) + } + + fn start_send(mut self: Pin<&mut Self>, item: Command) -> io::Result<()> { + self.commands.lock().unwrap().push(item.clone()); + + if matches!( + item, + Command::ReadRegister { + register_address: RegisterAddress::ChipId, + .. + } + ) { + if let Some(responses) = self.scripted_reads.pop_front() { + for response in responses { + self.response_tx.try_send(response).map_err(|err| { + io::Error::new( + io::ErrorKind::BrokenPipe, + format!("failed to script response: {err}"), + ) + })?; + } + } + } + + Ok(()) + } + + fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(())) + } + + fn poll_close(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(())) + } + } + /// Build a ChipId response for testing. fn chip_id_response(chip_type: ChipType, address: u8) -> Result { Ok(protocol::Response::ReadRegister { @@ -1155,6 +1267,7 @@ mod tests { peripherals: ChainPeripherals { asic_enable: Arc::new(Mutex::new(asic_enable)), voltage_regulator: None, + initialization_lock: Arc::new(Mutex::new(())), }, } } @@ -1176,6 +1289,7 @@ mod tests { let (_cmd_tx, cmd_rx) = mpsc::channel(10); let (evt_tx, _evt_rx) = mpsc::channel(100); let status = Arc::new(RwLock::new(HashThreadStatus::default())); + let nonce_work = nonce_work_for_chain(chain.chip_count()); // Pre-load responses into a channel let (response_tx, response_rx) = mpsc::channel(128); @@ -1192,15 +1306,66 @@ mod tests { peripherals: ChainPeripherals { asic_enable: Arc::new(Mutex::new(asic_enable)), voltage_regulator: None, + initialization_lock: Arc::new(Mutex::new(())), }, chain, sequencer, chip_state: ChipState::Disabled, current_task: None, chip_jobs: ChipJobs::new(), + hashrate_estimator: HashrateEstimator::new(Duration::from_secs(60)), + nonce_work, } } + /// Create a test actor whose sink injects scripted ChipId responses when verification reads occur. + fn test_actor_with_scripted_reads( + responses: Vec>, + chain: Chain, + sequencer: Sequencer, + asic_enable: MockAsicEnable, + ) -> (BM13xxActor, Arc>>) { + let scripted_reads = responses.into_iter().map(|response| vec![response]).collect(); + test_actor_with_scripted_read_batches(scripted_reads, chain, sequencer, asic_enable) + } + + fn test_actor_with_scripted_read_batches( + scripted_reads: Vec>>, + chain: Chain, + sequencer: Sequencer, + asic_enable: MockAsicEnable, + ) -> (BM13xxActor, Arc>>) { + let (_cmd_tx, cmd_rx) = mpsc::channel(10); + let (evt_tx, _evt_rx) = mpsc::channel(100); + let status = Arc::new(RwLock::new(HashThreadStatus::default())); + let nonce_work = nonce_work_for_chain(chain.chip_count()); + let (response_tx, response_rx) = mpsc::channel(128); + let (chip_tx, commands) = ScriptedResponseSink::new(response_tx, scripted_reads); + + ( + BM13xxActor { + cmd_rx, + evt_tx, + status, + response_rx, + chip_tx, + peripherals: ChainPeripherals { + asic_enable: Arc::new(Mutex::new(asic_enable)), + voltage_regulator: None, + initialization_lock: Arc::new(Mutex::new(())), + }, + chain, + sequencer, + chip_state: ChipState::Disabled, + current_task: None, + chip_jobs: ChipJobs::new(), + hashrate_estimator: HashrateEstimator::new(Duration::from_secs(60)), + nonce_work, + }, + commands, + ) + } + /// Create chain and sequencer for a given chip count. fn chain_and_sequencer(chip_count: usize) -> (Chain, Sequencer) { let topology = TopologySpec::single_domain(chip_count); @@ -1256,10 +1421,10 @@ mod tests { async fn initialize_single_chip_succeeds() { // Topology expects 1 chip, hardware provides 1 chip response let responses = vec![chip_id_response(ChipType::BM1362, 0x00)]; - let chip_tx = futures::sink::drain(); let (chain, sequencer) = chain_and_sequencer(1); - let mut actor = test_actor(responses, chip_tx, chain, sequencer, MockAsicEnable::new()); + let (mut actor, _commands) = + test_actor_with_scripted_reads(responses, chain, sequencer, MockAsicEnable::new()); let result = actor.initialize_chips().await; @@ -1273,10 +1438,10 @@ mod tests { let responses: Vec<_> = (0..12) .map(|i| chip_id_response(ChipType::BM1362, i * 2)) // interval 2 .collect(); - let chip_tx = futures::sink::drain(); let (chain, sequencer) = chain_and_sequencer(12); - let mut actor = test_actor(responses, chip_tx, chain, sequencer, MockAsicEnable::new()); + let (mut actor, _commands) = + test_actor_with_scripted_reads(responses, chain, sequencer, MockAsicEnable::new()); let result = actor.initialize_chips().await; @@ -1289,13 +1454,20 @@ mod tests { // Topology expects 12 chips, respond with minimum viable count (at threshold) let expected = 12; let responding = min_viable_chip_count(expected); - let responses: Vec<_> = (0..responding) - .map(|i| chip_id_response(ChipType::BM1362, (i * 2) as u8)) - .collect(); - let chip_tx = futures::sink::drain(); - let (chain, sequencer) = chain_and_sequencer(expected); - let mut actor = test_actor(responses, chip_tx, chain, sequencer, MockAsicEnable::new()); + let scripted_reads = (0..3) + .flat_map(|_| { + (0..responding) + .map(|i| vec![chip_id_response(ChipType::BM1362, (i * 2) as u8)]) + .collect::>() + }) + .collect(); + let (mut actor, _commands) = test_actor_with_scripted_read_batches( + scripted_reads, + chain, + sequencer, + MockAsicEnable::new(), + ); // Should succeed with warning, not fail let result = actor.initialize_chips().await; @@ -1311,10 +1483,10 @@ mod tests { let responses: Vec<_> = (0..responding) .map(|i| chip_id_response(ChipType::BM1362, (i * 2) as u8)) .collect(); - let chip_tx = futures::sink::drain(); let (chain, sequencer) = chain_and_sequencer(expected); - let mut actor = test_actor(responses, chip_tx, chain, sequencer, MockAsicEnable::new()); + let (mut actor, _commands) = + test_actor_with_scripted_reads(responses, chain, sequencer, MockAsicEnable::new()); let result = actor.initialize_chips().await; @@ -1446,10 +1618,10 @@ mod tests { let responses: Vec<_> = (0..5) .map(|i| chip_id_response(ChipType::BM1362, i * 2)) .collect(); - let (sink, _commands) = CapturingSink::new(); let (chain, sequencer) = chain_and_sequencer(5); - let mut actor = test_actor(responses, sink, chain, sequencer, MockAsicEnable::new()); + let (mut actor, _commands) = + test_actor_with_scripted_reads(responses, chain, sequencer, MockAsicEnable::new()); let count = actor.verify_chain().await; @@ -1458,19 +1630,23 @@ mod tests { #[tokio::test(start_paused = true)] async fn verify_chain_ignores_other_responses() { - let responses: Vec> = vec![ - chip_id_response(ChipType::BM1362, 0x00), - // Non-ChipId response should be ignored - Ok(protocol::Response::ReadRegister { - chip_address: 0x02, - register: Register::VersionMask(protocol::VersionMask::full_rolling()), - }), - chip_id_response(ChipType::BM1362, 0x04), - ]; - let (sink, _commands) = CapturingSink::new(); - let (chain, sequencer) = chain_and_sequencer(2); - let mut actor = test_actor(responses, sink, chain, sequencer, MockAsicEnable::new()); + let scripted_reads = vec![ + vec![chip_id_response(ChipType::BM1362, 0x00)], + vec![ + Ok(protocol::Response::ReadRegister { + chip_address: 0x02, + register: Register::VersionMask(protocol::VersionMask::full_rolling()), + }), + chip_id_response(ChipType::BM1362, 0x04), + ], + ]; + let (mut actor, _commands) = test_actor_with_scripted_read_batches( + scripted_reads, + chain, + sequencer, + MockAsicEnable::new(), + ); let count = actor.verify_chain().await; @@ -1479,15 +1655,20 @@ mod tests { #[tokio::test(start_paused = true)] async fn verify_chain_handles_stream_errors() { - let responses: Vec> = vec![ - chip_id_response(ChipType::BM1362, 0x00), - Err(io::Error::new(io::ErrorKind::Other, "glitch")), - chip_id_response(ChipType::BM1362, 0x02), - ]; - let (sink, _commands) = CapturingSink::new(); - let (chain, sequencer) = chain_and_sequencer(2); - let mut actor = test_actor(responses, sink, chain, sequencer, MockAsicEnable::new()); + let scripted_reads = vec![ + vec![chip_id_response(ChipType::BM1362, 0x00)], + vec![ + Err(io::Error::new(io::ErrorKind::Other, "glitch")), + chip_id_response(ChipType::BM1362, 0x02), + ], + ]; + let (mut actor, _commands) = test_actor_with_scripted_read_batches( + scripted_reads, + chain, + sequencer, + MockAsicEnable::new(), + ); let count = actor.verify_chain().await; diff --git a/mujina-miner/src/board/bitaxe.rs b/mujina-miner/src/board/bitaxe.rs index 711d516c..4dcdad29 100644 --- a/mujina-miner/src/board/bitaxe.rs +++ b/mujina-miner/src/board/bitaxe.rs @@ -786,12 +786,16 @@ impl BitaxeBoard { name: board_name.clone(), model: board_model.clone(), serial: board_serial.clone(), + frequency_mhz: None, + hashboard_count: Some(1), + active_hashboard_count: Some(1), fans: vec![Fan { name: "fan".into(), rpm: fan_rpm, percent: fan_percent, target_percent: None, }], + hashboards: vec![], temperatures: vec![ TemperatureSensor { name: "asic".into(), diff --git a/mujina-miner/src/board/emberone.rs b/mujina-miner/src/board/emberone.rs index 11d3bd04..12d8ad9b 100644 --- a/mujina-miner/src/board/emberone.rs +++ b/mujina-miner/src/board/emberone.rs @@ -383,6 +383,7 @@ impl Board for EmberOne { io_power_enable_pin, })), voltage_regulator, + initialization_lock: Arc::new(Mutex::new(())), }, }; diff --git a/mujina-miner/src/board/s19j_pro.rs b/mujina-miner/src/board/s19j_pro.rs index d0870df4..b778b62f 100644 --- a/mujina-miner/src/board/s19j_pro.rs +++ b/mujina-miner/src/board/s19j_pro.rs @@ -1,13 +1,15 @@ //! S19j Pro hashboard support via Bitcrane v3. //! -//! The S19j Pro is a hashboard with 126 BM1362 ASIC chips, communicating via -//! USB using the bitcrane protocol. Power is provided by an APW12 PSU controlled -//! via bit-banged I2C. +//! A Bitcrane v3 bridge exposes one management channel plus three independent +//! ASIC UART channels for three Antminer S19j Pro hashboards. Mujina models +//! that as a single board with shared power and cooling, and one BM13xx thread +//! per populated hashboard channel. use std::sync::Arc; use std::time::Duration; use async_trait::async_trait; +use parking_lot::RwLock; use tokio::sync::{Mutex, watch}; use tokio_serial::SerialPortBuilderExt; use tokio_util::codec::{FramedRead, FramedWrite}; @@ -17,7 +19,10 @@ use super::{ pattern::{BoardPattern, Match, StringMatch}, }; use crate::{ - api_client::types::{BoardState, Fan, MinerState, TemperatureSensor}, + api_client::types::{ + BoardState, Fan, HashboardState, MinerState, PowerMeasurement, TemperatureSensor, + ThreadState, + }, asic::{ bm13xx::{ self, @@ -25,12 +30,12 @@ use crate::{ chip_config, thread_v2, topology::TopologySpec, }, - hash_thread::{AsicEnable, HashThread}, + hash_thread::{AsicEnable, HashThread, HashThreadStatus}, }, error::Error, hw_trait::gpio::{GpioPin, PinValue}, mgmt_protocol::{ - ControlChannel, Apw12Psu, + Apw12Psu, ControlChannel, bitcrane::{ display::BitcraneDisplay, fan::{self, BitcraneFan}, @@ -40,21 +45,34 @@ use crate::{ }, peripheral::tmp75::{self, Tmp75}, tracing::prelude::*, - transport::{ - UsbDeviceInfo, - serial::{SerialControl, SerialStream}, - }, + transport::{UsbDeviceInfo, serial::SerialStream}, }; -// Register this board type with the inventory system +const S19J_PRO_TARGET_FREQ_MHZ: f32 = 500.0; +const APW12_POWER_RAIL_NAME: &str = "APW12"; +const HASHBOARD_COUNT: usize = 3; + +const TELEMETRY_INTERVAL: Duration = Duration::from_secs(2); +const DISPLAY_UPDATE_INTERVAL: u32 = 5; + +const FAN_MIN_PERCENT: u8 = 45; +const FAN_FAILSAFE_PERCENT: u8 = 80; +const FAN_MAX_PERCENT: u8 = 100; +const FAN_TARGET_TEMP_C: f32 = 62.0; +const FAN_FULL_SPEED_TEMP_C: f32 = 68.0; +const FAN_MAX_TEMP_C: f32 = 70.0; +const FAN_PID_KP: f32 = 6.0; +const FAN_PID_KI: f32 = 0.35; +const FAN_PID_KD: f32 = 3.0; + inventory::submit! { BoardDescriptor { pattern: BoardPattern { vid: Match::Any, pid: Match::Any, bcd_device: Match::Any, - manufacturer: Match::Specific(StringMatch::Exact("256F")), - product: Match::Specific(StringMatch::Exact("bitcrane_S19jpro")), + manufacturer: Match::Specific(StringMatch::Regex("^(256F|OSMU)$")), + product: Match::Specific(StringMatch::Regex("^(bitcrane_S19jpro|bitcrane3)$")), serial_pattern: Match::Any, }, name: "S19j Pro", @@ -62,97 +80,127 @@ inventory::submit! { } } -/// S19j Pro hashboard. -pub struct S19jPro { - device_info: UsbDeviceInfo, +#[derive(Clone)] +struct S19jProHashboard { + index: u8, data_port_path: String, - /// Control channel for board management (bitcrane protocol). - control_channel: ControlChannel, - /// Control handle for data channel (for baud rate changes). - data_control: Option, - /// ASIC reset pin (RST0, active-low). reset_pin: Option, - /// APW12 PSU controller. - psu: Option>>, - /// TMP75 temperature sensors (2 per hashboard). + plug_pin: Option, temp_sensors: Option<(Tmp75, Tmp75)>, + is_present: bool, + is_active: bool, +} - /// Channel for publishing board state to the API server. +impl S19jProHashboard { + fn state(&self, hashrate: u64) -> HashboardState { + HashboardState { + index: self.index, + serial_port: Some(self.data_port_path.clone()), + is_present: self.is_present, + is_active: self.is_active, + hashrate, + } + } +} + +#[derive(Clone)] +struct S19jProThreadMonitor { + index: u8, + name: String, + status: Arc>, +} + +/// S19j Pro board with shared PSU/fans and up to three hashboard channels. +pub struct S19jPro { + device_info: UsbDeviceInfo, + hashboards: Vec, + thread_monitors: Vec, + control_channel: ControlChannel, + psu: Option>>, state_tx: watch::Sender, - /// Receiver for miner state (hashrate for OLED display). miner_state_rx: Option>, } impl S19jPro { - /// Create a new S19j Pro board instance. pub fn new( device_info: UsbDeviceInfo, control_channel: ControlChannel, - data_port_path: String, + data_port_paths: Vec, state_tx: watch::Sender, ) -> Self { + let hashboards = data_port_paths + .into_iter() + .enumerate() + .map(|(index, data_port_path)| S19jProHashboard { + index: index as u8, + data_port_path, + reset_pin: None, + plug_pin: None, + temp_sensors: None, + is_present: false, + is_active: false, + }) + .collect(); + Self { device_info, - data_port_path, + hashboards, + thread_monitors: Vec::new(), control_channel, - data_control: None, - reset_pin: None, psu: None, - temp_sensors: None, state_tx, miner_state_rx: None, } } - /// Initialize the board hardware. - /// - /// Sets up GPIO pins, initializes APW12 PSU, and holds ASICs in reset until mining starts. pub async fn initialize(&mut self) -> Result<(), BoardError> { - // Get GPIO controller using bitcrane protocol let gpio = BitcraneGpioController::new(self.control_channel.clone()); + let i2c = BitcraneI2c::new(self.control_channel.clone()); - // Get RST0 pin for first hashboard - let mut reset_pin = gpio.pin(BitcraneGpioPin::Rst0); - - // Initialize to safe state: chips in reset (RST0 low = reset asserted) - debug!("Initializing S19j Pro: chips in reset"); - reset_pin.write(PinValue::Low).await.map_err(|e| { - BoardError::InitializationFailed(format!("Failed to assert reset: {}", e)) - })?; - - // Store GPIO pin for later use - self.reset_pin = Some(reset_pin); + for hashboard in &mut self.hashboards { + let mut reset_pin = gpio.pin(reset_pin_for_index(hashboard.index)); + reset_pin.write(PinValue::Low).await.map_err(|e| { + BoardError::InitializationFailed(format!( + "Failed to assert reset for HB{}: {}", + hashboard.index, e + )) + })?; + + let mut plug_pin = gpio.pin(plug_pin_for_index(hashboard.index)); + hashboard.is_present = match plug_pin.read().await { + Ok(PinValue::High) => true, + Ok(PinValue::Low) => false, + Err(e) => { + warn!(hashboard = hashboard.index, error = %e, "Plug detect failed; assuming present"); + true + } + }; + + hashboard.reset_pin = Some(reset_pin); + hashboard.plug_pin = Some(plug_pin); + hashboard.temp_sensors = + Some(tmp75::sensors_for_hashboard(i2c.clone(), hashboard.index)); + } - // Initialize APW12 PSU debug!("Initializing APW12 PSU"); let mut psu = Apw12Psu::new(self.control_channel.clone()); - - // Enable PSU psu.set_enabled(true).await.map_err(|e| { BoardError::InitializationFailed(format!("Failed to enable PSU: {}", e)) })?; + tokio::time::sleep(Duration::from_millis(500)).await; - // Wait for PSU to power up - tokio::time::sleep(tokio::time::Duration::from_millis(500)).await; - - // Disable watchdog (0x00 = disabled) psu.config_watchdog(0x00).await.map_err(|e| { BoardError::InitializationFailed(format!("Failed to configure PSU watchdog: {}", e)) })?; - // Set initial voltage for BM1362 chain const DEFAULT_VOUT: f32 = 12.6; psu.set_voltage(DEFAULT_VOUT).await.map_err(|e| { BoardError::InitializationFailed(format!("Failed to set PSU voltage: {}", e)) })?; info!("APW12 PSU enabled, voltage set to {}V", DEFAULT_VOUT); + tokio::time::sleep(Duration::from_millis(2000)).await; - // Wait for voltage to stabilize before chip enumeration - // Longer delay needed for 126-chip chain to fully power up - tokio::time::sleep(tokio::time::Duration::from_millis(2000)).await; - - // Verify voltage match psu.measure_voltage().await { Ok(v) => info!("APW12 measured voltage: {:.2}V", v), Err(e) => warn!("Failed to measure PSU voltage: {}", e), @@ -160,24 +208,58 @@ impl S19jPro { self.psu = Some(Arc::new(Mutex::new(psu))); - // Initialize TMP75 temperature sensors for hashboard 0 - let i2c = BitcraneI2c::new(self.control_channel.clone()); - let (temp0, temp1) = tmp75::sensors_for_hashboard(i2c, 0); - self.temp_sensors = Some((temp0.clone(), temp1.clone())); - - // Initialize fans and set to 50% speed let fans = fan::all_fans(self.control_channel.clone()); - const DEFAULT_FAN_SPEED: u8 = 50; for fan in &fans { - if let Err(e) = fan.set_speed(DEFAULT_FAN_SPEED).await { - warn!(fan = %fan.name(), error = %e, "Failed to set fan speed"); + if let Err(e) = fan.set_speed(FAN_FAILSAFE_PERCENT).await { + warn!(fan = %fan.name(), error = %e, "Failed to set initial fan speed"); } } - info!("Fans initialized at {}% speed", DEFAULT_FAN_SPEED); + info!("Fans initialized at {}% speed", FAN_FAILSAFE_PERCENT); + self.publish_hashboard_state(); info!("S19j Pro initialized successfully"); Ok(()) } + + fn publish_hashboard_state(&self) { + let hashboards: Vec = self.hashboard_states(); + let active_hashboard_count = hashboards + .iter() + .filter(|hashboard| hashboard.is_active) + .count() as u8; + + self.state_tx.send_modify(|state| { + state.frequency_mhz = Some(S19J_PRO_TARGET_FREQ_MHZ); + state.hashboard_count = Some(hashboard_count_u8(hashboards.len())); + state.active_hashboard_count = Some(active_hashboard_count); + state.hashboards = hashboards.clone(); + }); + } + + fn hashboard_states(&self) -> Vec { + let hashboard_hashrates = self.hashboard_hashrates(); + self.hashboards + .iter() + .map(|hashboard| { + let hashrate = hashboard_hashrates + .iter() + .find(|(index, _)| *index == hashboard.index) + .map(|(_, hashrate)| *hashrate) + .unwrap_or(0); + hashboard.state(hashrate) + }) + .collect() + } + + fn hashboard_hashrates(&self) -> Vec<(u8, u64)> { + self.thread_monitors + .iter() + .map(|monitor| { + let status = monitor.status.read(); + (monitor.index, effective_hashrate(&status)) + }) + .collect() + } } #[async_trait] @@ -191,98 +273,157 @@ impl Board for S19jPro { } async fn shutdown(&mut self) -> Result<(), BoardError> { - // Set all fans to 0% - for (i, fan) in fan::all_fans(self.control_channel.clone()).into_iter().enumerate() { + for fan in fan::all_fans(self.control_channel.clone()) { if let Err(e) = fan.set_speed(0).await { - warn!(fan = i, "Failed to set fan to 0% on shutdown: {}", e); + warn!(fan = %fan.name(), error = %e, "Failed to set fan to 0% on shutdown"); } } - info!("Fans set to 0%"); - // Assert reset to stop ASICs - if let Some(ref mut reset_pin) = self.reset_pin { - if let Err(e) = reset_pin.write(PinValue::Low).await { - warn!("Failed to assert reset on shutdown: {}", e); + for hashboard in &mut self.hashboards { + hashboard.is_active = false; + if let Some(reset_pin) = &mut hashboard.reset_pin { + if let Err(e) = reset_pin.write(PinValue::Low).await { + warn!(hashboard = hashboard.index, error = %e, "Failed to assert reset on shutdown"); + } } } - // Disable PSU - if let Some(ref psu) = self.psu { + if let Some(psu) = &self.psu { if let Err(e) = psu.lock().await.set_enabled(false).await { warn!("Failed to disable PSU on shutdown: {}", e); } } + self.publish_hashboard_state(); info!("S19j Pro shutdown complete"); Ok(()) } async fn create_hash_threads(&mut self) -> Result>, BoardError> { - // Take GPIO pin from initialization - let reset_pin = self.reset_pin.take().ok_or_else(|| { - BoardError::InitializationFailed("Reset pin not initialized".to_string()) - })?; - - // Open data port - let data_stream = SerialStream::new(&self.data_port_path, 115200).map_err(|e| { - BoardError::InitializationFailed(format!("Failed to open data port: {}", e)) - })?; - let (data_reader, data_writer, data_control) = data_stream.split(); - - // Flush any stale data in the serial buffer before enumeration - data_control.flush_input().map_err(|e| { - BoardError::InitializationFailed(format!("Failed to flush serial buffer: {}", e)) - })?; + let mut threads: Vec> = Vec::new(); + let initialization_lock = Arc::new(Mutex::new(())); + + for hashboard in &mut self.hashboards { + if !hashboard.is_present { + info!( + hashboard = hashboard.index, + "Skipping unpopulated hashboard channel" + ); + continue; + } - self.data_control = Some(data_control); + let Some(reset_pin) = hashboard.reset_pin.clone() else { + return Err(BoardError::InitializationFailed(format!( + "Reset pin not initialized for HB{}", + hashboard.index + ))); + }; + + let data_stream = match SerialStream::new(&hashboard.data_port_path, 115200) { + Ok(stream) => stream, + Err(e) => { + warn!( + hashboard = hashboard.index, + port = %hashboard.data_port_path, + error = %e, + "Failed to open data port" + ); + continue; + } + }; + let (data_reader, data_writer, data_control) = data_stream.split(); + + if let Err(e) = data_control.flush_input() { + warn!( + hashboard = hashboard.index, + port = %hashboard.data_port_path, + error = %e, + "Failed to flush serial buffer" + ); + continue; + } - // Create framed reader/writer for BM13xx protocol - let chip_rx = FramedRead::new(data_reader, bm13xx::FrameCodec); - let chip_tx = FramedWrite::new(data_writer, bm13xx::FrameCodec); + let chip_rx = FramedRead::new(data_reader, bm13xx::FrameCodec); + let chip_tx = FramedWrite::new(data_writer, bm13xx::FrameCodec); - // Build thread name from board model and serial - let thread_name = match &self.device_info.serial_number { - Some(serial) => format!("S19jPro-{}", &serial[..8.min(serial.len())]), - None => "S19jPro".to_string(), - }; + let serial_prefix = self + .device_info + .serial_number + .as_deref() + .unwrap_or("unknown"); + let thread_name = format!("S19jPro-{}-HB{}", serial_prefix, hashboard.index); - // Build chain configuration for S19j Pro: 42 series domains × 3 chips = 126 BM1362 chips - // Use APW12 PSU for voltage regulation - let voltage_regulator: Option>> = - self.psu.as_ref().map(|psu| { - Arc::clone(psu) as Arc> - }); + let voltage_regulator: Option>> = self + .psu + .as_ref() + .map(|psu| Arc::clone(psu) as Arc>); + + let config = ChainConfig { + name: thread_name, + topology: TopologySpec::uniform_domains(42, 3, false), + chip_config: chip_config::bm1362(), + peripherals: ChainPeripherals { + asic_enable: Arc::new(Mutex::new(S19jProAsicEnable { reset_pin })), + voltage_regulator, + initialization_lock: Arc::clone(&initialization_lock), + }, + }; + + match thread_v2::BM13xxThread::new(chip_rx, chip_tx, config) { + Ok(thread) => { + self.thread_monitors.push(S19jProThreadMonitor { + index: hashboard.index, + name: thread.name().to_string(), + status: thread.status_handle(), + }); + hashboard.is_active = true; + info!( + hashboard = hashboard.index, + port = %hashboard.data_port_path, + "Started S19j Pro hashboard thread" + ); + threads.push(Box::new(thread)); + } + Err(e) => { + warn!( + hashboard = hashboard.index, + port = %hashboard.data_port_path, + error = %e, + "Failed to create S19j Pro hashboard thread" + ); + } + } + } - let config = ChainConfig { - name: thread_name, - // S19j Pro: 42 series domains, 3 chips per domain (126 total) - topology: TopologySpec::uniform_domains(42, 3, false), - chip_config: chip_config::bm1362(), - peripherals: ChainPeripherals { - asic_enable: Arc::new(Mutex::new(S19jProAsicEnable { reset_pin })), - voltage_regulator, - }, - }; + if threads.is_empty() { + return Err(BoardError::InitializationFailed( + "No populated S19j Pro hashboards could be started".to_string(), + )); + } - // Create the hash thread - let thread = thread_v2::BM13xxThread::new(chip_rx, chip_tx, config).map_err(|e| { - BoardError::InitializationFailed(format!("Failed to create hash thread: {}", e)) - })?; + self.publish_hashboard_state(); - // Spawn telemetry task now that miner_state_rx is available - // (set_miner_state_rx is called by backplane before create_hash_threads) - let (temp0, temp1) = self.temp_sensors.take().ok_or_else(|| { - BoardError::InitializationFailed("Temperature sensors not initialized".to_string()) - })?; + let telemetry_hardware = self.hashboards.clone(); + let thread_monitors = self.thread_monitors.clone(); + let psu = self.psu.as_ref().map(Arc::clone); let fans = fan::all_fans(self.control_channel.clone()); let display = BitcraneDisplay::new(self.control_channel.clone()); let state_tx = self.state_tx.clone(); let miner_state_rx = self.miner_state_rx.clone(); tokio::spawn(async move { - telemetry_task(temp0, temp1, fans, display, state_tx, miner_state_rx).await; + telemetry_task( + telemetry_hardware, + thread_monitors, + psu, + fans, + display, + state_tx, + miner_state_rx, + ) + .await; }); - Ok(vec![Box::new(thread)]) + Ok(threads) } fn set_miner_state_rx(&mut self, rx: watch::Receiver) { @@ -290,45 +431,117 @@ impl Board for S19jPro { } } -/// Telemetry task that periodically reads temperature sensors and fan RPMs. -/// -/// Runs indefinitely, updating state_tx with readings every 2 seconds. -/// Also updates the OLED display with hashrate every 10 seconds. async fn telemetry_task( - temp0: Tmp75, - temp1: Tmp75, + mut hashboards: Vec, + thread_monitors: Vec, + psu: Option>>, fans: [BitcraneFan; 4], display: BitcraneDisplay, state_tx: watch::Sender, miner_state_rx: Option>, ) { - const TELEMETRY_INTERVAL: Duration = Duration::from_secs(2); - const DISPLAY_UPDATE_INTERVAL: u32 = 5; // Update display every N telemetry cycles - let mut cycle_count: u32 = 0; + let mut pid = FanPidController::default(); + let mut fan_target_percent = FAN_FAILSAFE_PERCENT; loop { - // Read both temperature sensors - let temp0_result = temp0.read_temperature().await; - let temp1_result = temp1.read_temperature().await; - - // Build temperature sensor readings - let temperatures = vec![ - TemperatureSensor { - name: temp0.name().to_string(), - temperature_c: temp0_result + let mut temperatures: Vec = Vec::with_capacity(hashboards.len() * 2); + let mut hashboard_states: Vec = Vec::with_capacity(hashboards.len()); + let mut thread_states: Vec = Vec::with_capacity(thread_monitors.len()); + let mut max_temp_c: Option = None; + let hashrates_by_index: Vec<(u8, bool, u64)> = thread_monitors + .iter() + .map(|monitor| { + let status = monitor.status.read(); + let hashrate = effective_hashrate(&status); + (monitor.index, status.is_active, hashrate) + }) + .collect(); + + for monitor in &thread_monitors { + let status = monitor.status.read(); + let hashrate = effective_hashrate(&status); + thread_states.push(ThreadState { + name: monitor.name.clone(), + hashrate, + is_active: status.is_active, + }); + } + + for hashboard in &mut hashboards { + if let Some(plug_pin) = &mut hashboard.plug_pin { + hashboard.is_present = match plug_pin.read().await { + Ok(PinValue::High) => true, + Ok(PinValue::Low) => false, + Err(e) => { + debug!(hashboard = hashboard.index, error = %e, "Plug detect read failed"); + hashboard.is_present + } + }; + } + + if let Some((temp0, temp1)) = &hashboard.temp_sensors { + let temp0_value = temp0 + .read_temperature() + .await .inspect_err(|e| debug!(sensor = %temp0.name(), error = %e, "Temp read failed")) - .ok(), - }, - TemperatureSensor { - name: temp1.name().to_string(), - temperature_c: temp1_result + .ok(); + let temp1_value = temp1 + .read_temperature() + .await .inspect_err(|e| debug!(sensor = %temp1.name(), error = %e, "Temp read failed")) - .ok(), - }, - ]; + .ok(); + + if let Some(temp) = temp0_value { + max_temp_c = Some(max_temp_c.map_or(temp, |max| max.max(temp))); + } + if let Some(temp) = temp1_value { + max_temp_c = Some(max_temp_c.map_or(temp, |max| max.max(temp))); + } + + temperatures.push(TemperatureSensor { + name: temp0.name().to_string(), + temperature_c: temp0_value, + }); + temperatures.push(TemperatureSensor { + name: temp1.name().to_string(), + temperature_c: temp1_value, + }); + } + + let (is_active, hashrate) = hashrates_by_index + .iter() + .find(|(index, _, _)| *index == hashboard.index) + .map(|(_, is_active, hashrate)| (*is_active, *hashrate)) + .unwrap_or((false, 0)); + hashboard.is_active = is_active; + hashboard_states.push(hashboard.state(hashrate)); + } + + let next_target_percent = pid.next_target_percent(max_temp_c, TELEMETRY_INTERVAL); + if next_target_percent != fan_target_percent { + for fan in &fans { + if let Err(e) = fan.set_speed(next_target_percent).await { + warn!(fan = %fan.name(), error = %e, "Failed to set fan speed"); + } + } + debug!( + target_percent = next_target_percent, + max_temp_c, "Updated S19j Pro fan target" + ); + fan_target_percent = next_target_percent; + } + + if let Some(max_temp_c) = max_temp_c { + if max_temp_c >= FAN_MAX_TEMP_C { + warn!( + max_temp_c, + target_percent = fan_target_percent, + "S19j Pro temperature reached the 70C ceiling" + ); + } + } - // Read all fan RPMs let mut fan_states = Vec::with_capacity(4); for fan in &fans { let rpm_result = fan.read_rpm().await; @@ -337,22 +550,50 @@ async fn telemetry_task( rpm: rpm_result .inspect_err(|e| debug!(fan = %fan.name(), error = %e, "Fan RPM read failed")) .ok(), - percent: None, // We don't track current duty cycle yet - target_percent: Some(50), // We set 50% at init + percent: None, + target_percent: Some(fan_target_percent), }); } - // Update board state + let voltage_v = if let Some(psu) = &psu { + let mut psu = psu.lock().await; + psu.measure_voltage() + .await + .inspect_err( + |e| debug!(rail = APW12_POWER_RAIL_NAME, error = %e, "Voltage read failed"), + ) + .ok() + } else { + None + }; + + let powers = vec![PowerMeasurement { + name: APW12_POWER_RAIL_NAME.to_string(), + voltage_v, + current_a: None, + power_w: None, + }]; + + let active_hashboard_count = hashboard_states + .iter() + .filter(|hashboard| hashboard.is_active) + .count() as u8; + state_tx.send_modify(|state| { - state.temperatures = temperatures; - state.fans = fan_states; + state.frequency_mhz = Some(S19J_PRO_TARGET_FREQ_MHZ); + state.hashboard_count = Some(hashboard_count_u8(hashboard_states.len())); + state.active_hashboard_count = Some(active_hashboard_count); + state.hashboards = hashboard_states.clone(); + state.temperatures = temperatures.clone(); + state.fans = fan_states.clone(); + state.powers = powers.clone(); + state.threads = thread_states.clone(); }); - // Update OLED display periodically with actual hashrate from scheduler if cycle_count % DISPLAY_UPDATE_INTERVAL == 0 { let hashrate_gh = miner_state_rx .as_ref() - .map(|rx| rx.borrow().hashrate as f64 / 1_000_000_000.0) // H/s to GH/s + .map(|rx| rx.borrow().hashrate as f64 / 1_000_000_000.0) .unwrap_or(0.0); if let Err(e) = display.display_hashrate(hashrate_gh).await { @@ -365,10 +606,46 @@ async fn telemetry_task( } } -/// Adapter implementing `AsicEnable` for S19j Pro's GPIO-based reset control. -/// -/// Controls RST0 pin via bitcrane protocol (active-low reset for ASIC chips). -/// Power is handled externally. +#[derive(Default)] +struct FanPidController { + integral: f32, + previous_error: Option, +} + +impl FanPidController { + fn next_target_percent(&mut self, max_temp_c: Option, dt: Duration) -> u8 { + let Some(max_temp_c) = max_temp_c else { + self.integral = 0.0; + self.previous_error = None; + return FAN_FAILSAFE_PERCENT; + }; + + if max_temp_c >= FAN_FULL_SPEED_TEMP_C { + self.integral = 0.0; + self.previous_error = Some(max_temp_c - FAN_TARGET_TEMP_C); + return FAN_MAX_PERCENT; + } + + let dt_secs = dt.as_secs_f32().max(1.0); + let error = max_temp_c - FAN_TARGET_TEMP_C; + self.integral = (self.integral + error * dt_secs).clamp(0.0, 40.0); + let derivative = self + .previous_error + .map(|previous_error| (error - previous_error) / dt_secs) + .unwrap_or(0.0); + self.previous_error = Some(error); + + let output = FAN_MIN_PERCENT as f32 + + FAN_PID_KP * error.max(0.0) + + FAN_PID_KI * self.integral + + FAN_PID_KD * derivative.max(0.0); + + output + .round() + .clamp(FAN_MIN_PERCENT as f32, FAN_MAX_PERCENT as f32) as u8 + } +} + struct S19jProAsicEnable { reset_pin: BitcraneGpioPinHandle, } @@ -376,21 +653,15 @@ struct S19jProAsicEnable { #[async_trait] impl AsicEnable for S19jProAsicEnable { async fn enable(&mut self) -> anyhow::Result<()> { - // Release reset (RST0 is active-low, so High = running) self.reset_pin .write(PinValue::High) .await .map_err(|e| anyhow::anyhow!("Failed to release reset: {}", e))?; - - // Wait for all 126 chips (42 series domains × 3 chips) to come out of - // reset and stabilize before enumeration begins - tokio::time::sleep(std::time::Duration::from_millis(2000)).await; - + tokio::time::sleep(Duration::from_millis(2000)).await; Ok(()) } async fn disable(&mut self) -> anyhow::Result<()> { - // Assert reset (RST0 low = reset asserted) self.reset_pin .write(PinValue::Low) .await @@ -398,16 +669,12 @@ impl AsicEnable for S19jProAsicEnable { } } -// Factory function to create S19j Pro board from USB device info async fn create_from_usb( device: UsbDeviceInfo, ) -> crate::error::Result<(Box, super::BoardRegistration)> { - // Get serial ports let serial_ports = device.serial_ports()?; - // S19j Pro uses 4 serial ports: control + 3 hashboard data channels - // For now, just use the first hashboard (ports 0=control, 1=data) - if serial_ports.len() != 4 { + if serial_ports.len() != HASHBOARD_COUNT + 1 { return Err(Error::Hardware(format!( "S19j Pro requires exactly 4 serial ports, found {}", serial_ports.len() @@ -415,34 +682,44 @@ async fn create_from_usb( } let control_port_path = serial_ports[0].clone(); - let data_port_path = serial_ports[1].clone(); // First hashboard + let data_port_paths = serial_ports[1..].to_vec(); debug!( serial = ?device.serial_number, control = %control_port_path, - data = %data_port_path, + data_ports = ?data_port_paths, "S19j Pro serial ports" ); - // Open control port let control_port = tokio_serial::new(&control_port_path, 115200) .open_native_async() .map_err(|e| Error::Hardware(format!("Failed to open control port: {}", e)))?; let control_channel = ControlChannel::new(control_port); - // Create watch channel for board state, seeded with identity let serial = device.serial_number.clone(); let initial_state = BoardState { name: format!("s19jpro-{}", serial.as_deref().unwrap_or("unknown")), model: "S19j Pro".into(), serial, + frequency_mhz: Some(S19J_PRO_TARGET_FREQ_MHZ), + hashboard_count: Some(HASHBOARD_COUNT as u8), + active_hashboard_count: Some(0), + hashboards: data_port_paths + .iter() + .enumerate() + .map(|(index, data_port_path)| HashboardState { + index: index as u8, + serial_port: Some(data_port_path.clone()), + is_present: false, + is_active: false, + hashrate: 0, + }) + .collect(), ..Default::default() }; let (state_tx, state_rx) = watch::channel(initial_state); - // Create and initialize board - let mut board = S19jPro::new(device, control_channel, data_port_path, state_tx); - + let mut board = S19jPro::new(device, control_channel, data_port_paths, state_tx); board .initialize() .await @@ -451,3 +728,29 @@ async fn create_from_usb( let registration = super::BoardRegistration { state_rx }; Ok((Box::new(board), registration)) } + +fn reset_pin_for_index(index: u8) -> BitcraneGpioPin { + match index { + 0 => BitcraneGpioPin::Rst0, + 1 => BitcraneGpioPin::Rst1, + 2 => BitcraneGpioPin::Rst2, + _ => panic!("invalid hashboard index {}", index), + } +} + +fn plug_pin_for_index(index: u8) -> BitcraneGpioPin { + match index { + 0 => BitcraneGpioPin::Plug0, + 1 => BitcraneGpioPin::Plug1, + 2 => BitcraneGpioPin::Plug2, + _ => panic!("invalid hashboard index {}", index), + } +} + +fn hashboard_count_u8(count: usize) -> u8 { + count.min(u8::MAX as usize) as u8 +} + +fn effective_hashrate(status: &HashThreadStatus) -> u64 { + u64::from(status.hashrate) +} diff --git a/mujina-miner/src/daemon.rs b/mujina-miner/src/daemon.rs index c2613d64..84239317 100644 --- a/mujina-miner/src/daemon.rs +++ b/mujina-miner/src/daemon.rs @@ -27,6 +27,8 @@ use crate::{ transport::{CpuDeviceInfo, TransportEvent, UsbTransport, cpu as cpu_transport}, }; +const SOURCE_COMMAND_BUFFER: usize = 1024; + /// The main daemon. pub struct Daemon { shutdown: CancellationToken, @@ -110,7 +112,7 @@ impl Daemon { // - MUJINA_POOL_USER: Worker username (optional, defaults to "mujina-testing") // - MUJINA_POOL_PASS: Worker password (optional, defaults to "x") let (source_event_tx, source_event_rx) = mpsc::channel::(100); - let (source_cmd_tx, source_cmd_rx) = mpsc::channel(10); + let (source_cmd_tx, source_cmd_rx) = mpsc::channel(SOURCE_COMMAND_BUFFER); if let Ok(pool_url) = env::var("MUJINA_POOL_URL") { // Use Stratum v1 source @@ -134,7 +136,8 @@ impl Daemon { // Create inner channels (stratum <-> wrapper) let (inner_event_tx, inner_event_rx) = mpsc::channel::(100); - let (inner_cmd_tx, inner_cmd_rx) = mpsc::channel::(10); + let (inner_cmd_tx, inner_cmd_rx) = + mpsc::channel::(SOURCE_COMMAND_BUFFER); let stratum_source = StratumV1Source::new( stratum_config, diff --git a/mujina-miner/src/mgmt_protocol/bitaxe_raw/mod.rs b/mujina-miner/src/mgmt_protocol/bitaxe_raw/mod.rs index c3b69623..1882f706 100644 --- a/mujina-miner/src/mgmt_protocol/bitaxe_raw/mod.rs +++ b/mujina-miner/src/mgmt_protocol/bitaxe_raw/mod.rs @@ -297,10 +297,31 @@ impl Decoder for ControlCodec { // Peek at length without consuming let length_field = u16::from_le_bytes([src[0], src[1]]) as usize; - // In bitaxe-raw, the length field contains the length of the response data only, - // NOT including the 2-byte length field itself or the 1-byte ID. - // Total packet size = 2 (length) + 1 (ID) + length_field - let total_packet_size = 2 + 1 + length_field; + // Normal responses encode the payload length, so total packet size is: + // 2 (length) + 1 (ID) + payload length. + // + // Some currently deployed RP2040 firmware variants incorrectly encode + // error responses with the total packet size in the length field + // instead. Those frames look like: + // [total_len:2] [id:1] [0xff] [error_code...] + // We tolerate that quirk here so a single unsupported telemetry + // command does not permanently desynchronize the control stream. + let legacy_error_packet = length_field >= 4 + && src.len() >= length_field + && match src.get(3).copied() { + Some(ERROR_MARKER) => true, + Some(code) => matches!( + ErrorCode::try_from(code), + Ok(ErrorCode::Timeout | ErrorCode::InvalidCommand | ErrorCode::BufferOverflow) + ), + None => false, + }; + + let total_packet_size = if legacy_error_packet { + length_field + } else { + 2 + 1 + length_field + }; if total_packet_size > self.max_length { return Err(io::Error::new( @@ -317,10 +338,29 @@ impl Decoder for ControlCodec { // Consume the complete packet let packet_data = src.split_to(total_packet_size); - // Skip the 2-byte length field - let response_data = &packet_data[2..]; - - let response = Response::parse(response_data)?; + let response = if legacy_error_packet { + let id = packet_data[2]; + match packet_data[3] { + ERROR_MARKER => Response::parse(&packet_data[2..])?, + code => Response { + id, + data: vec![], + error: Some(ResponseError { + code: ErrorCode::try_from(code).map_err(|unknown| { + io::Error::new( + io::ErrorKind::InvalidData, + format!("Unknown error code: 0x{:02x}", unknown), + ) + })?, + message: None, + }), + }, + } + } else { + // Skip the 2-byte length field + let response_data = &packet_data[2..]; + Response::parse(response_data)? + }; trace!( id = response.id, @@ -403,4 +443,16 @@ mod tests { assert!(response.is_error()); assert_eq!(response.error().unwrap().code, ErrorCode::InvalidCommand); } + + #[test] + fn test_decoder_accepts_legacy_error_length_encoding() { + let mut codec = ControlCodec::default(); + let mut src = BytesMut::from(&[0x04, 0x00, 0x42, 0x11][..]); + + let response = codec.decode(&mut src).unwrap().unwrap(); + assert_eq!(response.id, 0x42); + assert!(response.is_error()); + assert_eq!(response.error().unwrap().code, ErrorCode::InvalidCommand); + assert!(src.is_empty()); + } } diff --git a/mujina-miner/src/scheduler.rs b/mujina-miner/src/scheduler.rs index 7b04b366..7248f7ad 100644 --- a/mujina-miner/src/scheduler.rs +++ b/mujina-miner/src/scheduler.rs @@ -36,6 +36,7 @@ use std::collections::HashSet; use std::sync::Arc; use std::time::Duration; use tokio::sync::{mpsc, watch}; +use tokio::sync::mpsc::error::TrySendError; use tokio_stream::wrappers::ReceiverStream; use tokio_stream::{StreamExt, StreamMap}; @@ -277,10 +278,10 @@ impl Scheduler { /// /// Used with `broadcast_hashrate()` to avoid capturing `&self` across /// await points (Scheduler contains Box which isn't Sync). - fn hashrate_senders(&self) -> Vec> { + fn hashrate_senders(&self) -> Vec<(String, mpsc::Sender)> { self.sources .values() - .map(|s| s.command_tx.clone()) + .map(|s| (s.name.clone(), s.command_tx.clone())) .collect() } @@ -321,10 +322,13 @@ impl Scheduler { // Send current hashrate estimate to the new source let hashrate = self.operational_hashrate(); - let _ = self.sources[source_id] - .command_tx - .send(SourceCommand::UpdateHashRate(hashrate)) - .await; + let source = &self.sources[source_id]; + let _ = try_send_source_command( + &source.command_tx, + SourceCommand::UpdateHashRate(hashrate), + &source.name, + "initial hashrate update", + ); } /// Assign or replace work on all threads from a job template. @@ -498,23 +502,17 @@ impl Scheduler { // Check if share meets source threshold if task_entry.template.share_target.is_met_by(hash) { - self.stats.shares_submitted += 1; - // Submit share to originating source if let Some(source) = self.sources.get(task_entry.source_id) { let source_share = SourceShare::from((share, task_entry.template.id.clone())); - if let Err(e) = source - .command_tx - .send(SourceCommand::SubmitShare(source_share)) - .await - { - error!( - source_id = ?task_entry.source_id, - error = %e, - "Failed to submit share to source" - ); - } else { + if try_send_source_command( + &source.command_tx, + SourceCommand::SubmitShare(source_share), + &source.name, + "share submission", + ) { + self.stats.shares_submitted += 1; debug!(source = %source.name, "Share submitted to source"); } } else { @@ -586,7 +584,7 @@ impl Scheduler { // Broadcast updated hashrate to all sources let hashrate = self.operational_hashrate(); let senders = self.hashrate_senders(); - broadcast_hashrate(senders, hashrate).await; + broadcast_hashrate(senders, hashrate); // Reset difficulty alarm since hashrate changed for source in self.sources.values_mut() { @@ -688,7 +686,7 @@ impl Scheduler { // Broadcast updated hashrate to all sources let hashrate = self.operational_hashrate(); let senders = self.hashrate_senders(); - broadcast_hashrate(senders, hashrate).await; + broadcast_hashrate(senders, hashrate); // Reset difficulty alarm since hashrate changed for source in self.sources.values_mut() { @@ -830,7 +828,7 @@ impl Scheduler { } else { let hashrate = self.operational_hashrate(); let senders = self.hashrate_senders(); - broadcast_hashrate(senders, hashrate).await; + broadcast_hashrate(senders, hashrate); } let _ = miner_state_tx.send(self.compute_miner_state()); } @@ -866,9 +864,33 @@ impl Scheduler { /// /// Takes pre-collected senders to avoid capturing Scheduler across await /// points (it contains Box which isn't Sync). -async fn broadcast_hashrate(senders: Vec>, hashrate: HashRate) { - for sender in senders { - let _ = sender.send(SourceCommand::UpdateHashRate(hashrate)).await; +fn broadcast_hashrate(senders: Vec<(String, mpsc::Sender)>, hashrate: HashRate) { + for (source_name, sender) in senders { + let _ = try_send_source_command( + &sender, + SourceCommand::UpdateHashRate(hashrate), + &source_name, + "hashrate update", + ); + } +} + +fn try_send_source_command( + sender: &mpsc::Sender, + command: SourceCommand, + source_name: &str, + action: &str, +) -> bool { + match sender.try_send(command) { + Ok(()) => true, + Err(TrySendError::Full(_)) => { + warn!(source = source_name, action, "Source command queue full; dropping command"); + false + } + Err(TrySendError::Closed(_)) => { + warn!(source = source_name, action, "Source command queue closed; dropping command"); + false + } } } @@ -1043,4 +1065,42 @@ mod tests { ); } } + + #[test] + fn source_command_try_send_returns_false_when_queue_is_full() { + let (tx, mut rx) = mpsc::channel(1); + tx.try_send(SourceCommand::UpdateHashRate(HashRate::from(1))) + .unwrap(); + + let queued = try_send_source_command( + &tx, + SourceCommand::UpdateHashRate(HashRate::from(2)), + "pool", + "hashrate update", + ); + + assert!(!queued); + match rx.try_recv() { + Ok(SourceCommand::UpdateHashRate(rate)) => assert_eq!(u64::from(rate), 1), + other => panic!("expected queued hashrate update, got {other:?}"), + } + } + + #[test] + fn source_command_try_send_returns_true_when_queue_has_capacity() { + let (tx, mut rx) = mpsc::channel(1); + + let queued = try_send_source_command( + &tx, + SourceCommand::UpdateHashRate(HashRate::from(2)), + "pool", + "hashrate update", + ); + + assert!(queued); + match rx.try_recv() { + Ok(SourceCommand::UpdateHashRate(rate)) => assert_eq!(u64::from(rate), 2), + other => panic!("expected queued hashrate update, got {other:?}"), + } + } }