ThreadPoolExecutorTest.cpp File Reference

#include <atomic>
#include <memory>
#include <thread>
#include <boost/thread.hpp>
#include <folly/Exception.h>
#include <folly/VirtualExecutor.h>
#include <folly/executors/CPUThreadPoolExecutor.h>
#include <folly/executors/FutureExecutor.h>
#include <folly/executors/IOThreadPoolExecutor.h>
#include <folly/executors/ThreadPoolExecutor.h>
#include <folly/executors/task_queue/LifoSemMPMCQueue.h>
#include <folly/executors/task_queue/UnboundedBlockingQueue.h>
#include <folly/executors/thread_factory/InitThreadFactory.h>
#include <folly/executors/thread_factory/PriorityThreadFactory.h>
#include <folly/portability/GTest.h>
#include <folly/synchronization/detail/Spin.h>

Go to the source code of this file.

Classes
class	TestObserver
class	TestData
struct	SlowMover
struct	UBQ< T >

Functions
static Func	burnMs (uint64_t ms)
template<class TPE >
static void	basic ()
	TEST (ThreadPoolExecutorTest, CPUBasic)
	TEST (IOThreadPoolExecutorTest, IOBasic)
template<class TPE >
static void	resize ()
	TEST (ThreadPoolExecutorTest, CPUResize)
	TEST (ThreadPoolExecutorTest, IOResize)
template<class TPE >
static void	stop ()
template<>
void	stop< IOThreadPoolExecutor > ()
	TEST (ThreadPoolExecutorTest, CPUStop)
	TEST (ThreadPoolExecutorTest, IOStop)
template<class TPE >
static void	join ()
	TEST (ThreadPoolExecutorTest, CPUJoin)
	TEST (ThreadPoolExecutorTest, IOJoin)
template<class TPE >
static void	destroy ()
template<>
void	destroy< IOThreadPoolExecutor > ()
	TEST (ThreadPoolExecutorTest, CPUDestroy)
	TEST (ThreadPoolExecutorTest, IODestroy)
template<class TPE >
static void	resizeUnderLoad ()
	TEST (ThreadPoolExecutorTest, CPUResizeUnderLoad)
	TEST (ThreadPoolExecutorTest, IOResizeUnderLoad)
template<class TPE >
static void	poolStats ()
	TEST (ThreadPoolExecutorTest, CPUPoolStats)
	TEST (ThreadPoolExecutorTest, IOPoolStats)
template<class TPE >
static void	taskStats ()
	TEST (ThreadPoolExecutorTest, CPUTaskStats)
	TEST (ThreadPoolExecutorTest, IOTaskStats)
template<class TPE >
static void	expiration ()
	TEST (ThreadPoolExecutorTest, CPUExpiration)
	TEST (ThreadPoolExecutorTest, IOExpiration)
template<typename TPE >
static void	futureExecutor ()
	TEST (ThreadPoolExecutorTest, CPUFuturePool)
	TEST (ThreadPoolExecutorTest, IOFuturePool)
	TEST (ThreadPoolExecutorTest, PriorityPreemptionTest)
	TEST (ThreadPoolExecutorTest, IOObserver)
	TEST (ThreadPoolExecutorTest, CPUObserver)
	TEST (ThreadPoolExecutorTest, AddWithPriority)
	TEST (ThreadPoolExecutorTest, BlockingQueue)
	TEST (PriorityThreadFactoryTest, ThreadPriority)
	TEST (InitThreadFactoryTest, InitializerCalled)
	TEST (InitThreadFactoryTest, InitializerAndFinalizerCalled)
	TEST (ThreadPoolExecutorTest, RequestContext)
template<typename Q >
void	bugD3527722_test ()
	TEST (ThreadPoolExecutorTest, LifoSemMPMCQueueBugD3527722)
	TEST (ThreadPoolExecutorTest, UnboundedBlockingQueueBugD3527722)
template<typename TPE >
static void	removeThreadTest ()
	TEST (ThreadPoolExecutorTest, RemoveThreadTestIO)
	TEST (ThreadPoolExecutorTest, RemoveThreadTestCPU)
template<typename TPE >
static void	resizeThreadWhileExecutingTest ()
	TEST (ThreadPoolExecutorTest, resizeThreadWhileExecutingTestIO)
	TEST (ThreadPoolExecutorTest, resizeThreadWhileExecutingTestCPU)
template<typename TPE >
void	keepAliveTest ()
	TEST (ThreadPoolExecutorTest, KeepAliveTestIO)
	TEST (ThreadPoolExecutorTest, KeepAliveTestCPU)
int	getNumThreadPoolExecutors ()
template<typename TPE >
static void	registersToExecutorListTest ()
	TEST (ThreadPoolExecutorTest, registersToExecutorListTestIO)
	TEST (ThreadPoolExecutorTest, registersToExecutorListTestCPU)
template<typename TPE >
static void	testUsesNameFromNamedThreadFactory ()
	TEST (ThreadPoolExecutorTest, testUsesNameFromNamedThreadFactoryIO)
	TEST (ThreadPoolExecutorTest, testUsesNameFromNamedThreadFactoryCPU)
	TEST (ThreadPoolExecutorTest, DynamicThreadsTest)
	TEST (ThreadPoolExecutorTest, DynamicThreadAddRemoveRace)
	TEST (ThreadPoolExecutorTest, AddPerf)
template<typename TPE >
static void	WeakRefTest ()
template<typename TPE >
static void	virtualExecutorTest ()
	TEST (ThreadPoolExecutorTest, WeakRefTestIO)
	TEST (ThreadPoolExecutorTest, WeakRefTestCPU)
	TEST (ThreadPoolExecutorTest, VirtualExecutorTestIO)
	TEST (ThreadPoolExecutorTest, VirtualExecutorTestCPU)

Function Documentation

template<class TPE >

static void basic ( )

static

