Fixed flush of logs in async mode

include/spdlog/async_logger.h

@@ -76,6 +76,7 @@ class async_logger :public logger
                  const std::chrono::milliseconds& flush_interval_ms = std::chrono::milliseconds::zero());
 
 
+    void flush() override;
 protected:
     void _log_msg(details::log_msg& msg) override;
     void _set_formatter(spdlog::formatter_ptr msg_formatter) override;

include/spdlog/details/async_log_helper.h

@@ -35,15 +35,14 @@
 
 #include <chrono>
 #include <thread>
-#include <atomic>
 #include <functional>
 
 #include "../common.h"
 #include "../sinks/sink.h"
 #include "./mpmc_bounded_q.h"
 #include "./log_msg.h"
 #include "./format.h"
-#include "os.h"
+#include "./os.h"
 
 
 namespace spdlog
@@ -55,13 +54,20 @@ class async_log_helper
 {
     // Async msg to move to/from the queue
     // Movable only. should never be copied
+    enum class async_msg_type
+    {
+        log,
+        flush,
+        terminate
+    };
     struct async_msg
     {
         std::string logger_name;
         level::level_enum level;
         log_clock::time_point time;
         size_t thread_id;
         std::string txt;
+        async_msg_type msg_type;
 
         async_msg() = default;
         ~async_msg() = default;
@@ -70,16 +76,21 @@ async_msg(async_msg&& other) SPDLOG_NOEXCEPT:
         logger_name(std::move(other.logger_name)),
                     level(std::move(other.level)),
                     time(std::move(other.time)),
-                    txt(std::move(other.txt))
+                    txt(std::move(other.txt)),
+                    msg_type(std::move(other.msg_type))
         {}
 
+        async_msg(async_msg_type msg_type) :msg_type(msg_type)
+        {};
+
         async_msg& operator=(async_msg&& other) SPDLOG_NOEXCEPT
         {
             logger_name = std::move(other.logger_name);
             level = other.level;
             time = std::move(other.time);
             thread_id = other.thread_id;
             txt = std::move(other.txt);
+            msg_type = other.msg_type;
             return *this;
         }
         // never copy or assign. should only be moved..
@@ -92,10 +103,12 @@ async_msg(async_msg&& other) SPDLOG_NOEXCEPT:
             level(m.level),
             time(m.time),
             thread_id(m.thread_id),
-            txt(m.raw.data(), m.raw.size())
+            txt(m.raw.data(), m.raw.size()),
+            msg_type(async_msg_type::log)
         {}
 
 
+
         // copy into log_msg
         void fill_log_msg(log_msg &msg)
         {
@@ -130,6 +143,8 @@ async_msg(async_msg&& other) SPDLOG_NOEXCEPT:
 
     void set_formatter(formatter_ptr);
 
+    void flush();
+
 
 private:
     formatter_ptr _formatter;
@@ -138,6 +153,11 @@ async_msg(async_msg&& other) SPDLOG_NOEXCEPT:
     // queue of messages to log
     q_type _q;
 
+    bool _flush_requested;
+
+    bool _terminate_requested;
+
+
     // last exception thrown from the worker thread
     std::shared_ptr<spdlog_ex> _last_workerthread_ex;
 
@@ -153,34 +173,42 @@ async_msg(async_msg&& other) SPDLOG_NOEXCEPT:
     // worker thread
     std::thread _worker_thread;
 
+    void push_msg(async_msg& new_msg);
     // throw last worker thread exception or if worker thread is not active
+
     void throw_if_bad_worker();
 
     // worker thread main loop
     void worker_loop();
 
-    // pop next message from the queue and process it
-    // return true if a message was available (queue was not empty), will set the last_pop to the pop time
+    // pop next message from the queue and process it. will set the last_pop to the pop time
+    // return false if termination of the queue is required
     bool process_next_msg(log_clock::time_point& last_pop, log_clock::time_point& last_flush);
 
     void handle_flush_interval(log_clock::time_point& now, log_clock::time_point& last_flush);
 
     // sleep,yield or return immediatly using the time passed since last message as a hint
     static void sleep_or_yield(const spdlog::log_clock::time_point& now, const log_clock::time_point& last_op_time);
 
-
-
 };
 }
 }
 
 ///////////////////////////////////////////////////////////////////////////////
 // async_sink class implementation
 ///////////////////////////////////////////////////////////////////////////////
-inline spdlog::details::async_log_helper::async_log_helper(formatter_ptr formatter, const std::vector<sink_ptr>& sinks, size_t queue_size, const async_overflow_policy overflow_policy, const std::function<void()>& worker_warmup_cb, const std::chrono::milliseconds& flush_interval_ms):
+inline spdlog::details::async_log_helper::async_log_helper(
+    formatter_ptr formatter,
+    const std::vector<sink_ptr>& sinks,
+    size_t queue_size,
+    const async_overflow_policy overflow_policy,
+    const std::function<void()>& worker_warmup_cb,
+    const std::chrono::milliseconds& flush_interval_ms):
     _formatter(formatter),
     _sinks(sinks),
     _q(queue_size),
