commit 4b5aad70a424917e1bdc12d468e0abb60407fc52 · anil.recoil.org/ocaml-conpool

+28

lib/config.ml

···

       30
       30
        
           ?on_connection_closed

     

       31
       31
        
           ?on_connection_reused

     

       32
       32
        
           () =

     

       33
       33
       +
         (* Validate parameters *)

     

       34
       34
       +
         if max_connections_per_endpoint <= 0 then

     

       35
       35
       +
           invalid_arg (Printf.sprintf "max_connections_per_endpoint must be positive, got %d"

     

       36
       36
       +
             max_connections_per_endpoint);

     

       37
       37
       +
       

     

       38
       38
       +
         if max_idle_time <= 0.0 then

     

       39
       39
       +
           invalid_arg (Printf.sprintf "max_idle_time must be positive, got %.2f" max_idle_time);

     

       40
       40
       +
       

     

       41
       41
       +
         if max_connection_lifetime <= 0.0 then

     

       42
       42
       +
           invalid_arg (Printf.sprintf "max_connection_lifetime must be positive, got %.2f"

     

       43
       43
       +
             max_connection_lifetime);

     

       44
       44
       +
       

     

       45
       45
       +
         (match max_connection_uses with

     

       46
       46
       +
          | Some n when n <= 0 ->

     

       47
       47
       +
              invalid_arg (Printf.sprintf "max_connection_uses must be positive, got %d" n)

     

       48
       48
       +
          | _ -> ());

     

       49
       49
       +
       

     

       50
       50
       +
         if connect_timeout <= 0.0 then

     

       51
       51
       +
           invalid_arg (Printf.sprintf "connect_timeout must be positive, got %.2f" connect_timeout);

     

       52
       52
       +
       

     

       53
       53
       +
         if connect_retry_count < 0 then

     

       54
       54
       +
           invalid_arg (Printf.sprintf "connect_retry_count must be non-negative, got %d"

     

       55
       55
       +
             connect_retry_count);

     

       56
       56
       +
       

     

       57
       57
       +
         if connect_retry_delay <= 0.0 then

     

       58
       58
       +
           invalid_arg (Printf.sprintf "connect_retry_delay must be positive, got %.2f"

     

       59
       59
       +
             connect_retry_delay);

     

       60
       60
       +
       

     

       33
       61
        
         Log.debug (fun m ->

     

       34
       62
        
           m "Creating config: max_connections=%d, max_idle=%.1fs, max_lifetime=%.1fs"

     

       35
       63
        
             max_connections_per_endpoint max_idle_time max_connection_lifetime);

+15 -3

lib/connection.ml

···

       9
       9
        
         mutable last_used : float;

     

       10
       10
        
         mutable use_count : int;

     

       11
       11
        
         endpoint : Endpoint.t;

     

       12
       12
       +
         mutex : Eio.Mutex.t;

     

       12
       13
        
       }

     

       13
       14
        
       

     

       14
       15
        
       let flow t = t.flow

     

       15
       16
        
       let endpoint t = t.endpoint

     

       16
       17
        
       let created_at t = t.created_at

     

       17
       17
       -
       let last_used t = t.last_used

     

       18
       18
       -
       let use_count t = t.use_count

     

       18
       18
       +
       

     

       19
       19
       +
       let last_used t =

     

       20
       20
       +
         Eio.Mutex.use_ro t.mutex (fun () -> t.last_used)

     

       21
       21
       +
       

     

       22
       22
       +
       let use_count t =

     

       23
       23
       +
         Eio.Mutex.use_ro t.mutex (fun () -> t.use_count)

     

       24
       24
       +
       

     

       25
       25
       +
       let update_usage t ~now =

     

       26
       26
       +
         Eio.Mutex.use_rw ~protect:true t.mutex (fun () ->

     

       27
       27
       +
           t.last_used <- now;

     

       28
       28
       +
           t.use_count <- t.use_count + 1

     

       29
       29
       +
         )

     

       19
       30
        
       

     

       20
       31
        
       let pp ppf t =

     

       32
       32
       +
         let uses = Eio.Mutex.use_ro t.mutex (fun () -> t.use_count) in

     

       21
       33
        
         Fmt.pf ppf "Connection(endpoint=%a, age=%.2fs, uses=%d)"

     

       22
       34
        
           Endpoint.pp t.endpoint

     

       23
       35
        
           (Unix.gettimeofday () -. t.created_at)

     

       24
       24
       -
           t.use_count

     

       36
       36
       +
           uses

+50 -15

lib/conpool.ml

