folly/wangle/test/ThreadGateTest.cpp

   1 /*
   2  * Copyright 2014 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 <gtest/gtest.h>
  18 #include <thread>
  19 #include <future>
  20
  21 #include "folly/wangle/Executor.h"
  22 #include "folly/wangle/ManualExecutor.h"
  23 #include "folly/wangle/ThreadGate.h"
  24 #include "folly/wangle/GenericThreadGate.h"
  25
  26 using namespace folly::wangle;
  27 using std::make_shared;
  28 using std::shared_ptr;
  29 using std::thread;
  30 using std::vector;
  31
  32 struct ManualWaiter {
  33   explicit ManualWaiter(shared_ptr<ManualExecutor> ex) : ex(ex) {}
  34
  35   void makeProgress() {
  36     ex->wait();
  37     ex->run();
  38   }
  39
  40   shared_ptr<ManualExecutor> ex;
  41 };
  42
  43 struct GenericThreadGateFixture : public testing::Test {
  44   GenericThreadGateFixture() :
  45     westExecutor(new ManualExecutor),
  46     eastExecutor(new ManualExecutor),
  47     waiter(new ManualWaiter(westExecutor)),
  48     tg(westExecutor, eastExecutor, waiter),
  49     done(false)
  50   {
  51     t = thread([=] {
  52       ManualWaiter eastWaiter(eastExecutor);
  53       while (!done)
  54         eastWaiter.makeProgress();
  55     });
  56   }
  57
  58   ~GenericThreadGateFixture() {
  59     done = true;
  60     tg.gate<void>([] { return makeFuture(); });
  61     t.join();
  62   }
  63
  64   shared_ptr<ManualExecutor> westExecutor;
  65   shared_ptr<ManualExecutor> eastExecutor;
  66   shared_ptr<ManualWaiter> waiter;
  67   GenericThreadGate<
  68     shared_ptr<ManualExecutor>,
  69     shared_ptr<ManualExecutor>,
  70     shared_ptr<ManualWaiter>> tg;
  71   bool done;
  72   thread t;
  73 };
  74
  75 TEST_F(GenericThreadGateFixture, gate_and_wait) {
  76   auto f = tg.gate<void>([] { return makeFuture(); });
  77   EXPECT_FALSE(f.isReady());
  78
  79   tg.waitFor(f);
  80   EXPECT_TRUE(f.isReady());
  81 }
  82
  83 TEST_F(GenericThreadGateFixture, gate_many) {
  84   vector<Future<void>> fs;
  85   int n = 10;
  86
  87   for (int i = 0; i < n; i++)
  88     fs.push_back(tg.gate<void>([&] { return makeFuture(); }));
  89
  90   for (auto& f : fs)
  91     EXPECT_FALSE(f.isReady());
  92
  93   auto all = whenAll(fs.begin(), fs.end());
  94   tg.waitFor(all);
  95 }
  96
  97 TEST_F(GenericThreadGateFixture, gate_alternating) {
  98   vector<Promise<void>> ps(10);
  99   vector<Future<void>> fs;
 100   size_t count = 0;
 101
 102   for (auto& p : ps) {
 103     auto* pp = &p;
 104     auto f = tg.gate<void>([=] { return pp->getFuture(); });
 105
 106     // abuse the thread gate to do our dirty work in the other thread
 107     tg.gate<void>([=] { pp->setValue(); return makeFuture(); });
 108
 109     fs.push_back(f.then([&](Try<void>&&) { count++; }));
 110   }
 111
 112   for (auto& f : fs)
 113     EXPECT_FALSE(f.isReady());
 114   EXPECT_EQ(0, count);
 115
 116   auto all = whenAll(fs.begin(), fs.end());
 117   tg.waitFor(all);
 118
 119   EXPECT_EQ(ps.size(), count);
 120 }
 121
 122 TEST(GenericThreadGate, noWaiter) {
 123   auto west = make_shared<ManualExecutor>();
 124   auto east = make_shared<ManualExecutor>();
 125   Promise<void> p;
 126   auto dummyFuture = p.getFuture();
 127
 128   GenericThreadGate<ManualExecutor*, ManualExecutor*>
 129     tg(west.get(), east.get());
 130
 131   EXPECT_THROW(tg.waitFor(dummyFuture), std::logic_error);
 132 }
 133
 134 TEST_F(GenericThreadGateFixture, gate_with_promise) {
 135   Promise<int> p;
 136
 137   auto westId = std::this_thread::get_id();
 138   bool westThenCalled = false;
 139   auto f = p.getFuture().then(
 140     [westId, &westThenCalled](Try<int>&& t) {
 141       EXPECT_EQ(t.value(), 1);
 142       EXPECT_EQ(std::this_thread::get_id(), westId);
 143       westThenCalled = true;
 144       return t.value();
 145     });
 146
 147   bool eastPromiseMade = false;
 148   auto thread = std::thread([&p, &eastPromiseMade, this]() {
 149     EXPECT_NE(t.get_id(), std::this_thread::get_id());
 150     // South thread != west thread. p gets set in west thread.
 151     tg.gate<int>([&p, &eastPromiseMade, this] {
 152                    EXPECT_EQ(t.get_id(), std::this_thread::get_id());
 153                    Promise<int> eastPromise;
 154                    auto eastFuture = eastPromise.getFuture();
 155                    eastPromise.setValue(1);
 156                    eastPromiseMade = true;
 157                    return eastFuture;
 158                  },
 159                  std::move(p));
 160     });
 161
 162   tg.waitFor(f);
 163   EXPECT_TRUE(westThenCalled);
 164   EXPECT_TRUE(eastPromiseMade);
 165   EXPECT_EQ(f.value(), 1);
 166   thread.join();
 167 }