Revert "Refactor block production control flow"

This reverts commit 599ebf0.

Author: amc-ie

src/lib/block_producer/block_producer.ml

@@ -27,6 +27,47 @@ end
 type Structured_log_events.t += Block_produced
   [@@deriving register_event { msg = "Successfully produced a new block" }]
 
+module Singleton_supervisor : sig
+  type ('data, 'a) t
+
+  val create :
+    task:(unit Ivar.t -> 'data -> ('a, unit) Interruptible.t) -> ('data, 'a) t
+
+  val cancel : (_, _) t -> unit
+
+  val dispatch : ('data, 'a) t -> 'data -> ('a, unit) Interruptible.t
+end = struct
+  type ('data, 'a) t =
+    { mutable task : (unit Ivar.t * ('a, unit) Interruptible.t) option
+    ; f : unit Ivar.t -> 'data -> ('a, unit) Interruptible.t
+    }
+
+  let create ~task = { task = None; f = task }
+
+  let cancel t =
+    match t.task with
+    | Some (ivar, _) ->
+        if Ivar.is_full ivar then
+          [%log' error (Logger.create ())] "Ivar.fill bug is here!" ;
+        Ivar.fill ivar () ;
+        t.task <- None
+    | None ->
+        ()
+
+  let dispatch t data =
+    cancel t ;
+    let ivar = Ivar.create () in
+    let interruptible =
+      let open Interruptible.Let_syntax in
+      t.f ivar data
+      >>| fun x ->
+      t.task <- None ;
+      x
+    in
+    t.task <- Some (ivar, interruptible) ;
+    interruptible
+end
+
 let time_to_ms = Fn.compose Block_time.Span.to_ms Block_time.to_span_since_epoch
 
 let time_of_ms = Fn.compose Block_time.of_span_since_epoch Block_time.Span.of_ms
@@ -38,6 +79,54 @@ let lift_sync f =
            [%log' error (Logger.create ())] "Ivar.fill bug is here!" ;
          Ivar.fill ivar (f ()) ) )
 
+module Singleton_scheduler : sig
+  type t
+
+  val create : Block_time.Controller.t -> t
+
+  (** If you reschedule when already scheduled, take the min of the two schedulings *)
+  val schedule : t -> Block_time.t -> f:(unit -> unit) -> unit
+end = struct
+  type t =
+    { mutable timeout : unit Block_time.Timeout.t option
+    ; time_controller : Block_time.Controller.t
+    }
+
+  let create time_controller = { time_controller; timeout = None }
+
+  let cancel t =
+    match t.timeout with
+    | Some timeout ->
+        Block_time.Timeout.cancel t.time_controller timeout () ;
+        t.timeout <- None
+    | None ->
+        ()
+
+  let schedule t time ~f =
+    let remaining_time =
+      Option.map t.timeout ~f:Block_time.Timeout.remaining_time
+    in
+    cancel t ;
+    let span_till_time =
+      Block_time.diff time (Block_time.now t.time_controller)
+    in
+    let wait_span =
+      match remaining_time with
+      | Some remaining
+        when Block_time.Span.(remaining > Block_time.Span.of_ms Int64.zero) ->
+          let min a b = if Block_time.Span.(a < b) then a else b in
+          min remaining span_till_time
+      | None | Some _ ->
+          span_till_time
+    in
+    let timeout =
+      Block_time.Timeout.create t.time_controller wait_span ~f:(fun _ ->
+          t.timeout <- None ;
+          f () )
+    in
+    t.timeout <- Some timeout
+end
+
 (** Sends an error to the reporting service containing as many failed transactions as we can fit. *)
 let report_transaction_inclusion_failures ~commit_id ~logger failed_txns =
   let num_failures = List.length failed_txns in
@@ -595,7 +684,7 @@ let produce ~genesis_breadcrumb ~context:(module Context : CONTEXT) ~prover
     ~verifier ~trust_system ~get_completed_work ~transaction_resource_pool
     ~frontier_reader ~time_controller ~transition_writer ~log_block_creation
     ~block_reward_threshold ~block_produced_bvar ~slot_tx_end ~slot_chain_end
-    ~net ~zkapp_cmd_limit_hardcap interrupt_ivar
+    ~net ~zkapp_cmd_limit_hardcap ivar
     (scheduled_time, block_data, winner_pubkey) =
   let open Context in
   let module Breadcrumb = Transition_frontier.Breadcrumb in
@@ -683,9 +772,7 @@ let produce ~genesis_breadcrumb ~context:(module Context : CONTEXT) ~prover
         |> Sequence.map
              ~f:Transaction_hash.User_command_with_valid_signature.data
       in
-      let%bind () =
-        Interruptible.lift (Deferred.return ()) (Ivar.read interrupt_ivar)
-      in
+      let%bind () = Interruptible.lift (Deferred.return ()) (Ivar.read ivar) in
       [%log internal] "Generate_next_state" ;
       let%bind next_state_opt =
         generate_next_state ~commit_id ~constraint_constants ~scheduled_time
@@ -1120,19 +1207,14 @@ let iteration ~schedule_next_vrf_check ~produce_block_now
                  ~frontier_reader () ) ;
             schedule_block_production (scheduled_time, data, winner_pk) )
 
