Improve robustness of dependency tree.

The dependency tree grows unless pruned. We prune the dependency
information when the stream is closed, and in addition, fix the tree
manipulations to be robust in the face of missing stream dependencies.

Author: ioquatix

lib/protocol/http2/connection.rb

@@ -63,14 +63,6 @@ def id
 				0
 			end
 
-			def parent
-				nil
-			end
-			
-			def children
-				@dependency.streams
-			end
-			
 			def [] id
 				if id.zero?
 					self
@@ -113,10 +105,7 @@ def closed?
 
 			def delete(id)
 				@streams.delete(id)
-				
-				if dependency = @dependencies[id]
-					dependency.delete! if dependency.irrelevant?
-				end
+				@dependencies[id]&.delete!
 			end
 
 			# Close the underlying framer and all streams.
@@ -145,20 +134,10 @@ def next_stream_id
 			end
 
 			attr :streams
+			
 			attr :dependencies
 
-			# Fetch (or create) the flow control windows for the specified stream id.
-			# @param id [Integer] the stream id.
-			def dependency_for(id)
-				@dependencies.fetch(id) do
-					dependency = Dependency.new(self, id)
-					
-					# TODO this might be irrelevant, if initially processing priority frame.
-					@dependency.add_child(dependency)
-					
-					@dependencies[id] = dependency
-				end
-			end
+			attr :dependency
 
 			# 6.8. GOAWAY
 			# There is an inherent race condition between an endpoint starting new streams and the remote sending a GOAWAY frame. To deal with this case, the GOAWAY contains the stream identifier of the last peer-initiated stream that was or might be processed on the sending endpoint in this connection. For instance, if the server sends a GOAWAY frame, the identified stream is the highest-numbered stream initiated by the client.
@@ -415,10 +394,31 @@ def send_priority(stream_id, priority)
 				write_frame(frame)
 			end
 
+			def idle_stream_id?(id)
+				if id.even?
+					# Server-initiated streams are even.
+					if @local_stream_id.even?
+						id >= @local_stream_id
+					else
+						id > @remote_stream_id
+					end
+				elsif id.odd?
+					# Client-initiated streams are odd.
+					if @local_stream_id.odd?
+						id >= @local_stream_id
+					else
+						id > @remote_stream_id
+					end
+				end
+			end
+			
 			# Sets the priority for an incoming stream.
 			def receive_priority(frame)
-				dependency = dependency_for(frame.stream_id)
-				dependency.receive_priority(frame)
+				if dependency = @dependencies[frame.stream_id]
+					dependency.receive_priority(frame)
+				elsif idle_stream_id?(frame.stream_id)
+					Dependency.create(self, frame.stream_id, frame.unpack)
+				end
 			end
 
 			def receive_push_promise(frame)
@@ -472,6 +472,16 @@ def consume_window(size = self.available_size)
 				# Return if there is no window to consume:
 				return unless size > 0
 
+				# Console.logger.debug(self) do |buffer|
+				# 	@dependencies.each do |id, dependency|
+				# 		buffer.puts "- #{dependency}"
+				# 	end
+				# 
+				# 	buffer.puts
+				# 
+				# 	@dependency.print_hierarchy(buffer)
+				# end
+				
 				@dependency.consume_window(size)
 			end
 

lib/protocol/http2/dependency.rb

@@ -23,16 +23,44 @@ module HTTP2
 		DEFAULT_WEIGHT = 16
 
 		class Dependency
-			def initialize(connection, id, dependent_id = 0, weight = DEFAULT_WEIGHT)
+			def self.create(connection, id, priority = nil)
+				weight = DEFAULT_WEIGHT
+				exclusive = false
+				
+				if priority
+					if parent = connection.dependencies[priority.stream_dependency]
+						exclusive = priority.exclusive
+					end
+					
+					weight = priority.weight
+				end
+				
+				if parent.nil?
+					parent = connection.dependency
+				end
+				
+				dependency = self.new(connection, id, weight)
+				
+				connection.dependencies[id] = dependency
+				
+				if exclusive
+					parent.exclusive_child(dependency)
+				else
+					parent.add_child(dependency)
+				end
+				
+				return dependency
+			end
+			
+			def initialize(connection, id, weight = DEFAULT_WEIGHT)
 				@connection = connection
 				@id = id
 
-				# Stream priority:
-				@dependent_id = dependent_id
-				@weight = weight
-				
+				@parent = nil
 				@children = nil
 
+				@weight = weight
+				
 				# Cache of any associated stream:
 				@stream = nil
 
@@ -45,26 +73,17 @@ def <=> other
 				@weight <=> other.weight
 			end
 
-			def irrelevant?
-				(@weight == DEFAULT_WEIGHT) && (@children.nil? || @children.empty?)
-			end
-			
-			def delete!
-				@connection.dependencies.delete(@id)
-				self.parent&.remove_child(self)
-			end
-			
-			# Cache of dependent children.
-			attr_accessor :children
-			
 			# The connection this stream belongs to.
 			attr :connection
 
 			# Stream ID (odd for client initiated streams, even otherwise).
 			attr :id
 
