f
diff --git a/src/tracing/v2/shared_ring_buffer.cc b/src/tracing/v2/shared_ring_buffer.cc index 9fb9cf9..c560e67 100644 --- a/src/tracing/v2/shared_ring_buffer.cc +++ b/src/tracing/v2/shared_ring_buffer.cc
@@ -25,6 +25,7 @@ #include "perfetto/base/compiler.h" #include "perfetto/base/logging.h" #include "perfetto/ext/base/utils.h" +#include "perfetto/protozero/proto_utils.h" #include "src/tracing/v2/shared_ring_buffer_abi.h" #if PERFETTO_BUILDFLAG(PERFETTO_OS_LINUX_BUT_NOT_QNX) || \ @@ -44,6 +45,9 @@ namespace perfetto::tracing_v2 { namespace { +static_assert(kMaxFragmentSizeVarIntBytes == + protozero::proto_utils::kMessageLengthFieldSize); + uint32_t NumChunksForRingLayout(const uint8_t* start, size_t size, uint32_t chunk_size) { @@ -56,6 +60,8 @@ // size_t on 32-bit builds. PERFETTO_CHECK(size >= sizeof(RingBufferHeader)); const size_t chunks_size = size - sizeof(RingBufferHeader); + // One chunk is valid: full/empty use logical counter distance, and the + // Free wrap count distinguishes successive reservations of that chunk. PERFETTO_CHECK(chunks_size >= chunk_size); PERFETTO_CHECK(chunks_size % chunk_size == 0); const size_t num_chunks = chunks_size / chunk_size; @@ -110,13 +116,13 @@ // --- Writer-side reservation. --- +// TODO(sashwinbalaji): Measure and prove narrower memory ordering after the +// initial landing. Keep all shared-word operations sequentially consistent. + SharedRingBuffer::Reservation SharedRingBuffer::TryReserveWritePos() { - // The reader marks chunks Free before publishing read_pos. - // - // Memory order acquire ensures that a writer which sees the new - // read_pos also sees those Free state words. - return TryReserveWritePosFromSnapshot( - header()->rw_positions.load(std::memory_order_acquire)); + // The reader marks chunks Free before publishing read_pos, so a writer + // seeing the new position also sees those Free words. + return TryReserveWritePosFromSnapshot(header()->rw_positions.load()); } SharedRingBuffer::Reservation SharedRingBuffer::TryReserveWritePosFromSnapshot( @@ -140,14 +146,10 @@ // // Failure reloads both halves. The loop rechecks capacity, and no position // was taken. - // - // With memory order acquire, a snapshot that shows a newer read_pos - // also shows the chunks the reader freed before publishing it. if (ring_header->rw_positions.compare_exchange_weak( - rw_positions, PackRwPositions(write_pos + 1, read_pos), - std::memory_order_acquire, std::memory_order_acquire)) { + rw_positions, PackRwPositions(write_pos + 1, read_pos))) { reservation.result = ReserveResult::kReserved; - reservation.position = write_pos; + reservation.write_pos = write_pos; return reservation; } } @@ -155,33 +157,24 @@ // --- Writer-side chunk transitions. --- -bool SharedRingBuffer::TryAcquireChunkForWriting(uint32_t position, +bool SharedRingBuffer::TryAcquireChunkForWriting(uint32_t chunk_pos, uint32_t being_written_word) { PERFETTO_DCHECK(ChunkStateOf(being_written_word) == ChunkState::kBeingWritten); PERFETTO_DCHECK(NumFragmentsOf(being_written_word) == 0); - // Free(wrap(position)) -> BeingWritten(0). + // Free(wrap(chunk_pos)) -> BeingWritten(0). // The claim can fail because: - // - the reader resolved this position as unclaimed. + // - the reader consumed this position as unclaimed. // - an older reservation still owns the chunk. // // Either way, this reservation is a hole and the caller never retries it. - // - // Memory order acquire pairs with the reader's memory order release - // transition to Free. The writer cannot overwrite - // the chunk until the reader is done with its old contents. - // - // On failure, memory order relaxed is enough because the returned word - // is ignored. - uint32_t expected = MakeFreeStateWordForPosition(position, num_chunks_); - std::atomic<uint32_t>* state_word = chunk_state_word_for_position(position); - return state_word->compare_exchange_strong(expected, being_written_word, - std::memory_order_acquire, - std::memory_order_relaxed); + uint32_t expected = MakeFreeStateWordForPosition(chunk_pos, num_chunks_); + std::atomic<uint32_t>* state_word = chunk_state_word_for_position(chunk_pos); + return state_word->compare_exchange_strong(expected, being_written_word); } -bool SharedRingBuffer::TryReleaseChunkAsComplete(uint32_t chunk_idx, +bool SharedRingBuffer::TryReleaseChunkAsComplete(ChunkIndex chunk_idx, uint32_t complete_word, uint32_t* expected) { PERFETTO_DCHECK(ChunkStateOf(*expected) == ChunkState::kBeingWritten); @@ -189,140 +182,87 @@ // BeingWritten(N) -> Complete(M). // The reader can request a rewrite first. Failure then returns - // RewriteRequested(N), and the caller relocates the suffix. - // - // Memory order release publishes the M fragments and their sizes, plus - // BufferID on the first publication. - // - // On failure, memory order acquire ensures that the reader has finished - // copying the prefix before the writer sees its rewrite request. + // RewriteRequested(N), and the caller relocates the suffix. Success + // publishes the M fragments, their sizes and BufferID to the reader. std::atomic<uint32_t>* state_word = chunk_state_word_at(chunk_idx); - return state_word->compare_exchange_strong(*expected, complete_word, - std::memory_order_release, - std::memory_order_acquire); + return state_word->compare_exchange_strong(*expected, complete_word); } -bool SharedRingBuffer::TryReacquireChunkForWriting(uint32_t chunk_idx, +bool SharedRingBuffer::TryReacquireChunkForWriting(ChunkIndex chunk_idx, uint32_t observed) { PERFETTO_DCHECK(ChunkStateOf(observed) == ChunkState::kComplete); // Complete(N) -> BeingWritten(N). // The reader can reclaim the chunk first. The writer then drops its cached // handle and does not touch the chunk again. - // - // Success publishes no bytes, so it uses memory order relaxed. - // - // The read-modify-write still extends the release sequence of Complete(N). - // The reader's memory order acquire load of BeingWritten(N) therefore - // sees the prefix. - // - // Failure also uses memory order relaxed because the returned word is - // ignored. uint32_t expected = observed; std::atomic<uint32_t>* state_word = chunk_state_word_at(chunk_idx); return state_word->compare_exchange_strong( - expected, ReplaceChunkState(observed, ChunkState::kBeingWritten), - std::memory_order_relaxed, std::memory_order_relaxed); + expected, ReplaceChunkState(observed, ChunkState::kBeingWritten)); } -bool SharedRingBuffer::TryAcknowledgeRewrite(uint32_t chunk_idx, +bool SharedRingBuffer::TryAcknowledgeRewrite(ChunkIndex chunk_idx, uint32_t observed) { PERFETTO_DCHECK(ChunkStateOf(observed) == ChunkState::kRewriteRequested); // RewriteRequested -> RewriteAcknowledged. // Only this writer may acknowledge, so failure is a protocol error. - // - // Memory order release orders the suffix copy before the reader - // reclaims the chunk and a later writer overwrites it. - // - // On failure, memory order relaxed is enough because the unexpected - // word is not inspected. + // The suffix copy finishes before this lets the reader reclaim the chunk. uint32_t expected = observed; std::atomic<uint32_t>* state_word = chunk_state_word_at(chunk_idx); - return state_word->compare_exchange_strong( - expected, kRewriteAcknowledgedStateWord, std::memory_order_release, - std::memory_order_relaxed); + return state_word->compare_exchange_strong(expected, + kRewriteAcknowledgedStateWord); } // --- Reader-side chunk transitions. --- -uint32_t SharedRingBuffer::LoadChunkStateWord(uint32_t chunk_idx) const { - // Memory order acquire pairs with the writer's memory order release - // publication of Complete. This makes the - // fragments, their size varints and BufferID visible to the reader. - // - // A BeingWritten word created by reuse extends that release sequence and - // gives the same guarantee. - return chunk_state_word_at(chunk_idx)->load(std::memory_order_acquire); +uint32_t SharedRingBuffer::LoadChunkStateWord(ChunkIndex chunk_idx) const { + return chunk_state_word_at(chunk_idx)->load(); } uint32_t SharedRingBuffer::LoadWritePos() const { - // Memory order relaxed is enough because write_pos only bounds the - // current drain pass. - // - // LoadChunkStateWord() uses memory order acquire to order the payload - // reads. - return WritePosOf(header()->rw_positions.load(std::memory_order_relaxed)); + return WritePosOf(header()->rw_positions.load()); } -bool SharedRingBuffer::TryRequestRewrite(uint32_t chunk_idx, +bool SharedRingBuffer::TryRequestRewrite(ChunkIndex chunk_idx, uint32_t* expected) { PERFETTO_DCHECK(ChunkStateOf(*expected) == ChunkState::kBeingWritten); // BeingWritten(N) -> RewriteRequested(N). // The writer can publish first. The reader then discards its copy and retries - // this position on a later pass. - // - // Memory order release orders the prefix copy before the writer - // observes the request and relocates its suffix. - // - // On failure, memory order relaxed is enough because the reader retries - // this position without using the returned word to read the payload. + // this position immediately. std::atomic<uint32_t>* state_word = chunk_state_word_at(chunk_idx); return state_word->compare_exchange_strong( - *expected, ReplaceChunkState(*expected, ChunkState::kRewriteRequested), - std::memory_order_release, std::memory_order_relaxed); + *expected, ReplaceChunkState(*expected, ChunkState::kRewriteRequested)); } -bool SharedRingBuffer::TryMoveFreeChunkToNextWrap(uint32_t position, +bool SharedRingBuffer::TryMoveFreeChunkToNextWrap(uint32_t chunk_pos, uint32_t* expected) { PERFETTO_DCHECK(ChunkStateOf(*expected) == ChunkState::kFree); - // Free(wrap(position)) -> Free(next wrap). + // Free(wrap(chunk_pos)) -> Free(next wrap). // A delayed writer can claim first. The reader then retries this position on // a later pass and finds the chunk BeingWritten or Complete. - // - // Memory order release publishes Free to the next writer. - // - // On failure, memory order relaxed is enough because the reader retries - // the position. - std::atomic<uint32_t>* state_word = chunk_state_word_for_position(position); + std::atomic<uint32_t>* state_word = chunk_state_word_for_position(chunk_pos); return state_word->compare_exchange_strong( - *expected, MakeFreeWordForNextWrap(position), std::memory_order_release, - std::memory_order_relaxed); + *expected, MakeFreeWordForNextWrap(chunk_pos)); } -bool SharedRingBuffer::TryReleaseCompleteChunkAsFree(uint32_t position, +bool SharedRingBuffer::TryReleaseCompleteChunkAsFree(uint32_t chunk_pos, uint32_t* expected) { PERFETTO_DCHECK(ChunkStateOf(*expected) == ChunkState::kComplete); // Complete(N) -> Free(next wrap). // The writer can reuse the chunk first. The reader then discards its copy and - // retries this position on a later pass. - // - // Memory order release orders the copy before the next writer acquires - // Free and overwrites the chunk. - // - // On failure, memory order relaxed is enough because the reader retries - // the position without using the copied payload. - std::atomic<uint32_t>* state_word = chunk_state_word_for_position(position); + // retries this position on a later pass. A successful reclaim orders the + // reader's copy before the next writer can overwrite the chunk. + std::atomic<uint32_t>* state_word = chunk_state_word_for_position(chunk_pos); return state_word->compare_exchange_strong( - *expected, MakeFreeWordForNextWrap(position), std::memory_order_release, - std::memory_order_relaxed); + *expected, MakeFreeWordForNextWrap(chunk_pos)); } bool SharedRingBuffer::TryReleaseRewriteAcknowledgedChunkAsFree( - uint32_t position, + uint32_t chunk_pos, uint32_t* observed) { // RewriteAcknowledged -> Free(next wrap). // Only this reader changes an acknowledged chunk. The expected value is the @@ -330,18 +270,10 @@ // // Failure is therefore a protocol error. The caller gets the unexpected // word in |*observed|. - // - // Memory order acq_rel serves both handoffs: - // - acquire pairs with the writer's acknowledgement after its suffix copy. - // - release hands the chunk to the next writer after that copy. - // - // On failure, memory order relaxed is enough because the unexpected - // word is used only to report the protocol error. uint32_t expected = kRewriteAcknowledgedStateWord; - std::atomic<uint32_t>* state_word = chunk_state_word_for_position(position); + std::atomic<uint32_t>* state_word = chunk_state_word_for_position(chunk_pos); const bool reclaimed = state_word->compare_exchange_strong( - expected, MakeFreeWordForNextWrap(position), std::memory_order_acq_rel, - std::memory_order_relaxed); + expected, MakeFreeWordForNextWrap(chunk_pos)); if (!reclaimed) *observed = expected; return reclaimed; @@ -365,33 +297,22 @@ #else RingBufferHeader* ring_header = header(); - // To avoid a missed wake: - // - the writer counts itself as waiting before checking read_pos. - // - the reader publishes read_pos before checking the waiter count. - // - // Without the two fences, both checks could see the old value. The writer - // would then sleep after the reader skipped the wake: + // All four operations below are sequentially consistent: // // writer reader // ------ ------ // num_writers_waiting += 1 publish read_pos - // memory order seq_cst fence memory order seq_cst fence // load read_pos load num_writers_waiting // FUTEX_WAIT if unchanged FUTEX_WAKE if nonzero // - // Both fences use memory order seq_cst. Whichever comes first decides - // the outcome: - // - writer first: the reader sees the waiter and wakes it. - // - reader first: the writer sees the new read_pos and does not sleep. - // - // num_writers_waiting is only a wake hint, so its accesses use - // memory order relaxed. - ring_header->num_writers_waiting.fetch_add(1, std::memory_order_relaxed); - std::atomic_thread_fence(std::memory_order_seq_cst); + // Both loads cannot see the old values in that total order. Either the + // writer sees the new read_pos, or the reader sees the registered waiter. + // FUTEX_WAIT checks read_pos again before sleeping, covering a publication + // between the writer's load and the syscall. + ring_header->num_writers_waiting.fetch_add(1); WriterWaitResult result = WriterWaitResult::kRetry; - const uint32_t read_pos = - ReadPosOf(ring_header->rw_positions.load(std::memory_order_relaxed)); + const uint32_t read_pos = ReadPosOf(ring_header->rw_positions.load()); if (read_pos == read_pos_for_wait) { struct timespec timeout{}; timeout.tv_sec = static_cast<time_t>(timeout_ms / 1000); @@ -410,16 +331,13 @@ } } - ring_header->num_writers_waiting.fetch_sub(1, std::memory_order_relaxed); + ring_header->num_writers_waiting.fetch_sub(1); return result; #endif // PERFETTO_TRACING_V2_HAS_FUTEX() } void SharedRingBuffer::PublishReadPos(uint32_t read_pos) { - // This load only supplies the initial expected value for the CAS below, so - // memory order relaxed is enough. - PublishReadPosFromSnapshot( - header()->rw_positions.load(std::memory_order_relaxed), read_pos); + PublishReadPosFromSnapshot(header()->rw_positions.load(), read_pos); } void SharedRingBuffer::PublishReadPosFromSnapshot(uint64_t rw_positions, @@ -431,23 +349,13 @@ // // Failure reloads both halves. The retry keeps the new write_pos while // replacing read_pos again. - // - // Memory order release publishes this pass's Free words to writers that - // see the new read_pos. - // - // On failure, memory order relaxed is enough because the returned value - // only supplies the next CAS attempt. while (!ring_header->rw_positions.compare_exchange_weak( - rw_positions, ReplaceReadPos(rw_positions, read_pos), - std::memory_order_release, std::memory_order_relaxed)) { + rw_positions, ReplaceReadPos(rw_positions, read_pos))) { } #if PERFETTO_TRACING_V2_HAS_FUTEX() - // See the missed-wake schedule in WaitForReadPosChange(). The - // memory order seq_cst fence must precede the waiter-count load. - std::atomic_thread_fence(std::memory_order_seq_cst); - - if (ring_header->num_writers_waiting.load(std::memory_order_relaxed) == 0) + // Paired with waiter registration in WaitForReadPosChange(). + if (ring_header->num_writers_waiting.load() == 0) return; // Wake everyone because one drain pass can free many chunks. Waking one
diff --git a/src/tracing/v2/shared_ring_buffer.h b/src/tracing/v2/shared_ring_buffer.h index 5b1bd63..7d04a40 100644 --- a/src/tracing/v2/shared_ring_buffer.h +++ b/src/tracing/v2/shared_ring_buffer.h
@@ -31,20 +31,33 @@ class SharedRingBufferInternalsForTest; } -// A non-owning view over one ring region, plus the atomic transitions on it. -// Like SharedMemoryABI in v1, this class only interprets memory that somebody -// else owns: the ring's owner supplies the region and keeps it mapped for as -// long as this view, its reader and its writers are around. +// Provides the atomic operations on a shared ring buffer. // -// 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. +// Like SharedMemoryABI in v1, this class only interprets memory that somebody +// else owns: the ring buffer's owner supplies the region and keeps it mapped +// for as long as this view, its reader and its writers are around. +// +// The ring buffer is lock-free, with multiple writers and a single reader: +// - Writers reserve positions concurrently. When the ring buffer is full, +// each writer applies its BufferExhaustedPolicy and records any data it +// drops. +// - The reader consumes positions in order and is the only actor that makes a +// chunk Free again. +// - All updates to rw_positions and the chunk state words go through this +// class. class SharedRingBuffer { public: - // Attaches to a ring at |start| without modifying it. |size| covers the - // header followed by a power-of-two number of |chunk_size|-byte chunks. - // A newly created ring must be zero-filled. + // Attaches to a ring buffer at |start|. Its layout is: + // + // size = sizeof(RingBufferHeader) + num_chunks * chunk_size + // + // - num_chunks must be a nonzero power of two, at most 2^30. + // - |chunk_size| must be at least 256 and a multiple of four. It does not + // need to be a power of two. + // - |size| must match the equation exactly, with no trailing bytes. It does + // not need to be a power of two. + // + // A newly created ring buffer must be zero-filled. SharedRingBuffer(uint8_t* start, size_t size, uint32_t chunk_size); ~SharedRingBuffer() = default; @@ -53,27 +66,27 @@ SharedRingBuffer(SharedRingBuffer&&) = delete; SharedRingBuffer& operator=(SharedRingBuffer&&) = delete; - // Immutable for the life of the ring. + // Immutable for the life of the ring buffer. uint32_t num_chunks() const { return num_chunks_; } uint32_t chunk_size() const { return chunk_size_; } - uint8_t* chunk_at(uint32_t chunk_idx) { - PERFETTO_DCHECK(chunk_idx < num_chunks_); + uint8_t* chunk_at(ChunkIndex chunk_idx) { + PERFETTO_DCHECK(chunk_idx.value() < num_chunks_); return start_ + sizeof(RingBufferHeader) + - static_cast<size_t>(chunk_idx) * chunk_size_; + static_cast<size_t>(chunk_idx.value()) * chunk_size_; } - const uint8_t* chunk_at(uint32_t chunk_idx) const { - PERFETTO_DCHECK(chunk_idx < num_chunks_); + const uint8_t* chunk_at(ChunkIndex chunk_idx) const { + PERFETTO_DCHECK(chunk_idx.value() < num_chunks_); return start_ + sizeof(RingBufferHeader) + - static_cast<size_t>(chunk_idx) * chunk_size_; + static_cast<size_t>(chunk_idx.value()) * 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 + // The writer then changes Free(wrap_count(chunk_pos)) to BeingWritten. If it + // is descheduled between the two, the reader consumes 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. @@ -89,78 +102,79 @@ 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; + uint32_t write_pos = 0; // The read_pos sampled by the last reservation attempt. If the writer has // to wait, the futex sleeps only while read_pos still has this value. uint32_t read_pos_for_wait = 0; }; - // Reserves the next position if the ring has room, retrying a lost CAS. + // Reserves the next position if the ring buffer has room, retrying a lost + // CAS. Reservation TryReserveWritePos(); // Writer-side chunk transitions. - // Free(wrap_count(position)) -> BeingWritten. |being_written_word| must be a + // Free(wrap_count(chunk_pos)) -> BeingWritten. |being_written_word| must be a // 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, + bool TryAcquireChunkForWriting(uint32_t chunk_pos, uint32_t being_written_word); // BeingWritten -> Complete. |*expected| is the last BeingWritten word. // On failure it 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 TryReleaseChunkAsComplete(uint32_t chunk_idx, + bool TryReleaseChunkAsComplete(ChunkIndex chunk_idx, uint32_t complete_word, uint32_t* expected); // 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_idx, uint32_t observed); + bool TryReacquireChunkForWriting(ChunkIndex chunk_idx, uint32_t observed); // RewriteRequested -> RewriteAcknowledged after the writer has stopped // touching the old chunk. Failure is a protocol error. - bool TryAcknowledgeRewrite(uint32_t chunk_idx, uint32_t observed); + bool TryAcknowledgeRewrite(ChunkIndex chunk_idx, 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_idx) const; + uint32_t LoadChunkStateWord(ChunkIndex chunk_idx) const; - // Loads write_pos once. A writer may reserve the next position concurrently. - // The reader will see it on its next pass. + // Returns the next reservation position to bound a drain or control barrier. + // Chunk state determines which reserved positions have published payload. uint32_t LoadWritePos() const; // BeingWritten -> RewriteRequested, passing format, flags, num_fragments and // the WriterID through untouched. |*expected| is the last BeingWritten word; // on failure it receives the word that won the race. - bool TryRequestRewrite(uint32_t chunk_idx, uint32_t* expected); + bool TryRequestRewrite(ChunkIndex chunk_idx, uint32_t* expected); // 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. + // next pass around the ring buffer. The new wrap count comes from + // |chunk_pos|, not from the old state word. - // Free(wrap_count(position)) -> Free(next_wrap(position)). This consumes a + // Free(wrap_count(chunk_pos)) -> Free(next_wrap(chunk_pos)). This consumes a // position whose writer never entered BeingWritten and prepares the chunk - // for position + num_chunks. |*expected| is the Free word for this position; + // for chunk_pos + num_chunks. |*expected| is the Free word for this position; // on failure it receives the current word. - bool TryMoveFreeChunkToNextWrap(uint32_t position, uint32_t* expected); + bool TryMoveFreeChunkToNextWrap(uint32_t chunk_pos, uint32_t* expected); - // Complete -> Free(next_wrap(position)). |*expected| is the last Complete + // Complete -> Free(next_wrap(chunk_pos)). |*expected| is the last Complete // word; on failure it receives the current word. - bool TryReleaseCompleteChunkAsFree(uint32_t position, uint32_t* expected); + bool TryReleaseCompleteChunkAsFree(uint32_t chunk_pos, uint32_t* expected); - // RewriteAcknowledged -> Free(next_wrap(position)). Failure is a protocol + // RewriteAcknowledged -> Free(next_wrap(chunk_pos)). Failure is a protocol // error and updates |*observed| with the unexpected word. - bool TryReleaseRewriteAcknowledgedChunkAsFree(uint32_t position, + bool TryReleaseRewriteAcknowledgedChunkAsFree(uint32_t chunk_pos, uint32_t* observed); - // Backpressure: the writer's full-ring path. + // Backpressure when the ring buffer is full. // // 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 @@ -200,22 +214,23 @@ // Shared-memory address and wrap-count helpers. - std::atomic<uint32_t>* chunk_state_word_at(uint32_t chunk_idx) { + std::atomic<uint32_t>* chunk_state_word_at(ChunkIndex chunk_idx) { return reinterpret_cast<std::atomic<uint32_t>*>(chunk_at(chunk_idx)); } - const std::atomic<uint32_t>* chunk_state_word_at(uint32_t chunk_idx) const { + const std::atomic<uint32_t>* chunk_state_word_at(ChunkIndex chunk_idx) const { return reinterpret_cast<const std::atomic<uint32_t>*>(chunk_at(chunk_idx)); } - std::atomic<uint32_t>* chunk_state_word_for_position(uint32_t position) { - return chunk_state_word_at(ChunkIndexOf(position, num_chunks_)); + std::atomic<uint32_t>* chunk_state_word_for_position(uint32_t chunk_pos) { + return chunk_state_word_at( + ChunkIndex::FromPosition(chunk_pos, num_chunks_)); } // Returns the Free word for the next position that uses the same chunk. - uint32_t MakeFreeWordForNextWrap(uint32_t position) const { + uint32_t MakeFreeWordForNextWrap(uint32_t chunk_pos) const { // Deriving the value from the next position also handles uint32_t rollover. - const uint32_t next_position = position + num_chunks_; - return MakeFreeStateWordForPosition(next_position, num_chunks_); + const uint32_t next_pos = chunk_pos + num_chunks_; + return MakeFreeStateWordForPosition(next_pos, num_chunks_); } RingBufferHeader* header() {
diff --git a/src/tracing/v2/shared_ring_buffer_abi.h b/src/tracing/v2/shared_ring_buffer_abi.h index c0bee96..fbf159b 100644 --- a/src/tracing/v2/shared_ring_buffer_abi.h +++ b/src/tracing/v2/shared_ring_buffer_abi.h
@@ -19,8 +19,11 @@ #include <stddef.h> #include <stdint.h> +#include <string.h> +#include <algorithm> #include <atomic> +#include <optional> #include "perfetto/base/logging.h" #include "perfetto/ext/base/bits.h" @@ -30,18 +33,19 @@ namespace perfetto::tracing_v2 { -// Shared-memory ABI for a tracing-v2 producer ring. +// Shared-memory ABI for a tracing-v2 producer ring buffer. // Protocol and alternatives: RFC 0046, // https://github.com/google/perfetto/discussions/7120. // Parent design: RFC 0014, https://github.com/google/perfetto/discussions/4508. // -// Several trace writers can write to the ring at once. One reader drains their -// data in reservation order. A relocated suffix obtains a later reservation. +// Several trace writers can write to the ring buffer at once. One reader drains +// their data in reservation order. A relocated suffix obtains a later +// reservation. // -// 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 ring buffer 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. @@ -60,8 +64,8 @@ // Contiguous chunks must keep every 32-bit state word aligned. constexpr uint32_t kChunkAlignmentBytes = 4; -// Ring header -// ----------- +// Ring buffer header +// ------------------ // // byte offset // 0 4 8 12 64 @@ -72,19 +76,18 @@ // \_________ rw_positions _____/ \_ atomic32 _/ // atomic<uint64_t> // -// 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. +// The first four bytes of rw_positions contain read_pos. The following four +// bytes contain write_pos. Writers load both positions from the same atomic, +// so a capacity check cannot combine values read at different times. // // The reader is the only one that moves read_pos. It publishes a new value once -// per drain pass and then wakes any writer parked on a full ring. read_pos is -// also the first four bytes of rw_positions, which is the address the futex -// waits on. +// per drain pass and then wakes any writer parked on a full ring buffer. +// read_pos is also the first four bytes of rw_positions, which is the address +// the futex waits on. // // num_writers_waiting lets the reader skip a futex wake when nobody is waiting -// for space. It is only an optimization and never decides whether the ring is -// full or who owns a chunk. +// for space. It is only an optimization and never decides whether the ring +// buffer is full or who owns a chunk. // // Bytes 12..63 pad the header to one cache line. // @@ -112,8 +115,12 @@ // ------------------------------------ // // 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. +// position the reader must consume. Both counters are uint32_t and wrap at +// UINT32_MAX. +// +// This is different from traversing the ring buffer: each position +// is mapped to a physical chunk by ChunkIndex::FromPosition() when that chunk +// is accessed. // // The number of reserved positions not yet handled by the reader is: // @@ -128,7 +135,7 @@ // 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. +// positions do not describe a valid ring buffer state. // // This also means that at most one outstanding reservation maps to each // physical chunk (the one exception is the wrap-count alias described at @@ -163,45 +170,77 @@ return write_pos - read_pos; } -// num_chunks = 2^k, so the low k bits of a position select the physical chunk -// and the remaining bits count completed traversals: -// -// 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. -constexpr uint32_t ChunkIndexOf(uint32_t position, uint32_t num_chunks) { - return position & (num_chunks - 1); -} +// A physical chunk index, distinct from the logical reservation position. +// Variables use _idx for physical chunk indexes and _pos for logical positions. +// - _idx identifies a physical chunk in the shared memory mapping, in the range +// [0, num_chunks). Successive logical positions cycle through these chunks, +// returning to chunk index 0 after chunk index num_chunks - 1. +// - _pos identifies a logical position in the reservation order. It keeps +// advancing past num_chunks across successive traversals of the ring buffer; +// as a uint32_t, it wraps from UINT32_MAX to 0. FromPosition() maps it to the +// physical chunk index used for memory access. +class ChunkIndex { + public: + // Builds from an already computed physical chunk index. + static constexpr ChunkIndex FromIndex(uint32_t chunk_idx) { + return ChunkIndex{chunk_idx}; + } + + // num_chunks = 2^k, so the low k bits of a position select the physical chunk + // and the remaining bits count completed traversals: + // + // bits 31..k bits k-1..0 + // +----------------------------+----------------------------+ + // | traversal number | physical chunk index | + // +----------------------------+----------------------------+ + // 32 - k bits k bits + // + // For example, an eight-chunk ring buffer has three chunk-index bits. The low + // three bits select chunks 0 through 7. The remaining 29 bits count completed + // traversals. + static constexpr ChunkIndex FromPosition(uint32_t chunk_pos, + uint32_t num_chunks) { + return ChunkIndex{chunk_pos & (num_chunks - 1)}; + } + + constexpr uint32_t value() const { return chunk_idx_; } + + private: + explicit constexpr ChunkIndex(uint32_t chunk_idx) : chunk_idx_(chunk_idx) {} + + uint32_t chunk_idx_; +}; // Chunk state word // ---------------- // +// There are five defined states: Free, BeingWritten, Complete, +// RewriteRequested and RewriteAcknowledged. The low control byte contains the +// state. It also determines how to interpret the other three bytes: Free uses +// them differently from the three data-bearing states. +// // A Free word contains: // -// 31 16 15 8 7 0 -// +----------------------------------+-----------+------------------+ -// | wrap_count | num_frag- | control = 0 | -// | | ments = 0 | (Free, format 0, | -// | | | no flags) | -// +----------------------------------+-----------+------------------+ -// 16 bits 8 bits 8 bits +// +-------------------+-------------------+---------------------------+ +// | byte 0 | byte 1 | bytes 2-3 | +// +-------------------+-------------------+---------------------------+ +// | control = 0 | num_fragments = 0 | wrap_count | +// | (Free, format 0, | | | +// | no flags) | | | +// +-------------------+-------------------+---------------------------+ +// 8 bits 8 bits 16 bits // -// A Free word is wrap_count << 16, making a zero-filled ring valid and empty. +// A Free word is wrap_count << 16, making a zero-filled ring buffer valid and +// empty. // // BeingWritten, Complete and RewriteRequested contain: // -// 31 16 15 8 7 0 -// +----------------------------------+-----------+------------------+ -// | WriterID | num | control byte | -// | | fragments | | -// +----------------------------------+-----------+------------------+ -// 16 bits 8 bits 8 bits +// +-------------------+-------------------+---------------------------+ +// | byte 0 | byte 1 | bytes 2-3 | +// +-------------------+-------------------+---------------------------+ +// | control byte | num_fragments | WriterID | +// +-------------------+-------------------+---------------------------+ +// 8 bits 8 bits 16 bits // // The control byte is: // @@ -223,15 +262,18 @@ kBeingWritten = 1, // The writer has published num_fragments fragments and is no longer touching - // the chunk. It may take the chunk back before the reader reclaims it. + // the chunk. It may take the chunk back by changing it to BeingWritten + // 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 + // While the writer was appending fragment N + 1, the reader consumed the + // first N fragments. The writer must move fragment N + 1 before releasing // this chunk. kRewriteRequested = 3, - // The writer has finished with the old chunk. The reader may reclaim it. + // After noticing RewriteRequested, the writer moves its unfinished fragment + // and changes the old chunk to RewriteAcknowledged. The reader changes it to + // Free when it encounters that chunk on a later traversal. kRewriteAcknowledged = 4, // A reader that does not know a state cannot tell who owns the chunk. It @@ -325,14 +367,14 @@ // fields. The winning CAS settles ownership; no second atomic is needed. // If scraping wins, the writer relocates only the suffix after N fragments. // -// A reservation allows one claim against Free(wrap_count(position)). On +// A reservation allows one claim against Free(wrap_count(chunk_pos)). On // failure, the writer must reserve a new position, never retry the new word. // // Actor From Action To // ------ --------------------- ----------------------- ------------------- // writer Free(wrap) claim BeingWritten(0) // writer Free(wrap) gone hole, reserve later unchanged -// reader Free(wrap) resolve unclaimed Free(next wrap) +// reader Free(wrap) consume unclaimed Free(next wrap) // reader Free(other wrap) protocol error, stop unchanged // writer BeingWritten(N) publish Complete(M) // reader BeingWritten(N) take published prefix RewriteRequested(N) @@ -356,7 +398,7 @@ // // Free stores the low 16 bits of the traversal number: // -// wrap_count = uint16_t(position / num_chunks) +// wrap_count = uint16_t(chunk_pos / num_chunks) // // For the same chunk, that value repeats after: // @@ -368,9 +410,9 @@ // the delayed writer. This is the limit of the 16-bit wrap count. // // Computes the wrap from a position; it does not inspect the chunk's state. -inline uint16_t WrapCountForPosition(uint32_t position, uint32_t num_chunks) { +inline uint16_t WrapCountForPosition(uint32_t chunk_pos, uint32_t num_chunks) { PERFETTO_DCHECK(base::IsPowerOfTwo(num_chunks)); - return static_cast<uint16_t>(position >> base::CountTrailZeros(num_chunks)); + return static_cast<uint16_t>(chunk_pos >> base::CountTrailZeros(num_chunks)); } constexpr ChunkState ChunkStateOf(uint32_t state_word) { @@ -387,11 +429,11 @@ return static_cast<uint32_t>(wrap_count) << kWrapCountShift; } -// The Free word a reservation at |position| must find, and the word the -// reader leaves for the next traversal when called with position + num_chunks. -inline uint32_t MakeFreeStateWordForPosition(uint32_t position, +// The Free word a reservation at |chunk_pos| must find, and the word the +// reader leaves for the next traversal when called with chunk_pos + num_chunks. +inline uint32_t MakeFreeStateWordForPosition(uint32_t chunk_pos, uint32_t num_chunks) { - return MakeFreeStateWord(WrapCountForPosition(position, num_chunks)); + return MakeFreeStateWord(WrapCountForPosition(chunk_pos, num_chunks)); } // These accessors apply to BeingWritten, Complete and RewriteRequested. @@ -445,8 +487,8 @@ // Target-buffer chunk format // -------------------------- // -// A data-bearing format-0 chunk begins with this six-byte header, leaving -// 250 bytes for payload and size entries in a minimum-sized chunk: +// A data-bearing format-0 chunk begins with this six-byte header. The remaining +// bytes hold payload and size entries: // // +---------+---------+-------------------+-------------------+ // | byte 0 | byte 1 | bytes 2-3 | bytes 4-5 | @@ -454,7 +496,7 @@ // | control | num | WriterID | target BufferID | // | byte |fragments| | | // +---------+---------+-------------------+-------------------+ -// \_____________ atomic state word _______/ +// \__________ atomic state word __________/ // // The rest of a format-0 chunk is laid out as follows: // @@ -468,40 +510,40 @@ // // Fragment 0's size is stored at the end of the chunk. num_fragments 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 size before -// copying the payload. The first publication also makes BufferID visible; +// those bytes before transitioning out of BeingWritten. The reader loads the +// state word, then decodes and checks every size before copying the payload. +// The first publication also makes BufferID visible; // the reader must not load BufferID when num_fragments is zero. -constexpr uint32_t kTargetBufferIDOffset = 4; +constexpr uint32_t kTargetBufferIdOffset = 4; constexpr uint32_t kTargetBufferPayloadOffset = 6; inline void StoreTargetBufferID(uint8_t* chunk, BufferID buffer_id) { - chunk[kTargetBufferIDOffset] = static_cast<uint8_t>(buffer_id); - chunk[kTargetBufferIDOffset + 1] = static_cast<uint8_t>(buffer_id >> 8); + memcpy(&chunk[kTargetBufferIdOffset], &buffer_id, sizeof(buffer_id)); } 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)); + BufferID buffer_id; + memcpy(&buffer_id, &chunk[kTargetBufferIdOffset], sizeof(buffer_id)); + return buffer_id; } // Fragment size directory // ----------------------- // -// The fragment sizes are varints at the end of the chunk. The first fragment's -// varint ends at chunk_size. Each later varint is prepended below the previous -// one: +// The fragment sizes are reverse-encoded varints at the end of the chunk. The +// first fragment's varint ends at chunk_size. Each later varint is prepended +// below the previous one: // // low address high address // +----------+----------+----------+----------+----------+ // | size N-1 | ... | size 2 | size 1 | size 0 | // +----------+----------+----------+----------+----------+ // -// The reader walks the sizes from high addresses to low addresses. It sees the -// bytes of each size in normal protobuf varint order and stops at that -// varint's final byte. It never has to inspect the next, unpublished entry. +// The reader starts at the end of the chunk and walks towards lower addresses. +// WriteFragmentSizeReversed() mirrors each varint's bytes so the reader sees +// its least-significant group first and stops at its final byte. It never has +// to inspect the next, unpublished entry. // Each varint byte carries seven value bits. Its top bit is set when another // byte follows. @@ -509,53 +551,57 @@ constexpr uint8_t kVarIntContinuationBit = 1u << kVarIntDataBitsPerByte; constexpr uint8_t kVarIntDataBitsMask = kVarIntContinuationBit - 1; -// A uint32_t fragment size needs at most ceil(32 / 7) = 5 varint bytes. -constexpr uint32_t kMaxFragmentSizeVarIntBytes = 5; +// Fragment sizes share Protozero's message-length limit: four varint bytes. +constexpr uint32_t kMaxFragmentSizeVarIntBytes = 4; -constexpr uint32_t FragmentSizeVarIntByteCount(uint32_t fragment_size) { - uint32_t bytes = 1; - while (fragment_size >= kVarIntContinuationBit) { - fragment_size >>= kVarIntDataBitsPerByte; - ++bytes; - } - return bytes; +inline uint32_t FragmentSizeVarIntByteCount(uint32_t fragment_size) { + PERFETTO_DCHECK(fragment_size <= protozero::proto_utils::kMaxMessageLength); + if (fragment_size < (1u << 7)) + return 1; + if (fragment_size < (1u << 14)) + return 2; + if (fragment_size < (1u << 21)) + return 3; + return kMaxFragmentSizeVarIntBytes; } -static_assert(FragmentSizeVarIntByteCount(UINT32_MAX) == - kMaxFragmentSizeVarIntBytes, - "kMaxFragmentSizeVarIntBytes must bound every uint32_t size"); - -// Returns the largest n such that n + varint_size(n) <= available_bytes. The -// loop runs at most four times because a uint32_t varint is at most five bytes. -constexpr uint32_t MaxFragmentSizeForAvailableBytes(uint32_t available_bytes) { +// Returns the largest supported n such that n + varint_size(n) <= +// available_bytes, or zero if no payload byte fits. +inline uint32_t MaxFragmentSizeForAvailableBytes(uint32_t available_bytes) { if (available_bytes <= 1) return 0; - // Leave at least one byte for the size varint. - uint32_t fragment_size = available_bytes - 1; - while (FragmentSizeVarIntByteCount(fragment_size) > - available_bytes - fragment_size) { - --fragment_size; - } - return fragment_size; + // At each threshold the directory needs one more byte. Until both that + // byte and the larger payload fit, keep the previous maximum payload. + if (available_bytes <= (1u << 7)) + return available_bytes - 1; + if (available_bytes <= (1u << 14) + 1) + return available_bytes - 2; + if (available_bytes <= (1u << 21) + 2) + return available_bytes - 3; + return std::min( + available_bytes - 4, + static_cast<uint32_t>(protozero::proto_utils::kMaxMessageLength)); } -constexpr uint32_t MaxFragmentSizeForEmptyChunk(uint32_t chunk_size) { - if (chunk_size < kMinChunkSize) - return 0; +inline uint32_t MaxFragmentSizeForEmptyChunk(uint32_t chunk_size) { + PERFETTO_CHECK(chunk_size >= kMinChunkSize); const uint32_t available_bytes = chunk_size - kTargetBufferPayloadOffset; return MaxFragmentSizeForAvailableBytes(available_bytes); } // Writes the size into the directory immediately before |sizes_begin| and // returns the new directory start. The caller must provide -// FragmentSizeVarIntByteCount(size) bytes before |sizes_begin|. -inline uint8_t* WriteFragmentSize(uint8_t* sizes_begin, uint32_t size) { +// FragmentSizeVarIntByteCount(size) bytes before |sizes_begin|, and |size| +// must be at most protozero::proto_utils::kMaxMessageLength. +inline uint8_t* WriteFragmentSizeReversed(uint8_t* sizes_begin, uint32_t size) { + PERFETTO_DCHECK(size <= protozero::proto_utils::kMaxMessageLength); uint8_t encoded[kMaxFragmentSizeVarIntBytes]; const uint8_t* const encoded_end = protozero::proto_utils::WriteVarInt(size, encoded); const size_t encoded_size = static_cast<size_t>(encoded_end - encoded); // Put the first varint byte at the highest address: the reader starts - // there and reads towards lower addresses (see ReadFragmentSize()). + // there and reads towards lower addresses (see ReadFragmentSizeReversed()). + // TODO(sashwinbalaji): Move this into proto_utils as WriteVarIntReversed(). for (size_t i = 0; i < encoded_size; ++i) { --sizes_begin; *sizes_begin = encoded[i]; @@ -566,41 +612,37 @@ // Decodes the next size varint while moving |*sizes_cursor| towards lower // addresses. // -// WriteFragmentSize() stores each varint reversed, so a reader walking down -// the chunk sees the bytes in normal varint order. For example, a size of 300 -// is the varint AC 02 and is stored as: +// WriteFragmentSizeReversed() mirrors each varint, so a reader walking down +// the chunk sees its bytes in protobuf decoding order. For example, a size of +// 300 is the varint AC 02 and is stored as: // // ... | 02 | AC | <- chunk_size // ^ ^ // | first byte read: AC, continuation bit set // second byte read: 02, no continuation bit, stop // -// |lower_bound| is the lowest address a size byte may be read from, normally -// the start of the payload. It only keeps the decoder inside the chunk. -// Whether the payloads and the size bytes overlap is the caller's check, made -// once every size is decoded. +// |lower_bound| is the lowest address a size byte may be read from. The caller +// uses the start of the payload because its end is known only after decoding +// all the sizes. It checks for overlap with the payload afterwards. // -// Rejected, so that every size has exactly one byte pattern: -// - a varint that runs into |lower_bound| or is longer than five bytes; -// - a value above uint32_t; -// - a non-shortest encoding (81 00 for 1). +// Rejects truncated varints and encodings longer than four bytes. Non-minimal +// encodings are accepted; the writer emits the shortest representation. // // On success, |*sizes_cursor| points at the last byte read, which is the -// exclusive upper bound for the next size. |*fragment_size| is also updated -// only on success. -inline bool ReadFragmentSize(const uint8_t* lower_bound, - const uint8_t** sizes_cursor, - uint32_t* fragment_size) { +// exclusive upper bound for the next size. Failure leaves the cursor unchanged. +inline std::optional<uint32_t> ReadFragmentSizeReversed( + const uint8_t* lower_bound, + const uint8_t** sizes_cursor) { const uint8_t* cursor = *sizes_cursor; - uint64_t value = 0; + uint32_t value = 0; uint32_t num_bytes = 0; for (;;) { if (cursor == lower_bound || num_bytes == kMaxFragmentSizeVarIntBytes) - return false; + return std::nullopt; --cursor; const uint8_t byte = *cursor; // Reading down the directory yields the least significant group first. - const uint64_t data_bits = byte & kVarIntDataBitsMask; + const uint32_t data_bits = byte & kVarIntDataBitsMask; const uint32_t shift = kVarIntDataBitsPerByte * num_bytes; value |= data_bits << shift; ++num_bytes; @@ -608,16 +650,8 @@ break; } - // Five bytes can carry 35 value bits, hence the range check. - if (value > UINT32_MAX) - return false; - // Reject non-shortest encodings so that every size has one byte pattern. - if (FragmentSizeVarIntByteCount(static_cast<uint32_t>(value)) != num_bytes) - return false; - *sizes_cursor = cursor; - *fragment_size = static_cast<uint32_t>(value); - return true; + return value; } } // namespace perfetto::tracing_v2
diff --git a/src/tracing/v2/shared_ring_buffer_abi_unittest.cc b/src/tracing/v2/shared_ring_buffer_abi_unittest.cc index 0f29662..4c52893 100644 --- a/src/tracing/v2/shared_ring_buffer_abi_unittest.cc +++ b/src/tracing/v2/shared_ring_buffer_abi_unittest.cc
@@ -18,6 +18,7 @@ #include <stdint.h> +#include <type_traits> #include <vector> #include "src/tracing/v2/shared_ring_buffer_test_utils.h" @@ -28,6 +29,16 @@ using test::WrapCountOf; +static_assert(!std::is_convertible_v<uint32_t, ChunkIndex>); +static_assert(!std::is_constructible_v<ChunkIndex, uint32_t>); +static_assert(!std::is_convertible_v<ChunkIndex, uint32_t>); +static_assert( + std::is_same_v<decltype(ChunkIndex::FromPosition(0, 4)), ChunkIndex>); +static_assert( + !std::is_invocable_v<decltype(&SharedRingBuffer::LoadChunkStateWord), + SharedRingBuffer*, + uint32_t>); + // Chunk state word // ---------------- @@ -197,20 +208,25 @@ TEST(SharedRingBufferABITest, ChunkIndexAndWrapCount) { // A worked example. const uint32_t kNumChunks = 4; - const uint32_t kExpectedIndex[] = {0, 1, 2, 3, 0, 1, 2, 3, 0, 1, 2, 3, 0}; + const uint32_t kExpectedChunkIdx[] = {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(ChunkIndexOf(p, kNumChunks), kExpectedIndex[p]) << p; - EXPECT_EQ(WrapCountForPosition(p, kNumChunks), kExpectedWrap[p]) << p; + for (uint32_t chunk_pos = 0; chunk_pos < 13; ++chunk_pos) { + EXPECT_EQ(ChunkIndex::FromPosition(chunk_pos, kNumChunks).value(), + kExpectedChunkIdx[chunk_pos]) + << chunk_pos; + EXPECT_EQ(WrapCountForPosition(chunk_pos, kNumChunks), + kExpectedWrap[chunk_pos]) + << chunk_pos; } - // Including the one-chunk ring, where every position maps to chunk 0 and the - // wrap count is the position itself, truncated. + // Including the one-chunk ring buffer, 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}) { - for (uint32_t p = 0; p < 3 * num_chunks + 3; ++p) { - EXPECT_EQ(ChunkIndexOf(p, num_chunks), p % num_chunks); - EXPECT_EQ(WrapCountForPosition(p, num_chunks), - (p / num_chunks) & 0xffffu); + for (uint32_t chunk_pos = 0; chunk_pos < 3 * num_chunks + 3; ++chunk_pos) { + EXPECT_EQ(ChunkIndex::FromPosition(chunk_pos, num_chunks).value(), + chunk_pos % num_chunks); + EXPECT_EQ(WrapCountForPosition(chunk_pos, num_chunks), + (chunk_pos / num_chunks) & 0xffffu); } } } @@ -223,29 +239,31 @@ 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(ChunkIndexOf(p, kNumChunks), kExpected[i]) << i; + const uint32_t kExpectedChunkIdx[] = {4, 5, 6, 7, 0, 1}; + uint32_t chunk_pos = 0xfffffffcu; + for (uint32_t i = 0; i < 6; ++i, ++chunk_pos) + EXPECT_EQ(ChunkIndex::FromPosition(chunk_pos, kNumChunks).value(), + kExpectedChunkIdx[i]) + << i; } TEST(SharedRingBufferABITest, NextWrapFromPosition) { // The next wrap comes from the next position, not from the chunk word. // Away from the rollovers next_wrap is simply "one more". const uint32_t kNumChunks = 4; - for (uint32_t p = 0; p < 16; ++p) { - EXPECT_EQ(WrapCountForPosition(p + kNumChunks, kNumChunks), - WrapCountForPosition(p, kNumChunks) + 1) - << p; + for (uint32_t chunk_pos = 0; chunk_pos < 16; ++chunk_pos) { + EXPECT_EQ(WrapCountForPosition(chunk_pos + kNumChunks, kNumChunks), + WrapCountForPosition(chunk_pos, kNumChunks) + 1) + << chunk_pos; } // 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, kNumChunks), 0xffffu); - EXPECT_EQ(WrapCountForPosition(kLastLapOfPeriod + kNumChunks, kNumChunks), + const uint32_t kLastLapOfPeriodPos = 0xffffu * kNumChunks; // wrap 0xffff + EXPECT_EQ(WrapCountForPosition(kLastLapOfPeriodPos, kNumChunks), 0xffffu); + EXPECT_EQ(WrapCountForPosition(kLastLapOfPeriodPos + kNumChunks, kNumChunks), 0u); // At the uint32_t position rollover the traversal number restarts from zero @@ -253,12 +271,12 @@ // 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 kLastPosition = 0u - kBigRing; // the last lap's chunk 0 - EXPECT_EQ(ChunkIndexOf(kLastPosition, kBigRing), 0u); - EXPECT_EQ(WrapCountForPosition(kLastPosition, kBigRing), 0xfffu); - EXPECT_EQ(WrapCountForPosition(kLastPosition + kBigRing, kBigRing), 0u); - EXPECT_NE(WrapCountForPosition(kLastPosition + kBigRing, kBigRing), - WrapCountForPosition(kLastPosition, kBigRing) + 1u); + const uint32_t kLastPos = 0u - kBigRing; // the last lap's chunk 0 + EXPECT_EQ(ChunkIndex::FromPosition(kLastPos, kBigRing).value(), 0u); + EXPECT_EQ(WrapCountForPosition(kLastPos, kBigRing), 0xfffu); + EXPECT_EQ(WrapCountForPosition(kLastPos + kBigRing, kBigRing), 0u); + EXPECT_NE(WrapCountForPosition(kLastPos + kBigRing, kBigRing), + WrapCountForPosition(kLastPos, kBigRing) + 1u); } // Pin the finite period of the 16-bit wrap count. @@ -270,11 +288,12 @@ const uint32_t period = num_chunks * 65536; EXPECT_EQ(WrapCountForPosition(0, num_chunks), WrapCountForPosition(period, num_chunks)); - EXPECT_EQ(ChunkIndexOf(0, num_chunks), ChunkIndexOf(period, num_chunks)); + EXPECT_EQ(ChunkIndex::FromPosition(0, num_chunks).value(), + ChunkIndex::FromPosition(period, num_chunks).value()); // 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, num_chunks), + const uint32_t chunk_pos = lap * num_chunks; + EXPECT_NE(WrapCountForPosition(chunk_pos, num_chunks), WrapCountForPosition(0, num_chunks)); } } @@ -302,8 +321,9 @@ EXPECT_EQ(FragmentSizeVarIntByteCount(0x1fffff), 3u); EXPECT_EQ(FragmentSizeVarIntByteCount(0x200000), 4u); EXPECT_EQ(FragmentSizeVarIntByteCount(0x0fffffff), 4u); - EXPECT_EQ(FragmentSizeVarIntByteCount(0x10000000), 5u); - EXPECT_EQ(FragmentSizeVarIntByteCount(UINT32_MAX), 5u); + EXPECT_EQ( + FragmentSizeVarIntByteCount(protozero::proto_utils::kMaxMessageLength), + 4u); } TEST(SharedRingBufferABITest, MaxFragmentSizeForAvailableBytes) { @@ -315,11 +335,47 @@ EXPECT_EQ(MaxFragmentSizeForAvailableBytes(129), 127u); EXPECT_EQ(MaxFragmentSizeForAvailableBytes(130), 128u); EXPECT_EQ(MaxFragmentSizeForAvailableBytes(250), 248u); - EXPECT_EQ(MaxFragmentSizeForAvailableBytes(UINT32_MAX), UINT32_MAX - 5u); + EXPECT_EQ(MaxFragmentSizeForAvailableBytes(UINT32_MAX), + protozero::proto_utils::kMaxMessageLength); +} + +TEST(SharedRingBufferABITest, UndersizedChunkIsInvalid) { + EXPECT_DEATH_IF_SUPPORTED(MaxFragmentSizeForEmptyChunk(kMinChunkSize - 1), + "PERFETTO_CHECK"); +} + +TEST(SharedRingBufferABITest, CapacityAtEveryVarIntThreshold) { + // At a threshold the larger payload needs an extra directory byte. Test + // the last smaller payload, the one-byte gap, and the first larger payload. + for (uint32_t bytes : {1u, 2u, 3u}) { + const uint32_t threshold = 1u << (7 * bytes); + EXPECT_EQ(MaxFragmentSizeForAvailableBytes(threshold + bytes - 2), + threshold - 2); + EXPECT_EQ(MaxFragmentSizeForAvailableBytes(threshold + bytes - 1), + threshold - 1); + EXPECT_EQ(MaxFragmentSizeForAvailableBytes(threshold + bytes), + threshold - 1); + EXPECT_EQ(MaxFragmentSizeForAvailableBytes(threshold + bytes + 1), + threshold); + EXPECT_EQ(MaxFragmentSizeForAvailableBytes(threshold + bytes + 2), + threshold + 1); + } + const uint32_t max_size = protozero::proto_utils::kMaxMessageLength; + EXPECT_EQ(MaxFragmentSizeForAvailableBytes(max_size + 3), max_size - 1); + EXPECT_EQ(MaxFragmentSizeForAvailableBytes(max_size + 4), max_size); + EXPECT_EQ(MaxFragmentSizeForAvailableBytes(max_size + 5), max_size); +} + +TEST(SharedRingBufferABITest, NonMinimalFragmentSizes) { + // In decreasing address order these encode 0 and 1 using extra bytes. + const uint8_t sizes[] = {0x00, 0x80, 0x81, 0x00, 0x80}; + const uint8_t* cursor = sizes + sizeof(sizes); + EXPECT_EQ(ReadFragmentSizeReversed(sizes, &cursor), 0u); + EXPECT_EQ(ReadFragmentSizeReversed(sizes, &cursor), 1u); + EXPECT_EQ(cursor, sizes); } TEST(SharedRingBufferABITest, 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); @@ -330,7 +386,7 @@ std::vector<uint8_t> chunk(256, 0); uint8_t* sizes_begin = chunk.data() + chunk.size(); for (uint32_t size : {5u, 200u, 3u}) - sizes_begin = WriteFragmentSize(sizes_begin, size); + sizes_begin = WriteFragmentSizeReversed(sizes_begin, size); // Fragment 0 is nearest the end. Reading towards lower addresses yields the // normal varint byte sequence c8 01 for 200. @@ -342,49 +398,43 @@ const uint8_t* cursor = chunk.data() + chunk.size(); for (uint32_t expected : {5u, 200u, 3u}) { - uint32_t actual = 0; - ASSERT_TRUE(ReadFragmentSize(sizes_begin, &cursor, &actual)); - EXPECT_EQ(actual, expected); + EXPECT_EQ(ReadFragmentSizeReversed(sizes_begin, &cursor), expected); } EXPECT_EQ(cursor, sizes_begin); } TEST(SharedRingBufferABITest, FragmentSizeRoundTrip) { - const uint32_t kSizes[] = {0, 1, 127, 128, - 16383, 16384, 0x1fffff, 0x200000, - 0x0fffffff, 0x10000000, UINT32_MAX}; + const uint32_t kSizes[] = { + 0, 1, 127, + 128, 16383, 16384, + 0x1fffff, 0x200000, protozero::proto_utils::kMaxMessageLength}; std::vector<uint8_t> sizes(64, 0xee); uint8_t* sizes_begin = sizes.data() + sizes.size(); for (uint32_t size : kSizes) - sizes_begin = WriteFragmentSize(sizes_begin, size); + sizes_begin = WriteFragmentSizeReversed(sizes_begin, size); const uint8_t* cursor = sizes.data() + sizes.size(); for (uint32_t expected : kSizes) { - uint32_t actual = 0; - ASSERT_TRUE(ReadFragmentSize(sizes_begin, &cursor, &actual)); - EXPECT_EQ(actual, expected); + EXPECT_EQ(ReadFragmentSizeReversed(sizes_begin, &cursor), expected); } EXPECT_EQ(cursor, sizes_begin); } TEST(SharedRingBufferABITest, MalformedFragmentSizes) { 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}; + const uint8_t kTooLong[] = {0x00, 0x80, 0x80, 0x80, 0x80}; + const uint8_t kAboveMessageLimit[] = {0x01, 0x80, 0x80, 0x80, 0x80}; auto expect_rejected = [](const uint8_t* begin, size_t size) { const uint8_t* cursor = begin + size; - uint32_t fragment_size = 0xdeadbeef; - EXPECT_FALSE(ReadFragmentSize(begin, &cursor, &fragment_size)); - // A rejection leaves both outputs untouched. + EXPECT_EQ(ReadFragmentSizeReversed(begin, &cursor), std::nullopt); + // A rejection leaves the cursor untouched. EXPECT_EQ(cursor, begin + size); - EXPECT_EQ(fragment_size, 0xdeadbeefu); }; expect_rejected(kUnterminated, sizeof(kUnterminated)); expect_rejected(kTooLong, sizeof(kTooLong)); - expect_rejected(kUint32Overflow, sizeof(kUint32Overflow)); - expect_rejected(kNonCanonical, sizeof(kNonCanonical)); + expect_rejected(kAboveMessageLimit, sizeof(kAboveMessageLimit)); + expect_rejected(kUnterminated, 0); } TEST(SharedRingBufferABITest, TargetBufferID) {
diff --git a/src/tracing/v2/shared_ring_buffer_concurrency_unittest.cc b/src/tracing/v2/shared_ring_buffer_concurrency_unittest.cc index b9ab830..8a16b7b 100644 --- a/src/tracing/v2/shared_ring_buffer_concurrency_unittest.cc +++ b/src/tracing/v2/shared_ring_buffer_concurrency_unittest.cc
@@ -54,13 +54,13 @@ uint32_t chunk_size; // An upper bound on the attempts each writer makes. uint32_t fragments_per_writer; - uint32_t seed_position; + uint32_t seed_pos; BufferExhaustedPolicy policy; - // If nonzero, the writers are stopped once the reader has resolved this - // many positions past seed_position. That makes the run end on reader + // If nonzero, the writers are stopped once the reader has consumed this + // many positions past seed_pos. That makes the run end on reader // progress rather than on the attempt budget, which matters for a policy // like kDrop where writers never wait for the reader. - uint32_t min_positions_resolved = 0; + uint32_t min_positions_consumed = 0; }; // What a run actually did, so each test can assert that it exercised the path @@ -105,11 +105,11 @@ void RunStress(const StressParams& params, StressStats* stats) { test::SharedRingBufferForTesting ring(params.num_chunks, params.chunk_size); - Internals::SetPositions(ring.get(), params.seed_position); + Internals::SetPositions(ring.get(), params.seed_pos); StressDelegate delegate; SharedRingBufferReader reader(ring.get(), &delegate); - Internals::SetReaderPos(&reader, params.seed_position); + Internals::SetReaderPos(&reader, params.seed_pos); std::atomic<uint32_t> writers_done{0}; // Set when the run's progress target is reached or when the drain below @@ -175,10 +175,10 @@ }); } - // 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. + // Drain until every writer has finished and the ring buffer 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; @@ -187,13 +187,12 @@ writers_done.load(std::memory_order_acquire) == params.num_writers; const SharedRingBufferReader::DrainResult result = reader.Drain(64); if (result.last_result == - SharedRingBufferReader::ResolveResult::kProtocolError) { + SharedRingBufferReader::ConsumeResult::kProtocolError) { break; } // Unsigned subtraction keeps this right when the positions roll over. - if (params.min_positions_resolved != 0 && - reader.read_pos() - params.seed_position >= - params.min_positions_resolved) { + if (params.min_positions_consumed != 0 && + reader.read_pos() - params.seed_pos >= params.min_positions_consumed) { stop_writers.store(true, std::memory_order_relaxed); } if (all_done && !result.needs_another_drain()) @@ -203,18 +202,18 @@ ADD_FAILURE() << "Stress drain did not finish within its deadline: " << writers_done.load() << "/" << params.num_writers << " writers done, read_pos " << reader.read_pos() << " (" - << reader.read_pos() - params.seed_position + << reader.read_pos() - params.seed_pos << " positions past the seed), last result " << static_cast<int>(result.last_result); break; } - if (result.positions_resolved == 0) + if (result.positions_consumed == 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. + // joins below cannot wait on a writer parked on a full ring buffer. stop_writers.store(true, std::memory_order_relaxed); const base::TimeMillis shutdown_deadline = base::GetWallTimeMs() + base::TimeMillis(60000); @@ -273,15 +272,15 @@ StressStats stats; RunStressUntilRelocations({/*num_writers=*/4, /*num_chunks=*/8, /*chunk_size=*/256, - /*fragments_per_writer=*/4000, /*seed_position=*/0, + /*fragments_per_writer=*/4000, /*seed_pos=*/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 + // A drop policy on a small ring buffer 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 + // the per-writer checks ran on real data, not that the ring buffer achieved a // particular throughput. EXPECT_GT(stats.received, 100u); EXPECT_GT(stats.relocations, 0u); @@ -295,35 +294,35 @@ // // The run ends on reader progress, not on the writers' attempt budget: with // kDrop the writers never wait, so a reader short of CPU could otherwise - // watch them use up a fixed budget before it had resolved the positions up + // watch them use up a fixed budget before it had consumed the positions up // to the boundary. Requiring 16 traversals past it also means the writers // were still claiming with the wrapped-around count while the reader kept // advancing Free words from it. constexpr uint32_t kNumChunks = 4; - constexpr uint32_t kBoundary = kNumChunks * 65536u; - constexpr uint32_t kSeed = kBoundary - 16 * kNumChunks; - constexpr uint32_t kMinPositionsResolved = 32 * kNumChunks; + constexpr uint32_t kBoundaryPos = kNumChunks * 65536u; + constexpr uint32_t kSeedPos = kBoundaryPos - 16 * kNumChunks; + constexpr uint32_t kMinPositionsConsumed = 32 * kNumChunks; StressStats stats; RunStress({/*num_writers=*/4, kNumChunks, /*chunk_size=*/256, - /*fragments_per_writer=*/UINT32_MAX, kSeed, - BufferExhaustedPolicy::kDrop, kMinPositionsResolved}, + /*fragments_per_writer=*/UINT32_MAX, kSeedPos, + BufferExhaustedPolicy::kDrop, kMinPositionsConsumed}, &stats); EXPECT_GT(stats.received, 0u); - EXPECT_GE(stats.final_read_pos - kSeed, kMinPositionsResolved); + EXPECT_GE(stats.final_read_pos - kSeedPos, kMinPositionsConsumed); EXPECT_LT(WrapCountForPosition(stats.final_read_pos, kNumChunks), - WrapCountForPosition(kSeed, kNumChunks)); + WrapCountForPosition(kSeedPos, kNumChunks)); } TEST(SharedRingBufferConcurrencyTest, StressStallPolicy) { - // kStall needs the futex wait. Without it the first full ring reaches the - // deliberate PERFETTO_FATAL in AcquireNewChunk(), which would take the whole - // test binary down rather than fail this test. + // kStall needs the futex wait. Without it the first full ring buffer reaches + // the deliberate PERFETTO_FATAL in AcquireNewChunk(), which would take the + // whole test binary down rather than fail this test. if (!SharedRingBuffer::SupportsWriterWait()) GTEST_SKIP() << "The futex wait is not available on this platform"; StressStats stats; RunStress({/*num_writers=*/3, /*num_chunks=*/8, /*chunk_size=*/1024, - /*fragments_per_writer=*/2000, /*seed_position=*/0, + /*fragments_per_writer=*/2000, /*seed_pos=*/0, BufferExhaustedPolicy::kStall}, &stats); EXPECT_GT(stats.received, 0u);
diff --git a/src/tracing/v2/shared_ring_buffer_reader.cc b/src/tracing/v2/shared_ring_buffer_reader.cc index b10cdcd..eea23f1 100644 --- a/src/tracing/v2/shared_ring_buffer_reader.cc +++ b/src/tracing/v2/shared_ring_buffer_reader.cc
@@ -18,6 +18,7 @@ #include <stdint.h> +#include "perfetto/base/compiler.h" #include "perfetto/base/logging.h" #include "src/tracing/v2/shared_ring_buffer.h" #include "src/tracing/v2/shared_ring_buffer_abi.h" @@ -68,15 +69,16 @@ const uint32_t start_pos = read_pos_; DrainResult result{}; for (uint32_t i = 0; i < max_positions; ++i) { - result.last_result = ResolveNextPosition(); - if (result.last_result != ResolveResult::kChunkRead && - result.last_result != ResolveResult::kPositionSkipped) { + result.last_result = ConsumeNextPosition(); + if (result.last_result != ConsumeResult::kChunkRead && + result.last_result != ConsumeResult::kPositionSkipped) { break; } } - result.positions_resolved = read_pos_ - start_pos; - if (result.positions_resolved != 0) { + result.positions_consumed = read_pos_ - start_pos; + if (result.positions_consumed != 0) { + // uint32_t subtraction gives the forward distance across wraparound. // One publication and at most one wake cover the whole pass. Until this // point writers can only under-estimate free capacity. // @@ -97,126 +99,135 @@ // 4. RewriteRequested: skip; the writer still owns the chunk. Once it becomes // RewriteAcknowledged, only the reader may reclaim it on a later traversal. // -// In the first three cases, a lost CAS leaves read_pos unchanged for the next -// pass. Deliver the copy only after winning, so a retry cannot deliver the -// same fragments twice. RewriteAcknowledged has no competing writer -// transition; failure to reclaim it is a protocol error. -SharedRingBufferReader::ResolveResult -SharedRingBufferReader::ResolveNextPosition() { +// A lost CAS leaves read_pos unchanged. BeingWritten retries locally; Free +// and Complete retry on a later pass. Deliver the copy only after winning, +// so a retry cannot deliver the same fragments twice. RewriteAcknowledged has +// no competing writer transition; failure to reclaim it is a protocol error. +SharedRingBufferReader::ConsumeResult +SharedRingBufferReader::ConsumeNextPosition() { if (has_protocol_error_) - return ResolveResult::kProtocolError; + return ConsumeResult::kProtocolError; - // A stale write_pos only shortens this drain pass. + // The cursor bounds reservations; the chunk state publishes their payload. const uint32_t write_pos = ring_->LoadWritePos(); const uint32_t outstanding = NumOutstandingPositions(write_pos, read_pos_); if (outstanding == 0) - return ResolveResult::kNoData; - if (outstanding > num_chunks_) { + return ConsumeResult::kNoData; + if (PERFETTO_UNLIKELY(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", + "tracing v2: stopping ring buffer reader; write_pos %u is " + "%u positions ahead of read_pos %u, but the ring buffer has only %u " + "chunks", write_pos, outstanding, read_pos_, num_chunks_); - return ResolveResult::kProtocolError; + return ConsumeResult::kProtocolError; } // read_pos_ is the next logical position. - const uint32_t position = read_pos_; - const uint32_t chunk_idx = ChunkIndexOf(position, num_chunks_); + const uint32_t chunk_pos = read_pos_; + const auto chunk_idx = ChunkIndex::FromPosition(chunk_pos, num_chunks_); - // A failed compare-and-swap replaces this with the word that won. - uint32_t state_word = ring_->LoadChunkStateWord(chunk_idx); + for (;;) { + uint32_t state_word = ring_->LoadChunkStateWord(chunk_idx); - switch (ChunkStateOf(state_word)) { - case ChunkState::kFree: { - // Check reserved bits first; reclaiming must not hide an invalid word. - if ((state_word & ~kWriterIDMask) != 0) - return StopOnProtocolError("Free word has reserved bits", state_word); - // Only this reader advances the wrap; a different wrap here is an error. - const uint32_t expected_free_word = - MakeFreeStateWordForPosition(position, num_chunks_); - if (state_word != expected_free_word) { - return StopOnProtocolError( - "Free word carries another position's wrap count", state_word); + switch (ChunkStateOf(state_word)) { + case ChunkState::kFree: { + // Check reserved bits first; reclaiming must not hide an invalid word. + if ((state_word & ~kWriterIDMask) != 0) + return StopOnProtocolError("Free word has reserved bits", state_word); + // Only this reader advances the wrap. A different wrap is an error. + const uint32_t expected_free_word = + MakeFreeStateWordForPosition(chunk_pos, num_chunks_); + if (state_word != expected_free_word) { + return StopOnProtocolError( + "Free word carries another position's wrap count", state_word); + } + // Nobody claimed this reservation, so the reader advances the wrap + // count. A writer can still claim between the load and this CAS. The + // CAS then fails and the same position is retried as BeingWritten. + if (!ring_->TryMoveFreeChunkToNextWrap(chunk_pos, &state_word)) + return ConsumeResult::kRetryLater; + ++read_pos_; + ++stats_.positions_skipped; + return ConsumeResult::kPositionSkipped; } - // Nobody claimed this reservation, so the reader advances the wrap - // count. A writer can still claim between the load and this CAS. The - // CAS then fails and the same position is retried as BeingWritten. - if (!ring_->TryMoveFreeChunkToNextWrap(position, &state_word)) - return ResolveResult::kRetryLater; - ++read_pos_; - ++stats_.positions_skipped; - return ResolveResult::kPositionSkipped; - } - case ChunkState::kBeingWritten: { - const auto status = CopyCommittedPrefix(chunk_idx, state_word); - // Validation does not settle ownership. Even a malformed prefix must win - // the state transition before the reader can advance. - if (!ring_->TryRequestRewrite(chunk_idx, &state_word)) - return ResolveResult::kRetryLater; - ++read_pos_; - ++stats_.rewrite_requests; - return HandleCommittedPrefix(status); - } - - case ChunkState::kComplete: { - const auto status = CopyCommittedPrefix(chunk_idx, state_word); - // The writer may have taken the chunk back, turning Complete(N) into - // BeingWritten(N). The reader discards its copy and retries the same - // position instead of delivering data from a lost race. - if (!ring_->TryReleaseCompleteChunkAsFree(position, &state_word)) - return ResolveResult::kRetryLater; - ++read_pos_; - // A Complete chunk with no fragments can still carry kFlagDataLoss, and - // this reclaim was the last chance to see it: the writer's reuse CAS - // now fails and it forgets the chunk. The kBeingWritten case above - // stays silent for the same shape because that chunk still has an - // owner, which moves the flag to the relocated suffix. - if (status == CommittedPrefixStatus::kNoFragments && - (copied_chunk_.payload_flags & kFlagDataLoss)) { - delegate_->OnDataLoss(copied_chunk_.writer_id); + case ChunkState::kBeingWritten: { + const auto status = CopyCommittedPrefix(chunk_idx, state_word); + // Validation does not settle ownership. Even a malformed prefix must + // win the state transition before the reader can advance. + if (before_rewrite_for_testing_) + before_rewrite_for_testing_(); + if (!ring_->TryRequestRewrite(chunk_idx, &state_word)) { + // The writer published a newer prefix. Discard this copy and reload + // the state to try again. Each further publication increases the + // bounded fragment count; if the writer stops, the next CAS wins. + continue; + } + ++read_pos_; + ++stats_.rewrite_requests; + return HandleCommittedPrefix(status); } - return HandleCommittedPrefix(status); - } - case ChunkState::kRewriteRequested: - // The writer still owns this chunk. Resolve the position as a hole. - ++read_pos_; - ++stats_.positions_skipped; - return ResolveResult::kPositionSkipped; - - case ChunkState::kRewriteAcknowledged: - // RewriteAcknowledged has one canonical word with no payload fields, and - // the reclaim compares against exactly that word. After acknowledging, - // the writer is finished with the chunk and only this reader may change - // it, so a failed reclaim cannot be a lost race: either the word was not - // canonical or something other than this reader moved it. - if (!ring_->TryReleaseRewriteAcknowledgedChunkAsFree(position, - &state_word)) { - return StopOnProtocolError( - ChunkStateOf(state_word) == ChunkState::kRewriteAcknowledged - ? "RewriteAcknowledged word has payload bits set" - : "RewriteAcknowledged word changed under the reader", - state_word); + case ChunkState::kComplete: { + const auto status = CopyCommittedPrefix(chunk_idx, state_word); + // The writer may have taken the chunk back, turning Complete(N) into + // BeingWritten(N). The reader discards its copy and retries the same + // position instead of delivering data from a lost race. + if (!ring_->TryReleaseCompleteChunkAsFree(chunk_pos, &state_word)) + return ConsumeResult::kRetryLater; + ++read_pos_; + // A Complete chunk with no fragments can still carry kFlagDataLoss, and + // this reclaim was the last chance to see it: the writer's reuse CAS + // now fails and it forgets the chunk. The kBeingWritten case above + // stays silent for the same shape because that chunk still has an + // owner, which moves the flag to the relocated suffix. + if (status == CommittedPrefixStatus::kNoFragments && + (copied_chunk_.payload_flags & kFlagDataLoss)) { + delegate_->OnDataLoss(copied_chunk_.writer_id); + } + return HandleCommittedPrefix(status); } - ++read_pos_; - ++stats_.positions_skipped; - return ResolveResult::kPositionSkipped; - case ChunkState::kReserved5: - case ChunkState::kReserved6: - case ChunkState::kReserved7: - // The reader cannot safely reclaim an unknown state. - return StopOnProtocolError("reserved chunk state", state_word); + case ChunkState::kRewriteRequested: + // The writer still owns this chunk. Consume the position as a hole. + ++read_pos_; + ++stats_.positions_skipped; + return ConsumeResult::kPositionSkipped; + + case ChunkState::kRewriteAcknowledged: + // RewriteAcknowledged has one canonical word with no payload fields, + // and the reclaim compares against exactly that word. After + // acknowledging, the writer is finished with the chunk and only this + // reader may change it, so a failed reclaim cannot be a lost race: + // either the word was not canonical or something other than this reader + // moved it. + if (!ring_->TryReleaseRewriteAcknowledgedChunkAsFree(chunk_pos, + &state_word)) { + return StopOnProtocolError( + ChunkStateOf(state_word) == ChunkState::kRewriteAcknowledged + ? "RewriteAcknowledged word has payload bits set" + : "RewriteAcknowledged word changed under the reader", + state_word); + } + ++read_pos_; + ++stats_.positions_skipped; + return ConsumeResult::kPositionSkipped; + + case ChunkState::kReserved5: + case ChunkState::kReserved6: + case ChunkState::kReserved7: + // The reader cannot safely reclaim an unknown state. + return StopOnProtocolError("reserved chunk state", state_word); + } + PERFETTO_FATAL("tracing v2: unhandled chunk state word 0x%08x", state_word); } - PERFETTO_FATAL("tracing v2: unhandled chunk state word 0x%08x", state_word); } SharedRingBufferReader::CommittedPrefixStatus -SharedRingBufferReader::CopyCommittedPrefix(uint32_t chunk_idx, +SharedRingBufferReader::CopyCommittedPrefix(ChunkIndex chunk_idx, uint32_t state_word) { copied_fragments_.clear(); copied_chunk_ = ChunkContents{}; @@ -226,8 +237,8 @@ const uint32_t num_fragments = NumFragmentsOf(state_word); if (num_fragments == 0) { // A writer may still be storing BufferID after claiming BeingWritten(0). - // Only its first release publication makes that store visible. Do not - // touch any bytes beyond the state word when no fragment is published. + // Its first publication makes that store visible. Until then, do not touch + // any bytes beyond the state word. return CommittedPrefixStatus::kNoFragments; } @@ -241,13 +252,13 @@ const uint8_t* sizes_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, &sizes_cursor, &fragment_size) || - fragment_size > capacity - total) { + const auto fragment_size = + ReadFragmentSizeReversed(payload_begin, &sizes_cursor); + if (!fragment_size || *fragment_size > capacity - total) { return CommittedPrefixStatus::kMalformed; } - total += fragment_size; - copied_fragments_.push_back(Fragment{nullptr, fragment_size}); + total += *fragment_size; + copied_fragments_.push_back(Fragment{nullptr, *fragment_size}); } const uint32_t sizes_bytes = @@ -270,13 +281,13 @@ return CommittedPrefixStatus::kReady; } -SharedRingBufferReader::ResolveResult +SharedRingBufferReader::ConsumeResult SharedRingBufferReader::HandleCommittedPrefix(CommittedPrefixStatus status) { switch (status) { case CommittedPrefixStatus::kReady: ++stats_.chunks_read; delegate_->OnChunkRead(copied_chunk_); - return ResolveResult::kChunkRead; + return ConsumeResult::kChunkRead; case CommittedPrefixStatus::kMalformed: ++stats_.malformed_chunks; delegate_->OnDataLoss(copied_chunk_.writer_id); @@ -290,19 +301,19 @@ } ++stats_.positions_skipped; - return ResolveResult::kPositionSkipped; + return ConsumeResult::kPositionSkipped; } -SharedRingBufferReader::ResolveResult +SharedRingBufferReader::ConsumeResult SharedRingBufferReader::StopOnProtocolError(const char* reason, uint32_t state_word) { // Stop rather than trusting a malformed word from a producer. Log once. has_protocol_error_ = true; PERFETTO_ELOG( - "tracing v2: stopping ring reader at position %u: %s (chunk state word " - "0x%08x, %s)", + "tracing v2: stopping ring buffer reader at position %u: %s " + "(chunk state word 0x%08x, %s)", read_pos_, reason, state_word, ChunkStateName(ChunkStateOf(state_word))); - return ResolveResult::kProtocolError; + return ConsumeResult::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 index fcedb95..fdd0eab 100644 --- a/src/tracing/v2/shared_ring_buffer_reader.h +++ b/src/tracing/v2/shared_ring_buffer_reader.h
@@ -19,6 +19,7 @@ #include <stdint.h> +#include <functional> #include <vector> #include "perfetto/ext/tracing/core/basic_types.h" @@ -33,9 +34,9 @@ // Reads packet fragments from one SharedRingBuffer. // -// - Use each instance from one execution context. The ring's storage and +// - Use each instance from one execution context. The ring buffer's storage and // SharedRingBuffer view must outlive the reader. -// - Reservations are resolved in order. A relocated suffix gets a later +// - Reservations are consumed in order. A relocated suffix gets a later // reservation. // - The reader never waits for a writer, but may retry when it loses a // concurrent state transition. @@ -47,9 +48,10 @@ // producer-controlled shared memory. class SharedRingBufferReader { public: - // Result of resolving one position. - enum class ResolveResult { - // read_pos has caught up with write_pos. + // Result of consuming one position. + enum class ConsumeResult { + // No writer has reserved read_pos yet: read_pos has caught up with + // write_pos. kNoData, // A chunk was handed to the delegate and read_pos advanced. kChunkRead, @@ -59,7 +61,7 @@ // The reader lost a concurrent state transition. read_pos is unchanged // and the same position is retried later, without waiting for the writer. kRetryLater, - // Invalid ring state. This reader cannot continue. + // Invalid ring buffer state. This reader cannot continue. // - The offending position is neither advanced nor reclaimed. // - Earlier chunks in this drain may already have been delivered. // - Drain() still publishes any progress made before the error. @@ -83,13 +85,13 @@ }; struct DrainResult { - uint32_t positions_resolved = 0; - ResolveResult last_result = ResolveResult::kNoData; + uint32_t positions_consumed = 0; + ConsumeResult last_result = ConsumeResult::kNoData; // The caller should schedule another Drain() call. bool needs_another_drain() const { - return last_result != ResolveResult::kNoData && - last_result != ResolveResult::kProtocolError; + return last_result != ConsumeResult::kNoData && + last_result != ConsumeResult::kProtocolError; } }; @@ -119,11 +121,11 @@ 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. Without the bound, one pass over a large - // ring could monopolize the consumer's task sequence. The bound also caps - // the copying and delegate work done per task and decides how often - // read_pos gets published. + // Consumes up to |max_positions|, then publishes read_pos once and wakes any + // writer parked on a full ring buffer. These are logical positions rather + // than chunks: a position whose chunk was never claimed still counts towards + // the bound. Without it, one pass over a large ring buffer could monopolize + // the consumer's task sequence. DrainResult Drain(uint32_t max_positions); bool has_protocol_error() const { return has_protocol_error_; } @@ -144,8 +146,8 @@ private: friend class test::SharedRingBufferInternalsForTest; - // Resolves at most one position. Drain() publishes read_pos once per pass. - ResolveResult ResolveNextPosition(); + // Consumes at most one position. Drain() publishes read_pos once per pass. + ConsumeResult ConsumeNextPosition(); enum class CommittedPrefixStatus { kNoFragments, @@ -157,17 +159,17 @@ // 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_idx, + CommittedPrefixStatus CopyCommittedPrefix(ChunkIndex chunk_idx, uint32_t state_word); // Delivers a valid prefix to the delegate. Invalid or unsupported data is // reported as data loss. Called only after the position's compare-and-swap // won. - ResolveResult HandleCommittedPrefix(CommittedPrefixStatus); + ConsumeResult HandleCommittedPrefix(CommittedPrefixStatus); // Latches the error and logs |reason| once, together with the position and // the offending word. read_pos is not advanced. - ResolveResult StopOnProtocolError(const char* reason, uint32_t state_word); + ConsumeResult StopOnProtocolError(const char* reason, uint32_t state_word); SharedRingBuffer* const ring_; Delegate* const delegate_; @@ -178,7 +180,12 @@ uint32_t read_pos_ = 0; bool has_protocol_error_ = false; + // Pauses between the speculative copy and rewrite CAS in deterministic tests. + std::function<void()> before_rewrite_for_testing_; + // Published fragment bytes copied out of shared memory. + // TODO(sashwinbalaji): Measure whether embedding max-sized scratch arrays in + // the heap-allocated reader is faster than these reusable vectors. std::vector<uint8_t> copied_payload_; // Each fragment's decoded size and pointer into copied_payload_.
diff --git a/src/tracing/v2/shared_ring_buffer_reader_unittest.cc b/src/tracing/v2/shared_ring_buffer_reader_unittest.cc index cc2abd4..7364c75 100644 --- a/src/tracing/v2/shared_ring_buffer_reader_unittest.cc +++ b/src/tracing/v2/shared_ring_buffer_reader_unittest.cc
@@ -36,7 +36,7 @@ using Internals = test::SharedRingBufferInternalsForTest; using BeginFragmentResult = SharedRingBufferWriter::BeginFragmentResult; using EndFragmentResult = SharedRingBufferWriter::EndFragmentResult; -using ResolveResult = SharedRingBufferReader::ResolveResult; +using ConsumeResult = SharedRingBufferReader::ConsumeResult; using test::GetNoopWriterDelegate; using test::MakeWriter; using test::WriteFragment; @@ -87,15 +87,90 @@ // The nominal path. // --------------------------------------------------------------------------- +TEST(SharedRingBufferReaderTest, RewriteRaceRetriesWithinDrain) { + for (bool keep_writing : {false, true}) { + SCOPED_TRACE(keep_writing); + test::SharedRingBufferForTesting ring(4, 256); + RecordingDelegate delegate; + SharedRingBufferReader reader(ring.get(), &delegate); + auto writer = MakeWriter(ring.get(), kWriterA, kBuffer); + ASSERT_TRUE(WriteFragment(&writer, "first")); + auto range = writer.BeginFragment(6, false); + ASSERT_EQ(range.result, BeginFragmentResult::kSuccess); + memcpy(range.begin, "second", 6); + + uint32_t attempts = 0; + Internals::SetBeforeRewriteCallback(&reader, [&] { + if (++attempts != 1) + return; + // Publish after the reader copies "first", so its rewrite CAS loses. + ASSERT_EQ(writer.EndFragment(6, false), EndFragmentResult::kSuccess); + if (keep_writing) { + range = writer.BeginFragment(5, false); + ASSERT_EQ(range.result, BeginFragmentResult::kSuccess); + memcpy(range.begin, "third", 5); + } + }); + + const auto result = reader.Drain(1); + EXPECT_EQ(result.positions_consumed, 1u); + EXPECT_EQ(result.last_result, ConsumeResult::kChunkRead); + EXPECT_EQ(Internals::GetReadPos(ring.get()), 1u); + EXPECT_EQ(attempts, keep_writing ? 2u : 1u); + ASSERT_EQ(delegate.chunks.size(), 1u); + EXPECT_EQ(delegate.AllFragments(), + (std::vector<std::string>{"first", "second"})); + EXPECT_TRUE(delegate.writers_with_data_loss.empty()); + + if (keep_writing) { + ASSERT_EQ(writer.EndFragment(5, false), EndFragmentResult::kSuccess); + reader.Drain(4); + EXPECT_EQ(delegate.AllFragments(), + (std::vector<std::string>{"first", "second", "third"})); + } + } +} + +TEST(SharedRingBufferReaderTest, RewriteRaceReachesFragmentLimit) { + test::SharedRingBufferForTesting ring(2, 1024); + RecordingDelegate delegate; + SharedRingBufferReader reader(ring.get(), &delegate); + auto writer = MakeWriter(ring.get(), kWriterA, kBuffer); + ASSERT_TRUE(WriteFragment(&writer, "x")); + auto range = writer.BeginFragment(1, false); + ASSERT_EQ(range.result, BeginFragmentResult::kSuccess); + *range.begin = 'x'; + + uint32_t attempts = 0; + Internals::SetBeforeRewriteCallback(&reader, [&] { + ++attempts; + ASSERT_EQ(writer.EndFragment(1, false), EndFragmentResult::kSuccess); + if (attempts + 1 < kMaxFragmentsPerChunk) { + range = writer.BeginFragment(1, false); + ASSERT_EQ(range.result, BeginFragmentResult::kSuccess); + *range.begin = 'x'; + } + }); + + const auto result = reader.Drain(1); + EXPECT_EQ(result.positions_consumed, 1u); + EXPECT_EQ(result.last_result, ConsumeResult::kChunkRead); + EXPECT_EQ(attempts, kMaxFragmentsPerChunk - 1); + ASSERT_EQ(delegate.chunks.size(), 1u); + EXPECT_EQ(delegate.AllFragments(), + std::vector<std::string>(kMaxFragmentsPerChunk, "x")); + EXPECT_TRUE(delegate.writers_with_data_loss.empty()); +} + TEST(SharedRingBufferReaderTest, EmptyRing) { test::SharedRingBufferForTesting ring(4, 256); RecordingDelegate delegate; SharedRingBufferReader reader(ring.get(), &delegate); - EXPECT_EQ(Internals::ResolveNextPosition(&reader), ResolveResult::kNoData); + EXPECT_EQ(Internals::ConsumeNextPosition(&reader), ConsumeResult::kNoData); EXPECT_EQ(reader.read_pos(), 0u); const SharedRingBufferReader::DrainResult result = reader.Drain(16); - EXPECT_EQ(result.positions_resolved, 0u); + EXPECT_EQ(result.positions_consumed, 0u); EXPECT_FALSE(result.needs_another_drain()); } @@ -110,7 +185,7 @@ ASSERT_TRUE(WriteFragment(&writer, "beta")); writer.FinishCurrentChunk(); - EXPECT_EQ(Internals::ResolveNextPosition(&reader), ResolveResult::kChunkRead); + EXPECT_EQ(Internals::ConsumeNextPosition(&reader), ConsumeResult::kChunkRead); EXPECT_EQ(reader.read_pos(), 1u); ASSERT_EQ(delegate.chunks.size(), 1u); EXPECT_EQ(delegate.chunks[0].writer_id, kWriterA); @@ -119,8 +194,9 @@ (std::vector<std::string>{"alpha", "beta"})); // The chunk is now Free with the wrap count of its next position. - EXPECT_EQ(ring->LoadChunkStateWord(0), MakeFreeStateWord(1)); - EXPECT_EQ(Internals::ResolveNextPosition(&reader), ResolveResult::kNoData); + EXPECT_EQ(ring->LoadChunkStateWord(ChunkIndex::FromIndex(0)), + MakeFreeStateWord(1)); + EXPECT_EQ(Internals::ConsumeNextPosition(&reader), ConsumeResult::kNoData); } TEST(SharedRingBufferReaderTest, DrainPublishesOnce) { @@ -128,7 +204,7 @@ RecordingDelegate delegate; SharedRingBufferReader reader(ring.get(), &delegate); - // Four writers, one chunk each, so the pass has four positions to resolve. + // Four writers, one chunk each, so the pass has four positions to consume. std::vector<std::unique_ptr<SharedRingBufferWriter>> writers; for (uint32_t i = 0; i < 4; ++i) { writers.push_back(std::make_unique<SharedRingBufferWriter>( @@ -136,13 +212,13 @@ BufferExhaustedPolicy::kDrop, GetNoopWriterDelegate())); ASSERT_TRUE(WriteFragment(writers.back().get(), "x")); } - // The shared read_pos has not moved yet: ResolveNextPosition() does not + // The shared read_pos has not moved yet: ConsumeNextPosition() does not // publish it. EXPECT_EQ(Internals::GetReadPos(ring.get()), 0u); const SharedRingBufferReader::DrainResult result = reader.Drain(16); - EXPECT_EQ(result.positions_resolved, 4u); - EXPECT_EQ(result.last_result, ResolveResult::kNoData); + EXPECT_EQ(result.positions_consumed, 4u); + EXPECT_EQ(result.last_result, ConsumeResult::kNoData); EXPECT_FALSE(result.needs_another_drain()); EXPECT_EQ(Internals::GetReadPos(ring.get()), 4u); EXPECT_EQ(delegate.chunks.size(), 4u); @@ -163,12 +239,12 @@ } const SharedRingBufferReader::DrainResult first = reader.Drain(2); - EXPECT_EQ(first.positions_resolved, 2u); + EXPECT_EQ(first.positions_consumed, 2u); EXPECT_TRUE(first.needs_another_drain()); EXPECT_EQ(Internals::GetReadPos(ring.get()), 2u); const SharedRingBufferReader::DrainResult second = reader.Drain(16); - EXPECT_EQ(second.positions_resolved, 2u); + EXPECT_EQ(second.positions_consumed, 2u); EXPECT_FALSE(second.needs_another_drain()); EXPECT_EQ(delegate.chunks.size(), 4u); } @@ -184,14 +260,15 @@ SharedRingBufferReader reader(ring.get(), &delegate); // A writer reserved position 0 and never claimed it. - ASSERT_EQ(ring->TryReserveWritePos().position, 0u); + ASSERT_EQ(ring->TryReserveWritePos().write_pos, 0u); - EXPECT_EQ(Internals::ResolveNextPosition(&reader), - ResolveResult::kPositionSkipped); + EXPECT_EQ(Internals::ConsumeNextPosition(&reader), + ConsumeResult::kPositionSkipped); EXPECT_EQ(reader.read_pos(), 1u); EXPECT_EQ(reader.GetStats().positions_skipped, 1u); EXPECT_TRUE(delegate.chunks.empty()); - EXPECT_EQ(ring->LoadChunkStateWord(0), MakeFreeStateWord(1)); + EXPECT_EQ(ring->LoadChunkStateWord(ChunkIndex::FromIndex(0)), + MakeFreeStateWord(1)); } TEST(SharedRingBufferReaderTest, RewriteRequestedSkipped) { @@ -206,15 +283,15 @@ 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_TRUE(ring->TryRequestRewrite(ChunkIndex::FromIndex(0), &observed)); + const uint32_t marked = ring->LoadChunkStateWord(ChunkIndex::FromIndex(0)); - ASSERT_EQ(ring->TryReserveWritePos().position, 0u); - EXPECT_EQ(Internals::ResolveNextPosition(&reader), - ResolveResult::kPositionSkipped); + ASSERT_EQ(ring->TryReserveWritePos().write_pos, 0u); + EXPECT_EQ(Internals::ConsumeNextPosition(&reader), + ConsumeResult::kPositionSkipped); 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(ring->LoadChunkStateWord(ChunkIndex::FromIndex(0)), marked); EXPECT_EQ(reader.GetStats().positions_skipped, 1u); } @@ -227,14 +304,16 @@ 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->TryRequestRewrite(ChunkIndex::FromIndex(0), &observed)); ASSERT_TRUE(ring->TryAcknowledgeRewrite( - 0, ReplaceChunkState(being_written, ChunkState::kRewriteRequested))); + ChunkIndex::FromIndex(0), + ReplaceChunkState(being_written, ChunkState::kRewriteRequested))); - ASSERT_EQ(ring->TryReserveWritePos().position, 0u); - EXPECT_EQ(Internals::ResolveNextPosition(&reader), - ResolveResult::kPositionSkipped); - EXPECT_EQ(ring->LoadChunkStateWord(0), MakeFreeStateWord(1)); + ASSERT_EQ(ring->TryReserveWritePos().write_pos, 0u); + EXPECT_EQ(Internals::ConsumeNextPosition(&reader), + ConsumeResult::kPositionSkipped); + EXPECT_EQ(ring->LoadChunkStateWord(ChunkIndex::FromIndex(0)), + MakeFreeStateWord(1)); } // --------------------------------------------------------------------------- @@ -254,13 +333,13 @@ ASSERT_EQ(range.result, BeginFragmentResult::kSuccess); memcpy(range.begin, "suffix", 6); - EXPECT_EQ(Internals::ResolveNextPosition(&reader), ResolveResult::kChunkRead); + EXPECT_EQ(Internals::ConsumeNextPosition(&reader), ConsumeResult::kChunkRead); EXPECT_EQ(reader.GetStats().rewrite_requests, 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)), + EXPECT_EQ(ChunkStateOf(ring->LoadChunkStateWord(ChunkIndex::FromIndex(0))), ChunkState::kRewriteRequested); // The writer relocates its suffix, and the reader picks it up next pass. @@ -280,12 +359,12 @@ ASSERT_EQ(writer.BeginFragment(4, false).result, BeginFragmentResult::kSuccess); - EXPECT_EQ(Internals::ResolveNextPosition(&reader), - ResolveResult::kPositionSkipped); + EXPECT_EQ(Internals::ConsumeNextPosition(&reader), + ConsumeResult::kPositionSkipped); 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)), + EXPECT_EQ(ChunkStateOf(ring->LoadChunkStateWord(ChunkIndex::FromIndex(0))), ChunkState::kRewriteRequested); } @@ -299,19 +378,20 @@ RecordingDelegate delegate; SharedRingBufferReader reader(ring.get(), &delegate); - ASSERT_EQ(ring->TryReserveWritePos().position, 0u); + ASSERT_EQ(ring->TryReserveWritePos().write_pos, 0u); Internals::SetChunkStateWord( - ring.get(), 0, + ring.get(), ChunkIndex::FromIndex(0), MakeDataStateWord(ChunkState::kComplete, ChunkFormat::kReservedRouting, 0, 3, kWriterB)); - EXPECT_EQ(Internals::ResolveNextPosition(&reader), - ResolveResult::kPositionSkipped); + EXPECT_EQ(Internals::ConsumeNextPosition(&reader), + ConsumeResult::kPositionSkipped); EXPECT_EQ(reader.GetStats().unsupported_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(ring->LoadChunkStateWord(ChunkIndex::FromIndex(0)), + MakeFreeStateWord(1)); EXPECT_EQ(reader.read_pos(), 1u); } @@ -320,20 +400,21 @@ RecordingDelegate delegate; SharedRingBufferReader reader(ring.get(), &delegate); - ASSERT_EQ(ring->TryReserveWritePos().position, 0u); + ASSERT_EQ(ring->TryReserveWritePos().write_pos, 0u); // varint(0) takes one byte. 255 such entries do not fit in the 250-byte // payload area. Internals::SetChunkStateWord( - ring.get(), 0, + ring.get(), ChunkIndex::FromIndex(0), MakeDataStateWord(ChunkState::kComplete, ChunkFormat::kTargetBuffer, 0, 255, kWriterB)); - EXPECT_EQ(Internals::ResolveNextPosition(&reader), - ResolveResult::kPositionSkipped); + EXPECT_EQ(Internals::ConsumeNextPosition(&reader), + ConsumeResult::kPositionSkipped); EXPECT_EQ(reader.GetStats().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)); + EXPECT_EQ(ring->LoadChunkStateWord(ChunkIndex::FromIndex(0)), + MakeFreeStateWord(1)); } TEST(SharedRingBufferReaderTest, PayloadSizesOverlap) { @@ -342,20 +423,55 @@ RecordingDelegate delegate; SharedRingBufferReader reader(ring.get(), &delegate); - ASSERT_EQ(ring->TryReserveWritePos().position, 0u); + ASSERT_EQ(ring->TryReserveWritePos().write_pos, 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); + WriteFragmentSizeReversed(ring->chunk_at(ChunkIndex::FromIndex(0)) + 256, + 249); Internals::SetChunkStateWord( - ring.get(), 0, + ring.get(), ChunkIndex::FromIndex(0), MakeDataStateWord(ChunkState::kComplete, ChunkFormat::kTargetBuffer, 0, 1, kWriterB)); - EXPECT_EQ(Internals::ResolveNextPosition(&reader), - ResolveResult::kPositionSkipped); + EXPECT_EQ(Internals::ConsumeNextPosition(&reader), + ConsumeResult::kPositionSkipped); EXPECT_EQ(reader.GetStats().malformed_chunks, 1u); EXPECT_EQ(delegate.writers_with_data_loss, (std::vector<WriterID>{kWriterB})); - EXPECT_EQ(ring->LoadChunkStateWord(0), MakeFreeStateWord(1)); + EXPECT_EQ(ring->LoadChunkStateWord(ChunkIndex::FromIndex(0)), + MakeFreeStateWord(1)); +} + +TEST(SharedRingBufferReaderTest, NonMinimalSizeUsesActualDirectoryBytes) { + // 248 bytes fit with a minimal two-byte size. A three-byte encoding overlaps + // the payload by one byte. 247 bytes fit with either encoding. + for (uint32_t size : {247u, 248u}) { + SCOPED_TRACE(size); + test::SharedRingBufferForTesting ring(1, 256); + RecordingDelegate delegate; + SharedRingBufferReader reader(ring.get(), &delegate); + ASSERT_EQ(ring->TryReserveWritePos().write_pos, 0u); + uint8_t* chunk = ring->chunk_at(ChunkIndex::FromIndex(0)); + StoreTargetBufferID(chunk, kBuffer); + memset(chunk + kTargetBufferPayloadOffset, 'x', size); + chunk[255] = static_cast<uint8_t>(size); + chunk[254] = 0x81; + chunk[253] = 0x00; + Internals::SetChunkStateWord( + ring.get(), ChunkIndex::FromIndex(0), + MakeDataStateWord(ChunkState::kComplete, ChunkFormat::kTargetBuffer, 0, + 1, kWriterA)); + + EXPECT_EQ(reader.Drain(1).positions_consumed, 1u); + if (size == 247) { + EXPECT_EQ(delegate.AllFragments(), + std::vector<std::string>{std::string(size, 'x')}); + EXPECT_TRUE(delegate.writers_with_data_loss.empty()); + } else { + EXPECT_TRUE(delegate.chunks.empty()); + EXPECT_EQ(delegate.writers_with_data_loss, + std::vector<WriterID>{kWriterA}); + } + } } TEST(SharedRingBufferReaderTest, CumulativeSizeOverflow) { @@ -363,23 +479,24 @@ RecordingDelegate delegate; SharedRingBufferReader reader(ring.get(), &delegate); - ASSERT_EQ(ring->TryReserveWritePos().position, 0u); + ASSERT_EQ(ring->TryReserveWritePos().write_pos, 0u); // Two fragments claiming 400 bytes each in a chunk that has 506 bytes of // payload area, of which four go to the size varints. - uint8_t* chunk = ring->chunk_at(0); + uint8_t* chunk = ring->chunk_at(ChunkIndex::FromIndex(0)); uint8_t* sizes_begin = chunk + 512; - sizes_begin = WriteFragmentSize(sizes_begin, 400); - WriteFragmentSize(sizes_begin, 400); + sizes_begin = WriteFragmentSizeReversed(sizes_begin, 400); + WriteFragmentSizeReversed(sizes_begin, 400); Internals::SetChunkStateWord( - ring.get(), 0, + ring.get(), ChunkIndex::FromIndex(0), MakeDataStateWord(ChunkState::kComplete, ChunkFormat::kTargetBuffer, 0, 2, kWriterB)); - EXPECT_EQ(Internals::ResolveNextPosition(&reader), - ResolveResult::kPositionSkipped); + EXPECT_EQ(Internals::ConsumeNextPosition(&reader), + ConsumeResult::kPositionSkipped); EXPECT_EQ(reader.GetStats().malformed_chunks, 1u); EXPECT_TRUE(delegate.chunks.empty()); - EXPECT_EQ(ring->LoadChunkStateWord(0), MakeFreeStateWord(1)); + EXPECT_EQ(ring->LoadChunkStateWord(ChunkIndex::FromIndex(0)), + MakeFreeStateWord(1)); } TEST(SharedRingBufferReaderTest, UnterminatedFragmentSize) { @@ -387,19 +504,20 @@ RecordingDelegate delegate; SharedRingBufferReader reader(ring.get(), &delegate); - ASSERT_EQ(ring->TryReserveWritePos().position, 0u); + ASSERT_EQ(ring->TryReserveWritePos().write_pos, 0u); for (uint32_t i = 0; i < kMaxFragmentSizeVarIntBytes; ++i) - ring->chunk_at(0)[511 - i] = 0x80; + ring->chunk_at(ChunkIndex::FromIndex(0))[511 - i] = 0x80; Internals::SetChunkStateWord( - ring.get(), 0, + ring.get(), ChunkIndex::FromIndex(0), MakeDataStateWord(ChunkState::kComplete, ChunkFormat::kTargetBuffer, 0, 1, kWriterB)); - EXPECT_EQ(Internals::ResolveNextPosition(&reader), - ResolveResult::kPositionSkipped); + EXPECT_EQ(Internals::ConsumeNextPosition(&reader), + ConsumeResult::kPositionSkipped); EXPECT_EQ(reader.GetStats().malformed_chunks, 1u); EXPECT_TRUE(delegate.chunks.empty()); - EXPECT_EQ(ring->LoadChunkStateWord(0), MakeFreeStateWord(1)); + EXPECT_EQ(ring->LoadChunkStateWord(ChunkIndex::FromIndex(0)), + MakeFreeStateWord(1)); } // A count larger than the writer actually wrote is a producer claim like any @@ -410,13 +528,13 @@ RecordingDelegate delegate; SharedRingBufferReader reader(ring.get(), &delegate); - ASSERT_EQ(ring->TryReserveWritePos().position, 0u); + ASSERT_EQ(ring->TryReserveWritePos().write_pos, 0u); Internals::SetChunkStateWord( - ring.get(), 0, + ring.get(), ChunkIndex::FromIndex(0), MakeDataStateWord(ChunkState::kComplete, ChunkFormat::kTargetBuffer, 0, 8, kWriterB)); - EXPECT_EQ(Internals::ResolveNextPosition(&reader), ResolveResult::kChunkRead); + EXPECT_EQ(Internals::ConsumeNextPosition(&reader), ConsumeResult::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. @@ -425,7 +543,8 @@ for (const std::string& fragment : delegate.chunks[0].fragments) EXPECT_TRUE(fragment.empty()); EXPECT_EQ(reader.GetStats().malformed_chunks, 0u); - EXPECT_EQ(ring->LoadChunkStateWord(0), MakeFreeStateWord(1)); + EXPECT_EQ(ring->LoadChunkStateWord(ChunkIndex::FromIndex(0)), + MakeFreeStateWord(1)); EXPECT_EQ(reader.read_pos(), 1u); } @@ -435,18 +554,18 @@ RecordingDelegate delegate; SharedRingBufferReader reader(ring.get(), &delegate); - ASSERT_EQ(ring->TryReserveWritePos().position, 0u); + ASSERT_EQ(ring->TryReserveWritePos().write_pos, 0u); Internals::SetChunkStateWord( - ring.get(), 0, + ring.get(), ChunkIndex::FromIndex(0), MakeDataStateWord(ChunkState::kBeingWritten, ChunkFormat::kTargetBuffer, 0, 255, kWriterB)); - EXPECT_EQ(Internals::ResolveNextPosition(&reader), - ResolveResult::kPositionSkipped); + EXPECT_EQ(Internals::ConsumeNextPosition(&reader), + ConsumeResult::kPositionSkipped); EXPECT_EQ(reader.GetStats().malformed_chunks, 1u); // Malformed bytes may be dropped. What must not happen is leaving a // BeingWritten owner able to publish behind the reader. - EXPECT_EQ(ChunkStateOf(ring->LoadChunkStateWord(0)), + EXPECT_EQ(ChunkStateOf(ring->LoadChunkStateWord(ChunkIndex::FromIndex(0))), ChunkState::kRewriteRequested); } @@ -455,20 +574,21 @@ RecordingDelegate delegate; SharedRingBufferReader reader(ring.get(), &delegate); - ASSERT_EQ(ring->TryReserveWritePos().position, 0u); + ASSERT_EQ(ring->TryReserveWritePos().write_pos, 0u); const uint32_t reserved_word = 0x00000005u; - Internals::SetChunkStateWord(ring.get(), 0, reserved_word); + Internals::SetChunkStateWord(ring.get(), ChunkIndex::FromIndex(0), + reserved_word); - EXPECT_EQ(Internals::ResolveNextPosition(&reader), - ResolveResult::kProtocolError); + EXPECT_EQ(Internals::ConsumeNextPosition(&reader), + ConsumeResult::kProtocolError); EXPECT_TRUE(reader.has_protocol_error()); EXPECT_EQ(reader.read_pos(), 0u); - EXPECT_EQ(ring->LoadChunkStateWord(0), reserved_word); + EXPECT_EQ(ring->LoadChunkStateWord(ChunkIndex::FromIndex(0)), reserved_word); - // The ring is never read again, and a drain over it changes nothing. + // The ring buffer 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_result, ResolveResult::kProtocolError); + EXPECT_EQ(result.positions_consumed, 0u); + EXPECT_EQ(result.last_result, ConsumeResult::kProtocolError); EXPECT_EQ(Internals::GetReadPos(ring.get()), 0u); } @@ -477,18 +597,19 @@ RecordingDelegate delegate; SharedRingBufferReader reader(ring.get(), &delegate); - ASSERT_EQ(ring->TryReserveWritePos().position, 0u); + ASSERT_EQ(ring->TryReserveWritePos().write_pos, 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 + // it derives it from the position it is consuming, so no legal execution puts // this here. const uint32_t wrong_wrap = MakeFreeStateWord(3); - Internals::SetChunkStateWord(ring.get(), 0, wrong_wrap); + Internals::SetChunkStateWord(ring.get(), ChunkIndex::FromIndex(0), + wrong_wrap); - EXPECT_EQ(Internals::ResolveNextPosition(&reader), - ResolveResult::kProtocolError); + EXPECT_EQ(Internals::ConsumeNextPosition(&reader), + ConsumeResult::kProtocolError); EXPECT_TRUE(reader.has_protocol_error()); EXPECT_EQ(reader.read_pos(), 0u); - EXPECT_EQ(ring->LoadChunkStateWord(0), wrong_wrap); + EXPECT_EQ(ring->LoadChunkStateWord(ChunkIndex::FromIndex(0)), wrong_wrap); } // The control and fragment-count fields of a Free word are zero. A word using @@ -499,16 +620,18 @@ RecordingDelegate delegate; SharedRingBufferReader reader(ring.get(), &delegate); - ASSERT_EQ(ring->TryReserveWritePos().position, 0u); + ASSERT_EQ(ring->TryReserveWritePos().write_pos, 0u); // The wrap count is position 0's, but a reserved bit is set. const uint32_t reserved_bit_word = MakeFreeStateWord(0) | (1u << 8); - Internals::SetChunkStateWord(ring.get(), 0, reserved_bit_word); + Internals::SetChunkStateWord(ring.get(), ChunkIndex::FromIndex(0), + reserved_bit_word); - EXPECT_EQ(Internals::ResolveNextPosition(&reader), - ResolveResult::kProtocolError); + EXPECT_EQ(Internals::ConsumeNextPosition(&reader), + ConsumeResult::kProtocolError); EXPECT_TRUE(reader.has_protocol_error()); EXPECT_EQ(reader.read_pos(), 0u); - EXPECT_EQ(ring->LoadChunkStateWord(0), reserved_bit_word); + EXPECT_EQ(ring->LoadChunkStateWord(ChunkIndex::FromIndex(0)), + reserved_bit_word); } // RewriteAcknowledged has one canonical word: the state bits and nothing else. @@ -520,21 +643,21 @@ RecordingDelegate delegate; SharedRingBufferReader reader(ring.get(), &delegate); - ASSERT_EQ(ring->TryReserveWritePos().position, 0u); + ASSERT_EQ(ring->TryReserveWritePos().write_pos, 0u); const uint32_t forged = kRewriteAcknowledgedStateWord | kFlagDataLoss; - Internals::SetChunkStateWord(ring.get(), 0, forged); + Internals::SetChunkStateWord(ring.get(), ChunkIndex::FromIndex(0), forged); - EXPECT_EQ(Internals::ResolveNextPosition(&reader), - ResolveResult::kProtocolError); + EXPECT_EQ(Internals::ConsumeNextPosition(&reader), + ConsumeResult::kProtocolError); EXPECT_TRUE(reader.has_protocol_error()); EXPECT_EQ(reader.read_pos(), 0u); - EXPECT_EQ(ring->LoadChunkStateWord(0), forged); + EXPECT_EQ(ring->LoadChunkStateWord(ChunkIndex::FromIndex(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 +// mean consuming positions that were never reserved, one drain pass after // another, for as long as the producer keeps write_pos there. TEST(SharedRingBufferReaderTest, TooManyOutstandingStopsRing) { test::SharedRingBufferForTesting ring(4, 256); @@ -544,8 +667,8 @@ // Four chunks, five outstanding positions. Internals::SetWritePos(ring.get(), 5); - EXPECT_EQ(Internals::ResolveNextPosition(&reader), - ResolveResult::kProtocolError); + EXPECT_EQ(Internals::ConsumeNextPosition(&reader), + ConsumeResult::kProtocolError); EXPECT_TRUE(reader.has_protocol_error()); EXPECT_EQ(reader.read_pos(), 0u); EXPECT_EQ(Internals::GetReadPos(ring.get()), 0u); @@ -553,7 +676,7 @@ } // The boundary itself is legal and must still be drained: num_chunks -// outstanding positions is exactly a full ring. +// outstanding positions is exactly a full ring buffer. TEST(SharedRingBufferReaderTest, FullRingIsLegal) { test::SharedRingBufferForTesting ring(4, 256); RecordingDelegate delegate; @@ -568,7 +691,7 @@ const SharedRingBufferReader::DrainResult result = reader.Drain(8); EXPECT_FALSE(reader.has_protocol_error()); - EXPECT_EQ(result.positions_resolved, 4u); + EXPECT_EQ(result.positions_consumed, 4u); EXPECT_EQ(delegate.chunks.size(), 4u); } @@ -621,15 +744,16 @@ ASSERT_EQ(writer.BeginFragment(4, false).result, BeginFragmentResult::kSuccess); ASSERT_EQ(writer.FinishCurrentChunk(), EndFragmentResult::kSuccess); - ASSERT_EQ(ring->LoadChunkStateWord(0), + ASSERT_EQ(ring->LoadChunkStateWord(ChunkIndex::FromIndex(0)), MakeDataStateWord(ChunkState::kComplete, ChunkFormat::kTargetBuffer, kFlagDataLoss, 0, kWriterA)); - EXPECT_EQ(Internals::ResolveNextPosition(&reader), - ResolveResult::kPositionSkipped); + EXPECT_EQ(Internals::ConsumeNextPosition(&reader), + ConsumeResult::kPositionSkipped); EXPECT_TRUE(delegate.chunks.empty()); EXPECT_EQ(delegate.writers_with_data_loss, (std::vector<WriterID>{kWriterA})); - EXPECT_EQ(ring->LoadChunkStateWord(0), MakeFreeStateWord(1)); + EXPECT_EQ(ring->LoadChunkStateWord(ChunkIndex::FromIndex(0)), + MakeFreeStateWord(1)); // The writer's reuse of that chunk fails and its replacement does not repeat // the flag. @@ -655,11 +779,11 @@ ASSERT_EQ(range.result, BeginFragmentResult::kSuccess); memcpy(range.begin, "suffix", 6); - EXPECT_EQ(Internals::ResolveNextPosition(&reader), - ResolveResult::kPositionSkipped); + EXPECT_EQ(Internals::ConsumeNextPosition(&reader), + ConsumeResult::kPositionSkipped); EXPECT_TRUE(delegate.chunks.empty()); EXPECT_TRUE(delegate.writers_with_data_loss.empty()); - EXPECT_EQ(ChunkStateOf(ring->LoadChunkStateWord(0)), + EXPECT_EQ(ChunkStateOf(ring->LoadChunkStateWord(ChunkIndex::FromIndex(0))), ChunkState::kRewriteRequested); ASSERT_EQ(writer.EndFragment(6, false), EndFragmentResult::kSuccess); @@ -674,8 +798,8 @@ EXPECT_TRUE(delegate.writers_with_data_loss.empty()); } -// A one-chunk ring is legal: every position maps to chunk 0 and the ring -// alternates between one outstanding reservation and empty. +// A one-chunk ring buffer is legal: every position maps to chunk 0 and the ring +// buffer alternates between one outstanding reservation and empty. TEST(SharedRingBufferReaderTest, OneChunkRing) { test::SharedRingBufferForTesting ring(1, 256); RecordingDelegate delegate; @@ -689,10 +813,10 @@ writer.FinishCurrentChunk(); expected.push_back(bytes); const SharedRingBufferReader::DrainResult result = reader.Drain(4); - EXPECT_EQ(result.positions_resolved, 1u) << i; + EXPECT_EQ(result.positions_consumed, 1u) << i; } EXPECT_EQ(delegate.AllFragments(), expected); - EXPECT_EQ(Internals::ResolveNextPosition(&reader), ResolveResult::kNoData); + EXPECT_EQ(Internals::ConsumeNextPosition(&reader), ConsumeResult::kNoData); } // An aligned, non-power-of-two chunk size, end to end through writer and
diff --git a/src/tracing/v2/shared_ring_buffer_test_utils.h b/src/tracing/v2/shared_ring_buffer_test_utils.h index 4dd3419..d27b45a 100644 --- a/src/tracing/v2/shared_ring_buffer_test_utils.h +++ b/src/tracing/v2/shared_ring_buffer_test_utils.h
@@ -22,7 +22,9 @@ #include <string.h> #include <atomic> +#include <functional> #include <string> +#include <utility> #include "perfetto/ext/base/no_destructor.h" #include "perfetto/ext/base/paged_memory.h" @@ -33,9 +35,10 @@ namespace perfetto::tracing_v2::test { -// Owns zero-filled, page-aligned memory for a ring and exposes the non-owning -// SharedRingBuffer view over it. Production code allocates the region -// elsewhere. Tests use this so that the ring layout is specified only once. +// Owns zero-filled, page-aligned memory for a ring buffer and exposes the +// non-owning SharedRingBuffer view over it. Production code allocates the +// region elsewhere. Tests use this so that the ring buffer layout is specified +// only once. class SharedRingBufferForTesting { public: SharedRingBufferForTesting(uint32_t num_chunks, uint32_t chunk_size) @@ -63,44 +66,40 @@ public: // Injects a state word for corruption and unknown-ABI tests. static void SetChunkStateWord(SharedRingBuffer* ring, - uint32_t chunk_idx, + ChunkIndex chunk_idx, uint32_t state_word) { - ring->chunk_state_word_at(chunk_idx)->store(state_word, - std::memory_order_release); + ring->chunk_state_word_at(chunk_idx)->store(state_word); } // Injects a write_pos without reserving the intervening positions. static void SetWritePos(SharedRingBuffer* ring, uint32_t write_pos) { RingBufferHeader* header = ring->header(); - const uint64_t rw_positions = - header->rw_positions.load(std::memory_order_relaxed); + const uint64_t rw_positions = header->rw_positions.load(); header->rw_positions.store( - PackRwPositions(write_pos, ReadPosOf(rw_positions)), - std::memory_order_relaxed); + PackRwPositions(write_pos, ReadPosOf(rw_positions))); } - // Seeds a valid ring state near a position or wrap-count rollover: every - // chunk becomes Free for the first position at or after |position| that maps - // to it, and both positions are set to |position|. - static void SetPositions(SharedRingBuffer* ring, uint32_t position) { + // Seeds a valid ring buffer state near a position or wrap-count rollover: + // every chunk becomes Free for the first position at or after |chunk_pos| + // that maps to it, and both positions are set to |chunk_pos|. + static void SetPositions(SharedRingBuffer* ring, uint32_t chunk_pos) { for (uint32_t chunk_idx = 0; chunk_idx < ring->num_chunks(); ++chunk_idx) { - const uint32_t first_position = - position + ((chunk_idx - position) & (ring->num_chunks() - 1)); - ring->chunk_state_word_at(chunk_idx)->store( - MakeFreeStateWordForPosition(first_position, ring->num_chunks()), - std::memory_order_relaxed); + const uint32_t first_pos = + chunk_pos + ((chunk_idx - chunk_pos) & (ring->num_chunks() - 1)); + const uint32_t free_word = + MakeFreeStateWordForPosition(first_pos, ring->num_chunks()); + ring->chunk_state_word_at(ChunkIndex::FromIndex(chunk_idx)) + ->store(free_word); } - ring->header()->rw_positions.store(PackRwPositions(position, position), - std::memory_order_release); + ring->header()->rw_positions.store(PackRwPositions(chunk_pos, chunk_pos)); } static uint32_t GetReadPos(const SharedRingBuffer* ring) { - return ReadPosOf( - ring->header()->rw_positions.load(std::memory_order_relaxed)); + return ReadPosOf(ring->header()->rw_positions.load()); } static uint32_t GetNumWritersWaiting(const SharedRingBuffer* ring) { - return ring->header()->num_writers_waiting.load(std::memory_order_relaxed); + return ring->header()->num_writers_waiting.load(); } // Start the production CAS loops from an old rw_positions snapshot, so that @@ -116,15 +115,20 @@ ring->PublishReadPosFromSnapshot(rw_positions, read_pos); } - // Starts the reader at |position| instead of zero, to match a ring seeded - // near uint32_t rollover. - static void SetReaderPos(SharedRingBufferReader* reader, uint32_t position) { - reader->read_pos_ = position; + // Starts the reader at |chunk_pos| instead of zero, to match a ring buffer + // seeded near uint32_t rollover. + static void SetReaderPos(SharedRingBufferReader* reader, uint32_t chunk_pos) { + reader->read_pos_ = chunk_pos; } - static SharedRingBufferReader::ResolveResult ResolveNextPosition( + static SharedRingBufferReader::ConsumeResult ConsumeNextPosition( SharedRingBufferReader* reader) { - return reader->ResolveNextPosition(); + return reader->ConsumeNextPosition(); + } + + static void SetBeforeRewriteCallback(SharedRingBufferReader* reader, + std::function<void()> callback) { + reader->before_rewrite_for_testing_ = std::move(callback); } };
diff --git a/src/tracing/v2/shared_ring_buffer_unittest.cc b/src/tracing/v2/shared_ring_buffer_unittest.cc index 63f16d9..ee1d312 100644 --- a/src/tracing/v2/shared_ring_buffer_unittest.cc +++ b/src/tracing/v2/shared_ring_buffer_unittest.cc
@@ -51,9 +51,10 @@ 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_idx) { +// Reads a chunk's state word without going through the ring buffer's own +// accessors, so that a test observing the ring buffer cannot be fooled by a bug +// in them. +uint32_t PeekStateWord(SharedRingBuffer* ring, ChunkIndex chunk_idx) { const uint8_t* chunk = ring->chunk_at(chunk_idx); return static_cast<uint32_t>(chunk[0]) | (static_cast<uint32_t>(chunk[1]) << 8) | @@ -79,20 +80,20 @@ } // --------------------------------------------------------------------------- -// Ring dimensions. +// Ring buffer dimensions. // --------------------------------------------------------------------------- // A page-aligned, zero-filled region for the constructor-contract tests below. // SharedRingBufferForTesting is not used here because it never builds an -// invalid ring. +// invalid ring buffer. size_t RingSizeFor(uint32_t num_chunks, uint32_t chunk_size) { return sizeof(RingBufferHeader) + static_cast<size_t>(num_chunks) * chunk_size; } -// An invalid ring layout is a configuration error, so the constructor CHECKs. -// It only does arithmetic on |size|, which is why an impossibly large region -// can be described by a small mapping. +// An invalid ring buffer layout is a configuration error, so the constructor +// CHECKs. It only does arithmetic on |size|, which is why an impossibly large +// region can be described by a small mapping. TEST(SharedRingBufferTest, InvalidLayout) { base::PagedMemory memory = base::PagedMemory::Allocate(64 * 1024); uint8_t* start = static_cast<uint8_t*>(memory.Get()); @@ -178,19 +179,25 @@ for (uint32_t chunk_size : {260u, 1000u, 65536u}) { test::SharedRingBufferForTesting ring(4, chunk_size); EXPECT_EQ(ring->chunk_size(), chunk_size); - EXPECT_EQ(static_cast<uint32_t>(ring->chunk_at(1) - ring->chunk_at(0)), + EXPECT_EQ(static_cast<uint32_t>(ring->chunk_at(ChunkIndex::FromIndex(1)) - + ring->chunk_at(ChunkIndex::FromIndex(0))), chunk_size); } } // 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. +// already correct and the ring buffer does not walk it at construction. TEST(SharedRingBufferTest, FreshMappingIsFree) { test::SharedRingBufferForTesting ring(1024, kChunkSize); - 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); + for (uint32_t chunk_idx = 0; chunk_idx < ring->num_chunks(); ++chunk_idx) { + ASSERT_EQ(PeekStateWord(ring.get(), ChunkIndex::FromIndex(chunk_idx)), 0u) + << chunk_idx; + ASSERT_EQ(ChunkStateOf( + ring->LoadChunkStateWord(ChunkIndex::FromIndex(chunk_idx))), + ChunkState::kFree); + ASSERT_EQ( + WrapCountOf(ring->LoadChunkStateWord(ChunkIndex::FromIndex(chunk_idx))), + 0u); } EXPECT_EQ(Internals::GetReadPos(ring.get()), 0u); EXPECT_EQ(ring->LoadWritePos(), 0u); @@ -200,18 +207,19 @@ TEST(SharedRingBufferTest, StateWordAlignment) { for (uint32_t chunk_size : {256u, 260u, 512u, 4096u, 65536u}) { test::SharedRingBufferForTesting ring(8, chunk_size); - for (uint32_t i = 0; i < ring->num_chunks(); ++i) { - const auto address = reinterpret_cast<uintptr_t>(ring->chunk_at(i)); + for (uint32_t chunk_idx = 0; chunk_idx < ring->num_chunks(); ++chunk_idx) { + const auto address = reinterpret_cast<uintptr_t>( + ring->chunk_at(ChunkIndex::FromIndex(chunk_idx))); // 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; + << chunk_size << "/" << chunk_idx; // 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; + ASSERT_EQ(address % 64, 0u) << chunk_size << "/" << chunk_idx; } } } @@ -224,9 +232,11 @@ TEST(SharedRingBufferTest, ChunksFollowHeader) { for (uint32_t chunk_size : {256u, 260u, 4096u}) { test::SharedRingBufferForTesting ring(4, chunk_size); - const uintptr_t first = reinterpret_cast<uintptr_t>(ring->chunk_at(0)); + const uintptr_t first = + reinterpret_cast<uintptr_t>(ring->chunk_at(ChunkIndex::FromIndex(0))); EXPECT_EQ(first % 256, 64u) << "chunk_size " << chunk_size; - const uintptr_t last = reinterpret_cast<uintptr_t>(ring->chunk_at(3)); + const uintptr_t last = + reinterpret_cast<uintptr_t>(ring->chunk_at(ChunkIndex::FromIndex(3))); EXPECT_EQ(last - first, 3u * chunk_size); } } @@ -236,19 +246,20 @@ // --------------------------------------------------------------------------- TEST(SharedRingBufferTest, ReserveUntilFull) { - // Reservations are consecutive tickets until the ring is full. + // Reservations are consecutive tickets until the ring buffer is full. test::SharedRingBufferForTesting ring(4, kChunkSize); - for (uint32_t expected = 0; expected < 4; ++expected) { + for (uint32_t expected_write_pos = 0; expected_write_pos < 4; + ++expected_write_pos) { const auto reservation = ring->TryReserveWritePos(); ASSERT_EQ(reservation.result, ReserveResult::kReserved); - EXPECT_EQ(reservation.position, expected); + EXPECT_EQ(reservation.write_pos, expected_write_pos); } const auto full = ring->TryReserveWritePos(); EXPECT_EQ(full.result, ReserveResult::kFull); - // Nothing was reserved, so write_pos did not move: a full ring must not burn - // a position. + // Nothing was reserved, so write_pos did not move: a full ring buffer 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. @@ -269,7 +280,7 @@ const auto reservation = ring->TryReserveWritePos(); ASSERT_EQ(reservation.result, ReserveResult::kReserved); - EXPECT_EQ(reservation.position, 2u); + EXPECT_EQ(reservation.write_pos, 2u); EXPECT_EQ(reservation.read_pos_for_wait, 1u); EXPECT_EQ(Internals::GetReadPos(ring.get()), 1u); EXPECT_EQ(ring->LoadWritePos(), 3u); @@ -278,26 +289,26 @@ // One load of the packed positions decides capacity across uint32_t rollover. TEST(SharedRingBufferTest, CapacityAcrossPositionRollover) { test::SharedRingBufferForTesting ring(8, kChunkSize); - const uint32_t kSeed = 0xfffffffcu; // Four positions before the rollover. - Internals::SetPositions(ring.get(), kSeed); + const uint32_t kSeedPos = 0xfffffffcu; // Four positions before the rollover. + Internals::SetPositions(ring.get(), kSeedPos); for (uint32_t i = 0; i < 8; ++i) { const auto reservation = ring->TryReserveWritePos(); ASSERT_EQ(reservation.result, ReserveResult::kReserved) << i; - EXPECT_EQ(reservation.position, kSeed + i) << i; + EXPECT_EQ(reservation.write_pos, kSeedPos + i) << i; } // write_pos has wrapped: 0xfffffffc + 8 = 4. EXPECT_EQ(ring->LoadWritePos(), 4u); const auto full = ring->TryReserveWritePos(); ASSERT_EQ(full.result, ReserveResult::kFull); - EXPECT_EQ(full.read_pos_for_wait, kSeed); + EXPECT_EQ(full.read_pos_for_wait, kSeedPos); - ring->PublishReadPos(kSeed + 1); + ring->PublishReadPos(kSeedPos + 1); const auto reservation = ring->TryReserveWritePos(); ASSERT_EQ(reservation.result, ReserveResult::kReserved); - EXPECT_EQ(reservation.position, 4u); - EXPECT_EQ(reservation.read_pos_for_wait, kSeed + 1); + EXPECT_EQ(reservation.write_pos, 4u); + EXPECT_EQ(reservation.read_pos_for_wait, kSeedPos + 1); } // --------------------------------------------------------------------------- @@ -316,8 +327,8 @@ // not reuse either half of the old value. TEST(SharedRingBufferTest, ReservationLosesToPublication) { test::SharedRingBufferForTesting ring(4, kChunkSize); - ASSERT_EQ(ring->TryReserveWritePos().position, 0u); - ASSERT_EQ(ring->TryReserveWritePos().position, 1u); + ASSERT_EQ(ring->TryReserveWritePos().write_pos, 0u); + ASSERT_EQ(ring->TryReserveWritePos().write_pos, 1u); // The value the writer loaded before the reader published. const uint64_t stale_rw_positions = PackRwPositions(2, 0); @@ -326,7 +337,7 @@ const auto reservation = Internals::TryReserveWritePosFromSnapshot(ring.get(), stale_rw_positions); ASSERT_EQ(reservation.result, ReserveResult::kReserved); - EXPECT_EQ(reservation.position, 2u); + EXPECT_EQ(reservation.write_pos, 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_for_wait, 1u); @@ -335,14 +346,15 @@ } // 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. +// ring buffer is full. The retry must take the Full exit, without burning a +// position. TEST(SharedRingBufferTest, ReservationLossFindsFull) { test::SharedRingBufferForTesting ring(2, kChunkSize); - ASSERT_EQ(ring->TryReserveWritePos().position, 0u); + ASSERT_EQ(ring->TryReserveWritePos().write_pos, 0u); const uint64_t stale_rw_positions = PackRwPositions(1, 0); // One outstanding: capacity left. - ASSERT_EQ(ring->TryReserveWritePos().position, 1u); // Now (2, 0): full. + ASSERT_EQ(ring->TryReserveWritePos().write_pos, 1u); // Now (2, 0): full. const auto reservation = Internals::TryReserveWritePosFromSnapshot(ring.get(), stale_rw_positions); @@ -359,7 +371,8 @@ // 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). + ASSERT_EQ(ring->TryReserveWritePos().write_pos, + 0u); // Now (write=1, read=0). Internals::PublishReadPosFromSnapshot(ring.get(), stale_rw_positions, 1); EXPECT_EQ(ring->LoadWritePos(), 1u); @@ -373,22 +386,23 @@ const auto reservation = ring->TryReserveWritePos(); ASSERT_EQ(reservation.result, ReserveResult::kReserved); - EXPECT_EQ(reservation.position, 0u); + EXPECT_EQ(reservation.write_pos, 0u); EXPECT_EQ(ring->TryReserveWritePos().result, ReserveResult::kFull); ASSERT_TRUE(ring->TryAcquireChunkForWriting(0, BeingWrittenWord(kWriterA))); uint32_t observed = BeingWrittenWord(kWriterA); - ASSERT_TRUE( - ring->TryReleaseChunkAsComplete(0, CompleteWord(kWriterA, 1), &observed)); + ASSERT_TRUE(ring->TryReleaseChunkAsComplete( + ChunkIndex::FromIndex(0), CompleteWord(kWriterA, 1), &observed)); 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)); + EXPECT_EQ(PeekStateWord(ring.get(), ChunkIndex::FromIndex(0)), + MakeFreeStateWord(1)); ring->PublishReadPos(1); const auto next = ring->TryReserveWritePos(); ASSERT_EQ(next.result, ReserveResult::kReserved); - EXPECT_EQ(next.position, 1u); + EXPECT_EQ(next.write_pos, 1u); EXPECT_TRUE(ring->TryAcquireChunkForWriting(1, BeingWrittenWord(kWriterB))); } @@ -402,9 +416,10 @@ const auto reservation = ring->TryReserveWritePos(); ASSERT_EQ(reservation.result, ReserveResult::kReserved); - EXPECT_TRUE(ring->TryAcquireChunkForWriting(reservation.position, + EXPECT_TRUE(ring->TryAcquireChunkForWriting(reservation.write_pos, BeingWrittenWord(kWriterA))); - EXPECT_EQ(PeekStateWord(ring.get(), 0), BeingWrittenWord(kWriterA)); + EXPECT_EQ(PeekStateWord(ring.get(), ChunkIndex::FromIndex(0)), + BeingWrittenWord(kWriterA)); } // A writer that reserved a position and then slept wakes up expecting the free @@ -420,23 +435,25 @@ ring->PublishReadPos(4); const auto stale = ring->TryReserveWritePos(); ASSERT_EQ(stale.result, ReserveResult::kReserved); - ASSERT_EQ(stale.position, 4u); + ASSERT_EQ(stale.write_pos, 4u); - // The reader resolves positions 0 to 4 as holes, so chunk 0 ends up Free + // The reader consumes positions 0 to 4 as holes, so chunk 0 ends up Free // with position 8's wrap count. - for (uint32_t position = 0; position <= 4; ++position) { - const uint32_t chunk_idx = ChunkIndexOf(position, 4); + for (uint32_t chunk_pos = 0; chunk_pos <= 4; ++chunk_pos) { + const auto chunk_idx = ChunkIndex::FromPosition(chunk_pos, 4); uint32_t observed = ring->LoadChunkStateWord(chunk_idx); - ASSERT_TRUE(ring->TryMoveFreeChunkToNextWrap(position, &observed)) - << position; + ASSERT_TRUE(ring->TryMoveFreeChunkToNextWrap(chunk_pos, &observed)) + << chunk_pos; } - ASSERT_EQ(WrapCountOf(PeekStateWord(ring.get(), 0)), 2u); + ASSERT_EQ(WrapCountOf(PeekStateWord(ring.get(), ChunkIndex::FromIndex(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, + EXPECT_FALSE(ring->TryAcquireChunkForWriting(stale.write_pos, BeingWrittenWord(kWriterA))); - EXPECT_EQ(WrapCountOf(PeekStateWord(ring.get(), 0)), 2u); + EXPECT_EQ(WrapCountOf(PeekStateWord(ring.get(), ChunkIndex::FromIndex(0))), + 2u); // The writer holding position 8 - the one the chunk was actually prepared // for - still gets in. @@ -448,9 +465,10 @@ test::SharedRingBufferForTesting ring(2, kChunkSize); // Positions 0 and 2 both map to chunk 0 but belong to different traversals. - uint32_t observed = ring->LoadChunkStateWord(0); + uint32_t observed = ring->LoadChunkStateWord(ChunkIndex::FromIndex(0)); ASSERT_TRUE(ring->TryMoveFreeChunkToNextWrap(0, &observed)); - ASSERT_EQ(WrapCountOf(PeekStateWord(ring.get(), 0)), 1u); + ASSERT_EQ(WrapCountOf(PeekStateWord(ring.get(), ChunkIndex::FromIndex(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. @@ -469,57 +487,62 @@ // A fresh chunk 0 is Free(0), the word position 0 expects. One full // wrap-count period later, position 2 * 65536 maps to the same chunk and // its wrap count - bit 16 of the traversal number - is truncated back to - // zero, so this claim lands even though the ring never ran. + // zero, so this claim lands even though the ring buffer never ran. EXPECT_TRUE( ring->TryAcquireChunkForWriting(2u * 65536, BeingWrittenWord(kWriterA))); } -// The wrap count a seeded ring stamps is the low 16 bits of the traversal -// number. +// The wrap count a seeded ring buffer stamps is the low 16 bits of the +// traversal number. TEST(SharedRingBufferTest, SeededWrapCounts) { test::SharedRingBufferForTesting ring(4, kChunkSize); // 4 * 65536 is one whole wrap-count period, so every chunk is stamped with // exactly the word a fresh mapping holds. Internals::SetPositions(ring.get(), 4u * 65536); - for (uint32_t i = 0; i < 4; ++i) - EXPECT_EQ(PeekStateWord(ring.get(), i), 0u) << i; + for (uint32_t chunk_idx = 0; chunk_idx < 4; ++chunk_idx) + EXPECT_EQ(PeekStateWord(ring.get(), ChunkIndex::FromIndex(chunk_idx)), 0u) + << chunk_idx; } TEST(SharedRingBufferTest, PublishReuseReclaim) { // Publish, reuse and reclaim are exact-value transitions. test::SharedRingBufferForTesting ring(4, kChunkSize); - ASSERT_EQ(ring->TryReserveWritePos().position, 0u); + ASSERT_EQ(ring->TryReserveWritePos().write_pos, 0u); ASSERT_TRUE(ring->TryAcquireChunkForWriting(0, BeingWrittenWord(kWriterA))); uint32_t observed = BeingWrittenWord(kWriterA); - ASSERT_TRUE( - ring->TryReleaseChunkAsComplete(0, CompleteWord(kWriterA, 2), &observed)); - EXPECT_EQ(PeekStateWord(ring.get(), 0), CompleteWord(kWriterA, 2)); + ASSERT_TRUE(ring->TryReleaseChunkAsComplete( + ChunkIndex::FromIndex(0), CompleteWord(kWriterA, 2), &observed)); + EXPECT_EQ(PeekStateWord(ring.get(), ChunkIndex::FromIndex(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->TryReleaseChunkAsComplete(0, CompleteWord(kWriterA, 3), &stale)); + EXPECT_FALSE(ring->TryReleaseChunkAsComplete( + ChunkIndex::FromIndex(0), CompleteWord(kWriterA, 3), &stale)); 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)), + ASSERT_TRUE(ring->TryReacquireChunkForWriting(ChunkIndex::FromIndex(0), + CompleteWord(kWriterA, 2))); + EXPECT_EQ(ChunkStateOf(PeekStateWord(ring.get(), ChunkIndex::FromIndex(0))), ChunkState::kBeingWritten); - EXPECT_EQ(NumFragmentsOf(PeekStateWord(ring.get(), 0)), 2u); + EXPECT_EQ(NumFragmentsOf(PeekStateWord(ring.get(), ChunkIndex::FromIndex(0))), + 2u); observed = ReplaceChunkState(CompleteWord(kWriterA, 2), ChunkState::kBeingWritten); - ASSERT_TRUE( - ring->TryReleaseChunkAsComplete(0, CompleteWord(kWriterA, 5), &observed)); + ASSERT_TRUE(ring->TryReleaseChunkAsComplete( + ChunkIndex::FromIndex(0), CompleteWord(kWriterA, 5), &observed)); // The reader consumes it and stamps the wrap for the *next* traversal of this - // chunk, taken from the position it just resolved. + // chunk, taken from the position it just consumed. observed = CompleteWord(kWriterA, 5); ASSERT_TRUE(ring->TryReleaseCompleteChunkAsFree(0, &observed)); - EXPECT_EQ(PeekStateWord(ring.get(), 0), MakeFreeStateWord(1)); + EXPECT_EQ(PeekStateWord(ring.get(), ChunkIndex::FromIndex(0)), + MakeFreeStateWord(1)); } TEST(SharedRingBufferTest, RequestRewriteKeepsFields) { @@ -538,13 +561,15 @@ // through the publish/reuse pair. uint32_t observed = being_written; ASSERT_TRUE(ring->TryReleaseChunkAsComplete( - 0, ReplaceChunkState(published, ChunkState::kComplete), &observed)); + ChunkIndex::FromIndex(0), + ReplaceChunkState(published, ChunkState::kComplete), &observed)); ASSERT_TRUE(ring->TryReacquireChunkForWriting( - 0, ReplaceChunkState(published, ChunkState::kComplete))); + ChunkIndex::FromIndex(0), + ReplaceChunkState(published, ChunkState::kComplete))); observed = published; - ASSERT_TRUE(ring->TryRequestRewrite(0, &observed)); - const uint32_t marked = PeekStateWord(ring.get(), 0); + ASSERT_TRUE(ring->TryRequestRewrite(ChunkIndex::FromIndex(0), &observed)); + const uint32_t marked = PeekStateWord(ring.get(), ChunkIndex::FromIndex(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. @@ -561,20 +586,22 @@ ASSERT_TRUE(ring->TryAcquireChunkForWriting(0, BeingWrittenWord(kWriterA))); uint32_t observed = BeingWrittenWord(kWriterA); - ASSERT_TRUE(ring->TryRequestRewrite(0, &observed)); + ASSERT_TRUE(ring->TryRequestRewrite(ChunkIndex::FromIndex(0), &observed)); const uint32_t marked = ReplaceChunkState(BeingWrittenWord(kWriterA), ChunkState::kRewriteRequested); - EXPECT_EQ(PeekStateWord(ring.get(), 0), marked); + EXPECT_EQ(PeekStateWord(ring.get(), ChunkIndex::FromIndex(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); + ASSERT_TRUE(ring->TryAcknowledgeRewrite(ChunkIndex::FromIndex(0), marked)); + EXPECT_EQ(PeekStateWord(ring.get(), ChunkIndex::FromIndex(0)), + kRewriteAcknowledgedStateWord); // Only the reader turns that into a free word, and it stamps the wrap of the - // position it is resolving. + // position it is consuming. ASSERT_TRUE(ring->TryReleaseRewriteAcknowledgedChunkAsFree(4, &observed)); - EXPECT_EQ(PeekStateWord(ring.get(), 0), MakeFreeStateWord(2)); + EXPECT_EQ(PeekStateWord(ring.get(), ChunkIndex::FromIndex(0)), + MakeFreeStateWord(2)); } // The reclaim compares against the exact rewrite-acknowledgment word. A failed @@ -586,17 +613,19 @@ // 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; - Internals::SetChunkStateWord(ring.get(), 0, forged); + Internals::SetChunkStateWord(ring.get(), ChunkIndex::FromIndex(0), forged); uint32_t observed = 0; EXPECT_FALSE(ring->TryReleaseRewriteAcknowledgedChunkAsFree(0, &observed)); EXPECT_EQ(observed, forged); - EXPECT_EQ(PeekStateWord(ring.get(), 0), forged); + EXPECT_EQ(PeekStateWord(ring.get(), ChunkIndex::FromIndex(0)), forged); // The exact word goes through. - Internals::SetChunkStateWord(ring.get(), 0, kRewriteAcknowledgedStateWord); + Internals::SetChunkStateWord(ring.get(), ChunkIndex::FromIndex(0), + kRewriteAcknowledgedStateWord); EXPECT_TRUE(ring->TryReleaseRewriteAcknowledgedChunkAsFree(0, &observed)); - EXPECT_EQ(PeekStateWord(ring.get(), 0), MakeFreeStateWord(1)); + EXPECT_EQ(PeekStateWord(ring.get(), ChunkIndex::FromIndex(0)), + MakeFreeStateWord(1)); } // Every word that says "this chunk is claimable" comes out of exactly three @@ -607,24 +636,31 @@ ASSERT_TRUE(ring->TryAcquireChunkForWriting(0, BeingWrittenWord(kWriterA))); uint32_t observed = BeingWrittenWord(kWriterA); - ASSERT_TRUE( - ring->TryReleaseChunkAsComplete(0, CompleteWord(kWriterA, 1), &observed)); - EXPECT_NE(ChunkStateOf(PeekStateWord(ring.get(), 0)), ChunkState::kFree); + ASSERT_TRUE(ring->TryReleaseChunkAsComplete( + ChunkIndex::FromIndex(0), CompleteWord(kWriterA, 1), &observed)); + EXPECT_NE(ChunkStateOf(PeekStateWord(ring.get(), ChunkIndex::FromIndex(0))), + ChunkState::kFree); - ASSERT_TRUE(ring->TryReacquireChunkForWriting(0, CompleteWord(kWriterA, 1))); - EXPECT_NE(ChunkStateOf(PeekStateWord(ring.get(), 0)), ChunkState::kFree); + ASSERT_TRUE(ring->TryReacquireChunkForWriting(ChunkIndex::FromIndex(0), + CompleteWord(kWriterA, 1))); + EXPECT_NE(ChunkStateOf(PeekStateWord(ring.get(), ChunkIndex::FromIndex(0))), + ChunkState::kFree); observed = ReplaceChunkState(CompleteWord(kWriterA, 1), ChunkState::kBeingWritten); - ASSERT_TRUE(ring->TryRequestRewrite(0, &observed)); + ASSERT_TRUE(ring->TryRequestRewrite(ChunkIndex::FromIndex(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); + ChunkIndex::FromIndex(0), + ReplaceChunkState(CompleteWord(kWriterA, 1), + ChunkState::kRewriteRequested))); + EXPECT_NE(ChunkStateOf(PeekStateWord(ring.get(), ChunkIndex::FromIndex(0))), + ChunkState::kFree); + EXPECT_EQ(PeekStateWord(ring.get(), ChunkIndex::FromIndex(0)), + kRewriteAcknowledgedStateWord); ASSERT_TRUE(ring->TryReleaseRewriteAcknowledgedChunkAsFree(0, &observed)); - EXPECT_EQ(ChunkStateOf(PeekStateWord(ring.get(), 0)), ChunkState::kFree); + EXPECT_EQ(ChunkStateOf(PeekStateWord(ring.get(), ChunkIndex::FromIndex(0))), + ChunkState::kFree); } TEST(SharedRingBufferTest, ReclaimAcrossPositionRollover) { @@ -634,22 +670,24 @@ // the exact stamped word, derived from the position. test::SharedRingBufferForTesting ring(16, kChunkSize); - const uint32_t kLastLap = 0xfffffff0u; // chunk 0, the last lap's wrap - ASSERT_EQ(ChunkIndexOf(kLastLap, 16), 0u); - ASSERT_EQ(WrapCountForPosition(kLastLap, ring->num_chunks()), 0xffffu); + const uint32_t kLastLapPos = 0xfffffff0u; // chunk 0, the last lap's wrap + ASSERT_EQ(ChunkIndex::FromPosition(kLastLapPos, 16).value(), 0u); + ASSERT_EQ(WrapCountForPosition(kLastLapPos, ring->num_chunks()), 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->TryRequestRewrite(ChunkIndex::FromIndex(0), &observed)); ASSERT_TRUE(ring->TryAcknowledgeRewrite( - 0, ReplaceChunkState(BeingWrittenWord(kWriterA), - ChunkState::kRewriteRequested))); + ChunkIndex::FromIndex(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, + ring->TryReleaseRewriteAcknowledgedChunkAsFree(kLastLapPos, &observed)); + EXPECT_EQ(PeekStateWord(ring.get(), ChunkIndex::FromIndex(0)), + MakeFreeStateWord(0)); + // The writer holding position 0, the one that follows kLastLapPos on chunk 0, // is exactly the one that can claim it. EXPECT_TRUE(ring->TryAcquireChunkForWriting(0, BeingWrittenWord(kWriterA))); } @@ -660,19 +698,21 @@ // the position, not from incrementing the chunk's value. test::SharedRingBufferForTesting ring(4, kChunkSize); - const uint32_t kLastLap = 0xffffu * 4; // chunk 0, wrap 0xffff - ASSERT_EQ(ChunkIndexOf(kLastLap, 4), 0u); - ASSERT_EQ(WrapCountForPosition(kLastLap, ring->num_chunks()), 0xffffu); + const uint32_t kLastLapPos = 0xffffu * 4; // chunk 0, wrap 0xffff + ASSERT_EQ(ChunkIndex::FromPosition(kLastLapPos, 4).value(), 0u); + ASSERT_EQ(WrapCountForPosition(kLastLapPos, ring->num_chunks()), 0xffffu); - Internals::SetPositions(ring.get(), kLastLap); - ASSERT_EQ(PeekStateWord(ring.get(), 0), MakeFreeStateWord(0xffff)); + Internals::SetPositions(ring.get(), kLastLapPos); + ASSERT_EQ(PeekStateWord(ring.get(), ChunkIndex::FromIndex(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, + ASSERT_TRUE(ring->TryMoveFreeChunkToNextWrap(kLastLapPos, &observed)); + EXPECT_EQ(PeekStateWord(ring.get(), ChunkIndex::FromIndex(0)), + MakeFreeStateWord(0)); + // The next traversal of chunk 0 belongs to position kLastLapPos + 4, whose + // wrap is the truncated zero. + EXPECT_TRUE(ring->TryAcquireChunkForWriting(kLastLapPos + 4, BeingWrittenWord(kWriterB))); } @@ -686,38 +726,39 @@ // --------------------------------------------------------------------------- // Race 1: a delayed writer's claim against the reader's advance of an -// unclaimed position. Both compare against Free(wrap(position)). +// unclaimed position. Both compare against Free(wrap(chunk_pos)). // -// Positions 0..7 on a two-chunk ring cover both chunks and four wrap counts. +// Positions 0..7 on a two-chunk ring buffer cover both chunks and four wrap +// counts. TEST(SharedRingBufferTest, UnclaimedAdvanceLosesToClaim) { constexpr uint32_t kNumChunks = 2; test::SharedRingBufferForTesting ring(kNumChunks, kChunkSize); - for (uint32_t position = 0; position < 8; ++position) { - SCOPED_TRACE(position); - const uint32_t chunk_idx = ChunkIndexOf(position, kNumChunks); + for (uint32_t chunk_pos = 0; chunk_pos < 8; ++chunk_pos) { + SCOPED_TRACE(chunk_pos); + const auto chunk_idx = ChunkIndex::FromPosition(chunk_pos, kNumChunks); // The reader loaded the Free word before the claim landed. uint32_t observed = ring->LoadChunkStateWord(chunk_idx); - ASSERT_EQ(observed, MakeFreeStateWordForPosition(position, kNumChunks)); + ASSERT_EQ(observed, MakeFreeStateWordForPosition(chunk_pos, kNumChunks)); ASSERT_TRUE( - ring->TryAcquireChunkForWriting(position, BeingWrittenWord(kWriterA))); + ring->TryAcquireChunkForWriting(chunk_pos, BeingWrittenWord(kWriterA))); // The advance fails and receives the claimed word, so the reader retries // the position through the scrape path. - EXPECT_FALSE(ring->TryMoveFreeChunkToNextWrap(position, &observed)); + EXPECT_FALSE(ring->TryMoveFreeChunkToNextWrap(chunk_pos, &observed)); EXPECT_EQ(observed, BeingWrittenWord(kWriterA)); EXPECT_EQ(PeekStateWord(ring.get(), chunk_idx), BeingWrittenWord(kWriterA)); - // Publish and consume, so the chunk is Free for position + 2 by the time + // Publish and consume, so the chunk is Free for chunk_pos + 2 by the time // the loop comes back to it. uint32_t expected = BeingWrittenWord(kWriterA); ASSERT_TRUE(ring->TryReleaseChunkAsComplete( chunk_idx, CompleteWord(kWriterA, 0), &expected)); expected = CompleteWord(kWriterA, 0); - ASSERT_TRUE(ring->TryReleaseCompleteChunkAsFree(position, &expected)); + ASSERT_TRUE(ring->TryReleaseCompleteChunkAsFree(chunk_pos, &expected)); EXPECT_EQ(PeekStateWord(ring.get(), chunk_idx), - MakeFreeStateWordForPosition(position + kNumChunks, kNumChunks)); + MakeFreeStateWordForPosition(chunk_pos + kNumChunks, kNumChunks)); } } @@ -725,21 +766,21 @@ constexpr uint32_t kNumChunks = 2; test::SharedRingBufferForTesting ring(kNumChunks, kChunkSize); - for (uint32_t position = 0; position < 8; ++position) { - SCOPED_TRACE(position); - const uint32_t chunk_idx = ChunkIndexOf(position, kNumChunks); + for (uint32_t chunk_pos = 0; chunk_pos < 8; ++chunk_pos) { + SCOPED_TRACE(chunk_pos); + const auto chunk_idx = ChunkIndex::FromPosition(chunk_pos, kNumChunks); uint32_t observed = ring->LoadChunkStateWord(chunk_idx); - ASSERT_TRUE(ring->TryMoveFreeChunkToNextWrap(position, &observed)); + ASSERT_TRUE(ring->TryMoveFreeChunkToNextWrap(chunk_pos, &observed)); const uint32_t next_wrap = - MakeFreeStateWordForPosition(position + kNumChunks, kNumChunks); + MakeFreeStateWordForPosition(chunk_pos + kNumChunks, kNumChunks); EXPECT_EQ(PeekStateWord(ring.get(), chunk_idx), next_wrap); // The position's writer has spent its one claim attempt, and no other // writer can adopt the position either. The word stays with the next lap. EXPECT_FALSE( - ring->TryAcquireChunkForWriting(position, BeingWrittenWord(kWriterA))); + ring->TryAcquireChunkForWriting(chunk_pos, BeingWrittenWord(kWriterA))); EXPECT_FALSE( - ring->TryAcquireChunkForWriting(position, BeingWrittenWord(kWriterB))); + ring->TryAcquireChunkForWriting(chunk_pos, BeingWrittenWord(kWriterB))); EXPECT_EQ(PeekStateWord(ring.get(), chunk_idx), next_wrap); } } @@ -751,39 +792,41 @@ test::SharedRingBufferForTesting ring(2, kChunkSize); ASSERT_TRUE(ring->TryAcquireChunkForWriting(0, BeingWrittenWord(kWriterA))); // The reader loaded BeingWritten(0) before the publication landed. - uint32_t observed = ring->LoadChunkStateWord(0); + uint32_t observed = ring->LoadChunkStateWord(ChunkIndex::FromIndex(0)); ASSERT_EQ(observed, BeingWrittenWord(kWriterA)); uint32_t expected = BeingWrittenWord(kWriterA); - ASSERT_TRUE( - ring->TryReleaseChunkAsComplete(0, CompleteWord(kWriterA, 1), &expected)); + ASSERT_TRUE(ring->TryReleaseChunkAsComplete( + ChunkIndex::FromIndex(0), CompleteWord(kWriterA, 1), &expected)); // The scrape fails and receives the Complete word, so the reader discards // its speculative copy of the prefix and retries the position. - EXPECT_FALSE(ring->TryRequestRewrite(0, &observed)); + EXPECT_FALSE(ring->TryRequestRewrite(ChunkIndex::FromIndex(0), &observed)); EXPECT_EQ(observed, CompleteWord(kWriterA, 1)); - EXPECT_EQ(PeekStateWord(ring.get(), 0), CompleteWord(kWriterA, 1)); + EXPECT_EQ(PeekStateWord(ring.get(), ChunkIndex::FromIndex(0)), + CompleteWord(kWriterA, 1)); } TEST(SharedRingBufferTest, PublicationLosesToScrape) { test::SharedRingBufferForTesting ring(2, kChunkSize); ASSERT_TRUE(ring->TryAcquireChunkForWriting(0, BeingWrittenWord(kWriterA))); uint32_t observed = BeingWrittenWord(kWriterA); - ASSERT_TRUE(ring->TryRequestRewrite(0, &observed)); + ASSERT_TRUE(ring->TryRequestRewrite(ChunkIndex::FromIndex(0), &observed)); // The publication fails and receives the rewrite request. Its fragment count // says which prefix the reader took: nothing here. uint32_t expected = BeingWrittenWord(kWriterA); - EXPECT_FALSE( - ring->TryReleaseChunkAsComplete(0, CompleteWord(kWriterA, 1), &expected)); + EXPECT_FALSE(ring->TryReleaseChunkAsComplete( + ChunkIndex::FromIndex(0), CompleteWord(kWriterA, 1), &expected)); EXPECT_EQ(ChunkStateOf(expected), ChunkState::kRewriteRequested); EXPECT_EQ(WriterIDOf(expected), kWriterA); EXPECT_EQ(NumFragmentsOf(expected), 0u); // The writer moves its suffix elsewhere and lets go of the chunk, saying // nothing about who gets it next. - EXPECT_TRUE(ring->TryAcknowledgeRewrite(0, expected)); - EXPECT_EQ(PeekStateWord(ring.get(), 0), kRewriteAcknowledgedStateWord); + EXPECT_TRUE(ring->TryAcknowledgeRewrite(ChunkIndex::FromIndex(0), expected)); + EXPECT_EQ(PeekStateWord(ring.get(), ChunkIndex::FromIndex(0)), + kRewriteAcknowledgedStateWord); } // Race 3: a writer taking its cached chunk back against the reader consuming @@ -791,30 +834,36 @@ TEST(SharedRingBufferTest, ReclaimLosesToReuse) { test::SharedRingBufferForTesting ring(2, kChunkSize); - Internals::SetChunkStateWord(ring.get(), 0, CompleteWord(kWriterA, 1)); + Internals::SetChunkStateWord(ring.get(), ChunkIndex::FromIndex(0), + CompleteWord(kWriterA, 1)); // The reader loaded Complete(1) before the reuse landed. - uint32_t observed = ring->LoadChunkStateWord(0); - ASSERT_TRUE(ring->TryReacquireChunkForWriting(0, CompleteWord(kWriterA, 1))); + uint32_t observed = ring->LoadChunkStateWord(ChunkIndex::FromIndex(0)); + ASSERT_TRUE(ring->TryReacquireChunkForWriting(ChunkIndex::FromIndex(0), + CompleteWord(kWriterA, 1))); // The reclaim fails and receives BeingWritten(1): the reader discards its // copy and handles the position through the scrape path on its next look. EXPECT_FALSE(ring->TryReleaseCompleteChunkAsFree(0, &observed)); EXPECT_EQ(observed, ReplaceChunkState(CompleteWord(kWriterA, 1), ChunkState::kBeingWritten)); - EXPECT_EQ(PeekStateWord(ring.get(), 0), observed); + EXPECT_EQ(PeekStateWord(ring.get(), ChunkIndex::FromIndex(0)), observed); } TEST(SharedRingBufferTest, ReuseLosesToReclaim) { test::SharedRingBufferForTesting ring(2, kChunkSize); - Internals::SetChunkStateWord(ring.get(), 0, CompleteWord(kWriterA, 1)); + Internals::SetChunkStateWord(ring.get(), ChunkIndex::FromIndex(0), + CompleteWord(kWriterA, 1)); uint32_t observed = CompleteWord(kWriterA, 1); ASSERT_TRUE(ring->TryReleaseCompleteChunkAsFree(0, &observed)); // Position 0 was consumed, so the chunk is Free for position 2. - EXPECT_EQ(PeekStateWord(ring.get(), 0), MakeFreeStateWord(1)); + EXPECT_EQ(PeekStateWord(ring.get(), ChunkIndex::FromIndex(0)), + MakeFreeStateWord(1)); // The reuse fails; the writer drops its cached handle and the word stays. - EXPECT_FALSE(ring->TryReacquireChunkForWriting(0, CompleteWord(kWriterA, 1))); - EXPECT_EQ(PeekStateWord(ring.get(), 0), MakeFreeStateWord(1)); + EXPECT_FALSE(ring->TryReacquireChunkForWriting(ChunkIndex::FromIndex(0), + CompleteWord(kWriterA, 1))); + EXPECT_EQ(PeekStateWord(ring.get(), ChunkIndex::FromIndex(0)), + MakeFreeStateWord(1)); } // --------------------------------------------------------------------------- @@ -869,10 +918,10 @@ GTEST_SKIP() << "The futex wait is not available on this platform"; test::SharedRingBufferForTesting ring(2, kChunkSize); - const uint32_t stale_sample = Internals::GetReadPos(ring.get()); + const uint32_t stale_read_pos = Internals::GetReadPos(ring.get()); ring->PublishReadPos(5); - const auto outcome = ring->WaitForReadPosChange(stale_sample, 30000); + const auto outcome = ring->WaitForReadPosChange(stale_read_pos, 30000); EXPECT_EQ(outcome, WriterWaitResult::kRetry); EXPECT_EQ(Internals::GetNumWritersWaiting(ring.get()), 0u); } @@ -959,25 +1008,25 @@ test::SharedRingBufferForTesting ring(2, kChunkSize); for (uint32_t round = 0; round < kRounds; ++round) { - const uint32_t sample = Internals::GetReadPos(ring.get()); + const uint32_t sampled_read_pos = Internals::GetReadPos(ring.get()); SharedRingBuffer::WriterWaitResult outcome = WriterWaitResult::kUnavailable; const auto progress_deadline = base::GetWallTimeMs() + base::TimeMillis(5000); std::atomic<bool> completed{false}; std::thread writer([&] { - outcome = ring->WaitForReadPosChange(sample, 30000); + outcome = ring->WaitForReadPosChange(sampled_read_pos, 30000); completed.store(true); }); const bool registered = SpinUntil( [&] { return Internals::GetNumWritersWaiting(ring.get()) != 0; }, progress_deadline); - ring->PublishReadPos(sample + 1); + ring->PublishReadPos(sampled_read_pos + 1); const bool completed_promptly = SpinUntil([&] { return completed.load(); }, progress_deadline); // Cleanup precedes assertions, including when registration was not seen. if (!completed_promptly) - ring->PublishReadPos(sample + 1); + ring->PublishReadPos(sampled_read_pos + 1); writer.join(); ASSERT_TRUE(registered) << round;
diff --git a/src/tracing/v2/shared_ring_buffer_writer.cc b/src/tracing/v2/shared_ring_buffer_writer.cc index 4056971..c2e07ed 100644 --- a/src/tracing/v2/shared_ring_buffer_writer.cc +++ b/src/tracing/v2/shared_ring_buffer_writer.cc
@@ -55,7 +55,7 @@ max_fragment_size_(MaxFragmentSizeForEmptyChunk(chunk_size_)) { PERFETTO_CHECK(delegate_); // WriterIDs are nonzero, at most kMaxWriterID, and identify this writer - // until all positions reserved under the id have been resolved. + // until all positions reserved under the id have been consumed. PERFETTO_DCHECK(writer_id_ != 0 && writer_id_ <= kMaxWriterID); } @@ -117,14 +117,14 @@ PERFETTO_DCHECK(has_open_fragment()); PERFETTO_DCHECK(cur_chunk_); PERFETTO_DCHECK(cur_chunk_state() == ChunkState::kBeingWritten); - // |size| must still fit in the range BeginFragment() handed out, together - // with the size varint that encodes it. + // BeginFragment() left room for both the payload and its size entry. + // Subtracting the fragment's start gives that complete available range. PERFETTO_DCHECK(size <= MaxFragmentSizeForAvailableBytes( sizes_begin_ - cur_fragment_begin_)); // Nothing becomes visible until ReleaseCurrentChunkAsComplete(). uint8_t* sizes_begin = cur_chunk_ + sizes_begin_; - sizes_begin = WriteFragmentSize(sizes_begin, size); + sizes_begin = WriteFragmentSizeReversed(sizes_begin, size); sizes_begin_ = static_cast<uint32_t>(sizes_begin - cur_chunk_); payload_end_ = cur_fragment_begin_ + size; ++num_fragments_; @@ -189,13 +189,13 @@ // Each round reserves a position and then claims its chunk. // - // 1. If the ring is full, no position is reserved; apply the exhaustion - // policy below. + // 1. A full ring buffer prevents reservation. Apply the exhaustion policy. // 2. If both reservation and claim succeed, return the chunk. - // 3. A failed claim leaves a hole; reserve a later position instead. + // 3. A reservation succeeds, but its physical chunk is not Free for that + // traversal. The position is now a hole, so try a later reservation. // - // Stop when the ring reports full or after num_chunks failed claims. The - // latter bounds attempts, not physical chunks visited: other writers can + // Stop when the ring buffer reports full or after num_chunks failed claims. + // The latter bounds attempts, not physical chunks visited: other writers can // take intervening positions, so repeated attempts may hit the same pinned // chunk. uint32_t num_failed_claims = 0; @@ -205,10 +205,10 @@ if (reservation.result == SharedRingBuffer::ReserveResult::kReserved) { // Happy case: the reserved position's chunk is Free for this traversal. - if (ring_->TryAcquireChunkForWriting(reservation.position, + if (ring_->TryAcquireChunkForWriting(reservation.write_pos, being_written_word)) { - cur_chunk_idx_ = - ChunkIndexOf(reservation.position, ring_->num_chunks()); + cur_chunk_idx_ = ChunkIndex::FromPosition(reservation.write_pos, + ring_->num_chunks()); cur_chunk_ = ring_->chunk_at(cur_chunk_idx_); expected_state_word_ = being_written_word; payload_end_ = kTargetBufferPayloadOffset; @@ -219,18 +219,20 @@ return BeginFragmentResult::kSuccess; } - // This reservation is now a hole. Never retry it against a different - // Free word; reserve a later position instead. + // This position is now a hole. A later Free word belongs to another + // traversal, so this writer must leave this position behind and reserve + // a new one. ++stats_.failed_claims; saw_unclaimable_chunk = true; if (++num_failed_claims < ring_->num_chunks()) continue; } - // The reader resolves holes and moves read_pos, so notify it whenever + // The reader consumes holes and moves read_pos, so notify it whenever // holes were created or this writer is about to wait. The count can be - // below num_chunks here: a later reservation can find the ring full after - // only some failed claims, and those holes still need the notification. + // below num_chunks here: a later reservation can find the ring buffer full + // after only some failed claims, and those holes still need the + // notification. if (num_failed_claims != 0 || policy != BufferExhaustedPolicy::kDrop) { delegate_->NotifyReader(); num_failed_claims = 0; @@ -238,7 +240,8 @@ // Classify the exhaustion. // Preserve a failed claim across waits, even if the last reservation - // found the ring full: this acquisition has already left holes behind. + // found the ring buffer full: this acquisition has already left holes + // behind. const BeginFragmentResult exhausted_result = saw_unclaimable_chunk ? BeginFragmentResult::kNoChunkAvailable : BeginFragmentResult::kFull; @@ -351,7 +354,7 @@ 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", - cur_chunk_idx_, writer_id_, expected); + cur_chunk_idx_.value(), writer_id_, expected); } const uint32_t taken = NumFragmentsOf(expected); @@ -373,7 +376,7 @@ PERFETTO_FATAL( "tracing v2: writer %u could not acknowledge chunk %u; only its " "owner may leave RewriteRequested", - writer_id_, cur_chunk_idx_); + writer_id_, cur_chunk_idx_.value()); } ++stats_.relocations; @@ -411,7 +414,7 @@ } payload_end_ = kTargetBufferPayloadOffset + *suffix_size; uint8_t* sizes_begin = - WriteFragmentSize(cur_chunk_ + chunk_size_, *suffix_size); + WriteFragmentSizeReversed(cur_chunk_ + chunk_size_, *suffix_size); sizes_begin_ = static_cast<uint32_t>(sizes_begin - cur_chunk_); num_fragments_ = 1; // Round again to publish the replacement, which the reader may also scrape. @@ -420,7 +423,7 @@ void SharedRingBufferWriter::ResetCurrentChunk() { cur_chunk_ = nullptr; - cur_chunk_idx_ = 0; + cur_chunk_idx_ = ChunkIndex::FromIndex(0); expected_state_word_ = 0; payload_end_ = 0; sizes_begin_ = 0;
diff --git a/src/tracing/v2/shared_ring_buffer_writer.h b/src/tracing/v2/shared_ring_buffer_writer.h index 4dd4e2e..911e72b 100644 --- a/src/tracing/v2/shared_ring_buffer_writer.h +++ b/src/tracing/v2/shared_ring_buffer_writer.h
@@ -36,12 +36,12 @@ // boundary. A chunk can hold several fragments. The writer handles their // byte ranges and sizes without interpreting their contents. // - Use each instance from one thread at a time. Several instances can write -// to the same ring concurrently. -// - The ring's memory and SharedRingBuffer view must outlive every writer. +// to the same ring buffer concurrently. +// - Every writer must be destroyed before the ring buffer's memory and view. // The destructor still publishes the chunk the writer holds. // - Reserving a write position and claiming its physical chunk are separate // operations. A failed claim leaves a position that only the reader can -// resolve. +// consume. // - The writer notifies its delegate before waiting for the reader to make // space. class SharedRingBufferWriter { @@ -49,12 +49,13 @@ // Result of BeginFragment(), which may need a new chunk. enum class BeginFragmentResult { kSuccess, - // 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. + // The ring buffer 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. + // pinned by a stalled writer produce this without the ring buffer being + // full. // // The reader has been notified before this is returned. kNoChunkAvailable, @@ -122,9 +123,10 @@ // the reader scraped the chunk meanwhile, the unpublished suffix moves to a // new chunk, which applies the buffer-exhaustion policy and may wait. // - // |continues_on_next| says that this fragment continues in this writer's - // next chunk. A chunk published with that flag is never reused, so a prefix - // scraped from BeingWritten always ends on a packet boundary. + // The ring buffer writer sees fragments, not packets. Its caller therefore + // sets |continues_on_next| when the packet continues in this writer's next + // chunk. A chunk carrying the flag is never reused, so a prefix scraped from + // BeingWritten always ends on a packet boundary. EndFragmentResult EndFragment(uint32_t size, bool continues_on_next); // Publishes whatever is held and lets go of the chunk. Any open fragment is @@ -167,7 +169,7 @@ EndFragmentResult ReleaseCurrentChunkAsComplete( std::optional<uint32_t> suffix_size, bool continues_on_next); - // Clears only this writer's cached chunk state. It does not modify the ring. + // Clears this writer's cached chunk state, leaving shared memory untouched. void ResetCurrentChunk(); SharedRingBuffer* const ring_; @@ -181,7 +183,7 @@ // State cached for the chunk this writer currently owns. uint8_t* cur_chunk_ = nullptr; - uint32_t cur_chunk_idx_ = 0; + ChunkIndex cur_chunk_idx_ = ChunkIndex::FromIndex(0); // The exact word this writer's next compare-and-swap expects. It can differ // from the shared word once the reader has requested a rewrite. uint32_t expected_state_word_ = 0;
diff --git a/src/tracing/v2/shared_ring_buffer_writer_unittest.cc b/src/tracing/v2/shared_ring_buffer_writer_unittest.cc index 97fe469..30d2b4f 100644 --- a/src/tracing/v2/shared_ring_buffer_writer_unittest.cc +++ b/src/tracing/v2/shared_ring_buffer_writer_unittest.cc
@@ -61,7 +61,8 @@ void NotifyReader() override { ++num_notifications; const uint32_t read_pos = Internals::GetReadPos(ring_); - const uint32_t chunk_idx = ChunkIndexOf(read_pos, ring_->num_chunks()); + const auto chunk_idx = + ChunkIndex::FromPosition(read_pos, ring_->num_chunks()); uint32_t observed = ring_->LoadChunkStateWord(chunk_idx); if (ChunkStateOf(observed) == ChunkState::kRewriteAcknowledged) { ASSERT_TRUE( @@ -90,7 +91,7 @@ std::vector<std::string> fragments; }; -DecodedChunk Decode(SharedRingBuffer* ring, uint32_t chunk_idx) { +DecodedChunk Decode(SharedRingBuffer* ring, ChunkIndex chunk_idx) { const uint8_t* chunk = ring->chunk_at(chunk_idx); const uint32_t word = ring->LoadChunkStateWord(chunk_idx); DecodedChunk decoded; @@ -105,22 +106,21 @@ const uint8_t* sizes_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, - &sizes_cursor, &size); - EXPECT_TRUE(valid); - if (!valid) + const auto size = ReadFragmentSizeReversed( + chunk + kTargetBufferPayloadOffset, &sizes_cursor); + EXPECT_TRUE(size); + if (!size) return decoded; decoded.fragments.emplace_back( - reinterpret_cast<const char*>(chunk + offset), size); - offset += size; + reinterpret_cast<const char*>(chunk + offset), *size); + offset += *size; } return decoded; } // 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_idx) { +uint32_t MarkForRewrite(SharedRingBuffer* ring, ChunkIndex chunk_idx) { uint32_t observed = ring->LoadChunkStateWord(chunk_idx); EXPECT_EQ(ChunkStateOf(observed), ChunkState::kBeingWritten); const uint32_t taken = NumFragmentsOf(observed); @@ -142,13 +142,13 @@ ASSERT_TRUE(WriteFragment(&writer, std::string(3, 'c'))); // A 256-byte target-buffer chunk, byte for byte. - const uint8_t* chunk = ring->chunk_at(0); + const uint8_t* chunk = ring->chunk_at(ChunkIndex::FromIndex(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); + const DecodedChunk decoded = Decode(ring.get(), ChunkIndex::FromIndex(0)); EXPECT_EQ(decoded.state, ChunkState::kComplete); EXPECT_EQ(decoded.writer_id, kWriterA); EXPECT_EQ(decoded.target_buffer, kBuffer); @@ -175,8 +175,9 @@ // 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) + for (uint32_t chunk_idx = 0; chunk_idx < ring->num_chunks(); ++chunk_idx) { + for (const std::string& fragment : + Decode(ring.get(), ChunkIndex::FromIndex(chunk_idx)).fragments) seen.push_back(fragment); } EXPECT_EQ(seen, expected); @@ -195,8 +196,9 @@ memset(range.begin, 'z', kLargest); ASSERT_EQ(writer.EndFragment(kLargest, false), EndFragmentResult::kSuccess); - ASSERT_EQ(Decode(ring.get(), 0).fragments.size(), 1u); - EXPECT_EQ(Decode(ring.get(), 0).fragments[0].size(), kLargest); + ASSERT_EQ(Decode(ring.get(), ChunkIndex::FromIndex(0)).fragments.size(), 1u); + EXPECT_EQ(Decode(ring.get(), ChunkIndex::FromIndex(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. @@ -213,10 +215,10 @@ for (uint32_t i = 0; i < kMaxFragmentsPerChunk; ++i) ASSERT_TRUE(WriteFragment(&writer, "")) << i; - const DecodedChunk first = Decode(ring.get(), 0); + const DecodedChunk first = Decode(ring.get(), ChunkIndex::FromIndex(0)); EXPECT_EQ(first.fragments.size(), kMaxFragmentsPerChunk); ASSERT_TRUE(WriteFragment(&writer, "x")); - const DecodedChunk second = Decode(ring.get(), 1); + const DecodedChunk second = Decode(ring.get(), ChunkIndex::FromIndex(1)); ASSERT_EQ(second.fragments.size(), 1u); EXPECT_EQ(second.fragments[0], "x"); } @@ -237,7 +239,7 @@ memset(range.begin, 'a', kLargest); ASSERT_EQ(writer.EndFragment(kLargest, false), EndFragmentResult::kSuccess); - const DecodedChunk decoded = Decode(ring.get(), 0); + const DecodedChunk decoded = Decode(ring.get(), ChunkIndex::FromIndex(0)); ASSERT_EQ(decoded.fragments.size(), 1u); EXPECT_EQ(decoded.fragments[0], std::string(kLargest, 'a')); } @@ -262,13 +264,14 @@ const bool kept_chunk = residual == 2; EXPECT_EQ(ring->LoadWritePos(), kept_chunk ? 1u : 2u); - const DecodedChunk first = Decode(ring.get(), 0); + const DecodedChunk first = Decode(ring.get(), ChunkIndex::FromIndex(0)); ASSERT_EQ(first.fragments.size(), kept_chunk ? 2u : 1u); EXPECT_EQ(first.fragments[0].size(), first_size); if (kept_chunk) { EXPECT_TRUE(first.fragments[1].empty()); } else { - EXPECT_EQ(Decode(ring.get(), 1).fragments, std::vector<std::string>{""}); + EXPECT_EQ(Decode(ring.get(), ChunkIndex::FromIndex(1)).fragments, + std::vector<std::string>{""}); } } } @@ -285,9 +288,10 @@ ASSERT_TRUE(WriteFragment(&writer, std::string(kLargest, 'z'))); ASSERT_TRUE(WriteFragment(&writer, "next")); - 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(Decode(ring.get(), ChunkIndex::FromIndex(0)).fragments.size(), 1u); + EXPECT_EQ(Decode(ring.get(), ChunkIndex::FromIndex(0)).fragments[0].size(), + kLargest); + const DecodedChunk second = Decode(ring.get(), ChunkIndex::FromIndex(1)); ASSERT_EQ(second.fragments.size(), 1u); EXPECT_EQ(second.fragments[0], "next"); } @@ -307,11 +311,12 @@ // 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); + const DecodedChunk decoded = Decode(ring.get(), ChunkIndex::FromIndex(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); + EXPECT_EQ(ChunkStateOf(ring->LoadChunkStateWord(ChunkIndex::FromIndex(1))), + ChunkState::kFree); } TEST(SharedRingBufferWriterTest, ReuseLosesToReclaim) { @@ -322,7 +327,7 @@ ASSERT_TRUE(WriteFragment(&writer, "one")); // The reader consumes the Complete chunk before the writer takes it back. - uint32_t observed = ring->LoadChunkStateWord(0); + uint32_t observed = ring->LoadChunkStateWord(ChunkIndex::FromIndex(0)); ASSERT_EQ(ChunkStateOf(observed), ChunkState::kComplete); ASSERT_TRUE(ring->TryReleaseCompleteChunkAsFree(0, &observed)); ring->PublishReadPos(1); @@ -330,8 +335,9 @@ // The writer's reuse fails; it drops its handle and goes for a fresh chunk // rather than writing behind the reader. ASSERT_TRUE(WriteFragment(&writer, "two")); - EXPECT_EQ(ChunkStateOf(ring->LoadChunkStateWord(0)), ChunkState::kFree); - const DecodedChunk decoded = Decode(ring.get(), 1); + EXPECT_EQ(ChunkStateOf(ring->LoadChunkStateWord(ChunkIndex::FromIndex(0))), + ChunkState::kFree); + const DecodedChunk decoded = Decode(ring.get(), ChunkIndex::FromIndex(1)); ASSERT_EQ(decoded.fragments.size(), 1u); EXPECT_EQ(decoded.fragments[0], "two"); } @@ -348,12 +354,12 @@ ASSERT_TRUE(WriteFragment(&writer, "tail", /*continues_from_prev=*/true, /*continues_on_next=*/false)); - const DecodedChunk first = Decode(ring.get(), 0); + const DecodedChunk first = Decode(ring.get(), ChunkIndex::FromIndex(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); + const DecodedChunk second = Decode(ring.get(), ChunkIndex::FromIndex(1)); EXPECT_EQ(second.payload_flags, kFlagContinuesFromPrevChunk); ASSERT_EQ(second.fragments.size(), 1u); EXPECT_EQ(second.fragments[0], "tail"); @@ -366,11 +372,12 @@ writer.RecordDataLoss(); ASSERT_TRUE(WriteFragment(&writer, "after the gap")); - EXPECT_EQ(Decode(ring.get(), 0).payload_flags, kFlagDataLoss); + EXPECT_EQ(Decode(ring.get(), ChunkIndex::FromIndex(0)).payload_flags, + kFlagDataLoss); // The chunk after it describes no gap of its own. ASSERT_TRUE(WriteFragment(&writer, std::string(500, 'x'))); - EXPECT_EQ(Decode(ring.get(), 1).payload_flags, 0u); + EXPECT_EQ(Decode(ring.get(), ChunkIndex::FromIndex(1)).payload_flags, 0u); } // A loss recorded while the writer still holds a reusable Complete chunk must @@ -387,18 +394,18 @@ // The pre-loss fragment stays alone in chunk 0. The post-loss fragment opened // a new position, and its chunk is the one reporting the gap. EXPECT_EQ(ring->LoadWritePos(), 2u); - const DecodedChunk before = Decode(ring.get(), 0); + const DecodedChunk before = Decode(ring.get(), ChunkIndex::FromIndex(0)); ASSERT_EQ(before.fragments.size(), 1u); EXPECT_EQ(before.fragments[0], "before"); EXPECT_EQ(before.payload_flags, 0u); - const DecodedChunk after = Decode(ring.get(), 1); + const DecodedChunk after = Decode(ring.get(), ChunkIndex::FromIndex(1)); ASSERT_EQ(after.fragments.size(), 1u); EXPECT_EQ(after.fragments[0], "after"); EXPECT_EQ(after.payload_flags, kFlagDataLoss); // The gap is reported exactly once: the chunk after it carries no flag. ASSERT_TRUE(WriteFragment(&writer, std::string(500, 'x'))); - EXPECT_EQ(Decode(ring.get(), 2).payload_flags, 0u); + EXPECT_EQ(Decode(ring.get(), ChunkIndex::FromIndex(2)).payload_flags, 0u); } // --------------------------------------------------------------------------- @@ -406,24 +413,24 @@ // --------------------------------------------------------------------------- TEST(SharedRingBufferWriterTest, DropPolicyReportsFull) { - // kDrop reports a full ring without blocking. + // kDrop reports a full ring buffer without blocking. test::SharedRingBufferForTesting ring(2, 256); SharedRingBufferWriter a = MakeWriter(ring.get(), kWriterA, kBuffer); SharedRingBufferWriter b = MakeWriter(ring.get(), kWriterB, kBuffer); - // Two writers hold both chunks, so the ring is structurally full. + // Two writers hold both chunks, so the ring buffer is structurally full. ASSERT_EQ(a.BeginFragment(1, false).result, BeginFragmentResult::kSuccess); ASSERT_EQ(b.BeginFragment(1, false).result, BeginFragmentResult::kSuccess); SharedRingBufferWriter c = MakeWriter(ring.get(), 11, kBuffer); EXPECT_EQ(c.BeginFragment(1, false).result, BeginFragmentResult::kFull); - // Nothing was reserved, so a full ring costs no position. + // Nothing was reserved, so a full ring buffer costs no position. EXPECT_EQ(ring->LoadWritePos(), 2u); EXPECT_EQ(c.GetStats().failed_claims, 0u); } TEST(SharedRingBufferWriterTest, NotifiesReaderBeforeWaiting) { - // A full ring notifies the reader before the writer waits. + // A full ring buffer notifies the reader before the writer waits. if (!SharedRingBuffer::SupportsWriterWait()) GTEST_SKIP() << "The futex wait is not available on this platform"; test::SharedRingBufferForTesting ring(1, 256); @@ -461,12 +468,13 @@ EXPECT_EQ(second.BeginFragment(1, false).result, BeginFragmentResult::kFull); EXPECT_EQ(delegate.num_notifications, 0u); - uint32_t observed = ring->LoadChunkStateWord(0); + uint32_t observed = ring->LoadChunkStateWord(ChunkIndex::FromIndex(0)); ASSERT_TRUE(ring->TryReleaseCompleteChunkAsFree(0, &observed)); ring->PublishReadPos(1); ASSERT_TRUE(WriteFragment(&second, "after loss")); - EXPECT_EQ(Decode(ring.get(), 0).payload_flags, kFlagDataLoss); + EXPECT_EQ(Decode(ring.get(), ChunkIndex::FromIndex(0)).payload_flags, + kFlagDataLoss); ASSERT_EQ(second.FinishCurrentChunk(), EndFragmentResult::kSuccess); // Publishing the loss ends the drop episode. The next exhausted acquisition @@ -477,7 +485,8 @@ } // 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. +// a full ring buffer: positions are available, but their chunks cannot be +// acquired. TEST(SharedRingBufferWriterTest, PinnedChunks) { // The writer gives up after num_chunks failed claims. Nobody else reserves // here, so those claims land on each physical chunk once and use up the whole @@ -487,12 +496,14 @@ // 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) { + for (uint32_t chunk_pos = 0; chunk_pos < ring->num_chunks(); ++chunk_pos) { const uint32_t being_written = MakeDataStateWord( ChunkState::kBeingWritten, ChunkFormat::kTargetBuffer, 0, 0, kWriterB); - ASSERT_TRUE(ring->TryAcquireChunkForWriting(i, being_written)); + ASSERT_TRUE(ring->TryAcquireChunkForWriting(chunk_pos, being_written)); + const auto chunk_idx = + ChunkIndex::FromPosition(chunk_pos, ring->num_chunks()); uint32_t observed = being_written; - ASSERT_TRUE(ring->TryRequestRewrite(i, &observed)); + ASSERT_TRUE(ring->TryRequestRewrite(chunk_idx, &observed)); } ASSERT_EQ(ring->LoadWritePos(), 0u); @@ -526,8 +537,9 @@ const auto range = writer.BeginFragment(6, false); ASSERT_EQ(range.result, BeginFragmentResult::kSuccess); memcpy(range.begin, "suffix", 6); - EXPECT_EQ(MarkForRewrite(ring.get(), 0), prefix.size()); - EXPECT_EQ(Decode(ring.get(), 0).fragments, prefix); + EXPECT_EQ(MarkForRewrite(ring.get(), ChunkIndex::FromIndex(0)), + prefix.size()); + EXPECT_EQ(Decode(ring.get(), ChunkIndex::FromIndex(0)).fragments, prefix); ASSERT_EQ(writer.EndFragment(6, false), EndFragmentResult::kSuccess); EXPECT_EQ(writer.GetStats().relocations, 1u); @@ -535,9 +547,9 @@ // 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)), + EXPECT_EQ(ChunkStateOf(ring->LoadChunkStateWord(ChunkIndex::FromIndex(0))), ChunkState::kRewriteAcknowledged); - const DecodedChunk replacement = Decode(ring.get(), 1); + const DecodedChunk replacement = Decode(ring.get(), ChunkIndex::FromIndex(1)); EXPECT_EQ(replacement.state, ChunkState::kComplete); EXPECT_EQ(replacement.writer_id, kWriterA); EXPECT_EQ(replacement.target_buffer, kBuffer); @@ -551,13 +563,14 @@ const auto again = writer.BeginFragment(5, false); ASSERT_EQ(again.result, BeginFragmentResult::kSuccess); memcpy(again.begin, "again", 5); - EXPECT_EQ(MarkForRewrite(ring.get(), 1), 1u); + EXPECT_EQ(MarkForRewrite(ring.get(), ChunkIndex::FromIndex(1)), 1u); ASSERT_EQ(writer.EndFragment(5, false), EndFragmentResult::kSuccess); EXPECT_EQ(writer.GetStats().relocations, 2u); EXPECT_EQ(writer.GetStats().fragments_dropped, 0u); - EXPECT_EQ(ring->LoadChunkStateWord(1), kRewriteAcknowledgedStateWord); - const DecodedChunk second = Decode(ring.get(), 2); + EXPECT_EQ(ring->LoadChunkStateWord(ChunkIndex::FromIndex(1)), + kRewriteAcknowledgedStateWord); + const DecodedChunk second = Decode(ring.get(), ChunkIndex::FromIndex(2)); EXPECT_EQ(second.state, ChunkState::kComplete); ASSERT_EQ(second.fragments.size(), 1u); EXPECT_EQ(second.fragments[0], "again"); @@ -574,11 +587,11 @@ ASSERT_EQ(range.result, BeginFragmentResult::kSuccess); memcpy(range.begin, "tail", 4); // The reader takes nothing: the writer has published no fragment yet. - EXPECT_EQ(MarkForRewrite(ring.get(), 0), 0u); + EXPECT_EQ(MarkForRewrite(ring.get(), ChunkIndex::FromIndex(0)), 0u); ASSERT_EQ(writer.EndFragment(4, false), EndFragmentResult::kSuccess); - const DecodedChunk replacement = Decode(ring.get(), 1); + const DecodedChunk replacement = Decode(ring.get(), ChunkIndex::FromIndex(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 @@ -599,12 +612,15 @@ } ASSERT_EQ(writer.BeginFragment(0, false).result, BeginFragmentResult::kSuccess); - EXPECT_EQ(MarkForRewrite(ring.get(), 0), has_prefix ? 1u : 0u); + EXPECT_EQ(MarkForRewrite(ring.get(), ChunkIndex::FromIndex(0)), + has_prefix ? 1u : 0u); ASSERT_EQ(writer.EndFragment(0, false), EndFragmentResult::kSuccess); - EXPECT_EQ(ring->LoadChunkStateWord(0), kRewriteAcknowledgedStateWord); + EXPECT_EQ(ring->LoadChunkStateWord(ChunkIndex::FromIndex(0)), + kRewriteAcknowledgedStateWord); EXPECT_EQ(ring->LoadWritePos(), 2u); - const DecodedChunk replacement = Decode(ring.get(), 1); + const DecodedChunk replacement = + Decode(ring.get(), ChunkIndex::FromIndex(1)); EXPECT_EQ(replacement.state, ChunkState::kComplete); ASSERT_EQ(replacement.fragments.size(), 1u); EXPECT_TRUE(replacement.fragments[0].empty()); @@ -631,13 +647,13 @@ ASSERT_EQ(range.result, BeginFragmentResult::kSuccess); memcpy(range.begin, "tail", 4); - // Occupy the ring's only other chunk so no replacement can be had. + // Occupy the ring buffer's only other chunk so no replacement can be had. SharedRingBufferWriter blocker = MakeWriter(ring.get(), kWriterB, kBuffer); ASSERT_EQ(blocker.BeginFragment(1, false).result, BeginFragmentResult::kSuccess); // The reader takes nothing, so the flag moves with the one-fragment suffix. - EXPECT_EQ(MarkForRewrite(ring.get(), 0), 0u); + EXPECT_EQ(MarkForRewrite(ring.get(), ChunkIndex::FromIndex(0)), 0u); // A stall would notify the delegate, which frees the acknowledged chunk and // lets the relocation succeed. A drop asks nobody and gives the suffix up. @@ -645,14 +661,16 @@ EndFragmentResult::kRelocationDropped); EXPECT_EQ(delegate.num_notifications, 0u); EXPECT_EQ(writer.GetStats().fragments_dropped, 1u); - EXPECT_EQ(ring->LoadChunkStateWord(0), kRewriteAcknowledgedStateWord); + EXPECT_EQ(ring->LoadChunkStateWord(ChunkIndex::FromIndex(0)), + kRewriteAcknowledgedStateWord); // The loss is still unreported, so it goes out with the next chunk. uint32_t observed = 0; ASSERT_TRUE(ring->TryReleaseRewriteAcknowledgedChunkAsFree(0, &observed)); ring->PublishReadPos(1); ASSERT_TRUE(WriteFragment(&writer, "after loss")); - EXPECT_EQ(Decode(ring.get(), 0).payload_flags, kFlagDataLoss); + EXPECT_EQ(Decode(ring.get(), ChunkIndex::FromIndex(0)).payload_flags, + kFlagDataLoss); EXPECT_EQ(delegate.num_notifications, 0u); } @@ -665,14 +683,15 @@ // Take the chunk back but add nothing, then let the reader scrape it. ASSERT_EQ(writer.BeginFragment(1, false).result, BeginFragmentResult::kSuccess); - EXPECT_EQ(MarkForRewrite(ring.get(), 0), 1u); + EXPECT_EQ(MarkForRewrite(ring.get(), ChunkIndex::FromIndex(0)), 1u); // Releasing abandons the open fragment; there is nothing left to move. EXPECT_EQ(writer.FinishCurrentChunk(), EndFragmentResult::kSuccess); EXPECT_EQ(writer.GetStats().fragments_dropped, 0u); - EXPECT_EQ(ChunkStateOf(ring->LoadChunkStateWord(0)), + EXPECT_EQ(ChunkStateOf(ring->LoadChunkStateWord(ChunkIndex::FromIndex(0))), ChunkState::kRewriteAcknowledged); - EXPECT_EQ(ChunkStateOf(ring->LoadChunkStateWord(1)), ChunkState::kFree); + EXPECT_EQ(ChunkStateOf(ring->LoadChunkStateWord(ChunkIndex::FromIndex(1))), + ChunkState::kFree); } TEST(SharedRingBufferWriterTest, EmptySuffixKeepsDataLossPending) { @@ -684,20 +703,21 @@ ASSERT_EQ(writer.BeginFragment(4, false).result, BeginFragmentResult::kSuccess); - EXPECT_EQ(MarkForRewrite(ring.get(), 0), 0u); + EXPECT_EQ(MarkForRewrite(ring.get(), ChunkIndex::FromIndex(0)), 0u); // Releasing abandons the open fragment, so no suffix can carry the flag. EXPECT_EQ(writer.FinishCurrentChunk(), EndFragmentResult::kSuccess); - EXPECT_EQ(ChunkStateOf(ring->LoadChunkStateWord(0)), + EXPECT_EQ(ChunkStateOf(ring->LoadChunkStateWord(ChunkIndex::FromIndex(0))), ChunkState::kRewriteAcknowledged); ASSERT_TRUE(WriteFragment(&writer, "next")); - EXPECT_EQ(Decode(ring.get(), 1).payload_flags, kFlagDataLoss); + EXPECT_EQ(Decode(ring.get(), ChunkIndex::FromIndex(1)).payload_flags, + kFlagDataLoss); } // 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. +// other order would leave the old chunk occupied exactly when the ring buffer +// is full, so every later traversal of it would burn a position. TEST(SharedRingBufferWriterTest, RelocationDrop) { for (const std::string& suffix : {std::string("lost"), std::string()}) { SCOPED_TRACE(suffix); @@ -706,7 +726,7 @@ ASSERT_TRUE(WriteFragment(&writer, "published")); - // Occupy the ring's only other chunk so no replacement can be had. + // Occupy the ring buffer's only other chunk so no replacement can be had. SharedRingBufferWriter blocker = MakeWriter(ring.get(), kWriterB, kBuffer); ASSERT_EQ(blocker.BeginFragment(1, false).result, BeginFragmentResult::kSuccess); @@ -715,7 +735,7 @@ ASSERT_EQ(range.result, BeginFragmentResult::kSuccess); if (!suffix.empty()) memcpy(range.begin, suffix.data(), suffix.size()); - EXPECT_EQ(MarkForRewrite(ring.get(), 0), 1u); + EXPECT_EQ(MarkForRewrite(ring.get(), ChunkIndex::FromIndex(0)), 1u); EXPECT_EQ(writer.EndFragment(static_cast<uint32_t>(suffix.size()), false), EndFragmentResult::kRelocationDropped); @@ -723,7 +743,8 @@ EXPECT_EQ(writer.GetStats().relocations, 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); + EXPECT_EQ(ring->LoadChunkStateWord(ChunkIndex::FromIndex(0)), + kRewriteAcknowledgedStateWord); // Another failed acquisition must leave the loss pending. EXPECT_EQ(writer.BeginFragment(1, false).result, @@ -735,7 +756,8 @@ ASSERT_TRUE(ring->TryReleaseRewriteAcknowledgedChunkAsFree(0, &observed)); ring->PublishReadPos(1); ASSERT_TRUE(WriteFragment(&writer, "next")); - EXPECT_EQ(Decode(ring.get(), 0).payload_flags, kFlagDataLoss); + EXPECT_EQ(Decode(ring.get(), ChunkIndex::FromIndex(0)).payload_flags, + kFlagDataLoss); EXPECT_EQ(writer.GetStats().fragments_dropped, 1u); } } @@ -753,9 +775,9 @@ ASSERT_EQ(writer.FinishCurrentChunk(), EndFragmentResult::kSuccess); ASSERT_TRUE(WriteFragment(&writer, "after")); EXPECT_EQ(ring->LoadWritePos(), 2u); - EXPECT_EQ(Decode(ring.get(), 0).fragments, + EXPECT_EQ(Decode(ring.get(), ChunkIndex::FromIndex(0)).fragments, std::vector<std::string>{"before"}); - EXPECT_EQ(Decode(ring.get(), 1).fragments, + EXPECT_EQ(Decode(ring.get(), ChunkIndex::FromIndex(1)).fragments, std::vector<std::string>{"after"}); } } @@ -774,7 +796,7 @@ ASSERT_EQ(writer.BeginFragment(4, false).result, BeginFragmentResult::kSuccess); } - const DecodedChunk decoded = Decode(ring.get(), 0); + const DecodedChunk decoded = Decode(ring.get(), ChunkIndex::FromIndex(0)); EXPECT_EQ(decoded.state, ChunkState::kComplete); ASSERT_EQ(decoded.fragments.size(), 1u); EXPECT_EQ(decoded.fragments[0], "kept");