refactor system stop

This commit is contained in:
2025-01-21 15:19:33 -06:00
parent 1b11e500f3
commit 21d02534e5

View File

@ -22,6 +22,7 @@
#ifndef REPERTORY_INCLUDE_EVENTS_T_EVENT_SYSTEM_HPP_ #ifndef REPERTORY_INCLUDE_EVENTS_T_EVENT_SYSTEM_HPP_
#define REPERTORY_INCLUDE_EVENTS_T_EVENT_SYSTEM_HPP_ #define REPERTORY_INCLUDE_EVENTS_T_EVENT_SYSTEM_HPP_
#include "app_config.hpp"
#include "events/event.hpp" #include "events/event.hpp"
#include "utils/collection.hpp" #include "utils/collection.hpp"
#include "utils/utils.hpp" #include "utils/utils.hpp"
@ -83,27 +84,37 @@ private:
std::mutex event_mutex_; std::mutex event_mutex_;
std::unique_ptr<std::thread> event_thread_; std::unique_ptr<std::thread> event_thread_;
std::mutex run_mutex_; std::mutex run_mutex_;
stop_type stop_requested_ = false; stop_type stop_requested_{false};
private: private:
[[nodiscard]] auto get_stop_requested() const -> bool {
return stop_requested_ || app_config::get_stop_requested();
}
void process_events() { void process_events() {
std::vector<std::shared_ptr<event_type>> events; std::vector<std::shared_ptr<event_type>> events;
{
unique_mutex_lock lock(event_mutex_); unique_mutex_lock lock(event_mutex_);
if (not stop_requested_ && event_list_.empty()) { while (not get_stop_requested() && event_list_.empty()) {
event_notify_.wait_for(lock, 1s); event_notify_.wait_for(lock, 4s);
} }
if (not event_list_.empty()) { if (not event_list_.empty()) {
events.insert(events.end(), event_list_.begin(), event_list_.end()); events.insert(events.end(), event_list_.begin(), event_list_.end());
event_list_.clear(); event_list_.clear();
} }
event_notify_.notify_all();
lock.unlock();
if (events.empty()) {
return;
} }
const auto notify_events = [this](const std::string &name, const auto notify_events = [this](const std::string &name,
const event_type &event) { const event_type &event) {
std::deque<std::future<void>> futures; std::deque<std::future<void>> futures;
recur_mutex_lock lock(consumer_mutex_); recur_mutex_lock consumer_lock(consumer_mutex_);
if (event_consumers_.find(name) != event_consumers_.end()) { if (event_consumers_.find(name) != event_consumers_.end()) {
for (auto *consumer : event_consumers_[name]) { for (auto *consumer : event_consumers_[name]) {
if (event.get_allow_async()) { if (event.get_allow_async()) {
@ -127,6 +138,9 @@ private:
notify_events("", *evt.get()); notify_events("", *evt.get());
notify_events(evt->get_name(), *evt.get()); notify_events(evt->get_name(), *evt.get());
} }
lock.lock();
event_notify_.notify_all();
} }
void queue_event(std::shared_ptr<event_type> evt) { void queue_event(std::shared_ptr<event_type> evt) {
@ -165,28 +179,38 @@ public:
void start() { void start() {
mutex_lock lock(run_mutex_); mutex_lock lock(run_mutex_);
if (not event_thread_) { if (event_thread_) {
event_notify_.notify_all();
return;
}
stop_requested_ = false; stop_requested_ = false;
event_thread_ = std::make_unique<std::thread>([this]() { event_thread_ = std::make_unique<std::thread>([this]() {
while (not stop_requested_) { while (not get_stop_requested()) {
process_events(); process_events();
} }
}); });
} event_notify_.notify_all();
} }
void stop() { void stop() {
mutex_lock lock(run_mutex_); unique_mutex_lock lock(run_mutex_);
if (event_thread_) { if (not event_thread_) {
stop_requested_ = true;
event_notify_.notify_all(); event_notify_.notify_all();
return;
}
stop_requested_ = true;
event_notify_.notify_all();
lock.unlock();
event_thread_->join(); event_thread_->join();
event_thread_.reset(); event_thread_.reset();
process_events(); process_events();
} }
}
}; };
} // namespace repertory } // namespace repertory