From 41549365680f0aae3f82e5812f3470305e939f7d Mon Sep 17 00:00:00 2001 From: B3n30 Date: Sun, 9 Sep 2018 13:08:57 +0200 Subject: [PATCH 1/2] threadsafe_queue: Add WaitIfEmpty and use it in logging --- src/common/logging/backend.cpp | 20 +++++++------------- src/common/logging/backend.h | 1 + src/common/threadsafe_queue.h | 19 ++++++++++++++++++- 3 files changed, 26 insertions(+), 14 deletions(-) diff --git a/src/common/logging/backend.cpp b/src/common/logging/backend.cpp index a5e031189b..276e49f866 100644 --- a/src/common/logging/backend.cpp +++ b/src/common/logging/backend.cpp @@ -40,9 +40,7 @@ public: const Impl& operator=(Impl const&) = delete; void PushEntry(Entry e) { - std::lock_guard lock(message_mutex); message_queue.Push(std::move(e)); - message_cv.notify_one(); } void AddBackend(std::unique_ptr backend) { @@ -85,16 +83,13 @@ private: backend->Write(e); } }; - while (true) { - { - std::unique_lock lock(message_mutex); - message_cv.wait(lock, [&] { return !running || message_queue.Pop(entry); }); - } - if (!running) { + while (message_queue.PopWait(entry)) { + if (entry.final_entry) { break; } write_logs(entry); } + // Drain the logging queue. Only writes out up to MAX_LOGS_TO_WRITE to prevent a case // where a system is repeatedly spamming logs even on close. const int MAX_LOGS_TO_WRITE = filter.IsDebug() ? INT_MAX : 100; @@ -106,14 +101,13 @@ private: } ~Impl() { - running = false; - message_cv.notify_one(); + Entry entry; + entry.final_entry = true; + message_queue.Push(entry); backend_thread.join(); } - std::atomic_bool running{true}; - std::mutex message_mutex, writing_mutex; - std::condition_variable message_cv; + std::mutex writing_mutex; std::thread backend_thread; std::vector> backends; Common::MPSCQueue message_queue; diff --git a/src/common/logging/backend.h b/src/common/logging/backend.h index 91bb0c309c..a31ee69688 100644 --- a/src/common/logging/backend.h +++ b/src/common/logging/backend.h @@ -27,6 +27,7 @@ struct Entry { unsigned int line_num; std::string function; std::string message; + bool final_entry = false; Entry() = default; Entry(Entry&& o) = default; diff --git a/src/common/threadsafe_queue.h b/src/common/threadsafe_queue.h index f553efdc9a..4fdcecca0f 100644 --- a/src/common/threadsafe_queue.h +++ b/src/common/threadsafe_queue.h @@ -8,6 +8,7 @@ // single reader, single writer queue #include +#include #include #include #include @@ -39,12 +40,13 @@ public: template void Push(Arg&& t) { // create the element, add it to the queue - write_ptr->current = std::forward(t); + write_ptr->current = std::move(t); // set the next pointer to a new element ptr // then advance the write pointer ElementPtr* new_ptr = new ElementPtr(); write_ptr->next.store(new_ptr, std::memory_order_release); write_ptr = new_ptr; + cv.notify_one(); ++size; } @@ -67,6 +69,7 @@ public: --size; ElementPtr* tmpptr = read_ptr; + read_ptr = tmpptr->next.load(std::memory_order_acquire); t = std::move(tmpptr->current); tmpptr->next.store(nullptr); @@ -74,6 +77,14 @@ public: return true; } + bool PopWait(T& t) { + if (Empty()) { + std::unique_lock lock(cv_mutex); + cv.wait(lock, [this]() { return !Empty(); }); + } + return Pop(t); + } + // not thread-safe void Clear() { size.store(0); @@ -101,6 +112,8 @@ private: ElementPtr* write_ptr; ElementPtr* read_ptr; std::atomic_size_t size{0}; + std::mutex cv_mutex; + std::condition_variable cv; }; // a simple thread-safe, @@ -135,6 +148,10 @@ public: return spsc_queue.Pop(t); } + bool PopWait(T& t) { + return spsc_queue.PopWait(t); + } + // not thread-safe void Clear() { spsc_queue.Clear(); From 2195f10d152a52e01ccab0a6528f8758752d66a9 Mon Sep 17 00:00:00 2001 From: B3n30 Date: Mon, 8 Oct 2018 23:28:54 +0200 Subject: [PATCH 2/2] Adressed review comments --- src/common/logging/backend.cpp | 3 ++- src/common/threadsafe_queue.h | 13 +++++++------ 2 files changed, 9 insertions(+), 7 deletions(-) diff --git a/src/common/logging/backend.cpp b/src/common/logging/backend.cpp index 276e49f866..b369f199f3 100644 --- a/src/common/logging/backend.cpp +++ b/src/common/logging/backend.cpp @@ -83,7 +83,8 @@ private: backend->Write(e); } }; - while (message_queue.PopWait(entry)) { + while (true) { + entry = message_queue.PopWait(); if (entry.final_entry) { break; } diff --git a/src/common/threadsafe_queue.h b/src/common/threadsafe_queue.h index 4fdcecca0f..821e8536a4 100644 --- a/src/common/threadsafe_queue.h +++ b/src/common/threadsafe_queue.h @@ -40,7 +40,7 @@ public: template void Push(Arg&& t) { // create the element, add it to the queue - write_ptr->current = std::move(t); + write_ptr->current = std::forward(t); // set the next pointer to a new element ptr // then advance the write pointer ElementPtr* new_ptr = new ElementPtr(); @@ -69,7 +69,6 @@ public: --size; ElementPtr* tmpptr = read_ptr; - read_ptr = tmpptr->next.load(std::memory_order_acquire); t = std::move(tmpptr->current); tmpptr->next.store(nullptr); @@ -77,12 +76,14 @@ public: return true; } - bool PopWait(T& t) { + T PopWait() { if (Empty()) { std::unique_lock lock(cv_mutex); cv.wait(lock, [this]() { return !Empty(); }); } - return Pop(t); + T t; + Pop(t); + return t; } // not thread-safe @@ -148,8 +149,8 @@ public: return spsc_queue.Pop(t); } - bool PopWait(T& t) { - return spsc_queue.PopWait(t); + T PopWait() { + return spsc_queue.PopWait(); } // not thread-safe