libs/thread/test/sync/mutual_exclusion/sync_queue/multi_thread_pass.cpp

   1 // Copyright (C) 2013 Vicente J. Botet Escriba
   2 //
   3 //  Distributed under the Boost Software License, Version 1.0. (See accompanying
   4 //  file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
   5
   6 // <boost/thread/sync_queue.hpp>
   7
   8 // class sync_queue<T>
   9
  10 //    push || pull;
  11
  12 #include <boost/config.hpp>
  13 #if ! defined  BOOST_NO_CXX11_DECLTYPE
  14 #define BOOST_RESULT_OF_USE_DECLTYPE
  15 #endif
  16
  17 #define BOOST_THREAD_VERSION 4
  18 //#define BOOST_THREAD_QUEUE_DEPRECATE_OLD
  19
  20 #include <boost/thread/sync_queue.hpp>
  21 #include <boost/thread/future.hpp>
  22 #include <boost/thread/barrier.hpp>
  23
  24 #include <boost/detail/lightweight_test.hpp>
  25
  26 #ifndef BOOST_THREAD_QUEUE_DEPRECATE_OLD
  27 struct call_push
  28 {
  29   boost::sync_queue<int> *q_;
  30   boost::barrier *go_;
  31
  32   call_push(boost::sync_queue<int> *q, boost::barrier *go) :
  33     q_(q), go_(go)
  34   {
  35   }
  36   typedef void result_type;
  37   void operator()()
  38   {
  39     go_->count_down_and_wait();
  40     q_->push(42);
  41
  42   }
  43 };
  44
  45 struct call_pull
  46 {
  47   boost::sync_queue<int> *q_;
  48   boost::barrier *go_;
  49
  50   call_pull(boost::sync_queue<int> *q, boost::barrier *go) :
  51     q_(q), go_(go)
  52   {
  53   }
  54   typedef int result_type;
  55   int operator()()
  56   {
  57     go_->count_down_and_wait();
  58     return q_->pull();
  59   }
  60 };
  61 #endif
  62
  63 template <typename ValueType>
  64 struct call_push_back
  65 {
  66   boost::sync_queue<ValueType> *q_;
  67   boost::barrier *go_;
  68
  69   call_push_back(boost::sync_queue<ValueType> *q, boost::barrier *go) :
  70     q_(q), go_(go)
  71   {
  72   }
  73   typedef void result_type;
  74   void operator()()
  75   {
  76     go_->count_down_and_wait();
  77     q_->push_back(42);
  78   }
  79 };
  80
  81 template <typename ValueType>
  82 struct call_pull_front
  83 {
  84   boost::sync_queue<ValueType> *q_;
  85   boost::barrier *go_;
  86
  87   call_pull_front(boost::sync_queue<ValueType> *q, boost::barrier *go) :
  88     q_(q), go_(go)
  89   {
  90   }
  91   typedef ValueType result_type;
  92   ValueType operator()()
  93   {
  94     go_->count_down_and_wait();
  95     return q_->pull_front();
  96   }
  97 };
  98
  99 template <typename ValueType>
 100 struct call_wait_pull_front
 101 {
 102   boost::sync_queue<ValueType> *q_;
 103   boost::barrier *go_;
 104
 105   call_wait_pull_front(boost::sync_queue<ValueType> *q, boost::barrier *go) :
 106     q_(q), go_(go)
 107   {
 108   }
 109   typedef boost::queue_op_status result_type;
 110   boost::queue_op_status operator()(ValueType& v)
 111   {
 112     go_->wait();
 113     return q_->wait_pull_front(v);
 114   }
 115 };
 116
 117 #ifndef BOOST_THREAD_QUEUE_DEPRECATE_OLD
 118 void test_concurrent_push_and_pull_on_empty_queue()
 119 {
 120   boost::sync_queue<int> q;
 121
 122   boost::barrier go(2);
 123
 124   boost::future<void> push_done;
 125   boost::future<int> pull_done;
 126
 127   try
 128   {
 129     push_done=boost::async(boost::launch::async,
 130 #if ! defined BOOST_NO_CXX11_LAMBDAS
 131         [&q,&go]()
 132         {
 133           go.wait();
 134           q.push(42);
 135         }
 136 #else
 137         call_push(&q,&go)
 138 #endif
 139     );
 140     pull_done=boost::async(boost::launch::async,
 141 #if ! defined BOOST_NO_CXX11_LAMBDAS
 142         [&q,&go]() -> int
 143         {
 144           go.wait();
 145           return q.pull();
 146         }
 147 #else
 148         call_pull(&q,&go)
 149 #endif
 150     );
 151
 152     push_done.get();
 153     BOOST_TEST_EQ(pull_done.get(), 42);
 154     BOOST_TEST(q.empty());
 155   }
 156   catch (...)
 157   {
 158     BOOST_TEST(false);
 159   }
 160 }
 161
 162 void test_concurrent_push_on_empty_queue()
 163 {
 164   boost::sync_queue<int> q;
 165   const unsigned int n = 3;
 166   boost::barrier go(n);
 167   boost::future<void> push_done[n];
 168
 169   try
 170   {
 171     for (unsigned int i =0; i< n; ++i)
 172       push_done[i]=boost::async(boost::launch::async,
 173 #if ! defined BOOST_NO_CXX11_LAMBDAS
 174         [&q,&go]()
 175         {
 176           go.wait();
 177           q.push(42);
 178         }
 179 #else
 180         call_push(&q,&go)
 181 #endif
 182     );
 183
 184     for (unsigned int i = 0; i < n; ++i)
 185       push_done[i].get();
 186
 187     BOOST_TEST(!q.empty());
 188     for (unsigned int i =0; i< n; ++i)
 189       BOOST_TEST_EQ(q.pull(), 42);
 190     BOOST_TEST(q.empty());
 191
 192   }
 193   catch (...)
 194   {
 195     BOOST_TEST(false);
 196   }
 197 }
 198
 199 void test_concurrent_pull_on_queue()
 200 {
 201   boost::sync_queue<int> q;
 202   const unsigned int n = 3;
 203   boost::barrier go(n);
 204
 205   boost::future<int> pull_done[n];
 206
 207   try
 208   {
 209     for (unsigned int i =0; i< n; ++i)
 210       q.push(42);
 211
 212     for (unsigned int i =0; i< n; ++i)
 213       pull_done[i]=boost::async(boost::launch::async,
 214 #if ! defined BOOST_NO_CXX11_LAMBDAS
 215         [&q,&go]() -> int
 216         {
 217           go.wait();
 218           return q.pull();
 219         }
 220 #else
 221         call_pull(&q,&go)
 222 #endif
 223     );
 224
 225     for (unsigned int i = 0; i < n; ++i)
 226       BOOST_TEST_EQ(pull_done[i].get(), 42);
 227     BOOST_TEST(q.empty());
 228   }
 229   catch (...)
 230   {
 231     BOOST_TEST(false);
 232   }
 233 }
 234 #endif
 235
 236 void test_concurrent_push_back_and_pull_front_on_empty_queue()
 237 {
 238   boost::sync_queue<int> q;
 239
 240   boost::barrier go(2);
 241
 242   boost::future<void> push_done;
 243   boost::future<int> pull_done;
 244
 245   try
 246   {
 247     push_done=boost::async(boost::launch::async,
 248                            call_push_back<int>(&q,&go));
 249     pull_done=boost::async(boost::launch::async,
 250                            call_pull_front<int>(&q,&go));
 251
 252     push_done.get();
 253     BOOST_TEST_EQ(pull_done.get(), 42);
 254     BOOST_TEST(q.empty());
 255   }
 256   catch (...)
 257   {
 258     BOOST_TEST(false);
 259   }
 260 }
 261
 262 #if defined(BOOST_THREAD_PROVIDES_VARIADIC_THREAD)
 263 void test_concurrent_push_back_and_wait_pull_front_on_empty_queue()
 264 {
 265   boost::sync_queue<int> q;
 266   const unsigned int n = 3;
 267   boost::barrier go(n);
 268
 269   boost::future<boost::queue_op_status> pull_done[n];
 270   int results[n];
 271
 272   try
 273   {
 274     for (unsigned int i =0; i< n; ++i)
 275       pull_done[i]=boost::async(boost::launch::async,
 276                                 call_wait_pull_front<int>(&q,&go),
 277                                 boost::ref(results[i]));
 278
 279     for (unsigned int i =0; i< n; ++i)
 280       q.push_back(42);
 281
 282     for (unsigned int i = 0; i < n; ++i) {
 283       BOOST_TEST(pull_done[i].get() == boost::queue_op_status::success);
 284       BOOST_TEST_EQ(results[i], 42);
 285     }
 286     BOOST_TEST(q.empty());
 287   }
 288   catch (...)
 289   {
 290     BOOST_TEST(false);
 291   }
 292 }
 293
 294 void test_concurrent_wait_pull_front_and_close_on_empty_queue()
 295 {
 296   boost::sync_queue<int> q;
 297   const unsigned int n = 3;
 298   boost::barrier go(n);
 299
 300   boost::future<boost::queue_op_status> pull_done[n];
 301   int results[n];
 302
 303   try
 304   {
 305     for (unsigned int i =0; i< n; ++i)
 306       pull_done[i]=boost::async(boost::launch::async,
 307                                 call_wait_pull_front<int>(&q,&go),
 308                                 boost::ref(results[i]));
 309
 310     q.close();
 311
 312     for (unsigned int i = 0; i < n; ++i) {
 313       BOOST_TEST(pull_done[i].get() == boost::queue_op_status::closed);
 314     }
 315     BOOST_TEST(q.empty());
 316   }
 317   catch (...)
 318   {
 319     BOOST_TEST(false);
 320   }
 321 }
 322 #endif
 323
 324 void test_concurrent_push_back_on_empty_queue()
 325 {
 326   boost::sync_queue<int> q;
 327   const unsigned int n = 3;
 328   boost::barrier go(n);
 329   boost::future<void> push_done[n];
 330
 331   try
 332   {
 333     for (unsigned int i =0; i< n; ++i)
 334       push_done[i]=boost::async(boost::launch::async,
 335                                 call_push_back<int>(&q,&go));
 336
 337   }
 338   catch (...)
 339   {
 340     BOOST_TEST(false);
 341   }
 342   try
 343   {
 344     for (unsigned int i = 0; i < n; ++i)
 345       push_done[i].get();
 346
 347   }
 348   catch (...)
 349   {
 350     BOOST_TEST(false);
 351   }
 352   try
 353   {
 354     BOOST_TEST(!q.empty());
 355     for (unsigned int i =0; i< n; ++i)
 356       BOOST_TEST_EQ(q.pull_front(), 42);
 357     BOOST_TEST(q.empty());
 358
 359   }
 360   catch (...)
 361   {
 362     BOOST_TEST(false);
 363   }
 364 }
 365
 366 void test_concurrent_pull_front_on_queue()
 367 {
 368   boost::sync_queue<int> q;
 369   const unsigned int n = 3;
 370   boost::barrier go(n);
 371
 372   boost::future<int> pull_done[n];
 373
 374   try
 375   {
 376     for (unsigned int i =0; i< n; ++i)
 377       q.push_back(42);
 378
 379     for (unsigned int i =0; i< n; ++i)
 380       pull_done[i]=boost::async(boost::launch::async,
 381 #if ! defined BOOST_NO_CXX11_LAMBDAS
 382         [&q,&go]() -> int
 383         {
 384           go.wait();
 385           return q.pull_front();
 386         }
 387 #else
 388         call_pull_front<int>(&q,&go)
 389 #endif
 390       );
 391
 392     for (unsigned int i = 0; i < n; ++i)
 393       BOOST_TEST_EQ(pull_done[i].get(), 42);
 394     BOOST_TEST(q.empty());
 395   }
 396   catch (...)
 397   {
 398     BOOST_TEST(false);
 399   }
 400 }
 401
 402 int main()
 403 {
 404 #ifndef BOOST_THREAD_QUEUE_DEPRECATE_OLD
 405   test_concurrent_push_and_pull_on_empty_queue();
 406   test_concurrent_push_on_empty_queue();
 407   test_concurrent_pull_on_queue();
 408 #endif
 409   test_concurrent_push_back_and_pull_front_on_empty_queue();
 410 #if defined(BOOST_THREAD_PROVIDES_VARIADIC_THREAD)
 411   test_concurrent_push_back_and_wait_pull_front_on_empty_queue();
 412   test_concurrent_wait_pull_front_and_close_on_empty_queue();
 413 #endif
 414   test_concurrent_push_back_on_empty_queue();
 415   test_concurrent_pull_front_on_queue();
 416   return boost::report_errors();
 417 }
 418