Line data Source code
1 : #include "server_connection.hpp"
2 : #include "server_base.hpp"
3 : #include "request.hpp"
4 : #include "response_observer.hpp"
5 : #include "../../asio/sockets/tcp_socket.hpp"
6 : #include "../../asio/sockets/unix_socket.hpp"
7 : #include "../../util/logger.hpp"
8 : #include <boost/asio/bind_cancellation_slot.hpp>
9 : #include <boost/asio/defer.hpp>
10 : #include <boost/asio/post.hpp>
11 :
12 : namespace thinger::http {
13 :
14 : namespace {
15 :
16 : // The Connection header of a response head, written as the head is serialized: whatever
17 : // decided the connection closes after the response, the head says so. A protocol switch
18 : // keeps the Connection header set for it (Upgrade)
19 23259 : void announce_disposition(const http_stream& stream, http_response& head) {
20 23259 : if (head.get_status() == http_response::status::switching_protocols) return;
21 19589 : head.set_keep_alive(!stream.reason());
22 : }
23 :
24 : // Lingering close once the server shuts down: brief, not to hold it
25 : constexpr std::chrono::milliseconds shutdown_linger_timeout{500};
26 :
27 : // Stop sending on the connection, reading on: TCP (TLS included) and Unix sockets
28 902 : bool shutdown_send(asio::socket& socket) {
29 902 : if (auto* tcp = dynamic_cast<asio::tcp_socket*>(&socket)) return tcp->shutdown_send();
30 246 : if (auto* local = dynamic_cast<asio::unix_socket*>(&socket)) return local->shutdown_send();
31 3 : return false;
32 : }
33 :
34 : // Read what arrives to drop it: from the TCP stream itself under TLS, whose records are
35 : // not decrypted (nor read by the TLS stream, whose reads were cancelled)
36 684536 : awaitable<io_result> read_to_drop(asio::socket& socket, uint8_t* buffer, size_t size) {
37 : if (auto* tcp = dynamic_cast<asio::tcp_socket*>(&socket)) {
38 : co_return co_await tcp->get_socket().async_read_some(boost::asio::buffer(buffer, size),
39 : use_nothrow_awaitable);
40 : }
41 : co_return co_await socket.read_some(buffer, size);
42 1369072 : }
43 :
44 : }
45 :
46 23792 : server_connection::server_connection(std::shared_ptr<asio::socket> socket,
47 23792 : std::shared_ptr<const connection_options> options)
48 23792 : : socket_(std::move(socket))
49 : // A member, called only while this lives. It does not keep the connection alive: once
50 : // neither the read loop nor a write holds it (e.g. the read loop stopped with a response
51 : // left unfinished), it is released and the socket closed
52 23792 : , timeout_timer_(socket_->get_io_context().get_executor(), [this] {
53 242 : if (auto self = weak_from_this().lock()) self->deadline_expired();
54 242 : })
55 47584 : , options_(options ? std::move(options) : std::make_shared<const connection_options>()) {
56 23792 : timeout_ = options_->idle_timeout;
57 23792 : LOG_DEBUG("created http server connection");
58 23792 : }
59 :
60 23792 : server_connection::~server_connection() {
61 23792 : LOG_DEBUG("releasing http server connection");
62 : // The requests may outlive the connection, with their streams
63 34420 : for (; !request_queue_.empty(); request_queue_.pop_front()) request_queue_.front()->detach();
64 23792 : }
65 :
66 23786 : void server_connection::track(const std::shared_ptr<connection_tracker>& tracker) {
67 23786 : tracked_connection::track(tracker, weak_from_this(), socket_->get_io_context(), connection_type::http);
68 23786 : }
69 :
70 23792 : void server_connection::start() {
71 23792 : if (running_) return;
72 : // Accepted while the server shuts down (e.g. it was in a TLS handshake): nothing is
73 : // waiting for it yet
74 23792 : if (auto tracker = this->tracker(); tracker && tracker->current_phase() != connection_tracker::phase::serving) {
75 2 : untrack();
76 2 : socket_->close();
77 2 : return;
78 23792 : }
79 23790 : running_ = true;
80 :
81 : // Its first request starts now: a client that sends nothing is bounded by its header deadline
82 23790 : last_activity_ = request_start_ = clock::now();
83 23790 : reading_head_ = true;
84 23790 : arm_deadline();
85 :
86 : // The read loop runs the handlers of the requests: bound to the signal their scopes emit
87 : // (see start_request())
88 23790 : co_spawn(socket_->get_io_context(),
89 71370 : [self = shared_from_this()]() -> awaitable<void> {
90 : co_await self->read_loop();
91 47580 : },
92 47580 : boost::asio::bind_cancellation_slot(handler_signal_.slot(), detached));
93 : // The peer watcher starts once a handler yields first (see wake_watcher())
94 : }
95 :
96 6140 : void server_connection::close() {
97 6140 : running_ = false;
98 6140 : timeout_timer_.cancel();
99 6140 : if (reader_wake_) reader_wake_->set();
100 6140 : stop_watcher();
101 6140 : socket_->close();
102 : // Nothing left to drain, even if the connection is still referenced
103 6140 : untrack();
104 6140 : }
105 :
106 5102 : void server_connection::close_after(const http_stream& stream) {
107 5102 : if (stream.input_pending()) {
108 897 : linger();
109 : } else {
110 4205 : close();
111 : }
112 5102 : }
113 :
114 929 : void server_connection::linger() {
115 1005 : if (lingering_) return;
116 929 : size_t max_bytes = options_->linger_max_bytes;
117 929 : std::chrono::milliseconds timeout = options_->linger_timeout;
118 : // The server shuts down: briefly, so it does not hold the shutdown
119 929 : if (frame_watch_ & watch_draining) timeout = std::min(timeout, shutdown_linger_timeout);
120 929 : if (max_bytes == 0 || timeout.count() <= 0 || !socket_->is_open() || !shutdown_send(*socket_)) {
121 76 : close();
122 76 : return;
123 : }
124 : // The connection stops serving: no more requests, nor its timeout. Whatever reads it ends
125 : // now (the read loop, or a handler still reading the body, which reads nothing more), so
126 : // only the graceful close reads it from now on
127 853 : lingering_ = true;
128 853 : running_ = false;
129 853 : timeout_timer_.cancel();
130 853 : if (reader_wake_) reader_wake_->set();
131 853 : stop_watcher();
132 853 : socket_->cancel();
133 853 : co_spawn(socket_->get_io_context(),
134 2559 : [self = shared_from_this(), max_bytes, timeout]() -> awaitable<void> {
135 : co_await self->drop_input(max_bytes, timeout);
136 1706 : },
137 : detached);
138 : }
139 :
140 853 : awaitable<void> server_connection::drop_input(size_t max_bytes, std::chrono::milliseconds timeout) {
141 : // Into the connection buffer, which the read loop no longer uses: the memory of a
142 : // connection does not grow while it drops input
143 : size_t dropped = 0;
144 853 : auto drop = [&]() -> awaitable<void> {
145 : while (dropped < max_bytes) {
146 : auto [ec, bytes] = co_await read_to_drop(*socket_, buffer_, std::min(MAX_BUFFER_SIZE, max_bytes - dropped));
147 : if (ec) co_return; // the client closed (or reset) the connection
148 : dropped += bytes;
149 : }
150 1706 : };
151 : boost::asio::steady_timer timer(socket_->get_io_context(), timeout);
152 : auto result = co_await (timer.async_wait(use_nothrow_awaitable) || drop());
153 : LOG_DEBUG("http server connection closed gracefully: {} bytes dropped{}", dropped,
154 : result.index() == 0 ? ", until the timeout" : dropped >= max_bytes ? ", up to the limit" : "");
155 : close();
156 1706 : }
157 :
158 23790 : awaitable<void> server_connection::read_loop() {
159 : auto self = shared_from_this();
160 :
161 : // Headers only: the handler layer reads the body
162 : request_parser_.set_headers_only(true);
163 :
164 : size_t buffered = 0; // bytes of valid data in buffer_
165 :
166 : while (running_ && socket_->is_open()) {
167 : // A pipelined request, already read with the one before it. The request before it ended
168 : // (its stream may be gone): none is in progress until this one starts
169 : bool pipelined = buffered > 0;
170 : in_progress_ = nullptr;
171 : // Not while the responses back up (the client does not read them): they would pile up.
172 : // Then from the top: the loop ends if the connection stopped meanwhile
173 : if (reading_paused()) [[unlikely]] {
174 : co_await wait_output();
175 : continue;
176 : }
177 : if (buffered == 0) {
178 : auto [ec, bytes] = co_await socket_->read_some(buffer_, MAX_BUFFER_SIZE);
179 : // (or stopped meanwhile: its head was answered at its deadline)
180 : if (ec || !running_) break;
181 : extend_timeout();
182 : buffered = bytes;
183 : }
184 : head_started(pipelined);
185 :
186 : uint8_t* begin = buffer_;
187 : uint8_t* end = buffer_ + buffered;
188 : boost::tribool result = request_parser_.parse(begin, end);
189 : if (boost::indeterminate(result)) {
190 : // Every byte went to the parser: more are needed, within the header deadline
191 : buffered = 0;
192 : head_incomplete();
193 : continue;
194 : }
195 : reading_head_ = false;
196 : if (!result) {
197 : reject_request();
198 : break;
199 : }
200 :
201 : auto [stream, req] = start_request(self, begin, static_cast<size_t>(end - begin));
202 :
203 : // The client is watched while the request is handled, and until its response starts
204 : watch(stream);
205 :
206 : // The handler decides how to read the body
207 : if (handler_) {
208 : co_await handler_(req);
209 : // Not once the response closed the connection (or it was taken over): the
210 : // timer would be armed just to be cancelled
211 : extend_timeout();
212 : }
213 :
214 : // A response sent later, from a copy of the response, may still take over the
215 : // connection: wait for it to start before reading what follows this request
216 : if (!stream->responded()) co_await wait_response(*stream);
217 :
218 : // The read loop reads the socket again, alone: a read of the watcher in progress
219 : // (over TLS) ends first
220 : unwatch();
221 : while (watch_reading_) {
222 : watch_idle_->reset();
223 : co_await watch_idle_->wait();
224 : }
225 :
226 : // Read-ahead bytes the request did not consume belong to whatever follows it (they
227 : // may come from later socket reads, e.g. after a chunked body)
228 : if (stream->takes_over()) {
229 : hand_over_reader(*stream, *req);
230 : co_return;
231 : }
232 :
233 : // Without keep-alive, reading stops after this request. The leftover stays with the
234 : // request: its handler may still be reading the body (e.g. the request was answered
235 : // by its timeout)
236 : if (stream->disposition() == detail::after_response::close || !keep_pipelined(*req, buffered)) {
237 : // The response ended before the connection was to close (e.g. a body left unread
238 : // turned out too large to drop): close it now, gracefully if the body still comes.
239 : // A response not ended yet closes it once it ends
240 : linger_unread(*stream);
241 : break;
242 : }
243 : }
244 :
245 : stop_reading();
246 47580 : }
247 :
248 5196 : void server_connection::linger_unread(const http_stream& stream) {
249 5196 : if (request_queue_.empty() && stream.input_pending()) linger();
250 5196 : }
251 :
252 18032 : void server_connection::stop_reading() {
253 18032 : running_ = false;
254 18032 : reader_stopped_ = true;
255 18032 : in_progress_ = nullptr;
256 18032 : reading_head_ = false;
257 : // A response still waiting for the socket stays bounded by its write deadline: the client
258 : // may never read it (see frame_written() for those written afterwards)
259 18032 : if (write_waits_ && !lingering_ && socket_->is_open()) {
260 787 : arm_deadline();
261 : } else {
262 17245 : timeout_timer_.cancel();
263 : }
264 18032 : stop_watcher();
265 :
266 : // Draining, with every response written: done now, even if something still holds the
267 : // connection (e.g. a copy of the request). Unless it closes gracefully already
268 18032 : if ((frame_watch_ & watch_draining) && request_queue_.empty() && !lingering_) close();
269 18032 : }
270 :
271 23512 : std::pair<std::shared_ptr<http_stream>, std::shared_ptr<request>> server_connection::start_request(
272 : const std::shared_ptr<server_connection>& self, const uint8_t* read_ahead, size_t read_ahead_size) {
273 23512 : auto http_req = request_parser_.consume_request();
274 23512 : http_req->set_ssl(socket_->is_secure());
275 :
276 23512 : auto stream = std::make_shared<http_stream>(++request_id_, *http_req, socket_->get_io_context().get_executor());
277 : // Its handler runs on the read loop (which awaits handler_), one request at a time: its
278 : // scope interrupts it through the signal of the read loop
279 23512 : stream->bind_handler(&handler_signal_);
280 23512 : in_progress_ = stream.get();
281 : // Its body is on the connection until read to its end (where it ends may be unknown),
282 : // unless it arrived whole with the head, maybe with what follows it
283 46461 : bool input_pending = !http_req->has_valid_framing() || http_req->is_chunked_transfer() ||
284 22949 : read_ahead_size < http_req->pending_body_size();
285 23512 : stream->set_input_pending(input_pending, !input_pending && read_ahead_size > http_req->pending_body_size());
286 : // Read within the request deadline, which its route may change (see request_timed_out()); none
287 : // for a request received with its head
288 23512 : request_deadline_ = input_pending ? options_->request_timeout : std::chrono::milliseconds{0};
289 23512 : read_bounded_ = false;
290 23512 : if (input_pending) arm_deadline();
291 :
292 : // The client may wait for 100 Continue before sending the body: sent once the body is
293 : // read, unless the client already started sending it
294 23512 : if (http_req->get_expectation() == http_request::expectation::continue_100 &&
295 23512 : http_req->has_pending_body() && read_ahead_size == 0) [[unlikely]] {
296 234 : stream->set_continue(http_stream::continue_state::pending);
297 234 : frame_watch_ |= watch_continue;
298 : }
299 :
300 : // Queued in order, for pipelining
301 23512 : request_queue_.push_back(stream);
302 :
303 23512 : http_req->log("SERVER REQUEST", options_->logging);
304 :
305 23512 : auto req = std::make_shared<request>(make_source(*stream), stream, std::move(http_req));
306 23512 : if (read_ahead_size > 0) {
307 4478 : req->set_read_ahead(read_ahead, read_ahead_size);
308 : }
309 47024 : return {std::move(stream), std::move(req)};
310 23512 : }
311 :
312 372 : void server_connection::reject_request() {
313 : // Bad request, or a header section over the limits
314 372 : auto status = http_response::status::bad_request;
315 372 : if (request_parser_.header_section_too_large()) {
316 54 : LOG_DEBUG("http request header section too large");
317 54 : status = http_response::status::request_header_fields_too_large;
318 : } else {
319 318 : LOG_DEBUG("invalid http request");
320 : }
321 : // The rest of a header section too large may still be arriving (big cookies, say), and its
322 : // client is worth an answer; not that of a request that is not HTTP, as Go does
323 372 : answer_and_close(status, detail::close_reason::body_error,
324 : status == http_response::status::request_header_fields_too_large);
325 372 : }
326 :
327 4315 : void server_connection::hand_over_reader(http_stream& stream, request& req) {
328 4315 : auto leftover = req.take_read_ahead();
329 4315 : stream.takeover_reader_stopped(std::string(leftover.begin(), leftover.end()));
330 4315 : in_progress_ = nullptr;
331 4315 : complete_takeover(stream);
332 4315 : }
333 :
334 13339 : bool server_connection::keep_pipelined(request& req, size_t& buffered) {
335 : // The leftover is at most what was read at once: the connection buffer, or the body
336 : // reader framing reads
337 13339 : auto leftover = req.take_read_ahead();
338 : static_assert(body_reader::framing_read_size <= MAX_BUFFER_SIZE,
339 : "body framing reads must fit in the connection buffer, or the leftover is dropped");
340 13339 : if (leftover.size() > MAX_BUFFER_SIZE) {
341 0 : LOG_ERROR("pipelined data exceeds the connection buffer");
342 0 : return false;
343 : }
344 13339 : if (!leftover.empty()) {
345 3192 : std::memcpy(buffer_, leftover.data(), leftover.size());
346 : }
347 13339 : buffered = leftover.size();
348 13339 : return true;
349 13339 : }
350 :
351 51 : awaitable<void> server_connection::wait_response(const http_stream& stream) {
352 : // The watcher still watches the client meanwhile (see watch_peer()): a client resetting
353 : // the connection before a late response closes it now, not at the connection timeout
354 : frame_watch_ |= watch_response_start;
355 : if (!reader_wake_) reader_wake_.emplace(socket_->get_io_context().get_executor());
356 : while (!stream.responded() && running_ && socket_->is_open()) {
357 : reader_wake_->reset(); // woken by the next frame, not by one before
358 : co_await reader_wake_->wait();
359 : }
360 : frame_watch_ &= ~watch_response_start;
361 102 : }
362 :
363 2197 : awaitable<void> server_connection::wait_output() {
364 : // Within the deadlines of the connection: the write deadline while the front response waits
365 : // for the socket, the idle one otherwise. Its close (or a stop) wakes it
366 : output_waited_ = true;
367 : if (!reader_wake_) reader_wake_.emplace(socket_->get_io_context().get_executor());
368 : while (output_backed_up() && running_ && socket_->is_open()) {
369 : reader_wake_->reset();
370 : co_await reader_wake_->wait();
371 : }
372 : output_waited_ = false;
373 4394 : }
374 :
375 23512 : void server_connection::watch(const std::shared_ptr<http_stream>& stream) {
376 23512 : watched_ = stream;
377 : // The watcher wakes once the handler yields, unless the response ended meanwhile: a
378 : // handler answering at once costs a deferred call (on this thread, no lock), not a round
379 : // of the watcher
380 23512 : boost::asio::defer(socket_->get_io_context(), [self = shared_from_this(), watched = stream.get()] {
381 23105 : if (self->watched_.get() == watched && !watched->response_ended()) self->wake_watcher();
382 23105 : });
383 23512 : }
384 :
385 2886 : void server_connection::wake_watcher() {
386 2886 : if (watch_wake_) {
387 127 : watch_wake_->set();
388 127 : return;
389 : }
390 : // Its first need: a connection whose handlers all answer at once never starts it. It runs
391 : // while the read loop runs (stopped through its slot, see unwatch())
392 2759 : auto executor = socket_->get_io_context().get_executor();
393 2759 : watch_wake_.emplace(executor);
394 2759 : watch_idle_.emplace(executor);
395 2759 : co_spawn(socket_->get_io_context(),
396 8277 : [self = shared_from_this()]() -> awaitable<void> {
397 : co_await self->watch_loop();
398 5518 : },
399 5518 : boost::asio::bind_cancellation_slot(watch_stop_.slot(), detached));
400 2759 : }
401 :
402 52187 : void server_connection::unwatch() {
403 52187 : watched_ = nullptr;
404 52187 : if (watch_waiting_) watch_stop_.emit(boost::asio::cancellation_type::terminal);
405 52187 : }
406 :
407 29337 : void server_connection::stop_watcher() {
408 29337 : unwatch();
409 29337 : if (watch_wake_) watch_wake_->set();
410 29337 : }
411 :
412 3654 : bool server_connection::watching(const http_stream& stream) const {
413 : // Not a stream taking over the connection: its socket goes to a new owner
414 3654 : return watched_.get() == &stream && running_ && !lingering_ && socket_->is_open() && !stream.takes_over();
415 : }
416 :
417 2759 : awaitable<void> server_connection::watch_loop() {
418 : // Stopped through its slot (see unwatch()), which ends the wait in progress, and nothing else
419 : co_await boost::asio::this_coro::throw_if_cancelled(false);
420 : while (running_) {
421 : co_await boost::asio::this_coro::reset_cancellation_state();
422 : if (auto stream = watched_) {
423 : co_await watch_peer(*stream);
424 : if (watched_ == stream) watched_ = nullptr;
425 : continue;
426 : }
427 : watch_wake_->reset();
428 : co_await watch_wake_->wait();
429 : }
430 5518 : }
431 :
432 2884 : awaitable<void> server_connection::watch_peer(http_stream& stream) {
433 : // (woken after the handler yields, see watch())
434 : if (!watching(stream) || stream.response_ended()) co_return;
435 :
436 : // The body is read by the request: the socket is read here only once it was read to its
437 : // end (or dropped), never meanwhile
438 : if (stream.input_pending()) {
439 : watch_waiting_ = true;
440 : bool read = co_await stream.wait_input_read();
441 : watch_waiting_ = false;
442 : if (!read || !watching(stream)) co_return;
443 : }
444 : // What follows the request was read along with it (a pipelined request): it stays for
445 : // the next request
446 : if (stream.input_follows()) co_return;
447 :
448 : watch_waiting_ = true;
449 : auto ec = co_await socket_->wait(boost::asio::socket_base::wait_read);
450 : watch_waiting_ = false;
451 : // Stopped, or a socket that cannot wait (e.g. one defined by an application)
452 : if (ec || !watching(stream)) co_return;
453 :
454 : watch_reading_ = true;
455 : ec = co_await socket_->peer_closed();
456 : watch_reading_ = false;
457 : watch_idle_->set();
458 :
459 : // Data (the next request, or data for a takeover) stays readable from the socket. A
460 : // half-close after the request is no leaving (RFC 9112 section 9.6): its response is
461 : // written, and the connection closed after it. Anything else is the client gone (a
462 : // reset): unless the connection closed meanwhile
463 : if (!ec || ec == boost::asio::error::operation_not_supported || ec == boost::asio::error::operation_aborted) {
464 : co_return;
465 : }
466 : if (ec == boost::asio::error::eof) {
467 : stream.must_close(detail::close_reason::client_asked);
468 : co_return;
469 : }
470 : LOG_DEBUG("http client left while its request was handled: {}", ec.message());
471 : stream.cancel(cancel_reason::client_gone);
472 : if (!lingering_) close();
473 5768 : }
474 :
475 564 : void server_connection::observe_frame(http_stream& stream, http_frame& frame) {
476 : // The head of the response (an interim response is not one): on the server, a response
477 564 : if (frame.kind() == frame_kind::head) {
478 408 : auto& head = static_cast<http_response&>(frame);
479 408 : stream.observer()->head(head.get_status(), !frame.end_stream() && !stream.takes_over());
480 : }
481 564 : }
482 :
483 38973 : void server_connection::frame_written(http_stream& stream, const std::shared_ptr<http_frame>& frame,
484 : const boost::system::error_code& ec) {
485 38973 : bool written = !ec;
486 : // A write the client reset (or whose connection broke) tells that the client is gone
487 38973 : if (ec && (ec == boost::asio::error::connection_reset || ec == boost::asio::error::broken_pipe)) [[unlikely]] {
488 121 : stream.cancel(cancel_reason::client_gone);
489 : }
490 :
491 : // Whoever waits for the frame (a streamed response) may write the next one
492 38973 : frame->notify_written(written);
493 :
494 : // The response ended (or cannot be written anymore), for the response hook
495 38973 : if (auto* observer = stream.observer(); observer && (frame->end_stream() || !written)) [[unlikely]] {
496 813 : observer->ended(!written ? response_outcome::aborted
497 405 : : stream.takes_over() ? response_outcome::taken_over : response_outcome::completed);
498 : }
499 :
500 : // Reset timeout on activity. The timer holds a weak reference: once the read loop stopped
501 : // (the connection closes after this response), only a pending write keeps the connection,
502 : // so a response still unfinished when its queued frames are written (e.g. chunks written
503 : // from detached code after the handler returned) loses the connection then
504 38973 : last_activity_ = clock::now();
505 38973 : arm_deadline();
506 :
507 38973 : if (frame->end_stream()) {
508 23111 : stream.set_response_ended();
509 : // Response of a request taking over the connection: the socket may now belong to
510 : // someone else, never close it from here
511 23111 : if (stream.takes_over()) {
512 4312 : stream.takeover_response_written();
513 4312 : complete_takeover(stream);
514 4312 : return;
515 : }
516 :
517 18799 : if (stream.disposition() == detail::after_response::close) {
518 5438 : if (written) {
519 5102 : close_after(stream);
520 : } else {
521 336 : close();
522 : }
523 13361 : } else if (!written) [[unlikely]] {
524 : // The connection broke (the client reset it, say): closed, so the requests it read
525 : // ahead (pipelined) are not handled for a client gone, as Go and Node do
526 96 : close();
527 : } else {
528 13265 : if (!request_queue_.empty()) {
529 13265 : stream.detach();
530 13265 : request_queue_.pop_front();
531 13265 : output_progressed();
532 : }
533 : }
534 : }
535 : }
536 :
537 46150 : void server_connection::process_output_queue() {
538 85267 : while (!writing_) {
539 103189 : if (request_queue_.empty()) return;
540 64594 : auto stream = request_queue_.front();
541 64594 : if (stream->empty_queue()) return;
542 :
543 39117 : auto frame = stream->current_frame();
544 39117 : stream->pop_frame();
545 :
546 : // The only writer of the Connection header of a response (see announce_disposition())
547 39117 : if (frame->kind() == frame_kind::head) announce_disposition(*stream, static_cast<http_response&>(*frame));
548 :
549 39117 : frame->log("SERVER RESPONSE", options_->logging);
550 39117 : auto* observer = stream->observer();
551 39117 : if (observer) [[unlikely]] observe_frame(*stream, *frame);
552 :
553 39117 : auto& buffers = write_buffers_;
554 39117 : buffers.clear();
555 39117 : frame->fill_buffer(buffers);
556 :
557 : // Most frames fit in the socket send buffer: write them right away, without waiting.
558 : // A write error is left to the read loop, as for a write that waited
559 39117 : boost::system::error_code ec;
560 39117 : size_t written = socket_->write_now(buffers, ec);
561 39117 : if (observer) [[unlikely]] observer->add_bytes(written);
562 39117 : if (written == boost::asio::buffer_size(buffers)) {
563 31450 : frame_written(*stream, frame, {});
564 31450 : continue;
565 : }
566 7667 : if (ec && ec != boost::asio::error::would_block && ec != boost::asio::error::operation_not_supported) {
567 328 : frame_written(*stream, frame, ec);
568 328 : continue;
569 : }
570 :
571 : // Write the rest once the socket accepts it, within the write deadline
572 7339 : asio::socket::consume(buffers, written);
573 7339 : write_started();
574 7339 : co_spawn(socket_->get_io_context(),
575 29356 : [this, self = shared_from_this(), stream = std::move(stream), frame = std::move(frame),
576 7339 : buffers = std::move(buffers)]() mutable -> awaitable<void> {
577 : auto [ec, bytes] = write_in_slices(buffers)
578 13890 : ? co_await socket_->write_slices(buffers, [this]{ write_progress_ = clock::now(); })
579 : : co_await socket_->write(buffers);
580 : write_waits_ = false;
581 : if (auto* observer = stream->observer()) [[unlikely]] observer->add_bytes(bytes);
582 : frame_written(*stream, frame, ec);
583 : writing_ = false;
584 :
585 : process_output_queue();
586 : output_progressed();
587 14678 : },
588 : detached);
589 96372 : }
590 : }
591 :
592 39102 : void server_connection::queue_frame(const std::shared_ptr<http_stream>& stream, std::shared_ptr<http_frame> frame) {
593 39102 : if (frame_watch_) [[unlikely]] watch_frame(*stream);
594 39102 : stream->add_frame(std::move(frame));
595 :
596 39102 : if (request_queue_.empty()) {
597 0 : LOG_ERROR("trying to send response without a pending request!");
598 0 : return;
599 : }
600 :
601 : // Only the front stream writes: responses go in the order of their requests
602 39102 : if (request_queue_.front()->id() == stream->id()) {
603 38805 : process_output_queue();
604 : }
605 : }
606 :
607 558 : void server_connection::watch_frame(http_stream& stream) {
608 : // The waiter wakes up once this call returns, with the frame queued
609 558 : if (frame_watch_ & watch_response_start) reader_wake_->set();
610 :
611 : // A response starting while the connection drains is the last one: its head tells the client
612 558 : if ((frame_watch_ & watch_draining) && !stream.responded() && !stream.takes_over()) {
613 417 : stream.must_close(detail::close_reason::draining);
614 : }
615 :
616 : // A response starting before the 100 Continue its client waits for: the client may send
617 : // the body or not, so where the next request starts is unknown. Not sent anymore, and the
618 : // connection is closed after the response, which tells the client (as Go and Node do)
619 558 : if ((frame_watch_ & watch_continue) && stream.get_continue() == http_stream::continue_state::pending) {
620 84 : frame_watch_ &= ~watch_continue;
621 84 : stream.set_continue(http_stream::continue_state::skipped);
622 84 : if (!stream.takes_over()) stream.must_close(detail::close_reason::continue_skipped);
623 : }
624 558 : }
625 :
626 150 : void server_connection::send_continue(http_stream& stream) {
627 150 : if (stream.get_continue() != http_stream::continue_state::pending || !running_) return;
628 150 : stream.set_continue(http_stream::continue_state::sent);
629 150 : frame_watch_ &= ~watch_continue;
630 : // Queued as the first frame of the stream, before its response: written once the
631 : // responses of the previous requests are
632 150 : stream.add_interim_frame(std::make_shared<detail::continue_frame>());
633 150 : if (!request_queue_.empty() && request_queue_.front().get() == &stream) {
634 150 : process_output_queue();
635 : }
636 : }
637 :
638 681 : void server_connection::shutdown_drain() {
639 681 : if (!tracked() || (frame_watch_ & watch_draining)) return;
640 681 : frame_watch_ |= watch_draining;
641 : // Closing gracefully: its response was written whole, so it does not hold the shutdown
642 681 : if (lingering_) {
643 9 : close();
644 9 : return;
645 : }
646 672 : if (!running_) return;
647 :
648 : // Handling a request: the connection is closed after its response (whose head tells the
649 : // client unless it already started), and the read loop stops after it
650 672 : if (in_progress_) {
651 417 : in_progress_->must_close(detail::close_reason::draining);
652 417 : return;
653 : }
654 :
655 : // Waiting for a request with nothing to write: idle
656 255 : if (request_queue_.empty() && request_parser_.idle()) {
657 207 : LOG_DEBUG("closing an idle http server connection: the server shuts down");
658 207 : close();
659 207 : return;
660 : }
661 :
662 : // A response still being written (the read loop went on to the next request): the
663 : // connection is closed after it. Or part of a request arrived: its response closes it
664 48 : if (!request_queue_.empty()) request_queue_.back()->must_close(detail::close_reason::draining);
665 : }
666 :
667 493 : void server_connection::shutdown_cancel() {
668 : // The handlers still running: that of the request being read, and those of the requests
669 : // waiting for their responses (interrupted once, see http_stream::cancel())
670 493 : if (in_progress_ && in_progress_->handler_running()) in_progress_->cancel(cancel_reason::shutdown);
671 872 : for (const auto& stream : request_queue_) {
672 379 : if (stream->handler_running()) stream->cancel(cancel_reason::shutdown);
673 : }
674 493 : }
675 :
676 493 : void server_connection::shutdown_close() {
677 493 : if (tracked()) close();
678 493 : }
679 :
680 34787 : void server_connection::handle_stream(std::shared_ptr<http_stream> stream,
681 : std::shared_ptr<http_frame> frame) {
682 : // Frames are queued in the order of the calls, from whatever threads they come (e.g.
683 : // chunks written from the IO thread after headers sent from another one): inline only
684 : // on the connection thread and with no frame of the stream still posted
685 34787 : auto& io_context = socket_->get_io_context();
686 34787 : if (io_context.get_executor().running_in_this_thread() && !stream->has_posted_frames()) {
687 34740 : queue_frame(stream, std::move(frame));
688 34740 : return;
689 : }
690 47 : stream->frame_posted();
691 47 : boost::asio::post(io_context,
692 47 : [this, self = shared_from_this(), stream = std::move(stream), frame = std::move(frame)]() mutable {
693 47 : stream->posted_frame_queued();
694 47 : queue_frame(stream, std::move(frame));
695 47 : });
696 : }
697 :
698 4315 : void server_connection::take_over(std::shared_ptr<http_stream> stream, std::shared_ptr<http_frame> response,
699 : takeover_handler handler) {
700 4315 : boost::asio::dispatch(socket_->get_io_context(),
701 12945 : [this, self = shared_from_this(), stream = std::move(stream), response = std::move(response),
702 4315 : handler = std::move(handler)]() mutable {
703 : // The read loop already stopped (the client left, the connection timed out...):
704 : // the connection is gone and cannot be handed over
705 4315 : if (reader_stopped_) {
706 0 : LOG_WARNING("connection closed before it could be taken over: the takeover handler is not called");
707 0 : return;
708 : }
709 4315 : stream->set_takeover(std::move(handler));
710 4315 : queue_frame(stream, std::move(response));
711 : });
712 4315 : }
713 :
714 411 : void server_connection::handle_stock_error(std::shared_ptr<http_stream> stream,
715 : http_response::status status) {
716 : // Its head tells what happens to the connection, as any other (see process_output_queue())
717 411 : handle_stream(stream, http_response::stock_http_reply(status));
718 411 : }
719 :
720 4312 : std::shared_ptr<asio::socket> server_connection::release_socket() {
721 : // The socket belongs to someone else from now on: the server neither drains nor closes it
722 4312 : untrack();
723 4312 : running_ = false;
724 4312 : stop_watcher();
725 4312 : socket_->cancel();
726 4306 : timeout_timer_.cancel();
727 4306 : return socket_;
728 : }
729 :
730 8627 : void server_connection::complete_takeover(http_stream& stream) {
731 : // Both the response must be written and the read loop stopped, in either order
732 8627 : auto [handler, buffered] = stream.release_takeover();
733 8627 : if (!handler) return;
734 :
735 : // From now on the connection counts as taken over, while the socket handed out lives.
736 : // Tracked before this one leaves, so a shutdown never sees the server without connections
737 : // meanwhile
738 4312 : std::shared_ptr<detail::taken_over_tracking> taken_over;
739 4312 : if (auto tracker = this->tracker()) {
740 4306 : if (tracker->current_phase() == connection_tracker::phase::closing) {
741 0 : LOG_WARNING("connection closed before it could be taken over: the server shuts down");
742 0 : close();
743 0 : return;
744 : }
745 4306 : taken_over = detail::taken_over_tracking::track(std::move(tracker), socket_);
746 4312 : }
747 4312 : auto socket = release_socket();
748 : // The handler gets the socket owning the tracking, and its scope (none if not tracked)
749 4306 : handler(taken_over ? taken_over->hand_out(std::move(buffered))
750 0 : : upgraded_connection{std::move(socket), std::move(buffered), {}});
751 : // Told now if the server started shutting down meanwhile (the handler may have set what
752 : // to do then)
753 4306 : if (taken_over) taken_over->handed_over();
754 8633 : }
755 :
756 3 : void server_connection::release() {
757 3 : boost::asio::dispatch(socket_->get_io_context(),
758 6 : [this, self = shared_from_this()] {
759 3 : close();
760 3 : });
761 3 : }
762 :
763 89621 : void server_connection::extend_timeout() {
764 89621 : if (!running_) return;
765 85228 : last_activity_ = clock::now();
766 85228 : arm_deadline();
767 : }
768 :
769 225 : void server_connection::hold_until(std::chrono::steady_clock::time_point time) {
770 : // Of the idle deadline only (see next_deadline()): not of those of the request
771 225 : hold_ = std::max(hold_, time);
772 225 : }
773 :
774 798 : std::shared_ptr<asio::socket> server_connection::get_socket() {
775 798 : return socket_;
776 : }
777 :
778 0 : void server_connection::update_connection_timeout(std::chrono::seconds timeout) {
779 0 : boost::asio::dispatch(socket_->get_io_context(),
780 0 : [this, self = shared_from_this(), timeout] {
781 0 : timeout_ = timeout;
782 0 : last_activity_ = clock::now();
783 0 : arm_deadline();
784 0 : });
785 0 : }
786 :
787 : }
|
