async using lockfree queue and bug fixes regarding usage of cppformat

include/spdlog/async_logger.h

@@ -37,9 +37,9 @@
 namespace spdlog
 {
 
-namespace sinks
+namespace details
 {
-class async_sink;
+class async_log_helper;
 }
 
 class async_logger :public logger
@@ -59,7 +59,7 @@ protected:
 
 private:
     log_clock::duration _shutdown_duration;
-    std::unique_ptr<sinks::async_sink> _as;
+    std::unique_ptr<details::async_log_helper> _async_log_helper;
 };
 }
 

include/spdlog/details/async_log_helper.h

@@ -22,7 +22,7 @@
 /* SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.                */
 /*************************************************************************/
 
-// async sink:
+// async log helper :
 // Process logs asynchronously using a back thread.
 //
 // If the internal queue of log messages reaches its max size,
@@ -30,36 +30,69 @@
 //
 // If the back thread throws during logging, a spdlog::spdlog_ex exception
 // will be thrown in client's thread when tries to log the next message
+
 #pragma once
 
 #include <thread>
 #include <chrono>
 #include <atomic>
 
-#include "./base_sink.h"
+#include "../sinks/sink.h"
 #include "../logger.h"
-#include "../details/blocking_queue.h"
+#include "../details/mpcs_q.h"
-#include "../details/null_mutex.h"
 #include "../details/log_msg.h"
 #include "../details/format.h"
 
 
 namespace spdlog
 {
-namespace sinks
+namespace details
 {
 
-
+class async_log_helper
-class async_sink : public base_sink < details::null_mutex > //single worker thread so null_mutex
 {
+    struct async_msg
+    {
+        std::string logger_name;
+        level::level_enum level;
+        log_clock::time_point time;
+        std::tm tm_time;
+        std::string raw_msg_str;
+
+        async_msg() = default;
+
+        async_msg(const details::log_msg& m) :
+            logger_name(m.logger_name),
+            level(m.level),
+            time(m.time),
+            tm_time(m.tm_time),
+            raw_msg_str(m.raw.data(), m.raw.size())
+        {
+
+        }
+
+
+        log_msg to_log_msg()
+        {
+            log_msg msg;
+            msg.logger_name = logger_name;
+            msg.level = level;
+            msg.time = time;
+            msg.tm_time = tm_time;
+            msg.raw << raw_msg_str;
+            return msg;
+        }
+    };
+
 public:
 
-    using q_type = details::blocking_queue < details::log_msg > ;
+    using q_type = details::mpsc_q < std::unique_ptr<async_msg> >;
 
-    explicit async_sink(const q_type::size_type max_queue_size);
+    explicit async_log_helper(size_t max_queue_size);
+    void log(const details::log_msg& msg);
 
     //Stop logging and join the back thread
-    ~async_sink();
+    ~async_log_helper();
     void add_sink(sink_ptr sink);
     void remove_sink(sink_ptr sink_ptr);
     void set_formatter(formatter_ptr);
@@ -68,25 +101,28 @@ public:
 
 
 
-protected:
+
-    void _sink_it(const details::log_msg& msg) override;
-    void _thread_loop();
 
 private:
-    std::vector<std::shared_ptr<sink>> _sinks;
+    std::vector<std::shared_ptr<sinks::sink>> _sinks;
     std::atomic<bool> _active;
     q_type _q;
-    std::thread _back_thread;
+    std::thread _worker_thread;
     std::mutex _mutex;
-    //Last exception thrown from the back thread
-    std::shared_ptr<spdlog_ex> _last_backthread_ex;
 
+    // last exception thrown from the worker thread
+    std::shared_ptr<spdlog_ex> _last_workerthread_ex;
+
+    // worker thread formatter
     formatter_ptr _formatter;
 
-    //will throw last back thread exception or if backthread no active
+    // will throw last back thread exception or if worker hread no active
     void _push_sentry();
 
-    //Clear all remaining messages(if any), stop the _back_thread and join it
+    // worker thread loop
+    void _thread_loop();
+
+    // clear all remaining messages(if any), stop the _worker_thread and join it
     void _join();
 
 };
@@ -96,85 +132,87 @@ private:
 ///////////////////////////////////////////////////////////////////////////////
 // async_sink class implementation
 ///////////////////////////////////////////////////////////////////////////////
-inline spdlog::sinks::async_sink::async_sink(const q_type::size_type max_queue_size)
+inline spdlog::details::async_log_helper::async_log_helper(size_t max_queue_size)
     :_sinks(),
      _active(true),
      _q(max_queue_size),
-     _back_thread(&async_sink::_thread_loop, this)
+     _worker_thread(&async_log_helper::_thread_loop, this)
 {}
 
-inline spdlog::sinks::async_sink::~async_sink()
+inline spdlog::details::async_log_helper::~async_log_helper()
 {
     _join();
 }
 
-inline void spdlog::sinks::async_sink::_sink_it(const details::log_msg& msg)
+inline void spdlog::details::async_log_helper::log(const details::log_msg& msg)
 {
     _push_sentry();
-    _q.push(std::move(msg));
 
+    _q.push(std::unique_ptr<async_msg>(new async_msg(msg)));
 }
 
-inline void spdlog::sinks::async_sink::_thread_loop()
+inline void spdlog::details::async_log_helper::_thread_loop()
 {
-    std::chrono::seconds  pop_timeout { 1 };
 
     while (_active)
     {
-        q_type::item_type msg;
+        q_type::item_type async_msg;
-        if (_q.pop(msg, pop_timeout))
-        {
-            if (!_active)
-                return;
 
+        if (_q.pop(async_msg))
+        {
             try
             {
 
-                _formatter->format(msg);
+                details::log_msg log_msg = async_msg->to_log_msg();
+
+                _formatter->format(log_msg);
                 for (auto &s : _sinks)
-                    s->log(msg);
+                    s->log(log_msg);
 
             }
             catch (const std::exception& ex)
             {
-                _last_backthread_ex = std::make_shared<spdlog_ex>(ex.what());
+                _last_workerthread_ex = std::make_shared<spdlog_ex>(ex.what());
             }
             catch (...)
             {
-                _last_backthread_ex = std::make_shared<spdlog_ex>("Unknown exception");
+                _last_workerthread_ex = std::make_shared<spdlog_ex>("Unknown exception");
             }
-
+        }
+        else //Sleep and retry if empty
+        {
+            std::this_thread::sleep_for(std::chrono::microseconds(100));
         }
     }
 }
 
-inline void spdlog::sinks::async_sink::add_sink(spdlog::sink_ptr s)
+inline void spdlog::details::async_log_helper::add_sink(spdlog::sink_ptr s)
 {
     std::lock_guard<std::mutex> guard(_mutex);
     _sinks.push_back(s);
 }
 
 
-inline void spdlog::sinks::async_sink::remove_sink(spdlog::sink_ptr s)
+inline void spdlog::details::async_log_helper::remove_sink(spdlog::sink_ptr s)
 {
     std::lock_guard<std::mutex> guard(_mutex);
     _sinks.erase(std::remove(_sinks.begin(), _sinks.end(), s), _sinks.end());
 }
 
 
-inline void spdlog::sinks::async_sink::set_formatter(formatter_ptr msg_formatter)
+inline void spdlog::details::async_log_helper::set_formatter(formatter_ptr msg_formatter)
 {
     _formatter = msg_formatter;
 }
 
 
 
-inline void spdlog::sinks::async_sink::shutdown(const log_clock::duration& timeout)
+inline void spdlog::details::async_log_helper::shutdown(const log_clock::duration& timeout)
 {
     if (timeout > std::chrono::milliseconds::zero())
     {
         auto until = log_clock::now() + timeout;
-        while (_q.size() > 0 && log_clock::now() < until)
+        while (_q.approx_size() > 0 && log_clock::now() < until)
         {
             std::this_thread::sleep_for(std::chrono::milliseconds(5));
         }
@@ -182,13 +220,13 @@ inline void spdlog::sinks::async_sink::shutdown(const log_clock::duration& timeo
     _join();
 }
 
-#include <iostream>
+
-inline void spdlog::sinks::async_sink::_push_sentry()
+inline void spdlog::details::async_log_helper::_push_sentry()
 {
-    if (_last_backthread_ex)
+    if (_last_workerthread_ex)
     {
-        auto ex = std::move(_last_backthread_ex);
+        auto ex = std::move(_last_workerthread_ex);
-        _last_backthread_ex.reset();
+        _last_workerthread_ex.reset();
         throw *ex;
     }
     if (!_active)
@@ -196,17 +234,18 @@ inline void spdlog::sinks::async_sink::_push_sentry()
 }
 
 
-inline void spdlog::sinks::async_sink::_join()
+inline void spdlog::details::async_log_helper::_join()
 {
     _active = false;
-    if (_back_thread.joinable())
+    if (_worker_thread.joinable())
     {
         try
         {
-            _back_thread.join();
+            _worker_thread.join();
         }
         catch (const std::system_error&) //Dont crash if thread not joinable
-        {}
+        {
+        }
     }
 
 }

include/spdlog/details/async_logger_impl.h

@@ -25,8 +25,7 @@
 #pragma once
 
 
-#include <memory>
+#include "./async_log_helper.h"
-#include "../sinks/async_sink.h"
 
 //
 // Async Logger implementation
@@ -38,11 +37,11 @@ template<class It>
 inline spdlog::async_logger::async_logger(const std::string& logger_name, const It& begin, const It& end, size_t queue_size, const log_clock::duration& shutdown_duration) :
     logger(logger_name, begin, end),
     _shutdown_duration(shutdown_duration),
-    _as(std::unique_ptr<sinks::async_sink>(new sinks::async_sink(queue_size)))
+    _async_log_helper(new details::async_log_helper(queue_size))
 {
-    _as->set_formatter(_formatter);
+    _async_log_helper->set_formatter(_formatter);
     for (auto &s : _sinks)
-        _as->add_sink(s);
+        _async_log_helper->add_sink(s);
 }
 
 inline spdlog::async_logger::async_logger(const std::string& logger_name, sinks_init_list sinks, size_t queue_size, const log_clock::duration& shutdown_duration) :
@@ -52,21 +51,16 @@ inline spdlog::async_logger::async_logger(const std::string& logger_name, sink_p
     async_logger(logger_name, { single_sink }, queue_size, shutdown_duration) {}
 
 
-inline void spdlog::async_logger::_log_msg(details::log_msg& msg)
-{
-    _as->log(msg);
-}
-
 inline void spdlog::async_logger::_set_formatter(spdlog::formatter_ptr msg_formatter)
 {
     _formatter = msg_formatter;
-    _as->set_formatter(_formatter);
+    _async_log_helper->set_formatter(_formatter);
 }
 
 inline void spdlog::async_logger::_set_pattern(const std::string& pattern)
 {
     _formatter = std::make_shared<pattern_formatter>(pattern);
-    _as->set_formatter(_formatter);
+    _async_log_helper->set_formatter(_formatter);
 }
 
 
@@ -74,5 +68,10 @@ inline void spdlog::async_logger::_set_pattern(const std::string& pattern)
 inline void spdlog::async_logger::_stop()
 {
     set_level(level::OFF);
-    _as->shutdown(_shutdown_duration);
+    _async_log_helper->shutdown(_shutdown_duration);
+}
+
+inline void spdlog::async_logger::_log_msg(details::log_msg& msg)
+{
+    _async_log_helper->log(msg);
 }

include/spdlog/details/file_helper.h

@@ -102,7 +102,7 @@ public:
 
         size_t size = msg.formatted.size();
         auto data = msg.formatted.data();
-        if(std::fwrite(data, sizeof(char), size, _fd) != size)
+        if(std::fwrite(data, 1, size, _fd) != size)
             throw spdlog_ex("Failed writing to file " + _filename);
 
         if(_auto_flush)

include/spdlog/details/format.cc

@@ -31,7 +31,6 @@ SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 #undef _SCL_SECURE_NO_WARNINGS
 #define _SCL_SECURE_NO_WARNINGS
 
-#include "format.h"
 
 #include <string.h>
 

include/spdlog/details/log_msg.h

@@ -48,9 +48,13 @@ struct log_msg
         level(other.level),
         time(other.time),
         tm_time(other.tm_time)
+
     {
-        raw.write(other.raw.data(), other.raw.size());
+        if (other.raw.size())
-        formatted.write(other.formatted.data(), other.formatted.size());
+            raw << fmt::BasicStringRef<char>(other.raw.data(), other.raw.size());
+        if (other.formatted.size())
+            formatted << fmt::BasicStringRef<char>(other.formatted.data(), other.formatted.size());
+
     }
 
     log_msg(log_msg&& other) :

include/spdlog/details/mpcs_q.h

@@ -1,7 +1,5 @@
-#pragma once
-/*************************************************************************/
 /*
-Modified version of Intrusive MPSC node-based queue
+A modified version of Intrusive MPSC node-based queue
 
 Original code from
 http://www.1024cores.net/home/lock-free-algorithms/queues/intrusive-mpsc-node-based-queue
@@ -32,9 +30,10 @@ ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 
 The views and conclusions contained in the software and documentation are those of the authors and
 should not be interpreted as representing official policies, either expressed or implied, of Dmitry Vyukov.
+*/
 
 /*************************************************************************/
-/* The code in its current form adds the license below:                  */
+/********* The code in its current form adds the license below: **********/
 /*************************************************************************/
 /* spdlog - an extremely fast and easy to use c++11 logging library.     */
 /* Copyright (c) 2014 Gabi Melman.                                       */
@@ -59,6 +58,7 @@ should not be interpreted as representing official policies, either expressed or
 /* SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.                */
 /*************************************************************************/
 
+#pragma once
 
 #include <atomic>
 namespace spdlog
@@ -70,35 +70,40 @@ class mpsc_q
 {
 
 public:
-    mpsc_q(size_t size) :_stub(T()), _head(&_stub), _tail(&_stub)
+    using item_type = T;
+    mpsc_q(size_t max_size) :
+        _max_size(max_size),
+        _size(0),
+        _stub(),
+        _head(&_stub),
+        _tail(&_stub)
     {
-        _stub.next = nullptr;
     }
 
     ~mpsc_q()
     {
-        reset();
+        clear();
     }
 
-    void reset()
+    template<typename TT>
+    bool push(TT&& value)
     {
-        T dummy_val;
+        if (_size >= _max_size)
-        while (pop(dummy_val));
+            return false;
-    }
+        mpscq_node_t* new_node = new mpscq_node_t(std::forward<TT>(value));
-
-    bool push(const T& value)
-    {
-        mpscq_node_t* new_node = new mpscq_node_t(value);
         push_node(new_node);
+        ++_size;
         return true;
     }
 
+    // Try to pop or return false immediatly is queue is empty
     bool pop(T& value)
     {
         mpscq_node_t* node = pop_node();
         if (node != nullptr)
         {
-            value = node->value;
+            --_size;
+            value = std::move(node->value);
             delete(node);
             return true;
         }
@@ -108,23 +113,48 @@ public:
         }
     }
 
+    // Empty the queue by popping all its elements
+    void clear()
+    {
+
+        while (mpscq_node_t* node = pop_node())
+        {
+            --_size;
+            delete(node);
+
+        }
+
+    }
+
+    // Return approx size
+    size_t approx_size() const
+    {
+        return _size.load();
+    }
+
 private:
     struct mpscq_node_t
     {
         std::atomic<mpscq_node_t*>  next;
         T value;
 
-        explicit mpscq_node_t(const T& value) :next(nullptr), value(value)
+        mpscq_node_t() :next(nullptr) {}
-        {
+        explicit mpscq_node_t(const T& value):
-        }
+            next(nullptr),
+            value(value) {}
+
+        explicit mpscq_node_t(T&& value) :
+            next(nullptr),
+            value(std::move(value)) {}
     };
 
+    size_t _max_size;
+    std::atomic<size_t> _size;
     mpscq_node_t            _stub;
     std::atomic<mpscq_node_t*>  _head;
     mpscq_node_t*           _tail;
 
-
+    // Lockfree push
-
     void push_node(mpscq_node_t* n)
     {
         n->next = nullptr;
@@ -132,6 +162,8 @@ private:
         prev->next = n;
     }
 
+    // Clever lockfree pop algorithm by Dmitry Vyukov using single xchng instruction..
+    // Return pointer to the poppdc node or nullptr if no items left in the queue
     mpscq_node_t* pop_node()
     {
         mpscq_node_t* tail = _tail;
@@ -151,7 +183,7 @@ private:
         }
         mpscq_node_t* head = _head;
         if (tail != head)
-            return 0;
+            return nullptr;
 
         push_node(&_stub);
         next = tail->next;
@@ -163,8 +195,6 @@ private:
         return nullptr;
     }
 
-
-
 };
 }
 }

include/spdlog/details/pattern_formatter_impl.h

@@ -345,7 +345,7 @@ class v_formatter :public flag_formatter
 {
     void format(details::log_msg& msg) override
     {
-        msg.formatted.write(msg.raw.data(), msg.raw.size());
+        msg.formatted << fmt::BasicStringRef<char>(msg.raw.data(), msg.raw.size());
     }
 };
 
@@ -413,7 +413,7 @@ class full_formatter :public flag_formatter
                       << fmt::pad(static_cast<int>(millis), 3, '0') << "] ";
 
         msg.formatted << '[' << msg.logger_name << "] [" << level::to_str(msg.level) << "] ";
-        msg.formatted.write(msg.raw.data(), msg.raw.size());
+        msg.formatted << fmt::BasicStringRef<char>(msg.raw.data(), msg.raw.size());
     }
 };
 
@@ -581,7 +581,7 @@ inline void spdlog::pattern_formatter::format(details::log_msg& msg)
             f->format(msg);
         }
         //write eol
-        msg.formatted.write(details::os::eol(), details::os::eol_size());
+        msg.formatted << details::os::eol();
     }
     catch(const fmt::FormatError& e)
     {