Line data Source code
1 : #include "../../util/logger.hpp"
2 : #include "../../util/types.hpp"
3 : #include "websocket_connection.hpp"
4 : #include "connection_tracker.hpp"
5 : #include <algorithm>
6 : #include <memory>
7 : #include <stdexcept>
8 : #include <string>
9 : #include <utility>
10 : #include "../util/utf8.hpp"
11 : #include "detail/push_queue.hpp"
12 :
13 : namespace thinger::http{
14 :
15 : namespace {
16 :
17 : // The read buffer a connection keeps between messages: a larger message gives it up
18 : constexpr size_t RETAINED_BUFFER_SIZE = 64 * 1024;
19 :
20 : // A streambuf has no shrink (consume() keeps its capacity) and cannot be moved, so a new
21 : // one takes its place, with the same largest message (noexcept: it is never left destroyed)
22 660 : void release_capacity(boost::asio::streambuf& buffer) noexcept {
23 660 : auto max_size = buffer.max_size();
24 660 : std::destroy_at(&buffer);
25 660 : std::construct_at(&buffer, max_size);
26 660 : }
27 :
28 : }
29 :
30 3102 : websocket_connection::websocket_connection(std::shared_ptr<asio::websocket> socket,
31 3102 : const websocket_accept_options& options, server_scope scope) :
32 3102 : ws_(std::move(socket)),
33 3102 : queue_(std::make_unique<detail::push_queue<out_message>>(ws_->get_executor(),
34 0 : write_limits{.max_messages = MAX_OUTPUT_MESSAGES},
35 6204 : detail::push_queue<out_message>::hooks{
36 30275 : .write_item = [this](out_message& message){ return write_message(message); },
37 4901319 : .item_size = [](const out_message& message){ return message.data.size(); },
38 : // every message is a unit: drop_oldest drops messages
39 2423267 : .ends_unit = [](const out_message&){ return true; },
40 3167 : .on_failure = [this](detail::push_failure failure, boost::system::error_code ec){ queue_failed(failure, ec); },
41 11078 : .owner = [this]{ return std::shared_ptr<void>(shared_from_this()); },
42 3200 : .send_backlog = [this]{ return ws_->send_backlog(); }})),
43 6204 : buffer_(std::max<size_t>(options.max_message_size, 1)),
44 3102 : on_close_(ws_->get_executor()),
45 3102 : scope_(std::move(scope)),
46 3102 : resumed_(ws_->get_executor()),
47 3102 : ping_interval_(options.ping_interval.value_or(asio::websocket::CONNECTION_TIMEOUT_SECONDS)),
48 12408 : idle_timeout_(options.idle_timeout.value_or(2 * asio::websocket::CONNECTION_TIMEOUT_SECONDS))
49 : {
50 3102 : LOG_LEVEL(2, "websocket connection created");
51 3102 : }
52 :
53 3102 : websocket_connection::~websocket_connection()
54 : {
55 3102 : LOG_LEVEL(1, "releasing websocket connection");
56 3102 : }
57 :
58 45 : void websocket_connection::on_shutdown(sync_callback<void()> callback){
59 45 : on_shutdown_callback_ = std::move(callback);
60 45 : }
61 :
62 57 : void websocket_connection::server_shutdown(){
63 57 : if(on_shutdown_callback_){
64 36 : on_shutdown_callback_();
65 36 : return;
66 : }
67 21 : LOG_LEVEL(1, "closing websocket: the server shuts down");
68 21 : start_close(asio::websocket::CLOSE_GOING_AWAY);
69 : }
70 :
71 366 : void websocket_connection::start_close(uint16_t code, std::string reason){
72 : // One close frame: the first close decides
73 366 : if(closing_) return;
74 330 : closing_ = true;
75 : // The connection is told when the handshake ends (closed, or failed), as it may be paused
76 330 : co_spawn(ws_->get_executor(), ws_->close_graceful(code, std::move(reason)),
77 660 : [self = shared_from_this()](const std::exception_ptr&){
78 330 : LOG_LEVEL(1, "websocket closed");
79 330 : self->wake_reader();
80 330 : });
81 : }
82 :
83 2434 : void websocket_connection::on_message(sync_callback<void(std::string, bool binary)> callback){
84 2434 : on_frame_callback_ = std::move(callback);
85 2434 : }
86 :
87 444 : void websocket_connection::on_close(sync_callback<void(const websocket_close&)> callback){
88 444 : on_close_.set(std::move(callback));
89 444 : }
90 :
91 3099 : void websocket_connection::start_read_loop()
92 : {
93 3099 : co_spawn(ws_->get_executor(),
94 9297 : [this, self = shared_from_this()]() -> awaitable<void> {
95 : // However the loop ends (also with the io_context destroyed), the queue closes
96 : // (its watermark callbacks released, producers waiting told) and the callbacks
97 : // are released: they may hold this connection
98 3099 : auto guard = queue_->guard_cycles([this]{
99 3099 : on_frame_callback_ = nullptr;
100 3099 : on_shutdown_callback_ = nullptr;
101 : // never told if destroyed open (e.g. with the io_context)
102 3099 : on_close_.release();
103 : // closed: no longer counts for the server
104 3099 : scope_.closed();
105 3099 : });
106 : co_await read_messages();
107 : // Ended here, on the executor (not by the teardown of the io_context): closed,
108 : // unless released. Told after the cleanup of the guard, as it is posted
109 : if (!released_) on_close_.notify(close_status());
110 6198 : },
111 : detached);
112 3099 : }
113 :
114 3099 : awaitable<void> websocket_connection::read_messages()
115 : {
116 : try {
117 : co_await read_loop();
118 : } catch (const std::exception& e) {
119 : LOG_ERROR("error handling a websocket message: {}", e.what());
120 : // (not the socket released by on_message(): it belongs to someone else)
121 : if (!released_) ws_->close(asio::websocket::CLOSE_INTERNAL_ERROR);
122 : }
123 6198 : }
124 :
125 1888 : websocket_close websocket_connection::close_status() const
126 : {
127 1888 : if (lost_) return websocket_close{};
128 693 : return websocket_close{ws_->get_close_code(), ws_->get_close_reason(), ws_->closed_by_peer()};
129 : }
130 :
131 3099 : awaitable<void> websocket_connection::read_loop()
132 : {
133 : // (never past the largest message: the buffer refuses it)
134 : size_t next_read_size = std::min(DEFAULT_BUFFER_SIZE, buffer_.max_size());
135 :
136 : // (released from a callback, e.g. on_message(): the socket belongs to someone else)
137 : while (ws_->is_open() && !released_) {
138 : // nothing is read while paused
139 : if (paused_ && !co_await wait_resumed()) break;
140 :
141 : LOG_LEVEL(2, "waiting websocket data");
142 :
143 : auto buf = buffer_.prepare(next_read_size);
144 : auto [ec, bytes_transferred] = co_await ws_->read_message_some(
145 : static_cast<uint8_t*>(buf.data()), buf.size());
146 :
147 : if (ec) {
148 : // The connection is lost (reset, or ended without a close frame): closed. Not lost
149 : // for how it closed if the close frame was out: the peer just did not answer it
150 : if (!released_ && ws_->is_open()) {
151 : LOG_LEVEL(1, "websocket connection lost: {}", ec.message());
152 : lost_ = ws_->get_close_code() == asio::websocket::CLOSE_ABNORMAL;
153 : ws_->close();
154 : }
155 : break;
156 : }
157 :
158 : LOG_LEVEL(2, "socket read: {} bytes", bytes_transferred);
159 :
160 : buffer_.commit(bytes_transferred);
161 :
162 : // get remaining data in the frame
163 : auto remaining = ws_->remaining_in_frame();
164 :
165 : // no pending data to read from the frame
166 : if(remaining == 0){
167 :
168 : // is the message complete ? FIN flag is set
169 : if(ws_->is_message_complete() && !deliver_message()){
170 : ws_->close(asio::websocket::CLOSE_INVALID_PAYLOAD);
171 : co_return;
172 : }
173 :
174 : // the message goes on in another frame, if not complete: it must fit
175 : next_read_size = std::min(DEFAULT_BUFFER_SIZE, buffer_.max_size() - buffer_.size());
176 : if(next_read_size == 0){
177 : LOG_DEBUG("websocket buffer overflow. closing connection");
178 : ws_->close(asio::websocket::CLOSE_MESSAGE_TOO_BIG);
179 : co_return;
180 : }
181 :
182 : }else{
183 : // check if the buffer is not going to overflow
184 : if(buffer_.size() + remaining > buffer_.max_size()){
185 : LOG_DEBUG("websocket buffer overflow. closing connection");
186 : ws_->close(asio::websocket::CLOSE_MESSAGE_TOO_BIG);
187 : co_return;
188 : }
189 :
190 : // The rest of the frame, but the buffer grows only with what arrives (at most
191 : // doubling), never to the size a header announces: what the buffer already has
192 : // room for is used as is
193 : auto received = buffer_.size();
194 : auto room = buffer_.capacity() > received ? buffer_.capacity() - received : 0;
195 : next_read_size = static_cast<size_t>(std::min<uint64_t>(remaining,
196 : std::max({DEFAULT_BUFFER_SIZE, received, room})));
197 : }
198 : }
199 6198 : }
200 :
201 3278 : bool websocket_connection::deliver_message()
202 : {
203 3278 : auto readable = buffer_.data();
204 3278 : auto* data_ptr = static_cast<const uint8_t*>(readable.data());
205 3278 : auto data_size = readable.size();
206 :
207 : // check if the message is a valid UTF8 message
208 3278 : if(!ws_->is_binary() && !utf8::is_valid(data_ptr, data_size)){
209 24 : LOG_DEBUG("invalid UTF8 message received!");
210 24 : return false;
211 : }
212 :
213 3254 : if (on_frame_callback_) {
214 3254 : std::string data(reinterpret_cast<const char*>(data_ptr), data_size);
215 : // Only the size: messages may hold credentials (see body_logging)
216 3254 : LOG_TRACE("websocket {} message received: {} bytes", ws_->is_binary() ? "binary" : "text", data_size);
217 3272 : on_frame_callback_(std::move(data), ws_->is_binary());
218 3254 : }
219 :
220 : // clear processed buffer, giving up what only a large message needed
221 3236 : if (data_size > RETAINED_BUFFER_SIZE) release_capacity(buffer_);
222 2576 : else buffer_.consume(buffer_.size());
223 3236 : return true;
224 : }
225 :
226 24 : awaitable<bool> websocket_connection::wait_resumed()
227 : {
228 : LOG_LEVEL(2, "websocket reading paused");
229 : while (paused_ && !released_ && ws_->is_open()) {
230 : resumed_.reset(); // woken by the next wake_reader(), not by one before
231 : co_await resumed_.wait();
232 : }
233 : co_return !released_ && ws_->is_open();
234 48 : }
235 :
236 395 : void websocket_connection::wake_reader(){
237 395 : resumed_.set();
238 395 : }
239 :
240 24 : void websocket_connection::suspend_keepalive(){
241 24 : keepalive_ping_ = ws_->get_ping_interval();
242 24 : keepalive_idle_ = ws_->get_idle_timeout();
243 24 : ws_->set_keepalive(std::chrono::milliseconds{0}, std::chrono::milliseconds{0});
244 24 : }
245 :
246 12 : void websocket_connection::restore_keepalive(){
247 12 : ws_->set_keepalive(keepalive_ping_, keepalive_idle_);
248 12 : }
249 :
250 24 : void websocket_connection::pause_reading(){
251 24 : execute([this]{
252 24 : if(paused_) return;
253 24 : paused_ = true;
254 24 : if(started_ && !released_) suspend_keepalive();
255 : });
256 24 : }
257 :
258 18 : void websocket_connection::resume_reading(){
259 18 : execute([this]{
260 : // (once closing, the handshake is what reads the socket, until it closes)
261 18 : if(!paused_ || closing_) return;
262 12 : paused_ = false;
263 12 : if(started_ && !released_) restore_keepalive();
264 12 : wake_reader();
265 : });
266 18 : }
267 :
268 3 : bool websocket_connection::is_reading_paused() const{
269 3 : return paused_;
270 : }
271 :
272 27173 : awaitable<boost::system::error_code> websocket_connection::write_message(out_message& message)
273 : {
274 : // (closed meanwhile, e.g. by the peer: what is queued is not written)
275 : if(!ws_->is_open()) co_return boost::asio::error::not_connected;
276 : LOG_LEVEL(2, "handling websocket write, remaining in queue: {}", queue_->queued_messages());
277 : ws_->set_binary(message.binary);
278 : // past a slice, with a write deadline, in slices: each one is progress, so a slow client
279 : // reading a large message is told from one that stopped reading
280 : auto [ec, bytes] = message.data.size() > asio::socket::WRITE_SLICE_SIZE && queue_->get_write_limits().write_timeout.count() > 0
281 11073 : ? co_await ws_->write_message(message.data, [this]{ queue_->write_progressed(); })
282 : : co_await ws_->write_message(message.data);
283 : co_return ec;
284 54346 : }
285 :
286 65 : void websocket_connection::queue_failed(detail::push_failure failure, boost::system::error_code ec)
287 : {
288 65 : switch(failure){
289 15 : case detail::push_failure::overflow:
290 15 : close_congested("it is congested");
291 15 : return;
292 24 : case detail::push_failure::write_timeout:
293 24 : close_congested("its write timed out");
294 24 : return;
295 26 : case detail::push_failure::write_error:
296 : // closing: nothing more goes after the close frame, and the handshake goes on
297 26 : if(closing_) return;
298 : // a websocket that cannot write is broken: closed, which ends reading too (not
299 : // the socket released: its write was cancelled)
300 8 : LOG_LEVEL(1, "websocket write failed: {}", ec.message());
301 8 : if(ec != boost::asio::error::operation_aborted) ws_->close();
302 8 : wake_reader();
303 8 : return;
304 : }
305 : }
306 :
307 39 : void websocket_connection::close_congested(const char* reason)
308 : {
309 : // (once closing, the closing handshake has its own deadline, and the first code decides)
310 39 : if(closing_ || !ws_->is_open()) return;
311 27 : LOG_WARNING("closing websocket: {}", reason);
312 27 : ws_->close(asio::websocket::CLOSE_TRY_AGAIN_LATER);
313 27 : wake_reader();
314 : }
315 :
316 18 : std::shared_ptr<asio::socket> websocket_connection::release_socket(){
317 : // the read loop ends, even if paused
318 18 : released_ = true;
319 18 : wake_reader();
320 :
321 : // the new owner decides about the keepalive started by start()
322 18 : ws_->set_keepalive(std::chrono::milliseconds{0}, std::chrono::milliseconds{0});
323 :
324 : // cancel pending async i/io requests on this socket (a frame being written is
325 : // completed by the next write)
326 18 : ws_->cancel();
327 :
328 : // return socket
329 18 : return ws_;
330 : }
331 :
332 3099 : void websocket_connection::start(){
333 3099 : started_ = true;
334 :
335 : // Taken over from a server: told when it shuts down
336 3099 : scope_.set_type(connection_type::websocket);
337 3099 : scope_.on_shutdown([weak = weak_from_this()]{
338 57 : if(auto self = weak.lock()) self->server_shutdown();
339 57 : });
340 :
341 : // handle timeout on websocket (suspended while paused)
342 3099 : ws_->set_keepalive(ping_interval_, idle_timeout_);
343 3099 : if(paused_) suspend_keepalive();
344 :
345 : // initiates async reading on socket
346 3099 : start_read_loop();
347 3099 : }
348 :
349 96 : void websocket_connection::stop(){
350 96 : stop(asio::websocket::CLOSE_NORMAL);
351 96 : }
352 :
353 420 : void websocket_connection::stop(uint16_t code, std::string reason){
354 420 : if(!asio::websocket::valid_close_code(code)){
355 57 : throw std::invalid_argument("websocket close code may not be sent: " + std::to_string(code));
356 : }
357 363 : if(reason.size() > asio::websocket::MAX_CLOSE_REASON_SIZE){
358 6 : throw std::invalid_argument("websocket close reason above 123 bytes: " + std::to_string(reason.size()));
359 : }
360 357 : if(!utf8::is_valid(reason)){
361 12 : throw std::invalid_argument("websocket close reason is not valid UTF-8");
362 : }
363 345 : execute([this, self = shared_from_this(), code, reason = std::move(reason)]() mutable {
364 345 : LOG_LEVEL(1, "closing websocket: {}", code);
365 345 : start_close(code, std::move(reason));
366 345 : });
367 345 : }
368 :
369 12 : bool websocket_connection::queue_full() const{
370 12 : return queue_->full();
371 : }
372 :
373 1235680 : void websocket_connection::send_binary(std::string data){
374 1235680 : execute([this, data = std::move(data)]() mutable {
375 1235680 : push_message(std::move(data), true);
376 1235680 : });
377 1235680 : }
378 :
379 1200766 : void websocket_connection::send_text(std::string text){
380 1200766 : execute([this, data = std::move(text)]() mutable {
381 1200766 : push_message(std::move(data), false);
382 1200766 : });
383 1200766 : }
384 :
385 2436446 : void websocket_connection::push_message(std::string data, bool binary){
386 : // (nothing goes after the close frame, nor once closed, even if the read loop did not end
387 : // yet and the queue is still open: not a congestion)
388 2436446 : if(closing_ || !ws_->is_open()) return;
389 : // (a queue closed drops it too: not a congestion either)
390 2436098 : if(queue_->push(out_message{std::move(data), binary}) == detail::push_queue<out_message>::admission::drop && !queue_->closed()){
391 1198004 : LOG_DEBUG("websocket is congested. discarding packets!");
392 : }
393 : }
394 :
395 3129 : awaitable<bool> websocket_connection::send_async(std::string data, bool binary){
396 6258 : return queue_->push_and_wait(out_message{std::move(data), binary});
397 : }
398 :
399 309 : awaitable<bool> websocket_connection::send_text_async(std::string text){
400 309 : return send_async(std::move(text), false);
401 : }
402 :
403 2820 : awaitable<bool> websocket_connection::send_binary_async(std::string data){
404 2820 : return send_async(std::move(data), true);
405 : }
406 :
407 1215 : awaitable<bool> websocket_connection::wait_writable(){
408 1215 : return queue_->wait_writable();
409 : }
410 :
411 513 : void websocket_connection::set_write_limits(const write_limits& limits){
412 513 : queue_->set_write_limits(limits);
413 513 : }
414 :
415 3 : write_limits websocket_connection::get_write_limits() const{
416 3 : return queue_->get_write_limits();
417 : }
418 :
419 3572 : size_t websocket_connection::queued_bytes() const{
420 3572 : return queue_->queued_bytes();
421 : }
422 :
423 123 : size_t websocket_connection::queued_messages() const{
424 123 : return queue_->queued_messages();
425 : }
426 :
427 8108100 : uint64_t websocket_connection::dropped_messages() const{
428 8108100 : return queue_->dropped_messages();
429 : }
430 :
431 15 : void websocket_connection::on_high_watermark(sync_callback<void()> callback){
432 15 : queue_->on_high_watermark(std::move(callback));
433 15 : }
434 :
435 21 : void websocket_connection::on_low_watermark(sync_callback<void()> callback){
436 21 : queue_->on_low_watermark(std::move(callback));
437 21 : }
438 :
439 : }
|
