HTTPS: Flow control in request chain

examples/shrpx_downstream.cc

@@ -215,6 +215,21 @@ void Downstream::set_request_connection_close(bool f)
   request_connection_close_ = f;
 }
 
+namespace {
+const size_t DOWNSTREAM_OUTPUT_UPPER_THRES = 64*1024;
+} // namespace
+
+bool Downstream::get_output_buffer_full()
+{
+  if(dconn_) {
+    bufferevent *bev = dconn_->get_bev();
+    evbuffer *output = bufferevent_get_output(bev);
+    return evbuffer_get_length(output) >= DOWNSTREAM_OUTPUT_UPPER_THRES;
+  } else {
+    return false;
+  }
+}
+
 int Downstream::push_request_headers()
 {
   bool xff_found = false;
@@ -281,6 +296,10 @@ int Downstream::push_upload_data_chunk(const uint8_t *data, size_t datalen)
 {
   // Assumes that request headers have already been pushed to output
   // buffer using push_request_headers().
+  if(!dconn_) {
+    LOG(WARNING) << "dconn_ is NULL";
+    return 0;
+  }
   ssize_t res = 0;
   int rv;
   bufferevent *bev = dconn_->get_bev();

examples/shrpx_downstream.h

@@ -60,6 +60,9 @@ public:
   void force_resume_read();
   void set_downstream_connection(DownstreamConnection *dconn);
   DownstreamConnection* get_downstream_connection();
+  // Returns true if output buffer is full. If underlying dconn_ is
+  // NULL, this function always returns false.
+  bool get_output_buffer_full();
   // downstream request API
   const Headers& get_request_headers() const;
   void add_request_header(const std::string& name, const std::string& value);

examples/shrpx_https_upstream.cc

@@ -258,12 +258,19 @@ int HttpsUpstream::on_read()
     }
   } else if(htperr == htparse_error_none) {
     // downstream can be NULL here.
-    if(downstream && downstream->get_request_state() == Downstream::INITIAL &&
+    if(downstream) {
+      if(downstream->get_request_state() == Downstream::INITIAL &&
          current_header_length_ > SHRPX_HTTPS_MAX_HEADER_LENGTH) {
         LOG(WARNING) << "Request Header too long:" << current_header_length_
                      << " bytes";
         get_client_handler()->set_should_close_after_write(true);
         error_reply(400);
+      } else if(downstream->get_output_buffer_full()) {
+        if(ENABLE_LOG) {
+          LOG(INFO) << "Downstream output buffer is full";
+        }
+        pause_read(SHRPX_NO_BUFFER);
+      }
     }
   } else {
     if(ENABLE_LOG) {
@@ -369,6 +376,11 @@ void https_downstream_readcb(bufferevent *bev, void *ptr)
 namespace {
 void https_downstream_writecb(bufferevent *bev, void *ptr)
 {
+  DownstreamConnection *dconn = reinterpret_cast<DownstreamConnection*>(ptr);
+  Downstream *downstream = dconn->get_downstream();
+  HttpsUpstream *upstream;
+  upstream = static_cast<HttpsUpstream*>(downstream->get_upstream());
+  upstream->resume_read(SHRPX_NO_BUFFER);
 }
 } // namespace