reorder decoded frames according to stream priorization rules

Author: HoneyryderChuck

lib/http/2/connection.rb

@@ -93,7 +93,7 @@ def initialize(settings = {})
       @streams_recently_closed = {}
       @pending_settings = []
 
-      @framer = Framer.new(@local_settings[:settings_max_frame_size])
+      @framer = Framer.new(@streams, @local_settings[:settings_max_frame_size])
 
       @local_window_limit = @local_settings[:settings_initial_window_size]
       @local_window = @local_window_limit

lib/http/2/framer.rb

@@ -115,8 +115,12 @@ class Framer
 
     # Initializes new framer object.
     #
-    def initialize(local_max_frame_size = DEFAULT_MAX_FRAME_SIZE,
-                   remote_max_frame_size = DEFAULT_MAX_FRAME_SIZE)
+    def initialize(
+      streams = {},
+      local_max_frame_size = DEFAULT_MAX_FRAME_SIZE,
+      remote_max_frame_size = DEFAULT_MAX_FRAME_SIZE
+    )
+      @streams = streams
       @local_max_frame_size = local_max_frame_size
       @remote_max_frame_size = remote_max_frame_size
       @frames = []
@@ -372,13 +376,30 @@ def generate(frame)
     #
     # @param buf [Buffer]
     def parse(buf)
+      decoded = false
+
       while (frame = decode_frame(buf))
         if frame[:type] == :ping
           # PING responses SHOULD be given higher priority than any other frame.
           @frames.unshift(frame)
         else
           @frames << frame
         end
+        decoded = true
+      end
+
+      # TODO: support stream prioritization
+      # WIP
+      if decoded
+        @frames.sort! do |f1, f2|
+          next(0) unless f1.key?(:stream) && f2.key?(:stream)
+
+          s1 = @streams[f1[:stream]] or next(0)
+
+          s2 = @streams[f2[:stream]] or next(0)
+
+          s1 <=> s2
+        end
       end
 
       @frames.shift

lib/http/2/stream.rb

@@ -53,7 +53,14 @@ class Stream
 
     # Stream priority as set by initiator.
     attr_reader :weight
-    attr_reader :dependency, :remote_window
+
+    # whether the stream is exclusive in the dependency tree
+    attr_reader :exclusive
+
+    # the parent stream
+    attr_reader :dependency
+
+    attr_reader :remote_window
 
     # Size of current stream flow control window.
     attr_reader :local_window
@@ -82,6 +89,7 @@ def initialize(connection:, id:, weight: 16, dependency: 0, exclusive: false, pa
       @id = id
       @weight = weight
       @dependency = dependency
+      @exclusive = exclusive
 
       # from mixins
       @listeners = Hash.new { |hash, key| hash[key] = [] }
@@ -104,6 +112,28 @@ def initialize(connection:, id:, weight: 16, dependency: 0, exclusive: false, pa
       on(:local_window) { |v| @local_window_max_size = @local_window = v }
     end
 
+    def <=>(other)
+      if !@dependency.zero?
+        if @dependency == other.id
+          # parent stream processed before
+          return 1
+        elsif @dependency == other.dependency
+          if @exclusive
+            # exclusive streams from the same dep come first
+            return -1
+          elsif other.exclusive
+            return 1
+          else
+            return other.weight <=> @weight
+          end
+        end
+      elsif !other.dependency.zero?
+        return -1 if @id == other.dependency
+      end
+
+      other.weight <=> @weight
+    end
+
     def closed?
       @state == :closed
     end
@@ -647,6 +677,7 @@ def complete_transition(frame)
     def process_priority(frame)
       @weight = frame[:weight]
       @dependency = frame[:dependency]
+      @exclusive = frame[:exclusive]
       emit(
         :priority,
         weight: frame[:weight],

sig/framer.rbs

@@ -35,6 +35,7 @@ module HTTP2
     @local_max_frame_size: Integer
     @remote_max_frame_size: Integer
     @streams: Hash[Integer, Stream]
+    @frames: Array[frame]
 
     attr_accessor local_max_frame_size: Integer
 
@@ -52,7 +53,7 @@ module HTTP2
 
     private
 
-    def initialize: (?Integer local_max_frame_size, ?Integer remote_max_frame_size) -> untyped
+    def initialize: (?Hash[Integer, Stream] streams, ?Integer local_max_frame_size, ?Integer remote_max_frame_size) -> untyped
 
     def decode_frame: (String buf) -> frame?
 

sig/stream.rbs

@@ -10,6 +10,7 @@ module HTTP2
     attr_reader parent: Stream?
     attr_reader weight: Integer
     attr_reader dependency: Integer
+    attr_reader exclusive: bool
     attr_reader remote_window: Integer
     attr_reader local_window: Integer
     attr_reader closed: Symbol?
@@ -33,6 +34,8 @@ module HTTP2
 
     alias << receive
 
+    def <=>: (Stream other) -> Integer
+
     def verify_trailers: (headers_frame frame) -> void
 
     def calculate_content_length: (Integer?) -> void

spec/stream_spec.rb

@@ -662,6 +662,57 @@
     end
   end
 
+  context "prioritization" do
+    it "should order based on dependency" do
+      # 5.3.1
+      # if streams B and C are dependent on stream A, and if stream
+      # D is created with a dependency on stream A, this results in a
+      # dependency order of A followed by B, C, and D in any order.
+      a_stream = client.new_stream
+      b_stream = client.new_stream(dependency: a_stream.id)
+      c_stream = client.new_stream(dependency: a_stream.id)
+      d_stream = client.new_stream(dependency: a_stream.id)
+
+      expect([b_stream, c_stream, d_stream, a_stream].sort).to eq(
+        [a_stream, b_stream, c_stream, d_stream]
+      )
+      expect([b_stream, d_stream, c_stream, a_stream].sort).to eq(
+        [a_stream, b_stream, d_stream, c_stream]
+      )
+    end
+
+    it "should push exclusive streams up the stack" do
+      # he exclusive flag causes the stream to become the
+      # sole dependency of its parent stream, causing other dependencies to
+      # become dependent on the exclusive stream.  In the previous example,
+      # if stream D is created with an exclusive dependency on stream A, this
+      # results in D becoming the dependency parent of B and C.
+      a_stream = client.new_stream
+      b_stream = client.new_stream(dependency: a_stream.id)
+      c_stream = client.new_stream(dependency: a_stream.id)
+      d_stream = client.new_stream(dependency: a_stream.id, exclusive: true)
+
+      expect([b_stream, c_stream, d_stream, a_stream].sort).to eq(
+        [a_stream, d_stream, b_stream, c_stream]
+      )
+    end
+
+    it "should prioritze based on weight" do
+      # Streams with the same parent SHOULD be allocated resources
+      # proportionally based on their weight.  Thus, if stream B depends on
+      # stream A with weight 4, stream C depends on stream A with weight 12,
+      # and no progress can be made on stream A, stream B ideally receives
+      # one-third of the resources allocated to stream C.
+      a_stream = client.new_stream
+      b_stream = client.new_stream(dependency: a_stream.id, weight: 4)
+      c_stream = client.new_stream(dependency: a_stream.id, weight: 12)
+
+      expect([b_stream, c_stream, a_stream].sort).to eq(
+        [a_stream, c_stream, b_stream]
+      )
+    end
+  end
+
   context "client API" do
     it ".reprioritize should emit PRIORITY frame" do
       expect(stream).to receive(:send) do |frame|