Line data Source code
1 : #ifndef THINGER_SERVER_HTTP_SERVER_CONNECTION_HPP
2 : #define THINGER_SERVER_HTTP_SERVER_CONNECTION_HPP
3 :
4 : #include <deque>
5 : #include <atomic>
6 : #include "request_factory.hpp"
7 : #include "../common/http_frame.hpp"
8 : #include "../common/http_request.hpp"
9 : #include "../common/http_response.hpp"
10 : #include "http_stream.hpp"
11 : #include "connection_tracker.hpp"
12 : #include "../../asio/async_event.hpp"
13 : #include "../../asio/watchdog.hpp"
14 : #include "../../util/types.hpp"
15 :
16 : namespace thinger::http {
17 :
18 : class request;
19 : struct connection_options;
20 :
21 : namespace detail {
22 : class connection_source;
23 : }
24 :
25 : class server_connection : public std::enable_shared_from_this<server_connection>, public tracked_connection {
26 :
27 : static constexpr size_t MAX_BUFFER_SIZE = 4096;
28 :
29 : public:
30 : // A connection served with the settings of `options` (its server's, see
31 : // server_base::set_connection_options()): the defaults if null
32 : explicit server_connection(std::shared_ptr<asio::socket> socket,
33 : std::shared_ptr<const connection_options> options = nullptr);
34 : virtual ~server_connection();
35 :
36 : // Count the connection among those of a server (its statistics, and its shutdown), before
37 : // start(). A connection taken over then counts as such, while its socket lives
38 : void track(const std::shared_ptr<connection_tracker>& tracker);
39 :
40 : // Start processing requests (spawns the read loop coroutine), closed once idle for the
41 : // idle_timeout of its options, or at the deadlines of a request (see connection_options).
42 : // Closed right away if the server it is tracked by shuts down
43 : void start();
44 :
45 : // Release the socket for upgrades (WebSocket, etc.)
46 : std::shared_ptr<asio::socket> release_socket();
47 :
48 : // Take over the connection from a stream: send `response`, then, once its request
49 : // has been handled, stop serving HTTP and hand the socket to `handler` (can be called
50 : // from any thread). If the connection was closed meanwhile (e.g. the client left while
51 : // the response was pending), nothing is sent and the handler is released without
52 : // being called.
53 : void take_over(std::shared_ptr<http_stream> stream, std::shared_ptr<http_frame> response,
54 : takeover_handler handler);
55 :
56 : // Release this instance without touching the socket
57 : void release();
58 :
59 : // Get the raw socket
60 : std::shared_ptr<asio::socket> get_socket();
61 :
62 : // Queue a response frame, in call order (can be called from any thread)
63 : void handle_stream(std::shared_ptr<http_stream> stream, std::shared_ptr<http_frame> frame);
64 :
65 : // Send the 100 Continue the client of `stream` waits for before sending the body (Expect:
66 : // 100-continue), as the body is about to be read: once, in order with the responses, and
67 : // only while its response has not started (on the connection thread)
68 : void send_continue(http_stream& stream);
69 :
70 : // Update connection timeout
71 : void update_connection_timeout(std::chrono::seconds timeout);
72 :
73 : // Extend the connection deadline after some activity outside the read loop, such as
74 : // reading the request body (on the connection thread)
75 : void extend_timeout();
76 :
77 : // Do not close the connection as idle before `time`: a request with a timeout of its own
78 : // (see route::timeout) may wait for its response longer than the idle timeout. The
79 : // deadlines of a request still apply (on the connection thread)
80 : void hold_until(std::chrono::steady_clock::time_point time);
81 :
82 : // Whether the connection closes gracefully (see linger()): nothing more is read from it but
83 : // by itself (on the connection thread)
84 43848 : bool lingering() const { return lingering_; }
85 :
86 23792 : void set_handler(std::function<awaitable<void>(std::shared_ptr<request>)> handler) {
87 23792 : handler_ = std::move(handler);
88 23792 : }
89 :
90 : private:
91 : // The source of a request read here, for its stream (see detail::request_source)
92 : friend class detail::connection_source;
93 : std::unique_ptr<detail::request_source> make_source(http_stream& stream);
94 :
95 : // Read requests, one after another, and hand each to the handler
96 : awaitable<void> read_loop();
97 :
98 : // The parsed request, queued for its response, with the bytes read past its headers
99 : std::pair<std::shared_ptr<http_stream>, std::shared_ptr<request>> start_request(
100 : const std::shared_ptr<server_connection>& self, const uint8_t* read_ahead, size_t read_ahead_size);
101 :
102 : // Answer a request that cannot be parsed, or whose header section is too large
103 : void reject_request();
104 :
105 : // The request took the connection over: stop reading, and hand it what was read past it
106 : void hand_over_reader(http_stream& stream, request& req);
107 :
108 : // Move what was read past the request to the buffer, for the next request; false if it
109 : // does not fit
110 : bool keep_pipelined(request& req, size_t& buffered);
111 :
112 : // The connection ended without being taken over
113 : void stop_reading();
114 :
115 : // Once reading stops after a request whose response ended: close gracefully if its body
116 : // still comes
117 : void linger_unread(const http_stream& stream);
118 :
119 : // The server shuts down: close once idle, or after the response in progress (which tells
120 : // the client); or close right away. At the deadline, before the close, the handlers still
121 : // running are cancelled (cancel_reason::shutdown)
122 : void shutdown_drain() override;
123 : void shutdown_cancel() override;
124 : void shutdown_close() override;
125 :
126 : // A frame of `stream` is queued while frame_watch_ is set. Kept out of queue_frame(), which
127 : // only tests the flags on the way of every frame
128 : [[gnu::noinline, gnu::cold]] void watch_frame(http_stream& stream);
129 :
130 : // Wait until the response of a stream has started, or the connection is closed
131 : awaitable<void> wait_response(const http_stream& stream);
132 :
133 : // Requests read are bounded by the responses still to write out: no next request is read
134 : // while max_queued_responses wait in the queue, or the front one waits for the socket (a
135 : // client that does not read). Node pauses its parser so, Go reads one request at a time
136 : static constexpr size_t max_queued_responses = 16;
137 47008 : bool output_backed_up() const { return writing_ || request_queue_.size() >= max_queued_responses; }
138 : // The read loop waits for the output before the next request. A head begun is read to its
139 : // end first, within its deadline, which a pause would eat
140 42573 : bool reading_paused() const { return output_backed_up() && !reading_head_; }
141 : // Wait until the output is no longer backed up, or the connection stops; woken as a response
142 : // leaves the queue, or a write that waited completes
143 : awaitable<void> wait_output();
144 20460 : void output_progressed() { if (output_waited_) [[unlikely]] reader_wake_->set(); }
145 :
146 : // The peer watcher: the only reader of the socket while a request is handled, and until
147 : // its response starts (the read loop reads it otherwise). Once the body of the request
148 : // was read, it waits for the socket to be readable: data (the next request) stays there,
149 : // a half-close makes the connection close after the response, and anything else (a reset)
150 : // cancels the scope of the request (client_gone) and closes the connection. A socket that
151 : // cannot wait (e.g. one defined by an application) is not watched
152 : awaitable<void> watch_loop();
153 : awaitable<void> watch_peer(http_stream& stream);
154 : // Watch the client for `stream`; stop it (on its own, but for a read in progress over
155 : // TLS, see watch_reading_); stop it for good, as the read loop stops
156 : void watch(const std::shared_ptr<http_stream>& stream);
157 : void unwatch();
158 : void stop_watcher();
159 : // Wake the watcher for the stream it watches: started then, the first time
160 : void wake_watcher();
161 : // Whether the watcher still watches for `stream`
162 : bool watching(const http_stream& stream) const;
163 :
164 : // Queue a response frame (on the connection executor)
165 : void queue_frame(const std::shared_ptr<http_stream>& stream, std::shared_ptr<http_frame> frame);
166 :
167 : // Once a frame of `stream` is written (or its write failed, with `ec`): close the
168 : // connection or hand it over when its response ends, as required
169 : void frame_written(http_stream& stream, const std::shared_ptr<http_frame>& frame,
170 : const boost::system::error_code& ec);
171 :
172 : // A frame of a stream whose response is observed (see http_application::on_response) is
173 : // about to be written
174 : [[gnu::noinline, gnu::cold]] static void observe_frame(http_stream& stream, http_frame& frame);
175 :
176 : // Write the queued frames, in order
177 : void process_output_queue();
178 :
179 : // Handle stock error responses
180 : void handle_stock_error(std::shared_ptr<http_stream> stream, http_response::status status);
181 :
182 : // Close connection
183 : void close();
184 :
185 : // Close the connection once the response of `stream` ended: gracefully if input of its
186 : // request may still be arriving (see linger()), else at once
187 : void close_after(const http_stream& stream);
188 :
189 : // Close gracefully (RFC 9112 section 9.6): stop sending, drop what keeps arriving (up to
190 : // the linger limits of its options, see connection_options), then close. At once if the
191 : // socket cannot half-close, or either limit is zero; briefly if the server shuts down
192 : void linger();
193 : awaitable<void> drop_input(size_t max_bytes, std::chrono::milliseconds timeout);
194 :
195 : // Hand the socket over once the response is written and the read loop has stopped
196 : void complete_takeover(http_stream& stream);
197 :
198 : // The deadlines of the connection (server_connection_deadlines.cpp), all on timeout_timer_,
199 : // armed at the earliest of those that apply: idle (from the last activity, not before the
200 : // hold); while the head of a request is read, the header deadline, and while its body is,
201 : // the request deadline (both from the start of the request, see connection_options); while
202 : // a response waits for the socket, the write deadline instead of idle (from its last
203 : // progress). On expiry, the one that passed acts (see deadline_expired()); none passed (the
204 : // head completed, the body was read, or the write went on, meanwhile), it is armed again
205 : using clock = std::chrono::steady_clock;
206 : void arm_deadline();
207 : clock::time_point next_deadline() const;
208 : void deadline_expired();
209 : // Idle, or the write deadline while a response waits for the socket (see write_timeout)
210 : clock::time_point idle_deadline() const;
211 : // A frame waits for the socket (see process_output_queue()): its write deadline starts
212 : void write_started();
213 : // Whether `buffers` are written in slices, each one moving the write deadline (see
214 : // asio::socket::write_slices()): a write deadline applies and they exceed one
215 : bool write_in_slices(const std::vector<boost::asio::const_buffer>& buffers) const;
216 : // The deadline of the head being read, if any applies (false otherwise)
217 : bool head_deadline(clock::time_point& deadline) const;
218 : // Bytes of a request head are about to be parsed (`pipelined`: read with the request before
219 : // it): the first ones start the request, at the read that got them, or now for a pipelined
220 : // one. A read leaving its head incomplete arms its header deadline (see head_incomplete())
221 : void head_started(bool pipelined);
222 : void head_incomplete();
223 : // The head of a request was not complete at its deadline: 408, or a silent close if no
224 : // byte of it arrived
225 : void head_timed_out();
226 : // The deadline of the body of the request in progress, while it is not read to its end, if
227 : // any applies (false otherwise)
228 : bool body_deadline(clock::time_point& deadline) const;
229 : // The request deadline of the request of `stream`, set by its route (see
230 : // route::request_timeout), from the start of the request; zero for none
231 : void set_request_deadline(const http_stream& stream, std::chrono::milliseconds deadline);
232 : // The deadline the handler set for reading the body of the request of `stream` (only compared:
233 : // it may be gone), replacing its request deadline (see request::set_read_deadline()); the end
234 : // of time removes it
235 : void set_read_deadline(const http_stream* stream, clock::time_point deadline);
236 : // Whether a read deadline is set and has passed: a read then fails at once
237 : // (read_deadline_expired)
238 : bool read_deadline_passed() const;
239 : // A read of the body starts and ends, bound to read_cancel_, which the watchdog emits as its
240 : // deadline passes, whenever it was set: nothing else on the socket is cancelled, and the
241 : // same read goes on when the deadline only moves. `caller` is the cancellation of the
242 : // coroutine reading (its `||`, the scope of the request), forwarded to the read while it waits
243 : void body_read_started(boost::asio::cancellation_slot caller);
244 : void body_read_ended(boost::asio::cancellation_slot caller, io_result& result);
245 : // The body of the request of `stream` was not read to its end at its deadline: its scope is
246 : // cancelled (cancel_reason::timeout), and it is answered 408 unless its response started,
247 : // or another one is being written; the connection is closed after it (or at once). What
248 : // the handler reads or writes afterwards fails
249 : void request_timed_out(http_stream& stream);
250 : // Answer `status` with a stock reply on a stream of its own, closing the connection after
251 : // it (gracefully if `input_pending`)
252 : void answer_and_close(http_response::status status, detail::close_reason reason, bool input_pending);
253 :
254 : private:
255 : std::shared_ptr<asio::socket> socket_;
256 :
257 : // The connection is closed when a deadline passes (see arm_deadline()): activity moves the
258 : // idle one later (see extend_timeout()), a request holds it (see hold_until())
259 : asio::watchdog timeout_timer_;
260 :
261 : // Wakes the read loop waiting for a response to start, or for the responses to write out
262 : // (see wait_output()); set as it stops too. Created by its first wait
263 : std::optional<asio::async_event> reader_wake_;
264 :
265 : // Bound to the read loop, which runs the handlers: the scope of the request being handled
266 : // emits it to interrupt its handler (see http_stream::bind_handler())
267 : boost::asio::cancellation_signal handler_signal_;
268 :
269 : // The peer watcher (see watch_peer()): the stream it watches for (none while idle), set
270 : // by the read loop, which wakes it (see wake_watcher()); the slot of its coroutine, emitted
271 : // to end its wait; set once a read of it ends. Its events are created as it starts
272 : std::shared_ptr<http_stream> watched_;
273 : std::optional<asio::async_event> watch_wake_;
274 : boost::asio::cancellation_signal watch_stop_;
275 : std::optional<asio::async_event> watch_idle_;
276 :
277 : // Rare cases where a queued frame matters, all tested at once on the way of every frame:
278 : // the read loop waits for a response to start, the connection drains, or the request in
279 : // progress waits for its 100 Continue
280 : enum : uint8_t { watch_response_start = 1, watch_draining = 2, watch_continue = 4 };
281 : uint8_t frame_watch_{0};
282 :
283 : // Stream of the request the read loop is handling, none while it waits for a request
284 : http_stream* in_progress_{nullptr};
285 : // The idle timeout (from the options, see update_connection_timeout())
286 : std::chrono::seconds timeout_;
287 : // The last activity (a read, a frame written), the idle deadline counting from it; the
288 : // connection is never closed as idle before the hold (see hold_until())
289 : clock::time_point last_activity_{};
290 : clock::time_point hold_{};
291 : // The start of the request being read: its first byte, the start of serving the connection
292 : // for the first one, or the start of parsing a pipelined one
293 : clock::time_point request_start_{};
294 : // The last progress of the frame waiting for the socket (see write_started())
295 : clock::time_point write_progress_{};
296 : // The request deadline of the request in progress while its body is read: the options',
297 : // or its route's (zero: none)
298 23792 : std::chrono::milliseconds request_deadline_{0};
299 : // The read deadline its handler set (see set_read_deadline()), and the signal ending the body
300 : // read that waits (see body_read_started())
301 : clock::time_point read_deadline_{};
302 : boost::asio::cancellation_signal read_cancel_;
303 :
304 : uint8_t buffer_[MAX_BUFFER_SIZE];
305 : request_factory request_parser_;
306 :
307 : // Queue for HTTP pipelining (only used from the connection thread, as everything that
308 : // reads or writes the connection)
309 : std::deque<std::shared_ptr<http_stream>> request_queue_;
310 :
311 : std::function<awaitable<void>(std::shared_ptr<request>)> handler_;
312 :
313 : // The settings of its server, shared by its connections
314 : std::shared_ptr<const connection_options> options_;
315 :
316 : // Buffers of the frame being written (kept, so they are not allocated for every frame)
317 : std::vector<boost::asio::const_buffer> write_buffers_;
318 :
319 : // State
320 : bool writing_{false};
321 : bool running_{false};
322 : // The read loop stopped without handing the connection over: it cannot be taken over
323 : bool reader_stopped_{false};
324 : // Closing gracefully (see linger()). Where it takes no room
325 : bool lingering_{false};
326 : // The watcher waits (for the body to be read, or the socket to be readable): unwatch()
327 : // ends the wait. Or it reads (peer_closed(), which may wait for a TLS record)
328 : bool watch_waiting_{false};
329 : bool watch_reading_{false};
330 : // The head of a request is being read, incomplete: its header deadline applies
331 : bool reading_head_{false};
332 : // A frame waits for the socket, within its write deadline (see write_started())
333 : bool write_waits_{false};
334 : // The read loop waits for the output (see wait_output())
335 : bool output_waited_{false};
336 : // Whether the request in progress has a read deadline, a read of its body waits for the
337 : // socket, and it was ended at that deadline
338 : bool read_bounded_{false};
339 : bool read_waits_{false};
340 : bool read_expired_{false};
341 : stream_id request_id_{0};
342 : };
343 :
344 : }
345 :
346 : #endif
|
