Line data Source code
1 : #include "async_client.hpp"
2 : #include "../../util/logger.hpp"
3 : #include <algorithm>
4 : #include <condition_variable>
5 : #include <list>
6 : #include <mutex>
7 : #include <thread>
8 : #include <vector>
9 :
10 : namespace thinger::http {
11 :
12 : namespace detail {
13 :
14 : // How long stop() and the destructor wait for the cancelled requests to end, before leaving
15 : // behind what remains of them: the cancellation reaches every operation of a request at once,
16 : // unless one ignores it
17 : constexpr auto end_grace = std::chrono::seconds{1};
18 :
19 : // The HTTP client the requests of an async_client use, and the requests in flight. The client
20 : // holds it, and each request as long as any of its frames exists
21 : class client_state final : public http_client_base {
22 : public:
23 625 : explicit client_state(asio::workers& workers) : workers(workers) {}
24 :
25 : asio::workers& workers;
26 :
27 : std::mutex mutex;
28 : std::condition_variable changed;
29 :
30 : // Guarded by the mutex: the requests in flight, the threads running a synchronous callback
31 : // of one of them, and how many calls of stop() or the destructor run (or the destructor
32 : // ran), which start no request
33 : std::list<std::shared_ptr<client_request>> requests;
34 : std::vector<std::thread::id> callbacks;
35 : unsigned closers = 0;
36 : // The requests in flight not cancelled (and those the application tracks), written under
37 : // the mutex
38 : std::atomic<size_t> active{0};
39 :
40 : // Whether a thread other than the caller runs a callback (lock held)
41 685 : bool callbacks_elsewhere() const {
42 685 : return std::any_of(callbacks.begin(), callbacks.end(),
43 1417 : [this_thread = std::this_thread::get_id()](auto id) { return id != this_thread; });
44 : }
45 :
46 : protected:
47 : // The worker of the calling thread, or else one of the pool
48 2688 : boost::asio::io_context& get_io_context() override { return workers.get_thread_io_context(); }
49 : };
50 :
51 : // A request in flight: where it runs, what cancels it, and its place among the requests
52 : class client_request {
53 : public:
54 2764 : client_request(std::shared_ptr<client_state> state, boost::asio::io_context::executor_type executor)
55 2764 : : state(std::move(state)), executor(executor) {}
56 :
57 : std::shared_ptr<client_state> state;
58 : boost::asio::io_context::executor_type executor;
59 : // Connected by co_spawn to the coroutine; emitted on its thread
60 : boost::asio::cancellation_signal signal;
61 : std::list<std::shared_ptr<client_request>>::iterator position;
62 : // Guarded by the mutex of the state: its coroutine is spawned (the signal may be emitted), it
63 : // was cancelled, and stop() or the destructor gave up waiting for it
64 : bool spawned = false;
65 : bool cancelled = false;
66 : bool left_behind = false;
67 : // Set under the mutex, read too on its thread, where it is set
68 : std::atomic<bool> ended{false};
69 :
70 : // Cancel it, on its thread: whatever it waits for ends with operation_aborted
71 1656 : static void cancel(const std::shared_ptr<client_request>& request) {
72 1656 : boost::asio::dispatch(request->executor, [request] {
73 1655 : if (!request->ended) request->signal.emit(boost::asio::cancellation_type::terminal);
74 1655 : });
75 1656 : }
76 :
77 2764 : void end() {
78 2764 : std::lock_guard lock(state->mutex);
79 2764 : ended = true;
80 2764 : if (!cancelled && state->active > 0) --state->active;
81 2764 : state->requests.erase(position);
82 2764 : state->changed.notify_all();
83 2764 : }
84 : };
85 :
86 : } // namespace detail
87 :
88 625 : async_client::async_client(asio::workers& workers)
89 625 : : state_(std::make_shared<detail::client_state>(workers)), client_(*state_) {
90 625 : LOG_DEBUG("Created HTTP async client");
91 625 : }
92 :
93 625 : async_client::~async_client() {
94 625 : LOG_DEBUG("Destroying HTTP async client");
95 625 : running_ = false;
96 : // The idle connections close now, those in use when their requests give them back: they
97 : // belong to the io_contexts of the workers, which the requests may outlive as frames
98 625 : client_.max_idle_connections_per_host(0);
99 625 : close_requests(true);
100 625 : }
101 :
102 27640 : async_client::request_end::~request_end() {
103 27640 : if (request_) request_->end();
104 27640 : }
105 :
106 22159 : async_client::callback_scope::callback_scope(detail::client_request& request) : request_(request) {
107 22159 : auto& state = *request.state;
108 22159 : std::lock_guard lock(state.mutex);
109 22159 : entered_ = !request.cancelled;
110 22159 : if (entered_) state.callbacks.push_back(std::this_thread::get_id());
111 22159 : }
112 :
113 43931 : async_client::callback_scope::~callback_scope() {
114 22159 : if (!entered_) return;
115 21772 : auto& state = *request_.state;
116 21772 : std::lock_guard lock(state.mutex);
117 21772 : state.callbacks.erase(std::find(state.callbacks.begin(), state.callbacks.end(), std::this_thread::get_id()));
118 21772 : state.changed.notify_all();
119 22159 : }
120 :
121 1538 : async_client::request_context::request_context(detail::client_request& request)
122 1538 : : client(*request.state), request_(request) {}
123 :
124 9 : stream_callback async_client::request_context::chunks(stream_callback callback) const {
125 18 : return [&request = request_, callback = std::move(callback)](const stream_info& info) {
126 20634 : callback_scope running(request);
127 41268 : return running && callback(info);
128 20652 : };
129 : }
130 :
131 2776 : std::shared_ptr<detail::client_request> async_client::start_request() {
132 2776 : auto& io_context = state_->workers.get_thread_io_context();
133 2776 : std::lock_guard lock(state_->mutex);
134 2776 : if (state_->closers > 0) {
135 12 : LOG_DEBUG("Async client: stopping, a request started now does not run");
136 12 : return nullptr;
137 : }
138 2764 : auto request = std::make_shared<detail::client_request>(state_, io_context.get_executor());
139 2764 : request->position = state_->requests.insert(state_->requests.end(), request);
140 2764 : ++state_->active;
141 2764 : return request;
142 2776 : }
143 :
144 2764 : boost::asio::io_context::executor_type async_client::executor_of(const detail::client_request& request) {
145 2764 : return request.executor;
146 : }
147 :
148 2764 : boost::asio::cancellation_slot async_client::slot_of(detail::client_request& request) {
149 2764 : return request.signal.slot();
150 : }
151 :
152 2764 : void async_client::spawned(const std::shared_ptr<detail::client_request>& request) {
153 : bool cancel;
154 : {
155 2764 : std::lock_guard lock(request->state->mutex);
156 2764 : request->spawned = true;
157 2764 : cancel = request->cancelled;
158 2764 : }
159 2764 : if (cancel) detail::client_request::cancel(request);
160 2764 : }
161 :
162 2765 : bool async_client::cancelled(detail::client_request& request) {
163 2765 : std::lock_guard lock(request.state->mutex);
164 2765 : return request.cancelled;
165 2765 : }
166 :
167 763 : void async_client::close_requests(bool final) {
168 763 : auto& state = *state_;
169 : // The requests this call waits for (any cancelled before too, e.g. from a worker, which did
170 : // not wait for them), and those it cancels: one not spawned yet is cancelled by spawned()
171 763 : std::vector<std::shared_ptr<detail::client_request>> ending;
172 763 : std::vector<std::shared_ptr<detail::client_request>> cancelling;
173 : {
174 763 : std::lock_guard lock(state.mutex);
175 763 : ++state.closers;
176 2518 : for (const auto& request : state.requests) {
177 1755 : if (request->left_behind) continue;
178 1755 : ending.push_back(request);
179 1755 : if (request->cancelled) continue;
180 1656 : request->cancelled = true;
181 1656 : if (request->spawned) cancelling.push_back(request);
182 : }
183 : // Release wait() and wait_for(): the client no longer runs
184 763 : state.changed.notify_all();
185 763 : }
186 2419 : for (const auto& request : cancelling) detail::client_request::cancel(request);
187 :
188 : // No request calls its callback from now on, so those running only end. They are waited for
189 : // unless stop() runs on a worker: the client lives on, and two callbacks stopping it at once
190 : // would wait for each other. The cancelled requests end then (their worker may have been
191 : // running a callback), unless this is one of the workers: those of its thread could not run
192 : // meanwhile, and those of other threads may use connections of this one
193 763 : bool on_worker = state.workers.in_worker_thread();
194 763 : std::unique_lock lock(state.mutex);
195 1448 : if (final || !on_worker) state.changed.wait(lock, [&] { return !state.callbacks_elsewhere(); });
196 763 : if (!on_worker && state.workers.running()) {
197 530 : auto has_ended = [](const auto& request) { return request->ended.load(); };
198 523 : state.changed.wait_for(lock, detail::end_grace, [&] {
199 651 : return std::all_of(ending.begin(), ending.end(), has_ended);
200 : });
201 655 : for (const auto& request : ending) {
202 132 : if (!has_ended(request)) request->left_behind = true;
203 : }
204 : }
205 :
206 763 : state.active = 0;
207 763 : if (!final) --state.closers;
208 763 : state.changed.notify_all();
209 763 : }
210 :
211 0 : void async_client::track_request_start() {
212 0 : std::lock_guard lock(state_->mutex);
213 0 : ++state_->active;
214 0 : }
215 :
216 0 : void async_client::track_request_end() {
217 0 : std::lock_guard lock(state_->mutex);
218 0 : if (state_->active > 0 && --state_->active == 0) state_->changed.notify_all();
219 0 : }
220 :
221 60 : size_t async_client::pending_requests() const {
222 120 : return state_->active.load();
223 : }
224 :
225 138 : bool async_client::stop() {
226 138 : LOG_DEBUG("Stopping HTTP async client");
227 138 : bool result = running_.exchange(false);
228 138 : close_requests(false);
229 : // The idle connections belong to the io_contexts of the workers, which may go away
230 138 : clear_connections();
231 138 : return result;
232 : }
233 :
234 361 : void async_client::wait() {
235 361 : std::unique_lock lock(state_->mutex);
236 1711 : state_->changed.wait(lock, [this] { return state_->active == 0 || !running_; });
237 361 : }
238 :
239 48 : bool async_client::wait_for(std::chrono::milliseconds timeout) {
240 48 : std::unique_lock lock(state_->mutex);
241 276 : bool completed = state_->changed.wait_for(lock, timeout, [this] { return state_->active == 0 || !running_; });
242 96 : return completed && state_->active == 0;
243 48 : }
244 :
245 : } // namespace thinger::http
|
