folly/test/SynchronizedTestLib-inl.h

   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 #pragma once
  18
  19 #include <gtest/gtest.h>
  20
  21 #include <folly/Foreach.h>
  22 #include <folly/Random.h>
  23 #include <folly/Synchronized.h>
  24 #include <glog/logging.h>
  25 #include <algorithm>
  26 #include <condition_variable>
  27 #include <functional>
  28 #include <map>
  29 #include <random>
  30 #include <thread>
  31 #include <vector>
  32
  33 namespace folly {
  34 namespace sync_tests {
  35
  36 inline std::mt19937& getRNG() {
  37   static const auto seed = folly::randomNumberSeed();
  38   static std::mt19937 rng(seed);
  39   return rng;
  40 }
  41
  42 void randomSleep(std::chrono::milliseconds min, std::chrono::milliseconds max) {
  43   std::uniform_int_distribution<> range(min.count(), max.count());
  44   std::chrono::milliseconds duration(range(getRNG()));
  45   std::this_thread::sleep_for(duration);
  46 }
  47
  48 /*
  49  * Run a functon simultaneously in a number of different threads.
  50  *
  51  * The function will be passed the index number of the thread it is running in.
  52  * This function makes an attempt to synchronize the start of the threads as
  53  * best as possible.  It waits for all threads to be allocated and started
  54  * before invoking the function.
  55  */
  56 template <class Function>
  57 void runParallel(size_t numThreads, const Function& function) {
  58   std::vector<std::thread> threads;
  59   threads.reserve(numThreads);
  60
  61   // Variables used to synchronize all threads to try and start them
  62   // as close to the same time as possible
  63   //
  64   // TODO: At the moment Synchronized doesn't work with condition variables.
  65   // Update this to use Synchronized once the condition_variable support lands.
  66   std::mutex threadsReadyMutex;
  67   size_t threadsReady = 0;
  68   std::condition_variable readyCV;
  69   std::mutex goMutex;
  70   bool go = false;
  71   std::condition_variable goCV;
  72
  73   auto worker = [&](size_t threadIndex) {
  74     // Signal that we are ready
  75     {
  76       std::lock_guard<std::mutex> lock(threadsReadyMutex);
  77       ++threadsReady;
  78     }
  79     readyCV.notify_one();
  80
  81     // Wait until we are given the signal to start
  82     // The purpose of this is to try and make sure all threads start
  83     // as close to the same time as possible.
  84     {
  85       std::unique_lock<std::mutex> lock(goMutex);
  86       goCV.wait(lock, [&] { return go; });
  87     }
  88
  89     function(threadIndex);
  90   };
  91
  92   // Start all of the threads
  93   for (size_t threadIndex = 0; threadIndex < numThreads; ++threadIndex) {
  94     threads.emplace_back([threadIndex, &worker]() { worker(threadIndex); });
  95   }
  96
  97   // Wait for all threads to become ready
  98   {
  99     std::unique_lock<std::mutex> lock(threadsReadyMutex);
 100     readyCV.wait(lock, [&] { return threadsReady == numThreads; });
 101   }
 102   {
 103     std::lock_guard<std::mutex> lock(goMutex);
 104     go = true;
 105   }
 106   // Now signal the threads that they can go
 107   goCV.notify_all();
 108
 109   // Wait for all threads to finish
 110   for (auto& thread : threads) {
 111     thread.join();
 112   }
 113 }
 114
 115 template <class Mutex>
 116 void testBasic() {
 117   folly::Synchronized<std::vector<int>, Mutex> obj;
 118
 119   obj->resize(1000);
 120
 121   auto obj2 = obj;
 122   EXPECT_EQ(1000, obj2->size());
 123
 124   SYNCHRONIZED (obj) {
 125     obj.push_back(10);
 126     EXPECT_EQ(1001, obj.size());
 127     EXPECT_EQ(10, obj.back());
 128     EXPECT_EQ(1000, obj2->size());
 129
 130     UNSYNCHRONIZED(obj) {
 131       EXPECT_EQ(1001, obj->size());
 132     }
 133   }
 134
 135   SYNCHRONIZED_CONST (obj) {
 136     EXPECT_EQ(1001, obj.size());
 137     UNSYNCHRONIZED(obj) {
 138       EXPECT_EQ(1001, obj->size());
 139     }
 140   }
 141
 142   SYNCHRONIZED (lockedObj, *&obj) {
 143     lockedObj.front() = 2;
 144   }
 145
 146   EXPECT_EQ(1001, obj->size());
 147   EXPECT_EQ(10, obj->back());
 148   EXPECT_EQ(1000, obj2->size());
 149
 150   EXPECT_EQ(FB_ARG_2_OR_1(1, 2), 2);
 151   EXPECT_EQ(FB_ARG_2_OR_1(1), 1);
 152 }
 153
 154 template <class Mutex> void testConcurrency() {
 155   folly::Synchronized<std::vector<int>, Mutex> v;
 156   static const size_t numThreads = 100;
 157   // Note: I initially tried using itersPerThread = 1000,
 158   // which works fine for most lock types, but std::shared_timed_mutex
 159   // appears to be extraordinarily slow.  It could take around 30 seconds
 160   // to run this test with 1000 iterations per thread using shared_timed_mutex.
 161   static const size_t itersPerThread = 100;
 162
 163   auto pushNumbers = [&](size_t threadIdx) {
 164     // Test lock()
 165     for (size_t n = 0; n < itersPerThread; ++n) {
 166       v->push_back((itersPerThread * threadIdx) + n);
 167       sched_yield();
 168     }
 169   };
 170   runParallel(numThreads, pushNumbers);
 171
 172   std::vector<int> result;
 173   v.swap(result);
 174
 175   EXPECT_EQ(numThreads * itersPerThread, result.size());
 176   sort(result.begin(), result.end());
 177
 178   for (size_t i = 0; i < itersPerThread * numThreads; ++i) {
 179     EXPECT_EQ(i, result[i]);
 180   }
 181 }
 182
 183 template <class Mutex> void testDualLocking() {
 184   folly::Synchronized<std::vector<int>, Mutex> v;
 185   folly::Synchronized<std::map<int, int>, Mutex> m;
 186
 187   auto dualLockWorker = [&](size_t threadIdx) {
 188     if (threadIdx & 1) {
 189       SYNCHRONIZED_DUAL(lv, v, lm, m) {
 190         lv.push_back(threadIdx);
 191         lm[threadIdx] = threadIdx + 1;
 192       }
 193     } else {
 194       SYNCHRONIZED_DUAL(lm, m, lv, v) {
 195         lv.push_back(threadIdx);
 196         lm[threadIdx] = threadIdx + 1;
 197       }
 198     }
 199   };
 200   static const size_t numThreads = 100;
 201   runParallel(numThreads, dualLockWorker);
 202
 203   std::vector<int> result;
 204   v.swap(result);
 205
 206   EXPECT_EQ(numThreads, result.size());
 207   sort(result.begin(), result.end());
 208
 209   for (size_t i = 0; i < numThreads; ++i) {
 210     EXPECT_EQ(i, result[i]);
 211   }
 212 }
 213
 214 template <class Mutex> void testDualLockingWithConst() {
 215   folly::Synchronized<std::vector<int>, Mutex> v;
 216   folly::Synchronized<std::map<int, int>, Mutex> m;
 217
 218   auto dualLockWorker = [&](size_t threadIdx) {
 219     const auto& cm = m;
 220     if (threadIdx & 1) {
 221       SYNCHRONIZED_DUAL(lv, v, lm, cm) {
 222         (void)lm.size();
 223         lv.push_back(threadIdx);
 224       }
 225     } else {
 226       SYNCHRONIZED_DUAL(lm, cm, lv, v) {
 227         (void)lm.size();
 228         lv.push_back(threadIdx);
 229       }
 230     }
 231   };
 232   static const size_t numThreads = 100;
 233   runParallel(numThreads, dualLockWorker);
 234
 235   std::vector<int> result;
 236   v.swap(result);
 237
 238   EXPECT_EQ(numThreads, result.size());
 239   sort(result.begin(), result.end());
 240
 241   for (size_t i = 0; i < numThreads; ++i) {
 242     EXPECT_EQ(i, result[i]);
 243   }
 244 }
 245
 246 template <class Mutex> void testTimedSynchronized() {
 247   folly::Synchronized<std::vector<int>, Mutex> v;
 248   folly::Synchronized<uint64_t, Mutex> numTimeouts;
 249
 250   auto worker = [&](size_t threadIdx) {
 251     // Test operator->
 252     v->push_back(2 * threadIdx);
 253
 254     // Aaand test the TIMED_SYNCHRONIZED macro
 255     for (;;)
 256       TIMED_SYNCHRONIZED(5, lv, v) {
 257         if (lv) {
 258           // Sleep for a random time to ensure we trigger timeouts
 259           // in other threads
 260           randomSleep(
 261               std::chrono::milliseconds(5), std::chrono::milliseconds(15));
 262           lv->push_back(2 * threadIdx + 1);
 263           return;
 264         }
 265
 266         SYNCHRONIZED(numTimeouts) {
 267           ++numTimeouts;
 268         }
 269       }
 270   };
 271
 272   static const size_t numThreads = 100;
 273   runParallel(numThreads, worker);
 274
 275   std::vector<int> result;
 276   v.swap(result);
 277
 278   EXPECT_EQ(2 * numThreads, result.size());
 279   sort(result.begin(), result.end());
 280
 281   for (size_t i = 0; i < 2 * numThreads; ++i) {
 282     EXPECT_EQ(i, result[i]);
 283   }
 284   // We generally expect a large number of number timeouts here.
 285   // I'm not adding a check for it since it's theoretically possible that
 286   // we might get 0 timeouts depending on the CPU scheduling if our threads
 287   // don't get to run very often.
 288   uint64_t finalNumTimeouts = 0;
 289   SYNCHRONIZED(numTimeouts) {
 290     finalNumTimeouts = numTimeouts;
 291   }
 292   LOG(INFO) << "testTimedSynchronized: " << finalNumTimeouts << " timeouts";
 293 }
 294
 295 template <class Mutex> void testTimedSynchronizedWithConst() {
 296   folly::Synchronized<std::vector<int>, Mutex> v;
 297   folly::Synchronized<uint64_t, Mutex> numTimeouts;
 298
 299   auto worker = [&](size_t threadIdx) {
 300     // Test operator->
 301     v->push_back(threadIdx);
 302
 303     // Test TIMED_SYNCHRONIZED_CONST
 304     for (;;) {
 305       TIMED_SYNCHRONIZED_CONST(10, lv, v) {
 306         if (lv) {
 307           // Sleep while holding the lock.
 308           //
 309           // This will block other threads from acquiring the write lock to add
 310           // their thread index to v, but it won't block threads that have
 311           // entered the for loop and are trying to acquire a read lock.
 312           //
 313           // For lock types that give preference to readers rather than writers,
 314           // this will tend to serialize all threads on the wlock() above.
 315           randomSleep(
 316               std::chrono::milliseconds(5), std::chrono::milliseconds(15));
 317           auto found = std::find(lv->begin(), lv->end(), threadIdx);
 318           CHECK(found != lv->end());
 319           return;
 320         } else {
 321           SYNCHRONIZED(numTimeouts) {
 322             ++numTimeouts;
 323           }
 324         }
 325       }
 326     }
 327   };
 328
 329   static const size_t numThreads = 100;
 330   runParallel(numThreads, worker);
 331
 332   std::vector<int> result;
 333   v.swap(result);
 334
 335   EXPECT_EQ(numThreads, result.size());
 336   sort(result.begin(), result.end());
 337
 338   for (size_t i = 0; i < numThreads; ++i) {
 339     EXPECT_EQ(i, result[i]);
 340   }
 341   // We generally expect a small number of timeouts here.
 342   // For locks that give readers preference over writers this should usually
 343   // be 0.  With locks that give writers preference we do see a small-ish
 344   // number of read timeouts.
 345   uint64_t finalNumTimeouts = 0;
 346   SYNCHRONIZED(numTimeouts) {
 347     finalNumTimeouts = numTimeouts;
 348   }
 349   LOG(INFO) << "testTimedSynchronizedWithConst: " << finalNumTimeouts
 350             << " timeouts";
 351 }
 352
 353 template <class Mutex> void testConstCopy() {
 354   std::vector<int> input = {1, 2, 3};
 355   const folly::Synchronized<std::vector<int>, Mutex> v(input);
 356
 357   std::vector<int> result;
 358
 359   v.copy(&result);
 360   EXPECT_EQ(input, result);
 361
 362   result = v.copy();
 363   EXPECT_EQ(input, result);
 364 }
 365
 366 struct NotCopiableNotMovable {
 367   NotCopiableNotMovable(int, const char*) {}
 368   NotCopiableNotMovable(const NotCopiableNotMovable&) = delete;
 369   NotCopiableNotMovable& operator=(const NotCopiableNotMovable&) = delete;
 370   NotCopiableNotMovable(NotCopiableNotMovable&&) = delete;
 371   NotCopiableNotMovable& operator=(NotCopiableNotMovable&&) = delete;
 372 };
 373
 374 template <class Mutex> void testInPlaceConstruction() {
 375   // This won't compile without construct_in_place
 376   folly::Synchronized<NotCopiableNotMovable> a(
 377     folly::construct_in_place, 5, "a"
 378   );
 379 }
 380 }
 381 }