folly/experimental/logging/AsyncFileWriter.cpp

   1 /*
   2  * Copyright 2004-present Facebook, Inc.
   3  *
   4  * Licensed under the Apache License, Version 2.0 (the "License");
   5  * you may not use this file except in compliance with the License.
   6  * You may obtain a copy of the License at
   7  *
   8  *   http://www.apache.org/licenses/LICENSE-2.0
   9  *
  10  * Unless required by applicable law or agreed to in writing, software
  11  * distributed under the License is distributed on an "AS IS" BASIS,
  12  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13  * See the License for the specific language governing permissions and
  14  * limitations under the License.
  15  */
  16 #include <folly/experimental/logging/AsyncFileWriter.h>
  17
  18 #include <folly/Exception.h>
  19 #include <folly/FileUtil.h>
  20 #include <folly/experimental/logging/LoggerDB.h>
  21
  22 using folly::File;
  23 using folly::StringPiece;
  24
  25 namespace folly {
  26
  27 AsyncFileWriter::AsyncFileWriter(StringPiece path)
  28     : AsyncFileWriter{File{path.str(), O_WRONLY | O_APPEND | O_CREAT}} {}
  29
  30 AsyncFileWriter::AsyncFileWriter(folly::File&& file)
  31     : file_{std::move(file)}, ioThread_([this] { ioThread(); }) {}
  32
  33 AsyncFileWriter::~AsyncFileWriter() {
  34   data_->stop = true;
  35   messageReady_.notify_one();
  36   ioThread_.join();
  37 }
  38
  39 void AsyncFileWriter::writeMessage(StringPiece buffer) {
  40   return writeMessage(buffer.str());
  41 }
  42
  43 void AsyncFileWriter::writeMessage(std::string&& buffer) {
  44   auto data = data_.lock();
  45   if (data->currentBufferSize >= data->maxBufferBytes) {
  46     ++data->numDiscarded;
  47     return;
  48   }
  49
  50   data->currentBufferSize += buffer.size();
  51   auto* queue = data->getCurrentQueue();
  52   queue->emplace_back(std::move(buffer));
  53   messageReady_.notify_one();
  54 }
  55
  56 void AsyncFileWriter::flush() {
  57   auto data = data_.lock();
  58   auto start = data->ioThreadCounter;
  59
  60   // Wait until ioThreadCounter increments by at least two.
  61   // Waiting for a single increment is not sufficient, as this happens after
  62   // the I/O thread has swapped the queues, which is before it has actually
  63   // done the I/O.
  64   while (data->ioThreadCounter < start + 2) {
  65     if (data->ioThreadDone) {
  66       return;
  67     }
  68
  69     // Enqueue an empty string and wake the I/O thread.
  70     // The empty string ensures that the I/O thread will break out of its wait
  71     // loop and increment the ioThreadCounter, even if there is no other work
  72     // to do.
  73     data->getCurrentQueue()->emplace_back();
  74     messageReady_.notify_one();
  75
  76     // Wait for notification from the I/O thread that it has done work.
  77     ioCV_.wait(data.getUniqueLock());
  78   }
  79 }
  80
  81 void AsyncFileWriter::ioThread() {
  82   while (true) {
  83     // With the lock held, grab a pointer to the current queue, then increment
  84     // the ioThreadCounter index so that other threads will write into the
  85     // other queue as we process this one.
  86     std::vector<std::string>* ioQueue;
  87     size_t numDiscarded;
  88     bool stop;
  89     {
  90       auto data = data_.lock();
  91       ioQueue = data->getCurrentQueue();
  92       while (ioQueue->empty() && !data->stop) {
  93         messageReady_.wait(data.getUniqueLock());
  94       }
  95
  96       ++data->ioThreadCounter;
  97       numDiscarded = data->numDiscarded;
  98       data->numDiscarded = 0;
  99       data->currentBufferSize = 0;
 100       stop = data->stop;
 101     }
 102     ioCV_.notify_all();
 103
 104     // Write the log messages now that we have released the lock
 105     try {
 106       performIO(ioQueue);
 107     } catch (const std::exception& ex) {
 108       onIoError(ex);
 109     }
 110
 111     // clear() empties the vector, but the allocated capacity remains so we can
 112     // just reuse it without having to re-allocate in most cases.
 113     ioQueue->clear();
 114
 115     if (numDiscarded > 0) {
 116       auto msg = getNumDiscardedMsg(numDiscarded);
 117       if (!msg.empty()) {
 118         auto ret = folly::writeFull(file_.fd(), msg.data(), msg.size());
 119         // We currently ignore errors from writeFull() here.
 120         // There's not much we can really do.
 121         (void)ret;
 122       }
 123     }
 124
 125     if (stop) {
 126       data_->ioThreadDone = true;
 127       break;
 128     }
 129   }
 130 }
 131
 132 void AsyncFileWriter::performIO(std::vector<std::string>* ioQueue) {
 133   // kNumIovecs controls the maximum number of strings we write at once in a
 134   // single writev() call.
 135   constexpr int kNumIovecs = 64;
 136   std::array<iovec, kNumIovecs> iovecs;
 137
 138   size_t idx = 0;
 139   while (idx < ioQueue->size()) {
 140     int numIovecs = 0;
 141     while (numIovecs < kNumIovecs && idx < ioQueue->size()) {
 142       const auto& str = (*ioQueue)[idx];
 143       iovecs[numIovecs].iov_base = const_cast<char*>(str.data());
 144       iovecs[numIovecs].iov_len = str.size();
 145       ++numIovecs;
 146       ++idx;
 147     }
 148
 149     auto ret = folly::writevFull(file_.fd(), iovecs.data(), numIovecs);
 150     folly::checkUnixError(ret, "writeFull() failed");
 151   }
 152 }
 153
 154 void AsyncFileWriter::onIoError(const std::exception& ex) {
 155   LoggerDB::internalWarning(
 156       __FILE__,
 157       __LINE__,
 158       "error writing to log file ",
 159       file_.fd(),
 160       " in AsyncFileWriter: ",
 161       folly::exceptionStr(ex));
 162 }
 163
 164 std::string AsyncFileWriter::getNumDiscardedMsg(size_t numDiscarded) {
 165   // We may want to make this customizable in the future (e.g., to allow it to
 166   // conform to the LogFormatter style being used).
 167   // For now just return a simple fixed message.
 168   return folly::to<std::string>(
 169       numDiscarded,
 170       " log messages discarded: logging faster than we can write\n");
 171 }
 172 }