perf: avoid copying DATA payloads in parser

lib/client_connection.ml

@@ -569,7 +569,11 @@ let process_headers_frame t { Frame.frame_header; _ } headers_block =
        *   connection error (Section 5.4.1) of type STREAM_CLOSED [...]. *)
       report_connection_error t Error_code.StreamClosed)
 
-let process_data_frame t { Frame.frame_header; _ } bstr =
+let process_data_frame
+      t
+      { Frame.frame_header; _ }
+      { Httpun_types.IOVec.buffer; off; len }
+  =
   let open Scheduler in
   let { Frame.flags; stream_id; payload_length; _ } = frame_header in
   let payload_len32 = Int32.of_int payload_length in
@@ -584,7 +588,7 @@ let process_data_frame t { Frame.frame_header; _ } bstr =
         response_info
       in
       response_info.response_body_bytes <-
-        Int64.(add response_body_bytes (of_int (Bigstringaf.length bstr)));
+        Int64.(add response_body_bytes (of_int len));
       (* First, calculate whether we're allowed to receive this frame based
        * on the _current_ inflow. *)
       let allowed_to_receive =
@@ -640,7 +644,7 @@ let process_data_frame t { Frame.frame_header; _ } bstr =
           let faraday = Body.Reader.unsafe_faraday response_body in
           if not (Faraday.is_closed faraday)
           then (
-            Faraday.schedule_bigstring faraday bstr;
+            Faraday.write_bigstring faraday ~off ~len buffer;
             if end_stream then Body.Reader.close response_body);
           Respd.flush_response_body descriptor;
           if end_stream && not (Respd.requires_output descriptor)
@@ -1217,7 +1221,7 @@ let create ?(config = Config.default) ?push_handler ~error_handler () =
         (match frame_payload with
         | Headers (_priority, headers_block) ->
           process_headers_frame t frame headers_block
-        | Data bs -> process_data_frame t frame bs
+        | Data payload -> process_data_frame t frame payload
         | Priority priority -> process_priority_frame t frame priority
         | RSTStream error_code -> process_rst_stream_frame t frame error_code
         | Settings settings -> process_settings_frame t frame settings

lib/frame.ml

@@ -127,6 +127,8 @@ type frame_header =
   ; frame_type : FrameType.t
   }
 
+type payload_view = Bigstringaf.t Httpun_types.IOVec.t
+
 (* From RFC7540§4.1:
  *   The structure and content of the frame payload is dependent entirely on
  *   the frame type. *)
@@ -139,7 +141,7 @@ type frame_payload =
    *   Data: Application data. The amount of data is the remainder of the
    *         frame payload after subtracting the length of the other fields
    *         that are present. *)
-  | Data of Bigstringaf.t
+  | Data of payload_view
   (* From RFC7540§6.2:
    *   The HEADERS frame payload has the following fields:
    *

lib/parse.ml

@@ -33,6 +33,7 @@
  *---------------------------------------------------------------------------*)
 
 open Angstrom
+module Unsafe = Angstrom.Unsafe
 
 (* We use the tail-recursive variant of `skip_many` from
  * https://github.com/inhabitedtype/angstrom/pull/219 to avoid memory leaks in
@@ -157,7 +158,11 @@ let parse_data_frame ({ Frame.stream_id; payload_length; _ } as frame_header) =
       "Data frames must be associated with a stream"
   else
     let parse_data length =
-      lift (fun bs -> Ok (Frame.Data bs)) (take_bigstring length)
+      lift
+        (fun payload -> Ok (Frame.Data payload))
+        (Unsafe.peek length (fun buffer ~off ~len ->
+           { Httpun_types.IOVec.buffer; off; len })
+         <* advance length)
     in
     parse_padded_payload frame_header parse_data
 

lib/server_connection.ml

@@ -601,7 +601,11 @@ let process_headers_frame t { Frame.frame_header; _ } ~priority headers_block =
          *   connection error (Section 5.4.1) of type STREAM_CLOSED [...]. *)
         report_connection_error t Error_code.StreamClosed)
 
-let process_data_frame t { Frame.frame_header; _ } bstr =
+let process_data_frame
+      t
+      { Frame.frame_header; _ }
+      { Httpun_types.IOVec.buffer; off; len }
+  =
   let open Scheduler in
   let { Frame.flags; stream_id; payload_length; _ } = frame_header in
   if not (Stream_identifier.is_request stream_id)
@@ -626,10 +630,7 @@ let process_data_frame t { Frame.frame_header; _ } bstr =
       | Active (Open (ActiveMessage request_info), active_stream) ->
         let request_body = Reqd.request_body descriptor in
         request_info.request_body_bytes <-
