#2 改进Chunk模式的数据转发方法，降低连接转发延迟与数据缓存量

* 原实现需将所有的Chunk收起以后，统一处理再转发；
* 利用HttpParser的Pause模式。当解析完一个Chunk后，置HttpParser为Pause模式，
退出回调函数，调用业务处理函数，而后立即转发。

3 files changed, 31 insertions, 5 deletions

diff --git a/src/http1.cc b/src/http1.cc
index 6b388ac..868d367 100644
--- a/src/http1.cc
+++ b/src/http1.cc
@@ -344,11 +344,12 @@ Http1Request::Http1Request()
 ssize_t Http1Request::ConstructFromMemory(const char * buf, size_t buflen)
 {
     parser_->data = this;
+
     size_t sz_parsed = http_parser_execute(parser_.get(),
         Http1RequestParserCallbacks::CallbackSettings(), buf, buflen);
 
-    /* 解析错误 */
-    if (sz_parsed && parser_->http_errno > 0)
+    /* 解析失败 */
+    if(sz_parsed && parser_->http_errno > 0)
     {
         throw invalid_input_format(string_format("Failed at http parsing: errcode=%u, %s, %s",
             parser_->http_errno, http_errno_name(static_cast<http_errno>(parser_->http_errno)),
@@ -741,6 +742,8 @@ int Http1ResponseParserCallbacks::CallbackOnChunkComplete(http_parser * parser)
     assert(!__last_body_content.is_complete);
     __last_body_content.is_complete = true;
 
+    /* 设置Pause模式，暂停解析，以便调用外部业务处理函数 */
+    http_parser_pause(parser, 1);
     return 0;
 }
 
@@ -754,11 +757,24 @@ ssize_t Http1Response::ConstructFromMemory(const char * buf, size_t buflen)
 {
     parser_->data = this;
 
+    /* 31 == HTTP_PAUSED,
+    * TODO: 替换这一硬编码 */
+    if (parser_->http_errno == 31)
+    {
+        http_parser_pause(parser_.get(), 0);
+    }
+
     size_t sz_parsed = http_parser_execute(parser_.get(),
         Http1ResponseParserCallbacks::CallbackSettings(), buf, buflen);
 
+    /* 暂停状态，为了调用外部回调函数 */
+    if(sz_parsed && parser_->http_errno == 31)
+    {
+        return sz_parsed;
+    }
+
     /* 解析错误 */
-    if (sz_parsed && parser_->http_errno > 0)
+    else if (sz_parsed && parser_->http_errno > 0)
     {
         throw invalid_input_format(string_format("Failed at http parsing: errcode=%u, %s, %s",
             parser_->http_errno, http_errno_name(static_cast<http_errno>(parser_->http_errno)),
@@ -932,6 +948,8 @@ int Http1Connection::on_connection_read_response(pxy_conn_ctx_t * conn_ctx, pxy_
 
     /* 转发缓存中的数据，如果该Response处理结束，本Session处理完毕，销毁Session */
     evbuffer_add_buffer(upstream_evbuf, response.StolenEvBuf().release());
+    LOG(DEBUG) << "ForwardData, upstream len = " << evbuffer_get_length(upstream_evbuf);
+
     if (response.SectionState(response.kSectionMessage) == response.kStateComplete)
     {
         drop_last_session();
diff --git a/src/httpscan.cc b/src/httpscan.cc
index 777f988..4f04067 100644
--- a/src/httpscan.cc
+++ b/src/httpscan.cc
@@ -277,7 +277,7 @@ void HttpScanSession::ScanResponseBody(HttpSession * http_session_ctx)
             return hit_scan_error();
         }
 
-        CLOG(DEBUG, "HttpScanTrace") << hexdump("ContentBody", body_content_raw, body_content_length);
+        //CLOG(DEBUG, "HttpScanTrace") << hexdump("ContentBody", body_content_raw, body_content_length);
     }
 
     return;
diff --git a/src/pxyconn.cc b/src/pxyconn.cc
index 40d9b13..433ceca 100644
--- a/src/pxyconn.cc
+++ b/src/pxyconn.cc
@@ -1447,7 +1447,15 @@ static void pxy_bev_readcb(struct bufferevent * bev, void * arg)
     if (evbuffer_get_length(inbuf) == 0)
         return;
 
-    evbuffer_add_buffer(outbuf, inbuf);
+    if (ctx->passthrough)
+    {
+        evbuffer_add_buffer(outbuf, inbuf);
+    }
+    else if (evbuffer_get_length(inbuf) != 0)
+    {
+        bufferevent_trigger(bev, EV_READ, BEV_OPT_DEFER_CALLBACKS);
+    }
+
     if (evbuffer_get_length(outbuf) >= OUTBUF_LIMIT)
     {
         CLOG(DEBUG, "conntrace") << string_format("ctx = %p, exceed output limit, disable read", ctx);