AsyncFileWriterTest.cpp File Reference

#include <thread>
#include <folly/Conv.h>
#include <folly/Exception.h>
#include <folly/File.h>
#include <folly/FileUtil.h>
#include <folly/String.h>
#include <folly/Synchronized.h>
#include <folly/experimental/TestUtil.h>
#include <folly/futures/Future.h>
#include <folly/futures/Promise.h>
#include <folly/init/Init.h>
#include <folly/lang/SafeAssert.h>
#include <folly/logging/AsyncFileWriter.h>
#include <folly/logging/Init.h>
#include <folly/logging/LoggerDB.h>
#include <folly/logging/xlog.h>
#include <folly/portability/Config.h>
#include <folly/portability/GFlags.h>
#include <folly/portability/GMock.h>
#include <folly/portability/GTest.h>
#include <folly/portability/Unistd.h>
#include <folly/system/ThreadId.h>
#include <folly/system/ThreadName.h>
#include <folly/test/TestUtils.h>

Go to the source code of this file.

Classes
class	ReadStats
struct	ReadStats::ReaderData
struct	ReadStats::WriterData

Functions
	DEFINE_int64 (async_discard_num_normal_writers, 30,"number of threads to use to generate normal log messages during ""the AsyncFileWriter.discard test")
	DEFINE_int64 (async_discard_num_nodiscard_writers, 2,"number of threads to use to generate non-discardable log messages during ""the AsyncFileWriter.discard test")
	DEFINE_int64 (async_discard_read_sleep_usec, 500,"how long the read thread should sleep between reads in ""the AsyncFileWriter.discard test")
	DEFINE_int64 (async_discard_timeout_msec, 10000,"A timeout for the AsyncFileWriter.discard test if it cannot generate ""enough discards")
	DEFINE_int64 (async_discard_num_events, 10,"The number of discard events to wait for in the AsyncFileWriter.discard ""test")
	TEST (AsyncFileWriter, noMessages)
	TEST (AsyncFileWriter, simpleMessages)
	TEST (AsyncFileWriter, ioError)
	TEST (AsyncFileWriter, flush)
void	readThread (folly::File &&file, ReadStats *stats)
void	writeThread (AsyncFileWriter *writer, size_t id, uint32_t flags, ReadStats *readStats)
	TEST (AsyncFileWriter, discard)
	TEST (AsyncFileWriter, fork)
	TEST (AsyncFileWriter, crazyForks)

Variables
static constexpr StringPiece	kMsgSuffix

Function Documentation

DEFINE_int64	(	async_discard_num_normal_writers	,
		30	,
		"number of threads to use to generate normal log messages during ""the AsyncFileWriter.discard test"
	)

DEFINE_int64	(	async_discard_num_nodiscard_writers	,
		2	,
		"number of threads to use to generate non-discardable log messages during ""the AsyncFileWriter.discard test"
	)

DEFINE_int64	(	async_discard_read_sleep_usec	,
		500	,
		"how long the read thread should sleep between reads in ""the AsyncFileWriter.discard test"
	)

DEFINE_int64	(	async_discard_timeout_msec	,
		10000	,
		"A timeout for the AsyncFileWriter.discard test if it cannot generate ""enough discards"
	)

DEFINE_int64	(	async_discard_num_events	,
		10	,
		"The number of discard events to wait for in the AsyncFileWriter.discard ""test"
	)

void readThread	(	folly::File &&	file,
		ReadStats *	stats
	)

readThread() reads messages slowly from a pipe. This helps test the AsyncFileWriter behavior when I/O is slow.

Definition at line 506 of file AsyncFileWriterTest.cpp.

References buffer(), folly::Range< Iter >::data(), folly::DBG2, folly::test::end(), folly::ERR, ReadStats::getSleepUS(), ReadStats::messageReceived(), folly::Range< const char * >::npos, folly::readNoInt(), ReadStats::trailingData(), and XLOG.

