milvus-io
/
milvus
mirror of https://github.com/milvus-io/milvus.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

enhance: Optimize GetLocalDiskSize and segment loader mutex (#38683) 

fix of: https://github.com/milvus-io/milvus/pull/38599

Signed-off-by: bigsheeper <yihao.dai@zilliz.com>
pull/38784/head
yihao.dai 2024-12-24 11:19:58 +08:00 committed by GitHub
parent bf27f70c32
commit 1969ab3da7
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 16 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
internal/querynodev2/segments/disk_usage_fetcher.go
View File

@ -30,30 +30,34 @@ import (
)
type diskUsageFetcher struct {
ctx context.Context
path string
diskUsage *atomic.Int64
ctx context.Context
path string
usage *atomic.Int64
err *atomic.Error
}
func NewDiskUsageFetcher(ctx context.Context) *diskUsageFetcher {
return &diskUsageFetcher{
ctx: ctx,
path: paramtable.Get().LocalStorageCfg.Path.GetValue(),
diskUsage: atomic.NewInt64(0),
ctx: ctx,
path: paramtable.Get().LocalStorageCfg.Path.GetValue(),
usage: atomic.NewInt64(0),
err: atomic.NewError(nil),
}
}
func (d *diskUsageFetcher) GetDiskUsage() int64 {
return d.diskUsage.Load()
func (d *diskUsageFetcher) GetDiskUsage() (int64, error) {
return d.usage.Load(), d.err.Load()
}
func (d *diskUsageFetcher) fetch() {
diskUsage, err := GetLocalUsedSize(d.ctx, d.path)
if err != nil {
log.Warn("failed to get disk usage", zap.Error(err))
d.err.Store(err)
return
}
d.diskUsage.Store(diskUsage)
d.usage.Store(diskUsage)
d.err.Store(nil)
metrics.QueryNodeDiskUsedSize.WithLabelValues(fmt.Sprint(paramtable.GetNodeID())).Set(float64(diskUsage) / 1024 / 1024) // in MB
log.Ctx(d.ctx).WithRateGroup("diskUsageFetcher", 1, 300).
RatedInfo(300, "querynode disk usage", zap.Int64("size", diskUsage), zap.Int64("nodeID", paramtable.GetNodeID()))

6
internal/querynodev2/segments/segment_loader.go
View File

@ -848,9 +848,9 @@ func (loader *segmentLoader) requestResource(ctx context.Context, infos ...*quer
memoryUsage := hardware.GetUsedMemoryCount()
totalMemory := hardware.GetMemoryCount()
diskUsage := loader.duf.GetDiskUsage()
if diskUsage == 0 {
return requestResourceResult{}, errors.New("get local used size failed")
diskUsage, err := loader.duf.GetDiskUsage()
if err != nil {
return requestResourceResult{}, errors.Wrap(err, "get local used size failed")
}
diskCap := paramtable.Get().QueryNodeCfg.DiskCapacityLimit.GetAsUint64()