Definition at line 44 of file ThreadPoolExecutorTest.cpp.

                    {
  // Create and destroy
  TPE tpe(10);
}

template<typename Q >

void bugD3527722_test

(

)

Definition at line 592 of file ThreadPoolExecutorTest.cpp.

                        {
  // Test that the queue does not get stuck if writes are completed in
  // order opposite to how they are initiated.
  Q q(1024);
  std::atomic<int> turn{};

  std::thread consumer1([&] {
    ++turn;
    q.take();
  });
  std::thread consumer2([&] {
    ++turn;
    q.take();
  });

  std::thread producer1([&] {
    ++turn;
    while (turn < 4) {
      ;
    }
    ++turn;
    q.add(SlowMover(true));
  });
  std::thread producer2([&] {
    ++turn;
    while (turn < 5) {
      ;
    }
    q.add(SlowMover(false));
  });

  producer1.join();
  producer2.join();
  consumer1.join();
  consumer2.join();
}

static Func burnMs ( uint64_t  ms )

static

Definition at line 39 of file ThreadPoolExecutorTest.cpp.

Referenced by destroy(), destroy< IOThreadPoolExecutor >(), expiration(), futureExecutor(), join(), RecursiveAddTest(), removeThreadTest(), resizeThreadWhileExecutingTest(), resizeUnderLoad(), SimpleTest(), stop(), stop< IOThreadPoolExecutor >(), taskStats(), TEST(), and WeakRefTest().

                                {
  return [ms]() { std::this_thread::sleep_for(milliseconds(ms)); };
}

template<class TPE >

static void destroy ( )

static

Definition at line 140 of file ThreadPoolExecutorTest.cpp.

References burnMs(), EXPECT_GT, f, and i.

Referenced by folly::coro::detail::InlineTask< T >::Awaiter::await_resume(), folly::coro::TaskWithExecutor< T >::Awaiter::await_resume(), folly::coro::detail::TaskPromiseBase::await_transform(), folly::portability::ssl::BIO_meth_set_destroy(), wangle::AcceptorHandshakeManager::connectionError(), wangle::AcceptorHandshakeManager::connectionReady(), wangle::ServerAcceptor< Pipeline >::ServerConnection::deletePipeline(), folly::UndelayedDestruction< HHWheelTimer >::destroy(), fizz::AsyncFizzBase::destroy(), proxygen::AsyncTimeoutSet::destroy(), folly::Try< T >::emplace(), folly::Try< T >::emplaceException(), folly::threadlocal_detail::StaticMetaBase::erase(), folly::Try< folly::folly::Unit >::get(), folly::NotificationQueue< MessageT >::Consumer::getEventBase(), folly::allocator_delete< Alloc >::operator()(), folly::Try< T >::operator=(), folly::F14VectorSet< Key, Hasher, KeyEqual, Alloc >::operator=(), folly::ThreadLocalPtr< SubscriberMap >::operator=(), folly::F14VectorMap< Key, Mapped, Hasher, KeyEqual, Alloc >::operator=(), folly::dynamic::operator=(), proxygen::PassThroughTransportFilter::readEOF(), proxygen::PassThroughTransportFilter::readErr(), folly::detail::NodeRecycler< NodeType, NodeAlloc, typename std::enable_if< !NodeType::template DestroyIsNoOp< NodeAlloc >::value >::type >::releaseRef(), folly::fbvector< HTTPHeaderCode >::S_destroy_range_a(), proxygen::HTTPSession::shutdownTransportWithReset(), TEST(), test_relinquish3(), unique_ptr_test(), folly::AtomicHashMap< int64_t, int64_t >::~AtomicHashMap(), folly::detail::NodeRecycler< NodeType, NodeAlloc, typename std::enable_if< !NodeType::template DestroyIsNoOp< NodeAlloc >::value >::type >::~NodeRecycler(), and folly::ThreadLocalPtr< SubscriberMap >::~ThreadLocalPtr().

                      {
  TPE tpe(1);
  std::atomic<int> completed(0);
  auto f = [&]() {
    burnMs(10)();
    completed++;
  };
  for (int i = 0; i < 1000; i++) {
    tpe.add(f);
  }
  tpe.stop();
  EXPECT_GT(1000, completed);
}

template<>

void destroy< IOThreadPoolExecutor >

(

)

Definition at line 158 of file ThreadPoolExecutorTest.cpp.

References folly::IOThreadPoolExecutor::add(), burnMs(), folly::Optional< Value >::clear(), EXPECT_EQ, f, i, and folly::in_place().

Referenced by TEST().

                                     {
  Optional<IOThreadPoolExecutor> tpe(in_place, 1);
  std::atomic<int> completed(0);
  auto f = [&]() {
    burnMs(10)();
    completed++;
  };
  for (int i = 0; i < 10; i++) {
    tpe->add(f);
  }
  tpe.clear();
  EXPECT_EQ(10, completed);
}

template<class TPE >

static void expiration ( )

static

Definition at line 266 of file ThreadPoolExecutorTest.cpp.

References burnMs(), EXPECT_EQ, EXPECT_FALSE, EXPECT_TRUE, folly::ThreadPoolExecutor::TaskStats::expired, FAIL, and i.

Referenced by wangle::SSLSessionCacheManager::storeCacheRecord().

                         {
  TPE tpe(1);
  std::atomic<int> statCbCount(0);
  tpe.subscribeToTaskStats([&](ThreadPoolExecutor::TaskStats stats) {
    int i = statCbCount++;
    if (i == 0) {
      EXPECT_FALSE(stats.expired);
    } else if (i == 1) {
      EXPECT_TRUE(stats.expired);
    } else {
      FAIL();
    }
  });
  std::atomic<int> expireCbCount(0);
  auto expireCb = [&]() { expireCbCount++; };
  tpe.add(burnMs(10), seconds(60), expireCb);
  tpe.add(burnMs(10), milliseconds(10), expireCb);
  tpe.join();
  EXPECT_EQ(2, statCbCount);
  EXPECT_EQ(1, expireCbCount);
}

