Merge branch 'nghttpx-retry-h1-backend-request'

src/shrpx_downstream.cc

@@ -922,6 +922,10 @@ bool Downstream::get_request_pending() const { return request_pending_; }
 
 void Downstream::set_request_header_sent(bool f) { request_header_sent_ = f; }
 
+bool Downstream::get_request_header_sent() const {
+  return request_header_sent_;
+}
+
 bool Downstream::request_submission_ready() const {
   return (request_state_ == Downstream::HEADER_COMPLETE ||
           request_state_ == Downstream::MSG_COMPLETE) &&

src/shrpx_downstream.h

@@ -312,6 +312,7 @@ public:
   void set_request_pending(bool f);
   bool get_request_pending() const;
   void set_request_header_sent(bool f);
+  bool get_request_header_sent() const;
   // Returns true if request is ready to be submitted to downstream.
   // When sending pending request, get_request_pending() should be
   // checked too because this function may return true when

src/shrpx_error.h

@@ -37,6 +37,7 @@ enum ErrorCode {
   SHRPX_ERR_EOF = -101,
   SHRPX_ERR_INPROGRESS = -102,
   SHRPX_ERR_DCONN_CANCELED = -103,
+  SHRPX_ERR_RETRY = -104,
 };
 
 } // namespace shrpx

src/shrpx_http2_upstream.cc

@@ -1212,7 +1212,7 @@ int Http2Upstream::downstream_write(DownstreamConnection *dconn) {
     return downstream_error(dconn, Downstream::EVENT_ERROR);
   }
   if (rv != 0) {
-    return -1;
+    return rv;
   }
   return 0;
 }

src/shrpx_http_downstream_connection.cc