-          Int64.(
-            add
-              request_info.request_body_bytes
-              (of_int (Bigstringaf.length bstr)));
+          Int64.(add request_info.request_body_bytes (of_int len));
         let request = request_info.request in
         if not Scheduler.(allowed_to_receive t.streams stream payload_len32)
         then (
@@ -683,7 +684,7 @@ let process_data_frame t { Frame.frame_header; _ } bstr =
             let faraday = Body.Reader.unsafe_faraday request_body in
             if not (Faraday.is_closed faraday)
             then (
-              Faraday.schedule_bigstring faraday bstr;
+              Faraday.write_bigstring faraday ~off ~len buffer;
               if end_stream then Body.Reader.close request_body);
             Reqd.flush_request_body descriptor)
       | Idle ->
@@ -1192,7 +1193,7 @@ let create_generic ~h2c ~config ~error_handler request_handler =
         (match frame_payload with
         | Headers (priority, headers_block) ->
           process_headers_frame t frame ~priority headers_block
-        | Data bs -> process_data_frame t frame bs
+        | Data payload -> process_data_frame t frame payload
         | Priority priority -> process_priority_frame t frame priority
         | RSTStream error_code -> process_rst_stream_frame t frame error_code
         | Settings settings -> process_settings_frame t frame settings

lib_test/test_common.ml

@@ -17,6 +17,9 @@ let bs_to_string bs =
   let len = Bigstringaf.length bs in
   Bigstringaf.substring ~off ~len bs
 
+let payload_view_to_string { Httpun_types.IOVec.buffer; off; len } =
+  Bigstringaf.substring ~off ~len buffer
+
 let bs_of_string s = Bigstringaf.of_string ~off:0 ~len:(String.length s) s
 let string_of_hex s = Hex.to_string (`Hex s)
 
@@ -32,7 +35,8 @@ let write_frame ?padding t { Frame.frame_header; frame_payload } =
   let { Frame.flags; stream_id; _ } = frame_header in
   let info = Writer.make_frame_info ~flags ?padding stream_id in
   match frame_payload with
-  | Data body -> Writer.schedule_data t info body
+  | Data { Httpun_types.IOVec.buffer; off; len } ->
+    Writer.schedule_data t info (Bigstringaf.copy ~off ~len buffer)
   | Headers (priority, headers_block) ->
     (* Block already HPACK-encoded. *)
     write_headers_frame t.encoder info ~priority (make_iovecs headers_block)

lib_test/test_frames.ml

@@ -99,7 +99,7 @@ let frame_testable =
     let frame_payload_to_json frame_type payload =
       let others =
         match payload with
-        | Frame.Data data -> [ "data", `String (bs_to_string data) ]
+        | Frame.Data data -> [ "data", `String (payload_view_to_string data) ]
         | Headers (priority, fragment) ->
           ("header_block_fragment", `String (bs_to_string fragment))
           :: priority_to_yojson priority
@@ -195,7 +195,8 @@ let frame_type_of_string = function
 let frame_payload_of_json frame_type json =
   match frame_type with
   | Frame.FrameType.Data ->
-    Frame.Data Json.(json |> member "data" |> to_string |> bs_of_string)
+    let buffer = Json.(json |> member "data" |> to_string |> bs_of_string) in
+    Frame.Data { Httpun_types.IOVec.buffer; off = 0; len = Bigstringaf.length buffer }
   | Headers ->
     let priority = priority_of_json json in
     let fragment =

lib_test/test_h2_client.ml

@@ -1337,7 +1337,7 @@ module Client_connection_tests = struct
       List.filter_map
         (fun Frame.{ frame_payload; _ } ->
            match frame_payload with
-           | Frame.Data bs -> Some (bs_to_string bs)
+           | Frame.Data payload -> Some (payload_view_to_string payload)
            | _ -> None)
         frames
     in

lib_test/test_h2_server.ml

@@ -1055,7 +1055,9 @@ module Server_connection_tests = struct
           ; flags = Flags.default_flags
           ; frame_type = Data
           }
-      ; frame_payload = Frame.Data (Bigstringaf.of_string ~off:0 ~len:3 "foo")
+      ; 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 ];
@@ -1095,7 +1097,9 @@ module Server_connection_tests = struct
           ; flags = Flags.default_flags
           ; frame_type = Data
           }
-      ; frame_payload = Frame.Data (Bigstringaf.of_string ~off:0 ~len:3 "foo")
+      ; frame_payload =
+          let buffer = Bigstringaf.of_string ~off:0 ~len:3 "foo" in
+          Frame.Data { Httpun_types.IOVec.buffer; off = 0; len = 3 }
       }
     in
     let rst_stream =
@@ -1136,7 +1140,9 @@ module Server_connection_tests = struct
           ; flags = Flags.(default_flags |> set_end_stream)
           ; frame_type = Data
           }
-      ; frame_payload = Frame.Data (Bigstringaf.of_string ~off:0 ~len:3 "foo")
+      ; 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 ];
@@ -1181,7 +1187,9 @@ module Server_connection_tests = struct
           ; flags = Flags.(default_flags |> set_end_stream)
           ; frame_type = Data
           }
-      ; frame_payload = Frame.Data (Bigstringaf.of_string ~off:0 ~len:3 "foo")
+      ; 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 ];