+    _flush_requested(false),
+    _terminate_requested(false),
     _overflow_policy(overflow_policy),
     _worker_warmup_cb(worker_warmup_cb),
     _flush_interval_ms(flush_interval_ms),
@@ -191,22 +219,29 @@ inline spdlog::details::async_log_helper::async_log_helper(formatter_ptr formatt
 // and wait for it to finish gracefully
 inline spdlog::details::async_log_helper::~async_log_helper()
 {
-
     try
     {
-        log(log_msg(level::off));
+        push_msg(async_msg(async_msg_type::terminate));
         _worker_thread.join();
     }
-    catch (...) //Dont crash if thread not joinable
+    catch (...) // don't crash in destructor
     {}
 }
 
 
 //Try to push and block until succeeded
 inline void spdlog::details::async_log_helper::log(const details::log_msg& msg)
 {
-    throw_if_bad_worker();
     async_msg new_msg(msg);
+    push_msg(new_msg);
+
+
+}
+
+//Try to push and block until succeeded
+inline void spdlog::details::async_log_helper::push_msg(details::async_log_helper::async_msg& new_msg)
+{
+    throw_if_bad_worker();
     if (!_q.enqueue(std::move(new_msg)) && _overflow_policy != async_overflow_policy::discard_log_msg)
     {
         auto last_op_time = details::os::now();
@@ -215,12 +250,16 @@ inline void spdlog::details::async_log_helper::log(const details::log_msg& msg)
         {
             now = details::os::now();
             sleep_or_yield(now, last_op_time);
-        }
-        while (!_q.enqueue(std::move(new_msg)));
+        } while (!_q.enqueue(std::move(new_msg)));
     }
 
 }
 
+inline void spdlog::details::async_log_helper::flush()
+{
+    push_msg(async_msg(async_msg_type::flush));
+}
+
 inline void spdlog::details::async_log_helper::worker_loop()
 {
     try
@@ -251,31 +290,47 @@ inline bool spdlog::details::async_log_helper::process_next_msg(log_clock::time_
     if (_q.dequeue(incoming_async_msg))
     {
         last_pop = details::os::now();
-
-        if(incoming_async_msg.level == level::off)
-            return false;
-
-        incoming_async_msg.fill_log_msg(incoming_log_msg);
-        _formatter->format(incoming_log_msg);
-        for (auto &s : _sinks)
-            s->log(incoming_log_msg);
+        switch (incoming_async_msg.msg_type)
+        {
+        case async_msg_type::flush:
+            _flush_requested = true;
+            break;
+
+        case async_msg_type::terminate:
+            _flush_requested = true;
+            _terminate_requested = true;
+            break;
+
+        default:
+            incoming_async_msg.fill_log_msg(incoming_log_msg);
+            _formatter->format(incoming_log_msg);
+            for (auto &s : _sinks)
+                s->log(incoming_log_msg);
+        }
+        return true;
     }
-    else //empty queue
+
+    // Handle empty queue..
+    // This is the only place where the queue can terminate or flush to avoid losing messages already in the queue
+    else
     {
         auto now = details::os::now();
         handle_flush_interval(now, last_flush);
         sleep_or_yield(now, last_pop);
+        return !_terminate_requested;
+
     }
-    return true;
 }
 
 inline void spdlog::details::async_log_helper::handle_flush_interval(log_clock::time_point& now, log_clock::time_point& last_flush)
 {
-    if (_flush_interval_ms != std::chrono::milliseconds::zero() && now - last_flush >= _flush_interval_ms)
+    auto should_flush = _flush_requested || (_flush_interval_ms != std::chrono::milliseconds::zero() && now - last_flush >= _flush_interval_ms);
+    if ( should_flush)
     {
         for (auto &s : _sinks)
             s->flush();
         now = last_flush = details::os::now();
+        _flush_requested = false;
     }
 }
 inline void spdlog::details::async_log_helper::set_formatter(formatter_ptr msg_formatter)

include/spdlog/details/async_logger_impl.h

@@ -60,9 +60,17 @@ inline spdlog::async_logger::async_logger(const std::string& logger_name,
         const  async_overflow_policy overflow_policy,
         const std::function<void()>& worker_warmup_cb,
         const std::chrono::milliseconds& flush_interval_ms) :
-    async_logger(logger_name, { single_sink }, queue_size, overflow_policy, worker_warmup_cb, flush_interval_ms) {}
+    async_logger(logger_name, {
+    single_sink
+}, queue_size, overflow_policy, worker_warmup_cb, flush_interval_ms) {}
 
 
+inline void spdlog::async_logger::flush()
+{
+
+    _async_log_helper->flush();
+}
+
 inline void spdlog::async_logger::_set_formatter(spdlog::formatter_ptr msg_formatter)
 {
     _formatter = msg_formatter;

include/spdlog/logger.h

@@ -107,7 +107,7 @@ class logger
     void set_pattern(const std::string&);
     void set_formatter(formatter_ptr);
 
-    void flush();
+    virtual void flush();
 
 protected:
     virtual void _log_msg(details::log_msg&);