Trim IPC ChannelReader's buffer after processing large messages.

Most IPC messages are small, but there are few ones that can be quite large.
With this change ChannelReader trims its input buffer after processing large
messages.

BUG=529940

Review URL: https://codereview.chromium.org/1377483003

Cr-Commit-Position: refs/heads/master@{#357708}

4 files changed, 218 insertions, 5 deletions

diff --git a/ipc/ipc_channel.h b/ipc/ipc_channel.h
index 142f3d5..5345112 100644
--- a/ipc/ipc_channel.h
+++ b/ipc/ipc_channel.h
@@ -97,6 +97,11 @@ class IPC_EXPORT Channel : public Endpoint {
   // Amount of data to read at once from the pipe.
   static const size_t kReadBufferSize = 4 * 1024;
 
+  // Maximum persistent read buffer size. Read buffer can grow larger to
+  // accommodate large messages, but it's recommended to shrink back to this
+  // value because it fits 99.9% of all messages (see issue 529940 for data).
+  static const size_t kMaximumReadBufferSize = 64 * 1024;
+
   // Initialize a Channel.
   //
   // |channel_handle| identifies the communication Channel. For POSIX, if
diff --git a/ipc/ipc_channel_reader.cc b/ipc/ipc_channel_reader.cc
index a82093e..335852f 100644
--- a/ipc/ipc_channel_reader.cc
+++ b/ipc/ipc_channel_reader.cc
@@ -15,7 +15,9 @@
 namespace IPC {
 namespace internal {
 
-ChannelReader::ChannelReader(Listener* listener) : listener_(listener) {
+ChannelReader::ChannelReader(Listener* listener)
+  : listener_(listener),
+    max_input_buffer_size_(Channel::kMaximumReadBufferSize) {
   memset(input_buf_, 0, sizeof(input_buf_));
 }
 
@@ -114,6 +116,11 @@ bool ChannelReader::TranslateInputData(const char* input_data,
     }
   }
 
+  // Account for the case where last message's byte is in the next data chunk.
+  size_t next_message_buffer_size = next_message_size ?
+      next_message_size + Channel::kReadBufferSize - 1:
+      0;
+
   // Save any partial data in the overflow buffer.
   input_overflow_buf_.assign(p, end - p);
 
@@ -122,10 +129,28 @@ bool ChannelReader::TranslateInputData(const char* input_data,
     // append the next data chunk (instead of parsing it directly). So we
     // resize the buffer to fit the next message, to avoid repeatedly
     // growing the buffer as we receive all message' data chunks.
-    next_message_size += Channel::kReadBufferSize - 1;
-    if (next_message_size > input_overflow_buf_.capacity()) {
-      input_overflow_buf_.reserve(next_message_size);
+    if (next_message_buffer_size > input_overflow_buf_.capacity()) {
+      input_overflow_buf_.reserve(next_message_buffer_size);
+    }
+  }
+
+  // Trim the buffer if we can
+  if (next_message_buffer_size < max_input_buffer_size_ &&
+      input_overflow_buf_.size() < max_input_buffer_size_ &&
+      input_overflow_buf_.capacity() > max_input_buffer_size_) {
+    // std::string doesn't really have a method to shrink capacity to
+    // a specific value, so we have to swap with another string.
+    std::string trimmed_buf;
+    trimmed_buf.reserve(max_input_buffer_size_);
+    if (trimmed_buf.capacity() > max_input_buffer_size_) {
+      // Since we don't control how much space reserve() actually reserves,
+      // we have to go other way around and change the max size to avoid
+      // getting into the outer if() again.
+      max_input_buffer_size_ = trimmed_buf.capacity();
     }
+    trimmed_buf.assign(input_overflow_buf_.data(),
+                       input_overflow_buf_.size());
+    input_overflow_buf_.swap(trimmed_buf);
   }
 
   if (input_overflow_buf_.empty() && !DidEmptyInputBuffers())
diff --git a/ipc/ipc_channel_reader.h b/ipc/ipc_channel_reader.h
index 8e2fcba..352bfa5 100644
--- a/ipc/ipc_channel_reader.h
+++ b/ipc/ipc_channel_reader.h
@@ -129,6 +129,7 @@ class IPC_EXPORT ChannelReader : public SupportsAttachmentBrokering,
   FRIEND_TEST_ALL_PREFIXES(ChannelReaderTest, AttachmentNotYetBrokered);
   FRIEND_TEST_ALL_PREFIXES(ChannelReaderTest, ResizeOverflowBuffer);
   FRIEND_TEST_ALL_PREFIXES(ChannelReaderTest, InvalidMessageSize);
+  FRIEND_TEST_ALL_PREFIXES(ChannelReaderTest, TrimBuffer);
 
   using AttachmentIdSet = std::set<BrokerableAttachment::AttachmentId>;
   using AttachmentIdVector = std::vector<BrokerableAttachment::AttachmentId>;
@@ -195,6 +196,11 @@ class IPC_EXPORT ChannelReader : public SupportsAttachmentBrokering,
   // this buffer.
   std::string input_overflow_buf_;
 
+  // Maximum overflow buffer size, see Channel::kMaximumReadBufferSize.
+  // This is not a constant because we update it to reflect the reality
+  // of std::string::reserve() implementation.
+  size_t max_input_buffer_size_;
+
   // These messages are waiting to be dispatched. If this vector is non-empty,
   // then the front Message must be blocked on receiving an attachment from the
   // AttachmentBroker.
diff --git a/ipc/ipc_channel_reader_unittest.cc b/ipc/ipc_channel_reader_unittest.cc
index fdf3144..f49c275 100644
--- a/ipc/ipc_channel_reader_unittest.cc
+++ b/ipc/ipc_channel_reader_unittest.cc
@@ -13,6 +13,14 @@
 #include "ipc/placeholder_brokerable_attachment.h"
 #include "testing/gtest/include/gtest/gtest.h"
 
+// Whether IPC::Message::FindNext() can determine message size for
+// partial messages. The condition is from FindNext() implementation.
+#if USE_ATTACHMENT_BROKER && defined(OS_MACOSX) && !defined(OS_IOS)
+#define MESSAGE_FINDNEXT_PARTIAL 0
+#else
+#define MESSAGE_FINDNEXT_PARTIAL 1
+#endif
+
 namespace IPC {
 namespace internal {
 
@@ -64,7 +72,14 @@ class MockChannelReader : public ChannelReader {
       : ChannelReader(nullptr), last_dispatched_message_(nullptr) {}
 
   ReadState ReadData(char* buffer, int buffer_len, int* bytes_read) override {
-    return READ_FAILED;
+    if (data_.empty())
+      return READ_PENDING;
+
+    size_t read_len = std::min(static_cast<size_t>(buffer_len), data_.size());
+    memcpy(buffer, data_.data(), read_len);
+    *bytes_read = static_cast<int>(read_len);
+    data_.erase(0, read_len);
+    return READ_SUCCEEDED;
   }
 
   bool ShouldDispatchInputMessage(Message* msg) override { return true; }
@@ -92,9 +107,18 @@ class MockChannelReader : public ChannelReader {
 
   void set_broker(AttachmentBroker* broker) { broker_ = broker; }
 
+  void AppendData(const void* data, size_t size) {
+    data_.append(static_cast<const char*>(data), size);
+  }
+
+  void AppendMessageData(const Message& message) {
+    AppendData(message.data(), message.size());
+  }
+
  private:
   Message* last_dispatched_message_;
   AttachmentBroker* broker_;
+  std::string data_;
 };
 
 class ExposedMessage: public Message {
@@ -103,6 +127,9 @@ class ExposedMessage: public Message {
   using Message::header;
 };
 
+// Payload that makes messages large
+const size_t LargePayloadSize = Channel::kMaximumReadBufferSize * 3 / 2;
+
 }  // namespace
 
 #if USE_ATTACHMENT_BROKER
@@ -192,5 +219,155 @@ TEST(ChannelReaderTest, InvalidMessageSize) {
 
 #endif  // !USE_ATTACHMENT_BROKER
 
+TEST(ChannelReaderTest, TrimBuffer) {
+  // ChannelReader uses std::string as a buffer, and calls reserve()
+  // to trim it to kMaximumReadBufferSize. However, an implementation
+  // is free to actually reserve a larger amount.
+  size_t trimmed_buffer_size;
+  {
+    std::string buf;
+    buf.reserve(Channel::kMaximumReadBufferSize);
+    trimmed_buffer_size = buf.capacity();
+  }
+
+  // Buffer is trimmed after message is processed.
+  {
+    MockChannelReader reader;
+
+    Message message;
+    message.WriteString(std::string(LargePayloadSize, 'X'));
+
+    // Sanity check
+    EXPECT_TRUE(message.size() > trimmed_buffer_size);
+
+    // Initially buffer is small
+    EXPECT_LE(reader.input_overflow_buf_.capacity(), trimmed_buffer_size);
+
+    // Write and process large message
+    reader.AppendMessageData(message);
+    EXPECT_EQ(ChannelReader::DISPATCH_FINISHED,
+              reader.ProcessIncomingMessages());
+
+    // After processing large message buffer is trimmed
+    EXPECT_EQ(reader.input_overflow_buf_.capacity(), trimmed_buffer_size);
+  }
+
+  // Buffer is trimmed only after entire message is processed.
+  {
+    MockChannelReader reader;
+
+    ExposedMessage message;
+    message.WriteString(std::string(LargePayloadSize, 'X'));
+
+    // Write and process message header
+    reader.AppendData(message.header(), sizeof(ExposedMessage::Header));
+    EXPECT_EQ(ChannelReader::DISPATCH_FINISHED,
+              reader.ProcessIncomingMessages());
+
+#if MESSAGE_FINDNEXT_PARTIAL
+    // We determined message size for the message from its header, so
+    // we resized the buffer to fit.
+    EXPECT_GE(reader.input_overflow_buf_.capacity(), message.size());
+#else
+    // We couldn't determine message size, so we didn't resize the buffer.
+#endif
+
+    // Write and process payload
+    reader.AppendData(message.payload(), message.payload_size());
+    EXPECT_EQ(ChannelReader::DISPATCH_FINISHED,
+              reader.ProcessIncomingMessages());
+
+    // But once we process the message, we trim the buffer
+    EXPECT_EQ(reader.input_overflow_buf_.capacity(), trimmed_buffer_size);
+  }
+
+  // Buffer is not trimmed if the next message is also large.
+  {
+    MockChannelReader reader;
+
+    // Write large message
+    Message message1;
+    message1.WriteString(std::string(LargePayloadSize * 2, 'X'));
+    reader.AppendMessageData(message1);
+
+    // Write header for the next large message
+    ExposedMessage message2;
+    message2.WriteString(std::string(LargePayloadSize, 'Y'));
+    reader.AppendData(message2.header(), sizeof(ExposedMessage::Header));
+
+    // Process messages
+    EXPECT_EQ(ChannelReader::DISPATCH_FINISHED,
+              reader.ProcessIncomingMessages());
+
+#if MESSAGE_FINDNEXT_PARTIAL
+    // We determined message size for the second (partial) message, so
+    // we resized the buffer to fit.
+    EXPECT_GE(reader.input_overflow_buf_.capacity(), message1.size());
+#else
+    // We couldn't determine message size for the second (partial) message,
+    // so we trimmed the buffer.
+    EXPECT_EQ(reader.input_overflow_buf_.capacity(), trimmed_buffer_size);
+#endif
+  }
+
+  // Buffer resized appropriately if next message is larger than the first.
+  // (Similar to the test above except for the order of messages.)
+  {
+    MockChannelReader reader;
+
+    // Write large message
+    Message message1;
+    message1.WriteString(std::string(LargePayloadSize, 'Y'));
+    reader.AppendMessageData(message1);
+
+    // Write header for the next even larger message
+    ExposedMessage message2;
+    message2.WriteString(std::string(LargePayloadSize * 2, 'X'));
+    reader.AppendData(message2.header(), sizeof(ExposedMessage::Header));
+
+    // Process messages
+    EXPECT_EQ(ChannelReader::DISPATCH_FINISHED,
+              reader.ProcessIncomingMessages());
+
+#if MESSAGE_FINDNEXT_PARTIAL
+    // We determined message size for the second (partial) message, and
+    // resized the buffer to fit it.
+    EXPECT_GE(reader.input_overflow_buf_.capacity(), message2.size());
+#else
+    // We couldn't determine message size for the second (partial) message,
+    // so we trimmed the buffer.
+    EXPECT_EQ(reader.input_overflow_buf_.capacity(), trimmed_buffer_size);
+#endif
+  }
+
+  // Buffer is not trimmed if we've just resized it to accommodate large
+  // incoming message.
+  {
+    MockChannelReader reader;
+
+    // Write small message
+    Message message1;
+    message1.WriteString(std::string(11, 'X'));
+    reader.AppendMessageData(message1);
+
+    // Write header for the next large message
+    ExposedMessage message2;
+    message2.WriteString(std::string(LargePayloadSize, 'Y'));
+    reader.AppendData(message2.header(), sizeof(ExposedMessage::Header));
+
+    EXPECT_EQ(ChannelReader::DISPATCH_FINISHED,
+              reader.ProcessIncomingMessages());
+
+#if MESSAGE_FINDNEXT_PARTIAL
+    // We determined message size for the second (partial) message, so
+    // we resized the buffer to fit.
+    EXPECT_GE(reader.input_overflow_buf_.capacity(), message2.size());
+#else
+    // We couldn't determine size for the second (partial) message, and
+    // first message was small, so we did nothing.
+#endif
+  }
+}
+
 }  // namespace internal
 }  // namespace IPC