From faee23a9258629eff91094bf50792b0ac10fcda5 Mon Sep 17 00:00:00 2001
From: Tatsuhiro Tsujikawa <tatsuhiro.t@gmail.com>
Date: Wed, 6 Jun 2012 21:39:55 +0900
Subject: [PATCH] Fixed assertion failure. Resume downstream read on SPDY
 stream close.

---
 examples/shrpx_downstream.cc    | 16 +++++++++++-----
 examples/shrpx_downstream.h     |  1 +
 examples/shrpx_io_control.cc    |  6 ++++++
 examples/shrpx_io_control.h     |  2 ++
 examples/shrpx_spdy_upstream.cc |  5 +++++
 5 files changed, 25 insertions(+), 5 deletions(-)

diff --git a/examples/shrpx_downstream.cc b/examples/shrpx_downstream.cc
index 725dc567..e121638b 100644
--- a/examples/shrpx_downstream.cc
+++ b/examples/shrpx_downstream.cc
@@ -87,6 +87,11 @@ bool Downstream::resume_read(IOCtrlReason reason)
   return ioctrl_.resume_read(reason);
 }
 
+void Downstream::force_resume_read()
+{
+  ioctrl_.force_resume_read();
+}
+
 namespace {
 void check_transfer_encoding_chunked(bool *chunked,
                                      const Headers::value_type &item)
@@ -416,12 +421,13 @@ void body_buf_cb(evbuffer *body, size_t oldlen, size_t newlen, void *arg)
 
 int Downstream::init_response_body_buf()
 {
-  assert(response_body_buf_ == 0);
-  response_body_buf_ = evbuffer_new();
-  if(response_body_buf_ == 0) {
-    DIE();
+  if(!response_body_buf_) {
+    response_body_buf_ = evbuffer_new();
+    if(response_body_buf_ == 0) {
+      DIE();
+    }
+    evbuffer_setcb(response_body_buf_, body_buf_cb, this);
   }
-  evbuffer_setcb(response_body_buf_, body_buf_cb, this);
   return 0;
 }
 
diff --git a/examples/shrpx_downstream.h b/examples/shrpx_downstream.h
index e7df104c..d4c8d987 100644
--- a/examples/shrpx_downstream.h
+++ b/examples/shrpx_downstream.h
@@ -56,6 +56,7 @@ public:
   int32_t get_stream_id() const;
   void pause_read(IOCtrlReason reason);
   bool resume_read(IOCtrlReason reason);
+  void force_resume_read();
   // 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_io_control.cc b/examples/shrpx_io_control.cc
index 76b957ad..007c39d5 100644
--- a/examples/shrpx_io_control.cc
+++ b/examples/shrpx_io_control.cc
@@ -58,4 +58,10 @@ bool IOControl::resume_read(IOCtrlReason reason)
   }
 }
 
+void IOControl::force_resume_read()
+{
+  std::fill(ctrlv_.begin(), ctrlv_.end(), 0);
+  bufferevent_enable(bev_, EV_READ);
+}
+
 } // namespace shrpx
diff --git a/examples/shrpx_io_control.h b/examples/shrpx_io_control.h
index b73e36ed..800c4e7e 100644
--- a/examples/shrpx_io_control.h
+++ b/examples/shrpx_io_control.h
@@ -48,6 +48,8 @@ public:
   void pause_read(IOCtrlReason reason);
   // Returns true if read operation is enabled after this call
   bool resume_read(IOCtrlReason reason);
+  // Clear all pause flags and enable read
+  void force_resume_read();
 private:
   bufferevent *bev_;
   std::vector<int> ctrlv_;
diff --git a/examples/shrpx_spdy_upstream.cc b/examples/shrpx_spdy_upstream.cc
index a78356dd..7ac4f273 100644
--- a/examples/shrpx_spdy_upstream.cc
+++ b/examples/shrpx_spdy_upstream.cc
@@ -103,6 +103,11 @@ void on_stream_close_callback
       if(downstream->get_response_state() == Downstream::MSG_COMPLETE) {
         upstream->get_downstream_queue()->remove(downstream);
         delete downstream;
+      } else {
+        // At this point, downstream read may be paused. To reclaim
+        // file descriptor, enable read here and catch read
+        // notification. And delete downstream there.
+        downstream->force_resume_read();
       }
     }
   }