Add out-of-process memory logging stream parsing.

chrome/profiling/memlog_stream_parser.cc

+// 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>;
+
+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) {
+  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],
+           to_copy);
+
+    used += to_copy;
+
+    current_block++;
+    current_block_offset = 0;
+  }
+  return used == count;
+}
+
+bool MemlogStreamParser::ReadBytes(size_t count, void* dest) {
+  if (!PeekBytes(count, dest))
+    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());
+    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;
+  if (header.version != kStreamVersion)
+    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));
+  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