Fixed flush of logs in async mode

This commit is contained in:
gabime 2015-11-28 15:00:19 +02:00
parent 860015ccfd
commit 06ffde6333
4 changed files with 90 additions and 26 deletions

View File

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

View File

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

View File

@ -60,9 +60,17 @@ inline spdlog::async_logger::async_logger(const std::string& logger_name,
const async_overflow_policy overflow_policy, const async_overflow_policy overflow_policy,
const std::function<void()>& worker_warmup_cb, const std::function<void()>& worker_warmup_cb,
const std::chrono::milliseconds& flush_interval_ms) : 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) inline void spdlog::async_logger::_set_formatter(spdlog::formatter_ptr msg_formatter)
{ {
_formatter = msg_formatter; _formatter = msg_formatter;

View File

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