Rewrite SamplingCircularQueue

src/circular-queue-inl.h

 // Copyright 2011 the V8 project authors. All rights reserved.
 // Redistribution and use in source and binary forms, with or without
 // modification, are permitted provided that the following conditions are
 // met:
 //
 //     * Redistributions of source code must retain the above copyright
 //       notice, this list of conditions and the following disclaimer.
 //     * Redistributions in binary form must reproduce the above
 //       copyright notice, this list of conditions and the following
 //       disclaimer in the documentation and/or other materials provided
@@ skipping 15 matching lines @@
 // OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 
 #ifndef V8_CIRCULAR_QUEUE_INL_H_
 #define V8_CIRCULAR_QUEUE_INL_H_
 
 #include "circular-queue.h"
 
 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.

+
+
 void SamplingCircularQueue::WrapPositionIfNeeded(
     SamplingCircularQueue::Cell** pos) {
   if (*pos == buffer_ + buffer_size_) *pos = buffer_;
 }
 
 
 } }  // namespace v8::internal
 
 #endif  // V8_CIRCULAR_QUEUE_INL_H_