Support for non-immediate proof maintenance (#4563)

When doing large real-world proofs with a high degree of parallelism,
the bandwidth is overtaken by the writing of the proof to disk on every
proof iteration, slowing the process considerably.

This PR introduces the ability to write to disk and perform the callback
on every `maintenance_iterations` of the parallel proof, set by default
to 32.

EDIT: Actually, perhaps it's a better idea to write the proof every
`max_workers` iterations.

---------

Co-authored-by: rv-jenkins <[email protected]>
Co-authored-by: devops <[email protected]>
Co-authored-by: Noah Watson <[email protected]>
Co-authored-by: Georgy Lukyanov <[email protected]>

pyk/src/pyk/proof/proof.py

@@ -326,6 +326,7 @@ def parallel_advance_proof(
     fail_fast: bool = False,
     max_workers: int = 1,
     callback: Callable[[P], None] = (lambda x: None),
+    maintenance_rate: int = 1,
 ) -> None:
     """Advance proof with multithreaded strategy.
 
@@ -347,7 +348,8 @@ def parallel_advance_proof(
         fail_fast: If the proof is failing after finishing a step,
           halt execution even if there are still available steps.
         max_workers: Maximum number of worker threads the pool can spawn.
-        callback: Callable to run in between each completed step, useful for getting real-time information about the proof.
+        callback: Callable to run during proof maintenance, useful for getting real-time information about the proof.
+        maintenance_rate: Number of iterations between proof maintenance (writing to disk and executing callback).
     """
     pending: set[Future[Any]] = set()
     explored: set[PS] = set()
@@ -376,9 +378,10 @@ def submit_steps(_steps: Iterable[PS]) -> None:
                 proof_results = future.result()
                 for result in proof_results:
                     proof.commit(result)
-                proof.write_proof_data()
-                callback(proof)
                 iterations += 1
+                if iterations % maintenance_rate == 0:
+                    proof.write_proof_data()
+                    callback(proof)
                 if max_iterations is not None and max_iterations <= iterations:
                     break
                 if fail_fast and proof.failed:
@@ -502,6 +505,7 @@ def advance_proof(
         max_iterations: int | None = None,
         fail_fast: bool = False,
         callback: Callable[[P], None] = (lambda x: None),
+        maintenance_rate: int = 1,
     ) -> None:
         """Advance a proof.
 
@@ -513,6 +517,7 @@ def advance_proof(
             fail_fast: If the proof is failing after finishing a step,
               halt execution even if there are still available steps.
             callback: Callable to run in between each completed step, useful for getting real-time information about the proof.
+            maintenance_rate: Number of iterations between proof maintenance (writing to disk and executing callback).
         """
         iterations = 0
         _LOGGER.info(f'Initializing proof: {proof.id}')
@@ -533,7 +538,9 @@ def advance_proof(
                 results = self.step_proof(step)
                 for result in results:
                     proof.commit(result)
-                proof.write_proof_data()
-                callback(proof)
+                if iterations % maintenance_rate == 0:
+                    proof.write_proof_data()
+                    callback(proof)
+
         if proof.failed:
             proof.failure_info = self.failure_info(proof)