folly/experimental/logging/test/AsyncFileWriterTest.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 <thread>
  17
  18 #include <folly/Conv.h>
  19 #include <folly/Exception.h>
  20 #include <folly/File.h>
  21 #include <folly/FileUtil.h>
  22 #include <folly/String.h>
  23 #include <folly/Synchronized.h>
  24 #include <folly/experimental/TestUtil.h>
  25 #include <folly/experimental/logging/AsyncFileWriter.h>
  26 #include <folly/experimental/logging/Init.h>
  27 #include <folly/experimental/logging/LoggerDB.h>
  28 #include <folly/experimental/logging/xlog.h>
  29 #include <folly/futures/Future.h>
  30 #include <folly/futures/Promise.h>
  31 #include <folly/init/Init.h>
  32 #include <folly/portability/GFlags.h>
  33 #include <folly/portability/GMock.h>
  34 #include <folly/portability/GTest.h>
  35 #include <folly/portability/Unistd.h>
  36
  37 DEFINE_string(logging, "", "folly::logging configuration");
  38 DEFINE_int64(
  39     async_discard_num_normal_writers,
  40     30,
  41     "number of threads to use to generate normal log messages during "
  42     "the AsyncFileWriter.discard test");
  43 DEFINE_int64(
  44     async_discard_num_nodiscard_writers,
  45     2,
  46     "number of threads to use to generate non-discardable log messages during "
  47     "the AsyncFileWriter.discard test");
  48 DEFINE_int64(
  49     async_discard_read_sleep_usec,
  50     500,
  51     "how long the read thread should sleep between reads in "
  52     "the AsyncFileWriter.discard test");
  53 DEFINE_int64(
  54     async_discard_timeout_msec,
  55     10000,
  56     "A timeout for the AsyncFileWriter.discard test if it cannot generate "
  57     "enough discards");
  58 DEFINE_int64(
  59     async_discard_num_events,
  60     10,
  61     "The number of discard events to wait for in the AsyncFileWriter.discard "
  62     "test");
  63
  64 using namespace folly;
  65 using namespace std::literals::chrono_literals;
  66 using folly::test::TemporaryFile;
  67 using std::chrono::steady_clock;
  68 using std::chrono::milliseconds;
  69
  70 TEST(AsyncFileWriter, noMessages) {
  71   TemporaryFile tmpFile{"logging_test"};
  72
  73   // Test the simple construction and destruction of an AsyncFileWriter
  74   // without ever writing any messages.  This still exercises the I/O
  75   // thread start-up and shutdown code.
  76   AsyncFileWriter writer{folly::File{tmpFile.fd(), false}};
  77 }
  78
  79 TEST(AsyncFileWriter, simpleMessages) {
  80   TemporaryFile tmpFile{"logging_test"};
  81
  82   {
  83     AsyncFileWriter writer{folly::File{tmpFile.fd(), false}};
  84     for (int n = 0; n < 10; ++n) {
  85       writer.writeMessage(folly::to<std::string>("message ", n, "\n"));
  86       sched_yield();
  87     }
  88   }
  89
  90   std::string data;
  91   auto ret = folly::readFile(tmpFile.path().native().c_str(), data);
  92   ASSERT_TRUE(ret);
  93
  94   std::string expected =
  95       "message 0\n"
  96       "message 1\n"
  97       "message 2\n"
  98       "message 3\n"
  99       "message 4\n"
 100       "message 5\n"
 101       "message 6\n"
 102       "message 7\n"
 103       "message 8\n"
 104       "message 9\n";
 105   EXPECT_EQ(expected, data);
 106 }
 107
 108 #ifndef _WIN32
 109 namespace {
 110 static std::vector<std::string>* internalWarnings;
 111
 112 void handleLoggingError(
 113     StringPiece /* file */,
 114     int /* lineNumber */,
 115     std::string&& msg) {
 116   internalWarnings->emplace_back(std::move(msg));
 117 }
 118 }
 119
 120 TEST(AsyncFileWriter, ioError) {
 121   // Set the LoggerDB internal warning handler so we can record the messages
 122   std::vector<std::string> logErrors;
 123   internalWarnings = &logErrors;
 124   LoggerDB::setInternalWarningHandler(handleLoggingError);
 125
 126   // Create an AsyncFileWriter that refers to a pipe whose read end is closed
 127   std::array<int, 2> fds;
 128   auto rc = pipe(fds.data());
 129   folly::checkUnixError(rc, "failed to create pipe");
 130   signal(SIGPIPE, SIG_IGN);
 131   ::close(fds[0]);
 132
 133   // Log a bunch of messages to the writer
 134   size_t numMessages = 100;
 135   {
 136     AsyncFileWriter writer{folly::File{fds[1], true}};
 137     for (size_t n = 0; n < numMessages; ++n) {
 138       writer.writeMessage(folly::to<std::string>("message ", n, "\n"));
 139       sched_yield();
 140     }
 141   }
 142
 143   LoggerDB::setInternalWarningHandler(nullptr);
 144
 145   // AsyncFileWriter should have some internal warning messages about the
 146   // log failures.  This will generally be many fewer than the number of
 147   // messages we wrote, though, since it performs write batching.
 148   for (const auto& msg : logErrors) {
 149     EXPECT_THAT(
 150         msg,
 151         testing::ContainsRegex(
 152             "error writing to log file .* in AsyncFileWriter.*: Broken pipe"));
 153   }
 154   EXPECT_GT(logErrors.size(), 0);
 155   EXPECT_LE(logErrors.size(), numMessages);
 156 }
 157
 158 namespace {
 159 size_t fillUpPipe(int fd) {
 160   int flags = fcntl(fd, F_GETFL);
 161   folly::checkUnixError(flags, "failed get file descriptor flags");
 162   auto rc = fcntl(fd, F_SETFL, flags | O_NONBLOCK);
 163   folly::checkUnixError(rc, "failed to put pipe in non-blocking mode");
 164   std::vector<char> data;
 165   data.resize(4000);
 166   size_t totalBytes = 0;
 167   size_t bytesToWrite = data.size();
 168   while (true) {
 169     auto bytesWritten = writeNoInt(fd, data.data(), bytesToWrite);
 170     if (bytesWritten < 0) {
 171       if (errno == EAGAIN || errno == EWOULDBLOCK) {
 172         // We blocked.  Keep trying smaller writes, until we get down to a
 173         // single byte, just to make sure the logging code really won't be able
 174         // to write anything to the pipe.
 175         if (bytesToWrite <= 1) {
 176           break;
 177         } else {
 178           bytesToWrite /= 2;
 179         }
 180       } else {
 181         throwSystemError("error writing to pipe");
 182       }
 183     } else {
 184       totalBytes += bytesWritten;
 185     }
 186   }
 187   XLOG(DBG1, "pipe filled up after ", totalBytes, " bytes");
 188
 189   rc = fcntl(fd, F_SETFL, flags & ~O_NONBLOCK);
 190   folly::checkUnixError(rc, "failed to put pipe back in blocking mode");
 191
 192   return totalBytes;
 193 }
 194 }
 195
 196 TEST(AsyncFileWriter, flush) {
 197   // Set up a pipe(), then write data to the write endpoint until it fills up
 198   // and starts blocking.
 199   std::array<int, 2> fds;
 200   auto rc = pipe(fds.data());
 201   folly::checkUnixError(rc, "failed to create pipe");
 202   File readPipe{fds[0], true};
 203   File writePipe{fds[1], true};
 204
 205   auto paddingSize = fillUpPipe(writePipe.fd());
 206
 207   // Now set up an AsyncFileWriter pointing at the write end of the pipe
 208   AsyncFileWriter writer{std::move(writePipe)};
 209
 210   // Write a message
 211   writer.writeMessage(std::string{"test message"});
 212
 213   // Call flush().  Use a separate thread, since this should block until we
 214   // consume data from the pipe.
 215   Promise<Unit> promise;
 216   auto future = promise.getFuture();
 217   auto flushFunction = [&] { writer.flush(); };
 218   std::thread flushThread{
 219       [&]() { promise.setTry(makeTryWith(flushFunction)); }};
 220
 221   // Sleep briefly, and make sure flush() still hasn't completed.
 222   /* sleep override */
 223   std::this_thread::sleep_for(10ms);
 224   EXPECT_FALSE(future.isReady());
 225
 226   // Now read from the pipe
 227   std::vector<char> buf;
 228   buf.resize(paddingSize);
 229   readFull(readPipe.fd(), buf.data(), buf.size());
 230
 231   // Make sure flush completes successfully now
 232   future.get(10ms);
 233   flushThread.join();
 234 }
 235 #endif
 236
 237 // A large-ish message suffix, just to consume space and help fill up
 238 // log buffers faster.
 239 static constexpr StringPiece kMsgSuffix{
 240     "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789"
 241     "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789"
 242     "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789"
 243     "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789"};
 244
 245 class ReadStats {
 246  public:
 247   ReadStats()
 248       : deadline_{steady_clock::now() +
 249                   milliseconds{FLAGS_async_discard_timeout_msec}},
 250         readSleepUS_{static_cast<uint64_t>(
 251             std::min(0L, FLAGS_async_discard_read_sleep_usec))} {}
 252
 253   void clearSleepDuration() {
 254     readSleepUS_.store(0);
 255   }
 256   std::chrono::microseconds getSleepUS() const {
 257     return std::chrono::microseconds{readSleepUS_.load()};
 258   }
 259
 260   bool shouldWriterStop() const {
 261     // Stop after we have seen the required number of separate discard events.
 262     // We stop based on discardEventsSeen_ rather than numDiscarded_ since this
 263     // ensures the async writer blocks and then makes progress again multiple
 264     // times.
 265     if (FLAGS_async_discard_num_events > 0 &&
 266         discardEventsSeen_.load() >
 267             static_cast<uint64_t>(FLAGS_async_discard_num_events)) {
 268       return true;
 269     }
 270
 271     // Stop after a timeout, even if we don't hit the number of requested
 272     // discards.
 273     return steady_clock::now() > deadline_;
 274   }
 275   void writerFinished(size_t threadID, size_t messagesWritten, uint32_t flags) {
 276     auto map = perThreadWriteData_.wlock();
 277     assert(map->find(threadID) == map->end());
 278     auto& data = (*map)[threadID];
 279     data.numMessagesWritten = messagesWritten;
 280     data.flags = flags;
 281   }
 282
 283   void check() {
 284     auto writeDataMap = perThreadWriteData_.wlock();
 285
 286     EXPECT_EQ("", trailingData_);
 287     EXPECT_EQ(0, numUnableToParse_);
 288     EXPECT_EQ(0, numOutOfOrder_);
 289
 290     // Check messages received from each writer thread
 291     size_t readerStatsChecked = 0;
 292     size_t totalMessagesWritten = 0;
 293     size_t totalMessagesRead = 0;
 294     for (const auto& writeEntry : *writeDataMap) {
 295       const auto& writeInfo = writeEntry.second;
 296       totalMessagesWritten += writeInfo.numMessagesWritten;
 297
 298       auto iter = perThreadReadData_.find(writeEntry.first);
 299       if (iter == perThreadReadData_.end()) {
 300         // We never received any messages from this writer thread.
 301         // This is okay as long as this is not a NEVER_DISCARD writer.
 302         EXPECT_EQ(0, writeInfo.flags);
 303         continue;
 304       }
 305       const auto& readInfo = iter->second;
 306       ++readerStatsChecked;
 307       totalMessagesRead += readInfo.numMessagesRead;
 308       if (writeInfo.flags & LogWriter::NEVER_DISCARD) {
 309         // Non-discarding threads should never discard anything
 310         EXPECT_EQ(readInfo.numMessagesRead, writeInfo.numMessagesWritten);
 311         EXPECT_EQ(readInfo.lastId, writeInfo.numMessagesWritten);
 312       } else {
 313         // Other threads may have discarded some messages
 314         EXPECT_LE(readInfo.numMessagesRead, writeInfo.numMessagesWritten);
 315         EXPECT_LE(readInfo.lastId, writeInfo.numMessagesWritten);
 316       }
 317     }
 318
 319     EXPECT_EQ(totalMessagesWritten, totalMessagesRead + numDiscarded_);
 320     EXPECT_EQ(readerStatsChecked, perThreadReadData_.size());
 321
 322     // This test is intended to check the discard behavior.
 323     // Fail the test if we didn't actually trigger any discards before we timed
 324     // out.
 325     EXPECT_GT(numDiscarded_, 0);
 326
 327     XLOG(DBG1) << totalMessagesWritten << " messages written, "
 328                << totalMessagesRead << " messages read, " << numDiscarded_
 329                << " messages discarded";
 330   }
 331
 332   void messageReceived(StringPiece msg) {
 333     if (msg.endsWith(" log messages discarded: "
 334                      "logging faster than we can write")) {
 335       auto discardCount = folly::to<size_t>(msg.subpiece(0, msg.find(' ')));
 336       XLOG(DBG3, "received discard notification: ", discardCount);
 337       numDiscarded_ += discardCount;
 338       ++discardEventsSeen_;
 339       return;
 340     }
 341
 342     size_t threadID = 0;
 343     size_t messageIndex = 0;
 344     try {
 345       parseMessage(msg, &threadID, &messageIndex);
 346     } catch (const std::exception& ex) {
 347       ++numUnableToParse_;
 348       XLOG(ERR, "unable to parse log message: ", msg);
 349       return;
 350     }
 351
 352     auto& data = perThreadReadData_[threadID];
 353     data.numMessagesRead++;
 354     if (messageIndex > data.lastId) {
 355       data.lastId = messageIndex;
 356     } else {
 357       ++numOutOfOrder_;
 358       XLOG(ERR) << "received out-of-order messages from writer " << threadID
 359                 << ": " << messageIndex << " received after " << data.lastId;
 360     }
 361   }
 362
 363   void trailingData(StringPiece data) {
 364     trailingData_ = data.str();
 365   }
 366
 367  private:
 368   struct ReaderData {
 369     size_t numMessagesRead{0};
 370     size_t lastId{0};
 371   };
 372   struct WriterData {
 373     size_t numMessagesWritten{0};
 374     int flags{0};
 375   };
 376
 377   void parseMessage(StringPiece msg, size_t* threadID, size_t* messageIndex) {
 378     // Validate and strip off the message prefix and suffix
 379     constexpr StringPiece prefix{"thread "};
 380     if (!msg.startsWith(prefix)) {
 381       throw std::runtime_error("bad message prefix");
 382     }
 383     msg.advance(prefix.size());
 384     if (!msg.endsWith(kMsgSuffix)) {
 385       throw std::runtime_error("bad message suffix");
 386     }
 387     msg.subtract(kMsgSuffix.size());
 388
 389     // Parse then strip off the thread index
 390     auto threadIDEnd = msg.find(' ');
 391     if (threadIDEnd == StringPiece::npos) {
 392       throw std::runtime_error("no middle found");
 393     }
 394     *threadID = folly::to<size_t>(msg.subpiece(0, threadIDEnd));
 395     msg.advance(threadIDEnd);
 396
 397     // Validate that the middle of the message is what we expect,
 398     // then strip it off
 399     constexpr StringPiece middle{" message "};
 400     if (!msg.startsWith(middle)) {
 401       throw std::runtime_error("bad message middle");
 402     }
 403     msg.advance(middle.size());
 404
 405     // Parse the message index
 406     *messageIndex = folly::to<size_t>(msg);
 407   }
 408
 409   /**
 410    * Data about each writer thread, as recorded by the reader thread.
 411    *
 412    * At the end of the test we will compare perThreadReadData_ (recorded by the
 413    * reader) with perThreadWriteData_ (recorded by the writers) to make sure
 414    * the data matches up.
 415    *
 416    * This is a map from writer_thread_id to ReaderData.
 417    * The writer_thread_id is extracted from the received messages.
 418    *
 419    * This field does not need locking as it is only updated by the single
 420    * reader thread.
 421    */
 422   std::unordered_map<size_t, ReaderData> perThreadReadData_;
 423
 424   /*
 425    * Additional information recorded by the reader thread.
 426    */
 427   std::string trailingData_;
 428   size_t numUnableToParse_{0};
 429   size_t numOutOfOrder_{0};
 430   size_t numDiscarded_{0};
 431
 432   /**
 433    * deadline_ is a maximum end time for the test.
 434    *
 435    * The writer threads quit if the deadline is reached even if they have not
 436    * produced the desired number of discard events yet.
 437    */
 438   const std::chrono::steady_clock::time_point deadline_;
 439
 440   /**
 441    * How long the reader thread should sleep between each read event.
 442    *
 443    * This is initially set to a non-zero value (read from the
 444    * FLAGS_async_discard_read_sleep_usec flag) so that the reader thread reads
 445    * slowly, which will fill up the pipe buffer and cause discard events.
 446    *
 447    * Once we have produce enough discards and are ready to finish the test the
 448    * main thread reduces readSleepUS_ to 0, so the reader will finish the
 449    * remaining message backlog quickly.
 450    */
 451   std::atomic<uint64_t> readSleepUS_{0};
 452
 453   /**
 454    * A count of how many discard events have been seen so far.
 455    *
 456    * The reader increments discardEventsSeen_ each time it sees a discard
 457    * notification message.  A "discard event" basically corresponds to a single
 458    * group of dropped messages.  Once the reader pulls some messages off out of
 459    * the pipe the writers should be able to send more data, but the buffer will
 460    * eventually fill up again, producing another discard event.
 461    */
 462   std::atomic<uint64_t> discardEventsSeen_{0};
 463
 464   /**
 465    * Data about each writer thread, as recorded by the writers.
 466    *
 467    * When each writer thread finishes it records how many messages it wrote,
 468    * plus the flags it used to write the messages.
 469    */
 470   folly::Synchronized<std::unordered_map<size_t, WriterData>>
 471       perThreadWriteData_;
 472 };
 473
 474 /**
 475  * readThread() reads messages slowly from a pipe.  This helps test the
 476  * AsyncFileWriter behavior when I/O is slow.
 477  */
 478 void readThread(folly::File&& file, ReadStats* stats) {
 479   std::vector<char> buffer;
 480   buffer.resize(1024);
 481
 482   size_t bufferIdx = 0;
 483   while (true) {
 484     /* sleep override */
 485     std::this_thread::sleep_for(stats->getSleepUS());
 486
 487     auto readResult = folly::readNoInt(
 488         file.fd(), buffer.data() + bufferIdx, buffer.size() - bufferIdx);
 489     if (readResult < 0) {
 490       XLOG(ERR, "error reading from pipe: ", errno);
 491       return;
 492     }
 493     if (readResult == 0) {
 494       XLOG(DBG2, "read EOF");
 495       break;
 496     }
 497
 498     auto logDataLen = bufferIdx + readResult;
 499     StringPiece logData{buffer.data(), logDataLen};
 500     auto idx = 0;
 501     while (true) {
 502       auto end = logData.find('\n', idx);
 503       if (end == StringPiece::npos) {
 504         bufferIdx = logDataLen - idx;
 505         memmove(buffer.data(), buffer.data() + idx, bufferIdx);
 506         break;
 507       }
 508
 509       StringPiece logMsg{logData.data() + idx, end - idx};
 510       stats->messageReceived(logMsg);
 511       idx = end + 1;
 512     }
 513   }
 514
 515   if (bufferIdx != 0) {
 516     stats->trailingData(StringPiece{buffer.data(), bufferIdx});
 517   }
 518 }
 519
 520 /**
 521  * writeThread() writes a series of messages to the AsyncFileWriter
 522  */
 523 void writeThread(
 524     AsyncFileWriter* writer,
 525     size_t id,
 526     uint32_t flags,
 527     ReadStats* readStats) {
 528   size_t msgID = 0;
 529   while (true) {
 530     ++msgID;
 531     writer->writeMessage(
 532         folly::to<std::string>(
 533             "thread ", id, " message ", msgID, kMsgSuffix, '\n'),
 534         flags);
 535
 536     // Break out once the reader has seen enough discards
 537     if (((msgID & 0xff) == 0) && readStats->shouldWriterStop()) {
 538       readStats->writerFinished(id, msgID, flags);
 539       break;
 540     }
 541   }
 542 }
 543
 544 /*
 545  * The discard test spawns a number of threads that each write a large number
 546  * of messages quickly.  The AsyncFileWriter writes to a pipe, an a separate
 547  * thread reads from it slowly, causing a backlog to build up.
 548  *
 549  * The test then checks that:
 550  * - The read thread always receives full messages (no partial log messages)
 551  * - Messages that are received are received in order
 552  * - The number of messages received plus the number reported in discard
 553  *   notifications matches the number of messages sent.
 554  */
 555 TEST(AsyncFileWriter, discard) {
 556   std::array<int, 2> fds;
 557   auto pipeResult = pipe(fds.data());
 558   folly::checkUnixError(pipeResult, "pipe failed");
 559   folly::File readPipe{fds[0], true};
 560   folly::File writePipe{fds[1], true};
 561
 562   ReadStats readStats;
 563   std::thread reader(readThread, std::move(readPipe), &readStats);
 564   {
 565     AsyncFileWriter writer{std::move(writePipe)};
 566
 567     std::vector<std::thread> writeThreads;
 568     size_t numThreads = FLAGS_async_discard_num_normal_writers +
 569         FLAGS_async_discard_num_nodiscard_writers;
 570
 571     for (size_t n = 0; n < numThreads; ++n) {
 572       uint32_t flags = 0;
 573       if (n >= static_cast<size_t>(FLAGS_async_discard_num_normal_writers)) {
 574         flags = LogWriter::NEVER_DISCARD;
 575       }
 576       XLOGF(DBG4, "writer {:4d} flags {:#02x}", n, flags);
 577
 578       writeThreads.emplace_back(writeThread, &writer, n, flags, &readStats);
 579     }
 580
 581     for (auto& t : writeThreads) {
 582       t.join();
 583     }
 584     XLOG(DBG2, "writers done");
 585   }
 586   // Clear the read sleep duration so the reader will finish quickly now
 587   readStats.clearSleepDuration();
 588   reader.join();
 589   readStats.check();
 590 }
 591
 592 int main(int argc, char* argv[]) {
 593   testing::InitGoogleTest(&argc, argv);
 594   folly::init(&argc, &argv);
 595   // Don't use async logging in the async logging tests :-)
 596   folly::initLoggingGlogStyle(FLAGS_logging, LogLevel::INFO, /* async */ false);
 597
 598   return RUN_ALL_TESTS();
 599 }