async_sink

c11log.sln

 
 Microsoft Visual Studio Solution File, Format Version 12.00
 # Visual Studio 2013
-VisualStudioVersion = 12.0.21005.1
+VisualStudioVersion = 12.0.30110.0
 MinimumVisualStudioVersion = 10.0.40219.1
 Project("{8BC9CEB8-8B4A-11D0-8D11-00A0C91BC942}") = "c11log", "c11log.vcxproj", "{BBFA8622-1945-4EB0-BAF4-473BE753ED24}"
 EndProject
@@ -28,4 +28,7 @@ Global
 	GlobalSection(SolutionProperties) = preSolution
 		HideSolutionNode = FALSE
 	EndGlobalSection
+	GlobalSection(Performance) = preSolution
+		HasPerformanceSessions = true
+	EndGlobalSection
 EndGlobal

include/c11log/sinks/async_sink.h

@@ -2,7 +2,6 @@
 
 #include <thread>
 #include <chrono>
-#include <mutex>
 #include <atomic>
 
 #include "base_sink.h"
@@ -11,39 +10,45 @@
 
 namespace c11log {
 namespace sinks {
-class async_sink : public base_sink {
-    
+
+class async_sink : public base_sink
+{
 public:
-    using size_type = c11log::details::blocking_queue<std::string>::size_type;    
-    explicit async_sink(const std::size_t max_queue_size, const std::chrono::seconds& timeout = std::chrono::seconds::max());        
+    using size_type = c11log::details::blocking_queue<std::string>::size_type;
+
+    explicit async_sink(const size_type max_queue_size);
     ~async_sink();
     void add_sink(logger::sink_ptr_t sink);
-    void remove_sink(logger::sink_ptr_t sink_ptr);    
-    
+    void remove_sink(logger::sink_ptr_t sink_ptr);
+
+    //Wait to remaining items (if any) in the queue to be written and shutdown
+    void shutdown(const std::chrono::seconds& timeout);
 
-protected:    
+
+protected:
     void sink_it_(const std::string& msg) override;
     void thread_loop_();
-    
-private:    
+
+private:
     c11log::logger::sinks_vector_t sinks_;
-    bool active_ = true;    
-    const std::chrono::seconds timeout_;
+    std::atomic<bool> active_ { true };
+    const std::chrono::seconds push_pop_timeout_;
     c11log::details::blocking_queue<std::string> q_;
-    std::thread back_thread_; 
+    std::thread back_thread_;
+    //Clear all remaining messages(if any), stop the back_thread_ and join it
     void shutdown_();
 };
 }
 }
 
-//
-// async_sink inline impl
-//
+///////////////////////////////////////////////////////////////////////////////
+// async_sink class implementation
+///////////////////////////////////////////////////////////////////////////////
 
-inline c11log::sinks::async_sink::async_sink(const std::size_t max_queue_size, const std::chrono::seconds& timeout)
+inline c11log::sinks::async_sink::async_sink(const std::size_t max_queue_size)
     :q_(max_queue_size),
-    timeout_(timeout),
-    back_thread_(&async_sink::thread_loop_, this)
+     push_pop_timeout_(std::chrono::seconds(2)),
+     back_thread_(&async_sink::thread_loop_, this)
 {}
 
 inline c11log::sinks::async_sink::~async_sink()
@@ -52,23 +57,21 @@ inline c11log::sinks::async_sink::~async_sink()
 }
 inline void c11log::sinks::async_sink::sink_it_(const std::string& msg)
 {
-    q_.push(msg, timeout_);
+    q_.push(msg, push_pop_timeout_);
 }
 
 inline void c11log::sinks::async_sink::thread_loop_()
