Rework window handling improving both correctness and efficiency.

Author: ioquatix

lib/protocol/http2/connection.rb

@@ -20,13 +20,14 @@
 
 require_relative 'framer'
 require_relative 'dependency'
+require_relative 'flow_controlled'
 
 require 'protocol/hpack'
 
 module Protocol
 	module HTTP2
 		class Connection
-			include FlowControl
+			include FlowControlled
 
 			def initialize(framer, local_stream_id)
 				super()
@@ -465,6 +466,15 @@ def receive_reset_stream(frame)
 				end
 			end
 
+			# Traverse active streams in order of priority and allow them to consume the available flow-control window.
+			# @param amount [Integer] the amount of data to write. Defaults to the current window capacity.
+			def consume_window(size = self.available_size)
+				# Return if there is no window to consume:
+				return unless size > 0
+				
+				@dependency.consume_window(size)
+			end
+			
 			def receive_window_update(frame)
 				if frame.connection?
 					super

lib/protocol/http2/dependency.rb

@@ -18,23 +18,31 @@
 # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
 # THE SOFTWARE.
 
-require_relative 'flow_control'
-
 module Protocol
 	module HTTP2
 		DEFAULT_WEIGHT = 16
 
 		class Dependency
-			def initialize(connection, id, dependent_id = 0, weight = DEFAULT_WEIGHT, children = nil)
+			def initialize(connection, id, dependent_id = 0, weight = DEFAULT_WEIGHT)
 				@connection = connection
 				@id = id
 
 				# Stream priority:
 				@dependent_id = dependent_id
 				@weight = weight
 
-				# A cache of dependencies that have child.dependent_id = self.id
-				@children = children
+				@children = nil
+				
+				# Cache of any associated stream:
+				@stream = nil
+				
+				# Cache of children for window allocation:
+				@total_weight = 0
+				@ordered_children = nil
+			end
+			
+			def <=> other
+				@weight <=> other.weight
 			end
 
 			def irrelevant?
@@ -62,34 +70,36 @@ def delete!
 			attr_accessor :weight
 
 			def stream
-				@connection.streams[@id]
+				@stream ||= @connection.streams[@id]
 			end
 
-			def streams
-				if @children
-					# TODO this O(N) operation affects performance.
-					# It would be better to maintain a sorted list of children streams.
-					@children.map{|id, dependency| dependency.stream}.compact
-				end
+			def clear_cache!
+				@ordered_children = nil
 			end
 
 			def add_child(dependency)
 				@children ||= {}
 				@children[dependency.id] = dependency
+				
+				self.clear_cache!
 			end
 
 			def remove_child(dependency)
 				@children&.delete(dependency.id)
+				
+				self.clear_cache!
 			end
 
 			def exclusive_child(parent)
 				parent.children = @children
+				parent.clear_cache!
 
 				@children.each_value do |child|
 					child.dependent_id = parent.id
 				end
 
 				@children = {parent.id => parent}
+				self.clear_cache!
 
 				parent.dependent_id = @id
 			end
@@ -125,6 +135,8 @@ def process_priority priority
 					@dependent_id = dependent_id
 
 					self.parent.add_child(self)
+				else
+					self.parent&.clear_cache!
 				end
 			end
 
@@ -146,6 +158,34 @@ def send_priority(priority)
 			def receive_priority(frame)
 				self.process_priority(frame.unpack)
 			end
+			
+			def ordered_children
+				unless @ordered_children
+					if @children and !@children.empty?
+						@ordered_children = @children.values.sort
+						@total_weight = @ordered_children.sum(&:weight)
+					end
+				end
+				
+				return @ordered_children
+			end
+			
+			# Traverse active streams in order of priority and allow them to consume the available flow-control window.
+			# @param amount [Integer] the amount of data to write. Defaults to the current window capacity.
+			def consume_window(size)
+				# If there is an associated stream, give it priority:
+				if stream = self.stream
+					return if stream.window_updated(size)
+				end
+				
+				# Otherwise, allow the dependent children to use up the available window:
+				self.ordered_children&.each do |child|
+					# Compute the proportional allocation:
+					allocated = (child.weight * size) / @total_weight
+					
+					child.consume_window(allocated) if allocated > 0
+				end
+			end
 		end
 	end
 end

lib/protocol/http2/flow_control.rb renamed to lib/protocol/http2/flow_controlled.rb

