Move and rename StatePruneManager

Move the StatePruneManager to state::state_pruning_manager and rename it
to StatePruningManager.

Additionally:

- Remove reference to the state database in the ChainController's new
function, replacing it with a StatePruningManager instance.

Signed-off-by: Peter Schwarz <pschwarz@bitwise.io>

Author: peterschwarz

validator/src/journal/chain.rs

@@ -15,7 +15,6 @@
  * ------------------------------------------------------------------------------
  */
 
-use std::collections::VecDeque;
 use std::fs::File;
 use std::io;
 use std::io::prelude::*;
@@ -38,12 +37,11 @@ use std::time::Duration;
 use protobuf;
 
 use batch::Batch;
-use database::lmdb::LmdbDatabase;
 use journal;
 use journal::block_validator::{BlockValidationResult, BlockValidator, ValidationError};
 use journal::block_wrapper::BlockWrapper;
 use metrics;
-use state::merkle::MerkleDatabase;
+use state::state_pruning_manager::StatePruningManager;
 
 use proto::transaction_receipt::TransactionReceipt;
 use scheduler::TxnExecutionResult;
@@ -152,7 +150,7 @@ struct ChainControllerState<BC: BlockCache, BV: BlockValidator, CW: ChainWriter>
     chain_head_update_observer: Box<ChainHeadUpdateObserver>,
     observers: Vec<Box<ChainObserver>>,
 
-    state_prune_manager: StatePruneManager,
+    state_pruning_manager: StatePruningManager,
 }
 
 #[derive(Clone)]
@@ -177,7 +175,7 @@ impl<BC: BlockCache + 'static, BV: BlockValidator + 'static, CW: ChainWriter + '
         chain_head_update_observer: Box<ChainHeadUpdateObserver>,
         state_pruning_block_depth: u32,
         observers: Vec<Box<ChainObserver>>,