Referenced by TEST().

                                                    {
  std::vector<char> buffer;
  buffer.resize(1024);

  size_t bufferIdx = 0;
  while (true) {
    /* sleep override */
    std::this_thread::sleep_for(stats->getSleepUS());

    auto readResult = folly::readNoInt(
        file.fd(), buffer.data() + bufferIdx, buffer.size() - bufferIdx);
    if (readResult < 0) {
      XLOG(ERR, "error reading from pipe: ", errno);
      return;
    }
    if (readResult == 0) {
      XLOG(DBG2, "read EOF");
      break;
    }

    auto logDataLen = bufferIdx + readResult;
    StringPiece logData{buffer.data(), logDataLen};
    auto idx = 0;
    while (true) {
      auto end = logData.find('\n', idx);
      if (end == StringPiece::npos) {
        bufferIdx = logDataLen - idx;
        memmove(buffer.data(), buffer.data() + idx, bufferIdx);
        break;
      }

      StringPiece logMsg{logData.data() + idx, end - idx};
      stats->messageReceived(logMsg);
      idx = end + 1;
    }
  }

  if (bufferIdx != 0) {
    stats->trailingData(StringPiece{buffer.data(), bufferIdx});
  }
}

TEST	(	AsyncFileWriter	,
		noMessages
	)

Definition at line 75 of file AsyncFileWriterTest.cpp.

References folly::File::fd().

                                  {
  TemporaryFile tmpFile{"logging_test"};

  // Test the simple construction and destruction of an AsyncFileWriter
  // without ever writing any messages.  This still exercises the I/O
  // thread start-up and shutdown code.
  AsyncFileWriter writer{folly::File{tmpFile.fd(), false}};
}

TEST	(	AsyncFileWriter	,
		simpleMessages
	)

Definition at line 84 of file AsyncFileWriterTest.cpp.

References ASSERT_TRUE, folly::data(), EXPECT_EQ, folly::File::fd(), folly::gen::move, folly::readFile(), string, and folly::fibers::yield().

                                      {
  TemporaryFile tmpFile{"logging_test"};

  {
    AsyncFileWriter writer{folly::File{tmpFile.fd(), false}};
    for (int n = 0; n < 10; ++n) {
      writer.writeMessage(folly::to<std::string>("message ", n, "\n"));
      std::this_thread::yield();
    }
  }
  tmpFile.close();

  std::string data;
  auto ret = folly::readFile(tmpFile.path().string().c_str(), data);
  ASSERT_TRUE(ret);

  std::string expected =
      "message 0\n"
      "message 1\n"
      "message 2\n"
      "message 3\n"
      "message 4\n"
      "message 5\n"
      "message 6\n"
      "message 7\n"
      "message 8\n"
      "message 9\n";
  EXPECT_EQ(expected, data);
}

TEST	(	AsyncFileWriter	,
		ioError
	)

Definition at line 125 of file AsyncFileWriterTest.cpp.

References folly::checkUnixError(), folly::netops::close(), testing::ContainsRegex(), folly::data(), folly::DBG1, EXPECT_GT, EXPECT_LE, EXPECT_THAT, pipe(), folly::LoggerDB::setInternalWarningHandler(), string, folly::throwSystemError(), folly::writeNoInt(), XLOG, and folly::fibers::yield().

                               {
  // Set the LoggerDB internal warning handler so we can record the messages
  std::vector<std::string> logErrors;
  internalWarnings = &logErrors;
  LoggerDB::setInternalWarningHandler(handleLoggingError);

  // Create an AsyncFileWriter that refers to a pipe whose read end is closed
  std::array<int, 2> fds;
  auto rc = pipe(fds.data());
  folly::checkUnixError(rc, "failed to create pipe");
#ifndef _WIN32
  signal(SIGPIPE, SIG_IGN);
#endif
  ::close(fds[0]);

  // Log a bunch of messages to the writer
  size_t numMessages = 100;
  {
    AsyncFileWriter writer{folly::File{fds[1], true}};
    for (size_t n = 0; n < numMessages; ++n) {
      writer.writeMessage(folly::to<std::string>("message ", n, "\n"));
      std::this_thread::yield();
    }
  }

  LoggerDB::setInternalWarningHandler(nullptr);

  // AsyncFileWriter should have some internal warning messages about the
  // log failures.  This will generally be many fewer than the number of
  // messages we wrote, though, since it performs write batching.
  //
  // GTest on Windows doesn't support alternation in the regex syntax -_-....
  const std::string kExpectedErrorMessage =
#if _WIN32
      // The `pipe` call above is actually implemented via sockets, so we get
      // a different error message.
      "An established connection was aborted by the software in your host machine\\.";
#else
      "Broken pipe";
#endif

  for (const auto& msg : logErrors) {
    EXPECT_THAT(
        msg,
        ContainsRegex(
            "error writing to log file .* in AsyncFileWriter.*: " +
            kExpectedErrorMessage));
  }
  EXPECT_GT(logErrors.size(), 0);
  EXPECT_LE(logErrors.size(), numMessages);
}

