ReadStats Class Reference

Classes
struct	ReaderData
struct	WriterData

Public Member Functions
	ReadStats ()
void	clearSleepDuration ()
std::chrono::microseconds	getSleepUS () const
bool	shouldWriterStop () const
void	writerFinished (size_t threadID, size_t messagesWritten, uint32_t flags)
void	check ()
void	messageReceived (StringPiece msg)
void	trailingData (StringPiece data)

Private Member Functions
void	parseMessage (StringPiece msg, size_t *threadID, size_t *messageIndex)

Private Attributes
std::unordered_map< size_t, ReaderData >	perThreadReadData_
std::string	trailingData_
size_t	numUnableToParse_ {0}
size_t	numOutOfOrder_ {0}
size_t	numDiscarded_ {0}
const std::chrono::steady_clock::time_point	deadline_
std::atomic< uint64_t >	readSleepUS_ {0}
std::atomic< uint64_t >	discardEventsSeen_ {0}
folly::Synchronized< std::unordered_map< size_t, WriterData > >	perThreadWriteData_

Detailed Description

Definition at line 271 of file AsyncFileWriterTest.cpp.

Constructor & Destructor Documentation

ReadStats::ReadStats ( )

inline

Definition at line 273 of file AsyncFileWriterTest.cpp.

References int64_t, min, now(), and uint64_t.

      : deadline_{steady_clock::now() +
                  milliseconds{FLAGS_async_discard_timeout_msec}},
        readSleepUS_{static_cast<uint64_t>(
            std::min(int64_t{0}, FLAGS_async_discard_read_sleep_usec))} {}

Member Function Documentation

void ReadStats::check ( )

inline

Definition at line 311 of file AsyncFileWriterTest.cpp.

References folly::DBG1, EXPECT_EQ, EXPECT_GT, EXPECT_LE, folly::LogWriter::NEVER_DISCARD, and XLOG.

Referenced by TEST().

               {
    auto writeDataMap = perThreadWriteData_.wlock();

    EXPECT_EQ("", trailingData_);
    EXPECT_EQ(0, numUnableToParse_);
    EXPECT_EQ(0, numOutOfOrder_);

    // Check messages received from each writer thread
    size_t readerStatsChecked = 0;
    size_t totalMessagesWritten = 0;
    size_t totalMessagesRead = 0;
    for (const auto& writeEntry : *writeDataMap) {
      const auto& writeInfo = writeEntry.second;
      totalMessagesWritten += writeInfo.numMessagesWritten;

      auto iter = perThreadReadData_.find(writeEntry.first);
      if (iter == perThreadReadData_.end()) {
        // We never received any messages from this writer thread.
        // This is okay as long as this is not a NEVER_DISCARD writer.
        EXPECT_EQ(0, writeInfo.flags);
        continue;
      }
      const auto& readInfo = iter->second;
      ++readerStatsChecked;
      totalMessagesRead += readInfo.numMessagesRead;
      if (writeInfo.flags & LogWriter::NEVER_DISCARD) {
        // Non-discarding threads should never discard anything
        EXPECT_EQ(readInfo.numMessagesRead, writeInfo.numMessagesWritten);
        EXPECT_EQ(readInfo.lastId, writeInfo.numMessagesWritten);
      } else {
        // Other threads may have discarded some messages
        EXPECT_LE(readInfo.numMessagesRead, writeInfo.numMessagesWritten);
        EXPECT_LE(readInfo.lastId, writeInfo.numMessagesWritten);
      }
    }

    EXPECT_EQ(totalMessagesWritten, totalMessagesRead + numDiscarded_);
    EXPECT_EQ(readerStatsChecked, perThreadReadData_.size());

    // This test is intended to check the discard behavior.
    // Fail the test if we didn't actually trigger any discards before we timed
    // out.
    EXPECT_GT(numDiscarded_, 0);

    XLOG(DBG1) << totalMessagesWritten << " messages written, "
               << totalMessagesRead << " messages read, " << numDiscarded_
               << " messages discarded";
  }

