Class: Hyperion::Http2Handler

Inherits:

Object

• Object
• Hyperion::Http2Handler

Defined in:

lib/hyperion/http2_handler.rb

Overview

Real HTTP/2 dispatch driven by ‘protocol-http2`.

Each TLS connection that negotiated ‘h2` via ALPN ends up here. We frame the socket, read the connection preface, and then drive a frame loop on the connection’s fiber: it reads one frame at a time and lets ‘protocol-http2` update its connection/stream state machines. As soon as a client stream finishes its request half (state `:half_closed_remote` via `end_stream?`), we hand the stream off to a sibling fiber for dispatch — slow handlers no longer block other streams on the same connection.

## Outbound write architecture (1.6.0+)

Pre-1.6.0 every framer write (HEADERS / DATA / RST_STREAM / GOAWAY) ran under one connection-scoped ‘Mutex#synchronize { socket.write(…) }`. That capped per-connection h2 throughput to “one socket-write at a time” regardless of stream count: a slow socket (kernel send buffer full, remote peer reading slowly) blocked every other stream’s writes too.

1.6.0 splits the path:

* The HPACK encode + frame format step is fast (microseconds, in-memory)
  and remains serialized on the calling fiber via `@encode_mutex`. HPACK
  state is stateful across HEADERS frames per connection, and frames for
  a single stream must be wire-ordered (HEADERS → DATA → END_STREAM).
  Holding the encode mutex across a `send_*` call accomplishes both.
* The framer writes through a `SendQueueIO` wrapper (wraps the real
  socket). `SendQueueIO#write(bytes)` enqueues onto a connection-wide
  `@send_queue` and signals `@send_notify`; it never touches the real
  socket.
* A dedicated **writer fiber** owns the real socket. It pops byte chunks
  off the queue, writes them, and parks on `@send_notify` when empty.
  Only this fiber ever calls `socket.write` — the SSLSocket cross-fiber
  unsafety constraint is satisfied.

Net effect: the slow-socket case no longer serializes encode work across streams. A stream that has bytes ready to encode can encode and enqueue while the writer is mid-flush of an earlier chunk. The mutex hold time drops from “until the kernel accepts the write” to “until the bytes are appended to the in-memory queue.”

Backpressure: pathological clients (slow-read h2) could otherwise let the queue grow without bound. We track ‘@pending_bytes`; once it exceeds `MAX_PER_CONN_PENDING_BYTES`, encoding fibers wait on `@drained_notify` before enqueueing more. The writer signals `@drained_notify` after each drain pass.

Flow control: ‘RequestStream#window_updated` overrides the protocol-http2 default to fan a notification out to any fiber blocked in `send_body` waiting for the remote peer’s flow-control window to grow. The body writer chunks the response payload by the per-stream available frame size and yields on the notification when the window is exhausted, so large bodies never trip a FlowControlError.

Defined Under Namespace

Classes: RequestStream, SendQueueIO, WriterContext

Constant Summary

MAX_PER_CONN_PENDING_BYTES =

Cap on bytes that may sit in a connection’s send queue waiting for the writer fiber to drain. Slow-read h2 clients can otherwise let an encoder fiber pile arbitrary bytes into RAM. 16 MiB matches the upper bound a well-behaved peer will buffer — anything beyond that is the writer being starved, and the right answer is to backpressure the encoder rather than allocate more.

16 * 1024 * 1024

SETTINGS_KEY_MAP =

Maps Hyperion-friendly setting names to the integer SETTINGS_* identifiers protocol-http2 uses on the wire. See RFC 7540 §6.5.2 — these are the only four parameters Hyperion exposes; the rest of the SETTINGS frame (HEADER_TABLE_SIZE, ENABLE_PUSH, etc.) keeps protocol-http2’s default.

{
  max_concurrent_streams: ::Protocol::HTTP2::Settings::MAXIMUM_CONCURRENT_STREAMS,
  initial_window_size: ::Protocol::HTTP2::Settings::INITIAL_WINDOW_SIZE,
  max_frame_size: ::Protocol::HTTP2::Settings::MAXIMUM_FRAME_SIZE,
  max_header_list_size: ::Protocol::HTTP2::Settings::MAXIMUM_HEADER_LIST_SIZE
}.freeze

H2_MIN_FRAME_SIZE =

RFC 7540 §6.5.2 floor for SETTINGS_MAX_FRAME_SIZE. protocol-http2 raises ProtocolError on values below this; we clamp + warn instead so a misconfigured operator gets a working server, not a boot-time crash.

