lib/cpp-httplib
1
0
Fork 0
mirror of synced 2025-06-07 22:02:00 +03:00
Code

new_task_queue support

Browse Source
This commit is contained in:
yhirose 2019-08-03 13:15:05 +09:00
parent 47312e6df9
commit 579ff1a0a6
1 changed files with 46 additions and 32 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

78
httplib.h
View File

@ -265,12 +265,20 @@ private:
std::string buffer; std::string buffer;
}; };
class ThreadsTaskQueue { class TaskQueue {
public:
TaskQueue() {}
virtual ~TaskQueue() {}
virtual void enque(std::function<void(void)> fn) = 0;
virtual void shutdown() = 0;
};
class ThreadsTaskQueue : public TaskQueue {
public: public:
ThreadsTaskQueue() : running_threads_(0) {} ThreadsTaskQueue() : running_threads_(0) {}
~ThreadsTaskQueue() {} virtual ~ThreadsTaskQueue() {}
void enque(std::function<void(void)> fn) { virtual void enque(std::function<void(void)> fn) override {
std::thread([=]() { std::thread([=]() {
{ {
std::lock_guard<std::mutex> guard(running_threads_mutex_); std::lock_guard<std::mutex> guard(running_threads_mutex_);
@ -286,7 +294,7 @@ public:
}).detach(); }).detach();
} }
void shutdown() { virtual void shutdown() override {
for (;;) { for (;;) {
std::this_thread::sleep_for(std::chrono::milliseconds(10)); std::this_thread::sleep_for(std::chrono::milliseconds(10));
std::lock_guard<std::mutex> guard(running_threads_mutex_); std::lock_guard<std::mutex> guard(running_threads_mutex_);
@ -299,8 +307,6 @@ private:
int running_threads_; int running_threads_;
}; };
typedef ThreadsTaskQueue TaskQueue;
class Server { class Server {
public: public:
typedef std::function<void(const Request &, Response &)> Handler; typedef std::function<void(const Request &, Response &)> Handler;
@ -336,6 +342,8 @@ public:
bool is_running() const; bool is_running() const;
void stop(); void stop();
std::function<TaskQueue*(void)> new_task_queue;
protected: protected:
bool process_request(Stream &strm, bool last_connection, bool process_request(Stream &strm, bool last_connection,
bool &connection_close, bool &connection_close,
@ -373,8 +381,6 @@ private:
Handlers options_handlers_; Handlers options_handlers_;
Handler error_handler_; Handler error_handler_;
Logger logger_; Logger logger_;
TaskQueue task_queue_;
}; };
class Client { class Client {
@ -2226,40 +2232,48 @@ inline int Server::bind_internal(const char *host, int port, int socket_flags) {
inline bool Server::listen_internal() { inline bool Server::listen_internal() {
auto ret = true; auto ret = true;
is_running_ = true; is_running_ = true;
for (;;) { {
if (svr_sock_ == INVALID_SOCKET) { std::unique_ptr<TaskQueue> task_queue;
// The server socket was closed by 'stop' method.
break; if (new_task_queue) {
task_queue.reset(new_task_queue());
} else {
task_queue.reset(new ThreadsTaskQueue());
} }
auto val = detail::select_read(svr_sock_, 0, 100000); for (;;) {
if (svr_sock_ == INVALID_SOCKET) {
if (val == 0) { // Timeout // The server socket was closed by 'stop' method.
continue; break;
}
socket_t sock = accept(svr_sock_, nullptr, nullptr);
if (sock == INVALID_SOCKET) {
if (svr_sock_ != INVALID_SOCKET) {
detail::close_socket(svr_sock_);
ret = false;
} else {
; // The server socket was closed by user.
} }
break;
auto val = detail::select_read(svr_sock_, 0, 100000);
if (val == 0) { // Timeout
continue;
}
socket_t sock = accept(svr_sock_, nullptr, nullptr);
if (sock == INVALID_SOCKET) {
if (svr_sock_ != INVALID_SOCKET) {
detail::close_socket(svr_sock_);
ret = false;
} else {
; // The server socket was closed by user.
}
break;
}
task_queue->enque([=]() { read_and_close_socket(sock); });
} }
task_queue_.enque([=]() { read_and_close_socket(sock); }); task_queue->shutdown();
} }
task_queue_.shutdown();
is_running_ = false; is_running_ = false;
return ret; return ret;
} }