Optimize connection level remote flow control

Previously when connection level remote flow control window gets 0, we
mark the stream having DATA frame with
NGHTTP2_STREAM_FLAG_DEFERRED_FLOW_CONTROL.  When connection level
WINDOW_UPDATE is received, we checks all existing streams, including
closed ones, and call nghttp2_stream_resume_deferred_data().  The
profiler shows this is expensive.

Now we prepare dedicated priority queue for DATA frames.  And we don't
mark stream with NGHTTP2_STREAM_FLAG_DEFERRED_FLOW_CONTROL when DATA
cannot be sent solely due to connection level flow control.  Instead,
we just queue DATA item to queue.  We won't pop DATA item from queue
when connection level remote window size is 0.  This way, we avoid the
expensive operation for all streams when WINDOW_UPDATE is arrived.

lib/nghttp2_session.c

@@ -332,6 +332,10 @@ static int session_new(nghttp2_session **session_ptr,
   if(rv != 0) {
     goto fail_ob_ss_pq;
   }
+  rv = nghttp2_pq_init(&(*session_ptr)->ob_da_pq, outbound_item_compar);
+  if(rv != 0) {
+    goto fail_ob_da_pq;
+  }
 
   rv = nghttp2_hd_deflate_init(&(*session_ptr)->hd_deflater);
   if(rv != 0) {
@@ -429,6 +433,8 @@ static int session_new(nghttp2_session **session_ptr,
  fail_hd_inflater:
   nghttp2_hd_deflate_free(&(*session_ptr)->hd_deflater);
  fail_hd_deflater:
+  nghttp2_pq_free(&(*session_ptr)->ob_da_pq);
+ fail_ob_da_pq:
   nghttp2_pq_free(&(*session_ptr)->ob_ss_pq);
  fail_ob_ss_pq:
   nghttp2_pq_free(&(*session_ptr)->ob_pq);
@@ -542,6 +548,7 @@ void nghttp2_session_del(nghttp2_session *session)
 
   ob_pq_free(&session->ob_pq);
   ob_pq_free(&session->ob_ss_pq);
+  ob_pq_free(&session->ob_da_pq);
   active_outbound_item_reset(&session->aob);
   session_inbound_frame_reset(session);
   nghttp2_hd_deflate_free(&session->hd_deflater);
@@ -573,9 +580,9 @@ int nghttp2_session_reprioritize_stream
        session->roots.num_streams <= NGHTTP2_MAX_DEP_TREE_LENGTH) {
 
       rv = nghttp2_stream_dep_all_your_stream_are_belong_to_us
-        (stream, &session->ob_pq, session->last_cycle);
+        (stream, &session->ob_da_pq, session->last_cycle);
     } else {
-      rv = nghttp2_stream_dep_make_root(stream, &session->ob_pq,
+      rv = nghttp2_stream_dep_make_root(stream, &session->ob_da_pq,
                                         session->last_cycle);
     }
 
@@ -596,7 +603,7 @@ int nghttp2_session_reprioritize_stream
                    stream, stream->stream_id));
 
     nghttp2_stream_dep_remove_subtree(dep_stream);
-    nghttp2_stream_dep_make_root(dep_stream, &session->ob_pq,
+    nghttp2_stream_dep_make_root(dep_stream, &session->ob_da_pq,
                                  session->last_cycle);
   }
 
@@ -609,16 +616,16 @@ int nghttp2_session_reprioritize_stream
 
   if(root_stream->num_substreams + stream->num_substreams >
      NGHTTP2_MAX_DEP_TREE_LENGTH) {
-    rv = nghttp2_stream_dep_make_root(stream, &session->ob_pq,
+    rv = nghttp2_stream_dep_make_root(stream, &session->ob_da_pq,
                                       session->last_cycle);
   } else {
     if(pri_spec->exclusive) {
       rv = nghttp2_stream_dep_insert_subtree(dep_stream, stream,
-                                             &session->ob_pq,
+                                             &session->ob_da_pq,
                                              session->last_cycle);
     } else {
       rv = nghttp2_stream_dep_add_subtree(dep_stream, stream,
-                                          &session->ob_pq,
+                                          &session->ob_da_pq,
                                           session->last_cycle);
     }
   }
@@ -715,7 +722,7 @@ int nghttp2_session_add_frame(nghttp2_session *session,
         item->weight = stream->effective_weight;
         item->cycle = session->last_cycle;
 
-        rv = nghttp2_stream_attach_data(stream, item, &session->ob_pq,
+        rv = nghttp2_stream_attach_data(stream, item, &session->ob_da_pq,
                                         session->last_cycle);
       }
     }
@@ -823,7 +830,7 @@ nghttp2_stream* nghttp2_session_open_stream(nghttp2_session *session,
     if(pri_spec->exclusive &&
        session->roots.num_streams <= NGHTTP2_MAX_DEP_TREE_LENGTH) {
       rv = nghttp2_stream_dep_all_your_stream_are_belong_to_us
-        (stream, &session->ob_pq, session->last_cycle);
+        (stream, &session->ob_da_pq, session->last_cycle);
 
       /* Since no dpri is changed in dependency tree, the above
          function call never fail. */
@@ -879,7 +886,7 @@ int nghttp2_session_close_stream(nghttp2_session *session, int32_t stream_id,
 
     item = stream->data_item;
 
-    rv = nghttp2_stream_detach_data(stream, &session->ob_pq,
+    rv = nghttp2_stream_detach_data(stream, &session->ob_da_pq,
                                     session->last_cycle);
 
     if(rv != 0) {
@@ -1775,7 +1782,7 @@ static int session_prep_frame(nghttp2_session *session,
       int rv2;
 
       if(stream) {
-        rv2 = nghttp2_stream_detach_data(stream, &session->ob_pq,
+        rv2 = nghttp2_stream_detach_data(stream, &session->ob_da_pq,
                                          session->last_cycle);
 
         if(nghttp2_is_fatal(rv2)) {
@@ -1790,9 +1797,14 @@ static int session_prep_frame(nghttp2_session *session,
     next_readmax = nghttp2_session_next_data_read(session, stream);
 
     if(next_readmax == 0) {
-      rv = nghttp2_stream_defer_data(stream,
+
-                                     NGHTTP2_STREAM_FLAG_DEFERRED_FLOW_CONTROL,
+      /* This must be true since we only pop DATA frame item from
-                                     &session->ob_pq, session->last_cycle);
+         queue when session->remote_window_size > 0 */
+      assert(session->remote_window_size > 0);
+
+      rv = nghttp2_stream_defer_data
+        (stream, NGHTTP2_STREAM_FLAG_DEFERRED_FLOW_CONTROL,
+         &session->ob_da_pq, session->last_cycle);
 
       if(nghttp2_is_fatal(rv)) {
         return rv;
@@ -1808,7 +1820,7 @@ static int session_prep_frame(nghttp2_session *session,
                                         data_frame);
     if(framerv == NGHTTP2_ERR_DEFERRED) {
       rv = nghttp2_stream_defer_data(stream, NGHTTP2_STREAM_FLAG_DEFERRED_USER,
-                                     &session->ob_pq, session->last_cycle);
+                                     &session->ob_da_pq, session->last_cycle);
 
       if(nghttp2_is_fatal(rv)) {
         return rv;
@@ -1819,7 +1831,7 @@ static int session_prep_frame(nghttp2_session *session,
       return NGHTTP2_ERR_DEFERRED;
     }
     if(framerv == NGHTTP2_ERR_TEMPORAL_CALLBACK_FAILURE) {
-      rv = nghttp2_stream_detach_data(stream, &session->ob_pq,
+      rv = nghttp2_stream_detach_data(stream, &session->ob_da_pq,
                                       session->last_cycle);
 
       if(nghttp2_is_fatal(rv)) {
@@ -1834,7 +1846,7 @@ static int session_prep_frame(nghttp2_session *session,
       return framerv;
     }
     if(framerv < 0) {
-      rv = nghttp2_stream_detach_data(stream, &session->ob_pq,
+      rv = nghttp2_stream_detach_data(stream, &session->ob_da_pq,
                                       session->last_cycle);
 
       if(nghttp2_is_fatal(rv)) {
@@ -1851,6 +1863,7 @@ static int session_prep_frame(nghttp2_session *session,
   }
 }
 
+/* Used only for tests */
 nghttp2_outbound_item* nghttp2_session_get_ob_pq_top
 (nghttp2_session *session)
 {
@@ -1860,50 +1873,85 @@ nghttp2_outbound_item* nghttp2_session_get_ob_pq_top
 nghttp2_outbound_item* nghttp2_session_get_next_ob_item
 (nghttp2_session *session)
 {
+  nghttp2_outbound_item *item, *headers_item;
+
   if(nghttp2_pq_empty(&session->ob_pq)) {
     if(nghttp2_pq_empty(&session->ob_ss_pq)) {
+      if(session->remote_window_size == 0 ||
+         nghttp2_pq_empty(&session->ob_da_pq)) {
         return NULL;
-    } else {
+      }
+
+      return nghttp2_pq_top(&session->ob_da_pq);
+    }
+
     /* Return item only when concurrent connection limit is not
        reached */
     if(session_is_outgoing_concurrent_streams_max(session)) {
+      if(session->remote_window_size == 0 ||
+         nghttp2_pq_empty(&session->ob_da_pq)) {
         return NULL;
-      } else {
+      }
+
+      return nghttp2_pq_top(&session->ob_da_pq);
+    }
+
     return nghttp2_pq_top(&session->ob_ss_pq);
   }
-    }
+
-  } else {
   if(nghttp2_pq_empty(&session->ob_ss_pq)) {
     return nghttp2_pq_top(&session->ob_pq);
-    } else {
+  }
-      nghttp2_outbound_item *item, *headers_item;
+
   item = nghttp2_pq_top(&session->ob_pq);
   headers_item = nghttp2_pq_top(&session->ob_ss_pq);
+
   if(session_is_outgoing_concurrent_streams_max(session) ||
      item->weight > headers_item->weight ||
      (item->weight == headers_item->weight &&
       item->seq < headers_item->seq)) {
     return item;
-      } else {
+  }
+
   return headers_item;
-      }
-    }
-  }
 }
 
 nghttp2_outbound_item* nghttp2_session_pop_next_ob_item
 (nghttp2_session *session)
 {
+  nghttp2_outbound_item *item, *headers_item;
+
   if(nghttp2_pq_empty(&session->ob_pq)) {
     if(nghttp2_pq_empty(&session->ob_ss_pq)) {
+      if(session->remote_window_size == 0 ||
+         nghttp2_pq_empty(&session->ob_da_pq)) {
         return NULL;
-    } else {
+      }
+
+      item = nghttp2_pq_top(&session->ob_da_pq);
+      nghttp2_pq_pop(&session->ob_da_pq);
+
+      item->queued = 0;
+
+      return item;
+    }
+
     /* Pop item only when concurrent connection limit is not
        reached */
     if(session_is_outgoing_concurrent_streams_max(session)) {
+      if(session->remote_window_size == 0 ||
+         nghttp2_pq_empty(&session->ob_da_pq)) {
         return NULL;
-      } else {
+      }
-        nghttp2_outbound_item *item;
+
+      item = nghttp2_pq_top(&session->ob_da_pq);
+      nghttp2_pq_pop(&session->ob_da_pq);
+
+      item->queued = 0;
+
+      return item;
+    }
+
     item = nghttp2_pq_top(&session->ob_ss_pq);
     nghttp2_pq_pop(&session->ob_ss_pq);
 
@@ -1911,20 +1959,19 @@ nghttp2_outbound_item* nghttp2_session_pop_next_ob_item
 
     return item;
   }
-    }
+
-  } else {
   if(nghttp2_pq_empty(&session->ob_ss_pq)) {
-      nghttp2_outbound_item *item;
     item = nghttp2_pq_top(&session->ob_pq);
     nghttp2_pq_pop(&session->ob_pq);
 
     item->queued = 0;
 
     return item;
-    } else {
+  }
-      nghttp2_outbound_item *item, *headers_item;
+
   item = nghttp2_pq_top(&session->ob_pq);
   headers_item = nghttp2_pq_top(&session->ob_ss_pq);
+
   if(session_is_outgoing_concurrent_streams_max(session) ||
      item->weight > headers_item->weight ||
      (item->weight == headers_item->weight &&
@@ -1934,15 +1981,13 @@ nghttp2_outbound_item* nghttp2_session_pop_next_ob_item
     item->queued = 0;
 
     return item;
-      } else {
+  }
+
   nghttp2_pq_pop(&session->ob_ss_pq);
 
   headers_item->queued = 0;
 
   return headers_item;
-      }
-    }
-  }
 }
 
 static int session_call_before_frame_send(nghttp2_session *session,
@@ -2144,7 +2189,7 @@ static int session_after_frame_sent(nghttp2_session *session)
     }
 
     if(stream && data_frame->eof) {
-      rv = nghttp2_stream_detach_data(stream, &session->ob_pq,
+      rv = nghttp2_stream_detach_data(stream, &session->ob_da_pq,
                                       session->last_cycle);
 
       if(nghttp2_is_fatal(rv)) {
@@ -2205,7 +2250,7 @@ static int session_after_frame_sent(nghttp2_session *session)
     if(nghttp2_session_predicate_data_send(session,
                                            data_frame->hd.stream_id) != 0) {
       if(stream) {
-        rv = nghttp2_stream_detach_data(stream, &session->ob_pq,
+        rv = nghttp2_stream_detach_data(stream, &session->ob_da_pq,
                                         session->last_cycle);
 
         if(nghttp2_is_fatal(rv)) {
@@ -2235,14 +2280,30 @@ static int session_after_frame_sent(nghttp2_session *session)
       next_readmax = nghttp2_session_next_data_read(session, stream);
 
       if(next_readmax == 0) {
-        rv = nghttp2_stream_defer_data
+
-          (stream, NGHTTP2_STREAM_FLAG_DEFERRED_FLOW_CONTROL, &session->ob_pq,
+        if(session->remote_window_size == 0 &&
-           session->last_cycle);
+           stream->remote_window_size > 0) {
+
+          /* If DATA cannot be sent solely due to connection level
+             window size, just push item to queue again.  We never pop
+             DATA item while connection level window size is 0. */
+          rv = nghttp2_pq_push(&session->ob_da_pq, aob->item);
 
           if(nghttp2_is_fatal(rv)) {
             return rv;
           }
 
+          aob->item->queued = 1;
+        } else {
+          rv = nghttp2_stream_defer_data
+            (stream, NGHTTP2_STREAM_FLAG_DEFERRED_FLOW_CONTROL,
+             &session->ob_da_pq, session->last_cycle);
+
+          if(nghttp2_is_fatal(rv)) {
+            return rv;
+          }
+        }
+
         aob->item = NULL;
         active_outbound_item_reset(aob);
 
@@ -2257,9 +2318,9 @@ static int session_after_frame_sent(nghttp2_session *session)
         return rv;
       }
       if(rv == NGHTTP2_ERR_DEFERRED) {
-        rv = nghttp2_stream_defer_data(stream,
+        rv = nghttp2_stream_defer_data
-                                       NGHTTP2_STREAM_FLAG_DEFERRED_USER,
+          (stream, NGHTTP2_STREAM_FLAG_DEFERRED_USER,
-                                       &session->ob_pq, session->last_cycle);
+           &session->ob_da_pq, session->last_cycle);
 
         if(nghttp2_is_fatal(rv)) {
           return rv;
@@ -2282,7 +2343,7 @@ static int session_after_frame_sent(nghttp2_session *session)
           return rv;
         }
 
-        rv = nghttp2_stream_detach_data(stream, &session->ob_pq,
+        rv = nghttp2_stream_detach_data(stream, &session->ob_da_pq,
                                         session->last_cycle);
 
         if(nghttp2_is_fatal(rv)) {
@@ -2299,7 +2360,7 @@ static int session_after_frame_sent(nghttp2_session *session)
     }
 
     if(stream->dpri == NGHTTP2_STREAM_DPRI_TOP) {
-      rv = nghttp2_pq_push(&session->ob_pq, aob->item);
+      rv = nghttp2_pq_push(&session->ob_da_pq, aob->item);
 
       if(nghttp2_is_fatal(rv)) {
         return rv;
@@ -3218,7 +3279,7 @@ static int update_remote_initial_window_size_func
      stream->remote_window_size > 0 &&
      arg->session->remote_window_size > 0) {
 
-    rv = nghttp2_stream_resume_deferred_data(stream, &arg->session->ob_pq,
+    rv = nghttp2_stream_resume_deferred_data(stream, &arg->session->ob_da_pq,
                                              arg->session->last_cycle);
 
     if(nghttp2_is_fatal(rv)) {
@@ -3793,44 +3854,9 @@ static int session_process_altsvc_frame(nghttp2_session *session)
   return nghttp2_session_on_altsvc_received(session, frame);
 }
 
-static int push_back_deferred_data_func(nghttp2_map_entry *entry, void *ptr)
-{
-  int rv;
-  nghttp2_session *session;
-  nghttp2_stream *stream;
-
-  session = (nghttp2_session*)ptr;
-  stream = (nghttp2_stream*)entry;
-
-  /* If DATA frame is deferred due to flow control, push it back to
-     outbound queue. */
-  if(nghttp2_stream_check_deferred_by_flow_control(stream) &&
-     stream->remote_window_size > 0) {
-
-    rv = nghttp2_stream_resume_deferred_data(stream, &session->ob_pq,
-                                             session->last_cycle);
-
-    if(nghttp2_is_fatal(rv)) {
-      return rv;
-    }
-  }
-  return 0;
-}
-
-/*
- * Push back deferred DATA frames to queue if they are deferred due to
- * connection-level flow control.
- */
-static int session_push_back_deferred_data(nghttp2_session *session)
-{
-  return nghttp2_map_each(&session->streams,
-                          push_back_deferred_data_func, session);
-}
-
 static int session_on_connection_window_update_received
 (nghttp2_session *session, nghttp2_frame *frame)
 {
-  int rv;
   /* Handle connection-level flow control */
   if(frame->window_update.window_size_increment == 0 ||
      NGHTTP2_MAX_WINDOW_SIZE - frame->window_update.window_size_increment <
@@ -3839,17 +3865,7 @@ static int session_on_connection_window_update_received
       (session, frame, NGHTTP2_FLOW_CONTROL_ERROR, NULL);
   }
   session->remote_window_size += frame->window_update.window_size_increment;
-  /* To queue the DATA deferred by connection-level flow-control, we
-     have to check all streams. Bad. */
-  if(session->remote_window_size > 0) {
 
-    rv = session_push_back_deferred_data(session);
-    if(rv != 0) {
-      /* FATAL */
-      assert(rv < NGHTTP2_ERR_FATAL);
-      return rv;
-    }
-  }
   return session_call_on_frame_received(session, frame);
 }
 
@@ -3881,10 +3897,9 @@ static int session_on_stream_window_update_received
   stream->remote_window_size += frame->window_update.window_size_increment;
 
   if(stream->remote_window_size > 0 &&
-     session->remote_window_size > 0 &&
      nghttp2_stream_check_deferred_by_flow_control(stream)) {
 
-    rv = nghttp2_stream_resume_deferred_data(stream, &session->ob_pq,
+    rv = nghttp2_stream_resume_deferred_data(stream, &session->ob_da_pq,
                                              session->last_cycle);
 
     if(nghttp2_is_fatal(rv)) {
@@ -5505,6 +5520,8 @@ int nghttp2_session_want_write(nghttp2_session *session)
 
   if(session->aob.item == NULL &&
      nghttp2_pq_empty(&session->ob_pq) &&
+     (nghttp2_pq_empty(&session->ob_da_pq) ||
+      session->remote_window_size == 0) &&
      (nghttp2_pq_empty(&session->ob_ss_pq) ||
       session_is_outgoing_concurrent_streams_max(session))) {
     return 0;
@@ -5852,7 +5869,7 @@ int nghttp2_session_resume_data(nghttp2_session *session, int32_t stream_id)
     return NGHTTP2_ERR_INVALID_ARGUMENT;
   }
 
-  rv = nghttp2_stream_resume_deferred_data(stream, &session->ob_pq,
+  rv = nghttp2_stream_resume_deferred_data(stream, &session->ob_da_pq,
                                            session->last_cycle);
 
   if(nghttp2_is_fatal(rv)) {
@@ -5864,7 +5881,8 @@ int nghttp2_session_resume_data(nghttp2_session *session, int32_t stream_id)
 
 size_t nghttp2_session_get_outbound_queue_size(nghttp2_session *session)
 {
-  return nghttp2_pq_size(&session->ob_pq)+nghttp2_pq_size(&session->ob_ss_pq);
+  return nghttp2_pq_size(&session->ob_pq) +
+    nghttp2_pq_size(&session->ob_ss_pq) + nghttp2_pq_size(&session->ob_da_pq);
 }
 
 int32_t nghttp2_session_get_stream_effective_recv_data_length

lib/nghttp2_session.h

@@ -136,10 +136,13 @@ typedef enum {
 struct nghttp2_session {
   nghttp2_map /* <nghttp2_stream*> */ streams;
   nghttp2_stream_roots roots;
-  /* Queue for outbound frames other than stream-creating HEADERS */
+  /* Queue for outbound frames other than stream-creating HEADERS and
+     DATA */
   nghttp2_pq /* <nghttp2_outbound_item*> */ ob_pq;
   /* Queue for outbound stream-creating HEADERS frame */
   nghttp2_pq /* <nghttp2_outbound_item*> */ ob_ss_pq;
+  /* QUeue for DATA frame */
+  nghttp2_pq /* <nghttp2_outbound_item*> */ ob_da_pq;
   nghttp2_active_outbound_item aob;
   nghttp2_inbound_frame iframe;
   nghttp2_hd_deflater hd_deflater;

tests/nghttp2_session_test.c

@@ -2563,7 +2563,7 @@ void test_nghttp2_session_on_window_update_received(void)
   data_item->frame_cat = NGHTTP2_CAT_DATA;
 
   CU_ASSERT(0 == nghttp2_stream_attach_data(stream, data_item,
-                                            &session->ob_pq,
+                                            &session->ob_da_pq,
                                             session->last_cycle));
 
   nghttp2_frame_window_update_init(&frame.window_update, NGHTTP2_FLAG_NONE,
@@ -2575,7 +2575,7 @@ void test_nghttp2_session_on_window_update_received(void)
 
   CU_ASSERT(0 == nghttp2_stream_defer_data
             (stream, NGHTTP2_STREAM_FLAG_DEFERRED_FLOW_CONTROL,
-             &session->ob_pq, session->last_cycle));
+             &session->ob_da_pq, session->last_cycle));
 
   CU_ASSERT(0 == nghttp2_session_on_window_update_received(session, &frame));
   CU_ASSERT(2 == user_data.frame_recv_cb_called);
@@ -4538,7 +4538,7 @@ void test_nghttp2_session_defer_data(void)
 
   /* Resume deferred DATA */
   CU_ASSERT(0 == nghttp2_session_resume_data(session, 1));
-  item = nghttp2_session_get_ob_pq_top(session);
+  item = (nghttp2_outbound_item*)nghttp2_pq_top(&session->ob_da_pq);
   OB_DATA(item)->data_prd.read_callback =
     fixed_length_data_source_read_callback;
   ud.block_count = 1;
@@ -4556,7 +4556,7 @@ void test_nghttp2_session_defer_data(void)
 
   /* Resume deferred DATA */
   CU_ASSERT(0 == nghttp2_session_resume_data(session, 1));
-  item = nghttp2_session_get_ob_pq_top(session);
+  item = (nghttp2_outbound_item*)nghttp2_pq_top(&session->ob_da_pq);
   OB_DATA(item)->data_prd.read_callback =
     fixed_length_data_source_read_callback;
   ud.block_count = 1;
@@ -5579,7 +5579,8 @@ void test_nghttp2_session_stream_dep_add_subtree(void)
    *    d
    */
 
-  nghttp2_stream_dep_add_subtree(a, e, &session->ob_pq, session->last_cycle);
+  nghttp2_stream_dep_add_subtree(a, e, &session->ob_da_pq,
+                                 session->last_cycle);
 
   /* becomes
    * a
@@ -5630,7 +5631,8 @@ void test_nghttp2_session_stream_dep_add_subtree(void)
    *    d
    */
 
-  nghttp2_stream_dep_insert_subtree(a, e, &session->ob_pq, session->last_cycle);
+  nghttp2_stream_dep_insert_subtree(a, e, &session->ob_da_pq,
+                                    session->last_cycle);
 
   /* becomes
    * a
@@ -5823,7 +5825,7 @@ void test_nghttp2_session_stream_dep_all_your_stream_are_belong_to_us(void)
 
   nghttp2_stream_dep_remove_subtree(c);
   CU_ASSERT(0 == nghttp2_stream_dep_all_your_stream_are_belong_to_us
-            (c, &session->ob_pq, session->last_cycle));
+            (c, &session->ob_da_pq, session->last_cycle));
 
   /*
    * c
@@ -5861,7 +5863,7 @@ void test_nghttp2_session_stream_dep_all_your_stream_are_belong_to_us(void)
 
   nghttp2_stream_dep_remove_subtree(c);
   CU_ASSERT(0 == nghttp2_stream_dep_all_your_stream_are_belong_to_us
-            (c, &session->ob_pq, session->last_cycle));
+            (c, &session->ob_da_pq, session->last_cycle));
 
   /*
    * c
@@ -5898,7 +5900,7 @@ void test_nghttp2_session_stream_dep_all_your_stream_are_belong_to_us(void)
 
   nghttp2_stream_dep_remove_subtree(c);
   CU_ASSERT(0 == nghttp2_stream_dep_all_your_stream_are_belong_to_us
-            (c, &session->ob_pq, session->last_cycle));
+            (c, &session->ob_da_pq, session->last_cycle));
 
   /*
    * c
@@ -5951,7 +5953,7 @@ void test_nghttp2_session_stream_attach_data(void)
 
   db = create_data_ob_item();
 
-  nghttp2_stream_attach_data(b, db, &session->ob_pq, session->last_cycle);
+  nghttp2_stream_attach_data(b, db, &session->ob_da_pq, session->last_cycle);
 
   CU_ASSERT(NGHTTP2_STREAM_DPRI_NO_DATA == a->dpri);
   CU_ASSERT(NGHTTP2_STREAM_DPRI_TOP == b->dpri);
@@ -5966,7 +5968,7 @@ void test_nghttp2_session_stream_attach_data(void)
 
   dc = create_data_ob_item();
 
-  nghttp2_stream_attach_data(c, dc, &session->ob_pq, session->last_cycle);
+  nghttp2_stream_attach_data(c, dc, &session->ob_da_pq, session->last_cycle);
 
   CU_ASSERT(NGHTTP2_STREAM_DPRI_NO_DATA == a->dpri);
   CU_ASSERT(NGHTTP2_STREAM_DPRI_TOP == b->dpri);
@@ -5982,7 +5984,7 @@ void test_nghttp2_session_stream_attach_data(void)
 
   da = create_data_ob_item();
 
-  nghttp2_stream_attach_data(a, da, &session->ob_pq, session->last_cycle);
+  nghttp2_stream_attach_data(a, da, &session->ob_da_pq, session->last_cycle);
 
   CU_ASSERT(NGHTTP2_STREAM_DPRI_TOP == a->dpri);
   CU_ASSERT(NGHTTP2_STREAM_DPRI_REST == b->dpri);
@@ -5993,7 +5995,7 @@ void test_nghttp2_session_stream_attach_data(void)
 
   CU_ASSERT(1 == da->queued);
 
-  nghttp2_stream_detach_data(a, &session->ob_pq, session->last_cycle);
+  nghttp2_stream_detach_data(a, &session->ob_da_pq, session->last_cycle);
 
   CU_ASSERT(NGHTTP2_STREAM_DPRI_NO_DATA == a->dpri);
   CU_ASSERT(NGHTTP2_STREAM_DPRI_TOP == b->dpri);
@@ -6005,7 +6007,7 @@ void test_nghttp2_session_stream_attach_data(void)
 
   dd = create_data_ob_item();
 
-  nghttp2_stream_attach_data(d, dd, &session->ob_pq, session->last_cycle);
+  nghttp2_stream_attach_data(d, dd, &session->ob_da_pq, session->last_cycle);
 
   CU_ASSERT(NGHTTP2_STREAM_DPRI_NO_DATA == a->dpri);
   CU_ASSERT(NGHTTP2_STREAM_DPRI_TOP == b->dpri);
@@ -6017,7 +6019,7 @@ void test_nghttp2_session_stream_attach_data(void)
 
   CU_ASSERT(0 == dd->queued);
 
-  nghttp2_stream_detach_data(c, &session->ob_pq, session->last_cycle);
+  nghttp2_stream_detach_data(c, &session->ob_da_pq, session->last_cycle);
 
   CU_ASSERT(NGHTTP2_STREAM_DPRI_NO_DATA == a->dpri);
   CU_ASSERT(NGHTTP2_STREAM_DPRI_TOP == b->dpri);
@@ -6060,11 +6062,11 @@ void test_nghttp2_session_stream_attach_data_subtree(void)
 
   de = create_data_ob_item();
 
-  nghttp2_stream_attach_data(e, de, &session->ob_pq, session->last_cycle);
+  nghttp2_stream_attach_data(e, de, &session->ob_da_pq, session->last_cycle);
 
   db = create_data_ob_item();
 
-  nghttp2_stream_attach_data(b, db, &session->ob_pq, session->last_cycle);
+  nghttp2_stream_attach_data(b, db, &session->ob_da_pq, session->last_cycle);
 
   CU_ASSERT(NGHTTP2_STREAM_DPRI_NO_DATA == a->dpri);
   CU_ASSERT(NGHTTP2_STREAM_DPRI_TOP == b->dpri);
@@ -6079,7 +6081,8 @@ void test_nghttp2_session_stream_attach_data_subtree(void)
   /* Insert subtree e under a */
 
   nghttp2_stream_dep_remove_subtree(e);
-  nghttp2_stream_dep_insert_subtree(a, e, &session->ob_pq, session->last_cycle);
+  nghttp2_stream_dep_insert_subtree(a, e, &session->ob_da_pq,
+                                    session->last_cycle);
 
   /*
    * a
@@ -6104,7 +6107,7 @@ void test_nghttp2_session_stream_attach_data_subtree(void)
 
   nghttp2_stream_dep_remove_subtree(b);
 
-  nghttp2_stream_dep_make_root(b, &session->ob_pq, session->last_cycle);
+  nghttp2_stream_dep_make_root(b, &session->ob_da_pq, session->last_cycle);
 
   /*
    * a       b
@@ -6130,7 +6133,7 @@ void test_nghttp2_session_stream_attach_data_subtree(void)
 
   nghttp2_stream_dep_remove_subtree(a);
 
-  nghttp2_stream_dep_make_root(a, &session->ob_pq, session->last_cycle);
+  nghttp2_stream_dep_make_root(a, &session->ob_da_pq, session->last_cycle);
 
   CU_ASSERT(NGHTTP2_STREAM_DPRI_NO_DATA == a->dpri);
   CU_ASSERT(NGHTTP2_STREAM_DPRI_TOP == b->dpri);
@@ -6143,7 +6146,7 @@ void test_nghttp2_session_stream_attach_data_subtree(void)
 
   nghttp2_stream_dep_remove_subtree(c);
 
-  nghttp2_stream_dep_make_root(c, &session->ob_pq, session->last_cycle);
+  nghttp2_stream_dep_make_root(c, &session->ob_da_pq, session->last_cycle);
 
   /*
    * a       b     c
@@ -6162,12 +6165,13 @@ void test_nghttp2_session_stream_attach_data_subtree(void)
 
   dd = create_data_ob_item();
 
-  nghttp2_stream_attach_data(d, dd, &session->ob_pq, session->last_cycle);
+  nghttp2_stream_attach_data(d, dd, &session->ob_da_pq, session->last_cycle);
 
   /* Add subtree c to a */
 
   nghttp2_stream_dep_remove_subtree(c);
-  nghttp2_stream_dep_add_subtree(a, c, &session->ob_pq, session->last_cycle);
+  nghttp2_stream_dep_add_subtree(a, c, &session->ob_da_pq,
+                                 session->last_cycle);
 
   /*
    * a       b
@@ -6194,7 +6198,8 @@ void test_nghttp2_session_stream_attach_data_subtree(void)
   /* Insert b under a */
 
   nghttp2_stream_dep_remove_subtree(b);
-  nghttp2_stream_dep_insert_subtree(a, b, &session->ob_pq, session->last_cycle);
+  nghttp2_stream_dep_insert_subtree(a, b, &session->ob_da_pq,
+                                    session->last_cycle);
 
   /*
    * a
@@ -6221,7 +6226,7 @@ void test_nghttp2_session_stream_attach_data_subtree(void)
   /* Remove subtree b */
 
   nghttp2_stream_dep_remove_subtree(b);
-  nghttp2_stream_dep_make_root(b, &session->ob_pq, session->last_cycle);
+  nghttp2_stream_dep_make_root(b, &session->ob_da_pq, session->last_cycle);
 
   /*
    * b       a