diff --git a/Android.bp b/Android.bp
index 1379d31..c5411f1 100644
--- a/Android.bp
+++ b/Android.bp
@@ -169,6 +169,24 @@
   ],
 }
 
+// GN target: //protos:zero_gen
+genrule {
+  name: "perfetto_protos_zero_gen",
+  srcs: [
+    "protos/test_event.proto",
+    "protos/trace_packet.proto",
+  ],
+  tools: [
+    "aprotoc",
+    "perfetto_src_protozero_protoc_plugin_protoc_plugin___gn_standalone_toolchain_gcc_like_host_",
+  ],
+  cmd: "mkdir -p $(genDir)/external/perfetto && $(location aprotoc) --cpp_out=$(genDir)/external/perfetto --proto_path=external/perfetto --plugin=protoc-gen-plugin=$(location perfetto_src_protozero_protoc_plugin_protoc_plugin___gn_standalone_toolchain_gcc_like_host_) --plugin_out=wrapper_namespace=pbzero:$(genDir)/external/perfetto $(in)",
+  out: [
+    "external/perfetto/protos/test_event.pbzero.cc",
+    "external/perfetto/protos/trace_packet.pbzero.cc",
+  ],
+}
+
 // GN target: //src/ipc:test_messages_gen
 genrule {
   name: "perfetto_src_ipc_test_messages_gen",
@@ -196,6 +214,7 @@
 genrule {
   name: "perfetto_protos_lite_gen_headers",
   srcs: [
+    "protos/test_event.proto",
     "protos/trace_packet.proto",
   ],
   tools: [
@@ -203,6 +222,7 @@
   ],
   cmd: "mkdir -p $(genDir)/external/perfetto && $(location aprotoc) --cpp_out=$(genDir)/external/perfetto --proto_path=external/perfetto $(in)",
   out: [
+    "external/perfetto/protos/test_event.pb.h",
     "external/perfetto/protos/trace_packet.pb.h",
   ],
   export_include_dirs: [
@@ -225,6 +245,44 @@
   ],
 }
 
+// GN target: //protos:zero_gen
+genrule {
+  name: "perfetto_protos_zero_gen_headers",
+  srcs: [
+    "protos/test_event.proto",
+    "protos/trace_packet.proto",
+  ],
+  tools: [
+    "aprotoc",
+    "perfetto_src_protozero_protoc_plugin_protoc_plugin___gn_standalone_toolchain_gcc_like_host_",
+  ],
+  cmd: "mkdir -p $(genDir)/external/perfetto && $(location aprotoc) --cpp_out=$(genDir)/external/perfetto --proto_path=external/perfetto --plugin=protoc-gen-plugin=$(location perfetto_src_protozero_protoc_plugin_protoc_plugin___gn_standalone_toolchain_gcc_like_host_) --plugin_out=wrapper_namespace=pbzero:$(genDir)/external/perfetto $(in)",
+  out: [
+    "external/perfetto/protos/test_event.pbzero.h",
+    "external/perfetto/protos/trace_packet.pbzero.h",
+  ],
+  export_include_dirs: [
+    ".",
+  ],
+}
+
+// GN target: //protos:lite_gen
+genrule {
+  name: "perfetto_protos_lite_gen",
+  srcs: [
+    "protos/test_event.proto",
+    "protos/trace_packet.proto",
+  ],
+  tools: [
+    "aprotoc",
+  ],
+  cmd: "mkdir -p $(genDir)/external/perfetto && $(location aprotoc) --cpp_out=$(genDir)/external/perfetto --proto_path=external/perfetto $(in)",
+  out: [
+    "external/perfetto/protos/test_event.pb.cc",
+    "external/perfetto/protos/trace_packet.pb.cc",
+  ],
+}
+
 // GN target: //src/ftrace_reader:ftrace_reader_test_messages_zero_gen
 genrule {
   name: "perfetto_src_ftrace_reader_ftrace_reader_test_messages_zero_gen_headers",
@@ -244,21 +302,6 @@
   ],
 }
 
-// GN target: //protos:lite_gen
-genrule {
-  name: "perfetto_protos_lite_gen",
-  srcs: [
-    "protos/trace_packet.proto",
-  ],
-  tools: [
-    "aprotoc",
-  ],
-  cmd: "mkdir -p $(genDir)/external/perfetto && $(location aprotoc) --cpp_out=$(genDir)/external/perfetto --proto_path=external/perfetto $(in)",
-  out: [
-    "external/perfetto/protos/trace_packet.pb.cc",
-  ],
-}
-
 // GN target: //protos/ftrace:zero_gen
 genrule {
   name: "perfetto_protos_ftrace_zero_gen",
@@ -328,6 +371,7 @@
     ":perfetto_protos_lite_gen",
     ":perfetto_protos_tracing_service_lite_gen",
     ":perfetto_protos_tracing_service_tracing_service_gen",
+    ":perfetto_protos_zero_gen",
     ":perfetto_src_ftrace_reader_ftrace_reader_test_messages_lite_gen",
     ":perfetto_src_ftrace_reader_ftrace_reader_test_messages_zero_gen",
     ":perfetto_src_ipc_test_messages_gen",
@@ -388,15 +432,20 @@
     "src/tracing/core/service_impl_unittest.cc",
     "src/tracing/core/shared_memory_abi.cc",
     "src/tracing/core/shared_memory_abi_unittest.cc",
+    "src/tracing/core/shared_memory_arbiter.cc",
+    "src/tracing/core/shared_memory_arbiter_unittest.cc",
     "src/tracing/core/trace_config.cc",
     "src/tracing/core/trace_packet.cc",
     "src/tracing/core/trace_packet_unittest.cc",
+    "src/tracing/core/trace_writer_impl.cc",
+    "src/tracing/core/trace_writer_impl_unittest.cc",
     "src/tracing/ipc/consumer/consumer_ipc_client_impl.cc",
     "src/tracing/ipc/posix_shared_memory.cc",
     "src/tracing/ipc/posix_shared_memory_unittest.cc",
     "src/tracing/ipc/producer/producer_ipc_client_impl.cc",
     "src/tracing/ipc/service/producer_ipc_service.cc",
     "src/tracing/ipc/service/service_ipc_host_impl.cc",
+    "src/tracing/test/aligned_buffer_test.cc",
     "src/tracing/test/test_shared_memory.cc",
     "tools/sanitizers_unittests/sanitizers_unittest.cc",
   ],
@@ -414,6 +463,7 @@
     "perfetto_protos_lite_gen_headers",
     "perfetto_protos_tracing_service_lite_gen_headers",
     "perfetto_protos_tracing_service_tracing_service_gen_headers",
+    "perfetto_protos_zero_gen_headers",
     "perfetto_src_ftrace_reader_ftrace_reader_test_messages_lite_gen_headers",
     "perfetto_src_ftrace_reader_ftrace_reader_test_messages_zero_gen_headers",
     "perfetto_src_ipc_test_messages_gen_headers",
