Merge pull request #43 from Quantus-Network/fix_gpu_cancel_issue

Fix gpu cancel issue

crates/engine-gpu/benches/gpu_engine_bench.rs

@@ -5,10 +5,11 @@ use pow_core::JobContext;
 use primitive_types::U512;
 use rand::RngCore;
 use std::sync::atomic::AtomicBool;
+use std::time::Duration;
 
 fn bench_cpu_vs_gpu_small(c: &mut Criterion) {
     let cpu_engine = FastCpuEngine::new();
-    let gpu_engine = GpuEngine::new();
+    let gpu_engine = GpuEngine::new(Duration::from_millis(3000));
     let cancel_flag = AtomicBool::new(false);
 
     // Small range: 10K nonces - reasonable for benchmarking
@@ -58,7 +59,7 @@ 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 gpu_engine = GpuEngine::new(Duration::from_millis(3000));
     let cancel_flag = AtomicBool::new(false);
 
     // Medium range: 100K nonces
@@ -108,7 +109,7 @@ 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 gpu_engine = GpuEngine::new(Duration::from_millis(3000));
     let cancel_flag = AtomicBool::new(false);
 
     // Large range: 1M nonces - where GPU should really shine
@@ -158,7 +159,7 @@ 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 gpu_engine = GpuEngine::new(Duration::from_millis(3000));
     let cancel_flag = AtomicBool::new(false);
 
     // Range where we expect to find solutions quickly
@@ -208,7 +209,7 @@ 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 gpu_engine = GpuEngine::new(Duration::from_millis(3000));
     let cancel_flag = AtomicBool::new(false);
 
     // Fixed time benchmark - see how many hashes we can do in 1 second
@@ -257,7 +258,7 @@ 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::new(Duration::from_millis(3000));
     let cancel_flag = AtomicBool::new(false);
 
     let mut group = c.benchmark_group("gpu_batch_sizes");

crates/engine-gpu/examples/verify_nonce.rs

@@ -2,6 +2,7 @@ use engine_cpu::{BaselineCpuEngine, EngineStatus, MinerEngine, Range};
 use engine_gpu::GpuEngine;
 use primitive_types::U512;
 use std::sync::atomic::AtomicBool;
+use std::time::Duration;
 
 fn main() {
     // Initialize logging
@@ -25,7 +26,7 @@ fn main() {
 
     // 3. Verify with GPU engine
     log::info!("Initializing GPU engine...");
-    let gpu_engine = GpuEngine::new();
+    let gpu_engine = GpuEngine::new(Duration::from_millis(3000));
 
     // Search a small range around the valid nonce
     let gpu_range = Range {

crates/engine-gpu/src/lib.rs

@@ -7,9 +7,10 @@ use pow_core::JobContext;
 use primitive_types::U512;
 use std::cell::RefCell;
 use std::sync::{
-    atomic::{AtomicBool, AtomicUsize, Ordering},
+    atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering},
     Arc,
 };
+use std::time::Duration;
 
 /// Represents a single GPU device context.
 struct GpuContext {
@@ -19,6 +20,9 @@ struct GpuContext {
 
     // Cached vendor configuration
     optimal_workgroups: u32,
+
+    // Dynamic batch size tuning
+    batch_size: AtomicU64,
 }
 
 #[derive(Clone)]
@@ -35,6 +39,7 @@ struct GpuResources {
 pub struct GpuEngine {
     contexts: Vec<Arc<GpuContext>>,
     device_counter: AtomicUsize,
+    target_batch_duration: Duration,
 }
 
 // Thread-local storage for consistent GPU device assignment per worker thread
@@ -138,21 +143,22 @@ impl GpuContext {
 
 impl Default for GpuEngine {
     fn default() -> Self {
-        Self::new()
+        // Default to 3 seconds if not specified via new()
+        Self::new(Duration::from_secs(3))
     }
 }
 
 impl GpuEngine {
-    pub fn new() -> Self {
-        block_on(Self::init()).expect("Failed to initialize GPU engine")
+    pub fn new(target_batch_duration: Duration) -> Self {
+        block_on(Self::init(target_batch_duration)).expect("Failed to initialize GPU engine")
     }
 
     /// Try to initialize the GPU engine, returning an error if initialization fails.
-    pub fn try_new() -> Result<Self, Box<dyn std::error::Error>> {
-        block_on(Self::init())
+    pub fn try_new(target_batch_duration: Duration) -> Result<Self, Box<dyn std::error::Error>> {
+        block_on(Self::init(target_batch_duration))
     }
 
-    async fn init() -> Result<Self, Box<dyn std::error::Error>> {
+    async fn init(target_batch_duration: Duration) -> Result<Self, Box<dyn std::error::Error>> {
         log::info!("Initializing WGPU...");
         let instance = wgpu::Instance::new(&wgpu::InstanceDescriptor {
             backends: wgpu::Backends::PRIMARY,
@@ -235,6 +241,7 @@ impl GpuEngine {
                 queue,
                 pipeline,
                 optimal_workgroups,
+                batch_size: AtomicU64::new(1_000_000),
             }));
         }
 
@@ -279,6 +286,7 @@ impl GpuEngine {
         Ok(Self {
             contexts,
             device_counter: AtomicUsize::new(0),
+            target_batch_duration,
         })
     }
 
@@ -344,11 +352,14 @@ impl MinerEngine for GpuEngine {
         });
 
         let gpu_ctx = &self.contexts[device_index];
-        log::debug!(
-            "GPU device {} processing range {}..={} (inclusive)",
+        let initial_batch_size = gpu_ctx.batch_size.load(Ordering::Relaxed);
+        log::info!(
+            target: "gpu_engine",
+            "GPU {} search started: range {}..={} (inclusive), initial batch size: {}",
             device_index,
             range.start,
-            range.end
+            range.end,
+            initial_batch_size
         );
 
         // Ensure resources are initialized for this thread
@@ -387,174 +398,243 @@ impl MinerEngine for GpuEngine {
             bytemuck::cast_slice(&target_u32s),
         );
 
-        // Length of the inclusive nonce range
-        let total_range_size = (range.end - range.start + 1).as_u64();
-        if total_range_size == 0 {
-            return EngineStatus::Exhausted { hash_count: 0 };
-        }
+        // --- Batch Processing Loop ---
+        let mut current_nonce = range.start;
+        let mut total_hashes_processed = 0u64;
 
-        // Thread configuration for a single dispatch over this range.
-        let threads_per_workgroup = 256u32; // Must match shader @workgroup_size(256)
-        let limits = gpu_ctx.device.limits();
-        let max_workgroups = limits.max_compute_workgroups_per_dimension;
+        // Loop until range is covered or cancelled
+        while current_nonce <= range.end {
+            // Check for cancellation between batches to ensure responsiveness
+            if cancel.load(Ordering::Relaxed) {
+                log::info!(
+                    target: "gpu_engine",
+                    "GPU {} search cancelled: processed {} hashes",
+                    device_index,
+                    total_hashes_processed
+                );
+                return EngineStatus::Cancelled {
+                    hash_count: total_hashes_processed,
+                };
+            }
 
-        // Vendor hint, clamped by hardware limits.
-        let hinted_workgroups = gpu_ctx.optimal_workgroups.max(1).min(max_workgroups);
-        let hinted_threads = hinted_workgroups as u64 * threads_per_workgroup as u64;
+            // Determine dynamic batch size for this iteration
+            let batch_size = gpu_ctx.batch_size.load(Ordering::Relaxed);
 
-        // Choose a logical thread budget: enough threads to fill the GPU, but no more than
-        // the range length (spawning more threads than nonces is wasteful).
-        let mut logical_threads = total_range_size.min(hinted_threads);
-        if logical_threads == 0 {
-            logical_threads = 1;
-        }
-
-        // Round logical_threads up to a multiple of workgroup size so we have full workgroups.
-        let mut num_workgroups = (logical_threads as u32).div_ceil(threads_per_workgroup);
-        if num_workgroups == 0 {
-            num_workgroups = 1;
-        }
-        let total_threads = (num_workgroups * threads_per_workgroup) as u64;
-
-        // Derive how many nonces each logical thread should process so that the entire
-        // range is covered in a single dispatch.
-        let nonces_per_thread = total_range_size.div_ceil(total_threads).max(1) as u32;
-
-        let total_threads_u32 = total_threads as u32;
-
-        log::info!(
-            target: "gpu_engine",
-            "GPU dispatch configuration: total_range={} nonces, workgroups={}, threads={}, nonces_per_thread={}",
-            total_range_size,
-            num_workgroups,
-            total_threads_u32,
-            nonces_per_thread
-        );
-
-        // We'll process the full range in a single dispatch.
-        let hash_count = total_range_size;
-
-        // Eliminate intermediate syncs - only sync at end or when solution found
-        const RESULTS_SIZE: usize = (1 + 16 + 16) * 4;
-        const ZEROS: [u8; RESULTS_SIZE] = [0; RESULTS_SIZE];
-
-        #[cfg(feature = "metrics")]
-        {
-            let device_id = "gpu-0";
-            metrics::set_gpu_batch_size(device_id, total_range_size as f64);
-            metrics::set_gpu_workgroups(device_id, num_workgroups as f64);
-        }
-
-        if cancel.load(Ordering::Relaxed) {
-            log::debug!(target: "gpu_engine", "GPU {} cancelled before dispatch.", device_index);
-            return EngineStatus::Cancelled { hash_count: 0 };
-        }
-
-        // Dispatch configuration for this range:
-        // [total_threads, nonces_per_thread, total_nonces, threads_per_workgroup]
-        let total_nonces_u32 = total_range_size.min(u32::MAX as u64) as u32;
-        let dispatch_config = [
-            total_threads_u32,
-            nonces_per_thread,
-            total_nonces_u32,
-            threads_per_workgroup,
-        ];
-        gpu_ctx.queue.write_buffer(
-            &resources.dispatch_config_buffer,
-            0,
-            bytemuck::cast_slice(&dispatch_config),
-        );
-
-        // Starting nonce for this range.
-        let start_nonce_bytes = range.start.to_little_endian();
-        gpu_ctx
-            .queue
-            .write_buffer(&resources.start_nonce_buffer, 0, &start_nonce_bytes);
-
-        // Reset results buffer to detect solutions from this dispatch.
-        gpu_ctx
-            .queue
-            .write_buffer(&resources.results_buffer, 0, &ZEROS);
-
-        let total_start = std::time::Instant::now();
-
-        // Create command buffer for this range
-        let mut encoder = gpu_ctx
-            .device
-            .create_command_encoder(&wgpu::CommandEncoderDescriptor { label: None });
-        {
-            let mut cpass = encoder.begin_compute_pass(&wgpu::ComputePassDescriptor {
-                label: None,
-                timestamp_writes: None,
-            });
-            cpass.set_pipeline(&gpu_ctx.pipeline);
-            cpass.set_bind_group(0, &resources.bind_group, &[]);
-            cpass.dispatch_workgroups(num_workgroups, 1, 1);
-        }
-        encoder.copy_buffer_to_buffer(
-            &resources.results_buffer,
-            0,
-            &resources.staging_buffer,
-            0,
-            RESULTS_SIZE as u64,
-        );
-
-        // Submit and wait for completion
-        gpu_ctx.queue.submit(Some(encoder.finish()));
-
-        let buffer_slice = resources.staging_buffer.slice(..);
-        buffer_slice.map_async(wgpu::MapMode::Read, |_| {});
-
-        let _ = gpu_ctx.device.poll(wgpu::PollType::Wait {
-            submission_index: None,
-            timeout: None,
-        });
-
-        let data = buffer_slice.get_mapped_range();
-        let result_u32s: &[u32] = bytemuck::cast_slice(&data);
-
-        if result_u32s[0] != 0 {
-            // Solution found!
-            let nonce_u32s = &result_u32s[1..17];
-            let hash_u32s = &result_u32s[17..33];
-
-            let nonce = U512::from_little_endian(bytemuck::cast_slice(nonce_u32s));
-            let hash = U512::from_little_endian(bytemuck::cast_slice(hash_u32s));
-            let work = nonce.to_big_endian();
-
-            log::info!(
-                "GPU {} found solution! Nonce: {}, Hash: {:x}",
+            log::debug!(
+                target: "gpu_engine",
+                "GPU {} starting batch: nonce {}..{}, batch size: {}",
                 device_index,
-                nonce,
-                hash
+                current_nonce,
+                current_nonce.saturating_add(U512::from(batch_size)).saturating_sub(U512::from(1u64)).min(range.end),
+                batch_size
             );
 
+            // Calculate inclusive end for this batch
+            let batch_end = current_nonce
+                .saturating_add(U512::from(batch_size))
+                .saturating_sub(U512::from(1u64));
+
+            let actual_end = if batch_end > range.end {
+                range.end
+            } else {
+                batch_end
+            };
+
+            // Range size for this specific batch
+            let range_size_u512 = actual_end
+                .saturating_sub(current_nonce)
+                .saturating_add(U512::from(1u64));
+            let range_size = range_size_u512.as_u64(); // batch_size is u64, so this is safe
+
+            if range_size == 0 {
+                break;
+            }
+
+            // --- Dispatch Logic (same as before but using range_size) ---
+            let threads_per_workgroup = 256u32; // Must match shader
+            let limits = gpu_ctx.device.limits();
+            let max_workgroups = limits.max_compute_workgroups_per_dimension;
+
+            let hinted_workgroups = gpu_ctx.optimal_workgroups.max(1).min(max_workgroups);
+            let hinted_threads = hinted_workgroups as u64 * threads_per_workgroup as u64;
+
+            let mut logical_threads = range_size.min(hinted_threads);
+            if logical_threads == 0 {
+                logical_threads = 1;
+            }
+
+            let mut num_workgroups = (logical_threads as u32).div_ceil(threads_per_workgroup);
+            if num_workgroups == 0 {
+                num_workgroups = 1;
+            }
+            let total_threads = (num_workgroups * threads_per_workgroup) as u64;
+            let nonces_per_thread = range_size.div_ceil(total_threads).max(1) as u32;
+            let total_threads_u32 = total_threads as u32;
+
+            #[cfg(feature = "metrics")]
+            {
+                let device_id = "gpu-0"; // Simplified; ideally propagate device ID to metrics
+                metrics::set_gpu_batch_size(device_id, range_size as f64);
+                metrics::set_gpu_workgroups(device_id, num_workgroups as f64);
+            }
+
+            // Dispatch config: [total_threads, nonces_per_thread, total_nonces, threads_per_workgroup]
+            let total_nonces_u32 = range_size.min(u32::MAX as u64) as u32;
+            let dispatch_config = [
+                total_threads_u32,
+                nonces_per_thread,
+                total_nonces_u32,
+                threads_per_workgroup,
+            ];
+            gpu_ctx.queue.write_buffer(
+                &resources.dispatch_config_buffer,
+                0,
+                bytemuck::cast_slice(&dispatch_config),
+            );
+
+            // Starting nonce for this batch
+            let start_nonce_bytes = current_nonce.to_little_endian();
+            gpu_ctx
+                .queue
+                .write_buffer(&resources.start_nonce_buffer, 0, &start_nonce_bytes);
+
+            // Reset results
+            const RESULTS_SIZE: usize = (1 + 16 + 16) * 4;
+            const ZEROS: [u8; RESULTS_SIZE] = [0; RESULTS_SIZE];
+            gpu_ctx
+                .queue
+                .write_buffer(&resources.results_buffer, 0, &ZEROS);
+
+            let batch_start_time = std::time::Instant::now();
+
+            // Encode and submit
+            let mut encoder = gpu_ctx
+                .device
+                .create_command_encoder(&wgpu::CommandEncoderDescriptor { label: None });
+            {
+                let mut cpass = encoder.begin_compute_pass(&wgpu::ComputePassDescriptor {
+                    label: None,
+                    timestamp_writes: None,
+                });
+                cpass.set_pipeline(&gpu_ctx.pipeline);
+                cpass.set_bind_group(0, &resources.bind_group, &[]);
+                cpass.dispatch_workgroups(num_workgroups, 1, 1);
+            }
+            encoder.copy_buffer_to_buffer(
+                &resources.results_buffer,
+                0,
+                &resources.staging_buffer,
+                0,
+                RESULTS_SIZE as u64,
+            );
+
+            gpu_ctx.queue.submit(Some(encoder.finish()));
+
+            let buffer_slice = resources.staging_buffer.slice(..);
+            buffer_slice.map_async(wgpu::MapMode::Read, |_| {});
+
+            let _ = gpu_ctx.device.poll(wgpu::PollType::Wait {
+                submission_index: None,
+                timeout: None,
+            });
+
+            let data = buffer_slice.get_mapped_range();
+            let result_u32s: &[u32] = bytemuck::cast_slice(&data);
+
+            if result_u32s[0] != 0 {
+                // Solution found!
+                let nonce_u32s = &result_u32s[1..17];
+                let hash_u32s = &result_u32s[17..33];
+                let nonce = U512::from_little_endian(bytemuck::cast_slice(nonce_u32s));
+                let hash = U512::from_little_endian(bytemuck::cast_slice(hash_u32s));
+                let work = nonce.to_big_endian();
+
+                log::info!(
+                    target: "gpu_engine",
+                    "GPU {} found solution! Nonce: {}, Hash: {:x}",
+                    device_index,
+                    nonce,
+                    hash
+                );
+
+                drop(data);
+                resources.staging_buffer.unmap();
+
+                log::info!(
+                    target: "gpu_engine",
+                    "GPU {} search ended (solution found): processed {} hashes",
+                    device_index,
+                    total_hashes_processed + range_size
+                );
+
+                // Even if found, return total hashes processed including this batch
+                return EngineStatus::Found {
+                    candidate: Candidate { nonce, work, hash },
+                    hash_count: total_hashes_processed + range_size,
+                    origin: FoundOrigin::GpuG1,
+                };
+            }
+
             drop(data);
             resources.staging_buffer.unmap();
 
-            return EngineStatus::Found {
-                candidate: Candidate { nonce, work, hash },
-                // Approximate: we assume a uniform distribution within the dispatch
-                // and report that we processed the entire range.
-                hash_count,
-                origin: FoundOrigin::GpuG1,
-            };
+            // --- Auto-tuning Logic ---
+            let batch_elapsed = batch_start_time.elapsed();
+            let batch_elapsed_secs = batch_elapsed.as_secs_f64();
+
+            // Update stats
+            total_hashes_processed += range_size;
+
+            if batch_elapsed_secs > 0.0 {
+                let target_secs = self.target_batch_duration.as_secs_f64();
+
+                // Calculate ideal batch size: (current_size / elapsed) * target
+                let hashrate = range_size as f64 / batch_elapsed_secs;
+                let ideal_batch_size = (hashrate * target_secs) as u64;
+
+                let old_size = gpu_ctx.batch_size.load(Ordering::Relaxed);
+
+                // Smooth update: 50% old, 50% new to avoid oscillation
+                // Also clamp to avoid drastic changes (0.5x to 2.0x)
+                let min_clamp = old_size / 2;
+                let max_clamp = old_size * 2;
+
+                // Ensure we don't go below a reasonable minimum (e.g. 100k) to avoid overhead dominance
+                let clamped_ideal = ideal_batch_size.clamp(min_clamp, max_clamp).max(100_000);
+
+                // Simple moving average
+                let new_size = (old_size + clamped_ideal) / 2;
+
+                if new_size != old_size {
+                    gpu_ctx.batch_size.store(new_size, Ordering::Relaxed);
+                    log::debug!(
+                        target: "gpu_engine",
+                        "GPU {} auto-tune: batch {} -> {} (elapsed: {:.3}s, target: {:.1}s, rate: {:.1} MH/s)",
+                        device_index,
+                        old_size,
+                        new_size,
+                        batch_elapsed_secs,
+                        target_secs,
+                        hashrate / 1_000_000.0
+                    );
+                }
+            }
+
+            // Move to next batch
+            current_nonce = actual_end.saturating_add(U512::from(1u64));
         }
 
-        drop(data);
-        resources.staging_buffer.unmap();
-
-        let total_time = total_start.elapsed();
+        // Finished entire range without solution
         log::info!(
             target: "gpu_engine",
-            "GPU finished range. Total time: {:.2}ms, Batches: 1, Hashes: {}, Performance: {:.0} hashes/sec",
-            total_time.as_secs_f64() * 1000.0,
-            hash_count,
-            hash_count as f64 / total_time.as_secs_f64()
+            "GPU {} search completed: processed {} hashes, final batch size: {}",
+            device_index,
+            total_hashes_processed,
+            gpu_ctx.batch_size.load(Ordering::Relaxed)
         );
-
-        log::debug!(target: "gpu_engine", "GPU {} finished range with no solution.", device_index);
-        EngineStatus::Exhausted { hash_count }
+        EngineStatus::Exhausted {
+            hash_count: total_hashes_processed,
+        }
     }
 }
 

crates/miner-cli/src/main.rs

@@ -60,6 +60,10 @@ enum Command {
         #[arg(long = "manip-throttle-cap", env = "MINER_MANIP_THROTTLE_CAP")]
         manip_throttle_cap: Option<u64>,
 
+        /// Target duration for GPU mining batches in milliseconds (default: 3000)
+        #[arg(long = "gpu-batch-duration-ms", env = "MINER_GPU_BATCH_DURATION_MS")]
+        gpu_batch_duration_ms: Option<u64>,
+
         /// Telemetry endpoints (repeat --telemetry-endpoint or comma-separated)
         #[arg(long = "telemetry-endpoint", env = "MINER_TELEMETRY_ENDPOINTS", value_delimiter = ',', num_args = 0.., value_name = "URL")]
         telemetry_endpoints: Option<Vec<String>>,
@@ -148,6 +152,7 @@ async fn main() {
         manip_base_delay_ns: None,
         manip_step_batch: None,
         manip_throttle_cap: None,
+        gpu_batch_duration_ms: None,
         telemetry_endpoints: None,
         telemetry_enabled: None,
         telemetry_verbosity: None,
@@ -171,6 +176,7 @@ async fn main() {
             manip_base_delay_ns,
             manip_step_batch,
             manip_throttle_cap,
+            gpu_batch_duration_ms,
             telemetry_endpoints,
             telemetry_enabled,
             telemetry_verbosity,
@@ -194,6 +200,7 @@ async fn main() {
                 manip_base_delay_ns,
                 manip_step_batch,
                 manip_throttle_cap,
+                gpu_batch_duration_ms,
                 telemetry_endpoints,
                 telemetry_enabled,
                 telemetry_verbosity,
@@ -231,6 +238,7 @@ async fn run_serve_command(
     manip_base_delay_ns: Option<u64>,
     manip_step_batch: Option<u64>,
     manip_throttle_cap: Option<u64>,
+    gpu_batch_duration_ms: Option<u64>,
     telemetry_endpoints: Option<Vec<String>>,
     telemetry_enabled: Option<bool>,
     telemetry_verbosity: Option<u8>,
@@ -302,6 +310,7 @@ async fn run_serve_command(
         manip_base_delay_ns,
         manip_step_batch,
         manip_throttle_cap,
+        gpu_batch_duration_ms,
     };
     log::info!("Effective config: {config}");
 
@@ -332,7 +341,7 @@ async fn run_benchmark_command(
 
     // Initialize GPU engine and determine effective GPU devices
     let (gpu_engine, effective_gpu_devices) =
-        match miner_service::resolve_gpu_configuration(gpu_devices) {
+        match miner_service::resolve_gpu_configuration(gpu_devices, None) {
             Ok((engine, count)) => (engine, count),
             Err(e) => {
                 eprintln!("❌ ERROR: {}", e);

crates/miner-service/src/lib.rs

@@ -39,6 +39,8 @@ pub struct ServiceConfig {
     pub manip_step_batch: Option<u64>,
     /// Optional cap on solved-blocks throttle index for manipulator engine.
     pub manip_throttle_cap: Option<u64>,
+    /// Optional target duration for GPU batches in milliseconds.
+    pub gpu_batch_duration_ms: Option<u64>,
 }
 
 impl Default for ServiceConfig {
@@ -54,6 +56,7 @@ impl Default for ServiceConfig {
             manip_base_delay_ns: None,
             manip_step_batch: None,
             manip_throttle_cap: None,
+            gpu_batch_duration_ms: None,
         }
     }
 }
@@ -62,7 +65,7 @@ impl fmt::Display for ServiceConfig {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         write!(
             f,
-            "port={}, cpu_workers={:?}, gpu_devices={:?}, metrics_port={:?}, progress_interval_ms={:?}, chunk_size={:?}, manip_solved_blocks={:?}, manip_base_delay_ns={:?}, manip_step_batch={:?}, manip_throttle_cap={:?}",
+            "port={}, cpu_workers={:?}, gpu_devices={:?}, metrics_port={:?}, progress_interval_ms={:?}, chunk_size={:?}, manip_solved_blocks={:?}, manip_base_delay_ns={:?}, manip_step_batch={:?}, manip_throttle_cap={:?}, gpu_batch_duration_ms={:?}",
             self.port,
             self.cpu_workers,
             self.gpu_devices,
@@ -72,7 +75,8 @@ impl fmt::Display for ServiceConfig {
             self.manip_solved_blocks,
             self.manip_base_delay_ns,
             self.manip_step_batch,
-            self.manip_throttle_cap
+            self.manip_throttle_cap,
+            self.gpu_batch_duration_ms
         )
     }
 }
@@ -121,11 +125,7 @@ impl MiningService {
             return Err("Job already exists".to_string());
         }
 
-        log::info!(
-            "Adding mining job: {} with {} workers",
-            job_id,
-            self.cpu_workers + self.gpu_devices
-        );
+        log::info!("Adding mining job: {}", job_id,);
         job.job_id = Some(job_id.clone());
         #[cfg(feature = "metrics")]
         {
@@ -588,7 +588,9 @@ impl MiningJob {
         }
 
         while let Some(handle) = self.thread_handles.pop() {
-            let _ = handle.join();
+            // Do not wait for threads to finish; detach them so they can exit gracefully
+            // when they detect the cancellation flag. This prevents blocking the control loop.
+            drop(handle);
         }
     }
 
@@ -779,14 +781,26 @@ fn mine_range_with_engine_typed(
     progress_interval_ms: u64,
     chunk_size: Option<u64>,
 ) {
-    log::info!(
-        "⛏️  Job {} {} thread {} mining range {} to {} (inclusive)",
-        job_id,
-        engine_type,
-        thread_id,
-        range.start,
-        range.end
-    );
+    if engine_type == "CPU" {
+        log::info!(
+            target: "miner-service",
+            "CPU thread {} search started: Job {} range {} to {} (inclusive)",
+            thread_id,
+            job_id,
+            range.start,
+            range.end
+        );
+    } else {
+        log::debug!(
+            target: "miner-service",
+            "⛏️  Job {} {} thread {} mining range {} to {} (inclusive)",
+            job_id,
+            engine_type,
+            thread_id,
+            range.start,
+            range.end
+        );
+    }
 
     // Chunk the range into subranges to emit periodic progress updates for metrics
     let mut current_start = range.start;
@@ -796,19 +810,42 @@ fn mine_range_with_engine_typed(
     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
+            // GPU can handle much larger chunks efficiently.
+            // Increased to 4B to allow auto-tuned batches (e.g. 3s @ 1GH/s = 3B) to grow sufficiently.
+            4_000_000_000
         } else if engine.name() == "hybrid" {
             // Hybrid engines use GPU-sized chunks since they route to GPU workers
-            100_000_000 // 100M hashes per chunk for hybrid
+            4_000_000_000
         } 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)
         }
     });
+
+    // Log the effective chunk size to assist in debugging/tuning
+    if engine_type == "GPU" {
+        log::info!(
+            target: "miner-service",
+            "Job {} GPU thread {} configured with chunk size: {}",
+            job_id,
+            thread_id,
+            chunk_size_u64
+        );
+    } else {
+        log::debug!(
+            target: "miner-service",
+            "Job {} {} thread {} configured with chunk size: {}",
+            job_id,
+            engine_type,
+            thread_id,
+            chunk_size_u64
+        );
+    }
+
     let chunk_size = U512::from(chunk_size_u64);
     let mut done = false;
+    let mut total_hashes_processed = 0u64;
 
     while current_start <= end && !cancel_flag.load(Ordering::Relaxed) {
         let mut current_end = current_start
@@ -824,7 +861,7 @@ fn mine_range_with_engine_typed(
         };
 
         log::debug!(
-            "Job {} {} thread {} processing subrange {}..{} (inclusive)",
+            "Job {} {} thread {} starting search on subrange {}..{} (inclusive)",
             job_id,
             engine_type,
             thread_id,
@@ -832,6 +869,20 @@ fn mine_range_with_engine_typed(
             sub_range.end
         );
         let status = engine.search_range(&ctx, sub_range.clone(), &cancel_flag);
+        let status_str = match status {
+            engine_cpu::EngineStatus::Found { .. } => "found",
+            engine_cpu::EngineStatus::Exhausted { .. } => "exhausted",
+            engine_cpu::EngineStatus::Cancelled { .. } => "cancelled",
+            engine_cpu::EngineStatus::Running { .. } => "running",
+        };
+        log::info!(
+            target: "miner-service",
+            "{} thread {} finished search, status: {} job: {}",
+            engine_type,
+            thread_id,
+            status_str,
+            job_id,
+        );
 
         match status {
             engine_cpu::EngineStatus::Found {
@@ -866,6 +917,7 @@ fn mine_range_with_engine_typed(
                 break;
             }
             engine_cpu::EngineStatus::Exhausted { hash_count } => {
+                total_hashes_processed += hash_count;
                 // Send intermediate progress update for this chunk
                 let update = ThreadResult {
                     thread_id,
@@ -894,6 +946,7 @@ fn mine_range_with_engine_typed(
                 }
             }
             engine_cpu::EngineStatus::Cancelled { hash_count } => {
+                total_hashes_processed += hash_count;
                 // Send last progress update and stop
                 let update = ThreadResult {
                     thread_id,
@@ -905,6 +958,16 @@ fn mine_range_with_engine_typed(
                 if sender.try_send(update).is_err() {
                     log::warn!(target: "miner", "Job {job_id} thread {thread_id} failed to send cancel update");
                 }
+                if engine_type == "CPU" {
+                    log::info!(
+                        target: "miner-service",
+                        "CPU thread {} search cancelled: Job {} processed {} hashes (total: {})",
+                        thread_id,
+                        job_id,
+                        hash_count,
+                        total_hashes_processed
+                    );
+                }
                 done = true;
                 break;
             }
@@ -919,6 +982,20 @@ fn mine_range_with_engine_typed(
         current_start = current_end.saturating_add(U512::from(1u64));
     }
 
+    // Check if loop exited due to cancellation flag
+    if !done && cancel_flag.load(Ordering::Relaxed) {
+        if engine_type == "CPU" {
+            log::info!(
+                target: "miner-service",
+                "CPU thread {} search cancelled (flag set): Job {} processed {} hashes",
+                thread_id,
+                job_id,
+                total_hashes_processed
+            );
+        }
+        done = true;
+    }
+
     if !done {
         // Signal thread completion with no final candidate
         let final_result = ThreadResult {
@@ -931,6 +1008,15 @@ fn mine_range_with_engine_typed(
         if sender.try_send(final_result).is_err() {
             log::warn!(target: "miner", "Job {job_id} thread {thread_id} failed to send completion status after chunked search");
         }
+        if engine_type == "CPU" {
+            log::info!(
+                target: "miner-service",
+                "CPU thread {} search completed: Job {} exhausted range, processed {} hashes",
+                thread_id,
+                job_id,
+                total_hashes_processed
+            );
+        }
     }
 
     // Explicitly clear thread-local GPU resources to avoid TLS order panic
@@ -982,7 +1068,7 @@ pub async fn handle_mine_request(
     request: MiningRequest,
     state: MiningService,
 ) -> Result<impl Reply, Rejection> {
-    log::debug!("Received mine request: {request:?}");
+    log::debug!(target: "miner-servce", "Mine request: {request:?}");
     if let Err(e) = validate_mining_request(&request) {
         log::warn!("Invalid mine request ({}): {}", request.job_id, e);
         #[cfg(feature = "metrics")]
@@ -1011,7 +1097,7 @@ pub async fn handle_mine_request(
 
     match state.add_job(request.job_id.clone(), job).await {
         Ok(_) => {
-            log::debug!(target: "miner", "Accepted mine request for job ID: {}", request.job_id);
+            log::debug!(target: "miner-servce", "Accepted mine request for job ID: {}", request.job_id);
             #[cfg(feature = "metrics")]
             {
                 metrics::inc_mine_requests("accepted");
@@ -1053,7 +1139,7 @@ pub async fn handle_result_request(
     job_id: String,
     state: MiningService,
 ) -> Result<impl Reply, Rejection> {
-    log::debug!("Received result request for job: {job_id}");
+    log::debug!(target: "miner-servce", "Result request: {job_id}");
 
     let job = match state.get_job(&job_id).await {
         Some(job) => job,
@@ -1163,7 +1249,7 @@ pub async fn handle_cancel_request(
     job_id: String,
     state: MiningService,
 ) -> Result<impl Reply, Rejection> {
-    log::debug!("Received cancel request for job: {job_id}");
+    log::info!(target: "miner-servce", "Cancel job: {job_id}");
 
     if state.cancel_job(&job_id).await {
         log::debug!(target: "miner", "Successfully cancelled job: {job_id}");
@@ -1252,13 +1338,17 @@ fn compute_partitions(start: U512, end: U512, workers: usize) -> Partitions {
 
 pub fn resolve_gpu_configuration(
     requested_devices: Option<usize>,
+    gpu_batch_duration_ms: Option<u64>,
 ) -> anyhow::Result<(Option<Arc<dyn MinerEngine>>, usize)> {
+    // Default to 3000ms if not specified
+    let duration = std::time::Duration::from_millis(gpu_batch_duration_ms.unwrap_or(3000));
+
     if let Some(req_count) = requested_devices {
         if req_count == 0 {
             return Ok((None, 0));
         }
         // Explicit request > 0
-        let engine = engine_gpu::GpuEngine::try_new()
+        let engine = engine_gpu::GpuEngine::try_new(duration)
             .map_err(|e| anyhow::anyhow!("Failed to initialize GPU engine: {}", e))?;
 
         let available = engine.device_count();
@@ -1272,7 +1362,7 @@ pub fn resolve_gpu_configuration(
         Ok((Some(Arc::new(engine)), req_count))
     } else {
         // Auto-detect
-        match engine_gpu::GpuEngine::try_new() {
+        match engine_gpu::GpuEngine::try_new(duration) {
             Ok(engine) => {
                 let available = engine.device_count();
                 if available > 0 {
@@ -1362,7 +1452,7 @@ pub async fn run(config: ServiceConfig) -> anyhow::Result<()> {
 
     // Initialize GPU engine if requested or for auto-detection
     let (gpu_engine, gpu_devices): (Option<Arc<dyn MinerEngine>>, usize) =
-        match resolve_gpu_configuration(config.gpu_devices) {
+        match resolve_gpu_configuration(config.gpu_devices, config.gpu_batch_duration_ms) {
             Ok((engine, count)) => (engine, count),
             Err(e) => {
                 log::error!("❌ ERROR: {}", e);
@@ -2004,4 +2094,71 @@ mod tests {
             "final completion message carries zero hash_count"
         );
     }
+
+    #[tokio::test]
+    async fn test_cancel_returns_immediately() {
+        use engine_cpu::{EngineRange, EngineStatus, MinerEngine};
+        use pow_core::JobContext;
+        use std::any::Any;
+        use std::sync::atomic::{AtomicBool, Ordering};
+        use std::sync::Arc;
+        use std::time::{Duration, Instant};
+
+        struct SlowEngine;
+
+        impl MinerEngine for SlowEngine {
+            fn name(&self) -> &'static str {
+                "slow-engine"
+            }
+            fn prepare_context(&self, header_hash: [u8; 32], difficulty: U512) -> JobContext {
+                JobContext::new(header_hash, difficulty)
+            }
+            fn search_range(
+                &self,
+                _ctx: &JobContext,
+                _range: EngineRange,
+                cancel: &AtomicBool,
+            ) -> EngineStatus {
+                // Sleep to simulate a long-running batch on GPU
+                std::thread::sleep(Duration::from_secs(2));
+
+                if cancel.load(Ordering::Relaxed) {
+                    EngineStatus::Cancelled { hash_count: 0 }
+                } else {
+                    EngineStatus::Exhausted { hash_count: 0 }
+                }
+            }
+            fn as_any(&self) -> &dyn Any {
+                self
+            }
+        }
+
+        let engine = Arc::new(SlowEngine);
+        let service = MiningService::new(1, 0, Some(engine), None, 10000, None);
+
+        let job = MiningJob::new([0u8; 32], U512::MAX, U512::zero(), U512::from(1000u64));
+
+        service
+            .add_job("slow-job".to_string(), job)
+            .await
+            .expect("Failed to add job");
+
+        // Give thread a moment to start and enter sleep
+        tokio::time::sleep(Duration::from_millis(100)).await;
+
+        let start = Instant::now();
+        service.cancel_job("slow-job").await;
+        let elapsed = start.elapsed();
+
+        // Cancel should be near-instant (<< 2 seconds)
+        assert!(
+            elapsed < Duration::from_millis(500),
+            "Cancel took too long: {:?} (expected < 500ms)",
+            elapsed
+        );
+
+        // Verify job is actually cancelled
+        let job = service.get_job("slow-job").await.unwrap();
+        assert_eq!(job.status, JobStatus::Cancelled);
+    }
 }