···

       10
       10
        
       module Stats = Stats

     

       11
       11
        
       module Cmd = Cmd

     

       12
       12
        
       

     

       13
       13
       +
       (** {1 Error Types} *)

     

       14
       14
       +
       

     

       15
       15
       +
       type error =

     

       16
       16
       +
         | Dns_resolution_failed of { hostname : string }

     

       17
       17
       +
         | Connection_failed of { endpoint : Endpoint.t; attempts : int; last_error : string }

     

       18
       18
       +
         | Connection_timeout of { endpoint : Endpoint.t; timeout : float }

     

       19
       19
       +
         | Invalid_config of string

     

       20
       20
       +
         | Invalid_endpoint of string

     

       21
       21
       +
       

     

       22
       22
       +
       exception Pool_error of error

     

       23
       23
       +
       

     

       24
       24
       +
       let pp_error ppf = function

     

       25
       25
       +
         | Dns_resolution_failed { hostname } ->

     

       26
       26
       +
             Fmt.pf ppf "DNS resolution failed for hostname: %s" hostname

     

       27
       27
       +
         | Connection_failed { endpoint; attempts; last_error } ->

     

       28
       28
       +
             Fmt.pf ppf "Failed to connect to %a after %d attempts: %s"

     

       29
       29
       +
               Endpoint.pp endpoint attempts last_error

     

       30
       30
       +
         | Connection_timeout { endpoint; timeout } ->

     

       31
       31
       +
             Fmt.pf ppf "Connection timeout to %a after %.2fs"

     

       32
       32
       +
               Endpoint.pp endpoint timeout

     

       33
       33
       +
         | Invalid_config msg ->

     

       34
       34
       +
             Fmt.pf ppf "Invalid configuration: %s" msg

     

       35
       35
       +
         | Invalid_endpoint msg ->

     

       36
       36
       +
             Fmt.pf ppf "Invalid endpoint: %s" msg

     

       37
       37
       +
       

     

       13
       38
        
       type endp_stats = {

     

       14
       39
        
         mutable active : int;

     

       15
       40
        
         mutable idle : int;

     
···

       77
       102
        
             addr

     

       78
       103
        
         | [] ->

     

       79
       104
        
             Log.err (fun m -> m "Failed to resolve hostname: %s" (Endpoint.host endpoint));

     

       80
       80
       -
             failwith (Printf.sprintf "Failed to resolve hostname: %s" (Endpoint.host endpoint))

     

       105
       105
       +
             raise (Pool_error (Dns_resolution_failed { hostname = Endpoint.host endpoint }))

     

       81
       106
        
       

     

       82
       107
        
       (** {1 Connection Creation with Retry} *)

     

       83
       108
        
       

     

       84
       84
       -
       let rec create_connection_with_retry (pool : ('clock, 'net) internal) endpoint attempt =

     

       109
       109
       +
       let rec create_connection_with_retry (pool : ('clock, 'net) internal) endpoint attempt last_error =

     

       85
       110
        
         let retry_count = Config.connect_retry_count pool.config in

     

       86
       111
        
         if attempt > retry_count then begin

     

       87
       112
        
           Log.err (fun m -> m "Failed to connect to %a after %d attempts"

     

       88
       113
        
             Endpoint.pp endpoint retry_count);

     

       89
       89
       -
           failwith (Printf.sprintf "Failed to connect to %s:%d after %d attempts"

     

       90
       90
       -
             (Endpoint.host endpoint) (Endpoint.port endpoint) retry_count)

     

       114
       114
       +
           raise (Pool_error (Connection_failed { endpoint; attempts = retry_count; last_error }))

     

       91
       115
        
         end;

     

       92
       116
        
       

     

       93
       117
        
         Log.debug (fun m -> m "Connecting to %a (attempt %d/%d)"

     
···

       130
       154
        
             last_used = now;

     

       131
       155
        
             use_count = 0;

     

       132
       156
        
             endpoint;

     

       157
       157
       +
             mutex = Eio.Mutex.create ();

     

       133
       158
        
           }

     

       134
       159
        
       

     

       135
       160
        
         with

     

       136
       136
       -
         | Eio.Time.Timeout ->

     

       161
       161
       +
         | Eio.Time.Timeout as e ->

     

       137
       162
        
             Log.warn (fun m -> m "Connection timeout to %a (attempt %d)" Endpoint.pp endpoint attempt);

     

       138
       138
       -
             (* Exponential backoff *)

     

       139
       139
       -
             let delay = Config.connect_retry_delay pool.config *. (2.0 ** float_of_int (attempt - 1)) in

     

       140
       140
       -
             Eio.Time.sleep pool.clock delay;

     

       141
       141
       -
             create_connection_with_retry pool endpoint (attempt + 1)

     

       163
       163
       +
             let error_msg = Printexc.to_string e in

     

       164
       164
       +
             if attempt >= Config.connect_retry_count pool.config then

     

       165
       165
       +
               (* Last attempt - convert to our error type *)

     

       166
       166
       +
               match Config.connect_timeout pool.config with

     

       167
       167
       +
               | Some timeout ->

     

       168
       168
       +
                   raise (Pool_error (Connection_timeout { endpoint; timeout }))

     

       169
       169
       +
               | None ->

     

       170
       170
       +
                   raise (Pool_error (Connection_failed { endpoint; attempts = attempt; last_error = error_msg }))

     

       171
       171
       +
             else begin

     

       172
       172
       +
               (* Retry with exponential backoff *)

     

       173
       173
       +
               let delay = Config.connect_retry_delay pool.config *. (2.0 ** float_of_int (attempt - 1)) in

     

       174
       174
       +
               Eio.Time.sleep pool.clock delay;

     

       175
       175
       +
               create_connection_with_retry pool endpoint (attempt + 1) error_msg

     

       176
       176
       +
             end

     

       142
       177
        
         | e ->

     

       143
       178
        
             (* Other errors - retry with backoff *)

     

       179
       179
       +
             let error_msg = Printexc.to_string e in

     

       144
       180
        
             Log.warn (fun m -> m "Connection attempt %d to %a failed: %s"

     

       145
       145
       -
               attempt Endpoint.pp endpoint (Printexc.to_string e));

     

       181
       181
       +
               attempt Endpoint.pp endpoint error_msg);

     

       146
       182
        
             if attempt < Config.connect_retry_count pool.config then (

     

       147
       183
        
               let delay = Config.connect_retry_delay pool.config *. (2.0 ** float_of_int (attempt - 1)) in

     

       148
       184
        
               Eio.Time.sleep pool.clock delay;

     

       149
       149
       -
               create_connection_with_retry pool endpoint (attempt + 1)

     

       185
       185
       +
               create_connection_with_retry pool endpoint (attempt + 1) error_msg

     

       150
       186
        
             ) else

     

       151
       151
       -
               raise e

     

       187
       187
       +
               raise (Pool_error (Connection_failed { endpoint; attempts = attempt; last_error = error_msg }))

     

       152
       188
        
       

     

       153
       189
        
       let create_connection (pool : ('clock, 'net) internal) endpoint =

     

       154
       154
       -
         create_connection_with_retry pool endpoint 1

     

       190
       190
       +
         create_connection_with_retry pool endpoint 1 "No attempts made"

     

       155
       191
        
       

     

       156
       192
        
       (** {1 Connection Validation} *)

     

       157
       193
        
       

     