template<typename TPE >

static void futureExecutor ( )

static

Definition at line 297 of file ThreadPoolExecutorTest.cpp.

References folly::FutureExecutor< ExecutorImpl >::addFuture(), burnMs(), c, EXPECT_EQ, EXPECT_NO_THROW, EXPECT_THROW, folly::makeFuture(), folly::Baton< MayBlock, Atom >::post(), folly::pushmi::detail::t, and folly::Baton< MayBlock, Atom >::wait().

                             {
  FutureExecutor<TPE> fe(2);
  std::atomic<int> c{0};
  fe.addFuture([]() { return makeFuture<int>(42); }).then([&](Try<int>&& t) {
    c++;
    EXPECT_EQ(42, t.value());
  });
  fe.addFuture([]() { return 100; }).then([&](Try<int>&& t) {
    c++;
    EXPECT_EQ(100, t.value());
  });
  fe.addFuture([]() { return makeFuture(); }).then([&](Try<Unit>&& t) {
    c++;
    EXPECT_NO_THROW(t.value());
  });
  fe.addFuture([]() { return; }).then([&](Try<Unit>&& t) {
    c++;
    EXPECT_NO_THROW(t.value());
  });
  fe.addFuture([]() { throw std::runtime_error("oops"); })
      .then([&](Try<Unit>&& t) {
        c++;
        EXPECT_THROW(t.value(), std::runtime_error);
      });
  // Test doing actual async work
  folly::Baton<> baton;
  fe.addFuture([&]() {
      auto p = std::make_shared<Promise<int>>();
      std::thread t([p]() {
        burnMs(10)();
        p->setValue(42);
      });
      t.detach();
      return p->getFuture();
    })
      .then([&](Try<int>&& t) {
        EXPECT_EQ(42, t.value());
        c++;
        baton.post();
      });
  baton.wait();
  fe.join();
  EXPECT_EQ(6, c);
}

int getNumThreadPoolExecutors

(

)

Definition at line 731 of file ThreadPoolExecutorTest.cpp.

References count, and folly::ThreadPoolExecutor::withAll().

Referenced by registersToExecutorListTest().

                                {
  int count = 0;
  ThreadPoolExecutor::withAll([&count](ThreadPoolExecutor&) { count++; });
  return count;
}

template<class TPE >

static void join ( )

static

Definition at line 117 of file ThreadPoolExecutorTest.cpp.

References burnMs(), EXPECT_EQ, f, and i.

                   {
  TPE tpe(10);
  std::atomic<int> completed(0);
  auto f = [&]() {
    burnMs(1)();
    completed++;
  };
  for (int i = 0; i < 1000; i++) {
    tpe.add(f);
  }
  tpe.join();
  EXPECT_EQ(1000, completed);
}

template<typename TPE >

void keepAliveTest

(

)

Definition at line 708 of file ThreadPoolExecutorTest.cpp.

References folly::pushmi::executor, EXPECT_EQ, EXPECT_TRUE, f, folly::getKeepAliveToken(), folly::gen::move, folly::futures::sleep(), and folly::via().

                     {
  auto executor = std::make_unique<TPE>(4);

  auto f = futures::sleep(std::chrono::milliseconds{100})
               .via(executor.get())
               .thenValue([keepAlive = getKeepAliveToken(executor.get())](
                              auto&&) { return 42; })
               .semi();

  executor.reset();

  EXPECT_TRUE(f.isReady());
  EXPECT_EQ(42, std::move(f).get());
}

template<class TPE >

static void poolStats ( )

static

Definition at line 206 of file ThreadPoolExecutorTest.cpp.

References EXPECT_EQ, EXPECT_GE, folly::Baton< MayBlock, Atom >::post(), and folly::Baton< MayBlock, Atom >::wait().

                        {
  folly::Baton<> startBaton, endBaton;
  TPE tpe(1);
  auto stats = tpe.getPoolStats();
  EXPECT_GE(1, stats.threadCount);
  EXPECT_GE(1, stats.idleThreadCount);
  EXPECT_EQ(0, stats.activeThreadCount);
  EXPECT_EQ(0, stats.pendingTaskCount);
  EXPECT_EQ(0, tpe.getPendingTaskCount());
  EXPECT_EQ(0, stats.totalTaskCount);
  tpe.add([&]() {
    startBaton.post();
    endBaton.wait();
  });
  tpe.add([&]() {});
  startBaton.wait();
  stats = tpe.getPoolStats();
  EXPECT_EQ(1, stats.threadCount);
  EXPECT_EQ(0, stats.idleThreadCount);
  EXPECT_EQ(1, stats.activeThreadCount);
  EXPECT_EQ(1, stats.pendingTaskCount);
  EXPECT_EQ(1, tpe.getPendingTaskCount());
  EXPECT_EQ(2, stats.totalTaskCount);
  endBaton.post();
}

template<typename TPE >

static void registersToExecutorListTest ( )

static

Definition at line 738 of file ThreadPoolExecutorTest.cpp.

References EXPECT_EQ, and getNumThreadPoolExecutors().

                                          {
  EXPECT_EQ(0, getNumThreadPoolExecutors());
  {
    TPE tpe(10);
    EXPECT_EQ(1, getNumThreadPoolExecutors());
    {
      TPE tpe2(5);
      EXPECT_EQ(2, getNumThreadPoolExecutors());
    }
    EXPECT_EQ(1, getNumThreadPoolExecutors());
  }
  EXPECT_EQ(0, getNumThreadPoolExecutors());
}

template<typename TPE >

static void removeThreadTest ( )

static

Definition at line 643 of file ThreadPoolExecutorTest.cpp.

