Add out-of-process memory logging stream parsing.

chrome/profiling/memlog_stream_parser.cc

Index: chrome/profiling/memlog_stream_parser.cc
diff --git a/chrome/profiling/memlog_stream_parser.cc b/chrome/profiling/memlog_stream_parser.cc
new file mode 100644
index 0000000000000000000000000000000000000000..2d134db6270c6b0a0fe77aeeaabef44f199eb90d
--- /dev/null
+++ b/chrome/profiling/memlog_stream_parser.cc
@@ -0,0 +1,174 @@
+// Copyright 2017 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include "chrome/profiling/memlog_stream_parser.h"
+
+#include <algorithm>
+
+#include "base/containers/stack_container.h"
+#include "base/strings/stringprintf.h"
+#include "chrome/common/profiling/memlog_stream.h"
+#include "chrome/profiling/address.h"
+#include "chrome/profiling/profiling_globals.h"
+#include "chrome/profiling/stack.h"
+
+namespace profiling {
+
+using AddressVector = base::StackVector<Address, 128>;

[awong, 2017/06/19 21:51:27]

Random C++11 question... does this need to be in an anonymous namespace?  What
happens if another translation unit declares a profiling::AddressVector?

(I don't know the answer to this.)

[brettw, 2017/06/19 23:29:45]

I think it will be OK, but the anon namespace seems like a good idea anyway.

+
+struct MemlogStreamParser::Block {
+  Block(std::unique_ptr<char[]> d, size_t s) : data(std::move(d)), size(s) {}
+
+  std::unique_ptr<char[]> data;
+  size_t size;
+};
+
+MemlogStreamParser::MemlogStreamParser(MemlogReceiver* receiver)
+    : receiver_(receiver) {}
+
+MemlogStreamParser::~MemlogStreamParser() {}
+
+void MemlogStreamParser::OnStreamData(std::unique_ptr<char[]> data, size_t sz) {

[Boris Vidolov, 2017/06/17 00:53:07]

Should this be std::unique_ptr<char[]>&& data ?

[brettw, 2017/06/17 02:41:51]

I think these are usually passed by value also (git grep "(std::unique_ptr").
I've had people correct me in the opposite direction for this kind of thing
also, since the theory is that then the caller can control whether to give you a
copy or a move.

In this case it doesn't matter since a unique_ptr is just a wrapper around a
pointer the same thing should happen in both cases.

[awong, 2017/06/19 21:51:27]

Definitely should not be &&.  If the API is denoting ownership transfer, pass by
value is the right semantic for the RAII move-only types.

Any use of && is suspect to me if it isn't specifically an overload against
another function.

+  blocks_.emplace_back(std::move(data), sz);
+
+  if (!received_header_) {
+    received_header_ = true;
+    ReadStatus status = ParseHeader();
+    if (status != READ_OK)
+      return;  // TODO(brettw) signal error.
+  }
+
+  while (true) {
+    uint32_t msg_type;
+    if (!PeekBytes(sizeof(msg_type), &msg_type))
+      return;
+
+    ReadStatus status;
+    switch (msg_type) {
+      case kAllocPacketType:
+        status = ParseAlloc();
+        break;
+      case kFreePacketType:
+        status = ParseFree();
+        break;
+      default:
+        return;  // TODO(brettw) signal error.
+    }
+    if (status != READ_OK)
+      return;  // TODO(brettw) signal error.
+  }
+}
+
+void MemlogStreamParser::OnStreamComplete() {
+  receiver_->OnComplete();
+}
+
+bool MemlogStreamParser::AreBytesAvailable(size_t count) const {
+  size_t used = 0;
+
+  size_t current_block = 0;
+  size_t current_block_offset = block_zero_offset_;
+  while (current_block < blocks_.size() && used < count) {
+    used += blocks_[current_block].size - current_block_offset;
+    current_block++;
+    current_block_offset = 0;
+  }
+  return used >= count;
+}
+
+bool MemlogStreamParser::PeekBytes(size_t count, void* dest) const {
+  if (!AreBytesAvailable(count))
+    return false;
+
+  char* dest_char = static_cast<char*>(dest);
+  size_t used = 0;
+
+  size_t current_block = 0;
+  size_t current_block_offset = block_zero_offset_;
+  while (current_block < blocks_.size()) {
+    size_t in_current_block =
+        blocks_[current_block].size - current_block_offset;
+
+    size_t to_copy = std::min(count - used, in_current_block);
+    memcpy(&dest_char[used], &blocks_[current_block].data[current_block_offset],

[awong, 2017/06/19 21:51:27]

std::copy?

https://stackoverflow.com/questions/4707012/is-it-better-to-use-stdmemcpy-or-...

[brettw, 2017/06/19 23:29:46]

I didn't want to do an element-by-element copy here. std::copy may do this. I
looked at the GCC and MS implementations. I couldn't understand the GCC one and
I couldn't find the MS one. I prefer this simpler call which is definitely what
I want.

+           to_copy);
+
+    used += to_copy;
+
+    current_block++;
+    current_block_offset = 0;
+  }
+  return used == count;

[awong, 2017/06/19 21:51:27]

Can this ever be false if we didn't return in line 82?

[brettw, 2017/06/19 23:29:45]

It can be now since I removed the AreBytesAvailable call.

+}
+
+bool MemlogStreamParser::ReadBytes(size_t count, void* dest) {
+  if (!PeekBytes(count, dest))

[Boris Vidolov, 2017/06/17 03:13:14]

This doesn't sound to be very efficient:
ReadBytes calls PeekBytes, which calls AreBytesAvailable. Then ReadBytes calls
ConsumeBytes. All of PeekBytes, AreBytesAvailable and ConsumeBytes have similar
implementation walking the same blocks, tripling the work.
Maybe create one method that walks up to  'count' bytes and passes back an
iterator to the  last block? Then all three methods can call this one and decide
what to do based on the result.

[brettw, 2017/06/19 23:29:45]

I don't think returning an iterator would be very helpful since the log is in
going through the blocks, not detecting the end.

I removed the Peek call here and just said in the header that it may touch the
buffer on failure. In general, messages will almost never cross block
boundaries. Read could be optimized more by duplicating some code but that
didn't seem worth it to me.

+    return false;
+  ConsumeBytes(count);
+  return true;
+}
+
+void MemlogStreamParser::ConsumeBytes(size_t count) {
+  DCHECK(AreBytesAvailable(count));
+  while (count > 0) {
+    size_t bytes_left_in_block = blocks_[0].size - block_zero_offset_;
+    if (bytes_left_in_block > count) {
+      // Still data left in this block;
+      block_zero_offset_ += count;
+      return;
+    }
+
+    // Current block is consumed.
+    blocks_.erase(blocks_.begin());

[Boris Vidolov, 2017/06/17 03:13:15]

Can we erase a range instead of single block? Erasing one block at a time will
have quadratic complexity, as it always moves all of the following elements in
the array.

Overall, how many blocks do we expect to have? I am concerned that this code
would typically consume one block or less at a time, causing the vector to move
elements heavily if many blocks are available.

[brettw, 2017/06/19 23:29:46]

I would hope the common case is 0-2 blocks (they're 64K each so have a lot of
messages in each one). I believe this code should generally be able to keep up
with the stream sent by the instrumented process. I did consider a list or
something here but it seemed moving pointers was much more efficient than
mallocing for the expected workload.

A queue would probably be better overall so I used that.

+    block_zero_offset_ = 0;
+    count -= bytes_left_in_block;
+  }
+}
+
+MemlogStreamParser::ReadStatus MemlogStreamParser::ParseHeader() {
+  StreamHeader header;
+  if (!ReadBytes(sizeof(StreamHeader), &header))
+    return READ_NO_DATA;
+
+  if (header.signature != kStreamSignature)
+    return READ_ERROR;
+
+  receiver_->OnHeader(header);
+  return READ_OK;
+}
+
+MemlogStreamParser::ReadStatus MemlogStreamParser::ParseAlloc() {
+  // Read the packet. Can't commit the read until the stack is read and
+  // that has to be done below.
+  AllocPacket alloc_packet;
+  if (!PeekBytes(sizeof(AllocPacket), &alloc_packet))
+    return READ_NO_DATA;
+
+  std::vector<Address> stack_addrs;
+  stack_addrs.resize(alloc_packet.stack_len);
+  size_t stack_byte_size = sizeof(Address) * alloc_packet.stack_len;
+
+  if (!AreBytesAvailable(sizeof(AllocPacket) + stack_byte_size))
+    return READ_NO_DATA;
+
+  // Everything will fit, mark packet consumed, read stack.
+  ConsumeBytes(sizeof(AllocPacket));
+  if (!stack_addrs.empty())
+    ReadBytes(stack_byte_size, stack_addrs.data());
+
+  Stack stack(std::move(stack_addrs));

[Boris Vidolov, 2017/06/17 03:13:14]

Do we need to go through an intermediary 'Stack' variable? Can't we just pass
stack_addrs below?

[brettw, 2017/06/19 23:29:46]

The stack object exists to store the hash so we don't have to re-hash for
different operations. Since this exists, it seems nice to use it consistently.

+  receiver_->OnAlloc(alloc_packet, std::move(stack));
+  return READ_OK;
+}
+
+MemlogStreamParser::ReadStatus MemlogStreamParser::ParseFree() {
+  FreePacket free_packet;
+  if (!ReadBytes(sizeof(FreePacket), &free_packet))
+    return READ_NO_DATA;
+
+  receiver_->OnFree(free_packet);
+  return READ_OK;
+}
+
+}  // namespace profiling