···

       406
       442
        
                 Endpoint.pp endpoint (Connection.use_count conn));

     

       407
       443
        
       

     

       408
       444
        
               (* Update last used time and use count *)

     

       409
       409
       -
               conn.last_used <- get_time pool;

     

       410
       410
       -
               conn.use_count <- conn.use_count + 1;

     

       445
       445
       +
               Connection.update_usage conn ~now:(get_time pool);

     

       411
       446
        
       

     

       412
       447
        
               (* Update idle stats (connection taken from idle pool) *)

     

       413
       448
        
               Eio.Mutex.use_rw ~protect:true ep_pool.mutex (fun () ->

+27

lib/conpool.mli

···

       32
       32
        
       (** Cmdliner terms for connection pool configuration *)

     

       33
       33
        
       module Cmd : module type of Cmd

     

       34
       34
        
       

     

       35
       35
       +
       (** {1 Errors} *)

     

       36
       36
       +
       

     

       37
       37
       +
       type error =

     

       38
       38
       +
         | Dns_resolution_failed of { hostname : string }

     

       39
       39
       +
         (** DNS resolution failed for the given hostname *)

     

       40
       40
       +
       

     

       41
       41
       +
         | Connection_failed of { endpoint : Endpoint.t; attempts : int; last_error : string }

     

       42
       42
       +
         (** Failed to establish connection after all retry attempts *)

     

       43
       43
       +
       

     

       44
       44
       +
         | Connection_timeout of { endpoint : Endpoint.t; timeout : float }

     

       45
       45
       +
         (** Connection attempt timed out *)

     

       46
       46
       +
       

     

       47
       47
       +
         | Invalid_config of string

     

       48
       48
       +
         (** Invalid configuration parameter *)

     

       49
       49
       +
       

     

       50
       50
       +
         | Invalid_endpoint of string

     

       51
       51
       +
         (** Invalid endpoint specification *)

     

       52
       52
       +
       

     

       53
       53
       +
       exception Pool_error of error

     

       54
       54
       +
       (** Exception raised by pool operations.

     

       55
       55
       +
       

     

       56
       56
       +
           Most pool operations can raise this exception. Use {!pp_error} to get

     

       57
       57
       +
           human-readable error messages. *)

     

       58
       58
       +
       

     

       59
       59
       +
       val pp_error : error Fmt.t

     

       60
       60
       +
       (** Pretty-printer for error values. *)

     

       61
       61
       +
       

     

       35
       62
        
       (** {1 Connection Pool} *)

     

       36
       63
        
       

     

       37
       64
        
       type t

lib/endpoint.ml

···

       9
       9
        
       }

     

       10
       10
        
       

     

       11
       11
        
       let make ~host ~port =

     

       12
       12
       +
         (* Validate port range *)

     

       13
       13
       +
         if port < 1 || port > 65535 then

     

       14
       14
       +
           invalid_arg (Printf.sprintf "Invalid port number: %d (must be 1-65535)" port);

     

       15
       15
       +
       

     

       16
       16
       +
         (* Validate hostname is not empty *)

     

       17
       17
       +
         if String.trim host = "" then

     

       18
       18
       +
           invalid_arg "Hostname cannot be empty";

     

       19
       19
       +
       

     

       12
       20
        
         Log.debug (fun m -> m "Creating endpoint: %s:%d" host port);

     

       13
       21
        
         { host; port }

     

       14
       22