Nicholas Orlowsky
9b5719f9be
Some checks failed
Docker Build & Publish / build (push) Has been cancelled
139 lines
4.1 KiB
C++
139 lines
4.1 KiB
C++
#include "./event_loop.hpp"
|
|
#include "../log/log.hpp"
|
|
#include "assert.h"
|
|
#include "signal.h"
|
|
#include "sys/epoll.h"
|
|
#include <chrono>
|
|
#include <mutex>
|
|
#include <pthread.h>
|
|
#include <sstream>
|
|
#include <syncstream>
|
|
#include <thread>
|
|
|
|
using std::chrono::duration;
|
|
using std::chrono::duration_cast;
|
|
using std::chrono::high_resolution_clock;
|
|
using std::chrono::milliseconds;
|
|
|
|
namespace anthracite::thread_mgr {
|
|
|
|
event_loop::event_loop(std::vector<socket::listener*>& listen_sockets, backends::backend& backend, int max_threads, int max_clients)
|
|
: thread_mgr(backend)
|
|
, _error_backend("./www")
|
|
, _max_threads(max_threads)
|
|
, _listen_sockets(listen_sockets)
|
|
, _max_clients(max_clients)
|
|
, _nonblocking(false)
|
|
{
|
|
}
|
|
|
|
bool event_loop::event_handler(socket::server* sock)
|
|
{
|
|
std::string raw_request = sock->recv_message(http::HEADER_BYTES);
|
|
|
|
if (raw_request == "") {
|
|
return false;
|
|
}
|
|
|
|
http::request req(raw_request, sock->client_ip());
|
|
std::unique_ptr<http::response> resp = req.is_supported_version() ? _backend.handle_request(req) : _error_backend.handle_error(http::status_codes::HTTP_VERSION_NOT_SUPPORTED);
|
|
std::string header = resp->header_to_string();
|
|
sock->send_message(header);
|
|
sock->send_message(resp->content());
|
|
|
|
if (req.close_connection()) {
|
|
return false;
|
|
}
|
|
|
|
return true;
|
|
}
|
|
|
|
void event_loop::worker_thread_loop(int threadno)
|
|
{
|
|
std::stringstream ss;
|
|
ss << "worker " << threadno;
|
|
pthread_setname_np(pthread_self(), ss.str().c_str());
|
|
|
|
struct epoll_event* events = new struct epoll_event[_max_clients];
|
|
int timeout_ms = 1000;
|
|
|
|
if (_nonblocking) {
|
|
timeout_ms = 0;
|
|
}
|
|
|
|
std::osyncstream(log::info) << "Starting worker thread " << threadno << std::endl;
|
|
|
|
int epoll_fd = epoll_create(1);
|
|
|
|
for (socket::listener* sl : _listen_sockets) {
|
|
struct epoll_event event;
|
|
event.events = EPOLLIN | EPOLLEXCLUSIVE;
|
|
event.data.ptr = sl;
|
|
epoll_ctl(epoll_fd, EPOLL_CTL_ADD, sl->fd(), &event);
|
|
if (threadno == 0) {
|
|
std::osyncstream(log::info) << "Listening started on port " << sl->port() << std::endl;
|
|
}
|
|
}
|
|
|
|
while (_run) {
|
|
int ready_fds = epoll_wait(epoll_fd, events, _max_clients, timeout_ms);
|
|
|
|
if (ready_fds > 0) {
|
|
for (int i = 0; i < ready_fds; i++) {
|
|
socket::socket* sockptr = reinterpret_cast<socket::socket*>(events[i].data.ptr);
|
|
socket::server* server_ptr = dynamic_cast<socket::server*>(sockptr);
|
|
|
|
if (server_ptr != nullptr) {
|
|
if (!event_handler(server_ptr)) {
|
|
delete server_ptr;
|
|
}
|
|
} else {
|
|
socket::listener* listen_ptr = dynamic_cast<socket::listener*>(sockptr);
|
|
if (listen_ptr != nullptr) {
|
|
socket::server* server_sock;
|
|
while (listen_ptr->wait_for_conn(&server_sock)) {
|
|
struct epoll_event event;
|
|
event.events = EPOLLIN | EPOLLEXCLUSIVE;
|
|
event.data.ptr = server_sock;
|
|
epoll_ctl(epoll_fd, EPOLL_CTL_ADD, server_sock->fd(), &event);
|
|
}
|
|
} else {
|
|
std::osyncstream(log::err) << "Had socket type that wasn't listener or server" << std::endl;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
delete[] events;
|
|
|
|
std::osyncstream(log::info) << "Stopping worker thread " << threadno << std::endl;
|
|
}
|
|
|
|
void event_loop::start()
|
|
{
|
|
std::lock_guard<std::mutex> lg(_run_lock);
|
|
|
|
signal(SIGPIPE, SIG_IGN);
|
|
log::info << "Starting event_loop Thread Manager" << std::endl;
|
|
|
|
_run = true;
|
|
|
|
std::vector<std::thread> worker_threads;
|
|
|
|
for (int i = 0; i < _max_threads; i++) {
|
|
auto thread = std::thread(&event_loop::worker_thread_loop, this, i);
|
|
worker_threads.push_back(std::move(thread));
|
|
}
|
|
|
|
for (std::thread& t : worker_threads) {
|
|
t.join();
|
|
}
|
|
}
|
|
|
|
void event_loop::stop()
|
|
{
|
|
_run = false;
|
|
}
|
|
}
|