NotificationQueueTest.cpp File Reference

#include <folly/io/async/NotificationQueue.h>
#include <sys/types.h>
#include <iostream>
#include <list>
#include <thread>
#include <folly/io/async/ScopedEventBaseThread.h>
#include <folly/portability/GTest.h>
#include <folly/synchronization/Baton.h>
#include <sys/wait.h>

Go to the source code of this file.

Classes
class	QueueConsumer
class	QueueTest

Typedefs
typedef NotificationQueue< int >	IntQueue

Functions
	TEST (NotificationQueueTest, ConsumeUntilDrained)
	TEST (NotificationQueueTest, ConsumeUntilDrainedStress)
	TEST (NotificationQueueTest, SendOnePipe)
	TEST (NotificationQueueTest, PutMessagesPipe)
	TEST (NotificationQueueTest, MultiConsumerPipe)
	TEST (NotificationQueueTest, MaxQueueSizePipe)
	TEST (NotificationQueueTest, MaxReadAtOncePipe)
	TEST (NotificationQueueTest, DestroyCallbackPipe)
	TEST (NotificationQueueTest, UseAfterFork)
	TEST (NotificationQueueConsumer, make)

Typedef Documentation

typedef NotificationQueue<int> IntQueue

Definition at line 36 of file NotificationQueueTest.cpp.

Function Documentation

TEST	(	NotificationQueueTest	,
		ConsumeUntilDrained
	)

Definition at line 408 of file NotificationQueueTest.cpp.

References folly::NotificationQueue< MessageT >::Consumer::consumeUntilDrained(), EXPECT_EQ, EXPECT_FALSE, EXPECT_THROW, EXPECT_TRUE, QueueConsumer::fn, i, QueueConsumer::messages, folly::NotificationQueue< MessageT >::putMessage(), folly::NotificationQueue< MessageT >::putMessages(), folly::NotificationQueue< MessageT >::Consumer::setMaxReadAtOnce(), folly::NotificationQueue< MessageT >::Consumer::startConsuming(), folly::NotificationQueue< MessageT >::tryPutMessage(), and folly::NotificationQueue< MessageT >::tryPutMessageNoThrow().

Referenced by TEST().

                                                 {
  // Basic tests: make sure we
  // - drain all the messages
  // - ignore any maxReadAtOnce
  // - can't add messages during draining
  EventBase eventBase;
  IntQueue queue;
  QueueConsumer consumer;
  consumer.fn = [&](int i) {
    EXPECT_THROW(queue.tryPutMessage(i), std::runtime_error);
    EXPECT_FALSE(queue.tryPutMessageNoThrow(i));
    EXPECT_THROW(queue.putMessage(i), std::runtime_error);
    std::vector<int> ints{1, 2, 3};
    EXPECT_THROW(
        queue.putMessages(ints.begin(), ints.end()), std::runtime_error);
  };
  consumer.setMaxReadAtOnce(10); // We should ignore this
  consumer.startConsuming(&eventBase, &queue);
  for (int i = 0; i < 20; i++) {
    queue.putMessage(i);
  }
  EXPECT_TRUE(consumer.consumeUntilDrained());
  EXPECT_EQ(20, consumer.messages.size());

  // Make sure there can only be one drainer at once
  folly::Baton<> callbackBaton, threadStartBaton;
  consumer.fn = [&](int /* i */) { callbackBaton.wait(); };
  QueueConsumer competingConsumer;
  competingConsumer.startConsuming(&eventBase, &queue);
  queue.putMessage(1);
  atomic<bool> raceA{false};
  atomic<bool> raceB{false};
  size_t numConsA = 0;
  size_t numConsB = 0;
  auto thread = std::thread([&] {
    threadStartBaton.post();
    raceB = consumer.consumeUntilDrained(&numConsB) && numConsB;
  });
  threadStartBaton.wait();
  raceA = competingConsumer.consumeUntilDrained(&numConsA) && numConsA;
  callbackBaton.post();
  thread.join();
  EXPECT_FALSE(raceA && raceB);
  EXPECT_TRUE(raceA || raceB);
  EXPECT_TRUE(raceA ^ raceB);
}

TEST	(	NotificationQueueTest	,
		ConsumeUntilDrainedStress
	)

