Commit d49879bb authored by Dominik Charousset's avatar Dominik Charousset

Allow draining ringbuffers in one shot

parent 3ecdb1c9
...@@ -23,6 +23,8 @@ ...@@ -23,6 +23,8 @@
#include <condition_variable> #include <condition_variable>
#include <mutex> #include <mutex>
#include "caf/config.hpp"
namespace caf { namespace caf {
namespace detail { namespace detail {
...@@ -46,6 +48,15 @@ public: ...@@ -46,6 +48,15 @@ public:
cv_empty_.wait(guard); cv_empty_.wait(guard);
} }
template <class TimePoint>
bool wait_nonempty(TimePoint timeout) {
if (!empty())
return true;
auto pred = [&] { return !empty(); };
guard_type guard{mtx_};
return cv_empty_.wait_until(guard, timeout, pred);
}
T& front() { T& front() {
// Safe to access without lock, because we assume a single consumer. // Safe to access without lock, because we assume a single consumer.
return buf_[rd_pos_]; return buf_[rd_pos_];
...@@ -60,6 +71,33 @@ public: ...@@ -60,6 +71,33 @@ public:
cv_full_.notify_all(); cv_full_.notify_all();
} }
template <class OutputIterator>
OutputIterator get_all(OutputIterator i) {
// No lock needed, again because of single-consumer assumption.
auto first = rd_pos_.load();
auto last = wr_pos_.load();
size_t n;
CAF_ASSERT(first != last);
// Move buffer content to the output iterator.
if (first < last) {
n = last - first;
for (auto j = first; j != last; ++j)
*i++ = std::move(buf_[j]);
} else {
n = (Size - first) + last;
for (size_t j = first; j != Size; ++j)
*i++ = std::move(buf_[j]);
for (size_t j = 0; j != last; ++j)
*i++ = std::move(buf_[j]);
}
guard_type guard{mtx_};
rd_pos_ = (first + n) % Size;
// Wakeup a waiting producers if the queue became non-full.
if (first == next(last))
cv_full_.notify_all();
return i;
}
void push_back(T&& x) { void push_back(T&& x) {
guard_type guard{mtx_}; guard_type guard{mtx_};
while (full()) while (full())
......
...@@ -82,6 +82,40 @@ CAF_TEST(push_back) { ...@@ -82,6 +82,40 @@ CAF_TEST(push_back) {
CAF_CHECK_EQUAL(buf.front(), 0); CAF_CHECK_EQUAL(buf.front(), 0);
} }
CAF_TEST(get all) {
using array_type = std::array<int, buf_size>;
using vector_type = std::vector<int>;
array_type tmp;
auto fetch_all = [&] {
auto i = tmp.begin();
auto e = buf.get_all(i);
return vector_type(i, e);
};
CAF_MESSAGE("add five element");
for (int i = 0; i < 5; ++i)
buf.push_back(std::move(i));
CAF_CHECK_EQUAL(buf.empty(), false);
CAF_CHECK_EQUAL(buf.full(), false);
CAF_CHECK_EQUAL(buf.size(), 5u);
CAF_CHECK_EQUAL(buf.front(), 0);
CAF_MESSAGE("drain elements");
CAF_CHECK_EQUAL(fetch_all(), vector_type({0, 1, 2, 3, 4}));
CAF_CHECK_EQUAL(buf.empty(), true);
CAF_CHECK_EQUAL(buf.full(), false);
CAF_CHECK_EQUAL(buf.size(), 0u);
CAF_MESSAGE("add 60 elements (wraps around)");
vector_type expected;
for (int i = 0; i < 60; ++i) {
expected.push_back(i);
buf.push_back(std::move(i));
}
CAF_CHECK_EQUAL(buf.size(), 60u);
CAF_CHECK_EQUAL(fetch_all(), expected);
CAF_CHECK_EQUAL(buf.empty(), true);
CAF_CHECK_EQUAL(buf.full(), false);
CAF_CHECK_EQUAL(buf.size(), 0u);
}
CAF_TEST(concurrent access) { CAF_TEST(concurrent access) {
std::vector<std::thread> producers; std::vector<std::thread> producers;
producers.emplace_back(producer, std::ref(buf), 0, 100); producers.emplace_back(producer, std::ref(buf), 0, 100);
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment