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

Shell: Use some Error_monad in P2p

Browse Source
This commit is contained in:
Vincent Bernardoff 2016-11-28 22:40:49 +01:00 committed by Grégoire Henry
parent d41c05a066
commit 41d5bbe989
1 changed files with 46 additions and 36 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

82
src/node/net/p2p.ml
View File

@ -13,6 +13,13 @@ module LC = Lwt_condition
open Lwt.Infix open Lwt.Infix
open Logging.Net open Logging.Net
type error += Encoding_error
type error += Message_too_big
type error += Write_would_block
type error += Decipher_error
type error += Canceled
type error += Timeout
(* public types *) (* public types *)
type addr = Ipaddr.t type addr = Ipaddr.t
type port = int type port = int
@ -195,51 +202,46 @@ module Make (P: PARAMS) = struct
match uncrypt buf with match uncrypt buf with
| None -> | None ->
(* TODO track invalid message *) (* TODO track invalid message *)
Lwt.return Disconnect Error_monad.fail Decipher_error
| Some buf -> | Some buf ->
match Data_encoding.Binary.of_bytes msg_encoding buf with match Data_encoding.Binary.of_bytes msg_encoding buf with
| None -> | None ->
(* TODO track invalid message *) (* TODO track invalid message *)
Lwt.return Disconnect Error_monad.fail Encoding_error
| Some msg -> | Some msg ->
Lwt.return msg Error_monad.return (len, msg)
end end
(function (fun exn -> Lwt.return @@ Error_monad.error_exn exn)
| Unix.Unix_error _ | End_of_file -> Lwt.return Disconnect
| e -> Lwt.fail e)
(* send a message over a TCP socket *) (* send a message over a TCP socket *)
let send_msg ?crypt fd buf msg = let send_msg ?crypt fd buf msg =
Lwt.catch begin fun () -> Lwt.catch begin fun () ->
match Data_encoding.Binary.write msg_encoding msg buf hdrlen with match Data_encoding.Binary.write msg_encoding msg buf hdrlen with
| None -> Lwt.return_false | None -> Error_monad.fail Encoding_error
| Some len -> | Some len ->
match crypt with match crypt with
| None -> | None ->
if len > maxlen then if len > maxlen then Error_monad.fail Message_too_big
Lwt.return_false
else begin else begin
EndianBigstring.BigEndian.set_int16 buf 0 (len - hdrlen) ; EndianBigstring.BigEndian.set_int16 buf 0 (len - hdrlen) ;
(* TODO timeout write ??? *) (* TODO timeout write ??? *)
Lwt_utils.write_mbytes ~len fd buf >>= fun () -> Lwt_utils.write_mbytes ~len fd buf >>= fun () ->
Lwt.return_true Error_monad.return len
end end
| Some crypt -> | Some crypt ->
let encbuf = crypt (MBytes.sub buf hdrlen (len - hdrlen)) in let encbuf = crypt (MBytes.sub buf hdrlen (len - hdrlen)) in
let len = MBytes.length encbuf in let len = MBytes.length encbuf in
if len > maxlen then if len > maxlen then Error_monad.fail Message_too_big
Lwt.return_false
else begin else begin
let lenbuf = MBytes.create 2 in let lenbuf = MBytes.create 2 in
EndianBigstring.BigEndian.set_int16 lenbuf 0 len ; EndianBigstring.BigEndian.set_int16 lenbuf 0 len ;
Lwt_utils.write_mbytes fd lenbuf >>= fun () -> Lwt_utils.write_mbytes fd lenbuf >>= fun () ->
Lwt_utils.write_mbytes fd encbuf >>= fun () -> Lwt_utils.write_mbytes fd encbuf >>= fun () ->
Lwt.return_true Error_monad.return len
end end
end end
(function (fun exn -> Lwt.return @@ Error_monad.error_exn exn)
| Unix.Unix_error _ | End_of_file -> Lwt.return_false
| e -> Lwt.fail e)
(* The (internal) type of network events, those dispatched from peer (* The (internal) type of network events, those dispatched from peer
workers to the net and others internal to net workers. *) workers to the net and others internal to net workers. *)
@ -393,9 +395,11 @@ module Make (P: PARAMS) = struct
(* a non exception-based cancelation mechanism *) (* a non exception-based cancelation mechanism *)
let cancelation, cancel, on_cancel = Lwt_utils.canceler () in let cancelation, cancel, on_cancel = Lwt_utils.canceler () in
(* a cancelable encrypted reception *) (* a cancelable encrypted reception *)
let recv ~uncrypt buf = let recv ?uncrypt buf =
Lwt.pick [ recv_msg ~uncrypt socket buf ; Lwt.pick [ recv_msg ?uncrypt socket buf ;
(cancelation () >>= fun () -> Lwt.return Disconnect) ] in (cancelation () >>= fun () -> Error_monad.fail Canceled) ]
>>=? fun (_size, message) ->
return message in
(* First step: send and receive credentials, makes no difference (* First step: send and receive credentials, makes no difference
whether we're trying to connect to a peer or checking an incoming whether we're trying to connect to a peer or checking an incoming
connection, both parties must first present themselves. *) connection, both parties must first present themselves. *)
@ -410,10 +414,14 @@ module Make (P: PARAMS) = struct
versions = P.supported_versions }) >>= fun _ -> versions = P.supported_versions }) >>= fun _ ->
Lwt.pick Lwt.pick
[ ( LU.sleep limits.peer_answer_timeout >>= fun () -> [ ( LU.sleep limits.peer_answer_timeout >>= fun () ->
Lwt.return Disconnect ) ; Error_monad.fail Timeout ) ;
recv_msg socket buf ] >>= function recv buf ] >>= function
| Connect { gid; port = listening_port; versions ; public_key ; | Error err ->
proof_of_work ; message_nonce } -> debug "(%a) error receiving from %a:%d: %a"
pp_gid my_gid Ipaddr.pp_hum addr port Error_monad.pp_print_error err ;
cancel ()
| Ok (Connect { gid; port = listening_port; versions ;
public_key ; proof_of_work ; message_nonce }) ->
debug "(%a) connection requested from %a @@ %a:%d" debug "(%a) connection requested from %a @@ %a:%d"
pp_gid my_gid pp_gid gid Ipaddr.pp_hum addr port ; pp_gid my_gid pp_gid gid Ipaddr.pp_hum addr port ;
let work_proved = let work_proved =
@ -451,18 +459,18 @@ module Make (P: PARAMS) = struct
buf local_nonce version gid buf local_nonce version gid
public_key message_nonce listening_port public_key message_nonce listening_port
end end
| Advertise peers -> | Ok (Advertise peers) ->
(* alternatively, one can refuse a connection but reply with (* alternatively, one can refuse a connection but reply with
some peers, so we accept this info *) some peers, so we accept this info *)
debug "(%a) new peers received from %a:%d" debug "(%a) new peers received from %a:%d"
pp_gid my_gid Ipaddr.pp_hum addr port ; pp_gid my_gid Ipaddr.pp_hum addr port ;
push (Peers peers) ; push (Peers peers) ;
cancel () cancel ()
| Disconnect -> | Ok Disconnect ->
debug "(%a) connection rejected (closed by peer or timeout) from %a:%d" debug "(%a) connection rejected (closed by peer or timeout) from %a:%d"
pp_gid my_gid Ipaddr.pp_hum addr port ; pp_gid my_gid Ipaddr.pp_hum addr port ;
cancel () cancel ()
| _ -> | Ok _ ->
debug "(%a) connection rejected (bad connection request) from %a:%d" debug "(%a) connection rejected (bad connection request) from %a:%d"
pp_gid my_gid Ipaddr.pp_hum addr port ; pp_gid my_gid Ipaddr.pp_hum addr port ;
cancel () cancel ()
@ -496,17 +504,19 @@ module Make (P: PARAMS) = struct
| Some _ as res -> res in | Some _ as res -> res in
(* The message reception loop. *) (* The message reception loop. *)
let rec receiver () = let rec receiver () =
recv ~uncrypt buf >>= fun message -> recv ~uncrypt buf >>= function
last := Unix.gettimeofday () ; | Error err ->
match message with debug "(%a) error receiving: %a"
| Connect _ pp_gid my_gid Error_monad.pp_print_error err ;
| Disconnect -> cancel ()
| Ok Connect _
| Ok Disconnect ->
debug "(%a) disconnected (by peer) %a @@ %a:%d" debug "(%a) disconnected (by peer) %a @@ %a:%d"
pp_gid my_gid pp_gid gid Ipaddr.pp_hum addr port ; pp_gid my_gid pp_gid gid Ipaddr.pp_hum addr port ;
cancel () cancel ()
| Bootstrap -> push (Bootstrap peer) ; receiver () | Ok Bootstrap -> push (Bootstrap peer) ; receiver ()
| Advertise peers -> push (Peers peers) ; receiver () | Ok Advertise peers -> push (Peers peers) ; receiver ()
| Message msg -> push (Recv (peer, msg)) ; receiver () | Ok Message msg -> push (Recv (peer, msg)) ; receiver ()
in in
(* Events for the main worker *) (* Events for the main worker *)
push (Connected peer) ; push (Connected peer) ;
@ -1176,6 +1186,7 @@ module Make (P: PARAMS) = struct
let discovery_answerer = let discovery_answerer =
let buf = MBytes.create 0x100_000 in let buf = MBytes.create 0x100_000 in
match config.discovery_port with match config.discovery_port with
| None -> Lwt.return_unit
| Some disco_port -> | Some disco_port ->
let answerer () = let answerer () =
discovery_answerer discovery_answerer
@ -1198,8 +1209,7 @@ module Make (P: PARAMS) = struct
else LU.close socket in else LU.close socket in
Lwt_utils.worker Lwt_utils.worker
(Format.asprintf "(%a) discovery answerer" pp_gid my_gid) (Format.asprintf "(%a) discovery answerer" pp_gid my_gid)
answerer cancel answerer cancel in
| _ -> Lwt.return_unit in
let discovery_sender = let discovery_sender =
match config.incoming_port, config.discovery_port with match config.incoming_port, config.discovery_port with
| Some inco_port, Some disco_port -> | Some inco_port, Some disco_port ->