2 * Copyright 2013 Facebook, Inc.
4 * Licensed under the Apache License, Version 2.0 (the "License");
5 * you may not use this file except in compliance with the License.
6 * You may obtain a copy of the License at
8 * http://www.apache.org/licenses/LICENSE-2.0
10 * Unless required by applicable law or agreed to in writing, software
11 * distributed under the License is distributed on an "AS IS" BASIS,
12 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 * See the License for the specific language governing permissions and
14 * limitations under the License.
17 #include "DeterministicSchedule.h"
23 #include <unordered_map>
26 namespace folly { namespace test {
28 __thread sem_t* DeterministicSchedule::tls_sem;
29 __thread DeterministicSchedule* DeterministicSchedule::tls_sched;
31 // access is protected by futexLock
32 static std::unordered_map<detail::Futex<DeterministicAtomic>*,
33 std::list<std::pair<uint32_t,bool*>>> futexQueues;
35 static std::mutex futexLock;
37 DeterministicSchedule::DeterministicSchedule(
38 const std::function<int(int)>& scheduler)
39 : scheduler_(scheduler)
41 assert(tls_sem == nullptr);
42 assert(tls_sched == nullptr);
45 sem_init(tls_sem, 0, 1);
46 sems_.push_back(tls_sem);
51 DeterministicSchedule::~DeterministicSchedule() {
52 assert(tls_sched == this);
53 assert(sems_.size() == 1);
54 assert(sems_[0] == tls_sem);
58 std::function<int(int)>
59 DeterministicSchedule::uniform(long seed) {
60 auto rand = std::make_shared<std::ranlux48>(seed);
61 return [rand](int numActive) {
62 auto dist = std::uniform_int_distribution<int>(0, numActive - 1);
67 struct UniformSubset {
68 UniformSubset(long seed, int subsetSize, int stepsBetweenSelect)
69 : uniform_(DeterministicSchedule::uniform(seed))
70 , subsetSize_(subsetSize)
71 , stepsBetweenSelect_(stepsBetweenSelect)
76 int operator()(int numActive) {
77 adjustPermSize(numActive);
78 if (stepsLeft_-- == 0) {
79 stepsLeft_ = stepsBetweenSelect_ - 1;
82 return perm_[uniform_(std::min(numActive, subsetSize_))];
86 std::function<int(int)> uniform_;
87 const int subsetSize_;
88 const int stepsBetweenSelect_;
91 // only the first subsetSize_ is properly randomized
92 std::vector<int> perm_;
94 void adjustPermSize(int numActive) {
95 if (perm_.size() > numActive) {
96 perm_.erase(std::remove_if(perm_.begin(), perm_.end(),
97 [=](int x){ return x >= numActive; }), perm_.end());
99 while (perm_.size() < numActive) {
100 perm_.push_back(perm_.size());
103 assert(perm_.size() == numActive);
106 void shufflePrefix() {
107 for (int i = 0; i < std::min(int(perm_.size() - 1), subsetSize_); ++i) {
108 int j = uniform_(perm_.size() - i) + i;
109 std::swap(perm_[i], perm_[j]);
114 std::function<int(int)>
115 DeterministicSchedule::uniformSubset(long seed, int n, int m) {
116 auto gen = std::make_shared<UniformSubset>(seed, n, m);
117 return [=](int numActive) { return (*gen)(numActive); };
121 DeterministicSchedule::beforeSharedAccess() {
128 DeterministicSchedule::afterSharedAccess() {
129 auto sched = tls_sched;
134 sem_post(sched->sems_[sched->scheduler_(sched->sems_.size())]);
138 DeterministicSchedule::beforeThreadCreate() {
139 sem_t* s = new sem_t;
141 beforeSharedAccess();
148 DeterministicSchedule::afterThreadCreate(sem_t* sem) {
149 assert(tls_sem == nullptr);
150 assert(tls_sched == nullptr);
153 bool started = false;
155 beforeSharedAccess();
156 if (active_.count(std::this_thread::get_id()) == 1) {
164 DeterministicSchedule::beforeThreadExit() {
165 assert(tls_sched == this);
166 beforeSharedAccess();
167 sems_.erase(std::find(sems_.begin(), sems_.end(), tls_sem));
168 active_.erase(std::this_thread::get_id());
169 if (sems_.size() > 0) {
172 sem_destroy(tls_sem);
179 DeterministicSchedule::join(std::thread& child) {
180 auto sched = tls_sched;
184 beforeSharedAccess();
185 done = !sched->active_.count(child.get_id());
193 DeterministicSchedule::post(sem_t* sem) {
194 beforeSharedAccess();
200 DeterministicSchedule::tryWait(sem_t* sem) {
201 beforeSharedAccess();
202 int rv = sem_trywait(sem);
207 assert(errno == EAGAIN);
213 DeterministicSchedule::wait(sem_t* sem) {
214 while (!tryWait(sem)) {
215 // we're not busy waiting because this is a deterministic schedule
221 namespace folly { namespace detail {
223 using namespace test;
226 bool Futex<DeterministicAtomic>::futexWait(uint32_t expected,
229 DeterministicSchedule::beforeSharedAccess();
231 if (data != expected) {
234 auto& queue = futexQueues[this];
236 queue.push_back(std::make_pair(waitMask, &done));
239 DeterministicSchedule::afterSharedAccess();
240 DeterministicSchedule::beforeSharedAccess();
246 DeterministicSchedule::afterSharedAccess();
251 int Futex<DeterministicAtomic>::futexWake(int count, uint32_t wakeMask) {
253 DeterministicSchedule::beforeSharedAccess();
255 if (futexQueues.count(this) > 0) {
256 auto& queue = futexQueues[this];
257 auto iter = queue.begin();
258 while (iter != queue.end() && rv < count) {
260 if ((cur->first & wakeMask) != 0) {
261 *(cur->second) = true;
267 futexQueues.erase(this);
271 DeterministicSchedule::afterSharedAccess();
277 CacheLocality const& CacheLocality::system<test::DeterministicAtomic>() {
278 static CacheLocality cache(CacheLocality::uniform(16));
283 test::DeterministicAtomic<size_t>
284 SequentialThreadId<test::DeterministicAtomic>::prevId(0);
287 __thread size_t SequentialThreadId<test::DeterministicAtomic>::currentId(0);
290 const AccessSpreader<test::DeterministicAtomic>
291 AccessSpreader<test::DeterministicAtomic>::stripeByCore(
292 CacheLocality::system<>().numCachesByLevel.front());
295 const AccessSpreader<test::DeterministicAtomic>
296 AccessSpreader<test::DeterministicAtomic>::stripeByChip(
297 CacheLocality::system<>().numCachesByLevel.back());
300 AccessSpreaderArray<test::DeterministicAtomic,128>
301 AccessSpreaderArray<test::DeterministicAtomic,128>::sharedInstance = {};
306 AccessSpreader<test::DeterministicAtomic>::pickGetcpuFunc(size_t numStripes) {
307 return &SequentialThreadId<test::DeterministicAtomic>::getcpu;