TEST	(	AsyncFileWriter	,
		flush
	)

Definition at line 215 of file AsyncFileWriterTest.cpp.

References folly::checkUnixError(), EXPECT_EQ, EXPECT_FALSE, folly::Promise< T >::getFuture(), folly::makeTryWith(), folly::gen::move, pipe(), folly::readFull(), folly::Promise< T >::setTry(), and string.

                             {
  // Set up a pipe(), then write data to the write endpoint until it fills up
  // and starts blocking.
  std::array<int, 2> fds;
  auto rc = pipe(fds.data());
  folly::checkUnixError(rc, "failed to create pipe");
  File readPipe{fds[0], true};
  File writePipe{fds[1], true};

  auto paddingSize = fillUpPipe(writePipe.fd());

  // Now set up an AsyncFileWriter pointing at the write end of the pipe
  AsyncFileWriter writer{std::move(writePipe)};

  // Write a message
  writer.writeMessage("test message: " + std::string(200, 'x'));

  // Call flush().  Use a separate thread, since this should block until we
  // consume data from the pipe.
  Promise<Unit> promise;
  auto future = promise.getFuture();
  auto flushFunction = [&] { writer.flush(); };
  std::thread flushThread{
      [&]() { promise.setTry(makeTryWith(flushFunction)); }};
  // Detach the flush thread now rather than joining it at the end of the
  // function.  This way if something goes wrong during the test we will fail
  // with the real error, rather than crashing due to the std::thread
  // destructor running on a still-joinable thread.
  flushThread.detach();

  // Sleep briefly, and make sure flush() still hasn't completed.
  // If it has completed this doesn't necessarily indicate a bug in
  // AsyncFileWriter, but instead indicates that our test code failed to
  // successfully cause a blocking write.
  /* sleep override */
  std::this_thread::sleep_for(10ms);
  EXPECT_FALSE(future.isReady());

  // Now read from the pipe
  std::vector<char> buf;
  buf.resize(paddingSize);
  auto bytesRead = readFull(readPipe.fd(), buf.data(), buf.size());
  EXPECT_EQ(bytesRead, paddingSize);

  // Make sure flush completes successfully now
  std::move(future).get(10ms);
}

TEST	(	AsyncFileWriter	,
		discard
	)

Definition at line 583 of file AsyncFileWriterTest.cpp.

References ReadStats::check(), folly::checkUnixError(), ReadStats::clearSleepDuration(), folly::DBG2, folly::DBG4, folly::gen::move, folly::LogWriter::NEVER_DISCARD, pipe(), readThread(), folly::pushmi::detail::t, uint32_t, writeThread(), XLOG, and XLOGF.

                               {
  std::array<int, 2> fds;
  auto pipeResult = pipe(fds.data());
  folly::checkUnixError(pipeResult, "pipe failed");
  folly::File readPipe{fds[0], true};
  folly::File writePipe{fds[1], true};

  ReadStats readStats;
  std::thread reader(readThread, std::move(readPipe), &readStats);
  {
    AsyncFileWriter writer{std::move(writePipe)};

    std::vector<std::thread> writeThreads;
    size_t numThreads = FLAGS_async_discard_num_normal_writers +
        FLAGS_async_discard_num_nodiscard_writers;

    for (size_t n = 0; n < numThreads; ++n) {
      uint32_t flags = 0;
      if (n >= static_cast<size_t>(FLAGS_async_discard_num_normal_writers)) {
        flags = LogWriter::NEVER_DISCARD;
      }
      XLOGF(DBG4, "writer {:4d} flags {:#02x}", n, flags);

      writeThreads.emplace_back(writeThread, &writer, n, flags, &readStats);
    }

    for (auto& t : writeThreads) {
      t.join();
    }
    XLOG(DBG2, "writers done");
  }
  // Clear the read sleep duration so the reader will finish quickly now
  readStats.clearSleepDuration();
  reader.join();
  readStats.check();
}