@@ -118,27 +118,17 @@ void readcb(struct ev_loop *loop, ev_io *w, int revents) {
 } // namespace
 
 namespace {
-void writecb(struct ev_loop *loop, ev_io *w, int revents) {
+void backend_retry(Downstream *downstream) {
-  auto conn = static_cast<Connection *>(w->data);
-  auto dconn = static_cast<HttpDownstreamConnection *>(conn->data);
-  auto downstream = dconn->get_downstream();
   auto upstream = downstream->get_upstream();
   auto handler = upstream->get_client_handler();
 
-  if (upstream->downstream_write(dconn) != 0) {
+  downstream->add_retry();
+
+  if (downstream->no_more_retry()) {
     delete handler;
+    return;
   }
-}
-} // namespace
 
-namespace {
-void connectcb(struct ev_loop *loop, ev_io *w, int revents) {
-  auto conn = static_cast<Connection *>(w->data);
-  auto dconn = static_cast<HttpDownstreamConnection *>(conn->data);
-  auto downstream = dconn->get_downstream();
-  auto upstream = downstream->get_upstream();
-  auto handler = upstream->get_client_handler();
-  if (dconn->connected() != 0) {
   downstream->pop_downstream_connection();
 
   auto ndconn = handler->get_downstream_connection(downstream);
@@ -153,6 +143,37 @@ void connectcb(struct ev_loop *loop, ev_io *w, int revents) {
   if (upstream->on_downstream_abort_request(downstream, 503) != 0) {
     delete handler;
   }
+}
+} // namespace
+
+namespace {
+void writecb(struct ev_loop *loop, ev_io *w, int revents) {
+  int rv;
+  auto conn = static_cast<Connection *>(w->data);
+  auto dconn = static_cast<HttpDownstreamConnection *>(conn->data);
+  auto downstream = dconn->get_downstream();
+  auto upstream = downstream->get_upstream();
+  auto handler = upstream->get_client_handler();
+
+  rv = upstream->downstream_write(dconn);
+  if (rv == SHRPX_ERR_RETRY) {
+    backend_retry(downstream);
+    return;
+  }
+
+  if (rv != 0) {
+    delete handler;
+  }
+}
+} // namespace
+
+namespace {
+void connectcb(struct ev_loop *loop, ev_io *w, int revents) {
+  auto conn = static_cast<Connection *>(w->data);
+  auto dconn = static_cast<HttpDownstreamConnection *>(conn->data);
+  auto downstream = dconn->get_downstream();
+  if (dconn->connected() != 0) {
+    backend_retry(downstream);
     return;
   }
   writecb(loop, w, revents);
@@ -178,7 +199,8 @@ HttpDownstreamConnection::HttpDownstreamConnection(
       raddr_(nullptr),
       ioctrl_(&conn_.rlimit),
       response_htp_{0},
-      initial_addr_idx_(initial_addr_idx) {}
+      initial_addr_idx_(initial_addr_idx),
+      reuse_first_write_done_(true) {}
 
 HttpDownstreamConnection::~HttpDownstreamConnection() {
   if (LOG_ENABLED(INFO)) {
@@ -293,26 +315,7 @@ int HttpDownstreamConnection::initiate_connection() {
                 if (rv != 0) {
                   // This callback destroys |this|.
                   auto downstream = this->downstream_;
-                  auto upstream = downstream->get_upstream();
+                  backend_retry(downstream);
-                  auto handler = upstream->get_client_handler();
-
-                  downstream->pop_downstream_connection();
-
-                  auto ndconn = handler->get_downstream_connection(downstream);
-                  if (ndconn) {
-                    if (downstream->attach_downstream_connection(
-                            std::move(ndconn)) == 0) {
-                      return;
-                    }
-                  }
-
-                  downstream->set_request_state(Downstream::CONNECT_FAIL);
-
-                  if (upstream->on_downstream_abort_request(downstream, 503) !=
-                      0) {
-                    delete handler;
-                  }
-                  return;
                 }
               });
 
@@ -449,6 +452,9 @@ int HttpDownstreamConnection::initiate_connection() {
     }
 
     ev_set_cb(&conn_.rev, readcb);
+
+    do_write_ = &HttpDownstreamConnection::write_reuse_first;
+    reuse_first_write_done_ = false;
   }
 
   http_parser_init(&response_htp_, HTTP_RESPONSE);
@@ -458,6 +464,10 @@ int HttpDownstreamConnection::initiate_connection() {
 }
 
 int HttpDownstreamConnection::push_request_headers() {
+  if (downstream_->get_request_header_sent()) {
+    return 0;
+  }
+
   const auto &downstream_hostport = addr_->hostport;
   const auto &req = downstream_->request();
 
@@ -1071,6 +1081,30 @@ http_parser_settings htp_hooks = {
 };
 } // namespace
 
+int HttpDownstreamConnection::write_reuse_first() {
+  int rv;
+
+  if (conn_.tls.ssl) {
+    rv = write_tls();
+  } else {
+    rv = write_clear();
+  }
+
+  if (rv != 0) {
+    return SHRPX_ERR_RETRY;
+  }
+
+  if (conn_.tls.ssl) {
+    do_write_ = &HttpDownstreamConnection::write_tls;
+  } else {
+    do_write_ = &HttpDownstreamConnection::write_clear;
+  }
+
+  reuse_first_write_done_ = true;
+
+  return 0;
+}
+
 int HttpDownstreamConnection::read_clear() {
   conn_.last_read = ev_now(conn_.loop);
 
@@ -1116,6 +1150,9 @@ int HttpDownstreamConnection::write_clear() {
     }
 
     if (nwrite < 0) {
+      if (!reuse_first_write_done_) {
+        return nwrite;
+      }
       // We may have pending data in receive buffer which may contain
       // part of response body.  So keep reading.  Invoke read event
       // to get read(2) error just in case.
@@ -1241,6 +1278,9 @@ int HttpDownstreamConnection::write_tls() {
     }
 
     if (nwrite < 0) {
+      if (!reuse_first_write_done_) {
+        return nwrite;
+      }
       // We may have pending data in receive buffer which may contain
       // part of response body.  So keep reading.  Invoke read event
       // to get read(2) error just in case.

src/shrpx_http_downstream_connection.h

@@ -71,6 +71,7 @@ public:
 
   int initiate_connection();
 
+  int write_reuse_first();
   int read_clear();
   int write_clear();
   int read_tls();
@@ -108,6 +109,9 @@ private:
   IOControl ioctrl_;
   http_parser response_htp_;
   ssize_t initial_addr_idx_;
+  // true if first write of reused connection succeeded.  For
+  // convenience, this is initialized as true.
+  bool reuse_first_write_done_;
 };
 
 } // namespace shrpx

src/shrpx_https_upstream.cc

@@ -766,7 +766,7 @@ int HttpsUpstream::downstream_write(DownstreamConnection *dconn) {
   }
 
   if (rv != 0) {
-    return -1;
+    return rv;
   }
 
   return 0;

src/shrpx_spdy_upstream.cc

@@ -743,7 +743,7 @@ int SpdyUpstream::downstream_write(DownstreamConnection *dconn) {
     return downstream_error(dconn, Downstream::EVENT_ERROR);
   }
   if (rv != 0) {
-    return -1;
+    return rv;
   }
   return 0;
 }