folly/futures/test/BarrierTest.cpp

   1 /*
   2  * Copyright 2015 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/futures/Barrier.h>
  18
  19 #include <atomic>
  20 #include <condition_variable>
  21 #include <mutex>
  22
  23 #include <folly/Random.h>
  24 #include <glog/logging.h>
  25 #include <gtest/gtest.h>
  26
  27 DEFINE_int32(seed, 0, "Random seed");
  28
  29 namespace folly { namespace futures { namespace test {
  30
  31 TEST(BarrierTest, Simple) {
  32   constexpr uint32_t numThreads = 10;
  33
  34   std::mutex mutex;
  35   std::condition_variable b1DoneCond;
  36   std::condition_variable b2DoneCond;
  37   std::atomic<uint32_t> b1TrueSeen(0);
  38   std::atomic<uint32_t> b1Passed(0);
  39   std::atomic<uint32_t> b2TrueSeen(0);
  40   std::atomic<uint32_t> b2Passed(0);
  41
  42   Barrier barrier(numThreads + 1);
  43
  44   std::vector<std::thread> threads;
  45   threads.reserve(numThreads);
  46   for (uint32_t i = 0; i < numThreads; ++i) {
  47     threads.emplace_back([&] () {
  48       barrier.wait()
  49         .then(
  50             [&] (bool v) {
  51               std::unique_lock<std::mutex> lock(mutex);
  52               b1TrueSeen += uint32_t(v);
  53               if (++b1Passed == numThreads) {
  54                 b1DoneCond.notify_one();
  55               }
  56               return barrier.wait();
  57             })
  58         .then(
  59             [&] (bool v) {
  60               std::unique_lock<std::mutex> lock(mutex);
  61               b2TrueSeen += uint32_t(v);
  62               if (++b2Passed == numThreads) {
  63                 b2DoneCond.notify_one();
  64               }
  65             })
  66         .get();
  67     });
  68   }
  69
  70   /* sleep override */
  71   std::this_thread::sleep_for(std::chrono::milliseconds(50));
  72   EXPECT_EQ(0, b1Passed);
  73   EXPECT_EQ(0, b1TrueSeen);
  74
  75   b1TrueSeen += barrier.wait().get();
  76
  77   {
  78     std::unique_lock<std::mutex> lock(mutex);
  79     while (b1Passed != numThreads) {
  80       b1DoneCond.wait(lock);
  81     }
  82     EXPECT_EQ(1, b1TrueSeen);
  83   }
  84
  85   /* sleep override */
  86   std::this_thread::sleep_for(std::chrono::milliseconds(50));
  87   EXPECT_EQ(0, b2Passed);
  88   EXPECT_EQ(0, b2TrueSeen);
  89
  90   b2TrueSeen += barrier.wait().get();
  91
  92   {
  93     std::unique_lock<std::mutex> lock(mutex);
  94     while (b2Passed != numThreads) {
  95       b2DoneCond.wait(lock);
  96     }
  97     EXPECT_EQ(1, b2TrueSeen);
  98   }
  99
 100   for (auto& t : threads) {
 101     t.join();
 102   }
 103 }
 104
 105 TEST(BarrierTest, Random) {
 106   // Create numThreads threads.
 107   //
 108   // Each thread repeats the following numIterations times:
 109   //   - grab a randomly chosen number of futures from the barrier, waiting
 110   //     for a short random time between each
 111   //   - wait for all futures to complete
 112   //   - record whether the one future returning true was seen among them
 113   //
 114   // At the end, we verify that exactly one future returning true was seen
 115   // for each iteration.
 116   constexpr uint32_t numIterations = 1;
 117   auto numThreads = folly::Random::rand32(30, 91);
 118
 119   struct ThreadInfo {
 120     ThreadInfo() { }
 121     std::thread thread;
 122     uint32_t iteration = 0;
 123     uint32_t numFutures;
 124     std::vector<uint32_t> trueSeen;
 125   };
 126
 127   std::vector<ThreadInfo> threads;
 128   threads.resize(numThreads);
 129
 130   uint32_t totalFutures = 0;
 131   for (auto& tinfo : threads) {
 132     tinfo.numFutures = folly::Random::rand32(100);
 133     tinfo.trueSeen.resize(numIterations);
 134     totalFutures += tinfo.numFutures;
 135   }
 136
 137   Barrier barrier(totalFutures);
 138
 139   for (auto& tinfo : threads) {
 140     auto pinfo = &tinfo;
 141     tinfo.thread = std::thread(
 142         [numIterations, pinfo, &barrier] () {
 143           std::vector<folly::Future<bool>> futures;
 144           futures.reserve(pinfo->numFutures);
 145           for (uint32_t i = 0; i < numIterations; ++i, ++pinfo->iteration) {
 146             futures.clear();
 147             for (uint32_t j = 0; j < pinfo->numFutures; ++j) {
 148               futures.push_back(barrier.wait());
 149               auto nanos = folly::Random::rand32(10 * 1000 * 1000);
 150               /* sleep override */
 151               std::this_thread::sleep_for(std::chrono::nanoseconds(nanos));
 152             }
 153             auto results = folly::collect(futures).get();
 154             pinfo->trueSeen[i] =
 155               std::count(results.begin(), results.end(), true);
 156           }
 157         });
 158   }
 159
 160   for (auto& tinfo : threads) {
 161     tinfo.thread.join();
 162     EXPECT_EQ(numIterations, tinfo.iteration);
 163   }
 164
 165   for (uint32_t i = 0; i < numIterations; ++i) {
 166     uint32_t trueCount = 0;
 167     for (auto& tinfo : threads) {
 168       trueCount += tinfo.trueSeen[i];
 169     }
 170     EXPECT_EQ(1, trueCount);
 171   }
 172 }
 173
 174 }}}  // namespaces