-
Notifications
You must be signed in to change notification settings - Fork 478
/
series.go
259 lines (211 loc) · 5.97 KB
/
series.go
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
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
package wal
import (
"sync"
"github.com/prometheus/prometheus/pkg/intern"
"github.com/prometheus/prometheus/pkg/labels"
)
type memSeries struct {
sync.Mutex
ref uint64
lset labels.Labels
lastTs int64
// TODO(rfratto): this solution below isn't perfect, and there's still
// the possibility for a series to be deleted before it's
// completely gone from the WAL. Rather, we should have gc return
// a "should delete" map and be given a "deleted" map.
// If a series that is going to be marked for deletion is in the
// "deleted" map, then it should be deleted instead.
//
// The "deleted" map will be populated by the Truncate function.
// It will be cleared with every call to gc.
// willDelete marks a series as to be deleted on the next garbage
// collection. If it receives a write, willDelete is disabled.
willDelete bool
// Whether this series has samples waiting to be committed to the WAL
pendingCommit bool
}
func (s *memSeries) updateTs(ts int64) {
s.lastTs = ts
s.willDelete = false
s.pendingCommit = true
}
// seriesHashmap is a simple hashmap for memSeries by their label set. It is
// built on top of a regular hashmap and holds a slice of series to resolve
// hash collisions. Its methods require the hash to be submitted with it to
// avoid re-computations throughout the code.
//
// This code is copied from the Prometheus TSDB.
type seriesHashmap map[uint64][]*memSeries
func (m seriesHashmap) get(hash uint64, lset labels.Labels) *memSeries {
for _, s := range m[hash] {
if labels.Equal(s.lset, lset) {
return s
}
}
return nil
}
func (m seriesHashmap) set(hash uint64, s *memSeries) {
intern.InternLabels(intern.Global, s.lset)
l := m[hash]
for i, prev := range l {
if labels.Equal(prev.lset, s.lset) {
l[i] = s
return
}
}
m[hash] = append(l, s)
}
func (m seriesHashmap) del(hash uint64, ref uint64) {
var rem []*memSeries
for _, s := range m[hash] {
if s.ref != ref {
rem = append(rem, s)
} else {
intern.ReleaseLabels(intern.Global, s.lset)
}
}
if len(rem) == 0 {
delete(m, hash)
} else {
m[hash] = rem
}
}
const (
// defaultStripeSize is the default number of entries to allocate in the
// stripeSeries hash map.
defaultStripeSize = 1 << 14
)
// stripeSeries locks modulo ranges of IDs and hashes to reduce lock contention.
// The locks are padded to not be on the same cache line. Filling the padded space
// with the maps was profiled to be slower – likely due to the additional pointer
// dereferences.
//
// This code is copied from the Prometheus TSDB.
type stripeSeries struct {
size int
series []map[uint64]*memSeries
hashes []seriesHashmap
locks []stripeLock
}
type stripeLock struct {
sync.RWMutex
// Padding to avoid multiple locks being on the same cache line.
_ [40]byte
}
func newStripeSeries() *stripeSeries {
stripeSize := defaultStripeSize
s := &stripeSeries{
size: stripeSize,
series: make([]map[uint64]*memSeries, stripeSize),
hashes: make([]seriesHashmap, stripeSize),
locks: make([]stripeLock, stripeSize),
}
for i := range s.series {
s.series[i] = map[uint64]*memSeries{}
}
for i := range s.hashes {
s.hashes[i] = seriesHashmap{}
}
return s
}
// gc garbage collects old chunks that are strictly before mint and removes
// series entirely that have no chunks left.
func (s *stripeSeries) gc(mint int64) map[uint64]struct{} {
var (
deleted = map[uint64]struct{}{}
)
// Run through all series and find series that haven't been written to
// since mint. Mark those series as deleted and store their ID.
for i := 0; i < s.size; i++ {
s.locks[i].Lock()
for _, series := range s.series[i] {
series.Lock()
seriesHash := series.lset.Hash()
// If the series has received a write after mint, there's still
// data and it's not completely gone yet.
if series.lastTs >= mint || series.pendingCommit {
series.willDelete = false
series.Unlock()
continue
}
// The series hasn't received any data and *might* be gone, but
// we want to give it an opportunity to come back before marking
// it as deleted, so we wait one more GC cycle.
if !series.willDelete {
series.willDelete = true
series.Unlock()
continue
}
// The series is gone entirely. We'll need to delete the label
// hash (if one exists) so we'll obtain a lock for that too.
j := int(seriesHash) & (s.size - 1)
if i != j {
s.locks[j].Lock()
}
deleted[series.ref] = struct{}{}
delete(s.series[i], series.ref)
s.hashes[j].del(seriesHash, series.ref)
if i != j {
s.locks[j].Unlock()
}
series.Unlock()
}
s.locks[i].Unlock()
}
return deleted
}
func (s *stripeSeries) getByID(id uint64) *memSeries {
i := id & uint64(s.size-1)
s.locks[i].RLock()
series := s.series[i][id]
s.locks[i].RUnlock()
return series
}
func (s *stripeSeries) getByHash(hash uint64, lset labels.Labels) *memSeries {
i := hash & uint64(s.size-1)
s.locks[i].RLock()
series := s.hashes[i].get(hash, lset)
s.locks[i].RUnlock()
return series
}
func (s *stripeSeries) set(hash uint64, series *memSeries) {
i := hash & uint64(s.size-1)
s.locks[i].Lock()
s.hashes[i].set(hash, series)
s.locks[i].Unlock()
i = series.ref & uint64(s.size-1)
s.locks[i].Lock()
s.series[i][series.ref] = series
s.locks[i].Unlock()
}
func (s *stripeSeries) iterator() *stripeSeriesIterator {
return &stripeSeriesIterator{s}
}
// stripeSeriesIterator allows to iterate over series through a channel.
// The channel should always be completely consumed to not leak.
type stripeSeriesIterator struct {
s *stripeSeries
}
func (it *stripeSeriesIterator) Channel() <-chan *memSeries {
ret := make(chan *memSeries)
go func() {
for i := 0; i < it.s.size; i++ {
it.s.locks[i].RLock()
for _, series := range it.s.series[i] {
series.Lock()
j := int(series.lset.Hash()) & (it.s.size - 1)
if i != j {
it.s.locks[j].RLock()
}
ret <- series
if i != j {
it.s.locks[j].RUnlock()
}
series.Unlock()
}
it.s.locks[i].RUnlock()
}
close(ret)
}()
return ret
}