folly::AsyncFileWriter Class Reference

#include <AsyncFileWriter.h>

Inheritance diagram for folly::AsyncFileWriter:

Classes
struct	Data

Public Member Functions
	AsyncFileWriter (folly::StringPiece path)
	AsyncFileWriter (folly::File &&file)
	~AsyncFileWriter ()
void	writeMessage (folly::StringPiece buffer, uint32_t flags=0) override
void	writeMessage (std::string &&buffer, uint32_t flags=0) override
void	flush () override
bool	ttyOutput () const override
void	setMaxBufferSize (size_t size)
size_t	getMaxBufferSize () const
const folly::File &	getFile () const
Public Member Functions inherited from folly::LogWriter
virtual	~LogWriter ()

Static Public Attributes
static constexpr size_t	kDefaultMaxBufferSize = 1024 * 1024

Private Types
enum	Flags : uint32_t {   FLAG_IO_THREAD_STARTED = 0x01, FLAG_DESTROYING = 0x02, FLAG_STOP = 0x04, FLAG_IO_THREAD_STOPPED = 0x08,   FLAG_IO_THREAD_JOINED = 0x10 }

Private Member Functions
void	ioThread ()
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 ()

Private Attributes
folly::File	file_
folly::Synchronized< Data, std::mutex >	data_
std::condition_variable	messageReady_
std::condition_variable	ioCV_
folly::Synchronized< Data, std::mutex >::LockedPtr	lockedData_

Additional Inherited Members
Public Types inherited from folly::LogWriter
enum	Flags : uint32_t { NO_FLAGS = 0x00, NEVER_DISCARD = 0x01 }

Detailed Description

A LogWriter implementation that asynchronously writes to a file descriptor.

This class performs the log I/O in a separarate thread.

The advantage of this class over ImmediateFileWriter is that logging I/O can never slow down or block your normal program operation. If log messages are generated faster than they can be written, messages will be dropped (and an indication of how many messages were dropped will be written to the log file when we are able to catch up a bit.)

However, one downside is that if your program crashes, not all log messages may have been written, so you may lose messages generated immediately before the crash.

Definition at line 44 of file AsyncFileWriter.h.

Member Enumeration Documentation

enum folly::AsyncFileWriter::Flags : uint32_t

private

Enumerator
FLAG_IO_THREAD_STARTED
FLAG_DESTROYING
FLAG_STOP
FLAG_IO_THREAD_STOPPED
FLAG_IO_THREAD_JOINED

Definition at line 103 of file AsyncFileWriter.h.

             : 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,
  };

Constructor & Destructor Documentation

folly::AsyncFileWriter::AsyncFileWriter ( folly::StringPiece  path )

explicit

Construct an AsyncFileWriter that appends to the file at the specified path.

Definition at line 31 of file AsyncFileWriter.cpp.

References folly::Range< Iter >::str().

    : AsyncFileWriter{File{path.str(), O_WRONLY | O_APPEND | O_CREAT}} {}

folly::AsyncFileWriter::AsyncFileWriter ( folly::File &&  file )

explicit

Construct an AsyncFileWriter that writes to the specified File object.

Definition at line 33 of file AsyncFileWriter.cpp.

References folly::data(), data_, FLAG_IO_THREAD_STARTED, ioThread(), folly::gen::move, postForkChild(), postForkParent(), preFork(), and folly::detail::AtFork::registerHandler().

                                                 : 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(); });
  }
}

folly::AsyncFileWriter::~AsyncFileWriter

(

)

Definition at line 51 of file AsyncFileWriter.cpp.

References folly::data(), data_, FLAG_DESTROYING, onIoError(), performIO(), stopIoThread(), and folly::detail::AtFork::unregisterHandler().

                                  {
  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);
    }
  }
}

Member Function Documentation

void folly::AsyncFileWriter::flush ( )

overridevirtual

Block until the I/O thread has finished writing all messages that were already enqueued when flush() was called.

Implements folly::LogWriter.

Definition at line 103 of file AsyncFileWriter.cpp.

References folly::data(), data_, ioCV_, messageReady_, and start.

                            {
  auto data = data_.lock();
  auto start = data->ioThreadCounter;

  // Wait until ioThreadCounter increments by at least two.
  // Waiting for a single increment is not sufficient, as this happens after
  // the I/O thread has swapped the queues, which is before it has actually
  // done the I/O.
  while (data->ioThreadCounter < start + 2) {
    // 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
    // to do.
    data->getCurrentQueue()->emplace_back();
    messageReady_.notify_one();

    // Wait for notification from the I/O thread that it has done work.
    ioCV_.wait(data.getUniqueLock());
  }
}

const folly::File& folly::AsyncFileWriter::getFile ( ) const

inline

Get the output file.

Definition at line 98 of file AsyncFileWriter.h.

References file_.

                                   {
    return file_;
  }

