perf(commitment-generator): Run commitment generation for multiple batches in parallel

core/bin/external_node/src/config/mod.rs

@@ -745,6 +745,11 @@ pub(crate) struct ExperimentalENConfig {
     /// Maximum number of files concurrently opened by state keeper cache RocksDB. Useful to fit into OS limits; can be used
     /// as a rudimentary way to control RAM usage of the cache.
     pub state_keeper_db_max_open_files: Option<NonZeroU32>,
+
+    // Commitment generator
+    /// Maximum degree of parallelism during commitment generation, i.e., the maximum number of L1 batches being processed in parallel.
+    /// If not specified, commitment generator will use a value roughly equal to the number of CPU cores with some clamping applied.
+    pub commitment_generator_max_parallelism: Option<NonZeroU32>,
 }
 
 impl ExperimentalENConfig {
@@ -758,6 +763,7 @@ impl ExperimentalENConfig {
             state_keeper_db_block_cache_capacity_mb:
                 Self::default_state_keeper_db_block_cache_capacity_mb(),
             state_keeper_db_max_open_files: None,
+            commitment_generator_max_parallelism: None,
         }
     }
 

core/bin/external_node/src/main.rs

@@ -363,14 +363,13 @@ async fn run_core(
     );
     app_health.insert_component(batch_status_updater.health_check())?;
 
-    let commitment_generator_pool = singleton_pool_builder
-        .build()
-        .await
-        .context("failed to build a commitment_generator_pool")?;
-    let commitment_generator = CommitmentGenerator::new(
-        commitment_generator_pool,
+    let mut commitment_generator = CommitmentGenerator::new(
+        connection_pool.clone(),
         config.optional.l1_batch_commit_data_generator_mode,
     );
+    if let Some(parallelism) = config.experimental.commitment_generator_max_parallelism {
+        commitment_generator.set_max_parallelism(parallelism);
+    }
     app_health.insert_component(commitment_generator.health_check())?;
     let commitment_generator_handle = tokio::spawn(commitment_generator.run(stop_receiver.clone()));
 

core/lib/dal/src/blocks_dal.rs

@@ -164,6 +164,8 @@ impl BlocksDal<'_, '_> {
         Ok(row.number.map(|num| L1BatchNumber(num as u32)))
     }
 
+    /// Gets a number of the earliest L1 batch that is ready for commitment generation (i.e., doesn't have commitment
+    /// yet, and has tree data).
     pub async fn get_next_l1_batch_ready_for_commitment_generation(
         &mut self,
     ) -> DalResult<Option<L1BatchNumber>> {
@@ -190,6 +192,34 @@ impl BlocksDal<'_, '_> {
         Ok(row.map(|row| L1BatchNumber(row.number as u32)))
     }
 
+    /// Gets a number of the last L1 batch that is ready for commitment generation (i.e., doesn't have commitment
+    /// yet, and has tree data).
+    pub async fn get_last_l1_batch_ready_for_commitment_generation(
+        &mut self,
+    ) -> DalResult<Option<L1BatchNumber>> {
+        let row = sqlx::query!(
+            r#"
+            SELECT
+                number
+            FROM
+                l1_batches
+            WHERE
+                hash IS NOT NULL
+                AND commitment IS NULL
+            ORDER BY
+                number DESC
+            LIMIT
+                1
+            "#
+        )
+        .instrument("get_last_l1_batch_ready_for_commitment_generation")
+        .report_latency()
+        .fetch_optional(self.storage)
+        .await?;
+
+        Ok(row.map(|row| L1BatchNumber(row.number as u32)))
+    }
+
     /// Returns the number of the earliest L1 batch with metadata (= state hash) present in the DB,
     /// or `None` if there are no such L1 batches.
     pub async fn get_earliest_l1_batch_number_with_metadata(

core/node/commitment_generator/Cargo.toml

@@ -28,11 +28,16 @@ zk_evm_1_4_1.workspace = true
 zk_evm_1_3_3.workspace = true
 
 tokio = { workspace = true, features = ["time"] }
+futures.workspace = true
+num_cpus.workspace = true
 anyhow.workspace = true
 tracing.workspace = true
 itertools.workspace = true
 serde_json.workspace = true
 
 [dev-dependencies]
-jsonrpsee.workspace = true
 zksync_web3_decl.workspace = true
+zksync_node_genesis.workspace = true
+zksync_node_test_utils.workspace = true
+
+rand.workspace = true