From 397d011ed984561022ab07c380a943474220e68a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Rapha=C3=ABl=20Proust?= Date: Wed, 13 Jun 2018 16:03:29 +0800 Subject: [PATCH] Alpha/endorser: one upcoming endorsment per key --- .../lib_delegate/client_baking_endorsement.ml | 114 +++++++++++------- 1 file changed, 71 insertions(+), 43 deletions(-) diff --git a/src/proto_alpha/lib_delegate/client_baking_endorsement.ml b/src/proto_alpha/lib_delegate/client_baking_endorsement.ml index e6778713e..1017736d6 100644 --- a/src/proto_alpha/lib_delegate/client_baking_endorsement.ml +++ b/src/proto_alpha/lib_delegate/client_baking_endorsement.ml @@ -141,17 +141,19 @@ let forge_endorsement (cctxt : #Proto_alpha.full) type state = { delegates: public_key_hash list ; delay: int64 ; - mutable to_endorse : endorsement option + (* invariant: only one slot per delegate *) + mutable to_endorse : endorsement list ; } and endorsement = { time: Time.t ; + timeout: unit Lwt.t ; delegate: public_key_hash ; block: Client_baking_blocks.block_info ; - slots: int list; + slots: int list ; } let create_state delegates delay = - { delegates ; delay ; to_endorse=None } + { delegates ; delay ; to_endorse=[] } let get_delegates cctxt state = match state.delegates with @@ -161,33 +163,48 @@ let get_delegates cctxt state = | _ :: _ as delegates -> return delegates +let endorse_for_delegate cctxt { delegate ; block ; slots ; } = + let hash = block.hash in + let b = `Hash (hash, 0) in + let level = block.level.level in + Client_keys.get_key cctxt delegate >>=? fun (name, _pk, sk) -> + lwt_debug "Endorsing %a for %s (level %a using %d slots)!" + Block_hash.pp_short hash name + Raw_level.pp level + (List.length slots) >>= fun () -> + inject_endorsement cctxt + b level + sk slots delegate >>=? fun oph -> + lwt_log_info + "Injected endorsement for block '%a' \ + (level %a, contract %s) '%a'" + Block_hash.pp_short hash + Raw_level.pp level + name + Operation_hash.pp_short oph >>= fun () -> + cctxt#message + "Injected endorsement level %a, contract %s '%a'" + Raw_level.pp level + name + Operation_hash.pp_short oph >>= fun () -> + return () + let endorse_for cctxt = function - None -> return () - | Some {delegate; block ; slots} -> - let hash = block.hash in - let b = `Hash (hash, 0) in - let level = block.level.level in - Client_keys.get_key cctxt delegate >>=? fun (name, _pk, sk) -> - lwt_debug "Endorsing %a for %s (level %a using %d slots)!" - Block_hash.pp_short hash name - Raw_level.pp level - (List.length slots) >>= fun () -> - inject_endorsement cctxt - b level - sk slots delegate >>=? fun oph -> - lwt_log_info - "Injected endorsement for block '%a' \ - (level %a, contract %s) '%a'" - Block_hash.pp_short hash - Raw_level.pp level - name - Operation_hash.pp_short oph >>= fun () -> - cctxt#message - "Injected endorsement level %a, contract %s '%a'" - Raw_level.pp level - name - Operation_hash.pp_short oph >>= fun () -> - return () + | [] -> return [] + | endorsments -> + let done_waiting, still_waiting, errored = + List.fold_left + (fun (r, s, f) ({ timeout } as endorsment) -> match Lwt.state timeout with + | Lwt.Return () -> (endorsment :: r, s, f) + | Lwt.Sleep -> (r, endorsment :: s, f) + | Lwt.Fail _ -> (r, s, endorsment :: f) + ) + ([], [], []) + endorsments + in + iter_p (endorse_for_delegate cctxt) done_waiting >>=? fun () -> + ignore errored; (* TODO: log *) + return still_waiting let allowed_to_endorse cctxt state (block: Client_baking_blocks.block_info) delegate time = Client_keys.Public_key_hash.name cctxt delegate >>=? fun name -> @@ -203,14 +220,23 @@ let allowed_to_endorse cctxt state (block: Client_baking_blocks.block_info) dele lwt_debug "Level %a (or higher) previously endorsed: do not endorse." Raw_level.pp level >>= return | false -> - let neu = {time ; delegate ; block; slots} in - match state.to_endorse with + let timeout = + let delay = (Time.diff time (Time.now ())) in + if delay <= 0L then + Lwt.return_unit + else + Lwt_unix.sleep (Int64.to_float delay) + in + let neu = { time ; timeout ; delegate ; block; slots } in + match List.find_opt (fun { delegate = d } -> delegate = d) state.to_endorse with | None -> - state.to_endorse <- Some neu; + state.to_endorse <- neu :: state.to_endorse; return () | Some old -> if Fitness.compare old.block.fitness neu.block.fitness < 0 then begin - state.to_endorse <- Some neu; + let without_old = + List.filter (fun to_end -> to_end <> old) state.to_endorse in + state.to_endorse <- neu :: without_old; return () end else lwt_debug "Block %a is not the fittest: do not endorse." @@ -235,13 +261,10 @@ let prepare_endorsement (cctxt : #Proto_alpha.full) ~(max_past:int64) state bi let compute_timeout state = match state.to_endorse with - | None -> Lwt_utils.never_ending - | Some {time} -> - let delay = (Time.diff time (Time.now ())) in - if delay <= 0L then - Lwt.return_unit - else - Lwt_unix.sleep (Int64.to_float delay) + | [] -> Lwt_utils.never_ending + | to_ends -> + Lwt.choose (List.map (fun to_end -> to_end.timeout) to_ends) + let check_error f = f >>= function @@ -283,9 +306,14 @@ let create last_get_block := None; check_error (prepare_endorsement cctxt ~max_past state bi) | `Timeout -> - check_error (endorse_for cctxt state.to_endorse) >>= fun () -> - state.to_endorse <- None ; - Lwt.return_unit + begin + endorse_for cctxt state.to_endorse >>= function + | Ok still_waiting -> + state.to_endorse <- still_waiting ; + Lwt.return_unit + | Error errs -> + lwt_log_error "Error while endorsing:@\n%a" pp_print_error errs + end end >>= fun () -> worker_loop () in