Revision 50ad02fa2a56f1d3541a6d8b3c8a6a93160d4f54 authored by Albin Coquereau on 05 April 2024, 09:32:54 UTC, committed by Albin Coquereau on 05 April 2024, 09:32:54 UTC
1 parent 9a71363
consensus_heuristic.ml
(*****************************************************************************)
(* *)
(* Open Source License *)
(* Copyright (c) 2021 Nomadic Labs. <contact@nomadic-labs.com> *)
(* *)
(* Permission is hereby granted, free of charge, to any person obtaining a *)
(* copy of this software and associated documentation files (the "Software"),*)
(* to deal in the Software without restriction, including without limitation *)
(* the rights to use, copy, modify, merge, publish, distribute, sublicense, *)
(* and/or sell copies of the Software, and to permit persons to whom the *)
(* Software is furnished to do so, subject to the following conditions: *)
(* *)
(* The above copyright notice and this permission notice shall be included *)
(* in all copies or substantial portions of the Software. *)
(* *)
(* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR*)
(* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, *)
(* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL *)
(* THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER*)
(* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING *)
(* FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER *)
(* DEALINGS IN THE SOFTWARE. *)
(* *)
(*****************************************************************************)
type 'a t = {
threshold : int;
expected : int;
candidates : 'a P2p_peer.Id.Table.t;
compare : 'a -> 'a -> int;
values_map_module : (module Map.S with type key = 'a);
}
let update t (peer_id, data) =
(* To guarantee that the memory used by the table is bounded by
[t.expected]. *)
if
P2p_peer.Id.Table.length t.candidates < t.expected
|| P2p_peer.Id.Table.mem t.candidates peer_id
then P2p_peer.Id.Table.replace t.candidates peer_id data
type 'a state =
| Consensus of 'a
| No_consensus of ('a * int) list
| Need_more_candidates
let create (type a) ?(compare = compare) ~expected ~threshold () =
if threshold < 0 then invalid_arg "Hash_heuristic.create: threshold negative" ;
if expected < threshold then
invalid_arg "Hash_heuristic.create: expected lower than threshold" ;
if expected >= threshold * 2 then
invalid_arg
"Hash_heuristic.create: expected greater than twice the threshold" ;
(* We store the result of the functor application so that we don't
do it at every call of `get_state`. *)
let (values_map_module : (module Map.S with type key = a)) =
(module Map.Make (struct
type t = a
let compare = compare
end))
in
{
expected;
threshold;
candidates = P2p_peer.Id.Table.create 11;
compare;
values_map_module;
}
let get_state (type a) (t : a t) =
if P2p_peer_id.Table.length t.candidates < t.threshold then
Need_more_candidates
else
(* We retrieve the result of the functor application. *)
let (module Map : Map.S with type key = a) = t.values_map_module in
(* We use Error as an early return mechanism. *)
P2p_peer_id.Table.fold_e
(fun _ value map ->
let count = Map.find value map |> Option.fold ~some:succ ~none:1 in
let map' = Map.add value count map in
if count = t.threshold then Error value else Ok map')
t.candidates
Map.empty
|> function
| Ok map ->
if P2p_peer_id.Table.length t.candidates < t.expected then
Need_more_candidates
else No_consensus (Map.bindings map)
| Error value -> Consensus value
module Worker = struct
type 'a t = {
expire_time : Ptime.Span.t;
(* Expiration delay for a consensus value *)
state : 'a state;
(* Internal state of the consensus heuristic *)
mutable expired : unit Lwt.t;
(* Expiration promise which is fulfilled when the consensus value
has expired. The expiration mechanism could be simply
implemented as a timestamp check too. *)
job : unit -> 'a state Lwt.t;
(* Job associated to the consensus heuristic *)
mutable result : 'a Lwt.t;
(* Promise which is fulfilled when a consensus is found *)
restart_delay : Ptime.Span.t;
(* Restart delay when the consensus heuristic did not reach a consensus *)
mutable all_consensus_hooks : ('a -> unit) list;
(* Hooks to be executed on all the consensus values found by the
heuristic. *)
mutable next_consensus_hooks : ('a -> unit) list;
(* Hooks to be executed for the next consensus value found by the
heuristic. *)
}
let create ~expire_time ~job ~restart_delay =
{
expire_time;
state = Need_more_candidates (* only for initialisation *);
expired = Lwt.return_unit;
job;
result = Lwt.fail Lwt.Canceled (* only for initialisation *);
restart_delay;
all_consensus_hooks = [];
next_consensus_hooks = [];
}
let rec loop t () =
let open Lwt_syntax in
(* If we cancel the worker, we cancel [t.result]. This triggers
the cancellation of [loop] if t.result was not fulfilled. If
[t.job] is not a cancelable promise, the cancellation will stop
here and consequently, if [t.job] returns
[Need_more_candidates] for example, we will rerun the
[loop]. To provent this, we wrap [t.job] as a cancelable
promise. *)
let* v = Lwt.wrap_in_cancelable (t.job ()) in
match v with
| Need_more_candidates | No_consensus _ ->
let* () = Systime_os.sleep t.restart_delay in
loop t ()
| Consensus data ->
t.expired <- Systime_os.sleep t.expire_time ;
(* We call [List.rev] to ensure hooks are called in the same
order they were registered. *)
let one_shot_hooks = List.rev t.next_consensus_hooks in
t.next_consensus_hooks <- [] ;
let forever_hooks = List.rev t.all_consensus_hooks in
List.iter (fun hook -> hook data) one_shot_hooks ;
List.iter (fun hook -> hook data) forever_hooks ;
Lwt.return data
let wait t =
(* [t]'s job is ongoing if its [result] promise is pending
([Lwt.Sleep]). [t]'s result is expired if its [expired]
promise is resolved ([Lwt.Return]). We start/restart the job if
the current [result] has expired meaning the [result] promise
is not pending. *)
if Lwt.state t.result <> Lwt.Sleep && Lwt.state t.expired = Lwt.Return ()
then t.result <- loop t () ;
Lwt.protected t.result
let on_next_consensus t hook =
match Lwt.state t.result with
| Lwt.Return data -> hook data
| _ -> t.next_consensus_hooks <- hook :: t.next_consensus_hooks
let on_all_consensus t hook =
t.all_consensus_hooks <- hook :: t.all_consensus_hooks ;
match Lwt.state t.result with Lwt.Return data -> hook data | _ -> ()
let cancel t =
(* We cancel the promises handled by the worker. We also ensure
that these two promises are resolved with a [Canceled]
state. Finally, we remove all the hooks to avoid any memory
leaks. *)
Lwt.cancel t.expired ;
t.expired <- Lwt.fail Lwt.Canceled ;
Lwt.cancel t.result ;
t.result <- Lwt.fail Lwt.Canceled ;
t.all_consensus_hooks <- [] ;
t.next_consensus_hooks <- []
end
![swh spinner](/static/img/swh-spinner.gif)
Computing file changes ...