/src/crow/include/crow/http_server.h
Line | Count | Source |
1 | | #pragma once |
2 | | |
3 | | #ifdef CROW_USE_BOOST |
4 | | #include <boost/asio.hpp> |
5 | | #ifdef CROW_ENABLE_SSL |
6 | | #include <boost/asio/ssl.hpp> |
7 | | #endif |
8 | | #else |
9 | | #ifndef ASIO_STANDALONE |
10 | | #define ASIO_STANDALONE |
11 | | #endif |
12 | | #include <asio.hpp> |
13 | | #ifdef CROW_ENABLE_SSL |
14 | | #include <asio/ssl.hpp> |
15 | | #endif |
16 | | #endif |
17 | | |
18 | | #include <atomic> |
19 | | #include <chrono> |
20 | | #include <cstdint> |
21 | | #include <future> |
22 | | #include <memory> |
23 | | #include <system_error> |
24 | | #include <thread> |
25 | | #include <vector> |
26 | | |
27 | | #include "crow/version.h" |
28 | | #include "crow/http_connection.h" |
29 | | #include "crow/logging.h" |
30 | | #include "crow/task_timer.h" |
31 | | #include "crow/socket_acceptors.h" |
32 | | #include "crow/tcp_socket_options.h" |
33 | | |
34 | | |
35 | | namespace crow // NOTE: Already documented in "crow/app.h" |
36 | | { |
37 | | #ifdef CROW_USE_BOOST |
38 | | namespace asio = boost::asio; |
39 | | using error_code = boost::system::error_code; |
40 | | #else |
41 | | using error_code = asio::error_code; |
42 | | #endif |
43 | | using tcp = asio::ip::tcp; |
44 | | using stream_protocol = asio::local::stream_protocol; |
45 | | |
46 | | namespace detail |
47 | | { |
48 | | inline bool is_descriptor_exhaustion(const error_code& ec) |
49 | 0 | { |
50 | 0 | #ifdef CROW_USE_BOOST |
51 | 0 | return ec == boost::system::errc::too_many_files_open || |
52 | 0 | ec == boost::system::errc::too_many_files_open_in_system; |
53 | 0 | #else |
54 | 0 | return ec == std::errc::too_many_files_open || |
55 | 0 | ec == std::errc::too_many_files_open_in_system; |
56 | 0 | #endif |
57 | 0 | } |
58 | | } // namespace detail |
59 | | |
60 | | template<typename Handler, typename Acceptor = TCPAcceptor, typename Adaptor = SocketAdaptor, typename... Middlewares> |
61 | | class Server |
62 | | { |
63 | | public: |
64 | | Server(Handler* handler, |
65 | | typename Acceptor::endpoint endpoint, |
66 | | std::string server_name = std::string("Crow/") + VERSION, |
67 | | std::tuple<Middlewares...>* middlewares = nullptr, |
68 | | unsigned int concurrency = 1, |
69 | | uint8_t timeout = 5, |
70 | | typename Adaptor::context* adaptor_ctx = nullptr, |
71 | | detail::socket::tcp_socket_options tcp_socket_options = {}): |
72 | | concurrency_(concurrency), |
73 | | task_queue_length_pool_(concurrency_ - 1), |
74 | | acceptor_(io_context_), |
75 | | signals_(io_context_), |
76 | | tick_timer_(io_context_), |
77 | | accept_timer_(io_context_), |
78 | | handler_(handler), |
79 | | timeout_(timeout), |
80 | | server_name_(server_name), |
81 | | middlewares_(middlewares), |
82 | | adaptor_ctx_(adaptor_ctx), |
83 | | tcp_socket_options_(tcp_socket_options) |
84 | | { |
85 | | if (startup_failed_) { |
86 | | CROW_LOG_ERROR << "Startup failed; not running server."; |
87 | | return; |
88 | | } |
89 | | |
90 | | error_code ec; |
91 | | |
92 | | acceptor_.raw_acceptor().open(endpoint.protocol(), ec); |
93 | | if (ec) { |
94 | | CROW_LOG_ERROR << "Failed to open acceptor: " << ec.message(); |
95 | | startup_failed_ = true; |
96 | | return; |
97 | | } |
98 | | |
99 | | acceptor_.raw_acceptor().set_option(Acceptor::reuse_address_option(), ec); |
100 | | if (ec) { |
101 | | CROW_LOG_ERROR << "Failed to set socket option: " << ec.message(); |
102 | | startup_failed_ = true; |
103 | | return; |
104 | | } |
105 | | |
106 | | acceptor_.raw_acceptor().bind(endpoint, ec); |
107 | | if (ec) { |
108 | | CROW_LOG_ERROR << "Failed to bind to " << acceptor_.address() |
109 | | << ":" << acceptor_.port() << " - " << ec.message(); |
110 | | startup_failed_ = true; |
111 | | return; |
112 | | } |
113 | | |
114 | | acceptor_.raw_acceptor().listen(tcp::acceptor::max_listen_connections, ec); |
115 | | if (ec) { |
116 | | CROW_LOG_ERROR << "Failed to listen on port: " << ec.message(); |
117 | | startup_failed_ = true; |
118 | | return; |
119 | | } |
120 | | |
121 | | |
122 | | } |
123 | | |
124 | | void set_tick_function(std::chrono::milliseconds d, std::function<void()> f) |
125 | | { |
126 | | tick_interval_ = d; |
127 | | tick_function_ = f; |
128 | | } |
129 | | |
130 | | void on_tick() |
131 | | { |
132 | | tick_function_(); |
133 | | tick_timer_.expires_after(std::chrono::milliseconds(tick_interval_.count())); |
134 | | tick_timer_.async_wait([this](const error_code& ec) { |
135 | | if (ec) |
136 | | return; |
137 | | on_tick(); |
138 | | }); |
139 | | } |
140 | | |
141 | | void run() |
142 | | { |
143 | | |
144 | | if (startup_failed_) { |
145 | | CROW_LOG_ERROR << "Server startup failed. Aborting run()."; |
146 | | return; |
147 | | } |
148 | | |
149 | | uint16_t worker_thread_count = concurrency_ - 1; |
150 | | for (int i = 0; i < worker_thread_count; i++) |
151 | | io_context_pool_.emplace_back(new asio::io_context()); |
152 | | get_cached_date_str_pool_.resize(worker_thread_count); |
153 | | task_timer_pool_.resize(worker_thread_count); |
154 | | |
155 | | std::vector<std::future<void>> v; |
156 | | std::atomic<int> init_count(0); |
157 | | for (uint16_t i = 0; i < worker_thread_count; i++) |
158 | | v.push_back( |
159 | | std::async( |
160 | | std::launch::async, [this, i, &init_count] { |
161 | | // thread local date string get function |
162 | | auto last = std::chrono::steady_clock::now(); |
163 | | |
164 | | std::string date_str; |
165 | | auto update_date_str = [&] { |
166 | | auto last_time_t = time(0); |
167 | | tm my_tm; |
168 | | |
169 | | #if defined(_MSC_VER) || defined(__MINGW32__) |
170 | | gmtime_s(&my_tm, &last_time_t); |
171 | | #else |
172 | | gmtime_r(&last_time_t, &my_tm); |
173 | | #endif |
174 | | date_str.resize(100); |
175 | | size_t date_str_sz = strftime(&date_str[0], 99, "%a, %d %b %Y %H:%M:%S GMT", &my_tm); |
176 | | date_str.resize(date_str_sz); |
177 | | }; |
178 | | update_date_str(); |
179 | | get_cached_date_str_pool_[i] = [&]() -> std::string { |
180 | | if (std::chrono::steady_clock::now() - last >= std::chrono::seconds(1)) |
181 | | { |
182 | | last = std::chrono::steady_clock::now(); |
183 | | update_date_str(); |
184 | | } |
185 | | return date_str; |
186 | | }; |
187 | | |
188 | | // initializing task timers |
189 | | detail::task_timer task_timer(*io_context_pool_[i]); |
190 | | task_timer.set_default_timeout(timeout_); |
191 | | task_timer_pool_[i] = &task_timer; |
192 | | task_queue_length_pool_[i] = 0; |
193 | | |
194 | | init_count++; |
195 | | while (1) |
196 | | { |
197 | | try |
198 | | { |
199 | | if (io_context_pool_[i]->run() == 0) |
200 | | { |
201 | | // when io_service.run returns 0, there are no more works to do. |
202 | | break; |
203 | | } |
204 | | } |
205 | | catch (std::exception& e) |
206 | | { |
207 | | CROW_LOG_ERROR << "Worker Crash: An uncaught exception occurred: " << e.what(); |
208 | | } |
209 | | } |
210 | | })); |
211 | | |
212 | | if (tick_function_ && tick_interval_.count() > 0) |
213 | | { |
214 | | tick_timer_.expires_after(std::chrono::milliseconds(tick_interval_.count())); |
215 | | tick_timer_.async_wait( |
216 | | [this](const error_code& ec) { |
217 | | if (ec) |
218 | | return; |
219 | | on_tick(); |
220 | | }); |
221 | | } |
222 | | handler_->port(acceptor_.port()); |
223 | | handler_->address_is_bound(); |
224 | | CROW_LOG_INFO << server_name_ |
225 | | << " server is running at " << acceptor_.url_display(handler_->ssl_used()) |
226 | | << " using " << concurrency_ << " threads"; |
227 | | CROW_LOG_INFO << "Call `app.loglevel(crow::LogLevel::Warning)` to hide Info level logs."; |
228 | | |
229 | | signals_.async_wait( |
230 | | [&](const error_code& /*error*/, int /*signal_number*/) { |
231 | | stop(); |
232 | | }); |
233 | | |
234 | | while (worker_thread_count != init_count) |
235 | | std::this_thread::yield(); |
236 | | |
237 | | do_accept(); |
238 | | |
239 | | std::thread( |
240 | | [this] { |
241 | | notify_start(); |
242 | | io_context_.run(); |
243 | | CROW_LOG_INFO << "Exiting."; |
244 | | }) |
245 | | .join(); |
246 | | } |
247 | | |
248 | | void stop() |
249 | | { |
250 | | shutting_down_ = true; // Prevent the acceptor from taking new connections |
251 | | |
252 | | // Explicitly close the acceptor |
253 | | // else asio will throw an exception (linux only), when trying to start server again: |
254 | | // what(): bind: Address already in use |
255 | | if (acceptor_.raw_acceptor().is_open()) |
256 | | { |
257 | | CROW_LOG_INFO << "Closing acceptor. " << &acceptor_; |
258 | | error_code ec; |
259 | | acceptor_.raw_acceptor().close(ec); |
260 | | if (ec) |
261 | | { |
262 | | CROW_LOG_WARNING << "Failed to close acceptor: " << ec.message(); |
263 | | } |
264 | | } |
265 | | |
266 | | accept_timer_.cancel(); |
267 | | |
268 | | for (auto& io_context : io_context_pool_) |
269 | | { |
270 | | if (io_context != nullptr) |
271 | | { |
272 | | CROW_LOG_INFO << "Closing IO service " << &io_context; |
273 | | io_context->stop(); // Close all io_services (and HTTP connections) |
274 | | } |
275 | | } |
276 | | |
277 | | CROW_LOG_INFO << "Closing main IO service (" << &io_context_ << ')'; |
278 | | io_context_.stop(); // Close main io_service |
279 | | } |
280 | | |
281 | | |
282 | | uint16_t port() const { |
283 | | return acceptor_.local_endpoint().port(); |
284 | | } |
285 | | |
286 | | /// Wait until the server has properly started or until timeout |
287 | | std::cv_status wait_for_start(std::chrono::steady_clock::time_point wait_until) |
288 | | { |
289 | | std::unique_lock<std::mutex> lock(start_mutex_); |
290 | | |
291 | | std::cv_status status = std::cv_status::no_timeout; |
292 | | while (!server_started_ && !startup_failed_ && status == std::cv_status::no_timeout) |
293 | | status = cv_started_.wait_until(lock, wait_until); |
294 | | return status; |
295 | | } |
296 | | |
297 | | |
298 | | void signal_clear() |
299 | | { |
300 | | signals_.clear(); |
301 | | } |
302 | | |
303 | | void signal_add(int signal_number) |
304 | | { |
305 | | signals_.add(signal_number); |
306 | | } |
307 | | |
308 | | private: |
309 | | size_t pick_io_context_idx() |
310 | | { |
311 | | size_t min_queue_idx = 0; |
312 | | |
313 | | // TODO improve load balancing |
314 | | // size_t is used here to avoid the security issue https://codeql.github.com/codeql-query-help/cpp/cpp-comparison-with-wider-type/ |
315 | | // even though the max value of this can be only uint16_t as concurrency is uint16_t. |
316 | | for (size_t i = 1; i < task_queue_length_pool_.size() && task_queue_length_pool_[min_queue_idx] > 0; i++) |
317 | | // No need to check other io_services if the current one has no tasks |
318 | | { |
319 | | if (task_queue_length_pool_[i] < task_queue_length_pool_[min_queue_idx]) |
320 | | min_queue_idx = i; |
321 | | } |
322 | | return min_queue_idx; |
323 | | } |
324 | | |
325 | | void do_accept() |
326 | | { |
327 | | if (!shutting_down_) |
328 | | { |
329 | | size_t context_idx = pick_io_context_idx(); |
330 | | asio::io_context& ic = *io_context_pool_[context_idx]; |
331 | | auto p = std::make_shared<Connection<Adaptor, Handler, Middlewares...>>( |
332 | | ic, handler_, server_name_, middlewares_, |
333 | | get_cached_date_str_pool_[context_idx], *task_timer_pool_[context_idx], adaptor_ctx_, task_queue_length_pool_[context_idx]); |
334 | | |
335 | | CROW_LOG_DEBUG << &ic << " {" << context_idx << "} queue length: " << task_queue_length_pool_[context_idx]; |
336 | | |
337 | | acceptor_.raw_acceptor().async_accept( |
338 | | p->socket(), |
339 | | [this, p, &ic](error_code ec) { |
340 | | if (!ec) |
341 | | { |
342 | | detail::socket::apply_tcp_socket_options(p->socket(), tcp_socket_options_); |
343 | | asio::post(ic, |
344 | | [p] { |
345 | | p->start(); |
346 | | }); |
347 | | do_accept(); |
348 | | return; |
349 | | } |
350 | | |
351 | | if (shutting_down_ || ec == asio::error::operation_aborted) |
352 | | return; |
353 | | |
354 | | CROW_LOG_ERROR << "Failed to accept connection: " << ec.message(); |
355 | | if (detail::is_descriptor_exhaustion(ec)) |
356 | | { |
357 | | accept_timer_.expires_after(std::chrono::milliseconds(100)); |
358 | | accept_timer_.async_wait([this](const error_code& tec) { |
359 | | if (!tec) |
360 | | do_accept(); |
361 | | }); |
362 | | return; |
363 | | } |
364 | | do_accept(); |
365 | | }); |
366 | | } |
367 | | } |
368 | | |
369 | | /// Notify anything using `wait_for_start()` to proceed |
370 | | void notify_start() |
371 | | { |
372 | | std::unique_lock<std::mutex> lock(start_mutex_); |
373 | | server_started_ = true; |
374 | | cv_started_.notify_all(); |
375 | | } |
376 | | |
377 | | private: |
378 | | unsigned int concurrency_{2}; |
379 | | std::vector<std::atomic<unsigned int>> task_queue_length_pool_; |
380 | | std::vector<std::unique_ptr<asio::io_context>> io_context_pool_; |
381 | | asio::io_context io_context_; |
382 | | std::vector<detail::task_timer*> task_timer_pool_; |
383 | | std::vector<std::function<std::string()>> get_cached_date_str_pool_; |
384 | | Acceptor acceptor_; |
385 | | bool shutting_down_ = false; |
386 | | bool server_started_{false}; |
387 | | bool startup_failed_ = false; |
388 | | std::condition_variable cv_started_; |
389 | | std::mutex start_mutex_; |
390 | | asio::signal_set signals_; |
391 | | |
392 | | asio::basic_waitable_timer<std::chrono::high_resolution_clock> tick_timer_; |
393 | | asio::steady_timer accept_timer_; |
394 | | |
395 | | Handler* handler_; |
396 | | std::uint8_t timeout_; |
397 | | std::string server_name_; |
398 | | bool use_unix_; |
399 | | |
400 | | std::chrono::milliseconds tick_interval_; |
401 | | std::function<void()> tick_function_; |
402 | | |
403 | | std::tuple<Middlewares...>* middlewares_; |
404 | | |
405 | | typename Adaptor::context* adaptor_ctx_; |
406 | | detail::socket::tcp_socket_options tcp_socket_options_; |
407 | | }; |
408 | | } // namespace crow |