vm: add network

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

Author: hexfusion

mini-kvvm/Cargo.toml

@@ -15,7 +15,7 @@ path = "src/bin/mini-kvvm/main.rs"
 
 [dependencies]
 avalanche-proto = { version = "0.16.0" }
-avalanche-types = { version = "0.0.34" }
+avalanche-types = { version = "0.0.38" }
 byteorder = "1.4.3"
 chan = "0.1.23"
 chrono = "0.4.19"

mini-kvvm/src/block/builder.rs

@@ -7,9 +7,11 @@ use std::{
     time::{Duration, Instant},
 };
 
-use chan::chan_select;
 use avalanche_types::rpcchainvm;
-use tokio::sync::RwLock;
+use chan::chan_select;
+use tokio::sync::{RwLock};
+use crossbeam_channel::TryRecvError;
+
 
 use crate::vm;
 
@@ -276,17 +278,19 @@ impl Timed {
         let regossip = chan::tick(Duration::from_millis(100));
         let stop_ch = self.stop.clone();
 
-        loop {
-            
+        while stop_ch.try_recv() == Err(TryRecvError::Empty) {
             chan_select! {
                 gossip.recv() => {
-                    let mut mempool = self.vm.mempool.write().await;
-                    let new_txs = mempool.new_txs();
-                    drop(mempool)
+                    let mempool = &mut self.vm.mempool.write().await;
+                    let new_txs = mempool.new_txs().unwrap();
+                    drop(mempool);
 
+                    let mut network = self.vm.network.as_ref().unwrap().write().await;
+                    let _ = network.gossip_new_txs(new_txs).await;
                 },
                 regossip.recv() => {
-                    
+                    let mut network = self.vm.network.as_ref().unwrap().write().await;
+                    let _ = network.regossip_txs().await;
                 },
             }
         }

mini-kvvm/src/lib.rs

@@ -3,5 +3,6 @@ pub mod block;
 pub mod chain;
 pub mod genesis;
 pub mod mempool;
+pub mod network;
 pub mod utils;
 pub mod vm;

mini-kvvm/src/network.rs

@@ -0,0 +1,126 @@
+use std::{
+    io::{Error, ErrorKind, Result},
+    num::NonZeroUsize,
+};
+
+use avalanche_types::ids::{self, Id};
+use lru::LruCache;
+
+use crate::{
+    chain::{self, vm::Vm},
+    vm,
+};
+
+const GOSSIPED_TXS_LRU_SIZE: usize = 512;
+
+pub struct Push {
+    vm: vm::ChainVm,
+    gossiped_tx: LruCache<Id, ()>,
+}
+
+impl Push {
+    pub fn new(vm: vm::ChainVm) -> Self {
+        let cache: LruCache<Id, ()> =
+            LruCache::new(NonZeroUsize::new(GOSSIPED_TXS_LRU_SIZE).unwrap());
+        Self {
+            gossiped_tx: cache,
+            vm,
+        }
+    }
+
+    pub async fn send_txs(&self, txs: Vec<chain::tx::tx::Transaction>) -> Result<()> {
+        if txs.is_empty() {
+            return Ok(());
+        }
+
+        let b = serde_json::to_vec(&txs).map_err(|e| {
+            Error::new(
+                ErrorKind::Other,
+                format!("failed to marshal txs: {}", e.to_string()),
+            )
+        })?;
+
+        log::debug!("sending app gossip txs: {} size: {}", txs.len(), b.len());
+
+        let appsender = self
+            .vm
+            .app_sender
+            .clone()
+            .expect("appsender should exist after initialize");
+        appsender.send_app_gossip(b).await.map_err(|e| {
+            Error::new(
+                ErrorKind::Other,
+                format!("gossip txs failed: {}", e.to_string()),
+            )
+        })?;
+        Ok(())
+    }
+
+    pub async fn gossip_new_txs(&mut self, new_txs: Vec<chain::tx::tx::Transaction>) -> Result<()> {
+        let mut txs: Vec<chain::tx::tx::Transaction> = Vec::with_capacity(new_txs.len());
+
+        for tx in new_txs.iter() {
+            if self.gossiped_tx.contains(&tx.id) {
+                log::debug!("already gossiped skipping id: {}", tx.id);
+                continue;
+            }
+
+            self.gossiped_tx.put(tx.id, ());
+
+            txs.push(tx.to_owned());
+        }
+
+        Ok(())
+    }
+
+    /// Triggers "AppGossip" on the pending transactions in the mempool.
+    /// "force" is true to re-gossip whether recently gossiped or not.
+    pub async fn regossip_txs(&mut self) -> Result<()> {
+        let mut txs: Vec<chain::tx::tx::Transaction> = Vec::new();
+        let mempool = self.vm.mempool.read().await;
+
+        // Gossip at most the target units of a block at once
+        while mempool.len() > 0 {
+            match mempool.pop_back() {
+                Some(tx) => {
+                    // Note: when regossiping, we force resend even though we may have done it
+                    // recently.
+                    self.gossiped_tx.put(tx.id, ());
+                    txs.push(tx);
+                }
+                None => return Ok(()),
+            }
+        }
+
+        return self.send_txs(txs).await;
+    }
+
+    pub async fn app_gossip(&mut self, node_id: ids::node::Id, message: &[u8]) -> Result<()> {
+        log::debug!(
+            "appgossip message handler, sender: {} bytes: {:?}",
+            node_id,
+            message
+        );
+
+        let txs: Vec<chain::tx::tx::Transaction> = serde_json::from_slice(&message).unwrap();
+
+        // submit incoming gossip
+        log::debug!(
+            "appgossip transactions are being submitted txs: {}",
+            txs.len()
+        );
+
+        self.vm.submit(txs).await.map_err(|e| {
+            Error::new(
+                ErrorKind::Other,
+                format!(
+                    "appgossip failed to submit txs peer_id: {}: {}",
+                    node_id,
+                    e.to_string()
+                ),
+            )
+        })?;
+
+        Ok(())
+    }
+}

mini-kvvm/src/vm.rs

@@ -23,7 +23,7 @@ use crate::{
     block::{self, state::State},
     chain::{self, tx::Transaction, vm::Vm},
     genesis::Genesis,
-    mempool,
+    mempool, network,
 };
 
 const PUBLIC_API_ENDPOINT: &str = "/public";
@@ -63,7 +63,9 @@ impl Default for ChainVmInterior {
 #[derive(Clone)]
 pub struct ChainVm {
     pub db: Box<dyn rpcchainvm::database::Database + Sync + Send>,
+    pub app_sender: Option<Box<dyn rpcchainvm::common::appsender::AppSender + Send + Sync>>,
     pub mempool: Arc<RwLock<mempool::Mempool>>,
+    pub network: Option<Arc<RwLock<network::Push>>>,
     pub inner: Arc<RwLock<ChainVmInterior>>,
     pub verified_blocks: Arc<RwLock<HashMap<ids::Id, block::Block>>>,
 }
@@ -81,6 +83,8 @@ impl ChainVm {
             inner,
             mempool: Arc::new(RwLock::new(mempool)),
             verified_blocks,
+            app_sender: None,
+            network: None,
         })
     }
 
@@ -103,6 +107,8 @@ impl ChainVm {
             inner: Arc::new(RwLock::new(inner)),
             mempool: Arc::new(RwLock::new(mempool)),
             verified_blocks: Arc::clone(verified_blocks),
+            app_sender: None,
+            network: None,
         }
     }
 }
@@ -235,7 +241,7 @@ impl rpcchainvm::common::vm::Vm for ChainVm {
         _config_bytes: &[u8],
         _to_engine: Sender<rpcchainvm::common::message::Message>,
         _fxs: &[rpcchainvm::common::vm::Fx],
-        _app_sender: Box<dyn rpcchainvm::common::appsender::AppSender + Send + Sync>,
+        app_sender: Box<dyn rpcchainvm::common::appsender::AppSender + Send + Sync>,
     ) -> Result<()> {
         let mut vm = self.inner.write().await;
         vm.ctx = ctx;
@@ -247,6 +253,9 @@ impl rpcchainvm::common::vm::Vm for ChainVm {
         let current = db_manager.current().await?;
         self.db = current.db.clone();
 
+        self.app_sender = Some(app_sender);
+        self.network = Some(Arc::new(RwLock::new(network::Push::new(self.clone()))));
+
         let verified_blocks = self.verified_blocks.clone();
 
         vm.state = State::new(self.db.clone(), verified_blocks);