Send transaction in batches

zilliqa/benches/it.rs

@@ -256,9 +256,9 @@ pub fn produce_full(crit: &mut Criterion) {
         })
         .collect();
     for txn in txns {
-        let result = big.new_transaction(txn.clone()).unwrap();
+        let result = big.new_transaction(txn.clone(), false).unwrap();
         assert!(result.was_added());
-        let result = tiny.new_transaction(txn).unwrap();
+        let result = tiny.new_transaction(txn, false).unwrap();
         assert!(result.was_added());
     }
 

zilliqa/src/api/eth.rs

@@ -743,6 +743,8 @@ fn send_raw_transaction(params: Params, node: &Arc<Mutex<Node>>) -> Result<Strin
     let transaction = hex::decode(transaction)?;
     let transaction = parse_transaction(&transaction)?;
 
+    let transaction = transaction.verify()?;
+
     let (hash, result) = node.lock().unwrap().create_transaction(transaction)?;
     match result {
         TxAddResult::AddedToMempool

zilliqa/src/api/zil.rs

@@ -313,7 +313,14 @@ fn create_transaction(
         sig,
     };
 
-    let (transaction_hash, result) = node.create_transaction(signed_transaction.clone())?;
+    let Ok(transaction) = signed_transaction.verify() else {
+        Err(ErrorObject::owned::<String>(
+            RPCErrorCode::RpcVerifyRejected as i32,
+            "signature",
+            None,
+        ))?
+    };
+    let (transaction_hash, result) = node.create_transaction(transaction)?;
     let info = match result {
         TxAddResult::AddedToMempool => Ok("Txn processed".to_string()),
         TxAddResult::Duplicate(_) => Ok("Txn already present".to_string()),

zilliqa/src/consensus.rs

@@ -164,7 +164,7 @@ pub struct Consensus {
     receipts_cache: HashMap<Hash, (TransactionReceipt, Vec<Address>)>,
     receipts_cache_hash: Hash,
     /// Actions that act on newly created blocks
-    transaction_pool: TransactionPool,
+    pub transaction_pool: TransactionPool,
     /// Pending proposal. Gets created as soon as we become aware that we are leader for this view.
     early_proposal: Option<EarlyProposal>,
     /// Flag indicating that block creation should be postponed at least until empty_block_timeout is reached
@@ -634,9 +634,10 @@ impl Consensus {
         let head_block = self.head_block();
         let mut view = self.get_view()?;
 
-        trace!(
+        info!(
             block_view = block.view(),
             block_number = block.number(),
+            txns = transactions.len(),
             "handling block proposal {}",
             block.hash()
         );
@@ -1646,7 +1647,8 @@ impl Consensus {
         // need those transactions again
         for tx in opaque_transactions {
             let account_nonce = self.state.get_account(tx.signer)?.nonce;
-            self.transaction_pool.insert_transaction(tx, account_nonce);
+            self.transaction_pool
+                .insert_transaction(tx, account_nonce, false);
         }
 
         // finalise the proposal
@@ -1665,11 +1667,12 @@ impl Consensus {
     }
 
     /// Insert transaction and add to early_proposal if possible.
-    pub fn handle_new_transaction(&mut self, txn: SignedTransaction) -> Result<TxAddResult> {
-        let Ok(verified) = txn.verify() else {
-            return Ok(TxAddResult::CannotVerifySignature);
-        };
-        let inserted = self.new_transaction(verified)?;
+    pub fn handle_new_transaction(
+        &mut self,
+        verified: VerifiedTransaction,
+        from_broadcast: bool,
+    ) -> Result<TxAddResult> {
+        let inserted = self.new_transaction(verified, from_broadcast)?;
         if inserted.was_added()
             && self.create_next_block_on_timeout
             && self.early_proposal.is_some()
@@ -1884,7 +1887,11 @@ impl Consensus {
 
     /// Returns (flag, outcome).
     /// flag is true if the transaction was newly added to the pool - ie. if it validated correctly and has not been seen before.
-    pub fn new_transaction(&mut self, txn: VerifiedTransaction) -> Result<TxAddResult> {
+    pub fn new_transaction(
+        &mut self,
+        txn: VerifiedTransaction,
+        from_broadcast: bool,
+    ) -> Result<TxAddResult> {
         if self.db.contains_transaction(&txn.hash)? {
             debug!("Transaction {:?} already in mempool", txn.hash);
             return Ok(TxAddResult::Duplicate(txn.hash));
@@ -1919,9 +1926,9 @@ impl Consensus {
 
         let txn_hash = txn.hash;
 
-        let insert_result = self
-            .transaction_pool
-            .insert_transaction(txn, early_account.nonce);
+        let insert_result =
+            self.transaction_pool
+                .insert_transaction(txn, early_account.nonce, from_broadcast);
         if insert_result.was_added() {
             let _ = self.new_transaction_hashes.send(txn_hash);
 
@@ -2854,7 +2861,8 @@ impl Consensus {
 
             for txn in existing_txns {
                 let account_nonce = self.state.get_account(txn.signer)?.nonce;
-                self.transaction_pool.insert_transaction(txn, account_nonce);
+                self.transaction_pool
+                    .insert_transaction(txn, account_nonce, false);
             }
 
             // block transactions need to be removed from self.transactions and re-injected
@@ -2864,7 +2872,7 @@ impl Consensus {
                 // Insert this unwound transaction back into the transaction pool.
                 let account_nonce = self.state.get_account(orig_tx.signer)?.nonce;
                 self.transaction_pool
-                    .insert_transaction(orig_tx, account_nonce);
+                    .insert_transaction(orig_tx, account_nonce, false);
             }
             // then purge them all from the db, including receipts and indexes
             self.db
@@ -3005,7 +3013,7 @@ impl Consensus {
 
         let mut touched_addresses = vec![];
         for (tx_index, txn) in verified_txns.iter().enumerate() {
-            self.new_transaction(txn.clone())?;
+            self.new_transaction(txn.clone(), false)?;
             let tx_hash = txn.hash;
             let mut inspector = TouchedAddressInspector::default();
             let result = self

zilliqa/src/db.rs

@@ -232,13 +232,17 @@ impl Db {
         connection.pragma_update(None, "cache_size", (1 << 28) / page_size)?;
         let cache_size: i32 = connection.pragma_query_value(None, "cache_size", |r| r.get(0))?;
 
+        let mmap_size = 268435456;
+        connection.pragma_update(None, "mmap_size", mmap_size)?;
+
         tracing::info!(
             ?journal_mode,
             ?journal_size_limit,
             ?synchronous,
             ?temp_store,
             ?page_size,
             ?cache_size,
+            ?mmap_size,
             "PRAGMA"
         );
 

zilliqa/src/exec.rs

@@ -670,7 +670,7 @@ impl State {
     ) -> Result<TransactionApplyResult> {
         let hash = txn.hash;
         let from_addr = txn.signer;
-        info!(?hash, ?txn, "executing txn");
+        debug!(?hash, ?txn, "executing txn");
 
         let blessed = BLESSED_TRANSACTIONS.contains(&hash);
 

zilliqa/src/message.rs

@@ -247,6 +247,7 @@ pub enum ExternalMessage {
     BlockResponse(BlockResponse),
     ProcessProposal(ProcessProposal),
     NewTransaction(SignedTransaction),
+    BatchedTransactions(Vec<SignedTransaction>),
     /// An acknowledgement of the receipt of a message. Note this is only used as a response when the caller doesn't
     /// require any data in the response.
     Acknowledgement,
@@ -314,6 +315,9 @@ impl Display for ExternalMessage {
                     write!(f, "NewTransaction(Unable to verify txn due to: {:?})", err)
                 }
             },
+            ExternalMessage::BatchedTransactions(txns) => {
+                write!(f, "BatchedTransactions(txns_count: {:?})", txns.len())
+            }
             ExternalMessage::Acknowledgement => write!(f, "RequestResponse"),
         }
     }

zilliqa/src/node.rs

@@ -208,20 +208,20 @@ impl Node {
 
     pub fn handle_broadcast(&mut self, from: PeerId, message: ExternalMessage) -> Result<()> {
         debug!(%from, to = %self.peer_id, %message, "handling broadcast");
-        // We only expect `NewTransaction`s to be broadcast.
         // `Proposals` are re-routed to `handle_request()`.
-        match message {
-            ExternalMessage::NewTransaction(t) => {
-                // Don't process again txn sent by this node (it's already in the mempool)
-                if self.peer_id != from {
-                    self.consensus.handle_new_transaction(t)?;
-                }
-            }
-            _ => {
-                warn!("unexpected message type");
-            }
-        }
+        warn!("unexpected message type");
+
+        Ok(())
+    }
 
+    pub fn handle_broadcasted_transactions(
+        &mut self,
+        transactions: Vec<VerifiedTransaction>,
+    ) -> Result<()> {
+        for txn in transactions {
+            let from_broadcast = true;
+            self.consensus.handle_new_transaction(txn, from_broadcast)?;
+        }
         Ok(())
     }
 
@@ -388,7 +388,7 @@ impl Node {
         };
         let verified_tx = tx.verify()?;
         trace!("Injecting intershard transaction {}", verified_tx.hash);
-        self.consensus.new_transaction(verified_tx)?;
+        self.consensus.new_transaction(verified_tx, false)?;
         Ok(())
     }
 
@@ -407,21 +407,29 @@ impl Node {
         Ok(false)
     }
 
-    pub fn create_transaction(&mut self, txn: SignedTransaction) -> Result<(Hash, TxAddResult)> {
-        let hash = txn.calculate_hash();
+    pub fn create_transaction(&mut self, txn: VerifiedTransaction) -> Result<(Hash, TxAddResult)> {
+        let hash = txn.hash;
 
-        info!(?hash, "seen new txn {:?}", txn);
+        debug!(?hash, "seen new txn {:?}", txn);
 
-        let result = self.consensus.handle_new_transaction(txn.clone())?;
-        if result.was_added() {
-            // TODO: Avoid redundant self-broadcast
-            self.message_sender
-                .broadcast_external_message(ExternalMessage::NewTransaction(txn))?;
+        let from_broadcast = false;
+        let result = self.consensus.handle_new_transaction(txn, from_broadcast)?;
+        if !result.was_added() {
+            debug!(?result, "Transaction cannot be added to mempool");
         }
 
         Ok((hash, result))
     }
 
+    pub fn process_transactions_to_broadcast(&mut self) -> Result<()> {
+        let txns_to_broadcast = self.consensus.transaction_pool.pull_txns_to_broadcast()?;
+        if txns_to_broadcast.is_empty() {
+            return Ok(());
+        }
+        self.message_sender
+            .broadcast_external_message(ExternalMessage::BatchedTransactions(txns_to_broadcast))
+    }
+
     pub fn number(&self) -> u64 {
         self.consensus.head_block().header.number
     }

zilliqa/src/node_launcher.rs

@@ -180,8 +180,11 @@ impl NodeLauncher {
             return Err(anyhow!("Node already running!"));
         }
 
-        let sleep = time::sleep(Duration::from_millis(5));
-        tokio::pin!(sleep);
+        let consensus_sleep = time::sleep(Duration::from_millis(5));
+        tokio::pin!(consensus_sleep);
+
+        let mempool_sleep = time::sleep(Duration::from_millis(5));
+        tokio::pin!(mempool_sleep);
 
         self.node_launched = true;
 
@@ -202,7 +205,20 @@ impl NodeLauncher {
                     ];
 
                     let start = SystemTime::now();
-                    if let Err(e) = self.node.lock().unwrap().handle_broadcast(source, message) {
+                    if let ExternalMessage::BatchedTransactions(transactions) = message {
+                        let my_peer_id = self.node.lock().unwrap().consensus.peer_id();
+
+                        if source == my_peer_id {
+                            continue;
+                        }
+                        let mut verified = Vec::new();
+                        for txn in transactions {
+                            let txn = txn.verify()?;
+                            verified.push(txn);
+                        }
+                        self.node.lock().unwrap().handle_broadcasted_transactions(verified)?;
+                    }
+                    else if let Err(e) = self.node.lock().unwrap().handle_broadcast(source, message) {
                         attributes.push(KeyValue::new(ERROR_TYPE, "process-error"));
                         error!("Failed to process broadcast message: {e}");
                     }
@@ -269,7 +285,7 @@ impl NodeLauncher {
                     let (_source, _message) = message.expect("message stream should be infinite");
                     todo!("Local messages will need to be handled once cross-shard messaging is implemented");
                 }
-                () = &mut sleep => {
+                () = &mut consensus_sleep => {
                     let attributes = vec![
                         KeyValue::new(MESSAGING_OPERATION_NAME, "handle"),
                         KeyValue::new(MESSAGING_SYSTEM, "tokio_channel"),
@@ -281,7 +297,7 @@ impl NodeLauncher {
                     self.node.lock().unwrap().consensus.tick().unwrap();
                     // No messages for a while, so check if consensus wants to timeout
                     self.node.lock().unwrap().handle_timeout().unwrap();
-                    sleep.as_mut().reset(Instant::now() + Duration::from_millis(500));
+                    consensus_sleep.as_mut().reset(Instant::now() + Duration::from_millis(500));
                     messaging_process_duration.record(
                         start.elapsed().map_or(0.0, |d| d.as_secs_f64()),
                         &attributes,
@@ -290,7 +306,12 @@ impl NodeLauncher {
                 r = self.reset_timeout_receiver.next() => {
                     let sleep_time = r.expect("reset timeout stream should be infinite");
                     trace!(?sleep_time, "timeout reset");
-                    sleep.as_mut().reset(Instant::now() + sleep_time);
+                    consensus_sleep.as_mut().reset(Instant::now() + sleep_time);
+                },
+
+                () = &mut mempool_sleep => {
+                    self.node.lock().unwrap().process_transactions_to_broadcast()?;
+                    mempool_sleep.as_mut().reset(Instant::now() + Duration::from_millis(100));
                 },
             }
         }