folly/futures/test/CollectTest.cpp

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