Add new API to return effective recv data/win size for connection

Using this feature, connection level flow control is now enabled
in nghttpx.

lib/includes/nghttp2/nghttp2.h

@@ -1573,15 +1573,49 @@ int32_t nghttp2_session_get_stream_effective_recv_data_length
 /**
  * @function
  *
- * Returns the local (receive) window size. The local window size can
+ * Returns the local (receive) window size for the stream
- * be adjusted by `nghttp2_submit_window_update()`. This function
+ * |stream_id|. The local window size can be adjusted by
- * takes into account that and returns effective window size.
+ * `nghttp2_submit_window_update()`. This function takes into account
+ * that and returns effective window size.
  *
  * This function returns -1 if it fails.
  */
 int32_t nghttp2_session_get_stream_effective_local_window_size
 (nghttp2_session *session, int32_t stream_id);
 
+/**
+ * @function
+ *
+ * Returns the number of DATA payload in bytes received without
+ * WINDOW_UPDATE transmission for a connection. The local (receive)
+ * window size can be adjusted by
+ * `nghttp2_submit_window_update()`. This function takes into account
+ * that and returns effective data length. In particular, if the local
+ * window size is reduced by submitting negative window_size_increment
+ * with `nghttp2_submit_window_update()`, this function returns the
+ * number of bytes less than actually received.
+ *
+ * If flow control is disabled for a connection, this function returns
+ * 0.
+ *
+ * This function returns -1 if it fails.
+ */
+int32_t nghttp2_session_get_effective_recv_data_length
+(nghttp2_session *session);
+
+/**
+ * @function
+ *
+ * Returns the local (receive) window size for a connection. The local
+ * window size can be adjusted by
+ * `nghttp2_submit_window_update()`. This function takes into account
+ * that and returns effective window size.
+ *
+ * This function returns -1 if it fails.
+ */
+int32_t nghttp2_session_get_effective_local_window_size
+(nghttp2_session *session);
+
 /**
  * @function
  *

lib/nghttp2_session.c

@@ -3596,6 +3596,21 @@ int32_t nghttp2_session_get_stream_effective_local_window_size
   return stream->local_window_size;
 }
 
+int32_t nghttp2_session_get_effective_recv_data_length
+(nghttp2_session *session)
+{
+  if(session->local_flow_control == 0) {
+    return 0;
+  }
+  return session->recv_window_size;
+}
+
+int32_t nghttp2_session_get_effective_local_window_size
+(nghttp2_session *session)
+{
+  return session->local_window_size;
+}
+
 int nghttp2_session_set_option(nghttp2_session *session,
                                int optname, void *optval, size_t optlen)
 {

src/http2.cc

@@ -333,6 +333,27 @@ void build_http1_headers_from_norm_headers
   }
 }
 
+int32_t determine_window_update_transmission(nghttp2_session *session,
+                                             int32_t stream_id)
+{
+  int32_t recv_length, window_size;
+  if(stream_id == 0) {
+    recv_length = nghttp2_session_get_effective_recv_data_length(session);
+    window_size = nghttp2_session_get_effective_local_window_size(session);
+  } else {
+    recv_length = nghttp2_session_get_stream_effective_recv_data_length
+      (session, stream_id);
+    window_size = nghttp2_session_get_stream_effective_local_window_size
+      (session, stream_id);
+  }
+  if(recv_length != -1 && window_size != -1) {
+    if(recv_length >= window_size / 2) {
+      return recv_length;
+    }
+  }
+  return -1;
+}
+
 } // namespace http2
 
 } // namespace nghttp2

src/http2.h

@@ -107,6 +107,15 @@ void build_http1_headers_from_norm_headers
 (std::string& hdrs,
  const std::vector<std::pair<std::string, std::string>>& headers);
 
+// Return positive window_size_increment if WINDOW_UPDATE should be
+// sent for the stream |stream_id|. If |stream_id| == 0, this function
+// determines the necessity of the WINDOW_UPDATE for a connection.
+//
+// If the function determines WINDOW_UPDATE is not necessary at the
+// moment, it returns -1.
+int32_t determine_window_update_transmission(nghttp2_session *session,
+                                             int32_t stream_id);
+
 } // namespace http2
 
 } // namespace nghttp2

src/shrpx_http2_upstream.cc

@@ -428,6 +428,10 @@ Http2Upstream::Http2Upstream(ClientHandler *handler)
                                   NGHTTP2_OPT_NO_AUTO_STREAM_WINDOW_UPDATE,
                                   &val, sizeof(val));
   assert(rv == 0);
+  rv = nghttp2_session_set_option(session_,
+                                  NGHTTP2_OPT_NO_AUTO_CONNECTION_WINDOW_UPDATE,
+                                  &val, sizeof(val));
+  assert(rv == 0);
 
   // TODO Maybe call from outside?
   nghttp2_settings_entry entry[2];
@@ -441,11 +445,6 @@ Http2Upstream::Http2Upstream(ClientHandler *handler)
                                entry,
                                sizeof(entry)/sizeof(nghttp2_settings_entry));
   assert(rv == 0);
-  // Set large connection-level window size to effectively disable
-  // connection-level flow control.
-  rv = nghttp2_submit_window_update(session_, NGHTTP2_FLAG_NONE,
-                                    0, 1000000007);
-  assert(rv == 0);
 }
 
 Http2Upstream::~Http2Upstream()
@@ -725,7 +724,8 @@ int Http2Upstream::window_update(Downstream *downstream,
 {
   int rv;
   rv = nghttp2_submit_window_update(session_, NGHTTP2_FLAG_NONE,
-                                    downstream->get_stream_id(),
+                                    downstream ?
+                                    downstream->get_stream_id() : 0,
                                     window_size_increment);
   if(rv < NGHTTP2_ERR_FATAL) {
     ULOG(FATAL, this) << "nghttp2_submit_window_update() failed: "
@@ -959,15 +959,16 @@ void Http2Upstream::pause_read(IOCtrlReason reason)
 int Http2Upstream::resume_read(IOCtrlReason reason, Downstream *downstream)
 {
   if(get_flow_control()) {
-    int32_t recv_length, window_size;
+    int32_t window_size_increment;
-    recv_length = nghttp2_session_get_stream_effective_recv_data_length
+    window_size_increment = http2::determine_window_update_transmission
+      (session_, 0);
+    if(window_size_increment != -1) {
+      window_update(nullptr, window_size_increment);
+    }
+    window_size_increment = http2::determine_window_update_transmission
       (session_, downstream->get_stream_id());
-    window_size = nghttp2_session_get_stream_effective_local_window_size
+    if(window_size_increment != -1) {
-      (session_, downstream->get_stream_id());
+      window_update(downstream, window_size_increment);
-    if(recv_length != -1 && window_size != -1) {
-      if(recv_length >= window_size / 2) {
-        window_update(downstream, recv_length);
-      }
     }
   }
   return send();

src/shrpx_http2_upstream.h

@@ -58,6 +58,8 @@ public:
   nghttp2_session* get_spdy_session();
 
   int rst_stream(Downstream *downstream, nghttp2_error_code error_code);
+  // To send WINDOW_UPDATE for a connection, specify nullptr to
+  // |downstream|.
   int window_update(Downstream *downstream, int32_t window_size_increment);
   int error_reply(Downstream *downstream, unsigned int status_code);
 

src/shrpx_spdy_downstream_connection.cc

@@ -437,24 +437,30 @@ int SpdyDownstreamConnection::end_upload_data()
 
 int SpdyDownstreamConnection::resume_read(IOCtrlReason reason)
 {
-  int rv;
+  int rv1 = 0, rv2 = 0;
   if(spdy_->get_state() == SpdySession::CONNECTED &&
-     spdy_->get_flow_control() &&
+     spdy_->get_flow_control()) {
-     downstream_ && downstream_->get_downstream_stream_id() != -1) {
+    int32_t window_size_increment;
-    int32_t recv_length, window_size;
+    window_size_increment = http2::determine_window_update_transmission
-    recv_length = spdy_->get_stream_effective_recv_data_length
+      (spdy_->get_session(), 0);
-      (downstream_->get_stream_id());
+    if(window_size_increment != -1) {
-    window_size = spdy_->get_stream_effective_local_window_size
+      rv1 = spdy_->submit_window_update(nullptr, window_size_increment);
-      (downstream_->get_stream_id());
+      if(rv1 == 0) {
-    if(recv_length >= window_size / 2) {
+        spdy_->notify();
-      rv = spdy_->submit_window_update(this, recv_length);
+      }
-      if(rv == -1) {
+    }
-        return -1;
+    if(downstream_ && downstream_->get_downstream_stream_id() != -1) {
+      window_size_increment = http2::determine_window_update_transmission
+        (spdy_->get_session(), downstream_->get_downstream_stream_id());
+      if(window_size_increment != -1) {
+        rv2 = spdy_->submit_window_update(this, window_size_increment);
+        if(rv2 == 0) {
+          spdy_->notify();
+        }
       }
-      spdy_->notify();
     }
   }
-  return 0;
+  return (rv1 == 0 && rv2 == 0) ? 0 : -1;
 }
 
 int SpdyDownstreamConnection::on_read()

src/shrpx_spdy_session.cc

@@ -602,7 +602,12 @@ int SpdySession::submit_window_update(SpdyDownstreamConnection *dconn,
 {
   assert(state_ == CONNECTED);
   int rv;
-  auto stream_id = dconn->get_downstream()->get_downstream_stream_id();
+  int32_t stream_id;
+  if(dconn) {
+    stream_id = dconn->get_downstream()->get_downstream_stream_id();
+  } else {
+    stream_id = 0;
+  }
   rv = nghttp2_submit_window_update(session_, NGHTTP2_FLAG_NONE,
                                     stream_id, amount);
   if(rv < NGHTTP2_ERR_FATAL) {
@@ -618,16 +623,9 @@ int32_t SpdySession::get_initial_window_size() const
   return (1 << get_config()->spdy_downstream_window_bits) - 1;
 }
 
-int32_t SpdySession::get_stream_effective_recv_data_length(int32_t stream_id)
+nghttp2_session* SpdySession::get_session() const
 {
-  return nghttp2_session_get_stream_effective_recv_data_length
+  return session_;
-    (session_, stream_id);
-}
-
-int32_t SpdySession::get_stream_effective_local_window_size(int32_t stream_id)
-{
-  return nghttp2_session_get_stream_effective_local_window_size
-    (session_, stream_id);
 }
 
 bool SpdySession::get_flow_control() const
@@ -1077,6 +1075,10 @@ int SpdySession::on_connect()
                                   NGHTTP2_OPT_NO_AUTO_STREAM_WINDOW_UPDATE,
                                   &val, sizeof(val));
   assert(rv == 0);
+  rv = nghttp2_session_set_option(session_,
+                                  NGHTTP2_OPT_NO_AUTO_CONNECTION_WINDOW_UPDATE,
+                                  &val, sizeof(val));
+  assert(rv == 0);
 
   nghttp2_settings_entry entry[2];
   entry[0].settings_id = NGHTTP2_SETTINGS_MAX_CONCURRENT_STREAMS;
@@ -1090,13 +1092,6 @@ int SpdySession::on_connect()
   if(rv != 0) {
     return -1;
   }
-  // Set large connection-level window size to effectively disable
-  // connection-level flow control.
-  rv = nghttp2_submit_window_update(session_, NGHTTP2_FLAG_NONE,
-                                    0, 1000000007);
-  if(rv != 0) {
-    return -1;
-  }
 
   bufferevent_write(bev_, NGHTTP2_CLIENT_CONNECTION_HEADER,
                     NGHTTP2_CLIENT_CONNECTION_HEADER_LEN);

src/shrpx_spdy_session.h

@@ -70,12 +70,13 @@ public:
 
   int submit_rst_stream(int32_t stream_id, nghttp2_error_code error_code);
 
+  // To send WINDOW_UPDATE for a connection, specify nullptr to
+  // |dconn|.
   int submit_window_update(SpdyDownstreamConnection *dconn, int32_t amount);
 
   int32_t get_initial_window_size() const;
 
-  int32_t get_stream_effective_recv_data_length(int32_t stream_id);
+  nghttp2_session* get_session() const;
-  int32_t get_stream_effective_local_window_size(int32_t stream_id);
 
   bool get_flow_control() const;