folly/futures/test/ViaTest.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 <gtest/gtest.h>
  18
  19 #include <folly/futures/Future.h>
  20 #include <folly/futures/InlineExecutor.h>
  21 #include <folly/futures/ManualExecutor.h>
  22 #include <folly/futures/DrivableExecutor.h>
  23 #include <folly/Baton.h>
  24 #include <folly/MPMCQueue.h>
  25
  26 #include <thread>
  27
  28 using namespace folly;
  29
  30 struct ManualWaiter : public DrivableExecutor {
  31   explicit ManualWaiter(std::shared_ptr<ManualExecutor> ex) : ex(ex) {}
  32
  33   void add(Func f) override {
  34     ex->add(f);
  35   }
  36
  37   void drive() override {
  38     ex->wait();
  39     ex->run();
  40   }
  41
  42   std::shared_ptr<ManualExecutor> ex;
  43 };
  44
  45 struct ViaFixture : public testing::Test {
  46   ViaFixture() :
  47     westExecutor(new ManualExecutor),
  48     eastExecutor(new ManualExecutor),
  49     waiter(new ManualWaiter(westExecutor)),
  50     done(false)
  51   {
  52     t = std::thread([=] {
  53         ManualWaiter eastWaiter(eastExecutor);
  54         while (!done)
  55           eastWaiter.drive();
  56       });
  57   }
  58
  59   ~ViaFixture() {
  60     done = true;
  61     eastExecutor->add([=]() { });
  62     t.join();
  63   }
  64
  65   void addAsync(int a, int b, std::function<void(int&&)>&& cob) {
  66     eastExecutor->add([=]() {
  67       cob(a + b);
  68     });
  69   }
  70
  71   std::shared_ptr<ManualExecutor> westExecutor;
  72   std::shared_ptr<ManualExecutor> eastExecutor;
  73   std::shared_ptr<ManualWaiter> waiter;
  74   InlineExecutor inlineExecutor;
  75   bool done;
  76   std::thread t;
  77 };
  78
  79 TEST(Via, exceptionOnLaunch) {
  80   auto future = makeFuture<int>(std::runtime_error("E"));
  81   EXPECT_THROW(future.value(), std::runtime_error);
  82 }
  83
  84 TEST(Via, thenValue) {
  85   auto future = makeFuture(std::move(1))
  86     .then([](Try<int>&& t) {
  87       return t.value() == 1;
  88     })
  89     ;
  90
  91   EXPECT_TRUE(future.value());
  92 }
  93
  94 TEST(Via, thenFuture) {
  95   auto future = makeFuture(1)
  96     .then([](Try<int>&& t) {
  97       return makeFuture(t.value() == 1);
  98     });
  99   EXPECT_TRUE(future.value());
 100 }
 101
 102 static Future<std::string> doWorkStatic(Try<std::string>&& t) {
 103   return makeFuture(t.value() + ";static");
 104 }
 105
 106 TEST(Via, thenFunction) {
 107   struct Worker {
 108     Future<std::string> doWork(Try<std::string>&& t) {
 109       return makeFuture(t.value() + ";class");
 110     }
 111     static Future<std::string> doWorkStatic(Try<std::string>&& t) {
 112       return makeFuture(t.value() + ";class-static");
 113     }
 114   } w;
 115
 116   auto f = makeFuture(std::string("start"))
 117     .then(doWorkStatic)
 118     .then(Worker::doWorkStatic)
 119     .then(&Worker::doWork, &w)
 120     ;
 121
 122   EXPECT_EQ(f.value(), "start;static;class-static;class");
 123 }
 124
 125 TEST_F(ViaFixture, threadHops) {
 126   auto westThreadId = std::this_thread::get_id();
 127   auto f = via(eastExecutor.get()).then([=](Try<void>&& t) {
 128     EXPECT_NE(std::this_thread::get_id(), westThreadId);
 129     return makeFuture<int>(1);
 130   }).via(westExecutor.get()
 131   ).then([=](Try<int>&& t) {
 132     EXPECT_EQ(std::this_thread::get_id(), westThreadId);
 133     return t.value();
 134   });
 135   EXPECT_EQ(f.getVia(waiter.get()), 1);
 136 }
 137
 138 TEST_F(ViaFixture, chainVias) {
 139   auto westThreadId = std::this_thread::get_id();
 140   auto f = via(eastExecutor.get()).then([=]() {
 141     EXPECT_NE(std::this_thread::get_id(), westThreadId);
 142     return 1;
 143   }).then([=](int val) {
 144     return makeFuture(val).via(westExecutor.get())
 145       .then([=](int val) mutable {
 146         EXPECT_EQ(std::this_thread::get_id(), westThreadId);
 147         return val + 1;
 148       });
 149   }).then([=](int val) {
 150     // even though ultimately the future that triggers this one executed in
 151     // the west thread, this then() inherited the executor from its
 152     // predecessor, ie the eastExecutor.
 153     EXPECT_NE(std::this_thread::get_id(), westThreadId);
 154     return val + 1;
 155   }).via(westExecutor.get()).then([=](int val) {
 156     // go back to west, so we can wait on it
 157     EXPECT_EQ(std::this_thread::get_id(), westThreadId);
 158     return val + 1;
 159   });
 160
 161   EXPECT_EQ(f.getVia(waiter.get()), 4);
 162 }
 163
 164 TEST_F(ViaFixture, bareViaAssignment) {
 165   auto f = via(eastExecutor.get());
 166 }
 167 TEST_F(ViaFixture, viaAssignment) {
 168   // via()&&
 169   auto f = makeFuture().via(eastExecutor.get());
 170   // via()&
 171   auto f2 = f.via(eastExecutor.get());
 172 }
 173
 174 TEST(Via, chain1) {
 175   EXPECT_EQ(42,
 176             makeFuture()
 177             .thenMulti([] { return 42; })
 178             .get());
 179 }
 180
 181 TEST(Via, chain3) {
 182   int count = 0;
 183   auto f = makeFuture().thenMulti(
 184       [&]{ count++; return 3.14159; },
 185       [&](double) { count++; return std::string("hello"); },
 186       [&]{ count++; return makeFuture(42); });
 187   EXPECT_EQ(42, f.get());
 188   EXPECT_EQ(3, count);
 189 }
 190
 191 struct PriorityExecutor : public Executor {
 192   void add(Func f) override {}
 193
 194   void addWithPriority(Func, int8_t priority) override {
 195     int mid = getNumPriorities() / 2;
 196     int p = priority < 0 ?
 197             std::max(0, mid + priority) :
 198             std::min(getNumPriorities() - 1, mid + priority);
 199     EXPECT_LT(p, 3);
 200     EXPECT_GE(p, 0);
 201     if (p == 0) {
 202       count0++;
 203     } else if (p == 1) {
 204       count1++;
 205     } else if (p == 2) {
 206       count2++;
 207     }
 208   }
 209
 210   uint8_t getNumPriorities() const override {
 211     return 3;
 212   }
 213
 214   int count0{0};
 215   int count1{0};
 216   int count2{0};
 217 };
 218
 219 TEST(Via, priority) {
 220   PriorityExecutor exe;
 221   via(&exe, -1).then([]{});
 222   via(&exe, 0).then([]{});
 223   via(&exe, 1).then([]{});
 224   via(&exe, 42).then([]{});  // overflow should go to max priority
 225   via(&exe, -42).then([]{}); // underflow should go to min priority
 226   via(&exe).then([]{});      // default to mid priority
 227   via(&exe, Executor::LO_PRI).then([]{});
 228   via(&exe, Executor::HI_PRI).then([]{});
 229   EXPECT_EQ(3, exe.count0);
 230   EXPECT_EQ(2, exe.count1);
 231   EXPECT_EQ(3, exe.count2);
 232 }
 233
 234 TEST_F(ViaFixture, chainX1) {
 235   EXPECT_EQ(42,
 236             makeFuture()
 237             .thenMultiWithExecutor(eastExecutor.get(),[] { return 42; })
 238             .get());
 239 }
 240
 241 TEST_F(ViaFixture, chainX3) {
 242   auto westThreadId = std::this_thread::get_id();
 243   int count = 0;
 244   auto f = via(westExecutor.get()).thenMultiWithExecutor(
 245       eastExecutor.get(),
 246       [&]{
 247         EXPECT_NE(std::this_thread::get_id(), westThreadId);
 248         count++; return 3.14159;
 249       },
 250       [&](double) { count++; return std::string("hello"); },
 251       [&]{ count++; })
 252     .then([&](){
 253         EXPECT_EQ(std::this_thread::get_id(), westThreadId);
 254         return makeFuture(42);
 255     });
 256   EXPECT_EQ(42, f.getVia(waiter.get()));
 257   EXPECT_EQ(3, count);
 258 }
 259
 260 TEST(Via, then2) {
 261   ManualExecutor x1, x2;
 262   bool a = false, b = false, c = false;
 263   via(&x1)
 264     .then([&]{ a = true; })
 265     .then(&x2, [&]{ b = true; })
 266     .then([&]{ c = true; });
 267
 268   EXPECT_FALSE(a);
 269   EXPECT_FALSE(b);
 270
 271   x1.run();
 272   EXPECT_TRUE(a);
 273   EXPECT_FALSE(b);
 274   EXPECT_FALSE(c);
 275
 276   x2.run();
 277   EXPECT_TRUE(b);
 278   EXPECT_FALSE(c);
 279
 280   x1.run();
 281   EXPECT_TRUE(c);
 282 }
 283
 284 TEST(Via, then2Variadic) {
 285   struct Foo { bool a = false; void foo(Try<void>) { a = true; } };
 286   Foo f;
 287   ManualExecutor x;
 288   makeFuture().then(&x, &Foo::foo, &f);
 289   EXPECT_FALSE(f.a);
 290   x.run();
 291   EXPECT_TRUE(f.a);
 292 }
 293
 294 /// Simple executor that does work in another thread
 295 class ThreadExecutor : public Executor {
 296   folly::MPMCQueue<Func> funcs;
 297   std::atomic<bool> done {false};
 298   std::thread worker;
 299   folly::Baton<> baton;
 300
 301   void work() {
 302     baton.post();
 303     Func fn;
 304     while (!done) {
 305       while (!funcs.isEmpty()) {
 306         funcs.blockingRead(fn);
 307         fn();
 308       }
 309     }
 310   }
 311
 312  public:
 313   explicit ThreadExecutor(size_t n = 1024)
 314     : funcs(n) {
 315     worker = std::thread(std::bind(&ThreadExecutor::work, this));
 316   }
 317
 318   ~ThreadExecutor() {
 319     done = true;
 320     funcs.write([]{});
 321     worker.join();
 322   }
 323
 324   void add(Func fn) override {
 325     funcs.blockingWrite(std::move(fn));
 326   }
 327
 328   void waitForStartup() {
 329     baton.wait();
 330   }
 331 };
 332
 333 TEST(Via, viaThenGetWasRacy) {
 334   ThreadExecutor x;
 335   std::unique_ptr<int> val = folly::via(&x)
 336     .then([] { return folly::make_unique<int>(42); })
 337     .get();
 338   ASSERT_TRUE(!!val);
 339   EXPECT_EQ(42, *val);
 340 }
 341
 342 class DummyDrivableExecutor : public DrivableExecutor {
 343  public:
 344   void add(Func f) override {}
 345   void drive() override { ran = true; }
 346   bool ran{false};
 347 };
 348
 349 TEST(Via, getVia) {
 350   {
 351     // non-void
 352     ManualExecutor x;
 353     auto f = via(&x).then([]{ return true; });
 354     EXPECT_TRUE(f.getVia(&x));
 355   }
 356
 357   {
 358     // void
 359     ManualExecutor x;
 360     auto f = via(&x).then();
 361     f.getVia(&x);
 362   }
 363
 364   {
 365     DummyDrivableExecutor x;
 366     auto f = makeFuture(true);
 367     EXPECT_TRUE(f.getVia(&x));
 368     EXPECT_FALSE(x.ran);
 369   }
 370 }
 371
 372 TEST(Via, waitVia) {
 373   {
 374     ManualExecutor x;
 375     auto f = via(&x).then();
 376     EXPECT_FALSE(f.isReady());
 377     f.waitVia(&x);
 378     EXPECT_TRUE(f.isReady());
 379   }
 380
 381   {
 382     // try rvalue as well
 383     ManualExecutor x;
 384     auto f = via(&x).then().waitVia(&x);
 385     EXPECT_TRUE(f.isReady());
 386   }
 387
 388   {
 389     DummyDrivableExecutor x;
 390     makeFuture(true).waitVia(&x);
 391     EXPECT_FALSE(x.ran);
 392   }
 393 }
 394
 395 TEST(Via, viaRaces) {
 396   ManualExecutor x;
 397   Promise<void> p;
 398   auto tid = std::this_thread::get_id();
 399   bool done = false;
 400
 401   std::thread t1([&] {
 402     p.getFuture()
 403       .via(&x)
 404       .then([&](Try<void>&&) { EXPECT_EQ(tid, std::this_thread::get_id()); })
 405       .then([&](Try<void>&&) { EXPECT_EQ(tid, std::this_thread::get_id()); })
 406       .then([&](Try<void>&&) { done = true; });
 407   });
 408
 409   std::thread t2([&] {
 410     p.setValue();
 411   });
 412
 413   while (!done) x.run();
 414   t1.join();
 415   t2.join();
 416 }