Commit 0ba4153a authored by Dominik Charousset's avatar Dominik Charousset Committed by Dominik Charousset

Switch logger to the new DRR API

parent d35be7ff
...@@ -37,9 +37,9 @@ ...@@ -37,9 +37,9 @@
#include "caf/abstract_actor.hpp" #include "caf/abstract_actor.hpp"
#include "caf/deep_to_string.hpp" #include "caf/deep_to_string.hpp"
#include "caf/intrusive/drr_queue.hpp"
#include "caf/intrusive/fifo_inbox.hpp" #include "caf/intrusive/fifo_inbox.hpp"
#include "caf/intrusive/singly_linked.hpp" #include "caf/intrusive/singly_linked.hpp"
#include "caf/intrusive/task_queue.hpp"
#include "caf/detail/arg_wrapper.hpp" #include "caf/detail/arg_wrapper.hpp"
#include "caf/detail/pretty_type_name.hpp" #include "caf/detail/pretty_type_name.hpp"
...@@ -114,7 +114,7 @@ public: ...@@ -114,7 +114,7 @@ public:
using unique_pointer = std::unique_ptr<event, deleter_type>; using unique_pointer = std::unique_ptr<event, deleter_type>;
using queue_type = intrusive::task_queue<policy>; using queue_type = intrusive::drr_queue<policy>;
using deficit_type = long; using deficit_type = long;
......
...@@ -36,6 +36,8 @@ ...@@ -36,6 +36,8 @@
#include "caf/string_algorithms.hpp" #include "caf/string_algorithms.hpp"
#include "caf/actor_system_config.hpp" #include "caf/actor_system_config.hpp"
#include "caf/intrusive/task_result.hpp"
#include "caf/detail/get_process_id.hpp" #include "caf/detail/get_process_id.hpp"
#include "caf/detail/pretty_type_name.hpp" #include "caf/detail/pretty_type_name.hpp"
...@@ -398,18 +400,21 @@ void logger::run() { ...@@ -398,18 +400,21 @@ void logger::run() {
#if defined(CAF_LOG_LEVEL) #if defined(CAF_LOG_LEVEL)
log_first_line(); log_first_line();
// receive log entries from other threads and actors // receive log entries from other threads and actors
std::unique_ptr<event> ptr; bool stop = false;
for (;;) { auto f = [&](event& x) {
// make sure we have data to read
queue_.synchronized_await(queue_mtx_, queue_cv_);
// read & process event
ptr.reset(queue_.try_pop());
CAF_ASSERT(ptr != nullptr);
// empty message means: shut down // empty message means: shut down
if (ptr->message.empty()) if (x.message.empty()) {
break; stop = true;
handle_event(*ptr); return intrusive::task_result::stop;
} }
handle_event(x);
return intrusive::task_result::resume;
};
do {
// make sure we have data to read and consume events
queue_.synchronized_await(queue_mtx_, queue_cv_);
queue_.new_round(1000, f);
} while (!stop);
log_last_line(); log_last_line();
#endif #endif
} }
...@@ -534,7 +539,7 @@ void logger::stop() { ...@@ -534,7 +539,7 @@ void logger::stop() {
if (!thread_.joinable()) if (!thread_.joinable())
return; return;
// an empty string means: shut down // an empty string means: shut down
queue_.synchronized_enqueue(queue_mtx_, queue_cv_, new event); queue_.synchronized_push_back(queue_mtx_, queue_cv_, new event);
thread_.join(); thread_.join();
#endif #endif
} }
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment