Write and read HTTP/2 frames with fewer copies and writes

CHANGELOG.md

@@ -76,6 +76,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   instead of into a new vector, so receiving DATA no longer reallocates the
   buffer for every frame: a 1 MiB upload allocated 3.2 MiB in the server
   stream buffer and now allocates about 160 KiB. ([#1486])
+- HTTP/2 frames are serialized into one buffer without copying their
+  payload, a WINDOW_UPDATE pair goes out in one write, and frame headers and
+  WINDOW_UPDATE frames already buffered are parsed in place. A 1 MiB upload
+  made about 2,800 allocations in client and server together and now makes
+  about 1,160. The bytes sent are unchanged. ([#1487])
 - `HTTP.Headers` keeps each header name as the caller spelled it and matches
   names case-insensitively, as HTTP.jl 1.x did. HTTP.jl 2.0 through 2.7
   rewrote every name into `Content-Type` form on insertion, so a request header
@@ -1451,3 +1456,4 @@ See changes for 0.9.15: this release is equivalent to 0.9.15 with [#752] reverte
 [#1480]: https://github.com/JuliaWeb/HTTP.jl/issues/1480
 [#1483]: https://github.com/JuliaWeb/HTTP.jl/issues/1483
 [#1486]: https://github.com/JuliaWeb/HTTP.jl/issues/1486
+[#1487]: https://github.com/JuliaWeb/HTTP.jl/issues/1487

src/http2.jl

@@ -189,32 +189,40 @@ end
     return nothing
 end
 
-@inline function _encode_header_bytes!(bytes::Vector{UInt8}, header::FrameHeader)
+@inline function _encode_header_bytes!(bytes::Vector{UInt8}, header::FrameHeader, pos::Int=1)
     header.length < 0 && throw(ArgumentError("HTTP/2 frame length must be >= 0"))
     header.length <= 0x00ff_ffff || throw(ArgumentError("HTTP/2 frame length must fit 24 bits"))
-    length(bytes) >= 9 || throw(ArgumentError("HTTP/2 frame header needs 9 bytes"))
-    bytes[1] = UInt8((header.length >> 16) & 0xff)
-    bytes[2] = UInt8((header.length >> 8) & 0xff)
-    bytes[3] = UInt8(header.length & 0xff)
-    bytes[4] = header.type
-    bytes[5] = header.flags
-    bytes[6] = UInt8((header.stream_id >> 24) & 0x7f)
-    bytes[7] = UInt8((header.stream_id >> 16) & 0xff)
-    bytes[8] = UInt8((header.stream_id >> 8) & 0xff)
-    bytes[9] = UInt8(header.stream_id & 0xff)
+    length(bytes) >= pos + 8 || throw(ArgumentError("HTTP/2 frame header needs 9 bytes"))
+    bytes[pos] = UInt8((header.length >> 16) & 0xff)
+    bytes[pos+1] = UInt8((header.length >> 8) & 0xff)
+    bytes[pos+2] = UInt8(header.length & 0xff)
+    bytes[pos+3] = header.type
+    bytes[pos+4] = header.flags
+    bytes[pos+5] = UInt8((header.stream_id >> 24) & 0x7f)
+    bytes[pos+6] = UInt8((header.stream_id >> 16) & 0xff)
+    bytes[pos+7] = UInt8((header.stream_id >> 8) & 0xff)
+    bytes[pos+8] = UInt8(header.stream_id & 0xff)
     return bytes
 end
 
 @inline _encode_header_bytes(header::FrameHeader)::Vector{UInt8} =
     _encode_header_bytes!(Vector{UInt8}(undef, 9), header)
 
+@inline function _parse_frame_header(bytes::Vector{UInt8}, i::Int)::FrameHeader
+    length = (Int(bytes[i]) << 16) | (Int(bytes[i+1]) << 8) | Int(bytes[i+2])
+    return FrameHeader(length, bytes[i+3], bytes[i+4], _read_u32_be(bytes, i + 5) & 0x7fff_ffff)
+end
+
 function _read_frame_header!(io::IO)::FrameHeader
-    header_bytes = _read_exact_bytes!(io, 9)
-    length = (Int(header_bytes[1]) << 16) | (Int(header_bytes[2]) << 8) | Int(header_bytes[3])
-    ftype = header_bytes[4]
-    flags = header_bytes[5]
-    stream_id = _read_u32_be(header_bytes, 6) & 0x7fff_ffff
-    return FrameHeader(length, ftype, flags, stream_id)
+    return _parse_frame_header(_read_exact_bytes!(io, 9), 1)
+end
+
+# Parse the header in place when the reader already buffers all 9 bytes.
+function _read_frame_header!(reader::_ConnReader)::FrameHeader
+    _conn_reader_available(reader) >= 9 || return _parse_frame_header(_read_exact_bytes!(reader, 9), 1)
+    header = _parse_frame_header(reader.buf, reader.next)
+    reader.next += 9
+    return header
 end
 
 function _parse_settings_payload(payload::Vector{UInt8})::Vector{Pair{UInt16,UInt32}}
@@ -298,6 +306,13 @@ Throws:
 function read_frame!(io::IO)::AbstractFrame
     header = _read_frame_header!(io)
     header.length <= _H2_MAX_FRAME_SIZE || throw(ProtocolError("HTTP/2 frame exceeds max_frame_size"))
+    if io isa _ConnReader && header.type == FRAME_WINDOW_UPDATE && header.length == 4 && _conn_reader_available(io) >= 4
+        # Flow control sends these for every read; parse them in place.
+        increment = _read_u32_be(io.buf, io.next) & 0x7fff_ffff
+        io.next += 4
+        increment == 0 && throw(ProtocolError("HTTP/2 WINDOW_UPDATE increment must be > 0"))
+        return WindowUpdateFrame(header.stream_id, increment)
+    end
     payload = _read_exact_bytes!(io, header.length)
     # This function intentionally validates only frame-local invariants. Stream-
     # level rules such as "HEADERS must precede DATA" live in the client/server
@@ -382,17 +397,15 @@ end
         f = frame::DataFrame
         _require_nonzero_stream_id(f.stream_id, "DATA")
         flags = f.end_stream ? FLAG_END_STREAM : UInt8(0)
-        payload = copy(f.data)
-        return FrameHeader(length(payload), FRAME_DATA, flags, f.stream_id), payload
+        return FrameHeader(length(f.data), FRAME_DATA, flags, f.stream_id), f.data
     end
     if frame isa HeadersFrame
         f = frame::HeadersFrame
         _require_nonzero_stream_id(f.stream_id, "HEADERS")
         flags = UInt8(0)
         f.end_stream && (flags |= FLAG_END_STREAM)
         f.end_headers && (flags |= FLAG_END_HEADERS)
-        payload = copy(f.header_block_fragment)
-        return FrameHeader(length(payload), FRAME_HEADERS, flags, f.stream_id), payload
+        return FrameHeader(length(f.header_block_fragment), FRAME_HEADERS, flags, f.stream_id), f.header_block_fragment
     end
     if frame isa PriorityFrame
         f = frame::PriorityFrame
@@ -452,12 +465,11 @@ end
         f = frame::ContinuationFrame
         _require_nonzero_stream_id(f.stream_id, "CONTINUATION")
         flags = f.end_headers ? FLAG_END_HEADERS : UInt8(0)
-        payload = copy(f.header_block_fragment)
-        return FrameHeader(length(payload), FRAME_CONTINUATION, flags, f.stream_id), payload
+        return FrameHeader(length(f.header_block_fragment), FRAME_CONTINUATION, flags, f.stream_id), f.header_block_fragment
     end
     if frame isa UnknownFrame
         f = frame::UnknownFrame
-        return f.header, copy(f.payload)
+        return f.header, f.payload
     end
     throw(ArgumentError("unsupported HTTP/2 frame type"))
 end
@@ -473,9 +485,46 @@ cannot be represented legally, plus any exception raised by the underlying
 larger than the 16 KiB default.
 """
 function write_frame!(io::IO, frame::AbstractFrame)
+    write(io, _frame_bytes(frame))
+    return nothing
+end
+
+"""
+    _frame_bytes(frame) -> Vector{UInt8}
+
+Return the wire bytes of `frame`, its 9-byte header followed by the payload, in
+one new vector. Throws like [`write_frame!`](@ref).
+"""
+function _frame_bytes(frame::AbstractFrame)::Vector{UInt8}
     header, payload = _serialize_frame(frame)
-    write(io, _encode_header_bytes(header))
-    isempty(payload) || write(io, payload)
+    bytes = Vector{UInt8}(undef, 9 + length(payload))
+    _encode_header_bytes!(bytes, header)
+    copyto!(bytes, 10, payload, 1, length(payload))
+    return bytes
+end
+
+"""
+    _window_update_frames(stream_id, increment, stream_level) -> Vector{UInt8}
+
+Return the wire bytes of a connection-level WINDOW_UPDATE frame for `increment`,
+followed by the same update for `stream_id` when `stream_level` is `true`, so
+both go out in one write. `increment` must be positive.
+"""
+function _window_update_frames(stream_id::UInt32, increment::UInt32, stream_level::Bool)::Vector{UInt8}
+    increment > 0 || throw(ProtocolError("HTTP/2 WINDOW_UPDATE increment must be > 0"))
+    bytes = Vector{UInt8}(undef, stream_level ? 26 : 13)
+    _encode_window_update!(bytes, 1, UInt32(0), increment)
+    stream_level && _encode_window_update!(bytes, 14, stream_id, increment)
+    return bytes
+end
+
+@inline function _encode_window_update!(bytes::Vector{UInt8}, pos::Int, stream_id::UInt32, increment::UInt32)::Nothing
+    _encode_header_bytes!(bytes, FrameHeader(4, FRAME_WINDOW_UPDATE, UInt8(0), stream_id), pos)
+    value = increment & 0x7fff_ffff
+    bytes[pos+9] = UInt8((value >> 24) & 0xff)
+    bytes[pos+10] = UInt8((value >> 16) & 0xff)
+    bytes[pos+11] = UInt8((value >> 8) & 0xff)
+    bytes[pos+12] = UInt8(value & 0xff)
     return nothing
 end
 
@@ -489,7 +538,7 @@ function _header_block_frames(
     max_frame_size > 0 || throw(ArgumentError("max_frame_size must be > 0"))
     total = length(header_block)
     if total <= max_frame_size
-        f(HeadersFrame(stream_id, end_stream, true, copy(header_block)))
+        f(HeadersFrame(stream_id, end_stream, true, header_block))
         return nothing
     end
     offset = 1

src/http2_client.jl

@@ -264,17 +264,21 @@ function _set_h2_write_deadline!(conn::H2Connection, deadline_ns::Int64)::Nothin
 end
 
 function _write_frame_h2!(conn::H2Connection, frame::AbstractFrame, write_deadline_ns::Int64=Int64(0))
-    io = IOBuffer()
-    write_frame!(io, frame)
+    _write_frame_bytes_h2!(conn, _frame_bytes(frame), write_deadline_ns)
+    return nothing
+end
+
+function _write_frame_bytes_h2!(conn::H2Connection, bytes::Vector{UInt8}, write_deadline_ns::Int64)
     _set_h2_write_deadline!(conn, write_deadline_ns)
-    _write_all_h2!(conn, take!(io))
+    _write_all_h2!(conn, bytes)
     return nothing
 end
 
 function _write_frame_h2_threadsafe!(conn::H2Connection, frame::AbstractFrame, write_deadline_ns::Int64=Int64(0))
+    bytes = _frame_bytes(frame)
     lock(conn.write_lock)
     try
-        _write_frame_h2!(conn, frame, write_deadline_ns)
+        _write_frame_bytes_h2!(conn, bytes, write_deadline_ns)
     finally
         unlock(conn.write_lock)
     end
@@ -1080,8 +1084,13 @@ function _send_window_updates!(conn::H2Connection, stream_id::UInt32, nbytes::In
     nbytes <= 0 && return nothing
     increment = UInt32(nbytes)
     try
-        _write_frame_h2_threadsafe!(conn, WindowUpdateFrame(UInt32(0), increment))
-        stream_level && _write_frame_h2_threadsafe!(conn, WindowUpdateFrame(stream_id, increment))
+        bytes = _window_update_frames(stream_id, increment, stream_level)
+        lock(conn.write_lock)
+        try
+            _write_frame_bytes_h2!(conn, bytes, Int64(0))
+        finally
+            unlock(conn.write_lock)
+        end
     catch err
         if err isa EOFError || err isa IOPoll.NetClosingError || err isa SystemError
             return nothing

src/http2_server.jl

@@ -410,8 +410,14 @@ function _send_h2_server_window_updates!(
     @atomic send_state.conn_recv_window += nbytes
     deadline_ns = _h2_write_deadline_ns(send_state)
     try
-        _write_frame_h2_server_threadsafe!(write_lock, conn, WindowUpdateFrame(UInt32(0), increment), deadline_ns)
-        stream_level && _write_frame_h2_server_threadsafe!(write_lock, conn, WindowUpdateFrame(stream_id, increment), deadline_ns)
+        bytes = _window_update_frames(stream_id, increment, stream_level)
+        lock(write_lock)
+        try
+            _set_write_deadline!(conn, deadline_ns)
+            _write_all_h2_server!(conn, bytes)
+        finally
+            unlock(write_lock)
+        end
     catch err
         if err isa EOFError || err isa IOPoll.NetClosingError || err isa SystemError
             return nothing
@@ -770,10 +776,9 @@ end
 # deadline, `0` for none: the deadline outlives the write, and another task's
 # expired deadline would fail this one.
 function _write_frame_h2_server!(conn::Union{TCP.Conn,TLS.Conn}, frame::AbstractFrame, write_deadline_ns::Int64=Int64(0))::Nothing
-    io = IOBuffer()
-    write_frame!(io, frame)
+    bytes = _frame_bytes(frame)
     _set_write_deadline!(conn, write_deadline_ns)
-    _write_all_h2_server!(conn, take!(io))
+    _write_all_h2_server!(conn, bytes)
     return nothing
 end
 

test/http2_frame_tests.jl

@@ -238,3 +238,46 @@ end
     @test (frames[3]::HT.ContinuationFrame).end_headers
     @test (frames[3]::HT.ContinuationFrame).header_block_fragment == UInt8[0x09, 0x0a]
 end
+
+@testset "HTTP/2 frames read through a connection reader" begin
+    listener = Reseau.TCP.listen(Reseau.TCP.loopback_addr(0); backlog = 4)
+    addr = Reseau.TCP.addr(listener)::Reseau.TCP.SocketAddrV4
+    accepted = Channel{Reseau.TCP.Conn}(1)
+    errormonitor(Threads.@spawn put!(accepted, Reseau.TCP.accept(listener)))
+    client = Reseau.TCP.connect(Reseau.TCP.loopback_addr(Int(addr.port)))
+    server = take!(accepted)
+    try
+        updates = HT._window_update_frames(UInt32(3), UInt32(16384), true)
+        io = IOBuffer()
+        HT.write_frame!(io, HT.WindowUpdateFrame(UInt32(0), UInt32(16384)))
+        @test HT._window_update_frames(UInt32(3), UInt32(16384), false) == take!(copy(io))
+        HT.write_frame!(io, HT.WindowUpdateFrame(UInt32(3), UInt32(16384)))
+        @test updates == take!(io)
+        frames = HT.AbstractFrame[
+            HT.DataFrame(UInt32(3), false, UInt8[0x61, 0x62, 0x63]),
+            HT.PingFrame(false, (0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08)),
+        ]
+        bytes = vcat(updates, HT._frame_bytes(frames[1]), HT._frame_bytes(frames[2]))
+        function check(reader)
+            @test HT.read_frame!(reader) == HT.WindowUpdateFrame(UInt32(0), UInt32(16384))
+            @test HT.read_frame!(reader) == HT.WindowUpdateFrame(UInt32(3), UInt32(16384))
+            @test (HT.read_frame!(reader)::HT.DataFrame).data == UInt8[0x61, 0x62, 0x63]
+            @test (HT.read_frame!(reader)::HT.PingFrame).opaque_data == frames[2].opaque_data
+        end
+        # A 4-byte buffer splits every frame header and payload across reads.
+        write(client, bytes)
+        check(HT._ConnReader(server, 4))
+        # With the frames already buffered, headers and WINDOW_UPDATE frames are
+        # parsed in place: reading one allocates only the returned frame.
+        reader = HT._ConnReader(server)
+        copyto!(reader.buf, 1, bytes, 1, length(bytes))
+        reader.next, reader.stop = 1, length(bytes)
+        check(reader)
+        reader.next = 1
+        @test @allocated(HT.read_frame!(reader)) <= 64
+    finally
+        close(client)
+        close(server)
+        close(listener)
+    end
+end