From 689e5f9bfdab26a818e46a863e00c8a45ef364bb Mon Sep 17 00:00:00 2001 From: Nikolaus Heger Date: Tue, 6 Jan 2026 15:17:00 +0800 Subject: [PATCH] hash rate fix --- crates/metrics/src/lib.rs | 211 +++++++++++++++++++++++++++----- crates/miner-service/src/lib.rs | 119 +++++------------- 2 files changed, 216 insertions(+), 114 deletions(-) diff --git a/crates/metrics/src/lib.rs b/crates/metrics/src/lib.rs index 00e368d..45bfb7c 100644 --- a/crates/metrics/src/lib.rs +++ b/crates/metrics/src/lib.rs @@ -126,18 +126,6 @@ static HASHES_TOTAL: Lazy = Lazy::new(|| { c }); -static MINING_DURATION_NANOS: Lazy = Lazy::new(|| { - let c = IntCounter::new( - "miner_mining_duration_nanos", - "Total time spent mining across all threads (nanoseconds)", - ) - .expect("create miner_mining_duration_nanos"); - REGISTRY - .register(Box::new(c.clone())) - .expect("register miner_mining_duration_nanos"); - c -}); - static HASH_RATE: Lazy = Lazy::new(|| { let g = Gauge::new("miner_hash_rate", "Estimated hash rate (nonces per second)").expect("create"); @@ -147,6 +135,45 @@ static HASH_RATE: Lazy = Lazy::new(|| { g }); +type JobKey = (String, String); // (engine, job_id) +type ThreadKey = (String, String, String); // (engine, job_id, thread_id) + +#[derive(Debug)] +struct ThreadState { + start: Instant, + hashes: u64, +} + +#[derive(Debug)] +struct JobState { + start: Instant, + hashes: u64, + active_threads: usize, +} + +#[derive(Debug)] +struct MiningSession { + start: Option, + hashes: u64, + active_threads: usize, + threads: HashMap, + jobs: HashMap, +} + +impl MiningSession { + fn new() -> Self { + Self { + start: None, + hashes: 0, + active_threads: 0, + threads: HashMap::new(), + jobs: HashMap::new(), + } + } +} + +static SESSION: Lazy> = Lazy::new(|| Mutex::new(MiningSession::new())); + static HTTP_REQUESTS_TOTAL: Lazy = Lazy::new(|| { let c = IntCounterVec::new( opts!( @@ -549,21 +576,152 @@ pub fn inc_hashes(n: u64) { HASHES_TOTAL.inc_by(n); } -/// Record a completed mining segment to update global hash rate. -/// -/// This accumulates total hashes and total duration across all threads/jobs -/// to provide a stable, global hash rate average. -pub fn record_mining_segment(hashes: u64, duration: Duration) { - HASHES_TOTAL.inc_by(hashes); - MINING_DURATION_NANOS.inc_by(duration.as_nanos() as u64); +fn ensure_session(session: &mut MiningSession, now: Instant) { + if session.start.is_none() { + session.start = Some(now); + session.hashes = 0; + } +} - let total_hashes = HASHES_TOTAL.get(); - let total_nanos = MINING_DURATION_NANOS.get(); +fn update_global_rate(session: &MiningSession, now: Instant) { + if let Some(start) = session.start { + let elapsed = now.saturating_duration_since(start).as_secs_f64(); + if elapsed > 0.0 && session.hashes > 0 { + HASH_RATE.set(session.hashes as f64 / elapsed); + } + } +} - if total_nanos > 0 { - let total_seconds = total_nanos as f64 / 1_000_000_000.0; - let rate = total_hashes as f64 / total_seconds; - HASH_RATE.set(rate); +fn finish_thread_locked(session: &mut MiningSession, job_key: &JobKey, thread_key: &ThreadKey) { + if session.threads.remove(thread_key).is_some() { + session.active_threads = session.active_threads.saturating_sub(1); + } + if let Some(job) = session.jobs.get_mut(job_key) { + if job.active_threads > 0 { + job.active_threads -= 1; + } + } + if session.active_threads == 0 { + session.start = None; + session.hashes = 0; + HASH_RATE.set(0.0); + } +} + +pub fn start_thread(engine: &str, job_id: &str, thread_id: usize) { + let now = Instant::now(); + let engine_key = engine.to_string(); + let job_key = job_id.to_string(); + let thread_key = (engine_key.clone(), job_key.clone(), thread_id.to_string()); + let job_key = (engine_key, job_key); + + let mut session = SESSION.lock().unwrap(); + ensure_session(&mut session, now); + + if session.threads.contains_key(&thread_key) { + return; + } + + session.active_threads = session.active_threads.saturating_add(1); + session + .threads + .insert(thread_key, ThreadState { start: now, hashes: 0 }); + + let job = session.jobs.entry(job_key).or_insert(JobState { + start: now, + hashes: 0, + active_threads: 0, + }); + if job.active_threads == 0 { + job.start = now; + job.hashes = 0; + } + job.active_threads = job.active_threads.saturating_add(1); +} + +pub fn record_thread_progress( + engine: &str, + job_id: &str, + thread_id: usize, + hashes: u64, + completed: bool, + update_job_metrics: bool, +) { + let now = Instant::now(); + let engine_key = engine.to_string(); + let job_key = job_id.to_string(); + let thread_id_str = thread_id.to_string(); + let thread_key = (engine_key.clone(), job_key.clone(), thread_id_str.clone()); + let job_key = (engine_key, job_key); + + let mut session = SESSION.lock().unwrap(); + ensure_session(&mut session, now); + + let thread_was_present = session.threads.contains_key(&thread_key); + if !thread_was_present { + session.active_threads = session.active_threads.saturating_add(1); + session.threads.insert( + thread_key.clone(), + ThreadState { + start: now, + hashes: 0, + }, + ); + } + + let (thread_start, thread_hashes_total) = { + let thread_state = session.threads.get_mut(&thread_key).expect("thread must exist"); + thread_state.hashes = thread_state.hashes.saturating_add(hashes); + (thread_state.start, thread_state.hashes) + }; + + let (job_start, job_hashes_total) = { + let job_state = session.jobs.entry(job_key.clone()).or_insert(JobState { + start: thread_start, + hashes: 0, + active_threads: 0, + }); + if job_state.active_threads == 0 { + job_state.start = thread_start; + job_state.hashes = 0; + } + if !thread_was_present { + job_state.active_threads = job_state.active_threads.saturating_add(1); + } + job_state.hashes = job_state.hashes.saturating_add(hashes); + (job_state.start, job_state.hashes) + }; + + session.hashes = session.hashes.saturating_add(hashes); + + inc_hashes(hashes); + if update_job_metrics { + inc_job_hashes(engine, job_id, hashes); + inc_thread_hashes(engine, job_id, &thread_id_str, hashes); + } + + let thread_elapsed = now.saturating_duration_since(thread_start).as_secs_f64(); + if update_job_metrics && thread_elapsed > 0.0 && thread_hashes_total > 0 { + set_thread_hash_rate( + engine, + job_id, + &thread_id_str, + thread_hashes_total as f64 / thread_elapsed, + ); + } + + if update_job_metrics { + let job_elapsed = now.saturating_duration_since(job_start).as_secs_f64(); + if job_elapsed > 0.0 && job_hashes_total > 0 { + set_job_hash_rate(engine, job_id, job_hashes_total as f64 / job_elapsed); + } + } + + update_global_rate(&session, now); + + if completed { + finish_thread_locked(&mut session, &job_key, &thread_key); + update_global_rate(&session, now); } } @@ -591,9 +749,6 @@ pub fn inc_http_request(endpoint: &str, code: u16) { const METRICS_TTL_SECS: u64 = 300; const JANITOR_INTERVAL_SECS: u64 = 60; -type JobKey = (String, String); // (engine, job_id) -type ThreadKey = (String, String, String); // (engine, job_id, thread_id) - static JOB_KEYS: Lazy>> = Lazy::new(|| Mutex::new(HashSet::new())); static JOB_LAST: Lazy>> = Lazy::new(|| Mutex::new(HashMap::new())); static THREAD_KEYS: Lazy>> = Lazy::new(|| Mutex::new(HashSet::new())); diff --git a/crates/miner-service/src/lib.rs b/crates/miner-service/src/lib.rs index 7d669ce..ccf2ae5 100644 --- a/crates/miner-service/src/lib.rs +++ b/crates/miner-service/src/lib.rs @@ -348,7 +348,6 @@ pub struct MiningJob { pub status: JobStatus, pub start_time: Instant, pub total_hash_count: u64, - pub last_hash_rate: f64, pub best_result: Option, pub engine_name: &'static str, @@ -357,7 +356,6 @@ pub struct MiningJob { pub result_receiver: Option>, pub thread_handles: Vec>, pub thread_total_hashes: std::collections::HashMap, - pub thread_final_rates: Vec, completed_threads: usize, pub result_served: bool, } @@ -373,7 +371,6 @@ impl Clone for MiningJob { status: self.status.clone(), start_time: self.start_time, total_hash_count: self.total_hash_count, - last_hash_rate: self.last_hash_rate, best_result: self.best_result.clone(), engine_name: self.engine_name, job_id: self.job_id.clone(), @@ -383,7 +380,6 @@ impl Clone for MiningJob { result_receiver: None, thread_handles: Vec::new(), thread_total_hashes: self.thread_total_hashes.clone(), - thread_final_rates: self.thread_final_rates.clone(), completed_threads: self.completed_threads, result_served: self.result_served, } @@ -396,7 +392,6 @@ pub struct ThreadResult { result: Option, hash_count: u64, origin: Option, - duration: std::time::Duration, completed: bool, } @@ -415,7 +410,6 @@ impl MiningJob { status: JobStatus::Running, start_time: Instant::now(), total_hash_count: 0, - last_hash_rate: 0.0, best_result: None, engine_name: "unknown", job_id: None, @@ -423,7 +417,6 @@ impl MiningJob { result_receiver: None, thread_handles: Vec::new(), thread_total_hashes: std::collections::HashMap::new(), - thread_final_rates: Vec::new(), completed_threads: 0, result_served: false, } @@ -496,6 +489,9 @@ impl MiningJob { let engine = cpu_engine.clone(); let job_id = self.job_id.clone().unwrap_or_else(|| "unknown".to_string()); + #[cfg(feature = "metrics")] + metrics::start_thread(self.engine_name, &job_id, thread_id); + let handle = thread::spawn(move || { mine_range_with_engine_typed( thread_id, @@ -529,6 +525,9 @@ impl MiningJob { let engine = gpu_engine.clone(); let job_id = self.job_id.clone().unwrap_or_else(|| "unknown".to_string()); + #[cfg(feature = "metrics")] + metrics::start_thread(self.engine_name, &job_id, thread_id); + let handle = thread::spawn(move || { mine_range_with_engine_typed( thread_id, @@ -566,9 +565,6 @@ impl MiningJob { metrics::set_job_status_gauge(self.engine_name, job_id, "completed", 0); metrics::set_job_status_gauge(self.engine_name, job_id, "failed", 0); metrics::set_job_status_gauge(self.engine_name, job_id, "cancelled", 1); - // Remove job hash rate series on cancellation (avoid scrape-timing artifacts) - self.last_hash_rate = 0.0; - metrics::remove_job_hash_rate(self.engine_name, job_id); metrics::remove_job_metrics(self.engine_name, job_id); metrics::remove_thread_metrics_for_job(self.engine_name, job_id); // Remove all per-thread hash rate series on cancellation @@ -600,32 +596,15 @@ impl MiningJob { .or_default() += thread_result.hash_count; #[cfg(feature = "metrics")] - { - metrics::inc_hashes(thread_result.hash_count); - metrics::record_mining_segment(thread_result.hash_count, thread_result.duration); - - // Only update job-specific metrics if the job is still considered running. - // Once completed/failed/cancelled, we stop updating job metrics to avoid - // resurrecting series that were cleaned up. - if self.status == JobStatus::Running { - if let Some(job_id) = &self.job_id { - metrics::inc_job_hashes(self.engine_name, job_id, thread_result.hash_count); - metrics::inc_thread_hashes( - self.engine_name, - job_id, - &thread_result.thread_id.to_string(), - thread_result.hash_count, - ); - - // Simple per-job hash rate based on total progress - let elapsed = self.start_time.elapsed().as_secs_f64(); - if elapsed > 0.0 { - let job_rate = self.total_hash_count as f64 / elapsed; - self.last_hash_rate = job_rate; - metrics::set_job_hash_rate(self.engine_name, job_id, job_rate); - } - } - } + if let Some(job_id) = &self.job_id { + metrics::record_thread_progress( + self.engine_name, + job_id, + thread_result.thread_id, + thread_result.hash_count, + thread_result.completed, + self.status == JobStatus::Running, + ); } if thread_result.completed { @@ -638,8 +617,6 @@ impl MiningJob { if elapsed > 0.0 && thread_total > 0 { let thread_rate = thread_total as f64 / elapsed; - self.thread_final_rates.push(thread_rate); - log::info!( "Thread {} finished - Rate: {:.2} H/s ({} hashes in {:.2}s)", thread_result.thread_id, @@ -647,29 +624,6 @@ impl MiningJob { thread_total, elapsed ); - - #[cfg(feature = "metrics")] - if let Some(job_id) = &self.job_id { - // Report final thread rate before cleaning up - metrics::set_thread_hash_rate( - self.engine_name, - job_id, - &thread_result.thread_id.to_string(), - thread_rate, - ); - } - } - - #[cfg(feature = "metrics")] - { - if let Some(job_id) = &self.job_id { - // Remove this thread's hash rate series on completion - metrics::remove_thread_hash_rate( - self.engine_name, - job_id, - &thread_result.thread_id.to_string(), - ); - } } } @@ -723,9 +677,6 @@ impl MiningJob { metrics::set_job_status_gauge(self.engine_name, job_id, "completed", 1); metrics::set_job_status_gauge(self.engine_name, job_id, "failed", 0); metrics::set_job_status_gauge(self.engine_name, job_id, "cancelled", 0); - // Remove job hash rate on completion and clear per-thread series - self.last_hash_rate = 0.0; - metrics::remove_job_hash_rate(self.engine_name, job_id); metrics::remove_job_metrics(self.engine_name, job_id); metrics::remove_thread_metrics_for_job(self.engine_name, job_id); for (tid, _) in self.thread_total_hashes.iter() { @@ -750,9 +701,6 @@ impl MiningJob { metrics::set_job_status_gauge(self.engine_name, job_id, "completed", 0); metrics::set_job_status_gauge(self.engine_name, job_id, "failed", 1); metrics::set_job_status_gauge(self.engine_name, job_id, "cancelled", 0); - // Remove job hash rate on failure and clear per-thread series - self.last_hash_rate = 0.0; - metrics::remove_job_hash_rate(self.engine_name, job_id); for (tid, _) in self.thread_total_hashes.iter() { metrics::remove_thread_hash_rate( self.engine_name, @@ -892,14 +840,13 @@ fn mine_range_with_engine_typed( hash_count, origin, } => { - let duration = start_time.elapsed(); + let _duration = start_time.elapsed(); // Send final result with found candidate and the hashes covered in this subrange let final_result = ThreadResult { thread_id, result: Some(MiningJobResult { nonce, work, hash }), hash_count, origin: Some(origin), - duration, completed: true, }; log::info!( @@ -921,7 +868,7 @@ fn mine_range_with_engine_typed( break; } engine_cpu::EngineStatus::Exhausted { hash_count } => { - let duration = start_time.elapsed(); + let _duration = start_time.elapsed(); total_hashes_processed += hash_count; // Send intermediate progress update for this chunk let update = ThreadResult { @@ -929,7 +876,6 @@ fn mine_range_with_engine_typed( result: None, hash_count, origin: None, - duration, completed: false, }; if sender.try_send(update).is_err() { @@ -952,7 +898,7 @@ fn mine_range_with_engine_typed( } } engine_cpu::EngineStatus::Cancelled { hash_count } => { - let duration = start_time.elapsed(); + let _duration = start_time.elapsed(); total_hashes_processed += hash_count; // Send last progress update and stop let update = ThreadResult { @@ -960,8 +906,7 @@ fn mine_range_with_engine_typed( result: None, hash_count, origin: None, - duration, - completed: false, + completed: true, }; if sender.try_send(update).is_err() { log::warn!(target: "miner", "Job {job_id} thread {thread_id} failed to send cancel update"); @@ -1011,7 +956,6 @@ fn mine_range_with_engine_typed( result: None, hash_count: 0, origin: None, - duration: std::time::Duration::from_secs(0), completed: true, }; if sender.try_send(final_result).is_err() { @@ -1171,11 +1115,15 @@ pub async fn handle_result_request( let elapsed = job.start_time.elapsed().as_secs_f64(); match job.status { JobStatus::Running => { + #[cfg(feature = "metrics")] + let current_rate = metrics::get_hash_rate(); + #[cfg(not(feature = "metrics"))] + let current_rate = 0.0; log::debug!( "🔍 Job {} still running - elapsed: {:.1}s - hash rate: {:.0} H/s", job_id, elapsed, - job.last_hash_rate + current_rate ); } JobStatus::Completed if !job.result_served => { @@ -1653,19 +1601,18 @@ pub async fn run(config: ServiceConfig) -> anyhow::Result<()> { tokio::spawn(async move { loop { - let (active_jobs, total_rate) = { + let active_jobs = { let jobs = svc.jobs.lock().await; - let mut running = 0i64; - let mut rate = 0.0; - for (_id, job) in jobs.iter() { - if job.status == JobStatus::Running { - running += 1; - rate += job.last_hash_rate; - } - } - (running, rate) + jobs.values() + .filter(|job| job.status == JobStatus::Running) + .count() as i64 }; + #[cfg(feature = "metrics")] + let total_rate = metrics::get_hash_rate(); + #[cfg(not(feature = "metrics"))] + let total_rate = 0.0; + let uptime_ms = start_instant.elapsed().as_millis() as u64; let interval = miner_telemetry::SystemInterval {