References burnMs(), EXPECT_EQ, EXPECT_NE, f, folly::makeFuture(), and folly::gen::move.

                               {
  // test that adding a .then() after we have removed some threads
  // doesn't cause deadlock and they are executed on different threads
  folly::Optional<folly::Future<int>> f;
  std::thread::id id1, id2;
  TPE fe(2);
  f = folly::makeFuture()
          .via(&fe)
          .thenValue([&id1](auto&&) {
            burnMs(100)();
            id1 = std::this_thread::get_id();
          })
          .thenValue([&id2](auto&&) {
            return 77;
            id2 = std::this_thread::get_id();
          });
  fe.setNumThreads(1);

  // future::then should be fulfilled because there is other thread available
  EXPECT_EQ(77, std::move(*f).get());
  // two thread should be different because then part should be rescheduled to
  // the other thread
  EXPECT_NE(id1, id2);
}

template<class TPE >

static void resize ( )

static

Definition at line 58 of file ThreadPoolExecutorTest.cpp.

References EXPECT_EQ.

                     {
  TPE tpe(100);
  EXPECT_EQ(100, tpe.numThreads());
  tpe.setNumThreads(50);
  EXPECT_EQ(50, tpe.numThreads());
  tpe.setNumThreads(150);
  EXPECT_EQ(150, tpe.numThreads());
}

template<typename TPE >

static void resizeThreadWhileExecutingTest ( )

static

Definition at line 677 of file ThreadPoolExecutorTest.cpp.

References burnMs(), EXPECT_EQ, f, and i.

                                             {
  TPE tpe(10);
  EXPECT_EQ(10, tpe.numThreads());

  std::atomic<int> completed(0);
  auto f = [&]() {
    burnMs(10)();
    completed++;
  };
  for (int i = 0; i < 1000; i++) {
    tpe.add(f);
  }
  tpe.setNumThreads(8);
  EXPECT_EQ(8, tpe.numThreads());
  tpe.setNumThreads(5);
  EXPECT_EQ(5, tpe.numThreads());
  tpe.setNumThreads(15);
  EXPECT_EQ(15, tpe.numThreads());
  tpe.join();
  EXPECT_EQ(1000, completed);
}

template<class TPE >

static void resizeUnderLoad ( )

static

Definition at line 181 of file ThreadPoolExecutorTest.cpp.

References burnMs(), EXPECT_EQ, f, and i.

                              {
  TPE tpe(10);
  std::atomic<int> completed(0);
  auto f = [&]() {
    burnMs(1)();
    completed++;
  };
  for (int i = 0; i < 1000; i++) {
    tpe.add(f);
  }
  tpe.setNumThreads(5);
  tpe.setNumThreads(15);
  tpe.join();
  EXPECT_EQ(1000, completed);
}

template<class TPE >

static void stop ( )

static

Definition at line 76 of file ThreadPoolExecutorTest.cpp.

References burnMs(), EXPECT_GT, f, and i.

Referenced by accuracy_test(), ConcurrentFlowManySender::cancellation_test(), ConcurrentFlowSingleSender::cancellation_test(), proxygen::HTTP2PriorityQueue::Node::iterate(), proxygen::HTTP2PriorityQueue::iterateBFS(), ImmediateFlowSingleSender::make_producer(), ImmediateFlowManySender::make_producer(), proxygen::HTTP2PriorityQueue::nextEgress(), shared_cancellation_factory::operator()(), entangled_cancellation_factory::operator()(), proxygen::GzipHeaderCodec::parseNameValues(), proxygen::HTTP2PriorityQueue::Node::propagatePendingEgressClear(), proxygen::HTTP2PriorityQueue::Node::propagatePendingEgressSignal(), folly::AsyncServerSocket::RemoteAcceptor::RemoteAcceptor(), runFairness(), inline_executor_flow_many::submit(), TEST(), TEST_F(), and wangle::FilePoller::~FilePoller().

                   {
  TPE tpe(1);
  std::atomic<int> completed(0);
  auto f = [&]() {
    burnMs(10)();
    completed++;
  };
  for (int i = 0; i < 1000; i++) {
    tpe.add(f);
  }
  tpe.stop();
  EXPECT_GT(1000, completed);
}

template<>

void stop< IOThreadPoolExecutor >

(

)

Definition at line 94 of file ThreadPoolExecutorTest.cpp.

References folly::IOThreadPoolExecutor::add(), burnMs(), EXPECT_EQ, f, i, and folly::ThreadPoolExecutor::stop().

Referenced by TEST().

                                  {
  IOThreadPoolExecutor tpe(1);
  std::atomic<int> completed(0);
  auto f = [&]() {
    burnMs(10)();
    completed++;
  };
  for (int i = 0; i < 10; i++) {
    tpe.add(f);
  }
  tpe.stop();
  EXPECT_EQ(10, completed);
}

template<class TPE >

static void taskStats ( )

static

Definition at line 241 of file ThreadPoolExecutorTest.cpp.

References burnMs(), c, EXPECT_EQ, EXPECT_LT, i, folly::ThreadPoolExecutor::TaskStats::runTime, and folly::ThreadPoolExecutor::TaskStats::waitTime.

                        {
  TPE tpe(1);
  std::atomic<int> c(0);
  tpe.subscribeToTaskStats([&](ThreadPoolExecutor::TaskStats stats) {
    int i = c++;
    EXPECT_LT(milliseconds(0), stats.runTime);
    if (i == 1) {
      EXPECT_LT(milliseconds(0), stats.waitTime);
    }
  });
  tpe.add(burnMs(10));
  tpe.add(burnMs(10));
  tpe.join();
  EXPECT_EQ(2, c);
}

TEST	(	ThreadPoolExecutorTest	,
		CPUBasic
	)

Definition at line 49 of file ThreadPoolExecutorTest.cpp.

                                       {
  basic<CPUThreadPoolExecutor>();
}

TEST	(	IOThreadPoolExecutorTest	,
		IOBasic
	)

