Subject: [Boost-commit] svn:boost r84401 - in branches/release: boost/lockfree boost/lockfree/detail libs/lockfree libs/lockfree/test
From: tim_at_[hidden]
Date: 2013-05-21 06:26:58
Author: timblechmann
Date: 2013-05-21 06:26:57 EDT (Tue, 21 May 2013)
New Revision: 84401
URL: http://svn.boost.org/trac/boost/changeset/84401
Log:
lockfree: merge fixes from trunk
Properties modified:
branches/release/boost/lockfree/ (props changed)
branches/release/libs/lockfree/ (props changed)
Text files modified:
branches/release/boost/lockfree/detail/atomic.hpp | 14 +-
branches/release/boost/lockfree/spsc_queue.hpp | 9
branches/release/libs/lockfree/test/spsc_queue_test.cpp | 203 +++++++++++++++++++++++++++++++++++++++
3 files changed, 212 insertions(+), 14 deletions(-)
Modified: branches/release/boost/lockfree/detail/atomic.hpp
==============================================================================
--- branches/release/boost/lockfree/detail/atomic.hpp (original)
+++ branches/release/boost/lockfree/detail/atomic.hpp 2013-05-21 06:26:57 EDT (Tue, 21 May 2013)
@@ -20,20 +20,20 @@
// GCC supports atomic<> from version 4.8 onwards.
#if defined(__GNUC__)
# if defined(__GNUC_PATCHLEVEL__)
-# define __GNUC_VERSION__ (__GNUC__ * 10000 \
- + __GNUC_MINOR__ * 100 \
- + __GNUC_PATCHLEVEL__)
+# define BOOST_ATOMIC_GNUC_VERSION (__GNUC__ * 10000 \
+ + __GNUC_MINOR__ * 100 \
+ + __GNUC_PATCHLEVEL__)
# else
-# define __GNUC_VERSION__ (__GNUC__ * 10000 \
- + __GNUC_MINOR__ * 100)
+# define BOOST_LOCKFREE_GNUC_VERSION (__GNUC__ * 10000 \
+ + __GNUC_MINOR__ * 100)
# endif
#endif
-#if (__GNUC_VERSION__ >= 40800) && (__cplusplus >= 201103L)
+#if (BOOST_LOCKFREE_GNUC_VERSION >= 40800) && (__cplusplus >= 201103L)
#undef BOOST_LOCKFREE_NO_HDR_ATOMIC
#endif
-#undef __GNUC_VERSION__
+#undef BOOST_LOCKFREE_GNUC_VERSION
#if defined(BOOST_LOCKFREE_NO_HDR_ATOMIC)
#include <boost/atomic.hpp>
Modified: branches/release/boost/lockfree/spsc_queue.hpp
==============================================================================
--- branches/release/boost/lockfree/spsc_queue.hpp (original)
+++ branches/release/boost/lockfree/spsc_queue.hpp 2013-05-21 06:26:57 EDT (Tue, 21 May 2013)
@@ -284,11 +284,12 @@
}
};
-template <typename T, std::size_t max_size>
+template <typename T, std::size_t MaxSize>
class compile_time_sized_ringbuffer:
public ringbuffer_base<T>
{
typedef std::size_t size_t;
+ static const std::size_t max_size = MaxSize + 1;
boost::array<T, max_size> array_;
public:
@@ -349,7 +350,7 @@
public:
explicit runtime_sized_ringbuffer(size_t max_elements):
- max_elements_(max_elements)
+ max_elements_(max_elements + 1)
{
// TODO: we don't necessarily need to construct all elements
array_ = Alloc::allocate(max_elements);
@@ -359,7 +360,7 @@
template <typename U>
runtime_sized_ringbuffer(typename Alloc::template rebind<U>::other const & alloc, size_t max_elements):
- Alloc(alloc), max_elements_(max_elements)
+ Alloc(alloc), max_elements_(max_elements + 1)
{
// TODO: we don't necessarily need to construct all elements
array_ = Alloc::allocate(max_elements);
@@ -368,7 +369,7 @@
}
runtime_sized_ringbuffer(Alloc const & alloc, size_t max_elements):
- Alloc(alloc), max_elements_(max_elements)
+ Alloc(alloc), max_elements_(max_elements + 1)
{
// TODO: we don't necessarily need to construct all elements
array_ = Alloc::allocate(max_elements);
Modified: branches/release/libs/lockfree/test/spsc_queue_test.cpp
==============================================================================
--- branches/release/libs/lockfree/test/spsc_queue_test.cpp (original)
+++ branches/release/libs/lockfree/test/spsc_queue_test.cpp 2013-05-21 06:26:57 EDT (Tue, 21 May 2013)
@@ -131,6 +131,21 @@
output_iterator_
};
+BOOST_AUTO_TEST_CASE( spsc_queue_capacity_test )
+{
+ spsc_queue<int, capacity<2> > f;
+
+ BOOST_REQUIRE(f.push(1));
+ BOOST_REQUIRE(f.push(2));
+ BOOST_REQUIRE(!f.push(3));
+
+ spsc_queue<int> g(2);
+
+ BOOST_REQUIRE(g.push(1));
+ BOOST_REQUIRE(g.push(2));
+ BOOST_REQUIRE(!g.push(3));
+}
+
template <int EnqueueMode>
void spsc_queue_buffer_push_return_value(void)
@@ -162,15 +177,15 @@
switch (EnqueueMode) {
case pointer_and_size:
- BOOST_REQUIRE_EQUAL(rb.push(data, xqueue_size), buffer_size - xqueue_size - 1);
+ BOOST_REQUIRE_EQUAL(rb.push(data, xqueue_size), buffer_size - xqueue_size);
break;
case reference_to_array:
- BOOST_REQUIRE_EQUAL(rb.push(data), buffer_size - xqueue_size - 1);
+ BOOST_REQUIRE_EQUAL(rb.push(data), buffer_size - xqueue_size);
break;
case iterator_pair:
- BOOST_REQUIRE_EQUAL(rb.push(data, data + xqueue_size), data + buffer_size - xqueue_size - 1);
+ BOOST_REQUIRE_EQUAL(rb.push(data, data + xqueue_size), data + buffer_size - xqueue_size);
break;
default:
@@ -291,3 +306,185 @@
spsc_queue_buffer_pop<reference_to_array, 7, 16, 64>();
spsc_queue_buffer_pop<output_iterator_, 7, 16, 64>();
}
+
+
+#ifndef BOOST_LOCKFREE_STRESS_TEST
+static const boost::uint32_t nodes_per_thread = 100000;
+#else
+static const boost::uint32_t nodes_per_thread = 100000000;
+#endif
+
+struct spsc_queue_tester
+{
+ spsc_queue<int, capacity<128> > sf;
+
+ boost::lockfree::detail::atomic<long> spsc_queue_cnt, received_nodes;
+
+ static_hashed_set<int, 1<<16 > working_set;
+
+ spsc_queue_tester(void):
+ spsc_queue_cnt(0), received_nodes(0)
+ {}
+
+ void add(void)
+ {
+ for (boost::uint32_t i = 0; i != nodes_per_thread; ++i) {
+ int id = generate_id<int>();
+ working_set.insert(id);
+
+ while (sf.push(id) == false)
+ {}
+
+ ++spsc_queue_cnt;
+ }
+ }
+
+ bool get_element(void)
+ {
+ int data;
+ bool success = sf.pop(data);
+
+ if (success) {
+ ++received_nodes;
+ --spsc_queue_cnt;
+ bool erased = working_set.erase(data);
+ assert(erased);
+ return true;
+ } else
+ return false;
+ }
+
+ boost::lockfree::detail::atomic<bool> running;
+
+ void get(void)
+ {
+ for(;;) {
+ bool success = get_element();
+ if (!running && !success)
+ return;
+ }
+ }
+
+ void run(void)
+ {
+ running = true;
+
+ BOOST_REQUIRE(sf.empty());
+
+ thread reader(boost::bind(&spsc_queue_tester::get, this));
+ thread writer(boost::bind(&spsc_queue_tester::add, this));
+ cout << "reader and writer threads created" << endl;
+
+ writer.join();
+ cout << "writer threads joined. waiting for readers to finish" << endl;
+
+ running = false;
+ reader.join();
+
+ BOOST_REQUIRE_EQUAL(received_nodes, nodes_per_thread);
+ BOOST_REQUIRE_EQUAL(spsc_queue_cnt, 0);
+ BOOST_REQUIRE(sf.empty());
+ BOOST_REQUIRE(working_set.count_nodes() == 0);
+ }
+};
+
+BOOST_AUTO_TEST_CASE( spsc_queue_test_caching )
+{
+ boost::shared_ptr<spsc_queue_tester> test1(new spsc_queue_tester);
+ test1->run();
+}
+
+struct spsc_queue_tester_buffering
+{
+ spsc_queue<int, capacity<128> > sf;
+
+ boost::lockfree::detail::atomic<long> spsc_queue_cnt;
+
+ static_hashed_set<int, 1<<16 > working_set;
+ boost::lockfree::detail::atomic<long> received_nodes;
+
+ spsc_queue_tester_buffering(void):
+ spsc_queue_cnt(0), received_nodes(0)
+ {}
+
+ static const size_t buf_size = 5;
+
+ void add(void)
+ {
+ boost::array<int, buf_size> input_buffer;
+ for (boost::uint32_t i = 0; i != nodes_per_thread; i+=buf_size) {
+ for (size_t i = 0; i != buf_size; ++i) {
+ int id = generate_id<int>();
+ working_set.insert(id);
+ input_buffer[i] = id;
+ }
+
+ size_t pushed = 0;
+
+ do {
+ pushed += sf.push(input_buffer.c_array() + pushed,
+ input_buffer.size() - pushed);
+ } while (pushed != buf_size);
+
+ spsc_queue_cnt+=buf_size;
+ }
+ }
+
+ bool get_elements(void)
+ {
+ boost::array<int, buf_size> output_buffer;
+
+ size_t popd = sf.pop(output_buffer.c_array(), output_buffer.size());
+
+ if (popd) {
+ received_nodes += popd;
+ spsc_queue_cnt -= popd;
+
+ for (size_t i = 0; i != popd; ++i) {
+ bool erased = working_set.erase(output_buffer[i]);
+ assert(erased);
+ }
+
+ return true;
+ } else
+ return false;
+ }
+
+ boost::lockfree::detail::atomic<bool> running;
+
+ void get(void)
+ {
+ for(;;) {
+ bool success = get_elements();
+ if (!running && !success)
+ return;
+ }
+ }
+
+ void run(void)
+ {
+ running = true;
+
+ thread reader(boost::bind(&spsc_queue_tester_buffering::get, this));
+ thread writer(boost::bind(&spsc_queue_tester_buffering::add, this));
+ cout << "reader and writer threads created" << endl;
+
+ writer.join();
+ cout << "writer threads joined. waiting for readers to finish" << endl;
+
+ running = false;
+ reader.join();
+
+ BOOST_REQUIRE_EQUAL(received_nodes, nodes_per_thread);
+ BOOST_REQUIRE_EQUAL(spsc_queue_cnt, 0);
+ BOOST_REQUIRE(sf.empty());
+ BOOST_REQUIRE(working_set.count_nodes() == 0);
+ }
+};
+
+
+BOOST_AUTO_TEST_CASE( spsc_queue_test_buffering )
+{
+ boost::shared_ptr<spsc_queue_tester_buffering> test1(new spsc_queue_tester_buffering);
+ test1->run();
+}