WindowTest.cpp File Reference

#include <boost/thread/barrier.hpp>
#include <folly/Conv.h>
#include <folly/executors/ManualExecutor.h>
#include <folly/futures/Future.h>
#include <folly/portability/GTest.h>
#include <vector>

Go to the source code of this file.

Typedefs
typedef FutureException	eggs_t

Functions
	TEST (Window, basic)
	TEST (Window, exception)
	TEST (Window, stackOverflow)
	TEST (Window, parallel)
	TEST (Window, parallelWithError)
	TEST (Window, allParallelWithError)
	TEST (WindowExecutor, basic)
	TEST (WindowExecutor, parallel)
	TEST (WindowExecutor, parallelWithError)
	TEST (WindowExecutor, allParallelWithError)

Variables
static eggs_t	eggs ("eggs")

Typedef Documentation

typedef FutureException eggs_t

Definition at line 28 of file WindowTest.cpp.

Function Documentation

TEST	(	Window	,
		basic
	)

Definition at line 31 of file WindowTest.cpp.

References b, folly::symbolizer::test::expect(), EXPECT_EQ, EXPECT_TRUE, folly::Try< T >::hasValue(), i, folly::makeFuture(), folly::reduce(), s, SCOPED_TRACE, string, sum(), folly::Try< T >::value(), and folly::window().

                    {
  // int -> Future<int>
  auto fn = [](std::vector<int> input, size_t window_size, size_t expect) {
    auto res =
        reduce(
            window(input, [](int i) { return makeFuture(i); }, window_size),
            0,
            [](int sum, const Try<int>& b) { return sum + *b; })
            .get();
    EXPECT_EQ(expect, res);
  };
  {
    SCOPED_TRACE("2 in-flight at a time");
    std::vector<int> input = {1, 2, 3};
    fn(input, 2, 6);
  }
  {
    SCOPED_TRACE("4 in-flight at a time");
    std::vector<int> input = {1, 2, 3};
    fn(input, 4, 6);
  }
  {
    SCOPED_TRACE("empty input");
    std::vector<int> input;
    fn(input, 1, 0);
  }
  {
    // int -> Future<Unit>
    auto res = reduce(
                   window(
                       std::vector<int>({1, 2, 3}),
                       [](int /* i */) { return makeFuture(); },
                       2),
                   0,
                   [](int sum, const Try<Unit>& b) {
                     EXPECT_TRUE(b.hasValue());
                     return sum + 1;
                   })
                   .get();
    EXPECT_EQ(3, res);
  }
  {
    // string -> return Future<int>
    auto res = reduce(
                   window(
                       std::vector<std::string>{"1", "2", "3"},
                       [](std::string s) {
                         return makeFuture<int>(folly::to<int>(s));
                       },
                       2),
                   0,
                   [](int sum, const Try<int>& b) { return sum + *b; })
                   .get();
    EXPECT_EQ(6, res);
  }
  {
    // string -> return SemiFuture<int>
    auto res = reduce(
                   window(
                       std::vector<std::string>{"1", "2", "3"},
                       [](std::string s) {
                         return makeSemiFuture<int>(folly::to<int>(s));
                       },
                       2),
                   0,
                   [](int sum, const Try<int>& b) { return sum + *b; })
                   .get();
    EXPECT_EQ(6, res);
  }
  {
    SCOPED_TRACE("repeat same fn");
    auto res =
        reduce(
            window(
                5UL,
                [](size_t iteration) { return folly::makeFuture(iteration); },
                2),
            0UL,
            [](size_t sum, const Try<size_t>& b) { return sum + b.value(); })
            .get();
    EXPECT_EQ(0 + 1 + 2 + 3 + 4, res);
  }
}

TEST	(	Window	,
		exception
	)

Definition at line 115 of file WindowTest.cpp.

