libs/lockfree/test/ringbuffer_test.cpp

   1 #include <boost/lockfree/ringbuffer.hpp>
   2
   3 #include <climits>
   4 #define BOOST_TEST_MODULE lockfree_tests
   5 #include <boost/test/included/unit_test.hpp>
   6
   7
   8 #include <boost/thread.hpp>
   9 #include <iostream>
  10 #include <memory>
  11
  12
  13 #include "test_helpers.hpp"
  14
  15 using namespace boost;
  16 using namespace boost::lockfree;
  17 using namespace std;
  18
  19
  20 BOOST_AUTO_TEST_CASE( simple_ringbuffer_test )
  21 {
  22     ringbuffer<int, 64> f;
  23
  24     BOOST_REQUIRE(f.empty());
  25     f.enqueue(1);
  26     f.enqueue(2);
  27
  28     int i1(0), i2(0);
  29
  30     BOOST_REQUIRE(f.dequeue(&i1));
  31     BOOST_REQUIRE_EQUAL(i1, 1);
  32
  33     BOOST_REQUIRE(f.dequeue(&i2));
  34     BOOST_REQUIRE_EQUAL(i2, 2);
  35     BOOST_REQUIRE(f.empty());
  36 }
  37
  38
  39 struct ringbuffer_tester
  40 {
  41     ringbuffer<int, 128> sf;
  42
  43     atomic<long> ringbuffer_cnt, received_nodes;
  44
  45     static_hashed_set<int, 1<<16 > working_set;
  46
  47     static const uint nodes_per_thread = 20000000;
  48
  49     static const int reader_threads = 1;
  50     static const int writer_threads = 1;
  51
  52     ringbuffer_tester(void):
  53         ringbuffer_cnt(0), received_nodes(0)
  54     {}
  55
  56     void add(void)
  57     {
  58         for (uint i = 0; i != nodes_per_thread; ++i)
  59         {
  60             while(ringbuffer_cnt > 10000)
  61                 thread::yield();
  62
  63             int id = generate_id<int>();
  64
  65             working_set.insert(id);
  66
  67             while (sf.enqueue(id) == false)
  68             {
  69                 thread::yield();
  70             }
  71
  72             ++ringbuffer_cnt;
  73         }
  74     }
  75
  76     bool get_element(void)
  77     {
  78         int data;
  79
  80         bool success = sf.dequeue(&data);
  81
  82         if (success)
  83         {
  84             ++received_nodes;
  85             --ringbuffer_cnt;
  86             bool erased = working_set.erase(data);
  87             assert(erased);
  88             return true;
  89         }
  90         else
  91             return false;
  92     }
  93
  94     volatile bool running;
  95
  96     void get(void)
  97     {
  98         for(;;)
  99         {
 100             bool success = get_element();
 101             if (not running and not success)
 102                 return;
 103             if (not success)
 104                 thread::yield();
 105         }
 106     }
 107
 108     void run(void)
 109     {
 110         running = true;
 111
 112         thread_group writer;
 113         thread_group reader;
 114
 115         BOOST_REQUIRE(sf.empty());
 116         for (int i = 0; i != reader_threads; ++i)
 117             reader.create_thread(boost::bind(&ringbuffer_tester::get, this));
 118
 119         for (int i = 0; i != writer_threads; ++i)
 120             writer.create_thread(boost::bind(&ringbuffer_tester::add, this));
 121         cout << "reader and writer threads created" << endl;
 122
 123         writer.join_all();
 124         cout << "writer threads joined. waiting for readers to finish" << endl;
 125
 126         running = false;
 127         reader.join_all();
 128
 129         BOOST_REQUIRE_EQUAL(received_nodes, writer_threads * nodes_per_thread);
 130         BOOST_REQUIRE_EQUAL(ringbuffer_cnt, 0);
 131         BOOST_REQUIRE(sf.empty());
 132         BOOST_REQUIRE(working_set.count_nodes() == 0);
 133     }
 134 };
 135
 136
 137
 138 BOOST_AUTO_TEST_CASE( ringbuffer_test_caching )
 139 {
 140     ringbuffer_tester test1;
 141     test1.run();
 142 }