logging: improve the AsyncFileWriter flush test()
[folly.git] / 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("test message: " + std::string(200, 'x'));
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   // Detach the flush thread now rather than joining it at the end of the
221   // function.  This way if something goes wrong during the test we will fail
222   // with the real error, rather than crashing due to the std::thread
223   // destructor running on a still-joinable thread.
224   flushThread.detach();
225
226   // Sleep briefly, and make sure flush() still hasn't completed.
227   // If it has completed this doesn't necessarily indicate a bug in
228   // AsyncFileWriter, but instead indicates that our test code failed to
229   // successfully cause a blocking write.
230   /* sleep override */
231   std::this_thread::sleep_for(10ms);
232   EXPECT_FALSE(future.isReady());
233
234   // Now read from the pipe
235   std::vector<char> buf;
236   buf.resize(paddingSize);
237   readFull(readPipe.fd(), buf.data(), buf.size());
238
239   // Make sure flush completes successfully now
240   future.get(10ms);
241 }
242 #endif
243
244 // A large-ish message suffix, just to consume space and help fill up
245 // log buffers faster.
246 static constexpr StringPiece kMsgSuffix{
247     "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789"
248     "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789"
249     "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789"
250     "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789"};
251
252 class ReadStats {
253  public:
254   ReadStats()
255       : deadline_{steady_clock::now() +
256                   milliseconds{FLAGS_async_discard_timeout_msec}},
257         readSleepUS_{static_cast<uint64_t>(
258             std::min(0L, FLAGS_async_discard_read_sleep_usec))} {}
259
260   void clearSleepDuration() {
261     readSleepUS_.store(0);
262   }
263   std::chrono::microseconds getSleepUS() const {
264     return std::chrono::microseconds{readSleepUS_.load()};
265   }
266
267   bool shouldWriterStop() const {
268     // Stop after we have seen the required number of separate discard events.
269     // We stop based on discardEventsSeen_ rather than numDiscarded_ since this
270     // ensures the async writer blocks and then makes progress again multiple
271     // times.
272     if (FLAGS_async_discard_num_events > 0 &&
273         discardEventsSeen_.load() >
274             static_cast<uint64_t>(FLAGS_async_discard_num_events)) {
275       return true;
276     }
277
278     // Stop after a timeout, even if we don't hit the number of requested
279     // discards.
280     return steady_clock::now() > deadline_;
281   }
282   void writerFinished(size_t threadID, size_t messagesWritten, uint32_t flags) {
283     auto map = perThreadWriteData_.wlock();
284     assert(map->find(threadID) == map->end());
285     auto& data = (*map)[threadID];
286     data.numMessagesWritten = messagesWritten;
287     data.flags = flags;
288   }
289
290   void check() {
291     auto writeDataMap = perThreadWriteData_.wlock();
292
293     EXPECT_EQ("", trailingData_);
294     EXPECT_EQ(0, numUnableToParse_);
295     EXPECT_EQ(0, numOutOfOrder_);
296
297     // Check messages received from each writer thread
298     size_t readerStatsChecked = 0;
299     size_t totalMessagesWritten = 0;
300     size_t totalMessagesRead = 0;
301     for (const auto& writeEntry : *writeDataMap) {
302       const auto& writeInfo = writeEntry.second;
303       totalMessagesWritten += writeInfo.numMessagesWritten;
304
305       auto iter = perThreadReadData_.find(writeEntry.first);
306       if (iter == perThreadReadData_.end()) {
307         // We never received any messages from this writer thread.
308         // This is okay as long as this is not a NEVER_DISCARD writer.
309         EXPECT_EQ(0, writeInfo.flags);
310         continue;
311       }
312       const auto& readInfo = iter->second;
313       ++readerStatsChecked;
314       totalMessagesRead += readInfo.numMessagesRead;
315       if (writeInfo.flags & LogWriter::NEVER_DISCARD) {
316         // Non-discarding threads should never discard anything
317         EXPECT_EQ(readInfo.numMessagesRead, writeInfo.numMessagesWritten);
318         EXPECT_EQ(readInfo.lastId, writeInfo.numMessagesWritten);
319       } else {
320         // Other threads may have discarded some messages
321         EXPECT_LE(readInfo.numMessagesRead, writeInfo.numMessagesWritten);
322         EXPECT_LE(readInfo.lastId, writeInfo.numMessagesWritten);
323       }
324     }
325
326     EXPECT_EQ(totalMessagesWritten, totalMessagesRead + numDiscarded_);
327     EXPECT_EQ(readerStatsChecked, perThreadReadData_.size());
328
329     // This test is intended to check the discard behavior.
330     // Fail the test if we didn't actually trigger any discards before we timed
331     // out.
332     EXPECT_GT(numDiscarded_, 0);
333
334     XLOG(DBG1) << totalMessagesWritten << " messages written, "
335                << totalMessagesRead << " messages read, " << numDiscarded_
336                << " messages discarded";
337   }
338
339   void messageReceived(StringPiece msg) {
340     if (msg.endsWith(" log messages discarded: "
341                      "logging faster than we can write")) {
342       auto discardCount = folly::to<size_t>(msg.subpiece(0, msg.find(' ')));
343       XLOG(DBG3, "received discard notification: ", discardCount);
344       numDiscarded_ += discardCount;
345       ++discardEventsSeen_;
346       return;
347     }
348
349     size_t threadID = 0;
350     size_t messageIndex = 0;
351     try {
352       parseMessage(msg, &threadID, &messageIndex);
353     } catch (const std::exception& ex) {
354       ++numUnableToParse_;
355       XLOG(ERR, "unable to parse log message: ", msg);
356       return;
357     }
358
359     auto& data = perThreadReadData_[threadID];
360     data.numMessagesRead++;
361     if (messageIndex > data.lastId) {
362       data.lastId = messageIndex;
363     } else {
364       ++numOutOfOrder_;
365       XLOG(ERR) << "received out-of-order messages from writer " << threadID
366                 << ": " << messageIndex << " received after " << data.lastId;
367     }
368   }
369
370   void trailingData(StringPiece data) {
371     trailingData_ = data.str();
372   }
373
374  private:
375   struct ReaderData {
376     size_t numMessagesRead{0};
377     size_t lastId{0};
378   };
379   struct WriterData {
380     size_t numMessagesWritten{0};
381     int flags{0};
382   };
383
384   void parseMessage(StringPiece msg, size_t* threadID, size_t* messageIndex) {
385     // Validate and strip off the message prefix and suffix
386     constexpr StringPiece prefix{"thread "};
387     if (!msg.startsWith(prefix)) {
388       throw std::runtime_error("bad message prefix");
389     }
390     msg.advance(prefix.size());
391     if (!msg.endsWith(kMsgSuffix)) {
392       throw std::runtime_error("bad message suffix");
393     }
394     msg.subtract(kMsgSuffix.size());
395
396     // Parse then strip off the thread index
397     auto threadIDEnd = msg.find(' ');
398     if (threadIDEnd == StringPiece::npos) {
399       throw std::runtime_error("no middle found");
400     }
401     *threadID = folly::to<size_t>(msg.subpiece(0, threadIDEnd));
402     msg.advance(threadIDEnd);
403
404     // Validate that the middle of the message is what we expect,
405     // then strip it off
406     constexpr StringPiece middle{" message "};
407     if (!msg.startsWith(middle)) {
408       throw std::runtime_error("bad message middle");
409     }
410     msg.advance(middle.size());
411
412     // Parse the message index
413     *messageIndex = folly::to<size_t>(msg);
414   }
415
416   /**
417    * Data about each writer thread, as recorded by the reader thread.
418    *
419    * At the end of the test we will compare perThreadReadData_ (recorded by the
420    * reader) with perThreadWriteData_ (recorded by the writers) to make sure
421    * the data matches up.
422    *
423    * This is a map from writer_thread_id to ReaderData.
424    * The writer_thread_id is extracted from the received messages.
425    *
426    * This field does not need locking as it is only updated by the single
427    * reader thread.
428    */
429   std::unordered_map<size_t, ReaderData> perThreadReadData_;
430
431   /*
432    * Additional information recorded by the reader thread.
433    */
434   std::string trailingData_;
435   size_t numUnableToParse_{0};
436   size_t numOutOfOrder_{0};
437   size_t numDiscarded_{0};
438
439   /**
440    * deadline_ is a maximum end time for the test.
441    *
442    * The writer threads quit if the deadline is reached even if they have not
443    * produced the desired number of discard events yet.
444    */
445   const std::chrono::steady_clock::time_point deadline_;
446
447   /**
448    * How long the reader thread should sleep between each read event.
449    *
450    * This is initially set to a non-zero value (read from the
451    * FLAGS_async_discard_read_sleep_usec flag) so that the reader thread reads
452    * slowly, which will fill up the pipe buffer and cause discard events.
453    *
454    * Once we have produce enough discards and are ready to finish the test the
455    * main thread reduces readSleepUS_ to 0, so the reader will finish the
456    * remaining message backlog quickly.
457    */
458   std::atomic<uint64_t> readSleepUS_{0};
459
460   /**
461    * A count of how many discard events have been seen so far.
462    *
463    * The reader increments discardEventsSeen_ each time it sees a discard
464    * notification message.  A "discard event" basically corresponds to a single
465    * group of dropped messages.  Once the reader pulls some messages off out of
466    * the pipe the writers should be able to send more data, but the buffer will
467    * eventually fill up again, producing another discard event.
468    */
469   std::atomic<uint64_t> discardEventsSeen_{0};
470
471   /**
472    * Data about each writer thread, as recorded by the writers.
473    *
474    * When each writer thread finishes it records how many messages it wrote,
475    * plus the flags it used to write the messages.
476    */
477   folly::Synchronized<std::unordered_map<size_t, WriterData>>
478       perThreadWriteData_;
479 };
480
481 /**
482  * readThread() reads messages slowly from a pipe.  This helps test the
483  * AsyncFileWriter behavior when I/O is slow.
484  */
485 void readThread(folly::File&& file, ReadStats* stats) {
486   std::vector<char> buffer;
487   buffer.resize(1024);
488
489   size_t bufferIdx = 0;
490   while (true) {
491     /* sleep override */
492     std::this_thread::sleep_for(stats->getSleepUS());
493
494     auto readResult = folly::readNoInt(
495         file.fd(), buffer.data() + bufferIdx, buffer.size() - bufferIdx);
496     if (readResult < 0) {
497       XLOG(ERR, "error reading from pipe: ", errno);
498       return;
499     }
500     if (readResult == 0) {
501       XLOG(DBG2, "read EOF");
502       break;
503     }
504
505     auto logDataLen = bufferIdx + readResult;
506     StringPiece logData{buffer.data(), logDataLen};
507     auto idx = 0;
508     while (true) {
509       auto end = logData.find('\n', idx);
510       if (end == StringPiece::npos) {
511         bufferIdx = logDataLen - idx;
512         memmove(buffer.data(), buffer.data() + idx, bufferIdx);
513         break;
514       }
515
516       StringPiece logMsg{logData.data() + idx, end - idx};
517       stats->messageReceived(logMsg);
518       idx = end + 1;
519     }
520   }
521
522   if (bufferIdx != 0) {
523     stats->trailingData(StringPiece{buffer.data(), bufferIdx});
524   }
525 }
526
527 /**
528  * writeThread() writes a series of messages to the AsyncFileWriter
529  */
530 void writeThread(
531     AsyncFileWriter* writer,
532     size_t id,
533     uint32_t flags,
534     ReadStats* readStats) {
535   size_t msgID = 0;
536   while (true) {
537     ++msgID;
538     writer->writeMessage(
539         folly::to<std::string>(
540             "thread ", id, " message ", msgID, kMsgSuffix, '\n'),
541         flags);
542
543     // Break out once the reader has seen enough discards
544     if (((msgID & 0xff) == 0) && readStats->shouldWriterStop()) {
545       readStats->writerFinished(id, msgID, flags);
546       break;
547     }
548   }
549 }
550
551 /*
552  * The discard test spawns a number of threads that each write a large number
553  * of messages quickly.  The AsyncFileWriter writes to a pipe, an a separate
554  * thread reads from it slowly, causing a backlog to build up.
555  *
556  * The test then checks that:
557  * - The read thread always receives full messages (no partial log messages)
558  * - Messages that are received are received in order
559  * - The number of messages received plus the number reported in discard
560  *   notifications matches the number of messages sent.
561  */
562 TEST(AsyncFileWriter, discard) {
563   std::array<int, 2> fds;
564   auto pipeResult = pipe(fds.data());
565   folly::checkUnixError(pipeResult, "pipe failed");
566   folly::File readPipe{fds[0], true};
567   folly::File writePipe{fds[1], true};
568
569   ReadStats readStats;
570   std::thread reader(readThread, std::move(readPipe), &readStats);
571   {
572     AsyncFileWriter writer{std::move(writePipe)};
573
574     std::vector<std::thread> writeThreads;
575     size_t numThreads = FLAGS_async_discard_num_normal_writers +
576         FLAGS_async_discard_num_nodiscard_writers;
577
578     for (size_t n = 0; n < numThreads; ++n) {
579       uint32_t flags = 0;
580       if (n >= static_cast<size_t>(FLAGS_async_discard_num_normal_writers)) {
581         flags = LogWriter::NEVER_DISCARD;
582       }
583       XLOGF(DBG4, "writer {:4d} flags {:#02x}", n, flags);
584
585       writeThreads.emplace_back(writeThread, &writer, n, flags, &readStats);
586     }
587
588     for (auto& t : writeThreads) {
589       t.join();
590     }
591     XLOG(DBG2, "writers done");
592   }
593   // Clear the read sleep duration so the reader will finish quickly now
594   readStats.clearSleepDuration();
595   reader.join();
596   readStats.check();
597 }
598
599 int main(int argc, char* argv[]) {
600   testing::InitGoogleTest(&argc, argv);
601   folly::init(&argc, &argv);
602   // Don't use async logging in the async logging tests :-)
603   folly::initLoggingGlogStyle(FLAGS_logging, LogLevel::INFO, /* async */ false);
604
605   return RUN_ALL_TESTS();
606 }