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

app/vmselect/netstorage: marshal block outside tmpBlocksFile.WriteBlock

Browse source 
This allows re-using the destination buffer for marshaling in the outer loop.
This commit is contained in:
Aliaksandr Valialkin 2019-09-28 20:38:24 +03:00
parent b7ee2e7af2
commit c1d3705be0
3 changed files with 14 additions and 12 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -484,9 +484,12 @@ func ProcessSearchQuery(sq *storage.SearchQuery, fetchData bool, deadline Deadli
tbf := getTmpBlocksFile() tbf := getTmpBlocksFile()
m := make(map[string][]tmpBlockAddr) m := make(map[string][]tmpBlockAddr)
blocksRead := 0 blocksRead := 0
bb := tmpBufPool.Get()
defer tmpBufPool.Put(bb)
for sr.NextMetricBlock() { for sr.NextMetricBlock() {
blocksRead++ blocksRead++
addr, err := tbf.WriteBlock(sr.MetricBlock.Block) bb.B = storage.MarshalBlock(bb.B[:0], sr.MetricBlock.Block)
addr, err := tbf.WriteBlockData(bb.B)
if err != nil { if err != nil {
putTmpBlocksFile(tbf) putTmpBlocksFile(tbf)
return nil, fmt.Errorf("cannot write data block #%d to temporary blocks file: %s", blocksRead, err) return nil, fmt.Errorf("cannot write data block #%d to temporary blocks file: %s", blocksRead, err)

16
app/vmselect/netstorage/tmp_blocks_file.go
View file

@ -82,22 +82,18 @@ func (addr tmpBlockAddr) String() string {
var tmpBlocksFilesCreated = metrics.NewCounter(`vm_tmp_blocks_files_created_total`) var tmpBlocksFilesCreated = metrics.NewCounter(`vm_tmp_blocks_files_created_total`)
// WriteBlock writes b to tbf. // WriteBlockData writes b to tbf.
// //
// It returns errors since the operation may fail on space shortage // It returns errors since the operation may fail on space shortage
// and this must be handled. // and this must be handled.
func (tbf *tmpBlocksFile) WriteBlock(b *storage.Block) (tmpBlockAddr, error) { func (tbf *tmpBlocksFile) WriteBlockData(b []byte) (tmpBlockAddr, error) {
bb := tmpBufPool.Get()
defer tmpBufPool.Put(bb)
bb.B = storage.MarshalBlock(bb.B[:0], b)
var addr tmpBlockAddr var addr tmpBlockAddr
addr.offset = tbf.offset addr.offset = tbf.offset
addr.size = len(bb.B) addr.size = len(b)
tbf.offset += uint64(addr.size) tbf.offset += uint64(addr.size)
if len(tbf.buf)+len(bb.B) <= cap(tbf.buf) { if len(tbf.buf)+len(b) <= cap(tbf.buf) {
// Fast path - the data fits tbf.buf // Fast path - the data fits tbf.buf
tbf.buf = append(tbf.buf, bb.B...) tbf.buf = append(tbf.buf, b...)
return addr, nil return addr, nil
} }
@ -111,7 +107,7 @@ func (tbf *tmpBlocksFile) WriteBlock(b *storage.Block) (tmpBlockAddr, error) {
tmpBlocksFilesCreated.Inc() tmpBlocksFilesCreated.Inc()
} }
_, err := tbf.f.Write(tbf.buf) _, err := tbf.f.Write(tbf.buf)
tbf.buf = append(tbf.buf[:0], bb.B...) tbf.buf = append(tbf.buf[:0], b...)
if err != nil { if err != nil {
return addr, fmt.Errorf("cannot write block to %q: %s", tbf.f.Name(), err) return addr, fmt.Errorf("cannot write block to %q: %s", tbf.f.Name(), err)
} }

5
app/vmselect/netstorage/tmp_blocks_file_test.go
View file

@ -77,9 +77,12 @@ func testTmpBlocksFile() error {
// Write blocks until their summary size exceeds `size`. // Write blocks until their summary size exceeds `size`.
var addrs []tmpBlockAddr var addrs []tmpBlockAddr
var blocks []*storage.Block var blocks []*storage.Block
bb := tmpBufPool.Get()
defer tmpBufPool.Put(bb)
for tbf.offset < uint64(size) { for tbf.offset < uint64(size) {
b := createBlock() b := createBlock()
addr, err := tbf.WriteBlock(b) bb.B = storage.MarshalBlock(bb.B[:0], b)
addr, err := tbf.WriteBlockData(bb.B)
if err != nil { if err != nil {
return fmt.Errorf("cannot write block at offset %d: %s", tbf.offset, err) return fmt.Errorf("cannot write block at offset %d: %s", tbf.offset, err)
} }