tracing: add the v2 shared-ring chunk protocol
Add the producer-local shared ring used by tracing v2. This change contains
the shared-memory ABI, one writer per TraceWriter, the single reader, and the
protocol tests. SDK and service wiring follow in separate changes.
- Define a 64-byte ring header:
- an always-lock-free atomic64 packs write_pos in the high half and read_pos
in the low half;
- num_writers_waiting avoids a wake syscall when no writer is asleep;
- a zero-filled mapping is a valid empty ring.
- Define one atomic32 state word at the start of each chunk:
- Free(wrap) exposes the chunk to one ring traversal;
- BeingWritten identifies the writer and its published fragment prefix;
- Complete says the writer has finished touching the chunk, although it may
take the chunk back to append more fragments;
- RewriteRequested says the reader took the published prefix and the writer
must move any newer suffix;
- RewriteAcknowledged says the writer released the old chunk and the reader
may return it to Free.
- Reserve uint32 positions in FIFO order and map them onto a power-of-two
number of physical chunks. Free carries a 16-bit wrap count so a delayed
writer cannot claim the same chunk during a later traversal.
- Keep ownership rules local to the chunk word:
- one reservation permits one exact claim attempt;
- only the reader writes Free;
- the reader never waits for a writer;
- a lost reader CAS leaves read_pos unchanged and retries the position in a
later drain.
- Store target BufferID after the state word. Payload fragments grow towards
higher addresses while their sizes grow down from the end of the chunk as
shortest-form protobuf varints. Fragment counts publish an append-only
prefix. If the reader reaches an active writer, it copies that prefix and
asks the writer to move only the unpublished suffix.
- Keep the initial 256-byte minimum chunk size and require a four-byte-aligned
chunk stride. Check ring geometry and allocation arithmetic before mapping
memory. Require atomic32 and atomic64 to be always lock-free.
- Honor BufferExhaustedPolicy in SharedRingBufferWriter. Linux and Android
writers can wait on the read_pos half of rw_positions with a futex. Other
platforms report that blocking is unavailable instead of spinning. The
waiter hint and its seq_cst fence pair prevent a missed wake across the two
atomic words.
- Test ABI encoding, position and wrap rollover, malformed chunk input, every
state transition, failed claims, prefix copying, suffix relocation, futex
wake races, and concurrent multi-writer stress.
Bug: 536851377
diff --git a/Android.bp b/Android.bp
index f072867..56e203a 100644
--- a/Android.bp
+++ b/Android.bp
@@ -23009,6 +23009,28 @@
],
}
+// GN: //src/tracing/v2:unittests
+filegroup {
+ name: "perfetto_src_tracing_v2_unittests",
+ srcs: [
+ "src/tracing/v2/shared_ring_buffer_concurrency_unittest.cc",
+ "src/tracing/v2/shared_ring_buffer_reader_unittest.cc",
+ "src/tracing/v2/shared_ring_buffer_unittest.cc",
+ "src/tracing/v2/shared_ring_buffer_writer_unittest.cc",
+ "src/tracing/v2/tracing_v2_abi_unittest.cc",
+ ],
+}
+
+// GN: //src/tracing/v2:v2
+filegroup {
+ name: "perfetto_src_tracing_v2_v2",
+ srcs: [
+ "src/tracing/v2/shared_ring_buffer.cc",
+ "src/tracing/v2/shared_ring_buffer_reader.cc",
+ "src/tracing/v2/shared_ring_buffer_writer.cc",
+ ],
+}
+
// GN: //test:integrationtest_initializer
filegroup {
name: "perfetto_test_integrationtest_initializer",
@@ -24412,6 +24434,8 @@
":perfetto_src_tracing_system_backend",
":perfetto_src_tracing_test_test_support",
":perfetto_src_tracing_unittests",
+ ":perfetto_src_tracing_v2_unittests",
+ ":perfetto_src_tracing_v2_v2",
":perfetto_test_sanitizers_unittests",
],
shared_libs: [
diff --git a/gn/perfetto_unittests.gni b/gn/perfetto_unittests.gni
index e349a85..de56e67 100644
--- a/gn/perfetto_unittests.gni
+++ b/gn/perfetto_unittests.gni
@@ -38,6 +38,7 @@
_tracing_unittests_targets += [
"src/tracing/core:unittests",
"src/tracing/service:unittests",
+ "src/tracing/v2:unittests",
"src/tracing:unittests",
]
diff --git a/src/tracing/v2/BUILD.gn b/src/tracing/v2/BUILD.gn
new file mode 100644
index 0000000..de2894d
--- /dev/null
+++ b/src/tracing/v2/BUILD.gn
@@ -0,0 +1,65 @@
+# Copyright (C) 2026 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.
+
+import("../../../gn/test.gni")
+
+# The tracing v2 producer-local shared ring: its ABI, the writer half of the
+# chunk protocol and the single-consumer reader. It does not interpret fragment
+# payloads.
+source_set("v2") {
+ public_deps = [
+ # shared_ring_buffer.h holds a base::PagedMemory by value.
+ "../../../include/perfetto/ext/base",
+
+ # The headers use WriterID and BufferID.
+ "../../../include/perfetto/ext/tracing/core",
+
+ # tracing_v2_abi.h uses the public varint helpers.
+ "../../../include/perfetto/public:protozero",
+
+ # shared_ring_buffer_writer.h takes a BufferExhaustedPolicy.
+ "../../../include/perfetto/tracing",
+ ]
+ deps = [
+ "../../../gn:default_deps",
+ "../../base",
+ ]
+ sources = [
+ "shared_ring_buffer.cc",
+ "shared_ring_buffer.h",
+ "shared_ring_buffer_reader.cc",
+ "shared_ring_buffer_reader.h",
+ "shared_ring_buffer_writer.cc",
+ "shared_ring_buffer_writer.h",
+ "tracing_v2_abi.h",
+ ]
+}
+
+perfetto_unittest_source_set("unittests") {
+ testonly = true
+ deps = [
+ ":v2",
+ "../../../gn:default_deps",
+ "../../../gn:gtest_and_gmock",
+ "../../base",
+ "../../base:test_support",
+ ]
+ sources = [
+ "shared_ring_buffer_concurrency_unittest.cc",
+ "shared_ring_buffer_reader_unittest.cc",
+ "shared_ring_buffer_unittest.cc",
+ "shared_ring_buffer_writer_unittest.cc",
+ "tracing_v2_abi_unittest.cc",
+ ]
+}
diff --git a/src/tracing/v2/shared_ring_buffer.cc b/src/tracing/v2/shared_ring_buffer.cc
new file mode 100644
index 0000000..535962b
--- /dev/null
+++ b/src/tracing/v2/shared_ring_buffer.cc
@@ -0,0 +1,494 @@
+/*
+ * Copyright (C) 2026 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/v2/shared_ring_buffer.h"
+
+#include <errno.h>
+#include <stdint.h>
+
+#include <atomic>
+#include <memory>
+#include <optional>
+#include <utility>
+
+#include "perfetto/base/build_config.h"
+#include "perfetto/base/compiler.h"
+#include "perfetto/base/logging.h"
+#include "perfetto/ext/base/paged_memory.h"
+#include "perfetto/ext/base/utils.h"
+#include "src/tracing/v2/tracing_v2_abi.h"
+
+#if PERFETTO_BUILDFLAG(PERFETTO_OS_LINUX_BUT_NOT_QNX) || \
+ PERFETTO_BUILDFLAG(PERFETTO_OS_ANDROID)
+#define PERFETTO_TRACING_V2_HAS_FUTEX() 1
+#else
+#define PERFETTO_TRACING_V2_HAS_FUTEX() 0
+#endif
+
+#if PERFETTO_TRACING_V2_HAS_FUTEX()
+#include <linux/futex.h>
+#include <sys/syscall.h>
+#include <time.h>
+#include <unistd.h>
+#endif
+
+namespace perfetto::tracing_v2 {
+namespace {
+
+// PagedMemory adds two guard pages and rounds the mapping up to a page.
+std::optional<size_t> ComputeAllocationSize(uint32_t num_chunks,
+ uint32_t chunk_size,
+ size_t page_size) {
+ const uint64_t ring_size = uint64_t{sizeof(RingBufferHeader)} +
+ uint64_t{num_chunks} * uint64_t{chunk_size};
+ if (ring_size > SIZE_MAX || !base::IsPowerOfTwo(page_size))
+ return std::nullopt;
+
+ const size_t allocation_size = static_cast<size_t>(ring_size);
+
+ // PagedMemory rounds the request up and adds two guard pages using unchecked
+ // size_t arithmetic. Reject sizes that would overflow either operation.
+ const size_t page_rounding = page_size - 1;
+ if (allocation_size > SIZE_MAX - page_rounding)
+ return std::nullopt;
+ const size_t rounded_allocation_size =
+ (allocation_size + page_rounding) & ~page_rounding;
+
+ if (page_size > SIZE_MAX / 2)
+ return std::nullopt;
+ const size_t guard_pages_size = page_size * 2;
+ if (rounded_allocation_size > SIZE_MAX - guard_pages_size)
+ return std::nullopt;
+
+ return allocation_size;
+}
+
+#if PERFETTO_TRACING_V2_HAS_FUTEX()
+// A futex operation acts on one aligned 32-bit word. The word writers park on
+// is the low half of rw_positions. This pointer is passed only to the kernel;
+// C++ accesses the field as atomic<uint64_t>.
+uint32_t* ReadPosFutexWord(std::atomic<uint64_t>* rw_positions) {
+ static_assert(PERFETTO_IS_LITTLE_ENDIAN(),
+ "The low-word futex requires read_pos to be the first four "
+ "bytes of rw_positions");
+ return reinterpret_cast<uint32_t*>(rw_positions);
+}
+
+// TODO(sashwinbalaji): Drop *_PRIVATE when the reader moves into traced.
+// Private futexes are keyed by process, so a traced wake would not match a
+// producer wait.
+int FutexSyscall(uint32_t* word,
+ int op,
+ uint32_t value,
+ const struct timespec* timeout) {
+ return static_cast<int>(
+ syscall(SYS_futex, word, op, value, timeout, nullptr, 0));
+}
+#endif // PERFETTO_TRACING_V2_HAS_FUTEX()
+
+SharedRingBuffer::WriterWaitResult ClassifyWriterWaitErrno(int wait_errno) {
+ switch (wait_errno) {
+ case ETIMEDOUT:
+ return SharedRingBuffer::WriterWaitResult::kTimedOut;
+ case EAGAIN:
+ case EINTR:
+ return SharedRingBuffer::WriterWaitResult::kRetry;
+ default:
+ return SharedRingBuffer::WriterWaitResult::kUnavailable;
+ }
+}
+
+} // namespace
+
+// --- Allocation and lifetime. ---
+
+// static
+std::unique_ptr<SharedRingBuffer> SharedRingBuffer::Create(
+ uint32_t num_chunks,
+ uint32_t chunk_size) {
+ if (!base::IsPowerOfTwo(num_chunks) || num_chunks > kMaxChunksPerRing ||
+ chunk_size < kMinChunkSize || chunk_size % kChunkAlignmentBytes != 0) {
+ return nullptr;
+ }
+
+ const std::optional<size_t> allocation_size =
+ ComputeAllocationSize(num_chunks, chunk_size, base::GetSysPageSize());
+ if (!allocation_size)
+ return nullptr;
+
+ // PagedMemory is zero-filled. Zero is both (write_pos=0, read_pos=0) and
+ // Free(0), so no per-chunk initialization pass is needed.
+ auto ring_memory = base::PagedMemory::Allocate(*allocation_size,
+ base::PagedMemory::kMayFail);
+ if (!ring_memory.IsValid())
+ return nullptr;
+
+ return std::unique_ptr<SharedRingBuffer>(
+ new SharedRingBuffer(std::move(ring_memory), num_chunks, chunk_size));
+}
+
+SharedRingBuffer::SharedRingBuffer(base::PagedMemory ring_memory,
+ uint32_t num_chunks,
+ uint32_t chunk_size)
+ : ring_memory_(std::move(ring_memory)),
+ chunks_begin_(static_cast<uint8_t*>(ring_memory_.Get()) +
+ sizeof(RingBufferHeader)),
+ num_chunks_(num_chunks),
+ chunk_size_(chunk_size),
+ chunk_index_bits_(GetChunkIndexBits(num_chunks)) {}
+
+SharedRingBuffer::~SharedRingBuffer() = default;
+
+// --- Writer-side reservation. ---
+
+SharedRingBuffer::Reservation SharedRingBuffer::TryReserveWritePos() {
+ // PublishReadPos() releases after reclaiming chunks. If this load sees a new
+ // read_pos, acquire also sees those reclaims. If the load is stale, the
+ // compare-and-swap below fails unless the complete rw_positions word is
+ // still current.
+ return TryReserveWritePosImpl(
+ header()->rw_positions.load(std::memory_order_acquire));
+}
+
+SharedRingBuffer::Reservation SharedRingBuffer::TryReserveWritePosImpl(
+ uint64_t expected_rw_positions) {
+ RingBufferHeader* ring_header = header();
+ Reservation reservation{};
+ for (;;) {
+ const uint32_t write_pos = WritePosOf(expected_rw_positions);
+ const uint32_t read_pos = ReadPosOf(expected_rw_positions);
+ reservation.read_pos_sample = read_pos;
+
+ if (NumOutstandingPositions(write_pos, read_pos) >= num_chunks_) {
+ reservation.result = ReserveResult::kFull;
+ return reservation;
+ }
+
+ // Acquire pairs with the reader publishing newly reclaimed capacity.
+ if (ring_header->rw_positions.compare_exchange_weak(
+ expected_rw_positions, PackRwPositions(write_pos + 1, read_pos),
+ std::memory_order_acquire, std::memory_order_acquire)) {
+ reservation.result = ReserveResult::kReserved;
+ reservation.position = write_pos;
+ return reservation;
+ }
+ // Failure updates |expected_rw_positions|. No reservation, and therefore
+ // no hole, was created.
+ }
+}
+
+// --- Writer-side chunk transitions. ---
+
+bool SharedRingBuffer::TryAcquireChunkForWriting(uint32_t position,
+ uint32_t being_written_word) {
+ PERFETTO_DCHECK(ChunkStateOf(being_written_word) ==
+ ChunkState::kBeingWritten);
+ PERFETTO_DCHECK(NumFragmentsOf(being_written_word) == 0);
+
+ // A reservation authorizes a claim against this exact Free word.
+ uint32_t expected =
+ MakeFreeStateWord(WrapCountForPosition(position, chunk_index_bits_));
+
+ // Acquire on success consumes the reader's release reclaim, so the reader's
+ // last reads of the previous traversal's payload happen before this writer's
+ // first store into the chunk. On failure this position is left unclaimed;
+ // the caller does not use the word returned in |expected|.
+ return chunk_state_word_at(ChunkIndexOfPosition(position, num_chunks_))
+ ->compare_exchange_strong(expected, being_written_word,
+ std::memory_order_acquire,
+ std::memory_order_relaxed);
+}
+
+bool SharedRingBuffer::TrySetChunkComplete(uint32_t chunk_index,
+ uint32_t* observed,
+ uint32_t complete_word) {
+ PERFETTO_DCHECK(ChunkStateOf(*observed) == ChunkState::kBeingWritten);
+ PERFETTO_DCHECK(ChunkStateOf(complete_word) == ChunkState::kComplete);
+
+ // Release publishes the new fragments, their size varints and, on the first
+ // publication, the target BufferID. On failure the reader has marked the
+ // chunk, and acquire orders the reader's completed copy ahead of this
+ // writer's relocation work.
+ return chunk_state_word_at(chunk_index)
+ ->compare_exchange_strong(*observed, complete_word,
+ std::memory_order_release,
+ std::memory_order_acquire);
+}
+
+bool SharedRingBuffer::TryReacquireChunkForWriting(uint32_t chunk_index,
+ uint32_t observed) {
+ PERFETTO_DCHECK(ChunkStateOf(observed) == ChunkState::kComplete);
+
+ uint32_t expected = observed;
+ // This relaxed read-modify-write remains in the release sequence started by
+ // TrySetChunkComplete(). A reader that acquire-loads the resulting
+ // BeingWritten word therefore sees the prefix already published by this
+ // writer. On failure the writer drops its cached handle and does not use the
+ // returned state.
+ return chunk_state_word_at(chunk_index)
+ ->compare_exchange_strong(
+ expected, ReplaceChunkState(observed, ChunkState::kBeingWritten),
+ std::memory_order_relaxed, std::memory_order_relaxed);
+}
+
+bool SharedRingBuffer::TryAcknowledgeRewrite(uint32_t chunk_index,
+ uint32_t observed) {
+ PERFETTO_DCHECK(ChunkStateOf(observed) == ChunkState::kRewriteRequested);
+
+ uint32_t expected = observed;
+ // Release tells the reader that the writer has finished all accesses to the
+ // old chunk. Only this writer may leave RewriteRequested, so failure is a
+ // protocol error.
+ return chunk_state_word_at(chunk_index)
+ ->compare_exchange_strong(expected, kRewriteAcknowledgedStateWord,
+ std::memory_order_release,
+ std::memory_order_relaxed);
+}
+
+// --- Reader-side chunk transitions. ---
+
+uint32_t SharedRingBuffer::LoadChunkStateWord(uint32_t chunk_index) const {
+ // Pairs with every writer release transition, so each published fragment and
+ // its size varint is visible before the reader walks and copies them.
+ return chunk_state_word_at(chunk_index)->load(std::memory_order_acquire);
+}
+
+uint32_t SharedRingBuffer::LoadWritePos() const {
+ // Once a position is observed, the chunk word provides payload visibility.
+ return WritePosOf(header()->rw_positions.load(std::memory_order_relaxed));
+}
+
+bool SharedRingBuffer::TryRequestRewrite(uint32_t chunk_index,
+ uint32_t* observed) {
+ PERFETTO_DCHECK(ChunkStateOf(*observed) == ChunkState::kBeingWritten);
+
+ // On success, release finishes the reader's copy before the writer sees the
+ // rewrite request and relocates its suffix. On failure, the reader discards
+ // its copy and retries the position with a new acquire load.
+ return chunk_state_word_at(chunk_index)
+ ->compare_exchange_strong(
+ *observed,
+ ReplaceChunkState(*observed, ChunkState::kRewriteRequested),
+ std::memory_order_release, std::memory_order_relaxed);
+}
+
+bool SharedRingBuffer::TryMoveFreeChunkToNextWrap(uint32_t position,
+ uint32_t* observed) {
+ PERFETTO_DCHECK(ChunkStateOf(*observed) == ChunkState::kFree);
+
+ // Release finishes the reader's accesses before the next writer claims the
+ // chunk. On failure, the reader retries the position with a new acquire
+ // load.
+ return chunk_state_word_at(ChunkIndexOfPosition(position, num_chunks_))
+ ->compare_exchange_strong(*observed, MakeFreeWordForNextWrap(position),
+ std::memory_order_release,
+ std::memory_order_relaxed);
+}
+
+bool SharedRingBuffer::TryReleaseCompleteChunkAsFree(uint32_t position,
+ uint32_t* observed) {
+ PERFETTO_DCHECK(ChunkStateOf(*observed) == ChunkState::kComplete);
+
+ // Release finishes the reader's copy before the next writer uses the chunk.
+ // On failure, the reader discards its copy and retries the position with a
+ // new acquire load.
+ return chunk_state_word_at(ChunkIndexOfPosition(position, num_chunks_))
+ ->compare_exchange_strong(*observed, MakeFreeWordForNextWrap(position),
+ std::memory_order_release,
+ std::memory_order_relaxed);
+}
+
+bool SharedRingBuffer::TryReleaseRewriteAcknowledgedChunkAsFree(
+ uint32_t position,
+ uint32_t* observed) {
+ // RewriteAcknowledged has one canonical encoding. Compare against that exact
+ // word rather than trusting a value that merely decodes to the same state.
+ uint32_t expected = kRewriteAcknowledgedStateWord;
+ // Acquire observes the writer's final release. Release hands the chunk to the
+ // next writer. Only the reader may leave RewriteAcknowledged, so failure is a
+ // protocol error.
+ const bool reclaimed =
+ chunk_state_word_at(ChunkIndexOfPosition(position, num_chunks_))
+ ->compare_exchange_strong(expected, MakeFreeWordForNextWrap(position),
+ std::memory_order_acq_rel,
+ std::memory_order_relaxed);
+ if (!reclaimed)
+ *observed = expected;
+ return reclaimed;
+}
+
+// --- Backpressure. ---
+
+// static
+bool SharedRingBuffer::SupportsWriterWait() {
+ return PERFETTO_TRACING_V2_HAS_FUTEX();
+}
+
+SharedRingBuffer::WriterWaitResult SharedRingBuffer::WaitForReadPosChange(
+ uint32_t expected_read_pos,
+ uint32_t timeout_ms) {
+ PERFETTO_DCHECK(timeout_ms > 0);
+#if !PERFETTO_TRACING_V2_HAS_FUTEX()
+ base::ignore_result(expected_read_pos);
+ base::ignore_result(timeout_ms);
+ return WriterWaitResult::kUnavailable;
+#else
+ RingBufferHeader* ring_header = header();
+
+ // num_writers_waiting and rw_positions are separate atomics. Without the two
+ // seq_cst fences, this execution would be possible:
+ //
+ // writer reader
+ // ------ ------
+ // increment num_writers_waiting publish read_pos
+ // read the old read_pos read zero waiters
+ //
+ // The writer would sleep after the reader skipped the wake. The paired
+ // fences forbid both loads from missing the other side's store. The waiter
+ // count carries no data, so the operations around the fences stay relaxed.
+ ring_header->num_writers_waiting.fetch_add(1, std::memory_order_relaxed);
+ std::atomic_thread_fence(std::memory_order_seq_cst);
+
+ WriterWaitResult result = WriterWaitResult::kRetry;
+ if (ReadPosOf(ring_header->rw_positions.load(std::memory_order_relaxed)) ==
+ expected_read_pos) {
+ struct timespec timeout{};
+ timeout.tv_sec = static_cast<time_t>(timeout_ms / 1000);
+ timeout.tv_nsec = static_cast<long>((timeout_ms % 1000) * 1000000);
+
+ // FUTEX_WAIT checks the value again before sleeping. If read_pos changed
+ // after the load above, the syscall returns EAGAIN and no wake is lost.
+ const int futex_result =
+ FutexSyscall(ReadPosFutexWord(&ring_header->rw_positions),
+ FUTEX_WAIT_PRIVATE, expected_read_pos, &timeout);
+ if (futex_result != 0) {
+ const int wait_errno = errno;
+ result = ClassifyWriterWaitErrno(wait_errno);
+ if (result == WriterWaitResult::kUnavailable) {
+ errno = wait_errno;
+ PERFETTO_DPLOG("tracing v2: futex wait on read_pos failed");
+ }
+ }
+ }
+
+ ring_header->num_writers_waiting.fetch_sub(1, std::memory_order_relaxed);
+ return result;
+#endif // PERFETTO_TRACING_V2_HAS_FUTEX()
+}
+
+void SharedRingBuffer::PublishReadPos(uint32_t read_pos) {
+ // The initial load is relaxed: it is only the first expected value for the
+ // compare-and-swap below.
+ PublishReadPosImpl(header()->rw_positions.load(std::memory_order_relaxed),
+ read_pos);
+}
+
+void SharedRingBuffer::PublishReadPosImpl(uint64_t expected_rw_positions,
+ uint32_t read_pos) {
+ RingBufferHeader* ring_header = header();
+
+ // Only the reader changes read_pos, so this can lose only to a writer moving
+ // write_pos. Release publishes the chunk reclaims before their capacity.
+ // Failure only supplies the newer write_pos to preserve on the next attempt.
+ while (!ring_header->rw_positions.compare_exchange_weak(
+ expected_rw_positions, ReplaceReadPos(expected_rw_positions, read_pos),
+ std::memory_order_release, std::memory_order_relaxed)) {
+ }
+
+#if PERFETTO_TRACING_V2_HAS_FUTEX()
+ // Pairs with the fence in WaitForReadPosChange(); see the missed-wake
+ // schedule there.
+ std::atomic_thread_fence(std::memory_order_seq_cst);
+
+ if (ring_header->num_writers_waiting.load(std::memory_order_relaxed) == 0)
+ return;
+
+ // Wake everyone: one drain pass can free many chunks, so waking a single
+ // waiter would leave capacity unused. This runs once per pass, not once per
+ // reclaimed chunk.
+ //
+ // Waits are bounded, so a failed wake delays writers but cannot strand them.
+ if (FutexSyscall(ReadPosFutexWord(&ring_header->rw_positions),
+ FUTEX_WAKE_PRIVATE, static_cast<uint32_t>(INT32_MAX),
+ nullptr) < 0) {
+ PERFETTO_DPLOG("tracing v2: futex wake on read_pos failed");
+ }
+#endif // PERFETTO_TRACING_V2_HAS_FUTEX()
+}
+
+// --- Testing. ---
+
+// static
+std::optional<size_t> SharedRingBuffer::ComputeAllocationSizeForTesting(
+ uint32_t num_chunks,
+ uint32_t chunk_size,
+ size_t page_size) {
+ return ComputeAllocationSize(num_chunks, chunk_size, page_size);
+}
+
+// static
+SharedRingBuffer::WriterWaitResult
+SharedRingBuffer::ClassifyWriterWaitErrnoForTesting(int wait_errno) {
+ return ClassifyWriterWaitErrno(wait_errno);
+}
+
+uint32_t SharedRingBuffer::num_writers_waiting_for_testing() const {
+ return header()->num_writers_waiting.load(std::memory_order_relaxed);
+}
+
+uint32_t SharedRingBuffer::read_pos_for_testing() const {
+ return ReadPosOf(header()->rw_positions.load(std::memory_order_relaxed));
+}
+
+SharedRingBuffer::Reservation SharedRingBuffer::TryReserveWritePosForTesting(
+ uint64_t initial_rw_positions) {
+ return TryReserveWritePosImpl(initial_rw_positions);
+}
+
+void SharedRingBuffer::PublishReadPosForTesting(uint64_t initial_rw_positions,
+ uint32_t read_pos) {
+ PublishReadPosImpl(initial_rw_positions, read_pos);
+}
+
+void SharedRingBuffer::SetStateWordForTesting(uint32_t chunk_index,
+ uint32_t state_word) {
+ chunk_state_word_at(chunk_index)
+ ->store(state_word, std::memory_order_release);
+}
+
+void SharedRingBuffer::SetWritePosForTesting(uint32_t write_pos) {
+ RingBufferHeader* ring_header = header();
+ const uint64_t rw_positions =
+ ring_header->rw_positions.load(std::memory_order_relaxed);
+ ring_header->rw_positions.store(ReplaceWritePos(rw_positions, write_pos),
+ std::memory_order_relaxed);
+}
+
+void SharedRingBuffer::SetPositionsForTesting(uint32_t position) {
+ for (uint32_t chunk_index = 0; chunk_index < num_chunks_; ++chunk_index) {
+ // The first position at or after |position| that maps to this chunk.
+ const uint32_t first_position =
+ position + ((chunk_index - position) & (num_chunks_ - 1));
+ chunk_state_word_at(chunk_index)
+ ->store(MakeFreeStateWord(
+ WrapCountForPosition(first_position, chunk_index_bits_)),
+ std::memory_order_relaxed);
+ }
+ header()->rw_positions.store(PackRwPositions(position, position),
+ std::memory_order_release);
+}
+
+} // namespace perfetto::tracing_v2
diff --git a/src/tracing/v2/shared_ring_buffer.h b/src/tracing/v2/shared_ring_buffer.h
new file mode 100644
index 0000000..875e730
--- /dev/null
+++ b/src/tracing/v2/shared_ring_buffer.h
@@ -0,0 +1,261 @@
+/*
+ * Copyright (C) 2026 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_V2_SHARED_RING_BUFFER_H_
+#define SRC_TRACING_V2_SHARED_RING_BUFFER_H_
+
+#include <stddef.h>
+#include <stdint.h>
+
+#include <atomic>
+#include <memory>
+#include <optional>
+
+#include "perfetto/base/logging.h"
+#include "perfetto/ext/base/paged_memory.h"
+#include "src/tracing/v2/tracing_v2_abi.h"
+
+namespace perfetto::tracing_v2 {
+
+// Owns the producer-local ring and implements its atomic transitions.
+//
+// SharedRingBufferWriter calls the writer-side methods concurrently.
+// SharedRingBufferReader is the only caller of the reader-side methods. This
+// is the only class that updates rw_positions or a chunk state word, and only
+// the reader writes Free.
+class SharedRingBuffer {
+ public:
+ // |num_chunks| must be a power of two in [1, 2^30]. |chunk_size| must be at
+ // least 256 bytes and a multiple of four. Returns nullptr if the parameters
+ // are invalid, the allocation size overflows, or the allocation fails.
+ static std::unique_ptr<SharedRingBuffer> Create(uint32_t num_chunks,
+ uint32_t chunk_size);
+
+ ~SharedRingBuffer();
+
+ SharedRingBuffer(const SharedRingBuffer&) = delete;
+ SharedRingBuffer& operator=(const SharedRingBuffer&) = delete;
+ SharedRingBuffer(SharedRingBuffer&&) = delete;
+ SharedRingBuffer& operator=(SharedRingBuffer&&) = delete;
+
+ // Immutable for the life of the ring.
+ uint32_t num_chunks() const { return num_chunks_; }
+ uint32_t chunk_size() const { return chunk_size_; }
+ uint32_t chunk_index_bits() const { return chunk_index_bits_; }
+
+ uint8_t* chunk_at(uint32_t chunk_index) {
+ PERFETTO_DCHECK(chunk_index < num_chunks_);
+ return chunks_begin_ + static_cast<size_t>(chunk_index) * chunk_size_;
+ }
+ const uint8_t* chunk_at(uint32_t chunk_index) const {
+ PERFETTO_DCHECK(chunk_index < num_chunks_);
+ return chunks_begin_ + static_cast<size_t>(chunk_index) * chunk_size_;
+ }
+
+ // Writer-side reservation.
+ //
+ // Reserving a position and acquiring its physical chunk are deliberately
+ // separate operations. TryReserveWritePos() advances write_pos first.
+ // The writer then changes Free(wrap_count(position)) to BeingWritten. If it
+ // is descheduled between the two, the reader resolves that position as an
+ // unclaimed hole and moves the Free word to the next wrap. The delayed
+ // writer's exact compare-and-swap then fails, so it cannot publish behind
+ // the reader.
+
+ enum class ReserveResult {
+ kReserved,
+ // num_chunks positions are already outstanding. Nothing was reserved and
+ // no hole was created; a stalling policy may wait and try again.
+ kFull,
+ };
+
+ struct Reservation {
+ ReserveResult result = ReserveResult::kFull;
+ // Valid only for kReserved. This is a position in the reservation order,
+ // not a physical chunk index.
+ uint32_t position = 0;
+ // The read_pos used for the capacity check. A writer that receives kFull
+ // passes this to WaitForReadPosChange() to avoid a lost wakeup.
+ uint32_t read_pos_sample = 0;
+ };
+
+ // Reserves the next position if the ring has room. A failed
+ // compare-and-swap did not reserve anything, so the operation retries with
+ // the positions returned by the compare-and-swap.
+ Reservation TryReserveWritePos();
+
+ // Writer-side chunk transitions.
+
+ // Free(wrap_count(position)) -> BeingWritten. |being_written_word| must be an
+ // BeingWritten word for this writer with zero fragments.
+ //
+ // If the compare-and-swap fails, this reservation becomes a hole. Do not
+ // retry it against the returned word; reserve a later position.
+ bool TryAcquireChunkForWriting(uint32_t position,
+ uint32_t being_written_word);
+
+ // BeingWritten -> Complete. On failure, |*observed| receives the current
+ // word. It must be RewriteRequested with the same contents; no other actor
+ // may change a chunk while this writer owns it.
+ bool TrySetChunkComplete(uint32_t chunk_index,
+ uint32_t* observed,
+ uint32_t complete_word);
+
+ // Complete -> BeingWritten, for a writer taking its own cached chunk back to
+ // append more fragments. Failure means the reader reclaimed it first; the
+ // writer just drops its handle.
+ bool TryReacquireChunkForWriting(uint32_t chunk_index, uint32_t observed);
+
+ // RewriteRequested -> RewriteAcknowledged after the writer has stopped
+ // touching the old chunk. Failure is a protocol error.
+ bool TryAcknowledgeRewrite(uint32_t chunk_index, uint32_t observed);
+
+ // Reader side.
+
+ // Returns the current chunk state word. A writer publishes fragments before
+ // changing this word, so the returned word also makes those fragments
+ // visible to the reader.
+ uint32_t LoadChunkStateWord(uint32_t chunk_index) const;
+
+ // Loads write_pos once. A writer may reserve the next position concurrently;
+ // the reader will see it on its next pass.
+ uint32_t LoadWritePos() const;
+
+ // BeingWritten -> RewriteRequested, passing format, flags, num_fragments and
+ // the WriterID through untouched. On failure |*observed| receives the word
+ // that won the race.
+ bool TryRequestRewrite(uint32_t chunk_index, uint32_t* observed);
+
+ // The following transitions are the only ones that expose a chunk to the
+ // next pass around the ring. The new wrap count comes from |position|, not
+ // from the old state word.
+
+ // Free(wrap_count(position)) -> Free(next_wrap(position)). This consumes a
+ // position whose writer never entered BeingWritten and prepares the chunk
+ // for position + num_chunks.
+ bool TryMoveFreeChunkToNextWrap(uint32_t position, uint32_t* observed);
+
+ // Complete -> Free(next_wrap(position)).
+ bool TryReleaseCompleteChunkAsFree(uint32_t position, uint32_t* observed);
+
+ // RewriteAcknowledged -> Free(next_wrap(position)). Failure is a protocol
+ // error and updates |*observed| with the unexpected word.
+ bool TryReleaseRewriteAcknowledgedChunkAsFree(uint32_t position,
+ uint32_t* observed);
+
+ // Backpressure: the writer's full-ring path.
+ //
+ // Writers wait on read_pos; the reader publishes a new value and wakes them.
+ // The waiter count only avoids an unnecessary wake syscall. Capacity is
+ // always decided from rw_positions.
+
+ enum class WriterWaitResult {
+ // The writer must recheck capacity. This also covers interrupted and
+ // spurious wakes.
+ kRetry,
+ kTimedOut,
+ // This build or kernel cannot provide the wait. Do not retry the syscall.
+ kUnavailable,
+ };
+
+ // Whether WaitForReadPosChange() is implemented on this platform.
+ static bool SupportsWriterWait();
+
+ // Blocks until read_pos changes or |timeout_ms| elapses.
+ // |expected_read_pos| must be the sample used for the kFull result.
+ WriterWaitResult WaitForReadPosChange(uint32_t expected_read_pos,
+ uint32_t timeout_ms);
+
+ // Publishes read_pos without overwriting a concurrent write_pos update, then
+ // wakes waiting writers. Called once per drain pass, so the shared read_pos
+ // can lag the reader's local value; it only under-reports free capacity.
+ void PublishReadPos(uint32_t read_pos);
+
+ // Testing.
+
+ // Exposes the allocation-size calculation for overflow tests.
+ static std::optional<size_t> ComputeAllocationSizeForTesting(
+ uint32_t num_chunks,
+ uint32_t chunk_size,
+ size_t page_size);
+
+ // Exposes the futex error policy without requiring a failing syscall.
+ static WriterWaitResult ClassifyWriterWaitErrnoForTesting(int wait_errno);
+
+ // Diagnostics only. Never a correctness input.
+ uint32_t num_writers_waiting_for_testing() const;
+ uint32_t read_pos_for_testing() const;
+
+ // Starts the production CAS loops with |initial_rw_positions|. Tests pass an
+ // old value to force the first compare-and-swap to fail.
+ Reservation TryReserveWritePosForTesting(uint64_t initial_rw_positions);
+ void PublishReadPosForTesting(uint64_t initial_rw_positions,
+ uint32_t read_pos);
+
+ // Injects a state word for corruption and unknown-ABI tests.
+ void SetStateWordForTesting(uint32_t chunk_index, uint32_t state_word);
+
+ // Injects a write_pos without reserving the intervening positions.
+ void SetWritePosForTesting(uint32_t write_pos);
+
+ // Seeds a valid ring state near a position or wrap-count rollover.
+ void SetPositionsForTesting(uint32_t position);
+
+ private:
+ SharedRingBuffer(base::PagedMemory ring_memory,
+ uint32_t num_chunks,
+ uint32_t chunk_size);
+
+ // CAS loops shared by the production entry points and deterministic race
+ // tests. Production starts them with a freshly loaded rw_positions value;
+ // tests can supply an older value to force the first CAS to fail.
+ Reservation TryReserveWritePosImpl(uint64_t expected_rw_positions);
+ void PublishReadPosImpl(uint64_t expected_rw_positions, uint32_t read_pos);
+
+ // Shared-memory address and wrap-count helpers.
+
+ std::atomic<uint32_t>* chunk_state_word_at(uint32_t chunk_index) {
+ return reinterpret_cast<std::atomic<uint32_t>*>(chunk_at(chunk_index));
+ }
+ const std::atomic<uint32_t>* chunk_state_word_at(uint32_t chunk_index) const {
+ return reinterpret_cast<const std::atomic<uint32_t>*>(
+ chunk_at(chunk_index));
+ }
+
+ // Returns the Free word for the next position that uses the same chunk.
+ uint32_t MakeFreeWordForNextWrap(uint32_t position) const {
+ // Deriving the value from the next position also handles uint32_t rollover.
+ return MakeFreeStateWord(
+ WrapCountForPosition(position + num_chunks_, chunk_index_bits_));
+ }
+
+ RingBufferHeader* header() {
+ return static_cast<RingBufferHeader*>(ring_memory_.Get());
+ }
+ const RingBufferHeader* header() const {
+ return static_cast<const RingBufferHeader*>(ring_memory_.Get());
+ }
+
+ base::PagedMemory ring_memory_;
+ uint8_t* const chunks_begin_;
+ const uint32_t num_chunks_;
+ const uint32_t chunk_size_;
+ const uint32_t chunk_index_bits_;
+};
+
+} // namespace perfetto::tracing_v2
+
+#endif // SRC_TRACING_V2_SHARED_RING_BUFFER_H_
diff --git a/src/tracing/v2/shared_ring_buffer_concurrency_unittest.cc b/src/tracing/v2/shared_ring_buffer_concurrency_unittest.cc
new file mode 100644
index 0000000..8fe6616
--- /dev/null
+++ b/src/tracing/v2/shared_ring_buffer_concurrency_unittest.cc
@@ -0,0 +1,721 @@
+/*
+ * Copyright (C) 2026 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.
+ */
+
+// The three shared-ring races, run for real on two threads, plus an
+// MPSC stress test that checks the end-to-end guarantee: every fragment that
+// was published comes out exactly once, in the order its writer wrote it,
+// except for the ones the writer itself accounted as dropped.
+//
+// These are the tests worth running under ThreadSanitizer. Nothing here uses a
+// sleep to make a race likely: the threads meet on a spin barrier, and each
+// iteration asserts that exactly one of the two contenders won.
+
+#include <stdint.h>
+#include <string.h>
+
+#include <atomic>
+#include <map>
+#include <memory>
+#include <string>
+#include <thread>
+#include <vector>
+
+#include "perfetto/base/build_config.h"
+#include "perfetto/base/time.h"
+#include "perfetto/ext/base/no_destructor.h"
+#include "perfetto/ext/base/utils.h"
+#include "perfetto/tracing/buffer_exhausted_policy.h"
+#include "src/tracing/v2/shared_ring_buffer.h"
+#include "src/tracing/v2/shared_ring_buffer_reader.h"
+#include "src/tracing/v2/shared_ring_buffer_writer.h"
+#include "src/tracing/v2/tracing_v2_abi.h"
+#include "test/gtest_and_gmock.h"
+
+namespace perfetto::tracing_v2 {
+namespace {
+
+constexpr WriterID kWriterA = 7;
+constexpr WriterID kWriterB = 8;
+constexpr BufferID kBuffer = 5;
+
+class NoopSharedRingBufferWriterDelegate
+ : public SharedRingBufferWriter::Delegate {
+ public:
+ void NotifyReader() override {}
+};
+
+SharedRingBufferWriter::Delegate* GetNoopSharedRingBufferWriterDelegate() {
+ static base::NoDestructor<NoopSharedRingBufferWriterDelegate> delegate;
+ return &delegate.ref();
+}
+
+uint32_t BeingWrittenWord(WriterID writer) {
+ return MakeDataStateWord(ChunkState::kBeingWritten,
+ ChunkFormat::kTargetBuffer, 0, 0, writer);
+}
+
+// Sequences two threads through a fixed schedule. Each round has three steps:
+// the reader thread prepares the chunk, then the two contenders take their
+// compare-and-swap turns in an order that alternates every round. That makes
+// *both* winners of a two-party race reachable on demand instead of depending
+// on which thread happens to leave a barrier first - which, with a real
+// barrier, is always the same one.
+//
+// The atomics under test still cross a genuine thread boundary, which is what
+// makes these worth running under ThreadSanitizer.
+//
+// Every wait is bounded. When a step never arrives - the other thread failed
+// an assertion, timed out, or returned early - the waiter reports one failure
+// naming the missing step and aborts the schedule, which unblocks every other
+// waiter so the test can join its threads instead of spinning forever.
+class StepSequencer {
+ public:
+ static constexpr uint32_t kStepsPerRound = 3;
+
+ // Returns false when the schedule was aborted, either here on a timeout or
+ // by anybody else. The caller must stop sequencing; the failure has already
+ // been reported by whoever aborted first.
+ bool WaitForStep(uint64_t step) {
+ const base::TimeMillis deadline =
+ base::GetWallTimeMs() + base::TimeMillis(30000);
+ for (;;) {
+ if (aborted_.load(std::memory_order_acquire))
+ return false;
+ if (step_.load(std::memory_order_acquire) == step)
+ return true;
+ if (base::GetWallTimeMs() >= deadline) {
+ ADD_FAILURE() << "Timed out waiting for sequencer step " << step
+ << " (round " << step / kStepsPerRound
+ << "); the other thread stopped advancing the schedule";
+ Abort();
+ return false;
+ }
+ std::this_thread::yield();
+ }
+ }
+ void FinishStep(uint64_t step) {
+ step_.store(step + 1, std::memory_order_release);
+ }
+ // Unblocks every WaitForStep(), now and in the future. Harmless once the
+ // schedule has completed, so the tests below call it unconditionally before
+ // joining their threads.
+ void Abort() { aborted_.store(true, std::memory_order_release); }
+
+ // Step at which |round|'s preparation happens.
+ static uint64_t PrepareStep(uint32_t round) {
+ return uint64_t{round} * kStepsPerRound;
+ }
+ // Step at which the reader or the writer takes its turn in |round|. The
+ // reader goes first on even rounds and second on odd ones.
+ static uint64_t ReaderStep(uint32_t round) {
+ return PrepareStep(round) + (round % 2 == 0 ? 1 : 2);
+ }
+ static uint64_t WriterStep(uint32_t round) {
+ return PrepareStep(round) + (round % 2 == 0 ? 2 : 1);
+ }
+
+ private:
+ std::atomic<uint64_t> step_{0};
+ std::atomic<bool> aborted_{false};
+};
+
+// ---------------------------------------------------------------------------
+// Race 1: a writer claims while the reader consumes an unclaimed position.
+// Both compare against Free(wrap_count(p)).
+// ---------------------------------------------------------------------------
+
+TEST(SharedRingBufferConcurrencyTest, ClaimVersusUnclaimedAdvance) {
+ constexpr uint32_t kRounds = 1000;
+ constexpr uint32_t kNumChunks = 2;
+ auto ring = SharedRingBuffer::Create(kNumChunks, 256);
+ ASSERT_NE(ring, nullptr);
+
+ StepSequencer sequencer;
+ std::atomic<uint32_t> claims_won{0};
+ std::atomic<uint32_t> advances_won{0};
+ // Round r resolves position r, so which physical chunk and which wrap count
+ // are in play changes every round.
+ std::thread writer([&] {
+ for (uint32_t round = 0; round < kRounds; ++round) {
+ const uint64_t step = StepSequencer::WriterStep(round);
+ if (!sequencer.WaitForStep(step))
+ return;
+ if (ring->TryAcquireChunkForWriting(round, BeingWrittenWord(kWriterA)))
+ claims_won.fetch_add(1, std::memory_order_relaxed);
+ sequencer.FinishStep(step);
+ }
+ });
+ // Also runs when an ASSERT below returns early: the abort unblocks the
+ // writer, so the join cannot hang and no thread outlives the test.
+ auto join_writer = base::OnScopeExit([&] {
+ sequencer.Abort();
+ if (writer.joinable())
+ writer.join();
+ });
+
+ for (uint32_t round = 0; round < kRounds; ++round) {
+ const uint32_t position = round;
+ const uint32_t chunk_index = ChunkIndexOfPosition(position, kNumChunks);
+
+ if (!sequencer.WaitForStep(StepSequencer::PrepareStep(round)))
+ return;
+ uint32_t observed = ring->LoadChunkStateWord(chunk_index);
+ ASSERT_EQ(ChunkStateOf(observed), ChunkState::kFree) << round;
+ sequencer.FinishStep(StepSequencer::PrepareStep(round));
+
+ const uint64_t step = StepSequencer::ReaderStep(round);
+ if (!sequencer.WaitForStep(step))
+ return;
+ const bool advanced = ChunkStateOf(observed) == ChunkState::kFree &&
+ ring->TryMoveFreeChunkToNextWrap(position, &observed);
+ if (advanced)
+ advances_won.fetch_add(1, std::memory_order_relaxed);
+ sequencer.FinishStep(step);
+
+ if (!sequencer.WaitForStep(StepSequencer::PrepareStep(round + 1)))
+ return;
+ const uint32_t after = ring->LoadChunkStateWord(chunk_index);
+ if (advanced) {
+ // The reader consumed the hole and prepared the chunk for the writer
+ // holding position + num_chunks. The writer that lost holds |position|
+ // and has spent its one claim attempt.
+ ASSERT_EQ(ChunkStateOf(after), ChunkState::kFree) << round;
+ ASSERT_EQ(
+ WrapCountOf(after),
+ WrapCountForPosition(position + kNumChunks, ring->chunk_index_bits()))
+ << round;
+ ASSERT_FALSE(
+ ring->TryAcquireChunkForWriting(position, BeingWrittenWord(kWriterB)))
+ << round;
+ } else {
+ ASSERT_EQ(ChunkStateOf(after), ChunkState::kBeingWritten) << round;
+ // Put the chunk back the way the writer's publication and the reader's
+ // reclaim would, so the next round starts from a free word again.
+ uint32_t being_written = BeingWrittenWord(kWriterA);
+ const uint32_t complete = MakeDataStateWord(
+ ChunkState::kComplete, ChunkFormat::kTargetBuffer, 0, 0, kWriterA);
+ ASSERT_TRUE(
+ ring->TrySetChunkComplete(chunk_index, &being_written, complete))
+ << round;
+ uint32_t to_reclaim = complete;
+ ASSERT_TRUE(ring->TryReleaseCompleteChunkAsFree(position, &to_reclaim))
+ << round;
+ }
+ }
+ writer.join();
+
+ // Exactly one contender wins every round, and the schedule reaches both.
+ EXPECT_EQ(claims_won.load() + advances_won.load(), kRounds);
+ EXPECT_EQ(claims_won.load(), kRounds / 2);
+ EXPECT_EQ(advances_won.load(), kRounds / 2);
+}
+
+// ---------------------------------------------------------------------------
+// Race 2: a writer publishes while the reader scrapes.
+// Both compare against BeingWritten(w,n).
+// ---------------------------------------------------------------------------
+
+TEST(SharedRingBufferConcurrencyTest, PublishVersusScrape) {
+ constexpr uint32_t kRounds = 1000;
+ auto ring = SharedRingBuffer::Create(2, 256);
+ ASSERT_NE(ring, nullptr);
+
+ StepSequencer sequencer;
+ std::atomic<uint32_t> publishes_won{0};
+ std::atomic<uint32_t> scrapes_won{0};
+ const uint32_t kComplete = MakeDataStateWord(
+ ChunkState::kComplete, ChunkFormat::kTargetBuffer, 0, 1, kWriterA);
+
+ std::thread writer([&] {
+ for (uint32_t round = 0; round < kRounds; ++round) {
+ const uint64_t step = StepSequencer::WriterStep(round);
+ if (!sequencer.WaitForStep(step))
+ return;
+ uint32_t observed = BeingWrittenWord(kWriterA);
+ if (ring->TrySetChunkComplete(0, &observed, kComplete)) {
+ publishes_won.fetch_add(1, std::memory_order_relaxed);
+ } else {
+ // The reader marking the chunk is the only legal way to lose, and the
+ // marked count says exactly which prefix it took.
+ EXPECT_EQ(ChunkStateOf(observed), ChunkState::kRewriteRequested);
+ EXPECT_EQ(WriterIdOf(observed), kWriterA);
+ EXPECT_EQ(NumFragmentsOf(observed), 0u);
+ EXPECT_TRUE(ring->TryAcknowledgeRewrite(0, observed));
+ }
+ sequencer.FinishStep(step);
+ }
+ });
+ auto join_writer = base::OnScopeExit([&] {
+ sequencer.Abort();
+ if (writer.joinable())
+ writer.join();
+ });
+
+ for (uint32_t round = 0; round < kRounds; ++round) {
+ if (!sequencer.WaitForStep(StepSequencer::PrepareStep(round)))
+ return;
+ ring->SetStateWordForTesting(0, BeingWrittenWord(kWriterA));
+ uint32_t observed = ring->LoadChunkStateWord(0);
+ sequencer.FinishStep(StepSequencer::PrepareStep(round));
+
+ const uint64_t step = StepSequencer::ReaderStep(round);
+ if (!sequencer.WaitForStep(step))
+ return;
+ const bool marked = ring->TryRequestRewrite(0, &observed);
+ if (marked) {
+ scrapes_won.fetch_add(1, std::memory_order_relaxed);
+ } else {
+ // The writer completed, so the reader discards its speculative copy and
+ // retries the position.
+ ASSERT_EQ(ChunkStateOf(observed), ChunkState::kComplete) << round;
+ ASSERT_EQ(NumFragmentsOf(observed), 1u) << round;
+ }
+ sequencer.FinishStep(step);
+
+ if (!sequencer.WaitForStep(StepSequencer::PrepareStep(round + 1)))
+ return;
+ // A marked chunk comes back as RewriteAcknowledged: the writer released it
+ // and said nothing about who gets it next.
+ ASSERT_EQ(ring->LoadChunkStateWord(0),
+ marked ? kRewriteAcknowledgedStateWord : kComplete)
+ << round;
+ }
+ writer.join();
+
+ EXPECT_EQ(publishes_won.load() + scrapes_won.load(), kRounds);
+ EXPECT_EQ(publishes_won.load(), kRounds / 2);
+ EXPECT_EQ(scrapes_won.load(), kRounds / 2);
+}
+
+// ---------------------------------------------------------------------------
+// Race 3: a writer reuses while the reader consumes.
+// Both compare against Complete(w,n).
+// ---------------------------------------------------------------------------
+
+TEST(SharedRingBufferConcurrencyTest, ReuseVersusReclaim) {
+ constexpr uint32_t kRounds = 1000;
+ constexpr uint32_t kNumChunks = 2;
+ auto ring = SharedRingBuffer::Create(kNumChunks, 256);
+ ASSERT_NE(ring, nullptr);
+
+ StepSequencer sequencer;
+ std::atomic<uint32_t> reuses_won{0};
+ std::atomic<uint32_t> reclaims_won{0};
+ const uint32_t kComplete = MakeDataStateWord(
+ ChunkState::kComplete, ChunkFormat::kTargetBuffer, 0, 1, kWriterA);
+
+ std::thread writer([&] {
+ for (uint32_t round = 0; round < kRounds; ++round) {
+ const uint64_t step = StepSequencer::WriterStep(round);
+ if (!sequencer.WaitForStep(step))
+ return;
+ if (ring->TryReacquireChunkForWriting(0, kComplete))
+ reuses_won.fetch_add(1, std::memory_order_relaxed);
+ sequencer.FinishStep(step);
+ }
+ });
+ auto join_writer = base::OnScopeExit([&] {
+ sequencer.Abort();
+ if (writer.joinable())
+ writer.join();
+ });
+
+ for (uint32_t round = 0; round < kRounds; ++round) {
+ if (!sequencer.WaitForStep(StepSequencer::PrepareStep(round)))
+ return;
+ ring->SetStateWordForTesting(0, kComplete);
+ uint32_t observed = ring->LoadChunkStateWord(0);
+ sequencer.FinishStep(StepSequencer::PrepareStep(round));
+
+ const uint64_t step = StepSequencer::ReaderStep(round);
+ if (!sequencer.WaitForStep(step))
+ return;
+ // Position 0 every round, so the reclaimed word is always Free of
+ // the wrap after position 0's.
+ const bool reclaimed = ring->TryReleaseCompleteChunkAsFree(0, &observed);
+ if (reclaimed) {
+ reclaims_won.fetch_add(1, std::memory_order_relaxed);
+ } else {
+ // The writer took the chunk back; the reader handles that on its next
+ // look, through the scrape path.
+ ASSERT_EQ(ChunkStateOf(observed), ChunkState::kBeingWritten) << round;
+ ASSERT_EQ(NumFragmentsOf(observed), 1u) << round;
+ }
+ sequencer.FinishStep(step);
+ if (!sequencer.WaitForStep(StepSequencer::PrepareStep(round + 1)))
+ return;
+ }
+ writer.join();
+
+ EXPECT_EQ(reuses_won.load() + reclaims_won.load(), kRounds);
+ EXPECT_EQ(reuses_won.load(), kRounds / 2);
+ EXPECT_EQ(reclaims_won.load(), kRounds / 2);
+}
+
+// ---------------------------------------------------------------------------
+// Race 4: a writer reserves while the reader publishes read_pos. Both
+// compare-and-swap the packed read/write positions, each changing only its own
+// half.
+//
+// The sequencer gives each whole call its own turn, so this checks that both
+// halves survive interleaved calls across a real thread boundary; it does not
+// force a compare-and-swap to lose. The forced CAS-loser schedules - an old
+// rw_positions value failing and the retry using the returned value - are
+// pinned deterministically in SharedRingBufferTest::ReservationCasLoses* and
+// PublicationCasLoses*.
+// ---------------------------------------------------------------------------
+
+TEST(SharedRingBufferConcurrencyTest, ReservationVersusReadPosPublication) {
+ constexpr uint32_t kRounds = 1000;
+ auto ring = SharedRingBuffer::Create(4, 256);
+ ASSERT_NE(ring, nullptr);
+
+ StepSequencer sequencer;
+ std::thread writer([&] {
+ for (uint32_t round = 0; round < kRounds; ++round) {
+ const uint64_t step = StepSequencer::WriterStep(round);
+ if (!sequencer.WaitForStep(step))
+ return;
+ const auto reservation = ring->TryReserveWritePos();
+ // Capacity always exists: the reader below never falls more than one
+ // position behind.
+ EXPECT_EQ(reservation.result, SharedRingBuffer::ReserveResult::kReserved);
+ EXPECT_EQ(reservation.position, round);
+ sequencer.FinishStep(step);
+ }
+ });
+ auto join_writer = base::OnScopeExit([&] {
+ sequencer.Abort();
+ if (writer.joinable())
+ writer.join();
+ });
+
+ for (uint32_t round = 0; round < kRounds; ++round) {
+ // Round r starts at (write=r, read=r-1); the writer
+ // reserves position r while this side publishes read_pos = r, in an order
+ // that alternates every round.
+ if (!sequencer.WaitForStep(StepSequencer::PrepareStep(round)))
+ return;
+ sequencer.FinishStep(StepSequencer::PrepareStep(round));
+
+ const uint64_t step = StepSequencer::ReaderStep(round);
+ if (!sequencer.WaitForStep(step))
+ return;
+ ring->PublishReadPos(round);
+ sequencer.FinishStep(step);
+
+ // Whoever went second loaded a value already holding the other side's
+ // half and had to carry it through. Both moves must survive every round.
+ if (!sequencer.WaitForStep(StepSequencer::PrepareStep(round + 1)))
+ return;
+ ASSERT_EQ(ring->LoadWritePos(), round + 1) << round;
+ ASSERT_EQ(ring->read_pos_for_testing(), round) << round;
+ }
+ writer.join();
+}
+
+// ---------------------------------------------------------------------------
+// MPSC stress.
+// ---------------------------------------------------------------------------
+
+struct StressParams {
+ uint32_t num_writers;
+ uint32_t num_chunks;
+ uint32_t chunk_size;
+ uint32_t fragments_per_writer;
+ uint32_t seed_position;
+ BufferExhaustedPolicy policy;
+};
+
+// What a run actually did, so each test can assert that it exercised the path
+// it is named after rather than passing because nothing happened.
+struct StressStats {
+ uint64_t received = 0;
+ // The caller never got payload space for these.
+ uint64_t unwritten = 0;
+ // The reader scraped the chunk and there was no replacement capacity.
+ uint64_t dropped = 0;
+ // The reader scraped a chunk out from under a writer.
+ uint64_t relocations = 0;
+ // How many of the unwritten fragments were refused because the ring was
+ // structurally full, as opposed to because a reserved position's chunk could
+ // not be claimed. The two mean different things and a stalling policy is only
+ // supposed to produce the second.
+ uint64_t reported_full = 0;
+ uint32_t final_read_pos = 0;
+};
+
+class StressDelegate : public SharedRingBufferReader::Delegate {
+ public:
+ void OnChunkRead(
+ const SharedRingBufferReader::ChunkContents& contents) override {
+ for (uint32_t i = 0; i < contents.num_fragments; ++i) {
+ const SharedRingBufferReader::Fragment& fragment = contents.fragments[i];
+ ASSERT_GE(fragment.size, 8u);
+ uint32_t writer = 0;
+ uint32_t sequence = 0;
+ memcpy(&writer, fragment.data, sizeof(writer));
+ memcpy(&sequence, fragment.data + 4, sizeof(sequence));
+ ASSERT_EQ(writer, contents.writer_id);
+ received[writer].push_back(sequence);
+ }
+ }
+
+ void OnDataLoss(WriterID writer_id) override {
+ ADD_FAILURE() << "reader discarded a chunk from writer " << writer_id;
+ }
+
+ std::map<uint32_t, std::vector<uint32_t>> received;
+};
+
+void RunStress(const StressParams& params, StressStats* stats) {
+ auto ring = SharedRingBuffer::Create(params.num_chunks, params.chunk_size);
+ ASSERT_NE(ring, nullptr);
+ ring->SetPositionsForTesting(params.seed_position);
+
+ StressDelegate delegate;
+ SharedRingBufferReader reader(ring.get(), &delegate);
+ reader.SetReadPosForTesting(params.seed_position);
+
+ std::atomic<uint32_t> writers_done{0};
+ // Set when the drain below gives up, so the writers stop producing and can
+ // be joined instead of racing an unresponsive protocol forever. Every
+ // blocking call a writer makes has a 30 second deadline, so a stopped writer
+ // leaves its loop within one OpenFragment() call.
+ std::atomic<bool> stop_writers{false};
+ std::vector<uint64_t> unwritten(params.num_writers, 0);
+ std::vector<uint64_t> dropped(params.num_writers, 0);
+ std::vector<uint64_t> relocations(params.num_writers, 0);
+ std::vector<uint64_t> reported_full(params.num_writers, 0);
+ std::vector<std::thread> writer_threads;
+
+ for (uint32_t w = 0; w < params.num_writers; ++w) {
+ writer_threads.emplace_back([&, w] {
+ SharedRingBufferWriter writer(ring.get(), static_cast<WriterID>(w + 1),
+ kBuffer, params.policy,
+ GetNoopSharedRingBufferWriterDelegate());
+ // A payload that varies in size so that chunk boundaries, reuse and the
+ // 255-fragment cap all get exercised rather than one fixed shape.
+ std::vector<uint8_t> payload(64);
+ for (uint32_t n = 0; n < params.fragments_per_writer; ++n) {
+ if (stop_writers.load(std::memory_order_relaxed))
+ break;
+ const uint32_t size = 8 + (n % 40);
+ const uint32_t writer_tag = w + 1;
+ memcpy(payload.data(), &writer_tag, sizeof(writer_tag));
+ memcpy(payload.data() + 4, &n, sizeof(n));
+
+ const SharedRingBufferWriter::FragmentSpan span =
+ writer.OpenFragment(size, false);
+ if (span.outcome != SharedRingBufferWriter::Outcome::kOk) {
+ ++unwritten[w];
+ if (span.outcome == SharedRingBufferWriter::Outcome::kFull)
+ ++reported_full[w];
+ writer.RecordDataLoss();
+ std::this_thread::yield();
+ continue;
+ }
+ memcpy(span.begin, payload.data(), size);
+ // Yield while the chunk is still BeingWritten, so the reader lands
+ // inside it and the scrape-and-relocate path is exercised for real.
+ // Yielding at a protocol boundary is deliberate: a sleep would only
+ // make the race likely, this makes it frequent.
+ if ((n & 7) == 0)
+ std::this_thread::yield();
+ writer.CloseFragment(size, false);
+ }
+ writer.FinishCurrentChunk();
+ dropped[w] = writer.num_fragments_dropped();
+ relocations[w] = writer.num_relocations();
+ writers_done.fetch_add(1, std::memory_order_release);
+ });
+ }
+
+ // Drain until every writer has finished and the ring has been emptied. The
+ // deadline is far beyond what a correct run needs, even under TSAN; hitting
+ // it means the protocol stopped making progress, and the test must fail
+ // rather than spin here forever.
+ const base::TimeMillis drain_deadline =
+ base::GetWallTimeMs() + base::TimeMillis(60000);
+ bool drain_timed_out = false;
+ for (;;) {
+ const bool all_done =
+ writers_done.load(std::memory_order_acquire) == params.num_writers;
+ const SharedRingBufferReader::DrainResult result = reader.Drain(64);
+ if (result.last_outcome == SharedRingBufferReader::Outcome::kProtocolError)
+ break;
+ if (all_done && !result.needs_another_drain())
+ break;
+ if (base::GetWallTimeMs() >= drain_deadline) {
+ drain_timed_out = true;
+ ADD_FAILURE() << "Stress drain made no progress within its deadline: "
+ << writers_done.load() << "/" << params.num_writers
+ << " writers done, read_pos " << reader.read_pos()
+ << ", last outcome "
+ << static_cast<int>(result.last_outcome);
+ break;
+ }
+ if (result.positions_resolved == 0)
+ std::this_thread::yield();
+ }
+
+ // Stop the writers - a no-op on the success path, where they are already
+ // done - and keep draining until they have all left their loops, so the
+ // joins below cannot wait on a writer parked on a full ring.
+ stop_writers.store(true, std::memory_order_relaxed);
+ const base::TimeMillis shutdown_deadline =
+ base::GetWallTimeMs() + base::TimeMillis(60000);
+ while (writers_done.load(std::memory_order_acquire) != params.num_writers &&
+ base::GetWallTimeMs() < shutdown_deadline) {
+ reader.Drain(64);
+ std::this_thread::yield();
+ }
+ for (std::thread& thread : writer_threads)
+ thread.join();
+ if (drain_timed_out)
+ return; // Already failed; the accounting below would only add noise.
+ ASSERT_FALSE(reader.has_protocol_error());
+ reader.Drain(1u << 20);
+ stats->final_read_pos = reader.read_pos();
+
+ for (uint32_t w = 0; w < params.num_writers; ++w) {
+ const std::vector<uint32_t>& sequences = delegate.received[w + 1];
+ // Exactly once, and in the order the writer wrote them.
+ for (size_t i = 1; i < sequences.size(); ++i) {
+ ASSERT_LT(sequences[i - 1], sequences[i])
+ << "writer " << w << " fragment " << i;
+ }
+ // Everything that is missing is something the writer itself accounted for.
+ EXPECT_EQ(sequences.size() + unwritten[w] + dropped[w],
+ params.fragments_per_writer)
+ << "writer " << w;
+
+ stats->received += sequences.size();
+ stats->unwritten += unwritten[w];
+ stats->dropped += dropped[w];
+ stats->relocations += relocations[w];
+ stats->reported_full += reported_full[w];
+ }
+}
+
+// The reader only relocates when it lands inside a chunk a writer still owns,
+// which is a scheduling outcome and not something these tests can force from
+// the outside: roughly one run in a hundred sees none. Repeating a bounded
+// number of times keeps them from failing on one unlucky schedule.
+//
+// This is a coverage check, not a proof. That relocation happens, and what it
+// does to the payload and the flags, is settled deterministically by
+// PublishVersusScrape above and by the reader and writer unit tests. All
+// this asserts is that the stress ran with the contention it is named for.
+constexpr uint32_t kAttemptsForRelocations = 8;
+
+void RunStressUntilRelocations(const StressParams& params, StressStats* stats) {
+ for (uint32_t attempt = 0; attempt < kAttemptsForRelocations; ++attempt) {
+ RunStress(params, stats);
+ if (stats->relocations > 0 || testing::Test::HasFatalFailure())
+ return;
+ }
+}
+
+TEST(SharedRingBufferConcurrencyTest, StressFourWritersDropPolicy) {
+ StressStats stats;
+ RunStressUntilRelocations({/*num_writers=*/4, /*num_chunks=*/8,
+ /*chunk_size=*/256,
+ /*fragments_per_writer=*/4000, /*seed_position=*/0,
+ BufferExhaustedPolicy::kDrop},
+ &stats);
+ // A drop policy on a small ring loses a lot, which is the point - but the
+ // exactly-once and ordering checks above are worthless if nothing got
+ // through. An absolute floor rather than a share of the total: how much a
+ // drop-policy run keeps depends on how fast the reader is scheduled, and
+ // under ThreadSanitizer that is a different number. What has to hold is that
+ // the per-writer checks ran on real data, not that the ring achieved a
+ // particular throughput.
+ EXPECT_GT(stats.received, 100u);
+ EXPECT_GT(stats.relocations, 0u);
+}
+
+TEST(SharedRingBufferConcurrencyTest, StressTinyRingForcesHolesAndRelocations) {
+ StressStats stats;
+ RunStressUntilRelocations({/*num_writers=*/4, /*num_chunks=*/2,
+ /*chunk_size=*/256,
+ /*fragments_per_writer=*/2000, /*seed_position=*/0,
+ BufferExhaustedPolicy::kDrop},
+ &stats);
+ EXPECT_GT(stats.received, 0u);
+ // Two chunks and four writers is the shape that makes the reader land inside
+ // a live writer's chunk constantly.
+ EXPECT_GT(stats.relocations, 0u);
+}
+
+TEST(SharedRingBufferConcurrencyTest, StressAcrossPositionRollover) {
+ // Seeded so the 32-bit positions roll over part way through, which is where a
+ // reader that incremented the wrap it found in the chunk would lock writers
+ // out.
+ StressStats stats;
+ RunStress({/*num_writers=*/4, /*num_chunks=*/16, /*chunk_size=*/512,
+ /*fragments_per_writer=*/3000,
+ /*seed_position=*/0u - 64u, BufferExhaustedPolicy::kDrop},
+ &stats);
+ EXPECT_GT(stats.received, 0u);
+ EXPECT_LT(stats.final_read_pos, 0u - 64u);
+}
+
+TEST(SharedRingBufferConcurrencyTest, StressAcrossTheWrapIdentityRollover) {
+ // Seeded so the 16-bit wrap identity rolls over part way through - for four
+ // chunks that is every 4 * 65536 positions - while the 32-bit positions are
+ // nowhere near wrapping. The claim/advance arbitration runs right across the
+ // truncation boundary.
+ StressStats stats;
+ RunStress({/*num_writers=*/4, /*num_chunks=*/4, /*chunk_size=*/256,
+ /*fragments_per_writer=*/3000,
+ /*seed_position=*/4u * 65536 - 64u, BufferExhaustedPolicy::kDrop},
+ &stats);
+ EXPECT_GT(stats.received, 0u);
+ EXPECT_LT(WrapCountForPosition(stats.final_read_pos, 2),
+ WrapCountForPosition(4u * 65536 - 64u, 2));
+}
+
+TEST(SharedRingBufferConcurrencyTest, StressStallThenDropPolicy) {
+ StressStats stats;
+ RunStress({/*num_writers=*/3, /*num_chunks=*/4, /*chunk_size=*/1024,
+ /*fragments_per_writer=*/2000, /*seed_position=*/0,
+ BufferExhaustedPolicy::kStallThenDrop},
+ &stats);
+ EXPECT_GT(stats.received, 0u);
+}
+
+TEST(SharedRingBufferConcurrencyTest, StressStallPolicyNeverReportsFull) {
+ StressStats stats;
+ RunStress({/*num_writers=*/3, /*num_chunks=*/8, /*chunk_size=*/1024,
+ /*fragments_per_writer=*/2000, /*seed_position=*/0,
+ BufferExhaustedPolicy::kStall},
+ &stats);
+ EXPECT_GT(stats.received, 0u);
+#if PERFETTO_BUILDFLAG(PERFETTO_OS_LINUX_BUT_NOT_QNX) || \
+ PERFETTO_BUILDFLAG(PERFETTO_OS_ANDROID)
+ // A stalling writer waits for the reader to expose capacity, so a full ring
+ // never turns into a refusal. What it *can* still hit is a run of positions
+ // whose chunks are pinned by writers the reader has marked for rewrite: that
+ // is NoChunkAvailable, not Full, and the two are different events. Whatever
+ // is lost that way is accounted for exactly, which the per-writer check
+ // inside RunStress() has already asserted.
+ EXPECT_EQ(stats.reported_full, 0u);
+#endif
+}
+
+} // namespace
+} // namespace perfetto::tracing_v2
diff --git a/src/tracing/v2/shared_ring_buffer_reader.cc b/src/tracing/v2/shared_ring_buffer_reader.cc
new file mode 100644
index 0000000..947a5e7
--- /dev/null
+++ b/src/tracing/v2/shared_ring_buffer_reader.cc
@@ -0,0 +1,249 @@
+/*
+ * Copyright (C) 2026 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/v2/shared_ring_buffer_reader.h"
+
+#include <stdint.h>
+
+#include <utility>
+
+#include "perfetto/base/logging.h"
+#include "src/tracing/v2/shared_ring_buffer.h"
+#include "src/tracing/v2/tracing_v2_abi.h"
+
+namespace perfetto::tracing_v2 {
+
+SharedRingBufferReader::Delegate::~Delegate() = default;
+
+SharedRingBufferReader::SharedRingBufferReader(SharedRingBuffer* ring,
+ Delegate* delegate)
+ : ring_(ring),
+ delegate_(delegate),
+ num_chunks_(ring->num_chunks()),
+ chunk_size_(ring->chunk_size()),
+ chunk_index_bits_(ring->chunk_index_bits()) {
+ copied_payload_.reserve(MaxFragmentSizeForEmptyChunk(chunk_size_));
+ copied_fragments_.reserve(kMaxFragmentsPerChunk);
+}
+
+SharedRingBufferReader::~SharedRingBufferReader() = default;
+
+SharedRingBufferReader::DrainResult SharedRingBufferReader::Drain(
+ uint32_t max_positions) {
+ const uint32_t start_pos = read_pos_;
+ DrainResult result{};
+ for (uint32_t i = 0; i < max_positions; ++i) {
+ result.last_outcome = ResolveNextPosition();
+ if (result.last_outcome != Outcome::kChunkRead &&
+ result.last_outcome != Outcome::kSkipped) {
+ break;
+ }
+ }
+
+ result.positions_resolved = read_pos_ - start_pos;
+ if (result.positions_resolved != 0) {
+ // One publication and at most one wake cover the whole pass. Until this
+ // point writers can only under-estimate free capacity.
+ //
+ // TODO(sashwinbalaji): benchmark per-position publication against batch
+ // sizes such as 16, 64 and 256 before trusting any of them. Nothing here
+ // establishes that the caller's current pass size is optimal.
+ ring_->PublishReadPos(read_pos_);
+ }
+ return result;
+}
+
+SharedRingBufferReader::Outcome SharedRingBufferReader::ResolveNextPosition() {
+ if (has_protocol_error_)
+ return Outcome::kProtocolError;
+
+ // A stale write_pos only shortens this drain pass.
+ const uint32_t write_pos = ring_->LoadWritePos();
+ const uint32_t outstanding = NumOutstandingPositions(write_pos, read_pos_);
+ if (outstanding == 0)
+ return Outcome::kNoData;
+ if (outstanding > num_chunks_) {
+ // A legal writer cannot reserve more than num_chunks outstanding
+ // positions.
+ has_protocol_error_ = true;
+ PERFETTO_ELOG(
+ "tracing v2: stopping ring reader; write_pos %u is %u positions ahead "
+ "of read_pos %u, which is more than the %u chunks in the ring",
+ write_pos, outstanding, read_pos_, num_chunks_);
+ return Outcome::kProtocolError;
+ }
+
+ const uint32_t position = read_pos_;
+ const uint32_t chunk_index = ChunkIndexOfPosition(position, num_chunks_);
+
+ // A failed compare-and-swap replaces this with the current state word.
+ uint32_t observed = ring_->LoadChunkStateWord(chunk_index);
+
+ switch (ChunkStateOf(observed)) {
+ case ChunkState::kFree: {
+ if (observed != MakeFreeStateWord(
+ WrapCountForPosition(position, chunk_index_bits_))) {
+ return StopOnProtocolError(observed);
+ }
+ // Nobody claimed this position. Prepare the chunk for its next wrap.
+ if (!ring_->TryMoveFreeChunkToNextWrap(position, &observed))
+ return Outcome::kRetryLater;
+ ++read_pos_;
+ ++num_positions_skipped_;
+ return Outcome::kSkipped;
+ }
+
+ case ChunkState::kBeingWritten: {
+ // This copy is speculative until the CAS below wins against the writer.
+ const CommittedPrefixStatus status =
+ CopyCommittedPrefix(chunk_index, observed);
+ if (before_arbitration_hook_for_testing_)
+ before_arbitration_hook_for_testing_();
+ // Malformed payload is still marked: validation does not decide
+ // ownership.
+ if (!ring_->TryRequestRewrite(chunk_index, &observed))
+ return Outcome::kRetryLater;
+ ++read_pos_;
+ ++num_scrapes_;
+ return ForwardCopiedChunk(status);
+ }
+
+ case ChunkState::kComplete: {
+ const CommittedPrefixStatus status =
+ CopyCommittedPrefix(chunk_index, observed);
+ if (before_arbitration_hook_for_testing_)
+ before_arbitration_hook_for_testing_();
+ if (!ring_->TryReleaseCompleteChunkAsFree(position, &observed))
+ return Outcome::kRetryLater;
+ ++read_pos_;
+ return ForwardCopiedChunk(status);
+ }
+
+ case ChunkState::kRewriteRequested:
+ // The writer still owns this chunk. Resolve the position as a hole.
+ ++read_pos_;
+ ++num_positions_skipped_;
+ return Outcome::kSkipped;
+
+ case ChunkState::kRewriteAcknowledged:
+ if (!ring_->TryReleaseRewriteAcknowledgedChunkAsFree(position, &observed))
+ return StopOnProtocolError(observed);
+ ++read_pos_;
+ ++num_positions_skipped_;
+ return Outcome::kSkipped;
+
+ case ChunkState::kReserved5:
+ case ChunkState::kReserved6:
+ case ChunkState::kReserved7:
+ // The reader cannot safely reclaim an unknown state.
+ return StopOnProtocolError(observed);
+ }
+}
+
+SharedRingBufferReader::CommittedPrefixStatus
+SharedRingBufferReader::CopyCommittedPrefix(uint32_t chunk_index,
+ uint32_t state_word) {
+ copied_fragments_.clear();
+ copied_chunk_ = ChunkContents{};
+ copied_chunk_.writer_id = WriterIdOf(state_word);
+ copied_chunk_.payload_flags = PayloadFlagsOf(state_word);
+
+ const uint32_t num_fragments = NumFragmentsOf(state_word);
+ if (num_fragments == 0) {
+ // Nothing is published, so there is nothing to copy - and nothing else in
+ // the chunk may be touched. In particular the target BufferID is stored
+ // while the writer exclusively owns a freshly claimed chunk and is
+ // published only by the first release transition out of kBeingWritten, so
+ // a BeingWritten word carrying no fragments does not order that store
+ // against this reader at all.
+ return CommittedPrefixStatus::kNoFragments;
+ }
+
+ if (ChunkFormatOf(state_word) != ChunkFormat::kTargetBuffer)
+ return CommittedPrefixStatus::kUnsupportedFormat;
+
+ const uint32_t capacity = chunk_size_ - kTargetBufferPayloadOffset;
+ const uint8_t* chunk = ring_->chunk_at(chunk_index);
+ const uint8_t* const payload_begin = chunk + kTargetBufferPayloadOffset;
+ const uint8_t* directory_cursor = chunk + chunk_size_;
+ uint32_t total = 0;
+ for (uint32_t i = 0; i < num_fragments; ++i) {
+ uint32_t fragment_size = 0;
+ if (!ReadFragmentSize(payload_begin, &directory_cursor, &fragment_size) ||
+ fragment_size > capacity - total) {
+ copied_fragments_.clear();
+ return CommittedPrefixStatus::kMalformed;
+ }
+ total += fragment_size;
+ copied_fragments_.push_back(Fragment{nullptr, fragment_size});
+ }
+
+ const uint32_t directory_bytes =
+ static_cast<uint32_t>(chunk + chunk_size_ - directory_cursor);
+ if (total > capacity - directory_bytes) {
+ copied_fragments_.clear();
+ return CommittedPrefixStatus::kMalformed;
+ }
+
+ copied_payload_.assign(chunk + kTargetBufferPayloadOffset,
+ chunk + kTargetBufferPayloadOffset + total);
+ uint32_t offset = 0;
+ for (Fragment& fragment : copied_fragments_) {
+ fragment.data = copied_payload_.data() + offset;
+ offset += fragment.size;
+ }
+
+ copied_chunk_.target_buffer = LoadTargetBufferId(chunk);
+ copied_chunk_.fragments = copied_fragments_.data();
+ copied_chunk_.num_fragments = num_fragments;
+ return CommittedPrefixStatus::kReady;
+}
+
+SharedRingBufferReader::Outcome SharedRingBufferReader::ForwardCopiedChunk(
+ CommittedPrefixStatus status) {
+ switch (status) {
+ case CommittedPrefixStatus::kReady:
+ ++num_chunks_read_;
+ delegate_->OnChunkRead(copied_chunk_);
+ return Outcome::kChunkRead;
+ case CommittedPrefixStatus::kMalformed:
+ ++num_malformed_chunks_;
+ delegate_->OnDataLoss(copied_chunk_.writer_id);
+ break;
+ case CommittedPrefixStatus::kUnsupportedFormat:
+ ++num_unknown_format_chunks_;
+ delegate_->OnDataLoss(copied_chunk_.writer_id);
+ break;
+ case CommittedPrefixStatus::kNoFragments:
+ break;
+ }
+
+ ++num_positions_skipped_;
+ return Outcome::kSkipped;
+}
+
+SharedRingBufferReader::Outcome SharedRingBufferReader::StopOnProtocolError(
+ uint32_t state_word) {
+ // Latched, so this logs once per ring rather than once per drain pass.
+ has_protocol_error_ = true;
+ PERFETTO_ELOG(
+ "tracing v2: stopping ring reader at position %u; chunk state word "
+ "0x%08x is not something this build can arbitrate",
+ read_pos_, state_word);
+ return Outcome::kProtocolError;
+}
+
+} // namespace perfetto::tracing_v2
diff --git a/src/tracing/v2/shared_ring_buffer_reader.h b/src/tracing/v2/shared_ring_buffer_reader.h
new file mode 100644
index 0000000..a3d92ba
--- /dev/null
+++ b/src/tracing/v2/shared_ring_buffer_reader.h
@@ -0,0 +1,187 @@
+/*
+ * Copyright (C) 2026 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_V2_SHARED_RING_BUFFER_READER_H_
+#define SRC_TRACING_V2_SHARED_RING_BUFFER_READER_H_
+
+#include <stdint.h>
+
+#include <functional>
+#include <vector>
+
+#include "perfetto/ext/tracing/core/basic_types.h"
+#include "src/tracing/v2/shared_ring_buffer.h"
+#include "src/tracing/v2/tracing_v2_abi.h"
+
+namespace perfetto::tracing_v2 {
+
+// Reads packet fragments from one SharedRingBuffer.
+//
+// Each instance is used from one execution context. It resolves write
+// positions in order and never waits for a writer. If a writer still owns the
+// chunk, the reader takes its published prefix and leaves the writer to move
+// the unpublished suffix.
+//
+// Initially the reader runs in the producer process. It validates the fragment
+// directory in place and copies the published payload before calling the
+// delegate.
+//
+// TODO(sashwinbalaji): before moving SharedRingBufferReader into traced, copy
+// the published format header, directory and payload into reader-owned memory
+// before parsing them. A producer must not be able to change service-side
+// validation inputs.
+class SharedRingBufferReader {
+ public:
+ // Result of resolving one position.
+ enum class Outcome {
+ // read_pos has caught up with write_pos.
+ kNoData,
+ // A chunk was handed to the delegate and read_pos advanced.
+ kChunkRead,
+ // The position resolved to no payload - nobody claimed it, its writer is
+ // mid-rewrite, or the bytes could not be trusted - and read_pos advanced.
+ kSkipped,
+ // The chunk state changed before the reader completed its transition.
+ // read_pos is unchanged and the same position is retried later.
+ kRetryLater,
+ // Corruption or an incompatible ABI. This ring is left exactly as it was
+ // found and is never read again; the rest of the process is unaffected.
+ kProtocolError,
+ };
+
+ // A view into reader-owned scratch. Valid only for the duration of the
+ // Delegate call.
+ struct Fragment {
+ const uint8_t* data = nullptr;
+ uint32_t size = 0;
+ };
+
+ struct ChunkContents {
+ WriterID writer_id = 0;
+ BufferID target_buffer = 0;
+ // Any of the three PayloadFlags.
+ uint32_t payload_flags = 0;
+ const Fragment* fragments = nullptr;
+ uint32_t num_fragments = 0;
+ };
+
+ struct DrainResult {
+ uint32_t positions_resolved = 0;
+ Outcome last_outcome = Outcome::kNoData;
+
+ // The caller should schedule another Drain() call.
+ bool needs_another_drain() const {
+ return last_outcome != Outcome::kNoData &&
+ last_outcome != Outcome::kProtocolError;
+ }
+ };
+
+ class Delegate {
+ public:
+ virtual ~Delegate();
+
+ // Called after a chunk's published payload has been copied into
+ // reader-owned memory. The view is valid only for this call.
+ virtual void OnChunkRead(const ChunkContents&) = 0;
+
+ // The reader resolved committed data that it could not forward. The
+ // consumer should report the gap on the next packet from this writer.
+ virtual void OnDataLoss(WriterID) = 0;
+ };
+
+ SharedRingBufferReader(SharedRingBuffer* ring, Delegate* delegate);
+ ~SharedRingBufferReader();
+
+ SharedRingBufferReader(const SharedRingBufferReader&) = delete;
+ SharedRingBufferReader& operator=(const SharedRingBufferReader&) = delete;
+ SharedRingBufferReader(SharedRingBufferReader&&) = delete;
+ SharedRingBufferReader& operator=(SharedRingBufferReader&&) = delete;
+
+ // Resolves up to |max_positions|, then publishes read_pos once and wakes any
+ // writer parked on a full ring.
+ DrainResult Drain(uint32_t max_positions);
+
+ // Resolves at most one position. Drain() publishes read_pos once per pass.
+ Outcome ResolveNextPosition();
+
+ bool has_protocol_error() const { return has_protocol_error_; }
+ uint32_t read_pos() const { return read_pos_; }
+
+ // Diagnostics. Never inputs to a decision.
+ uint64_t num_positions_skipped() const { return num_positions_skipped_; }
+ uint64_t num_scrapes() const { return num_scrapes_; }
+ uint64_t num_chunks_read() const { return num_chunks_read_; }
+ uint64_t num_malformed_chunks() const { return num_malformed_chunks_; }
+ uint64_t num_unknown_format_chunks() const {
+ return num_unknown_format_chunks_;
+ }
+
+ // Invoked after the committed prefix has been copied and immediately before
+ // the compare-and-swap that arbitrates it. It exists so that a test can make
+ // the writer win that race deterministically instead of hoping for it.
+ // Nothing in production sets it.
+ void SetArbitrationHookForTesting(std::function<void()> hook) {
+ before_arbitration_hook_for_testing_ = std::move(hook);
+ }
+
+ // Starts the reader at |position| instead of zero, to match a ring seeded
+ // near uint32_t rollover.
+ void SetReadPosForTesting(uint32_t position) { read_pos_ = position; }
+
+ private:
+ enum class CommittedPrefixStatus {
+ kNoFragments,
+ kReady,
+ kMalformed,
+ kUnsupportedFormat,
+ };
+
+ // Validates and copies the committed prefix. A malformed or unknown format
+ // is dropped without changing the ownership transition chosen by the
+ // caller.
+ CommittedPrefixStatus CopyCommittedPrefix(uint32_t chunk_index,
+ uint32_t state_word);
+
+ Outcome ForwardCopiedChunk(CommittedPrefixStatus);
+ Outcome StopOnProtocolError(uint32_t state_word);
+
+ SharedRingBuffer* const ring_;
+ Delegate* const delegate_;
+ const uint32_t num_chunks_;
+ const uint32_t chunk_size_;
+ const uint32_t chunk_index_bits_;
+
+ // The reader owns this value and publishes it once per Drain().
+ uint32_t read_pos_ = 0;
+ bool has_protocol_error_ = false;
+ std::function<void()> before_arbitration_hook_for_testing_;
+
+ // Reader-owned scratch, sized once in the constructor so that draining
+ // allocates nothing.
+ std::vector<uint8_t> copied_payload_;
+ std::vector<Fragment> copied_fragments_;
+ ChunkContents copied_chunk_;
+
+ uint64_t num_positions_skipped_ = 0;
+ uint64_t num_scrapes_ = 0;
+ uint64_t num_chunks_read_ = 0;
+ uint64_t num_malformed_chunks_ = 0;
+ uint64_t num_unknown_format_chunks_ = 0;
+};
+
+} // namespace perfetto::tracing_v2
+
+#endif // SRC_TRACING_V2_SHARED_RING_BUFFER_READER_H_
diff --git a/src/tracing/v2/shared_ring_buffer_reader_unittest.cc b/src/tracing/v2/shared_ring_buffer_reader_unittest.cc
new file mode 100644
index 0000000..2d059ca
--- /dev/null
+++ b/src/tracing/v2/shared_ring_buffer_reader_unittest.cc
@@ -0,0 +1,847 @@
+/*
+ * Copyright (C) 2026 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/v2/shared_ring_buffer_reader.h"
+
+#include <stdint.h>
+#include <string.h>
+
+#include <memory>
+#include <string>
+#include <vector>
+
+#include "perfetto/ext/base/no_destructor.h"
+#include "perfetto/tracing/buffer_exhausted_policy.h"
+#include "src/tracing/v2/shared_ring_buffer.h"
+#include "src/tracing/v2/shared_ring_buffer_writer.h"
+#include "src/tracing/v2/tracing_v2_abi.h"
+#include "test/gtest_and_gmock.h"
+
+namespace perfetto::tracing_v2 {
+namespace {
+
+using Outcome = SharedRingBufferReader::Outcome;
+
+constexpr WriterID kWriterA = 7;
+constexpr WriterID kWriterB = 8;
+constexpr BufferID kBuffer = 0x1234;
+
+class NoopSharedRingBufferWriterDelegate
+ : public SharedRingBufferWriter::Delegate {
+ public:
+ void NotifyReader() override {}
+};
+
+SharedRingBufferWriter::Delegate* GetNoopSharedRingBufferWriterDelegate() {
+ static base::NoDestructor<NoopSharedRingBufferWriterDelegate> delegate;
+ return &delegate.ref();
+}
+
+struct ReadChunk {
+ WriterID writer_id = 0;
+ BufferID target_buffer = 0;
+ uint32_t payload_flags = 0;
+ std::vector<std::string> fragments;
+};
+
+class RecordingDelegate : public SharedRingBufferReader::Delegate {
+ public:
+ void OnChunkRead(
+ const SharedRingBufferReader::ChunkContents& contents) override {
+ ReadChunk chunk;
+ chunk.writer_id = contents.writer_id;
+ chunk.target_buffer = contents.target_buffer;
+ chunk.payload_flags = contents.payload_flags;
+ for (uint32_t i = 0; i < contents.num_fragments; ++i) {
+ chunk.fragments.emplace_back(
+ reinterpret_cast<const char*>(contents.fragments[i].data),
+ contents.fragments[i].size);
+ }
+ chunks.push_back(std::move(chunk));
+ }
+
+ void OnDataLoss(WriterID writer_id) override {
+ writers_with_data_loss.push_back(writer_id);
+ }
+
+ std::vector<std::string> AllFragments() const {
+ std::vector<std::string> all;
+ for (const ReadChunk& chunk : chunks)
+ all.insert(all.end(), chunk.fragments.begin(), chunk.fragments.end());
+ return all;
+ }
+
+ std::vector<ReadChunk> chunks;
+ std::vector<WriterID> writers_with_data_loss;
+};
+
+SharedRingBufferWriter::Outcome WriteFragment(SharedRingBufferWriter* writer,
+ const std::string& bytes,
+ bool continues_from_prev = false,
+ bool continues_on_next = false) {
+ const SharedRingBufferWriter::FragmentSpan span = writer->OpenFragment(
+ static_cast<uint32_t>(bytes.size()), continues_from_prev);
+ if (span.outcome != SharedRingBufferWriter::Outcome::kOk)
+ return span.outcome;
+ memcpy(span.begin, bytes.data(), bytes.size());
+ return writer->CloseFragment(static_cast<uint32_t>(bytes.size()),
+ continues_on_next);
+}
+
+// ---------------------------------------------------------------------------
+// The nominal path.
+// ---------------------------------------------------------------------------
+
+TEST(SharedRingBufferReaderTest, EmptyRingHasNoData) {
+ auto ring = SharedRingBuffer::Create(4, 256);
+ ASSERT_NE(ring, nullptr);
+ RecordingDelegate delegate;
+ SharedRingBufferReader reader(ring.get(), &delegate);
+
+ EXPECT_EQ(reader.ResolveNextPosition(), Outcome::kNoData);
+ EXPECT_EQ(reader.read_pos(), 0u);
+ const SharedRingBufferReader::DrainResult result = reader.Drain(16);
+ EXPECT_EQ(result.positions_resolved, 0u);
+ EXPECT_FALSE(result.needs_another_drain());
+}
+
+TEST(SharedRingBufferReaderTest,
+ ConsumesACompleteChunkAndFreesItForTheNextTraversal) {
+ auto ring = SharedRingBuffer::Create(4, 512);
+ ASSERT_NE(ring, nullptr);
+ RecordingDelegate delegate;
+ SharedRingBufferReader reader(ring.get(), &delegate);
+ SharedRingBufferWriter writer(ring.get(), kWriterA, kBuffer,
+ BufferExhaustedPolicy::kDrop,
+ GetNoopSharedRingBufferWriterDelegate());
+
+ ASSERT_EQ(WriteFragment(&writer, "alpha"),
+ SharedRingBufferWriter::Outcome::kOk);
+ ASSERT_EQ(WriteFragment(&writer, "beta"),
+ SharedRingBufferWriter::Outcome::kOk);
+ writer.FinishCurrentChunk();
+
+ EXPECT_EQ(reader.ResolveNextPosition(), Outcome::kChunkRead);
+ EXPECT_EQ(reader.read_pos(), 1u);
+ ASSERT_EQ(delegate.chunks.size(), 1u);
+ EXPECT_EQ(delegate.chunks[0].writer_id, kWriterA);
+ EXPECT_EQ(delegate.chunks[0].target_buffer, kBuffer);
+ EXPECT_EQ(delegate.chunks[0].fragments,
+ (std::vector<std::string>{"alpha", "beta"}));
+
+ // The chunk is now tagged for the traversal after the one just resolved.
+ EXPECT_EQ(ring->LoadChunkStateWord(0), MakeFreeStateWord(1));
+ EXPECT_EQ(reader.ResolveNextPosition(), Outcome::kNoData);
+}
+
+TEST(SharedRingBufferReaderTest, DrainPublishesReadPosOncePerPass) {
+ auto ring = SharedRingBuffer::Create(8, 256);
+ ASSERT_NE(ring, nullptr);
+ RecordingDelegate delegate;
+ SharedRingBufferReader reader(ring.get(), &delegate);
+
+ // Four writers, one chunk each, so the pass has four positions to resolve.
+ std::vector<std::unique_ptr<SharedRingBufferWriter>> writers;
+ for (uint32_t i = 0; i < 4; ++i) {
+ writers.push_back(std::make_unique<SharedRingBufferWriter>(
+ ring.get(), static_cast<WriterID>(10 + i), kBuffer,
+ BufferExhaustedPolicy::kDrop, GetNoopSharedRingBufferWriterDelegate()));
+ ASSERT_EQ(WriteFragment(writers.back().get(), "x"),
+ SharedRingBufferWriter::Outcome::kOk);
+ }
+ // The shared read_pos has not moved yet: ResolveNextPosition() does not
+ // publish it.
+ EXPECT_EQ(ring->read_pos_for_testing(), 0u);
+
+ const SharedRingBufferReader::DrainResult result = reader.Drain(16);
+ EXPECT_EQ(result.positions_resolved, 4u);
+ EXPECT_EQ(result.last_outcome, Outcome::kNoData);
+ EXPECT_FALSE(result.needs_another_drain());
+ EXPECT_EQ(ring->read_pos_for_testing(), 4u);
+ EXPECT_EQ(delegate.chunks.size(), 4u);
+}
+
+TEST(SharedRingBufferReaderTest, DrainStopsAtItsBudgetAndSaysWorkMayRemain) {
+ auto ring = SharedRingBuffer::Create(8, 256);
+ ASSERT_NE(ring, nullptr);
+ RecordingDelegate delegate;
+ SharedRingBufferReader reader(ring.get(), &delegate);
+
+ std::vector<std::unique_ptr<SharedRingBufferWriter>> writers;
+ for (uint32_t i = 0; i < 4; ++i) {
+ writers.push_back(std::make_unique<SharedRingBufferWriter>(
+ ring.get(), static_cast<WriterID>(10 + i), kBuffer,
+ BufferExhaustedPolicy::kDrop, GetNoopSharedRingBufferWriterDelegate()));
+ ASSERT_EQ(WriteFragment(writers.back().get(), "x"),
+ SharedRingBufferWriter::Outcome::kOk);
+ }
+
+ const SharedRingBufferReader::DrainResult first = reader.Drain(2);
+ EXPECT_EQ(first.positions_resolved, 2u);
+ EXPECT_TRUE(first.needs_another_drain());
+ EXPECT_EQ(ring->read_pos_for_testing(), 2u);
+
+ const SharedRingBufferReader::DrainResult second = reader.Drain(16);
+ EXPECT_EQ(second.positions_resolved, 2u);
+ EXPECT_FALSE(second.needs_another_drain());
+ EXPECT_EQ(delegate.chunks.size(), 4u);
+}
+
+// ---------------------------------------------------------------------------
+// Holes.
+// ---------------------------------------------------------------------------
+
+TEST(SharedRingBufferReaderTest,
+ UnclaimedPositionIsAHoleAndPreparesTheNextTraversal) {
+ auto ring = SharedRingBuffer::Create(4, 256);
+ ASSERT_NE(ring, nullptr);
+ RecordingDelegate delegate;
+ SharedRingBufferReader reader(ring.get(), &delegate);
+
+ // A writer reserved position 0 and never claimed it.
+ ASSERT_EQ(ring->TryReserveWritePos().position, 0u);
+
+ EXPECT_EQ(reader.ResolveNextPosition(), Outcome::kSkipped);
+ EXPECT_EQ(reader.read_pos(), 1u);
+ EXPECT_EQ(reader.num_positions_skipped(), 1u);
+ EXPECT_TRUE(delegate.chunks.empty());
+ EXPECT_EQ(ring->LoadChunkStateWord(0), MakeFreeStateWord(1));
+}
+
+TEST(SharedRingBufferReaderTest,
+ RewriteRequestedIsSkippedWithoutTouchingTheWord) {
+ auto ring = SharedRingBuffer::Create(4, 256);
+ ASSERT_NE(ring, nullptr);
+ RecordingDelegate delegate;
+ SharedRingBufferReader reader(ring.get(), &delegate);
+
+ // Chunk 0 is left mid-rewrite by a writer that has not come back, and a later
+ // position maps onto it.
+ const uint32_t being_written = MakeDataStateWord(
+ ChunkState::kBeingWritten, ChunkFormat::kTargetBuffer, 0, 0, kWriterB);
+ ASSERT_TRUE(ring->TryAcquireChunkForWriting(0, being_written));
+ uint32_t observed = being_written;
+ ASSERT_TRUE(ring->TryRequestRewrite(0, &observed));
+ const uint32_t marked = ring->LoadChunkStateWord(0);
+
+ ASSERT_EQ(ring->TryReserveWritePos().position, 0u);
+ EXPECT_EQ(reader.ResolveNextPosition(), Outcome::kSkipped);
+ EXPECT_EQ(reader.read_pos(), 1u);
+ // Only the owning writer may leave that state, so the reader left it alone.
+ EXPECT_EQ(ring->LoadChunkStateWord(0), marked);
+ EXPECT_EQ(reader.num_positions_skipped(), 1u);
+}
+
+TEST(SharedRingBufferReaderTest,
+ RewriteAcknowledgedIsReclaimedForResolvedPosition) {
+ auto ring = SharedRingBuffer::Create(4, 256);
+ ASSERT_NE(ring, nullptr);
+ RecordingDelegate delegate;
+ SharedRingBufferReader reader(ring.get(), &delegate);
+
+ const uint32_t being_written = MakeDataStateWord(
+ ChunkState::kBeingWritten, ChunkFormat::kTargetBuffer, 0, 0, kWriterB);
+ ASSERT_TRUE(ring->TryAcquireChunkForWriting(0, being_written));
+ uint32_t observed = being_written;
+ ASSERT_TRUE(ring->TryRequestRewrite(0, &observed));
+ ASSERT_TRUE(ring->TryAcknowledgeRewrite(
+ 0, ReplaceChunkState(being_written, ChunkState::kRewriteRequested)));
+
+ ASSERT_EQ(ring->TryReserveWritePos().position, 0u);
+ EXPECT_EQ(reader.ResolveNextPosition(), Outcome::kSkipped);
+ EXPECT_EQ(ring->LoadChunkStateWord(0), MakeFreeStateWord(1));
+}
+
+// ---------------------------------------------------------------------------
+// Scraping a live writer.
+// ---------------------------------------------------------------------------
+
+TEST(SharedRingBufferReaderTest, TakesTheCommittedPrefixOfALiveWriter) {
+ auto ring = SharedRingBuffer::Create(4, 512);
+ ASSERT_NE(ring, nullptr);
+ RecordingDelegate delegate;
+ SharedRingBufferReader reader(ring.get(), &delegate);
+ SharedRingBufferWriter writer(ring.get(), kWriterA, kBuffer,
+ BufferExhaustedPolicy::kDrop,
+ GetNoopSharedRingBufferWriterDelegate());
+
+ ASSERT_EQ(WriteFragment(&writer, "committed"),
+ SharedRingBufferWriter::Outcome::kOk);
+ // The writer is inside the chunk with a fragment still open.
+ const SharedRingBufferWriter::FragmentSpan span =
+ writer.OpenFragment(6, false);
+ ASSERT_EQ(span.outcome, SharedRingBufferWriter::Outcome::kOk);
+ memcpy(span.begin, "suffix", 6);
+
+ EXPECT_EQ(reader.ResolveNextPosition(), Outcome::kChunkRead);
+ EXPECT_EQ(reader.num_scrapes(), 1u);
+ ASSERT_EQ(delegate.chunks.size(), 1u);
+ // Only what the writer had published, and nothing of the open fragment.
+ EXPECT_EQ(delegate.chunks[0].fragments,
+ (std::vector<std::string>{"committed"}));
+ EXPECT_EQ(ChunkStateOf(ring->LoadChunkStateWord(0)),
+ ChunkState::kRewriteRequested);
+
+ // The writer relocates its suffix, and the reader picks it up next pass.
+ ASSERT_EQ(writer.CloseFragment(6, false),
+ SharedRingBufferWriter::Outcome::kOk);
+ reader.Drain(8);
+ EXPECT_EQ(delegate.AllFragments(),
+ (std::vector<std::string>{"committed", "suffix"}));
+}
+
+TEST(SharedRingBufferReaderTest, ScrapeWithNothingCommittedStillMarksTheChunk) {
+ auto ring = SharedRingBuffer::Create(4, 512);
+ ASSERT_NE(ring, nullptr);
+ RecordingDelegate delegate;
+ SharedRingBufferReader reader(ring.get(), &delegate);
+ SharedRingBufferWriter writer(ring.get(), kWriterA, kBuffer,
+ BufferExhaustedPolicy::kDrop,
+ GetNoopSharedRingBufferWriterDelegate());
+
+ ASSERT_EQ(writer.OpenFragment(4, false).outcome,
+ SharedRingBufferWriter::Outcome::kOk);
+
+ EXPECT_EQ(reader.ResolveNextPosition(), Outcome::kSkipped);
+ EXPECT_TRUE(delegate.chunks.empty());
+ // Nothing came out, but the old owner is still stopped from publishing
+ // behind the reader.
+ EXPECT_EQ(ChunkStateOf(ring->LoadChunkStateWord(0)),
+ ChunkState::kRewriteRequested);
+}
+
+// The reader's mark loses to the writer's publication. The speculative copy is
+// discarded and the position is retried, so the fragments come out exactly
+// once and in order.
+TEST(SharedRingBufferReaderTest,
+ PublicationBeatingTheScrapeIsReadOnTheNextAttempt) {
+ auto ring = SharedRingBuffer::Create(4, 512);
+ ASSERT_NE(ring, nullptr);
+ RecordingDelegate delegate;
+ SharedRingBufferReader reader(ring.get(), &delegate);
+ SharedRingBufferWriter writer(ring.get(), kWriterA, kBuffer,
+ BufferExhaustedPolicy::kDrop,
+ GetNoopSharedRingBufferWriterDelegate());
+
+ ASSERT_EQ(WriteFragment(&writer, "first"),
+ SharedRingBufferWriter::Outcome::kOk);
+ const SharedRingBufferWriter::FragmentSpan span =
+ writer.OpenFragment(6, false);
+ ASSERT_EQ(span.outcome, SharedRingBufferWriter::Outcome::kOk);
+ memcpy(span.begin, "second", 6);
+
+ // Publish from inside the reader, after it has copied the prefix and just
+ // before it tries to claim it.
+ bool published = false;
+ reader.SetArbitrationHookForTesting([&] {
+ if (published)
+ return;
+ published = true;
+ EXPECT_EQ(writer.CloseFragment(6, false),
+ SharedRingBufferWriter::Outcome::kOk);
+ });
+
+ EXPECT_EQ(reader.ResolveNextPosition(), Outcome::kRetryLater);
+ EXPECT_EQ(reader.read_pos(), 0u);
+ EXPECT_TRUE(delegate.chunks.empty());
+
+ reader.SetArbitrationHookForTesting(nullptr);
+ EXPECT_EQ(reader.ResolveNextPosition(), Outcome::kChunkRead);
+ EXPECT_EQ(delegate.AllFragments(),
+ (std::vector<std::string>{"first", "second"}));
+ // No relocation happened: the writer won, so there was nothing to move.
+ EXPECT_EQ(writer.num_relocations(), 0u);
+ EXPECT_EQ(reader.num_scrapes(), 0u);
+}
+
+TEST(SharedRingBufferReaderTest, ReuseBeatingTheReclaimIsHandledAsAScrape) {
+ auto ring = SharedRingBuffer::Create(4, 512);
+ ASSERT_NE(ring, nullptr);
+ RecordingDelegate delegate;
+ SharedRingBufferReader reader(ring.get(), &delegate);
+ SharedRingBufferWriter writer(ring.get(), kWriterA, kBuffer,
+ BufferExhaustedPolicy::kDrop,
+ GetNoopSharedRingBufferWriterDelegate());
+
+ ASSERT_EQ(WriteFragment(&writer, "first"),
+ SharedRingBufferWriter::Outcome::kOk);
+
+ // The writer takes the Complete chunk back just as the reader is about to
+ // reclaim it. The reader leaves the position unchanged and takes the
+ // committed prefix on its next attempt.
+ bool reused = false;
+ reader.SetArbitrationHookForTesting([&] {
+ if (reused)
+ return;
+ reused = true;
+ ASSERT_EQ(writer.OpenFragment(3, false).outcome,
+ SharedRingBufferWriter::Outcome::kOk);
+ });
+
+ EXPECT_EQ(reader.ResolveNextPosition(), Outcome::kRetryLater);
+ EXPECT_EQ(reader.read_pos(), 0u);
+ EXPECT_TRUE(delegate.chunks.empty());
+
+ reader.SetArbitrationHookForTesting(nullptr);
+ EXPECT_EQ(reader.ResolveNextPosition(), Outcome::kChunkRead);
+ EXPECT_EQ(delegate.AllFragments(), (std::vector<std::string>{"first"}));
+ EXPECT_EQ(reader.num_scrapes(), 1u);
+ EXPECT_EQ(ChunkStateOf(ring->LoadChunkStateWord(0)),
+ ChunkState::kRewriteRequested);
+}
+
+// ---------------------------------------------------------------------------
+// Untrusted input.
+// ---------------------------------------------------------------------------
+
+TEST(SharedRingBufferReaderTest,
+ UnknownFormatDropsThePayloadButReleasesTheChunk) {
+ auto ring = SharedRingBuffer::Create(4, 256);
+ ASSERT_NE(ring, nullptr);
+ RecordingDelegate delegate;
+ SharedRingBufferReader reader(ring.get(), &delegate);
+
+ ASSERT_EQ(ring->TryReserveWritePos().position, 0u);
+ ring->SetStateWordForTesting(
+ 0, MakeDataStateWord(ChunkState::kComplete, ChunkFormat::kReservedRouting,
+ 0, 3, kWriterB));
+
+ EXPECT_EQ(reader.ResolveNextPosition(), Outcome::kSkipped);
+ EXPECT_EQ(reader.num_unknown_format_chunks(), 1u);
+ EXPECT_TRUE(delegate.chunks.empty());
+ EXPECT_EQ(delegate.writers_with_data_loss, (std::vector<WriterID>{kWriterB}));
+ // The loss was reported and the chunk remains usable.
+ EXPECT_EQ(ring->LoadChunkStateWord(0), MakeFreeStateWord(1));
+ EXPECT_EQ(reader.read_pos(), 1u);
+}
+
+TEST(SharedRingBufferReaderTest,
+ DirectoryLargerThanTheChunkIsRejectedBoundedly) {
+ auto ring = SharedRingBuffer::Create(4, 256);
+ ASSERT_NE(ring, nullptr);
+ RecordingDelegate delegate;
+ SharedRingBufferReader reader(ring.get(), &delegate);
+
+ ASSERT_EQ(ring->TryReserveWritePos().position, 0u);
+ // varint(0) takes one byte. 255 such entries do not fit in the 250-byte
+ // payload area.
+ ring->SetStateWordForTesting(
+ 0, MakeDataStateWord(ChunkState::kComplete, ChunkFormat::kTargetBuffer, 0,
+ 255, kWriterB));
+
+ EXPECT_EQ(reader.ResolveNextPosition(), Outcome::kSkipped);
+ EXPECT_EQ(reader.num_malformed_chunks(), 1u);
+ EXPECT_TRUE(delegate.chunks.empty());
+ EXPECT_EQ(delegate.writers_with_data_loss, (std::vector<WriterID>{kWriterB}));
+ EXPECT_EQ(ring->LoadChunkStateWord(0), MakeFreeStateWord(1));
+}
+
+TEST(SharedRingBufferReaderTest, PayloadAndDirectoryMustNotOverlap) {
+ auto ring = SharedRingBuffer::Create(4, 256);
+ ASSERT_NE(ring, nullptr);
+ RecordingDelegate delegate;
+ SharedRingBufferReader reader(ring.get(), &delegate);
+
+ ASSERT_EQ(ring->TryReserveWritePos().position, 0u);
+ // The chunk has 250 bytes after its fixed header. A 249-byte fragment fits
+ // by itself, but its two-byte size varint does not fit beside it.
+ WriteFragmentSize(ring->chunk_at(0) + 256, 249);
+ ring->SetStateWordForTesting(
+ 0, MakeDataStateWord(ChunkState::kComplete, ChunkFormat::kTargetBuffer, 0,
+ 1, kWriterB));
+
+ EXPECT_EQ(reader.ResolveNextPosition(), Outcome::kSkipped);
+ EXPECT_EQ(reader.num_malformed_chunks(), 1u);
+ EXPECT_EQ(delegate.writers_with_data_loss, (std::vector<WriterID>{kWriterB}));
+ EXPECT_EQ(ring->LoadChunkStateWord(0), MakeFreeStateWord(1));
+}
+
+TEST(SharedRingBufferReaderTest, SpeculativeParseDoesNotReportDataLoss) {
+ auto ring = SharedRingBuffer::Create(4, 256);
+ ASSERT_NE(ring, nullptr);
+ RecordingDelegate delegate;
+ SharedRingBufferReader reader(ring.get(), &delegate);
+
+ ASSERT_EQ(ring->TryReserveWritePos().position, 0u);
+ const uint32_t being_written = MakeDataStateWord(
+ ChunkState::kBeingWritten, ChunkFormat::kTargetBuffer, 0, 255, kWriterB);
+ ring->SetStateWordForTesting(0, being_written);
+ reader.SetArbitrationHookForTesting([&] {
+ uint32_t observed = being_written;
+ ASSERT_TRUE(ring->TrySetChunkComplete(
+ 0, &observed, ReplaceChunkState(being_written, ChunkState::kComplete)));
+ });
+
+ EXPECT_EQ(reader.ResolveNextPosition(), Outcome::kRetryLater);
+ EXPECT_EQ(reader.num_malformed_chunks(), 0u);
+ EXPECT_TRUE(delegate.writers_with_data_loss.empty());
+
+ reader.SetArbitrationHookForTesting(nullptr);
+ EXPECT_EQ(reader.ResolveNextPosition(), Outcome::kSkipped);
+ EXPECT_EQ(reader.num_malformed_chunks(), 1u);
+ EXPECT_EQ(delegate.writers_with_data_loss, (std::vector<WriterID>{kWriterB}));
+}
+
+TEST(SharedRingBufferReaderTest,
+ CumulativeFragmentSizesBeyondTheChunkAreRejected) {
+ auto ring = SharedRingBuffer::Create(4, 512);
+ ASSERT_NE(ring, nullptr);
+ RecordingDelegate delegate;
+ SharedRingBufferReader reader(ring.get(), &delegate);
+
+ ASSERT_EQ(ring->TryReserveWritePos().position, 0u);
+ // Two fragments claiming 400 bytes each in a chunk that has 506 bytes of
+ // payload area, of which four go to the directory.
+ uint8_t* chunk = ring->chunk_at(0);
+ uint8_t* directory_begin = chunk + 512;
+ directory_begin = WriteFragmentSize(directory_begin, 400);
+ WriteFragmentSize(directory_begin, 400);
+ ring->SetStateWordForTesting(
+ 0, MakeDataStateWord(ChunkState::kComplete, ChunkFormat::kTargetBuffer, 0,
+ 2, kWriterB));
+
+ EXPECT_EQ(reader.ResolveNextPosition(), Outcome::kSkipped);
+ EXPECT_EQ(reader.num_malformed_chunks(), 1u);
+ EXPECT_TRUE(delegate.chunks.empty());
+ EXPECT_EQ(ring->LoadChunkStateWord(0), MakeFreeStateWord(1));
+}
+
+TEST(SharedRingBufferReaderTest, UnterminatedFragmentSizeVarIntIsRejected) {
+ auto ring = SharedRingBuffer::Create(4, 512);
+ ASSERT_NE(ring, nullptr);
+ RecordingDelegate delegate;
+ SharedRingBufferReader reader(ring.get(), &delegate);
+
+ ASSERT_EQ(ring->TryReserveWritePos().position, 0u);
+ for (uint32_t i = 0; i < kMaxFragmentSizeVarIntBytes; ++i)
+ ring->chunk_at(0)[511 - i] = 0x80;
+ ring->SetStateWordForTesting(
+ 0, MakeDataStateWord(ChunkState::kComplete, ChunkFormat::kTargetBuffer, 0,
+ 1, kWriterB));
+
+ EXPECT_EQ(reader.ResolveNextPosition(), Outcome::kSkipped);
+ EXPECT_EQ(reader.num_malformed_chunks(), 1u);
+ EXPECT_TRUE(delegate.chunks.empty());
+ EXPECT_EQ(ring->LoadChunkStateWord(0), MakeFreeStateWord(1));
+}
+
+// A count larger than the writer actually wrote is a producer claim like any
+// other. It cannot make the reader leave the chunk or spin; it just decodes the
+// stale bytes underneath as extra fragments.
+TEST(SharedRingBufferReaderTest, InflatedFragmentCountStaysInsideTheChunk) {
+ auto ring = SharedRingBuffer::Create(4, 512);
+ ASSERT_NE(ring, nullptr);
+ RecordingDelegate delegate;
+ SharedRingBufferReader reader(ring.get(), &delegate);
+
+ ASSERT_EQ(ring->TryReserveWritePos().position, 0u);
+ ring->SetStateWordForTesting(
+ 0, MakeDataStateWord(ChunkState::kComplete, ChunkFormat::kTargetBuffer, 0,
+ 8, kWriterB));
+
+ EXPECT_EQ(reader.ResolveNextPosition(), Outcome::kChunkRead);
+ // Over a zero-filled mapping the inflated count decodes as extra zero-length
+ // fragments. The checks make the walk memory-safe; they do not pretend to
+ // make its contents true.
+ ASSERT_EQ(delegate.chunks.size(), 1u);
+ EXPECT_EQ(delegate.chunks[0].fragments.size(), 8u);
+ for (const std::string& fragment : delegate.chunks[0].fragments)
+ EXPECT_TRUE(fragment.empty());
+ EXPECT_EQ(reader.num_malformed_chunks(), 0u);
+ EXPECT_EQ(ring->LoadChunkStateWord(0), MakeFreeStateWord(1));
+ EXPECT_EQ(reader.read_pos(), 1u);
+}
+
+TEST(SharedRingBufferReaderTest,
+ MalformedBeingWrittenChunkIsStillRewriteRequested) {
+ auto ring = SharedRingBuffer::Create(4, 256);
+ ASSERT_NE(ring, nullptr);
+ RecordingDelegate delegate;
+ SharedRingBufferReader reader(ring.get(), &delegate);
+
+ ASSERT_EQ(ring->TryReserveWritePos().position, 0u);
+ ring->SetStateWordForTesting(
+ 0, MakeDataStateWord(ChunkState::kBeingWritten,
+ ChunkFormat::kTargetBuffer, 0, 255, kWriterB));
+
+ EXPECT_EQ(reader.ResolveNextPosition(), Outcome::kSkipped);
+ EXPECT_EQ(reader.num_malformed_chunks(), 1u);
+ // Malformed bytes may be dropped. What must not happen is leaving an
+ // BeingWritten owner able to publish behind the reader.
+ EXPECT_EQ(ChunkStateOf(ring->LoadChunkStateWord(0)),
+ ChunkState::kRewriteRequested);
+}
+
+TEST(SharedRingBufferReaderTest,
+ ReservedStateStopsTheRingWithoutChangingAnything) {
+ auto ring = SharedRingBuffer::Create(4, 256);
+ ASSERT_NE(ring, nullptr);
+ RecordingDelegate delegate;
+ SharedRingBufferReader reader(ring.get(), &delegate);
+
+ ASSERT_EQ(ring->TryReserveWritePos().position, 0u);
+ const uint32_t reserved_word = 0x00000005u;
+ ring->SetStateWordForTesting(0, reserved_word);
+
+ EXPECT_EQ(reader.ResolveNextPosition(), Outcome::kProtocolError);
+ EXPECT_TRUE(reader.has_protocol_error());
+ EXPECT_EQ(reader.read_pos(), 0u);
+ EXPECT_EQ(ring->LoadChunkStateWord(0), reserved_word);
+
+ // The ring is never read again, and a drain over it changes nothing.
+ const SharedRingBufferReader::DrainResult result = reader.Drain(8);
+ EXPECT_EQ(result.positions_resolved, 0u);
+ EXPECT_EQ(result.last_outcome, Outcome::kProtocolError);
+ EXPECT_EQ(ring->read_pos_for_testing(), 0u);
+}
+
+TEST(SharedRingBufferReaderTest, FreeWordFromAnotherTraversalStopsTheRing) {
+ auto ring = SharedRingBuffer::Create(4, 256);
+ ASSERT_NE(ring, nullptr);
+ RecordingDelegate delegate;
+ SharedRingBufferReader reader(ring.get(), &delegate);
+
+ ASSERT_EQ(ring->TryReserveWritePos().position, 0u);
+ // Position 0 expects Free(0). Only the reader writes a free word, and
+ // it derives it from the position it is resolving, so no legal execution puts
+ // this here.
+ const uint32_t wrong_wrap = MakeFreeStateWord(3);
+ ring->SetStateWordForTesting(0, wrong_wrap);
+
+ EXPECT_EQ(reader.ResolveNextPosition(), Outcome::kProtocolError);
+ EXPECT_TRUE(reader.has_protocol_error());
+ EXPECT_EQ(reader.read_pos(), 0u);
+ EXPECT_EQ(ring->LoadChunkStateWord(0), wrong_wrap);
+}
+
+// The control and fragment-count fields of a Free word are zero. A word using
+// them is some future ABI's, not garbage to be normalized away: the reader
+// must not consume the position as if it understood the word.
+TEST(SharedRingBufferReaderTest, FreeWordWithReservedBitsSetStopsTheRing) {
+ auto ring = SharedRingBuffer::Create(4, 256);
+ ASSERT_NE(ring, nullptr);
+ RecordingDelegate delegate;
+ SharedRingBufferReader reader(ring.get(), &delegate);
+
+ ASSERT_EQ(ring->TryReserveWritePos().position, 0u);
+ // The wrap count is position 0's, but a reserved bit is set.
+ const uint32_t reserved_bit_word = MakeFreeStateWord(0) | (1u << 8);
+ ring->SetStateWordForTesting(0, reserved_bit_word);
+
+ EXPECT_EQ(reader.ResolveNextPosition(), Outcome::kProtocolError);
+ EXPECT_TRUE(reader.has_protocol_error());
+ EXPECT_EQ(reader.read_pos(), 0u);
+ EXPECT_EQ(ring->LoadChunkStateWord(0), reserved_bit_word);
+}
+
+// A word with the RewriteAcknowledged state bits but a nonzero payload is not
+// the word that transition may leave from. The reclaim's failed exact-value
+// CAS hands the reader the forged word itself, and the ring stops on it.
+TEST(SharedRingBufferReaderTest, ForgedRewriteAcknowledgedWordStopsTheRing) {
+ auto ring = SharedRingBuffer::Create(4, 256);
+ ASSERT_NE(ring, nullptr);
+ RecordingDelegate delegate;
+ SharedRingBufferReader reader(ring.get(), &delegate);
+
+ ASSERT_EQ(ring->TryReserveWritePos().position, 0u);
+ const uint32_t forged = kRewriteAcknowledgedStateWord | kFlagDataLoss;
+ ring->SetStateWordForTesting(0, forged);
+
+ EXPECT_EQ(reader.ResolveNextPosition(), Outcome::kProtocolError);
+ EXPECT_TRUE(reader.has_protocol_error());
+ EXPECT_EQ(reader.read_pos(), 0u);
+ EXPECT_EQ(ring->LoadChunkStateWord(0), forged);
+}
+
+// A writer reserves a position only after seeing fewer than num_chunks
+// outstanding, so write_pos further ahead than that cannot come from a legal
+// producer. Once the mapping is writable by another process, believing it would
+// mean resolving positions that were never reserved, one drain pass after
+// another, for as long as the producer keeps write_pos there.
+TEST(SharedRingBufferReaderTest, TooManyOutstandingPositionsStopsTheRing) {
+ auto ring = SharedRingBuffer::Create(4, 256);
+ ASSERT_NE(ring, nullptr);
+ RecordingDelegate delegate;
+ SharedRingBufferReader reader(ring.get(), &delegate);
+
+ // Four chunks, five outstanding positions.
+ ring->SetWritePosForTesting(5);
+
+ EXPECT_EQ(reader.ResolveNextPosition(), Outcome::kProtocolError);
+ EXPECT_TRUE(reader.has_protocol_error());
+ EXPECT_EQ(reader.read_pos(), 0u);
+ EXPECT_EQ(ring->read_pos_for_testing(), 0u);
+ EXPECT_TRUE(delegate.chunks.empty());
+}
+
+// The boundary itself is legal and must still be drained: num_chunks
+// outstanding positions is exactly a full ring.
+TEST(SharedRingBufferReaderTest, AFullRingIsNotAnImpossibleDistance) {
+ auto ring = SharedRingBuffer::Create(4, 256);
+ ASSERT_NE(ring, nullptr);
+ RecordingDelegate delegate;
+ SharedRingBufferReader reader(ring.get(), &delegate);
+
+ SharedRingBufferWriter writer(ring.get(), kWriterA, kBuffer,
+ BufferExhaustedPolicy::kDrop,
+ GetNoopSharedRingBufferWriterDelegate());
+ for (uint32_t i = 0; i < 4; ++i) {
+ ASSERT_EQ(WriteFragment(&writer, "x"),
+ SharedRingBufferWriter::Outcome::kOk);
+ writer.FinishCurrentChunk();
+ }
+ ASSERT_EQ(ring->LoadWritePos(), 4u);
+
+ const SharedRingBufferReader::DrainResult result = reader.Drain(8);
+ EXPECT_FALSE(reader.has_protocol_error());
+ EXPECT_EQ(result.positions_resolved, 4u);
+ EXPECT_EQ(delegate.chunks.size(), 4u);
+}
+
+// ---------------------------------------------------------------------------
+// Flags and routing survive the trip.
+// ---------------------------------------------------------------------------
+
+TEST(SharedRingBufferReaderTest,
+ ContinuationFlagsAndTargetBufferReachTheDelegate) {
+ auto ring = SharedRingBuffer::Create(8, 512);
+ ASSERT_NE(ring, nullptr);
+ RecordingDelegate delegate;
+ SharedRingBufferReader reader(ring.get(), &delegate);
+
+ SharedRingBufferWriter first(ring.get(), kWriterA, 11,
+ BufferExhaustedPolicy::kDrop,
+ GetNoopSharedRingBufferWriterDelegate());
+ first.RecordDataLoss();
+ ASSERT_EQ(WriteFragment(&first, "head", /*continues_from_prev=*/false,
+ /*continues_on_next=*/true),
+ SharedRingBufferWriter::Outcome::kOk);
+ ASSERT_EQ(WriteFragment(&first, "tail", /*continues_from_prev=*/true,
+ /*continues_on_next=*/false),
+ SharedRingBufferWriter::Outcome::kOk);
+ first.FinishCurrentChunk();
+
+ SharedRingBufferWriter second(ring.get(), kWriterB, 22,
+ BufferExhaustedPolicy::kDrop,
+ GetNoopSharedRingBufferWriterDelegate());
+ ASSERT_EQ(WriteFragment(&second, "other"),
+ SharedRingBufferWriter::Outcome::kOk);
+ second.FinishCurrentChunk();
+
+ reader.Drain(16);
+ ASSERT_EQ(delegate.chunks.size(), 3u);
+
+ EXPECT_EQ(delegate.chunks[0].target_buffer, 11);
+ EXPECT_EQ(delegate.chunks[0].payload_flags,
+ kFlagDataLoss | kFlagContinuesOnNextChunk);
+ EXPECT_EQ(delegate.chunks[1].target_buffer, 11);
+ EXPECT_EQ(delegate.chunks[1].payload_flags, kFlagContinuesFromPrevChunk);
+ EXPECT_EQ(delegate.chunks[2].target_buffer, 22);
+ EXPECT_EQ(delegate.chunks[2].writer_id, kWriterB);
+ EXPECT_EQ(delegate.chunks[2].payload_flags, 0u);
+}
+
+// A one-chunk ring is legal: every position maps to chunk 0 and the ring
+// alternates between one outstanding reservation and empty.
+TEST(SharedRingBufferReaderTest, OneChunkRingRoundTrips) {
+ auto ring = SharedRingBuffer::Create(1, 256);
+ ASSERT_NE(ring, nullptr);
+ RecordingDelegate delegate;
+ SharedRingBufferReader reader(ring.get(), &delegate);
+ SharedRingBufferWriter writer(ring.get(), kWriterA, kBuffer,
+ BufferExhaustedPolicy::kDrop,
+ GetNoopSharedRingBufferWriterDelegate());
+
+ std::vector<std::string> expected;
+ for (uint32_t i = 0; i < 5; ++i) {
+ const std::string bytes = "packet-" + std::to_string(i);
+ ASSERT_EQ(WriteFragment(&writer, bytes),
+ SharedRingBufferWriter::Outcome::kOk)
+ << i;
+ writer.FinishCurrentChunk();
+ expected.push_back(bytes);
+ const SharedRingBufferReader::DrainResult result = reader.Drain(4);
+ EXPECT_EQ(result.positions_resolved, 1u) << i;
+ }
+ EXPECT_EQ(delegate.AllFragments(), expected);
+ EXPECT_EQ(reader.ResolveNextPosition(), Outcome::kNoData);
+}
+
+// An aligned, non-power-of-two chunk size, end to end through writer and
+// reader.
+TEST(SharedRingBufferReaderTest, NonPowerOfTwoChunkSizeRoundTrips) {
+ auto ring = SharedRingBuffer::Create(4, 260);
+ ASSERT_NE(ring, nullptr);
+ RecordingDelegate delegate;
+ SharedRingBufferReader reader(ring.get(), &delegate);
+ SharedRingBufferWriter writer(ring.get(), kWriterA, kBuffer,
+ BufferExhaustedPolicy::kDrop,
+ GetNoopSharedRingBufferWriterDelegate());
+
+ // 300 bytes does not fit a 260-byte chunk, so it arrives as two fragments in
+ // two chunks - the split the upper layer would mark with the continuation
+ // flags.
+ const std::string head(252, 'h');
+ const std::string tail(48, 't');
+ ASSERT_EQ(WriteFragment(&writer, head, /*continues_from_prev=*/false,
+ /*continues_on_next=*/true),
+ SharedRingBufferWriter::Outcome::kOk);
+ ASSERT_EQ(WriteFragment(&writer, tail, /*continues_from_prev=*/true,
+ /*continues_on_next=*/false),
+ SharedRingBufferWriter::Outcome::kOk);
+ writer.FinishCurrentChunk();
+
+ reader.Drain(8);
+ ASSERT_EQ(delegate.chunks.size(), 2u);
+ EXPECT_EQ(delegate.chunks[0].payload_flags, kFlagContinuesOnNextChunk);
+ EXPECT_EQ(delegate.chunks[1].payload_flags, kFlagContinuesFromPrevChunk);
+ EXPECT_EQ(delegate.AllFragments(), (std::vector<std::string>{head, tail}));
+}
+
+TEST(SharedRingBufferReaderTest,
+ FragmentsComeOutInReservationOrderAcrossWriters) {
+ auto ring = SharedRingBuffer::Create(8, 256);
+ ASSERT_NE(ring, nullptr);
+ RecordingDelegate delegate;
+ SharedRingBufferReader reader(ring.get(), &delegate);
+
+ // Interleave two writers so their chunks land at alternating positions.
+ SharedRingBufferWriter a(ring.get(), kWriterA, kBuffer,
+ BufferExhaustedPolicy::kDrop,
+ GetNoopSharedRingBufferWriterDelegate());
+ SharedRingBufferWriter b(ring.get(), kWriterB, kBuffer,
+ BufferExhaustedPolicy::kDrop,
+ GetNoopSharedRingBufferWriterDelegate());
+ std::vector<std::string> expected;
+ for (uint32_t i = 0; i < 4; ++i) {
+ // Each fragment fills its chunk, so every write consumes a new position.
+ const std::string from_a = "a" + std::to_string(i) + std::string(240, '.');
+ const std::string from_b = "b" + std::to_string(i) + std::string(240, '.');
+ ASSERT_EQ(WriteFragment(&a, from_a), SharedRingBufferWriter::Outcome::kOk);
+ ASSERT_EQ(WriteFragment(&b, from_b), SharedRingBufferWriter::Outcome::kOk);
+ expected.push_back(from_a);
+ expected.push_back(from_b);
+ reader.Drain(16);
+ }
+ a.FinishCurrentChunk();
+ b.FinishCurrentChunk();
+ reader.Drain(16);
+
+ EXPECT_EQ(delegate.AllFragments(), expected);
+}
+
+} // namespace
+} // namespace perfetto::tracing_v2
diff --git a/src/tracing/v2/shared_ring_buffer_unittest.cc b/src/tracing/v2/shared_ring_buffer_unittest.cc
new file mode 100644
index 0000000..6c07be9
--- /dev/null
+++ b/src/tracing/v2/shared_ring_buffer_unittest.cc
@@ -0,0 +1,851 @@
+/*
+ * Copyright (C) 2026 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/v2/shared_ring_buffer.h"
+
+#include <errno.h>
+#include <stdint.h>
+
+#include <atomic>
+#include <memory>
+#include <optional>
+#include <thread>
+#include <vector>
+
+#include "perfetto/base/build_config.h"
+#include "perfetto/base/time.h"
+#include "perfetto/ext/base/waitable_event.h"
+#include "src/tracing/v2/tracing_v2_abi.h"
+#include "test/gtest_and_gmock.h"
+
+namespace perfetto::tracing_v2 {
+namespace {
+
+constexpr uint32_t kChunkSize = 256;
+constexpr WriterID kWriterA = 7;
+constexpr WriterID kWriterB = 9;
+
+uint32_t BeingWrittenWord(WriterID writer) {
+ return MakeDataStateWord(ChunkState::kBeingWritten,
+ ChunkFormat::kTargetBuffer, 0, 0, writer);
+}
+
+uint32_t CompleteWord(WriterID writer, uint32_t num_fragments) {
+ return MakeDataStateWord(ChunkState::kComplete, ChunkFormat::kTargetBuffer, 0,
+ num_fragments, writer);
+}
+
+// Reads a chunk's state word without going through the ring's own accessors,
+// so that a test observing the ring cannot be fooled by a bug in them.
+uint32_t PeekStateWord(SharedRingBuffer* ring, uint32_t chunk_index) {
+ const uint8_t* chunk = ring->chunk_at(chunk_index);
+ return static_cast<uint32_t>(chunk[0]) |
+ (static_cast<uint32_t>(chunk[1]) << 8) |
+ (static_cast<uint32_t>(chunk[2]) << 16) |
+ (static_cast<uint32_t>(chunk[3]) << 24);
+}
+
+// Spins until |condition| holds. Bounded: returns false after a deadline no
+// correct implementation comes anywhere near, so a broken wait path fails the
+// test instead of hanging the process. A caller must still unblock and join
+// every thread it spawned before asserting on the result.
+template <typename ConditionFn>
+bool SpinUntil(ConditionFn condition) {
+ const base::TimeMillis deadline =
+ base::GetWallTimeMs() + base::TimeMillis(30000);
+ while (base::GetWallTimeMs() < deadline) {
+ if (condition())
+ return true;
+ std::this_thread::yield();
+ }
+ return false;
+}
+
+// ---------------------------------------------------------------------------
+// Ring dimensions.
+// ---------------------------------------------------------------------------
+
+TEST(SharedRingBufferTest, RejectsInvalidChunkCountsAndSizes) {
+ // num_chunks must be a power of two no larger than 2^30.
+ EXPECT_EQ(SharedRingBuffer::Create(0, kChunkSize), nullptr);
+ EXPECT_EQ(SharedRingBuffer::Create(3, kChunkSize), nullptr);
+ EXPECT_EQ(SharedRingBuffer::Create(6, kChunkSize), nullptr);
+ EXPECT_EQ(SharedRingBuffer::Create(uint32_t{1} << 31, kChunkSize), nullptr);
+
+ // chunk_size must be at least 256 and keep the state word aligned.
+ EXPECT_EQ(SharedRingBuffer::Create(4, 0), nullptr);
+ EXPECT_EQ(SharedRingBuffer::Create(4, 128), nullptr);
+ EXPECT_EQ(SharedRingBuffer::Create(4, 255), nullptr);
+ EXPECT_EQ(SharedRingBuffer::Create(4, 258), nullptr);
+}
+
+TEST(SharedRingBufferTest, AcceptsValidChunkCountsAndSizes) {
+ for (uint32_t chunk_size : {256u, 260u, 512u, 1000u, 4096u, 65536u}) {
+ auto ring = SharedRingBuffer::Create(4, chunk_size);
+ ASSERT_NE(ring, nullptr) << chunk_size;
+ EXPECT_EQ(ring->chunk_size(), chunk_size);
+ }
+ for (uint32_t num_chunks : {1u, 2u, 4u, 8u, 16u, 1024u}) {
+ auto ring = SharedRingBuffer::Create(num_chunks, kChunkSize);
+ ASSERT_NE(ring, nullptr) << num_chunks;
+ EXPECT_EQ(ring->num_chunks(), num_chunks);
+ EXPECT_EQ(ring->chunk_index_bits(), GetChunkIndexBits(num_chunks));
+ }
+}
+
+// chunk_size must keep the state word aligned and be at least 256 bytes;
+// nothing requires it to be a power of two and there is no arbitrary maximum.
+TEST(SharedRingBufferTest, AcceptsAlignedNonPowerOfTwoChunkSizes) {
+ EXPECT_NE(SharedRingBuffer::Create(4, 260), nullptr);
+ EXPECT_NE(SharedRingBuffer::Create(4, 1000), nullptr);
+ EXPECT_NE(SharedRingBuffer::Create(4, 65536), nullptr);
+ // Still rejected: a misaligned state word, or a chunk below the minimum.
+ EXPECT_EQ(SharedRingBuffer::Create(4, 258), nullptr);
+ EXPECT_EQ(SharedRingBuffer::Create(4, 255), nullptr);
+}
+
+// Free(0) is the all-zero word, so a fresh zero-filled mapping is
+// already correct and the ring does not walk it at construction.
+TEST(SharedRingBufferTest, FreshMappingIsFreeZeroWithNoStampingPass) {
+ auto ring = SharedRingBuffer::Create(1024, kChunkSize);
+ ASSERT_NE(ring, nullptr);
+ for (uint32_t i = 0; i < ring->num_chunks(); ++i) {
+ ASSERT_EQ(PeekStateWord(ring.get(), i), 0u) << i;
+ ASSERT_EQ(ChunkStateOf(ring->LoadChunkStateWord(i)), ChunkState::kFree);
+ ASSERT_EQ(WrapCountOf(ring->LoadChunkStateWord(i)), 0u);
+ }
+ EXPECT_EQ(ring->read_pos_for_testing(), 0u);
+ EXPECT_EQ(ring->LoadWritePos(), 0u);
+ EXPECT_EQ(ring->num_writers_waiting_for_testing(), 0u);
+}
+
+TEST(SharedRingBufferTest, EveryChunkStateWordIsNaturallyAligned) {
+ for (uint32_t chunk_size : {256u, 260u, 512u, 4096u, 65536u}) {
+ auto ring = SharedRingBuffer::Create(8, chunk_size);
+ ASSERT_NE(ring, nullptr);
+ for (uint32_t i = 0; i < ring->num_chunks(); ++i) {
+ const auto address = reinterpret_cast<uintptr_t>(ring->chunk_at(i));
+ // The ABI's alignment requirement: every state word is naturally
+ // aligned, whatever the stride.
+ ASSERT_EQ(address % alignof(std::atomic<uint32_t>), 0u)
+ << chunk_size << "/" << i;
+ // For a cache-line-multiple stride the chunks additionally stay
+ // line-aligned, which keeps two writers on adjacent chunks off each
+ // other's lines. That is a property of those chunk sizes, not of the
+ // ABI.
+ if (chunk_size % 64 == 0) {
+ ASSERT_EQ(address % 64, 0u) << chunk_size << "/" << i;
+ }
+ }
+ }
+}
+
+// The header is one 64-byte line and chunk 0 starts on the next one. The
+// mapping is page-aligned, so the chunk area's misalignment from any
+// page-divisor boundary is exactly the header size - which is how the header's
+// size stays observable from outside the class.
+TEST(SharedRingBufferTest, ChunksStartRightAfterTheOneLineHeader) {
+ for (uint32_t chunk_size : {256u, 260u, 4096u}) {
+ auto ring = SharedRingBuffer::Create(4, chunk_size);
+ ASSERT_NE(ring, nullptr);
+ const uintptr_t first = reinterpret_cast<uintptr_t>(ring->chunk_at(0));
+ EXPECT_EQ(first % 256, 64u) << "chunk_size " << chunk_size;
+ const uintptr_t last = reinterpret_cast<uintptr_t>(ring->chunk_at(3));
+ EXPECT_EQ(last - first, 3u * chunk_size);
+ }
+}
+
+// The whole mapping-size policy, branch by branch, without asking the
+// allocator for anything.
+TEST(SharedRingBufferTest, MappingSizeArithmeticIsChecked) {
+ constexpr size_t kPage = 4096;
+ // The nominal case is the header plus the chunk area, exactly.
+ EXPECT_EQ(SharedRingBuffer::ComputeAllocationSizeForTesting(4, 256, kPage),
+ std::make_optional<size_t>(64 + 4 * 256));
+
+ // The largest legal ring is representable on a 64-bit host and nothing
+ // smaller can overflow there; on a 32-bit host the same request must be
+ // rejected by arithmetic, not by a crash inside the allocator.
+ const auto largest = SharedRingBuffer::ComputeAllocationSizeForTesting(
+ 1u << 30, 0xfffffffcu, kPage);
+ if (sizeof(size_t) >= 8) {
+ ASSERT_TRUE(largest.has_value());
+ EXPECT_EQ(*largest, 64 + (uint64_t{1} << 30) * 0xfffffffcull);
+ } else {
+ EXPECT_FALSE(largest.has_value());
+ }
+
+ // The guard-space headroom: a total that fits in size_t but whose padded
+ // allocator request would not is rejected here, because
+ // PagedMemory::Allocate() does its rounding and guard-page addition in
+ // unchecked size_t arithmetic.
+ EXPECT_FALSE(
+ SharedRingBuffer::ComputeAllocationSizeForTesting(4, 256, SIZE_MAX / 3)
+ .has_value());
+ EXPECT_FALSE(
+ SharedRingBuffer::ComputeAllocationSizeForTesting(4, 256, SIZE_MAX / 2)
+ .has_value());
+}
+
+// ---------------------------------------------------------------------------
+// Reserving positions.
+// ---------------------------------------------------------------------------
+
+TEST(SharedRingBufferTest, ReservationsAreConsecutiveTicketsUntilFull) {
+ auto ring = SharedRingBuffer::Create(4, kChunkSize);
+ ASSERT_NE(ring, nullptr);
+
+ for (uint32_t expected = 0; expected < 4; ++expected) {
+ const auto reservation = ring->TryReserveWritePos();
+ ASSERT_EQ(reservation.result, SharedRingBuffer::ReserveResult::kReserved);
+ EXPECT_EQ(reservation.position, expected);
+ }
+
+ const auto full = ring->TryReserveWritePos();
+ EXPECT_EQ(full.result, SharedRingBuffer::ReserveResult::kFull);
+ // Nothing was reserved, so write_pos did not move: a full ring must not burn
+ // a position.
+ EXPECT_EQ(ring->LoadWritePos(), 4u);
+ // The sample the decision was taken against is exactly what a stalling
+ // writer has to wait on.
+ EXPECT_EQ(full.read_pos_sample, 0u);
+}
+
+TEST(SharedRingBufferTest, CapacityFollowsReadPos) {
+ auto ring = SharedRingBuffer::Create(2, kChunkSize);
+ ASSERT_NE(ring, nullptr);
+
+ ASSERT_EQ(ring->TryReserveWritePos().result,
+ SharedRingBuffer::ReserveResult::kReserved);
+ ASSERT_EQ(ring->TryReserveWritePos().result,
+ SharedRingBuffer::ReserveResult::kReserved);
+ ASSERT_EQ(ring->TryReserveWritePos().result,
+ SharedRingBuffer::ReserveResult::kFull);
+
+ // Publishing read_pos must preserve a concurrently updated write_pos.
+ ring->PublishReadPos(1);
+ EXPECT_EQ(ring->read_pos_for_testing(), 1u);
+ EXPECT_EQ(ring->LoadWritePos(), 2u);
+
+ const auto reservation = ring->TryReserveWritePos();
+ ASSERT_EQ(reservation.result, SharedRingBuffer::ReserveResult::kReserved);
+ EXPECT_EQ(reservation.position, 2u);
+ EXPECT_EQ(reservation.read_pos_sample, 1u);
+ EXPECT_EQ(ring->read_pos_for_testing(), 1u);
+ EXPECT_EQ(ring->LoadWritePos(), 3u);
+}
+
+// One load of the packed positions decides capacity across uint32_t rollover.
+TEST(SharedRingBufferTest, CapacityDecisionsAreExactAcrossPositionRollover) {
+ auto ring = SharedRingBuffer::Create(8, kChunkSize);
+ ASSERT_NE(ring, nullptr);
+ const uint32_t kSeed = 0xfffffffcu; // Four positions before the rollover.
+ ring->SetPositionsForTesting(kSeed);
+
+ for (uint32_t i = 0; i < 8; ++i) {
+ const auto reservation = ring->TryReserveWritePos();
+ ASSERT_EQ(reservation.result, SharedRingBuffer::ReserveResult::kReserved)
+ << i;
+ EXPECT_EQ(reservation.position, kSeed + i) << i;
+ }
+ // write_pos has wrapped: 0xfffffffc + 8 = 4.
+ EXPECT_EQ(ring->LoadWritePos(), 4u);
+
+ const auto full = ring->TryReserveWritePos();
+ ASSERT_EQ(full.result, SharedRingBuffer::ReserveResult::kFull);
+ EXPECT_EQ(full.read_pos_sample, kSeed);
+
+ ring->PublishReadPos(kSeed + 1);
+ const auto reservation = ring->TryReserveWritePos();
+ ASSERT_EQ(reservation.result, SharedRingBuffer::ReserveResult::kReserved);
+ EXPECT_EQ(reservation.position, 4u);
+ EXPECT_EQ(reservation.read_pos_sample, kSeed + 1);
+}
+
+// ---------------------------------------------------------------------------
+// The two read/write-position compare-and-swap conflicts, forced
+// deterministically.
+//
+// Two racing threads cannot be scheduled into "load, then lose the CAS" on
+// demand. These tests start the production loop with an old rw_positions
+// value. The first compare-and-swap must fail, and its expected argument is
+// replaced with the current value. The test then checks that the retry uses
+// both halves of that current value.
+// ---------------------------------------------------------------------------
+
+// A reader publication lands between the writer's load and its CAS. The
+// reservation must recheck capacity with the value returned by the failed CAS,
+// not reuse either half of the old value.
+TEST(SharedRingBufferTest, ReservationCasLosesToPublicationAndRedispatches) {
+ auto ring = SharedRingBuffer::Create(4, kChunkSize);
+ ASSERT_NE(ring, nullptr);
+ ASSERT_EQ(ring->TryReserveWritePos().position, 0u);
+ ASSERT_EQ(ring->TryReserveWritePos().position, 1u);
+
+ // The value the writer loaded before the reader published.
+ const uint64_t stale_rw_positions = PackRwPositions(2, 0);
+ ring->PublishReadPos(1); // Now (write=2, read=1).
+
+ const auto reservation =
+ ring->TryReserveWritePosForTesting(stale_rw_positions);
+ ASSERT_EQ(reservation.result, SharedRingBuffer::ReserveResult::kReserved);
+ EXPECT_EQ(reservation.position, 2u);
+ // The failed CAS returned read_pos=1. Seeing that value here proves that the
+ // retry did not keep read_pos=0 from the old value.
+ EXPECT_EQ(reservation.read_pos_sample, 1u);
+ EXPECT_EQ(ring->LoadWritePos(), 3u);
+ EXPECT_EQ(ring->read_pos_for_testing(), 1u);
+}
+
+// The same losing CAS, but the value returned by the failure says that the
+// ring is full. The retry must take the Full exit, without burning a position.
+TEST(SharedRingBufferTest, ReservationCasLossRedispatchesIntoFull) {
+ auto ring = SharedRingBuffer::Create(2, kChunkSize);
+ ASSERT_NE(ring, nullptr);
+ ASSERT_EQ(ring->TryReserveWritePos().position, 0u);
+
+ const uint64_t stale_rw_positions =
+ PackRwPositions(1, 0); // One outstanding: capacity left.
+ ASSERT_EQ(ring->TryReserveWritePos().position, 1u); // Now (2, 0): full.
+
+ const auto reservation =
+ ring->TryReserveWritePosForTesting(stale_rw_positions);
+ EXPECT_EQ(reservation.result, SharedRingBuffer::ReserveResult::kFull);
+ EXPECT_EQ(reservation.read_pos_sample, 0u);
+ EXPECT_EQ(ring->LoadWritePos(), 2u);
+}
+
+// A writer reservation lands between the reader's load and its CAS. The
+// publication retry must preserve the returned write_pos: had it kept the
+// stale half, the reservation would be erased and write_pos would read 0.
+TEST(SharedRingBufferTest, PublicationCasLosesToReservationAndPreservesIt) {
+ auto ring = SharedRingBuffer::Create(4, kChunkSize);
+ ASSERT_NE(ring, nullptr);
+
+ // The value loaded by the reader before the writer reserved position 0.
+ const uint64_t stale_rw_positions = PackRwPositions(0, 0);
+ ASSERT_EQ(ring->TryReserveWritePos().position, 0u); // Now (write=1, read=0).
+
+ ring->PublishReadPosForTesting(stale_rw_positions, 1);
+ EXPECT_EQ(ring->LoadWritePos(), 1u);
+ EXPECT_EQ(ring->read_pos_for_testing(), 1u);
+}
+
+// The ABI accepts any power-of-two chunk count in [1, 2^30].
+TEST(SharedRingBufferTest, OneChunkRingWorks) {
+ auto ring = SharedRingBuffer::Create(1, kChunkSize);
+ ASSERT_NE(ring, nullptr);
+ EXPECT_EQ(ring->num_chunks(), 1u);
+ EXPECT_EQ(ring->chunk_index_bits(), 0u);
+
+ const auto reservation = ring->TryReserveWritePos();
+ ASSERT_EQ(reservation.result, SharedRingBuffer::ReserveResult::kReserved);
+ EXPECT_EQ(reservation.position, 0u);
+ EXPECT_EQ(ring->TryReserveWritePos().result,
+ SharedRingBuffer::ReserveResult::kFull);
+
+ ASSERT_TRUE(ring->TryAcquireChunkForWriting(0, BeingWrittenWord(kWriterA)));
+ uint32_t observed = BeingWrittenWord(kWriterA);
+ ASSERT_TRUE(
+ ring->TrySetChunkComplete(0, &observed, CompleteWord(kWriterA, 1)));
+ observed = CompleteWord(kWriterA, 1);
+ ASSERT_TRUE(ring->TryReleaseCompleteChunkAsFree(0, &observed));
+ // With one chunk the wrap count advances on every position.
+ EXPECT_EQ(PeekStateWord(ring.get(), 0), MakeFreeStateWord(1));
+ ring->PublishReadPos(1);
+
+ const auto next = ring->TryReserveWritePos();
+ ASSERT_EQ(next.result, SharedRingBuffer::ReserveResult::kReserved);
+ EXPECT_EQ(next.position, 1u);
+ EXPECT_TRUE(ring->TryAcquireChunkForWriting(1, BeingWrittenWord(kWriterB)));
+}
+
+// ---------------------------------------------------------------------------
+// The writer's transitions.
+// ---------------------------------------------------------------------------
+
+TEST(SharedRingBufferTest, ClaimTakesTheChunkForTheReservedPosition) {
+ auto ring = SharedRingBuffer::Create(4, kChunkSize);
+ ASSERT_NE(ring, nullptr);
+
+ const auto reservation = ring->TryReserveWritePos();
+ ASSERT_EQ(reservation.result, SharedRingBuffer::ReserveResult::kReserved);
+ EXPECT_TRUE(ring->TryAcquireChunkForWriting(reservation.position,
+ BeingWrittenWord(kWriterA)));
+ EXPECT_EQ(PeekStateWord(ring.get(), 0), BeingWrittenWord(kWriterA));
+}
+
+// A writer that reserved a position and then slept wakes up expecting the free
+// word for *its* traversal. Once the reader has moved the chunk on, that word
+// is gone, so the late claim cannot land. This is the schedule an all-zero free
+// word could not survive.
+TEST(SharedRingBufferTest, StaleClaimFailsOnceTheReaderHasAdvancedTheWrap) {
+ auto ring = SharedRingBuffer::Create(4, kChunkSize);
+ ASSERT_NE(ring, nullptr);
+
+ // A writer reserves position 4 (chunk 0, wrap 1) and stalls.
+ for (uint32_t i = 0; i < 4; ++i)
+ ASSERT_EQ(ring->TryReserveWritePos().result,
+ SharedRingBuffer::ReserveResult::kReserved);
+ ring->PublishReadPos(4);
+ const auto stale = ring->TryReserveWritePos();
+ ASSERT_EQ(stale.result, SharedRingBuffer::ReserveResult::kReserved);
+ ASSERT_EQ(stale.position, 4u);
+
+ // The reader resolves positions 0 to 4 as holes, so chunk 0 ends up tagged
+ // for position 8's traversal.
+ for (uint32_t position = 0; position <= 4; ++position) {
+ const uint32_t chunk_index = ChunkIndexOfPosition(position, 4);
+ uint32_t observed = ring->LoadChunkStateWord(chunk_index);
+ ASSERT_TRUE(ring->TryMoveFreeChunkToNextWrap(position, &observed))
+ << position;
+ }
+ ASSERT_EQ(WrapCountOf(PeekStateWord(ring.get(), 0)), 2u);
+
+ // The stale writer's one and only claim attempt uses the operand it computed
+ // back then, and it no longer matches.
+ EXPECT_FALSE(ring->TryAcquireChunkForWriting(stale.position,
+ BeingWrittenWord(kWriterA)));
+ EXPECT_EQ(WrapCountOf(PeekStateWord(ring.get(), 0)), 2u);
+
+ // The writer holding position 8 - the one the chunk was actually prepared
+ // for - still gets in.
+ EXPECT_TRUE(ring->TryAcquireChunkForWriting(8, BeingWrittenWord(kWriterB)));
+}
+
+TEST(SharedRingBufferTest, TwoStaleClaimantsCannotBothAdoptTheReturnedWord) {
+ auto ring = SharedRingBuffer::Create(2, kChunkSize);
+ ASSERT_NE(ring, nullptr);
+
+ // Positions 0 and 2 both map to chunk 0 but belong to different traversals.
+ uint32_t observed = ring->LoadChunkStateWord(0);
+ ASSERT_TRUE(ring->TryMoveFreeChunkToNextWrap(0, &observed));
+ ASSERT_EQ(WrapCountOf(PeekStateWord(ring.get(), 0)), 1u);
+
+ // Position 0's writer is out, and so is anyone whose ticket is not exactly
+ // position 2. Only position 2 can claim what position 0 left behind.
+ EXPECT_FALSE(ring->TryAcquireChunkForWriting(0, BeingWrittenWord(kWriterA)));
+ EXPECT_FALSE(ring->TryAcquireChunkForWriting(4, BeingWrittenWord(kWriterA)));
+ EXPECT_TRUE(ring->TryAcquireChunkForWriting(2, BeingWrittenWord(kWriterB)));
+}
+
+// The Free identity is a fixed 16-bit wrap count, so the same physical chunk
+// answers to the same word again after num_chunks * 65536 reservations. That
+// period is an accepted limit of the initial ABI, pinned here so the test
+// fails if the wrap-count width silently changes.
+TEST(SharedRingBufferTest, WrapIdentityRepeatsAfterTheDocumentedPeriod) {
+ auto ring = SharedRingBuffer::Create(2, kChunkSize);
+ ASSERT_NE(ring, nullptr);
+
+ // A fresh chunk 0 is Free(0), tagged for position 0. One full identity
+ // period later, position 2 * 65536 maps to the same chunk and its wrap
+ // count - bit 16 of the shifted position - is truncated back to zero, so
+ // this claim lands even though the ring never ran.
+ EXPECT_TRUE(
+ ring->TryAcquireChunkForWriting(2u * 65536, BeingWrittenWord(kWriterA)));
+}
+
+// The wrap count a seeded ring stamps is the 16-bit truncation of the shifted
+// position, not the shifted position itself.
+TEST(SharedRingBufferTest, SeededPositionsStampSixteenBitWraps) {
+ auto ring = SharedRingBuffer::Create(4, kChunkSize);
+ ASSERT_NE(ring, nullptr);
+ // 4 * 65536 is one whole identity period, so every chunk is stamped with
+ // exactly the word a fresh mapping holds.
+ ring->SetPositionsForTesting(4u * 65536);
+ for (uint32_t i = 0; i < 4; ++i)
+ EXPECT_EQ(PeekStateWord(ring.get(), i), 0u) << i;
+}
+
+TEST(SharedRingBufferTest, PublishReuseAndReclaimAreExactValueTransitions) {
+ auto ring = SharedRingBuffer::Create(4, kChunkSize);
+ ASSERT_NE(ring, nullptr);
+
+ ASSERT_EQ(ring->TryReserveWritePos().position, 0u);
+ ASSERT_TRUE(ring->TryAcquireChunkForWriting(0, BeingWrittenWord(kWriterA)));
+
+ uint32_t observed = BeingWrittenWord(kWriterA);
+ ASSERT_TRUE(
+ ring->TrySetChunkComplete(0, &observed, CompleteWord(kWriterA, 2)));
+ EXPECT_EQ(PeekStateWord(ring.get(), 0), CompleteWord(kWriterA, 2));
+
+ // A publication against a word that is no longer there fails and reports what
+ // is.
+ uint32_t stale = BeingWrittenWord(kWriterA);
+ EXPECT_FALSE(ring->TrySetChunkComplete(0, &stale, CompleteWord(kWriterA, 3)));
+ EXPECT_EQ(stale, CompleteWord(kWriterA, 2));
+
+ // Reuse takes the same chunk back, keeping the published count.
+ ASSERT_TRUE(ring->TryReacquireChunkForWriting(0, CompleteWord(kWriterA, 2)));
+ EXPECT_EQ(ChunkStateOf(PeekStateWord(ring.get(), 0)),
+ ChunkState::kBeingWritten);
+ EXPECT_EQ(NumFragmentsOf(PeekStateWord(ring.get(), 0)), 2u);
+
+ observed =
+ ReplaceChunkState(CompleteWord(kWriterA, 2), ChunkState::kBeingWritten);
+ ASSERT_TRUE(
+ ring->TrySetChunkComplete(0, &observed, CompleteWord(kWriterA, 5)));
+
+ // The reader consumes it and stamps the wrap for the *next* traversal of this
+ // chunk, taken from the position it just resolved.
+ observed = CompleteWord(kWriterA, 5);
+ ASSERT_TRUE(ring->TryReleaseCompleteChunkAsFree(0, &observed));
+ EXPECT_EQ(PeekStateWord(ring.get(), 0), MakeFreeStateWord(1));
+}
+
+TEST(SharedRingBufferTest, MarkForRewritePassesEveryOtherFieldThrough) {
+ auto ring = SharedRingBuffer::Create(4, 512);
+ ASSERT_NE(ring, nullptr);
+
+ const uint32_t being_written = MakeDataStateWord(
+ ChunkState::kBeingWritten, ChunkFormat::kReservedRouting,
+ kFlagContinuesFromPrevChunk | kFlagDataLoss, 0, kWriterA);
+ ASSERT_TRUE(ring->TryAcquireChunkForWriting(0, being_written));
+
+ const uint32_t published = MakeDataStateWord(
+ ChunkState::kBeingWritten, ChunkFormat::kReservedRouting,
+ kFlagContinuesFromPrevChunk | kFlagDataLoss, 3, kWriterA);
+ // Simulate the writer having appended three fragments by reusing the chunk
+ // through the publish/reuse pair.
+ uint32_t observed = being_written;
+ ASSERT_TRUE(ring->TrySetChunkComplete(
+ 0, &observed, ReplaceChunkState(published, ChunkState::kComplete)));
+ ASSERT_TRUE(ring->TryReacquireChunkForWriting(
+ 0, ReplaceChunkState(published, ChunkState::kComplete)));
+
+ observed = published;
+ ASSERT_TRUE(ring->TryRequestRewrite(0, &observed));
+ const uint32_t marked = PeekStateWord(ring.get(), 0);
+ EXPECT_EQ(ChunkStateOf(marked), ChunkState::kRewriteRequested);
+ // The reader can arbitrate a chunk whose format it does not understand
+ // precisely because it changes nothing but the state.
+ EXPECT_EQ(ChunkFormatOf(marked), ChunkFormat::kReservedRouting);
+ EXPECT_EQ(PayloadFlagsOf(marked),
+ kFlagContinuesFromPrevChunk | kFlagDataLoss);
+ EXPECT_EQ(NumFragmentsOf(marked), 3u);
+ EXPECT_EQ(WriterIdOf(marked), kWriterA);
+}
+
+TEST(SharedRingBufferTest, AcknowledgeThenReclaimReturnsTheChunk) {
+ auto ring = SharedRingBuffer::Create(4, kChunkSize);
+ ASSERT_NE(ring, nullptr);
+
+ ASSERT_TRUE(ring->TryAcquireChunkForWriting(0, BeingWrittenWord(kWriterA)));
+ uint32_t observed = BeingWrittenWord(kWriterA);
+ ASSERT_TRUE(ring->TryRequestRewrite(0, &observed));
+ const uint32_t marked = ReplaceChunkState(BeingWrittenWord(kWriterA),
+ ChunkState::kRewriteRequested);
+ EXPECT_EQ(PeekStateWord(ring.get(), 0), marked);
+
+ // Nobody but the owning writer can leave RewriteRequested, and the writer
+ // says nothing about who gets the chunk next.
+ ASSERT_TRUE(ring->TryAcknowledgeRewrite(0, marked));
+ EXPECT_EQ(PeekStateWord(ring.get(), 0), kRewriteAcknowledgedStateWord);
+
+ // Only the reader turns that into a free word, and it stamps the wrap of the
+ // position it is resolving.
+ ASSERT_TRUE(ring->TryReleaseRewriteAcknowledgedChunkAsFree(4, &observed));
+ EXPECT_EQ(PeekStateWord(ring.get(), 0), MakeFreeStateWord(2));
+}
+
+// The reclaim compares against the exact rewrite-acknowledgment word. A failed
+// CAS reports the word that defeated it, rather than doing a second load or
+// returning the caller's stale dispatch word.
+TEST(SharedRingBufferTest,
+ FailedRewriteAcknowledgedReclaimReportsTheDefeatingWord) {
+ auto ring = SharedRingBuffer::Create(4, kChunkSize);
+ ASSERT_NE(ring, nullptr);
+
+ // RewriteAcknowledged state bits over a nonzero payload: the same dispatch
+ // state, but not the one word the transition may leave from.
+ const uint32_t forged = kRewriteAcknowledgedStateWord | kFlagDataLoss;
+ ring->SetStateWordForTesting(0, forged);
+
+ uint32_t observed = 0;
+ EXPECT_FALSE(ring->TryReleaseRewriteAcknowledgedChunkAsFree(0, &observed));
+ EXPECT_EQ(observed, forged);
+ EXPECT_EQ(PeekStateWord(ring.get(), 0), forged);
+
+ // The exact word goes through.
+ ring->SetStateWordForTesting(0, kRewriteAcknowledgedStateWord);
+ EXPECT_TRUE(ring->TryReleaseRewriteAcknowledgedChunkAsFree(0, &observed));
+ EXPECT_EQ(PeekStateWord(ring.get(), 0), MakeFreeStateWord(1));
+}
+
+// Every word that says "this chunk is claimable" comes out of exactly three
+// methods, and all three are the reader's. A writer has no way to produce one:
+// the strongest thing it can say is RewriteAcknowledged.
+TEST(SharedRingBufferTest, NoWriterTransitionProducesAFreeWord) {
+ auto ring = SharedRingBuffer::Create(4, kChunkSize);
+ ASSERT_NE(ring, nullptr);
+
+ ASSERT_TRUE(ring->TryAcquireChunkForWriting(0, BeingWrittenWord(kWriterA)));
+ uint32_t observed = BeingWrittenWord(kWriterA);
+ ASSERT_TRUE(
+ ring->TrySetChunkComplete(0, &observed, CompleteWord(kWriterA, 1)));
+ EXPECT_NE(ChunkStateOf(PeekStateWord(ring.get(), 0)), ChunkState::kFree);
+
+ ASSERT_TRUE(ring->TryReacquireChunkForWriting(0, CompleteWord(kWriterA, 1)));
+ EXPECT_NE(ChunkStateOf(PeekStateWord(ring.get(), 0)), ChunkState::kFree);
+
+ observed =
+ ReplaceChunkState(CompleteWord(kWriterA, 1), ChunkState::kBeingWritten);
+ ASSERT_TRUE(ring->TryRequestRewrite(0, &observed));
+ ASSERT_TRUE(ring->TryAcknowledgeRewrite(
+ 0, ReplaceChunkState(CompleteWord(kWriterA, 1),
+ ChunkState::kRewriteRequested)));
+ EXPECT_NE(ChunkStateOf(PeekStateWord(ring.get(), 0)), ChunkState::kFree);
+ EXPECT_EQ(PeekStateWord(ring.get(), 0), kRewriteAcknowledgedStateWord);
+
+ ASSERT_TRUE(ring->TryReleaseRewriteAcknowledgedChunkAsFree(0, &observed));
+ EXPECT_EQ(ChunkStateOf(PeekStateWord(ring.get(), 0)), ChunkState::kFree);
+}
+
+TEST(SharedRingBufferTest, ReclaimStampsTheNextWrapAcrossPositionRollover) {
+ // At the 32-bit position rollover the shifted position restarts from zero. A
+ // reader that incremented the 16-bit value it found in the chunk would agree
+ // here by coincidence of the uint16_t wrap, so the assertion that matters is
+ // the exact stamped word, derived from the position.
+ auto ring = SharedRingBuffer::Create(16, kChunkSize);
+ ASSERT_NE(ring, nullptr);
+
+ const uint32_t kLastLap = 0xfffffff0u; // chunk 0, the last lap's wrap
+ ASSERT_EQ(ChunkIndexOfPosition(kLastLap, 16), 0u);
+ ASSERT_EQ(WrapCountForPosition(kLastLap, ring->chunk_index_bits()), 0xffffu);
+
+ // Walk chunk 0 into RewriteAcknowledged, the state the reader reclaims.
+ ASSERT_TRUE(ring->TryAcquireChunkForWriting(0, BeingWrittenWord(kWriterA)));
+ uint32_t observed = BeingWrittenWord(kWriterA);
+ ASSERT_TRUE(ring->TryRequestRewrite(0, &observed));
+ ASSERT_TRUE(ring->TryAcknowledgeRewrite(
+ 0, ReplaceChunkState(BeingWrittenWord(kWriterA),
+ ChunkState::kRewriteRequested)));
+
+ ASSERT_TRUE(
+ ring->TryReleaseRewriteAcknowledgedChunkAsFree(kLastLap, &observed));
+ EXPECT_EQ(PeekStateWord(ring.get(), 0), MakeFreeStateWord(0));
+ // The writer holding position 0, the one that follows kLastLap on chunk 0,
+ // is exactly the one that can claim it.
+ EXPECT_TRUE(ring->TryAcquireChunkForWriting(0, BeingWrittenWord(kWriterA)));
+}
+
+TEST(SharedRingBufferTest, ReclaimStampsZeroAtTheWrapIdentityRollover) {
+ // The 16-bit identity rolls over every num_chunks * 65536 positions, long
+ // before the position does. The wrap after 0xffff is zero, and it comes from
+ // the position, not from incrementing the chunk's value.
+ auto ring = SharedRingBuffer::Create(4, kChunkSize);
+ ASSERT_NE(ring, nullptr);
+
+ const uint32_t kLastLap = 0xffffu * 4; // chunk 0, wrap 0xffff
+ ASSERT_EQ(ChunkIndexOfPosition(kLastLap, 4), 0u);
+ ASSERT_EQ(WrapCountForPosition(kLastLap, ring->chunk_index_bits()), 0xffffu);
+
+ ring->SetPositionsForTesting(kLastLap);
+ ASSERT_EQ(PeekStateWord(ring.get(), 0), MakeFreeStateWord(0xffff));
+
+ uint32_t observed = MakeFreeStateWord(0xffff);
+ ASSERT_TRUE(ring->TryMoveFreeChunkToNextWrap(kLastLap, &observed));
+ EXPECT_EQ(PeekStateWord(ring.get(), 0), MakeFreeStateWord(0));
+ // The next traversal of chunk 0 belongs to position kLastLap + 4, whose wrap
+ // is the truncated zero.
+ EXPECT_TRUE(ring->TryAcquireChunkForWriting(kLastLap + 4,
+ BeingWrittenWord(kWriterB)));
+}
+
+// ---------------------------------------------------------------------------
+// Backpressure.
+// ---------------------------------------------------------------------------
+
+#if PERFETTO_BUILDFLAG(PERFETTO_OS_LINUX_BUT_NOT_QNX) || \
+ PERFETTO_BUILDFLAG(PERFETTO_OS_ANDROID)
+
+TEST(SharedRingBufferTest, BlockedWriterWakesWhenTheReaderReleasesCapacity) {
+ auto ring = SharedRingBuffer::Create(2, kChunkSize);
+ ASSERT_NE(ring, nullptr);
+ ASSERT_EQ(ring->TryReserveWritePos().result,
+ SharedRingBuffer::ReserveResult::kReserved);
+ ASSERT_EQ(ring->TryReserveWritePos().result,
+ SharedRingBuffer::ReserveResult::kReserved);
+
+ const auto full = ring->TryReserveWritePos();
+ ASSERT_EQ(full.result, SharedRingBuffer::ReserveResult::kFull);
+
+ base::WaitableEvent about_to_wait;
+ std::atomic<bool> reserved{false};
+ std::thread writer([&] {
+ about_to_wait.Notify();
+ // Bounded only as a deadlock guard: the reader below always frees capacity,
+ // so a correct implementation never comes close to the timeout.
+ const auto outcome =
+ ring->WaitForReadPosChange(full.read_pos_sample, 30000);
+ EXPECT_EQ(outcome, SharedRingBuffer::WriterWaitResult::kRetry);
+ reserved.store(ring->TryReserveWritePos().result ==
+ SharedRingBuffer::ReserveResult::kReserved);
+ });
+
+ about_to_wait.Wait();
+ // Wait until the writer is actually parked, so that this exercises the wake
+ // rather than the early-return path.
+ const bool parked =
+ SpinUntil([&] { return ring->num_writers_waiting_for_testing() != 0; });
+
+ // Published unconditionally: it is both the wake under test and what lets
+ // the writer thread finish - and be joined - if parking was never observed.
+ ring->PublishReadPos(1);
+ writer.join();
+ ASSERT_TRUE(parked);
+ EXPECT_TRUE(reserved.load());
+ EXPECT_EQ(ring->num_writers_waiting_for_testing(), 0u);
+}
+
+// If read_pos moves between the capacity sample and the wait, the wait must
+// come straight back rather than park on a value that will never be stored
+// again. Without that the writer sleeps with space available.
+TEST(SharedRingBufferTest, WaitReturnsImmediatelyIfReadPosAlreadyMoved) {
+ auto ring = SharedRingBuffer::Create(2, kChunkSize);
+ ASSERT_NE(ring, nullptr);
+
+ const uint32_t stale_sample = ring->read_pos_for_testing();
+ ring->PublishReadPos(5);
+
+ const auto outcome = ring->WaitForReadPosChange(stale_sample, 30000);
+ EXPECT_EQ(outcome, SharedRingBuffer::WriterWaitResult::kRetry);
+ EXPECT_EQ(ring->num_writers_waiting_for_testing(), 0u);
+}
+
+// Besides the timeout path, this checks the futex watches the read half of the
+// packed positions: write_pos is nonzero while read_pos still equals the
+// expected value, so a wait that watched the wrong half - or a misplaced
+// address - would find a mismatch and return immediately instead of sleeping.
+TEST(SharedRingBufferTest, WaitTimesOutWatchingTheReadHalf) {
+ auto ring = SharedRingBuffer::Create(2, kChunkSize);
+ ASSERT_NE(ring, nullptr);
+ ASSERT_EQ(ring->TryReserveWritePos().result,
+ SharedRingBuffer::ReserveResult::kReserved);
+ ASSERT_EQ(ring->TryReserveWritePos().result,
+ SharedRingBuffer::ReserveResult::kReserved);
+ ASSERT_EQ(ring->LoadWritePos(), 2u);
+ ASSERT_EQ(ring->read_pos_for_testing(), 0u);
+
+ // 1 ms because this test relies on the timeout firing.
+ EXPECT_EQ(ring->WaitForReadPosChange(0, 1),
+ SharedRingBuffer::WriterWaitResult::kTimedOut);
+ EXPECT_EQ(ring->num_writers_waiting_for_testing(), 0u);
+}
+
+TEST(SharedRingBufferTest, SeveralWaitersAreAllReleased) {
+ constexpr uint32_t kNumWaiters = 8;
+ constexpr uint32_t kWaitTimeoutMs = 1000;
+ auto ring = SharedRingBuffer::Create(2, kChunkSize);
+ ASSERT_NE(ring, nullptr);
+
+ std::atomic<uint32_t> woke{0};
+ std::vector<SharedRingBuffer::WriterWaitResult> outcomes(
+ kNumWaiters, SharedRingBuffer::WriterWaitResult::kTimedOut);
+ std::vector<std::thread> waiters;
+ for (uint32_t i = 0; i < kNumWaiters; ++i) {
+ waiters.emplace_back([&, i] {
+ outcomes[i] = ring->WaitForReadPosChange(0, kWaitTimeoutMs);
+ woke.fetch_add(1);
+ });
+ }
+
+ const bool all_parked = SpinUntil(
+ [&] { return ring->num_writers_waiting_for_testing() == kNumWaiters; });
+
+ // Published - and every thread joined - before any assertion, so a failure
+ // here cannot leave a joinable thread behind.
+ ring->PublishReadPos(1);
+ for (auto& waiter : waiters)
+ waiter.join();
+
+ ASSERT_TRUE(all_parked);
+ EXPECT_EQ(woke.load(), kNumWaiters);
+ for (auto outcome : outcomes)
+ EXPECT_EQ(outcome, SharedRingBuffer::WriterWaitResult::kRetry);
+ EXPECT_EQ(ring->num_writers_waiting_for_testing(), 0u);
+}
+
+// A publication may land anywhere between the waiter registering, the
+// user-space recheck and the kernel's atomic compare: none of those windows
+// may lose it. Each round parks a waiter, publishes as soon as the hint says
+// it is registered - which is before it has necessarily reached the kernel -
+// and requires progress, not a timeout. The generous timeout exists only so a
+// lost wake fails loudly instead of hanging.
+TEST(SharedRingBufferTest, PublicationBetweenRegistrationAndSleepIsNotLost) {
+ constexpr uint32_t kRounds = 100;
+ auto ring = SharedRingBuffer::Create(2, kChunkSize);
+ ASSERT_NE(ring, nullptr);
+
+ for (uint32_t round = 0; round < kRounds; ++round) {
+ const uint32_t sample = ring->read_pos_for_testing();
+ SharedRingBuffer::WriterWaitResult outcome =
+ SharedRingBuffer::WriterWaitResult::kTimedOut;
+ std::thread writer(
+ [&] { outcome = ring->WaitForReadPosChange(sample, 30000); });
+
+ const bool registered =
+ SpinUntil([&] { return ring->num_writers_waiting_for_testing() != 0; });
+ ring->PublishReadPos(sample + 1);
+ writer.join();
+
+ ASSERT_TRUE(registered) << round;
+ ASSERT_EQ(outcome, SharedRingBuffer::WriterWaitResult::kRetry) << round;
+ ASSERT_EQ(ring->num_writers_waiting_for_testing(), 0u) << round;
+ }
+}
+
+// The hint exists only to elide a syscall. A reader that never looks at it is
+// slower, not wrong, so a writer must make progress even then.
+TEST(SharedRingBufferTest, WaiterHintIsNotRequiredForProgress) {
+ auto ring = SharedRingBuffer::Create(2, kChunkSize);
+ ASSERT_NE(ring, nullptr);
+ EXPECT_EQ(ring->num_writers_waiting_for_testing(), 0u);
+
+ // Publishing with no waiters must not wake anything or leave the hint dirty.
+ ring->PublishReadPos(3);
+ EXPECT_EQ(ring->num_writers_waiting_for_testing(), 0u);
+ EXPECT_EQ(ring->read_pos_for_testing(), 3u);
+}
+
+#endif // PERFETTO_OS_LINUX_BUT_NOT_QNX || PERFETTO_OS_ANDROID
+
+// The errno policy of the wait, checked directly rather than by arranging for
+// the kernel to fail. Every branch matters to a stalling writer: two of them
+// send it back to the capacity predicate, one is the slice expiring, and the
+// rest have to stop it waiting at all.
+TEST(SharedRingBufferTest, WaitErrnoPolicy) {
+ using WriterWaitResult = SharedRingBuffer::WriterWaitResult;
+ EXPECT_EQ(SharedRingBuffer::ClassifyWriterWaitErrnoForTesting(ETIMEDOUT),
+ WriterWaitResult::kTimedOut);
+ // read_pos moved between the load and the syscall.
+ EXPECT_EQ(SharedRingBuffer::ClassifyWriterWaitErrnoForTesting(EAGAIN),
+ WriterWaitResult::kRetry);
+ // A signal. Deliberately not retried inside the wrapper.
+ EXPECT_EQ(SharedRingBuffer::ClassifyWriterWaitErrnoForTesting(EINTR),
+ WriterWaitResult::kRetry);
+ // Anything else means this address cannot be waited on, so a stalling writer
+ // must drop rather than come straight back into the same failing syscall.
+ EXPECT_EQ(SharedRingBuffer::ClassifyWriterWaitErrnoForTesting(EINVAL),
+ WriterWaitResult::kUnavailable);
+ EXPECT_EQ(SharedRingBuffer::ClassifyWriterWaitErrnoForTesting(EFAULT),
+ WriterWaitResult::kUnavailable);
+ EXPECT_EQ(SharedRingBuffer::ClassifyWriterWaitErrnoForTesting(ENOSYS),
+ WriterWaitResult::kUnavailable);
+}
+
+} // namespace
+} // namespace perfetto::tracing_v2
diff --git a/src/tracing/v2/shared_ring_buffer_writer.cc b/src/tracing/v2/shared_ring_buffer_writer.cc
new file mode 100644
index 0000000..7c29d45
--- /dev/null
+++ b/src/tracing/v2/shared_ring_buffer_writer.cc
@@ -0,0 +1,382 @@
+/*
+ * Copyright (C) 2026 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/v2/shared_ring_buffer_writer.h"
+
+#include <stdint.h>
+#include <string.h>
+
+#include <optional>
+
+#include "perfetto/base/logging.h"
+#include "perfetto/base/time.h"
+#include "perfetto/ext/tracing/core/basic_types.h"
+#include "perfetto/tracing/buffer_exhausted_policy.h"
+#include "src/tracing/v2/shared_ring_buffer.h"
+#include "src/tracing/v2/tracing_v2_abi.h"
+
+namespace perfetto::tracing_v2 {
+
+SharedRingBufferWriter::Delegate::~Delegate() = default;
+
+SharedRingBufferWriter::SharedRingBufferWriter(
+ SharedRingBuffer* ring,
+ WriterID writer_id,
+ BufferID target_buffer,
+ BufferExhaustedPolicy buffer_exhausted_policy,
+ Delegate* delegate)
+ : ring_(ring),
+ writer_id_(writer_id),
+ target_buffer_(target_buffer),
+ buffer_exhausted_policy_(buffer_exhausted_policy),
+ delegate_(delegate),
+ chunk_size_(ring->chunk_size()) {
+ PERFETTO_CHECK(delegate_);
+ relocation_payload_.reserve(MaxFragmentSizeForEmptyChunk(chunk_size_));
+ relocation_fragment_sizes_.reserve(kMaxFragmentsPerChunk);
+}
+
+SharedRingBufferWriter::~SharedRingBufferWriter() {
+ FinishCurrentChunk();
+}
+
+SharedRingBufferWriter::FragmentSpan SharedRingBufferWriter::OpenFragment(
+ uint32_t min_size,
+ bool continues_from_prev) {
+ PERFETTO_DCHECK(!has_open_fragment());
+
+ const uint32_t largest_possible_fragment =
+ MaxFragmentSizeForEmptyChunk(chunk_size_);
+ if (min_size > largest_possible_fragment)
+ return FragmentSpan{Outcome::kTooLarge, nullptr, nullptr};
+
+ if (current_chunk_) {
+ PERFETTO_DCHECK(ChunkStateOf(expected_state_word_) ==
+ ChunkState::kComplete);
+ if (MaxFragmentSizeInCurrentChunk() >= min_size &&
+ ring_->TryReacquireChunkForWriting(current_chunk_index_,
+ expected_state_word_)) {
+ expected_state_word_ =
+ ReplaceChunkState(expected_state_word_, ChunkState::kBeingWritten);
+ return OpenFragmentInCurrentChunk();
+ }
+ // The chunk is full or the reader reclaimed it.
+ ResetCurrentChunk();
+ }
+
+ const uint32_t carried_flags =
+ continues_from_prev ? uint32_t{kFlagContinuesFromPrevChunk} : 0u;
+ const Outcome outcome = AcquireNewChunk(carried_flags);
+ if (outcome != Outcome::kOk)
+ return FragmentSpan{outcome, nullptr, nullptr};
+ return OpenFragmentInCurrentChunk();
+}
+
+SharedRingBufferWriter::Outcome SharedRingBufferWriter::CloseFragment(
+ uint32_t size,
+ bool continues_on_next) {
+ PERFETTO_DCHECK(has_open_fragment());
+ PERFETTO_DCHECK(current_chunk_);
+ PERFETTO_DCHECK(ChunkStateOf(expected_state_word_) ==
+ ChunkState::kBeingWritten);
+ PERFETTO_DCHECK(size <= open_fragment_end_ - open_fragment_begin_);
+
+ // Nothing becomes visible until CompleteCurrentChunk().
+ const uint32_t varint_bytes = FragmentSizeVarIntBytes(size);
+ PERFETTO_DCHECK(open_fragment_begin_ + size <= fragment_sizes_begin_);
+ PERFETTO_DCHECK(varint_bytes <=
+ fragment_sizes_begin_ - (open_fragment_begin_ + size));
+ fragment_sizes_begin_ = static_cast<uint32_t>(
+ WriteFragmentSize(current_chunk_ + fragment_sizes_begin_, size) -
+ current_chunk_);
+ payload_end_ = open_fragment_begin_ + size;
+ ++num_fragments_;
+ open_fragment_begin_ = kNoOpenFragment;
+
+ return CompleteCurrentChunk(continues_on_next);
+}
+
+SharedRingBufferWriter::Outcome SharedRingBufferWriter::FinishCurrentChunk() {
+ // An open fragment is not counted and can be abandoned.
+ open_fragment_begin_ = kNoOpenFragment;
+
+ if (current_chunk_ &&
+ ChunkStateOf(expected_state_word_) == ChunkState::kBeingWritten)
+ return CompleteCurrentChunk(/*continues_on_next=*/false);
+
+ ResetCurrentChunk();
+ return Outcome::kOk;
+}
+
+uint32_t SharedRingBufferWriter::MaxFragmentSizeInCurrentChunk() const {
+ if (!current_chunk_)
+ return 0;
+ if (num_fragments_ >= kMaxFragmentsPerChunk)
+ return 0;
+ if (fragment_sizes_begin_ <= payload_end_)
+ return 0;
+ const uint32_t available_bytes = fragment_sizes_begin_ - payload_end_;
+ return MaxFragmentSizeForAvailableBytes(available_bytes);
+}
+
+SharedRingBufferWriter::Outcome SharedRingBufferWriter::AcquireNewChunk(
+ uint32_t carried_flags) {
+ PERFETTO_DCHECK(!current_chunk_);
+
+ BufferExhaustedPolicy policy = buffer_exhausted_policy_;
+ // Once data has been dropped, keep trying with kDrop until a chunk is
+ // acquired. That chunk reports the gap and lets the next full-ring event
+ // stall again.
+ if (policy == BufferExhaustedPolicy::kStallThenDrop && data_loss_pending_)
+ policy = BufferExhaustedPolicy::kDrop;
+
+ // BufferExhaustedPolicy gives a stalling writer a few seconds before kStall
+ // aborts or kStallThenDrop starts dropping. Use the same 30 second ceiling as
+ // the v1 arbiter. A deadline, rather than a retry count, also bounds spurious
+ // futex wakes.
+ constexpr uint32_t kStallTimeoutMs = 30000;
+ std::optional<base::TimeMillis> stall_deadline;
+
+ uint32_t flags = carried_flags;
+ if (data_loss_pending_)
+ flags |= kFlagDataLoss;
+
+ uint32_t failed_claims = 0;
+ bool saw_unclaimable_chunk = false;
+ for (;;) {
+ const auto reservation = ring_->TryReserveWritePos();
+
+ if (reservation.result == SharedRingBuffer::ReserveResult::kReserved) {
+ const uint32_t being_written_word =
+ MakeDataStateWord(ChunkState::kBeingWritten,
+ ChunkFormat::kTargetBuffer, flags, 0, writer_id_);
+ if (ring_->TryAcquireChunkForWriting(reservation.position,
+ being_written_word)) {
+ current_chunk_index_ =
+ ChunkIndexOfPosition(reservation.position, ring_->num_chunks());
+ current_chunk_ = ring_->chunk_at(current_chunk_index_);
+ expected_state_word_ = being_written_word;
+ payload_end_ = kTargetBufferPayloadOffset;
+ fragment_sizes_begin_ = chunk_size_;
+ num_fragments_ = 0;
+ data_loss_pending_ = false;
+ StoreTargetBufferId(current_chunk_, target_buffer_);
+ return Outcome::kOk;
+ }
+
+ // This reservation is now a hole. Never retry it against a different
+ // Free word; reserve a later position instead.
+ ++num_failed_claims_;
+ saw_unclaimable_chunk = true;
+ if (++failed_claims < ring_->num_chunks())
+ continue;
+ }
+
+ // A stalling writer cannot make progress until the reader runs. A dropping
+ // writer only needs to notify it after creating holes.
+ if (failed_claims != 0 || policy != BufferExhaustedPolicy::kDrop) {
+ delegate_->NotifyReader();
+ failed_claims = 0;
+ }
+
+ const Outcome exhausted_outcome =
+ saw_unclaimable_chunk ? Outcome::kNoChunkAvailable : Outcome::kFull;
+ if (policy == BufferExhaustedPolicy::kDrop)
+ return exhausted_outcome;
+
+ const base::TimeMillis now = base::GetWallTimeMs();
+ if (!stall_deadline)
+ stall_deadline = now + base::TimeMillis(kStallTimeoutMs);
+ if (now >= *stall_deadline) {
+ if (policy == BufferExhaustedPolicy::kStall) {
+ PERFETTO_FATAL(
+ "tracing v2: writer %u could not acquire a chunk for %u ms; "
+ "possible deadlock",
+ writer_id_, kStallTimeoutMs);
+ }
+ return exhausted_outcome;
+ }
+
+ const uint32_t timeout_ms =
+ static_cast<uint32_t>((*stall_deadline - now).count());
+ const SharedRingBuffer::WriterWaitResult wait =
+ ring_->WaitForReadPosChange(reservation.read_pos_sample, timeout_ms);
+ if (wait == SharedRingBuffer::WriterWaitResult::kUnavailable) {
+ if (policy == BufferExhaustedPolicy::kStall) {
+ PERFETTO_FATAL(
+ "tracing v2: writer %u cannot stall because waiting on read_pos "
+ "is unavailable",
+ writer_id_);
+ }
+ return exhausted_outcome;
+ }
+ }
+}
+
+SharedRingBufferWriter::Outcome SharedRingBufferWriter::CompleteCurrentChunk(
+ bool continues_on_next) {
+ PERFETTO_DCHECK(current_chunk_);
+ PERFETTO_DCHECK(ChunkStateOf(expected_state_word_) ==
+ ChunkState::kBeingWritten);
+ PERFETTO_DCHECK(!has_open_fragment());
+
+ // expected_state_word_ contains the number of fragments visible before the
+ // writer took the chunk. num_fragments_ also includes what it appended
+ // afterwards. Usually the writer changes BeingWritten directly to Complete.
+ // If the reader gets there first, it changes BeingWritten(N) to
+ // RewriteRequested(N) and takes those N fragments. The writer then moves only
+ // fragments N..M:
+ //
+ // BeingWritten(N) -- writer --> Complete(M)
+ // |
+ // +--------- reader --> RewriteRequested(N)
+ // |
+ // writer copies N..M and acknowledges
+ //
+ // A replacement can lose the same race, hence the loop.
+ for (;;) {
+ uint32_t flags = PayloadFlagsOf(expected_state_word_);
+ if (continues_on_next)
+ flags |= kFlagContinuesOnNextChunk;
+ const uint32_t complete_word =
+ MakeDataStateWord(ChunkState::kComplete, ChunkFormat::kTargetBuffer,
+ flags, num_fragments_, writer_id_);
+
+ uint32_t observed = expected_state_word_;
+ if (ring_->TrySetChunkComplete(current_chunk_index_, &observed,
+ complete_word)) {
+ expected_state_word_ = complete_word;
+ // A Complete chunk carrying "continues on next chunk" is never reused.
+ // Together with the rule that a BeingWritten word never carries that
+ // flag, this guarantees that every non-empty prefix the reader can
+ // scrape ends on a packet boundary.
+ if (continues_on_next || MaxFragmentSizeInCurrentChunk() == 0)
+ ResetCurrentChunk();
+ return Outcome::kOk;
+ }
+
+ // Only the reader's rewrite request may beat this publication.
+ if (ChunkStateOf(observed) != ChunkState::kRewriteRequested ||
+ WriterIdOf(observed) != writer_id_) {
+ PERFETTO_FATAL(
+ "tracing v2: publication of chunk %u by writer %u lost to state word "
+ "0x%08x, which is not a rewrite request for this writer",
+ current_chunk_index_, writer_id_, observed);
+ }
+
+ // The reader took this prefix. The remaining fragments must move.
+ const uint32_t taken = NumFragmentsOf(observed);
+ PERFETTO_DCHECK(taken == NumFragmentsOf(expected_state_word_));
+ PERFETTO_DCHECK(taken <= num_fragments_);
+ const uint32_t suffix_fragments = num_fragments_ - taken;
+
+ // Copy the suffix out before acknowledging the rewrite. After that, the
+ // reader may reclaim the chunk.
+ uint32_t suffix_begin = kTargetBufferPayloadOffset;
+ relocation_fragment_sizes_.clear();
+ const uint8_t* size_cursor = current_chunk_ + chunk_size_;
+ for (uint32_t i = 0; i < num_fragments_; ++i) {
+ uint32_t fragment_size = 0;
+ if (!ReadFragmentSize(current_chunk_ + fragment_sizes_begin_,
+ &size_cursor, &fragment_size)) {
+ PERFETTO_FATAL("tracing v2: writer %u corrupted its size directory",
+ writer_id_);
+ }
+ if (i < taken) {
+ suffix_begin += fragment_size;
+ } else {
+ relocation_fragment_sizes_.push_back(fragment_size);
+ }
+ }
+ PERFETTO_DCHECK(size_cursor == current_chunk_ + fragment_sizes_begin_);
+ relocation_payload_.assign(current_chunk_ + suffix_begin,
+ current_chunk_ + payload_end_);
+
+ // Acknowledge before looking for replacement capacity. The other order
+ // leaves the old chunk occupied whenever the ring is full, so every later
+ // traversal of it burns a position. Acknowledge even if the suffix ends up
+ // dropped: the acknowledgement is about the chunk, not about whether the
+ // data survived.
+ if (!ring_->TryAcknowledgeRewrite(current_chunk_index_, observed)) {
+ PERFETTO_FATAL(
+ "tracing v2: writer %u could not acknowledge chunk %u; only its "
+ "owner may leave RewriteRequested",
+ writer_id_, current_chunk_index_);
+ }
+ ++num_relocations_;
+
+ // Prefix flags move only when the reader took no fragments.
+ const uint32_t relocated_flags =
+ taken == 0 ? PayloadFlagsOf(observed) &
+ (kFlagContinuesFromPrevChunk | kFlagDataLoss)
+ : 0;
+
+ ResetCurrentChunk();
+
+ if (suffix_fragments == 0) {
+ // Nothing remains to relocate.
+ PERFETTO_DCHECK(!continues_on_next);
+ return Outcome::kOk;
+ }
+
+ if (AcquireNewChunk(relocated_flags) != Outcome::kOk) {
+ num_fragments_dropped_ += suffix_fragments;
+ data_loss_pending_ = true;
+ return Outcome::kRelocationDropped;
+ }
+
+ // Rebuild the suffix in the replacement. A suffix always fits: it came out
+ // of a chunk of the same size.
+ memcpy(current_chunk_ + kTargetBufferPayloadOffset,
+ relocation_payload_.data(), relocation_payload_.size());
+ payload_end_ = kTargetBufferPayloadOffset +
+ static_cast<uint32_t>(relocation_payload_.size());
+ for (uint32_t fragment_size : relocation_fragment_sizes_) {
+ fragment_sizes_begin_ = static_cast<uint32_t>(
+ WriteFragmentSize(current_chunk_ + fragment_sizes_begin_,
+ fragment_size) -
+ current_chunk_);
+ }
+ num_fragments_ = static_cast<uint32_t>(relocation_fragment_sizes_.size());
+ // Round again to publish the replacement, which the reader may also scrape.
+ }
+}
+
+void SharedRingBufferWriter::ResetCurrentChunk() {
+ current_chunk_ = nullptr;
+ current_chunk_index_ = 0;
+ expected_state_word_ = 0;
+ payload_end_ = 0;
+ fragment_sizes_begin_ = 0;
+ num_fragments_ = 0;
+ open_fragment_begin_ = kNoOpenFragment;
+ open_fragment_end_ = 0;
+}
+
+SharedRingBufferWriter::FragmentSpan
+SharedRingBufferWriter::OpenFragmentInCurrentChunk() {
+ PERFETTO_DCHECK(current_chunk_);
+ PERFETTO_DCHECK(ChunkStateOf(expected_state_word_) ==
+ ChunkState::kBeingWritten);
+ const uint32_t available = MaxFragmentSizeInCurrentChunk();
+ PERFETTO_DCHECK(available > 0);
+ open_fragment_begin_ = payload_end_;
+ open_fragment_end_ = payload_end_ + available;
+ return FragmentSpan{Outcome::kOk, current_chunk_ + open_fragment_begin_,
+ current_chunk_ + open_fragment_end_};
+}
+
+} // namespace perfetto::tracing_v2
diff --git a/src/tracing/v2/shared_ring_buffer_writer.h b/src/tracing/v2/shared_ring_buffer_writer.h
new file mode 100644
index 0000000..6fe4742
--- /dev/null
+++ b/src/tracing/v2/shared_ring_buffer_writer.h
@@ -0,0 +1,191 @@
+/*
+ * Copyright (C) 2026 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_V2_SHARED_RING_BUFFER_WRITER_H_
+#define SRC_TRACING_V2_SHARED_RING_BUFFER_WRITER_H_
+
+#include <stdint.h>
+
+#include <vector>
+
+#include "perfetto/ext/tracing/core/basic_types.h"
+#include "perfetto/tracing/buffer_exhausted_policy.h"
+#include "src/tracing/v2/shared_ring_buffer.h"
+#include "src/tracing/v2/tracing_v2_abi.h"
+
+namespace perfetto::tracing_v2 {
+
+// Writes the packet fragments produced by one TraceWriter into a
+// SharedRingBuffer.
+//
+// A fragment is one packet, or one part of a packet that crosses a chunk
+// boundary. A chunk can hold several fragments. SharedRingBufferWriter deals
+// only with their byte ranges and lengths; it does not interpret their bytes.
+//
+// Each instance is used by one thread at a time. Several instances can write
+// to the same ring concurrently.
+//
+// Reserving a write position and claiming its physical chunk are separate
+// operations. If the claim fails, the position is still part of the stream and
+// only the reader can resolve it. The writer notifies its delegate before it
+// waits for the reader to make space.
+class SharedRingBufferWriter {
+ public:
+ // Result of an operation that may need a new chunk.
+ enum class Outcome {
+ kOk,
+ // The ring is structurally full: num_chunks positions are outstanding and
+ // the reader is behind. A stalling policy has already waited by the time
+ // this is returned.
+ kFull,
+ // Positions were reserved but their chunks could not be claimed. Chunks
+ // pinned by a stalled writer produce this without the ring being full.
+ //
+ // The reader has been notified before this is returned.
+ kNoChunkAvailable,
+ // The request is larger than a freshly claimed chunk could hold. This is a
+ // caller bug, not backpressure.
+ kTooLarge,
+ // The reader scraped the chunk and there was no replacement capacity, so
+ // the unpublished suffix was dropped. The writer is consistent and its next
+ // publication will carry the data-loss flag.
+ kRelocationDropped,
+ };
+
+ // A contiguous range for the caller to fill. Valid until the next call on
+ // this SharedRingBufferWriter.
+ struct FragmentSpan {
+ Outcome outcome = Outcome::kFull;
+ uint8_t* begin = nullptr;
+ uint8_t* end = nullptr;
+ };
+
+ class Delegate {
+ public:
+ virtual ~Delegate();
+
+ // Makes the reader aware of newly reserved positions. Called before a
+ // writer applies a buffer-exhaustion policy that may wait for read_pos to
+ // advance. Several writers may call this concurrently.
+ virtual void NotifyReader() = 0;
+ };
+
+ SharedRingBufferWriter(SharedRingBuffer* ring,
+ WriterID writer_id,
+ BufferID target_buffer,
+ BufferExhaustedPolicy buffer_exhausted_policy,
+ Delegate* delegate);
+ ~SharedRingBufferWriter();
+
+ SharedRingBufferWriter(const SharedRingBufferWriter&) = delete;
+ SharedRingBufferWriter& operator=(const SharedRingBufferWriter&) = delete;
+ SharedRingBufferWriter(SharedRingBufferWriter&&) = delete;
+ SharedRingBufferWriter& operator=(SharedRingBufferWriter&&) = delete;
+
+ // Returns space for one fragment of at least |min_size| bytes. This reuses
+ // the current chunk when possible and acquires a new chunk otherwise. At
+ // most one fragment can be open at a time.
+ //
+ // |continues_from_prev| says that the first bytes written here are the tail
+ // of a packet that began in this writer's previous chunk. It is recorded only
+ // when this call moves to a new chunk, which is the only case where the flag
+ // means anything.
+ FragmentSpan OpenFragment(uint32_t min_size, bool continues_from_prev);
+
+ // Closes the open fragment at |size| bytes - which must not exceed the span
+ // OpenFragment() handed out - writes its size varint, and publishes.
+ //
+ // |continues_on_next| says that this fragment is the head of a packet that
+ // continues in this writer's next chunk. A chunk published with that flag is
+ // never reused, which is what makes every non-empty published prefix end on a
+ // packet boundary.
+ Outcome CloseFragment(uint32_t size, bool continues_on_next);
+
+ // Publishes whatever is held and lets go of the chunk. Any open fragment is
+ // abandoned: its bytes were never counted, so nothing is published for it.
+ // Safe to call with nothing held; the destructor calls it.
+ Outcome FinishCurrentChunk();
+
+ // Tells the writer that the caller dropped data, so that the next chunk this
+ // writer publishes reports the gap. The writer sets this itself when it has
+ // to drop a relocated suffix; callers that drop a packet of their own accord
+ // have to say so here.
+ void RecordDataLoss() { data_loss_pending_ = true; }
+
+ // Payload bytes the current chunk could give to one more fragment.
+ // Zero if nothing is held, if the chunk has reached 255 fragments, or if the
+ // payload bytes and fragment-size varints have met.
+ uint32_t MaxFragmentSizeInCurrentChunk() const;
+
+ bool has_open_fragment() const {
+ return open_fragment_begin_ != kNoOpenFragment;
+ }
+ WriterID writer_id() const { return writer_id_; }
+ BufferID target_buffer() const { return target_buffer_; }
+
+ // Diagnostics only.
+ uint64_t num_failed_claims() const { return num_failed_claims_; }
+ uint64_t num_fragments_dropped() const { return num_fragments_dropped_; }
+ uint64_t num_relocations() const { return num_relocations_; }
+
+ private:
+ static constexpr uint32_t kNoOpenFragment = UINT32_MAX;
+
+ Outcome AcquireNewChunk(uint32_t carried_flags);
+ Outcome CompleteCurrentChunk(bool continues_on_next);
+ // Clears only this writer's cached chunk state. It does not modify the ring.
+ void ResetCurrentChunk();
+ FragmentSpan OpenFragmentInCurrentChunk();
+
+ SharedRingBuffer* const ring_;
+ const WriterID writer_id_;
+ const BufferID target_buffer_;
+ const BufferExhaustedPolicy buffer_exhausted_policy_;
+ Delegate* const delegate_;
+ const uint32_t chunk_size_;
+
+ // State cached for the chunk this writer currently owns.
+ uint8_t* current_chunk_ = nullptr;
+ uint32_t current_chunk_index_ = 0;
+ // Exact word expected by this writer's next state transition.
+ uint32_t expected_state_word_ = 0;
+ // Offset of the end of finalized payload.
+ uint32_t payload_end_ = 0;
+ // Offset of the first fragment-size varint. Sizes are prepended from the end
+ // of the chunk towards the payload.
+ uint32_t fragment_sizes_begin_ = 0;
+ // Finalized fragments, published or not. expected_state_word_ contains the
+ // count already visible to the reader.
+ uint32_t num_fragments_ = 0;
+ uint32_t open_fragment_begin_ = kNoOpenFragment;
+ uint32_t open_fragment_end_ = 0;
+
+ // Set when this writer has lost data that the next chunk it publishes must
+ // report.
+ bool data_loss_pending_ = false;
+
+ // Preallocated so relocation does not allocate.
+ std::vector<uint8_t> relocation_payload_;
+ std::vector<uint32_t> relocation_fragment_sizes_;
+
+ uint64_t num_failed_claims_ = 0;
+ uint64_t num_fragments_dropped_ = 0;
+ uint64_t num_relocations_ = 0;
+};
+
+} // namespace perfetto::tracing_v2
+
+#endif // SRC_TRACING_V2_SHARED_RING_BUFFER_WRITER_H_
diff --git a/src/tracing/v2/shared_ring_buffer_writer_unittest.cc b/src/tracing/v2/shared_ring_buffer_writer_unittest.cc
new file mode 100644
index 0000000..9032ce2
--- /dev/null
+++ b/src/tracing/v2/shared_ring_buffer_writer_unittest.cc
@@ -0,0 +1,701 @@
+/*
+ * Copyright (C) 2026 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/v2/shared_ring_buffer_writer.h"
+
+#include <stdint.h>
+#include <string.h>
+
+#include <memory>
+#include <string>
+#include <vector>
+
+#include "perfetto/base/build_config.h"
+#include "perfetto/ext/base/no_destructor.h"
+#include "perfetto/tracing/buffer_exhausted_policy.h"
+#include "src/tracing/v2/shared_ring_buffer.h"
+#include "src/tracing/v2/tracing_v2_abi.h"
+#include "test/gtest_and_gmock.h"
+
+namespace perfetto::tracing_v2 {
+namespace {
+
+constexpr WriterID kWriterA = 7;
+constexpr WriterID kWriterB = 8;
+constexpr BufferID kBuffer = 0x1234;
+
+class CountingSharedRingBufferWriterDelegate
+ : public SharedRingBufferWriter::Delegate {
+ public:
+ void NotifyReader() override { ++num_notifications; }
+
+ uint32_t num_notifications = 0;
+};
+
+class ReleasingSharedRingBufferWriterDelegate
+ : public SharedRingBufferWriter::Delegate {
+ public:
+ explicit ReleasingSharedRingBufferWriterDelegate(SharedRingBuffer* ring)
+ : ring_(ring) {}
+
+ void NotifyReader() override {
+ ++num_notifications;
+ const uint32_t read_pos = ring_->read_pos_for_testing();
+ const uint32_t chunk_index =
+ ChunkIndexOfPosition(read_pos, ring_->num_chunks());
+ uint32_t observed = ring_->LoadChunkStateWord(chunk_index);
+ ASSERT_EQ(ChunkStateOf(observed), ChunkState::kComplete);
+ ASSERT_TRUE(ring_->TryReleaseCompleteChunkAsFree(read_pos, &observed));
+ ring_->PublishReadPos(read_pos + 1);
+ }
+
+ uint32_t num_notifications = 0;
+
+ private:
+ SharedRingBuffer* const ring_;
+};
+
+class NoopSharedRingBufferWriterDelegate
+ : public SharedRingBufferWriter::Delegate {
+ public:
+ void NotifyReader() override {}
+};
+
+SharedRingBufferWriter::Delegate* GetNoopSharedRingBufferWriterDelegate() {
+ static base::NoDestructor<NoopSharedRingBufferWriterDelegate> delegate;
+ return &delegate.ref();
+}
+
+// A minimal, independent decoder for what a chunk holds. It deliberately does
+// not go through SharedRingBufferReader, so that a writer test cannot pass
+// because the reader shares the same misunderstanding.
+struct DecodedChunk {
+ ChunkState state = ChunkState::kFree;
+ WriterID writer_id = 0;
+ BufferID target_buffer = 0;
+ uint32_t payload_flags = 0;
+ std::vector<std::string> fragments;
+};
+
+DecodedChunk Decode(SharedRingBuffer* ring, uint32_t chunk_index) {
+ const uint8_t* chunk = ring->chunk_at(chunk_index);
+ const uint32_t word = ring->LoadChunkStateWord(chunk_index);
+ DecodedChunk decoded;
+ decoded.state = ChunkStateOf(word);
+ if (!HasDataFields(decoded.state))
+ return decoded;
+
+ decoded.writer_id = WriterIdOf(word);
+ decoded.target_buffer = LoadTargetBufferId(chunk);
+ decoded.payload_flags = PayloadFlagsOf(word);
+
+ const uint8_t* directory_cursor = chunk + ring->chunk_size();
+ uint32_t offset = kTargetBufferPayloadOffset;
+ for (uint32_t i = 0; i < NumFragmentsOf(word); ++i) {
+ uint32_t size = 0;
+ const bool valid = ReadFragmentSize(chunk + kTargetBufferPayloadOffset,
+ &directory_cursor, &size);
+ EXPECT_TRUE(valid);
+ if (!valid)
+ return decoded;
+ decoded.fragments.emplace_back(
+ reinterpret_cast<const char*>(chunk + offset), size);
+ offset += size;
+ }
+ return decoded;
+}
+
+// Writes one fragment holding |bytes| and publishes it.
+SharedRingBufferWriter::Outcome WriteFragment(SharedRingBufferWriter* writer,
+ const std::string& bytes,
+ bool continues_from_prev = false,
+ bool continues_on_next = false) {
+ const SharedRingBufferWriter::FragmentSpan span = writer->OpenFragment(
+ static_cast<uint32_t>(bytes.size()), continues_from_prev);
+ if (span.outcome != SharedRingBufferWriter::Outcome::kOk)
+ return span.outcome;
+ memcpy(span.begin, bytes.data(), bytes.size());
+ return writer->CloseFragment(static_cast<uint32_t>(bytes.size()),
+ continues_on_next);
+}
+
+// Plays the reader's part of the scrape: marks whatever the writer currently
+// holds as rewrite-requested and reports what the marked prefix was.
+uint32_t MarkForRewrite(SharedRingBuffer* ring, uint32_t chunk_index) {
+ uint32_t observed = ring->LoadChunkStateWord(chunk_index);
+ EXPECT_EQ(ChunkStateOf(observed), ChunkState::kBeingWritten);
+ const uint32_t taken = NumFragmentsOf(observed);
+ EXPECT_TRUE(ring->TryRequestRewrite(chunk_index, &observed));
+ return taken;
+}
+
+// ---------------------------------------------------------------------------
+// Fragments and the directory.
+// ---------------------------------------------------------------------------
+
+TEST(SharedRingBufferWriterTest, WritesPayloadUpAndSizesDown) {
+ auto ring = SharedRingBuffer::Create(4, 256);
+ ASSERT_NE(ring, nullptr);
+ SharedRingBufferWriter writer(ring.get(), kWriterA, kBuffer,
+ BufferExhaustedPolicy::kDrop,
+ GetNoopSharedRingBufferWriterDelegate());
+
+ ASSERT_EQ(WriteFragment(&writer, std::string(5, 'a')),
+ SharedRingBufferWriter::Outcome::kOk);
+ ASSERT_EQ(WriteFragment(&writer, std::string(200, 'b')),
+ SharedRingBufferWriter::Outcome::kOk);
+ ASSERT_EQ(WriteFragment(&writer, std::string(3, 'c')),
+ SharedRingBufferWriter::Outcome::kOk);
+
+ // A 256-byte target-buffer chunk, byte for byte.
+ const uint8_t* chunk = ring->chunk_at(0);
+ EXPECT_EQ(chunk[255], 5u);
+ EXPECT_EQ(chunk[254], 0xc8u);
+ EXPECT_EQ(chunk[253], 1u);
+ EXPECT_EQ(chunk[252], 3u);
+
+ const DecodedChunk decoded = Decode(ring.get(), 0);
+ EXPECT_EQ(decoded.state, ChunkState::kComplete);
+ EXPECT_EQ(decoded.writer_id, kWriterA);
+ EXPECT_EQ(decoded.target_buffer, kBuffer);
+ ASSERT_EQ(decoded.fragments.size(), 3u);
+ EXPECT_EQ(decoded.fragments[0], std::string(5, 'a'));
+ EXPECT_EQ(decoded.fragments[1], std::string(200, 'b'));
+ EXPECT_EQ(decoded.fragments[2], std::string(3, 'c'));
+}
+
+TEST(SharedRingBufferWriterTest, FragmentSizesAtTheInterestingBoundaries) {
+ // These sizes straddle the one- and two-byte varint boundary.
+ auto ring = SharedRingBuffer::Create(8, 512);
+ ASSERT_NE(ring, nullptr);
+ SharedRingBufferWriter writer(ring.get(), kWriterA, kBuffer,
+ BufferExhaustedPolicy::kDrop,
+ GetNoopSharedRingBufferWriterDelegate());
+
+ const uint32_t kSizes[] = {0, 1, 127, 128, 255, 256};
+ std::vector<std::string> expected;
+ for (uint32_t size : kSizes) {
+ const std::string bytes(size, static_cast<char>('A' + (size % 26)));
+ ASSERT_EQ(WriteFragment(&writer, bytes),
+ SharedRingBufferWriter::Outcome::kOk)
+ << size;
+ expected.push_back(bytes);
+ }
+
+ // 0 + 1 + 127 + 128 + 255 + 256 = 767 bytes, which does not fit in one
+ // 512-byte chunk, so the writer moved on part way through. Walk every chunk
+ // it used and check the fragments come out in order.
+ std::vector<std::string> seen;
+ for (uint32_t i = 0; i < ring->num_chunks(); ++i) {
+ for (const std::string& fragment : Decode(ring.get(), i).fragments)
+ seen.push_back(fragment);
+ }
+ EXPECT_EQ(seen, expected);
+}
+
+TEST(SharedRingBufferWriterTest,
+ LargestFragmentFitsExactlyAndOneMoreByteDoesNot) {
+ auto ring = SharedRingBuffer::Create(4, 256);
+ ASSERT_NE(ring, nullptr);
+ // 256 - 6 header bytes - varint(248), which takes two bytes.
+ constexpr uint32_t kLargest = 248;
+
+ SharedRingBufferWriter writer(ring.get(), kWriterA, kBuffer,
+ BufferExhaustedPolicy::kDrop,
+ GetNoopSharedRingBufferWriterDelegate());
+ const SharedRingBufferWriter::FragmentSpan span =
+ writer.OpenFragment(kLargest, false);
+ ASSERT_EQ(span.outcome, SharedRingBufferWriter::Outcome::kOk);
+ EXPECT_EQ(static_cast<uint32_t>(span.end - span.begin), kLargest);
+ memset(span.begin, 'z', kLargest);
+ ASSERT_EQ(writer.CloseFragment(kLargest, false),
+ SharedRingBufferWriter::Outcome::kOk);
+
+ // The payload and the directory have met: the chunk cannot take another
+ // fragment, not even an empty one.
+ EXPECT_EQ(writer.MaxFragmentSizeInCurrentChunk(), 0u);
+ ASSERT_EQ(Decode(ring.get(), 0).fragments.size(), 1u);
+ EXPECT_EQ(Decode(ring.get(), 0).fragments[0].size(), kLargest);
+
+ // One byte more than any chunk of this size could ever hold is a caller
+ // bug, not backpressure, and says so.
+ SharedRingBufferWriter other(ring.get(), kWriterB, kBuffer,
+ BufferExhaustedPolicy::kDrop,
+ GetNoopSharedRingBufferWriterDelegate());
+ EXPECT_EQ(other.OpenFragment(kLargest + 1, false).outcome,
+ SharedRingBufferWriter::Outcome::kTooLarge);
+}
+
+TEST(SharedRingBufferWriterTest, ChunkClosesAt255FragmentsEvenWithSpaceLeft) {
+ auto ring = SharedRingBuffer::Create(4, 32768);
+ ASSERT_NE(ring, nullptr);
+ SharedRingBufferWriter writer(ring.get(), kWriterA, kBuffer,
+ BufferExhaustedPolicy::kDrop,
+ GetNoopSharedRingBufferWriterDelegate());
+
+ for (uint32_t i = 0; i < kMaxFragmentsPerChunk; ++i)
+ ASSERT_EQ(WriteFragment(&writer, ""), SharedRingBufferWriter::Outcome::kOk)
+ << i;
+
+ const DecodedChunk first = Decode(ring.get(), 0);
+ EXPECT_EQ(first.fragments.size(), kMaxFragmentsPerChunk);
+ // Thousands of payload bytes are still free; the eight-bit count is what
+ // ended the chunk.
+ EXPECT_EQ(writer.MaxFragmentSizeInCurrentChunk(), 0u);
+
+ ASSERT_EQ(WriteFragment(&writer, "x"), SharedRingBufferWriter::Outcome::kOk);
+ const DecodedChunk second = Decode(ring.get(), 1);
+ ASSERT_EQ(second.fragments.size(), 1u);
+ EXPECT_EQ(second.fragments[0], "x");
+}
+
+TEST(SharedRingBufferWriterTest, LargeChunkSupportsAFragmentLargerThanUint16) {
+ constexpr uint32_t kBigChunk = 128 * 1024;
+ constexpr uint32_t kLargest = kBigChunk - 9;
+ auto ring = SharedRingBuffer::Create(2, kBigChunk);
+ ASSERT_NE(ring, nullptr);
+ SharedRingBufferWriter writer(ring.get(), kWriterA, kBuffer,
+ BufferExhaustedPolicy::kDrop,
+ GetNoopSharedRingBufferWriterDelegate());
+
+ EXPECT_EQ(writer.OpenFragment(kLargest + 1, false).outcome,
+ SharedRingBufferWriter::Outcome::kTooLarge);
+
+ const SharedRingBufferWriter::FragmentSpan span =
+ writer.OpenFragment(1, false);
+ ASSERT_EQ(span.outcome, SharedRingBufferWriter::Outcome::kOk);
+ EXPECT_EQ(static_cast<uint32_t>(span.end - span.begin), kLargest);
+ memset(span.begin, 'a', kLargest);
+ ASSERT_EQ(writer.CloseFragment(kLargest, false),
+ SharedRingBufferWriter::Outcome::kOk);
+
+ const DecodedChunk decoded = Decode(ring.get(), 0);
+ ASSERT_EQ(decoded.fragments.size(), 1u);
+ EXPECT_EQ(decoded.fragments[0], std::string(kLargest, 'a'));
+}
+
+// An aligned, non-power-of-two chunk size, end to end through the writer.
+TEST(SharedRingBufferWriterTest, WritesAndDecodesInANonPowerOfTwoChunk) {
+ auto ring = SharedRingBuffer::Create(4, 260);
+ ASSERT_NE(ring, nullptr);
+ SharedRingBufferWriter writer(ring.get(), kWriterA, kBuffer,
+ BufferExhaustedPolicy::kDrop,
+ GetNoopSharedRingBufferWriterDelegate());
+
+ // 260 - 6 header bytes - the two-byte varint for 252.
+ constexpr uint32_t kLargest = 252;
+ EXPECT_EQ(writer.OpenFragment(kLargest + 1, false).outcome,
+ SharedRingBufferWriter::Outcome::kTooLarge);
+ ASSERT_EQ(WriteFragment(&writer, std::string(kLargest, 'z')),
+ SharedRingBufferWriter::Outcome::kOk);
+ EXPECT_EQ(writer.MaxFragmentSizeInCurrentChunk(), 0u);
+
+ ASSERT_EQ(WriteFragment(&writer, "next"),
+ SharedRingBufferWriter::Outcome::kOk);
+ ASSERT_EQ(Decode(ring.get(), 0).fragments.size(), 1u);
+ EXPECT_EQ(Decode(ring.get(), 0).fragments[0].size(), kLargest);
+ const DecodedChunk second = Decode(ring.get(), 1);
+ ASSERT_EQ(second.fragments.size(), 1u);
+ EXPECT_EQ(second.fragments[0], "next");
+}
+
+// ---------------------------------------------------------------------------
+// Chunk reuse and the payload flags.
+// ---------------------------------------------------------------------------
+
+TEST(SharedRingBufferWriterTest, ReusesItsOwnCompleteChunkWhileItCanTakeMore) {
+ auto ring = SharedRingBuffer::Create(4, 512);
+ ASSERT_NE(ring, nullptr);
+ SharedRingBufferWriter writer(ring.get(), kWriterA, kBuffer,
+ BufferExhaustedPolicy::kDrop,
+ GetNoopSharedRingBufferWriterDelegate());
+
+ ASSERT_EQ(WriteFragment(&writer, "one"),
+ SharedRingBufferWriter::Outcome::kOk);
+ ASSERT_EQ(WriteFragment(&writer, "two"),
+ SharedRingBufferWriter::Outcome::kOk);
+
+ // Both fragments are in the same physical chunk, and only one position was
+ // consumed.
+ EXPECT_EQ(ring->LoadWritePos(), 1u);
+ const DecodedChunk decoded = Decode(ring.get(), 0);
+ ASSERT_EQ(decoded.fragments.size(), 2u);
+ EXPECT_EQ(decoded.fragments[0], "one");
+ EXPECT_EQ(decoded.fragments[1], "two");
+ EXPECT_EQ(ChunkStateOf(ring->LoadChunkStateWord(1)), ChunkState::kFree);
+}
+
+TEST(SharedRingBufferWriterTest, ReuseLosesToTheReaderReclaimingTheChunk) {
+ auto ring = SharedRingBuffer::Create(4, 512);
+ ASSERT_NE(ring, nullptr);
+ SharedRingBufferWriter writer(ring.get(), kWriterA, kBuffer,
+ BufferExhaustedPolicy::kDrop,
+ GetNoopSharedRingBufferWriterDelegate());
+
+ ASSERT_EQ(WriteFragment(&writer, "one"),
+ SharedRingBufferWriter::Outcome::kOk);
+
+ // The reader consumes the Complete chunk before the writer takes it back.
+ uint32_t observed = ring->LoadChunkStateWord(0);
+ ASSERT_EQ(ChunkStateOf(observed), ChunkState::kComplete);
+ ASSERT_TRUE(ring->TryReleaseCompleteChunkAsFree(0, &observed));
+ ring->PublishReadPos(1);
+
+ // The writer's reuse fails; it drops its handle and goes for a fresh chunk
+ // rather than writing behind the reader.
+ ASSERT_EQ(WriteFragment(&writer, "two"),
+ SharedRingBufferWriter::Outcome::kOk);
+ EXPECT_EQ(ChunkStateOf(ring->LoadChunkStateWord(0)), ChunkState::kFree);
+ const DecodedChunk decoded = Decode(ring.get(), 1);
+ ASSERT_EQ(decoded.fragments.size(), 1u);
+ EXPECT_EQ(decoded.fragments[0], "two");
+}
+
+// A chunk published with "continues on next chunk" is never reused. Without
+// that rule a later scrape could take a prefix ending in the middle of a
+// packet.
+TEST(SharedRingBufferWriterTest, ChunkCarryingContinuesOnNextIsNeverReused) {
+ auto ring = SharedRingBuffer::Create(4, 512);
+ ASSERT_NE(ring, nullptr);
+ SharedRingBufferWriter writer(ring.get(), kWriterA, kBuffer,
+ BufferExhaustedPolicy::kDrop,
+ GetNoopSharedRingBufferWriterDelegate());
+
+ ASSERT_EQ(WriteFragment(&writer, "head", /*continues_from_prev=*/false,
+ /*continues_on_next=*/true),
+ SharedRingBufferWriter::Outcome::kOk);
+ ASSERT_EQ(WriteFragment(&writer, "tail", /*continues_from_prev=*/true,
+ /*continues_on_next=*/false),
+ SharedRingBufferWriter::Outcome::kOk);
+
+ const DecodedChunk first = Decode(ring.get(), 0);
+ EXPECT_EQ(first.payload_flags, kFlagContinuesOnNextChunk);
+ ASSERT_EQ(first.fragments.size(), 1u);
+ EXPECT_EQ(first.fragments[0], "head");
+
+ const DecodedChunk second = Decode(ring.get(), 1);
+ EXPECT_EQ(second.payload_flags, kFlagContinuesFromPrevChunk);
+ ASSERT_EQ(second.fragments.size(), 1u);
+ EXPECT_EQ(second.fragments[0], "tail");
+}
+
+TEST(SharedRingBufferWriterTest, DataLossIsReportedOnTheNextChunkAndOnlyOnce) {
+ auto ring = SharedRingBuffer::Create(4, 512);
+ ASSERT_NE(ring, nullptr);
+ SharedRingBufferWriter writer(ring.get(), kWriterA, kBuffer,
+ BufferExhaustedPolicy::kDrop,
+ GetNoopSharedRingBufferWriterDelegate());
+
+ writer.RecordDataLoss();
+ ASSERT_EQ(WriteFragment(&writer, "after the gap"),
+ SharedRingBufferWriter::Outcome::kOk);
+ EXPECT_EQ(Decode(ring.get(), 0).payload_flags, kFlagDataLoss);
+
+ // The chunk after it describes no gap of its own.
+ ASSERT_EQ(WriteFragment(&writer, std::string(500, 'x')),
+ SharedRingBufferWriter::Outcome::kOk);
+ EXPECT_EQ(Decode(ring.get(), 1).payload_flags, 0u);
+}
+
+// ---------------------------------------------------------------------------
+// Backpressure outcomes.
+// ---------------------------------------------------------------------------
+
+TEST(SharedRingBufferWriterTest, DropPolicyReportsFullWithoutBlocking) {
+ auto ring = SharedRingBuffer::Create(2, 256);
+ ASSERT_NE(ring, nullptr);
+ SharedRingBufferWriter a(ring.get(), kWriterA, kBuffer,
+ BufferExhaustedPolicy::kDrop,
+ GetNoopSharedRingBufferWriterDelegate());
+ SharedRingBufferWriter b(ring.get(), kWriterB, kBuffer,
+ BufferExhaustedPolicy::kDrop,
+ GetNoopSharedRingBufferWriterDelegate());
+
+ // Two writers hold both chunks, so the ring is structurally full.
+ ASSERT_EQ(a.OpenFragment(1, false).outcome,
+ SharedRingBufferWriter::Outcome::kOk);
+ ASSERT_EQ(b.OpenFragment(1, false).outcome,
+ SharedRingBufferWriter::Outcome::kOk);
+
+ SharedRingBufferWriter c(ring.get(), 11, kBuffer,
+ BufferExhaustedPolicy::kDrop,
+ GetNoopSharedRingBufferWriterDelegate());
+ EXPECT_EQ(c.OpenFragment(1, false).outcome,
+ SharedRingBufferWriter::Outcome::kFull);
+ // Nothing was reserved, so a full ring costs no position.
+ EXPECT_EQ(ring->LoadWritePos(), 2u);
+ EXPECT_EQ(c.num_failed_claims(), 0u);
+}
+
+#if PERFETTO_BUILDFLAG(PERFETTO_OS_LINUX_BUT_NOT_QNX) || \
+ PERFETTO_BUILDFLAG(PERFETTO_OS_ANDROID)
+
+TEST(SharedRingBufferWriterTest, FullRingNotifiesReaderBeforeWaiting) {
+ auto ring = SharedRingBuffer::Create(1, 256);
+ ASSERT_NE(ring, nullptr);
+
+ SharedRingBufferWriter first(ring.get(), kWriterA, kBuffer,
+ BufferExhaustedPolicy::kDrop,
+ GetNoopSharedRingBufferWriterDelegate());
+ ASSERT_EQ(WriteFragment(&first, "first"),
+ SharedRingBufferWriter::Outcome::kOk);
+ ASSERT_EQ(first.FinishCurrentChunk(), SharedRingBufferWriter::Outcome::kOk);
+
+ ReleasingSharedRingBufferWriterDelegate delegate(ring.get());
+ SharedRingBufferWriter second(ring.get(), kWriterB, kBuffer,
+ BufferExhaustedPolicy::kStall, &delegate);
+ EXPECT_EQ(second.OpenFragment(1, false).outcome,
+ SharedRingBufferWriter::Outcome::kOk);
+ EXPECT_EQ(delegate.num_notifications, 1u);
+}
+
+TEST(SharedRingBufferWriterTest,
+ StallThenDropDoesNotStallAgainBeforeAChunkIsAcquired) {
+ auto ring = SharedRingBuffer::Create(1, 256);
+ ASSERT_NE(ring, nullptr);
+
+ SharedRingBufferWriter first(ring.get(), kWriterA, kBuffer,
+ BufferExhaustedPolicy::kDrop,
+ GetNoopSharedRingBufferWriterDelegate());
+ ASSERT_EQ(WriteFragment(&first, "first"),
+ SharedRingBufferWriter::Outcome::kOk);
+ ASSERT_EQ(first.FinishCurrentChunk(), SharedRingBufferWriter::Outcome::kOk);
+
+ ReleasingSharedRingBufferWriterDelegate delegate(ring.get());
+ SharedRingBufferWriter second(ring.get(), kWriterB, kBuffer,
+ BufferExhaustedPolicy::kStallThenDrop,
+ &delegate);
+
+ // RecordDataLoss() marks an active drop episode. Retrying while that loss is
+ // still waiting to be reported must use kDrop rather than stall for another
+ // 30 seconds.
+ second.RecordDataLoss();
+ EXPECT_EQ(second.OpenFragment(1, false).outcome,
+ SharedRingBufferWriter::Outcome::kFull);
+ EXPECT_EQ(delegate.num_notifications, 0u);
+
+ uint32_t observed = ring->LoadChunkStateWord(0);
+ ASSERT_TRUE(ring->TryReleaseCompleteChunkAsFree(0, &observed));
+ ring->PublishReadPos(1);
+
+ ASSERT_EQ(WriteFragment(&second, "after loss"),
+ SharedRingBufferWriter::Outcome::kOk);
+ EXPECT_EQ(Decode(ring.get(), 0).payload_flags, kFlagDataLoss);
+ ASSERT_EQ(second.FinishCurrentChunk(), SharedRingBufferWriter::Outcome::kOk);
+
+ // Publishing the loss ends the drop episode. The next exhausted acquisition
+ // stalls again, which gives the delegate a chance to release the chunk.
+ EXPECT_EQ(second.OpenFragment(1, false).outcome,
+ SharedRingBufferWriter::Outcome::kOk);
+ EXPECT_EQ(delegate.num_notifications, 1u);
+}
+
+#endif // PERFETTO_OS_LINUX_BUT_NOT_QNX || PERFETTO_OS_ANDROID
+
+// A chunk pinned by a writer that stopped mid-rewrite is not the same thing as
+// a full ring: positions are available, but their chunks cannot be acquired.
+TEST(SharedRingBufferWriterTest,
+ PinnedChunksNotifyReaderAndReportNoChunkAvailable) {
+ // The writer tries each physical chunk once. Those failed claims fill the
+ // reservation window even though no writer acquired a chunk.
+ auto ring = SharedRingBuffer::Create(8, 256);
+ ASSERT_NE(ring, nullptr);
+
+ // Pin every chunk in RewriteRequested, which only its owner may leave.
+ // Claiming them directly leaves write_pos at zero, so there is capacity for
+ // every reservation the writer below makes.
+ for (uint32_t i = 0; i < ring->num_chunks(); ++i) {
+ const uint32_t being_written = MakeDataStateWord(
+ ChunkState::kBeingWritten, ChunkFormat::kTargetBuffer, 0, 0, kWriterB);
+ ASSERT_TRUE(ring->TryAcquireChunkForWriting(i, being_written));
+ uint32_t observed = being_written;
+ ASSERT_TRUE(ring->TryRequestRewrite(i, &observed));
+ }
+ ASSERT_EQ(ring->LoadWritePos(), 0u);
+
+ CountingSharedRingBufferWriterDelegate delegate;
+ SharedRingBufferWriter writer(ring.get(), kWriterA, kBuffer,
+ BufferExhaustedPolicy::kDrop, &delegate);
+ EXPECT_EQ(writer.OpenFragment(1, false).outcome,
+ SharedRingBufferWriter::Outcome::kNoChunkAvailable);
+ EXPECT_EQ(writer.num_failed_claims(), ring->num_chunks());
+ EXPECT_EQ(ring->LoadWritePos(), ring->num_chunks());
+ EXPECT_EQ(delegate.num_notifications, 1u);
+}
+
+// ---------------------------------------------------------------------------
+// Relocation after a scrape.
+// ---------------------------------------------------------------------------
+
+TEST(SharedRingBufferWriterTest, ScrapeMovesOnlyTheUnpublishedSuffix) {
+ auto ring = SharedRingBuffer::Create(4, 512);
+ ASSERT_NE(ring, nullptr);
+ SharedRingBufferWriter writer(ring.get(), kWriterA, kBuffer,
+ BufferExhaustedPolicy::kDrop,
+ GetNoopSharedRingBufferWriterDelegate());
+
+ ASSERT_EQ(WriteFragment(&writer, "published-one"),
+ SharedRingBufferWriter::Outcome::kOk);
+ ASSERT_EQ(WriteFragment(&writer, "published-two"),
+ SharedRingBufferWriter::Outcome::kOk);
+
+ // The writer takes the chunk back and fills a third fragment. The reader
+ // arrives while it is inside and takes the two published ones.
+ const SharedRingBufferWriter::FragmentSpan span =
+ writer.OpenFragment(6, false);
+ ASSERT_EQ(span.outcome, SharedRingBufferWriter::Outcome::kOk);
+ memcpy(span.begin, "suffix", 6);
+ EXPECT_EQ(MarkForRewrite(ring.get(), 0), 2u);
+
+ ASSERT_EQ(writer.CloseFragment(6, false),
+ SharedRingBufferWriter::Outcome::kOk);
+ EXPECT_EQ(writer.num_relocations(), 1u);
+ EXPECT_EQ(writer.num_fragments_dropped(), 0u);
+
+ // The old chunk is acknowledged - the writer says nothing about who gets it
+ // next - and only the suffix moved.
+ EXPECT_EQ(ChunkStateOf(ring->LoadChunkStateWord(0)),
+ ChunkState::kRewriteAcknowledged);
+ const DecodedChunk replacement = Decode(ring.get(), 1);
+ EXPECT_EQ(replacement.state, ChunkState::kComplete);
+ EXPECT_EQ(replacement.writer_id, kWriterA);
+ EXPECT_EQ(replacement.target_buffer, kBuffer);
+ ASSERT_EQ(replacement.fragments.size(), 1u);
+ EXPECT_EQ(replacement.fragments[0], "suffix");
+ // A non-empty prefix went out with the chunk's beginning, so the suffix does
+ // not repeat the flags describing it.
+ EXPECT_EQ(replacement.payload_flags, 0u);
+}
+
+TEST(SharedRingBufferWriterTest,
+ ScrapeWithNoPublishedPrefixCarriesTheFlagsAlong) {
+ auto ring = SharedRingBuffer::Create(4, 512);
+ ASSERT_NE(ring, nullptr);
+ SharedRingBufferWriter writer(ring.get(), kWriterA, kBuffer,
+ BufferExhaustedPolicy::kDrop,
+ GetNoopSharedRingBufferWriterDelegate());
+ writer.RecordDataLoss();
+
+ const SharedRingBufferWriter::FragmentSpan span =
+ writer.OpenFragment(4, /*continues_from_prev=*/true);
+ ASSERT_EQ(span.outcome, SharedRingBufferWriter::Outcome::kOk);
+ memcpy(span.begin, "tail", 4);
+ // The reader takes nothing: the writer has published no fragment yet.
+ EXPECT_EQ(MarkForRewrite(ring.get(), 0), 0u);
+
+ ASSERT_EQ(writer.CloseFragment(4, false),
+ SharedRingBufferWriter::Outcome::kOk);
+
+ const DecodedChunk replacement = Decode(ring.get(), 1);
+ ASSERT_EQ(replacement.fragments.size(), 1u);
+ EXPECT_EQ(replacement.fragments[0], "tail");
+ // The reader took no beginning, so both flags describing it travel with the
+ // relocated suffix.
+ EXPECT_EQ(replacement.payload_flags,
+ kFlagContinuesFromPrevChunk | kFlagDataLoss);
+}
+
+TEST(SharedRingBufferWriterTest, ScrapeWithNothingUnpublishedJustAcknowledges) {
+ auto ring = SharedRingBuffer::Create(4, 512);
+ ASSERT_NE(ring, nullptr);
+ SharedRingBufferWriter writer(ring.get(), kWriterA, kBuffer,
+ BufferExhaustedPolicy::kDrop,
+ GetNoopSharedRingBufferWriterDelegate());
+
+ ASSERT_EQ(WriteFragment(&writer, "only"),
+ SharedRingBufferWriter::Outcome::kOk);
+ // Take the chunk back but add nothing, then let the reader scrape it.
+ ASSERT_EQ(writer.OpenFragment(1, false).outcome,
+ SharedRingBufferWriter::Outcome::kOk);
+ EXPECT_EQ(MarkForRewrite(ring.get(), 0), 1u);
+
+ // Releasing abandons the open fragment; there is nothing left to move.
+ EXPECT_EQ(writer.FinishCurrentChunk(), SharedRingBufferWriter::Outcome::kOk);
+ EXPECT_EQ(writer.num_fragments_dropped(), 0u);
+ EXPECT_EQ(ChunkStateOf(ring->LoadChunkStateWord(0)),
+ ChunkState::kRewriteAcknowledged);
+ EXPECT_EQ(ChunkStateOf(ring->LoadChunkStateWord(1)), ChunkState::kFree);
+}
+
+// Acknowledging happens before the writer looks for replacement capacity. The
+// other order would leave the old chunk occupied exactly when the ring is full,
+// so every later traversal of it would burn a position.
+TEST(SharedRingBufferWriterTest,
+ RelocationWithNoCapacityDropsButFreesTheOldChunk) {
+ auto ring = SharedRingBuffer::Create(2, 512);
+ ASSERT_NE(ring, nullptr);
+ SharedRingBufferWriter writer(ring.get(), kWriterA, kBuffer,
+ BufferExhaustedPolicy::kDrop,
+ GetNoopSharedRingBufferWriterDelegate());
+
+ ASSERT_EQ(WriteFragment(&writer, "published"),
+ SharedRingBufferWriter::Outcome::kOk);
+
+ // Occupy the ring's only other chunk so no replacement can be had.
+ SharedRingBufferWriter blocker(ring.get(), kWriterB, kBuffer,
+ BufferExhaustedPolicy::kDrop,
+ GetNoopSharedRingBufferWriterDelegate());
+ ASSERT_EQ(blocker.OpenFragment(1, false).outcome,
+ SharedRingBufferWriter::Outcome::kOk);
+
+ const SharedRingBufferWriter::FragmentSpan span =
+ writer.OpenFragment(4, false);
+ ASSERT_EQ(span.outcome, SharedRingBufferWriter::Outcome::kOk);
+ memcpy(span.begin, "lost", 4);
+ EXPECT_EQ(MarkForRewrite(ring.get(), 0), 1u);
+
+ EXPECT_EQ(writer.CloseFragment(4, false),
+ SharedRingBufferWriter::Outcome::kRelocationDropped);
+ EXPECT_EQ(writer.num_fragments_dropped(), 1u);
+ // The old chunk is acknowledged and therefore reclaimable by the reader,
+ // even though the data did not survive.
+ EXPECT_EQ(ring->LoadChunkStateWord(0), kRewriteAcknowledgedStateWord);
+
+ // The gap is reported on the next chunk this writer manages to publish.
+ uint32_t observed = 0;
+ ASSERT_TRUE(ring->TryReleaseRewriteAcknowledgedChunkAsFree(0, &observed));
+ ring->PublishReadPos(1);
+ ASSERT_EQ(WriteFragment(&writer, "next"),
+ SharedRingBufferWriter::Outcome::kOk);
+ EXPECT_EQ(Decode(ring.get(), 0).payload_flags, kFlagDataLoss);
+}
+
+TEST(SharedRingBufferWriterTest, DestructorPublishesWhateverIsHeld) {
+ auto ring = SharedRingBuffer::Create(4, 512);
+ ASSERT_NE(ring, nullptr);
+ {
+ SharedRingBufferWriter writer(ring.get(), kWriterA, kBuffer,
+ BufferExhaustedPolicy::kDrop,
+ GetNoopSharedRingBufferWriterDelegate());
+ const SharedRingBufferWriter::FragmentSpan span =
+ writer.OpenFragment(4, false);
+ ASSERT_EQ(span.outcome, SharedRingBufferWriter::Outcome::kOk);
+ memcpy(span.begin, "kept", 4);
+ ASSERT_EQ(writer.CloseFragment(4, false),
+ SharedRingBufferWriter::Outcome::kOk);
+
+ // A second fragment is opened and never closed: it is abandoned.
+ ASSERT_EQ(writer.OpenFragment(4, false).outcome,
+ SharedRingBufferWriter::Outcome::kOk);
+ }
+ const DecodedChunk decoded = Decode(ring.get(), 0);
+ EXPECT_EQ(decoded.state, ChunkState::kComplete);
+ ASSERT_EQ(decoded.fragments.size(), 1u);
+ EXPECT_EQ(decoded.fragments[0], "kept");
+}
+
+} // namespace
+} // namespace perfetto::tracing_v2
diff --git a/src/tracing/v2/tracing_v2_abi.h b/src/tracing/v2/tracing_v2_abi.h
new file mode 100644
index 0000000..48a6a5b
--- /dev/null
+++ b/src/tracing/v2/tracing_v2_abi.h
@@ -0,0 +1,571 @@
+/*
+ * Copyright (C) 2026 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_V2_TRACING_V2_ABI_H_
+#define SRC_TRACING_V2_TRACING_V2_ABI_H_
+
+#include <stddef.h>
+#include <stdint.h>
+
+#include <atomic>
+
+#include "perfetto/base/logging.h"
+#include "perfetto/ext/tracing/core/basic_types.h"
+#include "perfetto/public/pb_utils.h"
+
+namespace perfetto::tracing_v2 {
+
+// Shared-memory ABI for a tracing-v2 producer ring. Several trace writers can
+// write to the ring at once. One reader drains their data in the order in which
+// the writers reserved space. The reader may run in the producer process or in
+// the tracing service.
+//
+// The ring has one header followed by fixed-size chunks. The header holds the
+// read and write positions. Every chunk starts with one atomic state word. It
+// records who may access the chunk and, while a writer owns it, how many
+// complete fragments have been published.
+//
+// The ABI assumes little-endian producer and service processes. Anything
+// written by the producer is untrusted when the reader runs in the service, so
+// lengths and offsets must be checked before they are used.
+//
+// Shared-memory layout:
+//
+// +------------------------+---------+---------+---------+-----+
+// | RingBufferHeader, 64 B | chunk 0 | chunk 1 | chunk 2 | ... |
+// +------------------------+---------+---------+---------+-----+
+// 0 64
+
+// Ring header
+// -----------
+//
+// Writers decide whether the ring has room by loading rw_positions once. The
+// high half is write_pos and the low half is read_pos. Keeping them in one
+// atomic prevents a capacity check from combining counters read at different
+// times. On little-endian systems, read_pos is also the first four bytes of
+// rw_positions, which is the address used by the futex wait.
+//
+// byte offset
+// 0 4 8 12 64
+// +--------------+--------------+--------------+----------------+
+// | read_pos | write_pos | num_writers_ | reserved |
+// | | | waiting | |
+// +--------------+--------------+--------------+----------------+
+// \_________ rw_positions _________/ \__ atomic32 __/
+// atomic<uint64_t>
+//
+// num_writers_waiting lets the reader skip a futex wake when nobody is waiting
+// for space. It is only an optimization; it never decides whether the ring is
+// full or who owns a chunk.
+//
+// Bytes 12..63 pad the header to one cache line. Chunk 0 starts on the next
+// line, so updating the positions does not invalidate a cache line holding
+// chunk data.
+//
+static_assert(sizeof(std::atomic<uint32_t>) == 4 &&
+ alignof(std::atomic<uint32_t>) <= 4,
+ "Chunk state must fit in a 4-byte-aligned ABI word");
+static_assert(sizeof(std::atomic<uint64_t>) == 8,
+ "The packed positions must occupy 8 bytes");
+static_assert(std::atomic<uint32_t>::is_always_lock_free &&
+ std::atomic<uint64_t>::is_always_lock_free,
+ "Shared-memory atomics must be lock-free");
+
+struct alignas(64) RingBufferHeader {
+ std::atomic<uint64_t> rw_positions;
+ std::atomic<uint32_t> num_writers_waiting;
+ uint8_t reserved[52];
+};
+
+static_assert(offsetof(RingBufferHeader, num_writers_waiting) == 8 &&
+ offsetof(RingBufferHeader, reserved) == 12 &&
+ sizeof(RingBufferHeader) == 64,
+ "RingBufferHeader does not match the shared-memory ABI");
+
+constexpr uint64_t PackRwPositions(uint32_t write_pos, uint32_t read_pos) {
+ return (uint64_t{write_pos} << 32) | uint64_t{read_pos};
+}
+
+constexpr uint32_t WritePosOf(uint64_t rw_positions) {
+ return static_cast<uint32_t>(rw_positions >> 32);
+}
+
+constexpr uint32_t ReadPosOf(uint64_t rw_positions) {
+ return static_cast<uint32_t>(rw_positions);
+}
+
+// Replace one half of a previously loaded rw_positions value, leaving the
+// other half unchanged.
+constexpr uint64_t ReplaceWritePos(uint64_t rw_positions, uint32_t write_pos) {
+ return PackRwPositions(write_pos, ReadPosOf(rw_positions));
+}
+
+constexpr uint64_t ReplaceReadPos(uint64_t rw_positions, uint32_t read_pos) {
+ return PackRwPositions(WritePosOf(rw_positions), read_pos);
+}
+
+// Logical positions and chunk indexing
+// ------------------------------------
+//
+// write_pos is the next position a writer can reserve. read_pos is the next
+// position the reader must resolve. Both counters are uint32_t and are allowed
+// to wrap.
+//
+// The number of reserved positions not yet handled by the reader is:
+//
+// outstanding = uint32_t(write_pos - read_pos)
+// outstanding <= num_chunks < 2^31
+//
+// Unsigned subtraction also works when write_pos wraps back to zero. For
+// example:
+//
+// read_pos = UINT32_MAX - 3
+// write_pos = 2
+// uint32_t(write_pos - read_pos) = 6
+//
+// A legal result is at most num_chunks. A larger result means that the two
+// positions do not describe a valid ring state.
+//
+// num_chunks is a power of two, so no division is needed to map a position:
+//
+// chunk_index = position & (num_chunks - 1)
+//
+// For num_chunks = 2^k, this uses the low k bits:
+//
+// bits 31..k bits k-1..0
+// +----------------------------+----------------------------+
+// | traversal number | physical chunk index |
+// +----------------------------+----------------------------+
+// 32 - k bits k bits
+//
+// For example, an eight-chunk ring has three chunk-index bits. The low three
+// bits select chunks 0 through 7. The remaining 29 bits count completed
+// traversals.
+
+// Positions can wrap from UINT32_MAX to zero. Unsigned subtraction still gives
+// the number of outstanding positions as long as fewer than 2^31 are in use.
+// Since num_chunks is a power of two, 2^30 is the largest legal chunk count.
+constexpr uint32_t kMaxChunksPerRing = 1u << 30;
+
+constexpr uint32_t NumOutstandingPositions(uint32_t write_pos,
+ uint32_t read_pos) {
+ return write_pos - read_pos;
+}
+
+// Returns k for num_chunks = 2^k. The low k bits of a position select the
+// physical chunk. num_chunks must be a non-zero power of two.
+constexpr uint32_t GetChunkIndexBits(uint32_t num_chunks) {
+ uint32_t chunk_index_bits = 0;
+ while (chunk_index_bits < 31 && (1u << chunk_index_bits) < num_chunks) {
+ ++chunk_index_bits;
+ }
+ return chunk_index_bits;
+}
+
+constexpr uint32_t ChunkIndexOfPosition(uint32_t position,
+ uint32_t num_chunks) {
+ return position & (num_chunks - 1);
+}
+
+// The initial ABI keeps the 256-byte minimum chosen for the producer-local
+// ring. Smaller chunks are not supported.
+constexpr uint32_t kMinChunkSize = 256;
+
+// Every chunk starts with a 32-bit atomic state word. Chunks are contiguous, so
+// the chunk size must be a multiple of four bytes to keep every state word
+// aligned.
+constexpr uint32_t kChunkAlignmentBytes = 4;
+
+// Chunk ownership
+// ---------------
+//
+// The state word is also the arbitration point between the reader and a
+// writer. For example, when a writer finishes while the reader is scraping the
+// same chunk, they race to change the exact same BeingWritten word:
+//
+// Writer: BeingWritten -> Complete
+// Reader: BeingWritten -> RewriteRequested
+//
+// If the writer wins, the reader sees Complete and consumes the finished
+// chunk. If the reader wins, the writer sees RewriteRequested and moves only
+// the data it appended after the published prefix. No second atomic is needed
+// to decide which event happened first.
+//
+// Claiming Free is different. A writer's reservation authorizes one exact
+// Free(wrap) word. If that compare-and-swap fails, the reader has already
+// resolved the position or another traversal owns the chunk. The writer drops
+// that reservation and must not retry against the new word.
+//
+// A chunk can move only along these paths:
+//
+// Free(wrap) --Writer: claim---------------------> BeingWritten
+// Free(wrap) --Reader: resolve unclaimed---------> Free(next wrap)
+// BeingWritten --Writer: publish-----------------> Complete
+// Complete --Writer: reuse----------------------> BeingWritten
+// Complete --Reader: consume--------------------> Free(next wrap)
+// BeingWritten --Reader: resolve committed prefix-> RewriteRequested
+// RewriteRequested --Writer: release old chunk---> RewriteAcknowledged
+// RewriteAcknowledged --Reader: reclaim----------> Free(next wrap)
+//
+// Only the reader writes Free.
+//
+// A Free word contains:
+//
+// 31 16 15 8 7 0
+// +----------------------------------+-----------+------------------+
+// | wrap_count | zero | control = Free |
+// +----------------------------------+-----------+------------------+
+// 16 bits 8 bits 8 bits
+//
+// BeingWritten, Complete and RewriteRequested contain:
+//
+// 31 16 15 8 7 0
+// +----------------------------------+-----------+------------------+
+// | WriterID | num | control byte |
+// | | fragments | |
+// +----------------------------------+-----------+------------------+
+// 16 bits 8 bits 8 bits
+//
+// The control byte is:
+//
+// +---------+---------+---------+---------+---------+
+// | bit 7 | bit 6 | bit 5 |bits 4-3 |bits 2-0 |
+// +---------+---------+---------+---------+---------+
+// |continues|continues| data | format | state |
+// |from prev| on next | loss | | |
+// +---------+---------+---------+---------+---------+
+//
+// RewriteAcknowledged carries no other fields; every other bit is zero.
+
+enum class ChunkState : uint32_t {
+ // The chunk may be claimed by the reservation with this wrap count.
+ kFree = 0,
+
+ // One writer owns the chunk. num_frags is the prefix it has already
+ // published. The writer may be appending another fragment after that prefix.
+ kBeingWritten = 1,
+
+ // The writer has published num_frags fragments and is no longer touching the
+ // chunk. It may take the chunk back before the reader reclaims it.
+ kComplete = 2,
+
+ // The reader took the published prefix while the writer still owned the
+ // chunk. The writer must move anything it appended afterwards, then release
+ // this chunk.
+ kRewriteRequested = 3,
+
+ // The writer has finished with the old chunk. The reader may reclaim it.
+ kRewriteAcknowledged = 4,
+
+ // A reader that does not know a state cannot tell who owns the chunk. It
+ // stops rather than reclaiming it.
+ kReserved5 = 5,
+ kReserved6 = 6,
+ kReserved7 = 7,
+};
+
+enum class ChunkFormat : uint32_t {
+ // Bytes 4..5 contain a little-endian target BufferID. Payload starts at 6.
+ kTargetBuffer = 0,
+
+ // Reserved for a format carrying per-packet routing information.
+ kReservedRouting = 1,
+ kReserved2 = 2,
+ kReserved3 = 3,
+};
+
+// Field widths and positions in the 32-bit word. The low byte is the control
+// byte, followed by the fragment count and WriterID.
+constexpr uint32_t kPayloadFlagsBits = 3;
+constexpr uint32_t kChunkFormatBits = 2;
+constexpr uint32_t kChunkStateBits = 3;
+constexpr uint32_t kNumFragmentsBits = 8;
+constexpr uint32_t kWriterIdBits = 16;
+
+constexpr uint32_t kChunkStateShift = 0;
+constexpr uint32_t kChunkFormatShift = kChunkStateShift + kChunkStateBits;
+constexpr uint32_t kPayloadFlagsShift = kChunkFormatShift + kChunkFormatBits;
+constexpr uint32_t kNumFragmentsShift = kPayloadFlagsShift + kPayloadFlagsBits;
+constexpr uint32_t kWriterIdShift = kNumFragmentsShift + kNumFragmentsBits;
+
+constexpr uint32_t kChunkStateMask = (1u << kChunkStateBits) - 1;
+constexpr uint32_t kChunkFormatMask = ((1u << kChunkFormatBits) - 1)
+ << kChunkFormatShift;
+constexpr uint32_t kPayloadFlagsMask = ((1u << kPayloadFlagsBits) - 1)
+ << kPayloadFlagsShift;
+constexpr uint32_t kNumFragmentsMask = ((1u << kNumFragmentsBits) - 1)
+ << kNumFragmentsShift;
+constexpr uint32_t kWriterIdMask = ((1u << kWriterIdBits) - 1)
+ << kWriterIdShift;
+
+constexpr uint32_t kMaxFragmentsPerChunk = (1u << kNumFragmentsBits) - 1;
+
+// Free uses the WriterID field for the wrap count. Its other data bits must be
+// zero.
+constexpr uint32_t kWrapCountBits = 16;
+constexpr uint32_t kWrapCountShift = kWriterIdShift;
+constexpr uint32_t kWrapCountMask = ((1u << kWrapCountBits) - 1)
+ << kWrapCountShift;
+constexpr uint32_t kFreeReservedBitsMask = ~(kChunkStateMask | kWrapCountMask);
+
+// Free stores the low 16 bits of the position's traversal number:
+//
+// wrap_count = uint16_t(position >> chunk_index_bits)
+//
+// A writer can claim a chunk only when this value matches its reservation.
+// This stops a delayed writer from claiming a Free word belonging to another
+// traversal. The 16-bit identity repeats after:
+//
+// min(num_chunks * 65536, 2^32) reservations
+constexpr uint16_t WrapCountForPosition(uint32_t position,
+ uint32_t chunk_index_bits) {
+ return static_cast<uint16_t>(position >> chunk_index_bits);
+}
+
+enum PayloadFlags : uint32_t {
+ // The writer dropped trace data before writing this chunk.
+ kFlagDataLoss = 1u << kPayloadFlagsShift,
+
+ // The last fragment is not the end of its packet; the packet continues in
+ // this writer's next chunk.
+ kFlagContinuesOnNextChunk = 1u << (kPayloadFlagsShift + 1),
+
+ // The first fragment contains the next part of a packet that started in this
+ // writer's previous chunk.
+ kFlagContinuesFromPrevChunk = 1u << (kPayloadFlagsShift + 2),
+};
+
+constexpr ChunkState ChunkStateOf(uint32_t state_word) {
+ return static_cast<ChunkState>((state_word & kChunkStateMask) >>
+ kChunkStateShift);
+}
+
+constexpr bool HasDataFields(ChunkState state) {
+ return state == ChunkState::kBeingWritten || state == ChunkState::kComplete ||
+ state == ChunkState::kRewriteRequested;
+}
+
+// Use these helpers only with Free. The uint16_t parameter keeps all reserved
+// bits clear.
+constexpr uint16_t WrapCountOf(uint32_t state_word) {
+ return static_cast<uint16_t>((state_word & kWrapCountMask) >>
+ kWrapCountShift);
+}
+
+constexpr uint32_t MakeFreeStateWord(uint16_t wrap_count) {
+ return static_cast<uint32_t>(wrap_count) << kWrapCountShift;
+}
+
+// These accessors apply to BeingWritten, Complete and RewriteRequested.
+// PayloadFlagsOf() leaves the flags in their encoded bit positions, so its
+// result can be passed to MakeDataStateWord().
+constexpr ChunkFormat ChunkFormatOf(uint32_t state_word) {
+ return static_cast<ChunkFormat>((state_word & kChunkFormatMask) >>
+ kChunkFormatShift);
+}
+
+constexpr uint32_t PayloadFlagsOf(uint32_t state_word) {
+ return state_word & kPayloadFlagsMask;
+}
+
+constexpr uint32_t NumFragmentsOf(uint32_t state_word) {
+ return (state_word & kNumFragmentsMask) >> kNumFragmentsShift;
+}
+
+constexpr WriterID WriterIdOf(uint32_t state_word) {
+ return static_cast<WriterID>((state_word & kWriterIdMask) >> kWriterIdShift);
+}
+
+inline uint32_t MakeDataStateWord(ChunkState state,
+ ChunkFormat format,
+ uint32_t payload_flags,
+ uint32_t num_fragments,
+ WriterID writer_id) {
+ PERFETTO_DCHECK(HasDataFields(state));
+ PERFETTO_DCHECK(static_cast<uint32_t>(format) < (1u << kChunkFormatBits));
+ PERFETTO_DCHECK((payload_flags & ~kPayloadFlagsMask) == 0);
+ PERFETTO_DCHECK(num_fragments <= kMaxFragmentsPerChunk);
+ return (static_cast<uint32_t>(state) << kChunkStateShift) |
+ (static_cast<uint32_t>(format) << kChunkFormatShift) |
+ (payload_flags & kPayloadFlagsMask) |
+ ((num_fragments << kNumFragmentsShift) & kNumFragmentsMask) |
+ (static_cast<uint32_t>(writer_id) << kWriterIdShift);
+}
+
+constexpr uint32_t kRewriteAcknowledgedStateWord =
+ static_cast<uint32_t>(ChunkState::kRewriteAcknowledged) << kChunkStateShift;
+
+// The reader can request a rewrite without understanding the chunk format.
+constexpr uint32_t ReplaceChunkState(uint32_t state_word, ChunkState state) {
+ return (state_word & ~kChunkStateMask) |
+ (static_cast<uint32_t>(state) << kChunkStateShift);
+}
+
+// Target-buffer chunk format
+// --------------------------
+//
+// A data-bearing format-0 chunk begins with this six-byte header:
+//
+// +---------+---------+-------------------+-------------------+
+// | byte 0 | byte 1 | bytes 2-3 | bytes 4-5 |
+// +---------+---------+-------------------+-------------------+
+// | control | num | WriterID | target BufferID |
+// | byte |fragments| | |
+// +---------+---------+-------------------+-------------------+
+// \_____________ atomic state word _______/
+//
+// Free uses bytes 2-3 for the wrap count and requires every other data bit to
+// be zero. RewriteAcknowledged carries only the control byte; bytes 1-3 are
+// zero.
+//
+// The rest of a format-0 chunk is laid out as follows:
+//
+// low address high address
+// 0 4 6 chunk_size
+// +-----------------+----------+-----------+------+--------------+
+// | atomic state | BufferID | payloads | free | size varints |
+// | word | | grow ---> | | <--- grow |
+// +-----------------+----------+-----------+------+--------------+
+// ... N 1 0
+//
+// Fragment 0's size is stored at the end of the chunk. num_frags publishes the
+// same number of payload fragments and size varints. The writer fills those
+// bytes before its release transition out of BeingWritten. The reader
+// acquire-loads the state word, then decodes and checks every length before
+// copying the payload.
+
+constexpr uint32_t kTargetBufferIdOffset = 4;
+constexpr uint32_t kTargetBufferPayloadOffset = 6;
+
+inline void StoreTargetBufferId(uint8_t* chunk, BufferID target_buffer_id) {
+ chunk[kTargetBufferIdOffset] = static_cast<uint8_t>(target_buffer_id);
+ chunk[kTargetBufferIdOffset + 1] =
+ static_cast<uint8_t>(target_buffer_id >> 8);
+}
+
+inline BufferID LoadTargetBufferId(const uint8_t* chunk) {
+ return static_cast<BufferID>(
+ static_cast<uint32_t>(chunk[kTargetBufferIdOffset]) |
+ (static_cast<uint32_t>(chunk[kTargetBufferIdOffset + 1]) << 8));
+}
+
+// Each fragment has a varint length in the directory at the end of the chunk.
+// The first fragment's varint ends at chunk_size. Each later varint is
+// prepended to the directory:
+//
+// low address high address
+// +----------+----------+----------+----------+----------+
+// | size N-1 | ... | size 2 | size 1 | size 0 |
+// +----------+----------+----------+----------+----------+
+//
+// The reader walks the directory from high addresses to low addresses. It sees
+// the bytes of each length in normal protobuf varint order and stops at that
+// varint's final byte. It never has to inspect the next, unpublished entry.
+//
+// Fragment lengths must use the shortest varint encoding. For example, 1 is
+// encoded as 01, not 81 00. The reader rejects redundant encodings.
+
+constexpr uint32_t kMaxFragmentSizeVarIntBytes = PERFETTO_PB_VARINT_MAX_SIZE_32;
+
+// Each varint byte carries seven value bits. Its top bit is set when another
+// byte follows.
+constexpr uint32_t kVarIntDataBitsPerByte = 7;
+constexpr uint8_t kVarIntContinuationBit = 1u << kVarIntDataBitsPerByte;
+
+constexpr uint32_t FragmentSizeVarIntBytes(uint32_t fragment_size) {
+ uint32_t bytes = 1;
+ while (fragment_size >= kVarIntContinuationBit) {
+ fragment_size >>= kVarIntDataBitsPerByte;
+ ++bytes;
+ }
+ return bytes;
+}
+
+// Returns the largest fragment that fits in |available_bytes| together with
+// its varint length. The loop runs at most four times: each decrement buys one
+// more byte for a varint that can be at most five bytes long.
+constexpr uint32_t MaxFragmentSizeForAvailableBytes(uint32_t available_bytes) {
+ if (available_bytes <= 1)
+ return 0;
+ uint32_t fragment_size = available_bytes - 1;
+ while (FragmentSizeVarIntBytes(fragment_size) >
+ available_bytes - fragment_size) {
+ --fragment_size;
+ }
+ return fragment_size;
+}
+
+constexpr uint32_t MaxFragmentSizeForEmptyChunk(uint32_t chunk_size) {
+ if (chunk_size < kMinChunkSize)
+ return 0;
+ const uint32_t available_bytes = chunk_size - kTargetBufferPayloadOffset;
+ return MaxFragmentSizeForAvailableBytes(available_bytes);
+}
+
+// Prepends one fragment length and returns the new beginning of the directory.
+// The destination moves down while the encoded bytes are copied in order. A
+// reader moving down through the directory therefore sees an ordinary protobuf
+// varint. The caller must leave FragmentSizeVarIntBytes(fragment_size) bytes
+// before |directory_begin|.
+inline uint8_t* WriteFragmentSize(uint8_t* directory_begin,
+ uint32_t fragment_size) {
+ uint8_t encoded[PERFETTO_PB_VARINT_MAX_SIZE_32];
+ const uint8_t* const encoded_end =
+ PerfettoPbWriteVarInt(fragment_size, encoded);
+ const size_t encoded_size = static_cast<size_t>(encoded_end - encoded);
+ for (size_t i = 0; i < encoded_size; ++i) {
+ --directory_begin;
+ *directory_begin = encoded[i];
+ }
+ return directory_begin;
+}
+
+// Reads the next varint while moving |*directory_cursor| towards lower
+// addresses. On success, leaves the cursor immediately before that varint.
+// |directory_begin| is the first byte the decoder may inspect.
+inline bool ReadFragmentSize(const uint8_t* directory_begin,
+ const uint8_t** directory_cursor,
+ uint32_t* fragment_size) {
+ uint8_t encoded[PERFETTO_PB_VARINT_MAX_SIZE_32];
+ uint32_t encoded_size = 0;
+ const uint8_t* cursor = *directory_cursor;
+ for (;;) {
+ if (cursor == directory_begin ||
+ encoded_size == kMaxFragmentSizeVarIntBytes) {
+ return false;
+ }
+ const uint8_t byte = *--cursor;
+ encoded[encoded_size++] = byte;
+ if ((byte & kVarIntContinuationBit) == 0)
+ break;
+ }
+
+ uint64_t value = 0;
+ if (PerfettoPbParseVarInt(encoded, encoded + encoded_size, &value) !=
+ encoded + encoded_size ||
+ value > UINT32_MAX ||
+ FragmentSizeVarIntBytes(static_cast<uint32_t>(value)) != encoded_size) {
+ return false;
+ }
+
+ *directory_cursor = cursor;
+ *fragment_size = static_cast<uint32_t>(value);
+ return true;
+}
+
+} // namespace perfetto::tracing_v2
+
+#endif // SRC_TRACING_V2_TRACING_V2_ABI_H_
diff --git a/src/tracing/v2/tracing_v2_abi_unittest.cc b/src/tracing/v2/tracing_v2_abi_unittest.cc
new file mode 100644
index 0000000..e82feb8
--- /dev/null
+++ b/src/tracing/v2/tracing_v2_abi_unittest.cc
@@ -0,0 +1,419 @@
+/*
+ * Copyright (C) 2026 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/v2/tracing_v2_abi.h"
+
+#include <stdint.h>
+
+#include <vector>
+
+#include "test/gtest_and_gmock.h"
+
+namespace perfetto::tracing_v2 {
+namespace {
+
+// Chunk state word
+// ----------------
+
+// Keep exact wire values in the test, rather than obscuring the ABI header with
+// compile-time examples. These literals are intentionally independent of the
+// masks used by the encoding helpers.
+TEST(TracingV2AbiTest, StateWordEncoding) {
+ EXPECT_EQ(MakeFreeStateWord(0), 0x00000000u);
+ EXPECT_EQ(MakeFreeStateWord(5), 0x00050000u);
+ EXPECT_EQ(MakeFreeStateWord(0xffff), 0xffff0000u);
+ EXPECT_EQ(MakeDataStateWord(ChunkState::kBeingWritten,
+ ChunkFormat::kTargetBuffer, 0, 0, 7),
+ 0x00070001u);
+ EXPECT_EQ(MakeDataStateWord(ChunkState::kComplete, ChunkFormat::kTargetBuffer,
+ 0, 3, 7),
+ 0x00070302u);
+ EXPECT_EQ(MakeDataStateWord(ChunkState::kRewriteRequested,
+ ChunkFormat::kTargetBuffer, 0, 3, 7),
+ 0x00070303u);
+ EXPECT_EQ(MakeDataStateWord(ChunkState::kComplete, ChunkFormat::kTargetBuffer,
+ kFlagContinuesOnNextChunk, 2, 0x1234),
+ 0x12340242u);
+ EXPECT_EQ(kRewriteAcknowledgedStateWord, 0x00000004u);
+}
+
+TEST(TracingV2AbiTest, PayloadFlagEncoding) {
+ EXPECT_EQ(kFlagDataLoss, 0x00000020u);
+ EXPECT_EQ(kFlagContinuesOnNextChunk, 0x00000040u);
+ EXPECT_EQ(kFlagContinuesFromPrevChunk, 0x00000080u);
+}
+
+TEST(TracingV2AbiTest, StateOrdinalsAndTheirDecoding) {
+ EXPECT_EQ(ChunkStateOf(0x00000000u), ChunkState::kFree);
+ EXPECT_EQ(ChunkStateOf(0x00000001u), ChunkState::kBeingWritten);
+ EXPECT_EQ(ChunkStateOf(0x00000002u), ChunkState::kComplete);
+ EXPECT_EQ(ChunkStateOf(0x00000003u), ChunkState::kRewriteRequested);
+ EXPECT_EQ(ChunkStateOf(0x00000004u), ChunkState::kRewriteAcknowledged);
+ EXPECT_EQ(ChunkStateOf(0x00000005u), ChunkState::kReserved5);
+ EXPECT_EQ(ChunkStateOf(0x00000006u), ChunkState::kReserved6);
+ EXPECT_EQ(ChunkStateOf(0x00000007u), ChunkState::kReserved7);
+
+ EXPECT_TRUE(HasDataFields(ChunkState::kBeingWritten));
+ EXPECT_TRUE(HasDataFields(ChunkState::kComplete));
+ EXPECT_TRUE(HasDataFields(ChunkState::kRewriteRequested));
+ EXPECT_FALSE(HasDataFields(ChunkState::kFree));
+ EXPECT_FALSE(HasDataFields(ChunkState::kRewriteAcknowledged));
+ EXPECT_FALSE(HasDataFields(ChunkState::kReserved5));
+ EXPECT_FALSE(HasDataFields(ChunkState::kReserved6));
+ EXPECT_FALSE(HasDataFields(ChunkState::kReserved7));
+}
+
+TEST(TracingV2AbiTest, DataStateFieldsRoundTrip) {
+ const ChunkState kStates[] = {ChunkState::kBeingWritten,
+ ChunkState::kComplete,
+ ChunkState::kRewriteRequested};
+ const uint32_t kFlagSets[] = {
+ 0,
+ kFlagContinuesFromPrevChunk,
+ kFlagContinuesOnNextChunk,
+ kFlagDataLoss,
+ kFlagContinuesFromPrevChunk | kFlagDataLoss,
+ kFlagContinuesFromPrevChunk | kFlagContinuesOnNextChunk | kFlagDataLoss};
+ const uint32_t kCounts[] = {0, 1, 127, 128, 254, 255};
+ const WriterID kWriters[] = {0, 1, 0x1234, 0x7fff, 0xffff};
+
+ for (ChunkState state : kStates) {
+ for (uint32_t flags : kFlagSets) {
+ for (uint32_t count : kCounts) {
+ for (WriterID writer : kWriters) {
+ const uint32_t word = MakeDataStateWord(
+ state, ChunkFormat::kTargetBuffer, flags, count, writer);
+ EXPECT_EQ(ChunkStateOf(word), state);
+ EXPECT_EQ(ChunkFormatOf(word), ChunkFormat::kTargetBuffer);
+ EXPECT_EQ(PayloadFlagsOf(word), flags);
+ EXPECT_EQ(NumFragmentsOf(word), count);
+ EXPECT_EQ(WriterIdOf(word), writer);
+ }
+ }
+ }
+ }
+}
+
+TEST(TracingV2AbiTest, EveryFormatRoundTrips) {
+ const ChunkFormat kFormats[] = {
+ ChunkFormat::kTargetBuffer, ChunkFormat::kReservedRouting,
+ ChunkFormat::kReserved2, ChunkFormat::kReserved3};
+ for (ChunkFormat format : kFormats) {
+ const uint32_t word = MakeDataStateWord(ChunkState::kComplete, format,
+ kFlagDataLoss, 9, 0xabcd);
+ EXPECT_EQ(ChunkFormatOf(word), format);
+ // Whatever the format is, the state and the fields the reader needs to
+ // arbitrate ownership stay where they are. That is what lets an old reader
+ // release a chunk whose layout it has never heard of.
+ EXPECT_EQ(ChunkStateOf(word), ChunkState::kComplete);
+ EXPECT_EQ(NumFragmentsOf(word), 9u);
+ EXPECT_EQ(WriterIdOf(word), 0xabcd);
+ }
+}
+
+TEST(TracingV2AbiTest, ReplaceChunkStateChangesOnlyTheState) {
+ const uint32_t being_written = MakeDataStateWord(
+ ChunkState::kBeingWritten, ChunkFormat::kReservedRouting,
+ kFlagContinuesFromPrevChunk | kFlagDataLoss, 17, 0x0f0f);
+ const uint32_t marked =
+ ReplaceChunkState(being_written, ChunkState::kRewriteRequested);
+
+ EXPECT_EQ(ChunkStateOf(marked), ChunkState::kRewriteRequested);
+ EXPECT_EQ(being_written & ~kChunkStateMask, marked & ~kChunkStateMask);
+ EXPECT_EQ(ChunkFormatOf(marked), ChunkFormat::kReservedRouting);
+ EXPECT_EQ(PayloadFlagsOf(marked),
+ kFlagContinuesFromPrevChunk | kFlagDataLoss);
+ EXPECT_EQ(NumFragmentsOf(marked), 17u);
+ EXPECT_EQ(WriterIdOf(marked), 0x0f0f);
+}
+
+TEST(TracingV2AbiTest, FreeSplitsIntoStateReservedBitsAndA16BitWrap) {
+ EXPECT_EQ(kWrapCountMask, 0xffff0000u);
+ EXPECT_EQ(kFreeReservedBitsMask, 0x0000fff8u);
+
+ EXPECT_EQ(WrapCountOf(MakeFreeStateWord(0)), 0u);
+ EXPECT_EQ(WrapCountOf(MakeFreeStateWord(1)), 1u);
+ EXPECT_EQ(WrapCountOf(MakeFreeStateWord(0xffff)), 0xffffu);
+
+ // MakeFreeStateWord takes a uint16_t, so no caller can produce a word with a
+ // reserved bit set; decoding one still yields only bytes 2-3.
+ EXPECT_EQ(WrapCountOf(0x0005fff8u), 5u);
+ EXPECT_EQ(ChunkStateOf(0x0005fff8u), ChunkState::kFree);
+}
+
+// Packed read/write positions
+// ---------------------------
+
+TEST(TracingV2AbiTest, RwPositionsEncoding) {
+ // write_pos occupies the high half of the numeric value and read_pos the low
+ // half.
+ EXPECT_EQ(PackRwPositions(0, 0), 0u);
+ EXPECT_EQ(PackRwPositions(1, 0), 0x0000000100000000ull);
+ EXPECT_EQ(PackRwPositions(0, 1), 0x0000000000000001ull);
+ EXPECT_EQ(PackRwPositions(0xdeadbeefu, 0x12345678u), 0xdeadbeef12345678ull);
+ EXPECT_EQ(PackRwPositions(0x00000002u, 0xfffffffcu), 0x00000002fffffffcull);
+}
+
+TEST(TracingV2AbiTest, PackedRwPositionsRoundTrips) {
+ const uint32_t kPositions[] = {0u, 1u, 0x7fffffffu,
+ 0x80000000u, 0xfffffffeu, 0xffffffffu};
+ for (uint32_t write_pos : kPositions) {
+ for (uint32_t read_pos : kPositions) {
+ const uint64_t rw_positions = PackRwPositions(write_pos, read_pos);
+ EXPECT_EQ(WritePosOf(rw_positions), write_pos);
+ EXPECT_EQ(ReadPosOf(rw_positions), read_pos);
+ }
+ }
+}
+
+TEST(TracingV2AbiTest, ReplaceWritePosReplacesOnlyTheWriteHalf) {
+ const uint64_t rw_positions = PackRwPositions(0x11111111u, 0x22222222u);
+ const uint64_t moved = ReplaceWritePos(rw_positions, 0x11111112u);
+ EXPECT_EQ(WritePosOf(moved), 0x11111112u);
+ EXPECT_EQ(ReadPosOf(moved), 0x22222222u);
+ // Including across the halves' own rollovers.
+ EXPECT_EQ(ReplaceWritePos(PackRwPositions(0xffffffffu, 7), 0),
+ PackRwPositions(0u, 7u));
+}
+
+TEST(TracingV2AbiTest, ReplaceReadPosReplacesOnlyTheReadHalf) {
+ const uint64_t rw_positions = PackRwPositions(0x11111111u, 0x22222222u);
+ const uint64_t moved = ReplaceReadPos(rw_positions, 0x22222223u);
+ EXPECT_EQ(WritePosOf(moved), 0x11111111u);
+ EXPECT_EQ(ReadPosOf(moved), 0x22222223u);
+ EXPECT_EQ(ReplaceReadPos(PackRwPositions(7, 0xffffffffu), 0),
+ PackRwPositions(7u, 0u));
+}
+
+// Logical positions
+// -----------------
+
+TEST(TracingV2AbiTest, ChunkIndexAndWrapCountForSeveralRingSizes) {
+ // A worked example.
+ const uint32_t kNumChunks = 4;
+ const uint32_t kChunkIndexBits = GetChunkIndexBits(kNumChunks);
+ const uint32_t kExpectedIndex[] = {0, 1, 2, 3, 0, 1, 2, 3, 0, 1, 2, 3, 0};
+ const uint32_t kExpectedWrap[] = {0, 0, 0, 0, 1, 1, 1, 1, 2, 2, 2, 2, 3};
+ for (uint32_t p = 0; p < 13; ++p) {
+ EXPECT_EQ(ChunkIndexOfPosition(p, kNumChunks), kExpectedIndex[p]) << p;
+ EXPECT_EQ(WrapCountForPosition(p, kChunkIndexBits), kExpectedWrap[p]) << p;
+ }
+
+ // Including the one-chunk ring, where every position maps to chunk 0 and the
+ // wrap count is the position itself, truncated.
+ for (uint32_t num_chunks : {1u, 2u, 4u, 8u, 1024u}) {
+ const uint32_t chunk_index_bits = GetChunkIndexBits(num_chunks);
+ for (uint32_t p = 0; p < 3 * num_chunks + 3; ++p) {
+ EXPECT_EQ(ChunkIndexOfPosition(p, num_chunks), p % num_chunks);
+ EXPECT_EQ(WrapCountForPosition(p, chunk_index_bits),
+ (p / num_chunks) & 0xffffu);
+ }
+ }
+}
+
+TEST(TracingV2AbiTest, OutstandingPositionsAreExactAcrossRollover) {
+ // A worked example.
+ EXPECT_EQ(NumOutstandingPositions(0x00000002u, 0xfffffffcu), 6u);
+ EXPECT_EQ(NumOutstandingPositions(0u, 0u), 0u);
+ EXPECT_EQ(NumOutstandingPositions(0u, 0xffffffffu), 1u);
+ EXPECT_EQ(NumOutstandingPositions(0xffffffffu, 0xfffffffeu), 1u);
+
+ const uint32_t kNumChunks = 8;
+ const uint32_t kExpected[] = {4, 5, 6, 7, 0, 1};
+ uint32_t p = 0xfffffffcu;
+ for (uint32_t i = 0; i < 6; ++i, ++p)
+ EXPECT_EQ(ChunkIndexOfPosition(p, kNumChunks), kExpected[i]) << i;
+}
+
+TEST(TracingV2AbiTest, NextWrapComesFromTheNextPositionNotFromTheChunk) {
+ // Away from the rollovers next_wrap is simply "one more".
+ const uint32_t kNumChunks = 4;
+ const uint32_t kChunkIndexBits = GetChunkIndexBits(kNumChunks);
+ for (uint32_t p = 0; p < 16; ++p) {
+ EXPECT_EQ(WrapCountForPosition(p + kNumChunks, kChunkIndexBits),
+ WrapCountForPosition(p, kChunkIndexBits) + 1)
+ << p;
+ }
+
+ // At the 16-bit truncation boundary it is not: the wrap after 0xffff is
+ // zero. A reader that incremented the value it found in the chunk would
+ // agree here by accident of the uint16_t, so the position rollover below is
+ // the discriminating case.
+ const uint32_t kLastLapOfPeriod = 0xffffu * kNumChunks; // wrap 0xffff
+ EXPECT_EQ(WrapCountForPosition(kLastLapOfPeriod, kChunkIndexBits), 0xffffu);
+ EXPECT_EQ(
+ WrapCountForPosition(kLastLapOfPeriod + kNumChunks, kChunkIndexBits), 0u);
+
+ // At the uint32_t position rollover the shifted position restarts from zero
+ // mid-way through the 16-bit range whenever num_chunks > 65536, which is why
+ // the protocol derives the wrap from the position and never increments the
+ // value it finds in the chunk.
+ const uint32_t kBigRing = 1u << 20;
+ const uint32_t kBigChunkIndexBits = GetChunkIndexBits(kBigRing);
+ const uint32_t kLastPosition = 0u - kBigRing; // the last lap's chunk 0
+ EXPECT_EQ(ChunkIndexOfPosition(kLastPosition, kBigRing), 0u);
+ EXPECT_EQ(WrapCountForPosition(kLastPosition, kBigChunkIndexBits), 0xfffu);
+ EXPECT_EQ(WrapCountForPosition(kLastPosition + kBigRing, kBigChunkIndexBits),
+ 0u);
+ EXPECT_NE(WrapCountForPosition(kLastPosition + kBigRing, kBigChunkIndexBits),
+ WrapCountForPosition(kLastPosition, kBigChunkIndexBits) + 1u);
+}
+
+// Pin the finite period of the 16-bit Free identity.
+TEST(TracingV2AbiTest, WrapIdentityRepeatsAfterTheDocumentedPeriod) {
+ // For num_chunks up to 65536 the identity period is num_chunks * 65536
+ // reservations: the same chunk carries the same wrap count again once the
+ // shifted position has run through the whole uint16_t.
+ for (uint32_t num_chunks : {1u, 2u, 16u}) {
+ const uint32_t chunk_index_bits = GetChunkIndexBits(num_chunks);
+ const uint32_t period = num_chunks * 65536;
+ EXPECT_EQ(WrapCountForPosition(0, chunk_index_bits),
+ WrapCountForPosition(period, chunk_index_bits));
+ EXPECT_EQ(ChunkIndexOfPosition(0, num_chunks),
+ ChunkIndexOfPosition(period, num_chunks));
+ // No earlier lap of the same chunk aliases position 0.
+ for (uint32_t lap = 1; lap < 8; ++lap) {
+ const uint32_t p = lap * num_chunks;
+ EXPECT_NE(WrapCountForPosition(p, chunk_index_bits),
+ WrapCountForPosition(0, chunk_index_bits));
+ }
+ }
+
+ // From 65537 chunks up the truncation throws nothing away - the shifted
+ // position already fits in 16 bits - so the identity repeats only when the
+ // 32-bit position itself wraps: the period is min(num_chunks * 65536, 2^32).
+ const uint32_t kBigRing = 1u << 20;
+ const uint32_t kBigChunkIndexBits = GetChunkIndexBits(kBigRing);
+ EXPECT_EQ(WrapCountForPosition(UINT32_MAX, kBigChunkIndexBits), 0xfffu);
+ for (uint32_t lap = 1; lap < 8; ++lap) {
+ EXPECT_NE(WrapCountForPosition(lap * kBigRing, kBigChunkIndexBits),
+ WrapCountForPosition(0, kBigChunkIndexBits));
+ }
+}
+
+// Target-buffer chunk format
+// --------------------------
+
+TEST(TracingV2AbiTest, FragmentSizeVarIntBytesAtBoundaries) {
+ EXPECT_EQ(FragmentSizeVarIntBytes(0), 1u);
+ EXPECT_EQ(FragmentSizeVarIntBytes(127), 1u);
+ EXPECT_EQ(FragmentSizeVarIntBytes(128), 2u);
+ EXPECT_EQ(FragmentSizeVarIntBytes(16383), 2u);
+ EXPECT_EQ(FragmentSizeVarIntBytes(16384), 3u);
+ EXPECT_EQ(FragmentSizeVarIntBytes(0x1fffff), 3u);
+ EXPECT_EQ(FragmentSizeVarIntBytes(0x200000), 4u);
+ EXPECT_EQ(FragmentSizeVarIntBytes(0x0fffffff), 4u);
+ EXPECT_EQ(FragmentSizeVarIntBytes(0x10000000), 5u);
+ EXPECT_EQ(FragmentSizeVarIntBytes(UINT32_MAX), 5u);
+}
+
+TEST(TracingV2AbiTest, MaxFragmentSizeLeavesRoomForItsVarInt) {
+ EXPECT_EQ(MaxFragmentSizeForAvailableBytes(0), 0u);
+ EXPECT_EQ(MaxFragmentSizeForAvailableBytes(1), 0u);
+ EXPECT_EQ(MaxFragmentSizeForAvailableBytes(2), 1u);
+ EXPECT_EQ(MaxFragmentSizeForAvailableBytes(128), 127u);
+ EXPECT_EQ(MaxFragmentSizeForAvailableBytes(129), 127u);
+ EXPECT_EQ(MaxFragmentSizeForAvailableBytes(130), 128u);
+ EXPECT_EQ(MaxFragmentSizeForAvailableBytes(250), 248u);
+ EXPECT_EQ(MaxFragmentSizeForAvailableBytes(UINT32_MAX), UINT32_MAX - 5u);
+}
+
+TEST(TracingV2AbiTest, MaxFragmentSizeForEmptyChunk) {
+ EXPECT_EQ(MaxFragmentSizeForEmptyChunk(kMinChunkSize - 1), 0u);
+ EXPECT_EQ(MaxFragmentSizeForEmptyChunk(256), 248u);
+ EXPECT_EQ(MaxFragmentSizeForEmptyChunk(260), 252u); // Non-power-of-two.
+ EXPECT_EQ(MaxFragmentSizeForEmptyChunk(65536), 65527u);
+ EXPECT_EQ(MaxFragmentSizeForEmptyChunk(128u * 1024), 128u * 1024 - 9u);
+}
+
+TEST(TracingV2AbiTest, VarIntDirectoryGrowsDownFromTheEndOfTheChunk) {
+ std::vector<uint8_t> chunk(256, 0);
+ uint8_t* directory_begin = chunk.data() + chunk.size();
+ for (uint32_t size : {5u, 200u, 3u})
+ directory_begin = WriteFragmentSize(directory_begin, size);
+
+ // Fragment 0 is nearest the end. Reading towards lower addresses yields the
+ // normal varint byte sequence c8 01 for 200.
+ EXPECT_EQ(directory_begin, chunk.data() + 252);
+ EXPECT_EQ(chunk[252], 0x03u);
+ EXPECT_EQ(chunk[253], 0x01u);
+ EXPECT_EQ(chunk[254], 0xc8u);
+ EXPECT_EQ(chunk[255], 0x05u);
+
+ const uint8_t* cursor = chunk.data() + chunk.size();
+ for (uint32_t expected : {5u, 200u, 3u}) {
+ uint32_t actual = 0;
+ ASSERT_TRUE(ReadFragmentSize(directory_begin, &cursor, &actual));
+ EXPECT_EQ(actual, expected);
+ }
+ EXPECT_EQ(cursor, directory_begin);
+}
+
+TEST(TracingV2AbiTest, FragmentSizeVarIntsRoundTrip) {
+ const uint32_t kSizes[] = {0, 1, 127, 128,
+ 16383, 16384, 0x1fffff, 0x200000,
+ 0x0fffffff, 0x10000000, UINT32_MAX};
+ std::vector<uint8_t> directory(64, 0xee);
+ uint8_t* directory_begin = directory.data() + directory.size();
+ for (uint32_t size : kSizes)
+ directory_begin = WriteFragmentSize(directory_begin, size);
+
+ const uint8_t* cursor = directory.data() + directory.size();
+ for (uint32_t expected : kSizes) {
+ uint32_t actual = 0;
+ ASSERT_TRUE(ReadFragmentSize(directory_begin, &cursor, &actual));
+ EXPECT_EQ(actual, expected);
+ }
+ EXPECT_EQ(cursor, directory_begin);
+}
+
+TEST(TracingV2AbiTest, RejectsMalformedOrRedundantFragmentSizeVarInts) {
+ const uint8_t kUnterminated[] = {0x80};
+ const uint8_t kTooLong[] = {0x00, 0x80, 0x80, 0x80, 0x80, 0x80};
+ const uint8_t kUint32Overflow[] = {0x10, 0xff, 0xff, 0xff, 0xff};
+ const uint8_t kNonCanonical[] = {0x00, 0x80};
+
+ auto expect_rejected = [](const uint8_t* begin, size_t size) {
+ const uint8_t* cursor = begin + size;
+ uint32_t fragment_size = 0;
+ EXPECT_FALSE(ReadFragmentSize(begin, &cursor, &fragment_size));
+ EXPECT_EQ(cursor, begin + size);
+ };
+ expect_rejected(kUnterminated, sizeof(kUnterminated));
+ expect_rejected(kTooLong, sizeof(kTooLong));
+ expect_rejected(kUint32Overflow, sizeof(kUint32Overflow));
+ expect_rejected(kNonCanonical, sizeof(kNonCanonical));
+}
+
+TEST(TracingV2AbiTest, TargetBufferIdIsLittleEndianAtBytesFourAndFive) {
+ std::vector<uint8_t> chunk(256, 0);
+ StoreTargetBufferId(chunk.data(), 0x1234);
+ EXPECT_EQ(chunk[4], 0x34u);
+ EXPECT_EQ(chunk[5], 0x12u);
+ EXPECT_EQ(LoadTargetBufferId(chunk.data()), 0x1234);
+ // The state word is not disturbed and the payload area still starts at 6.
+ EXPECT_EQ(chunk[0], 0u);
+ EXPECT_EQ(chunk[3], 0u);
+ EXPECT_EQ(chunk[kTargetBufferPayloadOffset], 0u);
+
+ StoreTargetBufferId(chunk.data(), 0xffff);
+ EXPECT_EQ(LoadTargetBufferId(chunk.data()), 0xffff);
+ StoreTargetBufferId(chunk.data(), 0);
+ EXPECT_EQ(LoadTargetBufferId(chunk.data()), 0);
+}
+
+} // namespace
+} // namespace perfetto::tracing_v2
