optimize: limit number of blocks proving in parallel (#600)

* optimize: limit number of blocks proving in parallel

* fix: nit

* fix: review

zero_bin/prover/src/lib.rs

@@ -13,11 +13,24 @@ use proof_gen::proof_types::GeneratedBlockProof;
 use serde::{Deserialize, Serialize};
 use tokio::io::AsyncWriteExt;
 use tokio::sync::mpsc::Receiver;
-use tokio::sync::oneshot;
+use tokio::sync::{oneshot, Semaphore};
 use trace_decoder::{BlockTrace, OtherBlockData};
 use tracing::{error, info};
 use zero_bin_common::fs::generate_block_proof_file_name;
 
+// All proving tasks are executed concurrently, which can cause issues for large
+// block intervals, where distant future blocks may be proven first.
+//
+// We then create a pool to limit the number of parallel proving block
+// tasks, retrieving new blocks in increasing order when some block proofs are
+// complete.
+//
+// While proving a block interval, we will output proofs corresponding to block
+// batches as soon as they are generated.
+const PARALLEL_BLOCK_PROVING_PERMIT_POOL_SIZE: usize = 16;
+static PARALLEL_BLOCK_PROVING_PERMIT_POOL: Semaphore =
+    Semaphore::const_new(PARALLEL_BLOCK_PROVING_PERMIT_POOL_SIZE);
+
 #[derive(Debug, Clone)]
 pub struct ProverConfig {
     pub batch_size: usize,
@@ -242,6 +255,9 @@ pub async fn prove(
         let previous_block_proof = prev_proof.take();
         let runtime = runtime.clone();
         let block_number = block_prover_input.get_block_number();
+
+        let prove_permit = PARALLEL_BLOCK_PROVING_PERMIT_POOL.acquire().await?;
+
         let _abort_handle = task_set.spawn(async move {
             let block_number = block_prover_input.get_block_number();
             info!("Proving block {block_number}");
@@ -253,6 +269,7 @@ pub async fn prove(
                 prover_config.clone(),
             )
             .then(move |proof| async move {
+                drop(prove_permit);
                 let proof = proof.inspect_err(|e| {
                     error!("failed to generate proof for block {block_number}, error {e:?}")
                 })?;