void ReadStats::clearSleepDuration ( )

inline

Definition at line 279 of file AsyncFileWriterTest.cpp.

Referenced by TEST().

                            {
    readSleepUS_.store(0);
  }

std::chrono::microseconds ReadStats::getSleepUS ( ) const

inline

Definition at line 282 of file AsyncFileWriterTest.cpp.

Referenced by readThread().

                                           {
    return std::chrono::microseconds{readSleepUS_.load()};
  }

void ReadStats::messageReceived ( StringPiece  msg )

inline

Definition at line 360 of file AsyncFileWriterTest.cpp.

References folly::data(), folly::DBG3, folly::Range< Iter >::endsWith(), folly::ERR, folly::Range< Iter >::find(), folly::Range< Iter >::subpiece(), and XLOG.

Referenced by readThread().

                                        {
    if (msg.endsWith(" log messages discarded: "
                     "logging faster than we can write")) {
      auto discardCount = folly::to<size_t>(msg.subpiece(0, msg.find(' ')));
      XLOG(DBG3, "received discard notification: ", discardCount);
      numDiscarded_ += discardCount;
      ++discardEventsSeen_;
      return;
    }

    size_t threadID = 0;
    size_t messageIndex = 0;
    try {
      parseMessage(msg, &threadID, &messageIndex);
    } catch (const std::exception& ex) {
      ++numUnableToParse_;
      XLOG(ERR, "unable to parse log message: ", msg);
      return;
    }

    auto& data = perThreadReadData_[threadID];
    data.numMessagesRead++;
    if (messageIndex > data.lastId) {
      data.lastId = messageIndex;
    } else {
      ++numOutOfOrder_;
      XLOG(ERR) << "received out-of-order messages from writer " << threadID
                << ": " << messageIndex << " received after " << data.lastId;
    }
  }

void ReadStats::parseMessage ( StringPiece  msg, size_t *  threadID, size_t *  messageIndex  )

inlineprivate

Definition at line 405 of file AsyncFileWriterTest.cpp.

References folly::Range< Iter >::advance(), folly::Range< Iter >::endsWith(), folly::Range< Iter >::find(), kMsgSuffix, folly::Range< const char * >::npos, prefix(), folly::Range< Iter >::size(), folly::Range< Iter >::startsWith(), folly::Range< Iter >::subpiece(), and folly::Range< Iter >::subtract().

                                                                             {
    // Validate and strip off the message prefix and suffix
    constexpr StringPiece prefix{"thread "};
    if (!msg.startsWith(prefix)) {
      throw std::runtime_error("bad message prefix");
    }
    msg.advance(prefix.size());
    if (!msg.endsWith(kMsgSuffix)) {
      throw std::runtime_error("bad message suffix");
    }
    msg.subtract(kMsgSuffix.size());

    // Parse then strip off the thread index
    auto threadIDEnd = msg.find(' ');
    if (threadIDEnd == StringPiece::npos) {
      throw std::runtime_error("no middle found");
    }
    *threadID = folly::to<size_t>(msg.subpiece(0, threadIDEnd));
    msg.advance(threadIDEnd);

    // Validate that the middle of the message is what we expect,
    // then strip it off
    constexpr StringPiece middle{" message "};
    if (!msg.startsWith(middle)) {
      throw std::runtime_error("bad message middle");
    }
    msg.advance(middle.size());

    // Parse the message index
    *messageIndex = folly::to<size_t>(msg);
  }

bool ReadStats::shouldWriterStop ( ) const

inline

Definition at line 286 of file AsyncFileWriterTest.cpp.

References now(), and uint64_t.

