balsoft/ligo
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Mempool_peer_worker: minor improvements

Browse Source
This commit is contained in:
Raphaël Proust 2018-11-08 12:08:11 +08:00 committed by MBourgoin
parent 460262130e
commit 1431bcd6c8
No known key found for this signature in database
GPG Key ID: 4B3F7008ABB5B2D0
2 changed files with 14 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

30
src/lib_shell/mempool_peer_worker.ml
View File

@ -53,8 +53,6 @@ module type T = sig
* from a previous protocol, etc. *) * from a previous protocol, etc. *)
val bypass_peer_workers: Mempool_worker.t -> input -> output Lwt.t val bypass_peer_workers: Mempool_worker.t -> input -> output Lwt.t
val rpc_directory : t RPC_directory.t
end end
module Make (Mempool_worker: Mempool_worker.T) : T with module Proto = Mempool_worker.Proto = struct module Make (Mempool_worker: Mempool_worker.T) : T with module Proto = Mempool_worker.Proto = struct
@ -73,6 +71,8 @@ module Make (Mempool_worker: Mempool_worker.T) : T with module Proto = Mempool_w
| Mempool_result of Mempool_worker.result | Mempool_result of Mempool_worker.result
type output = result Operation_hash.Map.t type output = result Operation_hash.Map.t
let pp_input = Format.pp_print_list Operation_hash.pp
module Log = Tezos_stdlib.Logging.Make(struct module Log = Tezos_stdlib.Logging.Make(struct
let name = "node.mempool.peer_worker" let name = "node.mempool.peer_worker"
end) end)
@ -140,6 +140,12 @@ module Make (Mempool_worker: Mempool_worker.T) : T with module Proto = Mempool_w
results = Operation_hash.Map.empty; results = Operation_hash.Map.empty;
} }
let cancel pipeline =
let cancel_snd (_, p) = Lwt.cancel p in
Queue.iter cancel_snd pipeline.downloading;
Queue.iter cancel_snd pipeline.parsing;
Queue.iter cancel_snd pipeline.applying
(* Exported interactions *) (* Exported interactions *)
@ -208,12 +214,7 @@ module Make (Mempool_worker: Mempool_worker.T) : T with module Proto = Mempool_w
loop () loop ()
in in
let work = loop () in let work = loop () in
Lwt.on_cancel work Lwt.on_cancel work (fun () -> cancel pipeline);
(fun () ->
let cancel_snd (_, p) = Lwt.cancel p in
Queue.iter cancel_snd pipeline.downloading;
Queue.iter cancel_snd pipeline.parsing;
Queue.iter cancel_snd pipeline.applying);
work work
end end
@ -280,7 +281,7 @@ module Make (Mempool_worker: Mempool_worker.T) : T with module Proto = Mempool_w
| Start input -> | Start input ->
Format.fprintf ppf Format.fprintf ppf
"@[<v 0>Starting: %a@]" "@[<v 0>Starting: %a@]"
(Format.pp_print_list Operation_hash.pp) pp_input
input input
| End_ok (view, _) -> | End_ok (view, _) ->
Format.fprintf ppf Format.fprintf ppf
@ -349,12 +350,9 @@ module Make (Mempool_worker: Mempool_worker.T) : T with module Proto = Mempool_w
Worker.push_request_and_wait t (Request.Batch (mempool_worker, os)) Worker.push_request_and_wait t (Request.Batch (mempool_worker, os))
let bypass_peer_workers mempool_worker input = let bypass_peer_workers mempool_worker input =
(* TODO: log, but not through the Worker's event system because no worker is available. *) Log.lwt_log_info "Bypassing workers to work on: %a" pp_input input >>= fun () ->
Work.work mempool_worker input Work.work mempool_worker input >>= fun output ->
Log.lwt_log_info "Finished work" >>= fun () ->
Lwt.return output
(* 5. Introspection *)
let rpc_directory = Pervasives.failwith "TODO"
end end

2
src/lib_shell/mempool_peer_worker.mli
View File

@ -82,8 +82,6 @@ module type T = sig
* calls to [bypass_peer_workers] is not sequentialised. *) * calls to [bypass_peer_workers] is not sequentialised. *)
val bypass_peer_workers: Mempool_worker.t -> input -> output Lwt.t val bypass_peer_workers: Mempool_worker.t -> input -> output Lwt.t
val rpc_directory : t RPC_directory.t
end end