github-mirrors/VictoriaMetrics
1
0
Fork 0
mirror of https://github.com/VictoriaMetrics/VictoriaMetrics.git synced 2025-01-10 15:14:09 +00:00
Code Issues Projects Releases Packages Wiki Activity

app/vmselect/netstorage: reduce memory usage when the time range from query touches big number of samples per each time series

Browse source
This commit is contained in:
Aliaksandr Valialkin 2020-09-15 21:06:04 +03:00
parent 8c3d7c1a59
commit 1fec47a289
1 changed files with 12 additions and 15 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

27
app/vmselect/netstorage/netstorage.go
View file

@ -202,10 +202,10 @@ func (upw *unpackWork) reset() {
}
}
func (upw *unpackWork) unpack() {
func (upw *unpackWork) unpack(tmpBlock *storage.Block) {
for _, w := range upw.ws {
sb := getSortBlock()
if err := sb.unpackFrom(w.br, w.tr, upw.fetchData); err != nil {
if err := sb.unpackFrom(tmpBlock, w.br, w.tr, upw.fetchData); err != nil {
putSortBlock(sb)
upw.doneCh <- fmt.Errorf("cannot unpack block: %w", err)
return
@ -239,8 +239,9 @@ func init() {
}
func unpackWorker() {
var tmpBlock storage.Block
for upw := range unpackWorkCh {
upw.unpack()
upw.unpack(&tmpBlock)
}
}
@ -372,30 +373,26 @@ func mergeSortBlocks(dst *Result, sbh sortBlocksHeap) {
var dedupsDuringSelect = metrics.NewCounter(`vm_deduplicated_samples_total{type="select"}`)
type sortBlock struct {
// b is used as a temporary storage for unpacked rows before they
// go to Timestamps and Values.
b storage.Block
Timestamps []int64
Values []float64
NextIdx int
}
func (sb *sortBlock) reset() {
sb.b.Reset()
sb.Timestamps = sb.Timestamps[:0]
sb.Values = sb.Values[:0]
sb.NextIdx = 0
}
func (sb *sortBlock) unpackFrom(br storage.BlockRef, tr storage.TimeRange, fetchData bool) error {
br.MustReadBlock(&sb.b, fetchData)
func (sb *sortBlock) unpackFrom(tmpBlock *storage.Block, br storage.BlockRef, tr storage.TimeRange, fetchData bool) error {
tmpBlock.Reset()
br.MustReadBlock(tmpBlock, fetchData)
if fetchData {
if err := sb.b.UnmarshalData(); err != nil {
if err := tmpBlock.UnmarshalData(); err != nil {
return fmt.Errorf("cannot unmarshal block: %w", err)
}
}
timestamps := sb.b.Timestamps()
timestamps := tmpBlock.Timestamps()
// Skip timestamps smaller than tr.MinTimestamp.
i := 0
@ -408,16 +405,16 @@ func (sb *sortBlock) unpackFrom(br storage.BlockRef, tr storage.TimeRange, fetch
for j > i && timestamps[j-1] > tr.MaxTimestamp {
j--
}
skippedRows := sb.b.RowsCount() - (j - i)
skippedRows := tmpBlock.RowsCount() - (j - i)
metricRowsSkipped.Add(skippedRows)
// Copy the remaining values.
if i == j {
return nil
}
values := sb.b.Values()
values := tmpBlock.Values()
sb.Timestamps = append(sb.Timestamps, timestamps[i:j]...)
sb.Values = decimal.AppendDecimalToFloat(sb.Values, values[i:j], sb.b.Scale())
sb.Values = decimal.AppendDecimalToFloat(sb.Values, values[i:j], tmpBlock.Scale())
return nil
}