diff --git a/mujina-miner/src/board/cpu.rs b/mujina-miner/src/board/cpu.rs index 5f07739f..79264242 100644 --- a/mujina-miner/src/board/cpu.rs +++ b/mujina-miner/src/board/cpu.rs @@ -3,12 +3,14 @@ //! Provides a virtual board that uses CPU cores for SHA-256 hashing. //! See [`CpuMinerConfig`] for environment variable configuration. +use std::time::Duration; + use anyhow::{Result, anyhow}; use tokio::sync::watch; use super::{BackplaneConnector, BoardInfo, VirtualBoardDescriptor}; use crate::{ - api_client::types::BoardTelemetry, + api_client::types::{BoardTelemetry, ThreadTelemetry}, asic::hash_thread::HashThread, cpu_miner::{CpuHashThread, CpuMinerConfig}, }; @@ -34,27 +36,68 @@ async fn create_cpu_board() -> Result { )), }; + let cpu_threads: Vec = (0..config.thread_count) + .map(|i| CpuHashThread::new(format!("CPU Core {i}"), config.duty_percent)) + .collect(); + + // Snapshot the thread names + live status handles before handing + // ownership of the threads to the scheduler. The board's telemetry + // task reads these to publish per-thread hashrate. + let thread_handles: Vec<(String, _)> = cpu_threads + .iter() + .map(|t| (t.name().to_string(), t.status_handle())) + .collect(); + + let initial_threads: Vec = thread_handles + .iter() + .map(|(name, _)| ThreadTelemetry { + name: name.clone(), + hashrate: 0, + is_active: false, + }) + .collect(); + let initial_state = BoardTelemetry { name: info.serial_number.clone().unwrap(), model: info.model.clone(), serial: info.serial_number.clone(), + threads: initial_threads, ..Default::default() }; - let (_telemetry_tx, telemetry_rx) = watch::channel(initial_state); - - let threads: Vec> = (0..config.thread_count) - .map(|i| { - Box::new(CpuHashThread::new( - format!("CPU Core {i}"), - config.duty_percent, - )) as _ - }) + let (telemetry_tx, telemetry_rx) = watch::channel(initial_state); + + // Periodically refresh BoardTelemetry from each thread's status. + // The task owns telemetry_tx so the channel stays alive until the + // board is shut down; the registry then evicts the board normally. + let publisher = tokio::spawn(async move { + let mut tick = tokio::time::interval(Duration::from_secs(2)); + tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + loop { + tick.tick().await; + let threads = thread_handles + .iter() + .map(|(name, status)| { + let s = status.read().unwrap(); + ThreadTelemetry { + name: name.clone(), + hashrate: u64::from(s.hashrate), + is_active: s.is_active, + } + }) + .collect(); + telemetry_tx.send_modify(|t| t.threads = threads); + } + }); + + let threads: Vec> = cpu_threads + .into_iter() + .map(|t| Box::new(t) as Box) .collect(); Ok(BackplaneConnector { info, threads, telemetry_rx, - shutdown: None, + shutdown: Some(Box::pin(async move { publisher.abort() })), }) } diff --git a/mujina-miner/src/cpu_miner/thread.rs b/mujina-miner/src/cpu_miner/thread.rs index 785934df..988e1778 100644 --- a/mujina-miner/src/cpu_miner/thread.rs +++ b/mujina-miner/src/cpu_miner/thread.rs @@ -108,6 +108,15 @@ impl CpuHashThread { self.shutdown.store(true, Ordering::Relaxed); let _ = self.command_tx.send(MinerCommand::Shutdown); } + + /// Clone of the live status handle. + /// + /// Lets the board layer publish per-thread telemetry without taking + /// ownership of the thread itself (which gets handed to the + /// scheduler). + pub fn status_handle(&self) -> Arc> { + Arc::clone(&self.status) + } } impl Drop for CpuHashThread { diff --git a/mujina-miner/src/scheduler.rs b/mujina-miner/src/scheduler.rs index 795a40e1..7fe229fd 100644 --- a/mujina-miner/src/scheduler.rs +++ b/mujina-miner/src/scheduler.rs @@ -201,28 +201,45 @@ impl Scheduler { /// Aggregate measured hashrate from per-thread estimators. /// - /// Returns the truth: zero if no shares have been recorded yet. + /// Falls back to the thread's self-reported `status().hashrate` when + /// the share-based estimator has no data yet. This surfaces a real + /// number for backends that measure hashrate directly (e.g. the CPU + /// miner, which counts hashes per cycle) without waiting for the + /// estimator to settle from observed shares -- which on slow + /// backends paired with a high source difficulty might never happen. fn measured_hashrate(&mut self) -> HashRate { self.threads .values_mut() - .map(|entry| entry.hashrate.hashrate()) + .map(|entry| { + let estimated = entry.hashrate.hashrate(); + if estimated.is_zero() { + entry.thread.status().hashrate + } else { + estimated + } + }) .sum() } /// Aggregate hashrate for operational decisions. /// - /// Per thread, uses measured hashrate if the estimator has settled, - /// otherwise falls back to the static capability estimate. Suitable - /// for broadcasting to sources and difficulty warnings, where a zero - /// value at startup would be unhelpful. + /// Per thread, prefers the share-based estimator once settled, then + /// the thread's self-reported hashrate, then the static capability + /// estimate. Suitable for broadcasting to sources and difficulty + /// warnings, where a zero value at startup would be unhelpful. fn operational_hashrate(&mut self) -> HashRate { self.threads .values_mut() .map(|entry| { - entry - .hashrate - .settled_hashrate() - .unwrap_or(entry.thread.capabilities().hashrate_estimate) + if let Some(settled) = entry.hashrate.settled_hashrate() { + return settled; + } + let reported = entry.thread.status().hashrate; + if reported.is_zero() { + entry.thread.capabilities().hashrate_estimate + } else { + reported + } }) .sum() }