Definition at line 53 of file ThreadPoolExecutorTest.cpp.

                                        {
  basic<IOThreadPoolExecutor>();
}

TEST	(	ThreadPoolExecutorTest	,
		CPUResize
	)

Definition at line 67 of file ThreadPoolExecutorTest.cpp.

                                        {
  resize<CPUThreadPoolExecutor>();
}

TEST	(	ThreadPoolExecutorTest	,
		IOResize
	)

Definition at line 71 of file ThreadPoolExecutorTest.cpp.

                                       {
  resize<IOThreadPoolExecutor>();
}

TEST	(	ThreadPoolExecutorTest	,
		CPUStop
	)

Definition at line 108 of file ThreadPoolExecutorTest.cpp.

                                      {
  stop<CPUThreadPoolExecutor>();
}

TEST	(	ThreadPoolExecutorTest	,
		IOStop
	)

Definition at line 112 of file ThreadPoolExecutorTest.cpp.

References stop< IOThreadPoolExecutor >().

                                     {
  stop<IOThreadPoolExecutor>();
}

TEST	(	ThreadPoolExecutorTest	,
		CPUJoin
	)

Definition at line 131 of file ThreadPoolExecutorTest.cpp.

                                      {
  join<CPUThreadPoolExecutor>();
}

TEST	(	ThreadPoolExecutorTest	,
		IOJoin
	)

Definition at line 135 of file ThreadPoolExecutorTest.cpp.

                                     {
  join<IOThreadPoolExecutor>();
}

TEST	(	ThreadPoolExecutorTest	,
		CPUDestroy
	)

Definition at line 172 of file ThreadPoolExecutorTest.cpp.

                                         {
  destroy<CPUThreadPoolExecutor>();
}

TEST	(	ThreadPoolExecutorTest	,
		IODestroy
	)

Definition at line 176 of file ThreadPoolExecutorTest.cpp.

References destroy< IOThreadPoolExecutor >().

                                        {
  destroy<IOThreadPoolExecutor>();
}

TEST	(	ThreadPoolExecutorTest	,
		CPUResizeUnderLoad
	)

Definition at line 197 of file ThreadPoolExecutorTest.cpp.

                                                 {
  resizeUnderLoad<CPUThreadPoolExecutor>();
}

TEST	(	ThreadPoolExecutorTest	,
		IOResizeUnderLoad
	)

Definition at line 201 of file ThreadPoolExecutorTest.cpp.

                                                {
  resizeUnderLoad<IOThreadPoolExecutor>();
}

TEST	(	ThreadPoolExecutorTest	,
		CPUPoolStats
	)

Definition at line 232 of file ThreadPoolExecutorTest.cpp.

                                           {
  poolStats<CPUThreadPoolExecutor>();
}

TEST	(	ThreadPoolExecutorTest	,
		IOPoolStats
	)

Definition at line 236 of file ThreadPoolExecutorTest.cpp.

                                          {
  poolStats<IOThreadPoolExecutor>();
}

TEST	(	ThreadPoolExecutorTest	,
		CPUTaskStats
	)

Definition at line 257 of file ThreadPoolExecutorTest.cpp.

                                           {
  taskStats<CPUThreadPoolExecutor>();
}

TEST	(	ThreadPoolExecutorTest	,
		IOTaskStats
	)

Definition at line 261 of file ThreadPoolExecutorTest.cpp.

                                          {
  taskStats<IOThreadPoolExecutor>();
}

TEST	(	ThreadPoolExecutorTest	,
		CPUExpiration
	)

Definition at line 288 of file ThreadPoolExecutorTest.cpp.

                                            {
  expiration<CPUThreadPoolExecutor>();
}

TEST	(	ThreadPoolExecutorTest	,
		IOExpiration
	)

Definition at line 292 of file ThreadPoolExecutorTest.cpp.

                                           {
  expiration<IOThreadPoolExecutor>();
}

TEST	(	ThreadPoolExecutorTest	,
		CPUFuturePool
	)

Definition at line 342 of file ThreadPoolExecutorTest.cpp.

                                            {
  futureExecutor<CPUThreadPoolExecutor>();
}

TEST	(	ThreadPoolExecutorTest	,
		IOFuturePool
	)

Definition at line 346 of file ThreadPoolExecutorTest.cpp.

                                           {
  futureExecutor<IOThreadPoolExecutor>();
}

TEST	(	ThreadPoolExecutorTest	,
		PriorityPreemptionTest
	)

Definition at line 350 of file ThreadPoolExecutorTest.cpp.

References folly::VirtualExecutor::addWithPriority(), EXPECT_EQ, EXPECT_FALSE, folly::Executor::HI_PRI, i, folly::Executor::LO_PRI, and folly::ThreadPoolExecutor::setNumThreads().

                                                     {
  bool tookLopri = false;
  auto completed = 0;
  auto hipri = [&] {
    EXPECT_FALSE(tookLopri);
    completed++;
  };
  auto lopri = [&] {
    tookLopri = true;
    completed++;
  };
  CPUThreadPoolExecutor pool(0, 2);
  {
    VirtualExecutor ve(pool);
    for (int i = 0; i < 50; i++) {
      ve.addWithPriority(lopri, Executor::LO_PRI);
    }
    for (int i = 0; i < 50; i++) {
      ve.addWithPriority(hipri, Executor::HI_PRI);
    }
    pool.setNumThreads(1);
  }
  EXPECT_EQ(100, completed);
}

TEST	(	ThreadPoolExecutorTest	,
		IOObserver
	)

Definition at line 397 of file ThreadPoolExecutorTest.cpp.

References folly::ThreadPoolExecutor::addObserver(), exe, folly::ThreadPoolExecutor::removeObserver(), and folly::ThreadPoolExecutor::setNumThreads().

                                         {
  auto observer = std::make_shared<TestObserver>();

  {
    IOThreadPoolExecutor exe(10);
    exe.addObserver(observer);
    exe.setNumThreads(3);
    exe.setNumThreads(0);
    exe.setNumThreads(7);
    exe.removeObserver(observer);
    exe.setNumThreads(10);
  }

  observer->checkCalls();
}

