Class: Hyperion::Http2Handler::WriterContext

Inherits:
Object
  • Object
show all
Defined in:
lib/hyperion/http2_handler.rb

Overview

Holds the per-connection outbound coordination state (queue, notifications, byte counters, shutdown flag) plus the encode mutex that protects HPACK state and per-stream frame ordering.

Single instance per connection, lives for the lifetime of ‘serve`.

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(max_pending_bytes: MAX_PER_CONN_PENDING_BYTES) ⇒ WriterContext

Returns a new instance of WriterContext.



144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
 
# File 'lib/hyperion/http2_handler.rb', line 144

def initialize(max_pending_bytes: MAX_PER_CONN_PENDING_BYTES)
  @queue              = ::Thread::Queue.new
  @send_notify        = ::Async::Notification.new
  @drained_notify     = ::Async::Notification.new
  @encode_mutex       = ::Mutex.new
  @pending_bytes      = 0
  @pending_bytes_lock = ::Mutex.new
  @max_pending_bytes  = max_pending_bytes
  @writer_done        = false
  # 2.11-A — pre-spawned dispatch worker pool. The connection-loop
  # fiber pushes ready streams onto `@dispatch_queue`; workers
  # parked on `dequeue` grab them and call `dispatch_stream`. The
  # queue is created here (cheap — wraps a Thread::Queue) so the
  # WriterContext is fully self-contained and unit-testable without
  # an Async reactor.
  @dispatch_queue           = ::Async::Queue.new
  @dispatch_worker_count    = 0
  @dispatch_worker_lock     = ::Mutex.new
  # 2.10-G timing slots, all initially nil so capture is a single
  # `||=` write under the encode mutex / writer fiber.
  @t0_serve_entry  = nil
  @t1_preface_done = nil
  @t2_first_encode = nil
  @t2_first_wire   = nil
end

Instance Attribute Details

#dispatch_queue ⇒ Object (readonly)

Returns the value of attribute dispatch_queue.



137
138
139
 
# File 'lib/hyperion/http2_handler.rb', line 137

def dispatch_queue
  @dispatch_queue
end

#encode_mutex ⇒ Object (readonly)

Returns the value of attribute encode_mutex.



137
138
139
 
# File 'lib/hyperion/http2_handler.rb', line 137

def encode_mutex
  @encode_mutex
end

#t0_serve_entry ⇒ Object

2.10-G — connection-lifecycle timing slots used by the optional h2 latency-instrumentation path (gated by ‘HYPERION_H2_TIMING=1`). Each slot is a single CLOCK_MONOTONIC timestamp captured at most once per connection. nil = unset, set on first observation.



142
143
144
 
# File 'lib/hyperion/http2_handler.rb', line 142

def t0_serve_entry
  @t0_serve_entry
end

#t1_preface_done ⇒ Object

2.10-G — connection-lifecycle timing slots used by the optional h2 latency-instrumentation path (gated by ‘HYPERION_H2_TIMING=1`). Each slot is a single CLOCK_MONOTONIC timestamp captured at most once per connection. nil = unset, set on first observation.



142
143
144
 
# File 'lib/hyperion/http2_handler.rb', line 142

def t1_preface_done
  @t1_preface_done
end

#t2_first_encode ⇒ Object

2.10-G — connection-lifecycle timing slots used by the optional h2 latency-instrumentation path (gated by ‘HYPERION_H2_TIMING=1`). Each slot is a single CLOCK_MONOTONIC timestamp captured at most once per connection. nil = unset, set on first observation.



142
143
144
 
# File 'lib/hyperion/http2_handler.rb', line 142

def t2_first_encode
  @t2_first_encode
end

#t2_first_wire ⇒ Object

