Generate operations dynamically (#4345)

## Motivation

Right now we always precalculate the operations for each block. However,
the performance gain we get from it should be minimal, and it limits us
on what we can do.

## Proposal

Generate the operations dynamically instead, giving more flexibility on
what we can do.

## Test Plan

Tested against a local network, works as expected

## Release Plan

- Nothing to do / These changes follow the usual release cycle.

Author: ndr-ds

linera-client/src/benchmark.rs

@@ -23,7 +23,7 @@ use linera_execution::{system::SystemOperation, Operation};
 use linera_sdk::abis::fungible::FungibleOperation;
 use num_format::{Locale, ToFormattedString};
 use prometheus_parse::{HistogramCount, Scrape, Value};
-use rand::{seq::SliceRandom, thread_rng};
+use rand::{rngs::SmallRng, seq::SliceRandom, thread_rng, SeedableRng};
 use serde::{Deserialize, Serialize};
 use tokio::{
     sync::{mpsc, Barrier, Notify},
@@ -73,6 +73,8 @@ pub enum BenchmarkError {
     ConfigLoadError(#[from] anyhow::Error),
     #[error("Could not find enough chains in wallet alone: needed {0}, but only found {1}")]
     NotEnoughChainsInWallet(usize, usize),
+    #[error("Random number generator error: {0}")]
+    RandError(#[from] rand::Error),
 }
 
 #[derive(Debug)]
@@ -109,17 +111,18 @@ pub struct Benchmark<Env: Environment> {
 impl<Env: Environment> Benchmark<Env> {
     #[expect(clippy::too_many_arguments)]
     pub async fn run_benchmark<C: ClientContext<Environment = Env> + 'static>(
-        num_chains: usize,
-        transactions_per_block: usize,
         bps: usize,
         chain_clients: Vec<ChainClient<Env>>,
-        blocks_infos: Vec<Vec<Operation>>,
+        all_chains: Vec<ChainId>,
+        transactions_per_block: usize,
+        fungible_application_id: Option<ApplicationId>,
         health_check_endpoints: Option<String>,
         runtime_in_seconds: Option<u64>,
         delay_between_chains_ms: Option<u64>,
         chain_listener: ChainListener<C>,
         shutdown_notifier: &CancellationToken,
     ) -> Result<(), BenchmarkError> {
+        let num_chains = chain_clients.len();
         let bps_counts = (0..num_chains)
             .map(|_| Arc::new(AtomicUsize::new(0)))
             .collect::<Vec<_>>();
@@ -150,31 +153,30 @@ impl<Env: Environment> Benchmark<Env> {
         let bps_initial_share = bps / num_chains;
         let mut bps_remainder = bps % num_chains;
         let mut join_set = task::JoinSet::<Result<(), BenchmarkError>>::new();
-        for (chain_idx, (block_info, chain_client)) in blocks_infos
-            .into_iter()
-            .zip(chain_clients.into_iter())
-            .enumerate()
-        {
+        for (chain_idx, chain_client) in chain_clients.into_iter().enumerate() {
+            let chain_id = chain_client.chain_id();
             let shutdown_notifier_clone = shutdown_notifier.clone();
             let barrier_clone = barrier.clone();
             let bps_count_clone = bps_counts[chain_idx].clone();
             let notifier_clone = notifier.clone();
             let runtime_control_sender_clone = runtime_control_sender.clone();
+            let all_chains_clone = all_chains.clone();
             let bps_share = if bps_remainder > 0 {
                 bps_remainder -= 1;
                 bps_initial_share + 1
             } else {
                 bps_initial_share
             };
-            let chain_id = chain_client.chain_id();
             join_set.spawn(
                 async move {
                     Box::pin(Self::run_benchmark_internal(
                         chain_idx,
                         chain_id,
                         bps_share,
-                        block_info,
                         chain_client,
+                        all_chains_clone,
+                        transactions_per_block,
+                        fungible_application_id,
                         shutdown_notifier_clone,
                         bps_count_clone,
                         barrier_clone,
@@ -554,8 +556,10 @@ impl<Env: Environment> Benchmark<Env> {
         chain_idx: usize,
         chain_id: ChainId,
         bps: usize,
-        operations: Vec<Operation>,
         chain_client: ChainClient<Env>,
+        all_chains: Vec<ChainId>,
+        transactions_per_block: usize,
+        fungible_application_id: Option<ApplicationId>,
         shutdown_notifier: CancellationToken,
         bps_count: Arc<AtomicUsize>,
         barrier: Arc<Barrier>,
@@ -576,6 +580,12 @@ impl<Env: Environment> Benchmark<Env> {
             runtime_control_sender.send(()).await?;
         }
 
+        let owner = chain_client
+            .identity()
+            .await
+            .map_err(BenchmarkError::ChainClient)?;
+        let mut destination_manager = ChainDestinationManager::new(chain_id, all_chains)?;
+
         loop {
             tokio::select! {
                 biased;
@@ -584,7 +594,15 @@ impl<Env: Environment> Benchmark<Env> {
                     info!("Shutdown signal received, stopping benchmark");
                     break;
                 }
-                result = chain_client.execute_operations(operations.clone(), vec![]) => {
+                result = chain_client.execute_operations(
+                    Self::generate_operations(
+                        owner,
+                        transactions_per_block,
+                        fungible_application_id,
+                        &mut destination_manager,
+                    ),
+                    vec![]
+                ) => {
                     result
                         .map_err(BenchmarkError::ChainClient)?
                         .expect("should execute block with operations");
@@ -601,6 +619,47 @@ impl<Env: Environment> Benchmark<Env> {
         Ok(())
     }
 
+    fn create_operation(
+        fungible_application_id: Option<ApplicationId>,
+        recipient_chain_id: ChainId,
+        owner: AccountOwner,
+        amount: Amount,
+    ) -> Operation {
+        match fungible_application_id {
+            Some(application_id) => {
+                Self::fungible_transfer(application_id, recipient_chain_id, owner, owner, amount)
+            }
+            None => Operation::system(SystemOperation::Transfer {
+                owner: AccountOwner::CHAIN,
+                recipient: Account::chain(recipient_chain_id),
+                amount,
+            }),
+        }
+    }
+
+    /// Generate operations for a single block, randomizing destinations after each full cycle
+    fn generate_operations(
+        owner: AccountOwner,
+        transactions_per_block: usize,
+        fungible_application_id: Option<ApplicationId>,
+        destination_manager: &mut ChainDestinationManager,
+    ) -> Vec<Operation> {
+        let mut operations = Vec::with_capacity(transactions_per_block);
+        let amount = Amount::from_attos(1);
+
+        for _ in 0..transactions_per_block {
+            let recipient_chain_id = destination_manager.get_next_destination();
+            operations.push(Self::create_operation(
+                fungible_application_id,
+                recipient_chain_id,
+                owner,
+                amount,
+            ));
+        }
+
+        operations
+    }
+
     /// Closes the chain that was created for the benchmark.
     pub async fn close_benchmark_chain(
         chain_client: &ChainClient<Env>,
@@ -640,76 +699,6 @@ impl<Env: Environment> Benchmark<Env> {
         Ok(all_chains)
     }
 
-    pub fn make_benchmark_block_info(
-        benchmark_chains: Vec<(ChainId, AccountOwner)>,
-        transactions_per_block: usize,
-        fungible_application_id: Option<ApplicationId>,
-        all_chains: Vec<ChainId>,
-    ) -> Result<Vec<Vec<Operation>>, BenchmarkError> {
-        let mut blocks_infos = Vec::new();
-        let amount = Amount::from_attos(1);
-
-        for (current_chain_id, owner) in benchmark_chains.iter() {
-            let mut operations = Vec::new();
-
-            let mut other_chains: Vec<_> = if all_chains.len() == 1 {
-                // If there's only one chain, just have it send to itself.
-                all_chains.clone()
-            } else {
-                // If there's more than one chain, have it send to all other chains, and don't
-                // send to self.
-                all_chains
-                    .iter()
-                    .filter(|chain_id| **chain_id != *current_chain_id)
-                    .copied()
-                    .collect()
-            };
-
-            other_chains.shuffle(&mut thread_rng());
-
-            // Calculate adjusted transactions_per_block to ensure even distribution
-            let num_destinations = other_chains.len();
-            let adjusted_transactions_per_block = if transactions_per_block % num_destinations != 0
-            {
-                let adjusted = transactions_per_block.div_ceil(num_destinations) * num_destinations;
-                warn!(
-                    "Requested transactions_per_block ({}) is not evenly divisible by number of destination chains ({}). \
-                    Adjusting to {} transactions per block to ensure transfers cancel each other out.",
-                    transactions_per_block, num_destinations, adjusted
-                );
-                adjusted
-            } else {
-                transactions_per_block
-            };
-
-            for recipient_chain_id in other_chains {
-                let operation = match fungible_application_id {
-                    Some(application_id) => Self::fungible_transfer(
-                        application_id,
-                        recipient_chain_id,
-                        *owner,
-                        *owner,
-                        amount,
-                    ),
-                    None => Operation::system(SystemOperation::Transfer {
-                        owner: AccountOwner::CHAIN,
-                        recipient: Account::chain(recipient_chain_id),
-                        amount,
-                    }),
-                };
-                operations.push(operation);
-            }
-
-            let operations = operations
-                .into_iter()
-                .cycle()
-                .take(adjusted_transactions_per_block)
-                .collect();
-            blocks_infos.push(operations);
-        }
-        Ok(blocks_infos)
-    }
-
     /// Creates a fungible token transfer operation.
     pub fn fungible_transfer(
         application_id: ApplicationId,
@@ -734,3 +723,45 @@ impl<Env: Environment> Benchmark<Env> {
         }
     }
 }
+
+struct ChainDestinationManager {
+    source_chain_id: ChainId,
+    destination_index: usize,
+    destination_chains: Vec<ChainId>,
+    rng: SmallRng,
+}
+
+impl ChainDestinationManager {
+    fn new(
+        source_chain_id: ChainId,
+        mut destination_chains: Vec<ChainId>,
+    ) -> Result<Self, BenchmarkError> {
+        let mut rng = SmallRng::from_rng(thread_rng())?;
+        destination_chains.shuffle(&mut rng);
+
+        Ok(Self {
+            source_chain_id,
+            destination_index: 0,
+            destination_chains,
+            rng,
+        })
+    }
+
+    fn get_next_destination(&mut self) -> ChainId {
+        // Check if we've gone through all destinations
+        if self.destination_index >= self.destination_chains.len() {
+            // Reshuffle the destinations for the next cycle
+            self.destination_chains.shuffle(&mut self.rng);
+            self.destination_index = 0;
+        }
+
+        let destination_chain_id = self.destination_chains[self.destination_index];
+        self.destination_index += 1;
+
+        if destination_chain_id == self.source_chain_id {
+            self.get_next_destination()
+        } else {
+            destination_chain_id
+        }
+    }
+}

linera-client/src/client_context.rs

@@ -671,12 +671,11 @@ where
     pub async fn prepare_for_benchmark(
         &mut self,
         num_chains: usize,
-        transactions_per_block: usize,
         tokens_per_chain: Amount,
         fungible_application_id: Option<ApplicationId>,
         pub_keys: Vec<AccountPublicKey>,
         chains_config_path: Option<&Path>,
-    ) -> Result<(Vec<ChainClient<Env>>, Vec<Vec<Operation>>), Error> {
+    ) -> Result<(Vec<ChainClient<Env>>, Vec<ChainId>), Error> {
         let start = Instant::now();
         // Below all block proposals are supposed to succeed without retries, we
         // must make sure that all incoming payments have been accepted on-chain
@@ -736,14 +735,7 @@ where
             }
         }
 
-        let blocks_infos = Benchmark::<Env>::make_benchmark_block_info(
-            benchmark_chains,
-            transactions_per_block,
-            fungible_application_id,
-            all_chains,
-        )?;
-
-        Ok((chain_clients, blocks_infos))
+        Ok((chain_clients, all_chains))
     }
 
     pub async fn wrap_up_benchmark(
@@ -841,20 +833,19 @@ where
             if chains_found_in_wallet == num_chains {
                 break;
             }
-            // This should never panic, because `owned_chain_ids` only returns the owned chains that
-            // we have a key pair for.
-            let owner = self
-                .wallet
-                .get(chain_id)
-                .and_then(|chain| chain.owner)
-                .unwrap();
             let chain_client = self.make_chain_client(chain_id);
             let ownership = chain_client.chain_info().await?.manager.ownership;
             if !ownership.owners.is_empty() || ownership.super_owners.len() != 1 {
                 continue;
             }
             chain_client.process_inbox().await?;
-            benchmark_chains.push((chain_id, owner));
+            benchmark_chains.push((
+                chain_id,
+                *ownership
+                    .super_owners
+                    .first()
+                    .expect("should have a super owner"),
+            ));
             chain_clients.push(chain_client);
             chains_found_in_wallet += 1;
         }
@@ -940,6 +931,12 @@ where
         info!("Processing default chain inbox");
         default_chain_client.process_inbox().await?;
 
+        assert_eq!(
+            benchmark_chains.len(),
+            chain_clients.len(),
+            "benchmark_chains and chain_clients must have the same size"
+        );
+
         Ok((benchmark_chains, chain_clients))
     }
 

linera-service/src/cli/main.rs

@@ -841,10 +841,9 @@ impl Runnable for Job {
                         wallet,
                         signer.into_value(),
                     );
-                    let (chain_clients, blocks_infos) = context
+                    let (chain_clients, all_chains) = context
                         .prepare_for_benchmark(
                             num_chains,
-                            transactions_per_block,
                             tokens_per_chain,
                             fungible_application_id,
                             pub_keys,
@@ -883,11 +882,11 @@ impl Runnable for Job {
                         shutdown_notifier.clone(),
                     );
                     linera_client::benchmark::Benchmark::run_benchmark(
-                        num_chains,
-                        transactions_per_block,
                         bps,
                         chain_clients.clone(),
-                        blocks_infos,
+                        all_chains,
+                        transactions_per_block,
+                        fungible_application_id,
                         health_check_endpoints.clone(),
                         runtime_in_seconds,
                         delay_between_chains_ms,