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.



105
106
107
108
109
110
111
 
# File 'lib/raptor/http2.rb', line 105

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. Raises when the stream or connection closes.

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



122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
 
# File 'lib/raptor/http2.rb', line 122

def acquire(stream_id, max_bytes)
  initial = @initial_stream_window.value
  capped = max_bytes < MAX_FRAME_SIZE ? max_bytes : MAX_FRAME_SIZE

  @available.wait do
    check(stream_id)
    stream_window = @stream_windows.value[stream_id] || initial
    capped_full = capped < stream_window ? capped : stream_window
    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



172
173
174
175
 
# File 'lib/raptor/http2.rb', line 172

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



184
185
186
187
188
189
190
191
 
# File 'lib/raptor/http2.rb', line 184

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



230
231
232
233
 
# File 'lib/raptor/http2.rb', line 230

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:



219
220
221
222
 
# File 'lib/raptor/http2.rb', line 219

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.



240
241
242
243
 
# File 'lib/raptor/http2.rb', line 240

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



251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
 
# File 'lib/raptor/http2.rb', line 251

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, returning the number of bytes granted (0 when the window is exhausted).

Parameters:

  • capped (Integer) —

    the largest reservation the caller will accept

Returns:

  • (Integer) 


278
279
280
281
282
283
284
285
 
# File 'lib/raptor/http2.rb', line 278

def reserve_connection(capped)
  granted = 0
  @connection_window.swap do |window|
    granted = window > capped ? capped : window
    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 7540 section 6.9.2.

Parameters:

  • new_size (Integer) —

    the peer's new initial window size



200
201
202
203
204
205
206
207
208
209
210
 
# File 'lib/raptor/http2.rb', line 200

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



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

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