nghttpx: Handle connection flow control for DATA not sent to backend

src/shrpx_http2_upstream.cc

@@ -449,11 +449,13 @@ int on_data_chunk_recv_callback(nghttp2_session *session,
   auto downstream = upstream->find_downstream(stream_id);
 
   if(!downstream) {
+    upstream->handle_ign_data_chunk(len);
     return 0;
   }
 
   if(downstream->push_upload_data_chunk(data, len) != 0) {
     upstream->rst_stream(downstream, NGHTTP2_INTERNAL_ERROR);
+    upstream->handle_ign_data_chunk(len);
     return 0;
   }
 
@@ -532,7 +534,8 @@ nghttp2_error_code infer_upstream_rst_stream_error_code
 Http2Upstream::Http2Upstream(ClientHandler *handler)
   : handler_(handler),
     session_(nullptr),
-    settings_timerev_(nullptr)
+    settings_timerev_(nullptr),
+    recv_ign_window_size_(0)
 {
   handler->set_upstream_timeouts(&get_config()->http2_upstream_read_timeout,
                                  &get_config()->upstream_write_timeout);
@@ -938,10 +941,18 @@ int Http2Upstream::window_update(Downstream *downstream,
                                  int32_t window_size_increment)
 {
   int rv;
+  int32_t stream_id;
+
+  if(downstream) {
+    stream_id = downstream->get_stream_id();
+  } else {
+    stream_id = 0;
+    recv_ign_window_size_ = 0;
+  }
+
   rv = nghttp2_submit_window_update(session_, NGHTTP2_FLAG_NONE,
-                                    downstream ?
-                                    downstream->get_stream_id() : 0,
-                                    window_size_increment);
+                                    stream_id, window_size_increment);
+
   if(rv < NGHTTP2_ERR_FATAL) {
     ULOG(FATAL, this) << "nghttp2_submit_window_update() failed: "
                       << nghttp2_strerror(rv);
@@ -1243,4 +1254,19 @@ int Http2Upstream::on_downstream_abort_request(Downstream *downstream,
   return send();
 }
 
+int Http2Upstream::handle_ign_data_chunk(size_t len)
+{
+  int32_t window_size;
+
+  recv_ign_window_size_ += len;
+
+  window_size = nghttp2_session_get_effective_local_window_size(session_);
+
+  if(recv_ign_window_size_ >= window_size / 2) {
+    window_update(nullptr, recv_ign_window_size_);
+  }
+
+  return 0;
+}
+
 } // namespace shrpx

src/shrpx_http2_upstream.h

@@ -81,12 +81,16 @@ public:
   int upgrade_upstream(HttpsUpstream *upstream);
   int start_settings_timer();
   void stop_settings_timer();
+  int handle_ign_data_chunk(size_t len);
 private:
   DownstreamQueue downstream_queue_;
   std::unique_ptr<HttpsUpstream> pre_upstream_;
   ClientHandler *handler_;
   nghttp2_session *session_;
   event *settings_timerev_;
+  // Received DATA frame size while it is not sent to backend before
+  // any connection-level WINDOW_UPDATE
+  int32_t recv_ign_window_size_;
   bool flow_control_;
 };
 

src/shrpx_spdy_upstream.cc

@@ -251,11 +251,13 @@ void on_data_chunk_recv_callback(spdylay_session *session,
   auto downstream = upstream->find_downstream(stream_id);
 
   if(!downstream) {
+    upstream->handle_ign_data_chunk(len);
     return;
   }
 
   if(downstream->push_upload_data_chunk(data, len) != 0) {
     upstream->rst_stream(downstream, SPDYLAY_INTERNAL_ERROR);
+    upstream->handle_ign_data_chunk(len);
     return;
   }
 
@@ -379,7 +381,8 @@ uint32_t infer_upstream_rst_stream_status_code
 
 SpdyUpstream::SpdyUpstream(uint16_t version, ClientHandler *handler)
   : handler_(handler),
-    session_(nullptr)
+    session_(nullptr),
+    recv_ign_window_size_(0)
 {
   //handler->set_bev_cb(spdy_readcb, 0, spdy_eventcb);
   handler->set_upstream_timeouts(&get_config()->http2_upstream_read_timeout,
@@ -726,10 +729,17 @@ int SpdyUpstream::rst_stream(Downstream *downstream, int status_code)
 int SpdyUpstream::window_update(Downstream *downstream, int32_t delta)
 {
   int rv;
-  rv = spdylay_submit_window_update(session_,
-                                    downstream ?
-                                    downstream->get_stream_id() : 0,
-                                    delta);
+  int32_t stream_id;
+
+  if(downstream) {
+    stream_id = downstream->get_stream_id();
+  } else {
+    stream_id = 0;
+    recv_ign_window_size_ = 0;
+  }
+
+  rv = spdylay_submit_window_update(session_, stream_id, delta);
+
   if(rv < SPDYLAY_ERR_FATAL) {
     ULOG(FATAL, this) << "spdylay_submit_window_update() failed: "
                       << spdylay_strerror(rv);
@@ -1044,4 +1054,22 @@ int SpdyUpstream::on_downstream_abort_request(Downstream *downstream,
   return send();
 }
 
+int SpdyUpstream::handle_ign_data_chunk(size_t len)
+{
+  int32_t window_size;
+
+  if(spdylay_session_get_recv_data_length(session_) == -1) {
+    // No connection flow control
+    return 0;
+  }
+
+  window_size = 1 << get_config()->http2_upstream_connection_window_bits;
+
+  if(recv_ign_window_size_ >= window_size / 2) {
+    window_update(0, recv_ign_window_size_);
+  }
+
+  return 0;
+}
+
 } // namespace shrpx

src/shrpx_spdy_upstream.h

@@ -71,12 +71,15 @@ public:
 
   bool get_flow_control() const;
 
+  int handle_ign_data_chunk(size_t len);
+
   nghttp2::util::EvbufferBuffer sendbuf;
 private:
   DownstreamQueue downstream_queue_;
   ClientHandler *handler_;
   spdylay_session *session_;
   int32_t initial_window_size_;
+  int32_t recv_ign_window_size_;
   bool flow_control_;
 };