blackbeard/miner
1
0
Fork 0
Code Issues 2 Pull Requests Actions Packages Projects Releases 2 Wiki Activity

hash rate fix

Browse Source
This commit is contained in:
Nikolaus Heger
2026-01-06 15:17:00 +08:00
parent 059fa8322c
commit 689e5f9bfd
2 changed files with 216 additions and 114 deletions
Download Patch File Download Diff File Expand all files Collapse all files

211
crates/metrics/src/lib.rs
View File

@@ -126,18 +126,6 @@ static HASHES_TOTAL: Lazy<IntCounter> = Lazy::new(|| {
c
});
static MINING_DURATION_NANOS: Lazy<IntCounter> = 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<Gauge> = 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<Gauge> = 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<Instant>,
hashes: u64,
active_threads: usize,
threads: HashMap<ThreadKey, ThreadState>,
jobs: HashMap<JobKey, JobState>,
}
impl MiningSession {
fn new() -> Self {
Self {
start: None,
hashes: 0,
active_threads: 0,
threads: HashMap::new(),
jobs: HashMap::new(),
}
}
}
static SESSION: Lazy<Mutex<MiningSession>> = Lazy::new(|| Mutex::new(MiningSession::new()));
static HTTP_REQUESTS_TOTAL: Lazy<IntCounterVec> = 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<Mutex<HashSet<JobKey>>> = Lazy::new(|| Mutex::new(HashSet::new()));
static JOB_LAST: Lazy<Mutex<HashMap<JobKey, Instant>>> = Lazy::new(|| Mutex::new(HashMap::new()));
static THREAD_KEYS: Lazy<Mutex<HashSet<ThreadKey>>> = Lazy::new(|| Mutex::new(HashSet::new()));

119
crates/miner-service/src/lib.rs
View File

@@ -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<MiningJobResult>,
pub engine_name: &'static str,
@@ -357,7 +356,6 @@ pub struct MiningJob {
pub result_receiver: Option<Receiver<ThreadResult>>,
pub thread_handles: Vec<thread::JoinHandle<()>>,
pub thread_total_hashes: std::collections::HashMap<usize, u64>,
pub thread_final_rates: Vec<f64>,
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<MiningJobResult>,
hash_count: u64,
origin: Option<engine_cpu::FoundOrigin>,
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 {