gitMirror/spdlog
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fixed issue #546 by adding an "is_empty" method to the queue instead of the buggy approx_size

Browse Source
This commit is contained in:
gabime
2017-11-05 00:21:00 +02:00
parent 6fd5f3c866
commit f5fe681a41
2 changed files with 102 additions and 106 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
include/spdlog/details/async_log_helper.h
View File

@@ -384,7 +384,7 @@ inline void spdlog::details::async_log_helper::sleep_or_yield(const spdlog::log_
inline void spdlog::details::async_log_helper::wait_empty_q() inline void spdlog::details::async_log_helper::wait_empty_q()
{ {
auto last_op = details::os::now(); auto last_op = details::os::now();
while (_q.approx_size() > 0) while (!_q.is_empty())
{ {
sleep_or_yield(details::os::now(), last_op); sleep_or_yield(details::os::now(), last_op);
} }

206
include/spdlog/details/mpmc_bounded_q.h
View File

@@ -50,123 +50,119 @@ Distributed under the MIT License (http://opensource.org/licenses/MIT)
namespace spdlog namespace spdlog
{ {
namespace details namespace details
{ {
template<typename T> template<typename T>
class mpmc_bounded_queue class mpmc_bounded_queue
{ {
public: public:
using item_type = T; using item_type = T;
mpmc_bounded_queue(size_t buffer_size) mpmc_bounded_queue(size_t buffer_size)
:max_size_(buffer_size), :max_size_(buffer_size),
buffer_(new cell_t [buffer_size]), buffer_(new cell_t[buffer_size]),
buffer_mask_(buffer_size - 1) buffer_mask_(buffer_size - 1)
{ {
//queue size must be power of two //queue size must be power of two
if(!((buffer_size >= 2) && ((buffer_size & (buffer_size - 1)) == 0))) if (!((buffer_size >= 2) && ((buffer_size & (buffer_size - 1)) == 0)))
throw spdlog_ex("async logger queue size must be power of two"); throw spdlog_ex("async logger queue size must be power of two");
for (size_t i = 0; i != buffer_size; i += 1) for (size_t i = 0; i != buffer_size; i += 1)
buffer_[i].sequence_.store(i, std::memory_order_relaxed); buffer_[i].sequence_.store(i, std::memory_order_relaxed);
enqueue_pos_.store(0, std::memory_order_relaxed); enqueue_pos_.store(0, std::memory_order_relaxed);
dequeue_pos_.store(0, std::memory_order_relaxed); dequeue_pos_.store(0, std::memory_order_relaxed);
} }
~mpmc_bounded_queue() ~mpmc_bounded_queue()
{ {
delete [] buffer_; delete[] buffer_;
} }
bool enqueue(T&& data) bool enqueue(T&& data)
{ {
cell_t* cell; cell_t* cell;
size_t pos = enqueue_pos_.load(std::memory_order_relaxed); size_t pos = enqueue_pos_.load(std::memory_order_relaxed);
for (;;) for (;;) {
{ cell = &buffer_[pos & buffer_mask_];
cell = &buffer_[pos & buffer_mask_]; size_t seq = cell->sequence_.load(std::memory_order_acquire);
size_t seq = cell->sequence_.load(std::memory_order_acquire); intptr_t dif = static_cast<intptr_t>(seq) - static_cast<intptr_t>(pos);
intptr_t dif = static_cast<intptr_t>(seq) - static_cast<intptr_t>(pos); if (dif == 0) {
if (dif == 0) if (enqueue_pos_.compare_exchange_weak(pos, pos + 1, std::memory_order_relaxed))
{ break;
if (enqueue_pos_.compare_exchange_weak(pos, pos + 1, std::memory_order_relaxed)) }
break; else if (dif < 0) {
} return false;
else if (dif < 0) }
{ else {
return false; pos = enqueue_pos_.load(std::memory_order_relaxed);
} }
else }
{ cell->data_ = std::move(data);
pos = enqueue_pos_.load(std::memory_order_relaxed); cell->sequence_.store(pos + 1, std::memory_order_release);
} return true;
} }
cell->data_ = std::move(data);
cell->sequence_.store(pos + 1, std::memory_order_release);
return true;
}
bool dequeue(T& data) bool dequeue(T& data)
{ {
cell_t* cell; cell_t* cell;
size_t pos = dequeue_pos_.load(std::memory_order_relaxed); size_t pos = dequeue_pos_.load(std::memory_order_relaxed);
for (;;) for (;;) {
{ cell = &buffer_[pos & buffer_mask_];
cell = &buffer_[pos & buffer_mask_]; size_t seq =
size_t seq = cell->sequence_.load(std::memory_order_acquire);
cell->sequence_.load(std::memory_order_acquire); intptr_t dif = static_cast<intptr_t>(seq) - static_cast<intptr_t>(pos + 1);
intptr_t dif = static_cast<intptr_t>(seq) - static_cast<intptr_t>(pos + 1); if (dif == 0) {
if (dif == 0) if (dequeue_pos_.compare_exchange_weak(pos, pos + 1, std::memory_order_relaxed))
{ break;
if (dequeue_pos_.compare_exchange_weak(pos, pos + 1, std::memory_order_relaxed)) }
break; else if (dif < 0)
} return false;
else if (dif < 0) else
return false; pos = dequeue_pos_.load(std::memory_order_relaxed);
else }
pos = dequeue_pos_.load(std::memory_order_relaxed); data = std::move(cell->data_);
} cell->sequence_.store(pos + buffer_mask_ + 1, std::memory_order_release);
data = std::move(cell->data_); return true;
cell->sequence_.store(pos + buffer_mask_ + 1, std::memory_order_release); }
return true;
}
size_t approx_size() bool is_empty()
{ {
size_t first_pos = dequeue_pos_.load(std::memory_order_relaxed); unsigned front, front1, back;
size_t last_pos = enqueue_pos_.load(std::memory_order_relaxed); // try to take a consistent snapshot of front/tail.
if (last_pos <= first_pos) do {
return 0; front = enqueue_pos_.load(std::memory_order_acquire);
auto size = last_pos - first_pos; back = dequeue_pos_.load(std::memory_order_acquire);
return size < max_size_ ? size : max_size_; front1 = enqueue_pos_.load(std::memory_order_relaxed);
} } while (front != front1);
return back == front;
}
private: private:
struct cell_t struct cell_t
{ {
std::atomic<size_t> sequence_; std::atomic<size_t> sequence_;
T data_; T data_;
}; };
size_t const max_size_; size_t const max_size_;
static size_t const cacheline_size = 64; static size_t const cacheline_size = 64;
typedef char cacheline_pad_t [cacheline_size]; typedef char cacheline_pad_t[cacheline_size];
cacheline_pad_t pad0_; cacheline_pad_t pad0_;
cell_t* const buffer_; cell_t* const buffer_;
size_t const buffer_mask_; size_t const buffer_mask_;
cacheline_pad_t pad1_; cacheline_pad_t pad1_;
std::atomic<size_t> enqueue_pos_; std::atomic<size_t> enqueue_pos_;
cacheline_pad_t pad2_; cacheline_pad_t pad2_;
std::atomic<size_t> dequeue_pos_; std::atomic<size_t> dequeue_pos_;
cacheline_pad_t pad3_; cacheline_pad_t pad3_;
mpmc_bounded_queue(mpmc_bounded_queue const&) = delete; mpmc_bounded_queue(mpmc_bounded_queue const&) = delete;
void operator= (mpmc_bounded_queue const&) = delete; void operator= (mpmc_bounded_queue const&) = delete;
}; };
} // ns details } // ns details
} // ns spdlog } // ns spdlog