cxx11_eventcount.cpp
Go to the documentation of this file.
1 // This file is part of Eigen, a lightweight C++ template library
2 // for linear algebra.
3 //
4 // Copyright (C) 2016 Dmitry Vyukov <dvyukov@google.com>
5 // Copyright (C) 2016 Benoit Steiner <benoit.steiner.goog@gmail.com>
6 //
7 // This Source Code Form is subject to the terms of the Mozilla
8 // Public License v. 2.0. If a copy of the MPL was not distributed
9 // with this file, You can obtain one at http://mozilla.org/MPL/2.0/.
10 
11 #define EIGEN_USE_THREADS
12 #include "main.h"
13 #include <Eigen/CXX11/ThreadPool>
14 
15 // Visual studio doesn't implement a rand_r() function since its
16 // implementation of rand() is already thread safe
17 int rand_reentrant(unsigned int* s) {
18 #ifdef EIGEN_COMP_MSVC_STRICT
20  return rand();
21 #else
22  return rand_r(s);
23 #endif
24 }
25 
26 static void test_basic_eventcount()
27 {
29  waiters.resize(1);
30  EventCount ec(waiters);
31  EventCount::Waiter& w = waiters[0];
32  ec.Notify(false);
33  ec.Prewait();
34  ec.Notify(true);
35  ec.CommitWait(&w);
36  ec.Prewait();
37  ec.CancelWait();
38 }
39 
40 // Fake bounded counter-based queue.
41 struct TestQueue {
42  std::atomic<int> val_;
43  static const int kQueueSize = 10;
44 
45  TestQueue() : val_() {}
46 
47  ~TestQueue() { VERIFY_IS_EQUAL(val_.load(), 0); }
48 
49  bool Push() {
50  int val = val_.load(std::memory_order_relaxed);
51  for (;;) {
52  VERIFY_GE(val, 0);
53  VERIFY_LE(val, kQueueSize);
54  if (val == kQueueSize) return false;
55  if (val_.compare_exchange_weak(val, val + 1, std::memory_order_relaxed))
56  return true;
57  }
58  }
59 
60  bool Pop() {
61  int val = val_.load(std::memory_order_relaxed);
62  for (;;) {
63  VERIFY_GE(val, 0);
64  VERIFY_LE(val, kQueueSize);
65  if (val == 0) return false;
66  if (val_.compare_exchange_weak(val, val - 1, std::memory_order_relaxed))
67  return true;
68  }
69  }
70 
71  bool Empty() { return val_.load(std::memory_order_relaxed) == 0; }
72 };
73 
74 const int TestQueue::kQueueSize;
75 
76 // A number of producers send messages to a set of consumers using a set of
77 // fake queues. Ensure that it does not crash, consumers don't deadlock and
78 // number of blocked and unblocked threads match.
80 {
81  const int kThreads = std::thread::hardware_concurrency();
82  static const int kEvents = 1 << 16;
83  static const int kQueues = 10;
84 
85  MaxSizeVector<EventCount::Waiter> waiters(kThreads);
86  waiters.resize(kThreads);
87  EventCount ec(waiters);
88  TestQueue queues[kQueues];
89 
90  std::vector<std::unique_ptr<std::thread>> producers;
91  for (int i = 0; i < kThreads; i++) {
92  producers.emplace_back(new std::thread([&ec, &queues]() {
93  unsigned int rnd = static_cast<unsigned int>(std::hash<std::thread::id>()(std::this_thread::get_id()));
94  for (int j = 0; j < kEvents; j++) {
95  unsigned idx = rand_reentrant(&rnd) % kQueues;
96  if (queues[idx].Push()) {
97  ec.Notify(false);
98  continue;
99  }
101  j--;
102  }
103  }));
104  }
105 
106  std::vector<std::unique_ptr<std::thread>> consumers;
107  for (int i = 0; i < kThreads; i++) {
108  consumers.emplace_back(new std::thread([&ec, &queues, &waiters, i]() {
109  EventCount::Waiter& w = waiters[i];
110  unsigned int rnd = static_cast<unsigned int>(std::hash<std::thread::id>()(std::this_thread::get_id()));
111  for (int j = 0; j < kEvents; j++) {
112  unsigned idx = rand_reentrant(&rnd) % kQueues;
113  if (queues[idx].Pop()) continue;
114  j--;
115  ec.Prewait();
116  bool empty = true;
117  for (int q = 0; q < kQueues; q++) {
118  if (!queues[q].Empty()) {
119  empty = false;
120  break;
121  }
122  }
123  if (!empty) {
124  ec.CancelWait();
125  continue;
126  }
127  ec.CommitWait(&w);
128  }
129  }));
130  }
131 
132  for (int i = 0; i < kThreads; i++) {
133  producers[i]->join();
134  consumers[i]->join();
135  }
136 }
137 
138 EIGEN_DECLARE_TEST(cxx11_eventcount)
139 {
142 }
std::atomic< int > val_
#define VERIFY_GE(a, b)
Definition: main.h:382
EIGEN_DECLARE_TEST(cxx11_eventcount)
#define VERIFY_IS_EQUAL(a, b)
Definition: main.h:386
void resize(size_t n)
Definition: MaxSizeVector.h:69
#define EIGEN_THREAD_YIELD()
Definition: ThreadYield.h:17
RealScalar s
static const int kQueueSize
EIGEN_DEVICE_FUNC const Scalar & q
#define VERIFY_LE(a, b)
Definition: main.h:383
RowVector3d w
#define CALL_SUBTEST(FUNC)
Definition: main.h:399
int rand_reentrant(unsigned int *s)
The MaxSizeVector class.
Definition: MaxSizeVector.h:31
static void test_basic_eventcount()
void Notify(bool notifyAll)
Definition: EventCount.h:130
std::ptrdiff_t j
#define EIGEN_UNUSED_VARIABLE(var)
Definition: Macros.h:1076
void CommitWait(Waiter *w)
Definition: EventCount.h:79
static void test_stress_eventcount()


gtsam
Author(s):
autogenerated on Tue Jul 4 2023 02:34:06