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