async_sink fixes

This commit is contained in:
gabime 2014-03-29 13:04:42 +03:00
parent d50efe9b29
commit 457cc1170b
4 changed files with 38 additions and 27 deletions

View File

@ -23,9 +23,11 @@ int main(int argc, char* argv[])
logger cout_logger ("", sinks::stdout_sink()); logger cout_logger ("", sinks::stdout_sink());
cout_logger.info() << "Hello " << "man"; cout_logger.info() << "Hello " << "man";
auto fsink = std::make_shared<sinks::rotating_file_sink>("log", "txt", 1024*1024*50 , 5, 0); //auto fsink = std::make_shared<sinks::rotating_file_sink>("log", "txt", 1024*1024*50 , 5, 0);
logger my_logger ("my_logger", sinks::null_sink::get()); auto as = std::make_shared<sinks::async_sink>(howmany);
as->add_sink(sinks::null_sink::get());
logger my_logger ("my_logger", as);
auto start = system_clock::now(); auto start = system_clock::now();
for(unsigned int i = 0; i < howmany ; i++) for(unsigned int i = 0; i < howmany ; i++)

View File

@ -9,17 +9,17 @@ namespace c11log
namespace details namespace details
{ {
class str_devicebuf:public std::streambuf class stack_devicebuf:public std::streambuf
{ {
public: public:
using Base = std::streambuf; using Base = std::streambuf;
str_devicebuf() = default; stack_devicebuf() = default;
~str_devicebuf() = default; ~stack_devicebuf() = default;
str_devicebuf(const str_devicebuf& other) = delete; stack_devicebuf(const stack_devicebuf& other) = delete;
str_devicebuf(str_devicebuf&& other) = delete; stack_devicebuf(stack_devicebuf&& other) = delete;
str_devicebuf& operator=(const str_devicebuf&) = delete; stack_devicebuf& operator=(const stack_devicebuf&) = delete;
str_devicebuf& operator=(str_devicebuf&&) = delete; stack_devicebuf& operator=(stack_devicebuf&&) = delete;
bufpair_t buf() bufpair_t buf()
{ {
@ -83,7 +83,7 @@ public:
} }
private: private:
str_devicebuf _dev; stack_devicebuf _dev;
}; };
} }
} }

View File

@ -6,7 +6,6 @@
#include<memory> #include<memory>
#include<mutex> #include<mutex>
#include<atomic> #include<atomic>
#include <algorithm>
#include "common_types.h" #include "common_types.h"
#include "sinks/base_sink.h" #include "sinks/base_sink.h"
@ -60,7 +59,6 @@ public:
details::line_logger fatal(); details::line_logger fatal();
private: private:
friend details::line_logger; friend details::line_logger;
@ -99,7 +97,6 @@ inline c11log::logger::logger(const std::string& name, sink_ptr sink, formatter_
logger(name, {sink}, f) {} logger(name, {sink}, f) {}
inline c11log::details::line_logger c11log::logger::log(c11log::level::level_enum msg_level) 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); 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 inline const std::string& c11log::logger::get_name() const
{ {
return _logger_name; return _logger_name;

View File

@ -24,12 +24,15 @@ public:
using size_type = queue_t::size_type; using size_type = queue_t::size_type;
explicit async_sink(const size_type max_queue_size); 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(); ~async_sink();
void add_sink(logger::sink_ptr sink); void add_sink(logger::sink_ptr sink);
void remove_sink(logger::sink_ptr sink_ptr); void remove_sink(logger::sink_ptr sink_ptr);
//Wait to remaining items (if any) in the queue to be written and shutdown //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: protected:
@ -43,6 +46,7 @@ private:
std::thread _back_thread; std::thread _back_thread;
//Clear all remaining messages(if any), stop the _back_thread and join it //Clear all remaining messages(if any), stop the _back_thread and join it
void _shutdown(); 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) 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 //re allocate on the heap the (stack based) message
auto new_msg = new details::log_msg(); auto new_msg = new details::log_msg();
*new_msg = msg; *new_msg = msg;
auto msg_size = msg.msg_buf.second; auto msg_size = msg.msg_buf.second;
char *buf = new char[msg_size];
std::memcpy(buf, msg.msg_buf.first, msg_size); char *buf = new char[msg_size];
new_msg->msg_buf = bufpair_t(buf, msg_size); std::memcpy(buf, msg.msg_buf.first, msg_size);
auto new_shared = std::shared_ptr<details::log_msg>(new_msg, [](details::log_msg* msg_to_delete) new_msg->msg_buf = bufpair_t(buf, msg_size);
auto new_shared_msg = std::shared_ptr<details::log_msg>(new_msg, [](details::log_msg* msg_to_delete)
{ {
delete []msg_to_delete->msg_buf.first; delete []msg_to_delete->msg_buf.first;
delete msg_to_delete; delete msg_to_delete;
}); });
_q.push(new_shared); _q.push(new_shared_msg);
} }
inline void c11log::sinks::async_sink::_thread_loop() 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; if(timeout > std::chrono::milliseconds::zero())
while (_q.size() > 0 && std::chrono::system_clock::now() < until) {
{ auto until = log_clock::now() + timeout;
std::this_thread::sleep_for(std::chrono::milliseconds(200)); while (_q.size() > 0 && log_clock::now() < until)
} {
std::this_thread::sleep_for(std::chrono::milliseconds(2));
}
}
_shutdown(); _shutdown();
} }
inline void c11log::sinks::async_sink::_shutdown() inline void c11log::sinks::async_sink::_shutdown()
{ {
std::lock_guard<std::mutex> guard(_shutdown_mutex);
if(_active) if(_active)
{ {
_active = false; _active = false;
if (_back_thread.joinable()) if (_back_thread.joinable())
_back_thread.join(); _back_thread.join();
} }
} }