Commit d586aab9 authored by Adam Simpkins's avatar Adam Simpkins Committed by Facebook Github Bot

logging: update AsyncFileWriter to interoperate with fork()

Summary:
Update AsyncFileWriter to use pthread_atfork() to stop the I/O thread around a
fork() operation.  Both the parent and child restart the I/O thread after the
fork.  Messages that were queued up at the time of the fork are logged in the
parent process after the fork occurs.

Previously after a fork() the child process would not have an I/O thread
running at all, so any messages sent to this handler would not actually be
written.  This also triggered some ASAN warnings which I did not fully track
down.  In the child process ASAN claimed that the exit handlers were run twice:
from both the main thread and the I/O thread, which is surprising since the I/O
thread should not run in the child process at all.

Reviewed By: mnv104

Differential Revision: D7946725

fbshipit-source-id: e4ad398f7bb334882c5c0b2afacc30c27bf3f6e4
parent 347f4867
......@@ -17,6 +17,7 @@
#include <folly/Exception.h>
#include <folly/FileUtil.h>
#include <folly/detail/AtFork.h>
#include <folly/logging/LoggerDB.h>
#include <folly/system/ThreadName.h>
......@@ -30,13 +31,52 @@ constexpr size_t AsyncFileWriter::kDefaultMaxBufferSize;
AsyncFileWriter::AsyncFileWriter(StringPiece path)
: AsyncFileWriter{File{path.str(), O_WRONLY | O_APPEND | O_CREAT}} {}
AsyncFileWriter::AsyncFileWriter(folly::File&& file)
: file_{std::move(file)}, ioThread_([this] { ioThread(); }) {}
AsyncFileWriter::AsyncFileWriter(folly::File&& file) : file_{std::move(file)} {
folly::detail::AtFork::registerHandler(
this,
[this] { return preFork(); },
[this] { postForkParent(); },
[this] { postForkChild(); });
// Start the I/O thread after registering the atfork handler.
// preFork() may be invoked in another thread as soon as registerHandler()
// returns. It will check FLAG_IO_THREAD_STARTED to see if the I/O thread is
// running yet.
{
auto data = data_.lock();
data->flags |= FLAG_IO_THREAD_STARTED;
data->ioThread = std::thread([this] { ioThread(); });
}
}
AsyncFileWriter::~AsyncFileWriter() {
data_->stop = true;
messageReady_.notify_one();
ioThread_.join();
std::vector<std::string>* ioQueue;
size_t numDiscarded;
{
// Stop the I/O thread
auto data = data_.lock();
stopIoThread(data, FLAG_DESTROYING);
// stopIoThread() causes the I/O thread to stop as soon as possible,
// without waiting for all pending messages to be written. Extract any
// remaining messages to write them below.
ioQueue = data->getCurrentQueue();
numDiscarded = data->numDiscarded;
}
// Unregister the atfork handler after stopping the I/O thread.
// preFork(), postForkParent(), and postForkChild() calls can run
// concurrently with the destructor until unregisterHandler() returns.
folly::detail::AtFork::unregisterHandler(this);
// If there are still any pending messages, flush them now.
if (!ioQueue->empty()) {
try {
performIO(ioQueue, numDiscarded);
} catch (const std::exception& ex) {
onIoError(ex);
}
}
}
void AsyncFileWriter::writeMessage(StringPiece buffer, uint32_t flags) {
......@@ -66,10 +106,6 @@ void AsyncFileWriter::flush() {
// the I/O thread has swapped the queues, which is before it has actually
// done the I/O.
while (data->ioThreadCounter < start + 2) {
if (data->ioThreadDone) {
return;
}
// Enqueue an empty string and wake the I/O thread.
// The empty string ensures that the I/O thread will break out of its wait
// loop and increment the ioThreadCounter, even if there is no other work
......@@ -101,25 +137,37 @@ void AsyncFileWriter::ioThread() {
// other queue as we process this one.
std::vector<std::string>* ioQueue;
size_t numDiscarded;
bool stop;
{
auto data = data_.lock();
ioQueue = data->getCurrentQueue();
while (ioQueue->empty() && !data->stop) {
while (ioQueue->empty() && !(data->flags & FLAG_STOP)) {
// Wait for a message or one of the above flags to be set.
messageReady_.wait(data.getUniqueLock());
}
if (data->flags & FLAG_STOP) {
// We have been asked to stop. We exit immediately in this case
// without writing out any pending messages. If we are stopping due
// to a fork() the I/O thread will be restarted after the fork (as
// long as we are not also being destroyed). If we are stopping due
// to the destructor, any remaining messages will be written out
// inside the destructor.
data->flags |= FLAG_IO_THREAD_STOPPED;
data.unlock();
ioCV_.notify_all();
return;
}
++data->ioThreadCounter;
numDiscarded = data->numDiscarded;
data->numDiscarded = 0;
data->currentBufferSize = 0;
stop = data->stop;
}
ioCV_.notify_all();
// Write the log messages now that we have released the lock
try {
performIO(ioQueue);
performIO(ioQueue, numDiscarded);
} catch (const std::exception& ex) {
onIoError(ex);
}
......@@ -127,25 +175,12 @@ void AsyncFileWriter::ioThread() {
// clear() empties the vector, but the allocated capacity remains so we can
// just reuse it without having to re-allocate in most cases.
ioQueue->clear();
if (numDiscarded > 0) {
auto msg = getNumDiscardedMsg(numDiscarded);
if (!msg.empty()) {
auto ret = folly::writeFull(file_.fd(), msg.data(), msg.size());
// We currently ignore errors from writeFull() here.
// There's not much we can really do.
(void)ret;
}
}
if (stop) {
data_->ioThreadDone = true;
break;
}
}
}
void AsyncFileWriter::performIO(std::vector<std::string>* ioQueue) {
void AsyncFileWriter::performIO(
std::vector<std::string>* ioQueue,
size_t numDiscarded) {
// kNumIovecs controls the maximum number of strings we write at once in a
// single writev() call.
constexpr int kNumIovecs = 64;
......@@ -165,6 +200,16 @@ void AsyncFileWriter::performIO(std::vector<std::string>* ioQueue) {
auto ret = folly::writevFull(file_.fd(), iovecs.data(), numIovecs);
folly::checkUnixError(ret, "writeFull() failed");
}
if (numDiscarded > 0) {
auto msg = getNumDiscardedMsg(numDiscarded);
if (!msg.empty()) {
auto ret = folly::writeFull(file_.fd(), msg.data(), msg.size());
// We currently ignore errors from writeFull() here.
// There's not much we can really do.
(void)ret;
}
}
}
void AsyncFileWriter::onIoError(const std::exception& ex) {
......@@ -185,4 +230,82 @@ std::string AsyncFileWriter::getNumDiscardedMsg(size_t numDiscarded) {
numDiscarded,
" log messages discarded: logging faster than we can write\n");
}
bool AsyncFileWriter::preFork() {
// Stop the I/O thread.
//
// It would perhaps be better to not stop the I/O thread in the parent
// process. However, this leaves us in a slightly tricky situation in the
// child process where data_->ioThread has been initialized and does not
// really point to a valid thread. While we could store it in a union and
// replace it without ever calling its destructor, in practice this still has
// some tricky corner cases to deal with.
// Grab the data lock to ensure no other thread is holding it
// while we fork.
lockedData_ = data_.lock();
// If the I/O thread has been started, stop it now
if (lockedData_->flags & FLAG_IO_THREAD_STARTED) {
stopIoThread(lockedData_, 0);
}
return true;
}
void AsyncFileWriter::postForkParent() {
// Restart the I/O thread
restartThread();
}
void AsyncFileWriter::postForkChild() {
// Clear any messages in the queue. We only want them to be written once,
// and we let the parent process handle writing them.
lockedData_->queues[0].clear();
lockedData_->queues[1].clear();
// Restart the I/O thread
restartThread();
}
void AsyncFileWriter::stopIoThread(
folly::Synchronized<Data, std::mutex>::LockedPtr& data,
uint32_t extraFlags) {
data->flags |= (FLAG_STOP | extraFlags);
messageReady_.notify_one();
ioCV_.wait(data.getUniqueLock(), [&] {
return bool(data->flags & FLAG_IO_THREAD_STOPPED);
});
// Check FLAG_IO_THREAD_JOINED before calling join().
// preFork() and the destructor may both run concurrently in separate
// threads, and only one should try to join the thread.
if ((data->flags & FLAG_IO_THREAD_JOINED) == 0) {
data->ioThread.join();
data->flags |= FLAG_IO_THREAD_JOINED;
}
}
void AsyncFileWriter::restartThread() {
// Move lockedData_ into a local member variable so it will be released
// when we return.
folly::Synchronized<Data, std::mutex>::LockedPtr data =
std::move(lockedData_);
if (!(data->flags & FLAG_IO_THREAD_STARTED)) {
// Do not start the I/O thread if the constructor has not finished yet
return;
}
if (data->flags & FLAG_DESTROYING) {
// Do not restart the I/O thread if we were being destroyed.
// If there are more pending messages that need to be flushed the
// destructor's stopIoThread() call will handle flushing the messages in
// this case.
return;
}
data->flags &= ~(FLAG_STOP | FLAG_IO_THREAD_JOINED | FLAG_IO_THREAD_STOPPED);
data->ioThread = std::thread([this] { ioThread(); });
}
} // namespace folly
......@@ -95,6 +95,23 @@ class AsyncFileWriter : public LogWriter {
}
private:
enum Flags : uint32_t {
// FLAG_IO_THREAD_STARTED indicates that the constructor has started the
// I/O thread.
FLAG_IO_THREAD_STARTED = 0x01,
// FLAG_DESTROYING indicates that the destructor is running and destroying
// the I/O thread.
FLAG_DESTROYING = 0x02,
// FLAG_STOP indicates that the I/O thread has been asked to stop.
// This is set either by the destructor or by preFork()
FLAG_STOP = 0x04,
// FLAG_IO_THREAD_STOPPED indicates that the I/O thread is about to return
// and can now be joined. ioCV_ will be signalled when this flag is set.
FLAG_IO_THREAD_STOPPED = 0x08,
// FLAG_IO_THREAD_JOINED indicates that the I/O thread has been joined.
FLAG_IO_THREAD_JOINED = 0x10,
};
/*
* A simple implementation using two queues.
* All writer threads enqueue into one queue while the I/O thread is
......@@ -105,12 +122,12 @@ class AsyncFileWriter : public LogWriter {
*/
struct Data {
std::array<std::vector<std::string>, 2> queues;
bool stop{false};
bool ioThreadDone{false};
uint32_t flags{0};
uint64_t ioThreadCounter{0};
size_t maxBufferBytes{kDefaultMaxBufferSize};
size_t currentBufferSize{0};
size_t numDiscarded{0};
std::thread ioThread;
std::vector<std::string>* getCurrentQueue() {
return &queues[ioThreadCounter & 0x1];
......@@ -118,11 +135,19 @@ class AsyncFileWriter : public LogWriter {
};
void ioThread();
void performIO(std::vector<std::string>* ioQueue);
void performIO(std::vector<std::string>* ioQueue, size_t numDiscarded);
void onIoError(const std::exception& ex);
std::string getNumDiscardedMsg(size_t numDiscarded);
bool preFork();
void postForkParent();
void postForkChild();
void stopIoThread(
folly::Synchronized<Data, std::mutex>::LockedPtr& data,
uint32_t extraFlags);
void restartThread();
folly::File file_;
folly::Synchronized<Data, std::mutex> data_;
/**
......@@ -137,11 +162,11 @@ class AsyncFileWriter : public LogWriter {
std::condition_variable ioCV_;
/**
* The I/O thread.
*
* This should come last, since all other member variables need to be
* constructed before the I/O thread starts.
* lockedData_ exists only to help pass the lock between preFork() and
* postForkParent()/postForkChild(). We potentially could add some new
* low-level methods to Synchronized to allow manually locking and unlocking
* to avoid having to store this object as a member variable.
*/
std::thread ioThread_;
folly::Synchronized<Data, std::mutex>::LockedPtr lockedData_;
};
} // namespace folly
......@@ -30,10 +30,14 @@
#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>
DEFINE_string(logging, "", "folly::logging configuration");
DEFINE_int64(
......@@ -65,8 +69,9 @@ DEFINE_int64(
using namespace folly;
using namespace std::literals::chrono_literals;
using folly::test::TemporaryFile;
using std::chrono::steady_clock;
using std::chrono::milliseconds;
using std::chrono::steady_clock;
using testing::ContainsRegex;
TEST(AsyncFileWriter, noMessages) {
TemporaryFile tmpFile{"logging_test"};
......@@ -162,7 +167,7 @@ TEST(AsyncFileWriter, ioError) {
for (const auto& msg : logErrors) {
EXPECT_THAT(
msg,
testing::ContainsRegex(
ContainsRegex(
"error writing to log file .* in AsyncFileWriter.*: " +
kExpectedErrorMessage));
}
......@@ -612,6 +617,192 @@ TEST(AsyncFileWriter, discard) {
readStats.check();
}
/**
* Test that AsyncFileWriter operates correctly after a fork() in both the
* parent and child processes.
*/
TEST(AsyncFileWriter, fork) {
#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
}
/**
* 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.
*/
TEST(AsyncFileWriter, crazyForks) {
#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
}
int main(int argc, char* argv[]) {
testing::InitGoogleTest(&argc, argv);
folly::init(&argc, &argv);
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment