Skip to content
This repository was archived by the owner on Aug 23, 2023. It is now read-only.

Commit 5b7b677

Browse files
DieterbeDieterbe
committed
cache accounting : batched chunk add and hit operations
1) less contention on channel 2) prevent channel overflowing 3) less CPU spent printing `Failed to submit event to accounting, channel was blocked` For the record, When I patch the code like so: diff --git a/mdata/cache/accnt/flat_accnt.go b/mdata/cache/accnt/flat_accnt.go index 4097e067..d202960c 100644 --- a/mdata/cache/accnt/flat_accnt.go +++ b/mdata/cache/accnt/flat_accnt.go @@ -1,7 +1,9 @@ package accnt import ( + "fmt" "sort" + "time" "github.com/grafana/metrictank/mdata/chunk" "github.com/raintank/worldping-api/pkg/log" @@ -178,6 +180,7 @@ func (a *FlatAccnt) eventLoop() { }, ) case evnt_add_chnks: + pre := time.Now() payload := event.pl.(*AddsPayload) a.addRange(payload.metric, payload.chunks) cacheChunkAdd.Add(len(payload.chunks)) @@ -189,6 +192,7 @@ func (a *FlatAccnt) eventLoop() { }, ) } + fmt.Println("CODE evnt_add_chnks took", time.Since(pre).Nanoseconds()) case evnt_hit_chnk: payload := event.pl.(*HitPayload) a.lru.touch( @@ -198,6 +202,7 @@ func (a *FlatAccnt) eventLoop() { }, ) case evnt_hit_chnks: + pre := time.Now() payload := event.pl.(*HitsPayload) for _, chunk := range payload.chunks { a.lru.touch( @@ -207,6 +212,7 @@ func (a *FlatAccnt) eventLoop() { }, ) } + fmt.Println("CODE evnt_hit_chnks took", time.Since(pre).Nanoseconds()) and run the workload described in #931 (queries for 2y of data across 500 series) I get these durations: awk '/evnt_add_chnks/ {print $4}' code2.txt |goplot hist 1187.00 -> 11160892.40: ▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇ 850 22320597.80: ▇▇ 39 (upto 22 ms) 33480303.20: ▇ 17 (upto 33 ms) 44640008.60: ▏ 5 (upto 45 ms) 55799714.00: ▏ 5 (upto 56 ms) 66959419.40: ▏ 1 (upto 67 ms) 78119124.80: ▏ 0 89278830.20: ▏ 0 100438535.60: ▏ 0 111598241.00: ▏ 1 (upto 111 ms) awk '/evnt_hit_chnks/ {print $4}' code2.txt |goplot hist 45.00 -> 22366478.50: ▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇▇ 7624 44732912.00: ▏ 65 (upto 45 ms) 67099345.50: ▏ 29 (upto 67 ms) 89465779.00: ▏ 8 (upto 89 ms) 111832212.50: ▏ 1 (upto 112 ms) 134198646.00: ▏ 1 (upto 134 ms) 156565079.50: ▏ 0 178931513.00: ▏ 1 (upto 179 ms) 201297946.50: ▏ 0 223664380.00: ▏ 1 (upto 224 ms)
1 parent 1a0f6fa commit 5b7b677

File tree

3 files changed

+99
-24
lines changed

3 files changed

+99
-24
lines changed

‎mdata/cache/accnt/flat_accnt.go

+90-14
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@ package accnt
33
import (
44
"sort"
55

6+
"github.com/grafana/metrictank/mdata/chunk"
67
"github.com/raintank/worldping-api/pkg/log"
78
"gopkg.in/raintank/schema.v1"
89
)
@@ -47,32 +48,49 @@ type FlatAccntMet struct {
4748
chunks map[uint32]uint64
4849
}
4950

