folly/test/DeterministicSchedule.cpp

   1 /*
   2  * Copyright 2016 Facebook, Inc.
   3  *
   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
   7  *
   8  *   http://www.apache.org/licenses/LICENSE-2.0
   9  *
  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.
  15  */
  16
  17 #include <folly/test/DeterministicSchedule.h>
  18
  19 #include <assert.h>
  20
  21 #include <algorithm>
  22 #include <list>
  23 #include <mutex>
  24 #include <random>
  25 #include <unordered_map>
  26 #include <utility>
  27
  28 #include <folly/Random.h>
  29
  30 namespace folly {
  31 namespace test {
  32
  33 FOLLY_TLS sem_t* DeterministicSchedule::tls_sem;
  34 FOLLY_TLS DeterministicSchedule* DeterministicSchedule::tls_sched;
  35 FOLLY_TLS unsigned DeterministicSchedule::tls_threadId;
  36 FOLLY_TLS std::function<void(uint64_t, bool)>* DeterministicSchedule::tls_aux;
  37
  38 // access is protected by futexLock
  39 static std::unordered_map<detail::Futex<DeterministicAtomic>*,
  40                           std::list<std::pair<uint32_t, bool*>>> futexQueues;
  41
  42 static std::mutex futexLock;
  43
  44 DeterministicSchedule::DeterministicSchedule(
  45     const std::function<int(int)>& scheduler)
  46     : scheduler_(scheduler), nextThreadId_(1), step_(0) {
  47   assert(tls_sem == nullptr);
  48   assert(tls_sched == nullptr);
  49   assert(tls_aux == nullptr);
  50
  51   tls_sem = new sem_t;
  52   sem_init(tls_sem, 0, 1);
  53   sems_.push_back(tls_sem);
  54
  55   tls_sched = this;
  56 }
  57
  58 DeterministicSchedule::~DeterministicSchedule() {
  59   assert(tls_sched == this);
  60   assert(sems_.size() == 1);
  61   assert(sems_[0] == tls_sem);
  62   beforeThreadExit();
  63 }
  64
  65 std::function<int(int)> DeterministicSchedule::uniform(long seed) {
  66   auto rand = std::make_shared<std::ranlux48>(seed);
  67   return [rand](size_t numActive) {
  68     auto dist = std::uniform_int_distribution<int>(0, numActive - 1);
  69     return dist(*rand);
  70   };
  71 }
  72
  73 struct UniformSubset {
  74   UniformSubset(long seed, int subsetSize, int stepsBetweenSelect)
  75       : uniform_(DeterministicSchedule::uniform(seed)),
  76         subsetSize_(subsetSize),
  77         stepsBetweenSelect_(stepsBetweenSelect),
  78         stepsLeft_(0) {}
  79
  80   size_t operator()(size_t numActive) {
  81     adjustPermSize(numActive);
  82     if (stepsLeft_-- == 0) {
  83       stepsLeft_ = stepsBetweenSelect_ - 1;
  84       shufflePrefix();
  85     }
  86     return perm_[uniform_(std::min(numActive, subsetSize_))];
  87   }
  88
  89  private:
  90   std::function<int(int)> uniform_;
  91   const size_t subsetSize_;
  92   const int stepsBetweenSelect_;
  93
  94   int stepsLeft_;
  95   // only the first subsetSize_ is properly randomized
  96   std::vector<int> perm_;
  97
  98   void adjustPermSize(size_t numActive) {
  99     if (perm_.size() > numActive) {
 100       perm_.erase(std::remove_if(perm_.begin(),
 101                                  perm_.end(),
 102                                  [=](size_t x) { return x >= numActive; }),
 103                   perm_.end());
 104     } else {
 105       while (perm_.size() < numActive) {
 106         perm_.push_back(perm_.size());
 107       }
 108     }
 109     assert(perm_.size() == numActive);
 110   }
 111
 112   void shufflePrefix() {
 113     for (size_t i = 0; i < std::min(perm_.size() - 1, subsetSize_); ++i) {
 114       int j = uniform_(perm_.size() - i) + i;
 115       std::swap(perm_[i], perm_[j]);
 116     }
 117   }
 118 };
 119
 120 std::function<int(int)> DeterministicSchedule::uniformSubset(long seed,
 121                                                              int n,
 122                                                              int m) {
 123   auto gen = std::make_shared<UniformSubset>(seed, n, m);
 124   return [=](size_t numActive) { return (*gen)(numActive); };
 125 }
 126
 127 void DeterministicSchedule::beforeSharedAccess() {
 128   if (tls_sem) {
 129     sem_wait(tls_sem);
 130   }
 131 }
 132
 133 void DeterministicSchedule::afterSharedAccess() {
 134   auto sched = tls_sched;
 135   if (!sched) {
 136     return;
 137   }
 138   sem_post(sched->sems_[sched->scheduler_(sched->sems_.size())]);
 139 }
 140
 141 void DeterministicSchedule::afterSharedAccess(bool success) {
 142   auto sched = tls_sched;
 143   if (!sched) {
 144     return;
 145   }
 146   sched->callAux(success);
 147   sem_post(sched->sems_[sched->scheduler_(sched->sems_.size())]);
 148 }
 149
 150 int DeterministicSchedule::getRandNumber(int n) {
 151   if (tls_sched) {
 152     return tls_sched->scheduler_(n);
 153   }
 154   return Random::rand32() % n;
 155 }
 156
 157 int DeterministicSchedule::getcpu(unsigned* cpu,
 158                                   unsigned* node,
 159                                   void* /* unused */) {
 160   if (!tls_threadId && tls_sched) {
 161     beforeSharedAccess();
 162     tls_threadId = tls_sched->nextThreadId_++;
 163     afterSharedAccess();
 164   }
 165   if (cpu) {
 166     *cpu = tls_threadId;
 167   }
 168   if (node) {
 169     *node = tls_threadId;
 170   }
 171   return 0;
 172 }
 173
 174 void DeterministicSchedule::setAux(std::function<void(uint64_t, bool)>& aux) {
 175   tls_aux = &aux;
 176 }
 177
 178 sem_t* DeterministicSchedule::beforeThreadCreate() {
 179   sem_t* s = new sem_t;
 180   sem_init(s, 0, 0);
 181   beforeSharedAccess();
 182   sems_.push_back(s);
 183   afterSharedAccess();
 184   return s;
 185 }
 186
 187 void DeterministicSchedule::afterThreadCreate(sem_t* sem) {
 188   assert(tls_sem == nullptr);
 189   assert(tls_sched == nullptr);
 190   tls_sem = sem;
 191   tls_sched = this;
 192   bool started = false;
 193   while (!started) {
 194     beforeSharedAccess();
 195     if (active_.count(std::this_thread::get_id()) == 1) {
 196       started = true;
 197     }
 198     afterSharedAccess();
 199   }
 200 }
 201
 202 void DeterministicSchedule::beforeThreadExit() {
 203   assert(tls_sched == this);
 204   beforeSharedAccess();
 205   sems_.erase(std::find(sems_.begin(), sems_.end(), tls_sem));
 206   active_.erase(std::this_thread::get_id());
 207   if (sems_.size() > 0) {
 208     FOLLY_TEST_DSCHED_VLOG("exiting");
 209     afterSharedAccess();
 210   }
 211   sem_destroy(tls_sem);
 212   delete tls_sem;
 213   tls_sem = nullptr;
 214   tls_sched = nullptr;
 215 }
 216
 217 void DeterministicSchedule::join(std::thread& child) {
 218   auto sched = tls_sched;
 219   if (sched) {
 220     bool done = false;
 221     while (!done) {
 222       beforeSharedAccess();
 223       done = !sched->active_.count(child.get_id());
 224       if (done) {
 225         FOLLY_TEST_DSCHED_VLOG("joined " << std::hex << child.get_id());
 226       }
 227       afterSharedAccess();
 228     }
 229   }
 230   child.join();
 231 }
 232
 233 void DeterministicSchedule::callAux(bool success) {
 234   ++step_;
 235   auto aux = tls_aux;
 236   if (!aux) {
 237     return;
 238   }
 239   (*aux)(step_, success);
 240   tls_aux = nullptr;
 241 }
 242
 243 void DeterministicSchedule::post(sem_t* sem) {
 244   beforeSharedAccess();
 245   sem_post(sem);
 246   FOLLY_TEST_DSCHED_VLOG("sem_post(" << sem << ")");
 247   afterSharedAccess();
 248 }
 249
 250 bool DeterministicSchedule::tryWait(sem_t* sem) {
 251   beforeSharedAccess();
 252   int rv = sem_trywait(sem);
 253   int e = rv == 0 ? 0 : errno;
 254   FOLLY_TEST_DSCHED_VLOG("sem_trywait(" << sem << ") = " << rv
 255                                         << " errno=" << e);
 256   afterSharedAccess();
 257   if (rv == 0) {
 258     return true;
 259   } else {
 260     assert(e == EAGAIN);
 261     return false;
 262   }
 263 }
 264
 265 void DeterministicSchedule::wait(sem_t* sem) {
 266   while (!tryWait(sem)) {
 267     // we're not busy waiting because this is a deterministic schedule
 268   }
 269 }
 270 }
 271 }
 272
 273 namespace folly {
 274 namespace detail {
 275
 276 using namespace test;
 277 using namespace std::chrono;
 278
 279 template <>
 280 FutexResult Futex<DeterministicAtomic>::futexWaitImpl(
 281     uint32_t expected,
 282     time_point<system_clock>* absSystemTimeout,
 283     time_point<steady_clock>* absSteadyTimeout,
 284     uint32_t waitMask) {
 285   bool hasTimeout = absSystemTimeout != nullptr || absSteadyTimeout != nullptr;
 286   bool awoken = false;
 287   FutexResult result = FutexResult::AWOKEN;
 288
 289   DeterministicSchedule::beforeSharedAccess();
 290   FOLLY_TEST_DSCHED_VLOG(this << ".futexWait(" << std::hex << expected
 291                               << ", .., " << std::hex << waitMask
 292                               << ") beginning..");
 293   futexLock.lock();
 294   if (data == expected) {
 295     auto& queue = futexQueues[this];
 296     queue.emplace_back(waitMask, &awoken);
 297     auto ours = queue.end();
 298     ours--;
 299     while (!awoken) {
 300       futexLock.unlock();
 301       DeterministicSchedule::afterSharedAccess();
 302       DeterministicSchedule::beforeSharedAccess();
 303       futexLock.lock();
 304
 305       // Simulate spurious wake-ups, timeouts each time with
 306       // a 10% probability if we haven't been woken up already
 307       if (!awoken && hasTimeout &&
 308           DeterministicSchedule::getRandNumber(100) < 10) {
 309         assert(futexQueues.count(this) != 0 && &futexQueues[this] == &queue);
 310         queue.erase(ours);
 311         if (queue.empty()) {
 312           futexQueues.erase(this);
 313         }
 314         // Simulate ETIMEDOUT 90% of the time and other failures
 315         // remaining time
 316         result = DeterministicSchedule::getRandNumber(100) >= 10
 317                      ? FutexResult::TIMEDOUT
 318                      : FutexResult::INTERRUPTED;
 319         break;
 320       }
 321     }
 322   } else {
 323     result = FutexResult::VALUE_CHANGED;
 324   }
 325   futexLock.unlock();
 326
 327   char const* resultStr = "?";
 328   switch (result) {
 329     case FutexResult::AWOKEN:
 330       resultStr = "AWOKEN";
 331       break;
 332     case FutexResult::TIMEDOUT:
 333       resultStr = "TIMEDOUT";
 334       break;
 335     case FutexResult::INTERRUPTED:
 336       resultStr = "INTERRUPTED";
 337       break;
 338     case FutexResult::VALUE_CHANGED:
 339       resultStr = "VALUE_CHANGED";
 340       break;
 341   }
 342   FOLLY_TEST_DSCHED_VLOG(this << ".futexWait(" << std::hex << expected
 343                               << ", .., " << std::hex << waitMask << ") -> "
 344                               << resultStr);
 345   DeterministicSchedule::afterSharedAccess();
 346   return result;
 347 }
 348
 349 template <>
 350 int Futex<DeterministicAtomic>::futexWake(int count, uint32_t wakeMask) {
 351   int rv = 0;
 352   DeterministicSchedule::beforeSharedAccess();
 353   futexLock.lock();
 354   if (futexQueues.count(this) > 0) {
 355     auto& queue = futexQueues[this];
 356     auto iter = queue.begin();
 357     while (iter != queue.end() && rv < count) {
 358       auto cur = iter++;
 359       if ((cur->first & wakeMask) != 0) {
 360         *(cur->second) = true;
 361         rv++;
 362         queue.erase(cur);
 363       }
 364     }
 365     if (queue.empty()) {
 366       futexQueues.erase(this);
 367     }
 368   }
 369   futexLock.unlock();
 370   FOLLY_TEST_DSCHED_VLOG(this << ".futexWake(" << count << ", " << std::hex
 371                               << wakeMask << ") -> " << rv);
 372   DeterministicSchedule::afterSharedAccess();
 373   return rv;
 374 }
 375
 376 template <>
 377 CacheLocality const& CacheLocality::system<test::DeterministicAtomic>() {
 378   static CacheLocality cache(CacheLocality::uniform(16));
 379   return cache;
 380 }
 381
 382 template <>
 383 Getcpu::Func AccessSpreader<test::DeterministicAtomic>::pickGetcpuFunc() {
 384   return &DeterministicSchedule::getcpu;
 385 }
 386 }
 387 }