123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252 |
- // Copyright (C) 2013 Vicente J. Botet Escriba
- //
- // Distributed under the Boost Software License, Version 1.0. (See accompanying
- // file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
- // <boost/thread/sync_bounded_queue.hpp>
- // class sync_queue<T>
- // push || pull;
- #include <boost/config.hpp>
- #if ! defined BOOST_NO_CXX11_DECLTYPE
- #define BOOST_RESULT_OF_USE_DECLTYPE
- #endif
- #define BOOST_THREAD_VERSION 4
- #include <boost/thread/sync_bounded_queue.hpp>
- #include <boost/thread/future.hpp>
- #include <boost/thread/barrier.hpp>
- #include <boost/detail/lightweight_test.hpp>
- struct call_push
- {
- boost::sync_bounded_queue<int> &q_;
- boost::barrier& go_;
- call_push(boost::sync_bounded_queue<int> &q, boost::barrier &go) :
- q_(q), go_(go)
- {
- }
- typedef void result_type;
- void operator()()
- {
- go_.count_down_and_wait();
- q_.push(42);
- }
- };
- struct call_push_2
- {
- boost::sync_bounded_queue<int> &q_;
- boost::barrier& go_;
- boost::barrier& end_;
- call_push_2(boost::sync_bounded_queue<int> &q, boost::barrier &go, boost::barrier &end) :
- q_(q), go_(go), end_(end)
- {
- }
- typedef void result_type;
- void operator()()
- {
- go_.count_down_and_wait();
- q_.push(42);
- end_.count_down_and_wait();
- }
- };
- struct call_pull
- {
- boost::sync_bounded_queue<int> &q_;
- boost::barrier& go_;
- call_pull(boost::sync_bounded_queue<int> &q, boost::barrier &go) :
- q_(q), go_(go)
- {
- }
- typedef int result_type;
- int operator()()
- {
- go_.count_down_and_wait();
- return q_.pull();
- }
- };
- void test_concurrent_push_and_pull_on_empty_queue()
- {
- boost::sync_bounded_queue<int> q(4);
- boost::barrier go(2);
- boost::future<void> push_done;
- boost::future<int> pull_done;
- try
- {
- push_done=boost::async(boost::launch::async,
- #if ! defined BOOST_NO_CXX11_LAMBDAS
- [&q,&go]()
- {
- go.wait();
- q.push(42);
- }
- #else
- call_push(q,go)
- #endif
- );
- pull_done=boost::async(boost::launch::async,
- #if ! defined BOOST_NO_CXX11_LAMBDAS
- [&q,&go]() -> int
- {
- go.wait();
- return q.pull();
- }
- #else
- call_pull(q,go)
- #endif
- );
- push_done.get();
- BOOST_TEST_EQ(pull_done.get(), 42);
- BOOST_TEST(q.empty());
- }
- catch (...)
- {
- BOOST_TEST(false);
- }
- }
- void test_concurrent_push_on_empty_queue()
- {
- boost::sync_bounded_queue<int> q(4);
- const unsigned int n = 3;
- boost::barrier go(n);
- boost::future<void> push_done[n];
- try
- {
- for (unsigned int i =0; i< n; ++i)
- push_done[i]=boost::async(boost::launch::async,
- #if ! defined BOOST_NO_CXX11_LAMBDAS
- [&q,&go]()
- {
- go.wait();
- q.push(42);
- }
- #else
- call_push(q,go)
- #endif
- );
- for (unsigned int i = 0; i < n; ++i)
- push_done[i].get();
- BOOST_TEST(!q.empty());
- for (unsigned int i =0; i< n; ++i)
- BOOST_TEST_EQ(q.pull(), 42);
- BOOST_TEST(q.empty());
- }
- catch (...)
- {
- BOOST_TEST(false);
- }
- }
- void test_concurrent_push_on_full_queue()
- {
- const unsigned int size = 2;
- boost::sync_bounded_queue<int> q(size);
- const unsigned int n = 2*size;
- boost::barrier go(n);
- boost::barrier end(size+1);
- boost::future<void> push_done[n];
- try
- {
- for (unsigned int i =0; i< n; ++i)
- push_done[i]=boost::async(boost::launch::async,
- #if ! defined BOOST_NO_CXX11_LAMBDAS
- [&q,&go,&end]()
- {
- go.wait();
- q.push(42);
- end.wait();
- }
- #else
- call_push_2(q,go,end)
- #endif
- );
- end.wait();
- BOOST_TEST(!q.empty());
- BOOST_TEST(q.full());
- for (unsigned int i =0; i< size; ++i)
- BOOST_TEST_EQ(q.pull(), 42);
- end.wait();
- for (unsigned int i = 0; i < n; ++i)
- push_done[i].get();
- BOOST_TEST(!q.empty());
- for (unsigned int i =0; i< size; ++i)
- BOOST_TEST_EQ(q.pull(), 42);
- BOOST_TEST(q.empty());
- }
- catch (...)
- {
- BOOST_TEST(false);
- }
- }
- void test_concurrent_pull_on_queue()
- {
- boost::sync_bounded_queue<int> q(4);
- const unsigned int n = 3;
- boost::barrier go(n);
- boost::future<int> pull_done[n];
- try
- {
- for (unsigned int i =0; i< n; ++i)
- q.push(42);
- for (unsigned int i =0; i< n; ++i)
- pull_done[i]=boost::async(boost::launch::async,
- #if ! defined BOOST_NO_CXX11_LAMBDAS
- [&q,&go]() -> int
- {
- go.wait();
- return q.pull();
- }
- #else
- call_pull(q,go)
- #endif
- );
- for (unsigned int i = 0; i < n; ++i)
- BOOST_TEST_EQ(pull_done[i].get(), 42);
- BOOST_TEST(q.empty());
- }
- catch (...)
- {
- BOOST_TEST(false);
- }
- }
- int main()
- {
- test_concurrent_push_and_pull_on_empty_queue();
- test_concurrent_push_on_empty_queue();
- test_concurrent_push_on_full_queue();
- test_concurrent_pull_on_queue();
- return boost::report_errors();
- }
|