Configurable CPU usage on mine (#58)

* Configurable CPU usage on mine

* Cargo clippy

node/miner/src/config.rs

@@ -14,11 +14,14 @@ pub struct MinerConfig {
     pub(crate) mine_address: Address,
     pub(crate) flow_address: Address,
     pub(crate) submission_gas: Option<U256>,
+    pub(crate) cpu_percentage: u64,
+    pub(crate) iter_batch: usize,
 }
 
 pub type MineServiceMiddleware = SignerMiddleware<Provider<Http>, LocalWallet>;
 
 impl MinerConfig {
+    #[allow(clippy::too_many_arguments)]
     pub fn new(
         miner_id: Option<H256>,
         miner_key: Option<H256>,
@@ -26,6 +29,8 @@ impl MinerConfig {
         mine_address: Address,
         flow_address: Address,
         submission_gas: Option<U256>,
+        cpu_percentage: u64,
+        iter_batch: usize,
     ) -> Option<MinerConfig> {
         match (miner_id, miner_key) {
             (Some(miner_id), Some(miner_key)) => Some(MinerConfig {
@@ -35,6 +40,8 @@ impl MinerConfig {
                 mine_address,
                 flow_address,
                 submission_gas,
+                cpu_percentage,
+                iter_batch,
             }),
             _ => None,
         }

node/miner/src/mine.rs

@@ -1,8 +1,10 @@
 use contract_interface::zgs_flow::MineContext;
 use ethereum_types::{H256, U256};
 use rand::{self, Rng};
+use std::time;
 use task_executor::TaskExecutor;
 use tokio::sync::{broadcast, mpsc};
+use tokio::time::{sleep, Duration, Instant};
 
 use zgs_spec::{SECTORS_PER_LOAD, SECTORS_PER_MAX_MINING_RANGE, SECTORS_PER_PRICING};
 
@@ -23,6 +25,9 @@ pub struct PoraService {
     puzzle: Option<PoraPuzzle>,
     mine_range: CustomMineRange,
     miner_id: H256,
+
+    cpu_percentage: u64,
+    iter_batch: usize,
 }
 
 struct PoraPuzzle {
@@ -92,6 +97,8 @@ impl PoraService {
             mine_range,
             miner_id: config.miner_id,
             loader,
+            cpu_percentage: config.cpu_percentage,
+            iter_batch: config.iter_batch,
         };
         executor.spawn(async move { Box::pin(pora.start()).await }, "pora_master");
         mine_answer_receiver
@@ -100,6 +107,12 @@ impl PoraService {
     async fn start(mut self) {
         let mut mining_enabled = true;
         let mut channel_opened = true;
+
+        let cpu_percent: u64 = self.cpu_percentage;
+        let diastole = sleep(Duration::from_secs(0));
+        tokio::pin!(diastole);
+        // info!("CPU percent {}", cpu_percent);
+
         loop {
             tokio::select! {
                 biased;
@@ -139,14 +152,25 @@ impl PoraService {
                     }
                 }
 
-                _ = async {}, if mining_enabled && self.as_miner().is_some() => {
+                () = &mut diastole, if !diastole.is_elapsed() => {
+                }
+
+                _ = async {}, if mining_enabled && cpu_percent > 0 && self.as_miner().is_some() && diastole.is_elapsed()  => {
                     let nonce = H256(rand::thread_rng().gen());
                     let miner = self.as_miner().unwrap();
-                    if let Some(answer) = miner.iteration(nonce).await{
+
+                    let timer = time::Instant::now();
+
+                    if let Some(answer) = miner.batch_iteration(nonce, self.iter_batch).await {
                         debug!("Hit Pora answer {:?}", answer);
                         if self.mine_answer_sender.send(answer).is_err() {
                             warn!("Mine submitter channel closed");
                         }
+                    } else if cpu_percent < 100 {
+                        // 2^64 ns = 500 years
+                        let elapsed = timer.elapsed().as_nanos() as u64;
+                        let diastole_time = elapsed / cpu_percent * (100 - cpu_percent);
+                        diastole.as_mut().reset(Instant::now() + Duration::from_nanos(diastole_time));
                     }
                 }
             }

node/miner/src/pora.rs

@@ -40,6 +40,24 @@ pub struct AnswerWithoutProof {
 }
 
 impl<'a> Miner<'a> {
+    pub async fn batch_iteration(
+        &self,
+        nonce: H256,
+        batch_size: usize,
+    ) -> Option<AnswerWithoutProof> {
+        for i in 0..batch_size {
+            let bytes = i.to_ne_bytes();
+            let mut current_nonce = nonce;
+            for (pos, b) in bytes.into_iter().enumerate() {
+                current_nonce.0[pos] ^= b;
+            }
+            if let Some(answer) = self.iteration(current_nonce).await {
+                return Some(answer);
+            }
+        }
+        None
+    }
+
     pub async fn iteration(&self, nonce: H256) -> Option<AnswerWithoutProof> {
         let (scratch_pad, recall_seed) = self.make_scratch_pad(&nonce);
 

node/src/config/convert.rs

@@ -140,6 +140,8 @@ impl ZgsConfig {
             None
         };
         let submission_gas = self.miner_submission_gas.map(U256::from);
+        let cpu_percentage = self.miner_cpu_percentage;
+        let iter_batch = self.mine_iter_batch_size;
         Ok(MinerConfig::new(
             miner_id,
             miner_key,
@@ -147,6 +149,8 @@ impl ZgsConfig {
             mine_address,
             flow_address,
             submission_gas,
+            cpu_percentage,
+            iter_batch,
         ))
     }
 

node/src/config/mod.rs

@@ -64,6 +64,8 @@ build_config! {
     (miner_id, (Option<String>), None)
     (miner_key, (Option<String>), None)
     (miner_submission_gas, (Option<u64>), None)
+    (miner_cpu_percentage, (u64), 100)
+    (mine_iter_batch_size, (usize), 100)
 }
 
 #[derive(Debug, Default, Deserialize)]

tests/long_time_mine_test_local.py

@@ -0,0 +1,53 @@
+#!/usr/bin/env python3
+from test_framework.test_framework import TestFramework
+from config.node_config import MINER_ID, GENESIS_PRIV_KEY
+from utility.submission import create_submission, submit_data
+from utility.utils import wait_until
+
+
+class LongTimeMineTest(TestFramework):
+    def setup_params(self):
+        self.num_blockchain_nodes = 1
+        self.num_nodes = 1
+        self.zgs_node_configs[0] = {
+            "miner_id": MINER_ID,
+            "miner_key": GENESIS_PRIV_KEY,
+            "miner_cpu_percentage": 70,
+            "mine_iter_batch_size": 50,
+        }
+        self.mine_period = 15
+
+    def submit_data(self, item, size):
+        submissions_before = self.contract.num_submissions()
+        client = self.nodes[0]
+        chunk_data = item * 256 * size
+        submissions, data_root = create_submission(chunk_data)
+        self.contract.submit(submissions)
+        wait_until(lambda: self.contract.num_submissions() == submissions_before + 1)
+        wait_until(lambda: client.zgs_get_file_info(data_root) is not None)
+
+        segment = submit_data(client, chunk_data)
+        wait_until(lambda: client.zgs_get_file_info(data_root)["finalized"])
+
+    def run_test(self):
+        blockchain = self.blockchain_nodes[0]
+
+        self.log.info("flow address: %s", self.contract.address())
+        self.log.info("mine address: %s", self.mine_contract.address())
+
+        quality = int(2**256 / 40960 / 1_000_000)
+        self.mine_contract.set_quality(quality)
+
+        self.log.info("Submit the first data chunk")
+        self.submit_data(b"\x11", 2000)
+
+        self.log.info("Start mine")
+        wait_until(lambda: int(blockchain.eth_blockNumber(), 16) > self.mine_period, timeout=180)
+
+        self.log.info("Wait for the first mine answer")
+        wait_until(lambda: self.mine_contract.last_mined_epoch() == 1)
+
+
+if __name__ == "__main__":
+    # This test is for local run only
+    LongTimeMineTest().main()