TEST	(	AsyncFileWriter	,
		fork
	)

Test that AsyncFileWriter operates correctly after a fork() in both the parent and child processes.

Definition at line 624 of file AsyncFileWriterTest.cpp.

References ASSERT_EQ, ASSERT_TRUE, folly::checkUnixError(), testing::ContainsRegex(), folly::data(), folly::DBG1, EXPECT_THAT, folly::File::fd(), folly::readFile(), SKIP, stop(), string, folly::pushmi::detail::t, and XLOG.

                            {
#if FOLLY_HAVE_PTHREAD_ATFORK
  TemporaryFile tmpFile{"logging_test"};

  // The number of messages to send before the fork and from each process
  constexpr size_t numMessages = 10;
  constexpr size_t numBgThreads = 2;

  // This can be increased to add some delay in the parent and child messages
  // so that they are likely to be interleaved in the log rather than grouped
  // together.  This doesn't really affect the test behavior or correctness
  // otherwise, though.
  constexpr milliseconds sleepDuration(0);

  {
    AsyncFileWriter writer{folly::File{tmpFile.fd(), false}};
    writer.writeMessage(folly::to<std::string>("parent pid=", getpid(), "\n"));

    // Start some background threads just to exercise the behavior
    // when other threads are also logging to the writer when the fork occurs
    std::vector<std::thread> bgThreads;
    std::atomic<bool> stop{false};
    for (size_t n = 0; n < numBgThreads; ++n) {
      bgThreads.emplace_back([&] {
        size_t iter = 0;
        while (!stop) {
          writer.writeMessage(
              folly::to<std::string>("bgthread_", getpid(), "_", iter, "\n"));
          ++iter;
        }
      });
    }

    for (size_t n = 0; n < numMessages; ++n) {
      writer.writeMessage(folly::to<std::string>("prefork", n, "\n"));
    }

    auto pid = fork();
    folly::checkUnixError(pid, "failed to fork");
    if (pid == 0) {
      writer.writeMessage(folly::to<std::string>("child pid=", getpid(), "\n"));
      for (size_t n = 0; n < numMessages; ++n) {
        writer.writeMessage(folly::to<std::string>("child", n, "\n"));
        std::this_thread::sleep_for(sleepDuration);
      }

      // Use _exit() rather than exit() in the child, purely to prevent
      // ASAN from complaining that we leak memory for the background threads.
      // (These threads don't actually exist in the child, so it is difficult
      // to clean up their allocated state entirely.)
      //
      // Explicitly flush the writer since exiting with _exit() won't do this
      // automatically.
      writer.flush();
      _exit(0);
    }

    for (size_t n = 0; n < numMessages; ++n) {
      writer.writeMessage(folly::to<std::string>("parent", n, "\n"));
      std::this_thread::sleep_for(sleepDuration);
    }

    // Stop the background threads.
    stop = true;
    for (auto& t : bgThreads) {
      t.join();
    }

    int status;
    auto waited = waitpid(pid, &status, 0);
    folly::checkUnixError(waited, "failed to wait on child");
    ASSERT_EQ(waited, pid);
  }

  // Read back the logged messages
  tmpFile.close();
  std::string data;
  auto ret = folly::readFile(tmpFile.path().string().c_str(), data);
  ASSERT_TRUE(ret) << "failed to read log file";

  XLOG(DBG1) << "log contents:\n" << data;

  // The log file should contain all of the messages we wrote, from both the
  // parent and child processes.
  for (size_t n = 0; n < numMessages; ++n) {
    EXPECT_THAT(
        data, ContainsRegex(folly::to<std::string>("prefork", n, "\n")));
    EXPECT_THAT(data, ContainsRegex(folly::to<std::string>("parent", n, "\n")));
    EXPECT_THAT(data, ContainsRegex(folly::to<std::string>("child", n, "\n")));
  }
#else
  SKIP() << "pthread_atfork() is not supported on this platform";
#endif // FOLLY_HAVE_PTHREAD_ATFORK
}

TEST	(	AsyncFileWriter	,
		crazyForks
	)

Have several threads concurrently performing fork() calls while several other threads continuously create and destroy AsyncFileWriter objects.

This exercises the synchronization around registration of the AtFork handlers and the creation/destruction of the AsyncFileWriter I/O thread.

