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

improved logging

Browse Source
This commit is contained in:
illuzen
2025-12-08 14:55:47 +08:00
parent 5705d7fb1a
commit ccca629c49
6 changed files with 413 additions and 92 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+1
View File
@@ -581,6 +581,7 @@ dependencies = [
"futures",
"hex",
"log",
"metrics",
"pow-core",
"primitive-types 0.13.1",
"qp-plonky2 1.1.1",
crates/engine-gpu/Cargo.toml
+3 -1
View File
@@ -11,15 +11,17 @@ description = "GPU mining engine implementation for the Quantus External Miner"
[features]
# Default to the baseline/reference implementation; enable others as needed.
default = ["baseline"]
default = ["baseline", "metrics"]
# Map engine features (pow-core dependency removed for now)
baseline = []
simd-poseidon2 = []
metrics = ["dep:metrics"]
[dependencies]
engine-cpu = { path = "../engine-cpu" }
pow-core = { path = "../pow-core" }
metrics = { path = "../metrics", optional = true }
primitive-types = { workspace = true }
log = { workspace = true }
thiserror = { workspace = true }
crates/engine-gpu/src/lib.rs
+65 -9
View File
@@ -32,7 +32,7 @@ impl GpuEngine {
}
async fn init() -> Result<Self, Box<dyn std::error::Error>> {
log::info!(target: "gpu_engine", "Initializing WGPU...");
log::info!("Initializing WGPU...");
let instance = wgpu::Instance::new(&wgpu::InstanceDescriptor {
backends: wgpu::Backends::PRIMARY,
..Default::default()
@@ -44,15 +44,22 @@ impl GpuEngine {
let adapters: Vec<_> = adapters.into_iter().collect();
if adapters.is_empty() {
log::error!(target: "gpu_engine", "No suitable GPU adapters found.");
log::error!("No suitable GPU adapters found.");
return Err("No suitable GPU adapters found".into());
}
let mut contexts = Vec::new();
let mut adapter_infos = Vec::new();
for (i, adapter) in adapters.into_iter().enumerate() {
let info = adapter.get_info();
log::info!(target: "gpu_engine", "Initializing adapter {}: {} (Backend: {:?})", i, info.name, info.backend);
log::info!(
"Initializing GPU adapter {}: {} (Backend: {:?})",
i,
info.name,
info.backend
);
log::debug!(target: "gpu_engine", "Adapter {} info: {:?}", i, info);
adapter_infos.push(info.clone());
let (device, queue) = adapter
.request_device(&wgpu::DeviceDescriptor {
@@ -164,7 +171,43 @@ impl GpuEngine {
}));
}
log::info!(target: "gpu_engine", "GPU engine initialized with {} devices", contexts.len());
log::info!("GPU engine initialized with {} devices", contexts.len());
// Set engine backend info for metrics
#[cfg(feature = "metrics")]
{
metrics::set_gpu_device_count(contexts.len() as i64);
for (i, adapter_info) in adapter_infos.iter().enumerate() {
let device_id = format!("gpu-{}", i);
let backend_str = format!("{:?}", adapter_info.backend);
let vendor_str = format!("{}", adapter_info.vendor);
let device_type_str = format!("{:?}", adapter_info.device_type);
let clean_name = adapter_info.name.replace(" ", "_").replace(",", "");
// Set general engine backend info
metrics::set_engine_backend(&device_id, &backend_str);
// Set detailed GPU device info
metrics::set_gpu_device_info(
&device_id,
&clean_name,
&backend_str,
&vendor_str,
&device_type_str,
);
// Log GPU device info for monitoring
log::info!(
"📊 GPU Device {}: {} | Backend: {:?} | Vendor: {} | Device Type: {:?}",
i,
adapter_info.name,
adapter_info.backend,
adapter_info.vendor,
adapter_info.device_type
);
}
}
Ok(Self { contexts })
}
@@ -240,7 +283,7 @@ impl GpuEngine {
let hash = U512::from_little_endian(&hash_bytes);
let work = nonce.to_big_endian();
log::info!(target: "gpu_engine", "Solution found on GPU! Nonce: {}, Hash: {:x}", nonce, hash);
log::info!("GPU solution found! Nonce: {}, Hash: {:x}", nonce, hash);
drop(data);
ctx.staging_buffer.unmap();
@@ -249,7 +292,7 @@ impl GpuEngine {
drop(data);
ctx.staging_buffer.unmap();
} else {
log::error!(target: "gpu_engine", "Failed to map staging buffer");
log::error!("GPU failed to map staging buffer");
}
None
@@ -268,10 +311,15 @@ impl MinerEngine for GpuEngine {
fn search_range(&self, ctx: &JobContext, range: Range, cancel: &AtomicBool) -> EngineStatus {
// Simple multi-gpu strategy: split range among available GPUs
let num_gpus = self.contexts.len();
log::info!(target: "gpu_engine", "Starting search on {} GPUs. Range: {} - {}", num_gpus, range.start, range.end);
log::info!(
"Starting GPU search on {} devices. Range: {} - {}",
num_gpus,
range.start,
range.end
);
if num_gpus == 0 {
log::warn!(target: "gpu_engine", "No GPUs available for search.");
log::warn!("No GPUs available for search.");
return EngineStatus::Exhausted { hash_count: 0 };
}
@@ -326,6 +374,14 @@ impl MinerEngine for GpuEngine {
// Increased batch size to reduce overhead
let batch_size = 65536 * 64; // ~4M hashes per batch
#[cfg(feature = "metrics")]
{
let device_id = format!("gpu-{}", i);
metrics::set_gpu_batch_size(&device_id, batch_size as f64);
let workgroups = (batch_size + 255) / 256;
metrics::set_gpu_workgroups(&device_id, workgroups as f64);
}
while current_start <= end {
if cancel.load(Ordering::Relaxed) {
log::debug!(target: "gpu_engine", "GPU {} cancelled.", i);
@@ -347,7 +403,7 @@ impl MinerEngine for GpuEngine {
{
let mut lock = found_candidate.lock().unwrap();
if lock.is_none() {
log::info!(target: "gpu_engine", "GPU {} found candidate!", i);
log::info!("GPU {} found solution {:?}", i, candidate);
*lock = Some(candidate);
}
return;
crates/metrics/src/lib.rs
+101 -7
View File
@@ -31,11 +31,7 @@ use prometheus::{
};
#[cfg(feature = "http-exporter")]
use {
anyhow::Result,
std::net::SocketAddr,
warp::{http::Response, Filter},
};
use {anyhow::Result, std::net::SocketAddr, warp, warp::Filter};
#[cfg(not(feature = "http-exporter"))]
use anyhow::Result;
@@ -341,7 +337,7 @@ static JOB_FOUND_ORIGIN: Lazy<GaugeVec> = Lazy::new(|| {
let g = GaugeVec::new(
opts!(
"miner_job_found_origin",
"Job found origin gauge (set to 1 for the origin that found the candidate)"
"Per-job found candidate origin (0=unknown, 1=cpu, 2=gpu-g1, 3=gpu-g2)"
),
&["engine", "job_id", "origin"],
)
@@ -352,6 +348,67 @@ static JOB_FOUND_ORIGIN: Lazy<GaugeVec> = Lazy::new(|| {
g
});
// -------------------------------------------------------------------------------------
// GPU-specific metrics
// -------------------------------------------------------------------------------------
static GPU_DEVICE_COUNT: Lazy<IntGauge> = Lazy::new(|| {
let g = IntGauge::new(
"miner_gpu_devices_total",
"Number of GPU devices available for mining",
)
.expect("create miner_gpu_devices_total");
REGISTRY
.register(Box::new(g.clone()))
.expect("register miner_gpu_devices_total");
g
});
static GPU_DEVICE_INFO: Lazy<GaugeVec> = Lazy::new(|| {
let g = GaugeVec::new(
opts!(
"miner_gpu_device_info",
"GPU device information (label-only gauge set to 1). Labels: device_id, name, backend, vendor, device_type"
),
&["device_id", "name", "backend", "vendor", "device_type"],
)
.expect("create miner_gpu_device_info");
REGISTRY
.register(Box::new(g.clone()))
.expect("register miner_gpu_device_info");
g
});
static GPU_BATCH_SIZE: Lazy<GaugeVec> = Lazy::new(|| {
let g = GaugeVec::new(
opts!(
"miner_gpu_batch_size",
"Current GPU batch size (hashes per batch) per device"
),
&["device_id"],
)
.expect("create miner_gpu_batch_size");
REGISTRY
.register(Box::new(g.clone()))
.expect("register miner_gpu_batch_size");
g
});
static GPU_WORKGROUPS: Lazy<GaugeVec> = Lazy::new(|| {
let g = GaugeVec::new(
opts!(
"miner_gpu_workgroups",
"Number of GPU workgroups dispatched per device"
),
&["device_id"],
)
.expect("create miner_gpu_workgroups");
REGISTRY
.register(Box::new(g.clone()))
.expect("register miner_gpu_workgroups");
g
});
pub fn default_registry() -> &'static Registry {
&REGISTRY
}
@@ -647,6 +704,42 @@ pub fn remove_thread_hash_rate(engine: &str, job_id: &str, thread_id: &str) {
let _ = THREAD_HASH_RATE.remove_label_values(&[engine, job_id, thread_id]);
}
// -------------------------------------------------------------------------------------
// GPU-specific metric helpers
// -------------------------------------------------------------------------------------
/// Set the total number of GPU devices available for mining
pub fn set_gpu_device_count(count: i64) {
GPU_DEVICE_COUNT.set(count);
}
/// Set GPU device information (call once per device during initialization)
pub fn set_gpu_device_info(
device_id: &str,
name: &str,
backend: &str,
vendor: &str,
device_type: &str,
) {
GPU_DEVICE_INFO
.with_label_values(&[device_id, name, backend, vendor, device_type])
.set(1.0);
}
/// Set GPU batch size for a specific device
pub fn set_gpu_batch_size(device_id: &str, batch_size: f64) {
GPU_BATCH_SIZE
.with_label_values(&[device_id])
.set(batch_size);
}
/// Set GPU workgroup count for a specific device
pub fn set_gpu_workgroups(device_id: &str, workgroups: f64) {
GPU_WORKGROUPS
.with_label_values(&[device_id])
.set(workgroups);
}
// -------------------------------------------------------------------------------------
// HTTP Exporter (feature: http-exporter)
// -------------------------------------------------------------------------------------
@@ -657,6 +750,7 @@ pub fn remove_thread_hash_rate(engine: &str, job_id: &str, thread_id: &str) {
/// - Spawns the exporter as a background task and returns immediately.
/// - Serves plaintext metrics at GET /metrics.
/// - If called multiple times, multiple servers may be created (call once).
#[cfg(feature = "http-exporter")]
pub async fn start_http_exporter(port: u16) -> Result<()> {
// Encoder is created inside the handler to avoid capturing non-Clone state
// Use REGISTRY.gather() directly in the handler
@@ -670,7 +764,7 @@ pub async fn start_http_exporter(port: u16) -> Result<()> {
.encode(&metric_families, &mut buffer)
.unwrap_or_default();
Response::builder()
warp::http::Response::builder()
.header("Content-Type", encoder.format_type())
.body(String::from_utf8(buffer).unwrap_or_default())
});
crates/miner-cli/src/main.rs
+22 -7
View File
@@ -17,10 +17,19 @@ struct Args {
#[arg(long, env = "MINER_METRICS_PORT")]
metrics_port: Option<u16>,
/// Target milliseconds for per-thread progress updates (chunking).
/// Smaller values increase metrics freshness but add a bit of overhead.
#[arg(long = "progress-chunk-ms", env = "MINER_PROGRESS_CHUNK_MS")]
progress_chunk_ms: Option<u64>,
/// Enable verbose logging (shows debug info, progress details, etc.)
#[arg(short, long, env = "MINER_VERBOSE")]
verbose: bool,
/// How often to report mining progress (in milliseconds).
/// Smaller values give more frequent updates but slightly reduce performance.
#[arg(long = "progress-interval-ms", env = "MINER_PROGRESS_INTERVAL_MS")]
progress_interval_ms: Option<u64>,
/// Size of work chunks to process before reporting progress (number of hashes).
/// If omitted, uses engine-specific defaults (200K for CPU, 100M for GPU).
#[arg(long = "chunk-size", env = "MINER_CHUNK_SIZE")]
chunk_size: Option<u64>,
/// For cpu-chain-manipulator: start throttle index at this many solved blocks
/// to "pick up where we left off" after restarts.
@@ -117,9 +126,14 @@ async fn main() {
let args = Args::parse();
// Initialize logger early to capture startup messages.
// If RUST_LOG is not set, default to info level for our app.
// If RUST_LOG is not set, default to appropriate level based on verbose flag
if std::env::var("RUST_LOG").is_err() {
std::env::set_var("RUST_LOG", "info");
let log_level = if args.verbose {
"debug,miner=debug,gpu_engine=debug,engine_cpu=debug"
} else {
"info,miner=info,gpu_engine=info"
};
std::env::set_var("RUST_LOG", log_level);
}
env_logger::init();
@@ -164,7 +178,8 @@ async fn main() {
port: args.port,
workers: args.workers,
metrics_port: args.metrics_port,
progress_chunk_ms: args.progress_chunk_ms,
progress_interval_ms: args.progress_interval_ms,
chunk_size: args.chunk_size,
manip_solved_blocks: args.manip_solved_blocks,
manip_base_delay_ns: args.manip_base_delay_ns,
manip_step_batch: args.manip_step_batch,
crates/miner-service/src/lib.rs
+221 -68
View File
@@ -25,8 +25,10 @@ pub struct ServiceConfig {
pub workers: Option<usize>,
/// Optional metrics port. When Some, metrics endpoint starts; when None, metrics are disabled.
pub metrics_port: Option<u16>,
/// Target milliseconds for per-thread progress updates (chunking). If None, defaults to 2000ms.
pub progress_chunk_ms: Option<u64>,
/// How often to report mining progress (in milliseconds). If None, defaults to 10000ms.
pub progress_interval_ms: Option<u64>,
/// Size of work chunks to process before reporting progress (in number of hashes). If None, uses engine-specific defaults.
pub chunk_size: Option<u64>,
/// Optional starting value for the manipulator engine's solved-blocks throttle index.
pub manip_solved_blocks: Option<u64>,
/// Optional base sleep per batch in nanoseconds for manipulator engine (default 500_000ns).
@@ -54,7 +56,8 @@ impl Default for ServiceConfig {
port: 9833,
workers: None,
metrics_port: None,
progress_chunk_ms: None,
progress_interval_ms: None,
chunk_size: None,
manip_solved_blocks: None,
manip_base_delay_ns: None,
manip_step_batch: None,
@@ -74,16 +77,17 @@ impl fmt::Display for ServiceConfig {
};
write!(
f,
"port={}, workers={:?}, engine={}, metrics_port={:?}, progress_chunk_ms={:?}, manip_solved_blocks={:?}, manip_base_delay_ns={:?}, manip_step_batch={:?}, manip_throttle_cap={:?}",
"port={}, workers={:?}, engine={}, metrics_port={:?}, progress_interval_ms={:?}, chunk_size={:?}, manip_solved_blocks={:?}, manip_base_delay_ns={:?}, manip_step_batch={:?}, manip_throttle_cap={:?}",
self.port,
self.workers,
engine,
self.metrics_port,
self.progress_chunk_ms,
self.progress_interval_ms,
self.chunk_size,
self.manip_solved_blocks,
self.manip_base_delay_ns,
self.manip_step_batch,
self.manip_throttle_cap
self.manip_throttle_cap,
)
}
}
@@ -94,19 +98,27 @@ pub struct MiningService {
pub jobs: Arc<Mutex<HashMap<String, MiningJob>>>,
pub workers: usize,
pub engine: Arc<dyn MinerEngine>,
/// Target milliseconds for per-thread progress updates (chunking).
pub progress_chunk_ms: u64,
/// How often to report mining progress (in milliseconds).
pub progress_interval_ms: u64,
/// Work chunk size (number of hashes to process before progress update).
pub chunk_size: Option<u64>,
/// Gauge of currently running jobs (for metrics)
pub active_jobs_gauge: Arc<tokio::sync::Mutex<i64>>,
}
impl MiningService {
pub fn new(workers: usize, engine: Arc<dyn MinerEngine>, progress_chunk_ms: u64) -> Self {
fn new(
workers: usize,
engine: Arc<dyn MinerEngine>,
progress_interval_ms: u64,
chunk_size: Option<u64>,
) -> Self {
Self {
jobs: Arc::new(Mutex::new(HashMap::new())),
workers,
engine,
progress_chunk_ms,
progress_interval_ms,
chunk_size,
active_jobs_gauge: Arc::new(tokio::sync::Mutex::new(0)),
}
}
@@ -118,7 +130,11 @@ impl MiningService {
return Err("Job already exists".to_string());
}
log::debug!(target: "miner", "Adding job: {} with {} workers", job_id, self.workers);
log::info!(
"Adding mining job: {} with {} workers",
job_id,
self.workers
);
job.job_id = Some(job_id.clone());
#[cfg(feature = "metrics")]
{
@@ -133,7 +149,12 @@ impl MiningService {
metrics::set_active_jobs(*g);
}
}
job.start_mining(self.engine.clone(), self.workers, self.progress_chunk_ms);
job.start_mining(
self.engine.clone(),
self.workers,
self.progress_interval_ms,
self.chunk_size,
);
jobs.insert(job_id, job);
Ok(())
}
@@ -148,7 +169,7 @@ impl MiningService {
if let Some(job) = jobs.get_mut(job_id) {
if !job.result_served {
job.result_served = true;
log::info!(target: "miner", "Result served and marked for job: {job_id}");
log::info!("Mining result served for job: {job_id}");
#[cfg(feature = "metrics")]
{
// reuse existing counter to trace served events
@@ -161,7 +182,7 @@ impl MiningService {
pub async fn remove_job(&self, job_id: &str) -> Option<MiningJob> {
let mut jobs = self.jobs.lock().await;
if let Some(mut job) = jobs.remove(job_id) {
log::debug!(target: "miner", "Removing job: {job_id}");
log::info!("Removing mining job: {job_id}");
job.cancel();
Some(job)
} else {
@@ -182,19 +203,18 @@ impl MiningService {
/// Periodically polls running jobs for results and advances their status.
pub async fn start_mining_loop(&self) {
let jobs = self.jobs.clone();
log::debug!(target: "miner", "Starting mining loop...");
log::info!("🔄 Starting job monitoring loop...");
tokio::spawn(async move {
let mut last_watchdog = std::time::Instant::now();
let mut iter: u64 = 0;
let service_start = std::time::Instant::now();
loop {
iter += 1;
let mut jobs_guard = jobs.lock().await;
jobs_guard.retain(|job_id, job| {
if job.status == JobStatus::Running && job.update_from_results() {
log::debug!(target: "miner",
"Job {} finished with status {:?}, hashes: {}, time: {:?}",
log::info!(
"Mining job {} finished with status {:?}, hashes: {}, time: {:?}",
job_id,
job.status,
job.total_hash_count,
@@ -207,7 +227,7 @@ impl MiningService {
|| (job.status == JobStatus::Completed && !job.result_served)
|| job.start_time.elapsed().as_secs() < 300;
if !retain {
log::debug!(target: "miner", "Cleaning up old job {job_id}");
log::info!("🧹 Cleaning up completed job {}", job_id);
}
retain
});
@@ -229,31 +249,77 @@ impl MiningService {
metrics::set_active_jobs(running_jobs);
}
let do_watchdog = last_watchdog.elapsed().as_secs() >= 30;
let (total, running, completed, failed, cancelled) = if do_watchdog {
let (total, running, completed, failed, cancelled, total_hash_rate) = if do_watchdog
{
let mut running = 0usize;
let mut completed = 0usize;
let mut cancelled = 0usize;
let mut failed = 0usize;
let mut total_hash_rate = 0.0;
let total = jobs_guard.len();
for (_id, job) in jobs_guard.iter() {
match job.status {
JobStatus::Running => running += 1,
JobStatus::Running => {
running += 1;
total_hash_rate += job.last_hash_rate;
}
JobStatus::Completed => completed += 1,
JobStatus::Cancelled => cancelled += 1,
JobStatus::Failed => failed += 1,
}
}
(total, running, completed, failed, cancelled)
(
total,
running,
completed,
failed,
cancelled,
total_hash_rate,
)
} else {
(0, 0, 0, 0, 0)
(0, 0, 0, 0, 0, 0.0)
};
drop(jobs_guard);
if do_watchdog {
log::info!(
target: "miner",
"Watchdog: jobs total={}, running={}, completed={}, failed={}, cancelled={}, loop_iter={}",
total, running, completed, failed, cancelled, iter
);
let uptime = service_start.elapsed();
let uptime_str = if uptime.as_secs() < 60 {
format!("{}s", uptime.as_secs())
} else if uptime.as_secs() < 3600 {
format!("{}m {}s", uptime.as_secs() / 60, uptime.as_secs() % 60)
} else {
format!(
"{}h {}m",
uptime.as_secs() / 3600,
(uptime.as_secs() % 3600) / 60
)
};
if total == 0 {
log::info!(
"📊 Mining service healthy - uptime: {} - waiting for jobs",
uptime_str
);
} else if running == 0 {
log::info!(
"📊 Mining status - uptime: {} - no active jobs - {} completed, {} cancelled, {} failed",
uptime_str, completed, cancelled, failed
);
} else {
let hash_rate_str = if total_hash_rate >= 1_000_000.0 {
format!("{:.1}M", total_hash_rate / 1_000_000.0)
} else if total_hash_rate >= 1_000.0 {
format!("{:.1}K", total_hash_rate / 1_000.0)
} else if total_hash_rate > 0.0 {
format!("{:.0}", total_hash_rate)
} else {
"starting...".to_string()
};
log::info!(
"📊 Mining status - uptime: {} - jobs: {} active, {} completed, {} cancelled, {} failed - hash rate: {} H/s",
uptime_str, running, completed, cancelled, failed, hash_rate_str
);
}
last_watchdog = std::time::Instant::now();
}
tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
@@ -374,7 +440,8 @@ impl MiningJob {
&mut self,
engine: Arc<dyn MinerEngine>,
workers: usize,
progress_chunk_ms: u64,
progress_interval_ms: u64,
chunk_size: Option<u64>,
) {
let chan_capacity = std::env::var("MINER_RESULT_CHANNEL_CAP")
.ok()
@@ -387,8 +454,7 @@ impl MiningJob {
// Partition nonce range safely
let partitions = compute_partitions(self.nonce_start, self.nonce_end, workers);
log::debug!(
target: "miner",
log::info!(
"Starting mining with {} workers, total range: {} ({} partitions)",
workers,
partitions.total_range,
@@ -414,7 +480,8 @@ impl MiningJob {
EngineRange { start, end },
cancel_flag,
sender,
progress_chunk_ms,
progress_interval_ms,
chunk_size,
);
});
@@ -423,7 +490,10 @@ impl MiningJob {
}
pub fn cancel(&mut self) {
log::debug!(target: "miner", "Cancelling mining job: {}", self.job_id.as_ref().unwrap());
log::info!(
"Cancelling mining job: {}",
self.job_id.as_ref().unwrap_or(&"unknown".to_string())
);
self.cancel_flag.store(true, Ordering::Relaxed);
self.status = JobStatus::Cancelled;
#[cfg(feature = "metrics")]
@@ -637,10 +707,10 @@ fn mine_range_with_engine(
range: EngineRange,
cancel_flag: Arc<AtomicBool>,
sender: Sender<ThreadResult>,
progress_chunk_ms: u64,
progress_interval_ms: u64,
chunk_size: Option<u64>,
) {
log::debug!(
target: "miner",
log::info!(
"Job {} thread {} mining range {} to {} (inclusive)",
job_id,
thread_id,
@@ -652,11 +722,19 @@ fn mine_range_with_engine(
let mut current_start = range.start;
let end = range.end;
// Derive a rough chunk size from target milliseconds and an estimate of hashes/sec.
// Start with a conservative default of 100k ops/sec per thread and scale by target ms.
let target_ms = progress_chunk_ms;
let est_ops_per_sec: u64 = 100_000;
let derived_chunk: u64 = ((est_ops_per_sec.saturating_mul(target_ms)) / 1000).max(5_000);
let chunk_size = U512::from(derived_chunk);
let target_ms = progress_interval_ms;
let chunk_size_u64 = chunk_size.unwrap_or_else(|| {
// Use engine-specific defaults if not configured
if engine.name().contains("gpu") {
// GPU can handle much larger chunks efficiently
100_000_000 // 100M hashes per chunk
} else {
// CPU uses time-based chunks
let est_ops_per_sec = 100_000u64; // 100K ops/sec for CPU
((est_ops_per_sec.saturating_mul(target_ms)) / 1000).max(5_000)
}
});
let chunk_size = U512::from(chunk_size_u64);
let mut done = false;
while current_start <= end && !cancel_flag.load(Ordering::Relaxed) {
@@ -673,7 +751,6 @@ fn mine_range_with_engine(
};
log::debug!(
target: "miner",
"Job {} thread {} processing subrange {}..{} (inclusive)",
job_id,
thread_id,
@@ -696,8 +773,19 @@ fn mine_range_with_engine(
origin: Some(origin),
completed: true,
};
log::info!(
"🎉 Solution found! Job {} thread {} - Nonce: {}, Hash: {:x}",
job_id,
thread_id,
nonce,
hash
);
if sender.try_send(final_result).is_err() {
log::warn!(target: "miner", "Job {job_id} thread {thread_id} failed to send final result");
log::warn!(
"Job {} thread {} failed to send final result",
job_id,
thread_id
);
}
done = true;
break;
@@ -712,12 +800,15 @@ fn mine_range_with_engine(
completed: false,
};
if sender.try_send(update).is_err() {
log::warn!(target: "miner", "Job {job_id} thread {thread_id} failed to send progress update");
log::warn!(
"Job {} thread {} failed to send progress update",
job_id,
thread_id
);
break;
} else {
log::debug!(
target: "miner",
"Job {} thread {} progress: hashed={} in subrange {}..{}",
log::info!(
"⛏️ Job {} thread {} processed {} hashes (range: {}..{})",
job_id,
thread_id,
hash_count,
@@ -901,14 +992,40 @@ pub async fn handle_result_request(
}
};
log::debug!(
target: "miner",
"Result polling watchdog: job_id={}, status={:?}, result_served={}, elapsed_s={:.3}",
job_id,
job.status,
job.result_served,
job.start_time.elapsed().as_secs_f64()
);
let elapsed = job.start_time.elapsed().as_secs_f64();
match job.status {
JobStatus::Running => {
log::debug!(
"🔍 Job {} still running - elapsed: {:.1}s - hash rate: {:.0} H/s",
job_id,
elapsed,
job.last_hash_rate
);
}
JobStatus::Completed if !job.result_served => {
log::info!(
"✅ Job {} completed - ready for pickup - elapsed: {:.1}s - {} hashes",
job_id,
elapsed,
job.total_hash_count
);
}
JobStatus::Completed => {
log::debug!(
"📤 Job {} result already served - elapsed: {:.1}s",
job_id,
elapsed
);
}
_ => {
log::debug!(
"🔄 Job {} status: {:?} - elapsed: {:.1}s",
job_id,
job.status,
elapsed
);
}
}
let status = match job.status {
JobStatus::Running => ApiResponseStatus::Running,
JobStatus::Completed => ApiResponseStatus::Completed,
@@ -1160,7 +1277,7 @@ pub async fn run(config: ServiceConfig) -> anyhow::Result<()> {
if workers > effective_cpus {
workers = effective_cpus.max(1);
}
// Force workers=1 for GPU engine to prevent concurrent buffer mapping panics
if matches!(config.engine, EngineSelection::Gpu) {
if workers > 1 {
@@ -1172,7 +1289,6 @@ pub async fn run(config: ServiceConfig) -> anyhow::Result<()> {
}
}
log::info!(
"Using {workers} worker thread(s) for mining (effective logical CPUs available: {effective_cpus})"
);
@@ -1218,11 +1334,34 @@ pub async fn run(config: ServiceConfig) -> anyhow::Result<()> {
}
}
};
log::info!("Using engine: {}", engine.name());
log::info!("Service configuration: {config}");
log::info!(
"🚀 Mining engine selected: {} ({})",
engine.name(),
match config.engine {
EngineSelection::CpuBaseline => "Single-threaded CPU reference implementation",
EngineSelection::CpuFast => "Optimized multi-threaded CPU mining",
EngineSelection::CpuChainManipulator => "Throttled CPU mining for difficulty control",
EngineSelection::Gpu => "High-performance GPU mining",
}
);
log::info!("⚙️ Service configuration: {config}");
let progress_chunk_ms = config.progress_chunk_ms.unwrap_or(2000);
let service = MiningService::new(workers, engine.clone(), progress_chunk_ms);
let progress_interval_ms = config.progress_interval_ms.unwrap_or(10000);
let service = MiningService::new(
workers,
engine.clone(),
progress_interval_ms,
config.chunk_size,
);
log::info!("⛏️ Mining service ready with {} worker threads", workers);
if let Some(chunk_size) = config.chunk_size {
log::info!(
"📦 Custom chunk size: {} hashes per progress update",
chunk_size
);
}
log::info!("⏰ Progress reporting interval: {}ms", progress_interval_ms);
// Start mining loop
service.start_mining_loop().await;
@@ -1370,7 +1509,20 @@ pub async fn run(config: ServiceConfig) -> anyhow::Result<()> {
// Start server
let addr = ([0, 0, 0, 0], config.port);
let socket = std::net::SocketAddr::from(addr);
log::info!("Server starting on {socket}");
log::info!("🌐 HTTP API server starting on http://{}", socket);
log::info!("📡 Mining endpoints available:");
log::info!(" POST http://{}/mine - Submit mining jobs", socket);
log::info!(
" GET http://{}/result/{{job_id}} - Check mining results",
socket
);
if config.metrics_port.is_some() {
log::info!(
" GET http://{}:{}/metrics - Prometheus metrics",
socket.ip(),
config.metrics_port.unwrap()
);
}
warp::serve(routes).run(socket).await;
Ok(())
@@ -1388,7 +1540,7 @@ mod tests {
#[tokio::test]
async fn test_mining_state_add_get_remove() {
let engine: Arc<dyn MinerEngine> = Arc::new(engine_cpu::BaselineCpuEngine::new());
let state = MiningService::new(2, engine, 2000);
let state = MiningService::new(2, engine, 2000, None);
let job = MiningJob::new(
[1u8; 32],
@@ -1407,7 +1559,7 @@ mod tests {
// Test that a job fails if no nonce is found (threshold too strict).
// To make this deterministic, avoid nonce=0, which some math paths treat as special.
let engine: Arc<dyn MinerEngine> = Arc::new(engine_cpu::BaselineCpuEngine::new());
let state = MiningService::new(2, engine, 2000);
let state = MiningService::new(1, engine, 2000, None);
state.start_mining_loop().await;
// Impossible difficulty with a nonce range that excludes 0
@@ -1505,7 +1657,7 @@ mod tests {
async fn test_job_lifecycle_success() {
// Test that a job completes successfully
let engine: Arc<dyn MinerEngine> = Arc::new(engine_cpu::BaselineCpuEngine::new());
let state = MiningService::new(2, engine, 2000);
let state = MiningService::new(2, engine, 2000, None);
state.start_mining_loop().await;
// Easy difficulty
@@ -1629,7 +1781,7 @@ mod tests {
let engine: Arc<dyn engine_cpu::MinerEngine> =
Arc::new(engine_cpu::BaselineCpuEngine::new());
let service = super::MiningService::new(2, engine, 2000);
let service = MiningService::new(2, engine, 1000, None);
service.start_mining_loop().await;
// Build routes
@@ -1706,7 +1858,8 @@ mod tests {
range,
cancel,
tx,
10, // ms (min chunk size is 5k; fine for testing progress + completion)
10, // ms (min chunk size is 5k; fine for testing progress + completion)
None, // chunk_size
);
// Expect at least two messages: one progress (completed=false) and one final completion (completed=true)