Referenced by writeThread().

                                {
    // Stop after we have seen the required number of separate discard events.
    // We stop based on discardEventsSeen_ rather than numDiscarded_ since this
    // ensures the async writer blocks and then makes progress again multiple
    // times.
    if (FLAGS_async_discard_num_events > 0 &&
        discardEventsSeen_.load() >
            static_cast<uint64_t>(FLAGS_async_discard_num_events)) {
      return true;
    }

    // Stop after a timeout, even if we don't hit the number of requested
    // discards.
    return steady_clock::now() > deadline_;
  }

void ReadStats::trailingData ( StringPiece  data )

inline

Definition at line 391 of file AsyncFileWriterTest.cpp.

References folly::Range< Iter >::str().

Referenced by readThread().

                                      {
    trailingData_ = data.str();
  }

void ReadStats::writerFinished ( size_t  threadID, size_t  messagesWritten, uint32_t  flags  )

inline

Definition at line 301 of file AsyncFileWriterTest.cpp.

References folly::data(), and FOLLY_SAFE_CHECK.

Referenced by writeThread().

                                                                               {
    auto map = perThreadWriteData_.wlock();
    FOLLY_SAFE_CHECK(
        map->find(threadID) == map->end(),
        "multiple writer threads with same ID");
    auto& data = (*map)[threadID];
    data.numMessagesWritten = messagesWritten;
    data.flags = flags;
  }

Member Data Documentation

const std::chrono::steady_clock::time_point ReadStats::deadline_

private

deadline_ is a maximum end time for the test.

The writer threads quit if the deadline is reached even if they have not produced the desired number of discard events yet.

Definition at line 466 of file AsyncFileWriterTest.cpp.

std::atomic<uint64_t> ReadStats::discardEventsSeen_ {0}

private

A count of how many discard events have been seen so far.

The reader increments discardEventsSeen_ each time it sees a discard notification message. A "discard event" basically corresponds to a single group of dropped messages. Once the reader pulls some messages off out of the pipe the writers should be able to send more data, but the buffer will eventually fill up again, producing another discard event.

Definition at line 490 of file AsyncFileWriterTest.cpp.

size_t ReadStats::numDiscarded_ {0}

private

Definition at line 458 of file AsyncFileWriterTest.cpp.

size_t ReadStats::numOutOfOrder_ {0}

private

Definition at line 457 of file AsyncFileWriterTest.cpp.

size_t ReadStats::numUnableToParse_ {0}

private

Definition at line 456 of file AsyncFileWriterTest.cpp.

std::unordered_map<size_t, ReaderData> ReadStats::perThreadReadData_

private

Data about each writer thread, as recorded by the reader thread.

At the end of the test we will compare perThreadReadData_ (recorded by the reader) with perThreadWriteData_ (recorded by the writers) to make sure the data matches up.

This is a map from writer_thread_id to ReaderData. The writer_thread_id is extracted from the received messages.

This field does not need locking as it is only updated by the single reader thread.

Definition at line 450 of file AsyncFileWriterTest.cpp.

folly::Synchronized<std::unordered_map<size_t, WriterData> > ReadStats::perThreadWriteData_

private

Data about each writer thread, as recorded by the writers.

When each writer thread finishes it records how many messages it wrote, plus the flags it used to write the messages.

Definition at line 499 of file AsyncFileWriterTest.cpp.

std::atomic<uint64_t> ReadStats::readSleepUS_ {0}

private

How long the reader thread should sleep between each read event.

This is initially set to a non-zero value (read from the FLAGS_async_discard_read_sleep_usec flag) so that the reader thread reads slowly, which will fill up the pipe buffer and cause discard events.

Once we have produce enough discards and are ready to finish the test the main thread reduces readSleepUS_ to 0, so the reader will finish the remaining message backlog quickly.

Definition at line 479 of file AsyncFileWriterTest.cpp.

std::string ReadStats::trailingData_

private

Definition at line 455 of file AsyncFileWriterTest.cpp.

---

The documentation for this class was generated from the following file:

• proxygen/folly/folly/logging/test/AsyncFileWriterTest.cpp