End flow control by WINDOW_UPDATE

lib/includes/nghttp2/nghttp2.h

@@ -1543,13 +1543,17 @@ int nghttp2_submit_goaway(nghttp2_session *session,
  * Submits WINDOW_UPDATE frame. The effective range of the
  * |window_size_increment| is [1, (1 << 31)-1], inclusive. But the
  * application must be responsible to keep the resulting window size
- * <= (1 << 31)-1.
+ * <= (1 << 31)-1. If NGHTTP2_FLAG_END_FLOW_CONTROL bit set in the
+ * |flags|, 0 can be specified in the |window_size_increment|. In
+ * fact, if this flag is set, the value specified in the
+ * |window_size_increment| is ignored.
  *
  * This function returns 0 if it succeeds, or one of the following
  * negative error codes:
  *
  * :enum:`NGHTTP2_ERR_INVALID_ARGUMENT`
- *     The |delta_window_size| is 0 or negative.
+ *     The |delta_window_size| is 0 or negative if
+ *     NGHTTP2_FLAG_END_FLOW_CONTROL bit is not set in |flags|.
  * :enum:`NGHTTP2_ERR_STREAM_CLOSED`
  *     The stream is already closed or does not exist.
  * :enum:`NGHTTP2_ERR_NOMEM`

lib/nghttp2_session.c

@@ -1124,6 +1124,17 @@ static int nghttp2_session_after_frame_sent(nghttp2_session *session)
       session->goaway_flags |= NGHTTP2_GOAWAY_SEND;
       break;
     case NGHTTP2_WINDOW_UPDATE:
+      if(frame->hd.flags & NGHTTP2_FLAG_END_FLOW_CONTROL) {
+        if(frame->hd.stream_id == 0) {
+          session->local_flow_control = 0;
+        } else {
+          nghttp2_stream *stream;
+          stream = nghttp2_session_get_stream(session, frame->hd.stream_id);
+          if(stream) {
+            stream->local_flow_control = 0;
+          }
+        }
+      }
       break;
     }
     nghttp2_active_outbound_item_reset(&session->aob);
