mirror of
https://github.com/VictoriaMetrics/VictoriaMetrics.git
synced 2024-11-21 14:44:00 +00:00
134 lines
3.2 KiB
Go
134 lines
3.2 KiB
Go
package mergeset
|
|
|
|
import (
|
|
"fmt"
|
|
"path/filepath"
|
|
"sync"
|
|
"unsafe"
|
|
|
|
"github.com/VictoriaMetrics/VictoriaMetrics/lib/blockcache"
|
|
"github.com/VictoriaMetrics/VictoriaMetrics/lib/filestream"
|
|
"github.com/VictoriaMetrics/VictoriaMetrics/lib/fs"
|
|
"github.com/VictoriaMetrics/VictoriaMetrics/lib/memory"
|
|
)
|
|
|
|
var idxbCache = blockcache.NewCache(getMaxIndexBlocksCacheSize)
|
|
var ibCache = blockcache.NewCache(getMaxInmemoryBlocksCacheSize)
|
|
|
|
func getMaxIndexBlocksCacheSize() int {
|
|
maxIndexBlockCacheSizeOnce.Do(func() {
|
|
maxIndexBlockCacheSize = int(0.2 * float64(memory.Allowed()))
|
|
})
|
|
return maxIndexBlockCacheSize
|
|
}
|
|
|
|
var (
|
|
maxIndexBlockCacheSize int
|
|
maxIndexBlockCacheSizeOnce sync.Once
|
|
)
|
|
|
|
func getMaxInmemoryBlocksCacheSize() int {
|
|
maxInmemoryBlockCacheSizeOnce.Do(func() {
|
|
maxInmemoryBlockCacheSize = int(0.3 * float64(memory.Allowed()))
|
|
})
|
|
return maxInmemoryBlockCacheSize
|
|
}
|
|
|
|
var (
|
|
maxInmemoryBlockCacheSize int
|
|
maxInmemoryBlockCacheSizeOnce sync.Once
|
|
)
|
|
|
|
type part struct {
|
|
ph partHeader
|
|
|
|
path string
|
|
|
|
size uint64
|
|
|
|
mrs []metaindexRow
|
|
|
|
indexFile fs.MustReadAtCloser
|
|
itemsFile fs.MustReadAtCloser
|
|
lensFile fs.MustReadAtCloser
|
|
}
|
|
|
|
func openFilePart(path string) (*part, error) {
|
|
path = filepath.Clean(path)
|
|
|
|
var ph partHeader
|
|
if err := ph.ParseFromPath(path); err != nil {
|
|
return nil, fmt.Errorf("cannot parse path to part: %w", err)
|
|
}
|
|
|
|
metaindexPath := path + "/metaindex.bin"
|
|
metaindexFile, err := filestream.Open(metaindexPath, true)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("cannot open %q: %w", metaindexPath, err)
|
|
}
|
|
metaindexSize := fs.MustFileSize(metaindexPath)
|
|
|
|
indexPath := path + "/index.bin"
|
|
indexFile := fs.MustOpenReaderAt(indexPath)
|
|
indexSize := fs.MustFileSize(indexPath)
|
|
|
|
itemsPath := path + "/items.bin"
|
|
itemsFile := fs.MustOpenReaderAt(itemsPath)
|
|
itemsSize := fs.MustFileSize(itemsPath)
|
|
|
|
lensPath := path + "/lens.bin"
|
|
lensFile := fs.MustOpenReaderAt(lensPath)
|
|
lensSize := fs.MustFileSize(lensPath)
|
|
|
|
size := metaindexSize + indexSize + itemsSize + lensSize
|
|
return newPart(&ph, path, size, metaindexFile, indexFile, itemsFile, lensFile)
|
|
}
|
|
|
|
func newPart(ph *partHeader, path string, size uint64, metaindexReader filestream.ReadCloser, indexFile, itemsFile, lensFile fs.MustReadAtCloser) (*part, error) {
|
|
var errors []error
|
|
mrs, err := unmarshalMetaindexRows(nil, metaindexReader)
|
|
if err != nil {
|
|
errors = append(errors, fmt.Errorf("cannot unmarshal metaindexRows: %w", err))
|
|
}
|
|
metaindexReader.MustClose()
|
|
|
|
var p part
|
|
p.path = path
|
|
p.size = size
|
|
p.mrs = mrs
|
|
|
|
p.indexFile = indexFile
|
|
p.itemsFile = itemsFile
|
|
p.lensFile = lensFile
|
|
|
|
p.ph.CopyFrom(ph)
|
|
if len(errors) > 0 {
|
|
// Return only the first error, since it has no sense in returning all errors.
|
|
err := fmt.Errorf("error opening part %s: %w", p.path, errors[0])
|
|
p.MustClose()
|
|
return nil, err
|
|
}
|
|
return &p, nil
|
|
}
|
|
|
|
func (p *part) MustClose() {
|
|
p.indexFile.MustClose()
|
|
p.itemsFile.MustClose()
|
|
p.lensFile.MustClose()
|
|
|
|
idxbCache.RemoveBlocksForPart(p)
|
|
ibCache.RemoveBlocksForPart(p)
|
|
}
|
|
|
|
type indexBlock struct {
|
|
bhs []blockHeader
|
|
}
|
|
|
|
func (idxb *indexBlock) SizeBytes() int {
|
|
bhs := idxb.bhs[:cap(idxb.bhs)]
|
|
n := int(unsafe.Sizeof(*idxb))
|
|
for i := range bhs {
|
|
n += bhs[i].SizeBytes()
|
|
}
|
|
return n
|
|
}
|