src/shrpx_http_downstream_connection.cc

   1 /*
   2  * nghttp2 - HTTP/2 C Library
   3  *
   4  * Copyright (c) 2012 Tatsuhiro Tsujikawa
   5  *
   6  * Permission is hereby granted, free of charge, to any person obtaining
   7  * a copy of this software and associated documentation files (the
   8  * "Software"), to deal in the Software without restriction, including
   9  * without limitation the rights to use, copy, modify, merge, publish,
  10  * distribute, sublicense, and/or sell copies of the Software, and to
  11  * permit persons to whom the Software is furnished to do so, subject to
  12  * the following conditions:
  13  *
  14  * The above copyright notice and this permission notice shall be
  15  * included in all copies or substantial portions of the Software.
  16  *
  17  * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
  18  * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
  19  * MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
  20  * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE
  21  * LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
  22  * OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
  23  * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
  24  */
  25 #include "shrpx_http_downstream_connection.h"
  26
  27 #include <openssl/rand.h>
  28
  29 #include "shrpx_client_handler.h"
  30 #include "shrpx_upstream.h"
  31 #include "shrpx_downstream.h"
  32 #include "shrpx_config.h"
  33 #include "shrpx_error.h"
  34 #include "shrpx_http.h"
  35 #include "shrpx_log_config.h"
  36 #include "shrpx_connect_blocker.h"
  37 #include "shrpx_downstream_connection_pool.h"
  38 #include "shrpx_worker.h"
  39 #include "shrpx_http2_session.h"
  40 #include "shrpx_tls.h"
  41 #include "shrpx_log.h"
  42 #include "http2.h"
  43 #include "util.h"
  44 #include "ssl_compat.h"
  45
  46 using namespace nghttp2;
  47
  48 namespace shrpx {
  49
  50 namespace {
  51 void timeoutcb(struct ev_loop *loop, ev_timer *w, int revents) {
  52   auto conn = static_cast<Connection *>(w->data);
  53   auto dconn = static_cast<HttpDownstreamConnection *>(conn->data);
  54
  55   if (w == &conn->rt && !conn->expired_rt()) {
  56     return;
  57   }
  58
  59   if (LOG_ENABLED(INFO)) {
  60     DCLOG(INFO, dconn) << "Time out";
  61   }
  62
  63   auto downstream = dconn->get_downstream();
  64   auto upstream = downstream->get_upstream();
  65   auto handler = upstream->get_client_handler();
  66   auto &resp = downstream->response();
  67
  68   // Do this so that dconn is not pooled
  69   resp.connection_close = true;
  70
  71   if (upstream->downstream_error(dconn, Downstream::EVENT_TIMEOUT) != 0) {
  72     delete handler;
  73   }
  74 }
  75 } // namespace
  76
  77 namespace {
  78 void retry_downstream_connection(Downstream *downstream,
  79                                  unsigned int status_code) {
  80   auto upstream = downstream->get_upstream();
  81   auto handler = upstream->get_client_handler();
  82
  83   assert(!downstream->get_request_header_sent());
  84
  85   downstream->add_retry();
  86
  87   if (downstream->no_more_retry()) {
  88     delete handler;
  89     return;
  90   }
  91
  92   downstream->pop_downstream_connection();
  93   auto buf = downstream->get_request_buf();
  94   buf->reset();
  95
  96   int rv;
  97
  98   for (;;) {
  99     auto ndconn = handler->get_downstream_connection(rv, downstream);
 100     if (!ndconn) {
 101       break;
 102     }
 103     if (downstream->attach_downstream_connection(std::move(ndconn)) != 0) {
 104       continue;
 105     }
 106     if (downstream->push_request_headers() == 0) {
 107       return;
 108     }
 109   }
 110
 111   downstream->set_request_state(DownstreamState::CONNECT_FAIL);
 112
 113   if (rv == SHRPX_ERR_TLS_REQUIRED) {
 114     rv = upstream->on_downstream_abort_request_with_https_redirect(downstream);
 115   } else {
 116     rv = upstream->on_downstream_abort_request(downstream, status_code);
 117   }
 118
 119   if (rv != 0) {
 120     delete handler;
 121   }
 122 }
 123 } // namespace
 124
 125 namespace {
 126 void connect_timeoutcb(struct ev_loop *loop, ev_timer *w, int revents) {
 127   auto conn = static_cast<Connection *>(w->data);
 128   auto dconn = static_cast<HttpDownstreamConnection *>(conn->data);
 129   auto addr = dconn->get_addr();
 130   auto raddr = dconn->get_raddr();
 131
 132   DCLOG(WARN, dconn) << "Connect time out; addr="
 133                      << util::to_numeric_addr(raddr);
 134
 135   downstream_failure(addr, raddr);
 136
 137   auto downstream = dconn->get_downstream();
 138
 139   retry_downstream_connection(downstream, 504);
 140 }
 141 } // namespace
 142
 143 namespace {
 144 void backend_retry(Downstream *downstream) {
 145   retry_downstream_connection(downstream, 502);
 146 }
 147 } // namespace
 148
 149 namespace {
 150 void readcb(struct ev_loop *loop, ev_io *w, int revents) {
 151   int rv;
 152   auto conn = static_cast<Connection *>(w->data);
 153   auto dconn = static_cast<HttpDownstreamConnection *>(conn->data);
 154   auto downstream = dconn->get_downstream();
 155   auto upstream = downstream->get_upstream();
 156   auto handler = upstream->get_client_handler();
 157
 158   rv = upstream->downstream_read(dconn);
 159   if (rv != 0) {
 160     if (rv == SHRPX_ERR_RETRY) {
 161       backend_retry(downstream);
 162       return;
 163     }
 164
 165     delete handler;
 166   }
 167 }
 168 } // namespace
 169
 170 namespace {
 171 void writecb(struct ev_loop *loop, ev_io *w, int revents) {
 172   int rv;
 173   auto conn = static_cast<Connection *>(w->data);
 174   auto dconn = static_cast<HttpDownstreamConnection *>(conn->data);
 175   auto downstream = dconn->get_downstream();
 176   auto upstream = downstream->get_upstream();
 177   auto handler = upstream->get_client_handler();
 178
 179   rv = upstream->downstream_write(dconn);
 180   if (rv == SHRPX_ERR_RETRY) {
 181     backend_retry(downstream);
 182     return;
 183   }
 184
 185   if (rv != 0) {
 186     delete handler;
 187   }
 188 }
 189 } // namespace
 190
 191 namespace {
 192 void connectcb(struct ev_loop *loop, ev_io *w, int revents) {
 193   auto conn = static_cast<Connection *>(w->data);
 194   auto dconn = static_cast<HttpDownstreamConnection *>(conn->data);
 195   auto downstream = dconn->get_downstream();
 196   if (dconn->connected() != 0) {
 197     backend_retry(downstream);
 198     return;
 199   }
 200   writecb(loop, w, revents);
 201 }
 202 } // namespace
 203
 204 HttpDownstreamConnection::HttpDownstreamConnection(
 205     const std::shared_ptr<DownstreamAddrGroup> &group, DownstreamAddr *addr,
 206     struct ev_loop *loop, Worker *worker)
 207     : conn_(loop, -1, nullptr, worker->get_mcpool(),
 208             group->shared_addr->timeout.write, group->shared_addr->timeout.read,
 209             {}, {}, connectcb, readcb, connect_timeoutcb, this,
 210             get_config()->tls.dyn_rec.warmup_threshold,
 211             get_config()->tls.dyn_rec.idle_timeout, Proto::HTTP1),
 212       on_read_(&HttpDownstreamConnection::noop),
 213       on_write_(&HttpDownstreamConnection::noop),
 214       signal_write_(&HttpDownstreamConnection::noop),
 215       worker_(worker),
 216       ssl_ctx_(worker->get_cl_ssl_ctx()),
 217       group_(group),
 218       addr_(addr),
 219       raddr_(nullptr),
 220       ioctrl_(&conn_.rlimit),
 221       response_htp_{0},
 222       first_write_done_(false),
 223       reusable_(true),
 224       request_header_written_(false) {}
 225
 226 HttpDownstreamConnection::~HttpDownstreamConnection() {
 227   if (LOG_ENABLED(INFO)) {
 228     DCLOG(INFO, this) << "Deleted";
 229   }
 230
 231   if (dns_query_) {
 232     auto dns_tracker = worker_->get_dns_tracker();
 233     dns_tracker->cancel(dns_query_.get());
 234   }
 235 }
 236
 237 int HttpDownstreamConnection::attach_downstream(Downstream *downstream) {
 238   int rv;
 239
 240   if (LOG_ENABLED(INFO)) {
 241     DCLOG(INFO, this) << "Attaching to DOWNSTREAM:" << downstream;
 242   }
 243
 244   downstream_ = downstream;
 245
 246   rv = initiate_connection();
 247   if (rv != 0) {
 248     downstream_ = nullptr;
 249     return rv;
 250   }
 251
 252   return 0;
 253 }
 254
 255 namespace {
 256 int htp_msg_begincb(llhttp_t *htp);
 257 int htp_hdr_keycb(llhttp_t *htp, const char *data, size_t len);
 258 int htp_hdr_valcb(llhttp_t *htp, const char *data, size_t len);
 259 int htp_hdrs_completecb(llhttp_t *htp);
 260 int htp_bodycb(llhttp_t *htp, const char *data, size_t len);
 261 int htp_msg_completecb(llhttp_t *htp);
 262 } // namespace
 263
 264 namespace {
 265 constexpr llhttp_settings_t htp_hooks = {
 266     htp_msg_begincb,     // llhttp_cb      on_message_begin;
 267     nullptr,             // llhttp_data_cb on_url;
 268     nullptr,             // llhttp_data_cb on_status;
 269     htp_hdr_keycb,       // llhttp_data_cb on_header_field;
 270     htp_hdr_valcb,       // llhttp_data_cb on_header_value;
 271     htp_hdrs_completecb, // llhttp_cb      on_headers_complete;
 272     htp_bodycb,          // llhttp_data_cb on_body;
 273     htp_msg_completecb,  // llhttp_cb      on_message_complete;
 274     nullptr,             // llhttp_cb      on_chunk_header
 275     nullptr,             // llhttp_cb      on_chunk_complete
 276 };
 277 } // namespace
 278
 279 int HttpDownstreamConnection::initiate_connection() {
 280   int rv;
 281
 282   auto worker_blocker = worker_->get_connect_blocker();
 283   if (worker_blocker->blocked()) {
 284     if (LOG_ENABLED(INFO)) {
 285       DCLOG(INFO, this)
 286           << "Worker wide backend connection was blocked temporarily";
 287     }
 288     return SHRPX_ERR_NETWORK;
 289   }
 290
 291   auto &downstreamconf = *worker_->get_downstream_config();
 292
 293   if (conn_.fd == -1) {
 294     auto check_dns_result = dns_query_.get() != nullptr;
 295
 296     if (check_dns_result) {
 297       assert(addr_->dns);
 298     }
 299
 300     auto &connect_blocker = addr_->connect_blocker;
 301
 302     if (connect_blocker->blocked()) {
 303       if (LOG_ENABLED(INFO)) {
 304         DCLOG(INFO, this) << "Backend server " << addr_->host << ":"
 305                           << addr_->port << " was not available temporarily";
 306       }
 307
 308       return SHRPX_ERR_NETWORK;
 309     }
 310
 311     Address *raddr;
 312
 313     if (addr_->dns) {
 314       if (!check_dns_result) {
 315         auto dns_query = std::make_unique<DNSQuery>(
 316             addr_->host,
 317             [this](DNSResolverStatus status, const Address *result) {
 318               int rv;
 319
 320               if (status == DNSResolverStatus::OK) {
 321                 *this->resolved_addr_ = *result;
 322               }
 323
 324               rv = this->initiate_connection();
 325               if (rv != 0) {
 326                 // This callback destroys |this|.
 327                 auto downstream = this->downstream_;
 328                 backend_retry(downstream);
 329               }
 330             });
 331
 332         auto dns_tracker = worker_->get_dns_tracker();
 333
 334         if (!resolved_addr_) {
 335           resolved_addr_ = std::make_unique<Address>();
 336         }
 337         switch (dns_tracker->resolve(resolved_addr_.get(), dns_query.get())) {
 338         case DNSResolverStatus::ERROR:
 339           downstream_failure(addr_, nullptr);
 340           return SHRPX_ERR_NETWORK;
 341         case DNSResolverStatus::RUNNING:
 342           dns_query_ = std::move(dns_query);
 343           return 0;
 344         case DNSResolverStatus::OK:
 345           break;
 346         default:
 347           assert(0);
 348         }
 349       } else {
 350         switch (dns_query_->status) {
 351         case DNSResolverStatus::ERROR:
 352           dns_query_.reset();
 353           downstream_failure(addr_, nullptr);
 354           return SHRPX_ERR_NETWORK;
 355         case DNSResolverStatus::OK:
 356           dns_query_.reset();
 357           break;
 358         default:
 359           assert(0);
 360         }
 361       }
 362
 363       raddr = resolved_addr_.get();
 364       util::set_port(*resolved_addr_, addr_->port);
 365     } else {
 366       raddr = &addr_->addr;
 367     }
 368
 369     conn_.fd = util::create_nonblock_socket(raddr->su.storage.ss_family);
 370
 371     if (conn_.fd == -1) {
 372       auto error = errno;
 373       DCLOG(WARN, this) << "socket() failed; addr="
 374                         << util::to_numeric_addr(raddr) << ", errno=" << error;
 375
 376       worker_blocker->on_failure();
 377
 378       return SHRPX_ERR_NETWORK;
 379     }
 380
 381     worker_blocker->on_success();
 382
 383     rv = connect(conn_.fd, &raddr->su.sa, raddr->len);
 384     if (rv != 0 && errno != EINPROGRESS) {
 385       auto error = errno;
 386       DCLOG(WARN, this) << "connect() failed; addr="
 387                         << util::to_numeric_addr(raddr) << ", errno=" << error;
 388
 389       downstream_failure(addr_, raddr);
 390
 391       return SHRPX_ERR_NETWORK;
 392     }
 393
 394     if (LOG_ENABLED(INFO)) {
 395       DCLOG(INFO, this) << "Connecting to downstream server";
 396     }
 397
 398     raddr_ = raddr;
 399
 400     if (addr_->tls) {
 401       assert(ssl_ctx_);
 402
 403       auto ssl = tls::create_ssl(ssl_ctx_);
 404       if (!ssl) {
 405         return -1;
 406       }
 407
 408       tls::setup_downstream_http1_alpn(ssl);
 409
 410       conn_.set_ssl(ssl);
 411       conn_.tls.client_session_cache = &addr_->tls_session_cache;
 412
 413       auto sni_name =
 414           addr_->sni.empty() ? StringRef{addr_->host} : StringRef{addr_->sni};
 415       if (!util::numeric_host(sni_name.c_str())) {
 416         SSL_set_tlsext_host_name(conn_.tls.ssl, sni_name.c_str());
 417       }
 418
 419       auto session = tls::reuse_tls_session(addr_->tls_session_cache);
 420       if (session) {
 421         SSL_set_session(conn_.tls.ssl, session);
 422         SSL_SESSION_free(session);
 423       }
 424
 425       conn_.prepare_client_handshake();
 426     }
 427
 428     ev_io_set(&conn_.wev, conn_.fd, EV_WRITE);
 429     ev_io_set(&conn_.rev, conn_.fd, EV_READ);
 430
 431     conn_.wlimit.startw();
 432
 433     conn_.wt.repeat = downstreamconf.timeout.connect;
 434     ev_timer_again(conn_.loop, &conn_.wt);
 435   } else {
 436     // we may set read timer cb to idle_timeoutcb.  Reset again.
 437     ev_set_cb(&conn_.rt, timeoutcb);
 438     if (conn_.read_timeout < group_->shared_addr->timeout.read) {
 439       conn_.read_timeout = group_->shared_addr->timeout.read;
 440       conn_.last_read = ev_now(conn_.loop);
 441     } else {
 442       conn_.again_rt(group_->shared_addr->timeout.read);
 443     }
 444
 445     ev_set_cb(&conn_.rev, readcb);
 446
 447     on_write_ = &HttpDownstreamConnection::write_first;
 448     first_write_done_ = false;
 449     request_header_written_ = false;
 450   }
 451
 452   llhttp_init(&response_htp_, HTTP_RESPONSE, &htp_hooks);
 453   response_htp_.data = downstream_;
 454
 455   return 0;
 456 }
 457
 458 int HttpDownstreamConnection::push_request_headers() {
 459   if (request_header_written_) {
 460     signal_write();
 461     return 0;
 462   }
 463
 464   const auto &downstream_hostport = addr_->hostport;
 465   const auto &req = downstream_->request();
 466
 467   auto &balloc = downstream_->get_block_allocator();
 468
 469   auto connect_method = req.regular_connect_method();
 470
 471   auto config = get_config();
 472   auto &httpconf = config->http;
 473
 474   request_header_written_ = true;
 475
 476   // For HTTP/1.0 request, there is no authority in request.  In that
 477   // case, we use backend server's host nonetheless.
 478   auto authority = StringRef(downstream_hostport);
 479   auto no_host_rewrite =
 480       httpconf.no_host_rewrite || config->http2_proxy || connect_method;
 481
 482   if (no_host_rewrite && !req.authority.empty()) {
 483     authority = req.authority;
 484   }
 485
 486   downstream_->set_request_downstream_host(authority);
 487
 488   auto buf = downstream_->get_request_buf();
 489
 490   // Assume that method and request path do not contain \r\n.
 491   auto meth = http2::to_method_string(
 492       req.connect_proto == ConnectProto::WEBSOCKET ? HTTP_GET : req.method);
 493   buf->append(meth);
 494   buf->append(' ');
 495
 496   if (connect_method) {
 497     buf->append(authority);
 498   } else if (config->http2_proxy) {
 499     // Construct absolute-form request target because we are going to
 500     // send a request to a HTTP/1 proxy.
 501     assert(!req.scheme.empty());
 502     buf->append(req.scheme);
 503     buf->append("://");
 504     buf->append(authority);
 505     buf->append(req.path);
 506   } else if (req.method == HTTP_OPTIONS && req.path.empty()) {
 507     // Server-wide OPTIONS
 508     buf->append("*");
 509   } else {
 510     buf->append(req.path);
 511   }
 512   buf->append(" HTTP/1.1\r\nHost: ");
 513   buf->append(authority);
 514   buf->append("\r\n");
 515
 516   auto &fwdconf = httpconf.forwarded;
 517   auto &xffconf = httpconf.xff;
 518   auto &xfpconf = httpconf.xfp;
 519   auto &earlydataconf = httpconf.early_data;
 520
 521   uint32_t build_flags =
 522       (fwdconf.strip_incoming ? http2::HDOP_STRIP_FORWARDED : 0) |
 523       (xffconf.strip_incoming ? http2::HDOP_STRIP_X_FORWARDED_FOR : 0) |
 524       (xfpconf.strip_incoming ? http2::HDOP_STRIP_X_FORWARDED_PROTO : 0) |
 525       (earlydataconf.strip_incoming ? http2::HDOP_STRIP_EARLY_DATA : 0) |
 526       ((req.http_major == 3 || req.http_major == 2)
 527            ? http2::HDOP_STRIP_SEC_WEBSOCKET_KEY
 528            : 0);
 529
 530   http2::build_http1_headers_from_headers(buf, req.fs.headers(), build_flags);
 531
 532   auto cookie = downstream_->assemble_request_cookie();
 533   if (!cookie.empty()) {
 534     buf->append("Cookie: ");
 535     buf->append(cookie);
 536     buf->append("\r\n");
 537   }
 538
 539   // set transfer-encoding only when content-length is unknown and
 540   // request body is expected.
 541   if (req.method != HTTP_CONNECT && req.http2_expect_body &&
 542       req.fs.content_length == -1) {
 543     downstream_->set_chunked_request(true);
 544     buf->append("Transfer-Encoding: chunked\r\n");
 545   }
 546
 547   if (req.connect_proto == ConnectProto::WEBSOCKET) {
 548     if (req.http_major == 3 || req.http_major == 2) {
 549       std::array<uint8_t, 16> nonce;
 550       if (RAND_bytes(nonce.data(), nonce.size()) != 1) {
 551         return -1;
 552       }
 553       auto iov = make_byte_ref(balloc, base64::encode_length(nonce.size()) + 1);
 554       auto p = base64::encode(std::begin(nonce), std::end(nonce), iov.base);
 555       *p = '\0';
 556       auto key = StringRef{iov.base, p};
 557       downstream_->set_ws_key(key);
 558
 559       buf->append("Sec-Websocket-Key: ");
 560       buf->append(key);
 561       buf->append("\r\n");
 562     }
 563
 564     buf->append("Upgrade: websocket\r\nConnection: Upgrade\r\n");
 565   } else if (!connect_method && req.upgrade_request) {
 566     auto connection = req.fs.header(http2::HD_CONNECTION);
 567     if (connection) {
 568       buf->append("Connection: ");
 569       buf->append((*connection).value);
 570       buf->append("\r\n");
 571     }
 572
 573     auto upgrade = req.fs.header(http2::HD_UPGRADE);
 574     if (upgrade) {
 575       buf->append("Upgrade: ");
 576       buf->append((*upgrade).value);
 577       buf->append("\r\n");
 578     }
 579   } else if (req.connection_close) {
 580     buf->append("Connection: close\r\n");
 581   }
 582
 583   auto upstream = downstream_->get_upstream();
 584   auto handler = upstream->get_client_handler();
 585
 586 #if OPENSSL_1_1_1_API
 587   auto conn = handler->get_connection();
 588
 589   if (conn->tls.ssl && !SSL_is_init_finished(conn->tls.ssl)) {
 590     buf->append("Early-Data: 1\r\n");
 591   }
 592 #endif // OPENSSL_1_1_1_API
 593
 594   auto fwd =
 595       fwdconf.strip_incoming ? nullptr : req.fs.header(http2::HD_FORWARDED);
 596
 597   if (fwdconf.params) {
 598     auto params = fwdconf.params;
 599
 600     if (config->http2_proxy || connect_method) {
 601       params &= ~FORWARDED_PROTO;
 602     }
 603
 604     auto value = http::create_forwarded(
 605         balloc, params, handler->get_forwarded_by(),
 606         handler->get_forwarded_for(), req.authority, req.scheme);
 607
 608     if (fwd || !value.empty()) {
 609       buf->append("Forwarded: ");
 610       if (fwd) {
 611         buf->append(fwd->value);
 612
 613         if (!value.empty()) {
 614           buf->append(", ");
 615         }
 616       }
 617       buf->append(value);
 618       buf->append("\r\n");
 619     }
 620   } else if (fwd) {
 621     buf->append("Forwarded: ");
 622     buf->append(fwd->value);
 623     buf->append("\r\n");
 624   }
 625
 626   auto xff = xffconf.strip_incoming ? nullptr
 627                                     : req.fs.header(http2::HD_X_FORWARDED_FOR);
 628
 629   if (xffconf.add) {
 630     buf->append("X-Forwarded-For: ");
 631     if (xff) {
 632       buf->append((*xff).value);
 633       buf->append(", ");
 634     }
 635     buf->append(client_handler_->get_ipaddr());
 636     buf->append("\r\n");
 637   } else if (xff) {
 638     buf->append("X-Forwarded-For: ");
 639     buf->append((*xff).value);
 640     buf->append("\r\n");
 641   }
 642   if (!config->http2_proxy && !connect_method) {
 643     auto xfp = xfpconf.strip_incoming
 644                    ? nullptr
 645                    : req.fs.header(http2::HD_X_FORWARDED_PROTO);
 646
 647     if (xfpconf.add) {
 648       buf->append("X-Forwarded-Proto: ");
 649       if (xfp) {
 650         buf->append((*xfp).value);
 651         buf->append(", ");
 652       }
 653       assert(!req.scheme.empty());
 654       buf->append(req.scheme);
 655       buf->append("\r\n");
 656     } else if (xfp) {
 657       buf->append("X-Forwarded-Proto: ");
 658       buf->append((*xfp).value);
 659       buf->append("\r\n");
 660     }
 661   }
 662   auto via = req.fs.header(http2::HD_VIA);
 663   if (httpconf.no_via) {
 664     if (via) {
 665       buf->append("Via: ");
 666       buf->append((*via).value);
 667       buf->append("\r\n");
 668     }
 669   } else {
 670     buf->append("Via: ");
 671     if (via) {
 672       buf->append((*via).value);
 673       buf->append(", ");
 674     }
 675     std::array<char, 16> viabuf;
 676     auto end = http::create_via_header_value(viabuf.data(), req.http_major,
 677                                              req.http_minor);
 678     buf->append(viabuf.data(), end - viabuf.data());
 679     buf->append("\r\n");
 680   }
 681
 682   for (auto &p : httpconf.add_request_headers) {
 683     buf->append(p.name);
 684     buf->append(": ");
 685     buf->append(p.value);
 686     buf->append("\r\n");
 687   }
 688
 689   buf->append("\r\n");
 690
 691   if (LOG_ENABLED(INFO)) {
 692     std::string nhdrs;
 693     for (auto chunk = buf->head; chunk; chunk = chunk->next) {
 694       nhdrs.append(chunk->pos, chunk->last);
 695     }
 696     if (log_config()->errorlog_tty) {
 697       nhdrs = http::colorizeHeaders(nhdrs.c_str());
 698     }
 699     DCLOG(INFO, this) << "HTTP request headers. stream_id="
 700                       << downstream_->get_stream_id() << "\n"
 701                       << nhdrs;
 702   }
 703
 704   // Don't call signal_write() if we anticipate request body.  We call
 705   // signal_write() when we received request body chunk, and it
 706   // enables us to send headers and data in one writev system call.
 707   if (req.method == HTTP_CONNECT ||
 708       downstream_->get_blocked_request_buf()->rleft() ||
 709       (!req.http2_expect_body && req.fs.content_length == 0) ||
 710       downstream_->get_expect_100_continue()) {
 711     signal_write();
 712   }
 713
 714   return 0;
 715 }
 716
 717 int HttpDownstreamConnection::process_blocked_request_buf() {
 718   auto src = downstream_->get_blocked_request_buf();
 719
 720   if (src->rleft()) {
 721     auto dest = downstream_->get_request_buf();
 722     auto chunked = downstream_->get_chunked_request();
 723     if (chunked) {
 724       auto chunk_size_hex = util::utox(src->rleft());
 725       dest->append(chunk_size_hex);
 726       dest->append("\r\n");
 727     }
 728
 729     src->copy(*dest);
 730
 731     if (chunked) {
 732       dest->append("\r\n");
 733     }
 734   }
 735
 736   if (downstream_->get_blocked_request_data_eof() &&
 737       downstream_->get_chunked_request()) {
 738     end_upload_data_chunk();
 739   }
 740
 741   return 0;
 742 }
 743
 744 int HttpDownstreamConnection::push_upload_data_chunk(const uint8_t *data,
 745                                                      size_t datalen) {
 746   if (!downstream_->get_request_header_sent()) {
 747     auto output = downstream_->get_blocked_request_buf();
 748     auto &req = downstream_->request();
 749     output->append(data, datalen);
 750     req.unconsumed_body_length += datalen;
 751     if (request_header_written_) {
 752       signal_write();
 753     }
 754     return 0;
 755   }
 756
 757   auto chunked = downstream_->get_chunked_request();
 758   auto output = downstream_->get_request_buf();
 759
 760   if (chunked) {
 761     auto chunk_size_hex = util::utox(datalen);
 762     output->append(chunk_size_hex);
 763     output->append("\r\n");
 764   }
 765
 766   output->append(data, datalen);
 767
 768   if (chunked) {
 769     output->append("\r\n");
 770   }
 771
 772   signal_write();
 773
 774   return 0;
 775 }
 776
 777 int HttpDownstreamConnection::end_upload_data() {
 778   if (!downstream_->get_request_header_sent()) {
 779     downstream_->set_blocked_request_data_eof(true);
 780     if (request_header_written_) {
 781       signal_write();
 782     }
 783     return 0;
 784   }
 785
 786   signal_write();
 787
 788   if (!downstream_->get_chunked_request()) {
 789     return 0;
 790   }
 791
 792   end_upload_data_chunk();
 793
 794   return 0;
 795 }
 796
 797 void HttpDownstreamConnection::end_upload_data_chunk() {
 798   const auto &req = downstream_->request();
 799
 800   auto output = downstream_->get_request_buf();
 801   const auto &trailers = req.fs.trailers();
 802   if (trailers.empty()) {
 803     output->append("0\r\n\r\n");
 804   } else {
 805     output->append("0\r\n");
 806     http2::build_http1_headers_from_headers(output, trailers,
 807                                             http2::HDOP_STRIP_ALL);
 808     output->append("\r\n");
 809   }
 810 }
 811
 812 namespace {
 813 void remove_from_pool(HttpDownstreamConnection *dconn) {
 814   auto addr = dconn->get_addr();
 815   auto &dconn_pool = addr->dconn_pool;
 816   dconn_pool->remove_downstream_connection(dconn);
 817 }
 818 } // namespace
 819
 820 namespace {
 821 void idle_readcb(struct ev_loop *loop, ev_io *w, int revents) {
 822   auto conn = static_cast<Connection *>(w->data);
 823   auto dconn = static_cast<HttpDownstreamConnection *>(conn->data);
 824   if (LOG_ENABLED(INFO)) {
 825     DCLOG(INFO, dconn) << "Idle connection EOF";
 826   }
 827
 828   remove_from_pool(dconn);
 829   // dconn was deleted
 830 }
 831 } // namespace
 832
 833 namespace {
 834 void idle_timeoutcb(struct ev_loop *loop, ev_timer *w, int revents) {
 835   auto conn = static_cast<Connection *>(w->data);
 836   auto dconn = static_cast<HttpDownstreamConnection *>(conn->data);
 837
 838   if (w == &conn->rt && !conn->expired_rt()) {
 839     return;
 840   }
 841
 842   if (LOG_ENABLED(INFO)) {
 843     DCLOG(INFO, dconn) << "Idle connection timeout";
 844   }
 845
 846   remove_from_pool(dconn);
 847   // dconn was deleted
 848 }
 849 } // namespace
 850
 851 void HttpDownstreamConnection::detach_downstream(Downstream *downstream) {
 852   if (LOG_ENABLED(INFO)) {
 853     DCLOG(INFO, this) << "Detaching from DOWNSTREAM:" << downstream;
 854   }
 855   downstream_ = nullptr;
 856
 857   ev_set_cb(&conn_.rev, idle_readcb);
 858   ioctrl_.force_resume_read();
 859
 860   auto &downstreamconf = *worker_->get_downstream_config();
 861
 862   ev_set_cb(&conn_.rt, idle_timeoutcb);
 863   if (conn_.read_timeout < downstreamconf.timeout.idle_read) {
 864     conn_.read_timeout = downstreamconf.timeout.idle_read;
 865     conn_.last_read = ev_now(conn_.loop);
 866   } else {
 867     conn_.again_rt(downstreamconf.timeout.idle_read);
 868   }
 869
 870   conn_.wlimit.stopw();
 871   ev_timer_stop(conn_.loop, &conn_.wt);
 872 }
 873
 874 void HttpDownstreamConnection::pause_read(IOCtrlReason reason) {
 875   ioctrl_.pause_read(reason);
 876 }
 877
 878 int HttpDownstreamConnection::resume_read(IOCtrlReason reason,
 879                                           size_t consumed) {
 880   auto &downstreamconf = *worker_->get_downstream_config();
 881
 882   if (downstream_->get_response_buf()->rleft() <=
 883       downstreamconf.request_buffer_size / 2) {
 884     ioctrl_.resume_read(reason);
 885   }
 886
 887   return 0;
 888 }
 889
 890 void HttpDownstreamConnection::force_resume_read() {
 891   ioctrl_.force_resume_read();
 892 }
 893
 894 namespace {
 895 int htp_msg_begincb(llhttp_t *htp) {
 896   auto downstream = static_cast<Downstream *>(htp->data);
 897
 898   if (downstream->get_response_state() != DownstreamState::INITIAL) {
 899     return -1;
 900   }
 901
 902   return 0;
 903 }
 904 } // namespace
 905
 906 namespace {
 907 int htp_hdrs_completecb(llhttp_t *htp) {
 908   auto downstream = static_cast<Downstream *>(htp->data);
 909   auto upstream = downstream->get_upstream();
 910   auto handler = upstream->get_client_handler();
 911   const auto &req = downstream->request();
 912   auto &resp = downstream->response();
 913   int rv;
 914
 915   auto config = get_config();
 916   auto &loggingconf = config->logging;
 917
 918   resp.http_status = htp->status_code;
 919   resp.http_major = htp->http_major;
 920   resp.http_minor = htp->http_minor;
 921
 922   if (resp.http_major > 1 || req.http_minor > 1) {
 923     resp.http_major = 1;
 924     resp.http_minor = 1;
 925     return -1;
 926   }
 927
 928   auto dconn = downstream->get_downstream_connection();
 929
 930   downstream->set_downstream_addr_group(dconn->get_downstream_addr_group());
 931   downstream->set_addr(dconn->get_addr());
 932
 933   // Server MUST NOT send Transfer-Encoding with a status code 1xx or
 934   // 204.  Also server MUST NOT send Transfer-Encoding with a status
 935   // code 2xx to a CONNECT request.  Same holds true with
 936   // Content-Length.
 937   if (resp.http_status == 204) {
 938     if (resp.fs.header(http2::HD_TRANSFER_ENCODING)) {
 939       return -1;
 940     }
 941     // Some server send content-length: 0 for 204.  Until they get
 942     // fixed, we accept, but ignore it.
 943
 944     // Calling parse_content_length() detects duplicated
 945     // content-length header fields.
 946     if (resp.fs.parse_content_length() != 0) {
 947       return -1;
 948     }
 949     if (resp.fs.content_length == 0) {
 950       resp.fs.erase_content_length_and_transfer_encoding();
 951     } else if (resp.fs.content_length != -1) {
 952       return -1;
 953     }
 954   } else if (resp.http_status / 100 == 1 ||
 955              (resp.http_status / 100 == 2 && req.method == HTTP_CONNECT)) {
 956     // Server MUST NOT send Content-Length and Transfer-Encoding in
 957     // these responses.
 958     resp.fs.erase_content_length_and_transfer_encoding();
 959   } else if (resp.fs.parse_content_length() != 0) {
 960     downstream->set_response_state(DownstreamState::MSG_BAD_HEADER);
 961     return -1;
 962   }
 963
 964   // Check upgrade before processing non-final response, since if
 965   // upgrade succeeded, 101 response is treated as final in nghttpx.
 966   downstream->check_upgrade_fulfilled_http1();
 967
 968   if (downstream->get_non_final_response()) {
 969     // Reset content-length because we reuse same Downstream for the
 970     // next response.
 971     resp.fs.content_length = -1;
 972     // For non-final response code, we just call
 973     // on_downstream_header_complete() without changing response
 974     // state.
 975     rv = upstream->on_downstream_header_complete(downstream);
 976
 977     if (rv != 0) {
 978       return -1;
 979     }
 980
 981     // Ignore response body for non-final response.
 982     return 1;
 983   }
 984
 985   resp.connection_close = !llhttp_should_keep_alive(htp);
 986   downstream->set_response_state(DownstreamState::HEADER_COMPLETE);
 987   downstream->inspect_http1_response();
 988   if (downstream->get_upgraded()) {
 989     // content-length must be ignored for upgraded connection.
 990     resp.fs.content_length = -1;
 991     resp.connection_close = true;
 992     // transfer-encoding not applied to upgraded connection
 993     downstream->set_chunked_response(false);
 994   } else if (http2::legacy_http1(req.http_major, req.http_minor)) {
 995     if (resp.fs.content_length == -1) {
 996       resp.connection_close = true;
 997     }
 998     downstream->set_chunked_response(false);
 999   } else if (!downstream->expect_response_body()) {
1000     downstream->set_chunked_response(false);
1001   }
1002
1003   if (loggingconf.access.write_early && downstream->accesslog_ready()) {
1004     handler->write_accesslog(downstream);
1005     downstream->set_accesslog_written(true);
1006   }
1007
1008   if (upstream->on_downstream_header_complete(downstream) != 0) {
1009     return -1;
1010   }
1011
1012   if (downstream->get_upgraded()) {
1013     // Upgrade complete, read until EOF in both ends
1014     if (upstream->resume_read(SHRPX_NO_BUFFER, downstream, 0) != 0) {
1015       return -1;
1016     }
1017     downstream->set_request_state(DownstreamState::HEADER_COMPLETE);
1018     if (LOG_ENABLED(INFO)) {
1019       LOG(INFO) << "HTTP upgrade success. stream_id="
1020                 << downstream->get_stream_id();
1021     }
1022   }
1023
1024   // Ignore the response body. HEAD response may contain
1025   // Content-Length or Transfer-Encoding: chunked.  Some server send
1026   // 304 status code with nonzero Content-Length, but without response
1027   // body. See
1028   // https://tools.ietf.org/html/rfc7230#section-3.3
1029
1030   // TODO It seems that the cases other than HEAD are handled by
1031   // llhttp.  Need test.
1032   return !http2::expect_response_body(req.method, resp.http_status);
1033 }
1034 } // namespace
1035
1036 namespace {
1037 int ensure_header_field_buffer(const Downstream *downstream,
1038                                const HttpConfig &httpconf, size_t len) {
1039   auto &resp = downstream->response();
1040
1041   if (resp.fs.buffer_size() + len > httpconf.response_header_field_buffer) {
1042     if (LOG_ENABLED(INFO)) {
1043       DLOG(INFO, downstream) << "Too large header header field size="
1044                              << resp.fs.buffer_size() + len;
1045     }
1046     return -1;
1047   }
1048
1049   return 0;
1050 }
1051 } // namespace
1052
1053 namespace {
1054 int ensure_max_header_fields(const Downstream *downstream,
1055                              const HttpConfig &httpconf) {
1056   auto &resp = downstream->response();
1057
1058   if (resp.fs.num_fields() >= httpconf.max_response_header_fields) {
1059     if (LOG_ENABLED(INFO)) {
1060       DLOG(INFO, downstream)
1061           << "Too many header field num=" << resp.fs.num_fields() + 1;
1062     }
1063     return -1;
1064   }
1065
1066   return 0;
1067 }
1068 } // namespace
1069
1070 namespace {
1071 int htp_hdr_keycb(llhttp_t *htp, const char *data, size_t len) {
1072   auto downstream = static_cast<Downstream *>(htp->data);
1073   auto &resp = downstream->response();
1074   auto &httpconf = get_config()->http;
1075
1076   if (ensure_header_field_buffer(downstream, httpconf, len) != 0) {
1077     return -1;
1078   }
1079
1080   if (downstream->get_response_state() == DownstreamState::INITIAL) {
1081     if (resp.fs.header_key_prev()) {
1082       resp.fs.append_last_header_key(data, len);
1083     } else {
1084       if (ensure_max_header_fields(downstream, httpconf) != 0) {
1085         return -1;
1086       }
1087       resp.fs.alloc_add_header_name(StringRef{data, len});
1088     }
1089   } else {
1090     // trailer part
1091     if (resp.fs.trailer_key_prev()) {
1092       resp.fs.append_last_trailer_key(data, len);
1093     } else {
1094       if (ensure_max_header_fields(downstream, httpconf) != 0) {
1095         // Could not ignore this trailer field easily, since we may
1096         // get its value in htp_hdr_valcb, and it will be added to
1097         // wrong place or crash if trailer fields are currently empty.
1098         return -1;
1099       }
1100       resp.fs.alloc_add_trailer_name(StringRef{data, len});
1101     }
1102   }
1103   return 0;
1104 }
1105 } // namespace
1106
1107 namespace {
1108 int htp_hdr_valcb(llhttp_t *htp, const char *data, size_t len) {
1109   auto downstream = static_cast<Downstream *>(htp->data);
1110   auto &resp = downstream->response();
1111   auto &httpconf = get_config()->http;
1112
1113   if (ensure_header_field_buffer(downstream, httpconf, len) != 0) {
1114     return -1;
1115   }
1116
1117   if (downstream->get_response_state() == DownstreamState::INITIAL) {
1118     resp.fs.append_last_header_value(data, len);
1119   } else {
1120     resp.fs.append_last_trailer_value(data, len);
1121   }
1122   return 0;
1123 }
1124 } // namespace
1125
1126 namespace {
1127 int htp_bodycb(llhttp_t *htp, const char *data, size_t len) {
1128   auto downstream = static_cast<Downstream *>(htp->data);
1129   auto &resp = downstream->response();
1130
1131   resp.recv_body_length += len;
1132
1133   return downstream->get_upstream()->on_downstream_body(
1134       downstream, reinterpret_cast<const uint8_t *>(data), len, true);
1135 }
1136 } // namespace
1137
1138 namespace {
1139 int htp_msg_completecb(llhttp_t *htp) {
1140   auto downstream = static_cast<Downstream *>(htp->data);
1141
1142   // llhttp does not treat "200 connection established" response
1143   // against CONNECT request, and in that case, this function is not
1144   // called.  But if HTTP Upgrade is made (e.g., WebSocket), this
1145   // function is called, and llhttp_execute() returns just after that.
1146   if (downstream->get_upgraded()) {
1147     return 0;
1148   }
1149
1150   if (downstream->get_non_final_response()) {
1151     downstream->reset_response();
1152
1153     return 0;
1154   }
1155
1156   downstream->set_response_state(DownstreamState::MSG_COMPLETE);
1157   // Block reading another response message from (broken?)
1158   // server. This callback is not called if the connection is
1159   // tunneled.
1160   downstream->pause_read(SHRPX_MSG_BLOCK);
1161   return downstream->get_upstream()->on_downstream_body_complete(downstream);
1162 }
1163 } // namespace
1164
1165 int HttpDownstreamConnection::write_first() {
1166   int rv;
1167
1168   process_blocked_request_buf();
1169
1170   if (conn_.tls.ssl) {
1171     rv = write_tls();
1172   } else {
1173     rv = write_clear();
1174   }
1175
1176   if (rv != 0) {
1177     return SHRPX_ERR_RETRY;
1178   }
1179
1180   if (conn_.tls.ssl) {
1181     on_write_ = &HttpDownstreamConnection::write_tls;
1182   } else {
1183     on_write_ = &HttpDownstreamConnection::write_clear;
1184   }
1185
1186   first_write_done_ = true;
1187   downstream_->set_request_header_sent(true);
1188
1189   auto buf = downstream_->get_blocked_request_buf();
1190   buf->reset();
1191
1192   // upstream->resume_read() might be called in
1193   // write_tls()/write_clear(), but before blocked_request_buf_ is
1194   // reset.  So upstream read might still be blocked.  Let's do it
1195   // again here.
1196   auto input = downstream_->get_request_buf();
1197   if (input->rleft() == 0) {
1198     auto upstream = downstream_->get_upstream();
1199     auto &req = downstream_->request();
1200
1201     upstream->resume_read(SHRPX_NO_BUFFER, downstream_,
1202                           req.unconsumed_body_length);
1203   }
1204
1205   return 0;
1206 }
1207
1208 int HttpDownstreamConnection::read_clear() {
1209   conn_.last_read = ev_now(conn_.loop);
1210
1211   std::array<uint8_t, 16_k> buf;
1212   int rv;
1213
1214   for (;;) {
1215     auto nread = conn_.read_clear(buf.data(), buf.size());
1216     if (nread == 0) {
1217       return 0;
1218     }
1219
1220     if (nread < 0) {
1221       if (nread == SHRPX_ERR_EOF && !downstream_->get_upgraded()) {
1222         auto htperr = llhttp_finish(&response_htp_);
1223         if (htperr != HPE_OK) {
1224           if (LOG_ENABLED(INFO)) {
1225             DCLOG(INFO, this) << "HTTP response ended prematurely: "
1226                               << llhttp_errno_name(htperr);
1227           }
1228
1229           return -1;
1230         }
1231       }
1232
1233       return nread;
1234     }
1235
1236     rv = process_input(buf.data(), nread);
1237     if (rv != 0) {
1238       return rv;
1239     }
1240
1241     if (!ev_is_active(&conn_.rev)) {
1242       return 0;
1243     }
1244   }
1245 }
1246
1247 int HttpDownstreamConnection::write_clear() {
1248   conn_.last_read = ev_now(conn_.loop);
1249
1250   auto upstream = downstream_->get_upstream();
1251   auto input = downstream_->get_request_buf();
1252
1253   std::array<struct iovec, MAX_WR_IOVCNT> iov;
1254
1255   while (input->rleft() > 0) {
1256     auto iovcnt = input->riovec(iov.data(), iov.size());
1257
1258     auto nwrite = conn_.writev_clear(iov.data(), iovcnt);
1259
1260     if (nwrite == 0) {
1261       return 0;
1262     }
1263
1264     if (nwrite < 0) {
1265       if (!first_write_done_) {
1266         return nwrite;
1267       }
1268       // We may have pending data in receive buffer which may contain
1269       // part of response body.  So keep reading.  Invoke read event
1270       // to get read(2) error just in case.
1271       ev_feed_event(conn_.loop, &conn_.rev, EV_READ);
1272       on_write_ = &HttpDownstreamConnection::noop;
1273       reusable_ = false;
1274       break;
1275     }
1276
1277     input->drain(nwrite);
1278   }
1279
1280   conn_.wlimit.stopw();
1281   ev_timer_stop(conn_.loop, &conn_.wt);
1282
1283   if (input->rleft() == 0) {
1284     auto &req = downstream_->request();
1285
1286     upstream->resume_read(SHRPX_NO_BUFFER, downstream_,
1287                           req.unconsumed_body_length);
1288   }
1289
1290   return 0;
1291 }
1292
1293 int HttpDownstreamConnection::tls_handshake() {
1294   ERR_clear_error();
1295
1296   conn_.last_read = ev_now(conn_.loop);
1297
1298   auto rv = conn_.tls_handshake();
1299   if (rv == SHRPX_ERR_INPROGRESS) {
1300     return 0;
1301   }
1302
1303   if (rv < 0) {
1304     downstream_failure(addr_, raddr_);
1305
1306     return rv;
1307   }
1308
1309   if (LOG_ENABLED(INFO)) {
1310     DCLOG(INFO, this) << "SSL/TLS handshake completed";
1311   }
1312
1313   if (!get_config()->tls.insecure &&
1314       tls::check_cert(conn_.tls.ssl, addr_, raddr_) != 0) {
1315     downstream_failure(addr_, raddr_);
1316
1317     return -1;
1318   }
1319
1320   auto &connect_blocker = addr_->connect_blocker;
1321
1322   signal_write_ = &HttpDownstreamConnection::actual_signal_write;
1323
1324   connect_blocker->on_success();
1325
1326   ev_set_cb(&conn_.rt, timeoutcb);
1327   ev_set_cb(&conn_.wt, timeoutcb);
1328
1329   on_read_ = &HttpDownstreamConnection::read_tls;
1330   on_write_ = &HttpDownstreamConnection::write_first;
1331
1332   // TODO Check negotiated ALPN
1333
1334   return on_write();
1335 }
1336
1337 int HttpDownstreamConnection::read_tls() {
1338   conn_.last_read = ev_now(conn_.loop);
1339
1340   ERR_clear_error();
1341
1342   std::array<uint8_t, 16_k> buf;
1343   int rv;
1344
1345   for (;;) {
1346     auto nread = conn_.read_tls(buf.data(), buf.size());
1347     if (nread == 0) {
1348       return 0;
1349     }
1350
1351     if (nread < 0) {
1352       if (nread == SHRPX_ERR_EOF && !downstream_->get_upgraded()) {
1353         auto htperr = llhttp_finish(&response_htp_);
1354         if (htperr != HPE_OK) {
1355           if (LOG_ENABLED(INFO)) {
1356             DCLOG(INFO, this) << "HTTP response ended prematurely: "
1357                               << llhttp_errno_name(htperr);
1358           }
1359
1360           return -1;
1361         }
1362       }
1363
1364       return nread;
1365     }
1366
1367     rv = process_input(buf.data(), nread);
1368     if (rv != 0) {
1369       return rv;
1370     }
1371
1372     if (!ev_is_active(&conn_.rev)) {
1373       return 0;
1374     }
1375   }
1376 }
1377
1378 int HttpDownstreamConnection::write_tls() {
1379   conn_.last_read = ev_now(conn_.loop);
1380
1381   ERR_clear_error();
1382
1383   auto upstream = downstream_->get_upstream();
1384   auto input = downstream_->get_request_buf();
1385
1386   struct iovec iov;
1387
1388   while (input->rleft() > 0) {
1389     auto iovcnt = input->riovec(&iov, 1);
1390     if (iovcnt != 1) {
1391       assert(0);
1392       return -1;
1393     }
1394     auto nwrite = conn_.write_tls(iov.iov_base, iov.iov_len);
1395
1396     if (nwrite == 0) {
1397       return 0;
1398     }
1399
1400     if (nwrite < 0) {
1401       if (!first_write_done_) {
1402         return nwrite;
1403       }
1404       // We may have pending data in receive buffer which may contain
1405       // part of response body.  So keep reading.  Invoke read event
1406       // to get read(2) error just in case.
1407       ev_feed_event(conn_.loop, &conn_.rev, EV_READ);
1408       on_write_ = &HttpDownstreamConnection::noop;
1409       reusable_ = false;
1410       break;
1411     }
1412
1413     input->drain(nwrite);
1414   }
1415
1416   conn_.wlimit.stopw();
1417   ev_timer_stop(conn_.loop, &conn_.wt);
1418
1419   if (input->rleft() == 0) {
1420     auto &req = downstream_->request();
1421
1422     upstream->resume_read(SHRPX_NO_BUFFER, downstream_,
1423                           req.unconsumed_body_length);
1424   }
1425
1426   return 0;
1427 }
1428
1429 int HttpDownstreamConnection::process_input(const uint8_t *data,
1430                                             size_t datalen) {
1431   int rv;
1432
1433   if (downstream_->get_upgraded()) {
1434     // For upgraded connection, just pass data to the upstream.
1435     rv = downstream_->get_upstream()->on_downstream_body(downstream_, data,
1436                                                          datalen, true);
1437     if (rv != 0) {
1438       return rv;
1439     }
1440
1441     if (downstream_->response_buf_full()) {
1442       downstream_->pause_read(SHRPX_NO_BUFFER);
1443       return 0;
1444     }
1445
1446     return 0;
1447   }
1448
1449   auto htperr = llhttp_execute(&response_htp_,
1450                                reinterpret_cast<const char *>(data), datalen);
1451   auto nproc =
1452       htperr == HPE_OK
1453           ? datalen
1454           : static_cast<size_t>(reinterpret_cast<const uint8_t *>(
1455                                     llhttp_get_error_pos(&response_htp_)) -
1456                                 data);
1457
1458   if (htperr != HPE_OK &&
1459       (!downstream_->get_upgraded() || htperr != HPE_PAUSED_UPGRADE)) {
1460     // Handling early return (in other words, response was hijacked by
1461     // mruby scripting).
1462     if (downstream_->get_response_state() == DownstreamState::MSG_COMPLETE) {
1463       return SHRPX_ERR_DCONN_CANCELED;
1464     }
1465
1466     if (LOG_ENABLED(INFO)) {
1467       DCLOG(INFO, this) << "HTTP parser failure: "
1468                         << "(" << llhttp_errno_name(htperr) << ") "
1469                         << llhttp_get_error_reason(&response_htp_);
1470     }
1471
1472     return -1;
1473   }
1474
1475   if (downstream_->get_upgraded()) {
1476     if (nproc < datalen) {
1477       // Data from data + nproc are for upgraded protocol.
1478       rv = downstream_->get_upstream()->on_downstream_body(
1479           downstream_, data + nproc, datalen - nproc, true);
1480       if (rv != 0) {
1481         return rv;
1482       }
1483
1484       if (downstream_->response_buf_full()) {
1485         downstream_->pause_read(SHRPX_NO_BUFFER);
1486         return 0;
1487       }
1488     }
1489     return 0;
1490   }
1491
1492   if (downstream_->response_buf_full()) {
1493     downstream_->pause_read(SHRPX_NO_BUFFER);
1494     return 0;
1495   }
1496
1497   return 0;
1498 }
1499
1500 int HttpDownstreamConnection::connected() {
1501   auto &connect_blocker = addr_->connect_blocker;
1502
1503   auto sock_error = util::get_socket_error(conn_.fd);
1504   if (sock_error != 0) {
1505     conn_.wlimit.stopw();
1506
1507     DCLOG(WARN, this) << "Backend connect failed; addr="
1508                       << util::to_numeric_addr(raddr_)
1509                       << ": errno=" << sock_error;
1510
1511     downstream_failure(addr_, raddr_);
1512
1513     return -1;
1514   }
1515
1516   if (LOG_ENABLED(INFO)) {
1517     DCLOG(INFO, this) << "Connected to downstream host";
1518   }
1519
1520   // Reset timeout for write.  Previously, we set timeout for connect.
1521   conn_.wt.repeat = group_->shared_addr->timeout.write;
1522   ev_timer_again(conn_.loop, &conn_.wt);
1523
1524   conn_.rlimit.startw();
1525   conn_.again_rt();
1526
1527   ev_set_cb(&conn_.wev, writecb);
1528
1529   if (conn_.tls.ssl) {
1530     on_read_ = &HttpDownstreamConnection::tls_handshake;
1531     on_write_ = &HttpDownstreamConnection::tls_handshake;
1532
1533     return 0;
1534   }
1535
1536   signal_write_ = &HttpDownstreamConnection::actual_signal_write;
1537
1538   connect_blocker->on_success();
1539
1540   ev_set_cb(&conn_.rt, timeoutcb);
1541   ev_set_cb(&conn_.wt, timeoutcb);
1542
1543   on_read_ = &HttpDownstreamConnection::read_clear;
1544   on_write_ = &HttpDownstreamConnection::write_first;
1545
1546   return 0;
1547 }
1548
1549 int HttpDownstreamConnection::on_read() { return on_read_(*this); }
1550
1551 int HttpDownstreamConnection::on_write() { return on_write_(*this); }
1552
1553 void HttpDownstreamConnection::on_upstream_change(Upstream *upstream) {}
1554
1555 void HttpDownstreamConnection::signal_write() { signal_write_(*this); }
1556
1557 int HttpDownstreamConnection::actual_signal_write() {
1558   ev_feed_event(conn_.loop, &conn_.wev, EV_WRITE);
1559   return 0;
1560 }
1561
1562 int HttpDownstreamConnection::noop() { return 0; }
1563
1564 const std::shared_ptr<DownstreamAddrGroup> &
1565 HttpDownstreamConnection::get_downstream_addr_group() const {
1566   return group_;
1567 }
1568
1569 DownstreamAddr *HttpDownstreamConnection::get_addr() const { return addr_; }
1570
1571 bool HttpDownstreamConnection::poolable() const {
1572   return !group_->retired && reusable_;
1573 }
1574
1575 const Address *HttpDownstreamConnection::get_raddr() const { return raddr_; }
1576
1577 } // namespace shrpx