sascha/query
1
0
Fork 0
Code Issues 4 Pull Requests Actions Packages Projects Releases Wiki Activity

started reactor refactoring

Browse Source
This commit is contained in:
Sascha Kühl
2025-07-18 15:33:04 +02:00
parent 3781bd0b66
commit f32594d9fd
11 changed files with 544 additions and 434 deletions
Download Patch File Download Diff File Expand all files Collapse all files
source/core/net/reactor.cpp
+222 -360
View File
@@ -1,342 +1,248 @@
#include "matador/net/reactor.hpp"
#include "matador/net/handler.hpp"
#include "matador/logger/log_manager.hpp"
#include <algorithm>
#include <limits>
#include <cerrno>
#include <ctime>
#include <iostream>
#include <system_error>
namespace matador {
namespace matador::net {
reactor::reactor()
: sentinel_(std::shared_ptr<handler>(nullptr))
, log_(create_logger("Reactor"))
, thread_pool_(4, [this]() { handle_events(); })
{
}
reactor::~reactor()
{
log_.debug("destroying reactor");
thread_pool_.shutdown();
: log_(logger::create_logger("Reactor"))
, thread_pool_(4, [this]() { handle_events(); }) {
}
void reactor::register_handler(const handler_ptr& h, event_type et)
{
h->register_reactor(this);
h->open();
std::lock_guard<std::mutex> l(mutex_);
auto it = find_handler_type(h);
if (it == handlers_.end()) {
handlers_.emplace_back(h, et);
} else if (it->first != h) {
throw std::logic_error("given handler isn't expected handler");
} else {
it->second = it->second | et;
}
interrupt_without_lock();
}
void reactor::unregister_handler(const handler_ptr& h, event_type)
{
std::lock_guard<std::mutex> l(mutex_);
auto it = find_handler_type(h);
if (it != handlers_.end()) {
(*it).first->close();
handlers_.erase(it);
}
interrupt_without_lock();
}
void reactor::schedule_timer(const std::shared_ptr<handler>& h, time_t offset, time_t interval)
{
h->register_reactor(this);
std::lock_guard<std::mutex> l(mutex_);
auto it = find_handler_type(h);
if (it == handlers_.end()) {
handlers_.emplace_back(h, event_type::TIMEOUT_MASK);
}
h->schedule(offset, interval);
}
void reactor::cancel_timer(const std::shared_ptr<handler>& h)
{
std::lock_guard<std::mutex> l(mutex_);
auto it = find_handler_type(h);
if (it != handlers_.end()) {
handlers_.erase(it);
}
h->cancel_timer();
}
void reactor::run()
{
// log_.info("start dispatching all clients");
thread_pool_.start();
{
// log_.info("waiting for reactor shutdown");
std::unique_lock<std::mutex> l(mutex_);
shutdown_.wait(l, [this]() { return shutdown_requested_.load(); });
cleanup();
}
// log_.info("all clients dispatched; shutting down");
thread_pool_.stop();
}
void reactor::handle_events()
{
// std::cout << this << " start handle events\n" << std::flush;
// log_.info("handle events");
running_ = true;
time_t timeout;
select_fdsets fd_sets;
prepare_select_bits(timeout, fd_sets);
// std::cout << this << " fd sets r: " << fd_sets.read_set().count() << ", w: " << fd_sets.write_set().count() << ", e: " << fd_sets.except_set().count() << ", max: " << fd_sets.maxp1() << "\n" << std::flush;
// log_.debug("fds [r: %d, w: %d, e: %d]",
// fdsets_.read_set().count(),
// fdsets_.write_set().count(),
// fdsets_.except_set().count());
// if (timeout != (std::numeric_limits<time_t>::max)()) {
// log_.debug("next timeout in %d sec", timeout);
// }
struct timeval tselect{};
struct timeval* p = nullptr;
if (timeout < (std::numeric_limits<time_t>::max)()) {
tselect.tv_sec = timeout;
tselect.tv_usec = 0;
p = &tselect;
// std::cout << this << " next timeout in " << p->tv_sec << " seconds\n" << std::flush;
}
if (!has_clients_to_handle(timeout, fd_sets)) {
// std::cout << this << " no clients to handle; returning\n" << std::flush;
// log_.info("no clients to handle, exiting");
return;
}
int ret;
while ((ret = select(p, fd_sets)) < 0) {
// std::cout << this << " select returned with error " << ret << "\n" << std::flush;
if(errno != EINTR) {
char error_buffer[1024];
log_.warn("select failed: %s", os::strerror(errno, error_buffer, 1024));
shutdown();
} else {
return;
reactor::~reactor() {
if (is_running()) {
try {
shutdown();
} catch (const std::exception& e) {
log_.error("Error during shutdown: %s", e.what());
}
}
}
// std::cout << this << " select returned with active requests " << ret << "\n" << std::flush;
bool interrupted = is_interrupted(fd_sets);
if (interrupted) {
if (!shutdown_requested_) {
// std::cout << this << " reactor was interrupted - promote new leader\n" << std::flush;
// log_.info("reactor was interrupted");
thread_pool_.promote_new_leader();
return;
} else {
// std::cout << this << " reactor was interrupted for shutdown\n" << std::flush;
// log_.info("shutting down");
// cleanup();
shutdown_.notify_one();
return;
}
}
time_t now = ::time(nullptr);
t_handler_type handler_type = resolve_next_handler(now, fd_sets);
if (handler_type.first) {
deactivate_handler(handler_type.first, handler_type.second);
// std::cout << this << " handling client " << handler_type.first->name() << " - promoting new leader\n" << std::flush;
thread_pool_.promote_new_leader();
// log_.info("start handling event");
// handle event
if (handler_type.second == event_type::WRITE_MASK) {
on_write_mask(handler_type.first);
} else if (handler_type.second == event_type::READ_MASK) {
on_read_mask(handler_type.first);
} else if (handler_type.second == event_type::TIMEOUT_MASK) {
on_timeout(handler_type.first, now);
} else {
// log_.info("unknown event type");
}
activate_handler(handler_type.first, handler_type.second);
remove_deleted();
} else {
// no handler found
// log_.info("no handler found");
thread_pool_.promote_new_leader();
}
}
void reactor::shutdown()
{
if (!is_running()) {
return;
}
// shutdown the reactor properly
log_.info("shutting down reactor");
shutdown_requested_ = true;
interrupt();
void reactor::register_handler(const handler_ptr& h, event_type type) {
if (!h) {
throw std::invalid_argument("Null handler");
}
std::unique_lock lock(handlers_mutex_);
auto* raw_ptr = h.get();
if (handlers_.find(raw_ptr) != handlers_.end()) {
throw std::runtime_error("Handler already registered");
}
safe_handler_operation(h, "register", [&] {
h->register_reactor(this);
h->open();
handlers_.emplace(raw_ptr, std::make_unique<HandlerEntry>(h, type));
});
interrupter_.interrupt();
}
bool reactor::is_running() const
{
return running_;
void reactor::unregister_handler(const handler_ptr& h, event_type type) {
if (!h) return;
std::unique_lock lock(handlers_mutex_);
if (find_handler_entry(h)) {
safe_handler_operation(h, "unregister", [&] {
h->close();
remove_handler_entry(h);
});
}
interrupter_.interrupt();
}
void reactor::schedule_timer(const handler_ptr& h, time_t offset, time_t interval) {
if (!h) return;
std::unique_lock lock(handlers_mutex_);
auto* entry = find_handler_entry(h);
if (!entry) {
entry = handlers_.emplace(h.get(),
std::make_unique<HandlerEntry>(h, event_type::TIMEOUT_MASK)).first->second.get();
}
std::lock_guard<std::mutex> timer_lock(timers_mutex_);
time_t now = std::time(nullptr);
entry->next_timeout = now + offset;
entry->interval = interval;
timers_.push(TimerEvent{entry->next_timeout, h});
interrupter_.interrupt();
}
void reactor::cancel_timer(const handler_ptr& h) {
if (!h) return;
std::unique_lock lock(handlers_mutex_);
if (auto* entry = find_handler_entry(h)) {
entry->next_timeout = 0;
entry->interval = 0;
}
}
void reactor::run() {
running_ = true;
thread_pool_.start();
{
std::unique_lock lock(handlers_mutex_);
shutdown_cv_.wait(lock, [this] { return shutdown_requested_.load(); });
cleanup_deleted_handlers();
}
thread_pool_.stop();
running_ = false;
}
void reactor::handle_events() {
time_t timeout;
select_fdsets fd_sets;
prepare_select_bits(timeout, fd_sets);
if (!fd_sets.maxp1() && timeout == (std::numeric_limits<time_t>::max)()) {
return;
}
struct timeval select_timeout{};
struct timeval* timeout_ptr = nullptr;
if (timeout < (std::numeric_limits<time_t>::max)()) {
select_timeout.tv_sec = timeout;
select_timeout.tv_usec = 0;
timeout_ptr = &select_timeout;
}
try {
if (int result = perform_select(timeout_ptr, fd_sets); result < 0) {
if (errno != EINTR) {
throw std::system_error(errno, std::system_category(), "select failed");
}
return;
}
time_t now = std::time(nullptr);
if (check_interruption(fd_sets)) {
return;
}
if (process_events(fd_sets, now)) {
thread_pool_.promote_new_leader();
}
process_timers(now);
if (now - last_cleanup_ >= CLEANUP_INTERVAL.count()) {
cleanup_deleted_handlers();
last_cleanup_ = now;
}
} catch (const std::exception& e) {
log_.error("Error in handle_events: %s", e.what());
stats_.errors_++;
}
}
void reactor::shutdown() {
if (!running_) return;
log_.info("Initiating reactor shutdown");
shutdown_requested_ = true;
interrupter_.interrupt();
shutdown_cv_.notify_all();
}
void reactor::mark_handler_for_delete(const handler_ptr& h) {
if (!h) return;
std::unique_lock lock(handlers_mutex_);
if (auto* entry = find_handler_entry(h)) {
entry->marked_for_deletion = true;
}
}
void reactor::safe_handler_operation(const handler_ptr& h, const std::string& op_name, const std::function<void()>& operation) {
try {
operation();
} catch (const std::exception& e) {
log_.error("Error during %s operation: %s", op_name.c_str(), e.what());
++stats_.errors_;
throw;
}
}
reactor::HandlerEntry* reactor::find_handler_entry(const handler_ptr& h) {
const auto it = handlers_.find(h.get());
return it != handlers_.end() ? it->second.get() : nullptr;
}
void reactor::remove_handler_entry(const handler_ptr& h) {
handlers_.erase(h.get());
}
void reactor::update_handler_events(HandlerEntry* entry, const event_type ev, const bool activate) {
if (!entry) return;
if (activate) {
entry->events |= ev;
} else {
entry->events &= ~ev;
}
}
void reactor::prepare_select_bits(time_t& timeout, select_fdsets& fd_sets) const
{
std::lock_guard<std::mutex> l(mutex_);
fd_sets.reset();
time_t now = ::time(nullptr);
timeout = (std::numeric_limits<time_t>::max)();
std::lock_guard l(handlers_mutex_);
fd_sets.reset();
const time_t now = ::time(nullptr);
timeout = (std::numeric_limits<time_t>::max)();
// set interrupter fd
fd_sets.read_set().set(interrupter_.socket_id());
// set interrupter fd
fd_sets.read_set().set(interrupter_.socket_id());
for (const auto &h : handlers_) {
if (h.first == nullptr) {
continue;
for (const auto & [hndlr, entry] : handlers_) {
if (hndlr == nullptr) {
continue;
}
if (hndlr->is_ready_read() && is_event_type_set(entry->events, event_type::READ_MASK)) {
fd_sets.read_set().set(hndlr->handle());
}
if (hndlr->is_ready_write() && is_event_type_set(entry->events, event_type::WRITE_MASK)) {
fd_sets.write_set().set(hndlr->handle());
}
if (hndlr->next_timeout() > 0 && is_event_type_set(entry->events, event_type::TIMEOUT_MASK)) {
timeout = (std::min)(timeout, hndlr->next_timeout() <= now ? 0 : (hndlr->next_timeout() - now));
}
}
if (h.first->is_ready_read() && is_event_type_set(h.second, event_type::READ_MASK)) {
fd_sets.read_set().set(h.first->handle());
}
if (h.first->is_ready_write() && is_event_type_set(h.second, event_type::WRITE_MASK)) {
fd_sets.write_set().set(h.first->handle());
}
if (h.first->next_timeout() > 0 && is_event_type_set(h.second, event_type::TIMEOUT_MASK)) {
timeout = (std::min)(timeout, h.first->next_timeout() <= now ? 0 : (h.first->next_timeout() - now));
}
}
}
void reactor::remove_deleted()
{
while (!handlers_to_delete_.empty()) {
auto h = handlers_to_delete_.front();
handlers_to_delete_.pop_front();
auto fi = std::find_if(handlers_.begin(), handlers_.end(), [&h](const t_handler_type &ht) {
return ht.first.get() == h.get();
});
if (fi != handlers_.end()) {
log_.debug("removing handler %d", fi->first->handle());
handlers_.erase(fi);
}
}
}
void reactor::cleanup()
{
while (!handlers_.empty()) {
auto hndlr = handlers_.front();
handlers_.pop_front();
hndlr.first->close();
}
}
int reactor::select(struct timeval *timeout, select_fdsets& fd_sets)
{
log_.debug("calling select; waiting for io events");
return ::select(
static_cast<int>(fd_sets.maxp1()) + 1,
fd_sets.read_set().get(),
fd_sets.write_set().get(),
fd_sets.except_set().get(),
timeout
);
}
//void reactor::process_handler(int /*ret*/)
//{
// handlers_.emplace_back(sentinel_, event_type::NONE_MASK);
// time_t now = ::time(nullptr);
// while (handlers_.front().first != nullptr) {
// auto h = handlers_.front();
// handlers_.pop_front();
// handlers_.push_back(h);
// // check for read/accept
// if (h.first->handle() > 0 && fdsets_.write_set().is_set(h.first->handle())) {
// on_write_mask(h.first);
// }
// if (h.first->handle() > 0 && fdsets_.read_set().is_set(h.first->handle())) {
// on_read_mask(h.first);
// }
// if (h.first->next_timeout() > 0 && h.first->next_timeout() <= now) {
// on_timeout(h.first, now);
// }
// }
// handlers_.pop_front();
//}
reactor::t_handler_type reactor::resolve_next_handler(time_t now, select_fdsets& fd_sets)
{
std::lock_guard<std::mutex> l(mutex_);
for (auto &h : handlers_) {
if (h.first->handle() > 0 && fd_sets.write_set().is_set(h.first->handle())) {
return std::make_pair(h.first, event_type::WRITE_MASK);
std::lock_guard l(handlers_mutex_);
for (auto & [hndlr, entry] : handlers_) {
if (hndlr->handle() > 0 && fd_sets.write_set().is_set(hndlr->handle())) {
return std::make_pair(hndlr, event_type::WRITE_MASK);
}
if (h.first->handle() > 0 && fd_sets.read_set().is_set(h.first->handle())) {
return std::make_pair(h.first, event_type::READ_MASK);
if (hndlr->handle() > 0 && fd_sets.read_set().is_set(hndlr->handle())) {
return std::make_pair(hndlr, event_type::READ_MASK);
}
if (h.first->next_timeout() > 0 && h.first->next_timeout() <= now) {
return std::make_pair(h.first, event_type::TIMEOUT_MASK);
if (hndlr->next_timeout() > 0 && hndlr->next_timeout() <= now) {
return std::make_pair(hndlr, event_type::TIMEOUT_MASK);
}
}
return std::make_pair(nullptr, event_type::NONE_MASK);
}
void reactor::on_read_mask(const handler_ptr& handler)
{
// log_.debug("read bit for handler %d is set; handle input", h->handle());
// std::cout << this << " (handler " << h.get() << "): handle read\n" << std::flush;
handler->on_input();
}
void reactor::on_write_mask(const handler_ptr& handler)
{
// log_.debug("write bit for handler %d is set; handle output", h->handle());
// std::cout << this << " (handler " << h.get() << "): handle write\n" << std::flush;
handler->on_output();
}
void reactor::on_except_mask(const handler_ptr& /*handler*/)
{
// std::cout << this << " (handler " << h.get() << "): handle exception\n" << std::flush;
}
void reactor::on_timeout(const handler_ptr &h, time_t now)
{
// log_.debug("timeout expired for handler %d; handle timeout", h->handle());
// std::cout << this << " (handler " << h.get() << "): handle timeout\n" << std::flush;
h->calculate_next_timeout(now);
h->on_timeout();
}
select_fdsets reactor::fdsets() const
select_fdsets reactor::fd_sets() const
{
time_t timeout;
select_fdsets fd_sets;
@@ -344,68 +250,24 @@ select_fdsets reactor::fdsets() const
return fd_sets;
}
void reactor::mark_handler_for_delete(const handler_ptr& h)
{
std::lock_guard<std::mutex> l(mutex_);
handlers_to_delete_.push_back(h);
}
bool reactor::is_interrupted(select_fdsets& fd_sets)
{
std::lock_guard<std::mutex> l(mutex_);
if (fd_sets.read_set().is_set(interrupter_.socket_id())) {
log_.debug("interrupt byte received; resetting interrupter");
if (shutdown_requested_) {
running_ = false;
}
return interrupter_.reset();
}
return false;
}
bool reactor::has_clients_to_handle(time_t timeout, select_fdsets& fd_sets) const {
std::lock_guard<std::mutex> lock(mutex_);
return fd_sets.maxp1() > 0 || timeout != (std::numeric_limits<time_t>::max)();
}
std::list<reactor::t_handler_type>::iterator reactor::find_handler_type(const reactor::handler_ptr &h)
{
return std::find_if(handlers_.begin(), handlers_.end(), [&h](const t_handler_type &ht) {
return ht.first.get() == h.get();
});
}
void reactor::activate_handler(const reactor::handler_ptr &h, event_type ev)
{
std::lock_guard<std::mutex> l(mutex_);
auto it = find_handler_type(h);
if (it == handlers_.end()) {
std::lock_guard l(handlers_mutex_);
const auto it = find_handler_entry(h);
if (!it) {
return;
}
it->second |= ev;
it->events |= ev;
}
void reactor::deactivate_handler(const reactor::handler_ptr &h, event_type ev)
{
std::lock_guard<std::mutex> l(mutex_);
auto it = find_handler_type(h);
if (it == handlers_.end()) {
std::lock_guard l(handlers_mutex_);
const auto it = find_handler_entry(h);
if (!it) {
return;
}
it->second &= ~ev;
it->events &= ~ev;
}
void reactor::interrupt()
{
log_.trace("interrupting reactor");
std::lock_guard<std::mutex> l(mutex_);
interrupter_.interrupt();
}
void reactor::interrupt_without_lock()
{
log_.trace("interrupting reactor");
interrupter_.interrupt();
}
}
} // namespace matador