feat(code): Add max_block_size config option

This value can be overridden with the environment variable `MALACHITE__CONSENSUS__MAX_BLOCK_SIZE`
informalsystems · Jun 4, 2024 · 4b4ebb9 · 4b4ebb9
1 parent 2ae7f78
commit 4b4ebb9
Show file tree

Hide file tree

Showing 12 changed files with 75 additions and 12 deletions.
diff --git a/code/Cargo.toml b/code/Cargo.toml
@@ -47,6 +47,7 @@ malachite-vote    = { version = "0.1.0", path = "vote" }
 
 async-trait        = "0.1.77"
 base64             = "0.22.0"
+bytesize           = "1.3"
 clap               = "4.5.4"
 color-eyre         = "0.6"
 derive-where       = "1.2.7"

diff --git a/code/actors/Cargo.toml b/code/actors/Cargo.toml
@@ -22,6 +22,7 @@ malachite-test.workspace    = true
 malachite-vote.workspace    = true
 
 async-trait        = { workspace = true }
+bytesize           = { workspace = true, features = ["serde"] }
 derive-where       = { workspace = true }
 libp2p             = { workspace = true }
 ractor             = { workspace = true, features = ["async-trait"] }

diff --git a/code/actors/src/util/make_actor.rs b/code/actors/src/util/make_actor.rs
@@ -17,6 +17,7 @@ use crate::host::Host;
 use crate::mempool::Mempool;
 use crate::node::{Msg as NodeMsg, Msg, Node};
 use crate::timers::Config as TimersConfig;
+use crate::util::value_builder::test::TestParams as TestValueBuilderParams;
 use crate::util::PartStore;
 use crate::util::TestValueBuilder;
 
