package async_kernel

  1. Overview
  2. Docs
package async_kernel
Legend:
Page
Library
Module
Module type
Parameter
Class
Class type
Source

Source file persistent_connection_kernel.ml

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
open! Core_kernel
open! Async_kernel
open! Async_kernel_require_explicit_time_source
include Persistent_connection_kernel_intf

module Make (Conn : T) = struct
  type address = Conn.Address.t [@@deriving sexp_of]
  type conn = Conn.t

  module Event = struct
    type t =
      | Attempting_to_connect
      | Obtained_address of address
      | Failed_to_connect of Error.t
      | Connected of (conn[@sexp.opaque])
      | Disconnected
    [@@deriving sexp_of]

    type event = t

    module Handler = struct
      type t =
        { server_name : string
        ; on_event : event -> unit Deferred.t
        }
    end

    let log_level = function
      | Attempting_to_connect | Connected _ | Disconnected | Obtained_address _ -> `Info
      | Failed_to_connect _ -> `Error
    ;;

    let handle t { Handler.server_name = _; on_event } = on_event t
  end

  type t =
    { get_address : unit -> address Or_error.t Deferred.t
    ; connect : address -> Conn.t Or_error.t Deferred.t
    ; retry_delay : unit -> unit Deferred.t
    ; mutable conn : [ `Ok of Conn.t | `Close_started ] Ivar.t
    ; mutable next_connect_result : Conn.t Or_error.t Ivar.t
    ; event_handler : Event.Handler.t
    ; close_started : unit Ivar.t
    ; close_finished : unit Ivar.t
    }
  [@@deriving fields]

  let handle_event t event = Event.handle event t.event_handler

  (* This function focuses in on the the error itself, discarding information about which
     monitor caught the error, if any.

     If we don't do this, we sometimes end up with noisy logs which report the same error
     again and again, differing only as to what monitor caught them. *)
  let same_error e1 e2 =
    let to_sexp e = Exn.sexp_of_t (Monitor.extract_exn (Error.to_exn e)) in
    Sexp.equal (to_sexp e1) (to_sexp e2)
  ;;

  let try_connecting_until_successful t =
    (* We take care not to spam logs with the same message over and over by comparing
       each log message the the previous one of the same type. *)
    let previous_address = ref None in
    let previous_error = ref None in
    let connect () =
      t.get_address ()
      >>= function
      | Error e -> return (Error e)
      | Ok addr ->
        let same_as_previous_address =
          match !previous_address with
          | None -> false
          | Some previous_address -> Conn.Address.equal addr previous_address
        in
        previous_address := Some addr;
        (if same_as_previous_address
         then Deferred.unit
         else handle_event t (Obtained_address addr))
        >>= fun () -> t.connect addr
    in
    let rec loop () =
      if Ivar.is_full t.close_started
      then return `Close_started
      else (
        let%bind connect_result = connect () in
        Ivar.fill t.next_connect_result connect_result;
        t.next_connect_result <- Ivar.create ();
        match connect_result with
        | Ok conn -> return (`Ok conn)
        | Error err ->
          let same_as_previous_error =
            match !previous_error with
            | None -> false
            | Some previous_err -> same_error err previous_err
          in
          previous_error := Some err;
          (if same_as_previous_error
           then Deferred.unit
           else handle_event t (Failed_to_connect err))
          >>= fun () ->
          Deferred.any [ t.retry_delay (); Ivar.read t.close_started ]
          >>= fun () -> loop ())
    in
    loop ()
  ;;

  let create
        ~server_name
        ?(on_event = fun _ -> Deferred.unit)
        ?retry_delay
        ?(random_state = Random.State.default)
        ?(time_source = Time_source.wall_clock ())
        ~connect
        get_address
    =
    let event_handler = { Event.Handler.server_name; on_event } in
    let retry_delay () =
      let default_retry_delay () =
        if am_running_test then Time_ns.Span.of_sec 0.1 else Time_ns.Span.of_sec 10.
      in
      let retry_delay = Option.value retry_delay ~default:default_retry_delay in
      let span = Time_ns.Span.to_sec (retry_delay ()) in
      let distance = Random.State.float random_state (span *. 0.3) in
      let wait =
        if Random.State.bool random_state then span +. distance else span -. distance
      in
      Time_source.after time_source (Time_ns.Span.of_sec wait)
    in
    let t =
      { event_handler
      ; get_address
      ; connect
      ; next_connect_result = Ivar.create ()
      ; retry_delay
      ; conn = Ivar.create ()
      ; close_started = Ivar.create ()
      ; close_finished = Ivar.create ()
      }
    in
    (* this loop finishes once [close t] has been called, in which case it makes sure to
       leave [t.conn] filled with [`Close_started]. *)
    don't_wait_for
    @@ Deferred.repeat_until_finished () (fun () ->
      handle_event t Attempting_to_connect
      >>= fun () ->
      let ready_to_retry_connecting = t.retry_delay () in
      try_connecting_until_successful t
      >>= fun maybe_conn ->
      Ivar.fill t.conn maybe_conn;
      match maybe_conn with
      | `Close_started -> return (`Finished ())
      | `Ok conn ->
        handle_event t (Connected conn)
        >>= fun () ->
        Conn.close_finished conn
        >>= fun () ->
        t.conn <- Ivar.create ();
        handle_event t Disconnected
        >>= fun () ->
        (* waits until [retry_delay ()] time has passed since the time just before we last
           tried to connect rather than the time we noticed being disconnected, so that if
           a long-lived connection dies, we will attempt to reconnect immediately. *)
        Deferred.choose
          [ Deferred.choice ready_to_retry_connecting (fun () -> `Repeat ())
          ; Deferred.choice (Ivar.read t.close_started) (fun () ->
              Ivar.fill t.conn `Close_started;
              `Finished ())
          ]);
    t
  ;;

  let connected t =
    (* Take care not to return a connection that is known to be closed at the time
       [connected] was called.  This could happen in client code that behaves like
       {[
         Persistent_connection.Rpc.connected t
         >>= fun c1 ->
         ...
           Rpc.Connection.close_finished c1
         (* at this point we are in a race with the same call inside
            persistent_client.ml *)
         >>= fun () ->
         Persistent_connection.Rpc.connected t
         (* depending on how the race turns out, we don't want to get a closed connection
            here *)
         >>= fun c2 ->
         ...
       ]}
       This doesn't remove the race condition, but it makes it less likely to happen.
    *)
    let rec loop () =
      let d = Ivar.read t.conn in
      match Deferred.peek d with
      | None ->
        d
        >>= (function
          | `Close_started -> Deferred.never ()
          | `Ok conn -> return conn)
      | Some `Close_started -> Deferred.never ()
      | Some (`Ok conn) ->
        if Conn.is_closed conn
        then
          (* give the reconnection loop a chance to overwrite the ivar *)
          Conn.close_finished conn >>= loop
        else return conn
    in
    loop ()
  ;;

  let current_connection t =
    match Deferred.peek (Ivar.read t.conn) with
    | None | Some `Close_started -> None
    | Some (`Ok conn) -> Some conn
  ;;

  let close_finished t = Ivar.read t.close_finished
  let is_closed t = Ivar.is_full t.close_started

  let close t =
    if Ivar.is_full t.close_started
    then
      (* Another call to close is already in progress.  Wait for it to finish. *)
      close_finished t
    else (
      Ivar.fill t.close_started ();
      Ivar.read t.conn
      >>= fun conn_opt ->
      (match conn_opt with
       | `Close_started -> Deferred.unit
       | `Ok conn -> Conn.close conn)
      >>| fun () -> Ivar.fill t.close_finished ())
  ;;

  let connected_or_failed_to_connect_connection_closed =
    Or_error.error_s [%message "Persistent connection closed"]
  ;;

  let connected_or_failed_to_connect t =
    if is_closed t
    then return connected_or_failed_to_connect_connection_closed
    else (
      match Deferred.peek (connected t) with
      | Some x -> return (Ok x)
      | None ->
        Deferred.choose
          [ choice (Ivar.read t.close_started) (fun () ->
              connected_or_failed_to_connect_connection_closed)
          ; choice (Ivar.read t.next_connect_result) Fn.id
          ])
  ;;
end
OCaml

Innovation. Community. Security.

GitHub Discord Twitter Peertube RSS
About
Changelog Releases Industrial Users Academic Users Why OCaml
Ecosystem
Packages Community Events OCaml Planet Jobs
Resources
Install OCaml Get Started Platform Tools Language Manual Standard Library API Books Exercises Papers OCaml Playground Logo
Policies
Carbon Footprint Governance Privacy Code of Conduct