50-
// event types to be used in FlatAccntEvent
51-
const evnt_hit_chnk uint8 = 4
52-
const evnt_add_chnk uint8 = 5
53-
const evnt_del_met uint8 = 6
54-
const evnt_get_total uint8 = 7
55-
const evnt_stop uint8 = 100
56-
const evnt_reset uint8 = 101
57-
5851
type FlatAccntEvent struct {
59-
t uint8 // event type
60-
pl interface{} // payload
52+
eType eventType
53+
pl interface{} // payload
6154
}
6255

56+
type eventType uint8
57+
58+
const (
59+
evnt_hit_chnk eventType = iota
60+
evnt_hit_chnks
61+
evnt_add_chnk
62+
evnt_add_chnks
63+
evnt_del_met
64+
evnt_get_total
65+
evnt_stop
66+
evnt_reset
67+
)
68+
6369
// payload to be sent with an add event
6470
type AddPayload struct {
6571
metric schema.AMKey
6672
ts uint32
6773
size uint64
6874
}
6975

76+
// payload to be sent with an add event
77+
type AddsPayload struct {
78+
metric schema.AMKey
79+
chunks []chunk.IterGen
80+
}
81+
7082
// payload to be sent with a hit event
7183
type HitPayload struct {
7284
metric schema.AMKey
7385
ts uint32
7486
}
7587

