Distributed_db: fix concurency issue
Some `events` might be lost in case of the timeout happens before concomitantly. We avoid this by storing the `events` promise into the worker state.
This commit is contained in:
parent
5c03d92457
commit
79ae54625d
@ -281,6 +281,7 @@ end = struct
|
|||||||
pending: status Table.t ;
|
pending: status Table.t ;
|
||||||
cancelation: unit -> unit Lwt.t ;
|
cancelation: unit -> unit Lwt.t ;
|
||||||
wait_events: unit -> event list Lwt.t ;
|
wait_events: unit -> event list Lwt.t ;
|
||||||
|
mutable events: event list Lwt.t ;
|
||||||
}
|
}
|
||||||
|
|
||||||
and status = {
|
and status = {
|
||||||
@ -329,18 +330,18 @@ end = struct
|
|||||||
(* TODO *)
|
(* TODO *)
|
||||||
Lwt.return_unit
|
Lwt.return_unit
|
||||||
|
|
||||||
let worker_loop state =
|
let rec worker_loop state =
|
||||||
let process = process_event state in
|
let shutdown = state.cancelation ()
|
||||||
let rec loop () =
|
and timeout = compute_timeout state in
|
||||||
let shutdown = state.cancelation () >|= fun () -> `Shutdown
|
Lwt.choose [ (state.events >|= fun _ -> ()) ; timeout ; shutdown ] >>= fun () ->
|
||||||
and timeout = compute_timeout state >|= fun () -> `Timeout
|
if Lwt.state shutdown <> Lwt.Sleep then
|
||||||
and events = state.wait_events () >|= fun events -> `Events events in
|
Lwt.return_unit
|
||||||
Lwt.pick [ timeout ; events ; shutdown ] >>= function
|
else if Lwt.state state.events <> Lwt.Sleep then
|
||||||
| `Shutdown -> Lwt.return_unit
|
state.events >>= fun events ->
|
||||||
| `Events events ->
|
state.events <- state.wait_events () ;
|
||||||
Lwt_list.iter_s process events >>= fun () ->
|
Lwt_list.iter_s (process_event state) events >>= fun () ->
|
||||||
loop ()
|
worker_loop state
|
||||||
| `Timeout ->
|
else
|
||||||
let now = Unix.gettimeofday () in
|
let now = Unix.gettimeofday () in
|
||||||
let active_peers = Request.active state.param in
|
let active_peers = Request.active state.param in
|
||||||
let requests =
|
let requests =
|
||||||
@ -349,16 +350,18 @@ end = struct
|
|||||||
if next_request > now +. 0.2 then
|
if next_request > now +. 0.2 then
|
||||||
acc
|
acc
|
||||||
else
|
else
|
||||||
let still_peers = P2p.Peer_id.Set.inter peers active_peers in
|
let remaining_peers =
|
||||||
if P2p.Peer_id.Set.is_empty still_peers &&
|
P2p.Peer_id.Set.inter peers active_peers in
|
||||||
|
if P2p.Peer_id.Set.is_empty remaining_peers &&
|
||||||
not (P2p.Peer_id.Set.is_empty peers) then
|
not (P2p.Peer_id.Set.is_empty peers) then
|
||||||
( Table.remove state.pending key ; acc )
|
( Table.remove state.pending key ; acc )
|
||||||
else
|
else
|
||||||
let requested_peers =
|
let requested_peers =
|
||||||
if P2p.Peer_id.Set.is_empty peers
|
if P2p.Peer_id.Set.is_empty remaining_peers
|
||||||
then active_peers
|
then active_peers
|
||||||
else peers in
|
else remaining_peers
|
||||||
let next = { peers = still_peers ;
|
in
|
||||||
|
let next = { peers = remaining_peers ;
|
||||||
next_request = now +. delay ;
|
next_request = now +. delay ;
|
||||||
delay = delay *. 1.2 } in
|
delay = delay *. 1.2 } in
|
||||||
Table.replace state.pending key next ;
|
Table.replace state.pending key next ;
|
||||||
@ -372,19 +375,17 @@ end = struct
|
|||||||
acc)
|
acc)
|
||||||
state.pending P2p_types.Peer_id.Map.empty in
|
state.pending P2p_types.Peer_id.Map.empty in
|
||||||
P2p_types.Peer_id.Map.iter (Request.send state.param) requests ;
|
P2p_types.Peer_id.Map.iter (Request.send state.param) requests ;
|
||||||
loop ()
|
worker_loop state
|
||||||
in
|
|
||||||
loop
|
|
||||||
|
|
||||||
let create param =
|
let create param =
|
||||||
let cancelation, cancel_worker, _ = Lwt_utils.canceler () in
|
let cancelation, cancel_worker, _ = Lwt_utils.canceler () in
|
||||||
let push_to_worker, wait_events = Lwt_utils.queue () in
|
let push_to_worker, wait_events = Lwt_utils.queue () in
|
||||||
let pending = Table.create 17 in
|
let pending = Table.create 17 in
|
||||||
let worker_state =
|
let worker_state =
|
||||||
{ cancelation ; wait_events ; pending ; param } in
|
{ cancelation ; wait_events ; pending ; param ; events = wait_events () } in
|
||||||
let worker =
|
let worker =
|
||||||
Lwt_utils.worker "db_request_scheduler"
|
Lwt_utils.worker "db_request_scheduler"
|
||||||
~run:(worker_loop worker_state)
|
~run:(fun () -> worker_loop worker_state)
|
||||||
~cancel:cancel_worker in
|
~cancel:cancel_worker in
|
||||||
{ cancel_worker ; push_to_worker ; worker }
|
{ cancel_worker ; push_to_worker ; worker }
|
||||||
|
|
||||||
|
Loading…
Reference in New Issue
Block a user