mirror of
https://github.com/VictoriaMetrics/VictoriaMetrics.git
synced 2025-02-19 15:30:17 +00:00
app/vmselect/netstorage: marshal block outside tmpBlocksFile.WriteBlock
This also allows marshaling outside lock, thus reducing the amount of work under the lock.
This commit is contained in:
parent
e92e39eddf
commit
946ca438a6
3 changed files with 19 additions and 19 deletions
app/vmselect/netstorage
|
@ -701,16 +701,17 @@ type tmpBlocksFileWrapper struct {
|
||||||
}
|
}
|
||||||
|
|
||||||
func (tbfw *tmpBlocksFileWrapper) WriteBlock(mb *storage.MetricBlock) error {
|
func (tbfw *tmpBlocksFileWrapper) WriteBlock(mb *storage.MetricBlock) error {
|
||||||
|
bb := tmpBufPool.Get()
|
||||||
|
bb.B = storage.MarshalBlock(bb.B[:0], mb.Block)
|
||||||
tbfw.mu.Lock()
|
tbfw.mu.Lock()
|
||||||
defer tbfw.mu.Unlock()
|
addr, err := tbfw.tbf.WriteBlockData(bb.B)
|
||||||
|
tmpBufPool.Put(bb)
|
||||||
addr, err := tbfw.tbf.WriteBlock(mb.Block)
|
if err == nil {
|
||||||
if err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
metricName := mb.MetricName
|
metricName := mb.MetricName
|
||||||
tbfw.m[string(metricName)] = append(tbfw.m[string(metricName)], addr)
|
tbfw.m[string(metricName)] = append(tbfw.m[string(metricName)], addr)
|
||||||
return nil
|
}
|
||||||
|
tbfw.mu.Unlock()
|
||||||
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
// ProcessSearchQuery performs sq on storage nodes until the given deadline.
|
// ProcessSearchQuery performs sq on storage nodes until the given deadline.
|
||||||
|
|
|
@ -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)
|
||||||
}
|
}
|
||||||
|
|
|
@ -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)
|
||||||
}
|
}
|
||||||
|
|
Loading…
Reference in a new issue