Skip to content

async: add close_finished #73

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
vbmithr wants to merge 5 commits into
base: master
Choose a base branch
from
Open

async: add close_finished #73

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
a69dd2d
async: add close_finished
vbmithr Dec 8, 2023
7238199
refactor: minimize the diff a bit
anmonteiro Mar 14, 2024
32bd8a5
async: cosmetics
vbmithr Dec 16, 2024
958480d
async: close connection on read error or exception
vbmithr Dec 17, 2024
34072c2
async: ssl: pass hostname, alpn protocols arguments
vbmithr Jan 19, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
132 changes: 71 additions & 61 deletions async/gluten_async.ml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,66 +37,64 @@ 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
~f:(fun buf ~off ~len k -> Async.upon (Io.read socket buf ~off ~len) k)
read_buffer
(fun n -> Ivar.fill ivar n);
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.
(module Gluten.RUNTIME with type t = t)
-> t
-> read_buffer_size:int
-> 'a fd
-> read_complete:unit Ivar.t
-> write_complete:unit Ivar.t
-> 'a Io.socket
-> unit Deferred.t
=
fun (module Runtime) t ~read_buffer_size socket ->
fun (module Runtime)
t
~read_buffer_size
~read_complete
~write_complete
socket ->
let read_buffer = Buffer.create read_buffer_size in
let read_complete = Ivar.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 ->
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 ())
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 ();
Ivar.fill_if_empty write_complete ()
| _n ->
let _n = Buffer.get read_buffer ~f:(Runtime.read t) in
reader_thread () )
| `Yield -> Runtime.yield_reader t reader_thread
| `Close ->
(* Needed ?*)

@anmonteiro anmonteiro Mar 14, 2024

Copy link
Copy Markdown
Owner

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I suspect this might not be needed but right now it doesn't hurt to have. I'm guessing if a program is running this branch, read_complete will always have been filled. Could be worth it to assert that.

@vbmithr vbmithr Mar 15, 2024

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yeah, this async code is notoriously hard to follow, I don't recall very much now. All I know is this code works in prod. Agree for asserting things that we suppose true! :)

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 ->
writev iovecs >>> fun result ->
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 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 ->
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. *)
Expand All @@ -117,7 +115,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
Expand All @@ -133,31 +139,35 @@ 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 :
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 =
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 ->
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
Expand All @@ -176,7 +186,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

Expand All @@ -186,13 +196,7 @@ 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
Expand Down Expand Up @@ -225,17 +229,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 }
return { connection; socket; closed = Ivar.read read_complete }

let upgrade t protocol =
Client_connection.upgrade_protocol t.connection protocol
Expand All @@ -245,6 +254,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
Expand All @@ -253,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
Expand Down
3 changes: 2 additions & 1 deletion async/gluten_async.mli
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
1 change: 1 addition & 0 deletions async/gluten_async_intf.ml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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

@anmonteiro anmonteiro Mar 14, 2024

Copy link
Copy Markdown
Owner

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Instead of introducing a new API, could we make this part of the val shutdown: _ t -> unit Deferred.t process?

@vbmithr vbmithr Mar 15, 2024

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

To be used as a functor for Persistent_connection_kernel.Make, it should follow this interface:

module type Closable = sig
  (** a connection type *)
  type t

  (** [close t] closes the connection. The returned deferred becomes determined once any
      resources needed to maintain the connection have been released. *)
  val close : t -> unit Deferred.t

  (** [is_closed t] returns true if [close] has ever been called (even if the returned
      deferred has not yet been fulfilled).

      Note that some modules implementing [Closable] may call close internally upon
      noticing that the connection was closed by the other side. The interface of such a
      module ought to say that this is the case. *)
  val is_closed : t -> bool

  (** [close_finished t] becomes determined at the same time as the result of the first
      call to [close]. [close_finished] differs from [close] in that it does not have the
      side effect of initiating a close. *)
  val close_finished : t -> unit Deferred.t
end

@vbmithr vbmithr Mar 15, 2024

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Actually, since there is a parameter it cannot be used directly as an argument to Persistent_connection_kernel.Make. I used this patch in an internal fork to ocaml-h2 that I could maybe upstream at some point.

@vbmithr vbmithr Mar 15, 2024

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

But as the doc says, close and close_finished have not the same semantics.
Also, would not be better to call shutdown -> close since the socket is closed in the process?

@anmonteiro anmonteiro Mar 15, 2024

Copy link
Copy Markdown
Owner

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

would not be better to call shutdown -> close since the socket is closed in the process?

works for me, would you like to propose a separate PR for that?

end
17 changes: 11 additions & 6 deletions async/ssl_io.real.ml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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
Expand Down