TEST	(	ThreadPoolExecutorTest	,
		CPUObserver
	)

Definition at line 413 of file ThreadPoolExecutorTest.cpp.

References folly::ThreadPoolExecutor::addObserver(), exe, folly::ThreadPoolExecutor::removeObserver(), and folly::ThreadPoolExecutor::setNumThreads().

                                          {
  auto observer = std::make_shared<TestObserver>();

  {
    CPUThreadPoolExecutor exe(10);
    exe.addObserver(observer);
    exe.setNumThreads(3);
    exe.setNumThreads(0);
    exe.setNumThreads(7);
    exe.removeObserver(observer);
    exe.setNumThreads(10);
  }

  observer->checkCalls();
}

TEST	(	ThreadPoolExecutorTest	,
		AddWithPriority
	)

Definition at line 429 of file ThreadPoolExecutorTest.cpp.

References folly::Executor::addWithPriority(), c, EXPECT_EQ, EXPECT_THROW, f, folly::Executor::HI_PRI, and folly::Executor::LO_PRI.

                                              {
  std::atomic_int c{0};
  auto f = [&] { c++; };

  // IO exe doesn't support priorities
  IOThreadPoolExecutor ioExe(10);
  EXPECT_THROW(ioExe.addWithPriority(f, 0), std::runtime_error);

  CPUThreadPoolExecutor cpuExe(10, 3);
  cpuExe.addWithPriority(f, -1);
  cpuExe.addWithPriority(f, 0);
  cpuExe.addWithPriority(f, 1);
  cpuExe.addWithPriority(f, -2); // will add at the lowest priority
  cpuExe.addWithPriority(f, 2); // will add at the highest priority
  cpuExe.addWithPriority(f, Executor::LO_PRI);
  cpuExe.addWithPriority(f, Executor::HI_PRI);
  cpuExe.join();

  EXPECT_EQ(7, c);
}

TEST	(	ThreadPoolExecutorTest	,
		BlockingQueue
	)

Definition at line 450 of file ThreadPoolExecutorTest.cpp.

References folly::BLOCK, burnMs(), c, EXPECT_EQ, EXPECT_NO_THROW, f, i, folly::make_unique(), and folly::gen::move.

                                            {
  std::atomic_int c{0};
  auto f = [&] {
    burnMs(1)();
    c++;
  };
  const int kQueueCapacity = 1;
  const int kThreads = 1;

  auto queue = std::make_unique<LifoSemMPMCQueue<
      CPUThreadPoolExecutor::CPUTask,
      QueueBehaviorIfFull::BLOCK>>(kQueueCapacity);

  CPUThreadPoolExecutor cpuExe(
      kThreads,
      std::move(queue),
      std::make_shared<NamedThreadFactory>("CPUThreadPool"));

  // Add `f` five times. It sleeps for 1ms every time. Calling
  // `cppExec.add()` is *almost* guaranteed to block because there's
  // only 1 cpu worker thread.
  for (int i = 0; i < 5; i++) {
    EXPECT_NO_THROW(cpuExe.add(f));
  }
  cpuExe.join();

  EXPECT_EQ(5, c);
}

TEST	(	PriorityThreadFactoryTest	,
		ThreadPriority
	)

Definition at line 479 of file ThreadPoolExecutorTest.cpp.

References EXPECT_EQ, folly::join(), min, folly::PriorityThreadFactory::newThread(), and folly::throwSystemError().

                                                {
  errno = 0;
  auto currentPriority = getpriority(PRIO_PROCESS, 0);
  if (errno != 0) {
    throwSystemError("failed to get current priority");
  }

  // Non-root users can only increase the priority value.  Make sure we are
  // trying to go to a higher priority than we are currently running as, up to
  // the maximum allowed of 20.
  int desiredPriority = std::min(20, currentPriority + 1);

  PriorityThreadFactory factory(
      std::make_shared<NamedThreadFactory>("stuff"), desiredPriority);
  int actualPriority = -21;
  factory.newThread([&]() { actualPriority = getpriority(PRIO_PROCESS, 0); })
      .join();
  EXPECT_EQ(desiredPriority, actualPriority);
}

TEST	(	InitThreadFactoryTest	,
		InitializerCalled
	)

Definition at line 499 of file ThreadPoolExecutorTest.cpp.

References EXPECT_EQ, folly::join(), and folly::InitThreadFactory::newThread().

                                               {
  int initializerCalledCount = 0;
  InitThreadFactory factory(
      std::make_shared<NamedThreadFactory>("test"),
      [&initializerCalledCount] { initializerCalledCount++; });
  factory
      .newThread(
          [&initializerCalledCount]() { EXPECT_EQ(initializerCalledCount, 1); })
      .join();
  EXPECT_EQ(initializerCalledCount, 1);
}

TEST	(	InitThreadFactoryTest	,
		InitializerAndFinalizerCalled
	)

Definition at line 511 of file ThreadPoolExecutorTest.cpp.

References EXPECT_FALSE, EXPECT_TRUE, folly::join(), and folly::InitThreadFactory::newThread().

                                                           {
  bool initializerCalled = false;
  bool taskBodyCalled = false;
  bool finalizerCalled = false;

  InitThreadFactory factory(
      std::make_shared<NamedThreadFactory>("test"),
      [&] {
        // thread initializer
        EXPECT_FALSE(initializerCalled);
        EXPECT_FALSE(taskBodyCalled);
        EXPECT_FALSE(finalizerCalled);
        initializerCalled = true;
      },
      [&] {
        // thread finalizer
        EXPECT_TRUE(initializerCalled);
        EXPECT_TRUE(taskBodyCalled);
        EXPECT_FALSE(finalizerCalled);
        finalizerCalled = true;
      });

  factory
      .newThread([&]() {
        EXPECT_TRUE(initializerCalled);
        EXPECT_FALSE(taskBodyCalled);
        EXPECT_FALSE(finalizerCalled);
        taskBodyCalled = true;
      })
      .join();

  EXPECT_TRUE(initializerCalled);
  EXPECT_TRUE(taskBodyCalled);
  EXPECT_TRUE(finalizerCalled);
}