@@ -48,7 +49,13 @@ pub async fn make_node_actor(
     let ctx = TestContext::new(validator_pk.clone());
 
     // Spawn the host actor
-    let value_builder = Box::new(TestValueBuilder::<TestContext>::new(mempool.clone()));
+    let value_builder = Box::new(TestValueBuilder::<TestContext>::new(
+        mempool.clone(),
+        TestValueBuilderParams {
+            max_block_size: cfg.consensus.max_block_size,
+        },
+    ));
+
     let host = Host::spawn(
         value_builder,
         PartStore::new(),

diff --git a/code/actors/src/util/value_builder.rs b/code/actors/src/util/value_builder.rs
@@ -46,29 +46,37 @@ pub mod test {
 
     use std::marker::PhantomData;
 
+    use bytesize::ByteSize;
+    use ractor::ActorRef;
+
+    use super::*;
     use malachite_common::Context;
     use malachite_driver::Validity;
     use malachite_test::{
         Address, BlockMetadata, BlockPart, Content, Height, TestContext, TransactionBatch, Value,
     };
-    use ractor::ActorRef;
 
-    use super::*;
+    #[derive(Copy, Clone, Debug)]
+    pub struct TestParams {
+        pub max_block_size: ByteSize,
+    }
 
     #[derive(Clone)]
     pub struct TestValueBuilder<Ctx: Context> {
         _phantom: PhantomData<Ctx>,
         tx_streamer: ActorRef<crate::mempool::Msg>,
+        params: TestParams,
     }
 
     impl<Ctx> TestValueBuilder<Ctx>
     where
         Ctx: Context,
     {
-        pub fn new(tx_streamer: ActorRef<crate::mempool::Msg>) -> Self {
+        pub fn new(tx_streamer: ActorRef<crate::mempool::Msg>, params: TestParams) -> Self {
             Self {
                 _phantom: Default::default(),
                 tx_streamer,
+                params,
             }
         }
     }
@@ -90,6 +98,7 @@ pub mod test {
 
             let mut tx_batch = vec![];
             let mut sequence = 1;
+            let mut block_size = 0;
             let mut result = None;
 
             loop {
@@ -100,7 +109,7 @@ pub mod test {
                     sequence
                 );
 
-                let mut txes = self
+                let txes = self
                     .tx_streamer
                     .call(
                         |reply| crate::mempool::Msg::TxStream {
@@ -135,12 +144,20 @@ pub mod test {
 
                 // Simulate execution
                 tokio::time::sleep(Duration::from_micros(EXEC_TIME_MICROSEC_PER_PART)).await;
-                tx_batch.append(&mut txes);
+
+                'inner: for tx in txes {
+                    if block_size + tx.size_bytes() > self.params.max_block_size.as_u64() {
+                        break 'inner;
+                    }
+
+                    block_size += tx.size_bytes();
+                    tx_batch.push(tx);
+                }
 
                 sequence += 1;
 
                 if Instant::now() > expiration_time {
-                    error!( "Value Builder started at {now:?} but failed to complete by expiration time {expiration_time:?}");
+                    error!("Value Builder started at {now:?} but failed to complete by expiration time {expiration_time:?}");
                     result = None;
                     break;
                 }
@@ -169,13 +186,14 @@ pub mod test {
                     part_store.store(block_part.clone());
 
                     consensus
-                        .cast(ConsensusMsg::BuilderBlockPart(block_part.clone()))
+                        .cast(ConsensusMsg::BuilderBlockPart(block_part))
                         .unwrap();
 
                     info!(
-                        "Value Builder created a block with {} tx-es, block hash (consensus value) {:?} ",
+                        "Value Builder created a block with {} tx-es ({}), block hash: {:?} ",
                         tx_batch.len(),
-                        result
+                        ByteSize::b(block_size),
+                        value.id()
                     );
 
                     break;
@@ -196,7 +214,7 @@ pub mod test {
 
             part_store.store(block_part.clone());
             let num_parts = part_store.all_parts(height, round).len();
-            trace!("({num_parts}):Received block part (h: {height}, r: {round}, seq: {sequence}");
+            trace!("({num_parts}): Received block part (h: {height}, r: {round}, seq: {sequence}");
 
             // Simulate Tx execution and proof verification (assumes success)
             // TODO - add config knob for invalid blocks

diff --git a/code/actors/tests/util.rs b/code/actors/tests/util.rs
@@ -3,6 +3,7 @@
 use std::sync::atomic::{AtomicUsize, Ordering};
 use std::sync::Arc;
 
+use bytesize::ByteSize;
 use rand::Rng;
 use tokio::sync::mpsc;
 use tokio::time::{sleep, Duration};
@@ -110,6 +111,7 @@ pub async fn run_test<const N: usize>(test: Test<N>) {
         let node_config = malachite_node::config::Config {
             moniker: format!("node-{i}"),
             consensus: ConsensusConfig {
+                max_block_size: ByteSize::mib(1),
                 timeouts: TimeoutConfig::default(),
                 p2p: P2pConfig {
                     listen_addr: format!(

diff --git a/code/cli/Cargo.toml b/code/cli/Cargo.toml
@@ -14,6 +14,7 @@ malachite-actors.workspace = true
 malachite-node.workspace   = true
 malachite-test.workspace   = true
 
+bytesize           = { workspace = true }
 clap               = { workspace = true, features = ["derive", "env"] }
 color-eyre         = { workspace = true }
 directories        = { workspace = true }

diff --git a/code/cli/src/args.rs b/code/cli/src/args.rs
@@ -7,7 +7,9 @@
 //! `clap` parses the command-line parameters into this structure.
 
 use std::path::{Path, PathBuf};
+use std::str::FromStr;
 
+use bytesize::ByteSize;
 use clap::{Parser, Subcommand};
 use color_eyre::eyre::{eyre, Context, Result};
 use directories::BaseDirs;
@@ -113,7 +115,9 @@ impl Args {
     pub fn load_config(&self) -> Result<Config> {
         let config_file = self.get_config_file_path()?;
         info!("Loading configuration from {:?}", config_file.display());
-        load_toml_file(&config_file)
+        let mut config = load_toml_file(&config_file)?;
+        override_config_from_env(&mut config)?;
+        Ok(config)
     }
 
     /// load_genesis returns the validator set from the genesis file
@@ -155,6 +159,18 @@ where
         .wrap_err_with(|| eyre!("Failed to load configuration at {}", file.display(),))
 }
 
+fn override_config_from_env(config: &mut Config) -> Result<()> {
+    use std::env;
+
+    if let Ok(max_block_size) = env::var("MALACHITE__CONSENSUS__MAX_BLOCK_SIZE") {
+        config.consensus.max_block_size = ByteSize::from_str(&max_block_size)
+            .map_err(|e| eyre!(e))
+            .wrap_err("Invalid MALACHITE__CONSENSUS__MAX_BLOCK_SIZE")?;
+    }
+
+    Ok(())
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

diff --git a/code/cli/src/cmd/testnet.rs b/code/cli/src/cmd/testnet.rs
@@ -2,6 +2,7 @@
 
 use std::path::Path;
 
+use bytesize::ByteSize;
 use color_eyre::eyre::Result;
 use rand::prelude::StdRng;
 use rand::rngs::OsRng;
@@ -101,6 +102,7 @@ pub fn generate_config(index: usize, total: usize) -> Config {
     Config {
         moniker: format!("test-{}", index),
         consensus: ConsensusConfig {
+            max_block_size: ByteSize::mib(1),
             timeouts: TimeoutConfig::default(),
             p2p: P2pConfig {
                 listen_addr: format!("/ip4/127.0.0.1/udp/{consensus_port}/quic-v1")

diff --git a/code/common/src/transaction.rs b/code/common/src/transaction.rs
@@ -15,4 +15,9 @@ impl Transaction {
     pub fn to_bytes(&self) -> Vec<u8> {
         self.0.to_vec()
     }
+
+    /// Size of this transaction in bytes
+    pub fn size_bytes(&self) -> u64 {
+        self.0.len() as u64
+    }
 }
diff --git a/code/config.toml b/code/config.toml
@@ -9,6 +9,11 @@ moniker = "malachite"
 ###         Consensus Configuration Options         ###
 #######################################################
 [consensus]
+# Maximum block size
+max_block_size = "1 MiB"
+
+## Timeouts
+
 # How long we wait for a proposal block before prevoting nil
 timeout_propose = "3s"
 # How much timeout_propose increases with each round

diff --git a/code/node/Cargo.toml b/code/node/Cargo.toml
@@ -18,6 +18,7 @@ malachite-network-mempool.workspace = true
 malachite-test.workspace    = true
 
 async-trait       = { workspace = true }
+bytesize          = { workspace = true, features = ["serde"] }
 derive-where      = { workspace = true }
 ed25519-consensus = { workspace = true, features = ["serde"] }
 humantime-serde   = { workspace = true }

diff --git a/code/node/src/config.rs b/code/node/src/config.rs
@@ -1,5 +1,6 @@
 use std::time::Duration;
 
+use bytesize::ByteSize;
 use malachite_common::TimeoutStep;
 use multiaddr::Multiaddr;
 use serde::{Deserialize, Serialize};
@@ -34,6 +35,9 @@ pub struct MempoolConfig {
 /// Consensus configuration options
 #[derive(Clone, Debug, Serialize, Deserialize)]
 pub struct ConsensusConfig {
+    /// Max block size
+    pub max_block_size: ByteSize,
+
     /// Timeouts
     #[serde(flatten)]
     pub timeouts: TimeoutConfig,