PS: works
diff --git a/src/tracing/v2/shared_ring_buffer.cc b/src/tracing/v2/shared_ring_buffer.cc index 5ecdf7f..6889bfa 100644 --- a/src/tracing/v2/shared_ring_buffer.cc +++ b/src/tracing/v2/shared_ring_buffer.cc
@@ -25,30 +25,6 @@ namespace perfetto { using ChunkHeader = SharedRingBuffer::ChunkHeader; -// TODO this mostly works %: -// - data loss auditing that needs to be improved (that's easy) -// - there is one logical bug in the protocol: the writer does this: -// 1. increment the wr_offset -// 2. try to acquire the chunk at that offset (there are 99.99% chances it -// will get it, it's rare it won't, ignore this case) -// When a writer acquires a chunk it does a CAS 0 -> (header with its writerId) -// Now the problem is in the reader: up until which point should it read? -// when does it stop? initially i thought naively "well you read up until the -// wr_off". But that open an interesting case, if a thread is stuck in between -// 1 and 2, the reader sees a "0" header at that offset so what does the reader -// do? if it stops conservatively, well now we are "stalling the reader", which -// means if one of the many writers is descheduled, the reader does read the -// chunks from other threads, and that is bad. -// if it just skips it, now it moves past ignoring that chunk. which is going to -// be filled up only later on. and it will go out of order for that thread. - -// TODO changes for tomorrow: -// - When the reader goes past a 0 (idle) chunk it must invalidate it, in a -// similar way as if the chunk was kAcquiredForWriting. -// - The writer has the resposnibility to set that chunk back to 0 when it -// sees the flag. -// - sort out data losses - // --- SharedRingBuffer --- SharedRingBuffer::SharedRingBuffer(void* start, size_t size) @@ -56,8 +32,8 @@ PERFETTO_CHECK(start != nullptr); PERFETTO_CHECK(size >= kRingBufferHeaderSize + kChunkSize); PERFETTO_CHECK(reinterpret_cast<uintptr_t>(start) % 8 == 0); - - num_chunks_ = (size - kRingBufferHeaderSize) / kChunkSize; + num_chunks_ = + static_cast<uint32_t>((size - kRingBufferHeaderSize) / kChunkSize); PERFETTO_CHECK(num_chunks_ > 0); } @@ -73,6 +49,7 @@ WriterID writer_id) : rb_(rb), + num_chunks_(rb->num_chunks()), writer_id_(writer_id), cached_header_(SharedRingBuffer::ChunkHeader::Pack(writer_id, 0, 0)) { // Generally speaking a 0 WriterID is not valid (As per IdAllocator semantic). @@ -253,9 +230,9 @@ // it falls back on the local bankruptcy chunk and marks a data loss. bool SharedRingBuffer_Writer::AcquireNewChunk(uint8_t extra_flags) { SharedRingBuffer::RingBufferHeader* rb_hdr = rb_->header(); - // TODO cache locally num_chunks to remove one pointer chasing. - const size_t num_chunks = rb_->num_chunks(); - const uint8_t flags = SharedRingBuffer::kFlagAcquiredForWriting | extra_flags; + + const uint8_t flags = + SharedRingBuffer::kFlagAcquiredForWriting | extra_flags | data_loss_; const uint32_t new_hdr = ChunkHeader::Pack(writer_id_, 0, flags); // This can be m-o-relaxed as the CAS below will be the authoritative check. @@ -264,7 +241,7 @@ for (;;) { rd_off = rb_hdr->rd_off.load(std::memory_order_acquire); - uint32_t next_wr_off = static_cast<uint32_t>((wr_off + 1) % num_chunks); + uint32_t next_wr_off = static_cast<uint32_t>((wr_off + 1) % num_chunks_); if (next_wr_off == rd_off) { // The buffer is full. Go to the epilogue and return the bankruptcy chunk. @@ -318,6 +295,7 @@ last_chunk_ = chunk; cached_header_ = new_hdr; write_off_ = 0; + data_loss_ = 0; return true; } // for(;;) @@ -329,6 +307,15 @@ // TODO mark data loss somewhere in local state. if we switch back to a normal // chunk we should invalidate somehow the first fragment. + // This will cause the next chunk to be marked with kFlagDataLoss to signal + // that we lost the data before it. + data_loss_ = SharedRingBuffer::kFlagDataLoss; + + // However we might not be able to acquire a next chunk if the system is + // catastrophically overloaded. For this reason we also increment a global + // counter of data losses. + // TODO(primiano): maybe use a bitmap instead? think about how we are going + // to make any use of this. rb_->IncrementDataLosses(); // Initialize bankruptcy_chunk_ with a proper header. @@ -523,24 +510,24 @@ (flags & SharedRingBuffer::kFlagContinuesFromPrevChunk) != 0; const bool continues_on_next = (flags & SharedRingBuffer::kFlagContinuesOnNextChunk) != 0; - const bool data_loss = (flags & SharedRingBuffer::kFlagDataLoss) != 0; // Get or create writer state. WriterState& ws = writer_states_[writer_id]; - // If there was data loss and this chunk continues from a previous one, - // the pending data is corrupted - discard it. - if (data_loss && continues_from_prev) { + // There are 3 cases of data loss: + // 1. The writer failed to acquire some chunk in the past, and on the next + // chunk it managed to obtain it marked the kFlagDataLoss flag. + // 2. The chunk is marked as a continuation but we have not seen any prior + // fragment in previous chunks. + // 3. The chunk is NOT marked as a continuation, but we have accumulated + // fragments (We missed the ending fragment) + bool data_loss = false; + if ((flags & SharedRingBuffer::kFlagDataLoss) != 0 || + (continues_from_prev && ws.pending_data.empty()) || + (!continues_from_prev && !ws.pending_data.empty())) { + data_loss = true; ws.pending_data.clear(); - } - - // If this chunk does NOT continue from prev, but we have pending data, - // the previous message's continuation chain was broken (e.g., a chunk was - // skipped due to NeedsRewrite and later rewritten elsewhere). Discard the - // incomplete message as data loss. - if (!continues_from_prev && !ws.pending_data.empty()) { - // TODO mark data loss. - ws.pending_data.clear(); + ++ws.data_losses; } for (size_t off = 0; off < payload_size;) { @@ -562,17 +549,12 @@ // Case 1: First fragment and continues_from_prev - append to pending data. if (is_first_frag && continues_from_prev) { - // If pending_data is empty, we missed the start of this message - // (e.g., the starting chunk was skipped). This is data loss - skip - // this continuation fragment entirely. - if (ws.pending_data.empty()) { - // TODO mark data loss. + if (data_loss) { continue; } - if (!data_loss) { - ws.pending_data.append(reinterpret_cast<const char*>(frag_data), - frag_size); - } + PERFETTO_DCHECK(!ws.pending_data.empty()); + ws.pending_data.append(reinterpret_cast<const char*>(frag_data), + frag_size); // If this is not the last fragment, or if there's no continuation, // the message is complete. if (!is_last_frag || !continues_on_next) { @@ -599,4 +581,9 @@ } } +uint32_t SharedRingBuffer_Reader::GetDataLossesForWriter(WriterID writer_id) { + WriterState* ws = writer_states_.Find(writer_id); + return ws ? ws->data_losses : 0; +} + } // namespace perfetto
diff --git a/src/tracing/v2/shared_ring_buffer.h b/src/tracing/v2/shared_ring_buffer.h index b61c0fc..91ee717 100644 --- a/src/tracing/v2/shared_ring_buffer.h +++ b/src/tracing/v2/shared_ring_buffer.h
@@ -108,7 +108,7 @@ uint8_t* start() const { return start_; } size_t size() const { return size_; } - size_t num_chunks() const { return num_chunks_; } + uint32_t num_chunks() const { return num_chunks_; } bool is_valid() const { return start_ != nullptr && num_chunks_ > 0; } RingBufferHeader* header() { @@ -141,7 +141,7 @@ uint8_t* start_ = nullptr; size_t size_ = 0; - size_t num_chunks_ = 0; + uint32_t num_chunks_ = 0; }; // Writer for the SharedRingBuffer. Move-only. @@ -285,7 +285,12 @@ bool AcquireNewChunk(uint8_t extra_flags); SharedRingBuffer* rb_ = nullptr; + uint32_t num_chunks_ = 0; // == rb_->num_chunk(), cached for performance. WriterID writer_id_ = 0; + + // This is effectively a boolean that is either 0 or kFlagDataLoss. + uint8_t data_loss_ = 0; + uint8_t* last_chunk_ = reinterpret_cast<uint8_t*>(&invalid_chunk_header_); uint32_t cached_header_ = 0; @@ -348,9 +353,12 @@ void ClearCompletedMessages() { completed_messages_.clear(); } + uint32_t GetDataLossesForWriter(WriterID); + private: struct WriterState { std::string pending_data; + uint32_t data_losses = 0; }; void ProcessChunkPayload(const uint8_t* payload,
diff --git a/src/tracing/v2/shared_ring_buffer_unittest.cc b/src/tracing/v2/shared_ring_buffer_unittest.cc index ca5ac80..3c58669 100644 --- a/src/tracing/v2/shared_ring_buffer_unittest.cc +++ b/src/tracing/v2/shared_ring_buffer_unittest.cc
@@ -137,15 +137,14 @@ // Multiple writers write variable-length messages with sequence numbers. // A reader on the main thread verifies monotonicity and payload integrity. TEST(SharedRingBufferTest, MultiThreadedStressTest) { - constexpr size_t kNumWriters = 1; // Single writer to isolate the issue. + constexpr size_t kNumWriters = 128; constexpr size_t kIterationsPerWriter = 100000; constexpr size_t kMinMsgSize = 4; // Minimum: just the sequence number. // Use small messages (< chunk payload 252) to avoid fragmentation for now. - constexpr size_t kMaxMsgSize = 200; + constexpr size_t kMaxMsgSize = 8192; constexpr uint32_t kMaxSleepUs = 100; - // Allocate a large ring buffer (1MB). - constexpr size_t kNumChunks = 4096; + constexpr size_t kNumChunks = 512; constexpr size_t kBufSize = SharedRingBuffer::kRingBufferHeaderSize + kNumChunks * kChunkSize; std::unique_ptr<uint8_t[]> buf(new uint8_t[kBufSize]()); @@ -213,7 +212,7 @@ std::vector<int64_t> last_seq(kNumWriters, -1); // Track message counts and data losses per writer. std::vector<uint64_t> msg_counts(kNumWriters, 0); - std::vector<uint64_t> data_losses(kNumWriters, 0); + std::vector<uint64_t> gaps_detected(kNumWriters, 0); std::vector<uint64_t> payload_errors(kNumWriters, 0); std::vector<uint64_t> reorderings(kNumWriters, 0); std::vector<uint64_t> bad_writer_ids(1, 0); @@ -262,7 +261,7 @@ // Check for gaps (data loss). if (last >= 0 && static_cast<int64_t>(seq) != last + 1) { uint64_t lost = seq - static_cast<uint32_t>(last) - 1; - data_losses[writer_idx] += lost; + gaps_detected[writer_idx] += lost; } last_seq[writer_idx] = static_cast<int64_t>(seq); @@ -322,14 +321,15 @@ for (size_t i = 0; i < kNumWriters; ++i) { WriterID wid = static_cast<WriterID>(i + 1); printf( - "Writer %u: received=%lu, lost=%lu, reorderings=%lu, " + "Writer %u: received=%lu, gaps=%lu, data_losses=%u, reorderings=%lu, " "payload_errors=%lu\n", wid, static_cast<unsigned long>(msg_counts[i]), - static_cast<unsigned long>(data_losses[i]), + static_cast<unsigned long>(gaps_detected[i]), + reader.GetDataLossesForWriter(wid), static_cast<unsigned long>(reorderings[i]), static_cast<unsigned long>(payload_errors[i])); total_received += msg_counts[i]; - total_lost += data_losses[i]; + total_lost += gaps_detected[i]; total_reorderings += reorderings[i]; total_payload_errors += payload_errors[i]; }