Definition at line 726 of file AsyncFileWriterTest.cpp.

References folly::checkUnixError(), folly::DBG3, EXPECT_EQ, folly::File::fd(), folly::getOSThreadID(), i, mutex, folly::setThreadName(), SKIP, stop(), folly::pushmi::detail::t, and XLOG.

                                  {
#if FOLLY_HAVE_PTHREAD_ATFORK
  constexpr size_t numAsyncWriterThreads = 10;
  constexpr size_t numForkThreads = 5;
  constexpr size_t numForkIterations = 20;
  std::atomic<bool> stop{false};

  // Spawn several threads that continuously create and destroy
  // AsyncFileWriter objects.
  std::vector<std::thread> asyncWriterThreads;
  for (size_t n = 0; n < numAsyncWriterThreads; ++n) {
    asyncWriterThreads.emplace_back([n, &stop] {
      folly::setThreadName(folly::to<std::string>("async_", n));

      TemporaryFile tmpFile{"logging_test"};
      while (!stop) {
        // Create an AsyncFileWriter, write a message to it, then destroy it.
        AsyncFileWriter writer{folly::File{tmpFile.fd(), false}};
        writer.writeMessage(folly::to<std::string>(
            "async thread ", folly::getOSThreadID(), "\n"));
      }
    });
  }

  // Spawn several threads that repeatedly fork.
  std::vector<std::thread> forkThreads;
  std::mutex forkStartMutex;
  std::condition_variable forkStartCV;
  bool forkStart = false;
  for (size_t n = 0; n < numForkThreads; ++n) {
    forkThreads.emplace_back([n, &forkStartMutex, &forkStartCV, &forkStart] {
      folly::setThreadName(folly::to<std::string>("fork_", n));

      // Wait until forkStart is set just to have a better chance of all the
      // fork threads running simultaneously.
      {
        std::unique_lock<std::mutex> l(forkStartMutex);
        forkStartCV.wait(l, [&forkStart] { return forkStart; });
      }

      for (size_t i = 0; i < numForkIterations; ++i) {
        XLOG(DBG3) << "fork " << n << ":" << i;
        auto pid = fork();
        folly::checkUnixError(pid, "forkFailed");
        if (pid == 0) {
          XLOG(DBG3) << "child " << getpid();
          _exit(0);
        }

        // parent
        int status;
        auto waited = waitpid(pid, &status, 0);
        folly::checkUnixError(waited, "failed to wait on child");
        EXPECT_EQ(waited, pid);
      }
    });
  }

  // Kick off the fork threads
  {
    std::unique_lock<std::mutex> l(forkStartMutex);
    forkStart = true;
  }
  forkStartCV.notify_all();

  // Wait for the fork threads to finish
  for (auto& t : forkThreads) {
    t.join();
  }

  // Stop and wait for the AsyncFileWriter threads
  stop = true;
  for (auto& t : asyncWriterThreads) {
    t.join();
  }
#else
  SKIP() << "pthread_atfork() is not supported on this platform";
#endif // FOLLY_HAVE_PTHREAD_ATFORK
}

void writeThread	(	AsyncFileWriter *	writer,
		size_t	id,
		uint32_t	flags,
		ReadStats *	readStats
	)

writeThread() writes a series of messages to the AsyncFileWriter

Definition at line 551 of file AsyncFileWriterTest.cpp.

References kMsgSuffix, ReadStats::shouldWriterStop(), folly::AsyncFileWriter::writeMessage(), and ReadStats::writerFinished().

Referenced by TEST().

                          {
  size_t msgID = 0;
  while (true) {
    ++msgID;
    writer->writeMessage(
        folly::to<std::string>(
            "thread ", id, " message ", msgID, kMsgSuffix, '\n'),
        flags);

    // Break out once the reader has seen enough discards
    if (((msgID & 0xff) == 0) && readStats->shouldWriterStop()) {
      readStats->writerFinished(id, msgID, flags);
      break;
    }
  }
}

Variable Documentation

constexpr StringPiece kMsgSuffix

static

Initial value:

{
    "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789"
    "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789"
    "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789"
    "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789"}

Definition at line 265 of file AsyncFileWriterTest.cpp.

Referenced by ReadStats::parseMessage(), and writeThread().