References b, EXPECT_EQ, folly::Try< T >::hasException(), i, folly::gen::move, folly::reduce(), sum(), and folly::window().

                        {
  std::vector<int> ints = {1, 2, 3, 4};
  std::vector<Promise<int>> ps(4);

  auto res = reduce(
      window(
          ints,
          [&ps](int i) {
            if (i > 2) {
              throw std::runtime_error("exception should not kill process");
            }
            return ps[i].getFuture();
          },
          2),
      0,
      [](int sum, const Try<int>& b) {
        sum += b.hasException<std::exception>() ? 1 : 0;
        return sum;
      });

  for (auto& p : ps) {
    p.setValue(0);
  }

  // Should have received 2 exceptions.
  EXPECT_EQ(2, std::move(res).get());
}

TEST	(	Window	,
		stackOverflow
	)

Definition at line 143 of file WindowTest.cpp.

References a, b, EXPECT_EQ, i, int64_t, m, folly::gen::move, cpp.ast::next(), folly::reduce(), sum(), and folly::window().

                            {
  // Number of futures to spawn.
  static constexpr size_t m = 1000;
  // Size of each block of input and output.
  static constexpr size_t n = 1000;

  std::vector<std::array<int, n>> ints;
  int64_t expectedSum = 0;
  for (size_t i = 0; i < m; i++) {
    std::array<int, n> next{};
    next[i % n] = i;
    ints.emplace_back(next);
    expectedSum += i;
  }

  // Try to overflow window's executor.
  auto res = reduce(
      window(
          ints,
          [](std::array<int, n> i) {
            return folly::Future<std::array<int, n>>(i);
          },
          1),
      static_cast<int64_t>(0),
      [](int64_t sum, const Try<std::array<int, n>>& b) {
        for (int a : b.value()) {
          sum += a;
        }
        return sum;
      });

  EXPECT_EQ(std::move(res).get(), expectedSum);
}

TEST	(	Window	,
		parallel
	)

Definition at line 177 of file WindowTest.cpp.

References folly::collect(), EXPECT_EQ, EXPECT_TRUE, f, i, folly::pushmi::detail::t, and folly::window().

                       {
  std::vector<int> input;
  std::vector<Promise<int>> ps(10);
  for (size_t i = 0; i < ps.size(); i++) {
    input.emplace_back(i);
  }
  auto f = collect(window(input, [&](int i) { return ps[i].getFuture(); }, 3));

  std::vector<std::thread> ts;
  boost::barrier barrier(ps.size() + 1);
  for (size_t i = 0; i < ps.size(); i++) {
    ts.emplace_back([&ps, &barrier, i]() {
      barrier.wait();
      ps[i].setValue(i);
    });
  }

  barrier.wait();

  for (auto& t : ts) {
    t.join();
  }

  EXPECT_TRUE(f.isReady());
  for (size_t i = 0; i < ps.size(); i++) {
    EXPECT_EQ(i, f.value()[i]);
  }
}

TEST	(	Window	,
		parallelWithError
	)

Definition at line 206 of file WindowTest.cpp.

References folly::collect(), eggs, EXPECT_THROW, EXPECT_TRUE, f, i, folly::pushmi::detail::t, and folly::window().

                                {
  std::vector<int> input;
  std::vector<Promise<int>> ps(10);
  for (size_t i = 0; i < ps.size(); i++) {
    input.emplace_back(i);
  }
  auto f = collect(window(input, [&](int i) { return ps[i].getFuture(); }, 3));

  std::vector<std::thread> ts;
  boost::barrier barrier(ps.size() + 1);
  for (size_t i = 0; i < ps.size(); i++) {
    ts.emplace_back([&ps, &barrier, i]() {
      barrier.wait();
      if (i == (ps.size() / 2)) {
        ps[i].setException(eggs);
      } else {
        ps[i].setValue(i);
      }
    });
  }

  barrier.wait();

  for (auto& t : ts) {
    t.join();
  }

  EXPECT_TRUE(f.isReady());
  EXPECT_THROW(f.value(), eggs_t);
}

