Bug 1235633 - IPC OOM mitigation by eliminating buffer copying (r=jld)

ipc/chromium/moz.build

@@ -9,6 +9,7 @@ include(libevent_path_prefix + '/libeventcommon.mozbuild')
 
 UNIFIED_SOURCES += [
     'src/base/at_exit.cc',
+    'src/base/buffer.cc',
     'src/base/command_line.cc',
     'src/base/file_path.cc',
     'src/base/file_util.cc',

ipc/chromium/src/base/buffer.cc

@@ -0,0 +1,125 @@
+/* -*- Mode: C++; tab-width: 8; indent-tabs-mode: nil; c-basic-offset: 2 -*- */
+/* vim: set ts=8 sts=2 et sw=2 tw=80: */
+/* This Source Code Form is subject to the terms of the Mozilla Public
+ * License, v. 2.0. If a copy of the MPL was not distributed with this
+ * file, You can obtain one at http://mozilla.org/MPL/2.0/. */
+
+#include "buffer.h"
+
+Buffer::Buffer()
+ : mBuffer(nullptr),
+   mSize(0),
+   mReserved(0)
+{
+}
+
+Buffer::~Buffer()
+{
+  if (mBuffer) {
+    free(mBuffer);
+  }
+}
+
+bool
+Buffer::empty() const
+{
+  return mSize == 0;
+}
+
+size_t
+Buffer::size() const
+{
+  return mSize;
+}
+
+const char*
+Buffer::data() const
+{
+  return mBuffer;
+}
+
+void
+Buffer::clear()
+{
+  free(mBuffer);
+  mBuffer = nullptr;
+  mSize = 0;
+  mReserved = 0;
+}
+
+void
+Buffer::try_realloc(size_t newlength)
+{
+  char* buffer = (char*)realloc(mBuffer, newlength);
+  if (buffer) {
+    mBuffer = buffer;
+    mReserved = newlength;
+    return;
+  }
+
+  // If we're growing the buffer, crash. If we're shrinking, then we continue to
+  // use the old (larger) buffer.
+  MOZ_RELEASE_ASSERT(newlength <= mReserved);
+}
+
+void
+Buffer::append(const char* bytes, size_t length)
+{
+  if (mSize + length > mReserved) {
+    try_realloc(mSize + length);
+  }
+
+  memcpy(mBuffer + mSize, bytes, length);
+  mSize += length;
+}
+
+void
+Buffer::assign(const char* bytes, size_t length)
+{
+  if (bytes >= mBuffer && bytes < mBuffer + mReserved) {
+    MOZ_RELEASE_ASSERT(bytes + length <= mBuffer + mSize);
+    memmove(mBuffer, bytes, length);
+    mSize = length;
+    try_realloc(length);
+  } else {
+    try_realloc(length);
+    mSize = length;
+    memcpy(mBuffer, bytes, length);
+  }
+}
+
+void
+Buffer::erase(size_t start, size_t count)
+{
+  mSize -= count;
+  memmove(mBuffer + start, mBuffer + start + count, mSize - start);
+  try_realloc(mSize);
+}
+
+void
+Buffer::reserve(size_t size)
+{
+  if (mReserved < size) {
+    try_realloc(size);
+  }
+}
+
+char*
+Buffer::trade_bytes(size_t count)
+{
+  char* result = mBuffer;
+  mSize = mReserved = mSize - count;
+  mBuffer = mReserved ? (char*)malloc(mReserved) : nullptr;
+  MOZ_RELEASE_ASSERT(!mReserved || mBuffer);
+  if (mSize) {
+    memcpy(mBuffer, result + count, mSize);
+  }
+
+  // Try to resize the buffer down, but ignore failure. This can cause extra
+  // copies, but so be it.
+  char* resized = (char*)realloc(result, count);
+  if (resized) {
+    return resized;
+  }
+  return result;
+}

ipc/chromium/src/base/buffer.h

@@ -0,0 +1,44 @@
+/* -*- Mode: C++; tab-width: 8; indent-tabs-mode: nil; c-basic-offset: 2 -*- */
+/* vim: set ts=8 sts=2 et sw=2 tw=80: */
+/* This Source Code Form is subject to the terms of the Mozilla Public
+ * License, v. 2.0. If a copy of the MPL was not distributed with this
+ * file, You can obtain one at http://mozilla.org/MPL/2.0/. */
+
+#ifndef CHROME_BASE_BUFFER_H_
+#define CHROME_BASE_BUFFER_H_
+
+// Buffer is a simple std::string-like class for buffering up IPC messages. Its
+// main distinguishing characteristic is the trade_bytes function.
+class Buffer {
+public:
+  Buffer();
+  ~Buffer();
+
+  bool empty() const;
+  const char* data() const;
+  size_t size() const;
+
+  void clear();
+  void append(const char* bytes, size_t length);
+  void assign(const char* bytes, size_t length);
+  void erase(size_t start, size_t count);
+
+  void reserve(size_t size);
+
+  // This function should be used by a caller who wants to extract the first
+  // |count| bytes from the buffer. Rather than copying the bytes out, this
+  // function returns the entire buffer. The bytes in range [count, size()) are
+  // copied out to a new buffer which becomes the current buffer. The
+  // presumption is that |count| is very large and approximately equal to size()
+  // so not much needs to be copied.
+  char* trade_bytes(size_t count);
+
+private:
+  void try_realloc(size_t newlength);
+
+  char* mBuffer;
+  size_t mSize;
+  size_t mReserved;
+};
+
+#endif // CHROME_BASE_BUFFER_H_

ipc/chromium/src/base/pickle.cc

@@ -122,10 +122,10 @@ Pickle::Pickle(int header_size)
   header_->payload_size = 0;
 }
 
