Rewrite SamplingCircularQueue

src/circular-queue.h

 // Copyright 2010 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 17 matching lines @@
 #ifndef V8_CIRCULAR_QUEUE_H_
 #define V8_CIRCULAR_QUEUE_H_
 
 namespace v8 {
 namespace internal {
 
 
 // Lock-free cache-friendly sampling circular queue for large
 // records. Intended for fast transfer of large records between a
 // single producer and a single consumer. If the queue is full,
-// previous unread records are overwritten. The queue is designed with
+// StartEnqueue will return NULL. The queue is designed with
 // a goal in mind to evade cache lines thrashing by preventing
 // simultaneous reads and writes to adjanced memory locations.
-//
-// IMPORTANT: as a producer never checks for chunks cleanness, it is
-// possible that it can catch up and overwrite a chunk that a consumer
-// is currently reading, resulting in a corrupt record being read.
 class SamplingCircularQueue {

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

While we're already cleaning up this class, let's turn it into a template with
template parameters for the record type T and the constant number N of records,
i.e.

template <typename T, unsigned N>
class SamplingCircularQueue {
...
};

[yurys, 2013/08/13 14:10:29]

Done.

  public:
   // Executed on the application thread.
   SamplingCircularQueue(size_t record_size_in_bytes,
-                        size_t desired_chunk_size_in_bytes,
-                        unsigned buffer_size_in_chunks);
+                        unsigned buffer_size_in_records);
   ~SamplingCircularQueue();
 
-  // Enqueue returns a pointer to a memory location for storing the next
-  // record.
-  INLINE(void* Enqueue());
+  // StartEnqueue returns a pointer to a memory location for storing the next
+  // record or NULL if all entries are full at the moment.
+  INLINE(void* StartEnqueue());
+  // Notifies the queue that the producer has complete writing data into the
+  // memory returned by StartEnqueue and it can be passed to the consumer.
+  INLINE(void FinishEnqueue());
 
   // Executed on the consumer (analyzer) thread.
   // StartDequeue returns a pointer to a memory location for retrieving
   // the next record. After the record had been read by a consumer,
   // FinishDequeue must be called. Until that moment, subsequent calls
   // to StartDequeue will return the same pointer.
   void* StartDequeue();
   void FinishDequeue();
-  // Due to a presence of slipping between the producer and the consumer,
-  // the queue must be notified whether producing has been finished in order
-  // to process remaining records from the buffer.
-  void FlushResidualRecords();
 
   typedef AtomicWord Cell;

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

Get rid of this typedef, it's just confusing. Also there's no need to use
AtomicWord, and Atomic32 is sufficient.

[yurys, 2013/08/13 14:10:29]

Done.

 
  private:
-  // Reserved values for the chunk marker (first Cell in each chunk).
+  // Reserved values for the entry marker (first Cell in each entry).
   enum {
-    kClear,          // Marks clean (processed) chunks.
-    kEnqueueStarted  // Marks chunks where enqueue started.
+    kEmpty,  // Marks clean (processed) entries.
+    kFull    // Marks entries already filled by the producer but not yet
+             // completely processed by the consumer.
   };
 
   struct ProducerPosition {
-    Cell* next_chunk_pos;
     Cell* enqueue_pos;
   };
   struct ConsumerPosition {
-    Cell* dequeue_chunk_pos;
-    Cell* dequeue_chunk_poll_pos;
     Cell* dequeue_pos;
-    Cell* dequeue_end_pos;
   };
 
-  INLINE(void WrapPositionIfNeeded(Cell** pos));
+  inline void WrapPositionIfNeeded(Cell** pos);
 
-  const size_t record_size_;
-  const size_t chunk_size_in_bytes_;
-  const size_t chunk_size_;
+  const size_t entry_size_;
   const size_t buffer_size_;
+  Cell* not_aligned_buffer_;
   Cell* buffer_;
   byte* positions_;
   ProducerPosition* producer_pos_;
   ConsumerPosition* consumer_pos_;

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

Let's also cleanup this mess. Assuming that the record size and number of
records is constant (see comment about templatizing), we should be able to
replace this with:

STATIC_ASSERT(kProcessorCacheLineSize % sizeof(Atomic32) == 0);
static const unsigned kRecordStateOffset = ROUND_UP(sizeof(T),
sizeof(Atomic32));
static const unsigned kRecordSize = ROUND_UP(kRecordStateOffset +
sizeof(Atomic32), kProcessorCacheLineSize) / sizeof(Atomic32);
Atomic32 buffer_[N * kRecordSize] ALIGNED_AT(kProcessorCacheLineSize);
// Allocate enqueue_pos_ and dequeue_pos_ to different cache lines.
Atomic32 enqueue_pos_ ALIGNED_AT(kProcessorCacheLineSize);
int dequeue_pos_ ALIGNED_AT(kProcessorCacheLineSize);

where ROUND_UP(size, alignment) is a macro version of RoundUp() and
ALIGNED_AT(x) is an appropriate macro that expands to
__attribute__((aligned(x))) for GCC, Clang & Co. or __declspec(align(x)) for
MSVC.

[yurys, 2013/08/13 14:10:29]

Done.

 
   DISALLOW_COPY_AND_ASSIGN(SamplingCircularQueue);
 };
 
 
 } }  // namespace v8::internal
 
 #endif  // V8_CIRCULAR_QUEUE_H_