TEST	(	Window	,
		allParallelWithError
	)

Definition at line 237 of file WindowTest.cpp.

References folly::collectAll(), eggs, EXPECT_EQ, EXPECT_THROW, EXPECT_TRUE, f, i, folly::pushmi::detail::t, and folly::window().

                                   {
  std::vector<int> input;
  std::vector<Promise<int>> ps(10);
  for (size_t i = 0; i < ps.size(); i++) {
    input.emplace_back(i);
  }
  auto f =
      collectAll(window(input, [&](int i) { return ps[i].getFuture(); }, 3));

  std::vector<std::thread> ts;
  boost::barrier barrier(ps.size() + 1);
  for (size_t i = 0; i < ps.size(); i++) {
    ts.emplace_back([&ps, &barrier, i]() {
      barrier.wait();
      if (i == (ps.size() / 2)) {
        ps[i].setException(eggs);
      } else {
        ps[i].setValue(i);
      }
    });
  }

  barrier.wait();

  for (auto& t : ts) {
    t.join();
  }

  EXPECT_TRUE(f.isReady());
  for (size_t i = 0; i < ps.size(); i++) {
    if (i == (ps.size() / 2)) {
      EXPECT_THROW(f.value()[i].value(), eggs_t);
    } else {
      EXPECT_TRUE(f.value()[i].hasValue());
      EXPECT_EQ(i, f.value()[i].value());
    }
  }
}

TEST	(	WindowExecutor	,
		basic
	)

Definition at line 276 of file WindowTest.cpp.

References b, folly::pushmi::executor, folly::symbolizer::test::expect(), EXPECT_EQ, EXPECT_TRUE, folly::Try< T >::hasValue(), i, folly::makeFuture(), folly::gen::move, folly::reduce(), s, SCOPED_TRACE, string, sum(), folly::ManualExecutor::waitFor(), and folly::window().

                            {
  ManualExecutor executor;

  // int -> Future<int>
  auto fn = [executor_ = &executor](
                std::vector<int> input, size_t window_size, size_t expect) {
    auto res = reduce(
        window(
            executor_, input, [](int i) { return makeFuture(i); }, window_size),
        0,
        [](int sum, const Try<int>& b) { return sum + *b; });
    executor_->waitFor(res);
    EXPECT_EQ(expect, std::move(res).get());
  };
  {
    SCOPED_TRACE("2 in-flight at a time");
    std::vector<int> input = {1, 2, 3};
    fn(input, 2, 6);
  }
  {
    SCOPED_TRACE("4 in-flight at a time");
    std::vector<int> input = {1, 2, 3};
    fn(input, 4, 6);
  }
  {
    SCOPED_TRACE("empty input");
    std::vector<int> input;
    fn(input, 1, 0);
  }
  {
    // int -> Future<Unit>
    auto res = reduce(
        window(
            &executor,
            std::vector<int>({1, 2, 3}),
            [](int /* i */) { return makeFuture(); },
            2),
        0,
        [](int sum, const Try<Unit>& b) {
          EXPECT_TRUE(b.hasValue());
          return sum + 1;
        });
    executor.waitFor(res);
    EXPECT_EQ(3, std::move(res).get());
  }
  {
    // string -> return Future<int>
    auto res = reduce(
        window(
            &executor,
            std::vector<std::string>{"1", "2", "3"},
            [](std::string s) { return makeFuture<int>(folly::to<int>(s)); },
            2),
        0,
        [](int sum, const Try<int>& b) { return sum + *b; });
    executor.waitFor(res);
    EXPECT_EQ(6, std::move(res).get());
  }
}

TEST	(	WindowExecutor	,
		parallel
	)

Definition at line 336 of file WindowTest.cpp.