@@ -1795,6 +1806,17 @@ int nghttp2_session_on_window_update_received(nghttp2_session *session,
          receiving WINDOW_UPDATE are asynchronous, so it is hard to
          determine that the peer is misbehaving or not without
          measuring RTT. For now, we just ignore such frames. */
+      nghttp2_session_call_on_frame_received(session, frame);
+      return 0;
+    }
+    if(frame->hd.flags & NGHTTP2_FLAG_END_FLOW_CONTROL) {
+      if(session->remote_flow_control) {
+        /* Disable connection-level flow control and push back
+           deferred DATA frame if any */
+        session->remote_flow_control = 0;
+        nghttp2_session_call_on_frame_received(session, frame);
+        return nghttp2_session_push_back_deferred_data(session);
+      }
       return 0;
     }
     if(INT32_MAX - frame->window_update.window_size_increment <
@@ -1817,6 +1839,25 @@ int nghttp2_session_on_window_update_received(nghttp2_session *session,
     if(stream) {
       if(stream->remote_flow_control == 0) {
         /* Same reason with connection-level flow control */
+        nghttp2_session_call_on_frame_received(session, frame);
+        return 0;
+      }
+      if(frame->hd.flags & NGHTTP2_FLAG_END_FLOW_CONTROL) {
+        stream->remote_flow_control = 0;
+        if(stream->remote_flow_control &&
+           stream->deferred_data != NULL &&
+           (stream->deferred_flags & NGHTTP2_DEFERRED_FLOW_CONTROL)) {
+          int r;
+          r = nghttp2_pq_push(&session->ob_pq, stream->deferred_data);
+          if(r == 0) {
+            nghttp2_stream_detach_deferred_data(stream);
+          } else if(r < 0) {
+            /* FATAL */
+            assert(r < NGHTTP2_ERR_FATAL);
+            return r;
+          }
+        }
+        nghttp2_session_call_on_frame_received(session, frame);
         return 0;
       }
       if(INT32_MAX - frame->window_update.window_size_increment <

lib/nghttp2_submit.c

@@ -168,7 +168,10 @@ int nghttp2_submit_window_update(nghttp2_session *session, uint8_t flags,
                                  int32_t window_size_increment)
 {
   nghttp2_stream *stream;
-  if(window_size_increment <= 0) {
+  flags &= NGHTTP2_FLAG_END_FLOW_CONTROL;
+  if(flags & NGHTTP2_FLAG_END_FLOW_CONTROL) {
+    window_size_increment = 0;
+  } else if(window_size_increment <= 0) {
     return NGHTTP2_ERR_INVALID_ARGUMENT;
   }
   if(stream_id == 0) {

tests/main.c

@@ -145,6 +145,8 @@ int main(int argc, char* argv[])
                    test_nghttp2_session_defer_data) ||
       !CU_add_test(pSuite, "session_flow_control",
                    test_nghttp2_session_flow_control) ||
+      !CU_add_test(pSuite, "session_flow_control_disable",
+                   test_nghttp2_session_flow_control_disable) ||
       !CU_add_test(pSuite, "session_data_read_temporal_failure",
                    test_nghttp2_session_data_read_temporal_failure) ||
       !CU_add_test(pSuite, "session_on_request_recv_callback",

tests/nghttp2_session_test.c

@@ -1036,6 +1036,26 @@ void test_nghttp2_session_on_window_update_received(void)
   CU_ASSERT(NULL == stream->deferred_data);
 
   nghttp2_frame_window_update_free(&frame.window_update);
+
+  /* Check END_FLOW_CONTROL flag */
+  user_data.frame_recv_cb_called = 0;
+  nghttp2_frame_window_update_init(&frame.window_update,
+                                   NGHTTP2_FLAG_END_FLOW_CONTROL, 1, 0);
+  CU_ASSERT(0 == nghttp2_session_on_window_update_received(session, &frame));
+  CU_ASSERT(1 == user_data.frame_recv_cb_called);
+  CU_ASSERT(0 == stream->remote_flow_control);
+
+  nghttp2_frame_window_update_free(&frame.window_update);
+
+  user_data.frame_recv_cb_called = 0;
+  nghttp2_frame_window_update_init(&frame.window_update,
+                                   NGHTTP2_FLAG_END_FLOW_CONTROL, 0, 0);
+  CU_ASSERT(0 == nghttp2_session_on_window_update_received(session, &frame));
+  CU_ASSERT(1 == user_data.frame_recv_cb_called);
+  CU_ASSERT(0 == session->remote_flow_control);
+
+  nghttp2_frame_window_update_free(&frame.window_update);
+
   nghttp2_session_del(session);
 }
 
@@ -1521,6 +1541,21 @@ void test_nghttp2_submit_window_update(void)
   CU_ASSERT(0 == nghttp2_session_send(session));
   CU_ASSERT(0 == stream->recv_window_size);
 
+  /* Disable stream-level flow control */
+  CU_ASSERT(0 == nghttp2_submit_window_update(session,
+                                              NGHTTP2_FLAG_END_FLOW_CONTROL,
+                                              2, 0));
+  CU_ASSERT(0 == nghttp2_session_send(session));
+  CU_ASSERT(0 == stream->local_flow_control);
+
+  /* Disable connection-level flow control */
+  CU_ASSERT(0 == nghttp2_submit_window_update(session,
+                                              NGHTTP2_FLAG_END_FLOW_CONTROL,
+                                              0, 0));
+  CU_ASSERT(0 == nghttp2_session_send(session));
+  CU_ASSERT(0 == session->local_flow_control);
+
+
   CU_ASSERT(NGHTTP2_ERR_INVALID_ARGUMENT ==
             nghttp2_submit_window_update(session, NGHTTP2_FLAG_NONE, 2, 0));
   CU_ASSERT(NGHTTP2_ERR_INVALID_ARGUMENT ==
@@ -1973,6 +2008,58 @@ void test_nghttp2_session_flow_control(void)
   nghttp2_session_del(session);
 }
 
+void test_nghttp2_session_flow_control_disable(void)
+{
+  nghttp2_session *session;
+  nghttp2_session_callbacks callbacks;
+  const char *nv[] = { NULL };
+  my_user_data ud;
+  nghttp2_data_provider data_prd;
+  nghttp2_frame frame;
+
+  memset(&callbacks, 0, sizeof(nghttp2_session_callbacks));
+  callbacks.send_callback = null_send_callback;
+  callbacks.on_frame_send_callback = on_frame_send_callback;
+  data_prd.read_callback = fixed_length_data_source_read_callback;
+
+  ud.frame_send_cb_called = 0;
+  ud.data_source_length = 128*1024;
+
+  /* Initial window size is 64KiB */
+  nghttp2_session_client_new(&session, &callbacks, &ud);
+  nghttp2_submit_request(session, NGHTTP2_PRI_DEFAULT, nv, &data_prd, NULL);
+
+  /* Sends 64KiB data */
+  CU_ASSERT(0 == nghttp2_session_send(session));
+  CU_ASSERT(64*1024 == ud.data_source_length);
+
+  /* Disable stream flow control */
+  nghttp2_frame_window_update_init(&frame.window_update,
+                                   NGHTTP2_FLAG_END_FLOW_CONTROL, 1, 0);
+  nghttp2_session_on_window_update_received(session, &frame);
+
+  /* Check stream-level remote_flow_control is disabled */
+  CU_ASSERT(0 == nghttp2_session_get_stream(session, 1)->remote_flow_control);
+
+  /* Still nothing is sent because of connection-level flow control */
+  CU_ASSERT(0 == nghttp2_session_send(session));
+  CU_ASSERT(64*1024 == ud.data_source_length);
+
+  /* Disable connection-level flow control */
+  frame.hd.stream_id = 0;
+  nghttp2_session_on_window_update_received(session, &frame);
+
+  /* Check connection-level remote_flow_control is disabled */
+  CU_ASSERT(0 == session->remote_flow_control);
+
+  /* Sends remaining data */
+  CU_ASSERT(0 == nghttp2_session_send(session));
+  CU_ASSERT(0 == ud.data_source_length);
+
+  nghttp2_frame_window_update_free(&frame.window_update);
+  nghttp2_session_del(session);
+}
+
 void test_nghttp2_session_data_read_temporal_failure(void)
 {
   nghttp2_session *session;

tests/nghttp2_session_test.h

@@ -62,6 +62,7 @@ void test_nghttp2_session_stream_close_on_headers_push(void);
 void test_nghttp2_session_stop_data_with_rst_stream(void);
 void test_nghttp2_session_defer_data(void);
 void test_nghttp2_session_flow_control(void);
+void test_nghttp2_session_flow_control_disable(void);
 void test_nghttp2_session_data_read_temporal_failure(void);
 void test_nghttp2_session_on_request_recv_callback(void);
 void test_nghttp2_session_on_stream_close(void);