/src/crow/include/crow/http_server.h

Source
#pragma once

#ifdef CROW_USE_BOOST
#include <boost/asio.hpp>
#ifdef CROW_ENABLE_SSL
#include <boost/asio/ssl.hpp>
#endif
#else
#ifndef ASIO_STANDALONE
#define ASIO_STANDALONE
#endif
#include <asio.hpp>
#ifdef CROW_ENABLE_SSL
#include <asio/ssl.hpp>
#endif
#endif

#include <atomic>
#include <chrono>
#include <cstdint>
#include <future>
#include <memory>
#include <thread>
#include <vector>

#include "crow/version.h"
#include "crow/http_connection.h"
#include "crow/logging.h"
#include "crow/task_timer.h"
#include "crow/socket_acceptors.h"


namespace crow // NOTE: Already documented in "crow/app.h"
{
#ifdef CROW_USE_BOOST
    namespace asio = boost::asio;
    using error_code = boost::system::error_code;
#else
    using error_code = asio::error_code;
#endif
    using tcp = asio::ip::tcp;
    using stream_protocol = asio::local::stream_protocol;

    template<typename Handler, typename Acceptor = TCPAcceptor, typename Adaptor = SocketAdaptor, typename... Middlewares>
    class Server
    {
    public:
      Server(Handler* handler,
             typename Acceptor::endpoint endpoint, 
             std::string server_name = std::string("Crow/") + VERSION,
             std::tuple<Middlewares...>* middlewares = nullptr,
             unsigned int concurrency = 1,
             uint8_t timeout = 5,
             typename Adaptor::context* adaptor_ctx = nullptr):
          concurrency_(concurrency),
          task_queue_length_pool_(concurrency_ - 1),
          acceptor_(io_context_),
          signals_(io_context_),
          tick_timer_(io_context_),
          handler_(handler),
          timeout_(timeout),
          server_name_(server_name),
          middlewares_(middlewares),
          adaptor_ctx_(adaptor_ctx)
        {
            if (startup_failed_) {
                CROW_LOG_ERROR << "Startup failed; not running server.";
                return;
            }

            error_code ec;

            acceptor_.raw_acceptor().open(endpoint.protocol(), ec);
            if (ec) {
                CROW_LOG_ERROR << "Failed to open acceptor: " << ec.message();
                startup_failed_ = true;
                return;
            }

            acceptor_.raw_acceptor().set_option(Acceptor::reuse_address_option(), ec);
            if (ec) {
                CROW_LOG_ERROR << "Failed to set socket option: " << ec.message();
                startup_failed_ = true;
                return;
            }

            acceptor_.raw_acceptor().bind(endpoint, ec);
            if (ec) {
                CROW_LOG_ERROR << "Failed to bind to " << acceptor_.address()
                            << ":" << acceptor_.port() << " - " << ec.message();
                startup_failed_ = true;
                return;
            }

            acceptor_.raw_acceptor().listen(tcp::acceptor::max_listen_connections, ec);
            if (ec) {
                CROW_LOG_ERROR << "Failed to listen on port: " << ec.message();
                startup_failed_ = true;
                return;
            }


        }

        void set_tick_function(std::chrono::milliseconds d, std::function<void()> f)
        {
            tick_interval_ = d;
            tick_function_ = f;
        }

        void on_tick()
        {
            tick_function_();
            tick_timer_.expires_after(std::chrono::milliseconds(tick_interval_.count()));
            tick_timer_.async_wait([this](const error_code& ec) {
                if (ec)
                    return;
                on_tick();
            });
        }

        void run()
        {

            if (startup_failed_) {
                CROW_LOG_ERROR << "Server startup failed. Aborting run().";
                return;
            }

            uint16_t worker_thread_count = concurrency_ - 1;
            for (int i = 0; i < worker_thread_count; i++)
                io_context_pool_.emplace_back(new asio::io_context());
            get_cached_date_str_pool_.resize(worker_thread_count);
            task_timer_pool_.resize(worker_thread_count);

            std::vector<std::future<void>> v;
            std::atomic<int> init_count(0);
            for (uint16_t i = 0; i < worker_thread_count; i++)
                v.push_back(
                  std::async(
                    std::launch::async, [this, i, &init_count] {
                        // thread local date string get function
                        auto last = std::chrono::steady_clock::now();

                        std::string date_str;
                        auto update_date_str = [&] {
                            auto last_time_t = time(0);
                            tm my_tm;

#if defined(_MSC_VER) || defined(__MINGW32__)
                            gmtime_s(&my_tm, &last_time_t);
#else
                            gmtime_r(&last_time_t, &my_tm);
#endif
                            date_str.resize(100);
                            size_t date_str_sz = strftime(&date_str[0], 99, "%a, %d %b %Y %H:%M:%S GMT", &my_tm);
                            date_str.resize(date_str_sz);
                        };
                        update_date_str();
                        get_cached_date_str_pool_[i] = [&]() -> std::string {
                            if (std::chrono::steady_clock::now() - last >= std::chrono::seconds(1))
                            {
                                last = std::chrono::steady_clock::now();
                                update_date_str();
                            }
                            return date_str;
                        };

                        // initializing task timers
                        detail::task_timer task_timer(*io_context_pool_[i]);
                        task_timer.set_default_timeout(timeout_);
                        task_timer_pool_[i] = &task_timer;
                        task_queue_length_pool_[i] = 0;

                        init_count++;
                        while (1)
                        {
                            try
                            {
                                if (io_context_pool_[i]->run() == 0)
                                {
                                    // when io_service.run returns 0, there are no more works to do.
                                    break;
                                }
                            }
                            catch (std::exception& e)
                            {
                                CROW_LOG_ERROR << "Worker Crash: An uncaught exception occurred: " << e.what();
                            }
                        }
                    }));

            if (tick_function_ && tick_interval_.count() > 0)
            {
                tick_timer_.expires_after(std::chrono::milliseconds(tick_interval_.count()));
                tick_timer_.async_wait(
                  [this](const error_code& ec) {
                      if (ec)
                          return;
                      on_tick();
                  });
            }
            handler_->port(acceptor_.port());
            handler_->address_is_bound();
            CROW_LOG_INFO << server_name_ 
                          << " server is running at " << acceptor_.url_display(handler_->ssl_used()) 
                          << " using " << concurrency_ << " threads";
            CROW_LOG_INFO << "Call `app.loglevel(crow::LogLevel::Warning)` to hide Info level logs.";

            signals_.async_wait(
              [&](const error_code& /*error*/, int /*signal_number*/) {
                  stop();
              });

            while (worker_thread_count != init_count)
                std::this_thread::yield();

            do_accept();

            std::thread(
              [this] {
                  notify_start();
                  io_context_.run();
                  CROW_LOG_INFO << "Exiting.";
              })
              .join();
        }