References folly::collect(), folly::pushmi::executor, EXPECT_EQ, EXPECT_TRUE, f, i, folly::pushmi::detail::t, folly::ManualExecutor::waitFor(), and folly::window().

                               {
  ManualExecutor executor;

  std::vector<int> input;
  std::vector<Promise<int>> ps(10);
  for (size_t i = 0; i < ps.size(); i++) {
    input.emplace_back(i);
  }
  auto f = collect(
      window(&executor, input, [&](int i) { return ps[i].getFuture(); }, 3));

  std::vector<std::thread> ts;
  boost::barrier barrier(ps.size() + 1);
  for (size_t i = 0; i < ps.size(); i++) {
    ts.emplace_back([&ps, &barrier, i]() {
      barrier.wait();
      ps[i].setValue(i);
    });
  }

  barrier.wait();

  for (auto& t : ts) {
    t.join();
  }

  executor.waitFor(f);
  EXPECT_TRUE(f.isReady());
  for (size_t i = 0; i < ps.size(); i++) {
    EXPECT_EQ(i, f.value()[i]);
  }
}

TEST	(	WindowExecutor	,
		parallelWithError
	)

Definition at line 369 of file WindowTest.cpp.

References folly::collect(), eggs, folly::pushmi::executor, EXPECT_THROW, EXPECT_TRUE, f, i, folly::pushmi::detail::t, folly::ManualExecutor::waitFor(), and folly::window().

                                        {
  ManualExecutor executor;

  std::vector<int> input;
  std::vector<Promise<int>> ps(10);
  for (size_t i = 0; i < ps.size(); i++) {
    input.emplace_back(i);
  }
  auto f = collect(
      window(&executor, input, [&](int i) { return ps[i].getFuture(); }, 3));

  std::vector<std::thread> ts;
  boost::barrier barrier(ps.size() + 1);
  for (size_t i = 0; i < ps.size(); i++) {
    ts.emplace_back([&ps, &barrier, i]() {
      barrier.wait();
      if (i == (ps.size() / 2)) {
        ps[i].setException(eggs);
      } else {
        ps[i].setValue(i);
      }
    });
  }

  barrier.wait();

  for (auto& t : ts) {
    t.join();
  }

  executor.waitFor(f);
  EXPECT_TRUE(f.isReady());
  EXPECT_THROW(f.value(), eggs_t);
}

TEST	(	WindowExecutor	,
		allParallelWithError
	)

Definition at line 404 of file WindowTest.cpp.

References folly::collectAll(), eggs, folly::pushmi::executor, EXPECT_EQ, EXPECT_THROW, EXPECT_TRUE, f, i, folly::pushmi::detail::t, folly::ManualExecutor::waitFor(), and folly::window().

                                           {
  ManualExecutor executor;

  std::vector<int> input;
  std::vector<Promise<int>> ps(10);
  for (size_t i = 0; i < ps.size(); i++) {
    input.emplace_back(i);
  }
  auto f = collectAll(
      window(&executor, input, [&](int i) { return ps[i].getFuture(); }, 3));

  std::vector<std::thread> ts;
  boost::barrier barrier(ps.size() + 1);
  for (size_t i = 0; i < ps.size(); i++) {
    ts.emplace_back([&ps, &barrier, i]() {
      barrier.wait();
      if (i == (ps.size() / 2)) {
        ps[i].setException(eggs);
      } else {
        ps[i].setValue(i);
      }
    });
  }

  barrier.wait();

  for (auto& t : ts) {
    t.join();
  }

  executor.waitFor(f);
  EXPECT_TRUE(f.isReady());
  for (size_t i = 0; i < ps.size(); i++) {
    if (i == (ps.size() / 2)) {
      EXPECT_THROW(f.value()[i].value(), eggs_t);
    } else {
      EXPECT_TRUE(f.value()[i].hasValue());
      EXPECT_EQ(i, f.value()[i].value());
    }
  }
}

Variable Documentation

eggs_t eggs("eggs")

static

Referenced by TEST().