summaryrefslogtreecommitdiffstats
path: root/src/shrpx_http_downstream_connection.cc
diff options
context:
space:
mode:
authorDaniel Baumann <daniel.baumann@progress-linux.org>2024-04-28 07:34:15 +0000
committerDaniel Baumann <daniel.baumann@progress-linux.org>2024-04-28 07:34:15 +0000
commitafea5f9539cbf1eeaa85ec77d79eb2f59724f470 (patch)
tree2a4eba394a6bc60d2eaa8304d91168a07225d51e /src/shrpx_http_downstream_connection.cc
parentInitial commit. (diff)
downloadnghttp2-afea5f9539cbf1eeaa85ec77d79eb2f59724f470.tar.xz
nghttp2-afea5f9539cbf1eeaa85ec77d79eb2f59724f470.zip
Adding upstream version 1.52.0.upstream/1.52.0upstream
Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>
Diffstat (limited to 'src/shrpx_http_downstream_connection.cc')
-rw-r--r--src/shrpx_http_downstream_connection.cc1602
1 files changed, 1602 insertions, 0 deletions
diff --git a/src/shrpx_http_downstream_connection.cc b/src/shrpx_http_downstream_connection.cc
new file mode 100644
index 0000000..96ce4b2
--- /dev/null
+++ b/src/shrpx_http_downstream_connection.cc
@@ -0,0 +1,1602 @@
+/*
+ * nghttp2 - HTTP/2 C Library
+ *
+ * Copyright (c) 2012 Tatsuhiro Tsujikawa
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining
+ * a copy of this software and associated documentation files (the
+ * "Software"), to deal in the Software without restriction, including
+ * without limitation the rights to use, copy, modify, merge, publish,
+ * distribute, sublicense, and/or sell copies of the Software, and to
+ * permit persons to whom the Software is furnished to do so, subject to
+ * the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be
+ * included in all copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
+ * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
+ * MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
+ * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE
+ * LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
+ * OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
+ * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ */
+#include "shrpx_http_downstream_connection.h"
+
+#include <openssl/rand.h>
+
+#include "shrpx_client_handler.h"
+#include "shrpx_upstream.h"
+#include "shrpx_downstream.h"
+#include "shrpx_config.h"
+#include "shrpx_error.h"
+#include "shrpx_http.h"
+#include "shrpx_log_config.h"
+#include "shrpx_connect_blocker.h"
+#include "shrpx_downstream_connection_pool.h"
+#include "shrpx_worker.h"
+#include "shrpx_http2_session.h"
+#include "shrpx_tls.h"
+#include "shrpx_log.h"
+#include "http2.h"
+#include "util.h"
+#include "ssl_compat.h"
+
+using namespace nghttp2;
+
+namespace shrpx {
+
+namespace {
+void timeoutcb(struct ev_loop *loop, ev_timer *w, int revents) {
+ auto conn = static_cast<Connection *>(w->data);
+ auto dconn = static_cast<HttpDownstreamConnection *>(conn->data);
+
+ if (w == &conn->rt && !conn->expired_rt()) {
+ return;
+ }
+
+ if (LOG_ENABLED(INFO)) {
+ DCLOG(INFO, dconn) << "Time out";
+ }
+
+ auto downstream = dconn->get_downstream();
+ auto upstream = downstream->get_upstream();
+ auto handler = upstream->get_client_handler();
+ auto &resp = downstream->response();
+
+ // Do this so that dconn is not pooled
+ resp.connection_close = true;
+
+ if (upstream->downstream_error(dconn, Downstream::EVENT_TIMEOUT) != 0) {
+ delete handler;
+ }
+}
+} // namespace
+
+namespace {
+void retry_downstream_connection(Downstream *downstream,
+ unsigned int status_code) {
+ auto upstream = downstream->get_upstream();
+ auto handler = upstream->get_client_handler();
+
+ assert(!downstream->get_request_header_sent());
+
+ downstream->add_retry();
+
+ if (downstream->no_more_retry()) {
+ delete handler;
+ return;
+ }
+
+ downstream->pop_downstream_connection();
+ auto buf = downstream->get_request_buf();
+ buf->reset();
+
+ int rv;
+
+ for (;;) {
+ auto ndconn = handler->get_downstream_connection(rv, downstream);
+ if (!ndconn) {
+ break;
+ }
+ if (downstream->attach_downstream_connection(std::move(ndconn)) != 0) {
+ continue;
+ }
+ if (downstream->push_request_headers() == 0) {
+ return;
+ }
+ }
+
+ downstream->set_request_state(DownstreamState::CONNECT_FAIL);
+
+ if (rv == SHRPX_ERR_TLS_REQUIRED) {
+ rv = upstream->on_downstream_abort_request_with_https_redirect(downstream);
+ } else {
+ rv = upstream->on_downstream_abort_request(downstream, status_code);
+ }
+
+ if (rv != 0) {
+ delete handler;
+ }
+}
+} // namespace
+
+namespace {
+void connect_timeoutcb(struct ev_loop *loop, ev_timer *w, int revents) {
+ auto conn = static_cast<Connection *>(w->data);
+ auto dconn = static_cast<HttpDownstreamConnection *>(conn->data);
+ auto addr = dconn->get_addr();
+ auto raddr = dconn->get_raddr();
+
+ DCLOG(WARN, dconn) << "Connect time out; addr="
+ << util::to_numeric_addr(raddr);
+
+ downstream_failure(addr, raddr);
+
+ auto downstream = dconn->get_downstream();
+
+ retry_downstream_connection(downstream, 504);
+}
+} // namespace
+
+namespace {
+void backend_retry(Downstream *downstream) {
+ retry_downstream_connection(downstream, 502);
+}
+} // namespace
+
+namespace {
+void readcb(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_read(dconn);
+ if (rv != 0) {
+ if (rv == SHRPX_ERR_RETRY) {
+ backend_retry(downstream);
+ return;
+ }
+
+ 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);
+}
+} // namespace
+
+HttpDownstreamConnection::HttpDownstreamConnection(
+ const std::shared_ptr<DownstreamAddrGroup> &group, DownstreamAddr *addr,
+ struct ev_loop *loop, Worker *worker)
+ : conn_(loop, -1, nullptr, worker->get_mcpool(),
+ group->shared_addr->timeout.write, group->shared_addr->timeout.read,
+ {}, {}, connectcb, readcb, connect_timeoutcb, this,
+ get_config()->tls.dyn_rec.warmup_threshold,
+ get_config()->tls.dyn_rec.idle_timeout, Proto::HTTP1),
+ on_read_(&HttpDownstreamConnection::noop),
+ on_write_(&HttpDownstreamConnection::noop),
+ signal_write_(&HttpDownstreamConnection::noop),
+ worker_(worker),
+ ssl_ctx_(worker->get_cl_ssl_ctx()),
+ group_(group),
+ addr_(addr),
+ raddr_(nullptr),
+ ioctrl_(&conn_.rlimit),
+ response_htp_{0},
+ first_write_done_(false),
+ reusable_(true),
+ request_header_written_(false) {}
+
+HttpDownstreamConnection::~HttpDownstreamConnection() {
+ if (LOG_ENABLED(INFO)) {
+ DCLOG(INFO, this) << "Deleted";
+ }
+
+ if (dns_query_) {
+ auto dns_tracker = worker_->get_dns_tracker();
+ dns_tracker->cancel(dns_query_.get());
+ }
+}
+
+int HttpDownstreamConnection::attach_downstream(Downstream *downstream) {
+ int rv;
+
+ if (LOG_ENABLED(INFO)) {
+ DCLOG(INFO, this) << "Attaching to DOWNSTREAM:" << downstream;
+ }
+
+ downstream_ = downstream;
+
+ rv = initiate_connection();
+ if (rv != 0) {
+ downstream_ = nullptr;
+ return rv;
+ }
+
+ return 0;
+}
+
+namespace {
+int htp_msg_begincb(llhttp_t *htp);
+int htp_hdr_keycb(llhttp_t *htp, const char *data, size_t len);
+int htp_hdr_valcb(llhttp_t *htp, const char *data, size_t len);
+int htp_hdrs_completecb(llhttp_t *htp);
+int htp_bodycb(llhttp_t *htp, const char *data, size_t len);
+int htp_msg_completecb(llhttp_t *htp);
+} // namespace
+
+namespace {
+constexpr llhttp_settings_t htp_hooks = {
+ htp_msg_begincb, // llhttp_cb on_message_begin;
+ nullptr, // llhttp_data_cb on_url;
+ nullptr, // llhttp_data_cb on_status;
+ nullptr, // llhttp_data_cb on_method;
+ nullptr, // llhttp_data_cb on_version;
+ htp_hdr_keycb, // llhttp_data_cb on_header_field;
+ htp_hdr_valcb, // llhttp_data_cb on_header_value;
+ nullptr, // llhttp_data_cb on_chunk_extension_name;
+ nullptr, // llhttp_data_cb on_chunk_extension_value;
+ htp_hdrs_completecb, // llhttp_cb on_headers_complete;
+ htp_bodycb, // llhttp_data_cb on_body;
+ htp_msg_completecb, // llhttp_cb on_message_complete;
+ nullptr, // llhttp_cb on_url_complete;
+ nullptr, // llhttp_cb on_status_complete;
+ nullptr, // llhttp_cb on_method_complete;
+ nullptr, // llhttp_cb on_version_complete;
+ nullptr, // llhttp_cb on_header_field_complete;
+ nullptr, // llhttp_cb on_header_value_complete;
+ nullptr, // llhttp_cb on_chunk_extension_name_complete;
+ nullptr, // llhttp_cb on_chunk_extension_value_complete;
+ nullptr, // llhttp_cb on_chunk_header;
+ nullptr, // llhttp_cb on_chunk_complete;
+ nullptr, // llhttp_cb on_reset;
+};
+} // namespace
+
+int HttpDownstreamConnection::initiate_connection() {
+ int rv;
+
+ auto worker_blocker = worker_->get_connect_blocker();
+ if (worker_blocker->blocked()) {
+ if (LOG_ENABLED(INFO)) {
+ DCLOG(INFO, this)
+ << "Worker wide backend connection was blocked temporarily";
+ }
+ return SHRPX_ERR_NETWORK;
+ }
+
+ auto &downstreamconf = *worker_->get_downstream_config();
+
+ if (conn_.fd == -1) {
+ auto check_dns_result = dns_query_.get() != nullptr;
+
+ if (check_dns_result) {
+ assert(addr_->dns);
+ }
+
+ auto &connect_blocker = addr_->connect_blocker;
+
+ if (connect_blocker->blocked()) {
+ if (LOG_ENABLED(INFO)) {
+ DCLOG(INFO, this) << "Backend server " << addr_->host << ":"
+ << addr_->port << " was not available temporarily";
+ }
+
+ return SHRPX_ERR_NETWORK;
+ }
+
+ Address *raddr;
+
+ if (addr_->dns) {
+ if (!check_dns_result) {
+ auto dns_query = std::make_unique<DNSQuery>(
+ addr_->host,
+ [this](DNSResolverStatus status, const Address *result) {
+ int rv;
+
+ if (status == DNSResolverStatus::OK) {
+ *this->resolved_addr_ = *result;
+ }
+
+ rv = this->initiate_connection();
+ if (rv != 0) {
+ // This callback destroys |this|.
+ auto downstream = this->downstream_;
+ backend_retry(downstream);
+ }
+ });
+
+ auto dns_tracker = worker_->get_dns_tracker();
+
+ if (!resolved_addr_) {
+ resolved_addr_ = std::make_unique<Address>();
+ }
+ switch (dns_tracker->resolve(resolved_addr_.get(), dns_query.get())) {
+ case DNSResolverStatus::ERROR:
+ downstream_failure(addr_, nullptr);
+ return SHRPX_ERR_NETWORK;
+ case DNSResolverStatus::RUNNING:
+ dns_query_ = std::move(dns_query);
+ return 0;
+ case DNSResolverStatus::OK:
+ break;
+ default:
+ assert(0);
+ }
+ } else {
+ switch (dns_query_->status) {
+ case DNSResolverStatus::ERROR:
+ dns_query_.reset();
+ downstream_failure(addr_, nullptr);
+ return SHRPX_ERR_NETWORK;
+ case DNSResolverStatus::OK:
+ dns_query_.reset();
+ break;
+ default:
+ assert(0);
+ }
+ }
+
+ raddr = resolved_addr_.get();
+ util::set_port(*resolved_addr_, addr_->port);
+ } else {
+ raddr = &addr_->addr;
+ }
+
+ conn_.fd = util::create_nonblock_socket(raddr->su.storage.ss_family);
+
+ if (conn_.fd == -1) {
+ auto error = errno;
+ DCLOG(WARN, this) << "socket() failed; addr="
+ << util::to_numeric_addr(raddr) << ", errno=" << error;
+
+ worker_blocker->on_failure();
+
+ return SHRPX_ERR_NETWORK;
+ }
+
+ worker_blocker->on_success();
+
+ rv = connect(conn_.fd, &raddr->su.sa, raddr->len);
+ if (rv != 0 && errno != EINPROGRESS) {
+ auto error = errno;
+ DCLOG(WARN, this) << "connect() failed; addr="
+ << util::to_numeric_addr(raddr) << ", errno=" << error;
+
+ downstream_failure(addr_, raddr);
+
+ return SHRPX_ERR_NETWORK;
+ }
+
+ if (LOG_ENABLED(INFO)) {
+ DCLOG(INFO, this) << "Connecting to downstream server";
+ }
+
+ raddr_ = raddr;
+
+ if (addr_->tls) {
+ assert(ssl_ctx_);
+
+ auto ssl = tls::create_ssl(ssl_ctx_);
+ if (!ssl) {
+ return -1;
+ }
+
+ tls::setup_downstream_http1_alpn(ssl);
+
+ conn_.set_ssl(ssl);
+ conn_.tls.client_session_cache = &addr_->tls_session_cache;
+
+ auto sni_name =
+ addr_->sni.empty() ? StringRef{addr_->host} : StringRef{addr_->sni};
+ if (!util::numeric_host(sni_name.c_str())) {
+ SSL_set_tlsext_host_name(conn_.tls.ssl, sni_name.c_str());
+ }
+
+ auto session = tls::reuse_tls_session(addr_->tls_session_cache);
+ if (session) {
+ SSL_set_session(conn_.tls.ssl, session);
+ SSL_SESSION_free(session);
+ }
+
+ conn_.prepare_client_handshake();
+ }
+
+ ev_io_set(&conn_.wev, conn_.fd, EV_WRITE);
+ ev_io_set(&conn_.rev, conn_.fd, EV_READ);
+
+ conn_.wlimit.startw();
+
+ conn_.wt.repeat = downstreamconf.timeout.connect;
+ ev_timer_again(conn_.loop, &conn_.wt);
+ } else {
+ // we may set read timer cb to idle_timeoutcb. Reset again.
+ ev_set_cb(&conn_.rt, timeoutcb);
+ if (conn_.read_timeout < group_->shared_addr->timeout.read) {
+ conn_.read_timeout = group_->shared_addr->timeout.read;
+ conn_.last_read = ev_now(conn_.loop);
+ } else {
+ conn_.again_rt(group_->shared_addr->timeout.read);
+ }
+
+ ev_set_cb(&conn_.rev, readcb);
+
+ on_write_ = &HttpDownstreamConnection::write_first;
+ first_write_done_ = false;
+ request_header_written_ = false;
+ }
+
+ llhttp_init(&response_htp_, HTTP_RESPONSE, &htp_hooks);
+ response_htp_.data = downstream_;
+
+ return 0;
+}
+
+int HttpDownstreamConnection::push_request_headers() {
+ if (request_header_written_) {
+ signal_write();
+ return 0;
+ }
+
+ const auto &downstream_hostport = addr_->hostport;
+ const auto &req = downstream_->request();
+
+ auto &balloc = downstream_->get_block_allocator();
+
+ auto connect_method = req.regular_connect_method();
+
+ auto config = get_config();
+ auto &httpconf = config->http;
+
+ request_header_written_ = true;
+
+ // For HTTP/1.0 request, there is no authority in request. In that
+ // case, we use backend server's host nonetheless.
+ auto authority = StringRef(downstream_hostport);
+ auto no_host_rewrite =
+ httpconf.no_host_rewrite || config->http2_proxy || connect_method;
+
+ if (no_host_rewrite && !req.authority.empty()) {
+ authority = req.authority;
+ }
+
+ downstream_->set_request_downstream_host(authority);
+
+ auto buf = downstream_->get_request_buf();
+
+ // Assume that method and request path do not contain \r\n.
+ auto meth = http2::to_method_string(
+ req.connect_proto == ConnectProto::WEBSOCKET ? HTTP_GET : req.method);
+ buf->append(meth);
+ buf->append(' ');
+
+ if (connect_method) {
+ buf->append(authority);
+ } else if (config->http2_proxy) {
+ // Construct absolute-form request target because we are going to
+ // send a request to a HTTP/1 proxy.
+ assert(!req.scheme.empty());
+ buf->append(req.scheme);
+ buf->append("://");
+ buf->append(authority);
+ buf->append(req.path);
+ } else if (req.method == HTTP_OPTIONS && req.path.empty()) {
+ // Server-wide OPTIONS
+ buf->append("*");
+ } else {
+ buf->append(req.path);
+ }
+ buf->append(" HTTP/1.1\r\nHost: ");
+ buf->append(authority);
+ buf->append("\r\n");
+
+ auto &fwdconf = httpconf.forwarded;
+ auto &xffconf = httpconf.xff;
+ auto &xfpconf = httpconf.xfp;
+ auto &earlydataconf = httpconf.early_data;
+
+ uint32_t build_flags =
+ (fwdconf.strip_incoming ? http2::HDOP_STRIP_FORWARDED : 0) |
+ (xffconf.strip_incoming ? http2::HDOP_STRIP_X_FORWARDED_FOR : 0) |
+ (xfpconf.strip_incoming ? http2::HDOP_STRIP_X_FORWARDED_PROTO : 0) |
+ (earlydataconf.strip_incoming ? http2::HDOP_STRIP_EARLY_DATA : 0) |
+ ((req.http_major == 3 || req.http_major == 2)
+ ? http2::HDOP_STRIP_SEC_WEBSOCKET_KEY
+ : 0);
+
+ http2::build_http1_headers_from_headers(buf, req.fs.headers(), build_flags);
+
+ auto cookie = downstream_->assemble_request_cookie();
+ if (!cookie.empty()) {
+ buf->append("Cookie: ");
+ buf->append(cookie);
+ buf->append("\r\n");
+ }
+
+ // set transfer-encoding only when content-length is unknown and
+ // request body is expected.
+ if (req.method != HTTP_CONNECT && req.http2_expect_body &&
+ req.fs.content_length == -1) {
+ downstream_->set_chunked_request(true);
+ buf->append("Transfer-Encoding: chunked\r\n");
+ }
+
+ if (req.connect_proto == ConnectProto::WEBSOCKET) {
+ if (req.http_major == 3 || req.http_major == 2) {
+ std::array<uint8_t, 16> nonce;
+ if (RAND_bytes(nonce.data(), nonce.size()) != 1) {
+ return -1;
+ }
+ auto iov = make_byte_ref(balloc, base64::encode_length(nonce.size()) + 1);
+ auto p = base64::encode(std::begin(nonce), std::end(nonce), iov.base);
+ *p = '\0';
+ auto key = StringRef{iov.base, p};
+ downstream_->set_ws_key(key);
+
+ buf->append("Sec-Websocket-Key: ");
+ buf->append(key);
+ buf->append("\r\n");
+ }
+
+ buf->append("Upgrade: websocket\r\nConnection: Upgrade\r\n");
+ } else if (!connect_method && req.upgrade_request) {
+ auto connection = req.fs.header(http2::HD_CONNECTION);
+ if (connection) {
+ buf->append("Connection: ");
+ buf->append((*connection).value);
+ buf->append("\r\n");
+ }
+
+ auto upgrade = req.fs.header(http2::HD_UPGRADE);
+ if (upgrade) {
+ buf->append("Upgrade: ");
+ buf->append((*upgrade).value);
+ buf->append("\r\n");
+ }
+ } else if (req.connection_close) {
+ buf->append("Connection: close\r\n");
+ }
+
+ auto upstream = downstream_->get_upstream();
+ auto handler = upstream->get_client_handler();
+
+#if OPENSSL_1_1_1_API
+ auto conn = handler->get_connection();
+
+ if (conn->tls.ssl && !SSL_is_init_finished(conn->tls.ssl)) {
+ buf->append("Early-Data: 1\r\n");
+ }
+#endif // OPENSSL_1_1_1_API
+
+ auto fwd =
+ fwdconf.strip_incoming ? nullptr : req.fs.header(http2::HD_FORWARDED);
+
+ if (fwdconf.params) {
+ auto params = fwdconf.params;
+
+ if (config->http2_proxy || connect_method) {
+ params &= ~FORWARDED_PROTO;
+ }
+
+ auto value = http::create_forwarded(
+ balloc, params, handler->get_forwarded_by(),
+ handler->get_forwarded_for(), req.authority, req.scheme);
+
+ if (fwd || !value.empty()) {
+ buf->append("Forwarded: ");
+ if (fwd) {
+ buf->append(fwd->value);
+
+ if (!value.empty()) {
+ buf->append(", ");
+ }
+ }
+ buf->append(value);
+ buf->append("\r\n");
+ }
+ } else if (fwd) {
+ buf->append("Forwarded: ");
+ buf->append(fwd->value);
+ buf->append("\r\n");
+ }
+
+ auto xff = xffconf.strip_incoming ? nullptr
+ : req.fs.header(http2::HD_X_FORWARDED_FOR);
+
+ if (xffconf.add) {
+ buf->append("X-Forwarded-For: ");
+ if (xff) {
+ buf->append((*xff).value);
+ buf->append(", ");
+ }
+ buf->append(client_handler_->get_ipaddr());
+ buf->append("\r\n");
+ } else if (xff) {
+ buf->append("X-Forwarded-For: ");
+ buf->append((*xff).value);
+ buf->append("\r\n");
+ }
+ if (!config->http2_proxy && !connect_method) {
+ auto xfp = xfpconf.strip_incoming
+ ? nullptr
+ : req.fs.header(http2::HD_X_FORWARDED_PROTO);
+
+ if (xfpconf.add) {
+ buf->append("X-Forwarded-Proto: ");
+ if (xfp) {
+ buf->append((*xfp).value);
+ buf->append(", ");
+ }
+ assert(!req.scheme.empty());
+ buf->append(req.scheme);
+ buf->append("\r\n");
+ } else if (xfp) {
+ buf->append("X-Forwarded-Proto: ");
+ buf->append((*xfp).value);
+ buf->append("\r\n");
+ }
+ }
+ auto via = req.fs.header(http2::HD_VIA);
+ if (httpconf.no_via) {
+ if (via) {
+ buf->append("Via: ");
+ buf->append((*via).value);
+ buf->append("\r\n");
+ }
+ } else {
+ buf->append("Via: ");
+ if (via) {
+ buf->append((*via).value);
+ buf->append(", ");
+ }
+ std::array<char, 16> viabuf;
+ auto end = http::create_via_header_value(viabuf.data(), req.http_major,
+ req.http_minor);
+ buf->append(viabuf.data(), end - viabuf.data());
+ buf->append("\r\n");
+ }
+
+ for (auto &p : httpconf.add_request_headers) {
+ buf->append(p.name);
+ buf->append(": ");
+ buf->append(p.value);
+ buf->append("\r\n");
+ }
+
+ buf->append("\r\n");
+
+ if (LOG_ENABLED(INFO)) {
+ std::string nhdrs;
+ for (auto chunk = buf->head; chunk; chunk = chunk->next) {
+ nhdrs.append(chunk->pos, chunk->last);
+ }
+ if (log_config()->errorlog_tty) {
+ nhdrs = http::colorizeHeaders(nhdrs.c_str());
+ }
+ DCLOG(INFO, this) << "HTTP request headers. stream_id="
+ << downstream_->get_stream_id() << "\n"
+ << nhdrs;
+ }
+
+ // Don't call signal_write() if we anticipate request body. We call
+ // signal_write() when we received request body chunk, and it
+ // enables us to send headers and data in one writev system call.
+ if (req.method == HTTP_CONNECT ||
+ downstream_->get_blocked_request_buf()->rleft() ||
+ (!req.http2_expect_body && req.fs.content_length == 0) ||
+ downstream_->get_expect_100_continue()) {
+ signal_write();
+ }
+
+ return 0;
+}
+
+int HttpDownstreamConnection::process_blocked_request_buf() {
+ auto src = downstream_->get_blocked_request_buf();
+
+ if (src->rleft()) {
+ auto dest = downstream_->get_request_buf();
+ auto chunked = downstream_->get_chunked_request();
+ if (chunked) {
+ auto chunk_size_hex = util::utox(src->rleft());
+ dest->append(chunk_size_hex);
+ dest->append("\r\n");
+ }
+
+ src->copy(*dest);
+
+ if (chunked) {
+ dest->append("\r\n");
+ }
+ }
+
+ if (downstream_->get_blocked_request_data_eof() &&
+ downstream_->get_chunked_request()) {
+ end_upload_data_chunk();
+ }
+
+ return 0;
+}
+
+int HttpDownstreamConnection::push_upload_data_chunk(const uint8_t *data,
+ size_t datalen) {
+ if (!downstream_->get_request_header_sent()) {
+ auto output = downstream_->get_blocked_request_buf();
+ auto &req = downstream_->request();
+ output->append(data, datalen);
+ req.unconsumed_body_length += datalen;
+ if (request_header_written_) {
+ signal_write();
+ }
+ return 0;
+ }
+
+ auto chunked = downstream_->get_chunked_request();
+ auto output = downstream_->get_request_buf();
+
+ if (chunked) {
+ auto chunk_size_hex = util::utox(datalen);
+ output->append(chunk_size_hex);
+ output->append("\r\n");
+ }
+
+ output->append(data, datalen);
+
+ if (chunked) {
+ output->append("\r\n");
+ }
+
+ signal_write();
+
+ return 0;
+}
+
+int HttpDownstreamConnection::end_upload_data() {
+ if (!downstream_->get_request_header_sent()) {
+ downstream_->set_blocked_request_data_eof(true);
+ if (request_header_written_) {
+ signal_write();
+ }
+ return 0;
+ }
+
+ signal_write();
+
+ if (!downstream_->get_chunked_request()) {
+ return 0;
+ }
+
+ end_upload_data_chunk();
+
+ return 0;
+}
+
+void HttpDownstreamConnection::end_upload_data_chunk() {
+ const auto &req = downstream_->request();
+
+ auto output = downstream_->get_request_buf();
+ const auto &trailers = req.fs.trailers();
+ if (trailers.empty()) {
+ output->append("0\r\n\r\n");
+ } else {
+ output->append("0\r\n");
+ http2::build_http1_headers_from_headers(output, trailers,
+ http2::HDOP_STRIP_ALL);
+ output->append("\r\n");
+ }
+}
+
+namespace {
+void remove_from_pool(HttpDownstreamConnection *dconn) {
+ auto addr = dconn->get_addr();
+ auto &dconn_pool = addr->dconn_pool;
+ dconn_pool->remove_downstream_connection(dconn);
+}
+} // namespace
+
+namespace {
+void idle_readcb(struct ev_loop *loop, ev_io *w, int revents) {
+ auto conn = static_cast<Connection *>(w->data);
+ auto dconn = static_cast<HttpDownstreamConnection *>(conn->data);
+ if (LOG_ENABLED(INFO)) {
+ DCLOG(INFO, dconn) << "Idle connection EOF";
+ }
+
+ remove_from_pool(dconn);
+ // dconn was deleted
+}
+} // namespace
+
+namespace {
+void idle_timeoutcb(struct ev_loop *loop, ev_timer *w, int revents) {
+ auto conn = static_cast<Connection *>(w->data);
+ auto dconn = static_cast<HttpDownstreamConnection *>(conn->data);
+
+ if (w == &conn->rt && !conn->expired_rt()) {
+ return;
+ }
+
+ if (LOG_ENABLED(INFO)) {
+ DCLOG(INFO, dconn) << "Idle connection timeout";
+ }
+
+ remove_from_pool(dconn);
+ // dconn was deleted
+}
+} // namespace
+
+void HttpDownstreamConnection::detach_downstream(Downstream *downstream) {
+ if (LOG_ENABLED(INFO)) {
+ DCLOG(INFO, this) << "Detaching from DOWNSTREAM:" << downstream;
+ }
+ downstream_ = nullptr;
+
+ ev_set_cb(&conn_.rev, idle_readcb);
+ ioctrl_.force_resume_read();
+
+ auto &downstreamconf = *worker_->get_downstream_config();
+
+ ev_set_cb(&conn_.rt, idle_timeoutcb);
+ if (conn_.read_timeout < downstreamconf.timeout.idle_read) {
+ conn_.read_timeout = downstreamconf.timeout.idle_read;
+ conn_.last_read = ev_now(conn_.loop);
+ } else {
+ conn_.again_rt(downstreamconf.timeout.idle_read);
+ }
+
+ conn_.wlimit.stopw();
+ ev_timer_stop(conn_.loop, &conn_.wt);
+}
+
+void HttpDownstreamConnection::pause_read(IOCtrlReason reason) {
+ ioctrl_.pause_read(reason);
+}
+
+int HttpDownstreamConnection::resume_read(IOCtrlReason reason,
+ size_t consumed) {
+ auto &downstreamconf = *worker_->get_downstream_config();
+
+ if (downstream_->get_response_buf()->rleft() <=
+ downstreamconf.request_buffer_size / 2) {
+ ioctrl_.resume_read(reason);
+ }
+
+ return 0;
+}
+
+void HttpDownstreamConnection::force_resume_read() {
+ ioctrl_.force_resume_read();
+}
+
+namespace {
+int htp_msg_begincb(llhttp_t *htp) {
+ auto downstream = static_cast<Downstream *>(htp->data);
+
+ if (downstream->get_response_state() != DownstreamState::INITIAL) {
+ return -1;
+ }
+
+ return 0;
+}
+} // namespace
+
+namespace {
+int htp_hdrs_completecb(llhttp_t *htp) {
+ auto downstream = static_cast<Downstream *>(htp->data);
+ auto upstream = downstream->get_upstream();
+ auto handler = upstream->get_client_handler();
+ const auto &req = downstream->request();
+ auto &resp = downstream->response();
+ int rv;
+
+ auto &balloc = downstream->get_block_allocator();
+
+ for (auto &kv : resp.fs.headers()) {
+ kv.value = util::rstrip(balloc, kv.value);
+ }
+
+ auto config = get_config();
+ auto &loggingconf = config->logging;
+
+ resp.http_status = htp->status_code;
+ resp.http_major = htp->http_major;
+ resp.http_minor = htp->http_minor;
+
+ if (resp.http_major > 1 || req.http_minor > 1) {
+ resp.http_major = 1;
+ resp.http_minor = 1;
+ return -1;
+ }
+
+ auto dconn = downstream->get_downstream_connection();
+
+ downstream->set_downstream_addr_group(dconn->get_downstream_addr_group());
+ downstream->set_addr(dconn->get_addr());
+
+ // Server MUST NOT send Transfer-Encoding with a status code 1xx or
+ // 204. Also server MUST NOT send Transfer-Encoding with a status
+ // code 2xx to a CONNECT request. Same holds true with
+ // Content-Length.
+ if (resp.http_status == 204) {
+ if (resp.fs.header(http2::HD_TRANSFER_ENCODING)) {
+ return -1;
+ }
+ // Some server send content-length: 0 for 204. Until they get
+ // fixed, we accept, but ignore it.
+
+ // Calling parse_content_length() detects duplicated
+ // content-length header fields.
+ if (resp.fs.parse_content_length() != 0) {
+ return -1;
+ }
+ if (resp.fs.content_length == 0) {
+ resp.fs.erase_content_length_and_transfer_encoding();
+ } else if (resp.fs.content_length != -1) {
+ return -1;
+ }
+ } else if (resp.http_status / 100 == 1 ||
+ (resp.http_status / 100 == 2 && req.method == HTTP_CONNECT)) {
+ // Server MUST NOT send Content-Length and Transfer-Encoding in
+ // these responses.
+ resp.fs.erase_content_length_and_transfer_encoding();
+ } else if (resp.fs.parse_content_length() != 0) {
+ downstream->set_response_state(DownstreamState::MSG_BAD_HEADER);
+ return -1;
+ }
+
+ // Check upgrade before processing non-final response, since if
+ // upgrade succeeded, 101 response is treated as final in nghttpx.
+ downstream->check_upgrade_fulfilled_http1();
+
+ if (downstream->get_non_final_response()) {
+ // Reset content-length because we reuse same Downstream for the
+ // next response.
+ resp.fs.content_length = -1;
+ // For non-final response code, we just call
+ // on_downstream_header_complete() without changing response
+ // state.
+ rv = upstream->on_downstream_header_complete(downstream);
+
+ if (rv != 0) {
+ return -1;
+ }
+
+ // Ignore response body for non-final response.
+ return 1;
+ }
+
+ resp.connection_close = !llhttp_should_keep_alive(htp);
+ downstream->set_response_state(DownstreamState::HEADER_COMPLETE);
+ downstream->inspect_http1_response();
+ if (downstream->get_upgraded()) {
+ // content-length must be ignored for upgraded connection.
+ resp.fs.content_length = -1;
+ resp.connection_close = true;
+ // transfer-encoding not applied to upgraded connection
+ downstream->set_chunked_response(false);
+ } else if (http2::legacy_http1(req.http_major, req.http_minor)) {
+ if (resp.fs.content_length == -1) {
+ resp.connection_close = true;
+ }
+ downstream->set_chunked_response(false);
+ } else if (!downstream->expect_response_body()) {
+ downstream->set_chunked_response(false);
+ }
+
+ if (loggingconf.access.write_early && downstream->accesslog_ready()) {
+ handler->write_accesslog(downstream);
+ downstream->set_accesslog_written(true);
+ }
+
+ if (upstream->on_downstream_header_complete(downstream) != 0) {
+ return -1;
+ }
+
+ if (downstream->get_upgraded()) {
+ // Upgrade complete, read until EOF in both ends
+ if (upstream->resume_read(SHRPX_NO_BUFFER, downstream, 0) != 0) {
+ return -1;
+ }
+ downstream->set_request_state(DownstreamState::HEADER_COMPLETE);
+ if (LOG_ENABLED(INFO)) {
+ LOG(INFO) << "HTTP upgrade success. stream_id="
+ << downstream->get_stream_id();
+ }
+ }
+
+ // Ignore the response body. HEAD response may contain
+ // Content-Length or Transfer-Encoding: chunked. Some server send
+ // 304 status code with nonzero Content-Length, but without response
+ // body. See
+ // https://tools.ietf.org/html/rfc7230#section-3.3
+
+ // TODO It seems that the cases other than HEAD are handled by
+ // llhttp. Need test.
+ return !http2::expect_response_body(req.method, resp.http_status);
+}
+} // namespace
+
+namespace {
+int ensure_header_field_buffer(const Downstream *downstream,
+ const HttpConfig &httpconf, size_t len) {
+ auto &resp = downstream->response();
+
+ if (resp.fs.buffer_size() + len > httpconf.response_header_field_buffer) {
+ if (LOG_ENABLED(INFO)) {
+ DLOG(INFO, downstream) << "Too large header header field size="
+ << resp.fs.buffer_size() + len;
+ }
+ return -1;
+ }
+
+ return 0;
+}
+} // namespace
+
+namespace {
+int ensure_max_header_fields(const Downstream *downstream,
+ const HttpConfig &httpconf) {
+ auto &resp = downstream->response();
+
+ if (resp.fs.num_fields() >= httpconf.max_response_header_fields) {
+ if (LOG_ENABLED(INFO)) {
+ DLOG(INFO, downstream)
+ << "Too many header field num=" << resp.fs.num_fields() + 1;
+ }
+ return -1;
+ }
+
+ return 0;
+}
+} // namespace
+
+namespace {
+int htp_hdr_keycb(llhttp_t *htp, const char *data, size_t len) {
+ auto downstream = static_cast<Downstream *>(htp->data);
+ auto &resp = downstream->response();
+ auto &httpconf = get_config()->http;
+
+ if (ensure_header_field_buffer(downstream, httpconf, len) != 0) {
+ return -1;
+ }
+
+ if (downstream->get_response_state() == DownstreamState::INITIAL) {
+ if (resp.fs.header_key_prev()) {
+ resp.fs.append_last_header_key(data, len);
+ } else {
+ if (ensure_max_header_fields(downstream, httpconf) != 0) {
+ return -1;
+ }
+ resp.fs.alloc_add_header_name(StringRef{data, len});
+ }
+ } else {
+ // trailer part
+ if (resp.fs.trailer_key_prev()) {
+ resp.fs.append_last_trailer_key(data, len);
+ } else {
+ if (ensure_max_header_fields(downstream, httpconf) != 0) {
+ // Could not ignore this trailer field easily, since we may
+ // get its value in htp_hdr_valcb, and it will be added to
+ // wrong place or crash if trailer fields are currently empty.
+ return -1;
+ }
+ resp.fs.alloc_add_trailer_name(StringRef{data, len});
+ }
+ }
+ return 0;
+}
+} // namespace
+
+namespace {
+int htp_hdr_valcb(llhttp_t *htp, const char *data, size_t len) {
+ auto downstream = static_cast<Downstream *>(htp->data);
+ auto &resp = downstream->response();
+ auto &httpconf = get_config()->http;
+
+ if (ensure_header_field_buffer(downstream, httpconf, len) != 0) {
+ return -1;
+ }
+
+ if (downstream->get_response_state() == DownstreamState::INITIAL) {
+ resp.fs.append_last_header_value(data, len);
+ } else {
+ resp.fs.append_last_trailer_value(data, len);
+ }
+ return 0;
+}
+} // namespace
+
+namespace {
+int htp_bodycb(llhttp_t *htp, const char *data, size_t len) {
+ auto downstream = static_cast<Downstream *>(htp->data);
+ auto &resp = downstream->response();
+
+ resp.recv_body_length += len;
+
+ return downstream->get_upstream()->on_downstream_body(
+ downstream, reinterpret_cast<const uint8_t *>(data), len, true);
+}
+} // namespace
+
+namespace {
+int htp_msg_completecb(llhttp_t *htp) {
+ auto downstream = static_cast<Downstream *>(htp->data);
+ auto &resp = downstream->response();
+ auto &balloc = downstream->get_block_allocator();
+
+ for (auto &kv : resp.fs.trailers()) {
+ kv.value = util::rstrip(balloc, kv.value);
+ }
+
+ // llhttp does not treat "200 connection established" response
+ // against CONNECT request, and in that case, this function is not
+ // called. But if HTTP Upgrade is made (e.g., WebSocket), this
+ // function is called, and llhttp_execute() returns just after that.
+ if (downstream->get_upgraded()) {
+ return 0;
+ }
+
+ if (downstream->get_non_final_response()) {
+ downstream->reset_response();
+
+ return 0;
+ }
+
+ downstream->set_response_state(DownstreamState::MSG_COMPLETE);
+ // Block reading another response message from (broken?)
+ // server. This callback is not called if the connection is
+ // tunneled.
+ downstream->pause_read(SHRPX_MSG_BLOCK);
+ return downstream->get_upstream()->on_downstream_body_complete(downstream);
+}
+} // namespace
+
+int HttpDownstreamConnection::write_first() {
+ int rv;
+
+ process_blocked_request_buf();
+
+ if (conn_.tls.ssl) {
+ rv = write_tls();
+ } else {
+ rv = write_clear();
+ }
+
+ if (rv != 0) {
+ return SHRPX_ERR_RETRY;
+ }
+
+ if (conn_.tls.ssl) {
+ on_write_ = &HttpDownstreamConnection::write_tls;
+ } else {
+ on_write_ = &HttpDownstreamConnection::write_clear;
+ }
+
+ first_write_done_ = true;
+ downstream_->set_request_header_sent(true);
+
+ auto buf = downstream_->get_blocked_request_buf();
+ buf->reset();
+
+ // upstream->resume_read() might be called in
+ // write_tls()/write_clear(), but before blocked_request_buf_ is
+ // reset. So upstream read might still be blocked. Let's do it
+ // again here.
+ auto input = downstream_->get_request_buf();
+ if (input->rleft() == 0) {
+ auto upstream = downstream_->get_upstream();
+ auto &req = downstream_->request();
+
+ upstream->resume_read(SHRPX_NO_BUFFER, downstream_,
+ req.unconsumed_body_length);
+ }
+
+ return 0;
+}
+
+int HttpDownstreamConnection::read_clear() {
+ conn_.last_read = ev_now(conn_.loop);
+
+ std::array<uint8_t, 16_k> buf;
+ int rv;
+
+ for (;;) {
+ auto nread = conn_.read_clear(buf.data(), buf.size());
+ if (nread == 0) {
+ return 0;
+ }
+
+ if (nread < 0) {
+ if (nread == SHRPX_ERR_EOF && !downstream_->get_upgraded()) {
+ auto htperr = llhttp_finish(&response_htp_);
+ if (htperr != HPE_OK) {
+ if (LOG_ENABLED(INFO)) {
+ DCLOG(INFO, this) << "HTTP response ended prematurely: "
+ << llhttp_errno_name(htperr);
+ }
+
+ return -1;
+ }
+ }
+
+ return nread;
+ }
+
+ rv = process_input(buf.data(), nread);
+ if (rv != 0) {
+ return rv;
+ }
+
+ if (!ev_is_active(&conn_.rev)) {
+ return 0;
+ }
+ }
+}
+
+int HttpDownstreamConnection::write_clear() {
+ conn_.last_read = ev_now(conn_.loop);
+
+ auto upstream = downstream_->get_upstream();
+ auto input = downstream_->get_request_buf();
+
+ std::array<struct iovec, MAX_WR_IOVCNT> iov;
+
+ while (input->rleft() > 0) {
+ auto iovcnt = input->riovec(iov.data(), iov.size());
+
+ auto nwrite = conn_.writev_clear(iov.data(), iovcnt);
+
+ if (nwrite == 0) {
+ return 0;
+ }
+
+ if (nwrite < 0) {
+ if (!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.
+ ev_feed_event(conn_.loop, &conn_.rev, EV_READ);
+ on_write_ = &HttpDownstreamConnection::noop;
+ reusable_ = false;
+ break;
+ }
+
+ input->drain(nwrite);
+ }
+
+ conn_.wlimit.stopw();
+ ev_timer_stop(conn_.loop, &conn_.wt);
+
+ if (input->rleft() == 0) {
+ auto &req = downstream_->request();
+
+ upstream->resume_read(SHRPX_NO_BUFFER, downstream_,
+ req.unconsumed_body_length);
+ }
+
+ return 0;
+}
+
+int HttpDownstreamConnection::tls_handshake() {
+ ERR_clear_error();
+
+ conn_.last_read = ev_now(conn_.loop);
+
+ auto rv = conn_.tls_handshake();
+ if (rv == SHRPX_ERR_INPROGRESS) {
+ return 0;
+ }
+
+ if (rv < 0) {
+ downstream_failure(addr_, raddr_);
+
+ return rv;
+ }
+
+ if (LOG_ENABLED(INFO)) {
+ DCLOG(INFO, this) << "SSL/TLS handshake completed";
+ }
+
+ if (!get_config()->tls.insecure &&
+ tls::check_cert(conn_.tls.ssl, addr_, raddr_) != 0) {
+ downstream_failure(addr_, raddr_);
+
+ return -1;
+ }
+
+ auto &connect_blocker = addr_->connect_blocker;
+
+ signal_write_ = &HttpDownstreamConnection::actual_signal_write;
+
+ connect_blocker->on_success();
+
+ ev_set_cb(&conn_.rt, timeoutcb);
+ ev_set_cb(&conn_.wt, timeoutcb);
+
+ on_read_ = &HttpDownstreamConnection::read_tls;
+ on_write_ = &HttpDownstreamConnection::write_first;
+
+ // TODO Check negotiated ALPN
+
+ return on_write();
+}
+
+int HttpDownstreamConnection::read_tls() {
+ conn_.last_read = ev_now(conn_.loop);
+
+ ERR_clear_error();
+
+ std::array<uint8_t, 16_k> buf;
+ int rv;
+
+ for (;;) {
+ auto nread = conn_.read_tls(buf.data(), buf.size());
+ if (nread == 0) {
+ return 0;
+ }
+
+ if (nread < 0) {
+ if (nread == SHRPX_ERR_EOF && !downstream_->get_upgraded()) {
+ auto htperr = llhttp_finish(&response_htp_);
+ if (htperr != HPE_OK) {
+ if (LOG_ENABLED(INFO)) {
+ DCLOG(INFO, this) << "HTTP response ended prematurely: "
+ << llhttp_errno_name(htperr);
+ }
+
+ return -1;
+ }
+ }
+
+ return nread;
+ }
+
+ rv = process_input(buf.data(), nread);
+ if (rv != 0) {
+ return rv;
+ }
+
+ if (!ev_is_active(&conn_.rev)) {
+ return 0;
+ }
+ }
+}
+
+int HttpDownstreamConnection::write_tls() {
+ conn_.last_read = ev_now(conn_.loop);
+
+ ERR_clear_error();
+
+ auto upstream = downstream_->get_upstream();
+ auto input = downstream_->get_request_buf();
+
+ struct iovec iov;
+
+ while (input->rleft() > 0) {
+ auto iovcnt = input->riovec(&iov, 1);
+ if (iovcnt != 1) {
+ assert(0);
+ return -1;
+ }
+ auto nwrite = conn_.write_tls(iov.iov_base, iov.iov_len);
+
+ if (nwrite == 0) {
+ return 0;
+ }
+
+ if (nwrite < 0) {
+ if (!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.
+ ev_feed_event(conn_.loop, &conn_.rev, EV_READ);
+ on_write_ = &HttpDownstreamConnection::noop;
+ reusable_ = false;
+ break;
+ }
+
+ input->drain(nwrite);
+ }
+
+ conn_.wlimit.stopw();
+ ev_timer_stop(conn_.loop, &conn_.wt);
+
+ if (input->rleft() == 0) {
+ auto &req = downstream_->request();
+
+ upstream->resume_read(SHRPX_NO_BUFFER, downstream_,
+ req.unconsumed_body_length);
+ }
+
+ return 0;
+}
+
+int HttpDownstreamConnection::process_input(const uint8_t *data,
+ size_t datalen) {
+ int rv;
+
+ if (downstream_->get_upgraded()) {
+ // For upgraded connection, just pass data to the upstream.
+ rv = downstream_->get_upstream()->on_downstream_body(downstream_, data,
+ datalen, true);
+ if (rv != 0) {
+ return rv;
+ }
+
+ if (downstream_->response_buf_full()) {
+ downstream_->pause_read(SHRPX_NO_BUFFER);
+ return 0;
+ }
+
+ return 0;
+ }
+
+ auto htperr = llhttp_execute(&response_htp_,
+ reinterpret_cast<const char *>(data), datalen);
+ auto nproc =
+ htperr == HPE_OK
+ ? datalen
+ : static_cast<size_t>(reinterpret_cast<const uint8_t *>(
+ llhttp_get_error_pos(&response_htp_)) -
+ data);
+
+ if (htperr != HPE_OK &&
+ (!downstream_->get_upgraded() || htperr != HPE_PAUSED_UPGRADE)) {
+ // Handling early return (in other words, response was hijacked by
+ // mruby scripting).
+ if (downstream_->get_response_state() == DownstreamState::MSG_COMPLETE) {
+ return SHRPX_ERR_DCONN_CANCELED;
+ }
+
+ if (LOG_ENABLED(INFO)) {
+ DCLOG(INFO, this) << "HTTP parser failure: "
+ << "(" << llhttp_errno_name(htperr) << ") "
+ << llhttp_get_error_reason(&response_htp_);
+ }
+
+ return -1;
+ }
+
+ if (downstream_->get_upgraded()) {
+ if (nproc < datalen) {
+ // Data from data + nproc are for upgraded protocol.
+ rv = downstream_->get_upstream()->on_downstream_body(
+ downstream_, data + nproc, datalen - nproc, true);
+ if (rv != 0) {
+ return rv;
+ }
+
+ if (downstream_->response_buf_full()) {
+ downstream_->pause_read(SHRPX_NO_BUFFER);
+ return 0;
+ }
+ }
+ return 0;
+ }
+
+ if (downstream_->response_buf_full()) {
+ downstream_->pause_read(SHRPX_NO_BUFFER);
+ return 0;
+ }
+
+ return 0;
+}
+
+int HttpDownstreamConnection::connected() {
+ auto &connect_blocker = addr_->connect_blocker;
+
+ auto sock_error = util::get_socket_error(conn_.fd);
+ if (sock_error != 0) {
+ conn_.wlimit.stopw();
+
+ DCLOG(WARN, this) << "Backend connect failed; addr="
+ << util::to_numeric_addr(raddr_)
+ << ": errno=" << sock_error;
+
+ downstream_failure(addr_, raddr_);
+
+ return -1;
+ }
+
+ if (LOG_ENABLED(INFO)) {
+ DCLOG(INFO, this) << "Connected to downstream host";
+ }
+
+ // Reset timeout for write. Previously, we set timeout for connect.
+ conn_.wt.repeat = group_->shared_addr->timeout.write;
+ ev_timer_again(conn_.loop, &conn_.wt);
+
+ conn_.rlimit.startw();
+ conn_.again_rt();
+
+ ev_set_cb(&conn_.wev, writecb);
+
+ if (conn_.tls.ssl) {
+ on_read_ = &HttpDownstreamConnection::tls_handshake;
+ on_write_ = &HttpDownstreamConnection::tls_handshake;
+
+ return 0;
+ }
+
+ signal_write_ = &HttpDownstreamConnection::actual_signal_write;
+
+ connect_blocker->on_success();
+
+ ev_set_cb(&conn_.rt, timeoutcb);
+ ev_set_cb(&conn_.wt, timeoutcb);
+
+ on_read_ = &HttpDownstreamConnection::read_clear;
+ on_write_ = &HttpDownstreamConnection::write_first;
+
+ return 0;
+}
+
+int HttpDownstreamConnection::on_read() { return on_read_(*this); }
+
+int HttpDownstreamConnection::on_write() { return on_write_(*this); }
+
+void HttpDownstreamConnection::on_upstream_change(Upstream *upstream) {}
+
+void HttpDownstreamConnection::signal_write() { signal_write_(*this); }
+
+int HttpDownstreamConnection::actual_signal_write() {
+ ev_feed_event(conn_.loop, &conn_.wev, EV_WRITE);
+ return 0;
+}
+
+int HttpDownstreamConnection::noop() { return 0; }
+
+const std::shared_ptr<DownstreamAddrGroup> &
+HttpDownstreamConnection::get_downstream_addr_group() const {
+ return group_;
+}
+
+DownstreamAddr *HttpDownstreamConnection::get_addr() const { return addr_; }
+
+bool HttpDownstreamConnection::poolable() const {
+ return !group_->retired && reusable_;
+}
+
+const Address *HttpDownstreamConnection::get_raddr() const { return raddr_; }
+
+} // namespace shrpx