summaryrefslogtreecommitdiffstats
path: root/src/shrpx_downstream.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_downstream.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 '')
-rw-r--r--src/shrpx_downstream.cc1195
1 files changed, 1195 insertions, 0 deletions
diff --git a/src/shrpx_downstream.cc b/src/shrpx_downstream.cc
new file mode 100644
index 0000000..245e00b
--- /dev/null
+++ b/src/shrpx_downstream.cc
@@ -0,0 +1,1195 @@
+/*
+ * 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_downstream.h"
+
+#include <cassert>
+
+#include "url-parser/url_parser.h"
+
+#include "shrpx_upstream.h"
+#include "shrpx_client_handler.h"
+#include "shrpx_config.h"
+#include "shrpx_error.h"
+#include "shrpx_downstream_connection.h"
+#include "shrpx_downstream_queue.h"
+#include "shrpx_worker.h"
+#include "shrpx_http2_session.h"
+#include "shrpx_log.h"
+#ifdef HAVE_MRUBY
+# include "shrpx_mruby.h"
+#endif // HAVE_MRUBY
+#include "util.h"
+#include "http2.h"
+
+namespace shrpx {
+
+namespace {
+void upstream_timeoutcb(struct ev_loop *loop, ev_timer *w, int revents) {
+ auto downstream = static_cast<Downstream *>(w->data);
+ auto upstream = downstream->get_upstream();
+
+ auto which = revents == EV_READ ? "read" : "write";
+
+ if (LOG_ENABLED(INFO)) {
+ DLOG(INFO, downstream) << "upstream timeout stream_id="
+ << downstream->get_stream_id() << " event=" << which;
+ }
+
+ downstream->disable_upstream_rtimer();
+ downstream->disable_upstream_wtimer();
+
+ upstream->on_timeout(downstream);
+}
+} // namespace
+
+namespace {
+void upstream_rtimeoutcb(struct ev_loop *loop, ev_timer *w, int revents) {
+ upstream_timeoutcb(loop, w, EV_READ);
+}
+} // namespace
+
+namespace {
+void upstream_wtimeoutcb(struct ev_loop *loop, ev_timer *w, int revents) {
+ upstream_timeoutcb(loop, w, EV_WRITE);
+}
+} // namespace
+
+namespace {
+void downstream_timeoutcb(struct ev_loop *loop, ev_timer *w, int revents) {
+ auto downstream = static_cast<Downstream *>(w->data);
+
+ auto which = revents == EV_READ ? "read" : "write";
+
+ if (LOG_ENABLED(INFO)) {
+ DLOG(INFO, downstream) << "downstream timeout stream_id="
+ << downstream->get_downstream_stream_id()
+ << " event=" << which;
+ }
+
+ downstream->disable_downstream_rtimer();
+ downstream->disable_downstream_wtimer();
+
+ auto dconn = downstream->get_downstream_connection();
+
+ if (dconn) {
+ dconn->on_timeout();
+ }
+}
+} // namespace
+
+namespace {
+void downstream_rtimeoutcb(struct ev_loop *loop, ev_timer *w, int revents) {
+ downstream_timeoutcb(loop, w, EV_READ);
+}
+} // namespace
+
+namespace {
+void downstream_wtimeoutcb(struct ev_loop *loop, ev_timer *w, int revents) {
+ downstream_timeoutcb(loop, w, EV_WRITE);
+}
+} // namespace
+
+// upstream could be nullptr for unittests
+Downstream::Downstream(Upstream *upstream, MemchunkPool *mcpool,
+ int64_t stream_id)
+ : dlnext(nullptr),
+ dlprev(nullptr),
+ response_sent_body_length(0),
+ balloc_(1024, 1024),
+ req_(balloc_),
+ resp_(balloc_),
+ request_start_time_(std::chrono::high_resolution_clock::now()),
+ blocked_request_buf_(mcpool),
+ request_buf_(mcpool),
+ response_buf_(mcpool),
+ upstream_(upstream),
+ blocked_link_(nullptr),
+ addr_(nullptr),
+ num_retry_(0),
+ stream_id_(stream_id),
+ assoc_stream_id_(-1),
+ downstream_stream_id_(-1),
+ response_rst_stream_error_code_(NGHTTP2_NO_ERROR),
+ affinity_cookie_(0),
+ request_state_(DownstreamState::INITIAL),
+ response_state_(DownstreamState::INITIAL),
+ dispatch_state_(DispatchState::NONE),
+ upgraded_(false),
+ chunked_request_(false),
+ chunked_response_(false),
+ expect_final_response_(false),
+ request_pending_(false),
+ request_header_sent_(false),
+ accesslog_written_(false),
+ new_affinity_cookie_(false),
+ blocked_request_data_eof_(false),
+ expect_100_continue_(false),
+ stop_reading_(false) {
+
+ auto &timeoutconf = get_config()->http2.timeout;
+
+ ev_timer_init(&upstream_rtimer_, &upstream_rtimeoutcb, 0.,
+ timeoutconf.stream_read);
+ ev_timer_init(&upstream_wtimer_, &upstream_wtimeoutcb, 0.,
+ timeoutconf.stream_write);
+ ev_timer_init(&downstream_rtimer_, &downstream_rtimeoutcb, 0.,
+ timeoutconf.stream_read);
+ ev_timer_init(&downstream_wtimer_, &downstream_wtimeoutcb, 0.,
+ timeoutconf.stream_write);
+
+ upstream_rtimer_.data = this;
+ upstream_wtimer_.data = this;
+ downstream_rtimer_.data = this;
+ downstream_wtimer_.data = this;
+
+ rcbufs_.reserve(32);
+#ifdef ENABLE_HTTP3
+ rcbufs3_.reserve(32);
+#endif // ENABLE_HTTP3
+}
+
+Downstream::~Downstream() {
+ if (LOG_ENABLED(INFO)) {
+ DLOG(INFO, this) << "Deleting";
+ }
+
+ // check nullptr for unittest
+ if (upstream_) {
+ auto loop = upstream_->get_client_handler()->get_loop();
+
+ ev_timer_stop(loop, &upstream_rtimer_);
+ ev_timer_stop(loop, &upstream_wtimer_);
+ ev_timer_stop(loop, &downstream_rtimer_);
+ ev_timer_stop(loop, &downstream_wtimer_);
+
+#ifdef HAVE_MRUBY
+ auto handler = upstream_->get_client_handler();
+ auto worker = handler->get_worker();
+ auto mruby_ctx = worker->get_mruby_context();
+
+ mruby_ctx->delete_downstream(this);
+#endif // HAVE_MRUBY
+ }
+
+#ifdef HAVE_MRUBY
+ if (dconn_) {
+ const auto &group = dconn_->get_downstream_addr_group();
+ if (group) {
+ const auto &mruby_ctx = group->shared_addr->mruby_ctx;
+ mruby_ctx->delete_downstream(this);
+ }
+ }
+#endif // HAVE_MRUBY
+
+ // DownstreamConnection may refer to this object. Delete it now
+ // explicitly.
+ dconn_.reset();
+
+#ifdef ENABLE_HTTP3
+ for (auto rcbuf : rcbufs3_) {
+ nghttp3_rcbuf_decref(rcbuf);
+ }
+#endif // ENABLE_HTTP3
+
+ for (auto rcbuf : rcbufs_) {
+ nghttp2_rcbuf_decref(rcbuf);
+ }
+
+ if (LOG_ENABLED(INFO)) {
+ DLOG(INFO, this) << "Deleted";
+ }
+}
+
+int Downstream::attach_downstream_connection(
+ std::unique_ptr<DownstreamConnection> dconn) {
+ if (dconn->attach_downstream(this) != 0) {
+ return -1;
+ }
+
+ dconn_ = std::move(dconn);
+
+ return 0;
+}
+
+void Downstream::detach_downstream_connection() {
+ if (!dconn_) {
+ return;
+ }
+
+#ifdef HAVE_MRUBY
+ const auto &group = dconn_->get_downstream_addr_group();
+ if (group) {
+ const auto &mruby_ctx = group->shared_addr->mruby_ctx;
+ mruby_ctx->delete_downstream(this);
+ }
+#endif // HAVE_MRUBY
+
+ dconn_->detach_downstream(this);
+
+ auto handler = dconn_->get_client_handler();
+
+ handler->pool_downstream_connection(
+ std::unique_ptr<DownstreamConnection>(dconn_.release()));
+}
+
+DownstreamConnection *Downstream::get_downstream_connection() {
+ return dconn_.get();
+}
+
+std::unique_ptr<DownstreamConnection> Downstream::pop_downstream_connection() {
+#ifdef HAVE_MRUBY
+ if (!dconn_) {
+ return nullptr;
+ }
+
+ const auto &group = dconn_->get_downstream_addr_group();
+ if (group) {
+ const auto &mruby_ctx = group->shared_addr->mruby_ctx;
+ mruby_ctx->delete_downstream(this);
+ }
+#endif // HAVE_MRUBY
+
+ return std::unique_ptr<DownstreamConnection>(dconn_.release());
+}
+
+void Downstream::pause_read(IOCtrlReason reason) {
+ if (dconn_) {
+ dconn_->pause_read(reason);
+ }
+}
+
+int Downstream::resume_read(IOCtrlReason reason, size_t consumed) {
+ if (dconn_) {
+ return dconn_->resume_read(reason, consumed);
+ }
+
+ return 0;
+}
+
+void Downstream::force_resume_read() {
+ if (dconn_) {
+ dconn_->force_resume_read();
+ }
+}
+
+namespace {
+const HeaderRefs::value_type *
+search_header_linear_backwards(const HeaderRefs &headers,
+ const StringRef &name) {
+ for (auto it = headers.rbegin(); it != headers.rend(); ++it) {
+ auto &kv = *it;
+ if (kv.name == name) {
+ return &kv;
+ }
+ }
+ return nullptr;
+}
+} // namespace
+
+StringRef Downstream::assemble_request_cookie() {
+ size_t len = 0;
+
+ for (auto &kv : req_.fs.headers()) {
+ if (kv.token != http2::HD_COOKIE || kv.value.empty()) {
+ continue;
+ }
+
+ len += kv.value.size() + str_size("; ");
+ }
+
+ auto iov = make_byte_ref(balloc_, len + 1);
+ auto p = iov.base;
+
+ for (auto &kv : req_.fs.headers()) {
+ if (kv.token != http2::HD_COOKIE || kv.value.empty()) {
+ continue;
+ }
+
+ auto end = std::end(kv.value);
+ for (auto it = std::begin(kv.value) + kv.value.size();
+ it != std::begin(kv.value); --it) {
+ auto c = *(it - 1);
+ if (c == ' ' || c == ';') {
+ continue;
+ }
+ end = it;
+ break;
+ }
+
+ p = std::copy(std::begin(kv.value), end, p);
+ p = util::copy_lit(p, "; ");
+ }
+
+ // cut trailing "; "
+ if (p - iov.base >= 2) {
+ p -= 2;
+ }
+
+ return StringRef{iov.base, p};
+}
+
+uint32_t Downstream::find_affinity_cookie(const StringRef &name) {
+ for (auto &kv : req_.fs.headers()) {
+ if (kv.token != http2::HD_COOKIE) {
+ continue;
+ }
+
+ for (auto it = std::begin(kv.value); it != std::end(kv.value);) {
+ if (*it == '\t' || *it == ' ' || *it == ';') {
+ ++it;
+ continue;
+ }
+
+ auto end = std::find(it, std::end(kv.value), '=');
+ if (end == std::end(kv.value)) {
+ return 0;
+ }
+
+ if (!util::streq(name, StringRef{it, end})) {
+ it = std::find(it, std::end(kv.value), ';');
+ continue;
+ }
+
+ it = std::find(end + 1, std::end(kv.value), ';');
+ auto val = StringRef{end + 1, it};
+ if (val.size() != 8) {
+ return 0;
+ }
+ uint32_t h = 0;
+ for (auto c : val) {
+ auto n = util::hex_to_uint(c);
+ if (n == 256) {
+ return 0;
+ }
+ h <<= 4;
+ h += n;
+ }
+ affinity_cookie_ = h;
+ return h;
+ }
+ }
+ return 0;
+}
+
+size_t Downstream::count_crumble_request_cookie() {
+ size_t n = 0;
+ for (auto &kv : req_.fs.headers()) {
+ if (kv.token != http2::HD_COOKIE) {
+ continue;
+ }
+
+ for (auto it = std::begin(kv.value); it != std::end(kv.value);) {
+ if (*it == '\t' || *it == ' ' || *it == ';') {
+ ++it;
+ continue;
+ }
+
+ it = std::find(it, std::end(kv.value), ';');
+
+ ++n;
+ }
+ }
+ return n;
+}
+
+void Downstream::crumble_request_cookie(std::vector<nghttp2_nv> &nva) {
+ for (auto &kv : req_.fs.headers()) {
+ if (kv.token != http2::HD_COOKIE) {
+ continue;
+ }
+
+ for (auto it = std::begin(kv.value); it != std::end(kv.value);) {
+ if (*it == '\t' || *it == ' ' || *it == ';') {
+ ++it;
+ continue;
+ }
+
+ auto first = it;
+
+ it = std::find(it, std::end(kv.value), ';');
+
+ nva.push_back({(uint8_t *)"cookie", (uint8_t *)first, str_size("cookie"),
+ (size_t)(it - first),
+ (uint8_t)(NGHTTP2_NV_FLAG_NO_COPY_NAME |
+ NGHTTP2_NV_FLAG_NO_COPY_VALUE |
+ (kv.no_index ? NGHTTP2_NV_FLAG_NO_INDEX : 0))});
+ }
+ }
+}
+
+namespace {
+void add_header(size_t &sum, HeaderRefs &headers, const StringRef &name,
+ const StringRef &value, bool no_index, int32_t token) {
+ sum += name.size() + value.size();
+ headers.emplace_back(name, value, no_index, token);
+}
+} // namespace
+
+namespace {
+StringRef alloc_header_name(BlockAllocator &balloc, const StringRef &name) {
+ auto iov = make_byte_ref(balloc, name.size() + 1);
+ auto p = iov.base;
+ p = std::copy(std::begin(name), std::end(name), p);
+ util::inp_strlower(iov.base, p);
+ *p = '\0';
+
+ return StringRef{iov.base, p};
+}
+} // namespace
+
+namespace {
+void append_last_header_key(BlockAllocator &balloc, bool &key_prev, size_t &sum,
+ HeaderRefs &headers, const char *data, size_t len) {
+ assert(key_prev);
+ sum += len;
+ auto &item = headers.back();
+ auto name =
+ realloc_concat_string_ref(balloc, item.name, StringRef{data, len});
+
+ auto p = const_cast<uint8_t *>(name.byte());
+ util::inp_strlower(p + name.size() - len, p + name.size());
+
+ item.name = name;
+ item.token = http2::lookup_token(item.name);
+}
+} // namespace
+
+namespace {
+void append_last_header_value(BlockAllocator &balloc, bool &key_prev,
+ size_t &sum, HeaderRefs &headers,
+ const char *data, size_t len) {
+ key_prev = false;
+ sum += len;
+ auto &item = headers.back();
+ item.value =
+ realloc_concat_string_ref(balloc, item.value, StringRef{data, len});
+}
+} // namespace
+
+int FieldStore::parse_content_length() {
+ content_length = -1;
+
+ for (auto &kv : headers_) {
+ if (kv.token != http2::HD_CONTENT_LENGTH) {
+ continue;
+ }
+
+ auto len = util::parse_uint(kv.value);
+ if (len == -1) {
+ return -1;
+ }
+ if (content_length != -1) {
+ return -1;
+ }
+ content_length = len;
+ }
+ return 0;
+}
+
+const HeaderRefs::value_type *FieldStore::header(int32_t token) const {
+ for (auto it = headers_.rbegin(); it != headers_.rend(); ++it) {
+ auto &kv = *it;
+ if (kv.token == token) {
+ return &kv;
+ }
+ }
+ return nullptr;
+}
+
+HeaderRefs::value_type *FieldStore::header(int32_t token) {
+ for (auto it = headers_.rbegin(); it != headers_.rend(); ++it) {
+ auto &kv = *it;
+ if (kv.token == token) {
+ return &kv;
+ }
+ }
+ return nullptr;
+}
+
+const HeaderRefs::value_type *FieldStore::header(const StringRef &name) const {
+ return search_header_linear_backwards(headers_, name);
+}
+
+void FieldStore::add_header_token(const StringRef &name, const StringRef &value,
+ bool no_index, int32_t token) {
+ shrpx::add_header(buffer_size_, headers_, name, value, no_index, token);
+}
+
+void FieldStore::alloc_add_header_name(const StringRef &name) {
+ auto name_ref = alloc_header_name(balloc_, name);
+ auto token = http2::lookup_token(name_ref);
+ add_header_token(name_ref, StringRef{}, false, token);
+ header_key_prev_ = true;
+}
+
+void FieldStore::append_last_header_key(const char *data, size_t len) {
+ shrpx::append_last_header_key(balloc_, header_key_prev_, buffer_size_,
+ headers_, data, len);
+}
+
+void FieldStore::append_last_header_value(const char *data, size_t len) {
+ shrpx::append_last_header_value(balloc_, header_key_prev_, buffer_size_,
+ headers_, data, len);
+}
+
+void FieldStore::clear_headers() {
+ headers_.clear();
+ header_key_prev_ = false;
+}
+
+void FieldStore::add_trailer_token(const StringRef &name,
+ const StringRef &value, bool no_index,
+ int32_t token) {
+ // Header size limit should be applied to all header and trailer
+ // fields combined.
+ shrpx::add_header(buffer_size_, trailers_, name, value, no_index, token);
+}
+
+void FieldStore::alloc_add_trailer_name(const StringRef &name) {
+ auto name_ref = alloc_header_name(balloc_, name);
+ auto token = http2::lookup_token(name_ref);
+ add_trailer_token(name_ref, StringRef{}, false, token);
+ trailer_key_prev_ = true;
+}
+
+void FieldStore::append_last_trailer_key(const char *data, size_t len) {
+ shrpx::append_last_header_key(balloc_, trailer_key_prev_, buffer_size_,
+ trailers_, data, len);
+}
+
+void FieldStore::append_last_trailer_value(const char *data, size_t len) {
+ shrpx::append_last_header_value(balloc_, trailer_key_prev_, buffer_size_,
+ trailers_, data, len);
+}
+
+void FieldStore::erase_content_length_and_transfer_encoding() {
+ for (auto &kv : headers_) {
+ switch (kv.token) {
+ case http2::HD_CONTENT_LENGTH:
+ case http2::HD_TRANSFER_ENCODING:
+ kv.name = StringRef{};
+ kv.token = -1;
+ break;
+ }
+ }
+}
+
+void Downstream::set_request_start_time(
+ std::chrono::high_resolution_clock::time_point time) {
+ request_start_time_ = std::move(time);
+}
+
+const std::chrono::high_resolution_clock::time_point &
+Downstream::get_request_start_time() const {
+ return request_start_time_;
+}
+
+void Downstream::reset_upstream(Upstream *upstream) {
+ upstream_ = upstream;
+ if (dconn_) {
+ dconn_->on_upstream_change(upstream);
+ }
+}
+
+Upstream *Downstream::get_upstream() const { return upstream_; }
+
+void Downstream::set_stream_id(int64_t stream_id) { stream_id_ = stream_id; }
+
+int64_t Downstream::get_stream_id() const { return stream_id_; }
+
+void Downstream::set_request_state(DownstreamState state) {
+ request_state_ = state;
+}
+
+DownstreamState Downstream::get_request_state() const { return request_state_; }
+
+bool Downstream::get_chunked_request() const { return chunked_request_; }
+
+void Downstream::set_chunked_request(bool f) { chunked_request_ = f; }
+
+bool Downstream::request_buf_full() {
+ auto handler = upstream_->get_client_handler();
+ auto faddr = handler->get_upstream_addr();
+ auto worker = handler->get_worker();
+
+ // We don't check buffer size here for API endpoint.
+ if (faddr->alt_mode == UpstreamAltMode::API) {
+ return false;
+ }
+
+ if (dconn_) {
+ auto &downstreamconf = *worker->get_downstream_config();
+ return blocked_request_buf_.rleft() + request_buf_.rleft() >=
+ downstreamconf.request_buffer_size;
+ }
+
+ return false;
+}
+
+DefaultMemchunks *Downstream::get_request_buf() { return &request_buf_; }
+
+// Call this function after this object is attached to
+// Downstream. Otherwise, the program will crash.
+int Downstream::push_request_headers() {
+ if (!dconn_) {
+ DLOG(INFO, this) << "dconn_ is NULL";
+ return -1;
+ }
+ return dconn_->push_request_headers();
+}
+
+int Downstream::push_upload_data_chunk(const uint8_t *data, size_t datalen) {
+ req_.recv_body_length += datalen;
+
+ if (!dconn_ && !request_header_sent_) {
+ blocked_request_buf_.append(data, datalen);
+ req_.unconsumed_body_length += datalen;
+ return 0;
+ }
+
+ // Assumes that request headers have already been pushed to output
+ // buffer using push_request_headers().
+ if (!dconn_) {
+ DLOG(INFO, this) << "dconn_ is NULL";
+ return -1;
+ }
+ if (dconn_->push_upload_data_chunk(data, datalen) != 0) {
+ return -1;
+ }
+
+ req_.unconsumed_body_length += datalen;
+
+ return 0;
+}
+
+int Downstream::end_upload_data() {
+ if (!dconn_ && !request_header_sent_) {
+ blocked_request_data_eof_ = true;
+ return 0;
+ }
+ if (!dconn_) {
+ DLOG(INFO, this) << "dconn_ is NULL";
+ return -1;
+ }
+ return dconn_->end_upload_data();
+}
+
+void Downstream::rewrite_location_response_header(
+ const StringRef &upstream_scheme) {
+ auto hd = resp_.fs.header(http2::HD_LOCATION);
+ if (!hd) {
+ return;
+ }
+
+ if (request_downstream_host_.empty() || req_.authority.empty()) {
+ return;
+ }
+
+ http_parser_url u{};
+ auto rv = http_parser_parse_url(hd->value.c_str(), hd->value.size(), 0, &u);
+ if (rv != 0) {
+ return;
+ }
+
+ auto new_uri = http2::rewrite_location_uri(balloc_, hd->value, u,
+ request_downstream_host_,
+ req_.authority, upstream_scheme);
+
+ if (new_uri.empty()) {
+ return;
+ }
+
+ hd->value = new_uri;
+}
+
+bool Downstream::get_chunked_response() const { return chunked_response_; }
+
+void Downstream::set_chunked_response(bool f) { chunked_response_ = f; }
+
+int Downstream::on_read() {
+ if (!dconn_) {
+ DLOG(INFO, this) << "dconn_ is NULL";
+ return -1;
+ }
+ return dconn_->on_read();
+}
+
+void Downstream::set_response_state(DownstreamState state) {
+ response_state_ = state;
+}
+
+DownstreamState Downstream::get_response_state() const {
+ return response_state_;
+}
+
+DefaultMemchunks *Downstream::get_response_buf() { return &response_buf_; }
+
+bool Downstream::response_buf_full() {
+ if (dconn_) {
+ auto handler = upstream_->get_client_handler();
+ auto worker = handler->get_worker();
+ auto &downstreamconf = *worker->get_downstream_config();
+
+ return response_buf_.rleft() >= downstreamconf.response_buffer_size;
+ }
+
+ return false;
+}
+
+bool Downstream::validate_request_recv_body_length() const {
+ if (req_.fs.content_length == -1) {
+ return true;
+ }
+
+ if (req_.fs.content_length != req_.recv_body_length) {
+ if (LOG_ENABLED(INFO)) {
+ DLOG(INFO, this) << "request invalid bodylen: content-length="
+ << req_.fs.content_length
+ << ", received=" << req_.recv_body_length;
+ }
+ return false;
+ }
+
+ return true;
+}
+
+bool Downstream::validate_response_recv_body_length() const {
+ if (!expect_response_body() || resp_.fs.content_length == -1) {
+ return true;
+ }
+
+ if (resp_.fs.content_length != resp_.recv_body_length) {
+ if (LOG_ENABLED(INFO)) {
+ DLOG(INFO, this) << "response invalid bodylen: content-length="
+ << resp_.fs.content_length
+ << ", received=" << resp_.recv_body_length;
+ }
+ return false;
+ }
+
+ return true;
+}
+
+void Downstream::check_upgrade_fulfilled_http2() {
+ // This handles nonzero req_.connect_proto and h1 frontend requests
+ // WebSocket upgrade.
+ upgraded_ = (req_.method == HTTP_CONNECT ||
+ req_.connect_proto == ConnectProto::WEBSOCKET) &&
+ resp_.http_status / 100 == 2;
+}
+
+void Downstream::check_upgrade_fulfilled_http1() {
+ if (req_.method == HTTP_CONNECT) {
+ if (req_.connect_proto == ConnectProto::WEBSOCKET) {
+ if (resp_.http_status != 101) {
+ return;
+ }
+
+ // This is done for HTTP/2 frontend only.
+ auto accept = resp_.fs.header(http2::HD_SEC_WEBSOCKET_ACCEPT);
+ if (!accept) {
+ return;
+ }
+
+ std::array<uint8_t, base64::encode_length(20)> accept_buf;
+ auto expected =
+ http2::make_websocket_accept_token(accept_buf.data(), ws_key_);
+
+ upgraded_ = expected != "" && expected == accept->value;
+ } else {
+ upgraded_ = resp_.http_status / 100 == 2;
+ }
+
+ return;
+ }
+
+ if (resp_.http_status == 101) {
+ // TODO Do more strict checking for upgrade headers
+ upgraded_ = req_.upgrade_request;
+
+ return;
+ }
+}
+
+void Downstream::inspect_http2_request() {
+ if (req_.method == HTTP_CONNECT) {
+ req_.upgrade_request = true;
+ }
+}
+
+void Downstream::inspect_http1_request() {
+ if (req_.method == HTTP_CONNECT) {
+ req_.upgrade_request = true;
+ } else if (req_.http_minor > 0) {
+ auto upgrade = req_.fs.header(http2::HD_UPGRADE);
+ if (upgrade) {
+ const auto &val = upgrade->value;
+ // TODO Perform more strict checking for upgrade headers
+ if (util::streq_l(NGHTTP2_CLEARTEXT_PROTO_VERSION_ID, val.c_str(),
+ val.size())) {
+ req_.http2_upgrade_seen = true;
+ } else {
+ req_.upgrade_request = true;
+
+ // TODO Should we check Sec-WebSocket-Key, and
+ // Sec-WebSocket-Version as well?
+ if (util::strieq_l("websocket", val)) {
+ req_.connect_proto = ConnectProto::WEBSOCKET;
+ }
+ }
+ }
+ }
+ auto transfer_encoding = req_.fs.header(http2::HD_TRANSFER_ENCODING);
+ if (transfer_encoding) {
+ req_.fs.content_length = -1;
+ if (util::iends_with_l(transfer_encoding->value, "chunked")) {
+ chunked_request_ = true;
+ }
+ }
+
+ auto expect = req_.fs.header(http2::HD_EXPECT);
+ expect_100_continue_ =
+ expect &&
+ util::strieq(expect->value, StringRef::from_lit("100-continue"));
+}
+
+void Downstream::inspect_http1_response() {
+ auto transfer_encoding = resp_.fs.header(http2::HD_TRANSFER_ENCODING);
+ if (transfer_encoding) {
+ resp_.fs.content_length = -1;
+ if (util::iends_with_l(transfer_encoding->value, "chunked")) {
+ chunked_response_ = true;
+ }
+ }
+}
+
+void Downstream::reset_response() {
+ resp_.http_status = 0;
+ resp_.http_major = 1;
+ resp_.http_minor = 1;
+}
+
+bool Downstream::get_non_final_response() const {
+ return !upgraded_ && resp_.http_status / 100 == 1;
+}
+
+bool Downstream::supports_non_final_response() const {
+ return req_.http_major == 3 || req_.http_major == 2 ||
+ (req_.http_major == 1 && req_.http_minor == 1);
+}
+
+bool Downstream::get_upgraded() const { return upgraded_; }
+
+bool Downstream::get_http2_upgrade_request() const {
+ return req_.http2_upgrade_seen && req_.fs.header(http2::HD_HTTP2_SETTINGS) &&
+ response_state_ == DownstreamState::INITIAL;
+}
+
+StringRef Downstream::get_http2_settings() const {
+ auto http2_settings = req_.fs.header(http2::HD_HTTP2_SETTINGS);
+ if (!http2_settings) {
+ return StringRef{};
+ }
+ return http2_settings->value;
+}
+
+void Downstream::set_downstream_stream_id(int64_t stream_id) {
+ downstream_stream_id_ = stream_id;
+}
+
+int64_t Downstream::get_downstream_stream_id() const {
+ return downstream_stream_id_;
+}
+
+uint32_t Downstream::get_response_rst_stream_error_code() const {
+ return response_rst_stream_error_code_;
+}
+
+void Downstream::set_response_rst_stream_error_code(uint32_t error_code) {
+ response_rst_stream_error_code_ = error_code;
+}
+
+void Downstream::set_expect_final_response(bool f) {
+ expect_final_response_ = f;
+}
+
+bool Downstream::get_expect_final_response() const {
+ return expect_final_response_;
+}
+
+bool Downstream::expect_response_body() const {
+ return !resp_.headers_only &&
+ http2::expect_response_body(req_.method, resp_.http_status);
+}
+
+bool Downstream::expect_response_trailer() const {
+ // In HTTP/2, if final response HEADERS does not bear END_STREAM it
+ // is possible trailer fields might come, regardless of request
+ // method or status code.
+ return !resp_.headers_only &&
+ (resp_.http_major == 3 || resp_.http_major == 2);
+}
+
+namespace {
+void reset_timer(struct ev_loop *loop, ev_timer *w) { ev_timer_again(loop, w); }
+} // namespace
+
+namespace {
+void try_reset_timer(struct ev_loop *loop, ev_timer *w) {
+ if (!ev_is_active(w)) {
+ return;
+ }
+ ev_timer_again(loop, w);
+}
+} // namespace
+
+namespace {
+void ensure_timer(struct ev_loop *loop, ev_timer *w) {
+ if (ev_is_active(w)) {
+ return;
+ }
+ ev_timer_again(loop, w);
+}
+} // namespace
+
+namespace {
+void disable_timer(struct ev_loop *loop, ev_timer *w) {
+ ev_timer_stop(loop, w);
+}
+} // namespace
+
+void Downstream::reset_upstream_rtimer() {
+ if (get_config()->http2.timeout.stream_read == 0.) {
+ return;
+ }
+ auto loop = upstream_->get_client_handler()->get_loop();
+ reset_timer(loop, &upstream_rtimer_);
+}
+
+void Downstream::reset_upstream_wtimer() {
+ auto loop = upstream_->get_client_handler()->get_loop();
+ auto &timeoutconf = get_config()->http2.timeout;
+
+ if (timeoutconf.stream_write != 0.) {
+ reset_timer(loop, &upstream_wtimer_);
+ }
+ if (timeoutconf.stream_read != 0.) {
+ try_reset_timer(loop, &upstream_rtimer_);
+ }
+}
+
+void Downstream::ensure_upstream_wtimer() {
+ if (get_config()->http2.timeout.stream_write == 0.) {
+ return;
+ }
+ auto loop = upstream_->get_client_handler()->get_loop();
+ ensure_timer(loop, &upstream_wtimer_);
+}
+
+void Downstream::disable_upstream_rtimer() {
+ if (get_config()->http2.timeout.stream_read == 0.) {
+ return;
+ }
+ auto loop = upstream_->get_client_handler()->get_loop();
+ disable_timer(loop, &upstream_rtimer_);
+}
+
+void Downstream::disable_upstream_wtimer() {
+ if (get_config()->http2.timeout.stream_write == 0.) {
+ return;
+ }
+ auto loop = upstream_->get_client_handler()->get_loop();
+ disable_timer(loop, &upstream_wtimer_);
+}
+
+void Downstream::reset_downstream_rtimer() {
+ if (get_config()->http2.timeout.stream_read == 0.) {
+ return;
+ }
+ auto loop = upstream_->get_client_handler()->get_loop();
+ reset_timer(loop, &downstream_rtimer_);
+}
+
+void Downstream::reset_downstream_wtimer() {
+ auto loop = upstream_->get_client_handler()->get_loop();
+ auto &timeoutconf = get_config()->http2.timeout;
+
+ if (timeoutconf.stream_write != 0.) {
+ reset_timer(loop, &downstream_wtimer_);
+ }
+ if (timeoutconf.stream_read != 0.) {
+ try_reset_timer(loop, &downstream_rtimer_);
+ }
+}
+
+void Downstream::ensure_downstream_wtimer() {
+ if (get_config()->http2.timeout.stream_write == 0.) {
+ return;
+ }
+ auto loop = upstream_->get_client_handler()->get_loop();
+ ensure_timer(loop, &downstream_wtimer_);
+}
+
+void Downstream::disable_downstream_rtimer() {
+ if (get_config()->http2.timeout.stream_read == 0.) {
+ return;
+ }
+ auto loop = upstream_->get_client_handler()->get_loop();
+ disable_timer(loop, &downstream_rtimer_);
+}
+
+void Downstream::disable_downstream_wtimer() {
+ if (get_config()->http2.timeout.stream_write == 0.) {
+ return;
+ }
+ auto loop = upstream_->get_client_handler()->get_loop();
+ disable_timer(loop, &downstream_wtimer_);
+}
+
+bool Downstream::accesslog_ready() const {
+ return !accesslog_written_ && resp_.http_status > 0;
+}
+
+void Downstream::add_retry() { ++num_retry_; }
+
+bool Downstream::no_more_retry() const { return num_retry_ > 50; }
+
+void Downstream::set_request_downstream_host(const StringRef &host) {
+ request_downstream_host_ = host;
+}
+
+void Downstream::set_request_pending(bool f) { request_pending_ = f; }
+
+bool Downstream::get_request_pending() const { return request_pending_; }
+
+void Downstream::set_request_header_sent(bool f) { request_header_sent_ = f; }
+
+bool Downstream::get_request_header_sent() const {
+ return request_header_sent_;
+}
+
+bool Downstream::request_submission_ready() const {
+ return (request_state_ == DownstreamState::HEADER_COMPLETE ||
+ request_state_ == DownstreamState::MSG_COMPLETE) &&
+ (request_pending_ || !request_header_sent_) &&
+ response_state_ == DownstreamState::INITIAL;
+}
+
+DispatchState Downstream::get_dispatch_state() const { return dispatch_state_; }
+
+void Downstream::set_dispatch_state(DispatchState s) { dispatch_state_ = s; }
+
+void Downstream::attach_blocked_link(BlockedLink *l) {
+ assert(!blocked_link_);
+
+ l->downstream = this;
+ blocked_link_ = l;
+}
+
+BlockedLink *Downstream::detach_blocked_link() {
+ auto link = blocked_link_;
+ blocked_link_ = nullptr;
+ return link;
+}
+
+bool Downstream::can_detach_downstream_connection() const {
+ // We should check request and response buffer. If request buffer
+ // is not empty, then we might leave downstream connection in weird
+ // state, especially for HTTP/1.1
+ return dconn_ && response_state_ == DownstreamState::MSG_COMPLETE &&
+ request_state_ == DownstreamState::MSG_COMPLETE && !upgraded_ &&
+ !resp_.connection_close && request_buf_.rleft() == 0;
+}
+
+DefaultMemchunks Downstream::pop_response_buf() {
+ return std::move(response_buf_);
+}
+
+void Downstream::set_assoc_stream_id(int64_t stream_id) {
+ assoc_stream_id_ = stream_id;
+}
+
+int64_t Downstream::get_assoc_stream_id() const { return assoc_stream_id_; }
+
+BlockAllocator &Downstream::get_block_allocator() { return balloc_; }
+
+void Downstream::add_rcbuf(nghttp2_rcbuf *rcbuf) {
+ nghttp2_rcbuf_incref(rcbuf);
+ rcbufs_.push_back(rcbuf);
+}
+
+#ifdef ENABLE_HTTP3
+void Downstream::add_rcbuf(nghttp3_rcbuf *rcbuf) {
+ nghttp3_rcbuf_incref(rcbuf);
+ rcbufs3_.push_back(rcbuf);
+}
+#endif // ENABLE_HTTP3
+
+void Downstream::set_downstream_addr_group(
+ const std::shared_ptr<DownstreamAddrGroup> &group) {
+ group_ = group;
+}
+
+void Downstream::set_addr(const DownstreamAddr *addr) { addr_ = addr; }
+
+const DownstreamAddr *Downstream::get_addr() const { return addr_; }
+
+void Downstream::set_accesslog_written(bool f) { accesslog_written_ = f; }
+
+void Downstream::renew_affinity_cookie(uint32_t h) {
+ affinity_cookie_ = h;
+ new_affinity_cookie_ = true;
+}
+
+uint32_t Downstream::get_affinity_cookie_to_send() const {
+ if (new_affinity_cookie_) {
+ return affinity_cookie_;
+ }
+ return 0;
+}
+
+DefaultMemchunks *Downstream::get_blocked_request_buf() {
+ return &blocked_request_buf_;
+}
+
+bool Downstream::get_blocked_request_data_eof() const {
+ return blocked_request_data_eof_;
+}
+
+void Downstream::set_blocked_request_data_eof(bool f) {
+ blocked_request_data_eof_ = f;
+}
+
+void Downstream::set_ws_key(const StringRef &key) { ws_key_ = key; }
+
+bool Downstream::get_expect_100_continue() const {
+ return expect_100_continue_;
+}
+
+bool Downstream::get_stop_reading() const { return stop_reading_; }
+
+void Downstream::set_stop_reading(bool f) { stop_reading_ = f; }
+
+} // namespace shrpx