*: react to new mempool

Signed-off-by: Sam Batschelet <[email protected]>

Author: hexfusion

mini-kvvm/src/mempool/mod.rs

@@ -14,7 +14,12 @@ use self::data::{Data, Entry};
 
 pub struct Mempool {
     data: Arc<RwLock<Data>>,
+
+    /// Channel of length one, which the mempool ensures has an item on
+    /// it as long as there is an unissued transaction remaining in [txs].
     pending_tx: broadcast::Sender<()>,
+
+    /// Vec of [Tx] that are ready to be gossiped.
     new_txs: Vec<Transaction>,
 }
 
@@ -25,8 +30,6 @@ impl Mempool {
             tokio::sync::broadcast::channel(1);
         Self {
             data: Arc::new(RwLock::new(Data::new(max_size))),
-            /// Channel of length one, which the mempool ensures has an item on
-            /// it as long as there is an unissued transaction remaining in [txs].
             pending_tx,
             new_txs: Vec::new(),
         }
@@ -74,6 +77,15 @@ impl Mempool {
         Ok(true)
     }
 
+    /// Return
+    pub fn pop_back(&self) -> Option<Transaction> {
+        let mut data = self.data.write().unwrap();
+        match data.items.pop_back() {
+            Some(entry) => entry.tx,
+            None => None,
+        }
+    }
+
     /// Returns len of mempool data.
     pub fn len(&self) -> usize {
         let data = self.data.read().unwrap();
@@ -85,6 +97,25 @@ impl Mempool {
         data.is_empty()
     }
 
+    /// Returns the vec of transactions ready to gossip and replaces it with an empty vec.
+    pub fn new_txs(&mut self) -> Result<Vec<Transaction>> {
+        let data = self.data.read().unwrap();
+
+        let mut selected: Vec<Transaction> = Vec::new();
+
+        // It is possible that a block may have been accepted that contains some
+        // new transactions before [new_txs] is called.
+        for tx in self.new_txs.iter() {
+            if data.has(&tx.id)? {
+                continue;
+            }
+            selected.push(tx.to_owned())
+        }
+        self.new_txs = Vec::new();
+
+        Ok(selected)
+    }
+
     /// Prunes any Ids not included in valid hashes set.
     pub fn prune(&self, valid_hashes: ids::Set) {
         let mut to_remove: Vec<ids::Id> = Vec::with_capacity(valid_hashes.len());

mini-kvvm/src/vm.rs

@@ -1,5 +1,5 @@
 use std::{
-    collections::{HashMap, VecDeque},
+    collections::HashMap,
     io::{Error, ErrorKind, Result},
     sync::Arc,
     time,
@@ -23,11 +23,15 @@ use crate::{
     block::{self, state::State},
     chain::{self, tx::Transaction, vm::Vm},
     genesis::Genesis,
+    mempool,
 };
 
 const PUBLIC_API_ENDPOINT: &str = "/public";
 const VERSION: &str = env!("CARGO_PKG_VERSION");
 
+// TODO: make configurable
+const MEMPOOL_SIZE: usize = 1024;
+
 pub struct ChainVmInterior {
     pub ctx: Option<rpcchainvm::context::Context>,
     pub bootstrapped: bool,
@@ -59,29 +63,29 @@ impl Default for ChainVmInterior {
 #[derive(Clone)]
 pub struct ChainVm {
     pub db: Box<dyn rpcchainvm::database::Database + Sync + Send>,
-    pub mempool: Arc<RwLock<VecDeque<chain::tx::tx::Transaction>>>,
+    pub mempool: Arc<RwLock<mempool::Mempool>>,
     pub inner: Arc<RwLock<ChainVmInterior>>,
-    pub verified_blocks: Arc<RwLock<HashMap<ids::Id, crate::block::Block>>>,
+    pub verified_blocks: Arc<RwLock<HashMap<ids::Id, block::Block>>>,
 }
 
 impl ChainVm {
     /// Returns initialized ChainVm Boxed as rpcchainvm::vm::Vm trait.
     pub fn new() -> Box<dyn rpcchainvm::vm::Vm + Send + Sync> {
         let inner = Arc::new(RwLock::new(ChainVmInterior::default()));
         let db = rpcchainvm::database::memdb::Database::new();
-        let mempool = Arc::new(RwLock::new(VecDeque::new()));
+        let mempool = mempool::Mempool::new(MEMPOOL_SIZE);
         let verified_blocks = Arc::new(RwLock::new(HashMap::new()));
 
         Box::new(ChainVm {
             db,
             inner,
-            mempool,
+            mempool: Arc::new(RwLock::new(mempool)),
             verified_blocks,
         })
     }
 
     pub fn new_with_state(db: &Box<dyn rpcchainvm::database::Database + Sync + Send>) -> Self {
-        let mempool = Arc::new(RwLock::new(VecDeque::new()));
+        let mempool = mempool::Mempool::new(MEMPOOL_SIZE);
         let verified_blocks = &Arc::new(RwLock::new(HashMap::new()));
         let inner = ChainVmInterior {
             ctx: None,
@@ -97,7 +101,7 @@ impl ChainVm {
         Self {
             db: db.clone(),
             inner: Arc::new(RwLock::new(inner)),
-            mempool,
+            mempool: Arc::new(RwLock::new(mempool)),
             verified_blocks: Arc::clone(verified_blocks),
         }
     }
@@ -112,7 +116,7 @@ impl crate::chain::vm::Vm for ChainVm {
         return vm.bootstrapped;
     }
 
-    async fn submit(&self, mut txs: Vec<crate::chain::tx::tx::Transaction>) -> Result<()> {
+    async fn submit(&self, mut txs: Vec<chain::tx::tx::Transaction>) -> Result<()> {
         let now = Utc::now().timestamp() as u64;
 
         // TODO append errors instead of fail
@@ -128,8 +132,11 @@ impl crate::chain::vm::Vm for ChainVm {
             tx.execute(self.db.clone(), dummy_block)
                 .await
                 .map_err(|e| Error::new(ErrorKind::Other, e.to_string()))?;
+
             let mut mempool = self.mempool.write().await;
-            mempool.push_front(tx.to_owned());
+            let _ = mempool
+                .add(tx.to_owned())
+                .map_err(|e| Error::new(ErrorKind::Other, e.to_string()))?;
         }
         Ok(())
     }
@@ -429,7 +436,7 @@ impl rpcchainvm::snowman::block::ChainVm for ChainVm {
     async fn build_block(
         &self,
     ) -> Result<Box<dyn rpcchainvm::concensus::snowman::Block + Send + Sync>> {
-        let mut mempool = self.mempool.write().await;
+        let mempool = self.mempool.read().await;
         if mempool.len() == 0 {
             return Err(Error::new(ErrorKind::Other, "no pending blocks"));
         }

mini-kvvm/tests/api/mod.rs

@@ -15,13 +15,23 @@ async fn service_test() {
     let db = MemDb::new();
     let vm = vm::ChainVm::new_with_state(&db);
 
-    {
-        // initialize genesis block
-        let mut inner = vm.inner.write().await;
-        let resp = create_genesis_block(&inner.state.clone(), vec![]).await;
-        assert!(resp.is_ok());
-        inner.preferred = resp.unwrap();
-    }
+    // get a broadcast tx pending receiver for new blocks;
+    let mempool = vm.mempool.read().await;
+    let mut pending_rx = mempool.subscribe_pending();
+    drop(mempool);
+    // unblock channel
+    tokio::spawn(async move {
+        loop {
+            pending_rx.recv().await.unwrap();
+        }
+    });
+
+    // initialize genesis block
+    let mut inner = vm.inner.write().await;
+    let resp = create_genesis_block(&inner.state.clone(), vec![]).await;
+    assert!(resp.is_ok());
+    inner.preferred = resp.unwrap();
+    drop(inner);
 
     let service = api::service::Service::new(vm);
     let mut io = jsonrpc_core::IoHandler::new();