diff --git a/internal/core/src/mmap/Column.h b/internal/core/src/mmap/Column.h index 9fcab78b0c5a76163d8d9a36e90faef4e6c9a271..233e2a16d29d14ac538e3ef30704d72b5961e71e 100644 --- a/internal/core/src/mmap/Column.h +++ b/internal/core/src/mmap/Column.h @@ -30,11 +30,7 @@ namespace milvus { -#ifdef MAP_POPULATE -static int mmap_flags = MAP_SHARED | MAP_POPULATE; -#else static int mmap_flags = MAP_SHARED; -#endif class ColumnBase { public: diff --git a/internal/querynodev2/segments/segment_loader.go b/internal/querynodev2/segments/segment_loader.go index 05c6e344e8fa301c600621f0be8ee9cb9dd42597..942c9c0c1dda799dc15aff1f8548f5b9c375fc32 100644 --- a/internal/querynodev2/segments/segment_loader.go +++ b/internal/querynodev2/segments/segment_loader.go @@ -299,6 +299,13 @@ func (loader *segmentLoader) notifyLoadFinish(segments ...*querypb.SegmentLoadIn // requestResource requests memory & storage to load segments, // returns the memory usage, disk usage and concurrency with the gained memory. func (loader *segmentLoader) requestResource(ctx context.Context, infos ...*querypb.SegmentLoadInfo) (LoadResource, int, error) { + segmentIDs := lo.Map(infos, func(info *querypb.SegmentLoadInfo, _ int) int64 { + return info.GetSegmentID() + }) + log := log.Ctx(ctx).With( + zap.Int64s("segmentIDs", segmentIDs), + ) + resource := LoadResource{} loader.mut.Lock() @@ -350,6 +357,14 @@ func (loader *segmentLoader) requestResource(ctx context.Context, infos ...*quer resource.DiskSize += du loader.committedResource.Add(resource) + log.Info("request resource for loading segments", + zap.Int("workerNum", resource.WorkNum), + zap.Int("committedWorkerNum", loader.committedResource.WorkNum), + zap.Uint64("memory", resource.MemorySize), + zap.Uint64("committedMemory", loader.committedResource.MemorySize), + zap.Uint64("disk", resource.DiskSize), + zap.Uint64("committedDisk", loader.committedResource.DiskSize), + ) return resource, concurrencyLevel, nil } @@ -830,6 +845,7 @@ func (loader *segmentLoader) checkSegmentSize(ctx context.Context, segmentLoadIn metrics.QueryNodeDiskUsedSize.WithLabelValues(fmt.Sprint(paramtable.GetNodeID())).Set(float64(toMB(uint64(localDiskUsage)))) diskUsage := uint64(localDiskUsage) + loader.committedResource.DiskSize + mmapEnabled := len(paramtable.Get().QueryNodeCfg.MmapDirPath.GetValue()) > 0 maxSegmentSize := uint64(0) predictMemUsage := memUsage predictDiskUsage := diskUsage @@ -856,10 +872,18 @@ func (loader *segmentLoader) checkSegmentSize(ctx context.Context, segmentLoadIn ) return 0, 0, err } - predictMemUsage += neededMemSize - predictDiskUsage += neededDiskSize + if mmapEnabled { + predictDiskUsage += neededMemSize + neededDiskSize + } else { + predictMemUsage += neededMemSize + predictDiskUsage += neededDiskSize + } } else { - predictMemUsage += uint64(getBinlogDataSize(fieldBinlog)) + if mmapEnabled { + predictDiskUsage += uint64(getBinlogDataSize(fieldBinlog)) + } else { + predictMemUsage += uint64(getBinlogDataSize(fieldBinlog)) + } } } @@ -878,7 +902,6 @@ func (loader *segmentLoader) checkSegmentSize(ctx context.Context, segmentLoadIn } } - mmapEnabled := len(paramtable.Get().QueryNodeCfg.MmapDirPath.GetValue()) > 0 log.Info("predict memory and disk usage while loading (in MiB)", zap.Uint64("maxSegmentSize", toMB(maxSegmentSize)), zap.Int("concurrency", concurrency),