bump version to v3.1.0 (#59)

Co-authored-by: illuzen <illuzen@users.noreply.github.com>

Cargo.lock

@@ -571,7 +571,7 @@ dependencies = [
 
 [[package]]
 name = "engine-cpu"
-version = "3.0.0"
+version = "3.1.0"
 dependencies = [
  "criterion",
  "hex",
@@ -582,7 +582,7 @@ dependencies = [
 
 [[package]]
 name = "engine-gpu"
-version = "3.0.0"
+version = "3.1.0"
 dependencies = [
  "bytemuck",
  "criterion",
@@ -1458,7 +1458,7 @@ dependencies = [
 
 [[package]]
 name = "metrics"
-version = "3.0.0"
+version = "3.1.0"
 dependencies = [
  "anyhow",
  "log",
@@ -1486,7 +1486,7 @@ dependencies = [
 
 [[package]]
 name = "miner-cli"
-version = "3.0.0"
+version = "3.1.0"
 dependencies = [
  "clap",
  "engine-cpu",
@@ -1503,7 +1503,7 @@ dependencies = [
 
 [[package]]
 name = "miner-service"
-version = "3.0.0"
+version = "3.1.0"
 dependencies = [
  "anyhow",
  "crossbeam-channel",
@@ -1524,7 +1524,7 @@ dependencies = [
 
 [[package]]
 name = "miner-telemetry"
-version = "3.0.0"
+version = "3.1.0"
 dependencies = [
  "anyhow",
  "futures",
@@ -2008,7 +2008,7 @@ dependencies = [
 
 [[package]]
 name = "pow-core"
-version = "3.0.0"
+version = "3.1.0"
 dependencies = [
  "hex",
  "primitive-types 0.13.1",

Cargo.toml

@@ -15,7 +15,7 @@ resolver = "2"
 edition = "2021"
 authors = ["Quantus Network"]
 description = "Quantus External Miner Workspace"
-version = "3.0.0"
+version = "3.1.0"
 
 [workspace.dependencies]
 anyhow = "1"

crates/engine-cpu/benches/cpu_engine_bench.rs

@@ -1,14 +1,15 @@
 use criterion::{black_box, criterion_group, criterion_main, Criterion};
-use engine_cpu::{FastCpuEngine, MinerEngine, Range};
+use engine_cpu::{AtomicBoolCancelCheck, FastCpuEngine, MinerEngine, Range};
 use pow_core::{hash_from_nonce, JobContext};
 use primitive_types::U512;
 use rand::RngCore;
 use std::sync::atomic::AtomicBool;
 
 fn bench_cpu_fast_engine(c: &mut Criterion) {
-    // Create the engine
-    let engine = FastCpuEngine::new();
+    // Create the engine with batch size of 10000
+    let engine = FastCpuEngine::new(10_000);
     let cancel_flag = AtomicBool::new(false);
+    let cancel_check = AtomicBoolCancelCheck(&cancel_flag);
 
     let large_range = Range {
         start: U512::from(0u64),
@@ -25,7 +26,7 @@ fn bench_cpu_fast_engine(c: &mut Criterion) {
             let result = engine.search_range(
                 black_box(&ctx),
                 black_box(large_range.clone()),
-                black_box(&cancel_flag),
+                black_box(&cancel_check),
             );
             black_box(result)
         })

crates/engine-cpu/src/lib.rs

@@ -8,7 +8,7 @@
 
 use pow_core::JobContext;
 use primitive_types::U512;
-use std::sync::atomic::{AtomicBool, Ordering as AtomicOrdering};
+use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
 
 /// An inclusive nonce range to search.
 #[derive(Clone, Debug)]
@@ -53,6 +53,36 @@ pub enum EngineStatus {
     },
 }
 
+/// Cancellation checker passed to search_range.
+/// Returns true if the search should be cancelled.
+pub trait CancelCheck: Send + Sync {
+    fn is_cancelled(&self) -> bool;
+}
+
+/// Simple cancel check using an AtomicBool flag.
+/// Useful for benchmarks and simple scenarios.
+pub struct AtomicBoolCancelCheck<'a>(pub &'a AtomicBool);
+
+impl CancelCheck for AtomicBoolCancelCheck<'_> {
+    fn is_cancelled(&self) -> bool {
+        self.0.load(Ordering::Relaxed)
+    }
+}
+
+/// Cancel check using job ID comparison.
+/// When a new job starts, the current_job_id is incremented.
+/// Workers compare their job_id against the current to detect cancellation.
+pub struct JobIdCancelCheck<'a> {
+    pub current_job_id: &'a AtomicU64,
+    pub my_job_id: u64,
+}
+
+impl CancelCheck for JobIdCancelCheck<'_> {
+    fn is_cancelled(&self) -> bool {
+        self.current_job_id.load(Ordering::Relaxed) != self.my_job_id
+    }
+}
+
 /// Abstract mining engine interface.
 ///
 /// The service layer depends only on this trait to manage jobs.
@@ -65,19 +95,28 @@ pub trait MinerEngine: Send + Sync {
     fn prepare_context(&self, header_hash: [u8; 32], difficulty: U512) -> JobContext;
 
     /// Search an inclusive nonce range with cancellation support.
-    fn search_range(&self, ctx: &JobContext, range: Range, cancel: &AtomicBool) -> EngineStatus;
+    fn search_range(
+        &self,
+        ctx: &JobContext,
+        range: Range,
+        cancel: &dyn CancelCheck,
+    ) -> EngineStatus;
 
     /// Enable downcasting to concrete engine types.
     fn as_any(&self) -> &dyn std::any::Any;
 }
 
 /// Fast CPU engine using optimized pow-core helpers.
-#[derive(Default)]
-pub struct FastCpuEngine;
+pub struct FastCpuEngine {
+    /// How often to check for cancellation (in hashes)
+    batch_size: u64,
+}
 
 impl FastCpuEngine {
-    pub fn new() -> Self {
-        Self
+    pub fn new(batch_size: u64) -> Self {
+        Self {
+            batch_size: batch_size.max(1), // Ensure at least 1
+        }
     }
 }
 
@@ -94,7 +133,12 @@ impl MinerEngine for FastCpuEngine {
         self
     }
 
-    fn search_range(&self, ctx: &JobContext, range: Range, cancel: &AtomicBool) -> EngineStatus {
+    fn search_range(
+        &self,
+        ctx: &JobContext,
+        range: Range,
+        cancel: &dyn CancelCheck,
+    ) -> EngineStatus {
         use pow_core::{hash_from_nonce, is_valid_hash, step_nonce};
 
         if range.start > range.end {
@@ -103,11 +147,19 @@ impl MinerEngine for FastCpuEngine {
 
         let mut current = range.start;
         let mut hash_count: u64 = 0;
+        // Use decrementing counter to avoid modulo division in hot loop
+        // Initialize to 0 so we check cancellation immediately on first iteration
+        let mut until_check: u64 = 0;
 
         loop {
-            if cancel.load(AtomicOrdering::Relaxed) {
-                return EngineStatus::Cancelled { hash_count };
+            // Check for cancellation every batch_size hashes
+            if until_check == 0 {
+                if cancel.is_cancelled() {
+                    return EngineStatus::Cancelled { hash_count };
+                }
+                until_check = self.batch_size;
             }
+            until_check -= 1;
 
             let hash = hash_from_nonce(ctx, current);
             hash_count = hash_count.saturating_add(1);
@@ -157,9 +209,10 @@ mod tests {
         };
 
         let cancel = AtomicBool::new(false);
-        let engine = FastCpuEngine::new();
+        let cancel_check = AtomicBoolCancelCheck(&cancel);
+        let engine = FastCpuEngine::new(1000); // check every 1000 hashes
 
-        let status = engine.search_range(&ctx, range.clone(), &cancel);
+        let status = engine.search_range(&ctx, range.clone(), &cancel_check);
         match status {
             EngineStatus::Exhausted { hash_count } => {
                 let expected = (range.end - range.start + U512::one()).as_u64();
@@ -181,9 +234,10 @@ mod tests {
         };
 
         let cancel = AtomicBool::new(true); // pre-cancelled
-        let engine = FastCpuEngine::new();
+        let cancel_check = AtomicBoolCancelCheck(&cancel);
+        let engine = FastCpuEngine::new(1); // check every hash for immediate cancellation
 
-        let status = engine.search_range(&ctx, range, &cancel);
+        let status = engine.search_range(&ctx, range, &cancel_check);
         match status {
             EngineStatus::Cancelled { hash_count } => {
                 assert_eq!(hash_count, 0);

crates/engine-gpu/benches/gpu_engine_bench.rs

@@ -1,5 +1,5 @@
 use criterion::{black_box, criterion_group, criterion_main, Criterion};
-use engine_cpu::{FastCpuEngine, MinerEngine, Range};
+use engine_cpu::{AtomicBoolCancelCheck, FastCpuEngine, MinerEngine, Range};
 use engine_gpu::GpuEngine;
 use pow_core::JobContext;
 use primitive_types::U512;
@@ -7,9 +7,10 @@ use rand::RngCore;
 use std::sync::atomic::AtomicBool;
 
 fn bench_cpu_vs_gpu_small(c: &mut Criterion) {
-    let cpu_engine = FastCpuEngine::new();
-    let gpu_engine = GpuEngine::new();
+    let cpu_engine = FastCpuEngine::new(10_000);
+    let gpu_engine = GpuEngine::try_new(10_000_000).expect("Failed to init GPU");
     let cancel_flag = AtomicBool::new(false);
+    let cancel_check = AtomicBoolCancelCheck(&cancel_flag);
 
     // Small range: 10K nonces - reasonable for benchmarking
     let small_range = Range {
@@ -31,7 +32,7 @@ fn bench_cpu_vs_gpu_small(c: &mut Criterion) {
             let result = cpu_engine.search_range(
                 black_box(&ctx),
                 black_box(small_range.clone()),
-                black_box(&cancel_flag),
+                black_box(&cancel_check),
             );
             black_box(result)
         })
@@ -47,7 +48,7 @@ fn bench_cpu_vs_gpu_small(c: &mut Criterion) {
             let result = gpu_engine.search_range(
                 black_box(&ctx),
                 black_box(small_range.clone()),
-                black_box(&cancel_flag),
+                black_box(&cancel_check),
             );
             black_box(result)
         })
@@ -57,9 +58,10 @@ fn bench_cpu_vs_gpu_small(c: &mut Criterion) {
 }
 
 fn bench_cpu_vs_gpu_medium(c: &mut Criterion) {
-    let cpu_engine = FastCpuEngine::new();
-    let gpu_engine = GpuEngine::new();
+    let cpu_engine = FastCpuEngine::new(10_000);
+    let gpu_engine = GpuEngine::try_new(10_000_000).expect("Failed to init GPU");
     let cancel_flag = AtomicBool::new(false);
+    let cancel_check = AtomicBoolCancelCheck(&cancel_flag);
 
     // Medium range: 100K nonces
     let medium_range = Range {
@@ -81,7 +83,7 @@ fn bench_cpu_vs_gpu_medium(c: &mut Criterion) {
             let result = cpu_engine.search_range(
                 black_box(&ctx),
                 black_box(medium_range.clone()),
-                black_box(&cancel_flag),
+                black_box(&cancel_check),
             );
             black_box(result)
         })
@@ -97,7 +99,7 @@ fn bench_cpu_vs_gpu_medium(c: &mut Criterion) {
             let result = gpu_engine.search_range(
                 black_box(&ctx),
                 black_box(medium_range.clone()),
-                black_box(&cancel_flag),
+                black_box(&cancel_check),
             );
             black_box(result)
         })
@@ -107,9 +109,10 @@ fn bench_cpu_vs_gpu_medium(c: &mut Criterion) {
 }
 
 fn bench_cpu_vs_gpu_large(c: &mut Criterion) {
-    let cpu_engine = FastCpuEngine::new();
-    let gpu_engine = GpuEngine::new();
+    let cpu_engine = FastCpuEngine::new(10_000);
+    let gpu_engine = GpuEngine::try_new(10_000_000).expect("Failed to init GPU");
     let cancel_flag = AtomicBool::new(false);
+    let cancel_check = AtomicBoolCancelCheck(&cancel_flag);
 
     // Large range: 1M nonces - where GPU should really shine
     let large_range = Range {
@@ -131,7 +134,7 @@ fn bench_cpu_vs_gpu_large(c: &mut Criterion) {
             let result = cpu_engine.search_range(
                 black_box(&ctx),
                 black_box(large_range.clone()),
-                black_box(&cancel_flag),
+                black_box(&cancel_check),
             );
             black_box(result)
         })
@@ -147,7 +150,7 @@ fn bench_cpu_vs_gpu_large(c: &mut Criterion) {
             let result = gpu_engine.search_range(
                 black_box(&ctx),
                 black_box(large_range.clone()),
-                black_box(&cancel_flag),
+                black_box(&cancel_check),
             );
             black_box(result)
         })
@@ -157,9 +160,10 @@ fn bench_cpu_vs_gpu_large(c: &mut Criterion) {
 }
 
 fn bench_solution_finding(c: &mut Criterion) {
-    let cpu_engine = FastCpuEngine::new();
-    let gpu_engine = GpuEngine::new();
+    let cpu_engine = FastCpuEngine::new(10_000);
+    let gpu_engine = GpuEngine::try_new(10_000_000).expect("Failed to init GPU");
     let cancel_flag = AtomicBool::new(false);
+    let cancel_check = AtomicBoolCancelCheck(&cancel_flag);
 
     // Range where we expect to find solutions quickly
     let solution_range = Range {
@@ -181,7 +185,7 @@ fn bench_solution_finding(c: &mut Criterion) {
             let result = cpu_engine.search_range(
                 black_box(&ctx),
                 black_box(solution_range.clone()),
-                black_box(&cancel_flag),
+                black_box(&cancel_check),
             );
             black_box(result)
         })
@@ -197,7 +201,7 @@ fn bench_solution_finding(c: &mut Criterion) {
             let result = gpu_engine.search_range(
                 black_box(&ctx),
                 black_box(solution_range.clone()),
-                black_box(&cancel_flag),
+                black_box(&cancel_check),
             );
             black_box(result)
         })
@@ -207,9 +211,10 @@ fn bench_solution_finding(c: &mut Criterion) {
 }
 
 fn bench_throughput_per_second(c: &mut Criterion) {
-    let cpu_engine = FastCpuEngine::new();
-    let gpu_engine = GpuEngine::new();
+    let cpu_engine = FastCpuEngine::new(10_000);
+    let gpu_engine = GpuEngine::try_new(10_000_000).expect("Failed to init GPU");
     let cancel_flag = AtomicBool::new(false);
+    let cancel_check = AtomicBoolCancelCheck(&cancel_flag);
 
     // Fixed time benchmark - see how many hashes we can do in 1 second
     let throughput_range = Range {
@@ -231,7 +236,7 @@ fn bench_throughput_per_second(c: &mut Criterion) {
             let result = cpu_engine.search_range(
                 black_box(&ctx),
                 black_box(throughput_range.clone()),
-                black_box(&cancel_flag),
+                black_box(&cancel_check),
             );
             black_box(result)
         })
@@ -247,7 +252,7 @@ fn bench_throughput_per_second(c: &mut Criterion) {
             let result = gpu_engine.search_range(
                 black_box(&ctx),
                 black_box(throughput_range.clone()),
-                black_box(&cancel_flag),
+                black_box(&cancel_check),
             );
             black_box(result)
         })
@@ -257,8 +262,9 @@ fn bench_throughput_per_second(c: &mut Criterion) {
 }
 
 fn bench_gpu_batch_efficiency(c: &mut Criterion) {
-    let gpu_engine = GpuEngine::new();
+    let gpu_engine = GpuEngine::try_new(10_000_000).expect("Failed to init GPU");
     let cancel_flag = AtomicBool::new(false);
+    let cancel_check = AtomicBoolCancelCheck(&cancel_flag);
 
     let mut group = c.benchmark_group("gpu_batch_sizes");
     group.sample_size(10);
@@ -290,7 +296,7 @@ fn bench_gpu_batch_efficiency(c: &mut Criterion) {
             let result = gpu_engine.search_range(
                 black_box(&ctx),
                 black_box(small_batch.clone()),
-                black_box(&cancel_flag),
+                black_box(&cancel_check),
             );
             black_box(result)
         })
@@ -306,7 +312,7 @@ fn bench_gpu_batch_efficiency(c: &mut Criterion) {
             let result = gpu_engine.search_range(
                 black_box(&ctx),
                 black_box(medium_batch.clone()),
-                black_box(&cancel_flag),
+                black_box(&cancel_check),
             );
             black_box(result)
         })
@@ -322,7 +328,7 @@ fn bench_gpu_batch_efficiency(c: &mut Criterion) {
             let result = gpu_engine.search_range(
                 black_box(&ctx),
                 black_box(large_batch.clone()),
-                black_box(&cancel_flag),
+                black_box(&cancel_check),
             );
             black_box(result)
         })

crates/engine-gpu/examples/verify_nonce.rs

@@ -1,4 +1,4 @@
-use engine_cpu::{EngineStatus, FastCpuEngine, MinerEngine, Range};
+use engine_cpu::{AtomicBoolCancelCheck, EngineStatus, FastCpuEngine, MinerEngine, Range};
 use engine_gpu::GpuEngine;
 use primitive_types::U512;
 use std::sync::atomic::AtomicBool;
@@ -16,16 +16,17 @@ fn main() {
     // Use a fixed header and easy difficulty (1) so any nonce is valid
     let header = [1u8; 32];
     let difficulty = U512::from(u64::MAX); // High difficulty - no solutions expected
-    let cpu_engine = FastCpuEngine::new();
+    let cpu_engine = FastCpuEngine::new(10_000);
     let ctx = cpu_engine.prepare_context(header, difficulty);
 
     log::info!("Context prepared. Difficulty: {}", difficulty);
 
     let cancel = AtomicBool::new(false);
+    let cancel_check = AtomicBoolCancelCheck(&cancel);
 
     // 3. Verify with GPU engine
     log::info!("Initializing GPU engine...");
-    let gpu_engine = GpuEngine::new();
+    let gpu_engine = GpuEngine::try_new(10_000_000).expect("Failed to init GPU");
 
     // Search a small range around the valid nonce
     let gpu_range = Range {
@@ -39,7 +40,7 @@ fn main() {
         gpu_range.end
     );
     let start = std::time::Instant::now();
-    let gpu_result = gpu_engine.search_range(&ctx, gpu_range, &cancel);
+    let gpu_result = gpu_engine.search_range(&ctx, gpu_range, &cancel_check);
     let elapsed = start.elapsed();
 
     log::info!("GPU search took {:?}", elapsed);

crates/engine-gpu/src/mining.wgsl

@@ -194,8 +194,7 @@ const MDS_MATRIX_DIAG_12: array<array<u32, 2>, 12> = array<array<u32, 2>, 12>(
 @group(0) @binding(1) var<storage, read> header: array<u32, 8>;     // 32 bytes
 @group(0) @binding(2) var<storage, read> start_nonce: array<u32, 16>; // 64 bytes
 @group(0) @binding(3) var<storage, read> difficulty_target: array<u32, 16>;    // 64 bytes (U512 target)
-@group(0) @binding(4) var<storage, read> dispatch_config: array<u32, 4>; // [total_threads, nonces_per_thread, total_nonces, cancel_check_interval]
-@group(0) @binding(5) var<storage, read> cancel_flag: array<u32, 1>; // 0 = running, 1 = cancel requested
+@group(0) @binding(4) var<storage, read> dispatch_config: array<u32, 3>; // [total_threads, nonces_per_thread, total_nonces]
 
 // Goldilocks field element represented as [limb0, limb1]
 // where the value is limb0 + limb1*2^32
@@ -804,20 +803,16 @@ fn is_below_target(hash: array<u32, 16>, difficulty_tgt: array<u32, 16>) -> bool
 // Main mining kernel
 @compute @workgroup_size(256)
 fn mining_main(@builtin(global_invocation_id) global_id: vec3<u32>) {
-    // If solution already found or cancelled, exit early
+    // If solution already found, exit early
     if (atomicLoad(&results[0]) != 0u) {
         return;
     }
-    if (cancel_flag[0] != 0u) {
-        return;
-    }
 
     let thread_id = global_id.x;
     // Read dispatch configuration from buffer
     let total_threads = dispatch_config[0];        // Total logical threads in this dispatch
     let nonces_per_thread = dispatch_config[1];    // Nonces processed by each thread
     let total_nonces = dispatch_config[2];         // Total logical nonces this dispatch should cover
-    let cancel_check_interval = dispatch_config[3]; // How often to check cancel flag
 
     // Guard against threads beyond configured total_threads
     if (thread_id >= total_threads) {
@@ -834,14 +829,6 @@ fn mining_main(@builtin(global_invocation_id) global_id: vec3<u32>) {
             break;
         }
 
-        // Periodic checks for early exit
-        // Check cancel flag every cancel_check_interval iterations
-        if (cancel_check_interval > 0u && (j % cancel_check_interval) == 0u) {
-            if (cancel_flag[0] != 0u) {
-                return;
-            }
-        }
-
         // Check if solution already found (early exit for entire dispatch)
         if (atomicLoad(&results[0]) != 0u) {
             return;

crates/miner-cli/src/main.rs

@@ -1,5 +1,5 @@
 use clap::{Parser, Subcommand};
-use engine_cpu::{EngineRange, MinerEngine};
+use engine_cpu::{AtomicBoolCancelCheck, EngineRange, MinerEngine};
 use miner_service::{run, ServiceConfig};
 use primitive_types::U512;
 use rand::RngCore;
@@ -8,12 +8,16 @@ use std::sync::Arc;
 use std::thread;
 use std::time::{Duration, Instant};
 
+// CLI defaults
+const DEFAULT_GPU_BATCH_SIZE: u64 = 1_000_000;
+const DEFAULT_CPU_BATCH_SIZE: u64 = 10_000;
+
 #[derive(Subcommand, Debug)]
 enum Command {
     /// Run the mining service
     Serve {
-        /// Address of the node to connect to (e.g., "127.0.0.1:9833")
-        #[arg(long, env = "MINER_NODE_ADDR")]
+        /// Address of the node to connect to
+        #[arg(long, env = "MINER_NODE_ADDR", default_value = "127.0.0.1:9833")]
         node_addr: std::net::SocketAddr,
 
         /// Number of CPU worker threads to use for mining (default: auto-detect)
@@ -24,9 +28,13 @@ enum Command {
         #[arg(long = "gpu-devices", env = "MINER_GPU_DEVICES")]
         gpu_devices: Option<usize>,
 
-        /// GPU cancel check interval in nonces (default: 10000)
-        #[arg(long = "gpu-cancel-interval", env = "MINER_GPU_CANCEL_INTERVAL")]
-        gpu_cancel_interval: Option<u32>,
+        /// GPU batch size in nonces - controls how often GPU checks for cancellation
+        #[arg(long = "gpu-batch-size", env = "MINER_GPU_BATCH_SIZE", default_value_t = DEFAULT_GPU_BATCH_SIZE)]
+        gpu_batch_size: u64,
+
+        /// CPU batch size in hashes - controls how often CPU checks for cancellation
+        #[arg(long = "cpu-batch-size", env = "MINER_CPU_BATCH_SIZE", default_value_t = DEFAULT_CPU_BATCH_SIZE)]
+        cpu_batch_size: u64,
 
         /// Port for Prometheus metrics HTTP endpoint (default: 9900)
         #[arg(
@@ -51,6 +59,14 @@ enum Command {
         #[arg(long = "gpu-devices", env = "MINER_GPU_DEVICES")]
         gpu_devices: Option<usize>,
 
+        /// GPU batch size in nonces - controls how often GPU checks for cancellation
+        #[arg(long = "gpu-batch-size", env = "MINER_GPU_BATCH_SIZE", default_value_t = DEFAULT_GPU_BATCH_SIZE)]
+        gpu_batch_size: u64,
+
+        /// CPU batch size in hashes - controls how often CPU checks for cancellation
+        #[arg(long = "cpu-batch-size", env = "MINER_CPU_BATCH_SIZE", default_value_t = DEFAULT_CPU_BATCH_SIZE)]
+        cpu_batch_size: u64,
+
         /// Benchmark duration in seconds (default: 10)
         #[arg(short, long, default_value_t = 10)]
         duration: u64,
@@ -74,7 +90,7 @@ async fn main() {
     let args = Args::parse();
 
     let Some(command) = args.command else {
-        eprintln!("Error: No command provided. Use 'serve --node-addr <ADDRESS>' to start mining.");
+        eprintln!("Error: No command provided. Use 'serve' to start mining (defaults to local node at 127.0.0.1:9833).");
         eprintln!("Example: quantus-miner serve --node-addr 127.0.0.1:9833");
         std::process::exit(1);
     };
@@ -84,7 +100,8 @@ async fn main() {
             node_addr,
             cpu_workers,
             gpu_devices,
-            gpu_cancel_interval,
+            gpu_batch_size,
+            cpu_batch_size,
             metrics_port,
             verbose,
         } => {
@@ -106,7 +123,8 @@ async fn main() {
                 node_addr,
                 cpu_workers,
                 gpu_devices,
-                gpu_cancel_interval,
+                gpu_batch_size,
+                cpu_batch_size,
             };
 
             if let Err(e) = run(config).await {
@@ -118,11 +136,20 @@ async fn main() {
         Command::Benchmark {
             cpu_workers,
             gpu_devices,
+            gpu_batch_size,
+            cpu_batch_size,
             duration,
             verbose,
         } => {
             init_logger(verbose);
-            run_benchmark(cpu_workers, gpu_devices, duration).await;
+            run_benchmark(
+                cpu_workers,
+                gpu_devices,
+                gpu_batch_size,
+                cpu_batch_size,
+                duration,
+            )
+            .await;
         }
     }
 }
@@ -139,12 +166,18 @@ fn init_logger(verbose: bool) {
     env_logger::init();
 }
 
-async fn run_benchmark(cpu_workers: Option<usize>, gpu_devices: Option<usize>, duration: u64) {
+async fn run_benchmark(
+    cpu_workers: Option<usize>,
+    gpu_devices: Option<usize>,
+    gpu_batch_size: u64,
+    cpu_batch_size: u64,
+    duration: u64,
+) {
     let effective_cpu_workers = cpu_workers.unwrap_or_else(num_cpus::get);
 
     // Initialize GPU engine
     let (gpu_engine, effective_gpu_devices) =
-        match miner_service::resolve_gpu_configuration(gpu_devices, None) {
+        match miner_service::resolve_gpu_configuration(gpu_devices, gpu_batch_size) {
             Ok((engine, count)) => (engine, count),
             Err(e) => {
                 eprintln!("❌ ERROR: {}", e);
@@ -172,7 +205,7 @@ async fn run_benchmark(cpu_workers: Option<usize>, gpu_devices: Option<usize>, d
 
     // Create CPU engine
     let cpu_engine: Option<Arc<dyn MinerEngine>> = if effective_cpu_workers > 0 {
-        Some(Arc::new(engine_cpu::FastCpuEngine::new()))
+        Some(Arc::new(engine_cpu::FastCpuEngine::new(cpu_batch_size)))
     } else {
         None
     };
@@ -224,7 +257,8 @@ async fn run_benchmark(cpu_workers: Option<usize>, gpu_devices: Option<usize>, d
                     break;
                 }
 
-                let result = engine.search_range(&ctx, worker_range.clone(), &cancel);
+                let cancel_check = AtomicBoolCancelCheck(&cancel);
+                let result = engine.search_range(&ctx, worker_range.clone(), &cancel_check);
 
                 match result {
                     engine_cpu::EngineStatus::Found { hash_count, .. }

crates/miner-service/src/lib.rs

@@ -10,11 +10,11 @@
 
 pub mod quic;
 
-use crossbeam_channel::{bounded, Receiver, Sender};
-use engine_cpu::{EngineCandidate, EngineRange, MinerEngine};
-use pow_core::format_u512;
+use crossbeam_channel::{bounded, unbounded, Receiver, Sender};
+use engine_cpu::{EngineCandidate, EngineRange, JobIdCancelCheck, MinerEngine};
+use pow_core::{format_hashrate, format_u512};
 use primitive_types::U512;
-use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
+use std::sync::atomic::{AtomicU64, Ordering};
 use std::sync::Arc;
 use std::thread;
 
@@ -27,8 +27,10 @@ pub struct ServiceConfig {
     pub cpu_workers: Option<usize>,
     /// Number of GPU devices to use for mining (None = auto-detect)
     pub gpu_devices: Option<usize>,
-    /// GPU cancel check interval in nonces (None = use default of 10,000)
-    pub gpu_cancel_interval: Option<u32>,
+    /// GPU batch size in nonces
+    pub gpu_batch_size: u64,
+    /// CPU batch size in hashes
+    pub cpu_batch_size: u64,
 }
 
 /// Engine type for tracking metrics per compute type.
@@ -38,17 +40,6 @@ pub enum EngineType {
     Gpu,
 }
 
-impl Default for ServiceConfig {
-    fn default() -> Self {
-        Self {
-            node_addr: "127.0.0.1:9833".parse().unwrap(),
-            cpu_workers: None,
-            gpu_devices: None,
-            gpu_cancel_interval: None,
-        }
-    }
-}
-
 /// Result from a single worker thread.
 #[derive(Debug, Clone)]
 pub struct WorkerResult {
@@ -102,8 +93,6 @@ pub struct WorkerPool {
     job_senders: Vec<Sender<MiningJob>>,
     /// Receiver for collecting results from all workers
     result_rx: Receiver<WorkerResult>,
-    /// Shared cancellation flag for all workers
-    cancel_flag: Arc<AtomicBool>,
     /// Job ID counter - incremented on each new job to detect stale results
     current_job_id: Arc<AtomicU64>,
     /// Thread handles (for cleanup)
@@ -124,7 +113,6 @@ impl WorkerPool {
     ) -> Self {
         let total_workers = cpu_workers + gpu_devices;
         let (result_tx, result_rx) = bounded(total_workers * 64);
-        let cancel_flag = Arc::new(AtomicBool::new(false));
         let current_job_id = Arc::new(AtomicU64::new(0));
 
         let mut job_senders = Vec::with_capacity(total_workers);
@@ -141,25 +129,16 @@ impl WorkerPool {
         if cpu_workers > 0 {
             if let Some(ref engine) = cpu_engine {
                 for _ in 0..cpu_workers {
-                    let (job_tx, job_rx) = bounded::<MiningJob>(1);
+                    let (job_tx, job_rx) = unbounded::<MiningJob>();
                     job_senders.push(job_tx);
 
                     let eng = engine.clone();
                     let tx = result_tx.clone();
-                    let cancel = cancel_flag.clone();
                     let job_id_counter = current_job_id.clone();
                     let tid = thread_id;
 
                     let handle = thread::spawn(move || {
-                        worker_loop(
-                            tid,
-                            EngineType::Cpu,
-                            eng,
-                            job_rx,
-                            tx,
-                            cancel,
-                            job_id_counter,
-                        );
+                        worker_loop(tid, EngineType::Cpu, eng, job_rx, tx, job_id_counter);
                     });
                     handles.push(handle);
                     thread_id += 1;
@@ -171,25 +150,16 @@ impl WorkerPool {
         if gpu_devices > 0 {
             if let Some(ref engine) = gpu_engine {
                 for _ in 0..gpu_devices {
-                    let (job_tx, job_rx) = bounded::<MiningJob>(1);
+                    let (job_tx, job_rx) = unbounded::<MiningJob>();
                     job_senders.push(job_tx);
 
                     let eng = engine.clone();
                     let tx = result_tx.clone();
-                    let cancel = cancel_flag.clone();
                     let job_id_counter = current_job_id.clone();
                     let tid = thread_id;
 
                     let handle = thread::spawn(move || {
-                        worker_loop(
-                            tid,
-                            EngineType::Gpu,
-                            eng,
-                            job_rx,
-                            tx,
-                            cancel,
-                            job_id_counter,
-                        );
+                        worker_loop(tid, EngineType::Gpu, eng, job_rx, tx, job_id_counter);
                     });
                     handles.push(handle);
                     thread_id += 1;
@@ -200,7 +170,6 @@ impl WorkerPool {
         Self {
             job_senders,
             result_rx,
-            cancel_flag,
             current_job_id,
             _handles: handles,
             cpu_worker_count: cpu_workers,
@@ -216,14 +185,7 @@ impl WorkerPool {
         // will be detected as stale when workers check the job ID before sending results
         let new_job_id = self.current_job_id.fetch_add(1, Ordering::SeqCst) + 1;
 
-        // Cancel any running job
-        self.cancel_flag.store(true, Ordering::SeqCst);
-
-        // Brief pause to let workers see the cancellation
-        std::thread::sleep(std::time::Duration::from_millis(1));
-
-        // Reset cancel flag for new job
-        self.cancel_flag.store(false, Ordering::SeqCst);
+        log::debug!("[JOB DISPATCH] Starting job {new_job_id}");
 
         // Create job context (shared across all workers)
         let ctx = pow_core::JobContext::new(header_hash, difficulty);
@@ -232,25 +194,26 @@ impl WorkerPool {
             job_id: new_job_id,
         };
 
-        // Dispatch job to all workers
+        // Dispatch job to all workers (unbounded channels - always succeeds unless worker died)
         for tx in &self.job_senders {
-            // Non-blocking send - if worker is still processing old job, it will
-            // see the cancel flag and exit soon
-            let _ = tx.try_send(job.clone());
+            // Send will only fail if receiver is dropped (worker thread died)
+            let _ = tx.send(job.clone());
         }
 
-        log::debug!(
-            "Job dispatched to {} workers (job_id {})",
-            self.job_senders.len(),
-            new_job_id
-        );
+        let worker_count = self.job_senders.len();
+        log::debug!("[JOB DISPATCH] Job {new_job_id} dispatched to {worker_count} workers");
 
         new_job_id
     }
 
-    /// Cancel the current job.
+    /// Cancel the current job by incrementing the job ID.
+    /// Workers will detect the change and stop processing.
+    ///
+    /// Note: This increments job_id by 1, and start_job() also increments by 1,
+    /// so job IDs in logs may become non-contiguous after disconnects/cancellations.
+    /// This is expected behavior - job IDs only need to be unique, not sequential.
     pub fn cancel(&self) {
-        self.cancel_flag.store(true, Ordering::SeqCst);
+        self.current_job_id.fetch_add(1, Ordering::SeqCst);
     }
 
     /// Get the result receiver for collecting worker results.
@@ -258,11 +221,6 @@ impl WorkerPool {
         &self.result_rx
     }
 
-    /// Get the shared cancel flag.
-    pub fn cancel_flag(&self) -> &Arc<AtomicBool> {
-        &self.cancel_flag
-    }
-
     /// Total number of workers.
     pub fn worker_count(&self) -> usize {
         self.job_senders.len()
@@ -279,6 +237,26 @@ impl WorkerPool {
     }
 }
 
+/// Log worker completion with hash rate info.
+fn log_worker_completion(
+    type_str: &str,
+    thread_id: usize,
+    status: &str,
+    hash_count: u64,
+    elapsed: std::time::Duration,
+) {
+    let hash_rate = if elapsed.as_secs_f64() > 0.0 {
+        hash_count as f64 / elapsed.as_secs_f64()
+    } else {
+        0.0
+    };
+    log::info!(
+        "{type_str} worker {thread_id} {status}: {hash_count} hashes in {:.2}s ({})",
+        elapsed.as_secs_f64(),
+        format_hashrate(hash_rate)
+    );
+}
+
 /// Main loop for a persistent worker thread.
 fn worker_loop(
     thread_id: usize,
@@ -286,7 +264,6 @@ fn worker_loop(
     engine: Arc<dyn MinerEngine>,
     job_rx: Receiver<MiningJob>,
     result_tx: Sender<WorkerResult>,
-    cancel_flag: Arc<AtomicBool>,
     current_job_id: Arc<AtomicU64>,
 ) {
     let type_str = match engine_type {
@@ -298,8 +275,10 @@ fn worker_loop(
 
     // Main job processing loop
     loop {
-        // Wait for a job
-        let job = match job_rx.recv() {
+        log::debug!("[WORKER {type_str}-{thread_id}] Waiting for job...");
+
+        // Wait for a job (blocking)
+        let mut job = match job_rx.recv() {
             Ok(job) => job,
             Err(_) => {
                 // Channel closed, pool is shutting down
@@ -308,35 +287,51 @@ fn worker_loop(
             }
         };
 
+        // Drain channel to get the latest job (in case multiple jobs queued while we were busy)
+        let mut skipped = 0;
+        while let Ok(newer_job) = job_rx.try_recv() {
+            skipped += 1;
+            job = newer_job;
+        }
+        if skipped > 0 {
+            log::debug!("[WORKER {type_str}-{thread_id}] Drained {skipped} stale jobs from queue");
+        }
+
         // Capture the job's ID for later validation
         let job_id = job.job_id;
-
-        // Check if already cancelled before starting
-        if cancel_flag.load(Ordering::Relaxed) {
-            log::debug!("{type_str} worker {thread_id} skipping cancelled job");
-            continue;
-        }
+        log::debug!("[WORKER {type_str}-{thread_id}] Received job {job_id}");
 
         // Generate random starting nonce for this job
         let start = generate_random_nonce();
         let end = U512::MAX;
 
-        log::debug!(
-            "{type_str} worker {thread_id} processing job {job_id}: range {} to {}",
-            format_u512(start),
-            format_u512(end)
-        );
+        log::debug!("[WORKER {type_str}-{thread_id}] Starting search for job {job_id}");
 
-        // Execute the search
+        // Execute the search - both CPU and GPU use job ID comparison for cancellation
+        let search_start = std::time::Instant::now();
         let range = EngineRange { start, end };
-        let result = engine.search_range(&job.ctx, range, &cancel_flag);
+        let cancel_check = JobIdCancelCheck {
+            current_job_id: &current_job_id,
+            my_job_id: job_id,
+        };
+        let result = engine.search_range(&job.ctx, range, &cancel_check);
+        let search_elapsed = search_start.elapsed();
+
+        let result_type = match &result {
+            engine_cpu::EngineStatus::Found { .. } => "FOUND",
+            engine_cpu::EngineStatus::Exhausted { .. } => "EXHAUSTED",
+            engine_cpu::EngineStatus::Cancelled { .. } => "CANCELLED",
+            engine_cpu::EngineStatus::Running { .. } => "RUNNING",
+        };
+        log::debug!(
+            "[WORKER {type_str}-{thread_id}] Job {job_id} search finished: {} in {:.2}s",
+            result_type,
+            search_elapsed.as_secs_f64()
+        );
 
         // Check if job ID changed during search - if so, this result is stale
         let actual_job_id = current_job_id.load(Ordering::SeqCst);
         if actual_job_id != job_id {
-            log::debug!(
-                "⏰ {type_str} worker {thread_id} discarding stale result (job {job_id} != current {actual_job_id})"
-            );
             // Still send hash count for metrics, but without the candidate
             let hash_count = match result {
                 engine_cpu::EngineStatus::Found { hash_count, .. } => hash_count,
@@ -344,6 +339,13 @@ fn worker_loop(
                 engine_cpu::EngineStatus::Cancelled { hash_count } => hash_count,
                 engine_cpu::EngineStatus::Running { .. } => 0,
             };
+            log_worker_completion(
+                type_str,
+                thread_id,
+                "cancelled (stale)",
+                hash_count,
+                search_elapsed,
+            );
             let _ = result_tx.try_send(WorkerResult {
                 thread_id,
                 engine_type,
@@ -370,11 +372,17 @@ fn worker_loop(
                 (Some(MiningCandidate { nonce, work, hash }), hash_count)
             }
             engine_cpu::EngineStatus::Exhausted { hash_count } => {
-                log::debug!("{type_str} worker {thread_id} exhausted range ({hash_count} hashes)");
+                log_worker_completion(
+                    type_str,
+                    thread_id,
+                    "exhausted range",
+                    hash_count,
+                    search_elapsed,
+                );
                 (None, hash_count)
             }
             engine_cpu::EngineStatus::Cancelled { hash_count } => {
-                log::debug!("{type_str} worker {thread_id} cancelled ({hash_count} hashes)");
+                log_worker_completion(type_str, thread_id, "cancelled", hash_count, search_elapsed);
                 (None, hash_count)
             }
             engine_cpu::EngineStatus::Running { .. } => {
@@ -405,7 +413,7 @@ fn worker_loop(
 /// Resolve GPU configuration and initialize the engine.
 pub fn resolve_gpu_configuration(
     requested_devices: Option<usize>,
-    cancel_interval: Option<u32>,
+    batch_size: u64,
 ) -> anyhow::Result<(Option<Arc<dyn MinerEngine>>, usize)> {
     // Explicit 0 means no GPU
     if requested_devices == Some(0) {
@@ -413,17 +421,14 @@ pub fn resolve_gpu_configuration(
     }
 
     // Try to initialize GPU engine
-    let engine = match cancel_interval {
-        Some(interval) => engine_gpu::GpuEngine::try_with_cancel_interval(interval),
-        None => engine_gpu::GpuEngine::try_new(),
-    };
+    let engine = engine_gpu::GpuEngine::try_new(batch_size);
     let engine = match engine {
         Ok(e) => e,
         Err(e) => {
             if requested_devices.is_some() {
                 anyhow::bail!("Failed to initialize GPU engine: {}", e);
             }
-            log::info!("No GPU available: {}", e);
+            log::info!("No GPU available: {e}");
             return Ok((None, 0));
         }
     };
@@ -443,7 +448,7 @@ pub fn resolve_gpu_configuration(
             return Ok((None, 0));
         }
         None => {
-            log::info!("Auto-detected {} GPU device(s)", available);
+            log::info!("Auto-detected {available} GPU device(s)");
             available
         }
     };
@@ -458,7 +463,7 @@ pub async fn run(config: ServiceConfig) -> anyhow::Result<()> {
 
     // Resolve GPU configuration
     let (gpu_engine, gpu_devices) =
-        resolve_gpu_configuration(config.gpu_devices, config.gpu_cancel_interval)?;
+        resolve_gpu_configuration(config.gpu_devices, config.gpu_batch_size)?;
 
     // Resolve CPU workers
     let cpu_workers = config.cpu_workers.unwrap_or_else(|| {
@@ -478,7 +483,9 @@ pub async fn run(config: ServiceConfig) -> anyhow::Result<()> {
 
     // Create CPU engine
     let cpu_engine: Option<Arc<dyn MinerEngine>> = if cpu_workers > 0 {
-        Some(Arc::new(engine_cpu::FastCpuEngine::new()))
+        Some(Arc::new(engine_cpu::FastCpuEngine::new(
+            config.cpu_batch_size,
+        )))
     } else {
         None
     };
@@ -491,19 +498,20 @@ pub async fn run(config: ServiceConfig) -> anyhow::Result<()> {
     );
 
     if let Some(ref engine) = cpu_engine {
-        log::info!("🖥️  CPU engine: {}", engine.name());
+        let name = engine.name();
+        log::info!("🖥️  CPU engine: {name}");
     }
     if let Some(ref engine) = gpu_engine {
-        log::info!("🎮 GPU engine: {}", engine.name());
+        let name = engine.name();
+        log::info!("🎮 GPU engine: {name}");
     }
 
-    log::info!(
-        "⛏️  Mining service ready with {} total workers",
-        cpu_workers + gpu_devices
-    );
+    let total_workers = cpu_workers + gpu_devices;
+    log::info!("⛏️  Mining service ready with {total_workers} total workers");
 
     // Connect to node and start mining
-    log::info!("🌐 Connecting to node at {}", config.node_addr);
+    let node_addr = config.node_addr;
+    log::info!("🌐 Connecting to node at {node_addr}");
     quic::connect_and_mine(
         config.node_addr,
         cpu_engine,

crates/miner-service/src/quic.rs

@@ -227,9 +227,9 @@ async fn handle_connection(
                 match msg_result {
                     Ok(MinerMessage::NewJob(request)) => {
                         log::info!(
-                            "⛏️ Received job: id={}, hash={}...",
+                            "⛏️ Received job: id={}, hash=0x{}",
                             request.job_id,
-                            &request.mining_hash[..8]
+                            request.mining_hash
                         );
 
                         // Parse header hash