Line data Source code
1 : #ifndef THINGER_HTTP_SERVER_FLOW_CONTROL_HPP
2 : #define THINGER_HTTP_SERVER_FLOW_CONTROL_HPP
3 :
4 : #include <atomic>
5 : #include <chrono>
6 : #include <cstddef>
7 : #include <memory>
8 : #include <boost/asio/any_io_executor.hpp>
9 : #include "../../asio/async_event.hpp"
10 : #include "../../asio/watchdog.hpp"
11 : #include "../../util/types.hpp"
12 :
13 : namespace thinger::http {
14 :
15 : // What a connection does with a message sent without waiting to a full queue
16 : enum class overflow_policy {
17 : drop_newest, // discard the message sent (the default)
18 : drop_oldest, // discard the oldest messages queued to make room for it
19 : close // close the connection (a WebSocket with 1013, Try Again Later)
20 : };
21 :
22 : /**
23 : * Limits of the queue of a server connection (WebSocket, SSE): what is not written yet, the
24 : * message being written included. 0 disables a limit.
25 : */
26 : struct write_limits {
27 : /// Full when another message would go over either (an empty queue takes any size)
28 : size_t max_messages = 100;
29 : size_t max_bytes = 0;
30 : /// Once the bytes queued reach the high watermark, the connection is not writable until
31 : /// they drain to the low one (as Netty's WriteBufferWaterMark): on_high_watermark() and
32 : /// on_low_watermark() are told once per cycle, posted to the connection's thread (never
33 : /// from within a send), and wait_writable() waits meanwhile
34 : size_t high_watermark = 0;
35 : size_t low_watermark = 0;
36 : /// For messages sent without waiting. drop_oldest never drops the message being
37 : /// written, one a sender waits for, nor the rest of an SSE event being written; when
38 : /// that is not enough, the new message is dropped, so the limits hold. What is dropped
39 : /// is counted by dropped_messages() of the connection
40 : overflow_policy policy = overflow_policy::drop_newest;
41 : /// A write that makes no progress for this long (no byte taken by the socket, nor by the
42 : /// peer from it, watched for a quarter of it more: see set_send_backlog()) closes
43 : /// the connection (a WebSocket with 1013), as a client that stopped reading; awaited sends and wait_writable() then return false. A stall, not the time to
44 : /// write a message: a slow client that keeps reading is never cut. 60 s, nginx's
45 : /// send_timeout; 0 disables it. A WebSocket closing with stop() does not wait for it: its
46 : /// closing handshake has its own deadline, asio::websocket::CLOSE_TIMEOUT
47 6092 : std::chrono::milliseconds write_timeout{60000};
48 : };
49 :
50 : /**
51 : * Accounting of the queue of a connection against its write_limits. On the executor of the
52 : * connection, except queued_bytes(), queued_messages() and the calls taking an `owner`.
53 : */
54 : class write_flow {
55 : public:
56 : explicit write_flow(boost::asio::any_io_executor executor, write_limits limits = {});
57 : ~write_flow();
58 :
59 : write_flow(const write_flow&) = delete;
60 : write_flow& operator=(const write_flow&) = delete;
61 :
62 : /// The watermark state is re-evaluated with the new limits
63 : void set_limits(const write_limits& limits);
64 6642 : const write_limits& limits() const { return limits_; }
65 :
66 : /// Whether a message of `size` bytes goes over the limits, once `dropped_bytes` and
67 : /// `dropped_messages` are out of the queue
68 : bool overflows(size_t size, size_t dropped_bytes = 0, size_t dropped_messages = 0) const;
69 :
70 : /// What to do with a message sent without waiting
71 : enum class admission { queue, drop, close };
72 :
73 : /// The overflow policy for a message of `size` bytes sent without waiting.
74 : /// `drop_oldest(size)` drops what the connection may and tells whether it fits then
75 : template<typename DropOldest>
76 2490487 : admission admit(size_t size, DropOldest&& drop_oldest) {
77 2490487 : if(!overflows(size)) return admission::queue;
78 2409859 : switch(limits_.policy){
79 1205764 : case overflow_policy::drop_newest: return admission::drop;
80 39 : case overflow_policy::close: return admission::close;
81 1204056 : case overflow_policy::drop_oldest: break;
82 : }
83 1204056 : return drop_oldest(size) ? admission::queue : admission::drop;
84 : }
85 :
86 : /// A message of `size` bytes is queued (may reach the high watermark)
87 : void add(size_t size);
88 :
89 : /// `messages` of `bytes` in all are written or dropped (may drain to the low watermark)
90 : void remove(size_t bytes, size_t messages = 1);
91 :
92 : /// Nothing is queued anymore, without notifications: once closed
93 : void clear();
94 :
95 37152304 : size_t queued_bytes() const { return queued_bytes_.load(std::memory_order_relaxed); }
96 32108650 : size_t queued_messages() const { return queued_messages_.load(std::memory_order_relaxed); }
97 :
98 : /// Above the high watermark, not drained to the low one yet
99 : bool congested() const { return congested_; }
100 :
101 : /// Wait while congested: true once drained, false if closed (or cancelled)
102 : awaitable<bool> wait_writable();
103 :
104 : /// From any thread: run on the connection thread, `owner` (the connection) kept alive
105 : void set_limits(std::shared_ptr<void> owner, const write_limits& limits);
106 : void on_high_watermark(std::shared_ptr<void> owner, sync_callback<void()> callback);
107 : void on_low_watermark(std::shared_ptr<void> owner, sync_callback<void()> callback);
108 : awaitable<bool> wait_writable(std::shared_ptr<void> owner);
109 :
110 : /// A write starts, makes progress (some bytes taken by the socket), and finishes: one
111 : /// without progress for the write timeout calls `on_timeout`, by a writer that keeps
112 : /// its connection alive until the write finishes. Progress moves the deadline lazily
113 : /// (see asio::watchdog): no timer operation
114 : void write_started(sync_callback<void()> on_timeout);
115 : void write_progressed();
116 : void write_finished();
117 :
118 : /// The bytes the socket holds that the peer has not taken (asio::socket::send_backlog()),
119 : /// checked once a write made no progress for the write timeout: their fall is progress
120 : /// too. A backlog that does not fall within 1/BACKLOG_WATCH of the timeout after it times
121 : /// the write out; one that does is checked every 1/BACKLOG_CHECKS of it from then. The
122 : /// system may wake a writer only once much of its buffer is free (a third of it, over TCP
123 : /// on Linux), long after a slow reader took the first byte. A write that progresses
124 : /// within its timeout is never checked: no timer operation per message
125 : void set_send_backlog(sync_callback<size_t()> backlog);
126 : static constexpr int BACKLOG_CHECKS = 16;
127 : static constexpr int BACKLOG_WATCH = 4;
128 :
129 : /// Waits end, pending notifications are dropped and the callbacks released
130 : void close();
131 : bool closed() const { return closed_; }
132 :
133 : private:
134 : /// Shared with the notifications posted, which may outlive this
135 : struct shared_state {
136 : bool closed = false;
137 : sync_callback<void()> on_high;
138 : sync_callback<void()> on_low;
139 : };
140 :
141 : void dispatch(std::shared_ptr<void> owner, sync_callback<void()> operation);
142 : /// The write deadline from the last progress, with the limits set now
143 : void arm_write_timer();
144 : /// The timer expired: a backlog check, or the write timed out
145 : void write_expired();
146 : void update_watermarks();
147 : void notify(bool high);
148 :
149 : boost::asio::any_io_executor executor_;
150 : std::shared_ptr<shared_state> state_;
151 : write_limits limits_;
152 : std::atomic<size_t> queued_bytes_{0};
153 : std::atomic<size_t> queued_messages_{0};
154 : bool congested_ = false;
155 : bool closed_ = false;
156 : bool write_watched_ = false;
157 : /// Set once the queue drains (or closes), waking the producers waiting; reset while congested
158 : asio::async_event drained_;
159 : /// The write timeout, and what it calls: released once the write finishes
160 : asio::watchdog write_timer_;
161 : std::chrono::steady_clock::time_point write_progress_{};
162 : sync_callback<void()> on_write_timeout_;
163 : sync_callback<size_t()> send_backlog_;
164 : /// The backlog at the last check since the last progress, if checked
165 : size_t backlog_seen_ = 0;
166 : bool backlog_checked_ = false;
167 : };
168 :
169 : }
170 :
171 : #endif
|
