diff --git a/lib/client_connection.ml b/lib/client_connection.ml index df1c69c..9755abf 100644 --- a/lib/client_connection.ml +++ b/lib/client_connection.ml @@ -62,7 +62,9 @@ type t = ; reader : Reader.frame ; writer : Writer.t ; config : Config.t + ; window_update_batch_size : int32 ; streams : Scheduler.t + ; mutable pending_connection_window_update : int32 ; mutable current_stream_id : Stream_identifier.t ; mutable max_pushed_stream_id : Stream_identifier.t ; mutable current_server_streams : int @@ -201,6 +203,12 @@ let send_window_update : type a. loop n; Writer.wakeup t.writer) +let queue_window_update t pending len = + let pending = Int32.add pending len in + if Int32.compare pending t.window_update_batch_size >= 0 + then 0l, pending + else pending, 0l + let handle_push_promise_headers t respd headers = (* From RFC7540§8.2.2: * The header fields in PUSH_PROMISE and any subsequent CONTINUATION frames @@ -297,6 +305,7 @@ let handle_response_headers t stream ~end_stream active_request headers = (`Invalid_response_body_length response) ProtocolError | `Fixed _ | `Unknown -> + let pending_stream_window_update = ref 0l in let response_body = if end_stream then Body.Reader.empty @@ -305,8 +314,22 @@ let handle_response_headers t stream ~end_stream active_request headers = (Bigstringaf.create t.config.response_body_buffer_size) ~done_reading:(fun len -> let len = Int32.of_int len in - send_window_update t t.streams len; - send_window_update t stream len) + let pending_connection_window_update, connection_credit = + queue_window_update + t + t.pending_connection_window_update + len + in + t.pending_connection_window_update <- + pending_connection_window_update; + if Int32.compare connection_credit 0l > 0 + then send_window_update t t.streams connection_credit; + let pending_window_update, stream_credit = + queue_window_update t !pending_stream_window_update len + in + pending_stream_window_update := pending_window_update; + if Int32.compare stream_credit 0l > 0 + then send_window_update t stream stream_credit) in let new_response_state = Respd.create_active_response response response_body @@ -1191,6 +1214,10 @@ let create ?(config = Config.default) ?push_handler ~error_handler () = config.enable_server_push && push_handler != default_push_handler } in + let half_window = Int32.shift_right_logical config.initial_window_size 1 in + let window_update_batch_size = + max 0x4000l (min 0x100000l half_window) + in let rec connection_preface_handler recv_frame settings_list = let t = Lazy.force t in (* Now process the client's SETTINGS frame. `process_settings_frame` will @@ -1242,6 +1269,8 @@ let create ?(config = Config.default) ?push_handler ~error_handler () = lazy { settings ; config + ; window_update_batch_size + ; pending_connection_window_update = 0l (* From RFC7540§5.1.1: * Streams initiated by a client MUST use odd-numbered stream * identifiers *) diff --git a/lib/server_connection.ml b/lib/server_connection.ml index 4747d73..6c65493 100644 --- a/lib/server_connection.ml +++ b/lib/server_connection.ml @@ -61,9 +61,11 @@ type t = ; reader : Reader.frame ; writer : Writer.t ; config : Config.t + ; window_update_batch_size : int32 ; request_handler : request_handler ; error_handler : error_handler ; streams : Scheduler.t + ; mutable pending_connection_window_update : int32 (* Number of currently open client streams. Used for * MAX_CONCURRENT_STREAMS bookkeeping *) ; mutable current_client_streams : int @@ -199,6 +201,12 @@ let send_window_update : type a. loop n; wakeup_writer t) +let queue_window_update t pending len = + let pending = Int32.add pending len in + if Int32.compare pending t.window_update_batch_size >= 0 + then 0l, pending + else pending, 0l + let create_push_stream t parent_stream_id = let candidate_push_stream_id = Int32.add t.max_pushed_stream_id 2l in if not t.settings.enable_push @@ -304,6 +312,7 @@ let handle_headers t ~end_stream stream active_stream headers = (Httpun_types.Method.of_string meth) path in + let pending_stream_window_update = ref 0l in let request_body = if end_stream then Body.Reader.empty @@ -319,8 +328,25 @@ let handle_headers t ~end_stream stream active_stream headers = * connection-level flow-control windows. *) match reqd.state with | Active _ -> - send_window_update t t.streams len; - send_window_update t stream len + let pending_connection_window_update, connection_credit = + queue_window_update + t + t.pending_connection_window_update + len + in + t.pending_connection_window_update <- + pending_connection_window_update; + if Int32.compare connection_credit 0l > 0 + then send_window_update t t.streams connection_credit; + let pending_window_update, stream_credit = + queue_window_update + t + !pending_stream_window_update + len + in + pending_stream_window_update := pending_window_update; + if Int32.compare stream_credit 0l > 0 + then send_window_update t stream stream_credit | Idle | Reserved _ | Closed _ -> ()) in let request_info = Reqd.create_active_request request request_body in @@ -1158,6 +1184,10 @@ let write_connection_preface t = let create_generic ~h2c ~config ~error_handler request_handler = let settings = Config.to_settings config in let writer = Writer.create settings.max_frame_size in + let half_window = Int32.shift_right_logical config.initial_window_size 1 in + let window_update_batch_size = + max 0x4000l (min 0x100000l half_window) + in let rec connection_preface_handler recv_frame settings_list = let t = Lazy.force t in (* If this connection is `h2c` (HTTP/2 over TCP), we have already written @@ -1230,9 +1260,11 @@ let create_generic ~h2c ~config ~error_handler request_handler = frame_handler ; writer ; config + ; window_update_batch_size ; request_handler ; error_handler ; streams = Scheduler.make_root () + ; pending_connection_window_update = 0l ; current_client_streams = 0 ; max_client_stream_id = 0l ; max_pushed_stream_id = 0l diff --git a/lib_test/test_h2_client.ml b/lib_test/test_h2_client.ml index 822fff9..9156d8c 100644 --- a/lib_test/test_h2_client.ml +++ b/lib_test/test_h2_client.ml @@ -1194,7 +1194,13 @@ module Client_connection_tests = struct !body_eof_called let test_flow_control () = - let t = create_and_handle_preface () in + let config = + { Config.default with + initial_window_size = 0x8000l + ; response_body_buffer_size = 0x4000 + } + in + let t = create_and_handle_preface ~config () in let request = Request.create ~scheme:"http" `GET "/" in let body_read_called = ref false in let response_handler _response response_body = @@ -1219,8 +1225,10 @@ module Client_connection_tests = struct t hpack_encoder ~flags:Flags.(default_flags |> set_end_header) - (Response.create `OK ~headers:(Headers.of_list [ "content-length", "3" ])); - read_response_body t "foo"; + (Response.create + `OK + ~headers:(Headers.of_list [ "content-length", "16384" ])); + read_response_body t (String.make 0x4000 'x'); let frames, lenv = flush_pending_writes t in Alcotest.(check (list int)) "Only writes are WINDOW_UPDATE frames" @@ -1231,7 +1239,38 @@ module Client_connection_tests = struct (fun Frame.{ frame_header = { frame_type; _ }; _ } -> Frame.FrameType.serialize frame_type) frames); - report_write_result t (`Ok lenv); + report_write_result t (`Ok lenv); + Alcotest.(check bool) "Response handler called" true !body_read_called + + let test_flow_control_delays_small_updates () = + let t = create_and_handle_preface () in + let request = Request.create ~scheme:"http" `GET "/" in + let body_read_called = ref false in + let response_handler _response response_body = + Body.Reader.schedule_read + response_body + ~on_eof:ignore + ~on_read:(fun _bs ~off:_ ~len:_ -> body_read_called := true) + in + let request_body = + Client_connection.request + t + request + ~flush_headers_immediately:true + ~error_handler:default_error_handler + ~response_handler + in + flush_request t; + Body.Writer.close request_body; + flush_request t; + let hpack_encoder = Hpack.Encoder.create 4096 in + read_response + t + hpack_encoder + ~flags:Flags.(default_flags |> set_end_header) + (Response.create `OK ~headers:(Headers.of_list [ "content-length", "3" ])); + read_response_body t "foo"; + writer_yielded t; Alcotest.(check bool) "Response handler called" true !body_read_called let test_flow_control_can_send_empty_data_frame () = @@ -1589,6 +1628,7 @@ module Client_connection_tests = struct , `Quick , test_reading_response_body ) ; "flow control", `Quick, test_flow_control + ; "flow control delays small updates", `Quick, test_flow_control_delays_small_updates ; ( "flow control -- can send empty data frame" , `Quick , test_flow_control_can_send_empty_data_frame ) diff --git a/lib_test/test_h2_server.ml b/lib_test/test_h2_server.ml index ccba8a8..fe2bc0a 100644 --- a/lib_test/test_h2_server.ml +++ b/lib_test/test_h2_server.ml @@ -1123,6 +1123,12 @@ module Server_connection_tests = struct let test_flow_control () = let body_read_called = ref false in + let config = + { Config.default with + initial_window_size = 0x8000l + ; request_body_buffer_size = 0x4000 + } + in let request = Request.create ~scheme:"http" `GET "/" in let request_handler reqd = let request_body = Reqd.request_body reqd in @@ -1131,18 +1137,24 @@ module Server_connection_tests = struct ~on_eof:ignore ~on_read:(fun _bs ~off:_ ~len:_ -> body_read_called := true) in - let t = create_and_handle_preface ~error_handler request_handler in + let t = + create_and_handle_preface ~config ~error_handler request_handler + in read_request ~body:"request body" t request; + let payload_length = 0x4000 in let data_frame = { Frame.frame_header = - { payload_length = 0 + { payload_length ; stream_id = 1l ; flags = Flags.(default_flags |> set_end_stream) ; frame_type = Data } ; frame_payload = - let buffer = Bigstringaf.of_string ~off:0 ~len:3 "foo" in - Frame.Data { Httpun_types.IOVec.buffer; off = 0; len = 3 } + let body = String.make payload_length 'x' in + let buffer = + Bigstringaf.of_string ~off:0 ~len:payload_length body + in + Frame.Data { Httpun_types.IOVec.buffer; off = 0; len = payload_length } } in read_frames t [ data_frame ]; @@ -1163,6 +1175,34 @@ module Server_connection_tests = struct writer_yields t | _ -> assert false + let test_flow_control_delays_small_updates () = + let body_read_called = ref false in + let request = Request.create ~scheme:"http" `GET "/" in + let request_handler reqd = + let request_body = Reqd.request_body reqd in + Body.Reader.schedule_read + request_body + ~on_eof:ignore + ~on_read:(fun _bs ~off:_ ~len:_ -> body_read_called := true) + in + let t = create_and_handle_preface ~error_handler request_handler in + read_request ~body:"request body" t request; + let data_frame = + { Frame.frame_header = + { payload_length = 3 + ; stream_id = 1l + ; flags = Flags.(default_flags |> set_end_stream) + ; frame_type = Data + } + ; frame_payload = + let buffer = Bigstringaf.of_string ~off:0 ~len:3 "foo" in + Frame.Data { Httpun_types.IOVec.buffer; off = 0; len = 3 } + } + in + read_frames t [ data_frame ]; + writer_yields t; + Alcotest.(check bool) "Response handler called" true !body_read_called + let test_flow_control_can_send_empty_data_frame () = let request = Request.create ~scheme:"http" `GET "/" in let request_handler reqd = @@ -1406,6 +1446,7 @@ module Server_connection_tests = struct , test_reading_request_body ) ; "accepting multiple RST_STREAM frames", `Quick, test_rst_stream_frames ; "flow control", `Quick, test_flow_control + ; "flow control delays small updates", `Quick, test_flow_control_delays_small_updates ; ( "flow control -- can send empty data frame" , `Quick , test_flow_control_can_send_empty_data_frame )