-Pickle::Pickle(const char* data, int data_len)
+Pickle::Pickle(const char* data, int data_len, Ownership ownership)
     : header_(reinterpret_cast<Header*>(const_cast<char*>(data))),
       header_size_(0),
-      capacity_(kCapacityReadOnly),
+      capacity_(ownership == BORROWS ? kCapacityReadOnly : data_len),
       variable_buffer_offset_(0) {
   if (data_len >= static_cast<int>(sizeof(Header)))
     header_size_ = data_len - header_->payload_size;
@@ -639,3 +639,23 @@ const char* Pickle::FindNext(uint32_t header_size,
 
   return start + header_size + hdr->payload_size;
 }
+
+// static
+uint32_t Pickle::GetLength(uint32_t header_size,
+			   const char* start,
+			   const char* end) {
+  DCHECK(header_size == AlignInt(header_size));
+  DCHECK(header_size <= static_cast<memberAlignmentType>(kPayloadUnit));
+
+  if (end < start)
+    return 0;
+  size_t length = static_cast<size_t>(end - start);
+  if (length < sizeof(Header))
+    return 0;
+
+  const Header* hdr = reinterpret_cast<const Header*>(start);
+  if (length < header_size)
+    return 0;
+
+  return header_size + hdr->payload_size;
+}

ipc/chromium/src/base/pickle.h

@@ -32,6 +32,11 @@
 //
 class Pickle {
  public:
+  enum Ownership {
+    BORROWS,
+    OWNS,
+  };
+
   ~Pickle();
 
   // Initialize a Pickle object using the default header size.
@@ -42,11 +47,13 @@ class Pickle {
   // will be rounded up to ensure that the header size is 32bit-aligned.
   explicit Pickle(int header_size);
 
-  // Initializes a Pickle from a const block of data.  The data is not copied;
+  // Initializes a Pickle from a const block of data. If ownership == BORROWS,
-  // instead the data is merely referenced by this Pickle.  Only const methods
+  // the data is not copied; instead the data is merely referenced by this
-  // should be used on the Pickle when initialized this way.  The header
+  // Pickle. Only const methods should be used on the Pickle when initialized
-  // padding size is deduced from the data length.
+  // this way. The header padding size is deduced from the data length.  If
-  Pickle(const char* data, int data_len);
+  // ownership == OWNS, then again no copying takes place. However, the buffer
+  // is writable and will be freed when this Pickle is destroyed.
+  Pickle(const char* data, int data_len, Ownership ownership = BORROWS);
 
   // Initializes a Pickle as a deep copy of another Pickle.
   Pickle(const Pickle& other);
@@ -280,6 +287,12 @@ class Pickle {
                               const char* range_start,
                               const char* range_end);
 
+  // If the given range contains at least header_size bytes, return the length
+  // of the pickled data including the header.
+  static uint32_t GetLength(uint32_t header_size,
+                            const char* range_start,
+                            const char* range_end);
+
   // The allocation granularity of the payload.
   static const int kPayloadUnit;
 

ipc/chromium/src/chrome/common/child_process_host.cc

@@ -154,13 +154,13 @@ ChildProcessHost::ListenerHook::ListenerHook(ChildProcessHost* host)
 }
 
 void ChildProcessHost::ListenerHook::OnMessageReceived(
-    const IPC::Message& msg) {
+    IPC::Message&& msg) {
 
   bool msg_is_ok = true;
   bool handled = false;
 
   if (!handled) {
-      host_->OnMessageReceived(msg);
+      host_->OnMessageReceived(mozilla::Move(msg));
   }
 
   if (!msg_is_ok)

ipc/chromium/src/chrome/common/child_process_host.h

@@ -75,7 +75,7 @@ class ChildProcessHost :
   void InstanceCreated();
 
   // IPC::Channel::Listener implementation:
-  virtual void OnMessageReceived(const IPC::Message& msg) { }
+  virtual void OnMessageReceived(IPC::Message&& msg) { }
   virtual void OnChannelConnected(int32_t peer_pid) { }
   virtual void OnChannelError() { }
 
@@ -102,7 +102,7 @@ class ChildProcessHost :
   class ListenerHook : public IPC::Channel::Listener {
    public:
     explicit ListenerHook(ChildProcessHost* host);
-    virtual void OnMessageReceived(const IPC::Message& msg);
+    virtual void OnMessageReceived(IPC::Message&& msg);
     virtual void OnChannelConnected(int32_t peer_pid);
     virtual void OnChannelError();
     virtual void GetQueuedMessages(std::queue<IPC::Message>& queue);

ipc/chromium/src/chrome/common/child_thread.cc

@@ -62,7 +62,7 @@ bool ChildThread::Send(IPC::Message* msg) {
   return channel_->Send(msg);
 }
 
-void ChildThread::OnMessageReceived(const IPC::Message& msg) {
+void ChildThread::OnMessageReceived(IPC::Message&& msg) {
   if (msg.routing_id() == MSG_ROUTING_CONTROL) {
     OnControlMessageReceived(msg);
   }

ipc/chromium/src/chrome/common/child_thread.h

@@ -58,7 +58,7 @@ class ChildThread : public IPC::Channel::Listener,
 
  private:
   // IPC::Channel::Listener implementation:
-  virtual void OnMessageReceived(const IPC::Message& msg);
+  virtual void OnMessageReceived(IPC::Message&& msg);
   virtual void OnChannelError();
 
 #ifdef MOZ_NUWA_PROCESS

ipc/chromium/src/chrome/common/ipc_channel.h

@@ -25,7 +25,7 @@ class Channel : public Message::Sender {
     virtual ~Listener() {}
 
     // Called when a message is received.
-    virtual void OnMessageReceived(const Message& message) = 0;
+    virtual void OnMessageReceived(Message&& message) = 0;
 
     // Called when the channel is connected and we have received the internal
     // Hello message from the peer.
@@ -51,7 +51,10 @@ class Channel : public Message::Sender {
     kMaximumMessageSize = 256 * 1024 * 1024,
 
     // Ammount of data to read at once from the pipe.
-    kReadBufferSize = 4 * 1024
+    kReadBufferSize = 4 * 1024,
+
+    // Maximum size of a message that we allow to be copied (rather than moved).
+    kMaxCopySize = 32 * 1024,
   };
 
   // Initialize a Channel.

ipc/chromium/src/chrome/common/ipc_channel_posix.cc

@@ -263,18 +263,7 @@ bool Channel::ChannelImpl::EnqueueHelloMessage() {
 
 void Channel::ChannelImpl::ClearAndShrinkInputOverflowBuf()
 {
-  // If input_overflow_buf_ has grown, shrink it back to its normal size.
+  input_overflow_buf_.clear();
-  static size_t previousCapacityAfterClearing = 0;
-  if (input_overflow_buf_.capacity() > previousCapacityAfterClearing) {
-    // This swap trick is the closest thing C++ has to a guaranteed way
-    // to shrink the capacity of a string.
-    std::string tmp;
-    tmp.reserve(Channel::kReadBufferSize);
-    input_overflow_buf_.swap(tmp);
-    previousCapacityAfterClearing = input_overflow_buf_.capacity();
-  } else {
-    input_overflow_buf_.clear();
-  }
 }
 
 bool Channel::ChannelImpl::Connect() {
@@ -396,9 +385,23 @@ bool Channel::ChannelImpl::ProcessIncomingMessages() {
         CHROMIUM_LOG(ERROR) << "IPC message is too big";
         return false;
       }
+
       input_overflow_buf_.append(input_buf_, bytes_read);
       overflowp = p = input_overflow_buf_.data();
       end = p + input_overflow_buf_.size();
+
+      // If we've received the entire header, then we know the message
+      // length. In that case, reserve enough space to hold the entire
+      // message. This is more efficient than repeatedly enlarging the buffer as
+      // more data comes in.
+      uint32_t length = Message::GetLength(p, end);
+      if (length) {
+        input_overflow_buf_.reserve(length + kReadBufferSize);
+
+        // Recompute these pointers in case the buffer moved.
+        overflowp = p = input_overflow_buf_.data();
+        end = p + input_overflow_buf_.size();
+      }
     }
 
     // A pointer to an array of |num_fds| file descriptors which includes any
@@ -423,7 +426,31 @@ bool Channel::ChannelImpl::ProcessIncomingMessages() {
       const char* message_tail = Message::FindNext(p, end);
       if (message_tail) {
         int len = static_cast<int>(message_tail - p);
-        Message m(p, len);
+        char* buf;
+
+        // The Message |m| allocated below needs to own its data. We can either
+        // copy the data out of the buffer or else steal the buffer and move the
+        // remaining data elsewhere. If len is large enough, we steal. Otherwise
+        // we copy.
+        if (len > kMaxCopySize) {
+          // Since len > kMaxCopySize > kReadBufferSize, we know that we must be
+          // using the overflow buffer. And since we always shift everything to
+          // the left at the end of a read, we must be at the start of the
+          // overflow buffer.
+          MOZ_RELEASE_ASSERT(p == overflowp);
+          buf = input_overflow_buf_.trade_bytes(len);
+
+          // At this point the remaining data is at the front of
+          // input_overflow_buf_. p will get fixed up at the end of the
+          // loop. Set it to null here to make sure no one uses it.
+          p = nullptr;
+          overflowp = message_tail = input_overflow_buf_.data();
+          end = overflowp + input_overflow_buf_.size();
+        } else {
+          buf = (char*)malloc(len);
+          memcpy(buf, p, len);
+        }
+        Message m(buf, len, Message::OWNS);
         if (m.header()->num_fds) {
           // the message has file descriptors
           const char* error = NULL;
@@ -492,7 +519,7 @@ bool Channel::ChannelImpl::ProcessIncomingMessages() {
           CloseDescriptors(m.fd_cookie());
 #endif
         } else {
-          listener_->OnMessageReceived(m);
+          listener_->OnMessageReceived(mozilla::Move(m));
         }
         p = message_tail;
       } else {

ipc/chromium/src/chrome/common/ipc_channel_posix.h

@@ -14,6 +14,7 @@
 #include <vector>
 #include <list>
 
+#include "base/buffer.h"
 #include "base/message_loop.h"
 #include "chrome/common/file_descriptor_set_posix.h"
 
@@ -127,7 +128,7 @@ class Channel::ChannelImpl : public MessageLoopForIO::Watcher {
 
   // Large messages that span multiple pipe buffers, get built-up using
   // this buffer.
-  std::string input_overflow_buf_;
+  Buffer input_overflow_buf_;
   std::vector<int> input_overflow_fds_;
 
   // In server-mode, we have to wait for the client to connect before we

ipc/chromium/src/chrome/common/ipc_channel_win.cc

@@ -350,16 +350,53 @@ bool Channel::ChannelImpl::ProcessIncomingMessages(
         CHROMIUM_LOG(ERROR) << "IPC message is too big";
         return false;
       }
+
       input_overflow_buf_.append(input_buf_, bytes_read);
       p = input_overflow_buf_.data();
       end = p + input_overflow_buf_.size();
+
+      // If we've received the entire header, then we know the message
+      // length. In that case, reserve enough space to hold the entire
+      // message. This is more efficient than repeatedly enlarging the buffer as
+      // more data comes in.
+      uint32_t length = Message::GetLength(p, end);
+      if (length) {
+        input_overflow_buf_.reserve(length + kReadBufferSize);
+
+        // Recompute these pointers in case the buffer moved.
+        p = input_overflow_buf_.data();
+        end = p + input_overflow_buf_.size();
+      }
     }
 
     while (p < end) {
       const char* message_tail = Message::FindNext(p, end);
       if (message_tail) {
         int len = static_cast<int>(message_tail - p);
-        const Message m(p, len);
+        char* buf;
+
+        // The Message |m| allocated below needs to own its data. We can either
+        // copy the data out of the buffer or else steal the buffer and move the
+        // remaining data elsewhere. If len is large enough, we steal. Otherwise
+        // we copy.
+        if (len > kMaxCopySize) {
+          // Since len > kMaxCopySize > kReadBufferSize, we know that we must be
+          // using the overflow buffer. And since we always shift everything to
+          // the left at the end of a read, we must be at the start of the
+          // overflow buffer.
+          buf = input_overflow_buf_.trade_bytes(len);
+
+          // At this point the remaining data is at the front of
+          // input_overflow_buf_. p will get fixed up at the end of the
+          // loop. Set it to null here to make sure no one uses it.
+          p = nullptr;
+          message_tail = input_overflow_buf_.data();
+          end = message_tail + input_overflow_buf_.size();
+        } else {
+          buf = (char*)malloc(len);
+          memcpy(buf, p, len);
+        }
+        Message m(buf, len, Message::OWNS);
 #ifdef IPC_MESSAGE_DEBUG_EXTRA
         DLOG(INFO) << "received message on channel @" << this <<
                       " with type " << m.type();
@@ -380,7 +417,7 @@ bool Channel::ChannelImpl::ProcessIncomingMessages(
           waiting_for_shared_secret_ = false;
           listener_->OnChannelConnected(claimed_pid);
         } else {
-          listener_->OnMessageReceived(m);
+          listener_->OnMessageReceived(mozilla::Move(m));
         }
         p = message_tail;
       } else {

ipc/chromium/src/chrome/common/ipc_channel_win.h

@@ -10,6 +10,7 @@
 #include <queue>
 #include <string>
 
+#include "base/buffer.h"
 #include "base/message_loop.h"
 #include "mozilla/UniquePtr.h"
 
@@ -86,7 +87,7 @@ class Channel::ChannelImpl : public MessageLoopForIO::IOHandler {
 
   // Large messages that span multiple pipe buffers, get built-up using
   // this buffer.
-  std::string input_overflow_buf_;
+  Buffer input_overflow_buf_;
 
   // In server-mode, we have to wait for the client to connect before we
   // can begin reading.  We make use of the input_state_ when performing

ipc/chromium/src/chrome/common/ipc_message.cc

@@ -71,7 +71,9 @@ Message::Message(int32_t routing_id, msgid_t type, PriorityValue priority,
   InitLoggingVariables(aName);
 }
 
-Message::Message(const char* data, int data_len) : Pickle(data, data_len) {
+Message::Message(const char* data, int data_len, Ownership ownership)
+  : Pickle(data, data_len, ownership)
+{
   MOZ_COUNT_CTOR(IPC::Message);
   InitLoggingVariables();
 }

ipc/chromium/src/chrome/common/ipc_message.h

@@ -70,10 +70,12 @@ class Message : public Pickle {
           MessageCompression compression = COMPRESSION_NONE,
           const char* const name="???");
 
-  // Initializes a message from a const block of data.  The data is not copied;
+  // Initializes a message from a const block of data. If ownership == BORROWS,
-  // instead the data is merely referenced by this message.  Only const methods
+  // the data is not copied; instead the data is merely referenced by this
-  // should be used on the message when initialized this way.
+  // message. Only const methods should be used on the message when initialized
-  Message(const char* data, int data_len);
+  // this way. If ownership == OWNS, then again no copying takes place. However,
+  // the buffer is writable and will be freed when the message is destroyed.
+  Message(const char* data, int data_len, Ownership ownership = BORROWS);
 
   Message(const Message& other);
   Message(Message&& other);
@@ -242,6 +244,12 @@ class Message : public Pickle {
     return Pickle::FindNext(sizeof(Header), range_start, range_end);
   }
 
+  // If the given range contains at least header_size bytes, return the length
+  // of the message including the header.
+  static uint32_t GetLength(const char* range_start, const char* range_end) {
+    return Pickle::GetLength(sizeof(Header), range_start, range_end);
+  }
+
 #if defined(OS_POSIX)
   // On POSIX, a message supports reading / writing FileDescriptor objects.
   // This is used to pass a file descriptor to the peer of an IPC channel.

ipc/glue/GeckoChildProcessHost.cpp

@@ -1140,11 +1140,11 @@ GeckoChildProcessHost::OnChannelConnected(int32_t peer_pid)
 }
 
 void
-GeckoChildProcessHost::OnMessageReceived(const IPC::Message& aMsg)
+GeckoChildProcessHost::OnMessageReceived(IPC::Message&& aMsg)
 {
   // We never process messages ourself, just save them up for the next
   // listener.
-  mQueue.push(aMsg);
+  mQueue.push(Move(aMsg));
 }
 
 void

ipc/glue/GeckoChildProcessHost.h

@@ -82,7 +82,7 @@ public:
                                   base::ProcessArchitecture aArch=base::GetCurrentProcessArchitecture());
 
   virtual void OnChannelConnected(int32_t peer_pid);
-  virtual void OnMessageReceived(const IPC::Message& aMsg);
+  virtual void OnMessageReceived(IPC::Message&& aMsg);
   virtual void OnChannelError();
   virtual void GetQueuedMessages(std::queue<IPC::Message>& queue);
 

ipc/glue/MessageChannel.cpp

@@ -875,7 +875,7 @@ public:
 };
 
 void
-MessageChannel::OnMessageReceivedFromLink(const Message& aMsg)
+MessageChannel::OnMessageReceivedFromLink(Message&& aMsg)
 {
     AssertLinkThread();
     mMonitor->AssertCurrentThreadOwns();
@@ -972,7 +972,7 @@ MessageChannel::OnMessageReceivedFromLink(const Message& aMsg)
     // blocked. This is okay, since we always check for pending events before
     // blocking again.
 
-    mPending.push_back(aMsg);
+    mPending.push_back(Move(aMsg));
 
     if (shouldWakeUp) {
         NotifyWorkerThread();

ipc/glue/MessageChannel.h

@@ -431,7 +431,7 @@ class MessageChannel : HasResultCodes
 
     bool WasTransactionCanceled(int transaction);
     bool ShouldDeferMessage(const Message& aMsg);
-    void OnMessageReceivedFromLink(const Message& aMsg);
+    void OnMessageReceivedFromLink(Message&& aMsg);
     void OnChannelErrorFromLink();
 
   private:

ipc/glue/MessageLink.cpp

@@ -271,7 +271,7 @@ ThreadLink::EchoMessage(Message *msg)
     mChan->AssertWorkerThread();
     mChan->mMonitor->AssertCurrentThreadOwns();
 
-    mChan->OnMessageReceivedFromLink(*msg);
+    mChan->OnMessageReceivedFromLink(Move(*msg));
     delete msg;
 }
 
@@ -282,7 +282,7 @@ ThreadLink::SendMessage(Message *msg)
     mChan->mMonitor->AssertCurrentThreadOwns();
 
     if (mTargetChan)
-        mTargetChan->OnMessageReceivedFromLink(*msg);
+        mTargetChan->OnMessageReceivedFromLink(Move(*msg));
     delete msg;
 }
 
@@ -322,19 +322,19 @@ ThreadLink::Unsound_NumQueuedMessages() const
 //
 
 void
-ProcessLink::OnMessageReceived(const Message& msg)
+ProcessLink::OnMessageReceived(Message&& msg)
 {
     AssertIOThread();
     NS_ASSERTION(mChan->mChannelState != ChannelError, "Shouldn't get here!");
     MonitorAutoLock lock(*mChan->mMonitor);
-    mChan->OnMessageReceivedFromLink(msg);
+    mChan->OnMessageReceivedFromLink(Move(msg));
 }
 
 void
 ProcessLink::OnEchoMessage(Message* msg)
 {
     AssertIOThread();
-    OnMessageReceived(*msg);
+    OnMessageReceived(Move(*msg));
     delete msg;
 }
 
@@ -381,7 +381,7 @@ ProcessLink::OnTakeConnectedChannel()
 
     // Dispatch whatever messages the previous listener had queued up.
     while (!pending.empty()) {
-        OnMessageReceived(pending.front());
+        OnMessageReceived(Move(pending.front()));
         pending.pop();
     }
 }

ipc/glue/MessageLink.h

@@ -204,7 +204,7 @@ class ProcessLink
     // These methods acquire the monitor and forward to the
     // similarly named methods in AsyncChannel below
     // (OnMessageReceivedFromLink(), etc)
-    virtual void OnMessageReceived(const Message& msg) override;
+    virtual void OnMessageReceived(Message&& msg) override;
     virtual void OnChannelConnected(int32_t peer_pid) override;
     virtual void OnChannelError() override;