···

       
3
       
 
       
let src = Logs.Src.create "conpool" ~doc:"Connection pooling library"

     

       
4
       
 
       
module Log = (val Logs.src_log src : Logs.LOG)

     

       
5
       
 
       


     

       
6
       
-
       
(** {1 Public Module Types} *)

     

       
7
       
-
       


     

       
8
       
 
       
module Endpoint = struct

     

       
9
       
 
       
  type t = {

     

       
10
       
 
       
    host : string;

     
···

       
51
       
 
       
    endpoint : Endpoint.t;

     

       
52
       
 
       
  }

     

       
53
       
 
       


     

       
54
       
-
       
  let get_flow t = t.flow

     

       
55
       
 
       
  let endpoint t = t.endpoint

     

       
56
       
 
       
  let created_at t = t.created_at

     

       
57
       
 
       
  let last_used t = t.last_used

     
···

       
171
       
 
       
      t.errors

     

       
172
       
 
       
end

     

       
173
       
 
       


     

       
174
       
-
       
(** {1 Internal Types} *)

     

       
175
       
-
       


     

       
176
       
-
       
type endpoint_stats_mutable = {

     

       
177
       
 
       
  mutable active : int;

     

       
178
       
 
       
  mutable idle : int;

     

       
179
       
 
       
  mutable total_created : int;

     
···

       
184
       
 
       


     

       
185
       
 
       
type endpoint_pool = {

     

       
186
       
 
       
  pool : Connection.t Eio.Pool.t;

     

       
187
       
-
       
  stats : endpoint_stats_mutable;

     

       
188
       
 
       
  mutex : Eio.Mutex.t;

     

       
189
       
 
       
}

     

       
190
       
 
       


     
···

       
198
       
 
       
  endpoints_mutex : Eio.Mutex.t;

     

       
199
       
 
       
}

     

       
200
       
 
       


     

       
201
       
-
       
(** {1 Endpoint Hashing} *)

     

       
202
       
-
       


     

       
203
       
 
       
module EndpointTbl = Hashtbl.Make(struct

     

       
204
       
 
       
  type t = Endpoint.t

     

       
205
       
 
       
  let equal = Endpoint.equal

     

       
206
       
 
       
  let hash = Endpoint.hash

     

       
207
       
 
       
end)

     

       
208
       
 
       


     

       
209
       
-
       
(** {1 Helper Functions} *)

     

       
210
       
-
       


     

       
211
       
 
       
let get_time pool =

     

       
212
       
 
       
  Eio.Time.now pool.clock

     

       
213
       
 
       


     

       
214
       
-
       
let create_mutable_stats () = {

     

       
215
       
 
       
  active = 0;

     

       
216
       
 
       
  idle = 0;

     

       
217
       
 
       
  total_created = 0;

     
···

       
220
       
 
       
  errors = 0;

     

       
221
       
 
       
}

     

       
222
       
 
       


     

       
223
       
-
       
let snapshot_stats (stats : endpoint_stats_mutable) : Stats.t = {

     

       
224
       
 
       
  active = stats.active;

     

       
225
       
 
       
  idle = stats.idle;

     

       
226
       
 
       
  total_created = stats.total_created;

     
···

       
232
       
 
       
(** {1 DNS Resolution} *)

     

       
233
       
 
       


     

       
234
       
 
       
let resolve_endpoint pool endpoint =

     

       
235
       
-
       
  (* Simple DNS resolution - returns socket address *)

     

       
236
       
 
       
  Log.debug (fun m -> m "Resolving %a..." Endpoint.pp endpoint);

     

       
237
       
 
       
  let addrs = Eio.Net.getaddrinfo_stream pool.net (Endpoint.host endpoint) ~service:(string_of_int (Endpoint.port endpoint)) in

     

       
238
       
 
       
  Log.debug (fun m -> m "Got address list for %a" Endpoint.pp endpoint);

     
···

       
274
       
 
       


     

       
275
       
 
       
    Log.debug (fun m -> m "TCP connection established to %a" Endpoint.pp endpoint);

     

       
276
       
 
       


     

       
277
       
-
       
    let flow : [`Close | `Flow | `R | `Shutdown | `W] Eio.Resource.t = match pool.tls with

     

       
278
       
 
       
      | None -> (socket :> [`Close | `Flow | `R | `Shutdown | `W] Eio.Resource.t)

     

       
