1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
|
// 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
// with the distribution.
// * Neither the name of Google Inc. nor the names of its
// contributors may be used to endorse or promote products derived
// from this software without specific prior written permission.
//
// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
// LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
// A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
// OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
// LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
// DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
// THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
#ifndef V8_CIRCULAR_QUEUE_H_
#define V8_CIRCULAR_QUEUE_H_
namespace v8 {
namespace internal {
// Lock-based blocking circular queue for small records. Intended for
// transfer of small records between a single producer and a single
// consumer. Blocks on enqueue operation if the queue is full.
template<typename Record>
class CircularQueue {
public:
inline explicit CircularQueue(int desired_buffer_size_in_bytes);
inline ~CircularQueue();
INLINE(void Dequeue(Record* rec));
INLINE(void Enqueue(const Record& rec));
INLINE(bool IsEmpty()) { return enqueue_pos_ == dequeue_pos_; }
private:
INLINE(Record* Next(Record* curr));
Record* buffer_;
Record* const buffer_end_;
Semaphore* enqueue_semaphore_;
Record* enqueue_pos_;
Record* dequeue_pos_;
DISALLOW_COPY_AND_ASSIGN(CircularQueue);
};
// 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
// 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 {
public:
// Executed on the application thread.
SamplingCircularQueue(int record_size_in_bytes,
int desired_chunk_size_in_bytes,
int buffer_size_in_chunks);
~SamplingCircularQueue();
// Executed on the producer (sampler) or application thread.
void SetUpProducer();
// Enqueue returns a pointer to a memory location for storing the next
// record.
INLINE(void* Enqueue());
void TearDownProducer();
// Executed on the consumer (analyzer) thread.
void SetUpConsumer();
// 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();
void TearDownConsumer();
typedef AtomicWord Cell;
// Reserved values for the first cell of a record.
static const Cell kClear = 0; // Marks clean (processed) chunks.
static const Cell kEnd = -1; // Marks the end of the buffer.
private:
struct ConsumerPosition {
Cell* dequeue_chunk_pos;
Cell* dequeue_chunk_poll_pos;
Cell* dequeue_pos;
Cell* dequeue_end_pos;
};
INLINE(void WrapPositionIfNeeded(Cell** pos));
const int record_size_;
const int chunk_size_in_bytes_;
const int chunk_size_;
const int buffer_size_;
const int producer_consumer_distance_;
Cell* buffer_;
// Store producer and consumer data in TLS to avoid modifying the
// same CPU cache line from two threads simultaneously.
Thread::LocalStorageKey consumer_key_;
Thread::LocalStorageKey producer_key_;
};
} } // namespace v8::internal
#endif // V8_CIRCULAR_QUEUE_H_
|