2.10-G — connection-lifecycle timing slots used by the optional h2 latency-instrumentation path (gated by ‘HYPERION_H2_TIMING=1`). Each slot is a single CLOCK_MONOTONIC timestamp captured at most once per connection. nil = unset, set on first observation.



142
143
144
 
# File 'lib/hyperion/http2_handler.rb', line 142

def t2_first_wire
  @t2_first_wire
end

Instance Method Details

#dispatch_worker_count ⇒ Object

2.11-A — bench/diagnostics introspection. Reads the live count of dispatch worker fibers parked on (or actively pulling from) ‘@dispatch_queue`. Reflects pre-spawned workers AND any ad-hoc workers spawned when the pool was saturated. Exposed as a method rather than `attr_reader` so the lock guards the counter.



175
176
177
 
# File 'lib/hyperion/http2_handler.rb', line 175

def dispatch_worker_count
  @dispatch_worker_lock.synchronize { @dispatch_worker_count }
end

#enqueue(bytes) ⇒ Object

Called by SendQueueIO#write on the calling (encoder) fiber. Enforces the per-connection backpressure cap before enqueuing.



197
198
199
200
201
202
 
# File 'lib/hyperion/http2_handler.rb', line 197

def enqueue(bytes)
  wait_for_drain_if_full(bytes.bytesize)
  @pending_bytes_lock.synchronize { @pending_bytes += bytes.bytesize }
  @queue << bytes
  @send_notify.signal
end

#note_drained(bytesize) ⇒ Object

Called by the writer fiber after each successful drain to release any encoders blocked on the cap.



213
214
215
216
217
218
219
 
# File 'lib/hyperion/http2_handler.rb', line 213

def note_drained(bytesize)
  @pending_bytes_lock.synchronize do
    @pending_bytes -= bytesize
    @pending_bytes = 0 if @pending_bytes.negative? # paranoia
  end
  @drained_notify.signal
end

#pending_bytes ⇒ Object 



240
241
242
 
# File 'lib/hyperion/http2_handler.rb', line 240

def pending_bytes
  @pending_bytes_lock.synchronize { @pending_bytes }
end

#queue_empty? ⇒ Boolean

Returns:

  • (Boolean) 


236
237
238
 
# File 'lib/hyperion/http2_handler.rb', line 236

def queue_empty?
  @queue.empty?
end

#register_dispatch_worker ⇒ Object

Called by a dispatch worker fiber when it enters its run loop. Pairs with ‘unregister_dispatch_worker` in an ensure block.



181
182
183
 
# File 'lib/hyperion/http2_handler.rb', line 181

def register_dispatch_worker
  @dispatch_worker_lock.synchronize { @dispatch_worker_count += 1 }
end

#shutdown! ⇒ Object 



225
226
227
228
229
230
 
# File 'lib/hyperion/http2_handler.rb', line 225

def shutdown!
  @writer_done = true
  # Wake the writer if it's parked, and any encoder waiting on drain.
  @send_notify.signal
  @drained_notify.signal
end

#try_pop ⇒ Object

Pops a single chunk; returns nil if the queue is empty (non-blocking).



205
206
207
208
209
 
# File 'lib/hyperion/http2_handler.rb', line 205

def try_pop
  @queue.pop(true)
rescue ::ThreadError
  nil
end

#unregister_dispatch_worker ⇒ Object

Called by a dispatch worker fiber when it exits (queue closed, or unrecoverable error). Floors at 0 to defend against a stray double-unregister — instrumentation must never go negative.



188
189
190
191
192
193
 
# File 'lib/hyperion/http2_handler.rb', line 188

def unregister_dispatch_worker
  @dispatch_worker_lock.synchronize do
    @dispatch_worker_count -= 1
    @dispatch_worker_count = 0 if @dispatch_worker_count.negative?
  end
end

#wait_for_signal ⇒ Object 



221
222
223
 
# File 'lib/hyperion/http2_handler.rb', line 221

def wait_for_signal
  @send_notify.wait
end

#writer_done? ⇒ Boolean

Returns:

  • (Boolean) 


232
233
234
 
# File 'lib/hyperion/http2_handler.rb', line 232

def writer_done?
  @writer_done
end