From cb71ce5ad6056c8ac81908f22522c34a44957f80 Mon Sep 17 00:00:00 2001 From: Antonio Nuno Monteiro Date: Wed, 11 Mar 2026 19:22:38 -0700 Subject: [PATCH] perf: avoid copying DATA payloads in parser --- lib/client_connection.ml | 12 ++++++++---- lib/frame.ml | 4 +++- lib/parse.ml | 7 ++++++- lib/server_connection.ml | 15 ++++++++------- lib_test/test_common.ml | 6 +++++- lib_test/test_frames.ml | 5 +++-- lib_test/test_h2_client.ml | 2 +- lib_test/test_h2_server.ml | 16 ++++++++++++---- 8 files changed, 46 insertions(+), 21 deletions(-) diff --git a/lib/client_connection.ml b/lib/client_connection.ml index 7d257189..df1c69cb 100644 --- a/lib/client_connection.ml +++ b/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 diff --git a/lib/frame.ml b/lib/frame.ml index 7f74c9eb..77cd80ce 100644 --- a/lib/frame.ml +++ b/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: * diff --git a/lib/parse.ml b/lib/parse.ml index 412212cc..ce17aca5 100644 --- a/lib/parse.ml +++ b/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 diff --git a/lib/server_connection.ml b/lib/server_connection.ml index a16bd61f..4747d737 100644 --- a/lib/server_connection.ml +++ b/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 diff --git a/lib_test/test_common.ml b/lib_test/test_common.ml index c8261420..174e5741 100644 --- a/lib_test/test_common.ml +++ b/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) diff --git a/lib_test/test_frames.ml b/lib_test/test_frames.ml index 56721000..decdd071 100644 --- a/lib_test/test_frames.ml +++ b/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 = diff --git a/lib_test/test_h2_client.ml b/lib_test/test_h2_client.ml index 18ff0734..822fff92 100644 --- a/lib_test/test_h2_client.ml +++ b/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 diff --git a/lib_test/test_h2_server.ml b/lib_test/test_h2_server.ml index 9efedb4d..ccba8a8c 100644 --- a/lib_test/test_h2_server.ml +++ b/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 ];