-			# The stream id that this stream depends on, according to the priority.
-			attr_accessor :dependent_id
+			# The parent dependency.
+			attr_accessor :parent
+			
+			# The dependent children.
+			attr_accessor :children
 
 			# The weight of the stream relative to other siblings.
 			attr_accessor :weight
@@ -77,10 +96,26 @@ def clear_cache!
 				@ordered_children = nil
 			end
 
+			def delete!
+				@connection.dependencies.delete(@id)
+				
+				@parent.remove_child(self)
+				
+				@children&.each do |id, child|
+					parent.add_child(child)
+				end
+				
+				@connection = nil
+				@parent = nil
+				@children = nil
+			end
+			
 			def add_child(dependency)
 				@children ||= {}
 				@children[dependency.id] = dependency
 
+				dependency.parent = self
+				
 				self.clear_cache!
 			end
 
@@ -90,33 +125,24 @@ def remove_child(dependency)
 				self.clear_cache!
 			end
 
+			# An exclusive flag allows for the insertion of a new level of dependencies.  The exclusive flag causes the stream to become the sole dependency of its parent stream, causing other dependencies to become dependent on the exclusive stream.
+			# @param parent [Dependency] the dependency which will be inserted, taking control of all current children.
 			def exclusive_child(parent)
 				parent.children = @children
-				parent.clear_cache!
 
-				@children.each_value do |child|
-					child.dependent_id = parent.id
+				@children&.each_value do |child|
+					child.parent = parent
 				end
 
+				parent.clear_cache!
+				
 				@children = {parent.id => parent}
 				self.clear_cache!
 
-				parent.dependent_id = @id
-			end
-			
-			def parent(id = @dependent_id)
-				@connection.dependency_for(id)
-			end
-			
-			def parent= dependency
-				self.parent&.remove_child(self)
-				
-				@dependent_id = dependency.id
-				
-				dependency.add_child(self)
+				parent.parent = self
 			end
 
-			def process_priority priority
+			def process_priority(priority)
 				dependent_id = priority.stream_dependency
 
 				if dependent_id == @id
@@ -125,18 +151,17 @@ def process_priority priority
 
 				@weight = priority.weight
 
-				if priority.exclusive
-					self.parent&.remove_child(self)
-					
-					self.parent(dependent_id).exclusive_child(self)
-				elsif dependent_id != @dependent_id
-					self.parent&.remove_child(self)
-					
-					@dependent_id = dependent_id
-					
-					self.parent.add_child(self)
-				else
-					self.parent&.clear_cache!
+				# We essentially ignore `dependent_id` if the dependency does not exist:
+				if parent = @connection.dependencies[dependent_id]
+					if priority.exclusive
+						@parent.remove_child(self)
+						
+						parent.exclusive_child(self)
+					elsif !@parent.equal?(parent)
+						@parent.remove_child(self)
+						
+						parent.add_child(self)
+					end
 				end
 			end
 
@@ -148,7 +173,7 @@ def priority= priority
 
 			# The current local priority of the stream.
 			def priority(exclusive = false)
-				Priority.new(exclusive, @dependent_id, @weight)
+				Priority.new(exclusive, @parent.id, @weight)
 			end
 
 			def send_priority(priority)
@@ -159,6 +184,12 @@ def receive_priority(frame)
 				self.process_priority(frame.unpack)
 			end
 
+			def total_weight
+				self.orderd_children
+				
+				return @total_weight
+			end
+			
 			def ordered_children
 				unless @ordered_children
 					if @children and !@children.empty?
@@ -186,6 +217,17 @@ def consume_window(size)
 					child.consume_window(allocated) if allocated > 0
 				end
 			end
+			
+			def to_s
+				"\#<#{self.class} id=#{@id} parent id=#{@parent&.id} weight=#{@weight} #{@children&.size || 0} children>"
+			end
+			
+			def print_hierarchy(buffer, indent: 0)
+				buffer.puts "#{" " * indent}#{self}"
+				@children&.each_value do |child|
+					child.print_hierarchy(buffer, indent: indent+1)
+				end
+			end
 		end
 	end
 end

lib/protocol/http2/stream.rb

@@ -91,7 +91,7 @@ def initialize(connection, id, state = :idle)
 				@local_window = Window.new(@connection.local_settings.initial_window_size)
 				@remote_window = Window.new(@connection.remote_settings.initial_window_size)
 
-				@dependency = @connection.dependency_for(@id)
+				@dependency = Dependency.create(@connection, @id)
 			end
 
 			# The connection this stream belongs to.

spec/protocol/http2/dependency_spec.rb

@@ -50,6 +50,9 @@
 		#   /
 		#  b
 		begin
+			a.priority = a.priority
+			server.read_frame
+			
 			priority = b.priority
 			priority.stream_dependency = a.id
 			b.priority = priority
@@ -58,6 +61,8 @@
 		end
 
 		expect(a.dependency.children).to be == {b.id => b.dependency}
+		
+		expect(server.dependencies).to include(a.id)
 		expect(server.dependencies[a.id].children).to be == {b.id => server.dependencies[b.id]}
 
 		#    a