Merge pull request #699 from AleoHQ/staging

Prepare for v1.4.1

Cargo.toml

@@ -136,7 +136,7 @@ version = "0.2"
 version = "0.11.2"
 
 [build-dependencies]
-rustc_version = "0.2"
+rustc_version = "0.3"
 
   [build-dependencies.capnpc]
   version = "0.14"

benchmarks/syncing/syncing.rs

@@ -23,8 +23,13 @@ use snarkos_testing::network::{blocks::*, handshaken_node_and_peer, TestSetup};
 fn providing_sync_blocks(c: &mut Criterion) {
     let rt = tokio::runtime::Runtime::new().unwrap();
 
+    let test_setup = TestSetup {
+        tokio_handle: Some(rt.handle().clone()),
+        ..Default::default()
+    };
+
     // prepare the block provider node and a fake requester node
-    let (provider, requester) = rt.block_on(handshaken_node_and_peer(TestSetup::default()));
+    let (provider, requester) = rt.block_on(handshaken_node_and_peer(test_setup));
     let requester = tokio::sync::Mutex::new(requester);
 
     const NUM_BLOCKS: usize = 10;

consensus/src/consensus.rs

@@ -95,7 +95,7 @@ impl<S: Storage> Consensus<S> {
                 self.parameters
                     .verify_header(&block.header, &parent_block.header, &merkle_root, &pedersen_merkle_root)
             {
-                println!("header failed to verify: {:?}", err);
+                error!("block header failed to verify: {:?}", err);
                 return Ok(false);
             }
         }
@@ -116,15 +116,15 @@ impl<S: Storage> Consensus<S> {
 
         // Check that there is only 1 coinbase transaction
         if coinbase_transaction_count > 1 {
-            println!("error - multiple coinbase transactions");
+            error!("multiple coinbase transactions");
             return Ok(false);
         }
 
         // Check that the block value balances are correct
         let expected_block_reward = crate::get_block_reward(self.ledger.len() as u32).0;
         if total_value_balance.0 + expected_block_reward != 0 {
-            println!("total_value_balance: {:?}", total_value_balance);
-            println!("expected_block_reward: {:?}", expected_block_reward);
+            trace!("total_value_balance: {:?}", total_value_balance);
+            trace!("expected_block_reward: {:?}", expected_block_reward);
 
             return Ok(false);
         }

consensus/src/difficulty.rs

@@ -38,8 +38,8 @@ pub fn naive_retarget(
         x *= parent_diff;
         x = parent_diff - x;
 
-        println!("old difficulty        {:#x}", parent_difficulty);
-        println!("new difficulty        {:#x}", x as u64);
+        trace!("old difficulty        {:#x}", parent_difficulty);
+        trace!("new difficulty        {:#x}", x as u64);
 
         x as u64
     }
@@ -91,8 +91,8 @@ pub fn ethereum_retarget(block_timestamp: i64, parent_timestamp: i64, parent_dif
     x *= y;
     x += parent_diff;
 
-    println!("old difficulty        {:#x}", parent_difficulty);
-    println!("new difficulty        {:#x}", x as u64);
+    trace!("old difficulty        {:#x}", parent_difficulty);
+    trace!("new difficulty        {:#x}", x as u64);
 
     x as u64
 }

network/Cargo.toml

@@ -115,4 +115,4 @@ path = "../testing"
 version = "0.1.0"
 
 [build-dependencies]
-rustc_version = "0.2"
+rustc_version = "0.3"

network/src/environment.rs → network/src/config.rs

@@ -16,20 +16,17 @@
 
 use crate::NetworkError;
 
-use once_cell::sync::OnceCell;
 use parking_lot::RwLock;
-use rand::{thread_rng, Rng};
 use std::{
     net::SocketAddr,
     time::Duration,
     {self},
 };
 
-/// A core data structure containing the networking parameters for this node.
-pub struct Environment {
-    pub name: u64,
-    /// The local address of this node.
-    local_address: OnceCell<SocketAddr>,
+/// A core data structure containing the pre-configured parameters for the node.
+pub struct Config {
+    /// The pre-configured desired address of this node.
+    pub desired_address: Option<SocketAddr>,
     /// The minimum number of peers required to maintain connections with.
     minimum_number_of_connected_peers: u16,
     /// The maximum number of peers permitted to maintain connections with.
@@ -43,10 +40,11 @@ pub struct Environment {
     peer_sync_interval: Duration,
 }
 
-impl Environment {
+impl Config {
     /// Creates a new instance of `Environment`.
     #[allow(clippy::too_many_arguments)]
     pub fn new(
+        desired_address: Option<SocketAddr>,
         minimum_number_of_connected_peers: u16,
         maximum_number_of_connected_peers: u16,
         bootnodes_addresses: Vec<String>,
@@ -61,13 +59,8 @@ impl Environment {
             }
         }
 
-        // Generate the node name.
-        let mut rng = thread_rng();
-        let name = rng.gen();
-
         Ok(Self {
-            local_address: Default::default(),
-            name,
+            desired_address,
             minimum_number_of_connected_peers,
             maximum_number_of_connected_peers,
             bootnodes: RwLock::new(bootnodes),
@@ -76,20 +69,6 @@ impl Environment {
         })
     }
 
-    /// Returns the local address of the node.
-    #[inline]
-    pub fn local_address(&self) -> Option<SocketAddr> {
-        self.local_address.get().copied()
-    }
-
-    /// Sets the local address of the node to the given value.
-    #[inline]
-    pub fn set_local_address(&self, addr: SocketAddr) {
-        self.local_address
-            .set(addr)
-            .expect("local address was set more than once!");
-    }
-
     /// Returns the default bootnodes of the network.
     #[inline]
     pub fn bootnodes(&self) -> Vec<SocketAddr> {

network/src/consensus/consensus.rs

@@ -14,7 +14,7 @@
 // You should have received a copy of the GNU General Public License
 // along with the snarkOS library. If not, see <https://www.gnu.org/licenses/>.
 
-use crate::Node;
+use crate::{Node, State};
 use snarkos_consensus::{ConsensusParameters, MemoryPool, MerkleTreeLedger};
 use snarkvm_dpc::base_dpc::{
     instantiated::{Components, Tx},
@@ -24,10 +24,8 @@ use snarkvm_objects::Storage;
 
 use parking_lot::{Mutex, RwLock};
 use std::{
-    sync::{
-        atomic::{AtomicBool, Ordering},
-        Arc,
-    },
+    net::SocketAddr,
+    sync::Arc,
     time::{Duration, Instant},
 };
 
@@ -42,11 +40,9 @@ pub struct Consensus<S: Storage> {
     /// The interval between each block sync.
     block_sync_interval: Duration,
     /// The last time a block sync was initiated.
-    last_block_sync: RwLock<Instant>,
+    last_block_sync: RwLock<Option<Instant>>,
     /// The interval between each transaction (memory pool) sync.
     transaction_sync_interval: Duration,
-    /// Is the node currently syncing blocks?
-    is_syncing_blocks: AtomicBool,
 }
 
 impl<S: Storage> Consensus<S> {
@@ -63,9 +59,8 @@ impl<S: Storage> Consensus<S> {
             consensus,
             is_miner,
             block_sync_interval,
-            last_block_sync: RwLock::new(Instant::now()),
+            last_block_sync: Default::default(),
             transaction_sync_interval,
-            is_syncing_blocks: Default::default(),
         }
     }
 
@@ -106,12 +101,12 @@ impl<S: Storage> Consensus<S> {
 
     /// Checks whether the node is currently syncing blocks.
     pub fn is_syncing_blocks(&self) -> bool {
-        self.is_syncing_blocks.load(Ordering::SeqCst)
+        self.node.state() == State::Syncing
     }
 
     /// Register that the node is no longer syncing blocks.
     pub fn finished_syncing_blocks(&self) {
-        self.is_syncing_blocks.store(false, Ordering::SeqCst);
+        self.node.set_state(State::Idle);
     }
 
     /// Returns the current block height of the ledger from storage.
@@ -120,15 +115,20 @@ impl<S: Storage> Consensus<S> {
         self.consensus.ledger.get_current_block_height()
     }
 
-    /// Checks whether enough time has elapsed for the node to attempt another block sync.
-    pub fn should_sync_blocks(&self) -> bool {
-        !self.is_syncing_blocks() && self.last_block_sync.read().elapsed() > self.block_sync_interval
+    /// Checks whether the conditions for the node to attempt another block sync are met.
+    pub fn should_sync_blocks(&self, peer_block_height: u32) -> bool {
+        peer_block_height > self.current_block_height() + 1
+            && if let Some(ref timestamp) = *self.last_block_sync.read() {
+                timestamp.elapsed() > self.block_sync_interval
+            } else {
+                true
+            }
     }
 
-    /// Register that the node attempted to sync blocks.
-    pub fn register_block_sync_attempt(&self) {
-        *self.last_block_sync.write() = Instant::now();
-        self.is_syncing_blocks.store(true, Ordering::SeqCst);
+    /// Register that the node attempted to sync blocks with the given peer.
+    pub fn register_block_sync_attempt(&self, provider: SocketAddr) {
+        trace!("Attempting to sync with {}", provider);
+        *self.last_block_sync.write() = Some(Instant::now());
     }
 
     /// Returns the interval between each transaction (memory pool) sync.

network/src/consensus/miner.rs

@@ -14,15 +14,15 @@
 // You should have received a copy of the GNU General Public License
 // along with the snarkOS library. If not, see <https://www.gnu.org/licenses/>.
 
-use crate::Node;
+use crate::{Node, State};
 use snarkos_consensus::Miner;
 use snarkvm_dpc::{base_dpc::instantiated::*, AccountAddress};
 use snarkvm_objects::Storage;
 
-use tokio::task;
+use tokio::runtime;
 use tracing::*;
 
-use std::sync::Arc;
+use std::{sync::Arc, thread, time::Duration};
 
 /// Parameters for spawning a miner that runs proof of work to find a block.
 pub struct MinerInstance<S: Storage> {
@@ -39,26 +39,44 @@ impl<S: Storage + Send + Sync + 'static> MinerInstance<S> {
     /// Spawns a new miner on a new thread using MinerInstance parameters.
     /// Once a block is found, A block message is sent to all peers.
     /// Calling this function multiple times will spawn additional listeners on separate threads.
-    /// Miner threads are asynchronous so the only way to stop them is to kill the runtime they were started in. This may be changed in the future.
-    pub fn spawn(self) -> task::JoinHandle<()> {
-        task::spawn(async move {
-            let local_address = self.node.environment.local_address().unwrap();
-            info!("Initializing Aleo miner - Your miner address is {}", self.miner_address);
-            let miner = Miner::new(
-                self.miner_address.clone(),
-                Arc::clone(&self.node.expect_consensus().consensus),
-            );
-            info!("Miner instantiated; starting to mine blocks");
-
-            let mut mining_failure_count = 0;
-            let mining_failure_threshold = 10;
-
+    pub fn spawn(self, tokio_handle: runtime::Handle) -> thread::JoinHandle<()> {
+        let local_address = self.node.local_address().unwrap();
+        info!("Initializing Aleo miner - Your miner address is {}", self.miner_address);
+        let miner = Miner::new(
+            self.miner_address.clone(),
+            Arc::clone(&self.node.expect_consensus().consensus),
+        );
+        info!("Miner instantiated; starting to mine blocks");
+
+        let mut mining_failure_count = 0;
+        let mining_failure_threshold = 10;
+
+        thread::spawn(move || {
             loop {
+                if self.node.is_shutting_down() {
+                    debug!("The node is shutting down, stopping mining");
+                    break;
+                }
+
+                // don't mine if the node is currently syncing
+                if self.node.state() == State::Syncing {
+                    thread::sleep(Duration::from_secs(5));
+                    continue;
+                } else {
+                    self.node.set_state(State::Mining);
+                }
+
                 info!("Starting to mine the next block");
 
                 let (block, _coinbase_records) = match miner.mine_block() {
                     Ok(mined_block) => mined_block,
                     Err(error) => {
+                        // it's possible that the node realized that it needs to sync with a nother one in the
+                        // meantime; don't change to `Idle` if the current status isn't still `Mining`
+                        if self.node.state() == State::Mining {
+                            self.node.set_state(State::Idle);
+                        }
+
                         warn!(
                             "Miner failed to mine a block {} time(s). (error message: {}).",
                             mining_failure_count, error
@@ -77,6 +95,11 @@ impl<S: Storage + Send + Sync + 'static> MinerInstance<S> {
                     }
                 };
 
+                // see the `Err` path note above
+                if self.node.state() == State::Mining {
+                    self.node.set_state(State::Idle);
+                }
+
                 info!("Mined a new block: {:?}", hex::encode(block.header.get_hash().0));
 
                 let serialized_block = if let Ok(block) = block.serialize() {
@@ -86,10 +109,12 @@ impl<S: Storage + Send + Sync + 'static> MinerInstance<S> {
                     continue;
                 };
 
-                self.node
-                    .expect_consensus()
-                    .propagate_block(serialized_block, local_address)
-                    .await;
+                let node = self.node.clone();
+                tokio_handle.spawn(async move {
+                    node.expect_consensus()
+                        .propagate_block(serialized_block, local_address)
+                        .await;
+                });
             }
         })
     }