Line data Source code
1 : #ifndef THINGER_HTTP_SERVER_DETAIL_PUSH_QUEUE_HPP
2 : #define THINGER_HTTP_SERVER_DETAIL_PUSH_QUEUE_HPP
3 :
4 : #include <atomic>
5 : #include <cstdint>
6 : #include <deque>
7 : #include <functional>
8 : #include <iterator>
9 : #include <memory>
10 : #include <optional>
11 : #include <utility>
12 : #include <boost/asio/co_spawn.hpp>
13 : #include <boost/asio/detached.hpp>
14 : #include <boost/system/error_code.hpp>
15 : #include "../../../asio/async_event.hpp"
16 : #include "../../../util/types.hpp"
17 : #include "../flow_control.hpp"
18 :
19 : namespace thinger::http::detail {
20 :
21 : // Why a push_queue stopped writing (see push_queue::hooks::on_failure)
22 : enum class push_failure {
23 : overflow, // a message sent without waiting to a full queue, with overflow_policy::close
24 : write_timeout, // a write not done within write_limits::write_timeout
25 : write_error // write_item() failed: the error it returned
26 : };
27 :
28 : // The queue, flow control and single writer that websocket_connection and sse_connection compose.
29 : // Not public: only those two classes use it, from their .cpp files.
30 : //
31 : // Items are written one at a time, in order, by one writer coroutine, started when the queue
32 : // gets an item and ended once it is empty or closed. A unit is what drop_oldest drops whole: the
33 : // items up to one that ends it (ends_unit()), e.g. a WebSocket message, or the fields of an SSE
34 : // event up to its data. The queue never closes its owner's connection: it stops writing, closes
35 : // (the producers waiting are told false) and tells its owner, through on_failure, which closes.
36 : //
37 : // On its owner's executor, except queued_bytes(), queued_messages(), dropped_messages() and the
38 : // calls said to be from any thread, which keep the owner alive while they hop to it
39 : template<typename Item>
40 : class push_queue {
41 : public:
42 : struct hooks {
43 : // Write one item to the wire: an error ends writing (push_failure::write_error)
44 : std::function<awaitable<boost::system::error_code>(Item&)> write_item;
45 : // What the write limits count of an item, in bytes
46 : size_t (*item_size)(const Item&) = nullptr;
47 : // Whether an item ends its unit (drop_oldest drops whole units)
48 : bool (*ends_unit)(const Item&) = nullptr;
49 : // The queue stopped writing, and closed: the owner closes its connection
50 : std::function<void(push_failure, boost::system::error_code)> on_failure;
51 : // The owner, kept alive while the writer runs and while a call from any thread hops to
52 : // the executor: never held while the queue is idle
53 : std::function<std::shared_ptr<void>()> owner;
54 : // The bytes its socket holds that the peer has not taken, if it tells (see
55 : // write_flow::set_send_backlog())
56 : sync_callback<size_t()> send_backlog = nullptr;
57 : };
58 :
59 : // What push() did with an item (see write_flow::admit())
60 : using admission = write_flow::admission;
61 :
62 3630 : push_queue(boost::asio::any_io_executor executor, write_limits defaults, hooks hooks) :
63 3630 : executor_(std::move(executor)),
64 3630 : hooks_(std::move(hooks)),
65 3630 : flow_(executor_, defaults)
66 : {
67 3630 : if(hooks_.send_backlog) flow_.set_send_backlog(std::move(hooks_.send_backlog));
68 3630 : }
69 :
70 : push_queue(const push_queue&) = delete;
71 : push_queue& operator=(const push_queue&) = delete;
72 :
73 : // Queue an item sent without waiting, or apply the overflow policy to it: dropped (this
74 : // one, or older units to make room), or the queue fails with push_failure::overflow
75 2491966 : admission push(Item item){
76 2491966 : if(closed_) return admission::drop;
77 3694543 : auto admitted = flow_.admit(hooks_.item_size(item), [this](size_t size){ return drop_oldest(size); });
78 2490487 : switch(admitted){
79 1284594 : case admission::queue:
80 1284594 : enqueue(entry{std::move(item), nullptr});
81 1284594 : break;
82 1205854 : case admission::drop:
83 1205854 : count_dropped(1);
84 1205854 : break;
85 39 : case admission::close:
86 39 : fail(push_failure::overflow, {});
87 39 : break;
88 : }
89 2490487 : return admitted;
90 1284594 : }
91 :
92 : // From any thread: queue an item and wait until written, true, or false if the queue closed
93 : // or failed first. Never dropped by the overflow policy, which it may go over; a cancelled
94 : // wait leaves the item queued
95 7941 : awaitable<bool> push_and_wait(Item item){
96 7941 : return on_executor(executor_, hooks_.owner(), wait_written(std::move(item)));
97 : }
98 :
99 : // Write `item` at once if nothing is queued nor being written: outside the write limits and
100 : // the units, never dropped (e.g. an SSE keep-alive comment). False if not idle, or closed
101 458 : bool write_idle(Item item){
102 458 : if(closed_ || writing_ || !queue_.empty()) return false;
103 333 : idle_item_ = std::move(item);
104 333 : start_writer();
105 333 : return true;
106 : }
107 :
108 : // The item being written made progress (the socket took some of it): its write deadline
109 : // counts from now. From write_item()
110 25383 : void write_progressed(){ flow_.write_progressed(); }
111 :
112 : // Whether the next item sent without waiting overflows the queue (not the watermarks)
113 12 : bool full() const { return flow_.overflows(1); }
114 :
115 : // From any thread
116 693 : void set_write_limits(const write_limits& limits){ flow_.set_limits(hooks_.owner(), limits); }
117 6639 : const write_limits& get_write_limits() const { return flow_.limits(); }
118 27 : void on_high_watermark(sync_callback<void()> callback){ flow_.on_high_watermark(hooks_.owner(), std::move(callback)); }
119 39 : void on_low_watermark(sync_callback<void()> callback){ flow_.on_low_watermark(hooks_.owner(), std::move(callback)); }
120 9633 : awaitable<bool> wait_writable(){ return flow_.wait_writable(hooks_.owner()); }
121 :
122 : // Bytes, and items, not written yet (the one being written included); from any thread
123 6191 : size_t queued_bytes() const { return flow_.queued_bytes(); }
124 39132 : size_t queued_messages() const { return flow_.queued_messages(); }
125 :
126 : // Items sent without waiting that the overflow policy dropped so far: the new one, or the older
127 : // ones it made room with. Not what close() empties, nor what is pushed once closed. From any
128 : // thread: it never decreases, and acquires what led to the value (e.g. what was sent)
129 52026712 : uint64_t dropped_messages() const { return dropped_.load(std::memory_order_acquire); }
130 :
131 : // Nothing else is written: the watermark callbacks are released, producers waiting (and
132 : // those of the items queued) are told false, the queue is emptied. Once; an item being
133 : // written completes, its producer told whether it was written
134 4290 : void close(){
135 4290 : if(closed_) return;
136 3630 : closed_ = true;
137 : // closed first: nobody is told the queue drained
138 3630 : flow_.close();
139 3630 : idle_item_.reset();
140 3630 : auto discarded = std::move(queue_);
141 3630 : queue_.clear();
142 3630 : flow_.clear();
143 35126 : for(auto& entry : discarded){
144 31496 : if(entry.waiter) entry.waiter->finish(false);
145 : }
146 3630 : }
147 :
148 1207333 : bool closed() const { return closed_; }
149 :
150 : // Kept by the owner in its read loop, which lives while the connection is open: however the
151 : // loop ends, also when its frame is destroyed with the io_context, the queue closes and
152 : // `release` runs, releasing the owner's callbacks (which may own it: a cycle otherwise)
153 : class cycle_guard {
154 : public:
155 3627 : cycle_guard(push_queue& queue, sync_callback<void()> release) :
156 3627 : queue_(queue), release_(std::move(release)) {}
157 3627 : ~cycle_guard(){
158 3627 : queue_.close();
159 3627 : if(release_) release_();
160 3627 : }
161 : cycle_guard(const cycle_guard&) = delete;
162 : cycle_guard& operator=(const cycle_guard&) = delete;
163 : private:
164 : push_queue& queue_;
165 : sync_callback<void()> release_;
166 : };
167 :
168 3627 : [[nodiscard]] cycle_guard guard_cycles(sync_callback<void()> release){
169 3627 : return cycle_guard{*this, std::move(release)};
170 : }
171 :
172 : private:
173 : // A producer waiting for its item to be written (push_and_wait())
174 : struct producer {
175 6735 : explicit producer(const boost::asio::any_io_executor& executor) : done(executor) {}
176 6735 : void finish(bool success){
177 6735 : written = success;
178 6735 : done.set();
179 6735 : }
180 : asio::async_event done;
181 : bool written = false;
182 : };
183 :
184 : struct entry {
185 : Item item;
186 : std::shared_ptr<producer> waiter;
187 : };
188 :
189 : using iterator = typename std::deque<entry>::iterator;
190 :
191 7941 : awaitable<bool> wait_written(Item item){
192 : if(closed_) co_return false;
193 : auto pending = std::make_shared<producer>(executor_);
194 : enqueue(entry{std::move(item), pending});
195 : co_await pending->done.wait();
196 : co_return pending->written;
197 15882 : }
198 :
199 1291329 : void enqueue(entry queued){
200 1291329 : flow_.add(hooks_.item_size(queued.item));
201 1291329 : queue_.push_back(std::move(queued));
202 1291329 : start_writer();
203 1291329 : }
204 :
205 1291662 : void start_writer(){
206 1291662 : if(writing_ || closed_ || (queue_.empty() && !idle_item_)) return;
207 8966 : writing_ = true;
208 8966 : co_spawn(executor_, write_loop(hooks_.owner()), boost::asio::detached);
209 : }
210 :
211 : // Write the items, one at a time (`owner` kept alive meanwhile)
212 8966 : awaitable<void> write_loop([[maybe_unused]] std::shared_ptr<void> owner){
213 : while(!closed_ && (idle_item_ || !queue_.empty())){
214 : entry current;
215 : // written between units: the unit being written stays as it was
216 : bool counted = !idle_item_;
217 : if(idle_item_){
218 : current.item = std::move(*idle_item_);
219 : idle_item_.reset();
220 : }else{
221 : current = std::move(queue_.front());
222 : queue_.pop_front();
223 : unit_open_ = !hooks_.ends_unit(current.item);
224 : }
225 :
226 57 : flow_.write_started([this]{ fail(push_failure::write_timeout, {}); });
227 : auto ec = co_await hooks_.write_item(current.item);
228 : flow_.write_finished();
229 :
230 : if(counted) flow_.remove(hooks_.item_size(current.item));
231 : if(current.waiter) current.waiter->finish(!ec);
232 : if(ec){
233 : fail(push_failure::write_error, ec);
234 : break;
235 : }
236 : }
237 : writing_ = false;
238 17932 : }
239 :
240 : // Only ever written here, on the executor, so no read-modify-write; nothing per item queued
241 2409910 : void count_dropped(uint64_t items){
242 4819820 : dropped_.store(dropped_.load(std::memory_order_relaxed) + items, std::memory_order_release);
243 2409910 : }
244 :
245 : // Stop writing for good, and tell the owner (once)
246 257 : void fail(push_failure failure, boost::system::error_code ec){
247 257 : if(closed_) return;
248 135 : close();
249 135 : if(hooks_.on_failure) hooks_.on_failure(failure, ec);
250 : }
251 :
252 : // Past the unit `it` is in: the item ending it (or the end of the queue)
253 1204539 : iterator unit_end(iterator it){
254 1205691 : while(it != queue_.end() && !hooks_.ends_unit(it->item)) ++it;
255 2409027 : return it == queue_.end() ? it : std::next(it);
256 : }
257 :
258 1204176 : static bool awaited(iterator begin, iterator end){
259 2408811 : for(auto it = begin; it != end; ++it){
260 1204812 : if(it->waiter) return true;
261 : }
262 1203999 : return false;
263 : }
264 :
265 : // Drop the oldest whole units, until an item of `size` bytes fits: never the rest of the
266 : // unit being written, a unit with a producer waiting, nor the last unit while incomplete.
267 : // False if it does not fit even so
268 1204056 : bool drop_oldest(size_t size){
269 1204056 : auto it = queue_.begin();
270 1204056 : if(unit_open_) it = unit_end(it);
271 1204056 : size_t bytes = 0;
272 1204056 : size_t items = 0;
273 2408232 : while(it != queue_.end() && flow_.overflows(size, bytes, items)){
274 1204176 : auto end = unit_end(it);
275 2408352 : if(!hooks_.ends_unit(std::prev(end)->item)) break;
276 1204176 : if(awaited(it, end)){
277 177 : it = end;
278 177 : continue;
279 : }
280 2408634 : for(auto dropped = it; dropped != end; ++dropped) bytes += hooks_.item_size(dropped->item);
281 1203999 : items += static_cast<size_t>(std::distance(it, end));
282 1203999 : it = queue_.erase(it, end);
283 : }
284 : // accounted for once the queue is consistent
285 1204056 : flow_.remove(bytes, items);
286 1204056 : count_dropped(items);
287 2408112 : return !flow_.overflows(size);
288 : }
289 :
290 : boost::asio::any_io_executor executor_;
291 : hooks hooks_;
292 : // Items not written yet (the one being written is out of it)
293 : std::deque<entry> queue_;
294 : // Bytes and items queued against the write limits
295 : write_flow flow_;
296 : // An item to write before those queued, outside the limits and the units (write_idle())
297 : std::optional<Item> idle_item_;
298 : bool writing_ = false;
299 : // The last item taken for writing did not end its unit: the items up to its end belong to it
300 : bool unit_open_ = false;
301 : bool closed_ = false;
302 : // See dropped_messages()
303 : std::atomic<uint64_t> dropped_{0};
304 : };
305 :
306 : }
307 :
308 : #endif
|
