Line data Source code
1 : #ifndef HTTP_STREAM_HPP
2 : #define HTTP_STREAM_HPP
3 :
4 : #include <atomic>
5 : #include <chrono>
6 : #include <list>
7 : #include <queue>
8 : #include <memory>
9 : #include <optional>
10 : #include <functional>
11 : #include <string>
12 : #include <string_view>
13 : #include <utility>
14 : #include <boost/asio/any_io_executor.hpp>
15 : #include <boost/asio/cancellation_signal.hpp>
16 : #include "../common/http_frame.hpp"
17 : #include "../../asio/async_event.hpp"
18 : #include "../../asio/ssl/tls_session.hpp"
19 : #include "../../util/types.hpp"
20 : #include "upgrade.hpp"
21 :
22 : namespace thinger::asio {
23 : class socket;
24 : }
25 :
26 : namespace thinger::http {
27 :
28 : class response_observer;
29 : class http_request;
30 :
31 : typedef uint32_t stream_id;
32 :
33 : // Why the scope of a request was cancelled (defined in request.hpp)
34 : enum class cancel_reason : uint8_t;
35 :
36 : namespace detail {
37 :
38 : // Why a connection closes after the response of a stream
39 : enum class close_reason : uint8_t {
40 : client_asked, // the request said Connection: close (or its client sends no more)
41 : http10, // an HTTP/1.0 request without keep-alive
42 : draining, // the server shuts down
43 : continue_skipped, // answered before the 100 Continue its client waits for
44 : body_unread, // a request body left unread that could not be dropped
45 : body_error, // a request (or its body) that could not be read: where it ends is unknown
46 : handler_failed, // the handler failed once its response had started (or before routing)
47 : route_timeout, // the route timed out with the request body still unread
48 : request_timeout, // the request was not received within its deadline (answered 408)
49 : framing_by_close, // a response body that ends when the connection closes
50 : aborted // a streamed response aborted: its body is incomplete
51 : };
52 :
53 : // What happens to the connection once the response of a stream is written
54 : enum class after_response : uint8_t { keep_alive, close, take_over };
55 :
56 : // Interim response asking the client to send the body (RFC 9110 section 15.2.1)
57 : class continue_frame : public http_frame {
58 : public:
59 156 : continue_frame() { set_last_frame(false); }
60 :
61 165 : frame_kind kind() const override { return frame_kind::interim; }
62 :
63 0 : size_t get_size() override { return 0; }
64 :
65 : void to_buffer(std::vector<boost::asio::const_buffer>& buffer) const override;
66 :
67 : void log(const char* scope, const body_logging&) const override;
68 : };
69 :
70 : }
71 :
72 : class server_connection;
73 :
74 : // The connection a request came from (see request::connection()): empty for a request
75 : // without one, or once it is gone
76 : struct connection_info {
77 : std::string peer_ip;
78 : std::string peer_port;
79 : std::string local_port;
80 : bool secure = false; // TLS
81 : };
82 :
83 : namespace detail {
84 :
85 : // Where a request reads its body from, and what it knows of its connection: the input
86 : // twin of response_sink, one per request, owned by it. A server connection reads its
87 : // client (connection_source, in connection_source.cpp); an in-memory dispatch reads
88 : // memory (memory_source)
89 : class request_source {
90 : public:
91 27619 : virtual ~request_source() = default;
92 :
93 : // Up to `size` bytes of what follows the request head (the body reader frames them):
94 : // none once nothing more comes, with the error why. The network source sends the
95 : // 100 Continue its client waits for first, and moves the connection deadline
96 : virtual awaitable<io_result> read_body_some(uint8_t* buffer, size_t size) = 0;
97 :
98 : virtual connection_info info() const = 0;
99 :
100 : // The TLS session of the connection: none without TLS or without a connection
101 6 : virtual std::optional<asio::tls_session> tls() const { return std::nullopt; }
102 :
103 : // Whether a request for `host` is misdirected on the connection (see
104 : // asio::socket::misdirected()): false without TLS or without a connection
105 15 : virtual bool misdirected(std::string_view /*host*/) const { return false; }
106 :
107 : // Keep the connection open at least until `time`, for a request with a timeout of
108 : // its own (see server_connection::hold_until()); nothing without a connection
109 39 : virtual void hold_until(std::chrono::steady_clock::time_point /*time*/) {}
110 :
111 : // The request deadline set by the route of the request (see route::request_timeout),
112 : // from the start of the request, zero for none (see
113 : // server_connection::set_request_deadline()); nothing without a connection
114 0 : virtual void set_request_deadline(std::chrono::milliseconds /*deadline*/) {}
115 :
116 : // The read deadline of the body (see request::set_read_deadline()), from the
117 : // thread reading it; nothing without a connection, as memory never waits
118 3 : virtual void set_read_deadline(std::chrono::steady_clock::time_point /*deadline*/) {}
119 :
120 : // The server connection, for the sink writing the response (none in memory)
121 3 : virtual std::weak_ptr<server_connection> connection() const { return {}; }
122 :
123 : // The io_context of the connection, for the sockets a reverse proxy opens to its
124 : // target (none in memory, or once the connection is gone)
125 9 : virtual boost::asio::io_context* io_context() const { return nullptr; }
126 : };
127 :
128 : }
129 :
130 : /**
131 : * A HTTP stream represents a communication channel inside a single HTTP connection. As a
132 : * single connection can be used for multiple HTTP request, each request will generate a new
133 : * HTTP stream. In HTTP 1.1, all requests in a connection must be answered in order, even if
134 : * the responses are generated in a different order. In HTTP 2.0, it is possible to answer
135 : * asynchronously to each single stream within a connection using a stream identifier.
136 : * TODO check HTTP 2.0 support.
137 : */
138 : class http_stream : public std::enable_shared_from_this<http_stream> {
139 :
140 : private:
141 : /**
142 : * Unique identifier within a http connection, generate for each request
143 : */
144 : stream_id stream_id_;
145 :
146 : /**
147 : * Queue for each HTTP frame composing a response. A response can be composed on several frames
148 : * i.e., while sending large files. A list, as a deque allocates its storage up front,
149 : * and most responses are a single frame
150 : */
151 : std::queue<std::shared_ptr<http_frame>, std::list<std::shared_ptr<http_frame>>> queue_;
152 :
153 : /**
154 : * Set when the response takes over the connection: receives the socket, and the
155 : * bytes read past the request, once the response is written and the read loop
156 : * stopped at this stream
157 : */
158 : takeover_handler takeover_;
159 : std::string takeover_buffer_;
160 : bool takes_over_ = false;
161 : bool takeover_response_written_ = false;
162 : bool takeover_reader_stopped_ = false;
163 :
164 : /**
165 : * Whether input of the request may still be unread on the connection: its body, until
166 : * read to the end, or the rest of a request that could not be parsed. A connection
167 : * closed after the response then closes gracefully (see server_connection::linger()).
168 : * Here, where it takes no room
169 : */
170 : bool input_pending_ = false;
171 :
172 : /**
173 : * Once the input is read: whether bytes of what follows the request (a pipelined
174 : * request) were already read past it. And the event set then, which the connection
175 : * waits for before it reads the socket itself (see server_connection::watch_peer()):
176 : * created by its first wait, set then if the input was read already
177 : */
178 : bool input_follows_ = false;
179 : std::optional<asio::async_event> input_read_;
180 :
181 : /**
182 : * Frames posted to the connection thread and not queued yet (frames are queued in
183 : * call order: none runs inline while one is posted)
184 : */
185 : std::atomic<unsigned> posted_frames_{0};
186 :
187 : /**
188 : * Whether the response has started: a frame was queued, or a takeover requested
189 : */
190 : bool responded_ = false;
191 :
192 : /**
193 : * Whether the last frame of the response was written (or failed to be)
194 : */
195 : bool response_ended_ = false;
196 :
197 : /**
198 : * Why the connection closes after the response: the first reason given, if any
199 : * (see disposition())
200 : */
201 : std::optional<detail::close_reason> close_reason_;
202 :
203 : /**
204 : * Told about the response (head, bytes written, end) while a response hook is set
205 : * (see http_application::on_response)
206 : */
207 : std::shared_ptr<response_observer> observer_;
208 :
209 : /**
210 : * Cancellation scope of the request (see cancel()). Everything but the reason belongs
211 : * to the executor of the stream: the one of its connection (or of its dispatch)
212 : */
213 : boost::asio::any_io_executor executor_;
214 : // none until cancelled; read from any thread (a handler may change executor)
215 : std::atomic<cancel_reason> cancelled_;
216 : // Set once cancelled, for request::client_gone(): created by its first wait, set then if
217 : // the scope was cancelled already
218 : std::optional<asio::async_event> cancel_event_;
219 : // Bound to the coroutine running the handler (see handler_slot()), unless that
220 : // coroutine is bound to a signal of its own, its owner's (see bind_handler())
221 : boost::asio::cancellation_signal handler_signal_;
222 : boost::asio::cancellation_signal* bound_signal_ = nullptr;
223 : // A coroutine about to run the handler is bound to the scope (see bind_handler())
224 : bool handler_bound_ = false;
225 : // The slot request::cancellation_slot() gives, for the awaits the handler binds to it
226 : boost::asio::cancellation_signal observer_signal_;
227 : // The reasons interrupting the handler, as bits 1 << reason (see route::cancel_on)
228 : uint8_t cancel_on_ = 0;
229 : // The handler is running: neither signal is emitted before or after it
230 : bool handler_running_ = false;
231 : // The handler signal was emitted
232 : bool handler_interrupted_ = false;
233 : // Emit it, on the executor of the stream
234 : void interrupt_handler();
235 :
236 : public:
237 : /**
238 : * 100 Continue of a request whose client waits for it before sending the body
239 : * (Expect: 100-continue): none expected, pending until the body is first read, sent,
240 : * or skipped because the response started first (the body may never come)
241 : */
242 : enum class continue_state : uint8_t { none, pending, sent, skipped };
243 :
244 : private:
245 : continue_state continue_ = continue_state::none;
246 :
247 : public:
248 : /**
249 : * Stream of `request` on `executor` (that of its connection), whose own Connection
250 : * header and HTTP version seed what happens to the connection after its response
251 : */
252 : http_stream(stream_id stream_id, const http_request& request, boost::asio::any_io_executor executor);
253 :
254 : /**
255 : * Stream whose response closes the connection, for `reason` (e.g. a request that could
256 : * not be parsed)
257 : */
258 : http_stream(stream_id stream_id, detail::close_reason reason, boost::asio::any_io_executor executor);
259 :
260 28066 : virtual ~http_stream() {}
261 :
262 : public:
263 :
264 : size_t get_queue_size() const;
265 :
266 : bool empty_queue() const;
267 :
268 : std::shared_ptr<http_frame> current_frame() const;
269 :
270 : void pop_frame();
271 :
272 : void add_frame(std::shared_ptr<http_frame> frame);
273 :
274 : /**
275 : * Queue an interim response (100 Continue): the response has not started yet
276 : */
277 : void add_interim_frame(std::shared_ptr<http_frame> frame);
278 :
279 44168 : continue_state get_continue() const {
280 44168 : return continue_;
281 : }
282 :
283 471 : void set_continue(continue_state state) {
284 471 : continue_ = state;
285 471 : }
286 :
287 55183 : bool input_pending() const {
288 55183 : return input_pending_;
289 : }
290 :
291 : /**
292 : * Whether input of the request may still be unread on the connection. Once it is not
293 : * (read to its end, or none past the head), `follows` tells whether bytes of what
294 : * follows the request were read past it, and the event wait_input_read() waits for is
295 : * set. Where the body is read: on the executor of the stream
296 : */
297 25960 : void set_input_pending(bool pending, bool follows = false) {
298 25960 : input_pending_ = pending;
299 25960 : input_follows_ = follows;
300 25960 : if (!pending && input_read_) input_read_->set();
301 25960 : }
302 :
303 1714 : bool input_follows() const {
304 1714 : return input_follows_;
305 : }
306 :
307 : /**
308 : * The request read its input to its end (see set_input_pending()), from any thread (a
309 : * handler may read its body from another executor): on the executor of the stream,
310 : * right away if called there
311 : */
312 : void end_input(bool follows);
313 :
314 : /**
315 : * Wait until input of the request is no longer pending: false if the wait is cancelled
316 : */
317 1328 : awaitable<bool> wait_input_read() {
318 1328 : if (!input_read_) {
319 1328 : input_read_.emplace(executor_);
320 1328 : if (!input_pending_) input_read_->set();
321 : }
322 1328 : return input_read_->wait();
323 : }
324 :
325 : size_t get_queued_frames() const;
326 :
327 23417 : bool responded() const {
328 23417 : return responded_;
329 : }
330 :
331 23111 : void set_response_ended() {
332 23111 : response_ended_ = true;
333 23111 : }
334 :
335 5789 : bool response_ended() const {
336 5789 : return response_ended_;
337 : }
338 :
339 : /**
340 : * Take over the connection once the response of this stream is written: the
341 : * disposition is take_over from now on
342 : */
343 4321 : void set_takeover(takeover_handler handler) {
344 4321 : takeover_ = std::move(handler);
345 4321 : takes_over_ = true;
346 4321 : responded_ = true;
347 4321 : }
348 :
349 50510 : bool takes_over() const {
350 50510 : return takes_over_;
351 : }
352 :
353 : /**
354 : * What the takeover waits for, in either order: the response written, and the read
355 : * loop stopped (with the bytes it read past the request)
356 : */
357 4312 : void takeover_response_written() {
358 4312 : takeover_response_written_ = true;
359 4312 : }
360 :
361 4315 : void takeover_reader_stopped(std::string buffered) {
362 4315 : takeover_buffer_ = std::move(buffered);
363 4315 : takeover_reader_stopped_ = true;
364 4315 : }
365 :
366 : /**
367 : * Handler of the takeover once both are done, and the bytes read past the request;
368 : * the handler is empty before, and after it was released once
369 : */
370 8627 : std::pair<takeover_handler, std::string> release_takeover() {
371 8627 : if (!takeover_response_written_ || !takeover_reader_stopped_) return {};
372 4312 : return {std::exchange(takeover_, nullptr), std::move(takeover_buffer_)};
373 : }
374 :
375 47 : void frame_posted() {
376 47 : ++posted_frames_;
377 47 : }
378 :
379 47 : void posted_frame_queued() {
380 47 : --posted_frames_;
381 47 : }
382 :
383 34776 : bool has_posted_frames() const {
384 34776 : return posted_frames_ > 0;
385 : }
386 :
387 : stream_id id() const;
388 :
389 : /**
390 : * What happens to the connection once the response is written: taken over if a
391 : * takeover was set, else closed if a reason was given, else kept alive. It never goes
392 : * back to keep_alive, and a takeover stays one
393 : */
394 37601 : detail::after_response disposition() const {
395 37601 : if (takes_over_) return detail::after_response::take_over;
396 37574 : return close_reason_ ? detail::after_response::close : detail::after_response::keep_alive;
397 : }
398 :
399 : /**
400 : * Close the connection after the response, for `reason`; the first reason given stays.
401 : * A takeover still takes over: the reason only makes its head say Connection: close
402 : */
403 8094 : void must_close(detail::close_reason reason) {
404 8094 : if (!close_reason_) close_reason_ = reason;
405 8094 : }
406 :
407 : /**
408 : * Why the connection closes after the response, if it does (also under a takeover)
409 : */
410 19844 : std::optional<detail::close_reason> reason() const {
411 19844 : return close_reason_;
412 : }
413 :
414 : /**
415 : * Observer of the response, set before the response is answered (none without a
416 : * response hook)
417 : */
418 411 : void set_observer(std::shared_ptr<response_observer> observer) {
419 411 : observer_ = std::move(observer);
420 411 : }
421 :
422 85757 : response_observer* observer() const {
423 85757 : return observer_.get();
424 : }
425 :
426 : // --- Cancellation scope (see route::cancel_on, request::cancelled()) ---
427 :
428 : /**
429 : * Cancel the scope for `reason`, from any thread, once: the first reason stays. On the
430 : * executor of the stream, it then sets the event client_gone() waits for and, while the
431 : * handler runs, emits terminal on the observer signal, and on the handler signal if the
432 : * route declared the reason or it is shutdown. Later calls do nothing, except a shutdown
433 : * while the handler runs not interrupted yet: it emits both signals, and neither
434 : * cancelled() nor wait_cancelled() see it. For a stream owned by a shared_ptr (it stays
435 : * alive until then)
436 : */
437 : void cancel(cancel_reason reason);
438 :
439 : /**
440 : * Why the scope was cancelled: none until then
441 : */
442 1095 : cancel_reason cancelled() const {
443 1095 : return cancelled_.load(std::memory_order_acquire);
444 : }
445 :
446 : /**
447 : * Wait, on the executor of the stream, until the scope is cancelled: its reason, or
448 : * none if the wait itself is cancelled
449 : */
450 : awaitable<cancel_reason> wait_cancelled();
451 :
452 : /**
453 : * The reasons interrupting the handler, as bits 1 << reason (see route::get_cancel_on);
454 : * set before the handler runs
455 : */
456 25959 : void set_cancel_on(uint8_t reasons) {
457 25959 : cancel_on_ = reasons;
458 25959 : }
459 :
460 : /**
461 : * Slot of the handler signal, bound once to the coroutine running the handler (see
462 : * http_application::serve), and that of the observer signal, for request::cancellation_slot()
463 : */
464 4287 : boost::asio::cancellation_slot handler_slot() {
465 4287 : return handler_signal_.slot();
466 : }
467 :
468 : /**
469 : * The coroutine about to run the handler is bound to the scope: through handler_slot(),
470 : * or to `signal`, which its owner emits only for this stream while the handler runs
471 : * (that of the read loop of the connection, which runs every handler of its requests).
472 : * The handler then runs on that coroutine; until a coroutine is bound,
473 : * http_application::serve runs it on one of its own, bound to handler_slot()
474 : */
475 27736 : void bind_handler(boost::asio::cancellation_signal* signal = nullptr) {
476 27736 : handler_bound_ = true;
477 27736 : bound_signal_ = signal;
478 27736 : }
479 :
480 31102 : bool handler_bound() const {
481 31102 : return handler_bound_;
482 : }
483 :
484 117 : boost::asio::cancellation_slot observer_slot() {
485 117 : return observer_signal_.slot();
486 : }
487 :
488 : /**
489 : * The handler starts and ends (on the executor of the stream): the signals are emitted
490 : * only in between, so a cancellation after its end is a no-op. Their slots are left as
491 : * they are: what they hold (the cancellation state of the coroutine that ran the
492 : * handler) goes with the stream
493 : */
494 31102 : void handler_started() {
495 31102 : handler_running_ = true;
496 31102 : }
497 :
498 26439 : void handler_ended() {
499 26439 : handler_running_ = false;
500 26439 : }
501 :
502 772 : bool handler_running() const {
503 772 : return handler_running_;
504 : }
505 :
506 : /**
507 : * Whether the handler signal was emitted: the handler was interrupted
508 : */
509 26448 : bool handler_interrupted() const {
510 26448 : return handler_interrupted_;
511 : }
512 :
513 : /**
514 : * The connection is done with the stream (its response ended, or the connection is
515 : * gone): what it kept for the response goes now, not with a request that may keep the
516 : * stream longer (the observer, or a takeover handler, may hold that request)
517 : */
518 23893 : void detach() {
519 23893 : observer_.reset();
520 23893 : takeover_ = nullptr;
521 23893 : queue_ = {};
522 23893 : }
523 : };
524 :
525 : }
526 :
527 : #endif
|
