···
1
+
(*---------------------------------------------------------------------------
2
+
Copyright (c) 2025 Anil Madhavapeddy <anil@recoil.org>. All rights reserved.
3
+
SPDX-License-Identifier: ISC
4
+
---------------------------------------------------------------------------*)
(** Conpool - Protocol-agnostic TCP/IP connection pooling library for Eio *)
let src = Logs.Src.create "conpool" ~doc:"Connection pooling library"
module Log = (val Logs.src_log src : Logs.LOG)
(* Re-export submodules *)
···
| Dns_resolution_failed of { hostname : string }
17
-
| Connection_failed of { endpoint : Endpoint.t; attempts : int; last_error : string }
23
+
| Connection_failed of {
24
+
endpoint : Endpoint.t;
26
+
last_error : string;
| Connection_timeout of { endpoint : Endpoint.t; timeout : float }
| Invalid_config of string
| Invalid_endpoint of string
···
| Dns_resolution_failed { hostname } ->
Fmt.pf ppf "DNS resolution failed for hostname: %s" hostname
| Connection_failed { endpoint; attempts; last_error } ->
28
-
Fmt.pf ppf "Failed to connect to %a after %d attempts: %s"
29
-
Endpoint.pp endpoint attempts last_error
38
+
Fmt.pf ppf "Failed to connect to %a after %d attempts: %s" Endpoint.pp
39
+
endpoint attempts last_error
| Connection_timeout { endpoint; timeout } ->
31
-
Fmt.pf ppf "Connection timeout to %a after %.2fs"
32
-
Endpoint.pp endpoint timeout
33
-
| Invalid_config msg ->
34
-
Fmt.pf ppf "Invalid configuration: %s" msg
35
-
| Invalid_endpoint msg ->
36
-
Fmt.pf ppf "Invalid endpoint: %s" msg
41
+
Fmt.pf ppf "Connection timeout to %a after %.2fs" Endpoint.pp endpoint
43
+
| Invalid_config msg -> Fmt.pf ppf "Invalid configuration: %s" msg
44
+
| Invalid_endpoint msg -> Fmt.pf ppf "Invalid endpoint: %s" msg
···
type t = T : ('clock Eio.Time.clock, 'net Eio.Net.t) internal -> t
65
-
module EndpointTbl = Hashtbl.Make(struct
73
+
module EndpointTbl = Hashtbl.Make (struct
let equal = Endpoint.equal
71
-
let get_time (pool : ('clock, 'net) internal) =
72
-
Eio.Time.now pool.clock
80
+
let get_time (pool : ('clock, 'net) internal) = Eio.Time.now pool.clock
74
-
let create_endp_stats () = {
82
+
let create_endp_stats () =
let snapshot_stats (stats : endp_stats) : Stats.t =
85
-
~active:stats.active
87
-
~total_created:stats.total_created
88
-
~total_reused:stats.total_reused
89
-
~total_closed:stats.total_closed
90
-
~errors:stats.errors
93
+
Stats.make ~active:stats.active ~idle:stats.idle
94
+
~total_created:stats.total_created ~total_reused:stats.total_reused
95
+
~total_closed:stats.total_closed ~errors:stats.errors
(** {1 DNS Resolution} *)
let resolve_endpoint (pool : ('clock, 'net) internal) endpoint =
Log.debug (fun m -> m "Resolving %a..." Endpoint.pp endpoint);
96
-
let addrs = Eio.Net.getaddrinfo_stream pool.net (Endpoint.host endpoint) ~service:(string_of_int (Endpoint.port endpoint)) in
102
+
Eio.Net.getaddrinfo_stream pool.net (Endpoint.host endpoint)
103
+
~service:(string_of_int (Endpoint.port endpoint))
Log.debug (fun m -> m "Got address list for %a" Endpoint.pp endpoint);
100
-
Log.debug (fun m -> m "Resolved %a to %a"
101
-
Endpoint.pp endpoint Eio.Net.Sockaddr.pp addr);
108
+
Log.debug (fun m ->
109
+
m "Resolved %a to %a" Endpoint.pp endpoint Eio.Net.Sockaddr.pp addr);
104
-
Log.err (fun m -> m "Failed to resolve hostname: %s" (Endpoint.host endpoint));
105
-
raise (Pool_error (Dns_resolution_failed { hostname = Endpoint.host endpoint }))
113
+
m "Failed to resolve hostname: %s" (Endpoint.host endpoint));
116
+
(Dns_resolution_failed { hostname = Endpoint.host endpoint }))
(** {1 Connection Creation with Retry} *)
109
-
let rec create_connection_with_retry (pool : ('clock, 'net) internal) endpoint attempt last_error =
120
+
let rec create_connection_with_retry (pool : ('clock, 'net) internal) endpoint
121
+
attempt last_error =
let retry_count = Config.connect_retry_count pool.config in
if attempt > retry_count then begin
112
-
Log.err (fun m -> m "Failed to connect to %a after %d attempts"
113
-
Endpoint.pp endpoint retry_count);
114
-
raise (Pool_error (Connection_failed { endpoint; attempts = retry_count; last_error }))
125
+
m "Failed to connect to %a after %d attempts" Endpoint.pp endpoint
129
+
(Connection_failed { endpoint; attempts = retry_count; last_error }))
117
-
Log.debug (fun m -> m "Connecting to %a (attempt %d/%d)"
118
-
Endpoint.pp endpoint attempt retry_count);
132
+
Log.debug (fun m ->
133
+
m "Connecting to %a (attempt %d/%d)" Endpoint.pp endpoint attempt
let addr = resolve_endpoint pool endpoint in
···
match Config.connect_timeout pool.config with
128
-
Eio.Time.with_timeout_exn pool.clock timeout
129
-
(fun () -> Eio.Net.connect ~sw:pool.sw pool.net addr)
131
-
Eio.Net.connect ~sw:pool.sw pool.net addr
144
+
Eio.Time.with_timeout_exn pool.clock timeout (fun () ->
145
+
Eio.Net.connect ~sw:pool.sw pool.net addr)
146
+
| None -> Eio.Net.connect ~sw:pool.sw pool.net addr
134
-
Log.debug (fun m -> m "TCP connection established to %a" Endpoint.pp endpoint);
149
+
Log.debug (fun m ->
150
+
m "TCP connection established to %a" Endpoint.pp endpoint);
136
-
let flow = match pool.tls with
137
-
| None -> (socket :> [`Close | `Flow | `R | `Shutdown | `W] Eio.Resource.t)
153
+
match pool.tls with
155
+
(socket :> [ `Close | `Flow | `R | `Shutdown | `W ] Eio.Resource.t)
139
-
Log.debug (fun m -> m "Initiating TLS handshake with %a" Endpoint.pp endpoint);
140
-
let host = match Tls_config.servername tls_cfg with
157
+
Log.debug (fun m ->
158
+
m "Initiating TLS handshake with %a" Endpoint.pp endpoint);
160
+
match Tls_config.servername tls_cfg with
| Some name -> Domain_name.(host_exn (of_string_exn name))
142
-
| None -> Domain_name.(host_exn (of_string_exn (Endpoint.host endpoint)))
163
+
Domain_name.(host_exn (of_string_exn (Endpoint.host endpoint)))
144
-
let tls_flow = Tls_eio.client_of_flow ~host (Tls_config.config tls_cfg) socket in
145
-
Log.info (fun m -> m "TLS connection established to %a" Endpoint.pp endpoint);
146
-
(tls_flow :> [`Close | `Flow | `R | `Shutdown | `W] Eio.Resource.t)
166
+
Tls_eio.client_of_flow ~host (Tls_config.config tls_cfg) socket
169
+
m "TLS connection established to %a" Endpoint.pp endpoint);
170
+
(tls_flow :> [ `Close | `Flow | `R | `Shutdown | `W ] Eio.Resource.t)
let now = get_time pool in
···
mutex = Eio.Mutex.create ();
| Eio.Time.Timeout as e ->
162
-
Log.warn (fun m -> m "Connection timeout to %a (attempt %d)" Endpoint.pp endpoint attempt);
186
+
m "Connection timeout to %a (attempt %d)" Endpoint.pp endpoint attempt);
let error_msg = Printexc.to_string e in
if attempt >= Config.connect_retry_count pool.config then
(* Last attempt - convert to our error type *)
···
raise (Pool_error (Connection_timeout { endpoint; timeout }))
170
-
raise (Pool_error (Connection_failed { endpoint; attempts = attempt; last_error = error_msg }))
197
+
{ endpoint; attempts = attempt; last_error = error_msg }))
(* Retry with exponential backoff *)
173
-
let delay = Config.connect_retry_delay pool.config *. (2.0 ** float_of_int (attempt - 1)) in
201
+
Config.connect_retry_delay pool.config
202
+
*. (2.0 ** float_of_int (attempt - 1))
Eio.Time.sleep pool.clock delay;
create_connection_with_retry pool endpoint (attempt + 1) error_msg
(* Other errors - retry with backoff *)
let error_msg = Printexc.to_string e in
180
-
Log.warn (fun m -> m "Connection attempt %d to %a failed: %s"
181
-
attempt Endpoint.pp endpoint error_msg);
211
+
m "Connection attempt %d to %a failed: %s" attempt Endpoint.pp
212
+
endpoint error_msg);
if attempt < Config.connect_retry_count pool.config then (
183
-
let delay = Config.connect_retry_delay pool.config *. (2.0 ** float_of_int (attempt - 1)) in
215
+
Config.connect_retry_delay pool.config
216
+
*. (2.0 ** float_of_int (attempt - 1))
Eio.Time.sleep pool.clock delay;
185
-
create_connection_with_retry pool endpoint (attempt + 1) error_msg
187
-
raise (Pool_error (Connection_failed { endpoint; attempts = attempt; last_error = error_msg }))
219
+
create_connection_with_retry pool endpoint (attempt + 1) error_msg)
224
+
{ endpoint; attempts = attempt; last_error = error_msg }))
let create_connection (pool : ('clock, 'net) internal) endpoint =
create_connection_with_retry pool endpoint 1 "No attempts made"
···
let age = now -. Connection.created_at conn in
let max_lifetime = Config.max_connection_lifetime pool.config in
if age > max_lifetime then begin
201
-
Log.debug (fun m -> m "Connection to %a unhealthy: exceeded max lifetime (%.2fs > %.2fs)"
202
-
Endpoint.pp (Connection.endpoint conn) age max_lifetime);
238
+
Log.debug (fun m ->
239
+
m "Connection to %a unhealthy: exceeded max lifetime (%.2fs > %.2fs)"
240
+
Endpoint.pp (Connection.endpoint conn) age max_lifetime);
let max_idle = Config.max_idle_time pool.config in
209
-
if (now -. Connection.last_used conn) > max_idle then begin
246
+
if now -. Connection.last_used conn > max_idle then begin
let idle_time = now -. Connection.last_used conn in
211
-
Log.debug (fun m -> m "Connection to %a unhealthy: exceeded max idle time (%.2fs > %.2fs)"
212
-
Endpoint.pp (Connection.endpoint conn) idle_time max_idle);
216
-
(* Check use count *)
217
-
else if (match Config.max_connection_uses pool.config with
218
-
| Some max -> Connection.use_count conn >= max
219
-
| None -> false) then begin
220
-
Log.debug (fun m -> m "Connection to %a unhealthy: exceeded max use count (%d)"
221
-
Endpoint.pp (Connection.endpoint conn) (Connection.use_count conn));
248
+
Log.debug (fun m ->
249
+
m "Connection to %a unhealthy: exceeded max idle time (%.2fs > %.2fs)"
250
+
Endpoint.pp (Connection.endpoint conn) idle_time max_idle);
225
-
(* Optional: custom health check *)
226
-
else if (match Config.health_check pool.config with
229
-
let healthy = check (Connection.flow conn) in
230
-
if not healthy then
231
-
Log.debug (fun m -> m "Connection to %a failed custom health check"
232
-
Endpoint.pp (Connection.endpoint conn));
235
-
Log.debug (fun m -> m "Connection to %a health check raised exception: %s"
236
-
Endpoint.pp (Connection.endpoint conn) (Printexc.to_string e));
237
-
true) (* Exception in health check = unhealthy *)
238
-
| None -> false) then
252
+
end (* Check use count *)
254
+
match Config.max_connection_uses pool.config with
255
+
| Some max -> Connection.use_count conn >= max
258
+
Log.debug (fun m ->
259
+
m "Connection to %a unhealthy: exceeded max use count (%d)"
260
+
Endpoint.pp (Connection.endpoint conn)
261
+
(Connection.use_count conn));
241
-
(* Optional: check if socket still connected *)
263
+
end (* Optional: custom health check *)
265
+
match Config.health_check pool.config with
268
+
let healthy = check (Connection.flow conn) in
269
+
if not healthy then
270
+
Log.debug (fun m ->
271
+
m "Connection to %a failed custom health check" Endpoint.pp
272
+
(Connection.endpoint conn));
275
+
Log.debug (fun m ->
276
+
m "Connection to %a health check raised exception: %s"
277
+
Endpoint.pp (Connection.endpoint conn) (Printexc.to_string e));
278
+
true (* Exception in health check = unhealthy *))
280
+
then false (* Optional: check if socket still connected *)
else if check_readable then
(* TODO avsm: a sockopt for this? *)
250
-
Log.debug (fun m -> m "Connection to %a is healthy (age=%.2fs, idle=%.2fs, uses=%d)"
251
-
Endpoint.pp (Connection.endpoint conn)
253
-
(now -. Connection.last_used conn)
254
-
(Connection.use_count conn));
287
+
Log.debug (fun m ->
288
+
m "Connection to %a is healthy (age=%.2fs, idle=%.2fs, uses=%d)"
289
+
Endpoint.pp (Connection.endpoint conn) age
290
+
(now -. Connection.last_used conn)
291
+
(Connection.use_count conn));
···
(** {1 Internal Pool Operations} *)
let close_internal (pool : ('clock, 'net) internal) conn =
262
-
Log.debug (fun m -> m "Closing connection to %a (age=%.2fs, uses=%d)"
263
-
Endpoint.pp (Connection.endpoint conn)
264
-
(get_time pool -. Connection.created_at conn)
265
-
(Connection.use_count conn));
299
+
Log.debug (fun m ->
300
+
m "Closing connection to %a (age=%.2fs, uses=%d)" Endpoint.pp
301
+
(Connection.endpoint conn)
302
+
(get_time pool -. Connection.created_at conn)
303
+
(Connection.use_count conn));
Eio.Cancel.protect (fun () ->
269
-
Eio.Flow.close (Connection.flow conn)
306
+
try Eio.Flow.close (Connection.flow conn) with _ -> ());
(* Call hook if configured *)
274
-
Option.iter (fun f -> f (Connection.endpoint conn)) (Config.on_connection_closed pool.config)
310
+
(fun f -> f (Connection.endpoint conn))
311
+
(Config.on_connection_closed pool.config)
let get_or_create_endpoint_pool (pool : ('clock, 'net) internal) endpoint =
277
-
Log.debug (fun m -> m "Getting or creating endpoint pool for %a" Endpoint.pp endpoint);
314
+
Log.debug (fun m ->
315
+
m "Getting or creating endpoint pool for %a" Endpoint.pp endpoint);
(* First try with read lock *)
280
-
match Eio.Mutex.use_ro pool.endpoints_mutex (fun () ->
281
-
Hashtbl.find_opt pool.endpoints endpoint
319
+
Eio.Mutex.use_ro pool.endpoints_mutex (fun () ->
320
+
Hashtbl.find_opt pool.endpoints endpoint)
284
-
Log.debug (fun m -> m "Found existing endpoint pool for %a" Endpoint.pp endpoint);
323
+
Log.debug (fun m ->
324
+
m "Found existing endpoint pool for %a" Endpoint.pp endpoint);
287
-
Log.debug (fun m -> m "No existing pool, need to create for %a" Endpoint.pp endpoint);
327
+
Log.debug (fun m ->
328
+
m "No existing pool, need to create for %a" Endpoint.pp endpoint);
(* Need to create - use write lock *)
Eio.Mutex.use_rw ~protect:true pool.endpoints_mutex (fun () ->
290
-
(* Check again in case another fiber created it *)
291
-
match Hashtbl.find_opt pool.endpoints endpoint with
293
-
Log.debug (fun m -> m "Another fiber created pool for %a" Endpoint.pp endpoint);
331
+
(* Check again in case another fiber created it *)
332
+
match Hashtbl.find_opt pool.endpoints endpoint with
334
+
Log.debug (fun m ->
335
+
m "Another fiber created pool for %a" Endpoint.pp endpoint);
(* Create new endpoint pool *)
let stats = create_endp_stats () in
let mutex = Eio.Mutex.create () in
300
-
Log.info (fun m -> m "Creating new endpoint pool for %a (max_connections=%d)"
301
-
Endpoint.pp endpoint (Config.max_connections_per_endpoint pool.config));
343
+
m "Creating new endpoint pool for %a (max_connections=%d)"
344
+
Endpoint.pp endpoint
345
+
(Config.max_connections_per_endpoint pool.config));
303
-
Log.debug (fun m -> m "About to create Eio.Pool for %a" Endpoint.pp endpoint);
347
+
Log.debug (fun m ->
348
+
m "About to create Eio.Pool for %a" Endpoint.pp endpoint);
305
-
let eio_pool = Eio.Pool.create
306
-
(Config.max_connections_per_endpoint pool.config)
307
-
~validate:(fun conn ->
308
-
Log.debug (fun m -> m "Validate called for connection to %a" Endpoint.pp endpoint);
309
-
(* Called before reusing from pool *)
310
-
let healthy = is_healthy pool ~check_readable:false conn in
352
+
(Config.max_connections_per_endpoint pool.config)
353
+
~validate:(fun conn ->
354
+
Log.debug (fun m ->
355
+
m "Validate called for connection to %a" Endpoint.pp
357
+
(* Called before reusing from pool *)
358
+
let healthy = is_healthy pool ~check_readable:false conn in
313
-
Log.debug (fun m -> m "Reusing connection to %a from pool" Endpoint.pp endpoint);
361
+
Log.debug (fun m ->
362
+
m "Reusing connection to %a from pool" Endpoint.pp
315
-
(* Update stats for reuse *)
316
-
Eio.Mutex.use_rw ~protect:true mutex (fun () ->
317
-
stats.total_reused <- stats.total_reused + 1
365
+
(* Update stats for reuse *)
366
+
Eio.Mutex.use_rw ~protect:true mutex (fun () ->
367
+
stats.total_reused <- stats.total_reused + 1);
320
-
(* Call hook if configured *)
321
-
Option.iter (fun f -> f endpoint) (Config.on_connection_reused pool.config);
369
+
(* Call hook if configured *)
371
+
(fun f -> f endpoint)
372
+
(Config.on_connection_reused pool.config);
323
-
(* Run health check if configured *)
324
-
match Config.health_check pool.config with
326
-
(try check (Connection.flow conn)
330
-
Log.debug (fun m -> m "Connection to %a failed validation, creating new one" Endpoint.pp endpoint);
334
-
~dispose:(fun conn ->
335
-
(* Called when removing from pool *)
336
-
Eio.Cancel.protect (fun () ->
337
-
close_internal pool conn;
374
+
(* Run health check if configured *)
375
+
match Config.health_check pool.config with
377
+
try check (Connection.flow conn) with _ -> false)
380
+
Log.debug (fun m ->
382
+
"Connection to %a failed validation, creating new \
384
+
Endpoint.pp endpoint);
387
+
~dispose:(fun conn ->
388
+
(* Called when removing from pool *)
389
+
Eio.Cancel.protect (fun () ->
390
+
close_internal pool conn;
340
-
Eio.Mutex.use_rw ~protect:true mutex (fun () ->
341
-
stats.total_closed <- stats.total_closed + 1
346
-
Log.debug (fun m -> m "Factory function called for %a" Endpoint.pp endpoint);
348
-
let conn = create_connection pool endpoint in
393
+
Eio.Mutex.use_rw ~protect:true mutex (fun () ->
394
+
stats.total_closed <- stats.total_closed + 1)))
396
+
Log.debug (fun m ->
397
+
m "Factory function called for %a" Endpoint.pp endpoint);
399
+
let conn = create_connection pool endpoint in
350
-
Log.debug (fun m -> m "Connection created successfully for %a" Endpoint.pp endpoint);
401
+
Log.debug (fun m ->
402
+
m "Connection created successfully for %a" Endpoint.pp
353
-
Eio.Mutex.use_rw ~protect:true mutex (fun () ->
354
-
stats.total_created <- stats.total_created + 1
406
+
Eio.Mutex.use_rw ~protect:true mutex (fun () ->
407
+
stats.total_created <- stats.total_created + 1);
357
-
(* Call hook if configured *)
358
-
Option.iter (fun f -> f endpoint) (Config.on_connection_created pool.config);
409
+
(* Call hook if configured *)
411
+
(fun f -> f endpoint)
412
+
(Config.on_connection_created pool.config);
362
-
Log.err (fun m -> m "Factory function failed for %a: %s"
363
-
Endpoint.pp endpoint (Printexc.to_string e));
364
-
(* Update error stats *)
365
-
Eio.Mutex.use_rw ~protect:true mutex (fun () ->
366
-
stats.errors <- stats.errors + 1
417
+
m "Factory function failed for %a: %s" Endpoint.pp
418
+
endpoint (Printexc.to_string e));
419
+
(* Update error stats *)
420
+
Eio.Mutex.use_rw ~protect:true mutex (fun () ->
421
+
stats.errors <- stats.errors + 1);
372
-
Log.debug (fun m -> m "Eio.Pool created successfully for %a" Endpoint.pp endpoint);
425
+
Log.debug (fun m ->
426
+
m "Eio.Pool created successfully for %a" Endpoint.pp endpoint);
428
+
let ep_pool = { pool = eio_pool; stats; mutex } in
Hashtbl.add pool.endpoints endpoint ep_pool;
381
-
Log.debug (fun m -> m "Endpoint pool added to hashtable for %a" Endpoint.pp endpoint);
431
+
Log.debug (fun m ->
432
+
m "Endpoint pool added to hashtable for %a" Endpoint.pp
(** {1 Public API - Pool Creation} *)
387
-
let create ~sw ~(net : 'net Eio.Net.t) ~(clock : 'clock Eio.Time.clock) ?tls ?(config = Config.default) () : t =
388
-
Log.info (fun m -> m "Creating new connection pool (max_per_endpoint=%d, max_idle=%.1fs, max_lifetime=%.1fs)"
389
-
(Config.max_connections_per_endpoint config)
390
-
(Config.max_idle_time config)
391
-
(Config.max_connection_lifetime config));
438
+
let create ~sw ~(net : 'net Eio.Net.t) ~(clock : 'clock Eio.Time.clock) ?tls
439
+
?(config = Config.default) () : t =
442
+
"Creating new connection pool (max_per_endpoint=%d, max_idle=%.1fs, \
443
+
max_lifetime=%.1fs)"
444
+
(Config.max_connections_per_endpoint config)
445
+
(Config.max_idle_time config)
446
+
(Config.max_connection_lifetime config));
399
-
endpoints = Hashtbl.create 16;
400
-
endpoints_mutex = Eio.Mutex.create ();
455
+
endpoints = Hashtbl.create 16;
456
+
endpoints_mutex = Eio.Mutex.create ();
(* Auto-cleanup on switch release *)
Eio.Switch.on_release sw (fun () ->
405
-
Eio.Cancel.protect (fun () ->
406
-
Log.info (fun m -> m "Closing connection pool");
407
-
(* Close all idle connections - active ones will be cleaned up by switch *)
408
-
Hashtbl.iter (fun _endpoint _ep_pool ->
409
-
(* Connections are bound to the switch and will be auto-closed *)
462
+
Eio.Cancel.protect (fun () ->
463
+
Log.info (fun m -> m "Closing connection pool");
464
+
(* Close all idle connections - active ones will be cleaned up by switch *)
466
+
(fun _endpoint _ep_pool ->
467
+
(* Connections are bound to the switch and will be auto-closed *)
413
-
Hashtbl.clear pool.endpoints
471
+
Hashtbl.clear pool.endpoints));
···
(* Increment active count *)
Eio.Mutex.use_rw ~protect:true ep_pool.mutex (fun () ->
427
-
ep_pool.stats.active <- ep_pool.stats.active + 1
483
+
ep_pool.stats.active <- ep_pool.stats.active + 1);
(* Decrement active count *)
Eio.Mutex.use_rw ~protect:true ep_pool.mutex (fun () ->
434
-
ep_pool.stats.active <- ep_pool.stats.active - 1
436
-
Log.debug (fun m -> m "Released connection to %a" Endpoint.pp endpoint)
489
+
ep_pool.stats.active <- ep_pool.stats.active - 1);
490
+
Log.debug (fun m -> m "Released connection to %a" Endpoint.pp endpoint))
(* Use Eio.Pool for resource management *)
Eio.Pool.use ep_pool.pool (fun conn ->
441
-
Log.debug (fun m -> m "Using connection to %a (uses=%d)"
442
-
Endpoint.pp endpoint (Connection.use_count conn));
494
+
Log.debug (fun m ->
495
+
m "Using connection to %a (uses=%d)" Endpoint.pp endpoint
496
+
(Connection.use_count conn));
444
-
(* Update last used time and use count *)
445
-
Connection.update_usage conn ~now:(get_time pool);
498
+
(* Update last used time and use count *)
499
+
Connection.update_usage conn ~now:(get_time pool);
447
-
(* Update idle stats (connection taken from idle pool) *)
448
-
Eio.Mutex.use_rw ~protect:true ep_pool.mutex (fun () ->
449
-
ep_pool.stats.idle <- max 0 (ep_pool.stats.idle - 1)
501
+
(* Update idle stats (connection taken from idle pool) *)
502
+
Eio.Mutex.use_rw ~protect:true ep_pool.mutex (fun () ->
503
+
ep_pool.stats.idle <- max 0 (ep_pool.stats.idle - 1));
453
-
let result = f conn.flow in
506
+
let result = f conn.flow in
455
-
(* Success - connection will be returned to pool by Eio.Pool *)
456
-
(* Update idle stats (connection returned to idle pool) *)
457
-
Eio.Mutex.use_rw ~protect:true ep_pool.mutex (fun () ->
458
-
ep_pool.stats.idle <- ep_pool.stats.idle + 1
508
+
(* Success - connection will be returned to pool by Eio.Pool *)
509
+
(* Update idle stats (connection returned to idle pool) *)
510
+
Eio.Mutex.use_rw ~protect:true ep_pool.mutex (fun () ->
511
+
ep_pool.stats.idle <- ep_pool.stats.idle + 1);
463
-
(* Error - close connection so it won't be reused *)
464
-
Log.warn (fun m -> m "Error using connection to %a: %s"
465
-
Endpoint.pp endpoint (Printexc.to_string e));
466
-
close_internal pool conn;
515
+
(* Error - close connection so it won't be reused *)
517
+
m "Error using connection to %a: %s" Endpoint.pp endpoint
518
+
(Printexc.to_string e));
519
+
close_internal pool conn;
468
-
(* Update error stats *)
469
-
Eio.Mutex.use_rw ~protect:true ep_pool.mutex (fun () ->
470
-
ep_pool.stats.errors <- ep_pool.stats.errors + 1
521
+
(* Update error stats *)
522
+
Eio.Mutex.use_rw ~protect:true ep_pool.mutex (fun () ->
523
+
ep_pool.stats.errors <- ep_pool.stats.errors + 1);
(** {1 Public API - Statistics} *)
let stats (T pool) endpoint =
match Hashtbl.find_opt pool.endpoints endpoint with
482
-
Eio.Mutex.use_ro ep_pool.mutex (fun () ->
483
-
snapshot_stats ep_pool.stats
532
+
Eio.Mutex.use_ro ep_pool.mutex (fun () -> snapshot_stats ep_pool.stats)
(* No pool for this endpoint yet *)
535
+
Stats.make ~active:0 ~idle:0 ~total_created:0 ~total_reused:0
536
+
~total_closed:0 ~errors:0
Eio.Mutex.use_ro pool.endpoints_mutex (fun () ->
497
-
Hashtbl.fold (fun endpoint ep_pool acc ->
498
-
let stats = Eio.Mutex.use_ro ep_pool.mutex (fun () ->
499
-
snapshot_stats ep_pool.stats
501
-
(endpoint, stats) :: acc
502
-
) pool.endpoints []
541
+
(fun endpoint ep_pool acc ->
543
+
Eio.Mutex.use_ro ep_pool.mutex (fun () ->
544
+
snapshot_stats ep_pool.stats)
546
+
(endpoint, stats) :: acc)
(** {1 Public API - Pool Management} *)
···
match Hashtbl.find_opt pool.endpoints endpoint with
Eio.Cancel.protect (fun () ->
512
-
(* Remove endpoint pool from hashtable *)
513
-
(* Idle connections will be discarded *)
514
-
(* Active connections will be closed when returned *)
515
-
Eio.Mutex.use_rw ~protect:true pool.endpoints_mutex (fun () ->
516
-
Hashtbl.remove pool.endpoints endpoint
556
+
(* Remove endpoint pool from hashtable *)
557
+
(* Idle connections will be discarded *)
558
+
(* Active connections will be closed when returned *)
559
+
Eio.Mutex.use_rw ~protect:true pool.endpoints_mutex (fun () ->
560
+
Hashtbl.remove pool.endpoints endpoint))
520
-
Log.debug (fun m -> m "No endpoint pool found for %a" Endpoint.pp endpoint)
562
+
Log.debug (fun m ->
563
+
m "No endpoint pool found for %a" Endpoint.pp endpoint)
anil.recoil.org