fix clippy

applications/tari_validator_node/Cargo.toml

@@ -32,6 +32,7 @@ blake2 = "0.9.2"
 clap = "2.33.3"
 digest = "0.9.0"
 futures = { version = "^0.3.1" }
+itertools = "0.10.3"
 log = { version = "0.4.8", features = ["std"] }
 lmdb-zero = "0.4.4"
 prost = "0.9"

applications/tari_validator_node/src/asset.rs

@@ -22,7 +22,10 @@
 
 use std::{
     collections::HashMap,
-    sync::{Arc, Mutex},
+    sync::{
+        atomic::{AtomicBool, Ordering},
+        Arc,
+    },
 };
 
 use tari_dan_core::models::AssetDefinition;
@@ -34,7 +37,7 @@ pub struct Asset {
     // Changes in the committee for this asset.
     // Mined height of the change TXs, and the involvment in the committe (true = part of committee)
     next_states: HashMap<u64, bool>,
-    kill_signal: Option<Arc<Mutex<bool>>>,
+    kill_signal: Option<Arc<AtomicBool>>,
 }
 
 impl Asset {
@@ -48,7 +51,7 @@ impl Asset {
     }
 
     pub fn update_height<Fstart>(&mut self, height: u64, start: Fstart)
-    where Fstart: Fn(AssetDefinition) -> Arc<Mutex<bool>> {
+    where Fstart: Fn(AssetDefinition) -> Arc<AtomicBool> {
         if let Some((&height, &involment)) = self
             .next_states
             .iter()
@@ -61,7 +64,7 @@ impl Asset {
                 } else {
                     // Switch on the kill signal for the asset to end processing
                     let stop = self.kill_signal.clone().unwrap();
-                    *stop.as_ref().lock().unwrap() = true;
+                    stop.as_ref().store(true, Ordering::Relaxed);
                     self.kill_signal = None;
                 }
             }

applications/tari_validator_node/src/dan_node.rs

@@ -21,10 +21,11 @@
 //  USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 
 use std::{
-    sync::{Arc, Mutex},
+    sync::{atomic::AtomicBool, Arc},
     time::Duration,
 };
 
+use itertools::Itertools;
 use log::*;
 use tari_common::{
     configuration::ValidatorNodeConfig,
@@ -128,12 +129,11 @@ impl DanNode {
                     monitoring.add_if_unmonitored(asset.clone());
                     monitoring.add_state(asset.public_key, mined_height, true);
                 }
-                let known_active_public_keys: Vec<PublicKey> =
-                    assets.into_iter().map(|(asset, _)| asset.public_key).collect();
+                let mut known_active_public_keys = assets.into_iter().map(|(asset, _)| asset.public_key);
                 let active_public_keys = monitoring
                     .get_active_public_keys()
                     .into_iter()
-                    .map(|public_key| public_key.clone())
+                    .cloned()
                     .collect::<Vec<PublicKey>>();
                 for public_key in active_public_keys {
                     if !known_active_public_keys.contains(&public_key) {
@@ -159,7 +159,7 @@ impl DanNode {
                     let subscription_factory = subscription_factory.clone();
                     let shutdown = shutdown.clone();
                     // Create a kill signal for each asset
-                    let kill = Arc::new(Mutex::new(false));
+                    let kill = Arc::new(AtomicBool::new(false));
                     let dan_config = dan_config.clone();
                     let db_factory = db_factory.clone();
                     task::spawn(DanNode::start_asset_worker(
@@ -189,7 +189,7 @@ impl DanNode {
         shutdown: ShutdownSignal,
         config: ValidatorNodeConfig,
         db_factory: SqliteDbFactory,
-        kill: Arc<Mutex<bool>>,
+        kill: Arc<AtomicBool>,
     ) -> Result<(), ExitError> {
         let timeout = Duration::from_secs(asset_definition.phase_timeout);
         let committee = asset_definition

applications/tari_validator_node/src/monitoring.rs

@@ -22,7 +22,7 @@
 
 use std::{
     collections::HashMap,
-    sync::{Arc, Mutex},
+    sync::{atomic::AtomicBool, Arc},
 };
 
 use tari_common_types::types::PublicKey;
@@ -46,7 +46,7 @@ impl Monitoring {
 
     pub fn add_if_unmonitored(&mut self, asset: AssetDefinition) {
         if !self.assets.contains_key(&asset.public_key) {
-            self.assets.insert(asset.public_key.clone(), Asset::new(asset.clone()));
+            self.assets.insert(asset.public_key.clone(), Asset::new(asset));
         }
     }
 
@@ -59,8 +59,8 @@ impl Monitoring {
     }
 
     pub fn update_height<Fstart: Clone>(&mut self, height: u64, start: Fstart)
-    where Fstart: Fn(AssetDefinition) -> Arc<Mutex<bool>> {
-        for (_, proc) in &mut self.assets {
+    where Fstart: Fn(AssetDefinition) -> Arc<AtomicBool> {
+        for proc in self.assets.values_mut() {
             proc.update_height(height, start.clone());
         }
         self.assets.retain(|_, proc| proc.should_monitor())

dan_layer/core/src/workers/consensus_worker.rs

@@ -20,7 +20,10 @@
 // WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE
 // USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 
-use std::sync::{Arc, Mutex};
+use std::sync::{
+    atomic::{AtomicBool, Ordering},
+    Arc,
+};
 
 use log::*;
 use tari_common_types::types::PublicKey;
@@ -117,7 +120,7 @@ impl<TSpecification: ServiceSpecification<Addr = PublicKey>> ConsensusWorker<TSp
         &mut self,
         shutdown: ShutdownSignal,
         max_views_to_process: Option<u64>,
-        stop: Arc<Mutex<bool>>,
+        stop: Arc<AtomicBool>,
     ) -> Result<(), DigitalAssetError> {
         let chain_db = self
             .db_factory
@@ -131,7 +134,7 @@ impl<TSpecification: ServiceSpecification<Addr = PublicKey>> ConsensusWorker<TSp
             "Consensus worker started for asset '{}'. Tip: {}", self.asset_definition.public_key, self.current_view_id
         );
         let starting_view = self.current_view_id;
-        while !*stop.as_ref().lock().unwrap() {
+        while !stop.as_ref().load(Ordering::Relaxed) {
             if let Some(max) = max_views_to_process {
                 if max <= self.current_view_id.0 - starting_view.0 {
                     break;