Definition at line 455 of file NotificationQueueTest.cpp.

References folly::NotificationQueue< MessageT >::Consumer::consumeUntilDrained(), QueueTest::destroyCallback(), EXPECT_EQ, EXPECT_FALSE, EXPECT_THROW, EXPECT_TRUE, QueueConsumer::fn, i, QueueTest::maxQueueSize(), QueueTest::maxReadAtOnce(), QueueConsumer::messages, QueueTest::multiConsumer(), folly::NotificationQueue< MessageT >::putMessage(), folly::NotificationQueue< MessageT >::putMessages(), QueueTest::sendOne(), folly::NotificationQueue< MessageT >::Consumer::setMaxReadAtOnce(), folly::NotificationQueue< MessageT >::Consumer::startConsuming(), TEST(), folly::NotificationQueue< MessageT >::tryPutMessage(), and folly::NotificationQueue< MessageT >::tryPutMessageNoThrow().

                                                       {
  for (size_t i = 0; i < 1 << 8; ++i) {
    // Basic tests: make sure we
    // - drain all the messages
    // - ignore any maxReadAtOnce
    // - can't add messages during draining
    EventBase eventBase;
    IntQueue queue;
    QueueConsumer consumer;
    consumer.fn = [&](int j) {
      EXPECT_THROW(queue.tryPutMessage(j), std::runtime_error);
      EXPECT_FALSE(queue.tryPutMessageNoThrow(j));
      EXPECT_THROW(queue.putMessage(j), std::runtime_error);
      std::vector<int> ints{1, 2, 3};
      EXPECT_THROW(
          queue.putMessages(ints.begin(), ints.end()), std::runtime_error);
    };
    consumer.setMaxReadAtOnce(10); // We should ignore this
    consumer.startConsuming(&eventBase, &queue);
    for (int j = 0; j < 20; j++) {
      queue.putMessage(j);
    }
    EXPECT_TRUE(consumer.consumeUntilDrained());
    EXPECT_EQ(20, consumer.messages.size());

    // Make sure there can only be one drainer at once
    folly::Baton<> callbackBaton, threadStartBaton;
    consumer.fn = [&](int /* i */) { callbackBaton.wait(); };
    QueueConsumer competingConsumer;
    competingConsumer.startConsuming(&eventBase, &queue);
    queue.putMessage(1);
    atomic<bool> raceA{false};
    atomic<bool> raceB{false};
    size_t numConsA = 0;
    size_t numConsB = 0;
    auto thread = std::thread([&] {
      threadStartBaton.post();
      raceB = consumer.consumeUntilDrained(&numConsB) && numConsB;
    });
    threadStartBaton.wait();
    raceA = competingConsumer.consumeUntilDrained(&numConsA) && numConsA;
    callbackBaton.post();
    thread.join();
    EXPECT_FALSE(raceA && raceB);
    EXPECT_TRUE(raceA || raceB);
    EXPECT_TRUE(raceA ^ raceB);
  }
}

TEST	(	NotificationQueueTest	,
		SendOnePipe
	)

Definition at line 536 of file NotificationQueueTest.cpp.

References folly::NotificationQueue< MessageT >::PIPE, and QueueTest::sendOne().

                                         {
  QueueTest qt(0, IntQueue::FdType::PIPE);
  qt.sendOne();
}

TEST	(	NotificationQueueTest	,
		PutMessagesPipe
	)

Definition at line 541 of file NotificationQueueTest.cpp.

References folly::NotificationQueue< MessageT >::PIPE, and QueueTest::sendOne().

                                             {
  QueueTest qt(0, IntQueue::FdType::PIPE);
  qt.sendOne();
}

TEST	(	NotificationQueueTest	,
		MultiConsumerPipe
	)

Definition at line 546 of file NotificationQueueTest.cpp.

References QueueTest::multiConsumer(), and folly::NotificationQueue< MessageT >::PIPE.

                                               {
  QueueTest qt(0, IntQueue::FdType::PIPE);
  qt.multiConsumer();
}

TEST	(	NotificationQueueTest	,
		MaxQueueSizePipe
	)

Definition at line 551 of file NotificationQueueTest.cpp.

References QueueTest::maxQueueSize(), and folly::NotificationQueue< MessageT >::PIPE.

                                              {
  QueueTest qt(5, IntQueue::FdType::PIPE);
  qt.maxQueueSize();
}