size_t folly::AsyncFileWriter::getMaxBufferSize

(

)

const

Get the maximum buffer size for this AsyncFileWriter, in bytes.

Definition at line 129 of file AsyncFileWriter.cpp.

References folly::data(), and data_.

                                               {
  auto data = data_.lock();
  return data->maxBufferBytes;
}

std::string folly::AsyncFileWriter::getNumDiscardedMsg ( size_t  numDiscarded )

private

Definition at line 228 of file AsyncFileWriter.cpp.

Referenced by folly::AsyncFileWriter::Data::getCurrentQueue(), and performIO().

                                                                 {
  // We may want to make this customizable in the future (e.g., to allow it to
  // conform to the LogFormatter style being used).
  // For now just return a simple fixed message.
  return folly::to<std::string>(
      numDiscarded,
      " log messages discarded: logging faster than we can write\n");
}

void folly::AsyncFileWriter::ioThread ( )

private

Definition at line 134 of file AsyncFileWriter.cpp.

References folly::data(), data_, FLAG_IO_THREAD_STOPPED, FLAG_STOP, ioCV_, messageReady_, onIoError(), performIO(), and folly::setThreadName().

Referenced by AsyncFileWriter(), and restartThread().

                               {
  folly::setThreadName("log_writer");

  while (true) {
    // With the lock held, grab a pointer to the current queue, then increment
    // the ioThreadCounter index so that other threads will write into the
    // other queue as we process this one.
    std::vector<std::string>* ioQueue;
    size_t numDiscarded;
    {
      auto data = data_.lock();
      ioQueue = data->getCurrentQueue();
      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;
    }
    ioCV_.notify_all();

    // Write the log messages now that we have released the lock
    try {
      performIO(ioQueue, numDiscarded);
    } catch (const std::exception& ex) {
      onIoError(ex);
    }

    // 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();
  }
}

void folly::AsyncFileWriter::onIoError ( const std::exception &  ex )

private

Definition at line 218 of file AsyncFileWriter.cpp.

References folly::exceptionStr(), folly::File::fd(), file_, and folly::LoggerDB::internalWarning().

Referenced by folly::AsyncFileWriter::Data::getCurrentQueue(), ioThread(), and ~AsyncFileWriter().

                                                      {
  LoggerDB::internalWarning(
      __FILE__,
      __LINE__,
      "error writing to log file ",
      file_.fd(),
      " in AsyncFileWriter: ",
      folly::exceptionStr(ex));
}

void folly::AsyncFileWriter::performIO ( std::vector< std::string > *  ioQueue, size_t  numDiscarded  )

private

Definition at line 184 of file AsyncFileWriter.cpp.

References folly::checkUnixError(), folly::File::fd(), file_, getNumDiscardedMsg(), folly::size(), folly::writeFull(), and folly::writevFull().

