Kiu/VictoriaMetrics
Kiu/VictoriaMetrics
1
0
Fork 0
mirror of https://github.com/VictoriaMetrics/VictoriaMetrics.git synced 2024-11-23 20:37:12 +01:00
Code Issues Actions 1 Packages Projects Releases Wiki Activity

lib/mergeset: skip createing temporary part objects when merging source inmemory parts

Browse Source 
This should reduce CPU usage when adding new entries to inverted index.
This should alos prevent from creating stalled cleaner goroutines for the created temporary parts,
since they were never closed.

This should fix the following issue: https://github.com/VictoriaMetrics/VictoriaMetrics/issues/316 .
This commit is contained in:
Aliaksandr Valialkin 2020-02-13 14:06:51 +02:00
parent eceaf13e5e
commit 7836ad8907
1 changed files with 15 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

29
lib/mergeset/table.go
View File

@ -565,7 +565,7 @@ func (tb *Table) mergeRawItemsBlocks(blocksToMerge []*inmemoryBlock) {
func (tb *Table) mergeInmemoryBlocks(blocksToMerge []*inmemoryBlock) *partWrapper {
// Convert blocksToMerge into inmemoryPart's
pws := make([]*partWrapper, 0, len(blocksToMerge))
mps := make([]*inmemoryPart, 0, len(blocksToMerge))
for _, ib := range blocksToMerge {
if len(ib.items) == 0 {
continue
@ -573,24 +573,25 @@ func (tb *Table) mergeInmemoryBlocks(blocksToMerge []*inmemoryBlock) *partWrappe
mp := getInmemoryPart()
mp.Init(ib)
putInmemoryBlock(ib)
mps = append(mps, mp)
}
if len(mps) == 0 {
return nil
}
if len(mps) == 1 {
// Nothing to merge. Just return a single inmemory part.
mp := mps[0]
p := mp.NewPart()
pw := &partWrapper{
return &partWrapper{
p: p,
mp: mp,
refCount: 1,
}
pws = append(pws, pw)
}
if len(pws) == 0 {
return nil
}
if len(pws) == 1 {
return pws[0]
}
defer func() {
// Return source inmemoryParts to pool.
for _, pw := range pws {
putInmemoryPart(pw.mp)
for _, mp := range mps {
putInmemoryPart(mp)
}
}()
@ -599,10 +600,10 @@ func (tb *Table) mergeInmemoryBlocks(blocksToMerge []*inmemoryBlock) *partWrappe
defer atomic.AddUint64(&tb.activeMerges, ^uint64(0))
// Prepare blockStreamReaders for source parts.
bsrs := make([]*blockStreamReader, 0, len(pws))
for _, pw := range pws {
bsrs := make([]*blockStreamReader, 0, len(mps))
for _, mp := range mps {
bsr := getBlockStreamReader()
bsr.InitFromInmemoryPart(pw.mp)
bsr.InitFromInmemoryPart(mp)
bsrs = append(bsrs, bsr)
}