Line data Source code
1 : #include "flow_control.hpp"
2 : #include <algorithm>
3 :
4 : namespace thinger::http {
5 :
6 3669 : write_flow::write_flow(boost::asio::any_io_executor executor, write_limits limits) :
7 3669 : executor_(std::move(executor)),
8 3669 : state_(std::make_shared<shared_state>()),
9 3669 : drained_(executor_),
10 : // a member: never called once this is gone
11 7970 : write_timer_(executor_, [this]{ write_expired(); })
12 : {
13 3669 : set_limits(limits);
14 3669 : }
15 :
16 3669 : write_flow::~write_flow(){
17 : // notifications still posted find it closed
18 3669 : state_->closed = true;
19 3669 : }
20 :
21 4374 : void write_flow::set_limits(const write_limits& limits){
22 4374 : auto timed = limits_.write_timeout.count() > 0;
23 4374 : limits_ = limits;
24 : // the low watermark is never above the high one
25 4374 : limits_.low_watermark = std::min(limits_.low_watermark, limits_.high_watermark);
26 4374 : update_watermarks();
27 : // a pending write gets the new deadline at once: from its last progress, or from now
28 : // if it had none (0 disarms it)
29 4374 : if(write_watched_ && !closed_){
30 24 : if(!timed) write_progress_ = std::chrono::steady_clock::now();
31 24 : arm_write_timer();
32 : }
33 4374 : }
34 :
35 6102685 : bool write_flow::overflows(size_t size, size_t dropped_bytes, size_t dropped_messages) const{
36 6102685 : auto messages = queued_messages() - std::min(dropped_messages, queued_messages());
37 6102685 : auto bytes = queued_bytes() - std::min(dropped_bytes, queued_bytes());
38 6102685 : if(limits_.max_messages && messages >= limits_.max_messages) return true;
39 2488623 : return limits_.max_bytes && bytes > 0 && bytes + size > limits_.max_bytes;
40 : }
41 :
42 1291329 : void write_flow::add(size_t size){
43 : // single writer (the connection thread): relaxed stores, for readers on other threads
44 1291329 : queued_bytes_.store(queued_bytes() + size, std::memory_order_relaxed);
45 1291329 : queued_messages_.store(queued_messages() + 1, std::memory_order_relaxed);
46 1291329 : update_watermarks();
47 1291329 : }
48 :
49 1259247 : void write_flow::remove(size_t bytes, size_t messages){
50 1259247 : queued_bytes_.store(queued_bytes() - std::min(bytes, queued_bytes()), std::memory_order_relaxed);
51 1259247 : queued_messages_.store(queued_messages() - std::min(messages, queued_messages()), std::memory_order_relaxed);
52 1259247 : update_watermarks();
53 1259247 : }
54 :
55 3630 : void write_flow::clear(){
56 3630 : queued_bytes_.store(0, std::memory_order_relaxed);
57 3630 : queued_messages_.store(0, std::memory_order_relaxed);
58 3630 : }
59 :
60 2554950 : void write_flow::update_watermarks(){
61 2554950 : if(closed_) return;
62 2554768 : auto queued = queued_bytes();
63 2554768 : if(!congested_ && limits_.high_watermark && queued >= limits_.high_watermark){
64 249 : congested_ = true;
65 249 : drained_.reset();
66 249 : notify(true);
67 2554519 : }else if(congested_ && (!limits_.high_watermark || queued <= limits_.low_watermark)){
68 237 : congested_ = false;
69 237 : drained_.set();
70 237 : notify(false);
71 : }
72 : }
73 :
74 486 : void write_flow::notify(bool high){
75 486 : if(!(high ? state_->on_high : state_->on_low)) return;
76 : // never from within a send or a drop: whatever the callback does (sending, closing...)
77 : // finds the queue consistent. Dropped if the connection closes meanwhile
78 66 : boost::asio::post(executor_, [state = state_, high]{
79 66 : if(state->closed) return;
80 66 : auto callback = high ? state->on_high : state->on_low;
81 66 : if(callback) callback();
82 66 : });
83 : }
84 :
85 9633 : awaitable<bool> write_flow::wait_writable(){
86 : while(congested_ && !closed_){
87 : co_await drained_.wait();
88 : auto state = co_await boost::asio::this_coro::cancellation_state;
89 : if(state.cancelled() != boost::asio::cancellation_type::none) co_return false;
90 : }
91 : co_return !closed_;
92 19266 : }
93 :
94 759 : void write_flow::dispatch(std::shared_ptr<void> owner, sync_callback<void()> operation){
95 759 : boost::asio::dispatch(executor_, [owner = std::move(owner), operation = std::move(operation)]{
96 759 : operation();
97 759 : });
98 759 : }
99 :
100 693 : void write_flow::set_limits(std::shared_ptr<void> owner, const write_limits& limits){
101 1386 : dispatch(std::move(owner), [this, limits]{ set_limits(limits); });
102 693 : }
103 :
104 27 : void write_flow::on_high_watermark(std::shared_ptr<void> owner, sync_callback<void()> callback){
105 27 : dispatch(std::move(owner), [this, callback = std::move(callback)]() mutable {
106 27 : if(!closed_) state_->on_high = std::move(callback);
107 27 : });
108 27 : }
109 :
110 39 : void write_flow::on_low_watermark(std::shared_ptr<void> owner, sync_callback<void()> callback){
111 39 : dispatch(std::move(owner), [this, callback = std::move(callback)]() mutable {
112 39 : if(!closed_) state_->on_low = std::move(callback);
113 39 : });
114 39 : }
115 :
116 9633 : awaitable<bool> write_flow::wait_writable(std::shared_ptr<void> owner){
117 9633 : return on_executor(executor_, std::move(owner), wait_writable());
118 : }
119 :
120 55570 : void write_flow::write_started(sync_callback<void()> on_timeout){
121 55570 : if(closed_) return;
122 : // watched also without a deadline, which set_limits() may give it meanwhile
123 55570 : write_watched_ = true;
124 55570 : on_write_timeout_ = std::move(on_timeout);
125 55570 : write_progressed();
126 : }
127 :
128 80980 : void write_flow::write_progressed(){
129 80980 : if(!write_watched_ || limits_.write_timeout.count() <= 0) return;
130 80467 : write_progress_ = std::chrono::steady_clock::now();
131 80467 : backlog_checked_ = false;
132 80467 : arm_write_timer();
133 : }
134 :
135 3642 : void write_flow::set_send_backlog(sync_callback<size_t()> backlog){
136 3642 : send_backlog_ = std::move(backlog);
137 3642 : }
138 :
139 : namespace {
140 491 : std::chrono::milliseconds backlog_interval(std::chrono::milliseconds timeout){
141 491 : return std::max(timeout / write_flow::BACKLOG_CHECKS, std::chrono::milliseconds(1));
142 : }
143 : }
144 :
145 80491 : void write_flow::arm_write_timer(){
146 80491 : if(limits_.write_timeout.count() <= 0){
147 3 : write_timer_.cancel();
148 3 : return;
149 : }
150 : // a later deadline does not touch the timer (see watchdog): the backlog is checked only
151 : // once it passes, so a write that progresses within it costs no timer operation
152 80488 : write_timer_.expires_at(write_progress_ + limits_.write_timeout);
153 : }
154 :
155 632 : void write_flow::write_expired(){
156 : // the deadline of a write already finished (see write_finished())
157 1156 : if(!write_watched_) return;
158 605 : if(send_backlog_ && limits_.write_timeout.count() > 0){
159 587 : auto now = std::chrono::steady_clock::now();
160 587 : auto backlog = send_backlog_();
161 : // the first check, at the deadline, watches the backlog for BACKLOG_WATCH more (a peer
162 : // may take bytes in bursts, e.g. a window of one segment, 64 KiB over loopback); then
163 : // the peer taking bytes since the last check is progress, as the socket taking some
164 587 : auto deadline = now + limits_.write_timeout / BACKLOG_WATCH;
165 587 : if(backlog_checked_){
166 491 : if(backlog < backlog_seen_) write_progress_ = now;
167 491 : deadline = std::min(now + backlog_interval(limits_.write_timeout), write_progress_ + limits_.write_timeout);
168 : }
169 587 : backlog_seen_ = backlog;
170 587 : backlog_checked_ = true;
171 587 : if(now < deadline){
172 524 : write_timer_.expires_at(deadline);
173 524 : return;
174 : }
175 : }
176 : // on_timeout may destroy this, so nothing of it is touched after the call
177 81 : auto on_timeout = std::move(on_write_timeout_);
178 81 : on_write_timeout_ = nullptr;
179 81 : if(on_timeout) on_timeout();
180 81 : }
181 :
182 55536 : void write_flow::write_finished(){
183 55536 : if(!write_watched_) return;
184 55536 : write_watched_ = false;
185 : // the timer is left armed: the next write's deadline, later, does not touch it (see
186 : // watchdog), so a stream of writes costs no timer operation each; an expiry with no
187 : // write pending does nothing
188 55536 : on_write_timeout_ = nullptr;
189 : }
190 :
191 3630 : void write_flow::close(){
192 3630 : closed_ = true;
193 3630 : congested_ = false;
194 3630 : state_->closed = true;
195 3630 : state_->on_high = nullptr;
196 3630 : state_->on_low = nullptr;
197 3630 : drained_.set();
198 3630 : write_timer_.cancel();
199 3630 : on_write_timeout_ = nullptr;
200 3630 : }
201 :
202 : }
|