-let schedule ~time_controller time =
-  let span_till_time = Block_time.diff time (Block_time.now time_controller) in
-  Block_time.Timeout.create time_controller span_till_time ~f:Fn.id
-  |> Block_time.Timeout.to_deferred
-
 let run ~context:(module Context : CONTEXT) ~vrf_evaluator ~prover ~verifier
     ~trust_system ~get_completed_work ~transaction_resource_pool
     ~time_controller ~consensus_local_state ~coinbase_receiver ~frontier_reader
     ~transition_writer ~set_next_producer_timing ~log_block_creation
     ~block_reward_threshold ~block_produced_bvar ~vrf_evaluation_state ~net
     ~zkapp_cmd_limit_hardcap =
   let open Context in
-  O1trace.sync_thread "produce_blocks_run" (fun () ->
+  O1trace.sync_thread "produce_blocks" (fun () ->
       let genesis_breadcrumb =
         genesis_breadcrumb_creator ~context:(module Context) prover
       in
@@ -1152,17 +1234,19 @@ let run ~context:(module Context : CONTEXT) ~vrf_evaluator ~prover ~verifier
           ~zkapp_cmd_limit_hardcap
       in
       let module Breadcrumb = Transition_frontier.Breadcrumb in
-      let iteration_wrapped (slot, i, prev_step) =
+      let production_supervisor = Singleton_supervisor.create ~task:produce in
+      let scheduler = Singleton_scheduler.create time_controller in
+      let rec check_next_block_timing slot i () =
         (* Begin checking for the ability to produce a block *)
         match Broadcast_pipe.Reader.peek frontier_reader with
         | None ->
             log_bootstrap_mode ~logger () ;
-            let%map () =
-              (* Iterates until there is some frontier *)
-              Broadcast_pipe.Reader.iter_until frontier_reader
-                ~f:(Fn.compose Deferred.return Option.is_some)
-            in
-            (slot, i, prev_step)
+            don't_wait_for
+              (let%map () =
+                 Broadcast_pipe.Reader.iter_until frontier_reader
+                   ~f:(Fn.compose Deferred.return Option.is_some)
+               in
+               check_next_block_timing slot i () )
         | Some transition_frontier ->
             let consensus_state =
               Transition_frontier.best_tip transition_frontier
@@ -1210,54 +1294,48 @@ let run ~context:(module Context : CONTEXT) ~vrf_evaluator ~prover ~verifier
                 "Block producer will begin producing only empty blocks after \
                  $slot_diff slots"
               slot_tx_end ;
+            let next_vrf_check_now =
+              check_next_block_timing new_global_slot i'
+            in
             (* TODO: Re-enable this assertion when it doesn't fail dev demos
              *       (see #5354)
              * assert (
                    Consensus.Hooks.required_local_state_sync
                     ~constants:consensus_constants ~consensus_state
                     ~local_state:consensus_local_state
                    = None ) ; *)
-            let next_vrf_check_now () =
-              return (new_global_slot, i', prev_step)
-            in
-            let produce_block_now data =
-              Option.iter !prev_step ~f:(fun ivar ->
-                  if Ivar.is_full ivar then
-                    [%log error] "Ivar.fill bug is here!" ;
-                  Ivar.fill ivar () ) ;
-              let intr_ivar = Ivar.create () in
-              let this_step = ref (Some intr_ivar) in
-              let produce_intr =
-                let%map.Interruptible x = produce intr_ivar data in
-                this_step := None ;
-                x
-              in
-              let%map _ = Interruptible.force produce_intr in
-              (* TODO consider uncommenting below or removing interruptible usage completely *)
-              (* Interruptible.don't_wait_for produce_intr ; *)
-              (new_global_slot, i', this_step)
-            in
-            let schedule_next_vrf_check time =
-              let%map _ = schedule ~time_controller time in
-              (new_global_slot, i', prev_step)
-            in
-            let schedule_block_production (time, data, winner) =
-              let%bind _ = schedule ~time_controller time in
-              produce_block_now (time, data, winner)
+            let produce_block_now triple =
+              ignore
+                ( Interruptible.finally
+                    (Singleton_supervisor.dispatch production_supervisor triple)
+                    ~f:next_vrf_check_now
+                  : (_, _) Interruptible.t )
             in
-            iteration ~schedule_next_vrf_check ~produce_block_now
-              ~schedule_block_production ~next_vrf_check_now ~genesis_breadcrumb
-              ~context:(module Context)
-              ~vrf_evaluator ~time_controller ~coinbase_receiver
-              ~frontier_reader ~set_next_producer_timing ~transition_frontier
-              ~vrf_evaluation_state ~epoch_data_for_vrf ~ledger_snapshot i slot
+            don't_wait_for
+              ( iteration
+                  ~schedule_next_vrf_check:
+                    (Fn.compose Deferred.return
+                       (Singleton_scheduler.schedule scheduler
+                          ~f:next_vrf_check_now ) )
+                  ~produce_block_now:
+                    (Fn.compose Deferred.return produce_block_now)
+                  ~schedule_block_production:(fun (time, data, winner) ->
+                    Singleton_scheduler.schedule scheduler time ~f:(fun () ->
+                        produce_block_now (time, data, winner) ) ;
+                    Deferred.unit )
+                  ~next_vrf_check_now:
+                    (Fn.compose Deferred.return next_vrf_check_now)
+                  ~genesis_breadcrumb
+                  ~context:(module Context)
+                  ~vrf_evaluator ~time_controller ~coinbase_receiver
+                  ~frontier_reader ~set_next_producer_timing
+                  ~transition_frontier ~vrf_evaluation_state ~epoch_data_for_vrf
+                  ~ledger_snapshot i slot
+                : unit Deferred.t )
       in
-      let start _ =
-        Deferred.forever
-          ( Mina_numbers.Global_slot_since_hard_fork.zero
-          , Mina_numbers.Length.zero
-          , ref @@ Some (Ivar.create ()) )
-          iteration_wrapped
+      let start () =
+        check_next_block_timing Mina_numbers.Global_slot_since_hard_fork.zero
+          Mina_numbers.Length.zero ()
       in
       let genesis_state_timestamp =
         consensus_constants.genesis_state_timestamp
@@ -1266,17 +1344,20 @@ let run ~context:(module Context : CONTEXT) ~vrf_evaluator ~prover ~verifier
       let now = Block_time.now time_controller in
       if Block_time.( >= ) now genesis_state_timestamp then start ()
       else
+        let time_till_genesis = Block_time.diff genesis_state_timestamp now in
         [%log warn]
           ~metadata:
             [ ( "time_till_genesis"
               , `Int
-                  (Int64.to_int_exn
-                     ( Block_time.Span.to_ms
-                     @@ Block_time.diff genesis_state_timestamp now ) ) )
+                  (Int64.to_int_exn (Block_time.Span.to_ms time_till_genesis))
+              )
             ]
           "Node started before genesis: waiting $time_till_genesis \
            milliseconds before starting block producer" ;
-      upon (schedule ~time_controller genesis_state_timestamp) start )
+        ignore
+          ( Block_time.Timeout.create time_controller time_till_genesis
+              ~f:(fun _ -> start ())
+            : unit Block_time.Timeout.t ) )
 
 let run_precomputed ~context:(module Context : CONTEXT) ~verifier ~trust_system
     ~time_controller ~frontier_reader ~transition_writer ~precomputed_blocks =