Adding inline transactions in payload to reduce latency (aptos-labs#12303)

* Create a channel between batch coordinator and proof manager

* Add new variant to Payload

* Adding inline transactions in proof manager

* Fixing a unit test

* Update get_transactions in payload_manager

* Update max_txns_to_execute in proof_manager

* Minor updates to max_txns_to_execute

* 100 node forge test

* 100 node forge

* Add a TODO

* 100 node forge

* Sending only small batches to proof manager

* Sending only small batches to proof manager

* Remove expired batches in proof manager

* Send only small batches to proof manager

* Minor change

* Send batches to proof manager only if there are less than 20 batches

* Exclude batches

* Storing only the most recent 20 batches in proof manager

* Sending all batches to proof manager and inserting inline blocks only if proof of queue is empty

* Maintaining ordering in inline blocks

* Verifying batch hash before including in inline block

* Adding some counters for testing

* Minor update

* Minor cleanup

* Addressing some PR comments

* Create a BatchQueue similar to ProofQueue

* Minor change

* Not removing excluded batches from batch queue

* Change the structure of Payload

* Make max_inline_txns configurable

* Storing batches only in batchstore

* Update Dag driver thresholds

* Addressed PR comments

* Minor edit to consensus.yaml

* Addressing PR comments

* Add allow_batches_without_pos_in_proposal flag to Payload::empty

* Fixing unit tests

* Fixing the smoke tests

* Removing some comments

* Addressing PR comments

* Addressing PR comments

Author: vusirikala

config/src/config/consensus_config.rs

@@ -21,6 +21,8 @@ pub struct ConsensusConfig {
     pub max_network_channel_size: usize,
     pub max_sending_block_txns: u64,
     pub max_sending_block_bytes: u64,
+    pub max_sending_inline_txns: u64,
+    pub max_sending_inline_bytes: u64,
     pub max_receiving_block_txns: u64,
     pub max_receiving_block_bytes: u64,
     pub max_pruned_blocks_in_mem: usize,
@@ -144,6 +146,8 @@ impl Default for ConsensusConfig {
             max_sending_block_txns: MAX_SENDING_BLOCK_TXNS,
             max_sending_block_bytes: 3 * 1024 * 1024, // 3MB
             max_receiving_block_txns: 10000.max(2 * MAX_SENDING_BLOCK_TXNS),
+            max_sending_inline_txns: 100,
+            max_sending_inline_bytes: 200 * 1024,       // 200 KB
             max_receiving_block_bytes: 6 * 1024 * 1024, // 6MB
             max_pruned_blocks_in_mem: 100,
             mempool_executed_txn_timeout_ms: 1000,

config/src/config/quorum_store_config.rs

@@ -71,6 +71,7 @@ pub struct QuorumStoreConfig {
     pub back_pressure: QuorumStoreBackPressureConfig,
     pub num_workers_for_remote_batches: usize,
     pub batch_buckets: Vec<u64>,
+    pub allow_batches_without_pos_in_proposal: bool,
 }
 
 impl Default for QuorumStoreConfig {
@@ -107,6 +108,7 @@ impl Default for QuorumStoreConfig {
             // number of batch coordinators to handle QS batch messages, should be >= 1
             num_workers_for_remote_batches: 10,
             batch_buckets: DEFAULT_BUCKETS.to_vec(),
+            allow_batches_without_pos_in_proposal: false,
         }
     }
 }

consensus/consensus-types/src/block.rs

@@ -113,6 +113,9 @@ impl Block {
                 Payload::InQuorumStore(pos) => pos.proofs.len(),
                 Payload::DirectMempool(_txns) => 0,
                 Payload::InQuorumStoreWithLimit(pos) => pos.proof_with_data.proofs.len(),
+                Payload::QuorumStoreInlineHybrid(inline_batches, proof_with_data, _) => {
+                    inline_batches.len() + proof_with_data.proofs.len()
+                },
             },
         }
     }

consensus/consensus-types/src/block_data.rs

@@ -235,7 +235,7 @@ impl BlockData {
     pub fn dummy_with_validator_txns(txns: Vec<ValidatorTransaction>) -> Self {
         Self::new_proposal_ext(
             txns,
-            Payload::empty(false),
+            Payload::empty(false, true),
             Author::ONE,
             vec![],
             1,
@@ -385,7 +385,7 @@ fn test_reconfiguration_suffix() {
         ),
     );
     let reconfig_suffix_block = BlockData::new_proposal(
-        Payload::empty(false),
+        Payload::empty(false, true),
         AccountAddress::random(),
         Vec::new(),
         2,

consensus/consensus-types/src/block_test.rs

@@ -72,7 +72,7 @@ fn test_nil_block() {
         nil_block_qc.certified_block().id()
     );
     let nil_block_child = Block::new_proposal(
-        Payload::empty(false),
+        Payload::empty(false, true),
         2,
         aptos_infallible::duration_since_epoch().as_micros() as u64,
         nil_block_qc,
@@ -91,7 +91,7 @@ fn test_block_relation() {
     // Test genesis and the next block
     let genesis_block = Block::make_genesis_block();
     let quorum_cert = certificate_for_genesis();
-    let payload = Payload::empty(false);
+    let payload = Payload::empty(false, true);
     let next_block = Block::new_proposal(
         payload.clone(),
         1,
@@ -118,7 +118,7 @@ fn test_same_qc_different_authors() {
     let signer = signers.first().unwrap();
     let genesis_qc = certificate_for_genesis();
     let round = 1;
-    let payload = Payload::empty(false);
+    let payload = Payload::empty(false, true);
     let current_timestamp = aptos_infallible::duration_since_epoch().as_micros() as u64;
     let block_round_1 = Block::new_proposal(
         payload.clone(),
@@ -180,7 +180,7 @@ fn test_block_metadata_bitvec() {
         &ledger_info,
         Block::make_genesis_block_from_ledger_info(&ledger_info).id(),
     );
-    let payload = Payload::empty(false);
+    let payload = Payload::empty(false, true);
     let start_round = 1;
     let start_timestamp = aptos_infallible::duration_since_epoch().as_micros() as u64;
 
@@ -254,7 +254,7 @@ fn test_failed_authors_well_formed() {
     let other = Author::random();
     // Test genesis and the next block
     let quorum_cert = certificate_for_genesis();
-    let payload = Payload::empty(false);
+    let payload = Payload::empty(false, true);
 
     let create_block = |round: Round, failed_authors: Vec<(Round, Author)>| {
         Block::new_proposal(

consensus/consensus-types/src/block_test_utils.rs

@@ -43,7 +43,7 @@ prop_compose! {
         parent_qc in Just(parent_qc)
     ) -> Block {
         Block::new_proposal(
-            Payload::empty(false),
+            Payload::empty(false, true),
             round,
             aptos_infallible::duration_since_epoch().as_micros() as u64,
             parent_qc,