hashiverse_lib/tools/pow_generator/
native_parallel_pow_generator.rs

1//! Multi-core PoW search for non-WASM targets.
2//!
3//! Each `run_chunk` call runs on a `tokio::task::spawn_blocking` thread; the shared
4//! [`crate::tools::pow_generator::pow_generator::run_pool`] dispatcher fires
5//! `pool_size()` of them concurrently and refeeds whichever slot completes first. With
6//! homogeneous server cores this is roughly equivalent to the previous rayon fan-out;
7//! on heterogeneous CPUs (Apple Silicon P+E, Intel hybrid) it stops fast cores from
8//! idling while a slow core finishes its slice.
9
10#![cfg(not(target_arch = "wasm32"))]
11
12use crate::tools::pow_generator::pow_generator;
13use crate::tools::pow_generator::pow_generator::{JobTracker, PowGenerator};
14use crate::tools::types::{Hash, Pow, Salt};
15use std::sync::{Arc, Mutex};
16use tokio_util::sync::CancellationToken;
17
18pub struct NativeParallelPowGenerator {
19    tracker: Arc<Mutex<JobTracker>>,
20    cancellation_token: Option<CancellationToken>,
21}
22
23impl NativeParallelPowGenerator {
24    pub fn new() -> Self {
25        Self { tracker: Arc::new(Mutex::new(JobTracker::default())), cancellation_token: None }
26    }
27
28    /// Abort every search on this generator once `cancellation_token` is cancelled; see
29    /// [`PowGenerator::cancellation_token`].
30    pub fn with_cancellation_token(mut self, cancellation_token: CancellationToken) -> Self {
31        self.cancellation_token = Some(cancellation_token);
32        self
33    }
34}
35
36impl Default for NativeParallelPowGenerator {
37    fn default() -> Self { Self::new() }
38}
39
40#[async_trait::async_trait]
41impl PowGenerator for NativeParallelPowGenerator {
42    fn pool_size(&self) -> usize {
43        std::thread::available_parallelism().map(|n| n.get()).unwrap_or(1)
44    }
45
46    async fn run_chunk(&self, _slot: usize, chunk_iterations: usize, pow_min: Pow, data_hash: Hash) -> anyhow::Result<(Salt, Pow, Hash)> {
47        let result = tokio::task::spawn_blocking(move || -> anyhow::Result<(Salt, Pow, Hash)> {
48            pow_generator::run_pool_chunk(chunk_iterations, pow_min, data_hash)
49        })
50        .await??;
51
52        Ok(result)
53    }
54
55    fn tracker(&self) -> &Arc<Mutex<JobTracker>> {
56        &self.tracker
57    }
58
59    fn cancellation_token(&self) -> Option<&CancellationToken> {
60        self.cancellation_token.as_ref()
61    }
62}
63
64#[cfg(test)]
65mod tests {
66    use crate::tools::pow::pow_compute_data_hash;
67    use crate::tools::pow_generator::native_parallel_pow_generator::NativeParallelPowGenerator;
68    use crate::tools::pow_generator::pow_generator::PowGenerator;
69    use crate::tools::tools;
70    use crate::tools::types::Pow;
71    use std::sync::Arc;
72    use std::time::Duration;
73    use tokio_util::sync::CancellationToken;
74
75    #[tokio::test]
76    async fn native_generates_valid_pow() -> anyhow::Result<()> {
77        const POW_MIN: Pow = Pow(12);
78        let mut data = [0u8; 64];
79        tools::random_fill_bytes(&mut data);
80        let data_hash = pow_compute_data_hash(&[&data]);
81        let generator = NativeParallelPowGenerator::new();
82        let (_, pow, _) = generator.generate("test", POW_MIN, data_hash).await?;
83        assert!(pow >= POW_MIN);
84        Ok(())
85    }
86
87    // Pow(255) is unreachable, so these searches only ever end through cancellation.
88    const UNREACHABLE_POW: Pow = Pow(255);
89
90    #[tokio::test]
91    async fn pre_cancelled_generator_fails_immediately() {
92        let cancellation_token = CancellationToken::new();
93        cancellation_token.cancel();
94        let generator = NativeParallelPowGenerator::new().with_cancellation_token(cancellation_token);
95        let data_hash = pow_compute_data_hash(&[b"pre-cancelled"]);
96
97        let result = tokio::time::timeout(Duration::from_secs(2), generator.generate("cancel", UNREACHABLE_POW, data_hash)).await.expect("generate did not return promptly");
98        assert!(result.is_err_and(|error| error.to_string().contains("cancelled")));
99        assert!(generator.active_jobs().is_empty());
100    }
101
102    #[tokio::test(flavor = "multi_thread")]
103    async fn cancel_mid_search_stops_generate() {
104        let cancellation_token = CancellationToken::new();
105        let generator = Arc::new(NativeParallelPowGenerator::new().with_cancellation_token(cancellation_token.clone()));
106        let data_hash = pow_compute_data_hash(&[b"cancel-mid-search"]);
107
108        let search = {
109            let generator = generator.clone();
110            tokio::spawn(async move { generator.generate("cancel", UNREACHABLE_POW, data_hash).await })
111        };
112        tokio::time::sleep(Duration::from_millis(100)).await;
113        assert!(!generator.active_jobs().is_empty(), "search should be running");
114        cancellation_token.cancel();
115
116        let result = tokio::time::timeout(Duration::from_secs(2), search).await.expect("generate did not stop within 2s of cancellation").unwrap();
117        assert!(result.is_err_and(|error| error.to_string().contains("cancelled")));
118        assert!(generator.active_jobs().is_empty(), "cancelled job must be removed from the tracker");
119    }
120
121    #[tokio::test(flavor = "multi_thread")]
122    async fn cancel_mid_search_stops_generate_best_effort() {
123        let cancellation_token = CancellationToken::new();
124        let generator = Arc::new(NativeParallelPowGenerator::new().with_cancellation_token(cancellation_token.clone()));
125        let data_hash = pow_compute_data_hash(&[b"cancel-best-effort"]);
126
127        let search = {
128            let generator = generator.clone();
129            tokio::spawn(async move { generator.generate_best_effort("cancel", usize::MAX, UNREACHABLE_POW, data_hash).await })
130        };
131        tokio::time::sleep(Duration::from_millis(100)).await;
132        cancellation_token.cancel();
133
134        let result = tokio::time::timeout(Duration::from_secs(2), search).await.expect("generate_best_effort did not stop within 2s of cancellation").unwrap();
135        assert!(result.is_err_and(|error| error.to_string().contains("cancelled")));
136    }
137
138    #[tokio::test]
139    async fn uncancelled_token_behaves_like_no_token() -> anyhow::Result<()> {
140        const POW_MIN: Pow = Pow(12);
141        let generator = NativeParallelPowGenerator::new().with_cancellation_token(CancellationToken::new());
142        let data_hash = pow_compute_data_hash(&[b"uncancelled"]);
143        let (_, pow, _) = generator.generate("test", POW_MIN, data_hash).await?;
144        assert!(pow >= POW_MIN);
145        let (_, best_effort_pow, _) = generator.generate_best_effort("test", 4096, POW_MIN, data_hash).await?;
146        assert!(best_effort_pow > Pow(0));
147        Ok(())
148    }
149}
hashiverse_lib/tools/pow_generator/native_parallel_pow_generator.rs

hashiverse_lib/tools/pow_generator/
native_parallel_pow_generator.rs
