From c2785955caa5f6e111d89b277338cd50e350a0e7 Mon Sep 17 00:00:00 2001
From: Tatsuhiro Tsujikawa <tatsuhiro.t@gmail.com>
Date: Sun, 10 Jun 2012 00:49:33 +0900
Subject: [PATCH] HTTPS: Flow control in request chain

---
 examples/shrpx_downstream.cc     | 19 +++++++++++++++++++
 examples/shrpx_downstream.h      |  3 +++
 examples/shrpx_https_upstream.cc | 24 ++++++++++++++++++------
 3 files changed, 40 insertions(+), 6 deletions(-)

diff --git a/examples/shrpx_downstream.cc b/examples/shrpx_downstream.cc
index 2ca9006..3c804b2 100644
--- a/examples/shrpx_downstream.cc
+++ b/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();
diff --git a/examples/shrpx_downstream.h b/examples/shrpx_downstream.h
index 80f3cdb..a0c0c4a 100644
--- a/examples/shrpx_downstream.h
+++ b/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);
diff --git a/examples/shrpx_https_upstream.cc b/examples/shrpx_https_upstream.cc
index ed28b43..5588931 100644
--- a/examples/shrpx_https_upstream.cc
+++ b/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 &&
-       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);
+    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