        void stop()
        {
            shutting_down_ = true; // Prevent the acceptor from taking new connections

            // Explicitly close the acceptor
            // else asio will throw an exception (linux only), when trying to start server again:
            // what():  bind: Address already in use
            if (acceptor_.raw_acceptor().is_open())
            {
                CROW_LOG_INFO << "Closing acceptor. " << &acceptor_;
                error_code ec;
                acceptor_.raw_acceptor().close(ec);
                if (ec)
                {
                    CROW_LOG_WARNING << "Failed to close acceptor: " << ec.message();
                }
            }

            for (auto& io_context : io_context_pool_)
            {
                if (io_context != nullptr)
                {
                    CROW_LOG_INFO << "Closing IO service " << &io_context;
                    io_context->stop(); // Close all io_services (and HTTP connections)
                }
            }

            CROW_LOG_INFO << "Closing main IO service (" << &io_context_ << ')';
            io_context_.stop(); // Close main io_service
        }

        
        uint16_t port() const {
            return acceptor_.local_endpoint().port();
        }

        /// Wait until the server has properly started or until timeout
        std::cv_status wait_for_start(std::chrono::steady_clock::time_point wait_until)
        {
            std::unique_lock<std::mutex> lock(start_mutex_);

            std::cv_status status = std::cv_status::no_timeout;
            while (!server_started_ && !startup_failed_ && status == std::cv_status::no_timeout)
                status = cv_started_.wait_until(lock, wait_until);
            return status;
        }


        void signal_clear()
        {
            signals_.clear();
        }

        void signal_add(int signal_number)
        {
            signals_.add(signal_number);
        }

    private:
        size_t pick_io_context_idx()
        {
            size_t min_queue_idx = 0;

            // TODO improve load balancing
            // size_t is used here to avoid the security issue https://codeql.github.com/codeql-query-help/cpp/cpp-comparison-with-wider-type/
            // even though the max value of this can be only uint16_t as concurrency is uint16_t.
            for (size_t i = 1; i < task_queue_length_pool_.size() && task_queue_length_pool_[min_queue_idx] > 0; i++)
            // No need to check other io_services if the current one has no tasks
            {
                if (task_queue_length_pool_[i] < task_queue_length_pool_[min_queue_idx])
                    min_queue_idx = i;
            }
            return min_queue_idx;
        }

        void do_accept()
        {
            if (!shutting_down_)
            {
                size_t context_idx = pick_io_context_idx();
                asio::io_context& ic = *io_context_pool_[context_idx];
                auto p = std::make_shared<Connection<Adaptor, Handler, Middlewares...>>(
                    ic, handler_, server_name_, middlewares_,
                    get_cached_date_str_pool_[context_idx], *task_timer_pool_[context_idx], adaptor_ctx_, task_queue_length_pool_[context_idx]);
                    
                CROW_LOG_DEBUG << &ic << " {" << context_idx << "} queue length: " << task_queue_length_pool_[context_idx];

                acceptor_.raw_acceptor().async_accept(
                  p->socket(),
                  [this, p, &ic](error_code ec) {
                      if (!ec)
                      {
                          asio::post(ic,
                            [p] {
                                p->start();
                            });
                      }
                      do_accept();
                  });
            }
        }

        /// Notify anything using `wait_for_start()` to proceed
        void notify_start()
        {
            std::unique_lock<std::mutex> lock(start_mutex_);
            server_started_ = true;
            cv_started_.notify_all();
        }

    private:
        unsigned int concurrency_{2};
        std::vector<std::atomic<unsigned int>> task_queue_length_pool_;
        std::vector<std::unique_ptr<asio::io_context>> io_context_pool_;
        asio::io_context io_context_;
        std::vector<detail::task_timer*> task_timer_pool_;
        std::vector<std::function<std::string()>> get_cached_date_str_pool_;
        Acceptor acceptor_;
        bool shutting_down_ = false;
        bool server_started_{false};
        bool startup_failed_ = false;
        std::condition_variable cv_started_;
        std::mutex start_mutex_;
        asio::signal_set signals_;

        asio::basic_waitable_timer<std::chrono::high_resolution_clock> tick_timer_;

        Handler* handler_;
        std::uint8_t timeout_;
        std::string server_name_;
        bool use_unix_;

        std::chrono::milliseconds tick_interval_;
        std::function<void()> tick_function_;

        std::tuple<Middlewares...>* middlewares_;

        typename Adaptor::context* adaptor_ctx_;
    };
} // namespace crow

Coverage Report

Created: 2026-01-16 06:16

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