Class: Raptor::Http2::FlowControl

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

Overview

Tracks the peer's connection-level and per-stream receive windows so outbound DATA frames respect RFC 7540 section 5.2.

Constant Summary collapse

ACQUIRE_POLL_INTERVAL =

Returns:

  • (::Float) 
0.001

Instance Method Summary collapse

Constructor Details

#initialize ⇒ FlowControl

Creates a new FlowControl with the spec-default windows.



91
92
93
94
95
 
# File 'lib/raptor/http2.rb', line 91

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

Instance Method Details

#acquire(stream_id, max_bytes, end_stream: false) ⇒ Integer

Reserves outbound capacity on the given stream, polling until at least one byte is available on both the connection and stream windows. The returned size is capped at MAX_FRAME_SIZE.

Parameters:

  • stream_id (Integer) —

    the HTTP/2 stream identifier

  • max_bytes (Integer) —

    the largest size the caller would like to send

  • end_stream (Boolean) (defaults to: false) —

    true when this is the final frame on the stream

  • end_stream: (Boolean) (defaults to: false)

Returns:

  • (Integer) —

    the number of bytes the caller may now send



107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
 
# File 'lib/raptor/http2.rb', line 107

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

  if end_stream && capped <= initial && !@stream_windows.value.key?(stream_id)
    loop do
      granted = reserve_connection(capped)
      return granted if granted > 0

      sleep ACQUIRE_POLL_INTERVAL
    end
  end

  loop do
    stream_window = @stream_windows.value[stream_id] || initial
    capped_full = capped < stream_window ? capped : stream_window
    granted = capped_full > 0 ? reserve_connection(capped_full) : 0

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

    sleep ACQUIRE_POLL_INTERVAL
  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



143
144
145
 
# File 'lib/raptor/http2.rb', line 143

def add_connection_window(increment)
  @connection_window.swap { |window| window + increment }
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



154
155
156
157
158
159
160
 
# File 'lib/raptor/http2.rb', line 154

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
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



186
187
188
189
190
191
192
193
194
195
196
 
# File 'lib/raptor/http2.rb', line 186

def discard_stream(stream_id)
  return unless @stream_windows.value.key?(stream_id)

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

    pruned = windows.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) 


208
209
210
211
212
213
214
215
 
# File 'lib/raptor/http2.rb', line 208

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



169
170
171
172
173
174
175
176
177
178
 
# File 'lib/raptor/http2.rb', line 169

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
end