TCP/TLS connection pooling for Eio
Squashed 'ocaml-conpool/' content from commit 7c1075e git-subtree-split: 7c1075e3eb3dbb5ff0c51213f88a8eeeb4ebedf5
+2731
+17
.gitignore
+17
.gitignore
+1
.ocamlformat
+1
.ocamlformat
···
1
+
version=0.28.1
+54
.tangled/workflows/build.yml
+54
.tangled/workflows/build.yml
···
1
+
when:
2
+
- event: ["push", "pull_request"]
3
+
branch: ["main"]
4
+
5
+
engine: nixery
6
+
7
+
dependencies:
8
+
nixpkgs:
9
+
- shell
10
+
- stdenv
11
+
- findutils
12
+
- binutils
13
+
- libunwind
14
+
- ncurses
15
+
- opam
16
+
- git
17
+
- gawk
18
+
- gnupatch
19
+
- gnum4
20
+
- gnumake
21
+
- gnutar
22
+
- gnused
23
+
- gnugrep
24
+
- diffutils
25
+
- gzip
26
+
- bzip2
27
+
- gcc
28
+
- ocaml
29
+
- pkg-config
30
+
31
+
steps:
32
+
- name: opam
33
+
command: |
34
+
opam init --disable-sandboxing -a -y
35
+
- name: repo
36
+
command: |
37
+
opam repo add aoah https://tangled.org/anil.recoil.org/aoah-opam-repo.git
38
+
- name: switch
39
+
command: |
40
+
export PKG_CONFIG_PATH="${PKG_CONFIG_PATH}:$(nix build nixpkgs#gmp.dev --no-link --print-out-paths)/lib/pkgconfig"
41
+
opam install . --confirm-level=unsafe-yes --deps-only
42
+
- name: build
43
+
command: |
44
+
opam exec -- dune build -p conpool
45
+
- name: switch-test
46
+
command: |
47
+
opam install . --confirm-level=unsafe-yes --deps-only --with-test
48
+
- name: test
49
+
command: |
50
+
opam exec -- dune runtest --verbose
51
+
- name: doc
52
+
command: |
53
+
opam install -y odoc
54
+
opam exec -- dune build @doc
+15
LICENSE.md
+15
LICENSE.md
···
1
+
ISC License
2
+
3
+
Copyright (c) 2025 Anil Madhavapeddy <anil@recoil.org>
4
+
5
+
Permission to use, copy, modify, and distribute this software for any
6
+
purpose with or without fee is hereby granted, provided that the above
7
+
copyright notice and this permission notice appear in all copies.
8
+
9
+
THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
10
+
WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
11
+
MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
12
+
ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
13
+
WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
14
+
ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
15
+
OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
+110
README.md
+110
README.md
···
1
+
# Conpool - Protocol-agnostic Connection Pooling for Eio
2
+
3
+
Conpool is a connection pooling library built on Eio that manages TCP connection lifecycles, validates connection health, and provides per-endpoint resource limiting for any TCP-based protocol.
4
+
5
+
## Key Features
6
+
7
+
- **Protocol-agnostic**: Works with HTTP, Redis, PostgreSQL, or any TCP-based protocol
8
+
- **Health validation**: Automatically validates connections before reuse
9
+
- **Per-endpoint limits**: Independent connection limits and pooling for each endpoint
10
+
- **TLS support**: Optional TLS configuration for secure connections
11
+
- **Statistics & monitoring**: Track connection usage, hits/misses, and health status
12
+
- **Built on Eio**: Leverages Eio's structured concurrency and resource management
13
+
14
+
## Usage
15
+
16
+
Basic example establishing a connection pool:
17
+
18
+
```ocaml
19
+
open Eio.Std
20
+
21
+
let run env =
22
+
Switch.run (fun sw ->
23
+
(* Create a connection pool *)
24
+
let pool = Conpool.create
25
+
~sw
26
+
~net:(Eio.Stdenv.net env)
27
+
~clock:(Eio.Stdenv.clock env)
28
+
()
29
+
in
30
+
31
+
(* Define an endpoint *)
32
+
let endpoint = Conpool.Endpoint.make ~host:"example.com" ~port:80 in
33
+
34
+
(* Use a connection from the pool *)
35
+
Conpool.with_connection pool endpoint (fun conn ->
36
+
Eio.Flow.copy_string "GET / HTTP/1.1\r\nHost: example.com\r\n\r\n" conn;
37
+
let buf = Eio.Buf_read.of_flow conn ~max_size:4096 in
38
+
Eio.Buf_read.take_all buf
39
+
)
40
+
)
41
+
```
42
+
43
+
With TLS configuration:
44
+
45
+
```ocaml
46
+
let run env =
47
+
Switch.run (fun sw ->
48
+
(* Create TLS configuration - SNI servername is automatically set to the endpoint's hostname *)
49
+
let tls_config = Tls.Config.client ~authenticator:(Ca_certs.authenticator ()) () in
50
+
51
+
(* Create pool with TLS *)
52
+
let pool = Conpool.create
53
+
~sw
54
+
~net:(Eio.Stdenv.net env)
55
+
~clock:(Eio.Stdenv.clock env)
56
+
~tls:tls_config
57
+
()
58
+
in
59
+
60
+
let endpoint = Conpool.Endpoint.make ~host:"example.com" ~port:443 in
61
+
Conpool.with_connection pool endpoint (fun conn ->
62
+
(* Use TLS-encrypted connection *)
63
+
...
64
+
)
65
+
)
66
+
```
67
+
68
+
Custom pool configuration:
69
+
70
+
```ocaml
71
+
let config = Conpool.Config.make
72
+
~max_connections_per_endpoint:20
73
+
~max_idle_per_endpoint:5
74
+
~connection_timeout:10.0
75
+
~validation_interval:300.0
76
+
()
77
+
in
78
+
79
+
let pool = Conpool.create ~sw ~net ~clock ~config ()
80
+
```
81
+
82
+
Monitor pool statistics:
83
+
84
+
```ocaml
85
+
let stats = Conpool.stats pool endpoint in
86
+
Printf.printf "Active: %d, Idle: %d, Hits: %d, Misses: %d\n"
87
+
(Conpool.Stats.active_connections stats)
88
+
(Conpool.Stats.idle_connections stats)
89
+
(Conpool.Stats.cache_hits stats)
90
+
(Conpool.Stats.cache_misses stats)
91
+
```
92
+
93
+
## Installation
94
+
95
+
```
96
+
opam install conpool
97
+
```
98
+
99
+
## Documentation
100
+
101
+
API documentation is available at https://tangled.org/@anil.recoil.org/ocaml-conpool or via:
102
+
103
+
```
104
+
opam install conpool
105
+
odig doc conpool
106
+
```
107
+
108
+
## License
109
+
110
+
ISC
+35
conpool.opam
+35
conpool.opam
···
1
+
# This file is generated by dune, edit dune-project instead
2
+
opam-version: "2.0"
3
+
synopsis: "Protocol-agnostic TCP/IP connection pooling library for Eio"
4
+
description:
5
+
"Conpool is a connection pooling library built on Eio.Pool that manages TCP connection lifecycles, validates connection health, and provides per-endpoint resource limiting for any TCP-based protocol (HTTP, Redis, PostgreSQL, etc.)"
6
+
maintainer: ["Anil Madhavapeddy <anil@recoil.org>"]
7
+
authors: ["Anil Madhavapeddy <anil@recoil.org>"]
8
+
license: "ISC"
9
+
homepage: "https://tangled.org/@anil.recoil.org/ocaml-conpool"
10
+
bug-reports: "https://tangled.org/@anil.recoil.org/ocaml-conpool/issues"
11
+
depends: [
12
+
"ocaml" {>= "5.1.0"}
13
+
"dune" {>= "3.20" & >= "3.0"}
14
+
"eio"
15
+
"tls-eio" {>= "1.0"}
16
+
"logs"
17
+
"fmt"
18
+
"cmdliner"
19
+
"odoc" {with-doc}
20
+
]
21
+
build: [
22
+
["dune" "subst"] {dev}
23
+
[
24
+
"dune"
25
+
"build"
26
+
"-p"
27
+
name
28
+
"-j"
29
+
jobs
30
+
"@install"
31
+
"@runtest" {with-test}
32
+
"@doc" {with-doc}
33
+
]
34
+
]
35
+
x-maintenance-intent: ["(latest)"]
+26
dune-project
+26
dune-project
···
1
+
(lang dune 3.20)
2
+
3
+
(name conpool)
4
+
5
+
(generate_opam_files true)
6
+
7
+
(license ISC)
8
+
(authors "Anil Madhavapeddy <anil@recoil.org>")
9
+
(homepage "https://tangled.org/@anil.recoil.org/ocaml-conpool")
10
+
(maintainers "Anil Madhavapeddy <anil@recoil.org>")
11
+
(bug_reports "https://tangled.org/@anil.recoil.org/ocaml-conpool/issues")
12
+
(maintenance_intent "(latest)")
13
+
14
+
(package
15
+
(name conpool)
16
+
(synopsis "Protocol-agnostic TCP/IP connection pooling library for Eio")
17
+
(description "Conpool is a connection pooling library built on Eio.Pool that manages TCP connection lifecycles, validates connection health, and provides per-endpoint resource limiting for any TCP-based protocol (HTTP, Redis, PostgreSQL, etc.)")
18
+
(depends
19
+
(ocaml (>= 5.1.0))
20
+
(dune (>= 3.0))
21
+
eio
22
+
(tls-eio (>= 1.0))
23
+
logs
24
+
fmt
25
+
cmdliner
26
+
(odoc :with-doc)))
+57
lib/cmd.ml
+57
lib/cmd.ml
···
1
+
(*---------------------------------------------------------------------------
2
+
Copyright (c) 2025 Anil Madhavapeddy <anil@recoil.org>. All rights reserved.
3
+
SPDX-License-Identifier: ISC
4
+
---------------------------------------------------------------------------*)
5
+
6
+
(** Cmdliner terms for connection pool configuration *)
7
+
8
+
open Cmdliner
9
+
10
+
let max_connections_per_endpoint =
11
+
let doc = "Maximum concurrent connections per endpoint." in
12
+
Arg.(
13
+
value & opt int 10
14
+
& info [ "max-connections-per-endpoint" ] ~doc ~docv:"NUM")
15
+
16
+
let max_idle_time =
17
+
let doc = "Maximum time a connection can sit idle in seconds." in
18
+
Arg.(value & opt float 60.0 & info [ "max-idle-time" ] ~doc ~docv:"SECONDS")
19
+
20
+
let max_connection_lifetime =
21
+
let doc = "Maximum connection age in seconds." in
22
+
Arg.(
23
+
value & opt float 300.0
24
+
& info [ "max-connection-lifetime" ] ~doc ~docv:"SECONDS")
25
+
26
+
let max_connection_uses =
27
+
let doc = "Maximum times a connection can be reused (omit for unlimited)." in
28
+
Arg.(
29
+
value
30
+
& opt (some int) None
31
+
& info [ "max-connection-uses" ] ~doc ~docv:"NUM")
32
+
33
+
let connect_timeout =
34
+
let doc = "Connection timeout in seconds." in
35
+
Arg.(value & opt float 10.0 & info [ "connect-timeout" ] ~doc ~docv:"SECONDS")
36
+
37
+
let connect_retry_count =
38
+
let doc = "Number of connection retry attempts." in
39
+
Arg.(value & opt int 3 & info [ "connect-retry-count" ] ~doc ~docv:"NUM")
40
+
41
+
let connect_retry_delay =
42
+
let doc = "Initial retry delay in seconds (with exponential backoff)." in
43
+
Arg.(
44
+
value & opt float 0.1 & info [ "connect-retry-delay" ] ~doc ~docv:"SECONDS")
45
+
46
+
let config =
47
+
let make max_conn max_idle max_lifetime max_uses timeout retry_count
48
+
retry_delay =
49
+
Config.make ~max_connections_per_endpoint:max_conn ~max_idle_time:max_idle
50
+
~max_connection_lifetime:max_lifetime ?max_connection_uses:max_uses
51
+
~connect_timeout:timeout ~connect_retry_count:retry_count
52
+
~connect_retry_delay:retry_delay ()
53
+
in
54
+
Term.(
55
+
const make $ max_connections_per_endpoint $ max_idle_time
56
+
$ max_connection_lifetime $ max_connection_uses $ connect_timeout
57
+
$ connect_retry_count $ connect_retry_delay)
+43
lib/cmd.mli
+43
lib/cmd.mli
···
1
+
(*---------------------------------------------------------------------------
2
+
Copyright (c) 2025 Anil Madhavapeddy <anil@recoil.org>. All rights reserved.
3
+
SPDX-License-Identifier: ISC
4
+
---------------------------------------------------------------------------*)
5
+
6
+
(** Cmdliner terms for connection pool configuration *)
7
+
8
+
(** {1 Configuration Terms} *)
9
+
10
+
val max_connections_per_endpoint : int Cmdliner.Term.t
11
+
(** Cmdliner term for maximum connections per endpoint. Default: 10 Flag:
12
+
[--max-connections-per-endpoint] *)
13
+
14
+
val max_idle_time : float Cmdliner.Term.t
15
+
(** Cmdliner term for maximum idle time in seconds. Default: 60.0 Flag:
16
+
[--max-idle-time] *)
17
+
18
+
val max_connection_lifetime : float Cmdliner.Term.t
19
+
(** Cmdliner term for maximum connection lifetime in seconds. Default: 300.0
20
+
Flag: [--max-connection-lifetime] *)
21
+
22
+
val max_connection_uses : int option Cmdliner.Term.t
23
+
(** Cmdliner term for maximum connection uses. Default: None (unlimited) Flag:
24
+
[--max-connection-uses] *)
25
+
26
+
val connect_timeout : float Cmdliner.Term.t
27
+
(** Cmdliner term for connection timeout in seconds. Default: 10.0 Flag:
28
+
[--connect-timeout] *)
29
+
30
+
val connect_retry_count : int Cmdliner.Term.t
31
+
(** Cmdliner term for number of connection retry attempts. Default: 3 Flag:
32
+
[--connect-retry-count] *)
33
+
34
+
val connect_retry_delay : float Cmdliner.Term.t
35
+
(** Cmdliner term for initial retry delay in seconds. Default: 0.1 Flag:
36
+
[--connect-retry-delay] *)
37
+
38
+
(** {1 Combined Terms} *)
39
+
40
+
val config : Config.t Cmdliner.Term.t
41
+
(** Cmdliner term that combines all configuration options into a {!Config.t}.
42
+
This term can be used in your application's main command to accept all
43
+
connection pool configuration options from the command line. *)
+164
lib/config.ml
+164
lib/config.ml
···
1
+
(*---------------------------------------------------------------------------
2
+
Copyright (c) 2025 Anil Madhavapeddy <anil@recoil.org>. All rights reserved.
3
+
SPDX-License-Identifier: ISC
4
+
---------------------------------------------------------------------------*)
5
+
6
+
(** Configuration for connection pools *)
7
+
8
+
let src = Logs.Src.create "conpool.config" ~doc:"Connection pool configuration"
9
+
10
+
module Log = (val Logs.src_log src : Logs.LOG)
11
+
12
+
type t = {
13
+
max_connections_per_endpoint : int;
14
+
max_idle_time : float;
15
+
max_connection_lifetime : float;
16
+
max_connection_uses : int option;
17
+
health_check :
18
+
([Eio.Resource.close_ty | Eio.Flow.two_way_ty] Eio.Resource.t -> bool) option;
19
+
connect_timeout : float option;
20
+
connect_retry_count : int;
21
+
connect_retry_delay : float;
22
+
on_connection_created : (Endpoint.t -> unit) option;
23
+
on_connection_closed : (Endpoint.t -> unit) option;
24
+
on_connection_reused : (Endpoint.t -> unit) option;
25
+
}
26
+
27
+
let make ?(max_connections_per_endpoint = 10) ?(max_idle_time = 60.0)
28
+
?(max_connection_lifetime = 300.0) ?max_connection_uses ?health_check
29
+
?(connect_timeout = 10.0) ?(connect_retry_count = 3)
30
+
?(connect_retry_delay = 0.1) ?on_connection_created ?on_connection_closed
31
+
?on_connection_reused () =
32
+
(* Validate parameters *)
33
+
if max_connections_per_endpoint <= 0 then
34
+
invalid_arg
35
+
(Printf.sprintf "max_connections_per_endpoint must be positive, got %d"
36
+
max_connections_per_endpoint);
37
+
38
+
if max_idle_time <= 0.0 then
39
+
invalid_arg
40
+
(Printf.sprintf "max_idle_time must be positive, got %.2f" max_idle_time);
41
+
42
+
if max_connection_lifetime <= 0.0 then
43
+
invalid_arg
44
+
(Printf.sprintf "max_connection_lifetime must be positive, got %.2f"
45
+
max_connection_lifetime);
46
+
47
+
(match max_connection_uses with
48
+
| Some n when n <= 0 ->
49
+
invalid_arg
50
+
(Printf.sprintf "max_connection_uses must be positive, got %d" n)
51
+
| _ -> ());
52
+
53
+
if connect_timeout <= 0.0 then
54
+
invalid_arg
55
+
(Printf.sprintf "connect_timeout must be positive, got %.2f"
56
+
connect_timeout);
57
+
58
+
if connect_retry_count < 0 then
59
+
invalid_arg
60
+
(Printf.sprintf "connect_retry_count must be non-negative, got %d"
61
+
connect_retry_count);
62
+
63
+
if connect_retry_delay <= 0.0 then
64
+
invalid_arg
65
+
(Printf.sprintf "connect_retry_delay must be positive, got %.2f"
66
+
connect_retry_delay);
67
+
68
+
{
69
+
max_connections_per_endpoint;
70
+
max_idle_time;
71
+
max_connection_lifetime;
72
+
max_connection_uses;
73
+
health_check;
74
+
connect_timeout = Some connect_timeout;
75
+
connect_retry_count;
76
+
connect_retry_delay;
77
+
on_connection_created;
78
+
on_connection_closed;
79
+
on_connection_reused;
80
+
}
81
+
82
+
let default = make ()
83
+
let max_connections_per_endpoint t = t.max_connections_per_endpoint
84
+
let max_idle_time t = t.max_idle_time
85
+
let max_connection_lifetime t = t.max_connection_lifetime
86
+
let max_connection_uses t = t.max_connection_uses
87
+
let health_check t = t.health_check
88
+
let connect_timeout t = t.connect_timeout
89
+
let connect_retry_count t = t.connect_retry_count
90
+
let connect_retry_delay t = t.connect_retry_delay
91
+
let on_connection_created t = t.on_connection_created
92
+
let on_connection_closed t = t.on_connection_closed
93
+
let on_connection_reused t = t.on_connection_reused
94
+
95
+
let pp ppf t =
96
+
Fmt.pf ppf
97
+
"@[<v>Config:@,\
98
+
- max_connections_per_endpoint: %d@,\
99
+
- max_idle_time: %.1fs@,\
100
+
- max_connection_lifetime: %.1fs@,\
101
+
- max_connection_uses: %s@,\
102
+
- connect_timeout: %s@,\
103
+
- connect_retry_count: %d@,\
104
+
- connect_retry_delay: %.2fs@]"
105
+
t.max_connections_per_endpoint t.max_idle_time t.max_connection_lifetime
106
+
(match t.max_connection_uses with
107
+
| Some n -> string_of_int n
108
+
| None -> "unlimited")
109
+
(match t.connect_timeout with
110
+
| Some f -> Fmt.str "%.1fs" f
111
+
| None -> "none")
112
+
t.connect_retry_count t.connect_retry_delay
113
+
114
+
(** {1 Protocol Handler Configuration}
115
+
116
+
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). *)
119
+
120
+
(** Access mode for connections.
121
+
- [Exclusive] - Each connection is used by one request at a time (HTTP/1.x)
122
+
- [Shared] - Multiple requests can share a connection (HTTP/2) *)
123
+
type access_mode =
124
+
| Exclusive
125
+
(** Exclusive access - one request per connection at a time *)
126
+
| Shared of int
127
+
(** Shared access - up to n concurrent requests per connection *)
128
+
129
+
(** Connection type alias for protocol config *)
130
+
type connection_flow = [Eio.Resource.close_ty | Eio.Flow.two_way_ty] Eio.Resource.t
131
+
132
+
(** Protocol configuration for typed connection pools.
133
+
@param 'state The protocol-specific state type (e.g., H2_client.t for HTTP/2) *)
134
+
type 'state protocol_config = {
135
+
init_state :
136
+
sw:Eio.Switch.t ->
137
+
flow:connection_flow ->
138
+
tls_epoch:Tls.Core.epoch_data option ->
139
+
'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
+
145
+
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
+
149
+
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
+
153
+
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
+
157
+
on_close : 'state -> unit;
158
+
(** Cleanup callback when connection is destroyed.
159
+
For HTTP/2, can send GOAWAY frame. *)
160
+
161
+
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. *)
164
+
}
+159
lib/config.mli
+159
lib/config.mli
···
1
+
(*---------------------------------------------------------------------------
2
+
Copyright (c) 2025 Anil Madhavapeddy <anil@recoil.org>. All rights reserved.
3
+
SPDX-License-Identifier: ISC
4
+
---------------------------------------------------------------------------*)
5
+
6
+
(** Configuration for connection pools *)
7
+
8
+
(** {1 Logging} *)
9
+
10
+
val src : Logs.Src.t
11
+
(** Logs source for configuration operations. Configure logging with:
12
+
{[
13
+
Logs.Src.set_level Conpool.Config.src (Some Logs.Debug)
14
+
]} *)
15
+
16
+
(** {1 Type} *)
17
+
18
+
type t
19
+
(** Pool configuration *)
20
+
21
+
(** {1 Construction} *)
22
+
23
+
val make :
24
+
?max_connections_per_endpoint:int ->
25
+
?max_idle_time:float ->
26
+
?max_connection_lifetime:float ->
27
+
?max_connection_uses:int ->
28
+
?health_check:([Eio.Resource.close_ty | Eio.Flow.two_way_ty] Eio.Resource.t -> bool) ->
29
+
?connect_timeout:float ->
30
+
?connect_retry_count:int ->
31
+
?connect_retry_delay:float ->
32
+
?on_connection_created:(Endpoint.t -> unit) ->
33
+
?on_connection_closed:(Endpoint.t -> unit) ->
34
+
?on_connection_reused:(Endpoint.t -> unit) ->
35
+
unit ->
36
+
t
37
+
(** Create pool configuration with optional parameters.
38
+
39
+
@param max_connections_per_endpoint
40
+
Maximum concurrent connections per endpoint (default: 10)
41
+
@param max_idle_time
42
+
Maximum time a connection can sit idle in seconds (default: 60.0)
43
+
@param max_connection_lifetime
44
+
Maximum connection age in seconds (default: 300.0)
45
+
@param max_connection_uses
46
+
Maximum times a connection can be reused (default: unlimited)
47
+
@param health_check Custom health check function (default: none)
48
+
@param connect_timeout Connection timeout in seconds (default: 10.0)
49
+
@param connect_retry_count Number of connection retry attempts (default: 3)
50
+
@param connect_retry_delay
51
+
Initial retry delay in seconds, with exponential backoff (default: 0.1)
52
+
@param on_connection_created Hook called when a connection is created
53
+
@param on_connection_closed Hook called when a connection is closed
54
+
@param on_connection_reused Hook called when a connection is reused *)
55
+
56
+
val default : t
57
+
(** Sensible defaults for most use cases:
58
+
- max_connections_per_endpoint: 10
59
+
- max_idle_time: 60.0s
60
+
- max_connection_lifetime: 300.0s
61
+
- max_connection_uses: unlimited
62
+
- health_check: none
63
+
- connect_timeout: 10.0s
64
+
- connect_retry_count: 3
65
+
- connect_retry_delay: 0.1s
66
+
- hooks: none *)
67
+
68
+
(** {1 Accessors} *)
69
+
70
+
val max_connections_per_endpoint : t -> int
71
+
(** Get maximum connections per endpoint. *)
72
+
73
+
val max_idle_time : t -> float
74
+
(** Get maximum idle time in seconds. *)
75
+
76
+
val max_connection_lifetime : t -> float
77
+
(** Get maximum connection lifetime in seconds. *)
78
+
79
+
val max_connection_uses : t -> int option
80
+
(** Get maximum connection uses, if any. *)
81
+
82
+
val health_check :
83
+
t -> ([Eio.Resource.close_ty | Eio.Flow.two_way_ty] Eio.Resource.t -> bool) option
84
+
(** Get custom health check function, if any. *)
85
+
86
+
val connect_timeout : t -> float option
87
+
(** Get connection timeout in seconds, if any. *)
88
+
89
+
val connect_retry_count : t -> int
90
+
(** Get number of connection retry attempts. *)
91
+
92
+
val connect_retry_delay : t -> float
93
+
(** Get initial retry delay in seconds. *)
94
+
95
+
val on_connection_created : t -> (Endpoint.t -> unit) option
96
+
(** Get connection created hook, if any. *)
97
+
98
+
val on_connection_closed : t -> (Endpoint.t -> unit) option
99
+
(** Get connection closed hook, if any. *)
100
+
101
+
val on_connection_reused : t -> (Endpoint.t -> unit) option
102
+
(** Get connection reused hook, if any. *)
103
+
104
+
(** {1 Pretty-printing} *)
105
+
106
+
val pp : t Fmt.t
107
+
(** Pretty-printer for configuration. *)
108
+
109
+
(** {1 Protocol Handler Configuration}
110
+
111
+
Protocol handlers define protocol-specific behavior for typed connection pools.
112
+
This enables different pooling strategies for different protocols
113
+
(e.g., exclusive for HTTP/1.x, shared for HTTP/2). *)
114
+
115
+
(** Access mode for connections.
116
+
- [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) *)
118
+
type access_mode =
119
+
| Exclusive
120
+
(** Exclusive access - one request per connection at a time *)
121
+
| Shared of int
122
+
(** Shared access - up to n concurrent requests per connection *)
123
+
124
+
(** Connection flow type for protocol handlers. *)
125
+
type connection_flow = [Eio.Resource.close_ty | Eio.Flow.two_way_ty] Eio.Resource.t
126
+
127
+
(** Protocol configuration for typed connection pools.
128
+
@param 'state The protocol-specific state type (e.g., H2_client.t for HTTP/2) *)
129
+
type 'state protocol_config = {
130
+
init_state :
131
+
sw:Eio.Switch.t ->
132
+
flow:connection_flow ->
133
+
tls_epoch:Tls.Core.epoch_data option ->
134
+
'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
+
140
+
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
+
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
+
148
+
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
+
152
+
on_close : 'state -> unit;
153
+
(** Cleanup callback when connection is destroyed.
154
+
For HTTP/2, can send GOAWAY frame. *)
155
+
156
+
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. *)
159
+
}
+39
lib/connection.ml
+39
lib/connection.ml
···
1
+
(*---------------------------------------------------------------------------
2
+
Copyright (c) 2025 Anil Madhavapeddy <anil@recoil.org>. All rights reserved.
3
+
SPDX-License-Identifier: ISC
4
+
---------------------------------------------------------------------------*)
5
+
6
+
(** Internal connection representation - not exposed in public API *)
7
+
8
+
let src =
9
+
Logs.Src.create "conpool.connection"
10
+
~doc:"Connection pool internal connection management"
11
+
12
+
module Log = (val Logs.src_log src : Logs.LOG)
13
+
14
+
type t = {
15
+
flow : [Eio.Resource.close_ty | Eio.Flow.two_way_ty] Eio.Resource.t;
16
+
tls_flow : Tls_eio.t option;
17
+
created_at : float;
18
+
mutable last_used : float;
19
+
mutable use_count : int;
20
+
endpoint : Endpoint.t;
21
+
mutex : Eio.Mutex.t;
22
+
}
23
+
24
+
let flow t = t.flow
25
+
let tls_flow t = t.tls_flow
26
+
let endpoint t = t.endpoint
27
+
let created_at t = t.created_at
28
+
let last_used t = t.last_used
29
+
let use_count t = t.use_count
30
+
31
+
let update_usage t ~now =
32
+
Eio.Mutex.use_rw ~protect:true t.mutex (fun () ->
33
+
t.last_used <- now;
34
+
t.use_count <- t.use_count + 1)
35
+
36
+
let pp ppf t =
37
+
let uses = t.use_count in
38
+
Fmt.pf ppf "Connection(endpoint=%a, created_at=%.2f, uses=%d)" Endpoint.pp
39
+
t.endpoint t.created_at uses
+665
lib/conpool.ml
+665
lib/conpool.ml
···
1
+
(*---------------------------------------------------------------------------
2
+
Copyright (c) 2025 Anil Madhavapeddy <anil@recoil.org>. All rights reserved.
3
+
SPDX-License-Identifier: ISC
4
+
---------------------------------------------------------------------------*)
5
+
6
+
(** Conpool - Protocol-aware TCP/IP connection pooling library for Eio *)
7
+
8
+
let src = Logs.Src.create "conpool" ~doc:"Connection pooling library"
9
+
10
+
module Log = (val Logs.src_log src : Logs.LOG)
11
+
12
+
(* Re-export submodules *)
13
+
module Endpoint = Endpoint
14
+
module Config = Config
15
+
module Stats = Stats
16
+
module Cmd = Cmd
17
+
18
+
(* Track whether TLS tracing has been suppressed *)
19
+
let tls_tracing_suppressed = ref false
20
+
21
+
(* Suppress TLS tracing debug output (hexdumps) unless explicitly enabled *)
22
+
let suppress_tls_tracing () =
23
+
if not !tls_tracing_suppressed then begin
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
+
| _ -> ())
30
+
| None -> ()
31
+
end
32
+
33
+
(** {1 Error Types} *)
34
+
35
+
type error =
36
+
| Dns_resolution_failed of { hostname : string }
37
+
| Connection_failed of {
38
+
endpoint : Endpoint.t;
39
+
attempts : int;
40
+
last_error : string;
41
+
}
42
+
| Connection_timeout of { endpoint : Endpoint.t; timeout : float }
43
+
| Invalid_config of string
44
+
| Invalid_endpoint of string
45
+
46
+
let pp_error ppf = function
47
+
| Dns_resolution_failed { hostname } ->
48
+
Fmt.pf ppf "DNS resolution failed for hostname: %s" hostname
49
+
| Connection_failed { endpoint; attempts; last_error } ->
50
+
Fmt.pf ppf "Failed to connect to %a after %d attempts: %s" Endpoint.pp
51
+
endpoint attempts last_error
52
+
| Connection_timeout { endpoint; timeout } ->
53
+
Fmt.pf ppf "Connection timeout to %a after %.2fs" Endpoint.pp endpoint
54
+
timeout
55
+
| Invalid_config msg -> Fmt.pf ppf "Invalid configuration: %s" msg
56
+
| Invalid_endpoint msg -> Fmt.pf ppf "Invalid endpoint: %s" msg
57
+
58
+
type Eio.Exn.err += E of error
59
+
60
+
let err e = Eio.Exn.create (E e)
61
+
62
+
let () =
63
+
Eio.Exn.register_pp (fun f -> function
64
+
| E e ->
65
+
Fmt.string f "Conpool ";
66
+
pp_error f e;
67
+
true
68
+
| _ -> false)
69
+
70
+
(** {1 Connection Types} *)
71
+
72
+
type connection_ty = [Eio.Resource.close_ty | Eio.Flow.two_way_ty]
73
+
type connection = connection_ty Eio.Resource.t
74
+
75
+
(** {1 Internal Types} *)
76
+
77
+
(** Internal connection wrapper with protocol state and tracking. *)
78
+
type 'state pooled_connection = {
79
+
pc_flow : connection;
80
+
pc_tls_flow : Tls_eio.t option;
81
+
pc_state : 'state;
82
+
pc_created_at : float;
83
+
mutable pc_last_used : float;
84
+
(** Last time this connection was used (for idle timeout). *)
85
+
mutable pc_use_count : int;
86
+
(** Number of times this connection has been used. *)
87
+
pc_endpoint : Endpoint.t;
88
+
mutable pc_active_users : int;
89
+
pc_user_available : Eio.Condition.t;
90
+
mutable pc_closed : bool;
91
+
pc_connection_cancel : exn -> unit;
92
+
(** Cancels the connection-lifetime switch, stopping any protocol fibers. *)
93
+
}
94
+
95
+
(** Statistics for an endpoint. *)
96
+
type endp_stats = {
97
+
mutable active : int;
98
+
mutable idle : int;
99
+
(** Number of idle connections (active_users = 0). *)
100
+
mutable total_created : int;
101
+
mutable total_reused : int;
102
+
mutable total_closed : int;
103
+
mutable errors : int;
104
+
(** Number of connection errors encountered. *)
105
+
}
106
+
107
+
(** Endpoint pool storing connections. *)
108
+
type 'state endpoint_pool = {
109
+
connections : 'state pooled_connection list ref;
110
+
ep_mutex : Eio.Mutex.t;
111
+
stats : endp_stats;
112
+
stats_mutex : Eio.Mutex.t;
113
+
}
114
+
115
+
(** Internal pool representation. *)
116
+
type ('state, 'clock, 'net) internal = {
117
+
sw : Eio.Switch.t;
118
+
net : 'net;
119
+
clock : 'clock;
120
+
config : Config.t;
121
+
tls : Tls.Config.client option;
122
+
protocol : 'state Config.protocol_config;
123
+
endpoints : (Endpoint.t, 'state endpoint_pool) Hashtbl.t;
124
+
endpoints_mutex : Eio.Mutex.t;
125
+
}
126
+
127
+
(** {1 Public Types} *)
128
+
129
+
type 'state t =
130
+
Pool : ('state, 'clock Eio.Time.clock, 'net Eio.Net.t) internal -> 'state t
131
+
132
+
type 'state connection_info = {
133
+
flow : connection;
134
+
tls_epoch : Tls.Core.epoch_data option;
135
+
state : 'state;
136
+
}
137
+
138
+
(** {1 Default Protocol Handler}
139
+
140
+
For simple exclusive-access protocols (HTTP/1.x, Redis, etc.),
141
+
use unit state with no special initialization. *)
142
+
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
+
}
151
+
152
+
(** {1 Helper Functions} *)
153
+
154
+
let get_time pool = Eio.Time.now pool.clock
155
+
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
+
}
164
+
165
+
let snapshot_stats (stats : endp_stats) : Stats.t =
166
+
Stats.make ~active:stats.active ~idle:stats.idle
167
+
~total_created:stats.total_created ~total_reused:stats.total_reused
168
+
~total_closed:stats.total_closed ~errors:stats.errors
169
+
170
+
(** {1 Connection Creation} *)
171
+
172
+
let create_connection pool endpoint =
173
+
Log.debug (fun m -> m "Creating connection to %a" Endpoint.pp endpoint);
174
+
175
+
(* DNS resolution *)
176
+
let addr =
177
+
try
178
+
let addrs =
179
+
Eio.Net.getaddrinfo_stream pool.net (Endpoint.host endpoint)
180
+
~service:(string_of_int (Endpoint.port endpoint))
181
+
in
182
+
match addrs with
183
+
| addr :: _ -> addr
184
+
| [] ->
185
+
raise (err (Dns_resolution_failed { hostname = Endpoint.host endpoint }))
186
+
with Eio.Io _ as ex ->
187
+
let bt = Printexc.get_raw_backtrace () in
188
+
Eio.Exn.reraise_with_context ex bt "resolving %a" Endpoint.pp endpoint
189
+
in
190
+
191
+
(* TCP connection with optional timeout *)
192
+
let socket =
193
+
try
194
+
match Config.connect_timeout pool.config with
195
+
| Some timeout ->
196
+
Eio.Time.with_timeout_exn pool.clock timeout (fun () ->
197
+
Eio.Net.connect ~sw:pool.sw pool.net addr)
198
+
| None -> Eio.Net.connect ~sw:pool.sw pool.net addr
199
+
with Eio.Io _ as ex ->
200
+
let bt = Printexc.get_raw_backtrace () in
201
+
Eio.Exn.reraise_with_context ex bt "connecting to %a" Endpoint.pp endpoint
202
+
in
203
+
204
+
Log.debug (fun m -> m "TCP connection established to %a" Endpoint.pp endpoint);
205
+
206
+
(* Optional TLS handshake *)
207
+
let flow, tls_flow =
208
+
match pool.tls with
209
+
| None ->
210
+
((socket :> connection), None)
211
+
| Some tls_config ->
212
+
try
213
+
Log.debug (fun m ->
214
+
m "Initiating TLS handshake with %a" Endpoint.pp endpoint);
215
+
let host =
216
+
Domain_name.(host_exn (of_string_exn (Endpoint.host endpoint)))
217
+
in
218
+
let tls = Tls_eio.client_of_flow ~host tls_config socket in
219
+
suppress_tls_tracing ();
220
+
Log.info (fun m ->
221
+
m "TLS connection established to %a" Endpoint.pp endpoint);
222
+
((tls :> connection), Some tls)
223
+
with Eio.Io _ as ex ->
224
+
let bt = Printexc.get_raw_backtrace () in
225
+
Eio.Exn.reraise_with_context ex bt "TLS handshake with %a" Endpoint.pp endpoint
226
+
in
227
+
228
+
(* Get TLS epoch if available *)
229
+
let tls_epoch =
230
+
match tls_flow with
231
+
| Some tls_flow -> (
232
+
match Tls_eio.epoch tls_flow with
233
+
| Ok epoch -> Some epoch
234
+
| Error () -> None)
235
+
| None -> None
236
+
in
237
+
238
+
(* Create connection-lifetime sub-switch via a fiber.
239
+
This switch lives for the connection's lifetime and can be used
240
+
by the protocol handler to spawn long-running fibers (e.g., HTTP/2 reader). *)
241
+
let conn_sw_ref = ref None in
242
+
let conn_cancel_ref = ref (fun (_ : exn) -> ()) in
243
+
let ready_promise, ready_resolver = Eio.Promise.create () in
244
+
245
+
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
+
);
256
+
257
+
(* Wait for the switch to be created *)
258
+
Eio.Promise.await ready_promise;
259
+
let conn_sw = Option.get !conn_sw_ref in
260
+
let conn_cancel = !conn_cancel_ref in
261
+
262
+
(* Initialize protocol-specific state with connection switch *)
263
+
Log.debug (fun m -> m "Initializing protocol state for %a" Endpoint.pp endpoint);
264
+
let state = pool.protocol.init_state ~sw:conn_sw ~flow ~tls_epoch in
265
+
266
+
let now = get_time pool in
267
+
268
+
Log.info (fun m -> m "Created connection to %a" Endpoint.pp endpoint);
269
+
270
+
{
271
+
pc_flow = flow;
272
+
pc_tls_flow = tls_flow;
273
+
pc_state = state;
274
+
pc_created_at = now;
275
+
pc_last_used = now;
276
+
pc_use_count = 0;
277
+
pc_endpoint = endpoint;
278
+
pc_active_users = 0;
279
+
pc_user_available = Eio.Condition.create ();
280
+
pc_closed = false;
281
+
pc_connection_cancel = conn_cancel;
282
+
}
283
+
284
+
(** {1 Connection Health Checking} *)
285
+
286
+
(** Health check result distinguishing errors from normal lifecycle. *)
287
+
type health_status =
288
+
| Healthy
289
+
| Unhealthy_error of string
290
+
(** Connection failed due to an error (protocol failure, etc.) *)
291
+
| Unhealthy_lifecycle of string
292
+
(** Connection should close due to normal lifecycle (timeout, max uses, etc.) *)
293
+
294
+
let check_health pool conn =
295
+
if conn.pc_closed then
296
+
Unhealthy_lifecycle "already closed"
297
+
else
298
+
(* Check protocol-specific health *)
299
+
let protocol_healthy = pool.protocol.is_healthy conn.pc_state in
300
+
if not protocol_healthy then begin
301
+
Log.debug (fun m -> m "Connection unhealthy: protocol check failed");
302
+
Unhealthy_error "protocol check failed"
303
+
end else
304
+
let now = get_time pool in
305
+
(* Check connection age *)
306
+
let age = now -. conn.pc_created_at in
307
+
let max_lifetime = Config.max_connection_lifetime pool.config in
308
+
if age > max_lifetime then begin
309
+
Log.debug (fun m -> m "Connection unhealthy: exceeded max lifetime (%.1fs > %.1fs)"
310
+
age max_lifetime);
311
+
Unhealthy_lifecycle "exceeded max lifetime"
312
+
end else
313
+
(* Check idle time - only for idle connections *)
314
+
let idle_time = now -. conn.pc_last_used in
315
+
let max_idle = Config.max_idle_time pool.config in
316
+
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);
319
+
Unhealthy_lifecycle "exceeded max idle time"
320
+
end else
321
+
(* Check use count *)
322
+
match Config.max_connection_uses pool.config with
323
+
| 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);
326
+
Unhealthy_lifecycle "exceeded max uses"
327
+
| _ ->
328
+
Healthy
329
+
330
+
let is_healthy pool conn =
331
+
match check_health pool conn with
332
+
| Healthy -> true
333
+
| Unhealthy_error _ | Unhealthy_lifecycle _ -> false
334
+
335
+
(** {1 Connection Cleanup} *)
336
+
337
+
let close_connection pool conn =
338
+
if not conn.pc_closed then begin
339
+
conn.pc_closed <- true;
340
+
Log.debug (fun m ->
341
+
m "Closing connection to %a" Endpoint.pp conn.pc_endpoint);
342
+
343
+
(* Cancel connection-lifetime switch first - this stops any protocol fibers *)
344
+
(try conn.pc_connection_cancel (Failure "Connection closed")
345
+
with _ -> ());
346
+
347
+
(* Call protocol cleanup *)
348
+
pool.protocol.on_close conn.pc_state;
349
+
350
+
(* Close the underlying flow *)
351
+
Eio.Cancel.protect (fun () ->
352
+
try Eio.Flow.close conn.pc_flow with _ -> ())
353
+
end
354
+
355
+
(** {1 Endpoint Pool Management} *)
356
+
357
+
let get_or_create_endpoint_pool pool endpoint =
358
+
match
359
+
Eio.Mutex.use_ro pool.endpoints_mutex (fun () ->
360
+
Hashtbl.find_opt pool.endpoints endpoint)
361
+
with
362
+
| Some ep_pool -> ep_pool
363
+
| None ->
364
+
Eio.Mutex.use_rw ~protect:true pool.endpoints_mutex (fun () ->
365
+
match Hashtbl.find_opt pool.endpoints endpoint with
366
+
| Some ep_pool -> ep_pool
367
+
| None ->
368
+
Log.info (fun m ->
369
+
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
376
+
Hashtbl.add pool.endpoints endpoint ep_pool;
377
+
ep_pool)
378
+
379
+
(** {1 Connection Acquisition} *)
380
+
381
+
let rec acquire_connection pool ep_pool endpoint =
382
+
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
404
+
405
+
(* Clean up closed connections *)
406
+
ep_pool.connections := List.filter (fun c -> not c.pc_closed) !(ep_pool.connections);
407
+
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;
415
+
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));
422
+
423
+
Log.debug (fun m ->
424
+
m "Reusing connection to %a (users=%d)"
425
+
Endpoint.pp endpoint conn.pc_active_users);
426
+
427
+
(* Notify protocol handler of acquisition *)
428
+
pool.protocol.on_acquire conn.pc_state;
429
+
conn
430
+
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);
441
+
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
448
+
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;
465
+
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);
469
+
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);
500
+
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);
504
+
505
+
Log.info (fun m ->
506
+
m "Created new connection to %a (total=%d)"
507
+
Endpoint.pp endpoint (List.length !(ep_pool.connections)));
508
+
509
+
(* Notify protocol handler of acquisition *)
510
+
pool.protocol.on_acquire conn.pc_state;
511
+
conn
512
+
end)
513
+
514
+
(** {1 Connection Release} *)
515
+
516
+
let release_connection pool ep_pool conn =
517
+
(* Notify protocol handler of release *)
518
+
pool.protocol.on_release conn.pc_state;
519
+
520
+
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
524
+
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);
530
+
531
+
(* Signal waiting fibers *)
532
+
Eio.Condition.broadcast conn.pc_user_available;
533
+
534
+
Log.debug (fun m ->
535
+
m "Released connection to %a (users=%d)"
536
+
Endpoint.pp conn.pc_endpoint conn.pc_active_users);
537
+
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;
543
+
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));
549
+
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)
553
+
554
+
| Unhealthy_lifecycle reason ->
555
+
conn.pc_closed <- true;
556
+
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))
565
+
566
+
(** {1 Public API} *)
567
+
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
571
+
| Some p -> p
572
+
| None -> Obj.magic default_protocol (* Safe: unit is compatible with any 'state *)
573
+
in
574
+
575
+
Log.info (fun m ->
576
+
m "Creating connection pool (max_per_endpoint=%d)"
577
+
(Config.max_connections_per_endpoint config));
578
+
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
589
+
590
+
(* Auto-cleanup on switch release *)
591
+
Eio.Switch.on_release sw (fun () ->
592
+
Eio.Cancel.protect (fun () ->
593
+
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;
599
+
Hashtbl.clear pool.endpoints));
600
+
601
+
Pool pool
602
+
603
+
let connection ~sw (Pool pool) endpoint =
604
+
Log.debug (fun m -> m "Acquiring connection to %a" Endpoint.pp endpoint);
605
+
606
+
let ep_pool = get_or_create_endpoint_pool pool endpoint in
607
+
let conn = acquire_connection pool ep_pool endpoint in
608
+
609
+
(* Release connection when switch ends *)
610
+
Eio.Switch.on_release sw (fun () ->
611
+
release_connection pool ep_pool conn);
612
+
613
+
(* Get TLS epoch if available *)
614
+
let tls_epoch =
615
+
match conn.pc_tls_flow with
616
+
| Some tls_flow -> (
617
+
match Tls_eio.epoch tls_flow with
618
+
| Ok epoch -> Some epoch
619
+
| Error () -> None)
620
+
| None -> None
621
+
in
622
+
623
+
{
624
+
flow = conn.pc_flow;
625
+
tls_epoch;
626
+
state = conn.pc_state;
627
+
}
628
+
629
+
let with_connection pool endpoint f =
630
+
Eio.Switch.run (fun sw -> f (connection ~sw pool endpoint))
631
+
632
+
let stats (Pool pool) endpoint =
633
+
match Hashtbl.find_opt pool.endpoints endpoint with
634
+
| Some ep_pool ->
635
+
Eio.Mutex.use_ro ep_pool.stats_mutex (fun () -> snapshot_stats ep_pool.stats)
636
+
| None ->
637
+
Stats.make ~active:0 ~idle:0 ~total_created:0 ~total_reused:0
638
+
~total_closed:0 ~errors:0
639
+
640
+
let all_stats (Pool pool) =
641
+
Eio.Mutex.use_ro pool.endpoints_mutex (fun () ->
642
+
Hashtbl.fold
643
+
(fun endpoint ep_pool acc ->
644
+
let stats =
645
+
Eio.Mutex.use_ro ep_pool.stats_mutex (fun () ->
646
+
snapshot_stats ep_pool.stats)
647
+
in
648
+
(endpoint, stats) :: acc)
649
+
pool.endpoints [])
650
+
651
+
let clear_endpoint (Pool pool) endpoint =
652
+
Log.info (fun m -> m "Clearing endpoint %a from pool" Endpoint.pp endpoint);
653
+
match Hashtbl.find_opt pool.endpoints endpoint with
654
+
| Some ep_pool ->
655
+
Eio.Cancel.protect (fun () ->
656
+
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 := []);
661
+
Eio.Mutex.use_rw ~protect:true pool.endpoints_mutex (fun () ->
662
+
Hashtbl.remove pool.endpoints endpoint))
663
+
| None ->
664
+
Log.debug (fun m ->
665
+
m "No endpoint pool found for %a" Endpoint.pp endpoint)
+178
lib/conpool.mli
+178
lib/conpool.mli
···
1
+
(*---------------------------------------------------------------------------
2
+
Copyright (c) 2025 Anil Madhavapeddy <anil@recoil.org>. All rights reserved.
3
+
SPDX-License-Identifier: ISC
4
+
---------------------------------------------------------------------------*)
5
+
6
+
(** Conpool - Protocol-aware TCP/IP connection pooling library for Eio
7
+
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.
11
+
12
+
{2 Quick Start}
13
+
14
+
For simple exclusive-access protocols (HTTP/1.x, Redis, etc.):
15
+
{[
16
+
let pool = Conpool.create ~sw ~net ~clock ~tls () in
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)
21
+
]}
22
+
23
+
For multiplexed protocols (HTTP/2):
24
+
{[
25
+
let pool = Conpool.create ~sw ~net ~clock ~tls ~protocol:h2_handler () in
26
+
Eio.Switch.run (fun conn_sw ->
27
+
let conn = Conpool.connection ~sw:conn_sw pool endpoint in
28
+
(* conn.state has H2_client.t, multiple streams share the connection *)
29
+
H2_client.request conn.flow conn.state ...)
30
+
]} *)
31
+
32
+
(** {1 Logging} *)
33
+
34
+
val src : Logs.Src.t
35
+
(** Logs source for the connection pool. Configure logging with:
36
+
{[
37
+
Logs.Src.set_level Conpool.src (Some Logs.Debug);
38
+
Logs.set_reporter (Logs_fmt.reporter ())
39
+
]} *)
40
+
41
+
(** {1 Core Types} *)
42
+
43
+
module Endpoint = Endpoint
44
+
(** Network endpoint representation *)
45
+
46
+
module Config = Config
47
+
(** Configuration for connection pools *)
48
+
49
+
module Stats = Stats
50
+
(** Statistics for connection pool endpoints *)
51
+
52
+
module Cmd = Cmd
53
+
(** Cmdliner terms for connection pool configuration *)
54
+
55
+
(** {1 Errors} *)
56
+
57
+
type error =
58
+
| Dns_resolution_failed of { hostname : string }
59
+
| Connection_failed of {
60
+
endpoint : Endpoint.t;
61
+
attempts : int;
62
+
last_error : string;
63
+
}
64
+
| Connection_timeout of { endpoint : Endpoint.t; timeout : float }
65
+
| Invalid_config of string
66
+
| Invalid_endpoint of string
67
+
68
+
type Eio.Exn.err += E of error
69
+
70
+
val err : error -> exn
71
+
(** [err e] creates an Eio exception from a connection pool error. *)
72
+
73
+
val pp_error : error Fmt.t
74
+
(** Pretty-printer for error values. *)
75
+
76
+
(** {1 Connection Types} *)
77
+
78
+
type connection_ty = [Eio.Resource.close_ty | Eio.Flow.two_way_ty]
79
+
(** Type tags for a pooled connection. *)
80
+
81
+
type connection = connection_ty Eio.Resource.t
82
+
(** A connection resource from the pool. *)
83
+
84
+
(** {1 Connection Pool}
85
+
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. *)
89
+
90
+
type 'state t
91
+
(** Connection pool with protocol-specific state ['state].
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) *)
95
+
96
+
(** Connection with protocol-specific state. *)
97
+
type 'state connection_info = {
98
+
flow : connection;
99
+
(** The underlying connection flow for I/O. *)
100
+
tls_epoch : Tls.Core.epoch_data option;
101
+
(** TLS epoch data if connection uses TLS. *)
102
+
state : 'state;
103
+
(** Protocol-specific state (e.g., H2_client.t for HTTP/2). *)
104
+
}
105
+
106
+
(** {2 Pool Creation} *)
107
+
108
+
val create :
109
+
sw:Eio.Switch.t ->
110
+
net:'net Eio.Net.t ->
111
+
clock:'clock Eio.Time.clock ->
112
+
?tls:Tls.Config.client ->
113
+
?config:Config.t ->
114
+
?protocol:'state Config.protocol_config ->
115
+
unit ->
116
+
'state t
117
+
(** Create a connection pool.
118
+
119
+
@param sw Switch for resource management
120
+
@param net Network interface for creating connections
121
+
@param clock Clock for timeouts
122
+
@param tls Optional TLS client configuration
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).
126
+
127
+
Examples:
128
+
129
+
Simple pool for HTTP/1.x (exclusive access, no state):
130
+
{[
131
+
let pool = Conpool.create ~sw ~net ~clock ~tls ()
132
+
]}
133
+
134
+
HTTP/2 pool (shared access with H2 state):
135
+
{[
136
+
let pool = Conpool.create ~sw ~net ~clock ~tls ~protocol:h2_handler ()
137
+
]} *)
138
+
139
+
(** {2 Connection Acquisition} *)
140
+
141
+
val connection : sw:Eio.Switch.t -> 'state t -> Endpoint.t -> 'state connection_info
142
+
(** [connection ~sw pool endpoint] acquires a connection from the pool.
143
+
144
+
The connection is automatically released when [sw] finishes:
145
+
- Exclusive mode: connection returns to idle pool
146
+
- Shared mode: user count is decremented
147
+
148
+
Behavior depends on access mode:
149
+
- Exclusive: blocks until a connection is available
150
+
- Shared: may share an existing connection if under max_concurrent limit
151
+
152
+
Example:
153
+
{[
154
+
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)
159
+
]} *)
160
+
161
+
val with_connection : 'state t -> Endpoint.t -> ('state connection_info -> 'a) -> 'a
162
+
(** [with_connection pool endpoint fn] is a convenience wrapper.
163
+
164
+
Equivalent to:
165
+
{[
166
+
Eio.Switch.run (fun sw -> fn (connection ~sw pool endpoint))
167
+
]} *)
168
+
169
+
(** {1 Statistics & Management} *)
170
+
171
+
val stats : 'state t -> Endpoint.t -> Stats.t
172
+
(** Get statistics for specific endpoint. *)
173
+
174
+
val all_stats : 'state t -> (Endpoint.t * Stats.t) list
175
+
(** Get statistics for all endpoints in pool. *)
176
+
177
+
val clear_endpoint : 'state t -> Endpoint.t -> unit
178
+
(** Clear all connections for an endpoint. *)
+4
lib/dune
+4
lib/dune
+30
lib/endpoint.ml
+30
lib/endpoint.ml
···
1
+
(*---------------------------------------------------------------------------
2
+
Copyright (c) 2025 Anil Madhavapeddy <anil@recoil.org>. All rights reserved.
3
+
SPDX-License-Identifier: ISC
4
+
---------------------------------------------------------------------------*)
5
+
6
+
(** Network endpoint representation *)
7
+
8
+
let src =
9
+
Logs.Src.create "conpool.endpoint" ~doc:"Connection pool endpoint operations"
10
+
11
+
module Log = (val Logs.src_log src : Logs.LOG)
12
+
13
+
type t = { host : string; port : int }
14
+
15
+
let make ~host ~port =
16
+
(* Validate port range *)
17
+
if port < 1 || port > 65535 then
18
+
invalid_arg
19
+
(Printf.sprintf "Invalid port number: %d (must be 1-65535)" port);
20
+
21
+
(* Validate hostname is not empty *)
22
+
if String.trim host = "" then invalid_arg "Hostname cannot be empty";
23
+
24
+
{ host; port }
25
+
26
+
let host t = t.host
27
+
let port t = t.port
28
+
let equal t1 t2 = String.equal t1.host t2.host && t1.port = t2.port
29
+
let hash t = Hashtbl.hash (t.host, t.port)
30
+
let pp = Fmt.of_to_string (fun t -> Printf.sprintf "%s:%d" t.host t.port)
+45
lib/endpoint.mli
+45
lib/endpoint.mli
···
1
+
(*---------------------------------------------------------------------------
2
+
Copyright (c) 2025 Anil Madhavapeddy <anil@recoil.org>. All rights reserved.
3
+
SPDX-License-Identifier: ISC
4
+
---------------------------------------------------------------------------*)
5
+
6
+
(** Network endpoint representation *)
7
+
8
+
(** {1 Logging} *)
9
+
10
+
val src : Logs.Src.t
11
+
(** Logs source for endpoint operations. Configure logging with:
12
+
{[
13
+
Logs.Src.set_level Conpool.Endpoint.src (Some Logs.Debug)
14
+
]} *)
15
+
16
+
(** {1 Type} *)
17
+
18
+
type t
19
+
(** Network endpoint identified by host and port *)
20
+
21
+
(** {1 Construction} *)
22
+
23
+
val make : host:string -> port:int -> t
24
+
(** Create an endpoint from a hostname and port. *)
25
+
26
+
(** {1 Accessors} *)
27
+
28
+
val host : t -> string
29
+
(** Get the hostname from an endpoint. *)
30
+
31
+
val port : t -> int
32
+
(** Get the port number from an endpoint. *)
33
+
34
+
(** {1 Comparison and Hashing} *)
35
+
36
+
val equal : t -> t -> bool
37
+
(** Compare two endpoints for equality. *)
38
+
39
+
val hash : t -> int
40
+
(** Hash an endpoint for use in hash tables. *)
41
+
42
+
(** {1 Pretty-printing} *)
43
+
44
+
val pp : t Fmt.t
45
+
(** Pretty-printer for endpoints. Formats as "host:port". *)
+36
lib/stats.ml
+36
lib/stats.ml
···
1
+
(*---------------------------------------------------------------------------
2
+
Copyright (c) 2025 Anil Madhavapeddy <anil@recoil.org>. All rights reserved.
3
+
SPDX-License-Identifier: ISC
4
+
---------------------------------------------------------------------------*)
5
+
6
+
(** Statistics for connection pool endpoints *)
7
+
8
+
type t = {
9
+
active : int;
10
+
idle : int;
11
+
total_created : int;
12
+
total_reused : int;
13
+
total_closed : int;
14
+
errors : int;
15
+
}
16
+
17
+
let make ~active ~idle ~total_created ~total_reused ~total_closed ~errors =
18
+
{ active; idle; total_created; total_reused; total_closed; errors }
19
+
20
+
let active t = t.active
21
+
let idle t = t.idle
22
+
let total_created t = t.total_created
23
+
let total_reused t = t.total_reused
24
+
let total_closed t = t.total_closed
25
+
let errors t = t.errors
26
+
27
+
let pp ppf t =
28
+
Fmt.pf ppf
29
+
"@[<v>Stats:@,\
30
+
- Active: %d@,\
31
+
- Idle: %d@,\
32
+
- Created: %d@,\
33
+
- Reused: %d@,\
34
+
- Closed: %d@,\
35
+
- Errors: %d@]"
36
+
t.active t.idle t.total_created t.total_reused t.total_closed t.errors
+48
lib/stats.mli
+48
lib/stats.mli
···
1
+
(*---------------------------------------------------------------------------
2
+
Copyright (c) 2025 Anil Madhavapeddy <anil@recoil.org>. All rights reserved.
3
+
SPDX-License-Identifier: ISC
4
+
---------------------------------------------------------------------------*)
5
+
6
+
(** Statistics for connection pool endpoints *)
7
+
8
+
(** {1 Type} *)
9
+
10
+
type t
11
+
(** Statistics snapshot for a specific endpoint *)
12
+
13
+
(** {1 Construction} *)
14
+
15
+
val make :
16
+
active:int ->
17
+
idle:int ->
18
+
total_created:int ->
19
+
total_reused:int ->
20
+
total_closed:int ->
21
+
errors:int ->
22
+
t
23
+
(** Create a statistics snapshot. *)
24
+
25
+
(** {1 Accessors} *)
26
+
27
+
val active : t -> int
28
+
(** Number of connections currently in use. *)
29
+
30
+
val idle : t -> int
31
+
(** Number of connections in pool waiting to be reused. *)
32
+
33
+
val total_created : t -> int
34
+
(** Total connections created over the endpoint's lifetime. *)
35
+
36
+
val total_reused : t -> int
37
+
(** Total number of times connections were reused from the pool. *)
38
+
39
+
val total_closed : t -> int
40
+
(** Total connections that have been closed. *)
41
+
42
+
val errors : t -> int
43
+
(** Total connection errors encountered. *)
44
+
45
+
(** {1 Pretty-printing} *)
46
+
47
+
val pp : t Fmt.t
48
+
(** Pretty-printer for statistics. *)
+9
test/dune
+9
test/dune
+993
test/stress_test.ml
+993
test/stress_test.ml
···
1
+
(*---------------------------------------------------------------------------
2
+
Copyright (c) 2025 Anil Madhavapeddy <anil@recoil.org>. All rights reserved.
3
+
SPDX-License-Identifier: ISC
4
+
---------------------------------------------------------------------------*)
5
+
6
+
(** Stress test framework for conpool
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
+
*)
12
+
13
+
(** Configuration for the stress test *)
14
+
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 *)
18
+
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 *)
22
+
}
23
+
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";
56
+
num_servers = 3;
57
+
num_clients = 100;
58
+
messages_per_client = 5;
59
+
max_parallel_clients = 100;
60
+
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
+
]
91
+
92
+
(** 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
+
}
102
+
103
+
(** Statistics collected during test *)
104
+
type latency_stats = {
105
+
mutable count : int;
106
+
mutable total : float;
107
+
mutable min : float;
108
+
mutable max : float;
109
+
mutable latencies : (float * float) list; (* (timestamp, latency) pairs *)
110
+
}
111
+
112
+
let create_latency_stats () = {
113
+
count = 0;
114
+
total = 0.0;
115
+
min = Float.infinity;
116
+
max = 0.0;
117
+
latencies = [];
118
+
}
119
+
120
+
let update_latency stats latency timestamp =
121
+
stats.count <- stats.count + 1;
122
+
stats.total <- stats.total +. latency;
123
+
stats.min <- min stats.min latency;
124
+
stats.max <- max stats.max latency;
125
+
stats.latencies <- (timestamp, latency) :: stats.latencies
126
+
127
+
(** Generate a random message of given size *)
128
+
let generate_message size =
129
+
let chars = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789" in
130
+
let len = String.length chars in
131
+
String.init size (fun _ -> chars.[Random.int len])
132
+
133
+
(** Echo server handler - echoes back everything it receives *)
134
+
let handle_echo_client flow _addr =
135
+
let buf = Cstruct.create 4096 in
136
+
let rec loop () =
137
+
match Eio.Flow.single_read flow buf with
138
+
| n ->
139
+
let data = Cstruct.sub buf 0 n in
140
+
Eio.Flow.write flow [data];
141
+
loop ()
142
+
| exception End_of_file -> ()
143
+
in
144
+
loop ()
145
+
146
+
(** Start an echo server on a random port, returns the port number *)
147
+
let start_echo_server ~sw net =
148
+
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
150
+
let actual_addr = Eio.Net.listening_addr listening_socket in
151
+
let port = match actual_addr with
152
+
| `Tcp (_, port) -> port
153
+
| _ -> failwith "Expected TCP address"
154
+
in
155
+
156
+
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
+
);
167
+
168
+
port
169
+
170
+
(** Client test: connect via pool, send message, verify echo *)
171
+
let run_client_test ~clock ~test_start_time pool endpoint message latency_stats errors =
172
+
let msg_len = String.length message in
173
+
let start_time = Eio.Time.now clock in
174
+
175
+
try
176
+
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;
180
+
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
184
+
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 *)
188
+
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
197
+
198
+
(** 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 =
200
+
for _ = 1 to cfg.messages_per_client do
201
+
let endpoint_idx = Random.int (Array.length endpoints) in
202
+
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
205
+
done
206
+
207
+
(** Pool statistics aggregated from all endpoints *)
208
+
type pool_stats = {
209
+
total_created : int;
210
+
total_reused : int;
211
+
total_closed : int;
212
+
active : int;
213
+
idle : int;
214
+
pool_errors : int;
215
+
}
216
+
217
+
(** Test result type *)
218
+
type test_result = {
219
+
test_name : string;
220
+
num_servers : int;
221
+
num_clients : int;
222
+
messages_per_client : int;
223
+
pool_size : int;
224
+
duration : float;
225
+
total_messages : int;
226
+
total_errors : int;
227
+
throughput : float;
228
+
avg_latency : float;
229
+
min_latency : float;
230
+
max_latency : float;
231
+
latency_data : (float * float) list; (* (timestamp, latency) pairs for visualization *)
232
+
pool_stats : pool_stats;
233
+
}
234
+
235
+
(** Main stress test runner - returns a test result *)
236
+
let run_stress_test ~env (cfg : config) : test_result =
237
+
let net = Eio.Stdenv.net env in
238
+
let clock = Eio.Stdenv.clock env in
239
+
240
+
let latency_stats = create_latency_stats () in
241
+
let errors = ref 0 in
242
+
let ports = ref [||] in
243
+
244
+
let result : test_result option ref = ref None in
245
+
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
+
);
253
+
254
+
Eio.Time.sleep clock 0.05;
255
+
256
+
let endpoints = Array.map (fun port ->
257
+
Conpool.Endpoint.make ~host:"127.0.0.1" ~port
258
+
) !ports in
259
+
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
269
+
270
+
let pool = Conpool.create ~sw ~net ~clock ~config:pool_config () in
271
+
272
+
(* Record start time *)
273
+
let start_time = Eio.Time.now clock in
274
+
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;
282
+
283
+
let end_time = Eio.Time.now clock in
284
+
let duration = end_time -. start_time in
285
+
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) ->
289
+
{
290
+
total_created = acc.total_created + Conpool.Stats.total_created stats;
291
+
total_reused = acc.total_reused + Conpool.Stats.total_reused stats;
292
+
total_closed = acc.total_closed + Conpool.Stats.total_closed stats;
293
+
active = acc.active + Conpool.Stats.active stats;
294
+
idle = acc.idle + Conpool.Stats.idle stats;
295
+
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
298
+
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;
319
+
320
+
Eio.Switch.fail sw Exit
321
+
with Exit -> ()
322
+
end;
323
+
324
+
match !result with
325
+
| Some r -> r
326
+
| None -> failwith "Test failed to produce result"
327
+
328
+
(** Convert result to JSON string *)
329
+
let result_to_json result =
330
+
Printf.sprintf {|{
331
+
"test_name": "%s",
332
+
"num_servers": %d,
333
+
"num_clients": %d,
334
+
"messages_per_client": %d,
335
+
"duration": %.3f,
336
+
"total_messages": %d,
337
+
"total_errors": %d,
338
+
"throughput": %.2f,
339
+
"avg_latency": %.2f,
340
+
"min_latency": %.2f,
341
+
"max_latency": %.2f
342
+
}|}
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
353
+
result.max_latency
354
+
355
+
(** Escape strings for JavaScript *)
356
+
let js_escape s =
357
+
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;
367
+
Buffer.contents buf
368
+
369
+
(** Calculate histogram buckets for latency data *)
370
+
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
376
+
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;
382
+
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)
389
+
390
+
(** Generate HTML report from test results *)
391
+
let generate_html_report results =
392
+
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
400
+
in
401
+
402
+
(* 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
406
+
407
+
(* 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
416
+
417
+
(* 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
422
+
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
430
+
431
+
Printf.sprintf {|
432
+
<div class="test-detail">
433
+
<h3>%s</h3>
434
+
<div class="compact-grid">
435
+
<div class="compact-metric"><span class="label">Servers:</span> <span class="value">%d</span></div>
436
+
<div class="compact-metric"><span class="label">Clients:</span> <span class="value">%d</span></div>
437
+
<div class="compact-metric"><span class="label">Msgs/Client:</span> <span class="value">%d</span></div>
438
+
<div class="compact-metric"><span class="label">Pool Size:</span> <span class="value">%d</span></div>
439
+
<div class="compact-metric"><span class="label">Total Msgs:</span> <span class="value">%d</span></div>
440
+
<div class="compact-metric"><span class="label">Duration:</span> <span class="value">%.2fs</span></div>
441
+
<div class="compact-metric highlight"><span class="label">Throughput:</span> <span class="value">%.0f/s</span></div>
442
+
<div class="compact-metric highlight"><span class="label">Avg Lat:</span> <span class="value">%.2fms</span></div>
443
+
<div class="compact-metric"><span class="label">Min Lat:</span> <span class="value">%.2fms</span></div>
444
+
<div class="compact-metric"><span class="label">Max Lat:</span> <span class="value">%.2fms</span></div>
445
+
<div class="compact-metric %s"><span class="label">Errors:</span> <span class="value">%d</span></div>
446
+
</div>
447
+
<div class="compact-grid" style="margin-top: 0.5rem;">
448
+
<div class="compact-metric"><span class="label">Conns Created:</span> <span class="value">%d</span></div>
449
+
<div class="compact-metric"><span class="label">Conns Reused:</span> <span class="value">%d</span></div>
450
+
<div class="compact-metric"><span class="label">Conns Closed:</span> <span class="value">%d</span></div>
451
+
<div class="compact-metric"><span class="label">Active:</span> <span class="value">%d</span></div>
452
+
<div class="compact-metric"><span class="label">Idle:</span> <span class="value">%d</span></div>
453
+
<div class="compact-metric"><span class="label">Reuse Rate:</span> <span class="value">%.1f%%%%</span></div>
454
+
</div>
455
+
<div class="chart-row">
456
+
<div class="chart-half">
457
+
<h4>Latency Distribution</h4>
458
+
<canvas id="hist_%d"></canvas>
459
+
</div>
460
+
<div class="chart-half">
461
+
<h4>Latency Timeline</h4>
462
+
<canvas id="timeline_%d"></canvas>
463
+
</div>
464
+
</div>
465
+
</div>
466
+
<script>
467
+
new Chart(document.getElementById('hist_%d'), {
468
+
type: 'bar',
469
+
data: {
470
+
labels: [%s],
471
+
datasets: [{
472
+
label: 'Count',
473
+
data: [%s],
474
+
backgroundColor: 'rgba(102, 126, 234, 0.6)',
475
+
borderColor: 'rgba(102, 126, 234, 1)',
476
+
borderWidth: 1
477
+
}]
478
+
},
479
+
options: {
480
+
responsive: true,
481
+
maintainAspectRatio: false,
482
+
plugins: { legend: { display: false } },
483
+
scales: {
484
+
x: { title: { display: true, text: 'Latency (ms)' } },
485
+
y: { beginAtZero: true, title: { display: true, text: 'Count' } }
486
+
}
487
+
}
488
+
});
489
+
490
+
new Chart(document.getElementById('timeline_%d'), {
491
+
type: 'scatter',
492
+
data: {
493
+
datasets: [{
494
+
label: 'Latency',
495
+
data: [%s],
496
+
backgroundColor: 'rgba(118, 75, 162, 0.5)',
497
+
borderColor: 'rgba(118, 75, 162, 0.8)',
498
+
pointRadius: 2
499
+
}]
500
+
},
501
+
options: {
502
+
responsive: true,
503
+
maintainAspectRatio: false,
504
+
plugins: { legend: { display: false } },
505
+
scales: {
506
+
x: { title: { display: true, text: 'Time (ms)' } },
507
+
y: { beginAtZero: true, title: { display: true, text: 'Latency (ms)' } }
508
+
}
509
+
}
510
+
});
511
+
</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
539
+
540
+
Printf.sprintf {|<!DOCTYPE html>
541
+
<html lang="en">
542
+
<head>
543
+
<meta charset="UTF-8">
544
+
<meta name="viewport" content="width=device-width, initial-scale=1.0">
545
+
<title>Connection Pool Stress Test Results</title>
546
+
<script src="https://cdn.jsdelivr.net/npm/chart.js@4.4.0/dist/chart.umd.min.js"></script>
547
+
<style>
548
+
* { margin: 0; padding: 0; box-sizing: border-box; }
549
+
body {
550
+
font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, sans-serif;
551
+
background: #f5f5f5;
552
+
padding: 1rem;
553
+
color: #333;
554
+
font-size: 14px;
555
+
}
556
+
.container { max-width: 1600px; margin: 0 auto; }
557
+
h1 {
558
+
color: #667eea;
559
+
text-align: center;
560
+
margin-bottom: 0.3rem;
561
+
font-size: 1.8rem;
562
+
}
563
+
.subtitle {
564
+
text-align: center;
565
+
margin-bottom: 1rem;
566
+
font-size: 0.9rem;
567
+
color: #666;
568
+
}
569
+
.summary {
570
+
background: white;
571
+
border-radius: 6px;
572
+
padding: 1rem;
573
+
margin-bottom: 1rem;
574
+
box-shadow: 0 2px 4px rgba(0,0,0,0.1);
575
+
}
576
+
.summary h2 {
577
+
color: #667eea;
578
+
margin-bottom: 0.8rem;
579
+
font-size: 1.2rem;
580
+
}
581
+
.summary-grid {
582
+
display: grid;
583
+
grid-template-columns: repeat(auto-fit, minmax(120px, 1fr));
584
+
gap: 0.8rem;
585
+
}
586
+
.summary-metric {
587
+
text-align: center;
588
+
padding: 0.8rem;
589
+
background: linear-gradient(135deg, #667eea 0%%, #764ba2 100%%);
590
+
border-radius: 4px;
591
+
color: white;
592
+
}
593
+
.summary-metric-label {
594
+
font-size: 0.75rem;
595
+
opacity: 0.9;
596
+
margin-bottom: 0.3rem;
597
+
}
598
+
.summary-metric-value {
599
+
font-size: 1.4rem;
600
+
font-weight: bold;
601
+
}
602
+
.comparison {
603
+
background: white;
604
+
border-radius: 6px;
605
+
padding: 1rem;
606
+
margin-bottom: 1rem;
607
+
box-shadow: 0 2px 4px rgba(0,0,0,0.1);
608
+
}
609
+
.comparison h2 {
610
+
color: #667eea;
611
+
margin-bottom: 0.8rem;
612
+
font-size: 1.2rem;
613
+
}
614
+
.comparison-charts {
615
+
display: grid;
616
+
grid-template-columns: repeat(3, 1fr);
617
+
gap: 1rem;
618
+
}
619
+
.comparison-chart {
620
+
height: 200px;
621
+
position: relative;
622
+
}
623
+
.test-detail {
624
+
background: white;
625
+
border-radius: 6px;
626
+
padding: 1rem;
627
+
margin-bottom: 1rem;
628
+
box-shadow: 0 2px 4px rgba(0,0,0,0.1);
629
+
border-left: 3px solid #667eea;
630
+
}
631
+
.test-detail h3 {
632
+
color: #764ba2;
633
+
margin-bottom: 0.6rem;
634
+
font-size: 1.1rem;
635
+
}
636
+
.test-detail h4 {
637
+
color: #666;
638
+
margin-bottom: 0.4rem;
639
+
font-size: 0.9rem;
640
+
font-weight: 500;
641
+
}
642
+
.compact-grid {
643
+
display: grid;
644
+
grid-template-columns: repeat(auto-fit, minmax(100px, 1fr));
645
+
gap: 0.4rem;
646
+
margin-bottom: 0.8rem;
647
+
font-size: 0.85rem;
648
+
}
649
+
.compact-metric {
650
+
background: #f8f9fa;
651
+
padding: 0.4rem 0.6rem;
652
+
border-radius: 3px;
653
+
display: flex;
654
+
justify-content: space-between;
655
+
align-items: center;
656
+
}
657
+
.compact-metric .label {
658
+
color: #666;
659
+
font-weight: 500;
660
+
}
661
+
.compact-metric .value {
662
+
color: #333;
663
+
font-weight: 600;
664
+
}
665
+
.compact-metric.highlight {
666
+
background: linear-gradient(135deg, #667eea 0%%, #764ba2 100%%);
667
+
color: white;
668
+
}
669
+
.compact-metric.highlight .label,
670
+
.compact-metric.highlight .value {
671
+
color: white;
672
+
}
673
+
.compact-metric.error {
674
+
background: #fee;
675
+
border: 1px solid #fcc;
676
+
}
677
+
.chart-row {
678
+
display: grid;
679
+
grid-template-columns: 1fr 1fr;
680
+
gap: 1rem;
681
+
}
682
+
.chart-half {
683
+
position: relative;
684
+
height: 220px;
685
+
}
686
+
@media (max-width: 1200px) {
687
+
.comparison-charts { grid-template-columns: 1fr; }
688
+
.chart-row { grid-template-columns: 1fr; }
689
+
}
690
+
@media (max-width: 768px) {
691
+
.compact-grid { grid-template-columns: repeat(2, 1fr); }
692
+
}
693
+
</style>
694
+
</head>
695
+
<body>
696
+
<div class="container">
697
+
<h1>Connection Pool Stress Test Results</h1>
698
+
<div class="subtitle">%s</div>
699
+
700
+
<div class="summary">
701
+
<h2>Summary</h2>
702
+
<div class="summary-grid">
703
+
<div class="summary-metric">
704
+
<div class="summary-metric-label">Tests</div>
705
+
<div class="summary-metric-value">%d</div>
706
+
</div>
707
+
<div class="summary-metric">
708
+
<div class="summary-metric-label">Messages</div>
709
+
<div class="summary-metric-value">%s</div>
710
+
</div>
711
+
<div class="summary-metric">
712
+
<div class="summary-metric-label">Errors</div>
713
+
<div class="summary-metric-value">%d</div>
714
+
</div>
715
+
<div class="summary-metric">
716
+
<div class="summary-metric-label">Duration</div>
717
+
<div class="summary-metric-value">%.1fs</div>
718
+
</div>
719
+
</div>
720
+
</div>
721
+
722
+
<div class="comparison">
723
+
<h2>Comparison</h2>
724
+
<div class="comparison-charts">
725
+
<div class="comparison-chart"><canvas id="cmpThroughput"></canvas></div>
726
+
<div class="comparison-chart"><canvas id="cmpLatency"></canvas></div>
727
+
<div class="comparison-chart"><canvas id="cmpErrors"></canvas></div>
728
+
</div>
729
+
</div>
730
+
731
+
%s
732
+
</div>
733
+
734
+
<script>
735
+
const testNames = [%s];
736
+
const throughputs = [%s];
737
+
const avgLatencies = [%s];
738
+
const errorRates = [%s];
739
+
740
+
const cc = {
741
+
primary: 'rgba(102, 126, 234, 0.8)',
742
+
secondary: 'rgba(118, 75, 162, 0.8)',
743
+
danger: 'rgba(220, 53, 69, 0.8)',
744
+
};
745
+
746
+
new Chart(document.getElementById('cmpThroughput'), {
747
+
type: 'bar',
748
+
data: {
749
+
labels: testNames,
750
+
datasets: [{
751
+
label: 'msg/s',
752
+
data: throughputs,
753
+
backgroundColor: cc.primary,
754
+
borderColor: cc.primary,
755
+
borderWidth: 1
756
+
}]
757
+
},
758
+
options: {
759
+
responsive: true,
760
+
maintainAspectRatio: false,
761
+
plugins: {
762
+
legend: { display: false },
763
+
title: { display: true, text: 'Throughput (msg/s)' }
764
+
},
765
+
scales: { y: { beginAtZero: true } }
766
+
}
767
+
});
768
+
769
+
new Chart(document.getElementById('cmpLatency'), {
770
+
type: 'bar',
771
+
data: {
772
+
labels: testNames,
773
+
datasets: [{
774
+
label: 'ms',
775
+
data: avgLatencies,
776
+
backgroundColor: cc.secondary,
777
+
borderColor: cc.secondary,
778
+
borderWidth: 1
779
+
}]
780
+
},
781
+
options: {
782
+
responsive: true,
783
+
maintainAspectRatio: false,
784
+
plugins: {
785
+
legend: { display: false },
786
+
title: { display: true, text: 'Avg Latency (ms)' }
787
+
},
788
+
scales: { y: { beginAtZero: true } }
789
+
}
790
+
});
791
+
792
+
new Chart(document.getElementById('cmpErrors'), {
793
+
type: 'bar',
794
+
data: {
795
+
labels: testNames,
796
+
datasets: [{
797
+
label: '%%',
798
+
data: errorRates,
799
+
backgroundColor: cc.danger,
800
+
borderColor: cc.danger,
801
+
borderWidth: 1
802
+
}]
803
+
},
804
+
options: {
805
+
responsive: true,
806
+
maintainAspectRatio: false,
807
+
plugins: {
808
+
legend: { display: false },
809
+
title: { display: true, text: 'Error Rate (%%)' }
810
+
},
811
+
scales: { y: { beginAtZero: true } }
812
+
}
813
+
});
814
+
</script>
815
+
</body>
816
+
</html>|}
817
+
date_str
818
+
(List.length results)
819
+
(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
830
+
831
+
(** Run all preset tests and return results *)
832
+
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
837
+
838
+
(** Parse command line arguments *)
839
+
type mode =
840
+
| Single of config
841
+
| AllPresets
842
+
| Extended
843
+
| ListPresets
844
+
845
+
let parse_args () =
846
+
let mode = ref (Single default_config) in
847
+
let name = ref default_config.name in
848
+
let num_servers = ref default_config.num_servers in
849
+
let num_clients = ref default_config.num_clients in
850
+
let messages_per_client = ref default_config.messages_per_client in
851
+
let max_parallel = ref default_config.max_parallel_clients in
852
+
let message_size = ref default_config.message_size in
853
+
let pool_size = ref default_config.pool_size in
854
+
let output_file = ref "stress_test_results.json" in
855
+
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
877
+
878
+
let usage = "Usage: stress_test [options]\n\nOptions:" in
879
+
Arg.parse specs (fun _ -> ()) usage;
880
+
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
890
+
891
+
(!mode, config, !output_file)
892
+
893
+
let () =
894
+
Random.self_init ();
895
+
let (mode, custom_config, output_file) = parse_args () in
896
+
897
+
match mode with
898
+
| ListPresets ->
899
+
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
+
905
+
| Single config ->
906
+
let config = if config.name = "default" then custom_config else config in
907
+
Eio_main.run @@ fun env ->
908
+
let result = run_stress_test ~env config in
909
+
let results = [result] in
910
+
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;
917
+
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;
930
+
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
+
934
+
| AllPresets ->
935
+
Eio_main.run @@ fun env ->
936
+
let results = run_all_presets ~env in
937
+
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;
944
+
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;
957
+
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
+
963
+
| 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;
966
+
Printf.printf "Total messages: %d\n%!"
967
+
(extended_preset.num_servers * extended_preset.num_clients * extended_preset.messages_per_client);
968
+
Eio_main.run @@ fun env ->
969
+
let result = run_stress_test ~env extended_preset in
970
+
let results = [result] in
971
+
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;
978
+
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;
991
+
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