279
       
 
       
      | Some tls_cfg ->

     

       
280
       
 
       
          Log.debug (fun m -> m "Initiating TLS handshake with %a" Endpoint.pp endpoint);

     
···

       
352
       
 
       
  else if (match pool.config.health_check with

     

       
353
       
 
       
           | Some check ->

     

       
354
       
 
       
               (try

     

       
355
       
-
       
                 let healthy = check (Connection.get_flow conn) in

     

       
356
       
 
       
                 if not healthy then

     

       
357
       
 
       
                   Log.debug (fun m -> m "Connection to %a failed custom health check"

     

       
358
       
 
       
                     Endpoint.pp (Connection.endpoint conn));

     
···

       
391
       
 
       


     

       
392
       
 
       
  Eio.Cancel.protect (fun () ->

     

       
393
       
 
       
    try

     

       
394
       
-
       
      Eio.Flow.close (Connection.get_flow conn)

     

       
395
       
 
       
    with _ -> ()

     

       
396
       
 
       
  );

     

       
397
       
 
       


     
···

       
419
       
 
       
            ep_pool

     

       
420
       
 
       
        | None ->

     

       
421
       
 
       
              (* Create new endpoint pool *)

     

       
422
       
-
       
              let stats = create_mutable_stats () in

     

       
423
       
 
       
              let mutex = Eio.Mutex.create () in

     

       
424
       
 
       


     

       
425
       
 
       
              Log.info (fun m -> m "Creating new endpoint pool for %a (max_connections=%d)"

     
···

       
448
       
 
       
                    (* Run health check if configured *)

     

       
449
       
 
       
                    match pool.config.health_check with

     

       
450
       
 
       
                    | Some check ->

     

       
451
       
-
       
                        (try check (Connection.get_flow conn)

     

       
452
       
 
       
                         with _ -> false)

     

       
453
       
 
       
                    | None -> true

     

       
454
       
 
       
                  ) else begin

     

···

       
3
       
 
       
let src = Logs.Src.create "conpool" ~doc:"Connection pooling library"

     

       
4
       
 
       
module Log = (val Logs.src_log src : Logs.LOG)

     

       
5
       
 
       


     

       
0
       
 
       

     

       
0
       
 
       

     

       
6
       
 
       
module Endpoint = struct

     

       
7
       
 
       
  type t = {

     

       
8
       
 
       
    host : string;

     
···

       
49
       
 
       
    endpoint : Endpoint.t;

     

       
50
       
 
       
  }

     

       
51
       
 
       


     

       
52
       
+
       
  let flow t = t.flow

     

       
53
       
 
       
  let endpoint t = t.endpoint

     

       
54
       
 
       
  let created_at t = t.created_at

     

       
55
       
 
       
  let last_used t = t.last_used

     
···

       
169
       
 
       
      t.errors

     

       
170
       
 
       
end

     

       
171
       
 
       


     

       
172
       
+
       
type endp_stats = {

     

       
0
       
 
       

     

       
0
       
 
       

     

       
173
       
 
       
  mutable active : int;

     

       
174
       
 
       
  mutable idle : int;

     

       
175
       
 
       
  mutable total_created : int;

     
···

       
180
       
 
       


     

       
181
       
 
       
type endpoint_pool = {

     

       
182
       
 
       
  pool : Connection.t Eio.Pool.t;

     

       
183
       
+
       
  stats : endp_stats;

     

       
184
       
 
       
  mutex : Eio.Mutex.t;

     

       
185
       
 
       
}

     

       
186
       
 
       


     
···

       
194
       
 
       
  endpoints_mutex : Eio.Mutex.t;

     

       
195
       
 
       
}

     

       
196
       
 
       


     

       
0
       
 
       

     

       
0
       
 
       

     

       
197
       
 
       
module EndpointTbl = Hashtbl.Make(struct

     

       
198
       
 
       
  type t = Endpoint.t

     

       
199
       
 
       
  let equal = Endpoint.equal

     

       
200
       
 
       
  let hash = Endpoint.hash

     

       
201
       
 
       
end)

     

       
202
       
 
       


     

       
0
       
 
       

     

       
0
       
 
       

     

       
203
       
 
       
let get_time pool =

     

       
204
       
 
       
  Eio.Time.now pool.clock

     

       
205
       
 
       


     

       
206
       
+
       
let create_endp_stats () = {

     

       
207
       
 
       
  active = 0;

     

       
208
       
 
       
  idle = 0;

     

       
209
       
 
       
  total_created = 0;

     
···

       
212
       
 
       
  errors = 0;

     

       
213
       
 
       
}

     

       
214
       
 
       


     

       
215
       
+
       
let snapshot_stats (stats : endp_stats) : Stats.t = {

     

       
216
       
 
       
  active = stats.active;

     

       
217
       
 
       
  idle = stats.idle;

     

       
218
       
 
       
  total_created = stats.total_created;

     
···

       
224
       
 
       
(** {1 DNS Resolution} *)

     

       
225
       
 
       


     

       
226
       
 
       
let resolve_endpoint pool endpoint =

     

       
0
       
 
       

     

       
227
       
 
       
  Log.debug (fun m -> m "Resolving %a..." Endpoint.pp endpoint);

     

       
228
       
 
       
  let addrs = Eio.Net.getaddrinfo_stream pool.net (Endpoint.host endpoint) ~service:(string_of_int (Endpoint.port endpoint)) in

     

       
229
       
 
       
  Log.debug (fun m -> m "Got address list for %a" Endpoint.pp endpoint);

     
···

       
265
       
 
       


     

       
266
       
 
       
    Log.debug (fun m -> m "TCP connection established to %a" Endpoint.pp endpoint);

     

       
267
       
 
       


     

       
268
       
+
       
    let flow = match pool.tls with

     

       
269
       
 
       
      | None -> (socket :> [`Close | `Flow | `R | `Shutdown | `W] Eio.Resource.t)

     

       
270
       
 
       
      | Some tls_cfg ->

     

       
271
       
 
       
          Log.debug (fun m -> m "Initiating TLS handshake with %a" Endpoint.pp endpoint);

     
···

       
343
       
 
       
  else if (match pool.config.health_check with

     

       
344
       
 
       
           | Some check ->

     

       
345
       
 
       
               (try

     

       
346
       
+
       
                 let healthy = check (Connection.flow conn) in

     

       
347
       
 
       
                 if not healthy then

     

       
348
       
 
       
                   Log.debug (fun m -> m "Connection to %a failed custom health check"

     

       
349
       
 
       
                     Endpoint.pp (Connection.endpoint conn));

     
···

       
382
       
 
       


     

       
383
       
 
       
  Eio.Cancel.protect (fun () ->

     

       
384
       
 
       
    try

     

       
385
       
+
       
      Eio.Flow.close (Connection.flow conn)

     

       
386
       
 
       
    with _ -> ()

     

       
387
       
 
       
  );

     

       
388
       
 
       


     
···

       
410
       
 
       
            ep_pool

     

       
411
       
 
       
        | None ->

     

       
412
       
 
       
              (* Create new endpoint pool *)

     

       
413
       
+
       
              let stats = create_endp_stats () in

     

       
414
       
 
       
              let mutex = Eio.Mutex.create () in

     

       
415
       
 
       


     

       
416
       
 
       
              Log.info (fun m -> m "Creating new endpoint pool for %a (max_connections=%d)"

     
···

       
439
       
 
       
                    (* Run health check if configured *)

     

       
440
       
 
       
                    match pool.config.health_check with

     

       
441
       
 
       
                    | Some check ->

     

       
442
       
+
       
                        (try check (Connection.flow conn)

     

       
443
       
 
       
                         with _ -> false)

     

       
444
       
 
       
                    | None -> true

     

       
445
       
 
       
                  ) else begin

     

···

       
61
       
 
       
  type t

     

       
62
       
 
       
  (** Opaque connection handle with metadata *)

     

       
63
       
 
       


     

       
64
       
-
       
  val get_flow : t -> [ `Close | `Flow | `R | `Shutdown | `W] Eio.Resource.t

     

       
65
       
 
       
  (** Extract underlying Eio flow from connection *)

     

       
66
       
 
       


     

       
67
       
 
       
  val endpoint : t -> Endpoint.t

     

···

       
61
       
 
       
  type t

     

       
62
       
 
       
  (** Opaque connection handle with metadata *)

     

       
63
       
 
       


     

       
64
       
+
       
  val flow : t -> [ `Close | `Flow | `R | `Shutdown | `W] Eio.Resource.t

     

       
65
       
 
       
  (** Extract underlying Eio flow from connection *)

     

       
66
       
 
       


     

       
67
       
 
       
  val endpoint : t -> Endpoint.t