Separate stream and priority logic to improve memory efficiency.

Author: ioquatix

lib/protocol/http2/connection.rb

@@ -19,7 +19,7 @@
 # THE SOFTWARE.
 
 require_relative 'framer'
-require_relative 'flow_control'
+require_relative 'dependency'
 
 require 'protocol/hpack'
 
@@ -32,10 +32,13 @@ def initialize(framer, local_stream_id)
 				super()
 
 				@state = :new
+				
+				# Hash(Integer, Stream)
 				@streams = {}
-				@children = {}
 
-				@active = 0
+				# Hash(Integer, Dependency)
+				@dependency = Dependency.new(self, 0)
+				@dependencies = {0 => @dependency}
 
 				@framer = framer
 				@local_stream_id = local_stream_id
@@ -59,6 +62,10 @@ def parent
 				nil
 			end
 
+			def children
+				@dependency.streams
+			end
+			
 			def [] id
 				if id.zero?
 					self
@@ -99,24 +106,18 @@ def closed?
 				@state == :closed
 			end
 
-			# The stream has become active (i.e. not idle or closed).
-			def activate(stream)
-				@active += 1
-			end
-			
-			# The stream is no longer active (i.e. has become closed).
-			def deactivate(stream)
-				@active -= 1
-			end
-			
-			# The number of active streams.
-			def active_streams
-				@active
+			def delete(id)
+				@streams.delete(id)
+				
+				if dependency = @dependencies[id]
+					dependency.delete! if dependency.irrelevant?
+				end
 			end
 
 			# Close the underlying framer and all streams.
 			def close(error = nil)
 				@framer.close
+				# @framer = nil
 
 				@streams.each_value{|stream| stream.close(error)}
 			end
@@ -139,26 +140,19 @@ def next_stream_id
 			end
 
 			attr :streams
-			attr :children
-			
-			def add_child(stream)
-				@children[stream.id] = stream
-			end
+			attr :dependencies
 
-			def remove_child(stream)
-				@children.delete(stream.id)
-			end
-			
-			def exclusive_child(stream)
-				stream.children = @children
-				
-				@children.each_value do |child|
-					child.dependent_id = stream.id
+			# 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
-				
-				@children = {stream.id => stream}
-				
-				stream.dependent_id = 0
 			end
 
 			# 6.8. GOAWAY
@@ -398,7 +392,7 @@ def receive_headers(frame)
 						raise ProtocolError, "Invalid stream id: #{stream_id} <= #{@remote_stream_id}!"
 					end
 
-					if self.active_streams.size < self.maximum_concurrent_streams
+					if @streams.size < self.maximum_concurrent_streams
 						stream = accept_stream(stream_id)
 						@remote_stream_id = stream_id
 
@@ -418,13 +412,8 @@ def send_priority(stream_id, priority)
 
 			# Sets the priority for an incoming stream.
 			def receive_priority(frame)
-				if stream = @streams[frame.stream_id]
-					stream.receive_priority(frame)
-				else
-					# Stream doesn't exist yet.
-					stream = accept_stream(frame.stream_id)
-					stream.receive_priority(frame)
-				end
+				dependency = dependency_for(frame.stream_id)
+				dependency.receive_priority(frame)
 			end
 
 			def receive_push_promise(frame)
@@ -450,7 +439,7 @@ def receive_window_update(frame)
 					rescue ProtocolError => error
 						stream.send_reset_stream(error.code)
 					end
-				else
+				elsif frame.stream_id > @remote_stream_id
 					# Receiving any frame other than HEADERS or PRIORITY on a stream in this state MUST be treated as a connection error of type PROTOCOL_ERROR.
 					raise ProtocolError, "Cannot update window of idle stream #{frame.stream_id}"
 				end

lib/protocol/http2/dependency.rb

@@ -0,0 +1,151 @@
+# Copyright, 2018, by Samuel G. D. Williams. <http://www.codeotaku.com>
+# 
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the "Software"), to deal
+# in the Software without restriction, including without limitation the rights
+# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+# 
+# The above copyright notice and this permission notice shall be included in
+# all copies or substantial portions of the Software.
+# 
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# 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)
+				@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
+			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 weight of the stream relative to other siblings.
+			attr_accessor :weight
+			
+			def 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
+			end
+			
+			def add_child(dependency)
+				@children ||= {}
+				@children[dependency.id] = dependency
+			end
+			
+			def remove_child(dependency)
+				@children&.delete(dependency.id)
+			end
+			
+			def exclusive_child(parent)
+				parent.children = @children
+				
+				@children.each_value do |child|
+					child.dependent_id = parent.id
+				end
+				
+				@children = {parent.id => parent}
+				
+				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)
+			end
+			
+			def process_priority priority
+				dependent_id = priority.stream_dependency
+				
+				if dependent_id == @id
+					raise ProtocolError, "Stream priority for stream id #{@id} cannot depend on itself!"
+				end
+				
+				@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)
+				end
+			end
+			
+			# Change the priority of the stream both locally and remotely.
+			def priority= priority
+				send_priority(priority)
+				process_priority(priority)
+			end
+			
+			# The current local priority of the stream.
+			def priority(exclusive = false)
+				Priority.new(exclusive, @dependent_id, @weight)
+			end
+			
+			def send_priority(priority)
+				@connection.send_priority(@id, priority)
+			end
+			
+			def receive_priority(frame)
+				self.process_priority(frame.unpack)
+			end
+		end
+	end
+end

lib/protocol/http2/flow_control.rb

@@ -115,7 +115,7 @@ def consume_window(size = self.available_size)
 
 				# Allow the current flow-controlled instance to use up the window:
 				if !self.window_updated(size) and children = self.children
-					children = children.values.sort_by(&:weight)
+					children = children.sort_by(&:weight)
 
 					# This must always be at least >= `children.size`, since stream weight can't be 0.
 					total = children.sum(&:weight)