Skip to content

M
milvus

milvus / milvus
11 个月 前同步成功

通知 261
Star 22476
Fork 2472
M
milvus

体验新版 GitCode，发现更多精彩内容 >>

未验证 提交 a1db9038 编写于 作者: X xige-16 提交者： GitHub
浏览文件
操作

Move disk index params to config file (#19714) 

Signed-off-by: Nxige-16 <xi.ge@zilliz.com>
Signed-off-by: Nxige-16 <xi.ge@zilliz.com>
上级 bdaddd7b
展开全部 隐藏空白更改
内联 并排
Showing with 1068 addition and 578 deletion
cmd/milvus/run.go
浏览文件 @ a1db9038
......@@ -6,13 +6,13 @@ import (
"io"
"os"
"github.com/milvus-io/milvus/internal/util/metricsinfo"
"github.com/milvus-io/milvus/internal/util/paramtable"
"go.uber.org/zap"
"github.com/milvus-io/milvus/internal/log"
"github.com/milvus-io/milvus/cmd/roles"
"github.com/milvus-io/milvus/internal/log"
"github.com/milvus-io/milvus/internal/util/hardware"
"github.com/milvus-io/milvus/internal/util/metricsinfo"
"github.com/milvus-io/milvus/internal/util/paramtable"
"github.com/milvus-io/milvus/internal/util/typeutil"
)
......@@ -133,7 +133,10 @@ func (c *run) formatFlags(args []string, flags *flag.FlagSet) {
flags.BoolVar(&c.enableIndexNode, typeutil.IndexNodeRole, false, "enable index node")
flags.BoolVar(&c.enableProxy, typeutil.ProxyRole, false, "enable proxy node")
initMaxprocs(c.serverType, flags)
if c.serverType == typeutil.EmbeddedRole {
flags.SetOutput(io.Discard)
}
hardware.InitMaxprocs(c.serverType, flags)
if err := flags.Parse(args[3:]); err != nil {
os.Exit(-1)
}
......
cmd/milvus/stop.go
浏览文件 @ a1db9038
......@@ -3,10 +3,12 @@ package milvus
import (
"flag"
"fmt"
"io"
"os"
"path"
"syscall"
"github.com/milvus-io/milvus/internal/util/hardware"
"github.com/milvus-io/milvus/internal/util/typeutil"
)
......@@ -47,7 +49,10 @@ func (c *stop) execute(args []string, flags *flag.FlagSet) {
func (c *stop) formatFlags(args []string, flags *flag.FlagSet) {
flags.StringVar(&(c.svrAlias), "alias", "", "set alias")
initMaxprocs(c.serverType, flags)
if c.serverType == typeutil.EmbeddedRole {
flags.SetOutput(io.Discard)
}
hardware.InitMaxprocs(c.serverType, flags)
if err := flags.Parse(args[3:]); err != nil {
os.Exit(-1)
}
......
cmd/milvus/util.go
浏览文件 @ a1db9038
package milvus
import (
"flag"
"fmt"
"io"
"io/ioutil"
syslog "log"
"os"
"path"
"runtime"
"github.com/gofrs/flock"
"go.uber.org/automaxprocs/maxprocs"
"github.com/milvus-io/milvus/internal/util/typeutil"
)
......@@ -64,18 +60,6 @@ func createRuntimeDir(sType string) string {
return runtimeDir
}
// Initialize maxprocs
func initMaxprocs(serverType string, flags *flag.FlagSet) {
if serverType == typeutil.EmbeddedRole {
flags.SetOutput(io.Discard)
// Initialize maxprocs while discarding log.
maxprocs.Set(maxprocs.Logger(nil))
} else {
// Initialize maxprocs.
maxprocs.Set(maxprocs.Logger(syslog.Printf))
}
}
func createPidFile(w io.Writer, filename string, runtimeDir string) (*flock.Flock, error) {
fileFullName := path.Join(runtimeDir, filename)
......
configs/milvus.yaml
浏览文件 @ a1db9038
......@@ -350,6 +350,14 @@ common:
# This configuration is only used by querynode and indexnode, it selects CPU instruction set for Searching and Index-building.
simdType: auto
indexSliceSize: 16 # MB
DiskIndex:
MaxDegree: 56
SearchListSize: 100
PGCodeBudgetGBRatio: 0.125
BuildNumThreadsRatio: 1.0
SearchCacheBudgetGBRatio: 0.125
LoadNumThreadRatio: 8.0
BeamWidthRatio: 4.0
# please adjust in embedded Milvus: local
storageType: minio
......
internal/core/src/index/Meta.h
浏览文件 @ a1db9038
......@@ -51,15 +51,15 @@ constexpr const char* INDEX_VERSION = "index_version";
// DiskAnn build params
constexpr const char* DISK_ANN_RAW_DATA_PATH = "data_path";
constexpr const char* DISK_ANN_MAX_DEGREE = "max_degree";
constexpr const char* DISK_ANN_BUILD_LIST = "build_list";
constexpr const char* DISK_ANN_SEARCH_DRAM_BUDGET = "search_dram_budget";
constexpr const char* DISK_ANN_BUILD_DRAM_BUDGET = "build_dram_budget";
constexpr const char* DISK_ANN_SEARCH_LIST_SIZE = "search_list_size";
constexpr const char* DISK_ANN_PQ_CODE_BUDGET = "pq_code_budget_gb";
constexpr const char* DISK_ANN_BUILD_DRAM_BUDGET = "build_dram_budget_gb";
constexpr const char* DISK_ANN_BUILD_THREAD_NUM = "num_build_thread";
constexpr const char* DISK_ANN_PQ_BYTES = "ps_disk_bytes";
constexpr const char* DISK_ANN_PQ_DIMS = "disk_pq_dims";
// DiskAnn prepare params
constexpr const char* DISK_ANN_PREPARE_THREAD_NUM = "num_prepare_thread";
constexpr const char* NUM_ROW_OF_RAW_DATA = "count";
constexpr const char* DISK_ANN_LOAD_THREAD_NUM = "num_load_thread";
constexpr const char* DISK_ANN_SEARCH_CACHE_BUDGET = "search_cache_budget_gb";
constexpr const char* DISK_ANN_PREPARE_WARM_UP = "warm_up";
constexpr const char* DISK_ANN_PREPARE_USE_BFS_CACHE = "use_bfs_cache";
......
internal/core/src/index/VectorDiskIndex.cpp
浏览文件 @ a1db9038
......@@ -114,11 +114,7 @@ VectorDiskAnnIndex<T>::Query(const DatasetPtr dataset, const SearchInfo& search_
"search_list should less than min(topk*10, 65535)");
// set beamwidth
query_config.beamwidth = 16;
auto beam_width = GetValueFromConfig<uint32_t>(search_info.search_params_, DISK_ANN_QUERY_BEAMWIDTH);
if (beam_width.has_value()) {
query_config.beamwidth = beam_width.value();
}
query_config.beamwidth = search_beamwidth_;
knowhere::Config cfg;
knowhere::DiskANNQueryConfig::Set(cfg, query_config);
......@@ -171,13 +167,13 @@ VectorDiskAnnIndex<T>::parse_build_config(const Config& config) {
build_disk_ann_config.max_degree = max_degree.value();
// set build list
auto search_list_size = GetValueFromConfig<uint32_t>(build_config, DISK_ANN_BUILD_LIST);
AssertInfo(search_list_size.has_value(), "param " + std::string(DISK_ANN_BUILD_LIST) + "is empty");
auto search_list_size = GetValueFromConfig<uint32_t>(build_config, DISK_ANN_SEARCH_LIST_SIZE);
AssertInfo(search_list_size.has_value(), "param " + std::string(DISK_ANN_SEARCH_LIST_SIZE) + "is empty");
build_disk_ann_config.search_list_size = search_list_size.value();
// set search dram budget
auto search_dram_budget_gb = GetValueFromConfig<float>(build_config, DISK_ANN_SEARCH_DRAM_BUDGET);
AssertInfo(search_dram_budget_gb.has_value(), "param " + std::string(DISK_ANN_SEARCH_DRAM_BUDGET) + "is empty");
auto search_dram_budget_gb = GetValueFromConfig<float>(build_config, DISK_ANN_PQ_CODE_BUDGET);
AssertInfo(search_dram_budget_gb.has_value(), "param " + std::string(DISK_ANN_PQ_CODE_BUDGET) + "is empty");
build_disk_ann_config.pq_code_budget_gb = search_dram_budget_gb.value();
// set build dram budget
......@@ -191,8 +187,8 @@ VectorDiskAnnIndex<T>::parse_build_config(const Config& config) {
build_disk_ann_config.num_threads = num_threads.value();
// set pq bytes
auto pq_disk_bytes = GetValueFromConfig<uint32_t>(build_config, DISK_ANN_PQ_BYTES);
AssertInfo(pq_disk_bytes.has_value(), "param " + std::string(DISK_ANN_PQ_BYTES) + "is empty");
auto pq_disk_bytes = GetValueFromConfig<uint32_t>(build_config, DISK_ANN_PQ_DIMS);
AssertInfo(pq_disk_bytes.has_value(), "param " + std::string(DISK_ANN_PQ_DIMS) + "is empty");
build_disk_ann_config.disk_pq_dims = pq_disk_bytes.value();
return build_disk_ann_config;
......@@ -202,7 +198,6 @@ template <typename T>
knowhere::DiskANNPrepareConfig
VectorDiskAnnIndex<T>::parse_prepare_config(const Config& config) {
Config prepare_config = config;
auto dim = GetDimFromConfig(prepare_config);
parse_config(prepare_config);
knowhere::DiskANNPrepareConfig prepare_disk_ann_config;
......@@ -210,20 +205,20 @@ VectorDiskAnnIndex<T>::parse_prepare_config(const Config& config) {
prepare_disk_ann_config.use_bfs_cache = false;
// set prepare thread num
auto num_threads = GetValueFromConfig<uint32_t>(prepare_config, DISK_ANN_PREPARE_THREAD_NUM);
AssertInfo(num_threads.has_value(), "param " + std::string(DISK_ANN_PREPARE_THREAD_NUM) + "is empty");
auto num_threads = GetValueFromConfig<uint32_t>(prepare_config, DISK_ANN_LOAD_THREAD_NUM);
AssertInfo(num_threads.has_value(), "param " + std::string(DISK_ANN_LOAD_THREAD_NUM) + "is empty");
prepare_disk_ann_config.num_threads = num_threads.value();
// get max degree
auto max_degree = GetValueFromConfig<uint32_t>(prepare_config, DISK_ANN_MAX_DEGREE);
AssertInfo(max_degree.has_value(), "param " + std::string(DISK_ANN_MAX_DEGREE) + "is empty");
// set prepare cached node
auto num_rows = GetValueFromConfig<int>(prepare_config, NUM_ROW_OF_RAW_DATA);
AssertInfo(num_rows.has_value(), "param " + std::string(NUM_ROW_OF_RAW_DATA) + "is empty");
// get search_cache_budget_gb
auto search_cache_budget_gb = GetValueFromConfig<float>(prepare_config, DISK_ANN_SEARCH_CACHE_BUDGET);
AssertInfo(search_cache_budget_gb.has_value(), "param " + std::string(DISK_ANN_SEARCH_CACHE_BUDGET) + "is empty");
prepare_disk_ann_config.search_cache_budget_gb = search_cache_budget_gb.value();
prepare_disk_ann_config.search_cache_budget_gb =
(dim + max_degree.value() + 1) * sizeof(float) * 1.2 * num_rows.value() * 0.1 / 1024 / 1024 / 1024;
// update search_beamwidth
auto beamwidth = GetValueFromConfig<uint32_t>(prepare_config, DISK_ANN_QUERY_BEAMWIDTH);
if (beamwidth.has_value()) {
search_beamwidth_ = beamwidth.value();
}
return prepare_disk_ann_config;
}
......@@ -241,20 +236,19 @@ VectorDiskAnnIndex<T>::parse_config(Config& config) {
CheckParameter<int>(config, knowhere::meta::TOPK, stoi_closure, std::nullopt);
/************************** DiskAnn build Params ************************/
CheckParameter<int>(config, DISK_ANN_MAX_DEGREE, stoi_closure, std::optional{48});
CheckParameter<int>(config, DISK_ANN_BUILD_LIST, stoi_closure, std::optional{128});
CheckParameter<float>(config, DISK_ANN_SEARCH_DRAM_BUDGET, stof_closure, std::optional{0.03});
CheckParameter<float>(config, DISK_ANN_BUILD_DRAM_BUDGET, stof_closure, std::optional{32});
CheckParameter<int>(config, DISK_ANN_MAX_DEGREE, stoi_closure, std::nullopt);
CheckParameter<int>(config, DISK_ANN_SEARCH_LIST_SIZE, stoi_closure, std::nullopt);
CheckParameter<float>(config, DISK_ANN_PQ_CODE_BUDGET, stof_closure, std::nullopt);
CheckParameter<float>(config, DISK_ANN_BUILD_DRAM_BUDGET, stof_closure, std::nullopt);
CheckParameter<int>(config, DISK_ANN_BUILD_THREAD_NUM, stoi_closure, std::optional{8});
CheckParameter<int>(config, DISK_ANN_PQ_BYTES, stoi_closure, std::optional{0});
CheckParameter<int>(config, DISK_ANN_PQ_DIMS, stoi_closure, std::optional{0});
/************************** DiskAnn prepare Params ************************/
CheckParameter<int>(config, DISK_ANN_PREPARE_THREAD_NUM, stoi_closure, std::optional{8});
CheckParameter<int>(config, NUM_ROW_OF_RAW_DATA, stoi_closure, std::nullopt);
CheckParameter<int>(config, DISK_ANN_LOAD_THREAD_NUM, stoi_closure, std::optional{8});
CheckParameter<float>(config, DISK_ANN_SEARCH_CACHE_BUDGET, stof_closure, std::nullopt);
/************************** DiskAnn query Params ************************/
// CheckParameter<int>(config, DISK_ANN_QUERY_LIST, stoi_closure, std::nullopt);
// CheckParameter<int>(config, DISK_ANN_QUERY_BEAMWIDTH, stoi_closure, std::optional{16});
CheckParameter<int>(config, DISK_ANN_QUERY_BEAMWIDTH, stoi_closure, std::nullopt);
}
template class VectorDiskAnnIndex<float>;
......
internal/core/src/index/VectorDiskIndex.h
浏览文件 @ a1db9038
......@@ -75,6 +75,7 @@ class VectorDiskAnnIndex : public VectorIndex {
private:
std::unique_ptr<knowhere::IndexDiskANN<T>> index_;
std::shared_ptr<storage::DiskFileManagerImpl> file_manager_;
uint32_t search_beamwidth_ = 8;
};
template <typename T>
......
internal/core/unittest/test_indexing.cpp
浏览文件 @ a1db9038
......@@ -381,6 +381,7 @@ TEST_P(IndexTest, BuildAndQuery) {
index = milvus::index::IndexFactory::GetInstance().CreateIndex(create_index_info, nullptr);
}
ASSERT_NO_THROW(index->BuildWithDataset(xb_dataset, build_conf));
milvus::index::IndexBasePtr new_index;
milvus::index::VectorIndex* vec_index = nullptr;
if (index_type == knowhere::IndexEnum::INDEX_DISKANN) {
......@@ -392,7 +393,8 @@ TEST_P(IndexTest, BuildAndQuery) {
milvus::storage::IndexMeta index_meta{3, 100, 1000, 1};
auto file_manager =
std::make_shared<milvus::storage::DiskFileManagerImpl>(field_data_meta, index_meta, storage_config_);
auto new_index = milvus::index::IndexFactory::GetInstance().CreateIndex(create_index_info, file_manager);
new_index = milvus::index::IndexFactory::GetInstance().CreateIndex(create_index_info, file_manager);
vec_index = dynamic_cast<milvus::index::VectorIndex*>(new_index.get());
std::vector<std::string> index_files;
......@@ -401,6 +403,7 @@ TEST_P(IndexTest, BuildAndQuery) {
}
load_conf["index_files"] = index_files;
vec_index->Load(binary_set, load_conf);
EXPECT_EQ(vec_index->Count(), NB);
#endif
} else {
vec_index = dynamic_cast<milvus::index::VectorIndex*>(index.get());
......
internal/core/unittest/test_utils/indexbuilder_test_utils.h
浏览文件 @ a1db9038
......@@ -186,6 +186,10 @@ generate_build_conf(const milvus::IndexType& index_type, const milvus::MetricTyp
return knowhere::Config{
{knowhere::meta::METRIC_TYPE, metric_type},
{knowhere::meta::DIM, std::to_string(DIM)},
{milvus::index::DISK_ANN_MAX_DEGREE, std::to_string(48)},
{milvus::index::DISK_ANN_SEARCH_LIST_SIZE, std::to_string(128)},
{milvus::index::DISK_ANN_PQ_CODE_BUDGET, std::to_string(0.001)},
{milvus::index::DISK_ANN_BUILD_DRAM_BUDGET, std::to_string(32)},
};
}
return knowhere::Config();
......@@ -197,7 +201,7 @@ generate_load_conf(const milvus::IndexType& index_type, const milvus::MetricType
return knowhere::Config{
{knowhere::meta::METRIC_TYPE, metric_type},
{knowhere::meta::DIM, std::to_string(DIM)},
{milvus::index::NUM_ROW_OF_RAW_DATA, std::to_string(nb)},
{milvus::index::DISK_ANN_SEARCH_CACHE_BUDGET, std::to_string(0.0002)},
};
}
return knowhere::Config();
......
internal/datacoord/metrics_info.go
浏览文件 @ a1db9038
......@@ -20,14 +20,16 @@ import (
"context"
"errors"
"go.uber.org/zap"
"github.com/milvus-io/milvus/api/commonpb"
"github.com/milvus-io/milvus/api/milvuspb"
"github.com/milvus-io/milvus/internal/log"
"github.com/milvus-io/milvus/internal/proto/internalpb"
"github.com/milvus-io/milvus/internal/util/hardware"
"github.com/milvus-io/milvus/internal/util/metricsinfo"
"github.com/milvus-io/milvus/internal/util/typeutil"
"github.com/milvus-io/milvus/internal/util/uniquegenerator"
"go.uber.org/zap"
)
// getQuotaMetrics returns DataCoordQuotaMetrics.
......@@ -120,12 +122,12 @@ func (s *Server) getDataCoordMetrics() metricsinfo.DataCoordInfos {
Name: metricsinfo.ConstructComponentName(typeutil.DataCoordRole, Params.DataCoordCfg.GetNodeID()),
HardwareInfos: metricsinfo.HardwareMetrics{
IP: s.session.Address,
CPUCoreCount: metricsinfo.GetCPUCoreCount(false),
CPUCoreUsage: metricsinfo.GetCPUUsage(),
Memory: metricsinfo.GetMemoryCount(),
MemoryUsage: metricsinfo.GetUsedMemoryCount(),
Disk: metricsinfo.GetDiskCount(),
DiskUsage: metricsinfo.GetDiskUsage(),
CPUCoreCount: hardware.GetCPUNum(),
CPUCoreUsage: hardware.GetCPUUsage(),
Memory: hardware.GetMemoryCount(),
MemoryUsage: hardware.GetUsedMemoryCount(),
Disk: hardware.GetDiskCount(),
DiskUsage: hardware.GetDiskUsage(),
},
SystemInfo: metricsinfo.DeployMetrics{},
CreatedTime: Params.DataCoordCfg.CreatedTime.String(),
......
internal/datanode/metrics_info.go
浏览文件 @ a1db9038
......@@ -22,6 +22,7 @@ import (
"github.com/milvus-io/milvus/api/commonpb"
"github.com/milvus-io/milvus/api/milvuspb"
"github.com/milvus-io/milvus/internal/proto/internalpb"
"github.com/milvus-io/milvus/internal/util/hardware"
"github.com/milvus-io/milvus/internal/util/metricsinfo"
"github.com/milvus-io/milvus/internal/util/ratelimitutil"
"github.com/milvus-io/milvus/internal/util/typeutil"
......@@ -82,8 +83,8 @@ func getComponentConfigurations(ctx context.Context, req *internalpb.ShowConfigu
func (node *DataNode) getSystemInfoMetrics(ctx context.Context, req *milvuspb.GetMetricsRequest) (*milvuspb.GetMetricsResponse, error) {
// TODO(dragondriver): add more metrics
usedMem := metricsinfo.GetUsedMemoryCount()
totalMem := metricsinfo.GetMemoryCount()
usedMem := hardware.GetUsedMemoryCount()
totalMem := hardware.GetMemoryCount()
quotaMetrics, err := node.getQuotaMetrics()
if err != nil {
......@@ -97,12 +98,12 @@ func (node *DataNode) getSystemInfoMetrics(ctx context.Context, req *milvuspb.Ge
}
hardwareMetrics := metricsinfo.HardwareMetrics{
IP: node.session.Address,
CPUCoreCount: metricsinfo.GetCPUCoreCount(false),
CPUCoreUsage: metricsinfo.GetCPUUsage(),
CPUCoreCount: hardware.GetCPUNum(),
CPUCoreUsage: hardware.GetCPUUsage(),
Memory: totalMem,
MemoryUsage: usedMem,
Disk: metricsinfo.GetDiskCount(),
DiskUsage: metricsinfo.GetDiskUsage(),
Disk: hardware.GetDiskCount(),
DiskUsage: hardware.GetDiskUsage(),
}
quotaMetrics.Hms = hardwareMetrics
......
internal/indexcoord/index_coord.go
浏览文件 @ a1db9038
......@@ -818,6 +818,7 @@ func (i *IndexCoord) GetIndexInfos(ctx context.Context, req *indexpb.GetIndexInf
IndexFilePaths: segIdx.IndexFilePaths,
SerializedSize: segIdx.IndexSize,
IndexVersion: segIdx.IndexVersion,
NumRows: segIdx.NumRows,
})
}
}
......
internal/indexcoord/metrics_info.go
浏览文件 @ a1db9038
......@@ -25,6 +25,7 @@ import (
"github.com/milvus-io/milvus/api/milvuspb"
"github.com/milvus-io/milvus/internal/log"
"github.com/milvus-io/milvus/internal/proto/internalpb"
"github.com/milvus-io/milvus/internal/util/hardware"
"github.com/milvus-io/milvus/internal/util/metricsinfo"
"github.com/milvus-io/milvus/internal/util/typeutil"
"github.com/milvus-io/milvus/internal/util/uniquegenerator"
......@@ -65,12 +66,12 @@ func getSystemInfoMetrics(
Name: metricsinfo.ConstructComponentName(typeutil.IndexCoordRole, coord.session.ServerID),
HardwareInfos: metricsinfo.HardwareMetrics{
IP: coord.session.Address,
CPUCoreCount: metricsinfo.GetCPUCoreCount(false),
CPUCoreUsage: metricsinfo.GetCPUUsage(),
Memory: metricsinfo.GetMemoryCount(),
MemoryUsage: metricsinfo.GetUsedMemoryCount(),
Disk: metricsinfo.GetDiskCount(),
DiskUsage: metricsinfo.GetDiskUsage(),
CPUCoreCount: hardware.GetCPUNum(),
CPUCoreUsage: hardware.GetCPUUsage(),
Memory: hardware.GetMemoryCount(),
MemoryUsage: hardware.GetUsedMemoryCount(),
Disk: hardware.GetDiskCount(),
DiskUsage: hardware.GetDiskUsage(),
},
SystemInfo: metricsinfo.DeployMetrics{},
CreatedTime: Params.IndexCoordCfg.CreatedTime.String(),
......
internal/indexnode/indexnode_mock.go
浏览文件 @ a1db9038
......@@ -26,6 +26,7 @@ import (
"github.com/milvus-io/milvus/internal/proto/indexpb"
"github.com/milvus-io/milvus/internal/proto/internalpb"
"github.com/milvus-io/milvus/internal/types"
"github.com/milvus-io/milvus/internal/util/hardware"
"github.com/milvus-io/milvus/internal/util/metricsinfo"
"github.com/milvus-io/milvus/internal/util/typeutil"
)
......@@ -216,12 +217,12 @@ func getMockSystemInfoMetrics(
BaseComponentInfos: metricsinfo.BaseComponentInfos{
Name: metricsinfo.ConstructComponentName(typeutil.IndexNodeRole, Params.IndexNodeCfg.GetNodeID()),
HardwareInfos: metricsinfo.HardwareMetrics{
CPUCoreCount: metricsinfo.GetCPUCoreCount(false),
CPUCoreUsage: metricsinfo.GetCPUUsage(),
CPUCoreCount: hardware.GetCPUNum(),
CPUCoreUsage: hardware.GetCPUUsage(),
Memory: 1000,
MemoryUsage: metricsinfo.GetUsedMemoryCount(),
Disk: metricsinfo.GetDiskCount(),
DiskUsage: metricsinfo.GetDiskUsage(),
MemoryUsage: hardware.GetUsedMemoryCount(),
Disk: hardware.GetDiskCount(),
DiskUsage: hardware.GetDiskUsage(),
},
SystemInfo: metricsinfo.DeployMetrics{},
CreatedTime: Params.IndexNodeCfg.CreatedTime.String(),
......
internal/indexnode/metrics_info.go
浏览文件 @ a1db9038
......@@ -22,6 +22,7 @@ import (
"github.com/milvus-io/milvus/api/commonpb"
"github.com/milvus-io/milvus/api/milvuspb"
"github.com/milvus-io/milvus/internal/proto/internalpb"
"github.com/milvus-io/milvus/internal/util/hardware"
"github.com/milvus-io/milvus/internal/util/metricsinfo"
"github.com/milvus-io/milvus/internal/util/typeutil"
)
......@@ -60,12 +61,12 @@ func getSystemInfoMetrics(
Name: metricsinfo.ConstructComponentName(typeutil.IndexNodeRole, Params.IndexNodeCfg.GetNodeID()),
HardwareInfos: metricsinfo.HardwareMetrics{
IP: node.session.Address,
CPUCoreCount: metricsinfo.GetCPUCoreCount(false),
CPUCoreUsage: metricsinfo.GetCPUUsage(),
Memory: metricsinfo.GetMemoryCount(),
MemoryUsage: metricsinfo.GetUsedMemoryCount(),
Disk: metricsinfo.GetDiskCount(),
DiskUsage: metricsinfo.GetDiskUsage(),
CPUCoreCount: hardware.GetCPUNum(),
CPUCoreUsage: hardware.GetCPUUsage(),
Memory: hardware.GetMemoryCount(),
MemoryUsage: hardware.GetUsedMemoryCount(),
Disk: hardware.GetDiskCount(),
DiskUsage: hardware.GetDiskUsage(),
},
SystemInfo: metricsinfo.DeployMetrics{},
CreatedTime: Params.IndexNodeCfg.CreatedTime.String(),
......
internal/indexnode/task.go
浏览文件 @ a1db9038
......@@ -18,6 +18,7 @@ package indexnode
import (
"context"
"encoding/json"
"errors"
"fmt"
"path"
......@@ -36,6 +37,7 @@ import (
"github.com/milvus-io/milvus/internal/util/funcutil"
"github.com/milvus-io/milvus/internal/util/indexcgowrapper"
"github.com/milvus-io/milvus/internal/util/indexparamcheck"
"github.com/milvus-io/milvus/internal/util/indexparams"
"github.com/milvus-io/milvus/internal/util/logutil"
"github.com/milvus-io/milvus/internal/util/retry"
"github.com/milvus-io/milvus/internal/util/timerecord"
......@@ -335,6 +337,20 @@ func (it *indexBuildTask) BuildDiskAnnIndex(ctx context.Context) error {
it.newIndexParams["index_id"] = strconv.FormatInt(it.req.IndexID, 10)
it.newIndexParams["index_version"] = strconv.FormatInt(it.req.GetIndexVersion(), 10)
err = indexparams.SetDiskIndexBuildParams(it.newIndexParams, it.statistic.NumRows)
if err != nil {
log.Ctx(ctx).Error("failed to fill disk index params", zap.Error(err))
return err
}
jsonIndexParams, err := json.Marshal(it.newIndexParams)
if err != nil {
log.Ctx(ctx).Error("failed to json marshal index params", zap.Error(err))
return err
}
log.Ctx(ctx).Info("disk index params are ready",
zap.Int64("buildID", it.BuildID),
zap.String("index params", string(jsonIndexParams)))
it.index, err = indexcgowrapper.NewCgoIndex(dType, it.newTypeParams, it.newIndexParams, it.req.GetStorageConfig())
if err != nil {
log.Ctx(ctx).Error("failed to create index", zap.Error(err))
......
internal/mq/mqimpl/rocksmq/server/rocksmq_impl.go
浏览文件 @ a1db9038
......@@ -22,18 +22,18 @@ import (
"sync/atomic"
"time"
"github.com/tecbot/gorocksdb"
"go.uber.org/zap"
"github.com/milvus-io/milvus/internal/allocator"
"github.com/milvus-io/milvus/internal/kv"
rocksdbkv "github.com/milvus-io/milvus/internal/kv/rocksdb"
"github.com/milvus-io/milvus/internal/log"
"github.com/milvus-io/milvus/internal/mq/msgstream/mqwrapper"
"github.com/milvus-io/milvus/internal/util/metricsinfo"
"github.com/milvus-io/milvus/internal/util/hardware"
"github.com/milvus-io/milvus/internal/util/paramtable"
"github.com/milvus-io/milvus/internal/util/retry"
"github.com/milvus-io/milvus/internal/util/typeutil"
"github.com/tecbot/gorocksdb"
"go.uber.org/zap"
)
// UniqueID is the type of message ID
......@@ -137,7 +137,7 @@ func NewRocksMQ(params paramtable.BaseTable, name string, idAllocator allocator.
} else if maxProcs > 8 {
parallelism = 2
}
memoryCount := metricsinfo.GetMemoryCount()
memoryCount := hardware.GetMemoryCount()
// default rocks db cache is set with memory
rocksDBLRUCacheCapacity := RocksDBLRUCacheMinCapacity
if memoryCount > 0 {
......
internal/proto/index_coord.proto
浏览文件 @ a1db9038
......@@ -146,6 +146,7 @@ message IndexFilePathInfo {
repeated string index_file_paths = 7;
uint64 serialized_size = 8;
int64 index_version = 9;
int64 num_rows = 10;
}
message SegmentInfo {
......
internal/proto/query_coord.proto
浏览文件 @ a1db9038
......@@ -225,6 +225,7 @@ message FieldIndexInfo {
repeated string index_file_paths = 7;
int64 index_size = 8;
int64 index_version = 9;
int64 num_rows = 10;
}
message LoadSegmentsRequest {
......
internal/proxy/metrics_info.go
浏览文件 @ a1db9038
......@@ -23,6 +23,7 @@ import (
"github.com/milvus-io/milvus/api/commonpb"
"github.com/milvus-io/milvus/api/milvuspb"
"github.com/milvus-io/milvus/internal/proto/internalpb"
"github.com/milvus-io/milvus/internal/util/hardware"
"github.com/milvus-io/milvus/internal/util/metricsinfo"
"github.com/milvus-io/milvus/internal/util/ratelimitutil"
"github.com/milvus-io/milvus/internal/util/typeutil"
......@@ -62,20 +63,20 @@ func getQuotaMetrics() (*metricsinfo.ProxyQuotaMetrics, error) {
// getProxyMetrics get metrics of Proxy, not including the topological metrics of Query cluster and Data cluster.
func getProxyMetrics(ctx context.Context, request *milvuspb.GetMetricsRequest, node *Proxy) (*milvuspb.GetMetricsResponse, error) {
totalMem := metricsinfo.GetMemoryCount()
usedMem := metricsinfo.GetUsedMemoryCount()
totalMem := hardware.GetMemoryCount()
usedMem := hardware.GetUsedMemoryCount()
quotaMetrics, err := getQuotaMetrics()
if err != nil {
return nil, err
}
hardwareMetrics := metricsinfo.HardwareMetrics{
IP: node.session.Address,
CPUCoreCount: metricsinfo.GetCPUCoreCount(false),
CPUCoreUsage: metricsinfo.GetCPUUsage(),
CPUCoreCount: hardware.GetCPUNum(),
CPUCoreUsage: hardware.GetCPUUsage(),
Memory: totalMem,
MemoryUsage: usedMem,
Disk: metricsinfo.GetDiskCount(),
DiskUsage: metricsinfo.GetDiskUsage(),
Disk: hardware.GetDiskCount(),
DiskUsage: hardware.GetDiskUsage(),
}
quotaMetrics.Hms = hardwareMetrics
......@@ -138,12 +139,12 @@ func getSystemInfoMetrics(
Name: proxyRoleName,
HardwareInfos: metricsinfo.HardwareMetrics{
IP: node.session.Address,
CPUCoreCount: metricsinfo.GetCPUCoreCount(false),
CPUCoreUsage: metricsinfo.GetCPUUsage(),
Memory: metricsinfo.GetMemoryCount(),
MemoryUsage: metricsinfo.GetUsedMemoryCount(),
Disk: metricsinfo.GetDiskCount(),
DiskUsage: metricsinfo.GetDiskUsage(),
CPUCoreCount: hardware.GetCPUNum(),
CPUCoreUsage: hardware.GetCPUUsage(),
Memory: hardware.GetMemoryCount(),
MemoryUsage: hardware.GetUsedMemoryCount(),
Disk: hardware.GetDiskCount(),
DiskUsage: hardware.GetDiskUsage(),
},
SystemInfo: metricsinfo.DeployMetrics{},
CreatedTime: Params.ProxyCfg.CreatedTime.String(),
......
internal/proxy/task_index.go
浏览文件 @ a1db9038
......@@ -21,19 +21,18 @@ import (
"errors"
"fmt"
"github.com/milvus-io/milvus/internal/proto/indexpb"
"go.uber.org/zap"
"github.com/milvus-io/milvus/internal/common"
"github.com/milvus-io/milvus/internal/log"
"github.com/milvus-io/milvus/internal/types"
"github.com/milvus-io/milvus/api/commonpb"
"github.com/milvus-io/milvus/api/milvuspb"
"github.com/milvus-io/milvus/api/schemapb"
"github.com/milvus-io/milvus/internal/common"
"github.com/milvus-io/milvus/internal/log"
"github.com/milvus-io/milvus/internal/proto/indexpb"
"github.com/milvus-io/milvus/internal/types"
"github.com/milvus-io/milvus/internal/util/funcutil"
"github.com/milvus-io/milvus/internal/util/indexparamcheck"
"github.com/milvus-io/milvus/internal/util/indexparams"
"github.com/milvus-io/milvus/internal/util/typeutil"
)
......@@ -144,6 +143,18 @@ func (cit *createIndexTask) parseIndexParams() error {
return fmt.Errorf("IndexType not specified")
}
}
indexType, exist := indexParamsMap[common.IndexTypeKey]
if !exist {
return fmt.Errorf("IndexType not specified")
}
if indexType == indexparamcheck.IndexDISKANN {
err := indexparams.FillDiskIndexParams(&Params, indexParamsMap)
if err != nil {
return err
}
}
err := checkTrain(cit.fieldSchema, indexParamsMap)
if err != nil {
return err
......
internal/querycoordv2/handlers.go
浏览文件 @ a1db9038
......@@ -34,6 +34,7 @@ import (
"github.com/milvus-io/milvus/internal/querycoordv2/session"
"github.com/milvus-io/milvus/internal/querycoordv2/task"
"github.com/milvus-io/milvus/internal/querycoordv2/utils"
"github.com/milvus-io/milvus/internal/util/hardware"
"github.com/milvus-io/milvus/internal/util/metricsinfo"
"github.com/milvus-io/milvus/internal/util/typeutil"
"github.com/milvus-io/milvus/internal/util/uniquegenerator"
......@@ -142,12 +143,12 @@ func (s *Server) getSystemInfoMetrics(
Name: metricsinfo.ConstructComponentName(typeutil.QueryCoordRole, Params.QueryCoordCfg.GetNodeID()),
HardwareInfos: metricsinfo.HardwareMetrics{
IP: s.session.Address,
CPUCoreCount: metricsinfo.GetCPUCoreCount(false),
CPUCoreUsage: metricsinfo.GetCPUUsage(),
Memory: metricsinfo.GetMemoryCount(),
MemoryUsage: metricsinfo.GetUsedMemoryCount(),
Disk: metricsinfo.GetDiskCount(),
DiskUsage: metricsinfo.GetDiskUsage(),
CPUCoreCount: hardware.GetCPUNum(),
CPUCoreUsage: hardware.GetCPUUsage(),
Memory: hardware.GetMemoryCount(),
MemoryUsage: hardware.GetUsedMemoryCount(),
Disk: hardware.GetDiskCount(),
DiskUsage: hardware.GetDiskUsage(),
},
SystemInfo: metricsinfo.DeployMetrics{},
CreatedTime: Params.QueryCoordCfg.CreatedTime.String(),
......
internal/querycoordv2/meta/coordinator_broker.go
浏览文件 @ a1db9038
......@@ -191,6 +191,7 @@ func (broker *CoordinatorBroker) GetIndexInfo(ctx context.Context, collectionID
IndexFilePaths: info.GetIndexFilePaths(),
IndexSize: int64(info.GetSerializedSize()),
IndexVersion: info.GetIndexVersion(),
NumRows: info.GetNumRows(),
}
indexes = append(indexes, indexInfo)
......
internal/querynode/load_index_info.go
浏览文件 @ a1db9038
......@@ -25,6 +25,12 @@ package querynode
import "C"
import (
"encoding/json"
"fmt"
"github.com/milvus-io/milvus/internal/log"
"github.com/milvus-io/milvus/internal/util/funcutil"
"github.com/milvus-io/milvus/internal/util/indexparams"
"go.uber.org/zap"
"path/filepath"
"unsafe"
......@@ -95,8 +101,19 @@ func (li *LoadIndexInfo) appendLoadIndexInfo(bytesIndex [][]byte, indexInfo *que
return err
}
for _, param := range indexInfo.IndexParams {
err = li.appendIndexParam(param.Key, param.Value)
// some build params also exist in indexParams, which are useless during loading process
indexParams := funcutil.KeyValuePair2Map(indexInfo.IndexParams)
indexparams.SetDiskIndexLoadParams(indexParams, indexInfo.GetNumRows())
jsonIndexParams, err := json.Marshal(indexParams)
if err != nil {
err = fmt.Errorf("failed to json marshal index params %w", err)
return err
}
log.Info("start append index params", zap.String("index params", string(jsonIndexParams)))
for key, value := range indexParams {
err = li.appendIndexParam(key, value)
if err != nil {
return err
}
......
internal/querynode/metrics_info.go
浏览文件 @ a1db9038
......@@ -22,6 +22,7 @@ import (
"github.com/milvus-io/milvus/api/commonpb"
"github.com/milvus-io/milvus/api/milvuspb"
"github.com/milvus-io/milvus/internal/proto/internalpb"
"github.com/milvus-io/milvus/internal/util/hardware"
"github.com/milvus-io/milvus/internal/util/metricsinfo"
"github.com/milvus-io/milvus/internal/util/ratelimitutil"
"github.com/milvus-io/milvus/internal/util/typeutil"
......@@ -86,8 +87,8 @@ func getQuotaMetrics(node *QueryNode) (*metricsinfo.QueryNodeQuotaMetrics, error
// getSystemInfoMetrics returns metrics info of QueryNode
func getSystemInfoMetrics(ctx context.Context, req *milvuspb.GetMetricsRequest, node *QueryNode) (*milvuspb.GetMetricsResponse, error) {
usedMem := metricsinfo.GetUsedMemoryCount()
totalMem := metricsinfo.GetMemoryCount()
usedMem := hardware.GetUsedMemoryCount()
totalMem := hardware.GetMemoryCount()
quotaMetrics, err := getQuotaMetrics(node)
if err != nil {
......@@ -101,12 +102,12 @@ func getSystemInfoMetrics(ctx context.Context, req *milvuspb.GetMetricsRequest,
}
hardwareInfos := metricsinfo.HardwareMetrics{
IP: node.session.Address,
CPUCoreCount: metricsinfo.GetCPUCoreCount(false),
CPUCoreUsage: metricsinfo.GetCPUUsage(),
CPUCoreCount: hardware.GetCPUNum(),
CPUCoreUsage: hardware.GetCPUUsage(),
Memory: totalMem,
MemoryUsage: usedMem,
Disk: metricsinfo.GetDiskCount(),
DiskUsage: metricsinfo.GetDiskUsage(),
Disk: hardware.GetDiskCount(),
DiskUsage: hardware.GetDiskUsage(),
}
quotaMetrics.Hms = hardwareInfos
......
internal/querynode/segment_loader.go
浏览文件 @ a1db9038
......@@ -42,8 +42,8 @@ import (
"github.com/milvus-io/milvus/internal/types"
"github.com/milvus-io/milvus/internal/util/concurrency"
"github.com/milvus-io/milvus/internal/util/funcutil"
"github.com/milvus-io/milvus/internal/util/hardware"
"github.com/milvus-io/milvus/internal/util/indexparamcheck"
"github.com/milvus-io/milvus/internal/util/metricsinfo"
"github.com/milvus-io/milvus/internal/util/timerecord"
)
......@@ -245,12 +245,19 @@ func (loader *segmentLoader) loadFiles(ctx context.Context, segment *Segment,
for _, fieldBinlog := range loadInfo.BinlogPaths {
fieldID := fieldBinlog.FieldID
// check num rows of data meta and index meta are consistent
if indexInfo, ok := fieldID2IndexInfo[fieldID]; ok {
// TODO:: ugly
indexInfo.IndexParams = append(indexInfo.IndexParams, &commonpb.KeyValuePair{
Key: "count",
Value: strconv.FormatInt(loadInfo.NumOfRows, 10),
})
if loadInfo.GetNumOfRows() != indexInfo.GetNumRows() {
err = fmt.Errorf("num rows of segment binlog file %d mismatch with num rows of index file %d",
loadInfo.GetNumOfRows(), indexInfo.GetNumRows())
log.Error("load segment failed, set segment to meta failed",
zap.Int64("collectionID", segment.collectionID),
zap.Int64("partitionID", segment.partitionID),
zap.Int64("segmentID", segment.segmentID),
zap.Int64("indexBuildID", indexInfo.BuildID),
zap.Error(err))
return err
}
fieldInfo := &IndexedFieldInfo{
fieldBinlog: fieldBinlog,
......@@ -836,8 +843,8 @@ func GetStorageSizeByIndexInfo(indexInfo *querypb.FieldIndexInfo) (uint64, uint6
}
func (loader *segmentLoader) checkSegmentSize(collectionID UniqueID, segmentLoadInfos []*querypb.SegmentLoadInfo, concurrency int) error {
usedMem := metricsinfo.GetUsedMemoryCount()
totalMem := metricsinfo.GetMemoryCount()
usedMem := hardware.GetUsedMemoryCount()
totalMem := hardware.GetMemoryCount()
if len(segmentLoadInfos) < concurrency {
concurrency = len(segmentLoadInfos)
}
......
internal/querynode/task_test.go
浏览文件 @ a1db9038
......@@ -32,7 +32,7 @@ import (
"github.com/milvus-io/milvus/internal/proto/internalpb"
"github.com/milvus-io/milvus/internal/proto/querypb"
"github.com/milvus-io/milvus/internal/util/funcutil"
"github.com/milvus-io/milvus/internal/util/metricsinfo"
"github.com/milvus-io/milvus/internal/util/hardware"
"github.com/milvus-io/milvus/internal/util/typeutil"
)
......@@ -528,7 +528,7 @@ func TestTask_loadSegmentsTask(t *testing.T) {
node, err := genSimpleQueryNode(ctx)
assert.NoError(t, err)
totalRAM := int64(metricsinfo.GetMemoryCount())
totalRAM := int64(hardware.GetMemoryCount())
col, err := node.metaReplica.getCollectionByID(defaultCollectionID)
assert.NoError(t, err)
......
internal/rootcoord/metrics_info.go
浏览文件 @ a1db9038
......@@ -19,13 +19,15 @@ package rootcoord
import (
"context"
"go.uber.org/zap"
"github.com/milvus-io/milvus/api/commonpb"
"github.com/milvus-io/milvus/api/milvuspb"
"github.com/milvus-io/milvus/internal/log"
"github.com/milvus-io/milvus/internal/proto/internalpb"
"github.com/milvus-io/milvus/internal/util/hardware"
"github.com/milvus-io/milvus/internal/util/metricsinfo"
"github.com/milvus-io/milvus/internal/util/typeutil"
"go.uber.org/zap"
)
//getComponentConfigurations returns the configurations of rootcoord matching req.Pattern
......@@ -57,12 +59,12 @@ func (c *Core) getSystemInfoMetrics(ctx context.Context, req *milvuspb.GetMetric
Name: metricsinfo.ConstructComponentName(typeutil.RootCoordRole, c.session.ServerID),
HardwareInfos: metricsinfo.HardwareMetrics{
IP: c.session.Address,
CPUCoreCount: metricsinfo.GetCPUCoreCount(false),
CPUCoreUsage: metricsinfo.GetCPUUsage(),
Memory: metricsinfo.GetMemoryCount(),
MemoryUsage: metricsinfo.GetUsedMemoryCount(),
Disk: metricsinfo.GetDiskCount(),
DiskUsage: metricsinfo.GetDiskUsage(),
CPUCoreCount: hardware.GetCPUNum(),
CPUCoreUsage: hardware.GetCPUUsage(),
Memory: hardware.GetMemoryCount(),
MemoryUsage: hardware.GetUsedMemoryCount(),
Disk: hardware.GetDiskCount(),
DiskUsage: hardware.GetDiskUsage(),
},
SystemInfo: metricsinfo.DeployMetrics{},
CreatedTime: Params.RootCoordCfg.CreatedTime.String(),
......
internal/util/autoindex/bigdata_index.go
浏览文件 @ a1db9038
......@@ -30,85 +30,97 @@ type BigDataIndexExtraParams struct {
}
const (
BuildRatioKey = "build_ratio"
PrepareRatioKey = "prepare_ratio"
BeamWidthRatioKey = "beamwidth_ratio"
BuildRatioKey = "build_ratio"
PrepareRatioKey = "prepare_ratio"
BeamWidthRatioKey = "beamwidth_ratio"
DefaultPGCodeBudgetGBRatio = 0.125
DefaultBuildNumThreadsRatio = 1.0
DefaultSearchCacheBudgetGBRatio = 0.125
DefaultLoadNumThreadRatio = 8.0
DefaultBeamWidthRatio = 4.0
)
func NewBigDataIndexExtraParams() *BigDataIndexExtraParams {
ret := &BigDataIndexExtraParams{
PGCodeBudgetGBRatio: 0.125,
BuildNumThreadsRatio: 1.0,
SearchCacheBudgetGBRatio: 0.125,
LoadNumThreadRatio: 8.0,
BeamWidthRatio: 4.0,
PGCodeBudgetGBRatio: DefaultPGCodeBudgetGBRatio,
BuildNumThreadsRatio: DefaultBuildNumThreadsRatio,
SearchCacheBudgetGBRatio: DefaultSearchCacheBudgetGBRatio,
LoadNumThreadRatio: DefaultLoadNumThreadRatio,
BeamWidthRatio: DefaultBeamWidthRatio,
}
return ret
}
func NewBigDataExtraParamsFromJSON(jsonStr string) *BigDataIndexExtraParams {
func NewBigDataExtraParamsFromJSON(jsonStr string) (*BigDataIndexExtraParams, error) {
buffer := make(map[string]string)
err := json.Unmarshal([]byte(jsonStr), &buffer)
if err != nil {
return NewBigDataIndexExtraParams()
return nil, err
}
return NewBigDataExtraParamsFromMap(buffer)
}
func NewBigDataExtraParamsFromMap(value map[string]string) *BigDataIndexExtraParams {
func NewBigDataExtraParamsFromMap(value map[string]string) (*BigDataIndexExtraParams, error) {
ret := &BigDataIndexExtraParams{}
var err error
buildRatio, ok := value[BuildRatioKey]
if !ok {
ret.PGCodeBudgetGBRatio = 0.125
ret.BuildNumThreadsRatio = 1.0
ret.PGCodeBudgetGBRatio = DefaultPGCodeBudgetGBRatio
ret.BuildNumThreadsRatio = DefaultBuildNumThreadsRatio
} else {
valueMap1 := make(map[string]float64)
err = json.Unmarshal([]byte(buildRatio), &valueMap1)
if err != nil {
ret.PGCodeBudgetGBRatio = 0.125
ret.BuildNumThreadsRatio = 1.0
return ret, err
}
PGCodeBudgetGBRatio, ok := valueMap1["pg_code_budget_gb"]
if !ok {
ret.PGCodeBudgetGBRatio = DefaultPGCodeBudgetGBRatio
} else {
ret.PGCodeBudgetGBRatio = PGCodeBudgetGBRatio
}
BuildNumThreadsRatio, ok := valueMap1["num_threads"]
if !ok {
ret.BuildNumThreadsRatio = DefaultBuildNumThreadsRatio
} else {
ret.PGCodeBudgetGBRatio = valueMap1["pg_code_budget_gb"]
ret.BuildNumThreadsRatio = valueMap1["num_threads"]
ret.BuildNumThreadsRatio = BuildNumThreadsRatio
}
}
prepareRatio, ok := value[PrepareRatioKey]
if !ok {
ret.SearchCacheBudgetGBRatio = 0.125
ret.LoadNumThreadRatio = 8
ret.SearchCacheBudgetGBRatio = DefaultSearchCacheBudgetGBRatio
ret.LoadNumThreadRatio = DefaultLoadNumThreadRatio
} else {
valueMap2 := make(map[string]float64)
err = json.Unmarshal([]byte(prepareRatio), &valueMap2)
if err != nil {
ret.SearchCacheBudgetGBRatio = 0.125
ret.LoadNumThreadRatio = 8
return ret, err
}
SearchCacheBudgetGBRatio, ok := valueMap2["search_cache_budget_gb"]
if !ok {
ret.SearchCacheBudgetGBRatio = DefaultSearchCacheBudgetGBRatio
} else {
ret.SearchCacheBudgetGBRatio = SearchCacheBudgetGBRatio
}
LoadNumThreadRatio, ok := valueMap2["num_threads"]
if !ok {
ret.LoadNumThreadRatio = DefaultLoadNumThreadRatio
} else {
SearchCacheBudgetGBRatio, ok := valueMap2["search_cache_budget_gb"]
if !ok {
ret.SearchCacheBudgetGBRatio = 0.125
} else {
ret.SearchCacheBudgetGBRatio = SearchCacheBudgetGBRatio
}
LoadNumThreadRatio, ok := valueMap2["num_threads"]
if !ok {
ret.LoadNumThreadRatio = 8
} else {
ret.LoadNumThreadRatio = LoadNumThreadRatio
}
ret.LoadNumThreadRatio = LoadNumThreadRatio
}
}
beamWidthRatioStr, ok := value[BeamWidthRatioKey]
if !ok {
ret.BeamWidthRatio = 4.0
ret.BeamWidthRatio = DefaultBeamWidthRatio
} else {
beamWidthRatio, err := strconv.ParseFloat(beamWidthRatioStr, 64)
if err != nil {
ret.BeamWidthRatio = 4.0
ret.BeamWidthRatio = DefaultBeamWidthRatio
} else {
ret.BeamWidthRatio = beamWidthRatio
}
}
return ret
return ret, nil
}
internal/util/autoindex/bigdata_index_test.go
浏览文件 @ a1db9038
......@@ -27,7 +27,8 @@ func TestBigDataIndex_parse(t *testing.T) {
mapString := make(map[string]string)
mapString[BuildRatioKey] = "{\"pg_code_budget_gb\": 0.125, \"num_threads\": 1}"
mapString[PrepareRatioKey] = "{\"search_cache_budget_gb\": 0.225, \"num_threads\": 8}"
extraParams := NewBigDataExtraParamsFromMap(mapString)
extraParams, err := NewBigDataExtraParamsFromMap(mapString)
assert.NoError(t, err)
assert.Equal(t, 1.0, extraParams.BuildNumThreadsRatio)
assert.Equal(t, 8.0, extraParams.LoadNumThreadRatio)
assert.Equal(t, 0.125, extraParams.PGCodeBudgetGBRatio)
......@@ -37,7 +38,8 @@ func TestBigDataIndex_parse(t *testing.T) {
t.Run("parse with partial", func(t *testing.T) {
mapString := make(map[string]string)
mapString[PrepareRatioKey] = "{\"search_cache_budget_gb\": 0.225, \"num_threads\": 8}"
extraParams := NewBigDataExtraParamsFromMap(mapString)
extraParams, err := NewBigDataExtraParamsFromMap(mapString)
assert.NoError(t, err)
assert.Equal(t, 1.0, extraParams.BuildNumThreadsRatio)
assert.Equal(t, 8.0, extraParams.LoadNumThreadRatio)
assert.Equal(t, 0.125, extraParams.PGCodeBudgetGBRatio)
......@@ -46,7 +48,8 @@ func TestBigDataIndex_parse(t *testing.T) {
t.Run("parse with empty", func(t *testing.T) {
mapString := make(map[string]string)
extraParams := NewBigDataExtraParamsFromMap(mapString)
extraParams, err := NewBigDataExtraParamsFromMap(mapString)
assert.NoError(t, err)
assert.Equal(t, 1.0, extraParams.BuildNumThreadsRatio)
assert.Equal(t, 8.0, extraParams.LoadNumThreadRatio)
assert.Equal(t, 0.125, extraParams.PGCodeBudgetGBRatio)
......@@ -54,7 +57,8 @@ func TestBigDataIndex_parse(t *testing.T) {
})
t.Run("parse with nil", func(t *testing.T) {
extraParams := NewBigDataExtraParamsFromMap(nil)
extraParams, err := NewBigDataExtraParamsFromMap(nil)
assert.NoError(t, err)
assert.Equal(t, 1.0, extraParams.BuildNumThreadsRatio)
assert.Equal(t, 8.0, extraParams.LoadNumThreadRatio)
assert.Equal(t, 0.125, extraParams.PGCodeBudgetGBRatio)
......@@ -69,7 +73,8 @@ func TestBigDataIndex_parse(t *testing.T) {
"beamwidth_ratio": "8.0"
}
`
extraParams := NewBigDataExtraParamsFromJSON(jsonStr)
extraParams, err := NewBigDataExtraParamsFromJSON(jsonStr)
assert.NoError(t, err)
assert.Equal(t, 1.0, extraParams.BuildNumThreadsRatio)
assert.Equal(t, 8.0, extraParams.LoadNumThreadRatio)
assert.Equal(t, 0.125, extraParams.PGCodeBudgetGBRatio)
......@@ -83,7 +88,8 @@ func TestBigDataIndex_parse(t *testing.T) {
"build_ratio": "{\"pg_code_budget_gb\": 0.125, \"num_threads\": 1}"
}
`
extraParams := NewBigDataExtraParamsFromJSON(jsonStr)
extraParams, err := NewBigDataExtraParamsFromJSON(jsonStr)
assert.NoError(t, err)
assert.Equal(t, 1.0, extraParams.BuildNumThreadsRatio)
assert.Equal(t, 8.0, extraParams.LoadNumThreadRatio)
assert.Equal(t, 0.125, extraParams.PGCodeBudgetGBRatio)
......@@ -96,7 +102,8 @@ func TestBigDataIndex_parse(t *testing.T) {
{
}
`
extraParams := NewBigDataExtraParamsFromJSON(jsonStr)
extraParams, err := NewBigDataExtraParamsFromJSON(jsonStr)
assert.NoError(t, err)
assert.Equal(t, 1.0, extraParams.BuildNumThreadsRatio)
assert.Equal(t, 8.0, extraParams.LoadNumThreadRatio)
assert.Equal(t, 0.125, extraParams.PGCodeBudgetGBRatio)
......@@ -109,24 +116,15 @@ func TestBigDataIndex_parse(t *testing.T) {
{ x
}
`
extraParams := NewBigDataExtraParamsFromJSON(jsonStr)
assert.Equal(t, 1.0, extraParams.BuildNumThreadsRatio)
assert.Equal(t, 8.0, extraParams.LoadNumThreadRatio)
assert.Equal(t, 0.125, extraParams.PGCodeBudgetGBRatio)
assert.Equal(t, 0.125, extraParams.SearchCacheBudgetGBRatio)
assert.Equal(t, 4.0, extraParams.BeamWidthRatio)
_, err := NewBigDataExtraParamsFromJSON(jsonStr)
assert.Error(t, err)
})
t.Run("new from json invalid1", func(t *testing.T) {
jsonStr := `
""
`
extraParams := NewBigDataExtraParamsFromJSON(jsonStr)
assert.Equal(t, 1.0, extraParams.BuildNumThreadsRatio)
assert.Equal(t, 8.0, extraParams.LoadNumThreadRatio)
assert.Equal(t, 0.125, extraParams.PGCodeBudgetGBRatio)
assert.Equal(t, 0.125, extraParams.SearchCacheBudgetGBRatio)
assert.Equal(t, 4.0, extraParams.BeamWidthRatio)
_, err := NewBigDataExtraParamsFromJSON(jsonStr)
assert.Error(t, err)
})
}
internal/util/funcutil/func.go
浏览文件 @ a1db9038
......@@ -109,7 +109,7 @@ func ParseIndexParamsMap(mStr string) (map[string]string, error) {
buffer := make(map[string]interface{})
err := json.Unmarshal([]byte(mStr), &buffer)
if err != nil {
return nil, errors.New("unmarshal params failed")
return nil, fmt.Errorf("unmarshal params failed, %w", err)
}
ret := make(map[string]string)
for key, value := range buffer {
......
internal/util/metricsinfo/container_darwin.go → internal/util/hardware/container_darwin.go
浏览文件 @ a1db9038
......@@ -9,7 +9,7 @@
// is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
// or implied. See the License for the specific language governing permissions and limitations under the License.
package metricsinfo
package hardware
import (
"errors"
......
internal/util/metricsinfo/container_linux.go → internal/util/hardware/container_linux.go
浏览文件 @ a1db9038
......@@ -9,7 +9,7 @@
// is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
// or implied. See the License for the specific language governing permissions and limitations under the License.
package metricsinfo
package hardware
import (
"errors"
......
internal/util/metricsinfo/container_test_linux.go → internal/util/hardware/container_test_linux.go
浏览文件 @ a1db9038
......@@ -12,7 +12,7 @@
//go:build linux
// +build linux
package metricsinfo
package hardware
import (
"testing"
......
internal/util/metricsinfo/container_windows.go → internal/util/hardware/container_windows.go
浏览文件 @ a1db9038
......@@ -9,7 +9,7 @@
// is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
// or implied. See the License for the specific language governing permissions and limitations under the License.
package metricsinfo
package hardware
import (
"errors"
......
internal/util/metricsinfo/hardware_info.go → internal/util/hardware/hardware_info.go
浏览文件 @ a1db9038
......@@ -9,16 +9,21 @@
// is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
// or implied. See the License for the specific language governing permissions and limitations under the License.
package metricsinfo
package hardware
import (
"flag"
syslog "log"
"runtime"
"sync"
"github.com/shirou/gopsutil/v3/cpu"
"github.com/shirou/gopsutil/v3/mem"
"go.uber.org/automaxprocs/maxprocs"
"go.uber.org/zap"
"github.com/milvus-io/milvus/internal/log"
"github.com/milvus-io/milvus/internal/util/typeutil"
)
var (
......@@ -27,16 +32,24 @@ var (
icErr error
)
// GetCPUCoreCount returns the count of cpu core.
func GetCPUCoreCount(logical bool) int {
c, err := cpu.Counts(logical)
if err != nil {
log.Warn("failed to get cpu counts",
zap.Error(err))
return 0
// Initialize maxprocs
func InitMaxprocs(serverType string, flags *flag.FlagSet) {
if serverType == typeutil.EmbeddedRole {
// Initialize maxprocs while discarding log.
maxprocs.Set(maxprocs.Logger(nil))
} else {
// Initialize maxprocs.
maxprocs.Set(maxprocs.Logger(syslog.Printf))
}
}
return c
// GetCPUNum returns the count of cpu core.
func GetCPUNum() int {
cur := runtime.GOMAXPROCS(0)
if cur <= 0 {
cur = runtime.NumCPU()
}
return cur
}
// GetCPUUsage returns the cpu usage in percentage.
......@@ -120,6 +133,11 @@ func GetUsedMemoryCount() uint64 {
return stats.Used
}
// GetFreeMemoryCount returns the free memory in bytes.
func GetFreeMemoryCount() uint64 {
return GetMemoryCount() - GetUsedMemoryCount()
}
// TODO(dragondriver): not accurate to calculate disk usage when we use distributed storage
// GetDiskCount returns the disk count in bytes.
......
internal/util/metricsinfo/hardware_info_test.go → internal/util/hardware/hardware_info_test.go
浏览文件 @ a1db9038
......@@ -9,7 +9,7 @@
// is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
// or implied. See the License for the specific language governing permissions and limitations under the License.
package metricsinfo
package hardware
import (
"testing"
......@@ -22,10 +22,7 @@ import (
func Test_GetCPUCoreCount(t *testing.T) {
log.Info("TestGetCPUCoreCount",
zap.Int("physical CPUCoreCount", GetCPUCoreCount(false)))
log.Info("TestGetCPUCoreCount",
zap.Int("logical CPUCoreCount", GetCPUCoreCount(true)))
zap.Int("physical CPUCoreCount", GetCPUNum()))
}
func Test_GetCPUUsage(t *testing.T) {
......
internal/util/indexparams/disk_index_params.go 0 → 100644
浏览文件 @ a1db9038
// Licensed to the LF AI & Data foundation under one
// or more contributor license agreements. See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership. The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License. You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package indexparams
import (
"fmt"
"strconv"
"unsafe"
"github.com/milvus-io/milvus/internal/util/hardware"
"github.com/milvus-io/milvus/internal/util/paramtable"
)
const (
PQCodeBudgetRatioKey = "pq_code_budget_gb_ratio"
NumBuildThreadRatioKey = "num_build_thread_ratio"
SearchCacheBudgetRatioKey = "search_cache_budget_gb_ratio"
NumLoadThreadRatioKey = "num_load_thread_ratio"
BeamWidthRatioKey = "beamwidth_ratio"
MaxDegreeKey = "max_degree"
SearchListSizeKey = "search_list_size"
PQCodeBudgetKey = "pq_code_budget_gb"
BuildDramBudgetKey = "build_dram_budget_gb"
NumBuildThreadKey = "num_build_thread"
SearchCacheBudgetKey = "search_cache_budget_gb"
NumLoadThreadKey = "num_load_thread"
BeamWidthKey = "beamwidth"
MaxLoadThread = 64
MaxBeamWidth = 16
)
func getRowDataSizeOfFloatVector(numRows int64, dim int64) int64 {
var floatValue float32
/* #nosec G103 */
return int64(unsafe.Sizeof(floatValue)) * dim * numRows
}
// FillDiskIndexParams fill ratio params to index param on proxy node
// Which will be used to calculate build and load params
func FillDiskIndexParams(params *paramtable.ComponentParam, indexParams map[string]string) error {
maxDegree := strconv.FormatInt(params.CommonCfg.MaxDegree, 10)
searchListSize := strconv.FormatInt(params.CommonCfg.SearchListSize, 10)
pgCodeBudgetGBRatio := params.CommonCfg.PGCodeBudgetGBRatio
buildNumThreadsRatio := params.CommonCfg.BuildNumThreadsRatio
searchCacheBudgetGBRatio := params.CommonCfg.SearchCacheBudgetGBRatio
loadNumThreadRatio := params.CommonCfg.LoadNumThreadRatio
beamWidthRatio := params.CommonCfg.BeamWidthRatio
if params.AutoIndexConfig.Enable {
var ok bool
maxDegree, ok = params.AutoIndexConfig.IndexParams[MaxDegreeKey]
if !ok {
return fmt.Errorf("index param max_degree not exist")
}
searchListSize, ok = params.AutoIndexConfig.IndexParams[SearchListSizeKey]
if !ok {
return fmt.Errorf("index param search_list_size not exist")
}
pgCodeBudgetGBRatio = params.AutoIndexConfig.BigDataExtraParams.PGCodeBudgetGBRatio
buildNumThreadsRatio = params.AutoIndexConfig.BigDataExtraParams.BuildNumThreadsRatio
searchCacheBudgetGBRatio = params.AutoIndexConfig.BigDataExtraParams.SearchCacheBudgetGBRatio
loadNumThreadRatio = params.AutoIndexConfig.BigDataExtraParams.LoadNumThreadRatio
beamWidthRatio = params.AutoIndexConfig.BigDataExtraParams.BeamWidthRatio
}
indexParams[MaxDegreeKey] = maxDegree
indexParams[SearchListSizeKey] = searchListSize
indexParams[PQCodeBudgetRatioKey] = fmt.Sprintf("%f", pgCodeBudgetGBRatio)
indexParams[NumBuildThreadRatioKey] = fmt.Sprintf("%f", buildNumThreadsRatio)
indexParams[SearchCacheBudgetRatioKey] = fmt.Sprintf("%f", searchCacheBudgetGBRatio)
indexParams[NumLoadThreadRatioKey] = fmt.Sprintf("%f", loadNumThreadRatio)
indexParams[BeamWidthRatioKey] = fmt.Sprintf("%f", beamWidthRatio)
return nil
}
// SetDiskIndexBuildParams set index build params with ratio params on indexNode
// IndexNode cal build param with ratio params and cpu count, memory count...
func SetDiskIndexBuildParams(indexParams map[string]string, numRows int64) error {
dimStr, ok := indexParams["dim"]
if !ok {
// type param dim has been put into index params before build index
return fmt.Errorf("type param dim not exist")
}
dim, err := strconv.ParseInt(dimStr, 10, 64)
if err != nil {
return err
}
pgCodeBudgetGBRatioStr, ok := indexParams[PQCodeBudgetRatioKey]
if !ok {
return fmt.Errorf("index param pgCodeBudgetGBRatio not exist")
}
pgCodeBudgetGBRatio, err := strconv.ParseFloat(pgCodeBudgetGBRatioStr, 64)
if err != nil {
return err
}
buildNumThreadsRatioStr, ok := indexParams[NumBuildThreadRatioKey]
if !ok {
return fmt.Errorf("index param buildNumThreadsRatio not exist")
}
buildNumThreadsRatio, err := strconv.ParseFloat(buildNumThreadsRatioStr, 64)
if err != nil {
return err
}
indexParams[PQCodeBudgetKey] = fmt.Sprintf("%f",
float32(getRowDataSizeOfFloatVector(numRows, dim))*float32(pgCodeBudgetGBRatio)/(1<<30))
indexParams[NumBuildThreadKey] = strconv.Itoa(int(float32(hardware.GetCPUNum()) * float32(buildNumThreadsRatio)))
indexParams[BuildDramBudgetKey] = fmt.Sprintf("%f", float32(hardware.GetFreeMemoryCount())/(1<<30))
return nil
}
// SetDiskIndexLoadParams set disk index load params with ratio params on queryNode
// QueryNode cal load params with ratio params ans cpu count...
func SetDiskIndexLoadParams(indexParams map[string]string, numRows int64) error {
dimStr, ok := indexParams["dim"]
if !ok {
// type param dim has been put into index params before build index
return fmt.Errorf("type param dim not exist")
}
dim, err := strconv.ParseInt(dimStr, 10, 64)
if err != nil {
return err
}
searchCacheBudgetGBRatioStr, ok := indexParams[SearchCacheBudgetRatioKey]
if !ok {
return fmt.Errorf("index param searchCacheBudgetGBRatio not exist")
}
searchCacheBudgetGBRatio, err := strconv.ParseFloat(searchCacheBudgetGBRatioStr, 64)
if err != nil {
return err
}
loadNumThreadRatioStr, ok := indexParams[NumLoadThreadRatioKey]
if !ok {
return fmt.Errorf("index param loadNumThreadRatio not exist")
}
loadNumThreadRatio, err := strconv.ParseFloat(loadNumThreadRatioStr, 64)
if err != nil {
return err
}
beamWidthRatioStr, ok := indexParams[BeamWidthRatioKey]
if !ok {
return fmt.Errorf("index param beamWidthRatio not exist")
}
beamWidthRatio, err := strconv.ParseFloat(beamWidthRatioStr, 64)
if err != nil {
return err
}
indexParams[SearchCacheBudgetKey] = fmt.Sprintf("%f",
float32(getRowDataSizeOfFloatVector(numRows, dim))*float32(searchCacheBudgetGBRatio)/(1<<30))
numLoadThread := int(float32(hardware.GetCPUNum()) * float32(loadNumThreadRatio))
if numLoadThread > MaxLoadThread {
numLoadThread = MaxLoadThread
}
indexParams[NumLoadThreadKey] = strconv.Itoa(numLoadThread)
beamWidth := int(float32(hardware.GetCPUNum()) * float32(beamWidthRatio))
if beamWidth > MaxBeamWidth {
beamWidth = MaxBeamWidth
}
indexParams[BeamWidthKey] = strconv.Itoa(beamWidth)
return nil
}
internal/util/indexparams/disk_index_params_test.go 0 → 100644
浏览文件 @ a1db9038
// Licensed to the LF AI & Data foundation under one
// or more contributor license agreements. See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership. The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License. You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package indexparams
import (
"strconv"
"testing"
"github.com/stretchr/testify/assert"
"github.com/milvus-io/milvus/internal/util/autoindex"
"github.com/milvus-io/milvus/internal/util/paramtable"
)
func TestDiskIndexParams(t *testing.T) {
t.Run("fill index params without auto index param", func(t *testing.T) {
var params paramtable.ComponentParam
params.Init()
indexParams := make(map[string]string)
err := FillDiskIndexParams(&params, indexParams)
assert.NoError(t, err)
pgCodeBudgetGBRatio, err := strconv.ParseFloat(indexParams[PQCodeBudgetRatioKey], 64)
assert.NoError(t, err)
assert.Equal(t, 0.125, pgCodeBudgetGBRatio)
buildNumThreadsRatio, err := strconv.ParseFloat(indexParams[NumBuildThreadRatioKey], 64)
assert.NoError(t, err)
assert.Equal(t, 1.0, buildNumThreadsRatio)
searchCacheBudgetGBRatio, err := strconv.ParseFloat(indexParams[SearchCacheBudgetRatioKey], 64)
assert.NoError(t, err)
assert.Equal(t, 0.125, searchCacheBudgetGBRatio)
loadNumThreadRatio, err := strconv.ParseFloat(indexParams[NumLoadThreadRatioKey], 64)
assert.NoError(t, err)
assert.Equal(t, 8.0, loadNumThreadRatio)
beamWidthRatio, err := strconv.ParseFloat(indexParams[BeamWidthRatioKey], 64)
assert.NoError(t, err)
assert.Equal(t, 4.0, beamWidthRatio)
})
t.Run("fill index params with auto index", func(t *testing.T) {
var params paramtable.ComponentParam
params.AutoIndexConfig.Enable = true
mapString := make(map[string]string)
mapString[autoindex.BuildRatioKey] = "{\"pg_code_budget_gb\": 0.125, \"num_threads\": 1}"
mapString[autoindex.PrepareRatioKey] = "{\"search_cache_budget_gb\": 0.225, \"num_threads\": 4}"
extraParams, err := autoindex.NewBigDataExtraParamsFromMap(mapString)
assert.NoError(t, err)
params.AutoIndexConfig.BigDataExtraParams = extraParams
params.AutoIndexConfig.IndexParams = make(map[string]string)
params.AutoIndexConfig.IndexParams["max_degree"] = "56"
params.AutoIndexConfig.IndexParams["search_list_size"] = "100"
params.AutoIndexConfig.IndexParams["index_type"] = "DISKANN"
indexParams := make(map[string]string)
err = FillDiskIndexParams(&params, indexParams)
assert.NoError(t, err)
pgCodeBudgetGBRatio, err := strconv.ParseFloat(indexParams[PQCodeBudgetRatioKey], 64)
assert.NoError(t, err)
assert.Equal(t, 0.125, pgCodeBudgetGBRatio)
buildNumThreadsRatio, err := strconv.ParseFloat(indexParams[NumBuildThreadRatioKey], 64)
assert.NoError(t, err)
assert.Equal(t, 1.0, buildNumThreadsRatio)
searchCacheBudgetGBRatio, err := strconv.ParseFloat(indexParams[SearchCacheBudgetRatioKey], 64)
assert.NoError(t, err)
assert.Equal(t, 0.225, searchCacheBudgetGBRatio)
loadNumThreadRatio, err := strconv.ParseFloat(indexParams[NumLoadThreadRatioKey], 64)
assert.NoError(t, err)
assert.Equal(t, 4.0, loadNumThreadRatio)
beamWidthRatio, err := strconv.ParseFloat(indexParams[BeamWidthRatioKey], 64)
assert.NoError(t, err)
assert.Equal(t, 4.0, beamWidthRatio)
})
t.Run("set disk index build params", func(t *testing.T) {
indexParams := make(map[string]string)
indexParams[PQCodeBudgetRatioKey] = "0.125"
indexParams[NumBuildThreadRatioKey] = "1.0"
err := SetDiskIndexBuildParams(indexParams, 100)
assert.Error(t, err)
indexParams["dim"] = "128"
err = SetDiskIndexBuildParams(indexParams, 100)
assert.NoError(t, err)
_, ok := indexParams[PQCodeBudgetKey]
assert.True(t, ok)
_, ok = indexParams[BuildDramBudgetKey]
assert.True(t, ok)
_, ok = indexParams[NumBuildThreadKey]
assert.True(t, ok)
})
t.Run("set disk index load params", func(t *testing.T) {
indexParams := make(map[string]string)
indexParams[SearchCacheBudgetRatioKey] = "0.125"
indexParams[NumLoadThreadRatioKey] = "8.0"
indexParams[BeamWidthRatioKey] = "4.0"
err := SetDiskIndexLoadParams(indexParams, 100)
assert.Error(t, err)
indexParams["dim"] = "128"
err = SetDiskIndexLoadParams(indexParams, 100)
assert.NoError(t, err)
_, ok := indexParams[SearchCacheBudgetKey]
assert.True(t, ok)
_, ok = indexParams[NumLoadThreadKey]
assert.True(t, ok)
_, ok = indexParams[BeamWidthKey]
assert.True(t, ok)
})
}
internal/util/paramtable/autoindex_param.go
浏览文件 @ a1db9038
......@@ -92,7 +92,12 @@ func (p *autoIndexConfig) parseBuildParams(paramsStr string) {
}
func (p *autoIndexConfig) parseExtraParams(paramsStr string) {
p.BigDataExtraParams = autoindex.NewBigDataExtraParamsFromJSON(paramsStr)
var err error
p.BigDataExtraParams, err = autoindex.NewBigDataExtraParamsFromJSON(paramsStr)
if err != nil {
err2 := fmt.Errorf("parse auto index extra params failed:%w", err)
panic(err2)
}
}
func (p *autoIndexConfig) parseSearchParams(paramsStr string) {
......
internal/util/paramtable/autoindex_param_test.go
浏览文件 @ a1db9038
......@@ -21,9 +21,10 @@ import (
"strconv"
"testing"
"github.com/stretchr/testify/assert"
"github.com/milvus-io/milvus/internal/common"
"github.com/milvus-io/milvus/internal/util/autoindex"
"github.com/stretchr/testify/assert"
)
const (
......
internal/util/paramtable/component_param.go
浏览文件 @ a1db9038
......@@ -37,6 +37,14 @@ const (
DefaultSessionTTL = 60 //s
DefaultSessionRetryTimes = 30
DefaultMaxDegree = 56
DefaultSearchListSize = 100
DefaultPGCodeBudgetGBRatio = 0.125
DefaultBuildNumThreadsRatio = 1.0
DefaultSearchCacheBudgetGBRatio = 0.125
DefaultLoadNumThreadRatio = 8.0
DefaultBeamWidthRatio = 4.0
)
// ComponentParam is used to quickly and easily access all components' configurations.
......@@ -134,8 +142,15 @@ type commonConfig struct {
RetentionDuration int64
EntityExpirationTTL time.Duration
IndexSliceSize int64
GracefulTime int64
IndexSliceSize int64
MaxDegree int64
SearchListSize int64
PGCodeBudgetGBRatio float64
BuildNumThreadsRatio float64
SearchCacheBudgetGBRatio float64
LoadNumThreadRatio float64
BeamWidthRatio float64
GracefulTime int64
StorageType string
SimdType string
......@@ -179,6 +194,13 @@ func (p *commonConfig) init(base *BaseTable) {
p.initSimdType()
p.initIndexSliceSize()
p.initMaxDegree()
p.initSearchListSize()
p.initPGCodeBudgetGBRatio()
p.initBuildNumThreadsRatio()
p.initSearchCacheBudgetGBRatio()
p.initLoadNumThreadRatio()
p.initBeamWidthRatio()
p.initGracefulTime()
p.initStorageType()
......@@ -379,6 +401,34 @@ func (p *commonConfig) initIndexSliceSize() {
p.IndexSliceSize = p.Base.ParseInt64WithDefault("common.indexSliceSize", DefaultIndexSliceSize)
}
func (p *commonConfig) initPGCodeBudgetGBRatio() {
p.PGCodeBudgetGBRatio = p.Base.ParseFloatWithDefault("common.DiskIndex.PGCodeBudgetGBRatio", DefaultPGCodeBudgetGBRatio)
}
func (p *commonConfig) initBuildNumThreadsRatio() {
p.BuildNumThreadsRatio = p.Base.ParseFloatWithDefault("common.DiskIndex.BuildNumThreadsRatio", DefaultBuildNumThreadsRatio)
}
func (p *commonConfig) initSearchCacheBudgetGBRatio() {
p.SearchCacheBudgetGBRatio = p.Base.ParseFloatWithDefault("common.DiskIndex.SearchCacheBudgetGBRatio", DefaultSearchCacheBudgetGBRatio)
}
func (p *commonConfig) initLoadNumThreadRatio() {
p.LoadNumThreadRatio = p.Base.ParseFloatWithDefault("common.DiskIndex.LoadNumThreadRatio", DefaultLoadNumThreadRatio)
}
func (p *commonConfig) initBeamWidthRatio() {
p.BeamWidthRatio = p.Base.ParseFloatWithDefault("common.DiskIndex.BeamWidthRatio", DefaultBeamWidthRatio)
}
func (p *commonConfig) initMaxDegree() {
p.MaxDegree = p.Base.ParseInt64WithDefault("common.DiskIndex.MaxDegree", DefaultMaxDegree)
}
func (p *commonConfig) initSearchListSize() {
p.SearchListSize = p.Base.ParseInt64WithDefault("common.DiskIndex.SearchListSize", DefaultSearchListSize)
}
func (p *commonConfig) initGracefulTime() {
p.GracefulTime = p.Base.ParseInt64WithDefault("common.gracefulTime", DefaultGracefulTime)
}
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或