TCP/TLS connection pooling for Eio
ocaml-conpool: Reformat with ocamlformat 0.28.1
+861
-755
+24
-27
lib/config.ml
+24
-27
lib/config.ml
···
15
15
max_connection_lifetime : float;
16
16
max_connection_uses : int option;
17
17
health_check :
18
-
([Eio.Resource.close_ty | Eio.Flow.two_way_ty] Eio.Resource.t -> bool) option;
18
+
([ Eio.Resource.close_ty | Eio.Flow.two_way_ty ] Eio.Resource.t -> bool)
19
+
option;
19
20
connect_timeout : float option;
20
21
connect_retry_count : int;
21
22
connect_retry_delay : float;
···
114
115
(** {1 Protocol Handler Configuration}
115
116
116
117
Protocol handlers define protocol-specific behavior for connection pools.
117
-
This enables different pooling strategies for different protocols
118
-
(e.g., exclusive for HTTP/1.x, shared for HTTP/2). *)
118
+
This enables different pooling strategies for different protocols (e.g.,
119
+
exclusive for HTTP/1.x, shared for HTTP/2). *)
119
120
120
121
(** Access mode for connections.
121
122
- [Exclusive] - Each connection is used by one request at a time (HTTP/1.x)
122
123
- [Shared] - Multiple requests can share a connection (HTTP/2) *)
123
124
type access_mode =
124
-
| Exclusive
125
-
(** Exclusive access - one request per connection at a time *)
125
+
| Exclusive (** Exclusive access - one request per connection at a time *)
126
126
| Shared of int
127
127
(** Shared access - up to n concurrent requests per connection *)
128
128
129
+
type connection_flow =
130
+
[ Eio.Resource.close_ty | Eio.Flow.two_way_ty ] Eio.Resource.t
129
131
(** Connection type alias for protocol config *)
130
-
type connection_flow = [Eio.Resource.close_ty | Eio.Flow.two_way_ty] Eio.Resource.t
131
132
132
-
(** Protocol configuration for typed connection pools.
133
-
@param 'state The protocol-specific state type (e.g., H2_client.t for HTTP/2) *)
134
133
type 'state protocol_config = {
135
134
init_state :
136
135
sw:Eio.Switch.t ->
137
136
flow:connection_flow ->
138
137
tls_epoch:Tls.Core.epoch_data option ->
139
138
'state;
140
-
(** Initialize protocol state when a new connection is created.
141
-
The [sw] parameter is a connection-lifetime switch that can be used
142
-
to spawn long-running fibers (e.g., HTTP/2 frame reader).
143
-
For HTTP/2, this performs the handshake and returns the H2_client.t. *)
144
-
139
+
(** Initialize protocol state when a new connection is created. The [sw]
140
+
parameter is a connection-lifetime switch that can be used to spawn
141
+
long-running fibers (e.g., HTTP/2 frame reader). For HTTP/2, this
142
+
performs the handshake and returns the H2_client.t. *)
145
143
on_acquire : 'state -> unit;
146
-
(** Called when a connection is acquired from the pool.
147
-
For HTTP/2, this can start the background reader fiber if not already running. *)
148
-
144
+
(** Called when a connection is acquired from the pool. For HTTP/2, this
145
+
can start the background reader fiber if not already running. *)
149
146
on_release : 'state -> unit;
150
-
(** Called when a connection is released back to the pool.
151
-
For HTTP/2, this is typically a no-op since the reader keeps running. *)
152
-
147
+
(** Called when a connection is released back to the pool. For HTTP/2,
148
+
this is typically a no-op since the reader keeps running. *)
153
149
is_healthy : 'state -> bool;
154
-
(** Protocol-specific health check. Return false if connection should be closed.
155
-
For HTTP/2, checks if GOAWAY has been received. *)
156
-
150
+
(** Protocol-specific health check. Return false if connection should be
151
+
closed. For HTTP/2, checks if GOAWAY has been received. *)
157
152
on_close : 'state -> unit;
158
-
(** Cleanup callback when connection is destroyed.
159
-
For HTTP/2, can send GOAWAY frame. *)
160
-
153
+
(** Cleanup callback when connection is destroyed. For HTTP/2, can send
154
+
GOAWAY frame. *)
161
155
access_mode : 'state -> access_mode;
162
-
(** Get the access mode for this connection.
163
-
For HTTP/2, returns [Shared n] with max_concurrent from peer settings. *)
156
+
(** Get the access mode for this connection. For HTTP/2, returns
157
+
[Shared n] with max_concurrent from peer settings. *)
164
158
}
159
+
(** Protocol configuration for typed connection pools.
160
+
@param 'state
161
+
The protocol-specific state type (e.g., H2_client.t for HTTP/2) *)
+29
-29
lib/config.mli
+29
-29
lib/config.mli
···
25
25
?max_idle_time:float ->
26
26
?max_connection_lifetime:float ->
27
27
?max_connection_uses:int ->
28
-
?health_check:([Eio.Resource.close_ty | Eio.Flow.two_way_ty] Eio.Resource.t -> bool) ->
28
+
?health_check:
29
+
([ Eio.Resource.close_ty | Eio.Flow.two_way_ty ] Eio.Resource.t -> bool) ->
29
30
?connect_timeout:float ->
30
31
?connect_retry_count:int ->
31
32
?connect_retry_delay:float ->
···
80
81
(** Get maximum connection uses, if any. *)
81
82
82
83
val health_check :
83
-
t -> ([Eio.Resource.close_ty | Eio.Flow.two_way_ty] Eio.Resource.t -> bool) option
84
+
t ->
85
+
([ Eio.Resource.close_ty | Eio.Flow.two_way_ty ] Eio.Resource.t -> bool)
86
+
option
84
87
(** Get custom health check function, if any. *)
85
88
86
89
val connect_timeout : t -> float option
···
108
111
109
112
(** {1 Protocol Handler Configuration}
110
113
111
-
Protocol handlers define protocol-specific behavior for typed connection pools.
112
-
This enables different pooling strategies for different protocols
114
+
Protocol handlers define protocol-specific behavior for typed connection
115
+
pools. This enables different pooling strategies for different protocols
113
116
(e.g., exclusive for HTTP/1.x, shared for HTTP/2). *)
114
117
115
118
(** Access mode for connections.
116
119
- [Exclusive] - Each connection is used by one request at a time (HTTP/1.x)
117
-
- [Shared n] - Up to n concurrent requests can share a connection (HTTP/2) *)
120
+
- [Shared n] - Up to n concurrent requests can share a connection (HTTP/2)
121
+
*)
118
122
type access_mode =
119
-
| Exclusive
120
-
(** Exclusive access - one request per connection at a time *)
123
+
| Exclusive (** Exclusive access - one request per connection at a time *)
121
124
| Shared of int
122
125
(** Shared access - up to n concurrent requests per connection *)
123
126
127
+
type connection_flow =
128
+
[ Eio.Resource.close_ty | Eio.Flow.two_way_ty ] Eio.Resource.t
124
129
(** Connection flow type for protocol handlers. *)
125
-
type connection_flow = [Eio.Resource.close_ty | Eio.Flow.two_way_ty] Eio.Resource.t
126
130
127
-
(** Protocol configuration for typed connection pools.
128
-
@param 'state The protocol-specific state type (e.g., H2_client.t for HTTP/2) *)
129
131
type 'state protocol_config = {
130
132
init_state :
131
133
sw:Eio.Switch.t ->
132
134
flow:connection_flow ->
133
135
tls_epoch:Tls.Core.epoch_data option ->
134
136
'state;
135
-
(** Initialize protocol state when a new connection is created.
136
-
The [sw] parameter is a connection-lifetime switch that can be used
137
-
to spawn long-running fibers (e.g., HTTP/2 frame reader).
138
-
For HTTP/2, this performs the handshake and returns the H2_client.t. *)
139
-
137
+
(** Initialize protocol state when a new connection is created. The [sw]
138
+
parameter is a connection-lifetime switch that can be used to spawn
139
+
long-running fibers (e.g., HTTP/2 frame reader). For HTTP/2, this
140
+
performs the handshake and returns the H2_client.t. *)
140
141
on_acquire : 'state -> unit;
141
-
(** Called when a connection is acquired from the pool.
142
-
For HTTP/2, this can start the background reader fiber if not already running. *)
143
-
142
+
(** Called when a connection is acquired from the pool. For HTTP/2, this
143
+
can start the background reader fiber if not already running. *)
144
144
on_release : 'state -> unit;
145
-
(** Called when a connection is released back to the pool.
146
-
For HTTP/2, this is typically a no-op since the reader keeps running. *)
147
-
145
+
(** Called when a connection is released back to the pool. For HTTP/2,
146
+
this is typically a no-op since the reader keeps running. *)
148
147
is_healthy : 'state -> bool;
149
-
(** Protocol-specific health check. Return false if connection should be closed.
150
-
For HTTP/2, checks if GOAWAY has been received. *)
151
-
148
+
(** Protocol-specific health check. Return false if connection should be
149
+
closed. For HTTP/2, checks if GOAWAY has been received. *)
152
150
on_close : 'state -> unit;
153
-
(** Cleanup callback when connection is destroyed.
154
-
For HTTP/2, can send GOAWAY frame. *)
155
-
151
+
(** Cleanup callback when connection is destroyed. For HTTP/2, can send
152
+
GOAWAY frame. *)
156
153
access_mode : 'state -> access_mode;
157
-
(** Get the access mode for this connection.
158
-
For HTTP/2, returns [Shared n] with max_concurrent from peer settings. *)
154
+
(** Get the access mode for this connection. For HTTP/2, returns
155
+
[Shared n] with max_concurrent from peer settings. *)
159
156
}
157
+
(** Protocol configuration for typed connection pools.
158
+
@param 'state
159
+
The protocol-specific state type (e.g., H2_client.t for HTTP/2) *)
+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 : [Eio.Resource.close_ty | Eio.Flow.two_way_ty] Eio.Resource.t;
15
+
flow : [ Eio.Resource.close_ty | Eio.Flow.two_way_ty ] Eio.Resource.t;
16
16
tls_flow : Tls_eio.t option;
17
17
created_at : float;
18
18
mutable last_used : float;
+281
-261
lib/conpool.ml
+281
-261
lib/conpool.ml
···
22
22
let suppress_tls_tracing () =
23
23
if not !tls_tracing_suppressed then begin
24
24
tls_tracing_suppressed := true;
25
-
match List.find_opt (fun s -> Logs.Src.name s = "tls.tracing") (Logs.Src.list ()) with
26
-
| Some tls_src ->
27
-
(match Logs.Src.level tls_src with
28
-
| Some Logs.Debug -> Logs.Src.set_level tls_src (Some Logs.Warning)
29
-
| _ -> ())
25
+
match
26
+
List.find_opt
27
+
(fun s -> Logs.Src.name s = "tls.tracing")
28
+
(Logs.Src.list ())
29
+
with
30
+
| Some tls_src -> (
31
+
match Logs.Src.level tls_src with
32
+
| Some Logs.Debug -> Logs.Src.set_level tls_src (Some Logs.Warning)
33
+
| _ -> ())
30
34
| None -> ()
31
35
end
32
36
···
61
65
62
66
let () =
63
67
Eio.Exn.register_pp (fun f -> function
64
-
| E e ->
65
-
Fmt.string f "Conpool ";
66
-
pp_error f e;
67
-
true
68
-
| _ -> false)
68
+
| E e ->
69
+
Fmt.string f "Conpool ";
70
+
pp_error f e;
71
+
true
72
+
| _ -> false)
69
73
70
74
(** {1 Connection Types} *)
71
75
72
-
type connection_ty = [Eio.Resource.close_ty | Eio.Flow.two_way_ty]
76
+
type connection_ty = [ Eio.Resource.close_ty | Eio.Flow.two_way_ty ]
73
77
type connection = connection_ty Eio.Resource.t
74
78
75
79
(** {1 Internal Types} *)
76
80
77
-
(** Internal connection wrapper with protocol state and tracking. *)
78
81
type 'state pooled_connection = {
79
82
pc_flow : connection;
80
83
pc_tls_flow : Tls_eio.t option;
···
89
92
pc_user_available : Eio.Condition.t;
90
93
mutable pc_closed : bool;
91
94
pc_connection_cancel : exn -> unit;
92
-
(** Cancels the connection-lifetime switch, stopping any protocol fibers. *)
95
+
(** Cancels the connection-lifetime switch, stopping any protocol fibers.
96
+
*)
93
97
}
98
+
(** Internal connection wrapper with protocol state and tracking. *)
94
99
95
-
(** Statistics for an endpoint. *)
96
100
type endp_stats = {
97
101
mutable active : int;
98
-
mutable idle : int;
99
-
(** Number of idle connections (active_users = 0). *)
102
+
mutable idle : int; (** Number of idle connections (active_users = 0). *)
100
103
mutable total_created : int;
101
104
mutable total_reused : int;
102
105
mutable total_closed : int;
103
-
mutable errors : int;
104
-
(** Number of connection errors encountered. *)
106
+
mutable errors : int; (** Number of connection errors encountered. *)
105
107
}
108
+
(** Statistics for an endpoint. *)
106
109
107
-
(** Endpoint pool storing connections. *)
108
110
type 'state endpoint_pool = {
109
111
connections : 'state pooled_connection list ref;
110
112
ep_mutex : Eio.Mutex.t;
111
113
stats : endp_stats;
112
114
stats_mutex : Eio.Mutex.t;
113
115
}
116
+
(** Endpoint pool storing connections. *)
114
117
115
-
(** Internal pool representation. *)
116
118
type ('state, 'clock, 'net) internal = {
117
119
sw : Eio.Switch.t;
118
120
net : 'net;
···
123
125
endpoints : (Endpoint.t, 'state endpoint_pool) Hashtbl.t;
124
126
endpoints_mutex : Eio.Mutex.t;
125
127
}
128
+
(** Internal pool representation. *)
126
129
127
130
(** {1 Public Types} *)
128
131
129
132
type 'state t =
130
-
Pool : ('state, 'clock Eio.Time.clock, 'net Eio.Net.t) internal -> 'state t
133
+
| Pool : ('state, 'clock Eio.Time.clock, 'net Eio.Net.t) internal -> 'state t
131
134
132
135
type 'state connection_info = {
133
136
flow : connection;
···
137
140
138
141
(** {1 Default Protocol Handler}
139
142
140
-
For simple exclusive-access protocols (HTTP/1.x, Redis, etc.),
141
-
use unit state with no special initialization. *)
143
+
For simple exclusive-access protocols (HTTP/1.x, Redis, etc.), use unit
144
+
state with no special initialization. *)
142
145
143
-
let default_protocol : unit Config.protocol_config = {
144
-
Config.init_state = (fun ~sw:_ ~flow:_ ~tls_epoch:_ -> ());
145
-
on_acquire = (fun () -> ());
146
-
on_release = (fun () -> ());
147
-
is_healthy = (fun () -> true);
148
-
on_close = (fun () -> ());
149
-
access_mode = (fun () -> Config.Exclusive);
150
-
}
146
+
let default_protocol : unit Config.protocol_config =
147
+
{
148
+
Config.init_state = (fun ~sw:_ ~flow:_ ~tls_epoch:_ -> ());
149
+
on_acquire = (fun () -> ());
150
+
on_release = (fun () -> ());
151
+
is_healthy = (fun () -> true);
152
+
on_close = (fun () -> ());
153
+
access_mode = (fun () -> Config.Exclusive);
154
+
}
151
155
152
156
(** {1 Helper Functions} *)
153
157
154
158
let get_time pool = Eio.Time.now pool.clock
155
159
156
-
let create_endp_stats () = {
157
-
active = 0;
158
-
idle = 0;
159
-
total_created = 0;
160
-
total_reused = 0;
161
-
total_closed = 0;
162
-
errors = 0;
163
-
}
160
+
let create_endp_stats () =
161
+
{
162
+
active = 0;
163
+
idle = 0;
164
+
total_created = 0;
165
+
total_reused = 0;
166
+
total_closed = 0;
167
+
errors = 0;
168
+
}
164
169
165
170
let snapshot_stats (stats : endp_stats) : Stats.t =
166
171
Stats.make ~active:stats.active ~idle:stats.idle
···
182
187
match addrs with
183
188
| addr :: _ -> addr
184
189
| [] ->
185
-
raise (err (Dns_resolution_failed { hostname = Endpoint.host endpoint }))
190
+
raise
191
+
(err (Dns_resolution_failed { hostname = Endpoint.host endpoint }))
186
192
with Eio.Io _ as ex ->
187
193
let bt = Printexc.get_raw_backtrace () in
188
194
Eio.Exn.reraise_with_context ex bt "resolving %a" Endpoint.pp endpoint
···
206
212
(* Optional TLS handshake *)
207
213
let flow, tls_flow =
208
214
match pool.tls with
209
-
| None ->
210
-
((socket :> connection), None)
211
-
| Some tls_config ->
215
+
| None -> ((socket :> connection), None)
216
+
| Some tls_config -> (
212
217
try
213
218
Log.debug (fun m ->
214
219
m "Initiating TLS handshake with %a" Endpoint.pp endpoint);
···
222
227
((tls :> connection), Some tls)
223
228
with Eio.Io _ as ex ->
224
229
let bt = Printexc.get_raw_backtrace () in
225
-
Eio.Exn.reraise_with_context ex bt "TLS handshake with %a" Endpoint.pp endpoint
230
+
Eio.Exn.reraise_with_context ex bt "TLS handshake with %a" Endpoint.pp
231
+
endpoint)
226
232
in
227
233
228
234
(* Get TLS epoch if available *)
···
243
249
let ready_promise, ready_resolver = Eio.Promise.create () in
244
250
245
251
Eio.Fiber.fork ~sw:pool.sw (fun () ->
246
-
Eio.Switch.run (fun conn_sw ->
247
-
conn_sw_ref := Some conn_sw;
248
-
conn_cancel_ref := (fun exn -> Eio.Switch.fail conn_sw exn);
249
-
(* Signal that the switch is ready *)
250
-
Eio.Promise.resolve ready_resolver ();
251
-
(* Block until the switch is cancelled *)
252
-
let wait_forever, _never_resolved = Eio.Promise.create () in
253
-
Eio.Promise.await wait_forever
254
-
)
255
-
);
252
+
Eio.Switch.run (fun conn_sw ->
253
+
conn_sw_ref := Some conn_sw;
254
+
(conn_cancel_ref := fun exn -> Eio.Switch.fail conn_sw exn);
255
+
(* Signal that the switch is ready *)
256
+
Eio.Promise.resolve ready_resolver ();
257
+
(* Block until the switch is cancelled *)
258
+
let wait_forever, _never_resolved = Eio.Promise.create () in
259
+
Eio.Promise.await wait_forever));
256
260
257
261
(* Wait for the switch to be created *)
258
262
Eio.Promise.await ready_promise;
···
260
264
let conn_cancel = !conn_cancel_ref in
261
265
262
266
(* Initialize protocol-specific state with connection switch *)
263
-
Log.debug (fun m -> m "Initializing protocol state for %a" Endpoint.pp endpoint);
267
+
Log.debug (fun m ->
268
+
m "Initializing protocol state for %a" Endpoint.pp endpoint);
264
269
let state = pool.protocol.init_state ~sw:conn_sw ~flow ~tls_epoch in
265
270
266
271
let now = get_time pool in
···
289
294
| Unhealthy_error of string
290
295
(** Connection failed due to an error (protocol failure, etc.) *)
291
296
| Unhealthy_lifecycle of string
292
-
(** Connection should close due to normal lifecycle (timeout, max uses, etc.) *)
297
+
(** Connection should close due to normal lifecycle (timeout, max uses,
298
+
etc.) *)
293
299
294
300
let check_health pool conn =
295
-
if conn.pc_closed then
296
-
Unhealthy_lifecycle "already closed"
301
+
if conn.pc_closed then Unhealthy_lifecycle "already closed"
297
302
else
298
303
(* Check protocol-specific health *)
299
304
let protocol_healthy = pool.protocol.is_healthy conn.pc_state in
300
305
if not protocol_healthy then begin
301
306
Log.debug (fun m -> m "Connection unhealthy: protocol check failed");
302
307
Unhealthy_error "protocol check failed"
303
-
end else
308
+
end
309
+
else
304
310
let now = get_time pool in
305
311
(* Check connection age *)
306
312
let age = now -. conn.pc_created_at in
307
313
let max_lifetime = Config.max_connection_lifetime pool.config in
308
314
if age > max_lifetime then begin
309
-
Log.debug (fun m -> m "Connection unhealthy: exceeded max lifetime (%.1fs > %.1fs)"
310
-
age max_lifetime);
315
+
Log.debug (fun m ->
316
+
m "Connection unhealthy: exceeded max lifetime (%.1fs > %.1fs)" age
317
+
max_lifetime);
311
318
Unhealthy_lifecycle "exceeded max lifetime"
312
-
end else
319
+
end
320
+
else
313
321
(* Check idle time - only for idle connections *)
314
322
let idle_time = now -. conn.pc_last_used in
315
323
let max_idle = Config.max_idle_time pool.config in
316
324
if conn.pc_active_users = 0 && idle_time > max_idle then begin
317
-
Log.debug (fun m -> m "Connection unhealthy: exceeded max idle time (%.1fs > %.1fs)"
318
-
idle_time max_idle);
325
+
Log.debug (fun m ->
326
+
m "Connection unhealthy: exceeded max idle time (%.1fs > %.1fs)"
327
+
idle_time max_idle);
319
328
Unhealthy_lifecycle "exceeded max idle time"
320
-
end else
329
+
end
330
+
else
321
331
(* Check use count *)
322
332
match Config.max_connection_uses pool.config with
323
333
| Some max_uses when conn.pc_use_count >= max_uses ->
324
-
Log.debug (fun m -> m "Connection unhealthy: exceeded max uses (%d >= %d)"
325
-
conn.pc_use_count max_uses);
334
+
Log.debug (fun m ->
335
+
m "Connection unhealthy: exceeded max uses (%d >= %d)"
336
+
conn.pc_use_count max_uses);
326
337
Unhealthy_lifecycle "exceeded max uses"
327
-
| _ ->
328
-
Healthy
338
+
| _ -> Healthy
329
339
330
340
let is_healthy pool conn =
331
341
match check_health pool conn with
···
341
351
m "Closing connection to %a" Endpoint.pp conn.pc_endpoint);
342
352
343
353
(* Cancel connection-lifetime switch first - this stops any protocol fibers *)
344
-
(try conn.pc_connection_cancel (Failure "Connection closed")
345
-
with _ -> ());
354
+
(try conn.pc_connection_cancel (Failure "Connection closed") with _ -> ());
346
355
347
356
(* Call protocol cleanup *)
348
357
pool.protocol.on_close conn.pc_state;
···
367
376
| None ->
368
377
Log.info (fun m ->
369
378
m "Creating endpoint pool for %a" Endpoint.pp endpoint);
370
-
let ep_pool = {
371
-
connections = ref [];
372
-
ep_mutex = Eio.Mutex.create ();
373
-
stats = create_endp_stats ();
374
-
stats_mutex = Eio.Mutex.create ();
375
-
} in
379
+
let ep_pool =
380
+
{
381
+
connections = ref [];
382
+
ep_mutex = Eio.Mutex.create ();
383
+
stats = create_endp_stats ();
384
+
stats_mutex = Eio.Mutex.create ();
385
+
}
386
+
in
376
387
Hashtbl.add pool.endpoints endpoint ep_pool;
377
388
ep_pool)
378
389
···
380
391
381
392
let rec acquire_connection pool ep_pool endpoint =
382
393
Eio.Mutex.use_rw ~protect:true ep_pool.ep_mutex (fun () ->
383
-
(* Find an existing healthy connection with available capacity *)
384
-
let rec find_available = function
385
-
| [] -> None
386
-
| conn :: rest ->
387
-
if not (is_healthy pool conn) then begin
388
-
conn.pc_closed <- true;
389
-
find_available rest
390
-
end else begin
391
-
match pool.protocol.access_mode conn.pc_state with
392
-
| Config.Exclusive ->
393
-
if conn.pc_active_users = 0 then
394
-
Some conn
395
-
else
396
-
find_available rest
397
-
| Config.Shared max_concurrent ->
398
-
if conn.pc_active_users < max_concurrent then
399
-
Some conn
400
-
else
401
-
find_available rest
402
-
end
403
-
in
394
+
(* Find an existing healthy connection with available capacity *)
395
+
let rec find_available = function
396
+
| [] -> None
397
+
| conn :: rest ->
398
+
if not (is_healthy pool conn) then begin
399
+
conn.pc_closed <- true;
400
+
find_available rest
401
+
end
402
+
else begin
403
+
match pool.protocol.access_mode conn.pc_state with
404
+
| Config.Exclusive ->
405
+
if conn.pc_active_users = 0 then Some conn
406
+
else find_available rest
407
+
| Config.Shared max_concurrent ->
408
+
if conn.pc_active_users < max_concurrent then Some conn
409
+
else find_available rest
410
+
end
411
+
in
404
412
405
-
(* Clean up closed connections *)
406
-
ep_pool.connections := List.filter (fun c -> not c.pc_closed) !(ep_pool.connections);
413
+
(* Clean up closed connections *)
414
+
ep_pool.connections :=
415
+
List.filter (fun c -> not c.pc_closed) !(ep_pool.connections);
407
416
408
-
match find_available !(ep_pool.connections) with
409
-
| Some conn ->
410
-
(* Reuse existing connection *)
411
-
let was_idle = conn.pc_active_users = 0 in
412
-
conn.pc_active_users <- conn.pc_active_users + 1;
413
-
conn.pc_last_used <- get_time pool;
414
-
conn.pc_use_count <- conn.pc_use_count + 1;
417
+
match find_available !(ep_pool.connections) with
418
+
| Some conn ->
419
+
(* Reuse existing connection *)
420
+
let was_idle = conn.pc_active_users = 0 in
421
+
conn.pc_active_users <- conn.pc_active_users + 1;
422
+
conn.pc_last_used <- get_time pool;
423
+
conn.pc_use_count <- conn.pc_use_count + 1;
415
424
416
-
Eio.Mutex.use_rw ~protect:true ep_pool.stats_mutex (fun () ->
417
-
ep_pool.stats.total_reused <- ep_pool.stats.total_reused + 1;
418
-
ep_pool.stats.active <- ep_pool.stats.active + 1;
419
-
(* Decrement idle count when connection becomes active *)
420
-
if was_idle then
421
-
ep_pool.stats.idle <- max 0 (ep_pool.stats.idle - 1));
425
+
Eio.Mutex.use_rw ~protect:true ep_pool.stats_mutex (fun () ->
426
+
ep_pool.stats.total_reused <- ep_pool.stats.total_reused + 1;
427
+
ep_pool.stats.active <- ep_pool.stats.active + 1;
428
+
(* Decrement idle count when connection becomes active *)
429
+
if was_idle then
430
+
ep_pool.stats.idle <- max 0 (ep_pool.stats.idle - 1));
422
431
423
-
Log.debug (fun m ->
424
-
m "Reusing connection to %a (users=%d)"
425
-
Endpoint.pp endpoint conn.pc_active_users);
432
+
Log.debug (fun m ->
433
+
m "Reusing connection to %a (users=%d)" Endpoint.pp endpoint
434
+
conn.pc_active_users);
426
435
427
-
(* Notify protocol handler of acquisition *)
428
-
pool.protocol.on_acquire conn.pc_state;
429
-
conn
436
+
(* Notify protocol handler of acquisition *)
437
+
pool.protocol.on_acquire conn.pc_state;
438
+
conn
439
+
| None ->
440
+
(* Need to create a new connection *)
441
+
let max_conns = Config.max_connections_per_endpoint pool.config in
442
+
let current_conns = List.length !(ep_pool.connections) in
430
443
431
-
| None ->
432
-
(* Need to create a new connection *)
433
-
let max_conns = Config.max_connections_per_endpoint pool.config in
434
-
let current_conns = List.length !(ep_pool.connections) in
435
-
436
-
if current_conns >= max_conns then begin
437
-
(* Wait for a connection to become available *)
438
-
Log.debug (fun m ->
439
-
m "At connection limit for %a (%d), waiting..."
440
-
Endpoint.pp endpoint max_conns);
444
+
if current_conns >= max_conns then begin
445
+
(* Wait for a connection to become available *)
446
+
Log.debug (fun m ->
447
+
m "At connection limit for %a (%d), waiting..." Endpoint.pp
448
+
endpoint max_conns);
441
449
442
-
(* Find a connection to wait on (prefer shared mode) *)
443
-
let wait_conn = List.find_opt (fun c ->
444
-
match pool.protocol.access_mode c.pc_state with
445
-
| Config.Shared _ -> true
446
-
| Config.Exclusive -> false
447
-
) !(ep_pool.connections) in
450
+
(* Find a connection to wait on (prefer shared mode) *)
451
+
let wait_conn =
452
+
List.find_opt
453
+
(fun c ->
454
+
match pool.protocol.access_mode c.pc_state with
455
+
| Config.Shared _ -> true
456
+
| Config.Exclusive -> false)
457
+
!(ep_pool.connections)
458
+
in
448
459
449
-
match wait_conn with
450
-
| Some conn ->
451
-
(* Wait for user slot *)
452
-
while conn.pc_active_users >=
453
-
(match pool.protocol.access_mode conn.pc_state with
454
-
| Config.Shared n -> n
455
-
| Config.Exclusive -> 1)
456
-
&& not conn.pc_closed do
457
-
Eio.Condition.await_no_mutex conn.pc_user_available
458
-
done;
459
-
if conn.pc_closed then
460
-
acquire_connection pool ep_pool endpoint
461
-
else begin
462
-
conn.pc_active_users <- conn.pc_active_users + 1;
463
-
conn.pc_last_used <- get_time pool;
464
-
conn.pc_use_count <- conn.pc_use_count + 1;
460
+
match wait_conn with
461
+
| Some conn ->
462
+
(* Wait for user slot *)
463
+
while
464
+
(conn.pc_active_users
465
+
>=
466
+
match pool.protocol.access_mode conn.pc_state with
467
+
| Config.Shared n -> n
468
+
| Config.Exclusive -> 1)
469
+
&& not conn.pc_closed
470
+
do
471
+
Eio.Condition.await_no_mutex conn.pc_user_available
472
+
done;
473
+
if conn.pc_closed then acquire_connection pool ep_pool endpoint
474
+
else begin
475
+
conn.pc_active_users <- conn.pc_active_users + 1;
476
+
conn.pc_last_used <- get_time pool;
477
+
conn.pc_use_count <- conn.pc_use_count + 1;
465
478
466
-
Eio.Mutex.use_rw ~protect:true ep_pool.stats_mutex (fun () ->
467
-
ep_pool.stats.total_reused <- ep_pool.stats.total_reused + 1;
468
-
ep_pool.stats.active <- ep_pool.stats.active + 1);
479
+
Eio.Mutex.use_rw ~protect:true ep_pool.stats_mutex (fun () ->
480
+
ep_pool.stats.total_reused <-
481
+
ep_pool.stats.total_reused + 1;
482
+
ep_pool.stats.active <- ep_pool.stats.active + 1);
469
483
470
-
(* Notify protocol handler of acquisition *)
471
-
pool.protocol.on_acquire conn.pc_state;
472
-
conn
473
-
end
474
-
| None ->
475
-
(* All connections are exclusive and in use - wait for any *)
476
-
let any_conn = List.hd !(ep_pool.connections) in
477
-
while any_conn.pc_active_users > 0 && not any_conn.pc_closed do
478
-
Eio.Condition.await_no_mutex any_conn.pc_user_available
479
-
done;
480
-
if any_conn.pc_closed then
481
-
acquire_connection pool ep_pool endpoint
482
-
else begin
483
-
(* Connection was idle (active_users = 0), now becoming active *)
484
-
Eio.Mutex.use_rw ~protect:true ep_pool.stats_mutex (fun () ->
485
-
ep_pool.stats.total_reused <- ep_pool.stats.total_reused + 1;
486
-
ep_pool.stats.active <- ep_pool.stats.active + 1;
487
-
ep_pool.stats.idle <- max 0 (ep_pool.stats.idle - 1));
488
-
any_conn.pc_active_users <- 1;
489
-
any_conn.pc_last_used <- get_time pool;
490
-
any_conn.pc_use_count <- any_conn.pc_use_count + 1;
491
-
(* Notify protocol handler of acquisition *)
492
-
pool.protocol.on_acquire any_conn.pc_state;
493
-
any_conn
494
-
end
495
-
end else begin
496
-
(* Create new connection *)
497
-
let conn = create_connection pool endpoint in
498
-
conn.pc_active_users <- 1;
499
-
ep_pool.connections := conn :: !(ep_pool.connections);
484
+
(* Notify protocol handler of acquisition *)
485
+
pool.protocol.on_acquire conn.pc_state;
486
+
conn
487
+
end
488
+
| None ->
489
+
(* All connections are exclusive and in use - wait for any *)
490
+
let any_conn = List.hd !(ep_pool.connections) in
491
+
while any_conn.pc_active_users > 0 && not any_conn.pc_closed do
492
+
Eio.Condition.await_no_mutex any_conn.pc_user_available
493
+
done;
494
+
if any_conn.pc_closed then
495
+
acquire_connection pool ep_pool endpoint
496
+
else begin
497
+
(* Connection was idle (active_users = 0), now becoming active *)
498
+
Eio.Mutex.use_rw ~protect:true ep_pool.stats_mutex (fun () ->
499
+
ep_pool.stats.total_reused <-
500
+
ep_pool.stats.total_reused + 1;
501
+
ep_pool.stats.active <- ep_pool.stats.active + 1;
502
+
ep_pool.stats.idle <- max 0 (ep_pool.stats.idle - 1));
503
+
any_conn.pc_active_users <- 1;
504
+
any_conn.pc_last_used <- get_time pool;
505
+
any_conn.pc_use_count <- any_conn.pc_use_count + 1;
506
+
(* Notify protocol handler of acquisition *)
507
+
pool.protocol.on_acquire any_conn.pc_state;
508
+
any_conn
509
+
end
510
+
end
511
+
else begin
512
+
(* Create new connection *)
513
+
let conn = create_connection pool endpoint in
514
+
conn.pc_active_users <- 1;
515
+
ep_pool.connections := conn :: !(ep_pool.connections);
500
516
501
-
Eio.Mutex.use_rw ~protect:true ep_pool.stats_mutex (fun () ->
502
-
ep_pool.stats.total_created <- ep_pool.stats.total_created + 1;
503
-
ep_pool.stats.active <- ep_pool.stats.active + 1);
517
+
Eio.Mutex.use_rw ~protect:true ep_pool.stats_mutex (fun () ->
518
+
ep_pool.stats.total_created <- ep_pool.stats.total_created + 1;
519
+
ep_pool.stats.active <- ep_pool.stats.active + 1);
504
520
505
-
Log.info (fun m ->
506
-
m "Created new connection to %a (total=%d)"
507
-
Endpoint.pp endpoint (List.length !(ep_pool.connections)));
521
+
Log.info (fun m ->
522
+
m "Created new connection to %a (total=%d)" Endpoint.pp endpoint
523
+
(List.length !(ep_pool.connections)));
508
524
509
-
(* Notify protocol handler of acquisition *)
510
-
pool.protocol.on_acquire conn.pc_state;
511
-
conn
512
-
end)
525
+
(* Notify protocol handler of acquisition *)
526
+
pool.protocol.on_acquire conn.pc_state;
527
+
conn
528
+
end)
513
529
514
530
(** {1 Connection Release} *)
515
531
···
518
534
pool.protocol.on_release conn.pc_state;
519
535
520
536
Eio.Mutex.use_rw ~protect:true ep_pool.ep_mutex (fun () ->
521
-
let was_active = conn.pc_active_users > 0 in
522
-
conn.pc_active_users <- max 0 (conn.pc_active_users - 1);
523
-
let now_idle = conn.pc_active_users = 0 in
537
+
let was_active = conn.pc_active_users > 0 in
538
+
conn.pc_active_users <- max 0 (conn.pc_active_users - 1);
539
+
let now_idle = conn.pc_active_users = 0 in
524
540
525
-
Eio.Mutex.use_rw ~protect:true ep_pool.stats_mutex (fun () ->
526
-
ep_pool.stats.active <- max 0 (ep_pool.stats.active - 1);
527
-
(* Track idle count: increment when connection becomes idle *)
528
-
if was_active && now_idle then
529
-
ep_pool.stats.idle <- ep_pool.stats.idle + 1);
541
+
Eio.Mutex.use_rw ~protect:true ep_pool.stats_mutex (fun () ->
542
+
ep_pool.stats.active <- max 0 (ep_pool.stats.active - 1);
543
+
(* Track idle count: increment when connection becomes idle *)
544
+
if was_active && now_idle then
545
+
ep_pool.stats.idle <- ep_pool.stats.idle + 1);
530
546
531
-
(* Signal waiting fibers *)
532
-
Eio.Condition.broadcast conn.pc_user_available;
547
+
(* Signal waiting fibers *)
548
+
Eio.Condition.broadcast conn.pc_user_available;
533
549
534
-
Log.debug (fun m ->
535
-
m "Released connection to %a (users=%d)"
536
-
Endpoint.pp conn.pc_endpoint conn.pc_active_users);
550
+
Log.debug (fun m ->
551
+
m "Released connection to %a (users=%d)" Endpoint.pp conn.pc_endpoint
552
+
conn.pc_active_users);
537
553
538
-
(* Check if connection should be closed *)
539
-
match check_health pool conn with
540
-
| Healthy -> ()
541
-
| Unhealthy_error reason ->
542
-
conn.pc_closed <- true;
554
+
(* Check if connection should be closed *)
555
+
match check_health pool conn with
556
+
| Healthy -> ()
557
+
| Unhealthy_error reason ->
558
+
conn.pc_closed <- true;
543
559
544
-
Eio.Mutex.use_rw ~protect:true ep_pool.stats_mutex (fun () ->
545
-
ep_pool.stats.total_closed <- ep_pool.stats.total_closed + 1;
546
-
ep_pool.stats.errors <- ep_pool.stats.errors + 1;
547
-
if now_idle then
548
-
ep_pool.stats.idle <- max 0 (ep_pool.stats.idle - 1));
560
+
Eio.Mutex.use_rw ~protect:true ep_pool.stats_mutex (fun () ->
561
+
ep_pool.stats.total_closed <- ep_pool.stats.total_closed + 1;
562
+
ep_pool.stats.errors <- ep_pool.stats.errors + 1;
563
+
if now_idle then
564
+
ep_pool.stats.idle <- max 0 (ep_pool.stats.idle - 1));
549
565
550
-
Log.warn (fun m -> m "Closing connection due to error: %s" reason);
551
-
close_connection pool conn;
552
-
ep_pool.connections := List.filter (fun c -> c != conn) !(ep_pool.connections)
566
+
Log.warn (fun m -> m "Closing connection due to error: %s" reason);
567
+
close_connection pool conn;
568
+
ep_pool.connections :=
569
+
List.filter (fun c -> c != conn) !(ep_pool.connections)
570
+
| Unhealthy_lifecycle reason ->
571
+
conn.pc_closed <- true;
553
572
554
-
| Unhealthy_lifecycle reason ->
555
-
conn.pc_closed <- true;
573
+
Eio.Mutex.use_rw ~protect:true ep_pool.stats_mutex (fun () ->
574
+
ep_pool.stats.total_closed <- ep_pool.stats.total_closed + 1;
575
+
if now_idle then
576
+
ep_pool.stats.idle <- max 0 (ep_pool.stats.idle - 1));
556
577
557
-
Eio.Mutex.use_rw ~protect:true ep_pool.stats_mutex (fun () ->
558
-
ep_pool.stats.total_closed <- ep_pool.stats.total_closed + 1;
559
-
if now_idle then
560
-
ep_pool.stats.idle <- max 0 (ep_pool.stats.idle - 1));
561
-
562
-
Log.debug (fun m -> m "Closing connection due to lifecycle: %s" reason);
563
-
close_connection pool conn;
564
-
ep_pool.connections := List.filter (fun c -> c != conn) !(ep_pool.connections))
578
+
Log.debug (fun m ->
579
+
m "Closing connection due to lifecycle: %s" reason);
580
+
close_connection pool conn;
581
+
ep_pool.connections :=
582
+
List.filter (fun c -> c != conn) !(ep_pool.connections))
565
583
566
584
(** {1 Public API} *)
567
585
568
-
let create ~sw ~(net : 'net Eio.Net.t) ~(clock : 'clock Eio.Time.clock)
569
-
?tls ?(config = Config.default) ?protocol () =
570
-
let protocol = match protocol with
586
+
let create ~sw ~(net : 'net Eio.Net.t) ~(clock : 'clock Eio.Time.clock) ?tls
587
+
?(config = Config.default) ?protocol () =
588
+
let protocol =
589
+
match protocol with
571
590
| Some p -> p
572
-
| None -> Obj.magic default_protocol (* Safe: unit is compatible with any 'state *)
591
+
| None ->
592
+
Obj.magic
593
+
default_protocol (* Safe: unit is compatible with any 'state *)
573
594
in
574
595
575
596
Log.info (fun m ->
576
597
m "Creating connection pool (max_per_endpoint=%d)"
577
598
(Config.max_connections_per_endpoint config));
578
599
579
-
let pool = {
580
-
sw;
581
-
net;
582
-
clock;
583
-
config;
584
-
tls;
585
-
protocol;
586
-
endpoints = Hashtbl.create 16;
587
-
endpoints_mutex = Eio.Mutex.create ();
588
-
} in
600
+
let pool =
601
+
{
602
+
sw;
603
+
net;
604
+
clock;
605
+
config;
606
+
tls;
607
+
protocol;
608
+
endpoints = Hashtbl.create 16;
609
+
endpoints_mutex = Eio.Mutex.create ();
610
+
}
611
+
in
589
612
590
613
(* Auto-cleanup on switch release *)
591
614
Eio.Switch.on_release sw (fun () ->
592
615
Eio.Cancel.protect (fun () ->
593
616
Log.info (fun m -> m "Closing connection pool");
594
-
Hashtbl.iter (fun _endpoint ep_pool ->
595
-
List.iter (fun conn ->
596
-
close_connection pool conn
597
-
) !(ep_pool.connections)
598
-
) pool.endpoints;
617
+
Hashtbl.iter
618
+
(fun _endpoint ep_pool ->
619
+
List.iter
620
+
(fun conn -> close_connection pool conn)
621
+
!(ep_pool.connections))
622
+
pool.endpoints;
599
623
Hashtbl.clear pool.endpoints));
600
624
601
625
Pool pool
···
607
631
let conn = acquire_connection pool ep_pool endpoint in
608
632
609
633
(* Release connection when switch ends *)
610
-
Eio.Switch.on_release sw (fun () ->
611
-
release_connection pool ep_pool conn);
634
+
Eio.Switch.on_release sw (fun () -> release_connection pool ep_pool conn);
612
635
613
636
(* Get TLS epoch if available *)
614
637
let tls_epoch =
···
620
643
| None -> None
621
644
in
622
645
623
-
{
624
-
flow = conn.pc_flow;
625
-
tls_epoch;
626
-
state = conn.pc_state;
627
-
}
646
+
{ flow = conn.pc_flow; tls_epoch; state = conn.pc_state }
628
647
629
648
let with_connection pool endpoint f =
630
649
Eio.Switch.run (fun sw -> f (connection ~sw pool endpoint))
···
632
651
let stats (Pool pool) endpoint =
633
652
match Hashtbl.find_opt pool.endpoints endpoint with
634
653
| Some ep_pool ->
635
-
Eio.Mutex.use_ro ep_pool.stats_mutex (fun () -> snapshot_stats ep_pool.stats)
654
+
Eio.Mutex.use_ro ep_pool.stats_mutex (fun () ->
655
+
snapshot_stats ep_pool.stats)
636
656
| None ->
637
657
Stats.make ~active:0 ~idle:0 ~total_created:0 ~total_reused:0
638
658
~total_closed:0 ~errors:0
···
654
674
| Some ep_pool ->
655
675
Eio.Cancel.protect (fun () ->
656
676
Eio.Mutex.use_rw ~protect:true ep_pool.ep_mutex (fun () ->
657
-
List.iter (fun conn ->
658
-
close_connection pool conn
659
-
) !(ep_pool.connections);
660
-
ep_pool.connections := []);
677
+
List.iter
678
+
(fun conn -> close_connection pool conn)
679
+
!(ep_pool.connections);
680
+
ep_pool.connections := []);
661
681
Eio.Mutex.use_rw ~protect:true pool.endpoints_mutex (fun () ->
662
682
Hashtbl.remove pool.endpoints endpoint))
663
683
| None ->
+26
-23
lib/conpool.mli
+26
-23
lib/conpool.mli
···
6
6
(** Conpool - Protocol-aware TCP/IP connection pooling library for Eio
7
7
8
8
Conpool provides efficient connection pooling with support for both
9
-
exclusive (HTTP/1.x) and shared (HTTP/2) access modes. All connections
10
-
carry protocol-specific state managed through callbacks.
9
+
exclusive (HTTP/1.x) and shared (HTTP/2) access modes. All connections carry
10
+
protocol-specific state managed through callbacks.
11
11
12
12
{2 Quick Start}
13
13
···
15
15
{[
16
16
let pool = Conpool.create ~sw ~net ~clock ~tls () in
17
17
Eio.Switch.run (fun conn_sw ->
18
-
let conn = Conpool.connection ~sw:conn_sw pool endpoint in
19
-
(* Use conn.flow for I/O *)
20
-
Eio.Flow.copy_string "GET / HTTP/1.1\r\n\r\n" conn.flow)
18
+
let conn = Conpool.connection ~sw:conn_sw pool endpoint in
19
+
(* Use conn.flow for I/O *)
20
+
Eio.Flow.copy_string "GET / HTTP/1.1\r\n\r\n" conn.flow)
21
21
]}
22
22
23
23
For multiplexed protocols (HTTP/2):
···
75
75
76
76
(** {1 Connection Types} *)
77
77
78
-
type connection_ty = [Eio.Resource.close_ty | Eio.Flow.two_way_ty]
78
+
type connection_ty = [ Eio.Resource.close_ty | Eio.Flow.two_way_ty ]
79
79
(** Type tags for a pooled connection. *)
80
80
81
81
type connection = connection_ty Eio.Resource.t
···
84
84
(** {1 Connection Pool}
85
85
86
86
All pools are typed - they carry protocol-specific state with each
87
-
connection. For simple exclusive-access protocols, use the default
88
-
[unit] state which requires no protocol handler. *)
87
+
connection. For simple exclusive-access protocols, use the default [unit]
88
+
state which requires no protocol handler. *)
89
89
90
90
type 'state t
91
91
(** Connection pool with protocol-specific state ['state].
92
92
93
-
- For HTTP/1.x: use [unit t] with exclusive access (one request per connection)
94
-
- For HTTP/2: use [h2_state t] with shared access (multiple streams per connection) *)
93
+
- For HTTP/1.x: use [unit t] with exclusive access (one request per
94
+
connection)
95
+
- For HTTP/2: use [h2_state t] with shared access (multiple streams per
96
+
connection) *)
95
97
96
-
(** Connection with protocol-specific state. *)
97
98
type 'state connection_info = {
98
-
flow : connection;
99
-
(** The underlying connection flow for I/O. *)
99
+
flow : connection; (** The underlying connection flow for I/O. *)
100
100
tls_epoch : Tls.Core.epoch_data option;
101
101
(** TLS epoch data if connection uses TLS. *)
102
-
state : 'state;
103
-
(** Protocol-specific state (e.g., H2_client.t for HTTP/2). *)
102
+
state : 'state; (** Protocol-specific state (e.g., H2_client.t for HTTP/2). *)
104
103
}
104
+
(** Connection with protocol-specific state. *)
105
105
106
106
(** {2 Pool Creation} *)
107
107
···
121
121
@param clock Clock for timeouts
122
122
@param tls Optional TLS client configuration
123
123
@param config Pool configuration (uses {!Config.default} if not provided)
124
-
@param protocol Protocol handler for state management. If not provided,
125
-
creates a [unit t] pool with exclusive access mode (one user per connection).
124
+
@param protocol
125
+
Protocol handler for state management. If not provided, creates a [unit t]
126
+
pool with exclusive access mode (one user per connection).
126
127
127
128
Examples:
128
129
···
138
139
139
140
(** {2 Connection Acquisition} *)
140
141
141
-
val connection : sw:Eio.Switch.t -> 'state t -> Endpoint.t -> 'state connection_info
142
+
val connection :
143
+
sw:Eio.Switch.t -> 'state t -> Endpoint.t -> 'state connection_info
142
144
(** [connection ~sw pool endpoint] acquires a connection from the pool.
143
145
144
146
The connection is automatically released when [sw] finishes:
···
152
154
Example:
153
155
{[
154
156
Eio.Switch.run (fun sw ->
155
-
let conn = Conpool.connection ~sw pool endpoint in
156
-
(* For HTTP/1.x: conn.state is () *)
157
-
(* For HTTP/2: conn.state is H2_client.t *)
158
-
Eio.Flow.copy_string data conn.flow)
157
+
let conn = Conpool.connection ~sw pool endpoint in
158
+
(* For HTTP/1.x: conn.state is () *)
159
+
(* For HTTP/2: conn.state is H2_client.t *)
160
+
Eio.Flow.copy_string data conn.flow)
159
161
]} *)
160
162
161
-
val with_connection : 'state t -> Endpoint.t -> ('state connection_info -> 'a) -> 'a
163
+
val with_connection :
164
+
'state t -> Endpoint.t -> ('state connection_info -> 'a) -> 'a
162
165
(** [with_connection pool endpoint fn] is a convenience wrapper.
163
166
164
167
Equivalent to:
+2
-1
test/dune
+2
-1
test/dune
+498
-413
test/stress_test.ml
+498
-413
test/stress_test.ml
···
5
5
6
6
(** Stress test framework for conpool
7
7
8
-
Spawns variable number of echo servers on random ports, then exercises
9
-
the connection pool with multiple parallel client fibers.
10
-
Collects detailed event traces for visualization.
11
-
*)
8
+
Spawns variable number of echo servers on random ports, then exercises the
9
+
connection pool with multiple parallel client fibers. Collects detailed
10
+
event traces for visualization. *)
12
11
13
-
(** Configuration for the stress test *)
14
12
type config = {
15
-
name : string; (** Test name for identification *)
16
-
num_servers : int; (** Number of echo servers to spawn *)
17
-
num_clients : int; (** Number of client connections per server *)
13
+
name : string; (** Test name for identification *)
14
+
num_servers : int; (** Number of echo servers to spawn *)
15
+
num_clients : int; (** Number of client connections per server *)
18
16
messages_per_client : int; (** Number of messages each client sends *)
19
-
max_parallel_clients : int; (** Maximum concurrent client fibers *)
20
-
message_size : int; (** Size of each message in bytes *)
21
-
pool_size : int; (** Max connections per endpoint *)
17
+
max_parallel_clients : int; (** Maximum concurrent client fibers *)
18
+
message_size : int; (** Size of each message in bytes *)
19
+
pool_size : int; (** Max connections per endpoint *)
22
20
}
21
+
(** Configuration for the stress test *)
23
22
24
-
let default_config = {
25
-
name = "default";
26
-
num_servers = 3;
27
-
num_clients = 10;
28
-
messages_per_client = 5;
29
-
max_parallel_clients = 20;
30
-
message_size = 64;
31
-
pool_size = 5;
32
-
}
33
-
34
-
(** Test presets for different scenarios *)
35
-
let presets = [
36
-
(* High connection reuse - few connections, many messages *)
37
-
{ name = "high_reuse";
38
-
num_servers = 2;
39
-
num_clients = 20;
40
-
messages_per_client = 50;
41
-
max_parallel_clients = 10;
42
-
message_size = 32;
43
-
pool_size = 3;
44
-
};
45
-
(* Many endpoints - test endpoint scaling *)
46
-
{ name = "many_endpoints";
47
-
num_servers = 10;
48
-
num_clients = 10;
49
-
messages_per_client = 10;
50
-
max_parallel_clients = 50;
51
-
message_size = 64;
52
-
pool_size = 5;
53
-
};
54
-
(* High concurrency - stress parallel connections *)
55
-
{ name = "high_concurrency";
23
+
let default_config =
24
+
{
25
+
name = "default";
56
26
num_servers = 3;
57
-
num_clients = 100;
27
+
num_clients = 10;
58
28
messages_per_client = 5;
59
-
max_parallel_clients = 100;
29
+
max_parallel_clients = 20;
60
30
message_size = 64;
61
-
pool_size = 20;
62
-
};
63
-
(* Large messages - test throughput *)
64
-
{ name = "large_messages";
65
-
num_servers = 3;
66
-
num_clients = 20;
67
-
messages_per_client = 20;
68
-
max_parallel_clients = 30;
69
-
message_size = 1024;
70
-
pool_size = 10;
71
-
};
72
-
(* Constrained pool - force queuing *)
73
-
{ name = "constrained_pool";
74
-
num_servers = 2;
75
-
num_clients = 50;
76
-
messages_per_client = 10;
77
-
max_parallel_clients = 50;
78
-
message_size = 64;
79
-
pool_size = 2;
80
-
};
81
-
(* Burst traffic - many clients, few messages each *)
82
-
{ name = "burst_traffic";
83
-
num_servers = 5;
84
-
num_clients = 200;
85
-
messages_per_client = 2;
86
-
max_parallel_clients = 100;
87
-
message_size = 32;
88
-
pool_size = 15;
89
-
};
90
-
]
31
+
pool_size = 5;
32
+
}
33
+
34
+
(** Test presets for different scenarios *)
35
+
let presets =
36
+
[
37
+
(* High connection reuse - few connections, many messages *)
38
+
{
39
+
name = "high_reuse";
40
+
num_servers = 2;
41
+
num_clients = 20;
42
+
messages_per_client = 50;
43
+
max_parallel_clients = 10;
44
+
message_size = 32;
45
+
pool_size = 3;
46
+
};
47
+
(* Many endpoints - test endpoint scaling *)
48
+
{
49
+
name = "many_endpoints";
50
+
num_servers = 10;
51
+
num_clients = 10;
52
+
messages_per_client = 10;
53
+
max_parallel_clients = 50;
54
+
message_size = 64;
55
+
pool_size = 5;
56
+
};
57
+
(* High concurrency - stress parallel connections *)
58
+
{
59
+
name = "high_concurrency";
60
+
num_servers = 3;
61
+
num_clients = 100;
62
+
messages_per_client = 5;
63
+
max_parallel_clients = 100;
64
+
message_size = 64;
65
+
pool_size = 20;
66
+
};
67
+
(* Large messages - test throughput *)
68
+
{
69
+
name = "large_messages";
70
+
num_servers = 3;
71
+
num_clients = 20;
72
+
messages_per_client = 20;
73
+
max_parallel_clients = 30;
74
+
message_size = 1024;
75
+
pool_size = 10;
76
+
};
77
+
(* Constrained pool - force queuing *)
78
+
{
79
+
name = "constrained_pool";
80
+
num_servers = 2;
81
+
num_clients = 50;
82
+
messages_per_client = 10;
83
+
max_parallel_clients = 50;
84
+
message_size = 64;
85
+
pool_size = 2;
86
+
};
87
+
(* Burst traffic - many clients, few messages each *)
88
+
{
89
+
name = "burst_traffic";
90
+
num_servers = 5;
91
+
num_clients = 200;
92
+
messages_per_client = 2;
93
+
max_parallel_clients = 100;
94
+
message_size = 32;
95
+
pool_size = 15;
96
+
};
97
+
]
91
98
92
99
(** Extended stress test - 100x messages, 10x clients/servers *)
93
-
let extended_preset = {
94
-
name = "extended_stress";
95
-
num_servers = 30;
96
-
num_clients = 1000;
97
-
messages_per_client = 100;
98
-
max_parallel_clients = 500;
99
-
message_size = 128;
100
-
pool_size = 50;
101
-
}
100
+
let extended_preset =
101
+
{
102
+
name = "extended_stress";
103
+
num_servers = 30;
104
+
num_clients = 1000;
105
+
messages_per_client = 100;
106
+
max_parallel_clients = 500;
107
+
message_size = 128;
108
+
pool_size = 50;
109
+
}
102
110
103
-
(** Statistics collected during test *)
104
111
type latency_stats = {
105
112
mutable count : int;
106
113
mutable total : float;
···
108
115
mutable max : float;
109
116
mutable latencies : (float * float) list; (* (timestamp, latency) pairs *)
110
117
}
118
+
(** Statistics collected during test *)
111
119
112
-
let create_latency_stats () = {
113
-
count = 0;
114
-
total = 0.0;
115
-
min = Float.infinity;
116
-
max = 0.0;
117
-
latencies = [];
118
-
}
120
+
let create_latency_stats () =
121
+
{ count = 0; total = 0.0; min = Float.infinity; max = 0.0; latencies = [] }
119
122
120
123
let update_latency stats latency timestamp =
121
124
stats.count <- stats.count + 1;
···
126
129
127
130
(** Generate a random message of given size *)
128
131
let generate_message size =
129
-
let chars = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789" in
132
+
let chars =
133
+
"abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789"
134
+
in
130
135
let len = String.length chars in
131
136
String.init size (fun _ -> chars.[Random.int len])
132
137
···
137
142
match Eio.Flow.single_read flow buf with
138
143
| n ->
139
144
let data = Cstruct.sub buf 0 n in
140
-
Eio.Flow.write flow [data];
145
+
Eio.Flow.write flow [ data ];
141
146
loop ()
142
147
| exception End_of_file -> ()
143
148
in
···
146
151
(** Start an echo server on a random port, returns the port number *)
147
152
let start_echo_server ~sw net =
148
153
let addr = `Tcp (Eio.Net.Ipaddr.V4.loopback, 0) in
149
-
let listening_socket = Eio.Net.listen net ~sw ~backlog:128 ~reuse_addr:true addr in
154
+
let listening_socket =
155
+
Eio.Net.listen net ~sw ~backlog:128 ~reuse_addr:true addr
156
+
in
150
157
let actual_addr = Eio.Net.listening_addr listening_socket in
151
-
let port = match actual_addr with
158
+
let port =
159
+
match actual_addr with
152
160
| `Tcp (_, port) -> port
153
161
| _ -> failwith "Expected TCP address"
154
162
in
155
163
156
164
Eio.Fiber.fork_daemon ~sw (fun () ->
157
-
try
158
-
while true do
159
-
Eio.Net.accept_fork ~sw listening_socket
160
-
~on_error:(fun _ -> ())
161
-
handle_echo_client
162
-
done;
163
-
`Stop_daemon
164
-
with Eio.Cancel.Cancelled _ ->
165
-
`Stop_daemon
166
-
);
165
+
try
166
+
while true do
167
+
Eio.Net.accept_fork ~sw listening_socket
168
+
~on_error:(fun _ -> ())
169
+
handle_echo_client
170
+
done;
171
+
`Stop_daemon
172
+
with Eio.Cancel.Cancelled _ -> `Stop_daemon);
167
173
168
174
port
169
175
170
176
(** Client test: connect via pool, send message, verify echo *)
171
-
let run_client_test ~clock ~test_start_time pool endpoint message latency_stats errors =
177
+
let run_client_test ~clock ~test_start_time pool endpoint message latency_stats
178
+
errors =
172
179
let msg_len = String.length message in
173
180
let start_time = Eio.Time.now clock in
174
181
175
182
try
176
183
Conpool.with_connection pool endpoint (fun conn ->
177
-
(* Send message *)
178
-
Eio.Flow.copy_string message conn.Conpool.flow;
179
-
Eio.Flow.copy_string "\n" conn.Conpool.flow;
184
+
(* Send message *)
185
+
Eio.Flow.copy_string message conn.Conpool.flow;
186
+
Eio.Flow.copy_string "\n" conn.Conpool.flow;
180
187
181
-
(* Read echo response *)
182
-
let response = Eio.Buf_read.of_flow conn.Conpool.flow ~max_size:(msg_len + 1) in
183
-
let echoed = Eio.Buf_read.line response in
188
+
(* Read echo response *)
189
+
let response =
190
+
Eio.Buf_read.of_flow conn.Conpool.flow ~max_size:(msg_len + 1)
191
+
in
192
+
let echoed = Eio.Buf_read.line response in
184
193
185
-
let end_time = Eio.Time.now clock in
186
-
let latency = (end_time -. start_time) *. 1000.0 in (* Convert to ms *)
187
-
let relative_time = (end_time -. test_start_time) *. 1000.0 in (* ms since test start *)
194
+
let end_time = Eio.Time.now clock in
195
+
let latency = (end_time -. start_time) *. 1000.0 in
196
+
(* Convert to ms *)
197
+
let relative_time = (end_time -. test_start_time) *. 1000.0 in
198
+
(* ms since test start *)
188
199
189
-
if String.equal echoed message then begin
190
-
update_latency latency_stats latency relative_time
191
-
end else begin
192
-
incr errors
193
-
end
194
-
)
195
-
with _ex ->
196
-
incr errors
200
+
if String.equal echoed message then begin
201
+
update_latency latency_stats latency relative_time
202
+
end
203
+
else begin
204
+
incr errors
205
+
end)
206
+
with _ex -> incr errors
197
207
198
208
(** Run a single client that sends multiple messages *)
199
-
let run_client ~clock ~test_start_time pool endpoints (cfg : config) latency_stats errors client_id =
209
+
let run_client ~clock ~test_start_time pool endpoints (cfg : config)
210
+
latency_stats errors client_id =
200
211
for _ = 1 to cfg.messages_per_client do
201
212
let endpoint_idx = Random.int (Array.length endpoints) in
202
213
let endpoint = endpoints.(endpoint_idx) in
203
-
let message = Printf.sprintf "c%d-%s" client_id (generate_message cfg.message_size) in
204
-
run_client_test ~clock ~test_start_time pool endpoint message latency_stats errors
214
+
let message =
215
+
Printf.sprintf "c%d-%s" client_id (generate_message cfg.message_size)
216
+
in
217
+
run_client_test ~clock ~test_start_time pool endpoint message latency_stats
218
+
errors
205
219
done
206
220
207
-
(** Pool statistics aggregated from all endpoints *)
208
221
type pool_stats = {
209
222
total_created : int;
210
223
total_reused : int;
···
213
226
idle : int;
214
227
pool_errors : int;
215
228
}
229
+
(** Pool statistics aggregated from all endpoints *)
216
230
217
-
(** Test result type *)
218
231
type test_result = {
219
232
test_name : string;
220
233
num_servers : int;
···
228
241
avg_latency : float;
229
242
min_latency : float;
230
243
max_latency : float;
231
-
latency_data : (float * float) list; (* (timestamp, latency) pairs for visualization *)
244
+
latency_data : (float * float) list;
245
+
(* (timestamp, latency) pairs for visualization *)
232
246
pool_stats : pool_stats;
233
247
}
248
+
(** Test result type *)
234
249
235
250
(** Main stress test runner - returns a test result *)
236
251
let run_stress_test ~env (cfg : config) : test_result =
···
243
258
244
259
let result : test_result option ref = ref None in
245
260
246
-
begin
247
-
try
248
-
Eio.Switch.run @@ fun sw ->
249
-
(* Start echo servers *)
250
-
ports := Array.init cfg.num_servers (fun _ ->
251
-
start_echo_server ~sw net
252
-
);
261
+
begin try
262
+
Eio.Switch.run @@ fun sw ->
263
+
(* Start echo servers *)
264
+
ports := Array.init cfg.num_servers (fun _ -> start_echo_server ~sw net);
253
265
254
-
Eio.Time.sleep clock 0.05;
266
+
Eio.Time.sleep clock 0.05;
255
267
256
-
let endpoints = Array.map (fun port ->
257
-
Conpool.Endpoint.make ~host:"127.0.0.1" ~port
258
-
) !ports in
268
+
let endpoints =
269
+
Array.map
270
+
(fun port -> Conpool.Endpoint.make ~host:"127.0.0.1" ~port)
271
+
!ports
272
+
in
259
273
260
-
(* Create connection pool *)
261
-
let pool_config = Conpool.Config.make
262
-
~max_connections_per_endpoint:cfg.pool_size
263
-
~max_idle_time:30.0
264
-
~max_connection_lifetime:120.0
265
-
~connect_timeout:5.0
266
-
~connect_retry_count:3
267
-
()
268
-
in
274
+
(* Create connection pool *)
275
+
let pool_config =
276
+
Conpool.Config.make ~max_connections_per_endpoint:cfg.pool_size
277
+
~max_idle_time:30.0 ~max_connection_lifetime:120.0 ~connect_timeout:5.0
278
+
~connect_retry_count:3 ()
279
+
in
269
280
270
-
let pool = Conpool.create ~sw ~net ~clock ~config:pool_config () in
281
+
let pool = Conpool.create ~sw ~net ~clock ~config:pool_config () in
271
282
272
-
(* Record start time *)
273
-
let start_time = Eio.Time.now clock in
283
+
(* Record start time *)
284
+
let start_time = Eio.Time.now clock in
274
285
275
-
(* Run clients in parallel *)
276
-
let total_clients = cfg.num_servers * cfg.num_clients in
277
-
let client_ids = List.init total_clients (fun i -> i) in
278
-
Eio.Fiber.List.iter ~max_fibers:cfg.max_parallel_clients
279
-
(fun client_id ->
280
-
run_client ~clock ~test_start_time:start_time pool endpoints cfg latency_stats errors client_id)
281
-
client_ids;
286
+
(* Run clients in parallel *)
287
+
let total_clients = cfg.num_servers * cfg.num_clients in
288
+
let client_ids = List.init total_clients (fun i -> i) in
289
+
Eio.Fiber.List.iter ~max_fibers:cfg.max_parallel_clients
290
+
(fun client_id ->
291
+
run_client ~clock ~test_start_time:start_time pool endpoints cfg
292
+
latency_stats errors client_id)
293
+
client_ids;
282
294
283
-
let end_time = Eio.Time.now clock in
284
-
let duration = end_time -. start_time in
295
+
let end_time = Eio.Time.now clock in
296
+
let duration = end_time -. start_time in
285
297
286
-
(* Collect pool statistics from all endpoints *)
287
-
let all_stats = Conpool.all_stats pool in
288
-
let pool_stats = List.fold_left (fun acc (_, stats) ->
298
+
(* Collect pool statistics from all endpoints *)
299
+
let all_stats = Conpool.all_stats pool in
300
+
let pool_stats =
301
+
List.fold_left
302
+
(fun acc (_, stats) ->
289
303
{
290
-
total_created = acc.total_created + Conpool.Stats.total_created stats;
304
+
total_created =
305
+
acc.total_created + Conpool.Stats.total_created stats;
291
306
total_reused = acc.total_reused + Conpool.Stats.total_reused stats;
292
307
total_closed = acc.total_closed + Conpool.Stats.total_closed stats;
293
308
active = acc.active + Conpool.Stats.active stats;
294
309
idle = acc.idle + Conpool.Stats.idle stats;
295
310
pool_errors = acc.pool_errors + Conpool.Stats.errors stats;
296
-
}
297
-
) { total_created = 0; total_reused = 0; total_closed = 0; active = 0; idle = 0; pool_errors = 0 } all_stats in
311
+
})
312
+
{
313
+
total_created = 0;
314
+
total_reused = 0;
315
+
total_closed = 0;
316
+
active = 0;
317
+
idle = 0;
318
+
pool_errors = 0;
319
+
}
320
+
all_stats
321
+
in
298
322
299
-
(* Build result *)
300
-
let r : test_result = {
301
-
test_name = cfg.name;
302
-
num_servers = cfg.num_servers;
303
-
num_clients = cfg.num_clients;
304
-
messages_per_client = cfg.messages_per_client;
305
-
pool_size = cfg.pool_size;
306
-
duration;
307
-
total_messages = latency_stats.count;
308
-
total_errors = !errors;
309
-
throughput = float_of_int latency_stats.count /. duration;
310
-
avg_latency = if latency_stats.count > 0
311
-
then latency_stats.total /. float_of_int latency_stats.count
312
-
else 0.0;
313
-
min_latency = if latency_stats.count > 0 then latency_stats.min else 0.0;
314
-
max_latency = latency_stats.max;
315
-
latency_data = List.rev latency_stats.latencies;
316
-
pool_stats;
317
-
} in
318
-
result := Some r;
323
+
(* Build result *)
324
+
let r : test_result =
325
+
{
326
+
test_name = cfg.name;
327
+
num_servers = cfg.num_servers;
328
+
num_clients = cfg.num_clients;
329
+
messages_per_client = cfg.messages_per_client;
330
+
pool_size = cfg.pool_size;
331
+
duration;
332
+
total_messages = latency_stats.count;
333
+
total_errors = !errors;
334
+
throughput = float_of_int latency_stats.count /. duration;
335
+
avg_latency =
336
+
(if latency_stats.count > 0 then
337
+
latency_stats.total /. float_of_int latency_stats.count
338
+
else 0.0);
339
+
min_latency =
340
+
(if latency_stats.count > 0 then latency_stats.min else 0.0);
341
+
max_latency = latency_stats.max;
342
+
latency_data = List.rev latency_stats.latencies;
343
+
pool_stats;
344
+
}
345
+
in
346
+
result := Some r;
319
347
320
-
Eio.Switch.fail sw Exit
321
-
with Exit -> ()
348
+
Eio.Switch.fail sw Exit
349
+
with Exit -> ()
322
350
end;
323
351
324
352
match !result with
···
327
355
328
356
(** Convert result to JSON string *)
329
357
let result_to_json result =
330
-
Printf.sprintf {|{
358
+
Printf.sprintf
359
+
{|{
331
360
"test_name": "%s",
332
361
"num_servers": %d,
333
362
"num_clients": %d,
···
340
369
"min_latency": %.2f,
341
370
"max_latency": %.2f
342
371
}|}
343
-
result.test_name
344
-
result.num_servers
345
-
result.num_clients
346
-
result.messages_per_client
347
-
result.duration
348
-
result.total_messages
349
-
result.total_errors
350
-
result.throughput
351
-
result.avg_latency
352
-
result.min_latency
372
+
result.test_name result.num_servers result.num_clients
373
+
result.messages_per_client result.duration result.total_messages
374
+
result.total_errors result.throughput result.avg_latency result.min_latency
353
375
result.max_latency
354
376
355
377
(** Escape strings for JavaScript *)
356
378
let js_escape s =
357
379
let buf = Buffer.create (String.length s) in
358
-
String.iter (fun c ->
359
-
match c with
360
-
| '\\' -> Buffer.add_string buf "\\\\"
361
-
| '"' -> Buffer.add_string buf "\\\""
362
-
| '\n' -> Buffer.add_string buf "\\n"
363
-
| '\r' -> Buffer.add_string buf "\\r"
364
-
| '\t' -> Buffer.add_string buf "\\t"
365
-
| _ -> Buffer.add_char buf c
366
-
) s;
380
+
String.iter
381
+
(fun c ->
382
+
match c with
383
+
| '\\' -> Buffer.add_string buf "\\\\"
384
+
| '"' -> Buffer.add_string buf "\\\""
385
+
| '\n' -> Buffer.add_string buf "\\n"
386
+
| '\r' -> Buffer.add_string buf "\\r"
387
+
| '\t' -> Buffer.add_string buf "\\t"
388
+
| _ -> Buffer.add_char buf c)
389
+
s;
367
390
Buffer.contents buf
368
391
369
392
(** Calculate histogram buckets for latency data *)
370
393
let calculate_histogram latencies num_buckets =
371
-
if List.length latencies = 0 then ([], []) else
372
-
let latency_values = List.map snd latencies in
373
-
let min_lat = List.fold_left min Float.infinity latency_values in
374
-
let max_lat = List.fold_left max 0.0 latency_values in
375
-
let bucket_width = (max_lat -. min_lat) /. float_of_int num_buckets in
394
+
if List.length latencies = 0 then ([], [])
395
+
else
396
+
let latency_values = List.map snd latencies in
397
+
let min_lat = List.fold_left min Float.infinity latency_values in
398
+
let max_lat = List.fold_left max 0.0 latency_values in
399
+
let bucket_width = (max_lat -. min_lat) /. float_of_int num_buckets in
376
400
377
-
let buckets = Array.make num_buckets 0 in
378
-
List.iter (fun lat ->
379
-
let bucket_idx = min (num_buckets - 1) (int_of_float ((lat -. min_lat) /. bucket_width)) in
380
-
buckets.(bucket_idx) <- buckets.(bucket_idx) + 1
381
-
) latency_values;
401
+
let buckets = Array.make num_buckets 0 in
402
+
List.iter
403
+
(fun lat ->
404
+
let bucket_idx =
405
+
min (num_buckets - 1)
406
+
(int_of_float ((lat -. min_lat) /. bucket_width))
407
+
in
408
+
buckets.(bucket_idx) <- buckets.(bucket_idx) + 1)
409
+
latency_values;
382
410
383
-
let bucket_labels = List.init num_buckets (fun i ->
384
-
let start = min_lat +. (float_of_int i *. bucket_width) in
385
-
Printf.sprintf "%.2f" start
386
-
) in
387
-
let bucket_counts = Array.to_list buckets in
388
-
(bucket_labels, bucket_counts)
411
+
let bucket_labels =
412
+
List.init num_buckets (fun i ->
413
+
let start = min_lat +. (float_of_int i *. bucket_width) in
414
+
Printf.sprintf "%.2f" start)
415
+
in
416
+
let bucket_counts = Array.to_list buckets in
417
+
(bucket_labels, bucket_counts)
389
418
390
419
(** Generate HTML report from test results *)
391
420
let generate_html_report results =
392
421
let timestamp = Unix.time () |> Unix.gmtime in
393
-
let date_str = Printf.sprintf "%04d-%02d-%02d %02d:%02d:%02d UTC"
394
-
(timestamp.Unix.tm_year + 1900)
395
-
(timestamp.Unix.tm_mon + 1)
396
-
timestamp.Unix.tm_mday
397
-
timestamp.Unix.tm_hour
398
-
timestamp.Unix.tm_min
399
-
timestamp.Unix.tm_sec
422
+
let date_str =
423
+
Printf.sprintf "%04d-%02d-%02d %02d:%02d:%02d UTC"
424
+
(timestamp.Unix.tm_year + 1900)
425
+
(timestamp.Unix.tm_mon + 1)
426
+
timestamp.Unix.tm_mday timestamp.Unix.tm_hour timestamp.Unix.tm_min
427
+
timestamp.Unix.tm_sec
400
428
in
401
429
402
430
(* Calculate summary statistics *)
403
-
let total_messages = List.fold_left (fun acc r -> acc + r.total_messages) 0 results in
404
-
let total_errors = List.fold_left (fun acc r -> acc + r.total_errors) 0 results in
405
-
let total_duration = List.fold_left (fun acc r -> acc +. r.duration) 0.0 results in
431
+
let total_messages =
432
+
List.fold_left (fun acc r -> acc + r.total_messages) 0 results
433
+
in
434
+
let total_errors =
435
+
List.fold_left (fun acc r -> acc + r.total_errors) 0 results
436
+
in
437
+
let total_duration =
438
+
List.fold_left (fun acc r -> acc +. r.duration) 0.0 results
439
+
in
406
440
407
441
(* Generate JavaScript arrays for comparison charts *)
408
-
let test_names = String.concat ", " (List.map (fun r -> Printf.sprintf "\"%s\"" (js_escape r.test_name)) results) in
409
-
let throughputs = String.concat ", " (List.map (fun r -> Printf.sprintf "%.2f" r.throughput) results) in
410
-
let avg_latencies = String.concat ", " (List.map (fun r -> Printf.sprintf "%.2f" r.avg_latency) results) in
411
-
let error_rates = String.concat ", " (List.map (fun r ->
412
-
if r.total_messages > 0 then
413
-
Printf.sprintf "%.2f" (float_of_int r.total_errors /. float_of_int r.total_messages *. 100.0)
414
-
else "0.0"
415
-
) results) in
442
+
let test_names =
443
+
String.concat ", "
444
+
(List.map
445
+
(fun r -> Printf.sprintf "\"%s\"" (js_escape r.test_name))
446
+
results)
447
+
in
448
+
let throughputs =
449
+
String.concat ", "
450
+
(List.map (fun r -> Printf.sprintf "%.2f" r.throughput) results)
451
+
in
452
+
let avg_latencies =
453
+
String.concat ", "
454
+
(List.map (fun r -> Printf.sprintf "%.2f" r.avg_latency) results)
455
+
in
456
+
let error_rates =
457
+
String.concat ", "
458
+
(List.map
459
+
(fun r ->
460
+
if r.total_messages > 0 then
461
+
Printf.sprintf "%.2f"
462
+
(float_of_int r.total_errors
463
+
/. float_of_int r.total_messages
464
+
*. 100.0)
465
+
else "0.0")
466
+
results)
467
+
in
416
468
417
469
(* Generate per-test detailed sections with histograms and timelines *)
418
-
let test_details = String.concat "\n" (List.mapi (fun idx r ->
419
-
let (hist_labels, hist_counts) = calculate_histogram r.latency_data 20 in
420
-
let hist_labels_str = String.concat ", " (List.map (fun s -> Printf.sprintf "\"%s\"" s) hist_labels) in
421
-
let hist_counts_str = String.concat ", " (List.map string_of_int hist_counts) in
470
+
let test_details =
471
+
String.concat "\n"
472
+
(List.mapi
473
+
(fun idx r ->
474
+
let hist_labels, hist_counts =
475
+
calculate_histogram r.latency_data 20
476
+
in
477
+
let hist_labels_str =
478
+
String.concat ", "
479
+
(List.map (fun s -> Printf.sprintf "\"%s\"" s) hist_labels)
480
+
in
481
+
let hist_counts_str =
482
+
String.concat ", " (List.map string_of_int hist_counts)
483
+
in
422
484
423
-
(* Sample data points for timeline (take every Nth point if too many) *)
424
-
let max_points = 500 in
425
-
let sample_rate = max 1 ((List.length r.latency_data) / max_points) in
426
-
let sampled_data = List.filteri (fun i _ -> i mod sample_rate = 0) r.latency_data in
427
-
let timeline_data = String.concat ", " (List.map (fun (t, l) ->
428
-
Printf.sprintf "{x: %.2f, y: %.3f}" t l
429
-
) sampled_data) in
485
+
(* Sample data points for timeline (take every Nth point if too many) *)
486
+
let max_points = 500 in
487
+
let sample_rate = max 1 (List.length r.latency_data / max_points) in
488
+
let sampled_data =
489
+
List.filteri (fun i _ -> i mod sample_rate = 0) r.latency_data
490
+
in
491
+
let timeline_data =
492
+
String.concat ", "
493
+
(List.map
494
+
(fun (t, l) -> Printf.sprintf "{x: %.2f, y: %.3f}" t l)
495
+
sampled_data)
496
+
in
430
497
431
-
Printf.sprintf {|
498
+
Printf.sprintf
499
+
{|
432
500
<div class="test-detail">
433
501
<h3>%s</h3>
434
502
<div class="compact-grid">
···
509
577
}
510
578
});
511
579
</script>|}
512
-
(js_escape r.test_name)
513
-
r.num_servers
514
-
r.num_clients
515
-
r.messages_per_client
516
-
r.pool_size
517
-
r.total_messages
518
-
r.duration
519
-
r.throughput
520
-
r.avg_latency
521
-
r.min_latency
522
-
r.max_latency
523
-
(if r.total_errors > 0 then "error" else "")
524
-
r.total_errors
525
-
r.pool_stats.total_created
526
-
r.pool_stats.total_reused
527
-
r.pool_stats.total_closed
528
-
r.pool_stats.active
529
-
r.pool_stats.idle
530
-
(if r.pool_stats.total_created > 0 then
531
-
(float_of_int r.pool_stats.total_reused /. float_of_int r.pool_stats.total_created *. 100.0)
532
-
else 0.0)
533
-
idx idx idx
534
-
hist_labels_str
535
-
hist_counts_str
536
-
idx
537
-
timeline_data
538
-
) results) in
580
+
(js_escape r.test_name) r.num_servers r.num_clients
581
+
r.messages_per_client r.pool_size r.total_messages r.duration
582
+
r.throughput r.avg_latency r.min_latency r.max_latency
583
+
(if r.total_errors > 0 then "error" else "")
584
+
r.total_errors r.pool_stats.total_created r.pool_stats.total_reused
585
+
r.pool_stats.total_closed r.pool_stats.active r.pool_stats.idle
586
+
(if r.pool_stats.total_created > 0 then
587
+
float_of_int r.pool_stats.total_reused
588
+
/. float_of_int r.pool_stats.total_created
589
+
*. 100.0
590
+
else 0.0)
591
+
idx idx idx hist_labels_str hist_counts_str idx timeline_data)
592
+
results)
593
+
in
539
594
540
-
Printf.sprintf {|<!DOCTYPE html>
595
+
Printf.sprintf
596
+
{|<!DOCTYPE html>
541
597
<html lang="en">
542
598
<head>
543
599
<meta charset="UTF-8">
···
814
870
</script>
815
871
</body>
816
872
</html>|}
817
-
date_str
818
-
(List.length results)
873
+
date_str (List.length results)
819
874
(if total_messages >= 1000 then
820
-
Printf.sprintf "%d,%03d" (total_messages / 1000) (total_messages mod 1000)
821
-
else
822
-
string_of_int total_messages)
823
-
total_errors
824
-
total_duration
825
-
test_details
826
-
test_names
827
-
throughputs
828
-
avg_latencies
829
-
error_rates
875
+
Printf.sprintf "%d,%03d" (total_messages / 1000) (total_messages mod 1000)
876
+
else string_of_int total_messages)
877
+
total_errors total_duration test_details test_names throughputs
878
+
avg_latencies error_rates
830
879
831
880
(** Run all preset tests and return results *)
832
881
let run_all_presets ~env =
833
-
List.map (fun config ->
834
-
Printf.eprintf "Running test: %s\n%!" config.name;
835
-
run_stress_test ~env config
836
-
) presets
882
+
List.map
883
+
(fun config ->
884
+
Printf.eprintf "Running test: %s\n%!" config.name;
885
+
run_stress_test ~env config)
886
+
presets
837
887
838
888
(** Parse command line arguments *)
839
-
type mode =
840
-
| Single of config
841
-
| AllPresets
842
-
| Extended
843
-
| ListPresets
889
+
type mode = Single of config | AllPresets | Extended | ListPresets
844
890
845
891
let parse_args () =
846
892
let mode = ref (Single default_config) in
···
853
899
let pool_size = ref default_config.pool_size in
854
900
let output_file = ref "stress_test_results.json" in
855
901
856
-
let specs = [
857
-
("--all", Arg.Unit (fun () -> mode := AllPresets),
858
-
"Run all preset test configurations");
859
-
("--extended", Arg.Unit (fun () -> mode := Extended),
860
-
"Run extended stress test (30 servers, 1000 clients, 100 msgs each = 3M messages)");
861
-
("--list", Arg.Unit (fun () -> mode := ListPresets),
862
-
"List available presets");
863
-
("--preset", Arg.String (fun p ->
864
-
match List.find_opt (fun c -> c.name = p) presets with
865
-
| Some c -> mode := Single c
866
-
| None -> failwith (Printf.sprintf "Unknown preset: %s" p)),
867
-
"Use a named preset configuration");
868
-
("-n", Arg.Set_string name, "Test name");
869
-
("-s", Arg.Set_int num_servers, Printf.sprintf "Number of servers (default: %d)" default_config.num_servers);
870
-
("-c", Arg.Set_int num_clients, Printf.sprintf "Clients per server (default: %d)" default_config.num_clients);
871
-
("-m", Arg.Set_int messages_per_client, Printf.sprintf "Messages per client (default: %d)" default_config.messages_per_client);
872
-
("-p", Arg.Set_int max_parallel, Printf.sprintf "Max parallel clients (default: %d)" default_config.max_parallel_clients);
873
-
("-b", Arg.Set_int message_size, Printf.sprintf "Message size (default: %d)" default_config.message_size);
874
-
("-P", Arg.Set_int pool_size, Printf.sprintf "Pool size per endpoint (default: %d)" default_config.pool_size);
875
-
("-o", Arg.Set_string output_file, "Output JSON file (default: stress_test_results.json)");
876
-
] in
902
+
let specs =
903
+
[
904
+
( "--all",
905
+
Arg.Unit (fun () -> mode := AllPresets),
906
+
"Run all preset test configurations" );
907
+
( "--extended",
908
+
Arg.Unit (fun () -> mode := Extended),
909
+
"Run extended stress test (30 servers, 1000 clients, 100 msgs each = \
910
+
3M messages)" );
911
+
( "--list",
912
+
Arg.Unit (fun () -> mode := ListPresets),
913
+
"List available presets" );
914
+
( "--preset",
915
+
Arg.String
916
+
(fun p ->
917
+
match List.find_opt (fun c -> c.name = p) presets with
918
+
| Some c -> mode := Single c
919
+
| None -> failwith (Printf.sprintf "Unknown preset: %s" p)),
920
+
"Use a named preset configuration" );
921
+
("-n", Arg.Set_string name, "Test name");
922
+
( "-s",
923
+
Arg.Set_int num_servers,
924
+
Printf.sprintf "Number of servers (default: %d)"
925
+
default_config.num_servers );
926
+
( "-c",
927
+
Arg.Set_int num_clients,
928
+
Printf.sprintf "Clients per server (default: %d)"
929
+
default_config.num_clients );
930
+
( "-m",
931
+
Arg.Set_int messages_per_client,
932
+
Printf.sprintf "Messages per client (default: %d)"
933
+
default_config.messages_per_client );
934
+
( "-p",
935
+
Arg.Set_int max_parallel,
936
+
Printf.sprintf "Max parallel clients (default: %d)"
937
+
default_config.max_parallel_clients );
938
+
( "-b",
939
+
Arg.Set_int message_size,
940
+
Printf.sprintf "Message size (default: %d)" default_config.message_size
941
+
);
942
+
( "-P",
943
+
Arg.Set_int pool_size,
944
+
Printf.sprintf "Pool size per endpoint (default: %d)"
945
+
default_config.pool_size );
946
+
( "-o",
947
+
Arg.Set_string output_file,
948
+
"Output JSON file (default: stress_test_results.json)" );
949
+
]
950
+
in
877
951
878
952
let usage = "Usage: stress_test [options]\n\nOptions:" in
879
953
Arg.parse specs (fun _ -> ()) usage;
880
954
881
-
let config = {
882
-
name = !name;
883
-
num_servers = !num_servers;
884
-
num_clients = !num_clients;
885
-
messages_per_client = !messages_per_client;
886
-
max_parallel_clients = !max_parallel;
887
-
message_size = !message_size;
888
-
pool_size = !pool_size;
889
-
} in
955
+
let config =
956
+
{
957
+
name = !name;
958
+
num_servers = !num_servers;
959
+
num_clients = !num_clients;
960
+
messages_per_client = !messages_per_client;
961
+
max_parallel_clients = !max_parallel;
962
+
message_size = !message_size;
963
+
pool_size = !pool_size;
964
+
}
965
+
in
890
966
891
967
(!mode, config, !output_file)
892
968
893
969
let () =
894
970
Random.self_init ();
895
-
let (mode, custom_config, output_file) = parse_args () in
971
+
let mode, custom_config, output_file = parse_args () in
896
972
897
973
match mode with
898
974
| ListPresets ->
899
975
Printf.printf "Available presets:\n";
900
-
List.iter (fun c ->
901
-
Printf.printf " %s: %d servers, %d clients, %d msgs/client, pool=%d\n"
902
-
c.name c.num_servers c.num_clients c.messages_per_client c.pool_size
903
-
) presets
904
-
976
+
List.iter
977
+
(fun c ->
978
+
Printf.printf
979
+
" %s: %d servers, %d clients, %d msgs/client, pool=%d\n" c.name
980
+
c.num_servers c.num_clients c.messages_per_client c.pool_size)
981
+
presets
905
982
| Single config ->
906
983
let config = if config.name = "default" then custom_config else config in
907
984
Eio_main.run @@ fun env ->
908
-
let result = run_stress_test ~env config in
909
-
let results = [result] in
985
+
let result = run_stress_test ~env config in
986
+
let results = [ result ] in
910
987
911
-
(* Write JSON *)
912
-
let json = Printf.sprintf "[%s]" (result_to_json result) in
913
-
let oc = open_out output_file in
914
-
output_string oc json;
915
-
close_out oc;
916
-
Printf.printf "Results written to %s\n" output_file;
988
+
(* Write JSON *)
989
+
let json = Printf.sprintf "[%s]" (result_to_json result) in
990
+
let oc = open_out output_file in
991
+
output_string oc json;
992
+
close_out oc;
993
+
Printf.printf "Results written to %s\n" output_file;
917
994
918
-
(* Write HTML *)
919
-
let html_file =
920
-
if Filename.check_suffix output_file ".json" then
921
-
Filename.chop_suffix output_file ".json" ^ ".html"
922
-
else
923
-
output_file ^ ".html"
924
-
in
925
-
let html = generate_html_report results in
926
-
let oc_html = open_out html_file in
927
-
output_string oc_html html;
928
-
close_out oc_html;
929
-
Printf.printf "HTML report written to %s\n" html_file;
995
+
(* Write HTML *)
996
+
let html_file =
997
+
if Filename.check_suffix output_file ".json" then
998
+
Filename.chop_suffix output_file ".json" ^ ".html"
999
+
else output_file ^ ".html"
1000
+
in
1001
+
let html = generate_html_report results in
1002
+
let oc_html = open_out html_file in
1003
+
output_string oc_html html;
1004
+
close_out oc_html;
1005
+
Printf.printf "HTML report written to %s\n" html_file;
930
1006
931
-
Printf.printf "Test: %s - %d messages, %.2f msg/s, %.2fms avg latency, %d errors\n"
932
-
result.test_name result.total_messages result.throughput result.avg_latency result.total_errors
933
-
1007
+
Printf.printf
1008
+
"Test: %s - %d messages, %.2f msg/s, %.2fms avg latency, %d errors\n"
1009
+
result.test_name result.total_messages result.throughput
1010
+
result.avg_latency result.total_errors
934
1011
| AllPresets ->
935
1012
Eio_main.run @@ fun env ->
936
-
let results = run_all_presets ~env in
1013
+
let results = run_all_presets ~env in
937
1014
938
-
(* Write JSON *)
939
-
let json = "[" ^ String.concat ",\n" (List.map result_to_json results) ^ "]" in
940
-
let oc = open_out output_file in
941
-
output_string oc json;
942
-
close_out oc;
943
-
Printf.printf "Results written to %s\n" output_file;
1015
+
(* Write JSON *)
1016
+
let json =
1017
+
"[" ^ String.concat ",\n" (List.map result_to_json results) ^ "]"
1018
+
in
1019
+
let oc = open_out output_file in
1020
+
output_string oc json;
1021
+
close_out oc;
1022
+
Printf.printf "Results written to %s\n" output_file;
944
1023
945
-
(* Write HTML *)
946
-
let html_file =
947
-
if Filename.check_suffix output_file ".json" then
948
-
Filename.chop_suffix output_file ".json" ^ ".html"
949
-
else
950
-
output_file ^ ".html"
951
-
in
952
-
let html = generate_html_report results in
953
-
let oc_html = open_out html_file in
954
-
output_string oc_html html;
955
-
close_out oc_html;
956
-
Printf.printf "HTML report written to %s\n" html_file;
1024
+
(* Write HTML *)
1025
+
let html_file =
1026
+
if Filename.check_suffix output_file ".json" then
1027
+
Filename.chop_suffix output_file ".json" ^ ".html"
1028
+
else output_file ^ ".html"
1029
+
in
1030
+
let html = generate_html_report results in
1031
+
let oc_html = open_out html_file in
1032
+
output_string oc_html html;
1033
+
close_out oc_html;
1034
+
Printf.printf "HTML report written to %s\n" html_file;
957
1035
958
-
List.iter (fun r ->
959
-
Printf.printf " %s: %d messages, %.2f msg/s, %.2fms avg latency, %d errors\n"
960
-
r.test_name r.total_messages r.throughput r.avg_latency r.total_errors
961
-
) results
962
-
1036
+
List.iter
1037
+
(fun r ->
1038
+
Printf.printf
1039
+
" %s: %d messages, %.2f msg/s, %.2fms avg latency, %d errors\n"
1040
+
r.test_name r.total_messages r.throughput r.avg_latency
1041
+
r.total_errors)
1042
+
results
963
1043
| Extended ->
964
-
Printf.printf "Running extended stress test: %d servers, %d clients/server, %d msgs/client\n"
965
-
extended_preset.num_servers extended_preset.num_clients extended_preset.messages_per_client;
1044
+
Printf.printf
1045
+
"Running extended stress test: %d servers, %d clients/server, %d \
1046
+
msgs/client\n"
1047
+
extended_preset.num_servers extended_preset.num_clients
1048
+
extended_preset.messages_per_client;
966
1049
Printf.printf "Total messages: %d\n%!"
967
-
(extended_preset.num_servers * extended_preset.num_clients * extended_preset.messages_per_client);
1050
+
(extended_preset.num_servers * extended_preset.num_clients
1051
+
* extended_preset.messages_per_client);
968
1052
Eio_main.run @@ fun env ->
969
-
let result = run_stress_test ~env extended_preset in
970
-
let results = [result] in
1053
+
let result = run_stress_test ~env extended_preset in
1054
+
let results = [ result ] in
971
1055
972
-
(* Write JSON *)
973
-
let json = Printf.sprintf "[%s]" (result_to_json result) in
974
-
let oc = open_out output_file in
975
-
output_string oc json;
976
-
close_out oc;
977
-
Printf.printf "Results written to %s\n" output_file;
1056
+
(* Write JSON *)
1057
+
let json = Printf.sprintf "[%s]" (result_to_json result) in
1058
+
let oc = open_out output_file in
1059
+
output_string oc json;
1060
+
close_out oc;
1061
+
Printf.printf "Results written to %s\n" output_file;
978
1062
979
-
(* Write HTML *)
980
-
let html_file =
981
-
if Filename.check_suffix output_file ".json" then
982
-
Filename.chop_suffix output_file ".json" ^ ".html"
983
-
else
984
-
output_file ^ ".html"
985
-
in
986
-
let html = generate_html_report results in
987
-
let oc_html = open_out html_file in
988
-
output_string oc_html html;
989
-
close_out oc_html;
990
-
Printf.printf "HTML report written to %s\n" html_file;
1063
+
(* Write HTML *)
1064
+
let html_file =
1065
+
if Filename.check_suffix output_file ".json" then
1066
+
Filename.chop_suffix output_file ".json" ^ ".html"
1067
+
else output_file ^ ".html"
1068
+
in
1069
+
let html = generate_html_report results in
1070
+
let oc_html = open_out html_file in
1071
+
output_string oc_html html;
1072
+
close_out oc_html;
1073
+
Printf.printf "HTML report written to %s\n" html_file;
991
1074
992
-
Printf.printf "Test: %s - %d messages, %.2f msg/s, %.2fms avg latency, %d errors\n"
993
-
result.test_name result.total_messages result.throughput result.avg_latency result.total_errors
1075
+
Printf.printf
1076
+
"Test: %s - %d messages, %.2f msg/s, %.2fms avg latency, %d errors\n"
1077
+
result.test_name result.total_messages result.throughput
1078
+
result.avg_latency result.total_errors