0x4000

H2_MAX_FRAME_SIZE =

RFC 7540 §6.5.2 ceiling for SETTINGS_MAX_FRAME_SIZE.

0xFFFFFF

H2_MAX_WINDOW_SIZE =

RFC 7540 §6.9.2 — INITIAL_WINDOW_SIZE has the same 31-bit max as the WINDOW_UPDATE frame’s Window Size Increment (see protocol-http2’s MAXIMUM_ALLOWED_WINDOW_SIZE).

0x7FFFFFFF

Instance Method Summary

• #initialize(app:, thread_pool: nil, h2_settings: nil) ⇒ Http2Handler constructor

  A new instance of Http2Handler.

• #serve(socket) ⇒ Object

Constructor Details

#initialize(app:, thread_pool: nil, h2_settings: nil) ⇒ Http2Handler

Returns a new instance of Http2Handler.

# File 'lib/hyperion/http2_handler.rb', line 415

def initialize(app:, thread_pool: nil, h2_settings: nil)
  @app         = app
  @thread_pool = thread_pool
  @h2_settings = h2_settings
  @metrics     = Hyperion.metrics
  @logger      = Hyperion.logger
end

Instance Method Details

#serve(socket) ⇒ Object

# File 'lib/hyperion/http2_handler.rb', line 423

def serve(socket)
  @metrics.increment(:connections_accepted)
  @metrics.increment(:connections_active)

  # Per-connection outbound coordination. Encoder fibers enqueue bytes;
  # the writer fiber owns the real socket and drains. See class docstring.
  writer_ctx   = WriterContext.new
  send_io      = SendQueueIO.new(socket, writer_ctx)
  framer       = ::Protocol::HTTP2::Framer.new(send_io)
  server       = build_server(framer)

  task = ::Async::Task.current

  # Spawn the dedicated writer fiber BEFORE the preface exchange.
  # `Server#read_connection_preface` writes the server's SETTINGS frame
  # via the framer; if the writer isn't running, those bytes sit in the
  # queue. Spawning first guarantees they flush as soon as the scheduler
  # ticks, avoiding any pathological deadlock where a client implementation
  # waits for our SETTINGS before sending more frames.
  writer_task = task.async { run_writer_loop(socket, writer_ctx) }

  server.read_connection_preface(initial_settings_payload)

  # Extract once — the same TCP peer drives every stream on this conn.
  peer_addr = peer_address(socket)

  # Track in-flight per-stream dispatch fibers so we can drain them on
  # connection close.
  stream_tasks = []

  until server.closed?
    ready_ids = []
    server.read_frame do |frame|
      ready_ids << frame.stream_id if frame.stream_id.positive?
    end

    ready_ids.uniq.each do |sid|
      stream = server.streams[sid]
      next unless stream.is_a?(RequestStream)
      next unless stream.request_complete
      next if stream.closed?
      next if stream.instance_variable_get(:@hyperion_dispatched)

      # Mark before spawning so we never dispatch the same stream twice
      # if subsequent frames (e.g. RST_STREAM races) arrive.
      stream.instance_variable_set(:@hyperion_dispatched, true)

      stream_tasks << task.async do
        dispatch_stream(stream, writer_ctx, peer_addr)
      end
    end
  end

  # Drain in-flight stream dispatches before we close the socket.
  stream_tasks.each do |t|
    t.wait
  rescue StandardError
    nil
  end
rescue EOFError, Errno::ECONNRESET, Errno::EPIPE, IOError, OpenSSL::SSL::SSLError
  # Peer disconnect — nothing to do.
rescue ::Protocol::HTTP2::GoawayError, ::Protocol::HTTP2::ProtocolError, ::Protocol::HTTP2::HandshakeError
  # Protocol-level error — protocol-http2 has already emitted GOAWAY.
rescue StandardError => e
  @logger.error do
    {
      message: 'h2 connection error',
      error: e.message,
      error_class: e.class.name,
      backtrace: (e.backtrace || []).first(10).join(' | ')
    }
  end
ensure
  # Coordinated shutdown: flag the writer, signal it, wait for the final
  # drain, then close the real socket. Order matters — closing the
  # socket before the writer drains would discard final RST_STREAM /
  # GOAWAY / END_STREAM frames in the queue.
  if writer_ctx
    writer_ctx.shutdown!
    begin
      writer_task&.wait
    rescue StandardError
      nil
    end
  end
  @metrics.decrement(:connections_active)
  socket.close unless socket.closed?
end