TEST	(	NotificationQueueTest	,
		MaxReadAtOncePipe
	)

Definition at line 556 of file NotificationQueueTest.cpp.

References QueueTest::maxReadAtOnce(), and folly::NotificationQueue< MessageT >::PIPE.

                                               {
  QueueTest qt(0, IntQueue::FdType::PIPE);
  qt.maxReadAtOnce();
}

TEST	(	NotificationQueueTest	,
		DestroyCallbackPipe
	)

Definition at line 561 of file NotificationQueueTest.cpp.

References QueueTest::destroyCallback(), and folly::NotificationQueue< MessageT >::PIPE.

                                                 {
  QueueTest qt(0, IntQueue::FdType::PIPE);
  qt.destroyCallback();
}

TEST	(	NotificationQueueTest	,
		UseAfterFork
	)

Definition at line 575 of file NotificationQueueTest.cpp.

References EXPECT_EQ, EXPECT_TRUE, folly::ScopedEventBaseThread::getEventBase(), folly::INFO, QueueConsumer::messages, folly::NotificationQueue< MessageT >::putMessage(), folly::EventBase::runInEventBaseThread(), and folly::NotificationQueue< MessageT >::Consumer::startConsuming().

                                          {
  IntQueue queue;
  int childStatus = 0;
  QueueConsumer consumer;

  // Boost sets a custom SIGCHLD handler, which fails the test if a child
  // process exits abnormally.  We don't want this.
  signal(SIGCHLD, SIG_DFL);

  // Log some info so users reading the test output aren't confused
  // by the child process' crash log messages.
  LOG(INFO) << "This test makes sure the child process crashes.  "
            << "Error log messagges and a backtrace are expected.";

  {
    // Start a separate thread consuming from the queue
    ScopedEventBaseThread t1;
    t1.getEventBase()->runInEventBaseThread(
        [&] { consumer.startConsuming(t1.getEventBase(), &queue); });

    // Send a message to it, just for sanity checking
    queue.putMessage(1234);

    // Fork
    pid_t pid = fork();
    if (pid == 0) {
      // The boost test framework installs signal handlers to catch errors.
      // We only want to catch in the parent.  In the child let SIGABRT crash
      // us normally.
      signal(SIGABRT, SIG_DFL);

      // Child.
      // We're horrible people, so we try to send a message to the queue
      // that is being consumed in the parent process.
      //
      // The putMessage() call should catch this error, and crash our process.
      queue.putMessage(9876);
      // We shouldn't reach here.
      _exit(0);
    }
    PCHECK(pid > 0);

    // Parent.  Wait for the child to exit.
    auto waited = waitpid(pid, &childStatus, 0);
    EXPECT_EQ(pid, waited);

    // Send another message to the queue before we terminate the thread.
    queue.putMessage(5678);
  }

  // The child process should have crashed when it tried to call putMessage()
  // on our NotificationQueue.
  EXPECT_TRUE(WIFSIGNALED(childStatus));
  EXPECT_EQ(SIGABRT, WTERMSIG(childStatus));

  // Make sure the parent saw the expected messages.
  // It should have gotten 1234 and 5678 from the parent process, but not
  // 9876 from the child.
  EXPECT_EQ(2, consumer.messages.size());
  EXPECT_EQ(1234, consumer.messages.front());
  consumer.messages.pop_front();
  EXPECT_EQ(5678, consumer.messages.front());
  consumer.messages.pop_front();
}

TEST	(	NotificationQueueConsumer	,
		make
	)

Definition at line 641 of file NotificationQueueTest.cpp.

References EXPECT_EQ, folly::EventBase::loopOnce(), folly::pushmi::__adl::noexcept(), folly::NotificationQueue< MessageT >::tryPutMessage(), and value.

                                      {
  int value = 0;
  EventBase evb;
  NotificationQueue<int> queue(32);

  auto consumer =
      decltype(queue)::Consumer::make([&](int&& msg) noexcept { value = msg; });

  consumer->startConsuming(&evb, &queue);

  int const newValue = 10;
  queue.tryPutMessage(newValue);

  evb.loopOnce();

  EXPECT_EQ(newValue, value);
}