diff --git a/include/perfetto/tracing/core/BUILD.gn b/include/perfetto/tracing/core/BUILD.gn
index 8d83c55..1f895ac 100644
--- a/include/perfetto/tracing/core/BUILD.gn
+++ b/include/perfetto/tracing/core/BUILD.gn
@@ -28,5 +28,6 @@
     "shared_memory_abi.h",
     "trace_config.h",
     "trace_packet.h",
+    "trace_writer.h",
   ]
 }
diff --git a/include/perfetto/tracing/core/trace_writer.h b/include/perfetto/tracing/core/trace_writer.h
new file mode 100644
index 0000000..8b64b3e
--- /dev/null
+++ b/include/perfetto/tracing/core/trace_writer.h
@@ -0,0 +1,67 @@
+/*
+ * Copyright (C) 2017 The Android Open Source Project
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#ifndef INCLUDE_PERFETTO_TRACING_CORE_TRACE_WRITER_H_
+#define INCLUDE_PERFETTO_TRACING_CORE_TRACE_WRITER_H_
+
+#include "perfetto/protozero/protozero_message_handle.h"
+
+namespace perfetto {
+
+namespace protos {
+namespace pbzero {
+class TracePacket;
+}  // namespace pbzero
+}  // namespace protos
+
+// This is a single-thread write interface that allows to write protobufs
+// directly into the tracing shared buffer without making any copies.
+// It takes care of acquiring and releasing chunks from the
+// SharedMemoryArbiter and splitting protos over chunks.
+// The idea is that each data source creates one (or more) TraceWriter for each
+// thread it wants to write from. Each TraceWriter will get its own dedicated
+// chunk and will write into the shared buffer without any locking most of the
+// time. Locking will happen only when a chunk is exhausted and a new one is
+// acquired from the arbiter.
+
+// TODO: TraceWriter needs to keep the shared memory buffer alive (refcount?).
+// Otherwise if the shared memory buffer goes away (e.g. the Service crashes)
+// the TraceWriter will keep writing into unmapped memory.
+
+class TraceWriter {
+ public:
+  using TracePacketHandle =
+      protozero::ProtoZeroMessageHandle<protos::pbzero::TracePacket>;
+
+  TraceWriter();
+  virtual ~TraceWriter();
+
+  // Returns a handle to the root proto message for the trace. The message will
+  // be finalized either by calling directly handle.Finalize() or by letting the
+  // handle go out of scope. The returned handle can be std::move()'d but cannot
+  // be used after either: (i) the TraceWriter instance is destroyed, (ii) a
+  // subsequence NewTracePacket() call is made on the same TraceWriter instance.
+  virtual TracePacketHandle NewTracePacket() = 0;
+  virtual WriterID writer_id() const = 0;
+
+ private:
+  TraceWriter(const TraceWriter&) = delete;
+  TraceWriter& operator=(const TraceWriter&) = delete;
+};
+
+}  // namespace perfetto
+
+#endif  // INCLUDE_PERFETTO_TRACING_CORE_TRACE_WRITER_H_
diff --git a/protos/BUILD.gn b/protos/BUILD.gn
index 077d57d..3ba9e45 100644
--- a/protos/BUILD.gn
+++ b/protos/BUILD.gn
@@ -16,7 +16,10 @@
 import("../gn/perfetto.gni")
 import("../gn/proto_library.gni")
 
-proto_sources = [ "trace_packet.proto" ]
+proto_sources = [
+  "test_event.proto",
+  "trace_packet.proto",
+]
 
 # Protozero generated stubs, for writers.
 protozero_library("zero") {
diff --git a/protos/test_event.proto b/protos/test_event.proto
new file mode 100644
index 0000000..bac6322
--- /dev/null
+++ b/protos/test_event.proto
@@ -0,0 +1,22 @@
+/*
+ * Copyright (C) 2017 The Android Open Source Project
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+syntax = "proto2";
+option optimize_for = LITE_RUNTIME;
+
+package perfetto.protos;
+
+message TestEvent {}
diff --git a/protos/trace_packet.proto b/protos/trace_packet.proto
index 5db2b1b..e59e991 100644
--- a/protos/trace_packet.proto
+++ b/protos/trace_packet.proto
@@ -18,12 +18,16 @@
 option optimize_for = LITE_RUNTIME;
 
 import "protos/ftrace/ftrace_event_bundle.proto";
+import "protos/test_event.proto";
 
 package perfetto.protos;
 
 // The root object emitted by Perfetto. A perfetto trace is just a stream of
 // TracePacket(s).
 message TracePacket {
-  oneof data { FtraceEventBundle ftrace_events = 1; }
+  oneof data {
+    FtraceEventBundle ftrace_events = 1;
+    TestEvent test_event = 536870911;  // 2^29 - 1, max field id for protos.
+  }
   optional string test = 2;
 }
diff --git a/src/tracing/BUILD.gn b/src/tracing/BUILD.gn
index 33697e5..bbfa964 100644
--- a/src/tracing/BUILD.gn
+++ b/src/tracing/BUILD.gn
@@ -23,6 +23,7 @@
   deps = [
     "../../gn:default_deps",
     "../../protos:lite",
+    "../../protos:zero",
     "../base",
   ]
   sources = [
@@ -35,8 +36,12 @@
     "core/service_impl.cc",
     "core/service_impl.h",
     "core/shared_memory_abi.cc",
+    "core/shared_memory_arbiter.cc",
+    "core/shared_memory_arbiter.h",
     "core/trace_config.cc",
     "core/trace_packet.cc",
+    "core/trace_writer_impl.cc",
+    "core/trace_writer_impl.h",
   ]
 }
 
@@ -76,6 +81,7 @@
     "../../gn:default_deps",
     "../../gn:gtest_deps",
     "../../protos:lite",
+    "../../protos:zero",
     "../base",
     "../base:test_support",
   ]
@@ -84,8 +90,12 @@
     "core/id_allocator_unittest.cc",
     "core/service_impl_unittest.cc",
     "core/shared_memory_abi_unittest.cc",
+    "core/shared_memory_arbiter_unittest.cc",
     "core/trace_packet_unittest.cc",
+    "core/trace_writer_impl_unittest.cc",
     "ipc/posix_shared_memory_unittest.cc",
+    "test/aligned_buffer_test.cc",
+    "test/aligned_buffer_test.h",
     "test/test_shared_memory.cc",
     "test/test_shared_memory.h",
   ]
diff --git a/src/tracing/core/shared_memory_abi_unittest.cc b/src/tracing/core/shared_memory_abi_unittest.cc
index d460759..dd75534 100644
--- a/src/tracing/core/shared_memory_abi_unittest.cc
+++ b/src/tracing/core/shared_memory_abi_unittest.cc
@@ -17,7 +17,7 @@
 #include "perfetto/tracing/core/shared_memory_abi.h"
 
 #include "gtest/gtest.h"
-#include "perfetto/base/utils.h"
+#include "src/tracing/test/aligned_buffer_test.h"
 
 namespace perfetto {
 namespace {
@@ -26,35 +26,18 @@
 using Chunk = SharedMemoryABI::Chunk;
 using ChunkHeader = SharedMemoryABI::ChunkHeader;
 
-class SharedMemoryABITest : public ::testing::TestWithParam<size_t> {
- public:
-  void SetUp() override {
-    page_size_ = GetParam();
-    buf_size_ = page_size_ * kNumPages;
-    void* mem = nullptr;
-    PERFETTO_CHECK(posix_memalign(&mem, page_size_, buf_size_) == 0);
-    buf_.reset(reinterpret_cast<uint8_t*>(mem));
-    memset(buf_.get(), 0, buf_size_);
-  }
-
-  void TearDown() override { buf_.reset(); }
-
-  const size_t kNumPages = 10;
-  std::unique_ptr<uint8_t, base::FreeDeleter> buf_;
-  size_t buf_size_;
-  size_t page_size_;
-};
+using SharedMemoryABITest = AlignedBufferTest;
 
 size_t const kPageSizes[] = {4096, 8192, 16384, 32768, 65536};
 INSTANTIATE_TEST_CASE_P(PageSize, SharedMemoryABITest, ValuesIn(kPageSizes));
 
 TEST_P(SharedMemoryABITest, NominalCases) {
-  SharedMemoryABI abi(buf_.get(), buf_size_, page_size_);
+  SharedMemoryABI abi(buf(), buf_size(), page_size());
 
-  ASSERT_EQ(buf_.get(), abi.start());
-  ASSERT_EQ(buf_.get() + buf_size_, abi.end());
-  ASSERT_EQ(buf_size_, abi.size());
-  ASSERT_EQ(page_size_, abi.page_size());
+  ASSERT_EQ(buf(), abi.start());
+  ASSERT_EQ(buf() + buf_size(), abi.end());
+  ASSERT_EQ(buf_size(), abi.size());
+  ASSERT_EQ(page_size(), abi.page_size());
   ASSERT_EQ(kNumPages, abi.num_pages());
 
   for (size_t i = 0; i < kNumPages; i++) {
@@ -94,8 +77,8 @@
   uint8_t* last_chunk_end = nullptr;
 
   for (size_t page_idx = 0; page_idx <= 4; page_idx++) {
-    uint8_t* const page_start = buf_.get() + page_idx * page_size_;
-    uint8_t* const page_end = page_start + page_size_;
+    uint8_t* const page_start = buf() + page_idx * page_size();
+    uint8_t* const page_end = page_start + page_size();
     const size_t num_chunks =
         SharedMemoryABI::GetNumChunksForLayout(abi.page_layout_dbg(page_idx));
     const size_t target_buffer = 10 + page_idx;
@@ -132,7 +115,7 @@
 
       // Sanity check chunk bounds.
       size_t expected_chunk_size =
-          (page_size_ - sizeof(SharedMemoryABI::PageHeader)) / num_chunks;
+          (page_size() - sizeof(SharedMemoryABI::PageHeader)) / num_chunks;
       expected_chunk_size = expected_chunk_size - (expected_chunk_size % 4);
       ASSERT_EQ(expected_chunk_size, chunk.size());
       ASSERT_GT(chunk.begin(), page_start);
@@ -224,7 +207,7 @@
 }
 
 TEST_P(SharedMemoryABITest, BatchAcquireAndRelease) {
-  SharedMemoryABI abi(buf_.get(), buf_size_, page_size_);
+  SharedMemoryABI abi(buf(), buf_size(), page_size());
   ChunkHeader header{};
 
   // TryAcquire on a non-partitioned page should fail.
diff --git a/src/tracing/core/shared_memory_arbiter.cc b/src/tracing/core/shared_memory_arbiter.cc
new file mode 100644
index 0000000..e2a059d
--- /dev/null
+++ b/src/tracing/core/shared_memory_arbiter.cc
@@ -0,0 +1,164 @@
+/*
+ * Copyright (C) 2017 The Android Open Source Project
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include "src/tracing/core/shared_memory_arbiter.h"
+
+#include "perfetto/base/logging.h"
+#include "perfetto/base/task_runner.h"
+#include "src/tracing/core/trace_writer_impl.h"
+
+#include <limits>
+
+namespace perfetto {
+
+using Chunk = SharedMemoryABI::Chunk;
+
+// static
+SharedMemoryABI::PageLayout SharedMemoryArbiter::default_page_layout =
+    SharedMemoryABI::PageLayout::kPageDiv1;
+
+SharedMemoryArbiter::SharedMemoryArbiter(void* start,
+                                         size_t size,
+                                         size_t page_size,
+                                         OnPageCompleteCallback callback,
+                                         base::TaskRunner* task_runner)
+    : task_runner_(task_runner),
+      on_page_complete_callback_(std::move(callback)),
+      shmem_abi_(reinterpret_cast<uint8_t*>(start), size, page_size),
+      active_writer_ids_(SharedMemoryABI::kMaxWriterID + 1) {}
+
+Chunk SharedMemoryArbiter::GetNewChunk(
+    const SharedMemoryABI::ChunkHeader& header,
+    BufferID target_buffer,
+    size_t size_hint) {
+  PERFETTO_DCHECK(size_hint == 0);  // Not implemented yet.
+
+  for (;;) {
+    // TODO(primiano): Probably this lock is not really required and this code
+    // could be rewritten leveraging only the Try* atomic operations in
+    // SharedMemoryABI. But let's not be too adventurous for the moment.
+    {
+      std::lock_guard<std::mutex> scoped_lock(lock_);
+      const size_t initial_page_idx = page_idx_;
+      // TODO(primiano): instead of scanning, we could maintain a bitmap of
+      // free chunks for each |target_buffer| and one for fully free pages.
+      for (size_t i = 0; i < shmem_abi_.num_pages(); i++) {
+        page_idx_ = (initial_page_idx + i) % shmem_abi_.num_pages();
+        bool is_new_page = false;
+
+        // TODO(primiano): make the page layout dynamic.
+        auto layout = SharedMemoryArbiter::default_page_layout;
+
+        if (shmem_abi_.is_page_free(page_idx_)) {
+          // TODO(primiano): Use the |size_hint| here to decide the layout.
+          is_new_page =
+              shmem_abi_.TryPartitionPage(page_idx_, layout, target_buffer);
+        }
+        uint32_t free_chunks;
+        size_t tbuf;
+        if (is_new_page) {
+          free_chunks = (1 << SharedMemoryABI::kNumChunksForLayout[layout]) - 1;
+          tbuf = target_buffer;
+        } else {
+          free_chunks = shmem_abi_.GetFreeChunks(page_idx_);
+
+          // |tbuf| here is advisory only and could change at any point, before
+          // or after this read. The only use of |tbuf| here is to to skip pages
+          // that are more likely to belong to other target_buffers, avoiding
+          // the more epxensive atomic operations in those cases. The
+          // authoritative check on |tbuf| happens atomically in the
+          // TryAcquireChunkForWriting() call below.
+          tbuf = shmem_abi_.page_header(page_idx_)->target_buffer.load(
+              std::memory_order_relaxed);
+        }
+        PERFETTO_DLOG("Free chunks for page %zu: %x. Target buffer: %zu",
+                      page_idx_, free_chunks, tbuf);
+
+        if (tbuf != target_buffer)
+          continue;
+
+        for (uint32_t chunk_idx = 0; free_chunks;
+             chunk_idx++, free_chunks >>= 1) {
+          if (!(free_chunks & 1))
+            continue;
+          // We found a free chunk.
+          Chunk chunk = shmem_abi_.TryAcquireChunkForWriting(
+              page_idx_, chunk_idx, tbuf, &header);
+          if (!chunk.is_valid())
+            continue;
+          PERFETTO_DLOG("Acquired chunk %zu:%u", page_idx_, chunk_idx);
+          return chunk;
+        }
+        // TODO: we should have some policy to guarantee fairness of the SMB
+        // page allocator w.r.t |target_buffer|? Or is the SMB best-effort. All
+        // chunks in the page are busy (either kBeingRead or kBeingWritten), or
+        // all the pages are assigned to a different target buffer. Try with the
+        // next page.
+      }
+    }  // std::lock_guard<std::mutex>
+    // All chunks are taken (either kBeingWritten by us or kBeingRead by the
+    // Service). TODO: at this point we should return a bankrupcy chunk, not
+    // crash the process.
+    PERFETTO_ELOG("Shared memory buffer overrun! Stalling");
+    usleep(250000);
+  }
+}
+
+void SharedMemoryArbiter::ReturnCompletedChunk(Chunk chunk) {
+  bool should_post_callback = false;
+  {
+    std::lock_guard<std::mutex> scoped_lock(lock_);
+    size_t page_index = shmem_abi_.ReleaseChunkAsComplete(std::move(chunk));
+    if (page_index != SharedMemoryABI::kInvalidPageIdx) {
+      should_post_callback = pages_to_notify_.empty();
+      pages_to_notify_.push_back(static_cast<uint32_t>(page_index));
+    }
+  }
+  if (should_post_callback) {
+    // TODO what happens if the arbiter gets destroyed?
+    task_runner_->PostTask(
+        std::bind(&SharedMemoryArbiter::InvokeOnPageCompleteCallback, this));
+  }
+}
+
+// This is always invoked on the |task_runner_| thread.
+void SharedMemoryArbiter::InvokeOnPageCompleteCallback() {
+  std::vector<uint32_t> pages_to_notify;
+  {
+    std::lock_guard<std::mutex> scoped_lock(lock_);
+    pages_to_notify = std::move(pages_to_notify_);
+    pages_to_notify_.clear();
+  }
+  on_page_complete_callback_(pages_to_notify);
+}
+
+std::unique_ptr<TraceWriter> SharedMemoryArbiter::CreateTraceWriter(
+    BufferID target_buffer) {
+  WriterID id;
+  {
+    std::lock_guard<std::mutex> scoped_lock(lock_);
+    id = static_cast<WriterID>(active_writer_ids_.Allocate());
+  }
+  return std::unique_ptr<TraceWriter>(
+      id ? new TraceWriterImpl(this, id, target_buffer) : nullptr);
+}
+
+void SharedMemoryArbiter::ReleaseWriterID(WriterID id) {
+  std::lock_guard<std::mutex> scoped_lock(lock_);
+  active_writer_ids_.Free(id);
+}
+
+}  // namespace perfetto
diff --git a/src/tracing/core/shared_memory_arbiter.h b/src/tracing/core/shared_memory_arbiter.h
new file mode 100644
index 0000000..34267bf
--- /dev/null
+++ b/src/tracing/core/shared_memory_arbiter.h
@@ -0,0 +1,112 @@
+/*
+ * Copyright (C) 2017 The Android Open Source Project
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#ifndef SRC_TRACING_CORE_SHARED_MEMORY_ARBITER_H_
+#define SRC_TRACING_CORE_SHARED_MEMORY_ARBITER_H_
+
+#include <stdint.h>
+
+#include <functional>
+#include <mutex>
+#include <vector>
+
+#include "perfetto/tracing/core/basic_types.h"
+#include "perfetto/tracing/core/shared_memory_abi.h"
+#include "src/tracing/core/id_allocator.h"
+
+namespace perfetto {
+
+class TraceWriter;
+
+namespace base {
+class TaskRunner;
+}  // namespace base
+
+// This class handles the shared memory buffer on the producer side. It is used
+// to obtain thread-local chunks and to partition pages from several threads.
+// There is one arbiter instance per Producer.
+// This class is thread-safe and uses locks to do so. Data sources are supposed
+// to interact with this sporadically, only when they run out of space on their
+// current thread-local chunk.
+class SharedMemoryArbiter {
+ public:
+  using OnPageCompleteCallback =
+      std::function<void(const std::vector<uint32_t>& /*page_indexes*/)>;
+
+  // Args:
+  // |start|,|size|: boundaries of the shared memory buffer.
+  // |page_size|: a multiple of 4KB that defines the granularity of tracing
+  // pages. See tradeoff considerations in shared_memory_abi.h.
+  // |OnPageCompleteCallback|: a callback that will be posted on the passed
+  // |TaskRunner| when one or more pages are complete (and hence the Producer
+  // should send a NotifySharedMemoryUpdate() to the Service).
+  SharedMemoryArbiter(void* start,
+                      size_t size,
+                      size_t page_size,
+                      OnPageCompleteCallback,
+                      base::TaskRunner*);
+
+  // Creates a new TraceWriter and assigns it a new WriterID. The WriterID is
+  // written in each chunk header owned by a given TraceWriter and is used by
+  // the Service to reconstruct TracePackets written by the same TraceWriter.
+  // Returns nullptr if all WriterID slots are exhausted.
+  // TODO(primiano): instead of nullptr this should return a NoopWriter.
+  std::unique_ptr<TraceWriter> CreateTraceWriter(BufferID target_buffer = 0);
+
+  // Returns a new Chunk to write tracing data. The call always returns a valid
+  // Chunk. TODO(primiano): right now this blocks if there are no free chunks
+  // in the SMB. In the long term the caller should be allowed to pick a policy
+  // and handle the retry itself asynchronously.
+  SharedMemoryABI::Chunk GetNewChunk(const SharedMemoryABI::ChunkHeader&,
+                                     BufferID target_buffer,
+                                     size_t size_hint = 0);
+
+  void ReturnCompletedChunk(SharedMemoryABI::Chunk chunk);
+
+  SharedMemoryABI* shmem_abi_for_testing() { return &shmem_abi_; }
+
+  static void set_default_layout_for_testing(SharedMemoryABI::PageLayout l) {
+    default_page_layout = l;
+  }
+
+ private:
+  friend class TraceWriterImpl;
+
+  static SharedMemoryABI::PageLayout default_page_layout;
+
+  SharedMemoryArbiter(const SharedMemoryArbiter&) = delete;
+  SharedMemoryArbiter& operator=(const SharedMemoryArbiter&) = delete;
+
+  // Called by the TraceWriter destructor.
+  void ReleaseWriterID(WriterID);
+
+  void InvokeOnPageCompleteCallback();
+
+  base::TaskRunner* const task_runner_;
+  OnPageCompleteCallback on_page_complete_callback_;
+
+  // --- Begin lock-protected members ---
+  std::mutex lock_;
+  SharedMemoryABI shmem_abi_;
+  size_t page_idx_ = 0;
+  IdAllocator active_writer_ids_;
+  std::vector<uint32_t> pages_to_notify_;
+  // --- End lock-protected members ---
+};
+
+}  // namespace perfetto
+
+#endif  // SRC_TRACING_CORE_SHARED_MEMORY_ARBITER_H_
diff --git a/src/tracing/core/shared_memory_arbiter_unittest.cc b/src/tracing/core/shared_memory_arbiter_unittest.cc
new file mode 100644
index 0000000..d2b511b
--- /dev/null
+++ b/src/tracing/core/shared_memory_arbiter_unittest.cc
@@ -0,0 +1,174 @@
+/*
+ * Copyright (C) 2017 The Android Open Source Project
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include "src/tracing/core/shared_memory_arbiter.h"
+
+#include "gtest/gtest.h"
+#include "perfetto/base/utils.h"
+#include "perfetto/tracing/core/basic_types.h"
+#include "perfetto/tracing/core/shared_memory_abi.h"
+#include "perfetto/tracing/core/trace_writer.h"
+#include "src/base/test/test_task_runner.h"
+#include "src/tracing/test/aligned_buffer_test.h"
+
+namespace perfetto {
+namespace {
+
+constexpr size_t kMaxWriterID = SharedMemoryABI::kMaxWriterID;
+
+class SharedMemoryArbiterTest : public AlignedBufferTest {
+ public:
+  void SetUp() override {
+    AlignedBufferTest::SetUp();
+    auto callback = [this](const std::vector<uint32_t>& arg) {
+      if (on_page_complete_)
+        on_page_complete_(arg);
+    };
+    task_runner_.reset(new base::TestTaskRunner());
+    arbiter_.reset(new SharedMemoryArbiter(buf(), buf_size(), page_size(),
+                                           callback, task_runner_.get()));
+  }
+
+  void TearDown() override {
+    arbiter_.reset();
+    task_runner_.reset();
+  }
+
+  std::unique_ptr<base::TestTaskRunner> task_runner_;
+  std::unique_ptr<SharedMemoryArbiter> arbiter_;
+  std::function<void(const std::vector<uint32_t>&)> on_page_complete_;
+};
+
+size_t const kPageSizes[] = {4096, 65536};
+INSTANTIATE_TEST_CASE_P(PageSize,
+                        SharedMemoryArbiterTest,
+                        ::testing::ValuesIn(kPageSizes));
+
+// Checks that chunks that target different buffer IDs are placed in different
+// pages.
+TEST_P(SharedMemoryArbiterTest, ChunksAllocationByTargetBufferID) {
+  SharedMemoryArbiter::set_default_layout_for_testing(
+      SharedMemoryABI::PageLayout::kPageDiv4);
+  SharedMemoryABI::Chunk chunks[8];
+  chunks[0] = arbiter_->GetNewChunk({}, 1 /* target buffer id */, 0);
+  chunks[1] = arbiter_->GetNewChunk({}, 1 /* target buffer id */, 0);
+  chunks[2] = arbiter_->GetNewChunk({}, 1 /* target buffer id */, 0);
+  chunks[3] = arbiter_->GetNewChunk({}, 2 /* target buffer id */, 0);
+  chunks[4] = arbiter_->GetNewChunk({}, 1 /* target buffer id */, 0);
+  chunks[5] = arbiter_->GetNewChunk({}, 1 /* target buffer id */, 0);
+  chunks[6] = arbiter_->GetNewChunk({}, 3 /* target buffer id */, 0);
+  chunks[7] = arbiter_->GetNewChunk({}, 3 /* target buffer id */, 0);
+
+  // "first" == "page index", "second" == "chunk index".
+  std::pair<size_t, size_t> idx[base::ArraySize(chunks)];
+  for (size_t i = 0; i < base::ArraySize(chunks); i++)
+    idx[i] = arbiter_->shmem_abi_for_testing()->GetPageAndChunkIndex(chunks[i]);
+
+  // The first three chunks should lay in the same page, as they target the same
+  // buffer id (1).
+  EXPECT_EQ(idx[0].first, idx[1].first);
+  EXPECT_EQ(idx[0].first, idx[2].first);
+
+  // Check also that the chunk IDs are different.
+  EXPECT_NE(idx[0].second, idx[1].second);
+  EXPECT_NE(idx[1].second, idx[2].second);
+  EXPECT_NE(idx[0].second, idx[2].second);
+
+  // The next one instead should be given a dedicated page because it targets
+  // a different buffer id (2);
+  EXPECT_NE(idx[2].first, idx[3].first);
+
+  // Hoever the next two chunks should be able to fit back into the same page.
+  EXPECT_EQ(idx[4].first, idx[5].first);
+  EXPECT_NE(idx[4].second, idx[5].second);
+
+  // Similarly the last two chunks should be able to share the same page, but
+  // not any page of the previous chunks.
+  EXPECT_NE(idx[0].first, idx[6].first);
+  EXPECT_NE(idx[3].first, idx[6].first);
+  EXPECT_EQ(idx[6].first, idx[7].first);
+  EXPECT_NE(idx[6].second, idx[7].second);
+
+  // TODO(primiano): check that after saturating all the pages, the arbiter
+  // goes back and reuses free chunks of previous pages. e.g., at some point
+  // a chunk targeting buffer id == 1 should be placed into (page:0, chunk:3).
+}
+
+// The buffer has 14 pages (kNumPages), each will be partitioned in 14 chunks.
+// The test requests all 14 * 14 chunks, alternating amongst 14 target buf IDs.
+// Because a chunk can share a page only if all other chunks in the page have
+// the same target buffer ID, there is only one possible final distribution:
+// each page is filled with chunks that all belong to the same buffer ID.
+TEST_P(SharedMemoryArbiterTest, GetAndReturnChunks) {
+  SharedMemoryArbiter::set_default_layout_for_testing(
+      SharedMemoryABI::PageLayout::kPageDiv14);
+  static constexpr size_t kTotChunks = kNumPages * 14;
+  SharedMemoryABI::Chunk chunks[kTotChunks];
+  for (size_t i = 0; i < kTotChunks; i++) {
+    BufferID target_buffer = i % 14;
+    chunks[i] = arbiter_->GetNewChunk({}, target_buffer, 0 /*size_hint*/);
+    ASSERT_TRUE(chunks[i].is_valid());
+  }
+
+  SharedMemoryABI* abi = arbiter_->shmem_abi_for_testing();
+  for (size_t page_idx = 0; page_idx < kNumPages; page_idx++) {
+    ASSERT_FALSE(abi->is_page_free(page_idx));
+    ASSERT_EQ(0u, abi->GetFreeChunks(page_idx));
+    const uint32_t page_layout = abi->page_layout_dbg(page_idx);
+    ASSERT_EQ(14u, SharedMemoryABI::GetNumChunksForLayout(page_layout));
+    ASSERT_EQ(page_idx % 14, abi->page_header(page_idx)->target_buffer.load());
+    for (size_t chunk_idx = 0; chunk_idx < 14; chunk_idx++) {
+      auto chunk = abi->GetChunkUnchecked(page_idx, page_layout, chunk_idx);
+      ASSERT_TRUE(chunk.is_valid());
+    }
+  }
+
+  // Finally return just two pages marking all their chunks as complete, and
+  // check that the notification callback is posted.
+
+  auto on_callback = task_runner_->CreateCheckpoint("on_callback");
+  on_page_complete_ =
+      [on_callback](const std::vector<uint32_t>& completed_pages) {
+        ASSERT_EQ(2u, completed_pages.size());
+        ASSERT_EQ(0u, completed_pages[0]);
+        ASSERT_EQ(3u, completed_pages[1]);
+        on_callback();
+      };
+  for (size_t i = 0; i < 14; i++) {
+    arbiter_->ReturnCompletedChunk(std::move(chunks[14 * i]));
+    arbiter_->ReturnCompletedChunk(std::move(chunks[14 * i + 3]));
+  }
+  task_runner_->RunUntilCheckpoint("on_callback");
+}
+
+// Check that we can actually create up to kMaxWriterID TraceWriter(s).
+TEST_P(SharedMemoryArbiterTest, WriterIDsAllocation) {
+  std::map<WriterID, std::unique_ptr<TraceWriter>> writers;
+  for (size_t i = 0; i < kMaxWriterID; i++) {
+    std::unique_ptr<TraceWriter> writer = arbiter_->CreateTraceWriter(0);
+    ASSERT_TRUE(writer);
+    WriterID writer_id = writer->writer_id();
+    ASSERT_TRUE(writers.emplace(writer_id, std::move(writer)).second);
+  }
+
+  // A further call should fail as we exhausted writer IDs.
+  ASSERT_EQ(nullptr, arbiter_->CreateTraceWriter(0).get());
+}
+
+// TODO(primiano): add multi-threaded tests.
+
+}  // namespace
+}  // namespace perfetto
diff --git a/src/tracing/core/trace_writer_impl.cc b/src/tracing/core/trace_writer_impl.cc
new file mode 100644
index 0000000..a17e543
--- /dev/null
+++ b/src/tracing/core/trace_writer_impl.cc
@@ -0,0 +1,172 @@
+/*
+ * Copyright (C) 2017 The Android Open Source Project
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include "src/tracing/core/trace_writer_impl.h"
+
+#include <string.h>
+
+#include <type_traits>
+#include <utility>
+
+#include "perfetto/base/logging.h"
+#include "perfetto/protozero/proto_utils.h"
+#include "src/tracing/core/shared_memory_arbiter.h"
+
+#include "protos/trace_packet.pbzero.h"
+
+// TODO(primiano): right now this class is accumulating a patchlist but not
+// sending it to the service.
+
+using protozero::proto_utils::kMessageLengthFieldSize;
+using protozero::proto_utils::WriteRedundantVarInt;
+using ChunkHeader = perfetto::SharedMemoryABI::ChunkHeader;
+
+namespace perfetto {
+
+namespace {
+constexpr size_t kPacketHeaderSize = SharedMemoryABI::kPacketHeaderSize;
+}  // namespace
+
+TraceWriterImpl::TraceWriterImpl(SharedMemoryArbiter* shmem_arbiter,
+                                 WriterID id,
+                                 BufferID target_buffer)
+    : shmem_arbiter_(shmem_arbiter),
+      id_(id),
+      target_buffer_(target_buffer),
+      protobuf_stream_writer_(this) {
+  // TODO(primiano): we could handle the case of running out of TraceWriterID(s)
+  // more gracefully and always return a no-op TracePacket in NewTracePacket().
+  PERFETTO_CHECK(id_ != 0);
+
+  cur_packet_.reset(new protos::pbzero::TracePacket());
+  cur_packet_->Finalize();  // To avoid the DCHECK in NewTracePacket().
+}
+
+TraceWriterImpl::~TraceWriterImpl() {
+  // TODO(primiano): this should also return the current chunk. Add tests.
+  shmem_arbiter_->ReleaseWriterID(id_);
+}
+
+TraceWriterImpl::TracePacketHandle TraceWriterImpl::NewTracePacket() {
+  // If we hit this, the caller is calling NewTracePacket() without having
+  // finalized the previous packet.
+  PERFETTO_DCHECK(cur_packet_->is_finalized());
+
+  fragmenting_packet_ = false;
+
+  // TODO: hack to get a new page every time and reduce fragmentation (that
+  // requires stitching support in the service).
+  protobuf_stream_writer_.Reset(GetNewBuffer());
+
+  // Reserve space for the size of the message. Note: this call might re-enter
+  // into this class invoking GetNewBuffer() if there isn't enough space or if
+  // this is the very first call to NewTracePacket().
+  static_assert(
+      kPacketHeaderSize == kMessageLengthFieldSize,
+      "The packet header must match the ProtoZeroMessage header size");
+  cur_packet_->Reset(&protobuf_stream_writer_);
+  uint8_t* header = protobuf_stream_writer_.ReserveBytes(kPacketHeaderSize);
+  memset(header, 0, kPacketHeaderSize);
+  cur_packet_->set_size_field(header);
+  cur_chunk_.IncrementPacketCount();
+  TracePacketHandle handle(cur_packet_.get());
+  cur_fragment_start_ = protobuf_stream_writer_.write_ptr();
+  fragmenting_packet_ = true;
+  return handle;
+}
+
+// Called by the ProtoZeroMessage. We can get here in two cases:
+// 1. In the middle of writing a ProtoZeroMessage,
+// when |fragmenting_packet_| == true. In this case we want to update the
+// chunk header with a partial packet and start a new partial packet in the
+// new chunk.
+// 2. While calling ReserveBytes() for the packet header in NewTracePacket().
+// In this case |fragmenting_packet_| == false and we just want a new chunk
+// without creating any fragments.
+protozero::ContiguousMemoryRange TraceWriterImpl::GetNewBuffer() {
+  if (fragmenting_packet_) {
+    uint8_t* const wptr = protobuf_stream_writer_.write_ptr();
+    PERFETTO_DCHECK(wptr >= cur_fragment_start_);
+    uint32_t partial_size = static_cast<uint32_t>(wptr - cur_fragment_start_);
+    PERFETTO_DCHECK(partial_size < cur_chunk_.size());
+
+    // Backfill the packet header with the fragment size.
+    cur_packet_->inc_size_already_written(partial_size);
+    cur_chunk_.SetFlag(ChunkHeader::kLastPacketContinuesOnNextChunk);
+    WriteRedundantVarInt(partial_size, cur_packet_->size_field());
+
+    // Descend in the stack of non-finalized nested submessages (if any) and
+    // detour their |size_field| into the |patch_list_|. At this point we have
+    // to release the chunk and they cannot write anymore into that.
+    // TODO(primiano): add tests to cover this logic.
+    for (auto* nested_msg = cur_packet_->nested_message(); nested_msg;
+         nested_msg = nested_msg->nested_message()) {
+      uint8_t* const cur_hdr = nested_msg->size_field();
+      PERFETTO_DCHECK(cur_hdr >= cur_chunk_.payload_begin() &&
+                      cur_hdr + kMessageLengthFieldSize <= cur_chunk_.end());
+      auto cur_hdr_offset = static_cast<uint16_t>(cur_hdr - cur_chunk_.begin());
+      patch_list_.emplace_front(cur_chunk_id_, cur_hdr_offset);
+      Patch& patch = patch_list_.front();
+      nested_msg->set_size_field(patch.size_field);
+      PERFETTO_DLOG("Created new patchlist entry for protobuf nested message");
+    }
+  }
+
+  if (cur_chunk_.is_valid())
+    shmem_arbiter_->ReturnCompletedChunk(std::move(cur_chunk_));
+
+  // Start a new chunk.
+  ChunkHeader::Identifier identifier = {};
+  identifier.writer_id = id_;
+  identifier.chunk_id = cur_chunk_id_++;
+
+  ChunkHeader::PacketsState packets_state = {};
+  if (fragmenting_packet_) {
+    packets_state.count = 1;
+    packets_state.flags = ChunkHeader::kFirstPacketContinuesFromPrevChunk;
+  }
+
+  // The memory order of the stores below doesn't really matter. This |header|
+  // is just a local temporary object. The GetNewChunk() call below will copy it
+  // into the shared buffer with the proper barriers.
+  ChunkHeader header = {};
+  header.identifier.store(identifier, std::memory_order_relaxed);
+  header.packets_state.store(packets_state, std::memory_order_relaxed);
+
+  cur_chunk_ = shmem_arbiter_->GetNewChunk(header, target_buffer_);
+  uint8_t* payload_begin = cur_chunk_.payload_begin();
+  if (fragmenting_packet_) {
+    cur_packet_->set_size_field(payload_begin);
+    memset(payload_begin, 0, kPacketHeaderSize);
+    payload_begin += kPacketHeaderSize;
+    cur_fragment_start_ = payload_begin;
+  }
+
+  return protozero::ContiguousMemoryRange{payload_begin, cur_chunk_.end()};
+}
+
+WriterID TraceWriterImpl::writer_id() const {
+  return id_;
+};
+
+TraceWriterImpl::Patch::Patch(uint16_t cid, uint16_t offset)
+    : chunk_id(cid), offset_in_chunk(offset) {}
+
+// Base class ctor/dtor definition.
+TraceWriter::TraceWriter() = default;
+TraceWriter::~TraceWriter() = default;
+
+}  // namespace perfetto
diff --git a/src/tracing/core/trace_writer_impl.h b/src/tracing/core/trace_writer_impl.h
new file mode 100644
index 0000000..3c7121d
--- /dev/null
+++ b/src/tracing/core/trace_writer_impl.h
@@ -0,0 +1,118 @@
+/*
+ * Copyright (C) 2017 The Android Open Source Project
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#ifndef SRC_TRACING_CORE_TRACE_WRITER_IMPL_H_
+#define SRC_TRACING_CORE_TRACE_WRITER_IMPL_H_
+
+#include <forward_list>
+
+#include "perfetto/protozero/protozero_message_handle.h"
+#include "perfetto/protozero/scattered_stream_writer.h"
+#include "perfetto/tracing/core/basic_types.h"
+#include "perfetto/tracing/core/shared_memory_abi.h"
+#include "perfetto/tracing/core/trace_writer.h"
+
+namespace perfetto {
+
+class SharedMemoryArbiter;
+
+// See //include/perfetto/tracing/core/trace_writer.h for docs.
+class TraceWriterImpl : public TraceWriter,
+                        public protozero::ScatteredStreamWriter::Delegate {
+ public:
+  // TracePacketHandle is defined in trace_writer.h
+
+  TraceWriterImpl(SharedMemoryArbiter*, WriterID, BufferID);
+  ~TraceWriterImpl() override;
+
+  // TraceWriter implementation. See documentation in trace_writer.h .
+  TracePacketHandle NewTracePacket() override;
+  WriterID writer_id() const override;
+
+ private:
+  // Used to handle the backfilling of the headers (the |size_field|) of nested
+  // messages when a proto is fragmented over several chunks. This patchlist is
+  // sent out of band to the tracing service, after having returned the initial
+  // chunks of the fragment.
+  struct Patch {
+    Patch(uint16_t chunk_id, uint16_t offset);
+    Patch(const Patch&) = delete;
+    Patch& operator=(const Patch&) = delete;
+    Patch(Patch&&) noexcept = delete;
+    Patch& operator=(Patch&&) = delete;
+
+    const uint16_t chunk_id;
+    const uint16_t offset_in_chunk;
+    uint8_t size_field[SharedMemoryABI::kPacketHeaderSize] = {};
+  };
+  TraceWriterImpl(const TraceWriterImpl&) = delete;
+  TraceWriterImpl& operator=(const TraceWriterImpl&) = delete;
+
+  // ScatteredStreamWriter::Delegate implementation.
+  protozero::ContiguousMemoryRange GetNewBuffer() override;
+
+  // The per-producer arbiter that coordinates access to the shared memory
+  // buffer from several threads.
+  SharedMemoryArbiter* const shmem_arbiter_;
+
+  // ID of the current writer.
+  const WriterID id_;
+
+  // This is just copied back into the chunk header.
+  // See comments in data_source_config.proto for |target_buffer|.
+  const BufferID target_buffer_;
+
+  // Monotonic (% wrapping) sequence id of the chunk. Together with the WriterID
+  // this allows the Service to reconstruct the linear sequence of packets.
+  uint16_t cur_chunk_id_ = 0;
+
+  // The chunk we are holding onto (if any).
+  SharedMemoryABI::Chunk cur_chunk_;
+
+  // Passed to protozero message to write directly into |cur_chunk_|. It
+  // keeps track of the write pointer. It calls us back (GetNewBuffer()) when
+  // |cur_chunk_| is filled.
+  protozero::ScatteredStreamWriter protobuf_stream_writer_;
+
+  // The packet returned via NewTracePacket(). Its owned by this class,
+  // TracePacketHandle has just a pointer to it.
+  std::unique_ptr<protos::pbzero::TracePacket> cur_packet_;
+
+  // The start address of |cur_packet_| within |cur_chunk_|. Used to figure out
+  // fragments sizes when a TracePacket write is interrupted by GetNewBuffer().
+  uint8_t* cur_fragment_start_ = nullptr;
+
+  // true if we received a call to GetNewBuffer() after NewTracePacket(),
+  // false if GetNewBuffer() happened during NewTracePacket() prologue, while
+  // starting the TracePacket header.
+  bool fragmenting_packet_ = false;
+
+  // When a packet is fragmented across different chunks, the |size_field| of
+  // the outstanding nested protobuf messages is redirected onto Patch entries
+  // in this list at the time the Chunk is returned (because at that point we
+  // have to release the ownership of the current Chunk). This list will be
+  // later sent out-of-band to the tracing service, who will patch the required
+  // chunks, if they are still around.
+  // Note: the ProtoZeroMessage will take pointers to the |size_field| of these
+  // entries. This container must guarantee that the Patch objects are never
+  // moved around (i.e. cannot be a vector because of reallocations can change
+  // addresses of pre-existing entries).
+  std::forward_list<Patch> patch_list_;
+};
+
+}  // namespace perfetto
+
+#endif  // SRC_TRACING_CORE_TRACE_WRITER_IMPL_H_
diff --git a/src/tracing/core/trace_writer_impl_unittest.cc b/src/tracing/core/trace_writer_impl_unittest.cc
new file mode 100644
index 0000000..15b46a8
--- /dev/null
+++ b/src/tracing/core/trace_writer_impl_unittest.cc
@@ -0,0 +1,90 @@
+/*
+ * Copyright (C) 2017 The Android Open Source Project
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include "src/tracing/core/trace_writer_impl.h"
+
+#include "gtest/gtest.h"
+#include "perfetto/base/utils.h"
+#include "perfetto/tracing/core/trace_writer.h"
+#include "src/base/test/test_task_runner.h"
+#include "src/tracing/core/shared_memory_arbiter.h"
+#include "src/tracing/test/aligned_buffer_test.h"
+
+#include "protos/trace_packet.pbzero.h"
+
+namespace perfetto {
+namespace {
+
+class TraceWriterImplTest : public AlignedBufferTest {
+ public:
+  void SetUp() override {
+    SharedMemoryArbiter::set_default_layout_for_testing(
+        SharedMemoryABI::PageLayout::kPageDiv4);
+    AlignedBufferTest::SetUp();
+    auto callback = [](const std::vector<uint32_t>& arg) {};
+    task_runner_.reset(new base::TestTaskRunner());
+    arbiter_.reset(new SharedMemoryArbiter(buf(), buf_size(), page_size(),
+                                           callback, task_runner_.get()));
+  }
+
+  void TearDown() override {
+    arbiter_.reset();
+    task_runner_.reset();
+  }
+
+  std::unique_ptr<base::TestTaskRunner> task_runner_;
+  std::unique_ptr<SharedMemoryArbiter> arbiter_;
+  std::function<void(const std::vector<uint32_t>&)> on_page_complete_;
+};
+
+size_t const kPageSizes[] = {4096, 65536};
+INSTANTIATE_TEST_CASE_P(PageSize,
+                        TraceWriterImplTest,
+                        ::testing::ValuesIn(kPageSizes));
+
+TEST_P(TraceWriterImplTest, SingleWriter) {
+  const BufferID tgt_buf_id = 42;
+  std::unique_ptr<TraceWriter> writer = arbiter_->CreateTraceWriter(tgt_buf_id);
+  for (int i = 0; i < 32; i++) {
+    auto packet = writer->NewTracePacket();
+    char str[16];
+    sprintf(str, "foobar %d", i);
+    packet->set_test(str);
+  }
+  writer->NewTracePacket()->set_test("workaround for returing the last chunk");
+  writer.reset();
+
+  SharedMemoryABI* abi = arbiter_->shmem_abi_for_testing();
+  size_t packets_count = 0;
+  for (size_t page_idx = 0; page_idx < kNumPages; page_idx++) {
+    uint32_t page_layout = abi->page_layout_dbg(page_idx);
+    size_t num_chunks = SharedMemoryABI::GetNumChunksForLayout(page_layout);
+    for (size_t chunk_idx = 0; chunk_idx < num_chunks; chunk_idx++) {
+      auto chunk =
+          abi->TryAcquireChunkForReading(page_idx, chunk_idx, tgt_buf_id);
+      if (!chunk.is_valid())
+        continue;
+      packets_count += chunk.header()->packets_state.load().count;
+    }
+  }
+  EXPECT_EQ(32u, packets_count);
+  // TODO(primiano): check also the content of the packets decoding the protos.
+}
+
+// TODO(primiano): add multi-writer test.
+
+}  // namespace
+}  // namespace perfetto
diff --git a/src/tracing/test/aligned_buffer_test.cc b/src/tracing/test/aligned_buffer_test.cc
new file mode 100644
index 0000000..6a603ea
--- /dev/null
+++ b/src/tracing/test/aligned_buffer_test.cc
@@ -0,0 +1,35 @@
+/*
+ * Copyright (C) 2017 The Android Open Source Project
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include "src/tracing/test/aligned_buffer_test.h"
+
+#include "perfetto/base/logging.h"
+
+namespace perfetto {
+
+// static
+constexpr size_t AlignedBufferTest::kNumPages;
+
+void AlignedBufferTest::SetUp() {
+  page_size_ = GetParam();
+  buf_.reset(new TestSharedMemory(page_size_ * kNumPages));
+}
+
+void AlignedBufferTest::TearDown() {
+  buf_.reset();
+}
+
+}  // namespace perfetto
diff --git a/src/tracing/test/aligned_buffer_test.h b/src/tracing/test/aligned_buffer_test.h
new file mode 100644
index 0000000..23c8b1c
--- /dev/null
+++ b/src/tracing/test/aligned_buffer_test.h
@@ -0,0 +1,51 @@
+/*
+ * Copyright (C) 2017 The Android Open Source Project
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#ifndef SRC_TRACING_TEST_ALIGNED_BUFFER_TEST_H_
+#define SRC_TRACING_TEST_ALIGNED_BUFFER_TEST_H_
+
+#include <stdlib.h>
+
+#include <memory>
+
+#include "gtest/gtest.h"
+#include "perfetto/base/utils.h"
+#include "src/tracing/test/test_shared_memory.h"
+
+namespace perfetto {
+
+// Base parametrized test for unittests that require an aligned buffer.
+class AlignedBufferTest : public ::testing::TestWithParam<size_t> {
+ public:
+  static constexpr size_t kNumPages = 14;
+  void SetUp() override;
+  void TearDown() override;
+
+  uint8_t* buf() const { return reinterpret_cast<uint8_t*>(buf_->start()); }
+  size_t buf_size() const { return buf_->size(); }
+  size_t page_size() const { return page_size_; }
+
+ private:
+  size_t page_size_ = 0;
+
+  // This doesn't need any sharing. TestSharedMemory just happens to have the
+  // right harness to create a zeroed page-aligned buffer.
+  std::unique_ptr<TestSharedMemory> buf_;
+};
+
+}  // namespace perfetto
+
+#endif  // SRC_TRACING_TEST_ALIGNED_BUFFER_TEST_H_