TEST	(	ThreadPoolExecutorTest	,
		RequestContext
	)

Definition at line 559 of file ThreadPoolExecutorTest.cpp.

References folly::CPUThreadPoolExecutor::add(), ASSERT_TRUE, folly::data(), data_, folly::pushmi::executor, EXPECT_EQ, folly::RequestContext::get(), folly::RequestContext::getContextData(), and folly::RequestContext::setContextData().

                                             {
  CPUThreadPoolExecutor executor(1);

  RequestContextScopeGuard rctx; // create new request context for this scope
  EXPECT_EQ(nullptr, RequestContext::get()->getContextData("test"));
  RequestContext::get()->setContextData("test", std::make_unique<TestData>(42));
  auto data = RequestContext::get()->getContextData("test");
  EXPECT_EQ(42, dynamic_cast<TestData*>(data)->data_);

  executor.add([] {
    auto data2 = RequestContext::get()->getContextData("test");
    ASSERT_TRUE(data2 != nullptr);
    EXPECT_EQ(42, dynamic_cast<TestData*>(data2)->data_);
  });
}

TEST	(	ThreadPoolExecutorTest	,
		LifoSemMPMCQueueBugD3527722
	)

Definition at line 629 of file ThreadPoolExecutorTest.cpp.

                                                          {
  bugD3527722_test<LifoSemMPMCQueue<SlowMover>>();
}

TEST	(	ThreadPoolExecutorTest	,
		UnboundedBlockingQueueBugD3527722
	)

Definition at line 638 of file ThreadPoolExecutorTest.cpp.

                                                                {
  bugD3527722_test<UBQ<SlowMover>>();
}

TEST	(	ThreadPoolExecutorTest	,
		RemoveThreadTestIO
	)

Definition at line 668 of file ThreadPoolExecutorTest.cpp.

                                                 {
  removeThreadTest<IOThreadPoolExecutor>();
}

TEST	(	ThreadPoolExecutorTest	,
		RemoveThreadTestCPU
	)

Definition at line 672 of file ThreadPoolExecutorTest.cpp.

                                                  {
  removeThreadTest<CPUThreadPoolExecutor>();
}

TEST	(	ThreadPoolExecutorTest	,
		resizeThreadWhileExecutingTestIO
	)

Definition at line 699 of file ThreadPoolExecutorTest.cpp.

                                                               {
  resizeThreadWhileExecutingTest<IOThreadPoolExecutor>();
}

TEST	(	ThreadPoolExecutorTest	,
		resizeThreadWhileExecutingTestCPU
	)

Definition at line 703 of file ThreadPoolExecutorTest.cpp.

                                                                {
  resizeThreadWhileExecutingTest<CPUThreadPoolExecutor>();
}

TEST	(	ThreadPoolExecutorTest	,
		KeepAliveTestIO
	)

Definition at line 723 of file ThreadPoolExecutorTest.cpp.

                                              {
  keepAliveTest<IOThreadPoolExecutor>();
}

TEST	(	ThreadPoolExecutorTest	,
		KeepAliveTestCPU
	)

Definition at line 727 of file ThreadPoolExecutorTest.cpp.

                                               {
  keepAliveTest<CPUThreadPoolExecutor>();
}

TEST	(	ThreadPoolExecutorTest	,
		registersToExecutorListTestIO
	)

Definition at line 752 of file ThreadPoolExecutorTest.cpp.

                                                            {
  registersToExecutorListTest<IOThreadPoolExecutor>();
}

TEST	(	ThreadPoolExecutorTest	,
		registersToExecutorListTestCPU
	)

Definition at line 756 of file ThreadPoolExecutorTest.cpp.

                                                             {
  registersToExecutorListTest<CPUThreadPoolExecutor>();
}

TEST	(	ThreadPoolExecutorTest	,
		testUsesNameFromNamedThreadFactoryIO
	)

Definition at line 767 of file ThreadPoolExecutorTest.cpp.

                                                                   {
  testUsesNameFromNamedThreadFactory<IOThreadPoolExecutor>();
}

TEST	(	ThreadPoolExecutorTest	,
		testUsesNameFromNamedThreadFactoryCPU
	)

Definition at line 771 of file ThreadPoolExecutorTest.cpp.

                                                                    {
  testUsesNameFromNamedThreadFactory<CPUThreadPoolExecutor>();
}

TEST	(	ThreadPoolExecutorTest	,
		DynamicThreadsTest
	)

Definition at line 775 of file ThreadPoolExecutorTest.cpp.

References folly::ThreadPoolExecutor::PoolStats::activeThreadCount, folly::CPUThreadPoolExecutor::add(), EXPECT_EQ, EXPECT_FALSE, folly::ThreadPoolExecutor::getPoolStats(), now(), folly::ThreadPoolExecutor::setThreadDeathTimeout(), folly::detail::spin_yield_until(), and folly::detail::success.

                                                 {
  boost::barrier barrier{3};
  auto twice_waiting_task = [&] { barrier.wait(), barrier.wait(); };
  CPUThreadPoolExecutor e(2);
  e.setThreadDeathTimeout(std::chrono::milliseconds(100));
  e.add(twice_waiting_task);
  e.add(twice_waiting_task);
  barrier.wait(); // ensure both tasks are mid-flight
  EXPECT_EQ(2, e.getPoolStats().activeThreadCount) << "sanity check";

  auto pred = [&] { return e.getPoolStats().activeThreadCount == 0; };
  EXPECT_FALSE(pred()) << "sanity check";
  barrier.wait(); // let both mid-flight tasks complete
  EXPECT_EQ(
      folly::detail::spin_result::success,
      folly::detail::spin_yield_until(
          std::chrono::steady_clock::now() + std::chrono::seconds(1), pred));
}

