Line data Source code
1 : // Graceful shutdown of the servers (see server_base::shutdown()), apart from the code
2 : // serving requests
3 : #include "server_base.hpp"
4 : #include "../../asio/async_event.hpp"
5 : #include "../../asio/with_timeout.hpp"
6 : #include "../../util/logger.hpp"
7 : #include <boost/asio/post.hpp>
8 : #include <boost/asio/steady_timer.hpp>
9 : #include <algorithm>
10 : #include <condition_variable>
11 : #include <thread>
12 : #include <utility>
13 :
14 : namespace thinger::http {
15 :
16 : // A graceful shutdown, until its deadline. What runs it and waits for it owns it, and knows the
17 : // server only while it runs: stop() (from the destructors too) ends the shutdown first, waiting
18 : // for one stopping the server on another thread.
19 : //
20 : // It ends in two steps. Its result is decided first: every connection finished, or the
21 : // deadline came (the connections left are closed). Those waiting are told at once. Then the
22 : // server stops, once the connections are closed and the coroutines awaiting the result resumed:
23 : // a server stopping its own io_context would drop them otherwise.
24 : struct server_base::shutdown_state : std::enable_shared_from_this<shutdown_state> {
25 : // A coroutine awaiting the result
26 : struct waiter {
27 : boost::asio::any_io_executor executor;
28 : std::shared_ptr<asio::async_event> woken; // set on `executor`
29 : };
30 :
31 : std::mutex mutex;
32 : std::condition_variable ended; // decided or done
33 : server_base* server = nullptr; // none once stopped
34 : std::thread::id finishing; // the thread stopping the server at the end
35 : bool expired = false; // the deadline came with connections open
36 : bool timing = false; // the deadline is waited for on io_context
37 : bool decided = false;
38 : bool clean = false; // every connection finished in time
39 : bool done = false; // the server stopped
40 : // What the server stops after: the connections (finished, or closed at the deadline), and
41 : // every coroutine awaiting the result
42 : size_t holds = 1;
43 : std::vector<waiter> waiters;
44 :
45 : // Set once when it starts
46 : std::string name; // the server, for the logs
47 : std::shared_ptr<connection_tracker> connections;
48 : boost::asio::io_context* io_context = nullptr; // where the connections are checked
49 : std::chrono::steady_clock::time_point deadline;
50 : // While waiting: the timer of the deadline, owned by its wait (on io_context)
51 : std::weak_ptr<boost::asio::steady_timer> deadline_timer;
52 :
53 846 : bool result() {
54 846 : std::scoped_lock lock(mutex);
55 846 : return clean;
56 846 : }
57 :
58 : // On io_context: whether the connections are gone. Checked again once the last one leaves
59 : // (the tracker tells, see wait()), or at the deadline
60 3816 : void check() {
61 : {
62 3816 : std::scoped_lock lock(mutex);
63 3816 : if (decided || expired) return;
64 3816 : }
65 3788 : if (connections->active() == 0) {
66 : // After what the connections posted as they closed (their close callbacks, see
67 : // websocket_connection::on_close()), which a server stopping its io_context would drop
68 2813 : boost::asio::post(*io_context, [self = shared_from_this()] {
69 : // (its wait would keep an io_context run until out of work busy until then)
70 2809 : if (auto timer = self->deadline_timer.lock()) timer->cancel();
71 2809 : if (self->decide(true)) self->release();
72 2809 : });
73 975 : } else if (std::chrono::steady_clock::now() >= deadline) {
74 366 : expire();
75 : } else {
76 609 : wait();
77 : }
78 : }
79 :
80 : // On io_context: check() again when the last connection leaves, or at the deadline. Neither
81 : // keeps this alive (the server, or those awaiting the result, do)
82 609 : void wait() {
83 : bool timed;
84 : {
85 609 : std::scoped_lock lock(mutex);
86 609 : timed = std::exchange(timing, true);
87 609 : }
88 609 : if (!timed) {
89 609 : auto timer = std::make_shared<boost::asio::steady_timer>(*io_context, deadline);
90 609 : deadline_timer = timer;
91 609 : timer->async_wait([weak = weak_from_this(), timer](const boost::system::error_code& ec) {
92 546 : if (auto self = weak.lock(); self && !ec) self->check();
93 546 : });
94 609 : }
95 : // Again after it told (a connection accepted late may have joined, and left, since):
96 : // right away if the last one left since it was counted
97 609 : connections->notify_when_empty([weak = weak_from_this()] {
98 228 : if (auto self = weak.lock()) {
99 447 : boost::asio::post(*self->io_context, [self] { self->check(); });
100 228 : }
101 228 : });
102 609 : }
103 :
104 : // The deadline came, on any thread (also one waiting, as the threads of the server may be
105 : // stuck): the streams still in progress are cancelled, and the connections closed on their
106 : // threads, without waiting for them
107 770 : void expire() {
108 770 : auto open = connections->active();
109 770 : if (open == 0) {
110 119 : if (decide(true)) release();
111 404 : return;
112 : }
113 : {
114 651 : std::scoped_lock lock(mutex);
115 651 : if (decided || expired) return;
116 366 : expired = true;
117 651 : }
118 366 : LOG_WARNING("closing {} connections still open at the end of the grace period", open);
119 366 : connections->set_phase(connection_tracker::phase::closing);
120 366 : decide(false);
121 : // The requests still in progress are cancelled with the shutdown as the reason, which
122 : // interrupts every handler still running (see route::cancel_on), right before their
123 : // connections close, in the same call on the io_context of each: the handlers resume
124 : // on a closed connection, whose response is never written. They are not waited for
125 694 : connections->cancel_and_close_all([self = shared_from_this()] { self->release(); });
126 : }
127 :
128 : // The result, if not decided yet (a clean one only before the deadline came): the waiters
129 : // are told
130 4144 : bool decide(bool result) {
131 4144 : std::vector<waiter> woken;
132 : {
133 4144 : std::scoped_lock lock(mutex);
134 4144 : if (decided || (result && expired)) return false;
135 3855 : decided = true;
136 3855 : clean = result;
137 3855 : woken.swap(waiters);
138 4144 : }
139 : // Not told of the last connection anymore
140 3855 : connections->notify_when_empty(nullptr);
141 3855 : ended.notify_all();
142 4453 : for (auto& waiter : woken) {
143 1191 : boost::asio::post(waiter.executor, [woken = std::move(waiter.woken)] { woken->set(); });
144 : }
145 3855 : return true;
146 4144 : }
147 :
148 : // A coroutine on `executor` awaits the result: false if it is known already
149 906 : bool await(boost::asio::any_io_executor executor, std::shared_ptr<asio::async_event> woken) {
150 906 : std::scoped_lock lock(mutex);
151 906 : if (decided) return false;
152 1206 : waiters.push_back({std::move(executor), std::move(woken)});
153 603 : ++holds;
154 603 : return true;
155 1509 : }
156 :
157 : // The coroutine awaiting with `woken` resumed (or is gone)
158 603 : void leave(const std::shared_ptr<asio::async_event>& woken) {
159 : {
160 603 : std::scoped_lock lock(mutex);
161 608 : std::erase_if(waiters, [&](const waiter& waiter) { return waiter.woken == woken; });
162 603 : }
163 603 : release();
164 603 : }
165 :
166 : // What the server stops after ended: once nothing else holds it, it stops
167 4308 : void release() {
168 : server_base* stopping;
169 : {
170 4308 : std::scoped_lock lock(mutex);
171 4308 : if (--holds > 0 || done || finishing != std::thread::id()) return;
172 3616 : finishing = std::this_thread::get_id();
173 3616 : stopping = server;
174 4308 : }
175 : // Anything arriving late is closed
176 3616 : connections->set_phase(connection_tracker::phase::closing);
177 : // Not stop(), which a subclass being destroyed meanwhile may override (see
178 : // set_shutdown_stop())
179 3616 : if (stopping) {
180 3608 : stopping->stop_listening();
181 3608 : if (stopping->shutdown_stop_) stopping->shutdown_stop_();
182 : }
183 : bool result;
184 : {
185 3616 : std::scoped_lock lock(mutex);
186 3616 : done = true;
187 3616 : finishing = std::thread::id();
188 3616 : result = clean;
189 3616 : }
190 3616 : ended.notify_all();
191 3616 : LOG_INFO("{} stopped{}", name, result ? "" : " (connections closed before they finished)");
192 : }
193 : };
194 :
195 10605 : server_base::~server_base() {
196 : // Not virtual: a subclass's override already ran, if its destructor called it
197 10605 : server_base::stop();
198 10605 : }
199 :
200 23802 : bool server_base::reset_shutdown() {
201 23802 : if (shutdown_) {
202 2710 : std::scoped_lock state_lock(shutdown_->mutex);
203 2710 : if (!shutdown_->done) return false;
204 2710 : }
205 : // Serving again after a shutdown that ended
206 22898 : shutdown_.reset();
207 22898 : connections_->set_phase(connection_tracker::phase::serving);
208 22898 : return true;
209 : }
210 :
211 24 : void server_base::on_shutdown(sync_callback<void()> callback) {
212 24 : on_shutdown_ = std::move(callback);
213 24 : }
214 :
215 10605 : void server_base::set_shutdown_stop(sync_callback<void()> stop) {
216 10605 : shutdown_stop_ = std::move(stop);
217 10605 : }
218 :
219 7906 : server_stats server_base::stats() const {
220 7906 : return connections_->stats();
221 : }
222 :
223 906 : awaitable<bool> server_base::shutdown(std::chrono::steady_clock::time_point by) {
224 : auto state = start_shutdown(by);
225 : auto executor = co_await boost::asio::this_coro::executor;
226 : auto woken = std::make_shared<asio::async_event>(executor);
227 : if (state->await(executor, woken)) {
228 : // Left however the wait ends, also with this coroutine destroyed before it resumes
229 : struct waiting {
230 : std::shared_ptr<shutdown_state> state;
231 : std::shared_ptr<asio::async_event> woken;
232 603 : ~waiting() { state->leave(woken); }
233 : } waiting{state, woken};
234 : if (!co_await asio::with_timeout(woken->wait(), state->deadline - std::chrono::steady_clock::now())) {
235 : // Not decided at the deadline, on this executor: those of the server may be stuck
236 : state->expire();
237 : }
238 : }
239 : co_return state->result();
240 1812 : }
241 :
242 3021 : bool server_base::shutdown(std::chrono::milliseconds grace) {
243 3021 : auto by = std::chrono::steady_clock::now() + grace;
244 3021 : if (in_server_thread()) {
245 609 : LOG_ERROR("shutdown({} ms) cannot block a thread of the server, which the shutdown needs: "
246 : "shutting down without waiting (co_await shutdown(by) there instead)", grace.count());
247 609 : start_shutdown(by);
248 609 : return false;
249 : }
250 2412 : auto state = start_shutdown(by);
251 2412 : std::unique_lock lock(state->mutex);
252 7971 : if (!state->ended.wait_until(lock, state->deadline, [&] { return state->done; })) {
253 : // Not decided at the deadline: the threads of the server may be stuck
254 132 : lock.unlock();
255 132 : state->expire();
256 132 : lock.lock();
257 346 : state->ended.wait(lock, [&] { return state->decided; });
258 : }
259 2412 : return state->clean;
260 2412 : }
261 :
262 3927 : std::shared_ptr<server_base::shutdown_state> server_base::start_shutdown(std::chrono::steady_clock::time_point by) {
263 3927 : std::shared_ptr<shutdown_state> state;
264 : {
265 3927 : std::scoped_lock lock(shutdown_mutex_);
266 : // In progress, or ended: the same end
267 3927 : if (shutdown_) return shutdown_;
268 3855 : state = std::make_shared<shutdown_state>();
269 3855 : state->server = this;
270 3855 : state->name = log_name();
271 3855 : state->connections = connections_;
272 3855 : state->deadline = by;
273 3855 : shutdown_ = state;
274 3927 : }
275 :
276 : // What is left of it, rounded up: the grace given to shutdown(grace), once it reaches here
277 3855 : auto budget = std::chrono::ceil<std::chrono::milliseconds>(by - std::chrono::steady_clock::now());
278 3855 : LOG_INFO("shutting down the {}: {} connections open, grace period of {} ms",
279 : state->name, connections_->active(), std::max<int64_t>(budget.count(), 0));
280 :
281 : // New connections are refused, and those accepted meanwhile closed
282 3855 : connections_->set_phase(connection_tracker::phase::draining);
283 3855 : stop_listening();
284 :
285 3855 : if (on_shutdown_) {
286 : try {
287 1818 : on_shutdown_();
288 3 : } catch (const std::exception& e) {
289 3 : LOG_ERROR("error in the shutdown callback: {}", e.what());
290 3 : }
291 : }
292 :
293 3855 : state->io_context = shutdown_context();
294 3855 : if (!state->io_context) {
295 603 : state->decide(connections_->active() == 0);
296 603 : state->release();
297 603 : return state;
298 : }
299 3252 : connections_->drain_all();
300 6492 : boost::asio::post(*state->io_context, [state] { state->check(); });
301 3252 : return state;
302 3927 : }
303 :
304 39772 : void server_base::end_shutdown() {
305 39772 : std::shared_ptr<shutdown_state> state;
306 : {
307 39772 : std::scoped_lock lock(shutdown_mutex_);
308 39772 : state = shutdown_;
309 39772 : }
310 39772 : if (!state) return;
311 : {
312 5556 : std::unique_lock lock(state->mutex);
313 5556 : state->server = nullptr;
314 5556 : if (state->done || state->finishing == std::this_thread::get_id()) return;
315 : // Ending on another thread, which is stopping this server: wait for it
316 264 : if (state->finishing != std::thread::id()) {
317 51 : state->ended.wait(lock, [&] { return state->done; });
318 17 : return;
319 : }
320 5556 : }
321 247 : state->decide(false);
322 : {
323 247 : std::scoped_lock lock(state->mutex);
324 247 : state->done = true;
325 247 : }
326 247 : state->ended.notify_all();
327 39772 : }
328 :
329 : } // namespace thinger::http
|
