folly/futures/test/CollectTest.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 <boost/thread/barrier.hpp>
  20
  21 #include <folly/futures/Future.h>
  22 #include <folly/Random.h>
  23 #include <folly/small_vector.h>
  24
  25 using namespace folly;
  26
  27 typedef FutureException eggs_t;
  28 static eggs_t eggs("eggs");
  29
  30 auto rng = std::mt19937(folly::randomNumberSeed());
  31
  32 TEST(Collect, collectAll) {
  33   // returns a vector variant
  34   {
  35     std::vector<Promise<int>> promises(10);
  36     std::vector<Future<int>> futures;
  37
  38     for (auto& p : promises)
  39       futures.push_back(p.getFuture());
  40
  41     auto allf = collectAll(futures);
  42
  43     std::shuffle(promises.begin(), promises.end(), rng);
  44     for (auto& p : promises) {
  45       EXPECT_FALSE(allf.isReady());
  46       p.setValue(42);
  47     }
  48
  49     EXPECT_TRUE(allf.isReady());
  50     auto& results = allf.value();
  51     for (auto& t : results) {
  52       EXPECT_EQ(42, t.value());
  53     }
  54   }
  55
  56   // check error semantics
  57   {
  58     std::vector<Promise<int>> promises(4);
  59     std::vector<Future<int>> futures;
  60
  61     for (auto& p : promises)
  62       futures.push_back(p.getFuture());
  63
  64     auto allf = collectAll(futures);
  65
  66
  67     promises[0].setValue(42);
  68     promises[1].setException(eggs);
  69
  70     EXPECT_FALSE(allf.isReady());
  71
  72     promises[2].setValue(42);
  73
  74     EXPECT_FALSE(allf.isReady());
  75
  76     promises[3].setException(eggs);
  77
  78     EXPECT_TRUE(allf.isReady());
  79     EXPECT_FALSE(allf.getTry().hasException());
  80
  81     auto& results = allf.value();
  82     EXPECT_EQ(42, results[0].value());
  83     EXPECT_TRUE(results[1].hasException());
  84     EXPECT_EQ(42, results[2].value());
  85     EXPECT_TRUE(results[3].hasException());
  86   }
  87
  88   // check that futures are ready in then()
  89   {
  90     std::vector<Promise<Unit>> promises(10);
  91     std::vector<Future<Unit>> futures;
  92
  93     for (auto& p : promises)
  94       futures.push_back(p.getFuture());
  95
  96     auto allf = collectAll(futures)
  97       .then([](Try<std::vector<Try<Unit>>>&& ts) {
  98         for (auto& f : ts.value())
  99           f.value();
 100       });
 101
 102     std::shuffle(promises.begin(), promises.end(), rng);
 103     for (auto& p : promises)
 104       p.setValue();
 105     EXPECT_TRUE(allf.isReady());
 106   }
 107 }
 108
 109 TEST(Collect, collect) {
 110   // success case
 111   {
 112     std::vector<Promise<int>> promises(10);
 113     std::vector<Future<int>> futures;
 114
 115     for (auto& p : promises)
 116       futures.push_back(p.getFuture());
 117
 118     auto allf = collect(futures);
 119
 120     std::shuffle(promises.begin(), promises.end(), rng);
 121     for (auto& p : promises) {
 122       EXPECT_FALSE(allf.isReady());
 123       p.setValue(42);
 124     }
 125
 126     EXPECT_TRUE(allf.isReady());
 127     for (auto i : allf.value()) {
 128       EXPECT_EQ(42, i);
 129     }
 130   }
 131
 132   // failure case
 133   {
 134     std::vector<Promise<int>> promises(10);
 135     std::vector<Future<int>> futures;
 136
 137     for (auto& p : promises)
 138       futures.push_back(p.getFuture());
 139
 140     auto allf = collect(futures);
 141
 142     std::shuffle(promises.begin(), promises.end(), rng);
 143     for (int i = 0; i < 10; i++) {
 144       if (i < 5) {
 145         // everthing goes well so far...
 146         EXPECT_FALSE(allf.isReady());
 147         promises[i].setValue(42);
 148       } else if (i == 5) {
 149         // short circuit with an exception
 150         EXPECT_FALSE(allf.isReady());
 151         promises[i].setException(eggs);
 152         EXPECT_TRUE(allf.isReady());
 153       } else if (i < 8) {
 154         // don't blow up on further values
 155         EXPECT_TRUE(allf.isReady());
 156         promises[i].setValue(42);
 157       } else {
 158         // don't blow up on further exceptions
 159         EXPECT_TRUE(allf.isReady());
 160         promises[i].setException(eggs);
 161       }
 162     }
 163
 164     EXPECT_THROW(allf.value(), eggs_t);
 165   }
 166
 167   // void futures success case
 168   {
 169     std::vector<Promise<Unit>> promises(10);
 170     std::vector<Future<Unit>> futures;
 171
 172     for (auto& p : promises)
 173       futures.push_back(p.getFuture());
 174
 175     auto allf = collect(futures);
 176
 177     std::shuffle(promises.begin(), promises.end(), rng);
 178     for (auto& p : promises) {
 179       EXPECT_FALSE(allf.isReady());
 180       p.setValue();
 181     }
 182
 183     EXPECT_TRUE(allf.isReady());
 184   }
 185
 186   // void futures failure case
 187   {
 188     std::vector<Promise<Unit>> promises(10);
 189     std::vector<Future<Unit>> futures;
 190
 191     for (auto& p : promises)
 192       futures.push_back(p.getFuture());
 193
 194     auto allf = collect(futures);
 195
 196     std::shuffle(promises.begin(), promises.end(), rng);
 197     for (int i = 0; i < 10; i++) {
 198       if (i < 5) {
 199         // everthing goes well so far...
 200         EXPECT_FALSE(allf.isReady());
 201         promises[i].setValue();
 202       } else if (i == 5) {
 203         // short circuit with an exception
 204         EXPECT_FALSE(allf.isReady());
 205         promises[i].setException(eggs);
 206         EXPECT_TRUE(allf.isReady());
 207       } else if (i < 8) {
 208         // don't blow up on further values
 209         EXPECT_TRUE(allf.isReady());
 210         promises[i].setValue();
 211       } else {
 212         // don't blow up on further exceptions
 213         EXPECT_TRUE(allf.isReady());
 214         promises[i].setException(eggs);
 215       }
 216     }
 217
 218     EXPECT_THROW(allf.value(), eggs_t);
 219   }
 220
 221   // move only compiles
 222   {
 223     std::vector<Promise<std::unique_ptr<int>>> promises(10);
 224     std::vector<Future<std::unique_ptr<int>>> futures;
 225
 226     for (auto& p : promises)
 227       futures.push_back(p.getFuture());
 228
 229     collect(futures);
 230   }
 231
 232 }
 233
 234 struct NotDefaultConstructible {
 235   NotDefaultConstructible() = delete;
 236   explicit NotDefaultConstructible(int arg) : i(arg) {}
 237   int i;
 238 };
 239
 240 // We have a specialized implementation for non-default-constructible objects
 241 // Ensure that it works and preserves order
 242 TEST(Collect, collectNotDefaultConstructible) {
 243   std::vector<Promise<NotDefaultConstructible>> promises(10);
 244   std::vector<Future<NotDefaultConstructible>> futures;
 245   std::vector<int> indices(10);
 246   std::iota(indices.begin(), indices.end(), 0);
 247   std::shuffle(indices.begin(), indices.end(), rng);
 248
 249   for (auto& p : promises)
 250     futures.push_back(p.getFuture());
 251
 252   auto allf = collect(futures);
 253
 254   for (auto i : indices) {
 255     EXPECT_FALSE(allf.isReady());
 256     promises[i].setValue(NotDefaultConstructible(i));
 257   }
 258
 259   EXPECT_TRUE(allf.isReady());
 260   int i = 0;
 261   for (auto val : allf.value()) {
 262     EXPECT_EQ(i, val.i);
 263     i++;
 264   }
 265 }
 266
 267 TEST(Collect, collectAny) {
 268   {
 269     std::vector<Promise<int>> promises(10);
 270     std::vector<Future<int>> futures;
 271
 272     for (auto& p : promises)
 273       futures.push_back(p.getFuture());
 274
 275     for (auto& f : futures) {
 276       EXPECT_FALSE(f.isReady());
 277     }
 278
 279     auto anyf = collectAny(futures);
 280
 281     /* futures were moved in, so these are invalid now */
 282     EXPECT_FALSE(anyf.isReady());
 283
 284     promises[7].setValue(42);
 285     EXPECT_TRUE(anyf.isReady());
 286     auto& idx_fut = anyf.value();
 287
 288     auto i = idx_fut.first;
 289     EXPECT_EQ(7, i);
 290
 291     auto& f = idx_fut.second;
 292     EXPECT_EQ(42, f.value());
 293   }
 294
 295   // error
 296   {
 297     std::vector<Promise<Unit>> promises(10);
 298     std::vector<Future<Unit>> futures;
 299
 300     for (auto& p : promises)
 301       futures.push_back(p.getFuture());
 302
 303     for (auto& f : futures) {
 304       EXPECT_FALSE(f.isReady());
 305     }
 306
 307     auto anyf = collectAny(futures);
 308
 309     EXPECT_FALSE(anyf.isReady());
 310
 311     promises[3].setException(eggs);
 312     EXPECT_TRUE(anyf.isReady());
 313     EXPECT_TRUE(anyf.value().second.hasException());
 314   }
 315
 316   // then()
 317   {
 318     std::vector<Promise<int>> promises(10);
 319     std::vector<Future<int>> futures;
 320
 321     for (auto& p : promises)
 322       futures.push_back(p.getFuture());
 323
 324     auto anyf = collectAny(futures)
 325       .then([](std::pair<size_t, Try<int>> p) {
 326         EXPECT_EQ(42, p.second.value());
 327       });
 328
 329     promises[3].setValue(42);
 330     EXPECT_TRUE(anyf.isReady());
 331   }
 332 }
 333
 334
 335 TEST(Collect, alreadyCompleted) {
 336   {
 337     std::vector<Future<Unit>> fs;
 338     for (int i = 0; i < 10; i++)
 339       fs.push_back(makeFuture());
 340
 341     collectAll(fs)
 342       .then([&](std::vector<Try<Unit>> ts) {
 343         EXPECT_EQ(fs.size(), ts.size());
 344       });
 345   }
 346   {
 347     std::vector<Future<int>> fs;
 348     for (int i = 0; i < 10; i++)
 349       fs.push_back(makeFuture(i));
 350
 351     collectAny(fs)
 352       .then([&](std::pair<size_t, Try<int>> p) {
 353         EXPECT_EQ(p.first, p.second.value());
 354       });
 355   }
 356 }
 357
 358 TEST(Collect, parallel) {
 359   std::vector<Promise<int>> ps(10);
 360   std::vector<Future<int>> fs;
 361   for (size_t i = 0; i < ps.size(); i++) {
 362     fs.emplace_back(ps[i].getFuture());
 363   }
 364   auto f = collect(fs);
 365
 366   std::vector<std::thread> ts;
 367   boost::barrier barrier(ps.size() + 1);
 368   for (size_t i = 0; i < ps.size(); i++) {
 369     ts.emplace_back([&ps, &barrier, i]() {
 370       barrier.wait();
 371       ps[i].setValue(i);
 372     });
 373   }
 374
 375   barrier.wait();
 376
 377   for (size_t i = 0; i < ps.size(); i++) {
 378     ts[i].join();
 379   }
 380
 381   EXPECT_TRUE(f.isReady());
 382   for (size_t i = 0; i < ps.size(); i++) {
 383     EXPECT_EQ(i, f.value()[i]);
 384   }
 385 }
 386
 387 TEST(Collect, parallelWithError) {
 388   std::vector<Promise<int>> ps(10);
 389   std::vector<Future<int>> fs;
 390   for (size_t i = 0; i < ps.size(); i++) {
 391     fs.emplace_back(ps[i].getFuture());
 392   }
 393   auto f = collect(fs);
 394
 395   std::vector<std::thread> ts;
 396   boost::barrier barrier(ps.size() + 1);
 397   for (size_t i = 0; i < ps.size(); i++) {
 398     ts.emplace_back([&ps, &barrier, i]() {
 399       barrier.wait();
 400       if (i == (ps.size()/2)) {
 401         ps[i].setException(eggs);
 402       } else {
 403         ps[i].setValue(i);
 404       }
 405     });
 406   }
 407
 408   barrier.wait();
 409
 410   for (size_t i = 0; i < ps.size(); i++) {
 411     ts[i].join();
 412   }
 413
 414   EXPECT_TRUE(f.isReady());
 415   EXPECT_THROW(f.value(), eggs_t);
 416 }
 417
 418 TEST(Collect, allParallel) {
 419   std::vector<Promise<int>> ps(10);
 420   std::vector<Future<int>> fs;
 421   for (size_t i = 0; i < ps.size(); i++) {
 422     fs.emplace_back(ps[i].getFuture());
 423   }
 424   auto f = collectAll(fs);
 425
 426   std::vector<std::thread> ts;
 427   boost::barrier barrier(ps.size() + 1);
 428   for (size_t i = 0; i < ps.size(); i++) {
 429     ts.emplace_back([&ps, &barrier, i]() {
 430       barrier.wait();
 431       ps[i].setValue(i);
 432     });
 433   }
 434
 435   barrier.wait();
 436
 437   for (size_t i = 0; i < ps.size(); i++) {
 438     ts[i].join();
 439   }
 440
 441   EXPECT_TRUE(f.isReady());
 442   for (size_t i = 0; i < ps.size(); i++) {
 443     EXPECT_TRUE(f.value()[i].hasValue());
 444     EXPECT_EQ(i, f.value()[i].value());
 445   }
 446 }
 447
 448 TEST(Collect, allParallelWithError) {
 449   std::vector<Promise<int>> ps(10);
 450   std::vector<Future<int>> fs;
 451   for (size_t i = 0; i < ps.size(); i++) {
 452     fs.emplace_back(ps[i].getFuture());
 453   }
 454   auto f = collectAll(fs);
 455
 456   std::vector<std::thread> ts;
 457   boost::barrier barrier(ps.size() + 1);
 458   for (size_t i = 0; i < ps.size(); i++) {
 459     ts.emplace_back([&ps, &barrier, i]() {
 460       barrier.wait();
 461       if (i == (ps.size()/2)) {
 462         ps[i].setException(eggs);
 463       } else {
 464         ps[i].setValue(i);
 465       }
 466     });
 467   }
 468
 469   barrier.wait();
 470
 471   for (size_t i = 0; i < ps.size(); i++) {
 472     ts[i].join();
 473   }
 474
 475   EXPECT_TRUE(f.isReady());
 476   for (size_t i = 0; i < ps.size(); i++) {
 477     if (i == (ps.size()/2)) {
 478       EXPECT_THROW(f.value()[i].value(), eggs_t);
 479     } else {
 480       EXPECT_TRUE(f.value()[i].hasValue());
 481       EXPECT_EQ(i, f.value()[i].value());
 482     }
 483   }
 484 }
 485
 486 TEST(Collect, collectN) {
 487   std::vector<Promise<Unit>> promises(10);
 488   std::vector<Future<Unit>> futures;
 489
 490   for (auto& p : promises)
 491     futures.push_back(p.getFuture());
 492
 493   bool flag = false;
 494   size_t n = 3;
 495   collectN(futures, n)
 496     .then([&](std::vector<std::pair<size_t, Try<Unit>>> v) {
 497       flag = true;
 498       EXPECT_EQ(n, v.size());
 499       for (auto& tt : v)
 500         EXPECT_TRUE(tt.second.hasValue());
 501     });
 502
 503   promises[0].setValue();
 504   EXPECT_FALSE(flag);
 505   promises[1].setValue();
 506   EXPECT_FALSE(flag);
 507   promises[2].setValue();
 508   EXPECT_TRUE(flag);
 509 }
 510
 511 /// Ensure that we can compile collectAll/Any with folly::small_vector
 512 TEST(Collect, smallVector) {
 513   static_assert(!FOLLY_IS_TRIVIALLY_COPYABLE(Future<Unit>),
 514                 "Futures should not be trivially copyable");
 515   static_assert(!FOLLY_IS_TRIVIALLY_COPYABLE(Future<int>),
 516                 "Futures should not be trivially copyable");
 517
 518   {
 519     folly::small_vector<Future<Unit>> futures;
 520
 521     for (int i = 0; i < 10; i++)
 522       futures.push_back(makeFuture());
 523
 524     auto anyf = collectAny(futures);
 525   }
 526   {
 527     folly::small_vector<Future<Unit>> futures;
 528
 529     for (int i = 0; i < 10; i++)
 530       futures.push_back(makeFuture());
 531
 532     auto allf = collectAll(futures);
 533   }
 534 }
 535
 536 TEST(Collect, collectAllVariadic) {
 537   Promise<bool> pb;
 538   Promise<int> pi;
 539   Future<bool> fb = pb.getFuture();
 540   Future<int> fi = pi.getFuture();
 541   bool flag = false;
 542   collectAll(std::move(fb), std::move(fi))
 543     .then([&](std::tuple<Try<bool>, Try<int>> tup) {
 544       flag = true;
 545       EXPECT_TRUE(std::get<0>(tup).hasValue());
 546       EXPECT_EQ(std::get<0>(tup).value(), true);
 547       EXPECT_TRUE(std::get<1>(tup).hasValue());
 548       EXPECT_EQ(std::get<1>(tup).value(), 42);
 549     });
 550   pb.setValue(true);
 551   EXPECT_FALSE(flag);
 552   pi.setValue(42);
 553   EXPECT_TRUE(flag);
 554 }
 555
 556 TEST(Collect, collectAllVariadicReferences) {
 557   Promise<bool> pb;
 558   Promise<int> pi;
 559   Future<bool> fb = pb.getFuture();
 560   Future<int> fi = pi.getFuture();
 561   bool flag = false;
 562   collectAll(fb, fi)
 563     .then([&](std::tuple<Try<bool>, Try<int>> tup) {
 564       flag = true;
 565       EXPECT_TRUE(std::get<0>(tup).hasValue());
 566       EXPECT_EQ(std::get<0>(tup).value(), true);
 567       EXPECT_TRUE(std::get<1>(tup).hasValue());
 568       EXPECT_EQ(std::get<1>(tup).value(), 42);
 569     });
 570   pb.setValue(true);
 571   EXPECT_FALSE(flag);
 572   pi.setValue(42);
 573   EXPECT_TRUE(flag);
 574 }
 575
 576 TEST(Collect, collectAllVariadicWithException) {
 577   Promise<bool> pb;
 578   Promise<int> pi;
 579   Future<bool> fb = pb.getFuture();
 580   Future<int> fi = pi.getFuture();
 581   bool flag = false;
 582   collectAll(std::move(fb), std::move(fi))
 583     .then([&](std::tuple<Try<bool>, Try<int>> tup) {
 584       flag = true;
 585       EXPECT_TRUE(std::get<0>(tup).hasValue());
 586       EXPECT_EQ(std::get<0>(tup).value(), true);
 587       EXPECT_TRUE(std::get<1>(tup).hasException());
 588       EXPECT_THROW(std::get<1>(tup).value(), eggs_t);
 589     });
 590   pb.setValue(true);
 591   EXPECT_FALSE(flag);
 592   pi.setException(eggs);
 593   EXPECT_TRUE(flag);
 594 }
 595
 596 TEST(Collect, collectVariadic) {
 597   Promise<bool> pb;
 598   Promise<int> pi;
 599   Future<bool> fb = pb.getFuture();
 600   Future<int> fi = pi.getFuture();
 601   bool flag = false;
 602   collect(std::move(fb), std::move(fi))
 603     .then([&](std::tuple<bool, int> tup) {
 604       flag = true;
 605       EXPECT_EQ(std::get<0>(tup), true);
 606       EXPECT_EQ(std::get<1>(tup), 42);
 607     });
 608   pb.setValue(true);
 609   EXPECT_FALSE(flag);
 610   pi.setValue(42);
 611   EXPECT_TRUE(flag);
 612 }
 613
 614 TEST(Collect, collectVariadicWithException) {
 615   Promise<bool> pb;
 616   Promise<int> pi;
 617   Future<bool> fb = pb.getFuture();
 618   Future<int> fi = pi.getFuture();
 619   auto f = collect(std::move(fb), std::move(fi));
 620   pb.setValue(true);
 621   EXPECT_FALSE(f.isReady());
 622   pi.setException(eggs);
 623   EXPECT_TRUE(f.isReady());
 624   EXPECT_TRUE(f.getTry().hasException());
 625   EXPECT_THROW(f.get(), eggs_t);
 626 }
 627
 628 TEST(Collect, collectAllNone) {
 629   std::vector<Future<int>> fs;
 630   auto f = collectAll(fs);
 631   EXPECT_TRUE(f.isReady());
 632 }
 633
 634 TEST(Collect, noDefaultConstructor) {
 635   struct A {
 636     explicit A(size_t /* x */) {}
 637   };
 638
 639   auto f1 = makeFuture(A(1));
 640   auto f2 = makeFuture(A(2));
 641
 642   auto f = collect(std::move(f1), std::move(f2));
 643 }