TEST	(	ThreadPoolExecutorTest	,
		DynamicThreadAddRemoveRace
	)

Definition at line 794 of file ThreadPoolExecutorTest.cpp.

References folly::CPUThreadPoolExecutor::add(), b, count, EXPECT_EQ, i, folly::ThreadPoolExecutor::join(), folly::Baton< MayBlock, Atom >::post(), folly::ThreadPoolExecutor::setThreadDeathTimeout(), and folly::Baton< MayBlock, Atom >::wait().

                                                         {
  CPUThreadPoolExecutor e(1);
  e.setThreadDeathTimeout(std::chrono::milliseconds(0));
  std::atomic<uint64_t> count{0};
  for (int i = 0; i < 10000; i++) {
    Baton<> b;
    e.add([&]() {
      count.fetch_add(1, std::memory_order_relaxed);
      b.post();
    });
    b.wait();
  }
  e.join();
  EXPECT_EQ(count, 10000);
}

TEST	(	ThreadPoolExecutorTest	,
		AddPerf
	)

Definition at line 810 of file ThreadPoolExecutorTest.cpp.

References i, folly::make_unique(), and folly::gen::move.

                                      {
  auto queue = std::make_unique<
      UnboundedBlockingQueue<CPUThreadPoolExecutor::CPUTask>>();
  CPUThreadPoolExecutor e(
      1000,
      std::move(queue),
      std::make_shared<NamedThreadFactory>("CPUThreadPool"));
  e.setThreadDeathTimeout(std::chrono::milliseconds(1));
  for (int i = 0; i < 10000; i++) {
    e.add([&]() { e.add([]() { /* sleep override */ usleep(1000); }); });
  }
  e.stop();
}

TEST	(	ThreadPoolExecutorTest	,
		WeakRefTestIO
	)

Definition at line 886 of file ThreadPoolExecutorTest.cpp.

                                            {
  WeakRefTest<IOThreadPoolExecutor>();
}

TEST	(	ThreadPoolExecutorTest	,
		WeakRefTestCPU
	)

Definition at line 890 of file ThreadPoolExecutorTest.cpp.

                                             {
  WeakRefTest<CPUThreadPoolExecutor>();
}

TEST	(	ThreadPoolExecutorTest	,
		VirtualExecutorTestIO
	)

Definition at line 894 of file ThreadPoolExecutorTest.cpp.

                                                    {
  virtualExecutorTest<IOThreadPoolExecutor>();
}

TEST	(	ThreadPoolExecutorTest	,
		VirtualExecutorTestCPU
	)

Definition at line 898 of file ThreadPoolExecutorTest.cpp.

                                                     {
  virtualExecutorTest<CPUThreadPoolExecutor>();
}

template<typename TPE >

static void testUsesNameFromNamedThreadFactory ( )

static

Definition at line 761 of file ThreadPoolExecutorTest.cpp.

References EXPECT_EQ.

                                                 {
  auto ntf = std::make_shared<NamedThreadFactory>("my_executor");
  TPE tpe(10, ntf);
  EXPECT_EQ("my_executor", tpe.getName());
}

template<typename TPE >

static void virtualExecutorTest ( )

static

Definition at line 845 of file ThreadPoolExecutorTest.cpp.

References folly::VirtualExecutor::add(), counter, EXPECT_EQ, EXPECT_NO_THROW, EXPECT_TRUE, f, folly::gen::guard(), folly::makeGuard(), folly::gen::move, folly::futures::sleep(), and folly::via().

                                  {
  using namespace std::literals;

  folly::Optional<folly::SemiFuture<folly::Unit>> f;
  int counter{0};
  {
    TPE fe(1);
    {
      VirtualExecutor ve(fe);
      f = futures::sleep(100ms)
              .via(&ve)
              .thenValue([&](auto&&) {
                ++counter;
                return futures::sleep(100ms);
              })
              .via(&fe)
              .thenValue([&](auto&&) { ++counter; })
              .semi();
    }
    EXPECT_EQ(1, counter);

    bool functionDestroyed{false};
    bool functionCalled{false};
    {
      VirtualExecutor ve(fe);
      auto guard = makeGuard([&functionDestroyed] {
        std::this_thread::sleep_for(100ms);
        functionDestroyed = true;
      });
      ve.add([&functionCalled, guard = std::move(guard)] {
        functionCalled = true;
      });
    }
    EXPECT_TRUE(functionCalled);
    EXPECT_TRUE(functionDestroyed);
  }
  EXPECT_TRUE(f->isReady());
  EXPECT_NO_THROW(std::move(*f).get());
  EXPECT_EQ(2, counter);
}

template<typename TPE >

static void WeakRefTest ( )

static

Definition at line 825 of file ThreadPoolExecutorTest.cpp.

References burnMs(), counter, EXPECT_EQ, EXPECT_THROW, f, folly::makeFuture(), folly::gen::move, and folly::via().

                          {
  // test that adding a .then() after we have
  // started shutting down does not deadlock
  folly::Optional<folly::Future<folly::Unit>> f;
  int counter{0};
  {
    TPE fe(1);
    f = folly::makeFuture()
            .via(&fe)
            .thenValue([](auto&&) { burnMs(100)(); })
            .thenValue([&](auto&&) { ++counter; })
            .via(fe.weakRef())
            .thenValue([](auto&&) { burnMs(100)(); })
            .thenValue([&](auto&&) { ++counter; });
  }
  EXPECT_THROW(std::move(*f).get(), folly::BrokenPromise);
  EXPECT_EQ(1, counter);
}