-{   
+{
     std::string msg;
-    while (active_) 
-    {        
-        if (q_.pop(msg, timeout_))
+    while (active_)
+    {
+        if (q_.pop(msg, push_pop_timeout_))
         {
-            std::lock_guard<std::mutex> lock(mutex_);
             for (auto &sink : sinks_)
             {
-                if (active_)
-                    sink->log(msg, _level);
-                else
-                    break;
+                sink->log(msg, _level);
+                if (!active_)
+                    return;
             }
         }
     }
@@ -76,23 +79,32 @@ inline void c11log::sinks::async_sink::thread_loop_()
 
 inline void c11log::sinks::async_sink::add_sink(logger::sink_ptr_t sink)
 {
-    std::lock_guard<std::mutex> lock(mutex_);
     sinks_.push_back(sink);
 }
 
 inline void c11log::sinks::async_sink::remove_sink(logger::sink_ptr_t sink_ptr)
 {
-    std::lock_guard<std::mutex> lock(mutex_);
     sinks_.erase(std::remove(sinks_.begin(), sinks_.end(), sink_ptr), sinks_.end());
 }
 
+
+inline void c11log::sinks::async_sink::shutdown(const std::chrono::seconds &timeout)
+{
+    auto until = std::chrono::system_clock::now() + timeout;
+    while (q_.size() > 0 && std::chrono::system_clock::now() < until)
+    {
+        std::this_thread::sleep_for(std::chrono::milliseconds(200));
+    }
+    shutdown_();
+}
+
 inline void c11log::sinks::async_sink::shutdown_()
 {
-    {        
-        std::lock_guard<std::mutex> lock(mutex_);
-        active_ = false;        
+    if(active_)
+    {
+        active_ = false;
+        if (back_thread_.joinable())
+            back_thread_.join();
     }
-    q_.clear();
-    back_thread_.join();
 }
 

include/c11log/sinks/file_sinks.h

@@ -125,7 +125,7 @@ private:
  */
 class midnight_file_sink:public details::rotating_file_sink_base {
 public:
-    midnight_file_sink(const std::string& base_filename, const std::string& extension):
+    midnight_file_sink(const std::string& base_filename, const std::string& extension = "txt"):
         _base_filename(base_filename),
         _extension(extension),
         _midnight_tp { _calc_midnight_tp() }

test/test.cpp

@@ -4,21 +4,32 @@
 #include "stdafx.h"
 
 
-void fn();
+
 int main(int argc, char* argv[])
 {
 
     c11log::logger logger("test");
 
-    auto sink = std::make_shared<c11log::sinks::stdout_sink>();
-    auto async = std::make_shared<c11log::sinks::async_sink>(100);
-    async->add_sink(sink);
+    auto screen_sink = std::make_shared<c11log::sinks::stdout_sink>();
+    auto file_sink = std::make_shared<c11log::sinks::midnight_file_sink>("logtest");
+    auto async = std::make_shared<c11log::sinks::async_sink>(1000);
+    async->add_sink(file_sink);
     logger.add_sink(async);
-    logger.info() << "Hello logger!";
-    utils::run(std::chrono::seconds(10), fn);
+    //logger.add_sink(file_sink);
+
+
+    auto fn = [&logger]()
+    {
+        logger.info() << "Hello logger!";
+    };
+    utils::bench("test log", std::chrono::seconds(3), fn);
+    logger.info() << "bye";
+    utils::bench("shutdown", [&async]() {
+        async->shutdown(std::chrono::seconds(10));
+    });
+
+
     return 0;
 }
 
-static void fn()
-{}
 

test/utils.h

@@ -20,7 +20,7 @@ std::string format(const T& value)
     return ss.str();
 }
 
-inline void run(const std::chrono::milliseconds &duration, const std::function<void() >& fn)
+inline void bench(const std::string& fn_name, const std::chrono::milliseconds &duration, const std::function<void() >& fn)
 {
     using namespace std::chrono;
     typedef steady_clock the_clock;
@@ -38,10 +38,21 @@ inline void run(const std::chrono::milliseconds &duration, const std::function<v
         auto p = now - lastPrintTime;
         if (now - lastPrintTime >= print_interval)
         {
-            std::cout << format(counter) << " per sec" << std::endl;
+            std::cout << fn_name << ": " << format(counter) << " per sec" << std::endl;
             counter = 0;
             lastPrintTime = the_clock::now();
         }
     }
 }
+
+inline void bench(const std::string& fn_name,  const std::function<void() >& fn)
+{
+
+    using namespace std::chrono;
+    auto start = steady_clock::now();
+    fn();
+    auto delta = steady_clock::now() - start;
+
+    std::cout << fn_name << ": " << duration_cast<milliseconds>(delta).count() << " ms" << std::endl;
+}
 }
\ No newline at end of file