88+
// payload to be sent with a hits event
89+
type HitsPayload struct {
90+
metric schema.AMKey
91+
chunks []chunk.IterGen
92+
}
93+
7694
// payload to be sent with del metric event
7795
type DelMetPayload struct {
7896
metric schema.AMKey
@@ -111,9 +129,16 @@ func (a *FlatAccnt) AddChunk(metric schema.AMKey, ts uint32, size uint64) {
111129
a.act(evnt_add_chnk, &AddPayload{metric, ts, size})
112130
}
113131

132+
func (a *FlatAccnt) AddChunks(metric schema.AMKey, chunks []chunk.IterGen) {
133+
a.act(evnt_add_chnks, &AddsPayload{metric, chunks})
134+
}
135+
114136
func (a *FlatAccnt) HitChunk(metric schema.AMKey, ts uint32) {
115137
a.act(evnt_hit_chnk, &HitPayload{metric, ts})
116138
}
139+
func (a *FlatAccnt) HitChunks(metric schema.AMKey, chunks []chunk.IterGen) {
140+
a.act(evnt_hit_chnks, &HitsPayload{metric, chunks})
141+
}
117142

118143
func (a *FlatAccnt) Stop() {
119144
a.act(evnt_stop, nil)
@@ -123,10 +148,10 @@ func (a *FlatAccnt) Reset() {
123148
a.act(evnt_reset, nil)
124149
}
125150

126-
func (a *FlatAccnt) act(t uint8, payload interface{}) {
151+
func (a *FlatAccnt) act(eType eventType, payload interface{}) {
127152
event := FlatAccntEvent{
128-
t: t,
129-
pl: payload,
153+
eType: eType,
154+
pl: payload,
130155
}
131156

132157
select {
@@ -141,7 +166,7 @@ func (a *FlatAccnt) eventLoop() {
141166
for {
142167
select {
143168
case event := <-a.eventQ:
144-
switch event.t {
169+
switch event.eType {
145170
case evnt_add_chnk:
146171
payload := event.pl.(*AddPayload)
147172
a.add(payload.metric, payload.ts, payload.size)
@@ -152,6 +177,18 @@ func (a *FlatAccnt) eventLoop() {
152177
Ts: payload.ts,
153178
},
154179
)
180+
case evnt_add_chnks:
181+
payload := event.pl.(*AddsPayload)
182+
a.addRange(payload.metric, payload.chunks)
183+
cacheChunkAdd.Add(len(payload.chunks))
184+
for _, chunk := range payload.chunks {
185+
a.lru.touch(
186+
EvictTarget{
187+
Metric: payload.metric,
188+
Ts: chunk.Ts,
189+
},
190+
)
191+
}
155192
case evnt_hit_chnk:
156193
payload := event.pl.(*HitPayload)
157194
a.lru.touch(
@@ -160,6 +197,16 @@ func (a *FlatAccnt) eventLoop() {
160197
Ts: payload.ts,
161198
},
162199
)
200+
case evnt_hit_chnks:
201+
payload := event.pl.(*HitsPayload)
202+
for _, chunk := range payload.chunks {
203+
a.lru.touch(
204+
EvictTarget{
205+
Metric: payload.metric,
206+
Ts: chunk.Ts,
207+
},
208+
)
209+
}
163210
case evnt_del_met:
164211
payload := event.pl.(*DelMetPayload)
165212
a.delMet(payload.metric)
@@ -228,6 +275,35 @@ func (a *FlatAccnt) add(metric schema.AMKey, ts uint32, size uint64) {
228275
cacheSizeUsed.AddUint64(size)
229276
}
230277

278+
func (a *FlatAccnt) addRange(metric schema.AMKey, chunks []chunk.IterGen) {
279+
var met *FlatAccntMet
280+
var ok bool
281+
282+
if met, ok = a.metrics[metric]; !ok {
283+
met = &FlatAccntMet{
284+
total: 0,
285+
chunks: make(map[uint32]uint64),
286+
}
287+
a.metrics[metric] = met
288+
cacheMetricAdd.Inc()
289+
}
290+
291+
var sizeDiff uint64
292+
293+
for _, chunk := range chunks {
294+
if _, ok = met.chunks[chunk.Ts]; ok {
295+
// we already have that chunk
296+
continue
297+
}
298+
size := chunk.Size()
299+
sizeDiff += size
300+
met.chunks[chunk.Ts] = size
301+
}
302+
303+
met.total = met.total + sizeDiff
304+
cacheSizeUsed.AddUint64(sizeDiff)
305+
}
306+
231307
func (a *FlatAccnt) evict() {
232308
var met *FlatAccntMet
233309
var targets []uint32

‎mdata/cache/accnt/if.go

+6-1
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,9 @@
11
package accnt
22

3-
import "gopkg.in/raintank/schema.v1"
3+
import (
4+
"github.com/grafana/metrictank/mdata/chunk"
5+
"gopkg.in/raintank/schema.v1"
6+
)
47

58
// Accnt represents an instance of cache accounting.
69
// Currently there is only one implementation called `FlatAccnt`,
@@ -9,7 +12,9 @@ import "gopkg.in/raintank/schema.v1"
912
type Accnt interface {
1013
GetEvictQ() chan *EvictTarget
1114
AddChunk(metric schema.AMKey, ts uint32, size uint64)
15+
AddChunks(metric schema.AMKey, chunks []chunk.IterGen)
1216
HitChunk(metric schema.AMKey, ts uint32)
17+
HitChunks(metric schema.AMKey, chunks []chunk.IterGen)
1318
DelMetric(metric schema.AMKey)
1419
Stop()
1520
Reset()

‎mdata/cache/ccache.go

+3-9
Original file line numberDiff line numberDiff line change
@@ -202,9 +202,7 @@ func (c *CCache) AddRange(metric schema.AMKey, prev uint32, itergens []chunk.Ite
202202
ccm.AddRange(prev, itergens)
203203
}
204204

205-
for _, itergen := range itergens {
206-
c.accnt.AddChunk(metric, itergen.Ts, itergen.Size())
207-
}
205+
c.accnt.AddChunks(metric, itergens)
208206
}
209207

210208
func (cc *CCache) Reset() (int, int) {
@@ -293,12 +291,8 @@ func (c *CCache) Search(ctx context.Context, metric schema.AMKey, from, until ui
293291

294292
accnt.CacheChunkHit.Add(len(res.Start) + len(res.End))
295293
go func() {
296-
for _, hit := range res.Start {
297-
c.accnt.HitChunk(metric, hit.Ts)
298-
}
299-
for _, hit := range res.End {
300-
c.accnt.HitChunk(metric, hit.Ts)
301-
}
294+
c.accnt.HitChunks(metric, res.Start)
295+
c.accnt.HitChunks(metric, res.End)
302296
}()
303297

304298
if res.Complete {

0 commit comments

Comments
 (0)