folly/experimental/logging/AsyncFileWriter.h

   1 /*
   2  * Copyright 2017-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 #pragma once
  17
  18 #include <condition_variable>
  19 #include <mutex>
  20 #include <thread>
  21
  22 #include <folly/File.h>
  23 #include <folly/Range.h>
  24 #include <folly/Synchronized.h>
  25 #include <folly/experimental/logging/LogWriter.h>
  26
  27 namespace folly {
  28
  29 /**
  30  * A LogWriter implementation that asynchronously writes to a file descriptor.
  31  *
  32  * This class performs the log I/O in a separarate thread.
  33  *
  34  * The advantage of this class over ImmediateFileWriter is that logging I/O can
  35  * never slow down or block your normal program operation.  If log messages are
  36  * generated faster than they can be written, messages will be dropped (and an
  37  * indication of how many messages were dropped will be written to the log file
  38  * when we are able to catch up a bit.)
  39  *
  40  * However, one downside is that if your program crashes, not all log messages
  41  * may have been written, so you may lose messages generated immediately before
  42  * the crash.
  43  */
  44 class AsyncFileWriter : public LogWriter {
  45  public:
  46   /**
  47    * The default maximum buffer size.
  48    *
  49    * The comments for setMaxBufferSize() explain how this parameter is used.
  50    */
  51   static constexpr size_t kDefaultMaxBufferSize = 1024 * 1024;
  52
  53   /**
  54    * Construct an AsyncFileWriter that appends to the file at the specified
  55    * path.
  56    */
  57   explicit AsyncFileWriter(folly::StringPiece path);
  58
  59   /**
  60    * Construct an AsyncFileWriter that writes to the specified File object.
  61    */
  62   explicit AsyncFileWriter(folly::File&& file);
  63
  64   ~AsyncFileWriter();
  65
  66   void writeMessage(folly::StringPiece buffer, uint32_t flags = 0) override;
  67   void writeMessage(std::string&& buffer, uint32_t flags = 0) override;
  68
  69   /**
  70    * Block until the I/O thread has finished writing all messages that
  71    * were already enqueued when flush() was called.
  72    */
  73   void flush() override;
  74
  75   /**
  76    * Set the maximum buffer size for this AsyncFileWriter, in bytes.
  77    *
  78    * This controls the upper bound on how much unwritten data will be buffered
  79    * in memory.  If messages are being logged faster than they can be written
  80    * to output file, new messages will be discarded if they would cause the
  81    * amount of buffered data to exceed this limit.
  82    */
  83   void setMaxBufferSize(size_t size);
  84
  85   /**
  86    * Get the maximum buffer size for this AsyncFileWriter, in bytes.
  87    */
  88   size_t getMaxBufferSize() const;
  89
  90   /**
  91    * Get the output file.
  92    */
  93   const folly::File& getFile() const {
  94     return file_;
  95   }
  96
  97  private:
  98   /*
  99    * A simple implementation using two queues.
 100    * All writer threads enqueue into one queue while the I/O thread is
 101    * processing the other.
 102    *
 103    * We could potentially also provide an implementation using folly::MPMCQueue
 104    * in the future, which may improve contention under very high write loads.
 105    */
 106   struct Data {
 107     std::array<std::vector<std::string>, 2> queues;
 108     bool stop{false};
 109     bool ioThreadDone{false};
 110     uint64_t ioThreadCounter{0};
 111     size_t maxBufferBytes{kDefaultMaxBufferSize};
 112     size_t currentBufferSize{0};
 113     size_t numDiscarded{0};
 114
 115     std::vector<std::string>* getCurrentQueue() {
 116       return &queues[ioThreadCounter & 0x1];
 117     }
 118   };
 119
 120   void ioThread();
 121   void performIO(std::vector<std::string>* ioQueue);
 122
 123   void onIoError(const std::exception& ex);
 124   std::string getNumDiscardedMsg(size_t numDiscarded);
 125
 126   folly::File file_;
 127   folly::Synchronized<Data, std::mutex> data_;
 128   /**
 129    * messageReady_ is signaled by writer threads whenever they add a new
 130    * message to the current queue.
 131    */
 132   std::condition_variable messageReady_;
 133   /**
 134    * ioCV_ is signaled by the I/O thread each time it increments
 135    * the ioThreadCounter (once each time around its loop).
 136    */
 137   std::condition_variable ioCV_;
 138
 139   /**
 140    * The I/O thread.
 141    *
 142    * This should come last, since all other member variables need to be
 143    * constructed before the I/O thread starts.
 144    */
 145   std::thread ioThread_;
 146 };
 147 } // namespace folly