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.

RBS:

  • () -> void



89
90
91
92
93
 
# File 'lib/raptor/http2.rb', line 89

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.

RBS:

  • (Integer stream_id, Integer max_bytes, ?end_stream: bool) -> Integer

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



105
106
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
 
# File 'lib/raptor/http2.rb', line 105

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.positive?

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

RBS:

  • (Integer increment) -> void

Parameters:

  • increment (Integer) —

    the byte count to add



141
142
143
 
# File 'lib/raptor/http2.rb', line 141

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.

RBS:

  • (Integer stream_id, Integer increment) -> void

Parameters:

  • stream_id (Integer) —

    the HTTP/2 stream identifier

  • increment (Integer) —

    the byte count to add



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

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.

RBS:

  • (Integer stream_id) -> void

Parameters:

  • stream_id (Integer) —

    the HTTP/2 stream identifier



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

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

RBS:

  • (Integer capped) -> Integer

Parameters:

  • capped (Integer) —

    the largest reservation the caller will accept

Returns:

  • (Integer) 


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

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.

RBS:

  • (Integer new_size) -> void

Parameters:

  • new_size (Integer) —

    the peer's new initial window size



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

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