flush helper count writes instead of time

This commit is contained in:
gabime
2014-03-14 14:35:46 +02:00
parent 9d687d1634
commit 38670cef27
14 changed files with 139 additions and 94 deletions

View File

@@ -26,13 +26,15 @@ public:
explicit blocking_queue(size_type max_size) :
_max_size(max_size),
_q(),
_mutex() {
_mutex()
{
}
blocking_queue(const blocking_queue&) = delete;
blocking_queue& operator=(const blocking_queue&) = delete;
~blocking_queue() = default;
size_type size() {
size_type size()
{
std::lock_guard<std::mutex> lock(_mutex);
return _q.size();
}
@@ -41,16 +43,20 @@ public:
// If the queue is full, block the calling thread util there is room or timeout have passed.
// Return: false on timeout, true on successful push.
template<typename Duration_Rep, typename Duration_Period, typename TT>
bool push(TT&& item, const std::chrono::duration<Duration_Rep, Duration_Period>& timeout) {
bool push(TT&& item, const std::chrono::duration<Duration_Rep, Duration_Period>& timeout)
{
std::unique_lock<std::mutex> ul(_mutex);
if (_q.size() >= _max_size) {
if (!_item_popped_cond.wait_until(ul, clock::now() + timeout, [this]() {
if (_q.size() >= _max_size)
{
if (!_item_popped_cond.wait_until(ul, clock::now() + timeout, [this]()
{
return this->_q.size() < this->_max_size;
}))
return false;
}
_q.push(std::forward<TT>(item));
if (_q.size() <= 1) {
if (_q.size() <= 1)
{
ul.unlock(); //So the notified thread will have better chance to accuire the lock immediatly..
_item_pushed_cond.notify_one();
}
@@ -60,7 +66,8 @@ public:
// Push copy of item into the back of the queue.
// If the queue is full, block the calling thread until there is room.
template<typename TT>
void push(TT&& item) {
void push(TT&& item)
{
while (!push(std::forward<TT>(item), std::chrono::hours(1)));
}
@@ -68,17 +75,21 @@ public:
// If the queue is empty, block the calling thread util there is item to pop or timeout have passed.
// Return: false on timeout , true on successful pop/
template<class Duration_Rep, class Duration_Period>
bool pop(T& item, const std::chrono::duration<Duration_Rep, Duration_Period>& timeout) {
bool pop(T& item, const std::chrono::duration<Duration_Rep, Duration_Period>& timeout)
{
std::unique_lock<std::mutex> ul(_mutex);
if (_q.empty()) {
if (!_item_pushed_cond.wait_until(ul, clock::now() + timeout, [this]() {
if (_q.empty())
{
if (!_item_pushed_cond.wait_until(ul, clock::now() + timeout, [this]()
{
return !this->_q.empty();
}))
return false;
}
item = std::move(_q.front());
_q.pop();
if (_q.size() >= _max_size - 1) {
if (_q.size() >= _max_size - 1)
{
ul.unlock(); //So the notified thread will have better chance to accuire the lock immediatly..
_item_popped_cond.notify_one();
}
@@ -87,12 +98,14 @@ public:
// Pop a copy of the front item in the queue into the given item ref.
// If the queue is empty, block the calling thread util there is item to pop.
void pop(T& item) {
void pop(T& item)
{
while (!pop(item, std::chrono::hours(1)));
}
// Clear the queue
void clear() {
void clear()
{
{
std::unique_lock<std::mutex> ul(_mutex);
queue_t().swap(_q);

View File

@@ -22,25 +22,30 @@ public:
str_devicebuf& operator=(str_devicebuf&&) = delete;
const std::string& str_ref() const {
const std::string& str_ref() const
{
return _str;
}
void clear() {
void clear()
{
_str.clear();
}
protected:
virtual int sync() override {
virtual int sync() override
{
return 0;
}
virtual std::streamsize xsputn(const char_type* s, std::streamsize count) override {
virtual std::streamsize xsputn(const char_type* s, std::streamsize count) override
{
_str.append(s, static_cast<unsigned int>(count));
return count;
}
virtual int_type overflow(int_type ch) override {
virtual int_type overflow(int_type ch) override
{
if (ch != traits_type::eof())
_str.append((char*)&ch, 1);
return 1;
@@ -59,12 +64,11 @@ public:
fast_oss(fast_oss&& other) = delete;
fast_oss& operator=(const fast_oss& other) = delete;
const std::string& str_ref() const {
const std::string& str_ref() const
{
return _dev.str_ref();
}
private:
str_devicebuf _dev;
};

View File

@@ -1,36 +1,31 @@
#pragma once
#include <chrono>
#include <iostream>
// Flush to file every X writes..
namespace c11log
{
namespace details
{
class file_flush_helper
{
public:
explicit file_flush_helper(const std::chrono::milliseconds &flush_every): _flush_every(flush_every), _last_flush() {};
explicit file_flush_helper(const std::size_t flush_every):
_flush_every(flush_every),
_write_counter(0) {};
void write(std::ofstream& ofs, const std::string& msg) {
ofs << msg;
//If zero - flush every time
if(_flush_every == std::chrono::milliseconds::min()) {
void write(std::ofstream& ofs, const std::string& msg)
{
ofs.write(msg.c_str(), msg.size());
if(++_write_counter == _flush_every)
{
ofs.flush();
} else {
auto now = std::chrono::system_clock::now();
if(now - _last_flush >= _flush_every) {
ofs.flush();
_last_flush = now;
}
_write_counter = 0;
}
}
private:
std::chrono::milliseconds _flush_every;
std::chrono::system_clock::time_point _last_flush;
const std::size_t _flush_every;
std::size_t _write_counter;
};
}
}

View File

@@ -22,8 +22,10 @@ public:
_callback_logger(callback_logger),
_oss(),
_level(msg_level),
_enabled(enabled) {
if(enabled) {
_enabled(enabled)
{
if(enabled)
{
callback_logger->_formatter->format_header(callback_logger->_logger_name,
msg_level,
log_clock::now(),
@@ -40,12 +42,15 @@ public:
// The move ctor should only be called on start of logging line,
// where no logging happened yet for this line so no need to copy the string from the other
_oss(),
_level(other._level) {
_level(other._level)
{
};
~line_logger() {
if (_enabled) {
~line_logger()
{
if (_enabled)
{
_oss << '\n';
_callback_logger->_log_it(_oss.str_ref(), _level);
}
@@ -53,7 +58,8 @@ public:
template<typename T>
line_logger&& operator<<(const T& msg) && {
line_logger&& operator<<(const T& msg) &&
{
if (_enabled)
_oss << msg;
return std::move(*this);