folly/futures/test/RetryingTest.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 <algorithm>
  18 #include <atomic>
  19 #include <vector>
  20
  21 #include <folly/futures/Future.h>
  22 #include <folly/portability/GTest.h>
  23 #include <folly/portability/SysResource.h>
  24 #include "TestExecutor.h"
  25
  26 using namespace std;
  27 using namespace std::chrono;
  28 using namespace folly;
  29
  30 // Runs func num_times in parallel, expects that all of them will take
  31 // at least min_duration and at least 1 execution will take less than
  32 // max_duration.
  33 template <typename D, typename F>
  34 void multiAttemptExpectDurationWithin(size_t num_tries,
  35                                       D min_duration,
  36                                       D max_duration,
  37                                       const F& func) {
  38   vector<thread> threads(num_tries);
  39   vector<D> durations(num_tries, D::min());
  40   for (size_t i = 0; i < num_tries; ++i) {
  41     threads[i] = thread([&,i]{
  42       auto start = steady_clock::now();
  43       func();
  44       durations[i] = duration_cast<D>(steady_clock::now() - start);
  45     });
  46   }
  47   for (auto& t : threads) {
  48     t.join();
  49   }
  50   sort(durations.begin(), durations.end());
  51   for (auto d : durations) {
  52     EXPECT_GE(d, min_duration);
  53   }
  54   EXPECT_LE(durations[0], max_duration);
  55 }
  56
  57 TEST(RetryingTest, has_op_call) {
  58   using ew = exception_wrapper;
  59   auto policy_raw = [](size_t n, const ew&) { return n < 3; };
  60   auto policy_fut = [](size_t n, const ew&) { return makeFuture(n < 3); };
  61   using namespace futures::detail;
  62   EXPECT_TRUE(retrying_policy_traits<decltype(policy_raw)>::is_raw::value);
  63   EXPECT_TRUE(retrying_policy_traits<decltype(policy_fut)>::is_fut::value);
  64 }
  65
  66 TEST(RetryingTest, basic) {
  67   auto r = futures::retrying(
  68       [](size_t n, const exception_wrapper&) { return n < 3; },
  69       [](size_t n) {
  70           return n < 2
  71             ? makeFuture<size_t>(runtime_error("ha"))
  72             : makeFuture(n);
  73       }
  74   ).wait();
  75   EXPECT_EQ(2, r.value());
  76 }
  77
  78 TEST(RetryingTest, policy_future) {
  79   atomic<size_t> sleeps {0};
  80   auto r = futures::retrying(
  81       [&](size_t n, const exception_wrapper&) {
  82           return n < 3
  83             ? makeFuture(++sleeps).then([] { return true; })
  84             : makeFuture(false);
  85       },
  86       [](size_t n) {
  87           return n < 2
  88             ? makeFuture<size_t>(runtime_error("ha"))
  89             : makeFuture(n);
  90       }
  91   ).wait();
  92   EXPECT_EQ(2, r.value());
  93   EXPECT_EQ(2, sleeps);
  94 }
  95
  96 TEST(RetryingTest, policy_basic) {
  97   auto r = futures::retrying(
  98       futures::retryingPolicyBasic(3),
  99       [](size_t n) {
 100           return n < 2
 101             ? makeFuture<size_t>(runtime_error("ha"))
 102             : makeFuture(n);
 103       }
 104   ).wait();
 105   EXPECT_EQ(2, r.value());
 106 }
 107
 108 TEST(RetryingTest, policy_capped_jittered_exponential_backoff) {
 109   multiAttemptExpectDurationWithin(5, milliseconds(200), milliseconds(400), []{
 110     using ms = milliseconds;
 111     auto r = futures::retrying(
 112         futures::retryingPolicyCappedJitteredExponentialBackoff(
 113           3, ms(100), ms(1000), 0.1, mt19937_64(0),
 114           [](size_t, const exception_wrapper&) { return true; }),
 115         [](size_t n) {
 116             return n < 2
 117               ? makeFuture<size_t>(runtime_error("ha"))
 118               : makeFuture(n);
 119         }
 120     ).wait();
 121     EXPECT_EQ(2, r.value());
 122   });
 123 }
 124
 125 TEST(RetryingTest, policy_sleep_defaults) {
 126   multiAttemptExpectDurationWithin(5, milliseconds(200), milliseconds(400), []{
 127     //  To ensure that this compiles with default params.
 128     using ms = milliseconds;
 129     auto r = futures::retrying(
 130         futures::retryingPolicyCappedJitteredExponentialBackoff(
 131           3, ms(100), ms(1000), 0.1),
 132         [](size_t n) {
 133             return n < 2
 134               ? makeFuture<size_t>(runtime_error("ha"))
 135               : makeFuture(n);
 136         }
 137     ).wait();
 138     EXPECT_EQ(2, r.value());
 139   });
 140 }
 141
 142 TEST(RetryingTest, large_retries) {
 143   rlimit oldMemLimit;
 144   PCHECK(getrlimit(RLIMIT_AS, &oldMemLimit) == 0);
 145
 146   rlimit newMemLimit;
 147   newMemLimit.rlim_cur = std::min(1UL << 30, oldMemLimit.rlim_max);
 148   newMemLimit.rlim_max = oldMemLimit.rlim_max;
 149   if (!folly::kIsSanitizeAddress) { // ASAN reserves outside of the rlimit
 150     PCHECK(setrlimit(RLIMIT_AS, &newMemLimit) == 0);
 151   }
 152   SCOPE_EXIT {
 153     PCHECK(setrlimit(RLIMIT_AS, &oldMemLimit) == 0);
 154   };
 155
 156   TestExecutor executor(4);
 157   // size of implicit promise is at least the size of the return.
 158   using LargeReturn = array<uint64_t, 16000>;
 159   auto func = [&executor](size_t retryNum) -> Future<LargeReturn> {
 160     return via(&executor).then([retryNum] {
 161       return retryNum < 10000
 162           ? makeFuture<LargeReturn>(
 163                 make_exception_wrapper<std::runtime_error>("keep trying"))
 164           : makeFuture<LargeReturn>(LargeReturn());
 165     });
 166   };
 167
 168   vector<Future<LargeReturn>> futures;
 169   for (auto idx = 0; idx < 40; ++idx) {
 170     futures.emplace_back(futures::retrying(
 171         [&executor](size_t, const exception_wrapper&) {
 172           return via(&executor).then([] { return true; });
 173         },
 174         func));
 175   }
 176
 177   // 40 * 10,000 * 16,000B > 1GB; we should avoid OOM
 178
 179   for (auto& f : futures) {
 180     f.wait();
 181     EXPECT_TRUE(f.hasValue());
 182   }
 183 }
 184
 185 /*
 186 TEST(RetryingTest, policy_sleep_cancel) {
 187   multiAttemptExpectDurationWithin(5, milliseconds(0), milliseconds(10), []{
 188     mt19937_64 rng(0);
 189     using ms = milliseconds;
 190     auto r = futures::retrying(
 191         futures::retryingPolicyCappedJitteredExponentialBackoff(
 192           5, ms(100), ms(1000), 0.1, rng,
 193           [](size_t n, const exception_wrapper&) { return true; }),
 194         [](size_t n) {
 195             return n < 4
 196               ? makeFuture<size_t>(runtime_error("ha"))
 197               : makeFuture(n);
 198         }
 199     );
 200     r.cancel();
 201     r.wait();
 202     EXPECT_EQ(2, r.value());
 203   });
 204 }
 205 */