-        state_database: LmdbDatabase,
+        state_pruning_manager: StatePruningManager,
     ) -> Self {
         let mut chain_controller = ChainController {
             state: Arc::new(RwLock::new(ChainControllerState {
@@ -190,7 +188,7 @@ impl<BC: BlockCache + 'static, BV: BlockValidator + 'static, CW: ChainWriter + '
                 chain_head_update_observer,
                 observers,
                 chain_head: None,
-                state_prune_manager: StatePruneManager::new(state_database),
+                state_pruning_manager
             })),
             stop_handle: Arc::new(Mutex::new(None)),
             block_queue_sender: None,
@@ -276,7 +274,7 @@ impl<BC: BlockCache + 'static, BV: BlockValidator + 'static, CW: ChainWriter + '
             let chain_head_block = new_block.clone();
             state.chain_head = Some(new_block);
 
-            state.state_prune_manager.update_queue(
+            state.state_pruning_manager.update_queue(
                 &result
                     .new_chain
                     .iter()
@@ -363,14 +361,14 @@ impl<BC: BlockCache + 'static, BV: BlockValidator + 'static, CW: ChainWriter + '
             if chain_head_block_num + 1 > self.state_pruning_block_depth as u64 {
                 let prune_at =  chain_head_block_num - (self.state_pruning_block_depth as u64);
                 match state.chain_reader.get_block_by_block_num(prune_at) {
-                    Ok(Some(block)) => state.state_prune_manager.add_to_queue(block.state_root_hash()),
+                    Ok(Some(block)) => state.state_pruning_manager.add_to_queue(block.state_root_hash()),
                     Ok(None) => warn!("No block at block height {}; ignoring...", prune_at),
                     Err(err) => error!("Unable to fetch block at height {}: {:?}", prune_at, err)
                 }
             }
 
             // Execute pruning:
-            state.state_prune_manager.decimate()
+            state.state_pruning_manager.decimate()
         } else {
             info!("Rejected new chain head: {}", new_block);
         }
@@ -698,92 +696,5 @@ impl ChainIdManager {
     }
 }
 
-/// The StatePruneManager manages a collection of state root hashes that will be
-/// prune from the MerkleDatabase at intervals.  Pruning will occur by decimating
-/// the state root hashes.  I.e. ten percent (rounded down) of the state roots in
-/// the queue will be pruned.  This allows state roots to remain in the queue for
-/// a period of time, on the chance that they are from a chain that has been
-/// abandoned and then re-chosen as the primary chain.
-struct StatePruneManager {
-    // Contains the state root hashes slated for pruning
-    state_root_prune_queue: VecDeque<String>,
-    state_database: LmdbDatabase,
-}
-
-impl StatePruneManager {
-    fn new(state_database: LmdbDatabase) -> Self {
-        StatePruneManager {
-            state_root_prune_queue: VecDeque::new(),
-            state_database,
-        }
-    }
-
-    /// Updates the pruning queue.  Abandoned roots will be added to the queue.
-    /// Added roots will be removed from the queue.  This ensures that the state
-    /// roots won't be removed, regardless of the chain state.
-    fn update_queue(&mut self, added_roots: &[&str], _abandoned_roots: &[&str]) {
-        // Remove any state root hashes from the pruning queue that we may have switched
-        // back too from an alternate chain
-        added_roots.iter().for_each(|state_root_hash| {
-            self.state_root_prune_queue.retain(|hash| {
-                if hash != state_root_hash {
-                    true
-                } else {
-                    debug!("Removing {} from pruning queue", state_root_hash);
-                    false
-                }
-            })
-        });
-    }
-
-    /// Add a single state root to the pruning queue.
-    fn add_to_queue(&mut self, state_root_hash: &str) {
-        let state_root_hash = state_root_hash.into();
-        if !self.state_root_prune_queue.contains(&state_root_hash) {
-            debug!("Adding {} to pruning queue", state_root_hash);
-            self.state_root_prune_queue.push_back(state_root_hash);
-        }
-    }
-
-    /// Remove the first ten percent (floored) of the prune queue.
-    fn decimate(&mut self) {
-        let prune_count = (self.state_root_prune_queue.len() as f64 / 10.0).floor() as usize;
-
-        let mut total_pruned_entries: usize = 0;
-        for _ in 0..prune_count {
-            if let Some(state_root) = self.state_root_prune_queue.pop_front() {
-                match MerkleDatabase::prune(&self.state_database, &state_root) {
-                    Ok(removed_keys) => {
-                        total_pruned_entries += removed_keys.len();
-
-                        let mut state_roots_pruned_count =
-                            COLLECTOR.counter("StatePruneManager.state_roots_pruned", None, None);
-                        state_roots_pruned_count.inc();
-
-                        // the state root was not pruned (it is likely the root of a
-                        // fork), so push it back into the queue.
-                        if removed_keys.is_empty() {
-                            self.state_root_prune_queue.push_back(state_root);
-                        }
-                    }
-                    Err(err) => {
-                        error!("Unable to prune state root {}: {:?}", state_root, err);
-                        self.state_root_prune_queue.push_back(state_root);
-                    }
-                }
-            } else {
-                break;
-            }
-        }
-
-        if total_pruned_entries > 0 {
-            info!(
-                "Pruned {} keys from the Global state Database",
-                total_pruned_entries
-            );
-        }
-    }
-}
-
 #[cfg(tests)]
 mod tests {}

validator/src/journal/chain_ffi.rs

@@ -22,6 +22,7 @@ use journal::block_wrapper::BlockWrapper;
 use journal::chain::*;
 use py_ffi;
 use pylogger;
+use state::state_pruning_manager::StatePruningManager;
 use std::ffi::CStr;
 use std::os::raw::{c_char, c_void};
 use std::sync::mpsc::Sender;
@@ -109,6 +110,8 @@ pub extern "C" fn chain_controller_new(
 
     let state_database = unsafe { (*(state_database as *const LmdbDatabase)).clone() };
 
+    let state_pruning_manager = StatePruningManager::new(state_database);
+
     let chain_controller = ChainController::new(
         PyBlockCache::new(py_block_cache),
         PyBlockValidator::new(py_block_validator),
@@ -119,7 +122,7 @@ pub extern "C" fn chain_controller_new(
         Box::new(PyChainHeadUpdateObserver::new(py_on_chain_updated)),
         state_pruning_block_depth,
         observer_wrappers,
-        state_database,
+        state_pruning_manager,
     );
 
     unsafe {

validator/src/state/mod.rs

@@ -20,6 +20,7 @@ pub mod identity_view;
 pub mod merkle;
 pub mod merkle_ffi;
 pub mod settings_view;
+pub mod state_pruning_manager;
 
 use state::error::StateDatabaseError;
 

validator/src/state/state_pruning_manager.rs

@@ -0,0 +1,113 @@
+/*
+ * Copyright 2018 Intel Corporation
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ------------------------------------------------------------------------------
+ */
+use std::collections::VecDeque;
+
+use database::lmdb::LmdbDatabase;
+use metrics;
+use state::merkle::MerkleDatabase;
+
+lazy_static! {
+    static ref COLLECTOR: metrics::MetricsCollectorHandle =
+        metrics::get_collector("sawtooth_validator.state");
+}
+
+/// The StatePruneManager manages a collection of state root hashes that will be
+/// prune from the MerkleDatabase at intervals.  Pruning will occur by decimating
+/// the state root hashes.  I.e. ten percent (rounded down) of the state roots in
+/// the queue will be pruned.  This allows state roots to remain in the queue for
+/// a period of time, on the chance that they are from a chain that has been
+/// abandoned and then re-chosen as the primary chain.
+pub struct StatePruningManager {
+    // Contains the state root hashes slated for pruning
+    state_root_prune_queue: VecDeque<String>,
+    state_database: LmdbDatabase,
+}
+
+impl StatePruningManager {
+    pub fn new(state_database: LmdbDatabase) -> Self {
+        StatePruningManager {
+            state_root_prune_queue: VecDeque::new(),
+            state_database,
+        }
+    }
+
+    /// Updates the pruning queue.  Abandoned roots will be added to the queue.
+    /// Added roots will be removed from the queue.  This ensures that the state
+    /// roots won't be removed, regardless of the chain state.
+    pub fn update_queue(&mut self, added_roots: &[&str], _abandoned_roots: &[&str]) {
+        // Remove any state root hashes from the pruning queue that we may have switched
+        // back too from an alternate chain
+        added_roots.iter().for_each(|state_root_hash| {
+            self.state_root_prune_queue.retain(|hash| {
+                if hash != state_root_hash {
+                    true
+                } else {
+                    debug!("Removing {} from pruning queue", state_root_hash);
+                    false
+                }
+            })
+        });
+    }
+
+    /// Add a single state root to the pruning queue.
+    pub fn add_to_queue(&mut self, state_root_hash: &str) {
+        let state_root_hash = state_root_hash.into();
+        if !self.state_root_prune_queue.contains(&state_root_hash) {
+            debug!("Adding {} to pruning queue", state_root_hash);
+            self.state_root_prune_queue.push_back(state_root_hash);
+        }
+    }
+
+    /// Remove the first ten percent (floored) of the prune queue.
+    pub fn decimate(&mut self) {
+        let prune_count = (self.state_root_prune_queue.len() as f64 / 10.0).floor() as usize;
+
+        let mut total_pruned_entries: usize = 0;
+        for _ in 0..prune_count {
+            if let Some(state_root) = self.state_root_prune_queue.pop_front() {
+                match MerkleDatabase::prune(&self.state_database, &state_root) {
+                    Ok(removed_keys) => {
+                        total_pruned_entries += removed_keys.len();
+
+                        let mut state_roots_pruned_count =
+                            COLLECTOR.counter("StatePruneManager.state_roots_pruned", None, None);
+                        state_roots_pruned_count.inc();
+
+                        // the state root was not pruned (it is likely the root of a
+                        // fork), so push it back into the queue.
+                        if removed_keys.is_empty() {
+                            self.state_root_prune_queue.push_back(state_root);
+                        }
+                    }
+                    Err(err) => {
+                        error!("Unable to prune state root {}: {:?}", state_root, err);
+                        self.state_root_prune_queue.push_back(state_root);
+                    }
+                }
+            } else {
+                break;
+            }
+        }
+
+        if total_pruned_entries > 0 {
+            info!(
+                "Pruned {} keys from the Global state Database",
+                total_pruned_entries
+            );
+        }
+    }
+}