From 457cc1170bd16175341ddbe0b6f43a605205cb36 Mon Sep 17 00:00:00 2001 From: gabime Date: Sat, 29 Mar 2014 13:04:42 +0300 Subject: [PATCH] async_sink fixes --- example/example.cpp | 6 +++-- include/c11log/details/stack_oss.h | 16 ++++++------ include/c11log/logger.h | 4 --- include/c11log/sinks/async_sink.h | 39 ++++++++++++++++++++---------- 4 files changed, 38 insertions(+), 27 deletions(-) diff --git a/example/example.cpp b/example/example.cpp index 72e428af..df72c907 100644 --- a/example/example.cpp +++ b/example/example.cpp @@ -23,9 +23,11 @@ int main(int argc, char* argv[]) logger cout_logger ("", sinks::stdout_sink()); cout_logger.info() << "Hello " << "man"; - auto fsink = std::make_shared("log", "txt", 1024*1024*50 , 5, 0); + //auto fsink = std::make_shared("log", "txt", 1024*1024*50 , 5, 0); - logger my_logger ("my_logger", sinks::null_sink::get()); + auto as = std::make_shared(howmany); + as->add_sink(sinks::null_sink::get()); + logger my_logger ("my_logger", as); auto start = system_clock::now(); for(unsigned int i = 0; i < howmany ; i++) diff --git a/include/c11log/details/stack_oss.h b/include/c11log/details/stack_oss.h index 61c5c9a2..837ae1d3 100644 --- a/include/c11log/details/stack_oss.h +++ b/include/c11log/details/stack_oss.h @@ -9,17 +9,17 @@ namespace c11log namespace details { -class str_devicebuf:public std::streambuf +class stack_devicebuf:public std::streambuf { public: using Base = std::streambuf; - str_devicebuf() = default; - ~str_devicebuf() = default; + stack_devicebuf() = default; + ~stack_devicebuf() = default; - str_devicebuf(const str_devicebuf& other) = delete; - str_devicebuf(str_devicebuf&& other) = delete; - str_devicebuf& operator=(const str_devicebuf&) = delete; - str_devicebuf& operator=(str_devicebuf&&) = delete; + stack_devicebuf(const stack_devicebuf& other) = delete; + stack_devicebuf(stack_devicebuf&& other) = delete; + stack_devicebuf& operator=(const stack_devicebuf&) = delete; + stack_devicebuf& operator=(stack_devicebuf&&) = delete; bufpair_t buf() { @@ -83,7 +83,7 @@ public: } private: - str_devicebuf _dev; + stack_devicebuf _dev; }; } } diff --git a/include/c11log/logger.h b/include/c11log/logger.h index 26f47baa..fa75aa49 100644 --- a/include/c11log/logger.h +++ b/include/c11log/logger.h @@ -6,7 +6,6 @@ #include #include #include -#include #include "common_types.h" #include "sinks/base_sink.h" @@ -60,7 +59,6 @@ public: details::line_logger fatal(); - private: friend details::line_logger; @@ -99,7 +97,6 @@ inline c11log::logger::logger(const std::string& name, sink_ptr sink, formatter_ logger(name, {sink}, f) {} - inline c11log::details::line_logger c11log::logger::log(c11log::level::level_enum msg_level) { return details::line_logger(this, msg_level, msg_level >= _logger_level); @@ -133,7 +130,6 @@ inline c11log::details::line_logger c11log::logger::fatal() } - inline const std::string& c11log::logger::get_name() const { return _logger_name; diff --git a/include/c11log/sinks/async_sink.h b/include/c11log/sinks/async_sink.h index 6a408e34..8f532049 100644 --- a/include/c11log/sinks/async_sink.h +++ b/include/c11log/sinks/async_sink.h @@ -24,12 +24,15 @@ public: using size_type = queue_t::size_type; explicit async_sink(const size_type max_queue_size); + + //Stop logging and join the back thread + // TODO: limit with timeout of the join and kill it afterwards? ~async_sink(); void add_sink(logger::sink_ptr sink); void remove_sink(logger::sink_ptr sink_ptr); //Wait to remaining items (if any) in the queue to be written and shutdown - void shutdown(const std::chrono::seconds& timeout); + void shutdown(const std::chrono::milliseconds& timeout); protected: @@ -43,6 +46,7 @@ private: std::thread _back_thread; //Clear all remaining messages(if any), stop the _back_thread and join it void _shutdown(); + std::mutex _shutdown_mutex; }; } } @@ -64,19 +68,24 @@ inline c11log::sinks::async_sink::~async_sink() } inline void c11log::sinks::async_sink::_sink_it(const details::log_msg& msg) { + auto msg_size = msg.msg_buf.second; + if(!msg_size) + return; //re allocate on the heap the (stack based) message auto new_msg = new details::log_msg(); *new_msg = msg; auto msg_size = msg.msg_buf.second; - char *buf = new char[msg_size]; - std::memcpy(buf, msg.msg_buf.first, msg_size); - new_msg->msg_buf = bufpair_t(buf, msg_size); - auto new_shared = std::shared_ptr(new_msg, [](details::log_msg* msg_to_delete) + + char *buf = new char[msg_size]; + std::memcpy(buf, msg.msg_buf.first, msg_size); + new_msg->msg_buf = bufpair_t(buf, msg_size); + + auto new_shared_msg = std::shared_ptr(new_msg, [](details::log_msg* msg_to_delete) { delete []msg_to_delete->msg_buf.first; delete msg_to_delete; }); - _q.push(new_shared); + _q.push(new_shared_msg); } inline void c11log::sinks::async_sink::_thread_loop() @@ -106,23 +115,27 @@ inline void c11log::sinks::async_sink::remove_sink(logger::sink_ptr sink_ptr) } -inline void c11log::sinks::async_sink::shutdown(const std::chrono::seconds &timeout) +inline void c11log::sinks::async_sink::shutdown(const std::chrono::milliseconds& 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)); - } + if(timeout > std::chrono::milliseconds::zero()) + { + auto until = log_clock::now() + timeout; + while (_q.size() > 0 && log_clock::now() < until) + { + std::this_thread::sleep_for(std::chrono::milliseconds(2)); + } + } _shutdown(); } inline void c11log::sinks::async_sink::_shutdown() { + std::lock_guard guard(_shutdown_mutex); if(_active) { _active = false; if (_back_thread.joinable()) - _back_thread.join(); + _back_thread.join(); } }