perf: batch inbound WINDOW_UPDATE frames

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 *)

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

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 )

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 )