@@ -23,7 +23,7 @@
 
 module Protocol
 	module HTTP2
-		module FlowControl
+		module FlowControlled
 			def available_size
 				@remote_window.available
 			end
@@ -98,35 +98,9 @@ def receive_window_update(frame)
 			# The window has been expanded by the given amount.
 			# @param size [Integer] the maximum amount of data to send.
 			# @return [Boolean] whether the window update was used or not.
-			def window_updated(size = self.available_size)
+			def window_updated(size)
 				return false
 			end
-			
-			# Traverse active streams in order of priority and allow them to consume the available flow-control window.
-			# @todo This function can get slow when there are a lot of children [INEFFICIENT].
-			# @param amount [Integer] the amount of data to write. Defaults to the current window capacity.
-			def consume_window(size = self.available_size)
-				# Don't consume more than the available window size:
-				size = [self.available_size, size].min
-				# puts "consume_window(#{size}) local_window=#{@local_window} remote_window=#{@remote_window}"
-				
-				# Return if there is no window to consume:
-				return unless size > 0
-				
-				# Allow the current flow-controlled instance to use up the window:
-				if !self.window_updated(size) and children = self.children
-					children = children.sort_by(&:weight)
-					
-					# This must always be at least >= `children.size`, since stream weight can't be 0.
-					total = children.sum(&:weight)
-					
-					children.each do |child|
-						# Compute the proportional allocation:
-						allocated = (child.weight * size) / total
-						child.consume_window(allocated)
-					end
-				end
-			end
 		end
 	end
 end

lib/protocol/http2/stream.rb

@@ -72,7 +72,7 @@ module HTTP2
 		#
 		# State transition methods use a trailing "!".
 		class Stream
-			include FlowControl
+			include FlowControlled
 
 			def self.create(connection, id)
 				stream = self.new(connection, id)
@@ -124,10 +124,6 @@ def parent=(stream)
 				@dependency.parent = stream.dependency
 			end
 
-			def children
-				@dependency&.streams
-			end
-			
 			# The stream is being closed because the connection is being closed.
 			def close(error = nil)
 			end

lib/protocol/http2/version.rb

@@ -20,6 +20,6 @@
 
 module Protocol
 	module HTTP2
-		VERSION = "0.11.3"
+		VERSION = "0.11.4"
 	end
 end

spec/protocol/http2/dependency_spec.rb

@@ -57,7 +57,7 @@
 			server.read_frame
 		end
 
-		expect(a.children).to be == [b]
+		expect(a.dependency.children).to be == {b.id => b.dependency}
 		expect(server.dependencies[a.id].children).to be == {b.id => server.dependencies[b.id]}
 
 		#    a
@@ -71,7 +71,7 @@
 			server.read_frame
 		end
 
-		expect(a.children).to be == [b, c]
+		expect(a.dependency.children).to be == {b.id => b.dependency, c.id => c.dependency}
 		expect(server.dependencies[a.id].children).to be == {b.id => server.dependencies[b.id], c.id => server.dependencies[c.id]}
 
 		#    a
@@ -88,9 +88,32 @@
 			server.read_frame
 		end
 
-		expect(a.children).to be == [d]
-		expect(d.children).to be == [b, c]
+		expect(a.dependency.children).to be == {d.id => d.dependency}
+		expect(d.dependency.children).to be == {b.id => b.dependency, c.id => c.dependency}
 		expect(server.dependencies[a.id].children).to be == {d.id => server.dependencies[d.id]}
 		expect(server.dependencies[d.id].children).to be == {b.id => server.dependencies[b.id], c.id => server.dependencies[c.id]}
 	end
+	
+	it "correctly allocates window" do
+		parent = client.create_stream
+		children = 2.times.collect {client.create_stream}
+		
+		children.each do |child|
+			priority = child.priority
+			priority.stream_dependency = parent.id
+			child.priority = priority
+		end
+		
+		2.times {server.read_frame}
+		
+		# Now we have this prioritization on the server:
+		#    a
+		#   / \
+		#  b   c
+		
+		expect(server.dependencies[children[0].id]).to receive(:consume_window).with(0xFFFF / 2).once
+		expect(server.dependencies[children[1].id]).to receive(:consume_window).with(0xFFFF / 2).once
+		
+		server.consume_window
+	end
 end