Line data Source code
1 : #ifndef THINGER_HTTP_SERVER_CONNECTION_TRACKER_HPP
2 : #define THINGER_HTTP_SERVER_CONNECTION_TRACKER_HPP
3 :
4 : #include <array>
5 : #include <atomic>
6 : #include <cstdint>
7 : #include <functional>
8 : #include <memory>
9 : #include <mutex>
10 : #include <vector>
11 : #include <boost/asio/io_context.hpp>
12 : #include "../../asio/sockets/socket.hpp"
13 : #include "../../util/types.hpp"
14 : #include "upgrade.hpp"
15 :
16 : namespace thinger::http {
17 :
18 : // What a connection of a server serves
19 : enum class connection_type : uint8_t {
20 : http, // HTTP requests (keep-alive connections waiting for the next one included)
21 : websocket, // a WebSocket (upgrade_websocket(), upgrade_websocket_stream())
22 : sse, // a Server-Sent Events stream (start_sse())
23 : tunnel, // a reverse proxy upgrade (e.g. a WebSocket proxied to its target)
24 : taken_over // any other connection taken over (response::take_over())
25 : };
26 :
27 : // Connections of a server, as of the moment they are read
28 : struct server_stats {
29 : // Open connections, by what they serve
30 : uint64_t http_connections = 0;
31 : uint64_t websocket_connections = 0;
32 : uint64_t sse_connections = 0;
33 : uint64_t tunnel_connections = 0;
34 : uint64_t taken_over_connections = 0;
35 : // Connections served since the server was created (as HTTP connections, the way every
36 : // connection starts)
37 : uint64_t total_connections = 0;
38 :
39 808 : uint64_t active_connections() const {
40 808 : return http_connections + websocket_connections + sse_connections + tunnel_connections +
41 808 : taken_over_connections;
42 : }
43 : };
44 :
45 : class connection_tracker;
46 :
47 : // A connection of a server, known to its tracker while open (see track()). Nothing is done per
48 : // request: the tracker only takes a lock when the connection joins or leaves it, or changes
49 : // what it serves, and on the shard of its io_context (statistics and shutdown are rare, and
50 : // read the shards instead).
51 : class tracked_connection {
52 : public:
53 : tracked_connection(const tracked_connection&) = delete;
54 : tracked_connection& operator=(const tracked_connection&) = delete;
55 :
56 : protected:
57 28098 : tracked_connection() = default;
58 : // Leaves the tracker. Derived classes are gone by then, which is fine: the tracker reaches
59 : // them only through `self` (expired by then), and reads nothing else but this base
60 : ~tracked_connection();
61 :
62 : // Join the tracker of a server. `self` owns this object (the tracker locks it to act on
63 : // the connection), which runs on `io_context`. From any thread, once
64 : void track(std::shared_ptr<connection_tracker> tracker, std::weak_ptr<void> self,
65 : boost::asio::io_context& io_context, connection_type type);
66 : // Leave the tracker (the connection is closed, or belongs to someone else now). From the
67 : // thread of the connection, and again from the destructor
68 : void untrack();
69 : void set_type(connection_type type);
70 :
71 : // The tracker, while tracked (it is held weakly: it is not kept alive by its connections)
72 32410 : std::shared_ptr<connection_tracker> tracker() const { return tracker_.lock(); }
73 5307 : bool tracked() const { return !tracker_.expired(); }
74 :
75 : // On the io_context of the connection, while the server shuts down: finish what the
76 : // connection is doing and close (or tell its owner to), or close it right away
77 : virtual void shutdown_drain() = 0;
78 : virtual void shutdown_close() = 0;
79 : // On the io_context of the connection, at the deadline of the shutdown, before it is
80 : // closed: cancel the requests whose handlers still run (nothing by default)
81 109 : virtual void shutdown_cancel() {}
82 :
83 : private:
84 : friend class connection_tracker;
85 :
86 : std::weak_ptr<connection_tracker> tracker_;
87 : std::weak_ptr<void> self_;
88 : boost::asio::io_context* io_context_ = nullptr;
89 : // In the list of its shard (guarded by the mutex of the shard)
90 : tracked_connection* prev_ = nullptr;
91 : tracked_connection* next_ = nullptr;
92 : bool linked_ = false;
93 : uint8_t shard_ = 0;
94 : connection_type type_ = connection_type::http;
95 : };
96 :
97 : // The connections of a server: for its statistics, and to drain and close them on shutdown.
98 : // Thread-safe.
99 : class connection_tracker {
100 : public:
101 : enum class phase : uint8_t {
102 : serving,
103 : draining, // connections finish what they do, and close
104 : closing // connections are closed right away
105 : };
106 :
107 32306 : phase current_phase() const { return phase_.load(std::memory_order_acquire); }
108 52782 : void set_phase(phase value) { phase_.store(value, std::memory_order_release); }
109 :
110 : server_stats stats() const;
111 : // Connections open
112 : size_t active() const;
113 :
114 : // Call `empty` once no connection is left: right away, on this thread, if none is; else
115 : // on the thread of the last one to leave, as it leaves. Once, replacing the one set before
116 : // (none with null). For a shutdown waiting for its connections: a connection leaving does
117 : // nothing more while none is set
118 : void notify_when_empty(std::function<void()> empty);
119 :
120 : // Call shutdown_drain() on every connection, on its io_context (does not wait)
121 : void drain_all();
122 : // Call shutdown_cancel() and then shutdown_close() on every connection, in one call on its
123 : // io_context, so the handlers interrupted resume on a closed connection; then `done` (on
124 : // the thread of the last one, or right away if there is none), once what was posted on
125 : // each io_context before its close ran too (the first resumption of those handlers)
126 : void cancel_and_close_all(std::function<void()> done);
127 :
128 : private:
129 : friend class tracked_connection;
130 :
131 : static constexpr size_t shard_count = 16;
132 :
133 : struct alignas(64) shard {
134 : mutable std::mutex mutex;
135 : tracked_connection* head = nullptr;
136 : std::array<uint64_t, 5> active{};
137 : uint64_t total_connections = 0;
138 : };
139 :
140 : static size_t shard_of(const boost::asio::io_context& io_context);
141 :
142 : void add(tracked_connection& connection);
143 : void remove(tracked_connection& connection);
144 : void retype(tracked_connection& connection, connection_type type);
145 : // Call the callback of notify_when_empty() if no connection is left
146 : void check_empty();
147 :
148 : // The connections open (owned while they are acted on)
149 : std::vector<std::pair<std::shared_ptr<void>, tracked_connection*>> snapshot() const;
150 :
151 : std::array<shard, shard_count> shards_;
152 : std::atomic<phase> phase_{phase::serving};
153 : // What notify_when_empty() set, and whether it did (read by every connection leaving)
154 : std::mutex empty_mutex_;
155 : std::function<void()> on_empty_;
156 : std::atomic<bool> watching_{false};
157 : };
158 :
159 : namespace detail {
160 :
161 : // A connection taken over from the server (a WebSocket, an event stream, a tunnel...). It owns
162 : // the socket, and the socket handed out (see hand_out()) owns it: tracked while that socket
163 : // lives (the handles copied from it, and a WebSocket built on it, included), or until its
164 : // owner closes it (server_scope::closed()). On shutdown it calls the callback set with
165 : // server_scope::on_shutdown() (if any), and once the grace period ends its socket is closed.
166 : // On the thread of the socket.
167 : class taken_over_tracking final : public tracked_connection,
168 : public std::enable_shared_from_this<taken_over_tracking> {
169 : public:
170 : // Track `socket`, taken over from a connection of `tracker`
171 : static std::shared_ptr<taken_over_tracking> track(std::shared_ptr<connection_tracker> tracker,
172 : std::shared_ptr<asio::socket> socket);
173 :
174 : ~taken_over_tracking();
175 :
176 : // The connection to hand to the new owner: the socket (owning this tracking) and its scope
177 : upgraded_connection hand_out(std::string buffered);
178 :
179 : // Once the handler took the socket: drained or closed if the server shuts down meanwhile
180 : void handed_over();
181 :
182 : // What server_scope does
183 7191 : void set_type(connection_type type) { tracked_connection::set_type(type); }
184 : void on_shutdown(sync_callback<void()> callback);
185 : void closed();
186 :
187 : private:
188 4306 : explicit taken_over_tracking(std::shared_ptr<asio::socket> socket) : socket_(std::move(socket)) {}
189 :
190 : void shutdown_drain() override;
191 : void shutdown_close() override;
192 : // Call the callback of on_shutdown(), logging what it throws
193 : static void notify(const sync_callback<void()>& callback);
194 :
195 : std::shared_ptr<asio::socket> socket_;
196 : sync_callback<void()> on_shutdown_;
197 : // The drain started (shutdown_drain()), and the owner was told (or will be, posted)
198 : bool draining_ = false;
199 : bool notified_ = false;
200 : };
201 :
202 : }
203 :
204 : }
205 :
206 : #endif
|
