Class: Raptor::Http2::FlowControl

Inherits:
Object
  • Object
show all
Defined in:
lib/raptor/http2.rb,
sig/generated/raptor/http2.rbs

Overview

Coordinates outbound flow control and stream cancellation without polling or mutexes.

Instance Method Summary collapse

Constructor Details

#initialize ⇒ FlowControl

Creates a new FlowControl with the spec-default windows.



130
131
132
133
134
135
136
 
# File 'lib/raptor/http2.rb', line 130

def initialize
  @connection_window = Atom.new(DEFAULT_WINDOW_SIZE)
  @stream_windows = Atom.new({})
  @initial_stream_window = Atom.new(DEFAULT_WINDOW_SIZE)
  @cancelled_streams = Atom.new({})
  @available = AtomicConditionVariable.new
end

Instance Method Details

#acquire(stream_id, max_bytes) ⇒ Integer

Reserves outbound capacity on the given stream, waiting until at least one byte is available on both the connection and stream windows.

Parameters:

  • stream_id (Integer) —

    the HTTP/2 stream identifier

  • max_bytes (Integer) —

    the largest size the caller would like to send

Returns:

  • (Integer) —

    the number of bytes the caller may now send

Raises:



148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
 
# File 'lib/raptor/http2.rb', line 148

def acquire(stream_id, max_bytes)
  capped = [max_bytes, MAX_FRAME_SIZE].min

  @available.wait do
    check(stream_id)
    initial = @initial_stream_window.value
    stream_window = @stream_windows.value[stream_id] || initial
    capped_full = [capped, stream_window].min
    granted = capped_full.positive? ? reserve_connection(capped_full) : 0

    if granted.positive?
      @stream_windows.swap do |windows|
        current = windows[stream_id] || initial
        windows.merge(stream_id => current - granted)
      end
      granted
    end
  end
end

#add_connection_window(increment) ⇒ void

This method returns an undefined value.

Increments the connection-level send window by increment bytes.

Parameters:

  • increment (Integer) —

    the byte count to add



199
200
201
202
 
# File 'lib/raptor/http2.rb', line 199

def add_connection_window(increment)
  @connection_window.swap { |window| window + increment }
  @available.broadcast
end

#add_stream_window(stream_id, increment) ⇒ void

This method returns an undefined value.

Increments the send window for the given stream by increment bytes.

Parameters:

  • stream_id (Integer) —

    the HTTP/2 stream identifier

  • increment (Integer) —

    the byte count to add



211
212
213
214
215
216
217
218
 
# File 'lib/raptor/http2.rb', line 211

def add_stream_window(stream_id, increment)
  initial = @initial_stream_window.value
  @stream_windows.swap do |windows|
    current = windows[stream_id] || initial
    windows.merge(stream_id => current + increment)
  end
  @available.broadcast
end

#cancel_stream(stream_id) ⇒ void

This method returns an undefined value.

Cancels pending and future writes for the given stream.

Parameters:

  • stream_id (Integer) —

    the HTTP/2 stream identifier



257
258
259
260
 
# File 'lib/raptor/http2.rb', line 257

def cancel_stream(stream_id)
  @cancelled_streams.swap { |streams| streams.merge(stream_id => true) }
  @available.broadcast
end

#check(stream_id) ⇒ void

This method returns an undefined value.

Raises when the given stream can no longer write a response.

Parameters:

  • stream_id (Integer) —

    the HTTP/2 stream identifier

Raises:



246
247
248
249
 
# File 'lib/raptor/http2.rb', line 246

def check(stream_id)
  cancelled = @cancelled_streams.value
  raise StreamClosedError if cancelled.key?(0) || cancelled.key?(stream_id)
end

#close ⇒ void

This method returns an undefined value.

Cancels pending and future writes for the connection.



267
268
269
270
 
# File 'lib/raptor/http2.rb', line 267

def close
  @cancelled_streams.swap { |streams| streams.merge(0 => true) }
  @available.broadcast
end

#discard_stream(stream_id) ⇒ void

This method returns an undefined value.

Discards any per-stream tracking for the given stream.

Parameters:

  • stream_id (Integer) —

    the HTTP/2 stream identifier



278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
 
# File 'lib/raptor/http2.rb', line 278

def discard_stream(stream_id)
  @stream_windows.swap do |windows|
    next windows unless windows.key?(stream_id)

    pruned = windows.dup
    pruned.delete(stream_id)
    pruned
  end
  @cancelled_streams.swap do |streams|
    next streams unless streams.key?(stream_id)

    pruned = streams.dup
    pruned.delete(stream_id)
    pruned
  end
end

#reserve_connection(capped) ⇒ Integer

Reserves up to capped bytes from the connection window.

Parameters:

  • capped (Integer) —

    the largest reservation the caller will accept

Returns:

  • (Integer) —

    bytes granted, or 0 when the window is exhausted



303
304
305
306
307
308
309
310
 
# File 'lib/raptor/http2.rb', line 303

def reserve_connection(capped)
  granted = 0
  @connection_window.swap do |window|
    granted = [window, capped].min
    granted.positive? ? window - granted : window
  end
  granted
end

#set_initial_stream_window(new_size) ⇒ void

This method returns an undefined value.

Updates the peer's SETTINGS_INITIAL_WINDOW_SIZE. Shifts every existing stream window by the delta as required by RFC 9113 section 6.9.2.

Parameters:

  • new_size (Integer) —

    the peer's new initial window size



227
228
229
230
231
232
233
234
235
236
237
 
# File 'lib/raptor/http2.rb', line 227

def set_initial_stream_window(new_size)
  old = @initial_stream_window.value
  @initial_stream_window.swap { new_size }
  delta = new_size - old
  return if delta.zero?

  @stream_windows.swap do |windows|
    windows.transform_values { |size| size + delta }
  end
  @available.broadcast
end

#try_acquire(stream_id, max_bytes) ⇒ Integer

Reserves outbound capacity without waiting for a window update.

Parameters:

  • stream_id (Integer) —

    the HTTP/2 stream identifier

  • max_bytes (Integer) —

    the largest size the caller would like to send

Returns:

  • (Integer) —

    available bytes, or 0 when flow control is exhausted

Raises:



176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
 
# File 'lib/raptor/http2.rb', line 176

def try_acquire(stream_id, max_bytes)
  check(stream_id)
  initial = @initial_stream_window.value
  stream_window = @stream_windows.value[stream_id] || initial
  capped = [max_bytes, MAX_FRAME_SIZE, stream_window].min
  return 0 unless capped.positive?

  granted = reserve_connection(capped)
  if granted.positive?
    @stream_windows.swap do |windows|
      current = windows[stream_id] || initial
      windows.merge(stream_id => current - granted)
    end
  end
  granted
end