change TransactionCommitCache to LRU

TransactionCommitCache is currently based on a HashSet which depends
on its clients to properly clean up their items in a timely fashion.
However, under high load circumstances, this can be too long resulting
in quiet OOM issues and in practice freezing the whole network.

This changes the TransactionCommitCache to an uluru LRU. This is safe
since the commit cache is based on data that cannot change and the
worst that happens on a miss is a refetch from the store.

uluru is also updated to 3.0.0 which requires a minor change to
`block_validator.rs` to account for the changed construction

Signed-off-by: Kevin O'Donnell <kevin.odonnell@btp.works>

Author: scealiontach

validator/Cargo.toml

@@ -15,7 +15,7 @@ cbor-codec = "0.7"
 openssl = "0.10"
 protobuf = "2.0"
 python3-sys = "0.2"
-uluru = "0.2.0"
+uluru = "3.0.0"
 
 [dev-dependencies]
 rand = "0.4"

validator/src/journal/block_validator.rs

@@ -57,7 +57,7 @@ pub enum ValidationError {
 }
 
 type BlockValidationResultCache =
-    uluru::LRUCache<[uluru::Entry<BlockValidationResult>; BLOCK_VALIDATION_RESULT_CACHE_SIZE]>;
+    uluru::LRUCache<BlockValidationResult, BLOCK_VALIDATION_RESULT_CACHE_SIZE>;
 
 #[derive(Clone, Default)]
 pub struct BlockValidationResultStore {
@@ -73,7 +73,7 @@ impl BlockValidationResultStore {
         self.validation_result_cache
             .lock()
             .expect("The mutex is poisoned")
-            .insert(result)
+            .insert(result);
     }
 
     pub fn get(&self, block_id: &str) -> Option<BlockValidationResult> {

validator/src/journal/chain_commit_state.rs

@@ -20,6 +20,7 @@ use std::collections::HashSet;
 use batch::Batch;
 use journal::block_manager::BlockManager;
 use journal::commit_store::CommitStore;
+use std::sync::{Arc, RwLock};
 use transaction::Transaction;
 
 #[derive(Debug, PartialEq)]
@@ -124,21 +125,26 @@ pub fn validate_transaction_dependencies(
     Ok(())
 }
 
+type TransactionCommitCacheLRU = uluru::LRUCache<String, 300>;
+
 pub struct TransactionCommitCache {
-    committed: HashSet<String>,
+    committed: Arc<RwLock<TransactionCommitCacheLRU>>,
     commit_store: CommitStore,
 }
 
 impl TransactionCommitCache {
     pub fn new(commit_store: CommitStore) -> Self {
         TransactionCommitCache {
-            committed: HashSet::new(),
+            committed: Arc::new(RwLock::new(TransactionCommitCacheLRU::default())),
             commit_store,
         }
     }
 
     pub fn add(&mut self, transaction_id: String) {
-        self.committed.insert(transaction_id);
+        self.committed
+            .write()
+            .expect("Failed to acquire write lock on commit cache")
+            .insert(transaction_id);
     }
 
     pub fn add_batch(&mut self, batch: &Batch) {
@@ -148,8 +154,8 @@ impl TransactionCommitCache {
             .for_each(|txn| self.add(txn.header_signature.clone()));
     }
 
-    pub fn remove(&mut self, transaction_id: &str) {
-        self.committed.remove(transaction_id);
+    fn remove(&mut self, transaction_id: &str) {
+        ()
     }
 
     pub fn remove_batch(&mut self, batch: &Batch) {
@@ -161,7 +167,11 @@ impl TransactionCommitCache {
 
     pub fn contains(&self, transaction_id: &str) -> bool {
         // Shouldn't expect here
-        self.committed.contains(transaction_id)
+        self.committed
+            .write()
+            .expect("Failed to acquire write lock on commit cache")
+            .find(|x| x == transaction_id)
+            .is_some()
             || self
                 .commit_store
                 .contains_transaction(transaction_id)
@@ -281,7 +291,7 @@ mod test {
         let block_manager = setup_state();
 
         let transactions: Vec<Transaction> = ["B6b0t0", "B6b0t1", "B6b0t2"]
-            .into_iter()
+            .iter()
             .map(|t_id| create_transaction((*t_id).into(), vec!["B2b0t0".into()]))
             .collect();
 
@@ -300,7 +310,7 @@ mod test {
         let block_manager = setup_state();
 
         let transactions: Vec<Transaction> = ["B6b0t0", "B6b0t1", "B6b0t2"]
-            .into_iter()
+            .iter()
             .map(|t_id| create_transaction((*t_id).into(), vec!["B1b0t0".into()]))
             .collect();
 
@@ -319,7 +329,7 @@ mod test {
         let block_manager = setup_state();
 
         let transactions: Vec<Transaction> = ["B3-1b0t0", "B3-1b0t1", "B3-1b0t2"]
-            .into_iter()
+            .iter()
             .map(|t_id| create_transaction((*t_id).into(), vec!["B1b0t0".into()]))
             .collect();
 
@@ -338,7 +348,7 @@ mod test {
         let block_manager = setup_state();
 
         let transactions: Vec<Transaction> = ["B6b0t0", "B6b0t1", "B6b0t2"]
-            .into_iter()
+            .iter()
             .map(|t_id| create_transaction((*t_id).into(), vec!["B4-3b0t0".into()]))
             .collect();