Referenced by folly::AsyncFileWriter::Data::getCurrentQueue(), ioThread(), and ~AsyncFileWriter().

                         {
  // kNumIovecs controls the maximum number of strings we write at once in a
  // single writev() call.
  constexpr int kNumIovecs = 64;
  std::array<iovec, kNumIovecs> iovecs;

  size_t idx = 0;
  while (idx < ioQueue->size()) {
    int numIovecs = 0;
    while (numIovecs < kNumIovecs && idx < ioQueue->size()) {
      const auto& str = (*ioQueue)[idx];
      iovecs[numIovecs].iov_base = const_cast<char*>(str.data());
      iovecs[numIovecs].iov_len = str.size();
      ++numIovecs;
      ++idx;
    }

    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 folly::AsyncFileWriter::postForkChild ( )

private

Definition at line 264 of file AsyncFileWriter.cpp.

References lockedData_, and restartThread().

Referenced by AsyncFileWriter(), and folly::AsyncFileWriter::Data::getCurrentQueue().

                                    {
  // 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 folly::AsyncFileWriter::postForkParent ( )

private

Definition at line 259 of file AsyncFileWriter.cpp.

References restartThread().

Referenced by AsyncFileWriter(), and folly::AsyncFileWriter::Data::getCurrentQueue().

                                     {
  // Restart the I/O thread
  restartThread();
}

bool folly::AsyncFileWriter::preFork ( )

private

Definition at line 237 of file AsyncFileWriter.cpp.

References data_, FLAG_IO_THREAD_STARTED, lockedData_, and stopIoThread().

Referenced by AsyncFileWriter(), and folly::AsyncFileWriter::Data::getCurrentQueue().

                              {
  // 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 folly::AsyncFileWriter::restartThread ( )

private

Definition at line 292 of file AsyncFileWriter.cpp.

References folly::data(), FLAG_DESTROYING, FLAG_IO_THREAD_JOINED, FLAG_IO_THREAD_STARTED, FLAG_IO_THREAD_STOPPED, FLAG_STOP, ioThread(), lockedData_, and folly::gen::move.

Referenced by folly::AsyncFileWriter::Data::getCurrentQueue(), postForkChild(), and postForkParent().

                                    {
  // 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(); });
}

void folly::AsyncFileWriter::setMaxBufferSize

(

size_t

size

)

Set the maximum buffer size for this AsyncFileWriter, in bytes.

This controls the upper bound on how much unwritten data will be buffered in memory. If messages are being logged faster than they can be written to output file, new messages will be discarded if they would cause the amount of buffered data to exceed this limit.

Definition at line 124 of file AsyncFileWriter.cpp.

References folly::data(), data_, and folly::size().

                                                  {
  auto data = data_.lock();
  data->maxBufferBytes = size;
}

void folly::AsyncFileWriter::stopIoThread ( folly::Synchronized< Data, std::mutex >::LockedPtr &  data, uint32_t  extraFlags  )

private

Definition at line 274 of file AsyncFileWriter.cpp.

References FLAG_IO_THREAD_JOINED, FLAG_IO_THREAD_STOPPED, FLAG_STOP, ioCV_, and messageReady_.

Referenced by folly::AsyncFileWriter::Data::getCurrentQueue(), preFork(), and ~AsyncFileWriter().

                         {
  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;
  }
}

bool folly::AsyncFileWriter::ttyOutput ( ) const

overridevirtual

Returns true if the output steam is a tty.

Implements folly::LogWriter.

Definition at line 81 of file AsyncFileWriter.cpp.

References folly::File::fd(), and file_.

                                      {
  return isatty(file_.fd());
}

void folly::AsyncFileWriter::writeMessage ( folly::StringPiece  buffer, uint32_t  flags = 0  )

overridevirtual

Write a serialized log message.

The flags parameter is a bitwise-ORed set of Flag values defined above.

Implements folly::LogWriter.

Definition at line 85 of file AsyncFileWriter.cpp.

References folly::Range< Iter >::str().

Referenced by writeThread().

                                                                     {
  return writeMessage(buffer.str(), flags);
}

void folly::AsyncFileWriter::writeMessage ( std::string &&  buffer, uint32_t  flags = 0  )

overridevirtual

Write a serialized message.

This version of writeMessage() accepts a std::string&&. The default implementation calls the StringPiece version of writeMessage(), but subclasses may override this implementation if desired.

Reimplemented from folly::LogWriter.

Definition at line 89 of file AsyncFileWriter.cpp.

References buffer(), folly::data(), data_, messageReady_, folly::gen::move, and folly::LogWriter::NEVER_DISCARD.

                                                                     {
  auto data = data_.lock();
  if ((data->currentBufferSize >= data->maxBufferBytes) &&
      !(flags & NEVER_DISCARD)) {
    ++data->numDiscarded;
    return;
  }

  data->currentBufferSize += buffer.size();
  auto* queue = data->getCurrentQueue();
  queue->emplace_back(std::move(buffer));
  messageReady_.notify_one();
}

Member Data Documentation

folly::Synchronized<Data, std::mutex> folly::AsyncFileWriter::data_

private

Definition at line 158 of file AsyncFileWriter.h.

Referenced by AsyncFileWriter(), flush(), getMaxBufferSize(), ioThread(), preFork(), setMaxBufferSize(), writeMessage(), and ~AsyncFileWriter().

folly::File folly::AsyncFileWriter::file_

private

Definition at line 156 of file AsyncFileWriter.h.

Referenced by getFile(), onIoError(), performIO(), and ttyOutput().

std::condition_variable folly::AsyncFileWriter::ioCV_

private

ioCV_ is signaled by the I/O thread each time it increments the ioThreadCounter (once each time around its loop).

Definition at line 168 of file AsyncFileWriter.h.

Referenced by flush(), ioThread(), and stopIoThread().

constexpr size_t folly::AsyncFileWriter::kDefaultMaxBufferSize = 1024 * 1024

static

The default maximum buffer size.

The comments for setMaxBufferSize() explain how this parameter is used.

Definition at line 51 of file AsyncFileWriter.h.

Referenced by TEST().

folly::Synchronized<Data, std::mutex>::LockedPtr folly::AsyncFileWriter::lockedData_

private

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.

Definition at line 176 of file AsyncFileWriter.h.

Referenced by postForkChild(), preFork(), and restartThread().

std::condition_variable folly::AsyncFileWriter::messageReady_

private

messageReady_ is signaled by writer threads whenever they add a new message to the current queue.

Definition at line 163 of file AsyncFileWriter.h.

Referenced by flush(), ioThread(), stopIoThread(), and writeMessage().

---

The documentation for this class was generated from the following files:

• proxygen/folly/folly/logging/AsyncFileWriter.h
• proxygen/folly/folly/logging/AsyncFileWriter.cpp