anil.recoil.org
TCP/TLS connection pooling for Eio
refine
+142
-57
+1
-1
lib/config.ml
+1
-1
lib/config.ml
···
15
15
max_connection_lifetime : float;
16
16
max_connection_uses : int option;
17
17
health_check :
18
-
([ `Close | `Flow | `R | `Shutdown | `W ] Eio.Resource.t -> bool) option;
18
+
([Eio.Resource.close_ty | Eio.Flow.two_way_ty] Eio.Resource.t -> bool) option;
19
19
connect_timeout : float option;
20
20
connect_retry_count : int;
21
21
connect_retry_delay : float;
+2
-3
lib/config.mli
+2
-3
lib/config.mli
···
25
25
?max_idle_time:float ->
26
26
?max_connection_lifetime:float ->
27
27
?max_connection_uses:int ->
28
-
?health_check:
29
-
([ `Close | `Flow | `R | `Shutdown | `W ] Eio.Resource.t -> bool) ->
28
+
?health_check:([Eio.Resource.close_ty | Eio.Flow.two_way_ty] Eio.Resource.t -> bool) ->
30
29
?connect_timeout:float ->
31
30
?connect_retry_count:int ->
32
31
?connect_retry_delay:float ->
···
81
80
(** Get maximum connection uses, if any. *)
82
81
83
82
val health_check :
84
-
t -> ([ `Close | `Flow | `R | `Shutdown | `W ] Eio.Resource.t -> bool) option
83
+
t -> ([Eio.Resource.close_ty | Eio.Flow.two_way_ty] Eio.Resource.t -> bool) option
85
84
(** Get custom health check function, if any. *)
86
85
87
86
val connect_timeout : t -> float option
+1
-1
lib/connection.ml
+1
-1
lib/connection.ml
···
12
12
module Log = (val Logs.src_log src : Logs.LOG)
13
13
14
14
type t = {
15
-
flow : [ `Close | `Flow | `R | `Shutdown | `W ] Eio.Resource.t;
15
+
flow : [Eio.Resource.close_ty | Eio.Flow.two_way_ty] Eio.Resource.t;
16
16
created_at : float;
17
17
mutable last_used : float;
18
18
mutable use_count : int;
+79
-37
lib/conpool.ml
+79
-37
lib/conpool.ml
···
43
43
| Invalid_config msg -> Fmt.pf ppf "Invalid configuration: %s" msg
44
44
| Invalid_endpoint msg -> Fmt.pf ppf "Invalid endpoint: %s" msg
45
45
46
+
type Eio.Exn.err += E of error
47
+
48
+
let err e = Eio.Exn.create (E e)
49
+
50
+
let () =
51
+
Eio.Exn.register_pp (fun f -> function
52
+
| E e ->
53
+
Fmt.string f "Conpool ";
54
+
pp_error f e;
55
+
true
56
+
| _ -> false)
57
+
58
+
(** {1 Connection Types} *)
59
+
60
+
type connection_ty = [Eio.Resource.close_ty | Eio.Flow.two_way_ty]
61
+
type connection = connection_ty Eio.Resource.t
62
+
46
63
type endp_stats = {
47
64
mutable active : int;
48
65
mutable idle : int;
···
152
169
let flow =
153
170
match pool.tls with
154
171
| None ->
155
-
(socket :> [ `Close | `Flow | `R | `Shutdown | `W ] Eio.Resource.t)
172
+
(socket :> connection)
156
173
| Some tls_cfg ->
157
174
Log.debug (fun m ->
158
175
m "Initiating TLS handshake with %a" Endpoint.pp endpoint);
···
167
184
in
168
185
Log.info (fun m ->
169
186
m "TLS connection established to %a" Endpoint.pp endpoint);
170
-
(tls_flow :> [ `Close | `Flow | `R | `Shutdown | `W ] Eio.Resource.t)
187
+
(tls_flow :> connection)
171
188
in
172
189
173
190
let now = get_time pool in
···
474
491
475
492
(** {1 Public API - Connection Management} *)
476
493
477
-
let with_connection (T pool) endpoint f =
494
+
let connection_internal ~sw (T pool) endpoint =
478
495
Log.debug (fun m -> m "Acquiring connection to %a" Endpoint.pp endpoint);
479
496
let ep_pool = get_or_create_endpoint_pool pool endpoint in
497
+
498
+
(* Create promises for connection handoff and cleanup signal *)
499
+
let conn_promise, conn_resolver = Eio.Promise.create () in
500
+
let done_promise, done_resolver = Eio.Promise.create () in
480
501
481
502
(* Increment active count *)
482
503
Eio.Mutex.use_rw ~protect:true ep_pool.mutex (fun () ->
483
504
ep_pool.stats.active <- ep_pool.stats.active + 1);
484
505
485
-
Fun.protect
486
-
~finally:(fun () ->
487
-
(* Decrement active count *)
488
-
Eio.Mutex.use_rw ~protect:true ep_pool.mutex (fun () ->
489
-
ep_pool.stats.active <- ep_pool.stats.active - 1);
490
-
Log.debug (fun m -> m "Released connection to %a" Endpoint.pp endpoint))
491
-
(fun () ->
492
-
(* Use Eio.Pool for resource management *)
493
-
Eio.Pool.use ep_pool.pool (fun conn ->
494
-
Log.debug (fun m ->
495
-
m "Using connection to %a (uses=%d)" Endpoint.pp endpoint
496
-
(Connection.use_count conn));
506
+
(* Fork a daemon fiber to manage the connection lifecycle *)
507
+
Eio.Fiber.fork_daemon ~sw (fun () ->
508
+
Fun.protect
509
+
~finally:(fun () ->
510
+
(* Decrement active count *)
511
+
Eio.Mutex.use_rw ~protect:true ep_pool.mutex (fun () ->
512
+
ep_pool.stats.active <- ep_pool.stats.active - 1);
513
+
Log.debug (fun m -> m "Released connection to %a" Endpoint.pp endpoint))
514
+
(fun () ->
515
+
(* Use Eio.Pool for resource management *)
516
+
Eio.Pool.use ep_pool.pool (fun conn ->
517
+
Log.debug (fun m ->
518
+
m "Using connection to %a (uses=%d)" Endpoint.pp endpoint
519
+
(Connection.use_count conn));
520
+
521
+
(* Update last used time and use count *)
522
+
Connection.update_usage conn ~now:(get_time pool);
523
+
524
+
(* Update idle stats (connection taken from idle pool) *)
525
+
Eio.Mutex.use_rw ~protect:true ep_pool.mutex (fun () ->
526
+
ep_pool.stats.idle <- max 0 (ep_pool.stats.idle - 1));
527
+
528
+
(* Hand off connection to caller *)
529
+
Eio.Promise.resolve conn_resolver conn.flow;
530
+
531
+
try
532
+
(* Wait for switch to signal cleanup *)
533
+
Eio.Promise.await done_promise;
534
+
535
+
(* Success - connection will be returned to pool by Eio.Pool *)
536
+
(* Update idle stats (connection returned to idle pool) *)
537
+
Eio.Mutex.use_rw ~protect:true ep_pool.mutex (fun () ->
538
+
ep_pool.stats.idle <- ep_pool.stats.idle + 1);
539
+
540
+
`Stop_daemon
541
+
with e ->
542
+
(* Error - close connection so it won't be reused *)
543
+
Log.warn (fun m ->
544
+
m "Error with connection to %a: %s" Endpoint.pp endpoint
545
+
(Printexc.to_string e));
546
+
close_internal pool conn;
497
547
498
-
(* Update last used time and use count *)
499
-
Connection.update_usage conn ~now:(get_time pool);
548
+
(* Update error stats *)
549
+
Eio.Mutex.use_rw ~protect:true ep_pool.mutex (fun () ->
550
+
ep_pool.stats.errors <- ep_pool.stats.errors + 1);
500
551
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));
552
+
raise e)));
504
553
505
-
try
506
-
let result = f conn.flow in
554
+
(* Signal cleanup when switch ends *)
555
+
Eio.Switch.on_release sw (fun () ->
556
+
Eio.Promise.resolve done_resolver ());
507
557
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);
558
+
(* Return the connection *)
559
+
Eio.Promise.await conn_promise
512
560
513
-
result
514
-
with e ->
515
-
(* Error - close connection so it won't be reused *)
516
-
Log.warn (fun m ->
517
-
m "Error using connection to %a: %s" Endpoint.pp endpoint
518
-
(Printexc.to_string e));
519
-
close_internal pool conn;
561
+
let connection ~sw t endpoint = connection_internal ~sw t endpoint
520
562
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);
563
+
let with_connection t endpoint f =
564
+
Eio.Switch.run (fun sw -> f (connection ~sw t endpoint))
524
565
525
-
raise e))
566
+
let with_connection_exn t endpoint f =
567
+
try with_connection t endpoint f with Pool_error e -> raise (err e)
526
568
527
569
(** {1 Public API - Statistics} *)
528
570
+59
-15
lib/conpool.mli
+59
-15
lib/conpool.mli
···
21
21
22
22
(** {1 Core Types} *)
23
23
24
-
module Endpoint : module type of Endpoint
24
+
module Endpoint = Endpoint
25
25
(** Network endpoint representation *)
26
26
27
-
module Tls_config : module type of Tls_config
27
+
module Tls_config = Tls_config
28
28
(** TLS configuration for connection pools *)
29
29
30
-
module Config : module type of Config
30
+
module Config = Config
31
31
(** Configuration for connection pools *)
32
32
33
-
module Stats : module type of Stats
33
+
module Stats = Stats
34
34
(** Statistics for connection pool endpoints *)
35
35
36
-
module Cmd : module type of Cmd
36
+
module Cmd = Cmd
37
37
(** Cmdliner terms for connection pool configuration *)
38
38
39
39
(** {1 Errors} *)
···
56
56
57
57
Most pool operations can raise this exception. Use {!pp_error} to get
58
58
human-readable error messages. *)
59
+
60
+
type Eio.Exn.err += E of error
61
+
(** Extension of Eio's error type for connection pool errors. *)
62
+
63
+
val err : error -> exn
64
+
(** [err e] is [Eio.Exn.create (E e)].
65
+
66
+
This converts a connection pool error to an Eio exception, allowing it to
67
+
be handled uniformly with other Eio I/O errors. *)
59
68
60
69
val pp_error : error Fmt.t
61
70
(** Pretty-printer for error values. *)
62
71
72
+
(** {1 Connection Types} *)
73
+
74
+
type connection_ty = [Eio.Resource.close_ty | Eio.Flow.two_way_ty]
75
+
(** The type tags for a pooled connection.
76
+
Connections support reading, writing, shutdown, and closing. *)
77
+
78
+
type connection = connection_ty Eio.Resource.t
79
+
(** A connection resource from the pool. *)
80
+
63
81
(** {1 Connection Pool} *)
64
82
65
83
type t
···
85
103
86
104
(** {1 Connection Usage} *)
87
105
88
-
val with_connection :
89
-
t ->
90
-
Endpoint.t ->
91
-
([ `Close | `Flow | `R | `Shutdown | `W ] Eio.Resource.t -> 'a) ->
92
-
'a
93
-
(** Acquire connection, use it, automatically release back to pool.
106
+
val connection : sw:Eio.Switch.t -> t -> Endpoint.t -> connection
107
+
(** [connection ~sw pool endpoint] acquires a connection from the pool.
94
108
95
-
If idle connection available and healthy:
96
-
- Reuse from pool (validates health first) Else:
109
+
The connection is automatically returned to the pool when [sw] finishes.
110
+
If the connection becomes unhealthy or an error occurs during use, it is
111
+
closed instead of being returned to the pool.
112
+
113
+
If an idle connection is available and healthy:
114
+
- Reuse from pool (validates health first)
115
+
116
+
Otherwise:
97
117
- Create new connection (may block if endpoint at limit)
98
118
99
-
On success: connection returned to pool for reuse On error: connection
100
-
closed, not returned to pool
119
+
Example:
120
+
{[
121
+
let endpoint = Conpool.Endpoint.make ~host:"example.com" ~port:443 in
122
+
Eio.Switch.run (fun sw ->
123
+
let conn = Conpool.connection ~sw pool endpoint in
124
+
Eio.Flow.copy_string "GET / HTTP/1.1\r\n\r\n" conn;
125
+
let buf = Eio.Buf_read.of_flow conn ~max_size:4096 in
126
+
Eio.Buf_read.take_all buf)
127
+
]} *)
128
+
129
+
val with_connection : t -> Endpoint.t -> (connection -> 'a) -> 'a
130
+
(** [with_connection pool endpoint fn] is a convenience wrapper around
131
+
{!val:connection}.
132
+
133
+
Equivalent to:
134
+
{[
135
+
Eio.Switch.run (fun sw -> fn (connection ~sw pool endpoint))
136
+
]}
101
137
102
138
Example:
103
139
{[
···
108
144
let buf = Eio.Buf_read.of_flow conn ~max_size:4096 in
109
145
Eio.Buf_read.take_all buf)
110
146
]} *)
147
+
148
+
val with_connection_exn : t -> Endpoint.t -> (connection -> 'a) -> 'a
149
+
(** [with_connection_exn pool endpoint fn] is like {!with_connection} but
150
+
converts {!Pool_error} exceptions to [Eio.Io] exceptions for better
151
+
integration with Eio error handling.
152
+
153
+
This is useful when you want pool errors to be handled uniformly with other
154
+
Eio I/O errors. *)
111
155
112
156
(** {1 Statistics & Monitoring} *)
113
157