2 * nghttp2 - HTTP/2 C Library
4 * Copyright (c) 2012 Tatsuhiro Tsujikawa
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:
14 * The above copyright notice and this permission notice shall be
15 * included in all copies or substantial portions of the Software.
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.
25 #include "shrpx_downstream_queue.h"
30 #include "shrpx_downstream.h"
34 DownstreamQueue::HostEntry::HostEntry() : num_active(0) {}
36 DownstreamQueue::DownstreamQueue(size_t conn_max_per_host, bool unified_host)
37 : conn_max_per_host_(conn_max_per_host == 0
38 ? std::numeric_limits<size_t>::max()
40 unified_host_(unified_host) {}
42 DownstreamQueue::~DownstreamQueue() {}
44 void DownstreamQueue::add_pending(std::unique_ptr<Downstream> downstream) {
45 auto stream_id = downstream->get_stream_id();
46 pending_downstreams_[stream_id] = std::move(downstream);
49 void DownstreamQueue::add_failure(std::unique_ptr<Downstream> downstream) {
50 auto stream_id = downstream->get_stream_id();
51 failure_downstreams_[stream_id] = std::move(downstream);
54 DownstreamQueue::HostEntry &
55 DownstreamQueue::find_host_entry(const std::string &host) {
56 auto itr = host_entries_.find(host);
57 if (itr == std::end(host_entries_)) {
58 #ifdef HAVE_STD_MAP_EMPLACE
59 std::tie(itr, std::ignore) = host_entries_.emplace(host, HostEntry());
60 #else // !HAVE_STD_MAP_EMPLACE
62 std::tie(itr, std::ignore) = host_entries_.insert({host, HostEntry()});
63 #endif // !HAVE_STD_MAP_EMPLACE
69 DownstreamQueue::make_host_key(const std::string &host) const {
70 static std::string empty_key;
71 return unified_host_ ? empty_key : host;
75 DownstreamQueue::make_host_key(Downstream *downstream) const {
76 return make_host_key(downstream->get_request_http2_authority());
79 void DownstreamQueue::add_active(std::unique_ptr<Downstream> downstream) {
80 auto &ent = find_host_entry(make_host_key(downstream.get()));
83 auto stream_id = downstream->get_stream_id();
84 active_downstreams_[stream_id] = std::move(downstream);
87 void DownstreamQueue::add_blocked(std::unique_ptr<Downstream> downstream) {
88 auto &ent = find_host_entry(make_host_key(downstream.get()));
89 auto stream_id = downstream->get_stream_id();
90 ent.blocked.insert(stream_id);
91 blocked_downstreams_[stream_id] = std::move(downstream);
94 bool DownstreamQueue::can_activate(const std::string &host) const {
95 auto itr = host_entries_.find(make_host_key(host));
96 if (itr == std::end(host_entries_)) {
99 auto &ent = (*itr).second;
100 return ent.num_active < conn_max_per_host_;
104 std::unique_ptr<Downstream>
105 pop_downstream(DownstreamQueue::DownstreamMap::iterator i,
106 DownstreamQueue::DownstreamMap &downstreams) {
107 auto downstream = std::move((*i).second);
108 downstreams.erase(i);
114 bool remove_host_entry_if_empty(const DownstreamQueue::HostEntry &ent,
115 DownstreamQueue::HostEntryMap &host_entries,
116 const std::string &host) {
117 if (ent.blocked.empty() && ent.num_active == 0) {
118 host_entries.erase(host);
125 std::unique_ptr<Downstream> DownstreamQueue::pop_pending(int32_t stream_id) {
126 auto itr = pending_downstreams_.find(stream_id);
127 if (itr == std::end(pending_downstreams_)) {
130 return pop_downstream(itr, pending_downstreams_);
133 std::unique_ptr<Downstream>
134 DownstreamQueue::remove_and_pop_blocked(int32_t stream_id) {
135 auto kv = active_downstreams_.find(stream_id);
137 if (kv != std::end(active_downstreams_)) {
138 auto downstream = pop_downstream(kv, active_downstreams_);
139 auto &host = make_host_key(downstream.get());
140 auto &ent = find_host_entry(host);
143 if (remove_host_entry_if_empty(ent, host_entries_, host)) {
147 if (ent.blocked.empty() || ent.num_active >= conn_max_per_host_) {
151 auto next_stream_id = *std::begin(ent.blocked);
152 ent.blocked.erase(std::begin(ent.blocked));
154 auto itr = blocked_downstreams_.find(next_stream_id);
155 assert(itr != std::end(blocked_downstreams_));
157 auto next_downstream = pop_downstream(itr, blocked_downstreams_);
159 remove_host_entry_if_empty(ent, host_entries_, host);
161 return next_downstream;
164 kv = blocked_downstreams_.find(stream_id);
166 if (kv != std::end(blocked_downstreams_)) {
167 auto downstream = pop_downstream(kv, blocked_downstreams_);
168 auto &host = make_host_key(downstream.get());
169 auto &ent = find_host_entry(host);
170 ent.blocked.erase(stream_id);
172 remove_host_entry_if_empty(ent, host_entries_, host);
177 kv = pending_downstreams_.find(stream_id);
179 if (kv != std::end(pending_downstreams_)) {
180 pop_downstream(kv, pending_downstreams_);
184 kv = failure_downstreams_.find(stream_id);
186 if (kv != std::end(failure_downstreams_)) {
187 pop_downstream(kv, failure_downstreams_);
194 Downstream *DownstreamQueue::find(int32_t stream_id) {
195 auto kv = active_downstreams_.find(stream_id);
197 if (kv != std::end(active_downstreams_)) {
198 return (*kv).second.get();
201 kv = blocked_downstreams_.find(stream_id);
203 if (kv != std::end(blocked_downstreams_)) {
204 return (*kv).second.get();
207 kv = pending_downstreams_.find(stream_id);
209 if (kv != std::end(pending_downstreams_)) {
210 return (*kv).second.get();
213 kv = failure_downstreams_.find(stream_id);
215 if (kv != std::end(failure_downstreams_)) {
216 return (*kv).second.get();
222 const DownstreamQueue::DownstreamMap &
223 DownstreamQueue::get_active_downstreams() const {
224 return active_downstreams_;