From a69dd2ddde32639c18f5c3cae812db2b69034119 Mon Sep 17 00:00:00 2001 From: Vincent Bernardoff Date: Fri, 8 Dec 2023 19:19:12 +0100 Subject: [PATCH 1/5] async: add close_finished --- async/gluten_async.ml | 99 +++++++++++++++++++++----------------- async/gluten_async_intf.ml | 1 + 2 files changed, 55 insertions(+), 45 deletions(-) diff --git a/async/gluten_async.ml b/async/gluten_async.ml index 4e388cb..bf33ec8 100644 --- a/async/gluten_async.ml +++ b/async/gluten_async.ml @@ -44,46 +44,35 @@ module Make_IO_Loop (Io : Gluten_async_intf.IO) = struct Buffer.put ~f:(fun buf ~off ~len k -> Async.upon (Io.read socket buf ~off ~len) k) read_buffer - (fun n -> Ivar.fill ivar n); + (Ivar.fill ivar); Ivar.read ivar - let start : - type t. - (module Gluten.RUNTIME with type t = t) - -> t - -> read_buffer_size:int - -> 'a fd - -> unit Deferred.t + let start (type t) m t ~read_buffer_size ~read_complete ~write_complete socket = - fun (module Runtime) t ~read_buffer_size socket -> + let module Runtime = (val m : Gluten.RUNTIME with type t = t) in let read_buffer = Buffer.create read_buffer_size in - let read_complete = Ivar.create () in + let monitor = Monitor.create () in let rec reader_thread () = - let rec reader_thread_step () = - match Runtime.next_read_operation t with - | `Read -> - Monitor.try_with (fun () -> read socket read_buffer) >>= ( function - | Ok _n -> + match Runtime.next_read_operation t with + | `Read -> + read socket read_buffer >>> ( function + | 0 -> + (* End of file, error, nothing can be read, exiting. *) + let (_ : int) = Buffer.get read_buffer ~f:(Runtime.read_eof t) in + Ivar.fill_if_empty read_complete () + | _n -> + let _n = Buffer.get read_buffer ~f:(fun bigstring ~off ~len -> Runtime.read t bigstring ~off ~len) - |> ignore; - reader_thread_step () - | Error End_of_file -> - let (_ : int) = Buffer.get read_buffer ~f:(Runtime.read_eof t) in - reader_thread_step () - | Error exn -> raise exn ) - | `Yield -> - Runtime.yield_reader t reader_thread; - Deferred.return () - | `Close -> - Ivar.fill read_complete (); - Io.shutdown_receive socket; - Deferred.return () - in - Deferred.don't_wait_for (reader_thread_step ()) + in + reader_thread () ) + | `Yield -> Runtime.yield_reader t reader_thread + | `Close -> + (* Needed ?*) + Ivar.fill_if_empty read_complete (); + Io.shutdown_receive socket in let writev = Io.writev socket in - let write_complete = Ivar.create () in let rec writer_thread () = match Runtime.next_write_operation t with | `Write iovecs -> @@ -93,14 +82,13 @@ module Make_IO_Loop (Io : Gluten_async_intf.IO) = struct | `Yield -> Runtime.yield_writer t writer_thread | `Close _ -> Ivar.fill write_complete () in - let conn_monitor = Monitor.create () in - Scheduler.within ~monitor:conn_monitor reader_thread; - Scheduler.within ~monitor:conn_monitor writer_thread; - Monitor.detach_and_iter_errors conn_monitor ~f:(fun exn -> + Scheduler.within ~monitor reader_thread; + Scheduler.within ~monitor writer_thread; + Monitor.detach_and_iter_errors monitor ~f:(fun exn -> Runtime.report_exn t exn); (* The Tcp module will close the file descriptor once this becomes determined. *) - Deferred.all_unit [ Ivar.read read_complete; Ivar.read write_complete ] + Deferred.all_unit Ivar.[ read read_complete; read write_complete ] >>= fun () -> Io.close socket end @@ -117,7 +105,15 @@ module Make_server (Io : Gluten_async_intf.IO) = struct socket = let connection = Gluten.Server.create ~protocol connection in - IO_loop.start (module Gluten.Server) connection ~read_buffer_size socket + let read_complete = Ivar.create () in + let write_complete = Ivar.create () in + IO_loop.start + (module Gluten.Server) + connection + ~read_buffer_size + ~read_complete + ~write_complete + socket let create_upgradable_connection_handler ~read_buffer_size @@ -133,7 +129,16 @@ module Make_server (Io : Gluten_async_intf.IO) = struct ~create:create_protocol (request_handler client_addr) in - IO_loop.start (module Gluten.Server) connection ~read_buffer_size socket + (* TODO: expose this somewhere? *) + let read_complete = Ivar.create () in + let write_complete = Ivar.create () in + IO_loop.start + (module Gluten.Server) + connection + ~read_buffer_size + socket + ~read_complete + ~write_complete end module Unix_io : @@ -144,16 +149,14 @@ struct let read socket bigstring ~off ~len = let fd = Socket.fd socket in - let badfd fd = failwithf "read got back fd: %s" (Fd.to_string fd) () in - let rec finish fd buffer result = - let open Unix.Error in - match result with - | `Already_closed | `Ok 0 -> raise End_of_file + let badfd fd = failwithf "read got bad fd: %s" (Fd.to_string fd) () in + let rec finish fd buffer = function + | `Already_closed -> return 0 | `Ok n -> return n | `Error (Unix.Unix_error ((EWOULDBLOCK | EAGAIN), _, _)) -> Fd.ready_to fd `Read >>= ( function | `Bad_fd -> badfd fd - | `Closed -> raise End_of_file + | `Closed -> return 0 | `Ready -> go fd buffer ) | `Error (Unix.Unix_error (EBADF, _, _)) -> badfd fd | `Error exn -> @@ -225,17 +228,22 @@ module Make_client (Io : Gluten_async_intf.IO) = struct type 'a t = { connection : Client_connection.t ; socket : 'a socket + ; closed : unit Deferred.t } let create ~read_buffer_size ~protocol t socket = let connection = Client_connection.create ~protocol t in + let read_complete = Ivar.create () in + let write_complete = Ivar.create () in don't_wait_for (IO_loop.start (module Client_connection) connection ~read_buffer_size + ~read_complete + ~write_complete socket); - Deferred.return { connection; socket } + Deferred.return { connection; socket; closed = Ivar.read read_complete } let upgrade t protocol = Client_connection.upgrade_protocol t.connection protocol @@ -245,6 +253,7 @@ module Make_client (Io : Gluten_async_intf.IO) = struct Io.close t.socket let is_closed t = Client_connection.is_closed t.connection + let close_finished t = t.closed end module Client = struct diff --git a/async/gluten_async_intf.ml b/async/gluten_async_intf.ml index 9d863bd..157a6ee 100644 --- a/async/gluten_async_intf.ml +++ b/async/gluten_async_intf.ml @@ -82,4 +82,5 @@ module type Client = sig val upgrade : _ t -> Gluten.impl -> unit val shutdown : _ t -> unit Deferred.t val is_closed : _ t -> bool + val close_finished : _ t -> unit Deferred.t end From 7238199dccecbc49be74f51d69d2d5f8977f9bf4 Mon Sep 17 00:00:00 2001 From: Antonio Nuno Monteiro Date: Thu, 14 Mar 2024 14:39:46 -0700 Subject: [PATCH 2/5] refactor: minimize the diff a bit --- async/gluten_async.ml | 49 ++++++++++++++++++++++++------------------- 1 file changed, 28 insertions(+), 21 deletions(-) diff --git a/async/gluten_async.ml b/async/gluten_async.ml index bf33ec8..9a7b013 100644 --- a/async/gluten_async.ml +++ b/async/gluten_async.ml @@ -37,8 +37,6 @@ open Async module Buffer = Gluten.Buffer module Make_IO_Loop (Io : Gluten_async_intf.IO) = struct - type 'a fd = 'a Io.socket - let read socket read_buffer = let ivar = Ivar.create () in Buffer.put @@ -47,11 +45,22 @@ module Make_IO_Loop (Io : Gluten_async_intf.IO) = struct (Ivar.fill ivar); Ivar.read ivar - let start (type t) m t ~read_buffer_size ~read_complete ~write_complete socket + let start : type t. + (module Gluten.RUNTIME with type t = t) + -> t + -> read_buffer_size:int + -> read_complete:unit Ivar.t + -> write_complete:unit Ivar.t + -> 'a Io.socket + -> unit Deferred.t = - let module Runtime = (val m : Gluten.RUNTIME with type t = t) in + fun (module Runtime) + t + ~read_buffer_size + ~read_complete + ~write_complete + socket -> let read_buffer = Buffer.create read_buffer_size in - let monitor = Monitor.create () in let rec reader_thread () = match Runtime.next_read_operation t with | `Read -> @@ -61,10 +70,7 @@ module Make_IO_Loop (Io : Gluten_async_intf.IO) = struct let (_ : int) = Buffer.get read_buffer ~f:(Runtime.read_eof t) in Ivar.fill_if_empty read_complete () | _n -> - let _n = - Buffer.get read_buffer ~f:(fun bigstring ~off ~len -> - Runtime.read t bigstring ~off ~len) - in + let _n = Buffer.get read_buffer ~f:(Runtime.read t) in reader_thread () ) | `Yield -> Runtime.yield_reader t reader_thread | `Close -> @@ -82,13 +88,14 @@ module Make_IO_Loop (Io : Gluten_async_intf.IO) = struct | `Yield -> Runtime.yield_writer t writer_thread | `Close _ -> Ivar.fill write_complete () in + let monitor = Monitor.create () in Scheduler.within ~monitor reader_thread; Scheduler.within ~monitor writer_thread; Monitor.detach_and_iter_errors monitor ~f:(fun exn -> Runtime.report_exn t exn); (* The Tcp module will close the file descriptor once this becomes determined. *) - Deferred.all_unit Ivar.[ read read_complete; read write_complete ] + Deferred.all_unit [ Ivar.read read_complete; Ivar.read write_complete ] >>= fun () -> Io.close socket end @@ -98,11 +105,11 @@ module Make_server (Io : Gluten_async_intf.IO) = struct module IO_loop = Make_IO_Loop (Io) let create_connection_handler - ~read_buffer_size - ~protocol - connection - _client_addr - socket + ~read_buffer_size + ~protocol + connection + _client_addr + socket = let connection = Gluten.Server.create ~protocol connection in let read_complete = Ivar.create () in @@ -116,12 +123,12 @@ module Make_server (Io : Gluten_async_intf.IO) = struct socket let create_upgradable_connection_handler - ~read_buffer_size - ~protocol - ~create_protocol - ~request_handler - client_addr - socket + ~read_buffer_size + ~protocol + ~create_protocol + ~request_handler + client_addr + socket = let connection = Gluten.Server.create_upgradable From 32bd8a5f2ed485bb6635a23f430ce6dc1f5a9c0a Mon Sep 17 00:00:00 2001 From: Vincent Bernardoff Date: Mon, 16 Dec 2024 11:30:28 +0100 Subject: [PATCH 3/5] async: cosmetics --- async/gluten_async.ml | 17 ++++++----------- 1 file changed, 6 insertions(+), 11 deletions(-) diff --git a/async/gluten_async.ml b/async/gluten_async.ml index 9a7b013..dbe0e11 100644 --- a/async/gluten_async.ml +++ b/async/gluten_async.ml @@ -39,10 +39,8 @@ module Buffer = Gluten.Buffer module Make_IO_Loop (Io : Gluten_async_intf.IO) = struct let read socket read_buffer = let ivar = Ivar.create () in - Buffer.put - ~f:(fun buf ~off ~len k -> Async.upon (Io.read socket buf ~off ~len) k) - read_buffer - (Ivar.fill ivar); + let readf buf ~off ~len k = Io.read socket buf ~off ~len >>> k in + Buffer.put ~f:readf read_buffer (Ivar.fill ivar); Ivar.read ivar let start : type t. @@ -148,10 +146,7 @@ module Make_server (Io : Gluten_async_intf.IO) = struct ~write_complete end -module Unix_io : - Gluten_async_intf.IO - with type 'a socket = ([ `Active ], ([< Socket.Address.t ] as 'a)) Socket.t = -struct +module Unix_io = struct type 'a socket = ([ `Active ], ([< Socket.Address.t ] as 'a)) Socket.t let read socket bigstring ~off ~len = @@ -167,7 +162,7 @@ struct | `Ready -> go fd buffer ) | `Error (Unix.Unix_error (EBADF, _, _)) -> badfd fd | `Error exn -> - Deferred.don't_wait_for (Fd.close fd); + don't_wait_for (Fd.close fd); raise exn and go fd buffer = if Fd.supports_nonblock fd @@ -186,7 +181,7 @@ struct else Fd.syscall_in_thread fd ~name:"read" (fun file_descr -> Bigstring_unix.read file_descr bigstring ~pos:off ~len) - >>= fun result -> finish fd buffer result + >>= finish fd buffer in go fd bigstring @@ -250,7 +245,7 @@ module Make_client (Io : Gluten_async_intf.IO) = struct ~read_complete ~write_complete socket); - Deferred.return { connection; socket; closed = Ivar.read read_complete } + return { connection; socket; closed = Ivar.read read_complete } let upgrade t protocol = Client_connection.upgrade_protocol t.connection protocol From 958480dc3e3a30310a16f1fe6272197e43ebfc6d Mon Sep 17 00:00:00 2001 From: Vincent Bernardoff Date: Tue, 17 Dec 2024 11:44:14 +0100 Subject: [PATCH 4/5] async: close connection on read error or exception --- async/gluten_async.ml | 55 +++++++++++++++++++++---------------------- 1 file changed, 27 insertions(+), 28 deletions(-) diff --git a/async/gluten_async.ml b/async/gluten_async.ml index dbe0e11..4c11a7f 100644 --- a/async/gluten_async.ml +++ b/async/gluten_async.ml @@ -43,14 +43,15 @@ module Make_IO_Loop (Io : Gluten_async_intf.IO) = struct Buffer.put ~f:readf read_buffer (Ivar.fill ivar); Ivar.read ivar - let start : type t. - (module Gluten.RUNTIME with type t = t) - -> t - -> read_buffer_size:int - -> read_complete:unit Ivar.t - -> write_complete:unit Ivar.t - -> 'a Io.socket - -> unit Deferred.t + let start : + type t. + (module Gluten.RUNTIME with type t = t) + -> t + -> read_buffer_size:int + -> read_complete:unit Ivar.t + -> write_complete:unit Ivar.t + -> 'a Io.socket + -> unit Deferred.t = fun (module Runtime) t @@ -66,7 +67,8 @@ module Make_IO_Loop (Io : Gluten_async_intf.IO) = struct | 0 -> (* End of file, error, nothing can be read, exiting. *) let (_ : int) = Buffer.get read_buffer ~f:(Runtime.read_eof t) in - Ivar.fill_if_empty read_complete () + Ivar.fill_if_empty read_complete (); + Ivar.fill_if_empty write_complete () | _n -> let _n = Buffer.get read_buffer ~f:(Runtime.read t) in reader_thread () ) @@ -84,12 +86,15 @@ module Make_IO_Loop (Io : Gluten_async_intf.IO) = struct Runtime.report_write_result t result; writer_thread () | `Yield -> Runtime.yield_writer t writer_thread - | `Close _ -> Ivar.fill write_complete () + | `Close _ -> Ivar.fill_if_empty write_complete () in let monitor = Monitor.create () in Scheduler.within ~monitor reader_thread; Scheduler.within ~monitor writer_thread; Monitor.detach_and_iter_errors monitor ~f:(fun exn -> + (* Kill the connection when either reader or writer encounter an error. *) + Ivar.fill_if_empty read_complete (); + Ivar.fill_if_empty write_complete (); Runtime.report_exn t exn); (* The Tcp module will close the file descriptor once this becomes determined. *) @@ -103,11 +108,11 @@ module Make_server (Io : Gluten_async_intf.IO) = struct module IO_loop = Make_IO_Loop (Io) let create_connection_handler - ~read_buffer_size - ~protocol - connection - _client_addr - socket + ~read_buffer_size + ~protocol + connection + _client_addr + socket = let connection = Gluten.Server.create ~protocol connection in let read_complete = Ivar.create () in @@ -121,12 +126,12 @@ module Make_server (Io : Gluten_async_intf.IO) = struct socket let create_upgradable_connection_handler - ~read_buffer_size - ~protocol - ~create_protocol - ~request_handler - client_addr - socket + ~read_buffer_size + ~protocol + ~create_protocol + ~request_handler + client_addr + socket = let connection = Gluten.Server.create_upgradable @@ -191,13 +196,7 @@ module Unix_io = struct let fd = Socket.fd socket in if not (Fd.is_closed fd) then Socket.shutdown socket `Receive - let close socket = - let fd = Socket.fd socket in - if not (Fd.is_closed fd) - then ( - Socket.shutdown socket `Both; - Fd.close fd) - else Deferred.unit + let close socket = Fd.close (Socket.fd socket) end module Server = struct From 34072c247d41e45091e3e91aa5353f9ac9a1b0d4 Mon Sep 17 00:00:00 2001 From: Vincent Bernardoff Date: Mon, 19 Jan 2026 23:09:51 +0100 Subject: [PATCH 5/5] async: ssl: pass hostname, alpn protocols arguments --- async/gluten_async.ml | 4 ++-- async/gluten_async.mli | 3 ++- async/ssl_io.real.ml | 17 +++++++++++------ 3 files changed, 15 insertions(+), 9 deletions(-) diff --git a/async/gluten_async.ml b/async/gluten_async.ml index 4c11a7f..93102ef 100644 --- a/async/gluten_async.ml +++ b/async/gluten_async.ml @@ -263,8 +263,8 @@ module Client = struct module SSL = struct include Make_client (Ssl_io.Io) - let create_default ?alpn_protocols socket = - Ssl_io.make_default_client ?alpn_protocols socket + let create_default ?hostname ?alpn_protocols socket = + Ssl_io.make_default_client ?hostname ?alpn_protocols socket end module TLS = struct diff --git a/async/gluten_async.mli b/async/gluten_async.mli index c66eb89..5d81b6f 100644 --- a/async/gluten_async.mli +++ b/async/gluten_async.mli @@ -68,7 +68,8 @@ module Client : sig include Gluten_async_intf.Client with type 'a socket = 'a Ssl_io.descriptor val create_default : - ?alpn_protocols:string list + ?hostname:string + -> ?alpn_protocols:string list -> ([ `Active ], [< Socket.Address.t ]) Socket.t -> [< Socket.Address.t ] socket Deferred.t end diff --git a/async/ssl_io.real.ml b/async/ssl_io.real.ml index c17e53e..6e2240e 100644 --- a/async/ssl_io.real.ml +++ b/async/ssl_io.real.ml @@ -92,12 +92,19 @@ let reader_writer_of_sock ( Reader.create ?buf_len:reader_buffer_size fd , Writer.create ?buffer_age_limit ?buf_len:writer_buffer_size fd ) -let connect r w = +let connect ?hostname ?alpn_protocols r w = let net_to_ssl = Reader.pipe r in let ssl_to_net = Writer.pipe w in let app_to_ssl, app_wr = Pipe.create () in let app_rd, ssl_to_app = Pipe.create () in - Ssl.client ~app_to_ssl ~ssl_to_app ~net_to_ssl ~ssl_to_net () + Ssl.client + ?hostname + ?alpn_protocols + ~app_to_ssl + ~ssl_to_app + ~net_to_ssl + ~ssl_to_net + () |> Deferred.Or_error.ok_exn >>= fun _connection -> Reader.of_pipe (Info.of_string "httpaf_async_ssl_reader") app_rd @@ -113,11 +120,9 @@ let connect r w = let writer = app_writer in reader, writer, Ivar.read closed_ivar -(* XXX(anmonteiro): Unfortunately Async_ssl doesn't seem to support configuring - * the ALPN protocols *) -let make_default_client ?alpn_protocols:_ socket = +let make_default_client ?hostname ?alpn_protocols socket = let reader, writer = reader_writer_of_sock socket in - connect reader writer + connect ?hostname ?alpn_protocols reader writer let listen ~crt_file ~key_file r w = let net_to_ssl = Reader.pipe r in