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() override {
  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   std::atomic<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())
 128                .then([=](Try<Unit>&& /* t */) {
 129                  EXPECT_NE(std::this_thread::get_id(), westThreadId);
 130                  return makeFuture<int>(1);
 131                })
 132                .via(westExecutor.get())
 133                .then([=](Try<int>&& t) {
 134                  EXPECT_EQ(std::this_thread::get_id(), westThreadId);
 135                  return t.value();
 136                });
 137   EXPECT_EQ(f.getVia(waiter.get()), 1);
 138 }
 139
 140 TEST_F(ViaFixture, chainVias) {
 141   auto westThreadId = std::this_thread::get_id();
 142   auto f = via(eastExecutor.get()).then([=]() {
 143     EXPECT_NE(std::this_thread::get_id(), westThreadId);
 144     return 1;
 145   }).then([=](int val) {
 146     return makeFuture(val).via(westExecutor.get())
 147       .then([=](int val) mutable {
 148         EXPECT_EQ(std::this_thread::get_id(), westThreadId);
 149         return val + 1;
 150       });
 151   }).then([=](int val) {
 152     // even though ultimately the future that triggers this one executed in
 153     // the west thread, this then() inherited the executor from its
 154     // predecessor, ie the eastExecutor.
 155     EXPECT_NE(std::this_thread::get_id(), westThreadId);
 156     return val + 1;
 157   }).via(westExecutor.get()).then([=](int val) {
 158     // go back to west, so we can wait on it
 159     EXPECT_EQ(std::this_thread::get_id(), westThreadId);
 160     return val + 1;
 161   });
 162
 163   EXPECT_EQ(f.getVia(waiter.get()), 4);
 164 }
 165
 166 TEST_F(ViaFixture, bareViaAssignment) {
 167   auto f = via(eastExecutor.get());
 168 }
 169 TEST_F(ViaFixture, viaAssignment) {
 170   // via()&&
 171   auto f = makeFuture().via(eastExecutor.get());
 172   // via()&
 173   auto f2 = f.via(eastExecutor.get());
 174 }
 175
 176 TEST(Via, chain1) {
 177   EXPECT_EQ(42,
 178             makeFuture()
 179             .thenMulti([] { return 42; })
 180             .get());
 181 }
 182
 183 TEST(Via, chain3) {
 184   int count = 0;
 185   auto f = makeFuture().thenMulti(
 186       [&]{ count++; return 3.14159; },
 187       [&](double) { count++; return std::string("hello"); },
 188       [&]{ count++; return makeFuture(42); });
 189   EXPECT_EQ(42, f.get());
 190   EXPECT_EQ(3, count);
 191 }
 192
 193 struct PriorityExecutor : public Executor {
 194   void add(Func /* f */) override {}
 195
 196   void addWithPriority(Func f, int8_t priority) override {
 197     int mid = getNumPriorities() / 2;
 198     int p = priority < 0 ?
 199             std::max(0, mid + priority) :
 200             std::min(getNumPriorities() - 1, mid + priority);
 201     EXPECT_LT(p, 3);
 202     EXPECT_GE(p, 0);
 203     if (p == 0) {
 204       count0++;
 205     } else if (p == 1) {
 206       count1++;
 207     } else if (p == 2) {
 208       count2++;
 209     }
 210     f();
 211   }
 212
 213   uint8_t getNumPriorities() const override {
 214     return 3;
 215   }
 216
 217   int count0{0};
 218   int count1{0};
 219   int count2{0};
 220 };
 221
 222 TEST(Via, priority) {
 223   PriorityExecutor exe;
 224   via(&exe, -1).then([]{});
 225   via(&exe, 0).then([]{});
 226   via(&exe, 1).then([]{});
 227   via(&exe, 42).then([]{});  // overflow should go to max priority
 228   via(&exe, -42).then([]{}); // underflow should go to min priority
 229   via(&exe).then([]{});      // default to mid priority
 230   via(&exe, Executor::LO_PRI).then([]{});
 231   via(&exe, Executor::HI_PRI).then([]{});
 232   EXPECT_EQ(3, exe.count0);
 233   EXPECT_EQ(2, exe.count1);
 234   EXPECT_EQ(3, exe.count2);
 235 }
 236
 237 TEST_F(ViaFixture, chainX1) {
 238   EXPECT_EQ(42,
 239             makeFuture()
 240             .thenMultiWithExecutor(eastExecutor.get(),[] { return 42; })
 241             .get());
 242 }
 243
 244 TEST_F(ViaFixture, chainX3) {
 245   auto westThreadId = std::this_thread::get_id();
 246   int count = 0;
 247   auto f = via(westExecutor.get()).thenMultiWithExecutor(
 248       eastExecutor.get(),
 249       [&]{
 250         EXPECT_NE(std::this_thread::get_id(), westThreadId);
 251         count++; return 3.14159;
 252       },
 253       [&](double) { count++; return std::string("hello"); },
 254       [&]{ count++; })
 255     .then([&](){
 256         EXPECT_EQ(std::this_thread::get_id(), westThreadId);
 257         return makeFuture(42);
 258     });
 259   EXPECT_EQ(42, f.getVia(waiter.get()));
 260   EXPECT_EQ(3, count);
 261 }
 262
 263 TEST(Via, then2) {
 264   ManualExecutor x1, x2;
 265   bool a = false, b = false, c = false;
 266   via(&x1)
 267     .then([&]{ a = true; })
 268     .then(&x2, [&]{ b = true; })
 269     .then([&]{ c = true; });
 270
 271   EXPECT_FALSE(a);
 272   EXPECT_FALSE(b);
 273
 274   x1.run();
 275   EXPECT_TRUE(a);
 276   EXPECT_FALSE(b);
 277   EXPECT_FALSE(c);
 278
 279   x2.run();
 280   EXPECT_TRUE(b);
 281   EXPECT_FALSE(c);
 282
 283   x1.run();
 284   EXPECT_TRUE(c);
 285 }
 286
 287 TEST(Via, then2Variadic) {
 288   struct Foo { bool a = false; void foo(Try<Unit>) { a = true; } };
 289   Foo f;
 290   ManualExecutor x;
 291   makeFuture().then(&x, &Foo::foo, &f);
 292   EXPECT_FALSE(f.a);
 293   x.run();
 294   EXPECT_TRUE(f.a);
 295 }
 296
 297 #ifndef __APPLE__ // TODO #7372389
 298 /// Simple executor that does work in another thread
 299 class ThreadExecutor : public Executor {
 300   folly::MPMCQueue<Func> funcs;
 301   std::atomic<bool> done {false};
 302   std::thread worker;
 303   folly::Baton<> baton;
 304
 305   void work() {
 306     baton.post();
 307     Func fn;
 308     while (!done) {
 309       while (!funcs.isEmpty()) {
 310         funcs.blockingRead(fn);
 311         fn();
 312       }
 313     }
 314   }
 315
 316  public:
 317   explicit ThreadExecutor(size_t n = 1024)
 318     : funcs(n) {
 319     worker = std::thread(std::bind(&ThreadExecutor::work, this));
 320   }
 321
 322   ~ThreadExecutor() override {
 323     done = true;
 324     funcs.write([]{});
 325     worker.join();
 326   }
 327
 328   void add(Func fn) override {
 329     funcs.blockingWrite(std::move(fn));
 330   }
 331
 332   void waitForStartup() {
 333     baton.wait();
 334   }
 335 };
 336
 337 TEST(Via, viaThenGetWasRacy) {
 338   ThreadExecutor x;
 339   std::unique_ptr<int> val = folly::via(&x)
 340     .then([] { return folly::make_unique<int>(42); })
 341     .get();
 342   ASSERT_TRUE(!!val);
 343   EXPECT_EQ(42, *val);
 344 }
 345
 346 TEST(Via, callbackRace) {
 347   ThreadExecutor x;
 348
 349   auto fn = [&x]{
 350     auto promises = std::make_shared<std::vector<Promise<Unit>>>(4);
 351     std::vector<Future<Unit>> futures;
 352
 353     for (auto& p : *promises) {
 354       futures.emplace_back(
 355         p.getFuture()
 356         .via(&x)
 357         .then([](Try<Unit>&&){}));
 358     }
 359
 360     x.waitForStartup();
 361     x.add([promises]{
 362       for (auto& p : *promises) {
 363         p.setValue();
 364       }
 365     });
 366
 367     return collectAll(futures);
 368   };
 369
 370   fn().wait();
 371 }
 372 #endif
 373
 374 class DummyDrivableExecutor : public DrivableExecutor {
 375  public:
 376   void add(Func /* f */) override {}
 377   void drive() override { ran = true; }
 378   bool ran{false};
 379 };
 380
 381 TEST(Via, getVia) {
 382   {
 383     // non-void
 384     ManualExecutor x;
 385     auto f = via(&x).then([]{ return true; });
 386     EXPECT_TRUE(f.getVia(&x));
 387   }
 388
 389   {
 390     // void
 391     ManualExecutor x;
 392     auto f = via(&x).then();
 393     f.getVia(&x);
 394   }
 395
 396   {
 397     DummyDrivableExecutor x;
 398     auto f = makeFuture(true);
 399     EXPECT_TRUE(f.getVia(&x));
 400     EXPECT_FALSE(x.ran);
 401   }
 402 }
 403
 404 TEST(Via, waitVia) {
 405   {
 406     ManualExecutor x;
 407     auto f = via(&x).then();
 408     EXPECT_FALSE(f.isReady());
 409     f.waitVia(&x);
 410     EXPECT_TRUE(f.isReady());
 411   }
 412
 413   {
 414     // try rvalue as well
 415     ManualExecutor x;
 416     auto f = via(&x).then().waitVia(&x);
 417     EXPECT_TRUE(f.isReady());
 418   }
 419
 420   {
 421     DummyDrivableExecutor x;
 422     makeFuture(true).waitVia(&x);
 423     EXPECT_FALSE(x.ran);
 424   }
 425 }
 426
 427 TEST(Via, viaRaces) {
 428   ManualExecutor x;
 429   Promise<Unit> p;
 430   auto tid = std::this_thread::get_id();
 431   bool done = false;
 432
 433   std::thread t1([&] {
 434     p.getFuture()
 435       .via(&x)
 436       .then([&](Try<Unit>&&) { EXPECT_EQ(tid, std::this_thread::get_id()); })
 437       .then([&](Try<Unit>&&) { EXPECT_EQ(tid, std::this_thread::get_id()); })
 438       .then([&](Try<Unit>&&) { done = true; });
 439   });
 440
 441   std::thread t2([&] {
 442     p.setValue();
 443   });
 444
 445   while (!done) x.run();
 446   t1.join();
 447   t2.join();
 448 }
 449
 450 TEST(ViaFunc, liftsVoid) {
 451   ManualExecutor x;
 452   int count = 0;
 453   Future<Unit> f = via(&x, [&]{ count++; });
 454
 455   EXPECT_EQ(0, count);
 456   x.run();
 457   EXPECT_EQ(1, count);
 458 }
 459
 460 TEST(ViaFunc, value) {
 461   ManualExecutor x;
 462   EXPECT_EQ(42, via(&x, []{ return 42; }).getVia(&x));
 463 }
 464
 465 TEST(ViaFunc, exception) {
 466   ManualExecutor x;
 467   EXPECT_THROW(
 468     via(&x, []() -> int { throw std::runtime_error("expected"); })
 469       .getVia(&x),
 470     std::runtime_error);
 471 }
 472
 473 TEST(ViaFunc, future) {
 474   ManualExecutor x;
 475   EXPECT_EQ(42, via(&x, []{ return makeFuture(42); })
 476             .getVia(&x));
 477 }
 478
 479 TEST(ViaFunc, voidFuture) {
 480   ManualExecutor x;
 481   int count = 0;
 482   via(&x, [&]{ count++; }).getVia(&x);
 483   EXPECT_EQ(1, count);
 484 }
 485
 486 TEST(ViaFunc, isSticky) {
 487   ManualExecutor x;
 488   int count = 0;
 489
 490   auto f = via(&x, [&]{ count++; });
 491   x.run();
 492
 493   f.then([&]{ count++; });
 494   EXPECT_EQ(1, count);
 495   x.run();
 496   EXPECT_EQ(2, count);
 497 }