cpu-miner: make the CPU backend a first-class citizen for API/tooling

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<BackplaneConnector> {
         )),
     };
 
+    let cpu_threads: Vec<CpuHashThread> = (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<ThreadTelemetry> = 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<Box<dyn HashThread>> = (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<Box<dyn HashThread>> = cpu_threads
+        .into_iter()
+        .map(|t| Box::new(t) as Box<dyn HashThread>)
         .collect();
 
     Ok(BackplaneConnector {
         info,
         threads,
         telemetry_rx,
-        shutdown: None,
+        shutdown: Some(Box::pin(async move { publisher.abort() })),
     })
 }

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<RwLock<HashThreadStatus>> {
+        Arc::clone(&self.status)
+    }
 }
 
 impl Drop for CpuHashThread {

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()
     }