Rewrite SamplingCircularQueue

src/circular-queue-inl.h

Index: src/circular-queue-inl.h
diff --git a/src/circular-queue-inl.h b/src/circular-queue-inl.h
index b48070ab5d24a064e4f8a91a72020e80d8f02954..88257c340a1fc9393d6522b768b34d91ec0a2c94 100644
--- a/src/circular-queue-inl.h
+++ b/src/circular-queue-inl.h
@@ -33,21 +33,20 @@
 namespace v8 {
 namespace internal {
 
-
-void* SamplingCircularQueue::Enqueue() {
-  if (producer_pos_->enqueue_pos == producer_pos_->next_chunk_pos) {
-    if (producer_pos_->enqueue_pos == buffer_ + buffer_size_) {
-      producer_pos_->next_chunk_pos = buffer_;
-      producer_pos_->enqueue_pos = buffer_;
-    }
-    Acquire_Store(producer_pos_->next_chunk_pos, kEnqueueStarted);
+void* SamplingCircularQueue::StartEnqueue() {
+  MemoryBarrier();
+  if (Acquire_Load(producer_pos_->enqueue_pos) == kEmpty) {
     // Skip marker.
-    producer_pos_->enqueue_pos += 1;
-    producer_pos_->next_chunk_pos += chunk_size_;
+    return producer_pos_->enqueue_pos + 1;
   }
-  void* result = producer_pos_->enqueue_pos;
-  producer_pos_->enqueue_pos += record_size_;
-  return result;
+  return NULL;
+}
+
+
+void SamplingCircularQueue::FinishEnqueue() {
+  Release_Store(producer_pos_->enqueue_pos, kFull);
+  producer_pos_->enqueue_pos += entry_size_;
+  WrapPositionIfNeeded(&producer_pos_->enqueue_pos);
 }

[Benedikt Meurer, 2013/08/13 09:31:32]

This assumes that there is always only a single producer, but this is not always
the case, for example in Chrome on Linux with profiled web worker threads, and
maybe some other scenarios. We should get this right now while we're at it.

I'd suggest to use a tri-state scheme with kEmpty, kPreparing and kFull, and
turn the enqueue_pos into an AtomicWord. Then the code would look like this:

T* StartEnqueue() {
  Atomic32 curr = Acquire_Load(&enqueue_pos_);
  if (!Acquire_CompareAndSwap(&buffer_[curr + kRecordStateOffset], kEmpty,
kPreparing) != kEmpty) return NULL;
  Atomic32 next = (curr + kRecordSize) % (kRecordSize * N);
  ASSERT(Release_Load(&enqueue_pos_) == curr);
  Release_Store(&enqueue_pos_, next);
  return reinterpret_cast<T*>(&buffer_[curr]);
}

void FinishEnqueue(T* obj) {
  ASSERT(reinterpret_cast<Atomic32*>(obj) >= &buffer_[0] &&
              reinterpret_cast<Atomic32*>(obj) < &buffer_[kRecordSize * N]);
  Atomic32* state_ptr = reinterpret_cast<Atomic32*>(obj) + kRecordStateOffset;
  ASSERT(Release_Load(state_ptr) == kPreparing);
  Release_Store(state_ptr, kFull);
}

This way the objects are aligned on cache line boundaries, which is probably
better performancewise (guessing). If that turns out to be wrong, put the state
before the object again.