folly/experimental/test/FunctionSchedulerTest.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 #include <algorithm>
  17 #include <atomic>
  18 #include <cassert>
  19 #include <random>
  20
  21 #include <folly/Random.h>
  22 #include <folly/experimental/FunctionScheduler.h>
  23 #include <folly/portability/GTest.h>
  24
  25 #if defined(__linux__)
  26 #include <dlfcn.h>
  27 #endif
  28
  29 using namespace folly;
  30 using std::chrono::milliseconds;
  31
  32 namespace {
  33
  34 /*
  35  * Helper functions for controlling how long this test takes.
  36  *
  37  * Using larger intervals here will make the tests less flaky when run on
  38  * heavily loaded systems.  However, this will also make the tests take longer
  39  * to run.
  40  */
  41 static const auto timeFactor = std::chrono::milliseconds(100);
  42 std::chrono::milliseconds testInterval(int n) { return n * timeFactor; }
  43 int getTicksWithinRange(int n, int min, int max) {
  44   assert(min <= max);
  45   n = std::max(min, n);
  46   n = std::min(max, n);
  47   return n;
  48 }
  49 void delay(int n) {
  50   std::chrono::microseconds usec(n * timeFactor);
  51   usleep(usec.count());
  52 }
  53
  54 } // unnamed namespace
  55
  56 TEST(FunctionScheduler, StartAndShutdown) {
  57   FunctionScheduler fs;
  58   EXPECT_TRUE(fs.start());
  59   EXPECT_FALSE(fs.start());
  60   EXPECT_TRUE(fs.shutdown());
  61   EXPECT_FALSE(fs.shutdown());
  62   // start again
  63   EXPECT_TRUE(fs.start());
  64   EXPECT_FALSE(fs.start());
  65   EXPECT_TRUE(fs.shutdown());
  66   EXPECT_FALSE(fs.shutdown());
  67 }
  68
  69 TEST(FunctionScheduler, SimpleAdd) {
  70   int total = 0;
  71   FunctionScheduler fs;
  72   fs.addFunction([&] { total += 2; }, testInterval(2), "add2");
  73   fs.start();
  74   delay(1);
  75   EXPECT_EQ(2, total);
  76   fs.shutdown();
  77   delay(2);
  78   EXPECT_EQ(2, total);
  79 }
  80
  81 TEST(FunctionScheduler, AddCancel) {
  82   int total = 0;
  83   FunctionScheduler fs;
  84   fs.addFunction([&] { total += 2; }, testInterval(2), "add2");
  85   fs.start();
  86   delay(1);
  87   EXPECT_EQ(2, total);
  88   delay(2);
  89   EXPECT_EQ(4, total);
  90   EXPECT_TRUE(fs.cancelFunction("add2"));
  91   EXPECT_FALSE(fs.cancelFunction("NO SUCH FUNC"));
  92   delay(2);
  93   EXPECT_EQ(4, total);
  94   fs.addFunction([&] { total += 1; }, testInterval(2), "add2");
  95   delay(1);
  96   EXPECT_EQ(5, total);
  97   delay(2);
  98   EXPECT_EQ(6, total);
  99   fs.shutdown();
 100 }
 101
 102 TEST(FunctionScheduler, AddCancel2) {
 103   int total = 0;
 104   FunctionScheduler fs;
 105
 106   // Test adds and cancels while the scheduler is stopped
 107   EXPECT_FALSE(fs.cancelFunction("add2"));
 108   fs.addFunction([&] { total += 1; }, testInterval(2), "add2");
 109   EXPECT_TRUE(fs.cancelFunction("add2"));
 110   EXPECT_FALSE(fs.cancelFunction("add2"));
 111   fs.addFunction([&] { total += 2; }, testInterval(2), "add2");
 112   fs.addFunction([&] { total += 3; }, testInterval(3), "add3");
 113
 114   EXPECT_EQ(0, total);
 115   fs.start();
 116   delay(1);
 117   EXPECT_EQ(5, total);
 118
 119   // Cancel add2 while the scheduler is running
 120   EXPECT_TRUE(fs.cancelFunction("add2"));
 121   EXPECT_FALSE(fs.cancelFunction("add2"));
 122   EXPECT_FALSE(fs.cancelFunction("bogus"));
 123
 124   delay(3);
 125   EXPECT_EQ(8, total);
 126   EXPECT_TRUE(fs.cancelFunction("add3"));
 127
 128   // Test a function that cancels itself
 129   int selfCancelCount = 0;
 130   fs.addFunction(
 131       [&] {
 132         ++selfCancelCount;
 133         if (selfCancelCount > 2) {
 134           fs.cancelFunction("selfCancel");
 135         }
 136       },
 137       testInterval(1), "selfCancel", testInterval(1));
 138   delay(4);
 139   EXPECT_EQ(3, selfCancelCount);
 140   EXPECT_FALSE(fs.cancelFunction("selfCancel"));
 141
 142   // Test a function that schedules another function
 143   int adderCount = 0;
 144   int fn2Count = 0;
 145   auto fn2 = [&] { ++fn2Count; };
 146   auto fnAdder = [&] {
 147     ++adderCount;
 148     if (adderCount == 2) {
 149       fs.addFunction(fn2, testInterval(3), "fn2", testInterval(2));
 150     }
 151   };
 152   fs.addFunction(fnAdder, testInterval(4), "adder");
 153   // t0: adder fires
 154   delay(1); // t1
 155   EXPECT_EQ(1, adderCount);
 156   EXPECT_EQ(0, fn2Count);
 157   // t4: adder fires, schedules fn2
 158   delay(4); // t5
 159   EXPECT_EQ(2, adderCount);
 160   EXPECT_EQ(0, fn2Count);
 161   // t6: fn2 fires
 162   delay(2); // t7
 163   EXPECT_EQ(2, adderCount);
 164   EXPECT_EQ(1, fn2Count);
 165   // t8: adder fires
 166   // t9: fn2 fires
 167   delay(3); // t10
 168   EXPECT_EQ(3, adderCount);
 169   EXPECT_EQ(2, fn2Count);
 170   EXPECT_TRUE(fs.cancelFunction("fn2"));
 171   EXPECT_TRUE(fs.cancelFunction("adder"));
 172   delay(5); // t10
 173   EXPECT_EQ(3, adderCount);
 174   EXPECT_EQ(2, fn2Count);
 175
 176   EXPECT_EQ(8, total);
 177   EXPECT_EQ(3, selfCancelCount);
 178 }
 179
 180 TEST(FunctionScheduler, AddMultiple) {
 181   int total = 0;
 182   FunctionScheduler fs;
 183   fs.addFunction([&] { total += 2; }, testInterval(2), "add2");
 184   fs.addFunction([&] { total += 3; }, testInterval(3), "add3");
 185   EXPECT_THROW(fs.addFunction([&] { total += 2; }, testInterval(2), "add2"),
 186                std::invalid_argument); // function name already exists
 187
 188   fs.start();
 189   delay(1);
 190   EXPECT_EQ(5, total);
 191   delay(4);
 192   EXPECT_EQ(12, total);
 193   EXPECT_TRUE(fs.cancelFunction("add2"));
 194   delay(2);
 195   EXPECT_EQ(15, total);
 196   fs.shutdown();
 197   delay(3);
 198   EXPECT_EQ(15, total);
 199   fs.shutdown();
 200 }
 201
 202 TEST(FunctionScheduler, AddAfterStart) {
 203   int total = 0;
 204   FunctionScheduler fs;
 205   fs.addFunction([&] { total += 2; }, testInterval(2), "add2");
 206   fs.addFunction([&] { total += 3; }, testInterval(2), "add3");
 207   fs.start();
 208   delay(3);
 209   EXPECT_EQ(10, total);
 210   fs.addFunction([&] { total += 2; }, testInterval(3), "add22");
 211   delay(2);
 212   EXPECT_EQ(17, total);
 213 }
 214
 215 TEST(FunctionScheduler, ShutdownStart) {
 216   int total = 0;
 217   FunctionScheduler fs;
 218   fs.addFunction([&] { total += 2; }, testInterval(2), "add2");
 219   fs.start();
 220   delay(1);
 221   fs.shutdown();
 222   fs.start();
 223   delay(1);
 224   EXPECT_EQ(4, total);
 225   EXPECT_FALSE(fs.cancelFunction("add3")); // non existing
 226   delay(2);
 227   EXPECT_EQ(6, total);
 228 }
 229
 230 TEST(FunctionScheduler, ResetFunc) {
 231   int total = 0;
 232   FunctionScheduler fs;
 233   fs.addFunction([&] { total += 2; }, testInterval(3), "add2");
 234   fs.addFunction([&] { total += 3; }, testInterval(3), "add3");
 235   fs.start();
 236   delay(1);
 237   EXPECT_EQ(5, total);
 238   EXPECT_FALSE(fs.resetFunctionTimer("NON_EXISTING"));
 239   EXPECT_TRUE(fs.resetFunctionTimer("add2"));
 240   delay(1);
 241   // t2: after the reset, add2 should have been invoked immediately
 242   EXPECT_EQ(7, total);
 243   usleep(150000);
 244   // t3.5: add3 should have been invoked. add2 should not
 245   EXPECT_EQ(10, total);
 246   delay(1);
 247   // t4.5: add2 should have been invoked once more (it was reset at t1)
 248   EXPECT_EQ(12, total);
 249 }
 250
 251 TEST(FunctionScheduler, AddInvalid) {
 252   int total = 0;
 253   FunctionScheduler fs;
 254   // interval may not be negative
 255   EXPECT_THROW(fs.addFunction([&] { total += 2; }, testInterval(-1), "add2"),
 256                std::invalid_argument);
 257
 258   EXPECT_FALSE(fs.cancelFunction("addNoFunc"));
 259 }
 260
 261 TEST(FunctionScheduler, NoFunctions) {
 262   FunctionScheduler fs;
 263   EXPECT_TRUE(fs.start());
 264   fs.shutdown();
 265   FunctionScheduler fs2;
 266   fs2.shutdown();
 267 }
 268
 269 TEST(FunctionScheduler, AddWhileRunning) {
 270   int total = 0;
 271   FunctionScheduler fs;
 272   fs.start();
 273   delay(1);
 274   fs.addFunction([&] { total += 2; }, testInterval(2), "add2");
 275   // The function should be invoked nearly immediately when we add it
 276   // and the FunctionScheduler is already running
 277   usleep(50000);
 278   EXPECT_EQ(2, total);
 279   delay(2);
 280   EXPECT_EQ(4, total);
 281 }
 282
 283 TEST(FunctionScheduler, NoShutdown) {
 284   int total = 0;
 285   {
 286     FunctionScheduler fs;
 287     fs.addFunction([&] { total += 2; }, testInterval(1), "add2");
 288     fs.start();
 289     usleep(50000);
 290     EXPECT_EQ(2, total);
 291   }
 292   // Destroyed the FunctionScheduler without calling shutdown.
 293   // Everything should have been cleaned up, and the function will no longer
 294   // get called.
 295   delay(2);
 296   EXPECT_EQ(2, total);
 297 }
 298
 299 TEST(FunctionScheduler, StartDelay) {
 300   int total = 0;
 301   FunctionScheduler fs;
 302   fs.addFunction([&] { total += 2; }, testInterval(2), "add2",
 303                  testInterval(2));
 304   fs.addFunction([&] { total += 3; }, testInterval(3), "add3",
 305                  testInterval(2));
 306   EXPECT_THROW(fs.addFunction([&] { total += 2; }, testInterval(3),
 307                               "addX", testInterval(-1)),
 308                std::invalid_argument);
 309   fs.start();
 310   delay(1); // t1
 311   EXPECT_EQ(0, total);
 312   // t2 : add2 total=2
 313   // t2 : add3 total=5
 314   delay(2); // t3
 315   EXPECT_EQ(5, total);
 316   // t4 : add2: total=7
 317   // t5 : add3: total=10
 318   // t6 : add2: total=12
 319   delay(4); // t7
 320   EXPECT_EQ(12, total);
 321   fs.cancelFunction("add2");
 322   // t8 : add3: total=15
 323   delay(2); // t9
 324   EXPECT_EQ(15, total);
 325   fs.shutdown();
 326   delay(3);
 327   EXPECT_EQ(15, total);
 328   fs.shutdown();
 329 }
 330
 331 TEST(FunctionScheduler, NoSteadyCatchup) {
 332   std::atomic<int> ticks(0);
 333   FunctionScheduler fs;
 334   // fs.setSteady(false); is the default
 335   fs.addFunction([&ticks] {
 336                    if (++ticks == 2) {
 337                      std::this_thread::sleep_for(
 338                          std::chrono::milliseconds(200));
 339                    }
 340                  },
 341                  milliseconds(5));
 342   fs.start();
 343   std::this_thread::sleep_for(std::chrono::milliseconds(500));
 344
 345   // no steady catch up means we'd tick once for 200ms, then remaining
 346   // 300ms / 5 = 60 times
 347   EXPECT_LE(ticks.load(), 61);
 348 }
 349
 350 TEST(FunctionScheduler, SteadyCatchup) {
 351   std::atomic<int> ticks(0);
 352   FunctionScheduler fs;
 353   fs.setSteady(true);
 354   fs.addFunction([&ticks] {
 355                    if (++ticks == 2) {
 356                      std::this_thread::sleep_for(
 357                          std::chrono::milliseconds(200));
 358                    }
 359                  },
 360                  milliseconds(5));
 361   fs.start();
 362
 363   std::this_thread::sleep_for(std::chrono::milliseconds(500));
 364
 365   // tick every 5ms. Despite tick == 2 is slow, later ticks should be fast
 366   // enough to catch back up to schedule
 367   EXPECT_NEAR(100, ticks.load(), 10);
 368 }
 369
 370 TEST(FunctionScheduler, UniformDistribution) {
 371   int total = 0;
 372   const int kTicks = 2;
 373   std::chrono::milliseconds minInterval =
 374       testInterval(kTicks) - (timeFactor / 5);
 375   std::chrono::milliseconds maxInterval =
 376       testInterval(kTicks) + (timeFactor / 5);
 377   FunctionScheduler fs;
 378   fs.addFunctionUniformDistribution([&] { total += 2; },
 379                                     minInterval,
 380                                     maxInterval,
 381                                     "UniformDistribution",
 382                                     std::chrono::milliseconds(0));
 383   fs.start();
 384   delay(1);
 385   EXPECT_EQ(2, total);
 386   delay(kTicks);
 387   EXPECT_EQ(4, total);
 388   delay(kTicks);
 389   EXPECT_EQ(6, total);
 390   fs.shutdown();
 391   delay(2);
 392   EXPECT_EQ(6, total);
 393 }
 394
 395 TEST(FunctionScheduler, ExponentialBackoff) {
 396   int total = 0;
 397   int expectedInterval = 0;
 398   int nextInterval = 2;
 399   FunctionScheduler fs;
 400   fs.addFunctionGenericDistribution(
 401       [&] { total += 2; },
 402       [&expectedInterval, nextInterval]() mutable {
 403         expectedInterval = nextInterval;
 404         nextInterval *= nextInterval;
 405         return testInterval(expectedInterval);
 406       },
 407       "ExponentialBackoff",
 408       "2^n * 100ms",
 409       std::chrono::milliseconds(0));
 410   fs.start();
 411   delay(1);
 412   EXPECT_EQ(2, total);
 413   delay(expectedInterval);
 414   EXPECT_EQ(4, total);
 415   delay(expectedInterval);
 416   EXPECT_EQ(6, total);
 417   fs.shutdown();
 418   delay(2);
 419   EXPECT_EQ(6, total);
 420 }
 421
 422 TEST(FunctionScheduler, GammaIntervalDistribution) {
 423   int total = 0;
 424   int expectedInterval = 0;
 425   FunctionScheduler fs;
 426   std::default_random_engine generator(folly::Random::rand32());
 427   // The alpha and beta arguments are selected, somewhat randomly, to be 2.0.
 428   // These values do not matter much in this test, as we are not testing the
 429   // std::gamma_distribution itself...
 430   std::gamma_distribution<double> gamma(2.0, 2.0);
 431   fs.addFunctionGenericDistribution(
 432       [&] { total += 2; },
 433       [&expectedInterval, generator, gamma]() mutable {
 434         expectedInterval =
 435             getTicksWithinRange(static_cast<int>(gamma(generator)), 2, 10);
 436         return testInterval(expectedInterval);
 437       },
 438       "GammaDistribution",
 439       "gamma(2.0,2.0)*100ms",
 440       std::chrono::milliseconds(0));
 441   fs.start();
 442   delay(1);
 443   EXPECT_EQ(2, total);
 444   delay(expectedInterval);
 445   EXPECT_EQ(4, total);
 446   delay(expectedInterval);
 447   EXPECT_EQ(6, total);
 448   fs.shutdown();
 449   delay(2);
 450   EXPECT_EQ(6, total);
 451 }
 452
 453 TEST(FunctionScheduler, AddWithRunOnce) {
 454   int total = 0;
 455   FunctionScheduler fs;
 456   fs.addFunctionOnce([&] { total += 2; }, "add2");
 457   fs.start();
 458   delay(1);
 459   EXPECT_EQ(2, total);
 460   delay(2);
 461   EXPECT_EQ(2, total);
 462
 463   fs.addFunctionOnce([&] { total += 2; }, "add2");
 464   delay(1);
 465   EXPECT_EQ(4, total);
 466   delay(2);
 467   EXPECT_EQ(4, total);
 468
 469   fs.shutdown();
 470 }
 471
 472 TEST(FunctionScheduler, cancelFunctionAndWait) {
 473   int total = 0;
 474   FunctionScheduler fs;
 475   fs.addFunction(
 476       [&] {
 477         delay(5);
 478         total += 2;
 479       },
 480       testInterval(100),
 481       "add2");
 482
 483   fs.start();
 484   delay(1);
 485   EXPECT_EQ(0, total); // add2 is still sleeping
 486
 487   EXPECT_TRUE(fs.cancelFunctionAndWait("add2"));
 488   EXPECT_EQ(2, total); // add2 should have completed
 489
 490   EXPECT_FALSE(fs.cancelFunction("add2")); // add2 has been canceled
 491   fs.shutdown();
 492 }
 493
 494 #if defined(__linux__)
 495 namespace {
 496 /**
 497  * A helper class that forces our pthread_create() wrapper to fail when
 498  * an PThreadCreateFailure object exists.
 499  */
 500 class PThreadCreateFailure {
 501  public:
 502   PThreadCreateFailure() {
 503     ++forceFailure_;
 504   }
 505   ~PThreadCreateFailure() {
 506     --forceFailure_;
 507   }
 508
 509   static bool shouldFail() {
 510     return forceFailure_ > 0;
 511   }
 512
 513  private:
 514   static std::atomic<int> forceFailure_;
 515 };
 516
 517 std::atomic<int> PThreadCreateFailure::forceFailure_{0};
 518 } // unnamed namespce
 519
 520 // Replace the system pthread_create() function with our own stub, so we can
 521 // trigger failures in the StartThrows() test.
 522 extern "C" int pthread_create(
 523     pthread_t* thread,
 524     const pthread_attr_t* attr,
 525     void* (*start_routine)(void*),
 526     void* arg) {
 527   static const auto realFunction = reinterpret_cast<decltype(&pthread_create)>(
 528       dlsym(RTLD_NEXT, "pthread_create"));
 529   // For sanity, make sure we didn't find ourself,
 530   // since that would cause infinite recursion.
 531   CHECK_NE(realFunction, pthread_create);
 532
 533   if (PThreadCreateFailure::shouldFail()) {
 534     errno = EINVAL;
 535     return -1;
 536   }
 537   return realFunction(thread, attr, start_routine, arg);
 538 }
 539
 540 TEST(FunctionScheduler, StartThrows) {
 541   FunctionScheduler fs;
 542   PThreadCreateFailure fail;
 543   EXPECT_ANY_THROW(fs.start());
 544   EXPECT_NO_THROW(fs.shutdown());
 545 }
 546 #endif
 547
 548 TEST(FunctionScheduler, cancelAllFunctionsAndWait) {
 549   int total = 0;
 550   FunctionScheduler fs;
 551
 552   fs.addFunction(
 553       [&] {
 554         delay(5);
 555         total += 2;
 556       },
 557       testInterval(100),
 558       "add2");
 559
 560   fs.start();
 561   delay(1);
 562   EXPECT_EQ(0, total); // add2 is still sleeping
 563
 564   fs.cancelAllFunctionsAndWait();
 565   EXPECT_EQ(2, total);
 566
 567   EXPECT_FALSE(fs.cancelFunction("add2")); // add2 has been canceled
 568   fs.shutdown();
 569 }