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

Merge remote-tracking branch 'mega/branch-0.3.0' into branch-0.3.0 

Former-commit-id: 3f2c2aa48d193802d1786515271b77f61b0b1438
pull/191/head
Yang Xuan 2019-06-12 09:09:02 +08:00
parent 0187266364 cb489d015d
commit e3ecd90368
33 changed files with 1019 additions and 223 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
cpp/CHANGELOG.md
View File

@ -15,6 +15,13 @@ Please mark all change in change log and use the ticket from JIRA.
- MS-64 - Different table can have different index type
- MS-52 - Return search score
- MS-66 - Support time range query
- MS-68 - Remove rocksdb from third-party
- MS-70 - cmake: remove redundant libs in src
- MS-71 - cmake: fix faiss dependency
- MS-72 - cmake: change prometheus source to git
- MS-73 - cmake: delete civetweb
- MS-65 - Implement GetTableRowCount interface
- MS-45 - Implement DeleteTable interface
## Task
@ -44,6 +51,7 @@ Please mark all change in change log and use the ticket from JIRA.
- MS-37 - Add query, cache usage, disk write speed and file data size metrics
- MS-30 - Use faiss v1.5.2
- MS-54 - cmake: Change Thrift third party URL to github.com
- MS-69 - prometheus: add all proposed metrics
## Task

34
cpp/CMakeLists.txt
View File

@ -38,6 +38,40 @@ configure_file(${CMAKE_CURRENT_SOURCE_DIR}/version.h.macro ${CMAKE_CURRENT_SOURC
project(megasearch VERSION "${MEGASEARCH_VERSION}")
project(vecwise_engine LANGUAGES CUDA CXX)
# Ensure that a default make is set
if("${MAKE}" STREQUAL "")
if(NOT MSVC)
find_program(MAKE make)
endif()
endif()
set(GCC_VERSION 5.3.0)
set(GCC_DIR "${CMAKE_CURRENT_BINARY_DIR}/gcc")
set(GCC_BUILD_DIR "${CMAKE_CURRENT_BINARY_DIR}/gcc/gcc_build")
set(GCC_TAR_NAME "${GCC_BUILD_DIR}/gcc-${GCC_VERSION}.tar.gz")
if (NOT EXISTS ${GCC_TAR_NAME})
file(MAKE_DIRECTORY ${GCC_BUILD_DIR})
file(DOWNLOAD http://ftp.tsukuba.wide.ad.jp/software/gcc/releases/gcc-${GCC_VERSION}/${GCC_TAR_NAME}
${GCC_TAR_NAME})
message(STATUS "Building GCC-${GCC_VERSION} from source")
execute_process(
COMMAND
"${CMAKE_COMMAND} -E tar -xf ${GCC_TAR_NAME}"
COMMAND
"./configure --prefix=${GCC_DIR} --with-local-prefix=${GCC_DIR}
--enable-languages=cxx,fortran
--enable-checking=release
--disable-bootstrap"
COMMAND
"${MAKE} -j4"
COMMAND
"${MAKE} install"
RESULT_VARIABLE _status
#OUTPUT_QUIET
#ERROR_QUIET
WORKING_DIRECTORY ${GCC_BUILD_DIR})
endif ()
set(MEGASEARCH_VERSION_MAJOR "${megasearch_VERSION_MAJOR}")
set(MEGASEARCH_VERSION_MINOR "${megasearch_VERSION_MINOR}")
set(MEGASEARCH_VERSION_PATCH "${megasearch_VERSION_PATCH}")

2
cpp/cmake/DefineOptions.cmake
View File

@ -81,7 +81,7 @@ define_option(MEGASEARCH_WITH_OPENBLAS "Build with OpenBLAS library" ON)
define_option(MEGASEARCH_WITH_PROMETHEUS "Build with PROMETHEUS library" ON)
define_option(MEGASEARCH_WITH_ROCKSDB "Build with RocksDB library" ON)
define_option(MEGASEARCH_WITH_ROCKSDB "Build with RocksDB library" OFF)
define_option(MEGASEARCH_WITH_SNAPPY "Build with Snappy compression" ON)

30
cpp/cmake/ThirdPartyPackages.cmake
View File

@ -253,7 +253,8 @@ if (DEFINED ENV{MEGASEARCH_PROMETHEUS_URL})
set(PROMETHEUS_SOURCE_URL "$ENV{PROMETHEUS_OPENBLAS_URL}")
else ()
set(PROMETHEUS_SOURCE_URL
"https://github.com/JinHai-CN/prometheus-cpp/archive/${PROMETHEUS_VERSION}.tar.gz")
#"https://github.com/JinHai-CN/prometheus-cpp/archive/${PROMETHEUS_VERSION}.tar.gz"
https://github.com/jupp0r/prometheus-cpp.git)
endif()
if (DEFINED ENV{MEGASEARCH_ROCKSDB_URL})
@ -716,17 +717,23 @@ macro(build_faiss)
${FAISS_STATIC_LIB})
# DEPENDS
# ${faiss_dependencies})
ExternalProject_Add_StepDependencies(faiss_ep build openblas_ep)
ExternalProject_Add_StepDependencies(faiss_ep build lapack_ep)
ExternalProject_Add_StepDependencies(faiss_ep build openblas_ep lapack_ep)
file(MAKE_DIRECTORY "${FAISS_INCLUDE_DIR}")
add_library(faiss STATIC IMPORTED)
set_target_properties(
faiss
PROPERTIES IMPORTED_LOCATION "${FAISS_STATIC_LIB}"
INTERFACE_INCLUDE_DIRECTORIES "${FAISS_INCLUDE_DIR}")
INTERFACE_INCLUDE_DIRECTORIES "${FAISS_INCLUDE_DIR}"
INTERFACE_LINK_LIBRARIES "openblas;lapack" )
add_dependencies(faiss faiss_ep)
#add_dependencies(faiss openblas_ep)
#add_dependencies(faiss lapack_ep)
#target_link_libraries(faiss ${OPENBLAS_PREFIX}/lib)
#target_link_libraries(faiss ${LAPACK_PREFIX}/lib)
endmacro()
if(MEGASEARCH_WITH_FAISS)
@ -929,11 +936,20 @@ macro(build_prometheus)
${EP_COMMON_CMAKE_ARGS}
-DCMAKE_INSTALL_LIBDIR=lib
-DBUILD_SHARED_LIBS=OFF
"-DCMAKE_INSTALL_PREFIX=${PROMETHEUS_PREFIX}")
"-DCMAKE_INSTALL_PREFIX=${PROMETHEUS_PREFIX}"
-DCMAKE_BUILD_TYPE=Release)
externalproject_add(prometheus_ep
URL
GIT_REPOSITORY
${PROMETHEUS_SOURCE_URL}
GIT_TAG
${PROMETHEUS_VERSION}
GIT_SHALLOW
TRUE
# GIT_CONFIG
# recurse-submodules=true
# URL
# ${PROMETHEUS_SOURCE_URL}
${EP_LOG_OPTIONS}
CMAKE_ARGS
${PROMETHEUS_CMAKE_ARGS}
@ -991,7 +1007,7 @@ if(MEGASEARCH_WITH_PROMETHEUS)
link_directories(SYSTEM ${PROMETHEUS_PREFIX}/core/)
include_directories(SYSTEM ${PROMETHEUS_PREFIX}/core/include)
link_directories(${PROMETHEUS_PREFIX}/civetweb_ep-prefix/src/civetweb_ep)
#link_directories(${PROMETHEUS_PREFIX}/civetweb_ep-prefix/src/civetweb_ep)
endif()
# ----------------------------------------------------------------------

74
cpp/src/CMakeLists.txt
View File

@ -34,6 +34,10 @@ set(service_files
thrift/gen-cpp/MegasearchService.cpp
thrift/gen-cpp/megasearch_constants.cpp
thrift/gen-cpp/megasearch_types.cpp
metrics/SystemInfo.cpp
metrics/SystemInfo.h
server/MegasearchThreadPoolServer.cpp
server/MegasearchThreadPoolServer.h
)
set(vecwise_engine_files
@ -50,32 +54,12 @@ set(get_sys_info_files
license/GetSysInfo.cpp)
include_directories(/usr/include)
include_directories(/usr/local/cuda/include)
include_directories("${CUDA_TOOLKIT_ROOT_DIR}/include")
include_directories(thrift/gen-cpp)
#target_link_libraries(megasearch boost_system_static)
#target_link_libraries(megasearch boost_filesystem_static)
#target_link_libraries(megasearch boost_serialization_static)
#target_link_libraries(megasearch bzip2)
#target_link_libraries(megasearch easyloggingpp)
#target_link_libraries(megasearch faiss)
#target_link_libraries(megasearch gtest)
#target_link_libraries(megasearch lapack)
#target_link_libraries(megasearch lz4)
#target_link_libraries(megasearch openblas)
#target_link_libraries(megasearch rocksdb)
#target_link_libraries(megasearch snappy)
#target_link_libraries(megasearch sqlite)
#target_link_libraries(megasearch sqlite_orm)
#target_link_libraries(megasearch thrift)
#target_link_libraries(megasearch yaml-cpp)
#target_link_libraries(megasearch zlib)
#target_link_libraries(megasearch zstd)
set(third_party_libs
easyloggingpp
sqlite
# sqlite_orm
thrift
yaml-cpp
faiss
@ -84,7 +68,6 @@ set(third_party_libs
prometheus-cpp-push
prometheus-cpp-pull
prometheus-cpp-core
civetweb
boost_system_static
boost_filesystem_static
boost_serialization_static
@ -93,48 +76,35 @@ set(third_party_libs
snappy
zlib
zstd
${CUDA_TOOLKIT_ROOT_DIR}/lib64/stubs/libnvidia-ml.so
)
if (GPU_VERSION STREQUAL "ON")
link_directories(/usr/local/cuda/lib64)
link_directories("${CUDA_TOOLKIT_ROOT_DIR}/lib64")
set(engine_libs
pthread
libfaiss.a
libgpufaiss.a
libgomp.a
libopenblas.a
libgfortran.a
libquadmath.a
cudart
cublas
libsqlite3.a
libprometheus-cpp-push.a
libprometheus-cpp-pull.a
libprometheus-cpp-core.a
${CUDA_TOOLKIT_ROOT_DIR}/lib64/stubs/libnvidia-ml.so
)
else()
set(engine_libs
pthread
libfaiss.a
libgomp.a
libopenblas.a
libgfortran.a
libquadmath.a
libsqlite3.a
libprometheus-cpp-push.a
libprometheus-cpp-pull.a
libprometheus-cpp-core.a
${CUDA_TOOLKIT_ROOT_DIR}/lib64/stubs/libnvidia-ml.so
)
endif ()
if (ENABLE_LICENSE STREQUAL "ON")
link_directories(/usr/local/cuda/lib64/stubs)
link_directories(/usr/local/cuda/lib64)
link_directories("${CUDA_TOOLKIT_ROOT_DIR}/lib64/stubs")
link_directories("${CUDA_TOOLKIT_ROOT_DIR}/lib64")
set(license_libs
nvidia-ml
libboost_system.a
libboost_filesystem.a
libboost_serialization.a
crypto
cudart
cublas
@ -153,30 +123,20 @@ if (ENABLE_LICENSE STREQUAL "ON")
target_link_libraries(vecwise_license ${license_libs} ${third_party_libs})
endif ()
#set(metrics_lib
# libprometheus-cpp-push.a
# libprometheus-cpp-pull.a
# libprometheus-cpp-core.a
# )
set(metrics_lib
prometheus-cpp-push
prometheus-cpp-pull
prometheus-cpp-core
)
#add_library(vecwise_engine STATIC ${metrics_files} )
#target_link_libraries(metrics ${metrics_lib})
target_link_libraries(metrics ${metrics_lib})
set(server_libs
vecwise_engine
libthrift.a
pthread
libyaml-cpp.a
libboost_system.a
libboost_filesystem.a
libsnappy.a
libbz2.a
libz.a
libzstd.a
liblz4.a
dl
metrics
)
add_executable(vecwise_server

13
cpp/src/db/DB.h
View File

@ -23,19 +23,22 @@ public:
static void Open(const Options& options, DB** dbptr);
virtual Status CreateTable(meta::TableSchema& table_schema_) = 0;
virtual Status DeleteTable(const std::string& table_id, const meta::DatesT& dates) = 0;
virtual Status DescribeTable(meta::TableSchema& table_schema_) = 0;
virtual Status HasTable(const std::string& table_id_, bool& has_or_not_) = 0;
virtual Status HasTable(const std::string& table_id, bool& has_or_not_) = 0;
virtual Status AllTables(std::vector<meta::TableSchema>& table_schema_array) = 0;
virtual Status GetTableRowCount(const std::string& table_id, uint64_t& row_count) = 0;
virtual Status InsertVectors(const std::string& table_id_,
size_t n, const float* vectors, IDNumbers& vector_ids_) = 0;
uint64_t n, const float* vectors, IDNumbers& vector_ids_) = 0;
virtual Status Query(const std::string& table_id, size_t k, size_t nq,
virtual Status Query(const std::string& table_id, uint64_t k, uint64_t nq,
const float* vectors, QueryResults& results) = 0;
virtual Status Query(const std::string& table_id, size_t k, size_t nq,
virtual Status Query(const std::string& table_id, uint64_t k, uint64_t nq,
const float* vectors, const meta::DatesT& dates, QueryResults& results) = 0;
virtual Status Size(long& result) = 0;
virtual Status Size(uint64_t& result) = 0;
virtual Status DropAll() = 0;

78
cpp/src/db/DBImpl.cpp
View File

@ -6,6 +6,7 @@
#include "DBImpl.h"
#include "DBMetaImpl.h"
#include "Env.h"
#include "Log.h"
#include "EngineFactory.h"
#include "metrics/Metrics.h"
#include "scheduler/SearchScheduler.h"
@ -15,8 +16,8 @@
#include <thread>
#include <iostream>
#include <cstring>
#include <easylogging++.h>
#include <cache/CpuCacheMgr.h>
#include <boost/filesystem.hpp>
namespace zilliz {
namespace vecwise {
@ -88,6 +89,34 @@ Status DBImpl::CreateTable(meta::TableSchema& table_schema) {
return pMeta_->CreateTable(table_schema);
}
Status DBImpl::DeleteTable(const std::string& table_id, const meta::DatesT& dates) {
meta::DatePartionedTableFilesSchema files;
auto status = pMeta_->FilesToDelete(table_id, dates, files);
if (!status.ok()) { return status; }
for (auto &day_files : files) {
for (auto &file : day_files.second) {
boost::filesystem::remove(file.location_);
}
}
//dates empty means delete all files of the table
if(dates.empty()) {
meta::TableSchema table_schema;
table_schema.table_id_ = table_id;
status = DescribeTable(table_schema);
pMeta_->DeleteTable(table_id);
boost::system::error_code ec;
boost::filesystem::remove_all(table_schema.location_, ec);
if(ec.failed()) {
ENGINE_LOG_WARNING << "Failed to remove table folder";
}
}
return Status::OK();
}
Status DBImpl::DescribeTable(meta::TableSchema& table_schema) {
return pMeta_->DescribeTable(table_schema);
}
@ -96,8 +125,16 @@ Status DBImpl::HasTable(const std::string& table_id, bool& has_or_not) {
return pMeta_->HasTable(table_id, has_or_not);
}
Status DBImpl::AllTables(std::vector<meta::TableSchema>& table_schema_array) {
return pMeta_->AllTables(table_schema_array);
}
Status DBImpl::GetTableRowCount(const std::string& table_id, uint64_t& row_count) {
return pMeta_->Count(table_id, row_count);
}
Status DBImpl::InsertVectors(const std::string& table_id_,
size_t n, const float* vectors, IDNumbers& vector_ids_) {
uint64_t n, const float* vectors, IDNumbers& vector_ids_) {
auto start_time = METRICS_NOW_TIME;
Status status = pMemMgr_->InsertVectors(table_id_, n, vectors, vector_ids_);
@ -108,9 +145,10 @@ Status DBImpl::InsertVectors(const std::string& table_id_,
CollectInsertMetrics(total_time, n, status.ok());
return status;
}
Status DBImpl::Query(const std::string &table_id, size_t k, size_t nq,
Status DBImpl::Query(const std::string &table_id, uint64_t k, uint64_t nq,
const float *vectors, QueryResults &results) {
auto start_time = METRICS_NOW_TIME;
meta::DatesT dates = {meta::Meta::GetDate()};
@ -119,10 +157,11 @@ Status DBImpl::Query(const std::string &table_id, size_t k, size_t nq,
auto total_time = METRICS_MICROSECONDS(start_time,end_time);
CollectQueryMetrics(total_time, nq);
return result;
}
Status DBImpl::Query(const std::string& table_id, size_t k, size_t nq,
Status DBImpl::Query(const std::string& table_id, uint64_t k, uint64_t nq,
const float* vectors, const meta::DatesT& dates, QueryResults& results) {
#if 0
return QuerySync(table_id, k, nq, vectors, dates, results);
@ -131,13 +170,13 @@ Status DBImpl::Query(const std::string& table_id, size_t k, size_t nq,
#endif
}
Status DBImpl::QuerySync(const std::string& table_id, size_t k, size_t nq,
Status DBImpl::QuerySync(const std::string& table_id, uint64_t k, uint64_t nq,
const float* vectors, const meta::DatesT& dates, QueryResults& results) {
meta::DatePartionedTableFilesSchema files;
auto status = pMeta_->FilesToSearch(table_id, dates, files);
if (!status.ok()) { return status; }
LOG(DEBUG) << "Search DateT Size=" << files.size();
ENGINE_LOG_DEBUG << "Search DateT Size = " << files.size();
meta::TableFilesSchema index_files;
meta::TableFilesSchema raw_files;
@ -154,7 +193,7 @@ Status DBImpl::QuerySync(const std::string& table_id, size_t k, size_t nq,
} else if (!raw_files.empty()) {
dim = raw_files[0].dimension_;
} else {
LOG(DEBUG) << "no files to search";
ENGINE_LOG_DEBUG << "no files to search";
return Status::OK();
}
@ -190,7 +229,7 @@ Status DBImpl::QuerySync(const std::string& table_id, size_t k, size_t nq,
auto file_size = index->PhysicalSize();
search_set_size += file_size;
LOG(DEBUG) << "Search file_type " << file.file_type_ << " Of Size: "
ENGINE_LOG_DEBUG << "Search file_type " << file.file_type_ << " Of Size: "
<< file_size/(1024*1024) << " M";
int inner_k = index->Count() < k ? index->Count() : k;
@ -252,7 +291,7 @@ Status DBImpl::QuerySync(const std::string& table_id, size_t k, size_t nq,
search_in_index(raw_files);
search_in_index(index_files);
LOG(DEBUG) << "Search Overall Set Size=" << search_set_size << " M";
ENGINE_LOG_DEBUG << "Search Overall Set Size = " << search_set_size << " M";
cluster_topk();
free(output_distence);
@ -265,7 +304,7 @@ Status DBImpl::QuerySync(const std::string& table_id, size_t k, size_t nq,
return Status::OK();
}
Status DBImpl::QueryAsync(const std::string& table_id, size_t k, size_t nq,
Status DBImpl::QueryAsync(const std::string& table_id, uint64_t k, uint64_t nq,
const float* vectors, const meta::DatesT& dates, QueryResults& results) {
//step 1: get files to search
@ -273,7 +312,7 @@ Status DBImpl::QueryAsync(const std::string& table_id, size_t k, size_t nq,
auto status = pMeta_->FilesToSearch(table_id, dates, files);
if (!status.ok()) { return status; }
LOG(DEBUG) << "Search DateT Size=" << files.size();
ENGINE_LOG_DEBUG << "Search DateT Size=" << files.size();
SearchContextPtr context = std::make_shared<SearchContext>(k, nq, vectors);
@ -304,17 +343,24 @@ void DBImpl::StartTimerTasks(int interval) {
void DBImpl::BackgroundTimerTask(int interval) {
Status status;
server::SystemInfo::GetInstance().Init();
while (true) {
if (!bg_error_.ok()) break;
if (shutting_down_.load(std::memory_order_acquire)) break;
std::this_thread::sleep_for(std::chrono::seconds(interval));
int64_t cache_total = cache::CpuCacheMgr::GetInstance()->CacheUsage();
LOG(DEBUG) << "Cache usage " << cache_total;
server::Metrics::GetInstance().CacheUsageGaugeSet(static_cast<double>(cache_total));
long size;
server::Metrics::GetInstance().KeepingAliveCounterIncrement(interval);
int64_t cache_usage = cache::CpuCacheMgr::GetInstance()->CacheUsage();
int64_t cache_total = cache::CpuCacheMgr::GetInstance()->CacheCapacity();
server::Metrics::GetInstance().CacheUsageGaugeSet(cache_usage*100/cache_total);
uint64_t size;
Size(size);
server::Metrics::GetInstance().DataFileSizeGaugeSet(size);
server::Metrics::GetInstance().CPUUsagePercentSet();
server::Metrics::GetInstance().RAMUsagePercentSet();
server::Metrics::GetInstance().GPUPercentGaugeSet();
server::Metrics::GetInstance().GPUMemoryUsageGaugeSet();
TrySchedule();
}
}
@ -509,7 +555,7 @@ Status DBImpl::DropAll() {
return pMeta_->DropAll();
}
Status DBImpl::Size(long& result) {
Status DBImpl::Size(uint64_t& result) {
return pMeta_->Size(result);
}

15
cpp/src/db/DBImpl.h
View File

@ -33,29 +33,32 @@ public:
DBImpl(const Options& options);
virtual Status CreateTable(meta::TableSchema& table_schema) override;
virtual Status DeleteTable(const std::string& table_id, const meta::DatesT& dates) override;
virtual Status DescribeTable(meta::TableSchema& table_schema) override;
virtual Status HasTable(const std::string& table_id, bool& has_or_not) override;
virtual Status AllTables(std::vector<meta::TableSchema>& table_schema_array) override;
virtual Status GetTableRowCount(const std::string& table_id, uint64_t& row_count) override;
virtual Status InsertVectors(const std::string& table_id,
size_t n, const float* vectors, IDNumbers& vector_ids) override;
uint64_t n, const float* vectors, IDNumbers& vector_ids) override;
virtual Status Query(const std::string& table_id, size_t k, size_t nq,
virtual Status Query(const std::string& table_id, uint64_t k, uint64_t nq,
const float* vectors, QueryResults& results) override;
virtual Status Query(const std::string& table_id, size_t k, size_t nq,
virtual Status Query(const std::string& table_id, uint64_t k, uint64_t nq,
const float* vectors, const meta::DatesT& dates, QueryResults& results) override;
virtual Status DropAll() override;
virtual Status Size(long& result) override;
virtual Status Size(uint64_t& result) override;
virtual ~DBImpl();
private:
Status QuerySync(const std::string& table_id, size_t k, size_t nq,
Status QuerySync(const std::string& table_id, uint64_t k, uint64_t nq,
const float* vectors, const meta::DatesT& dates, QueryResults& results);
Status QueryAsync(const std::string& table_id, size_t k, size_t nq,
Status QueryAsync(const std::string& table_id, uint64_t k, uint64_t nq,
const float* vectors, const meta::DatesT& dates, QueryResults& results);

207
cpp/src/db/DBMetaImpl.cpp
View File

@ -6,6 +6,7 @@
#include "DBMetaImpl.h"
#include "IDGenerator.h"
#include "Utils.h"
#include "Log.h"
#include "MetaConsts.h"
#include "Factories.h"
#include "metrics/Metrics.h"
@ -17,7 +18,6 @@
#include <chrono>
#include <fstream>
#include <sqlite_orm.h>
#include <easylogging++.h>
namespace zilliz {
@ -27,6 +27,15 @@ namespace meta {
using namespace sqlite_orm;
namespace {
void HandleException(std::exception &e) {
ENGINE_LOG_DEBUG << "Engine meta exception: " << e.what();
throw e;
}
}
inline auto StoragePrototype(const std::string &path) {
return make_storage(path,
make_table("Table",
@ -100,7 +109,7 @@ Status DBMetaImpl::Initialize() {
if (!boost::filesystem::is_directory(options_.path)) {
auto ret = boost::filesystem::create_directory(options_.path);
if (!ret) {
LOG(ERROR) << "Create directory " << options_.path << " Error";
ENGINE_LOG_ERROR << "Create directory " << options_.path << " Error";
}
assert(ret);
}
@ -148,8 +157,7 @@ Status DBMetaImpl::DropPartitionsByDates(const std::string &table_id,
in(&TableFileSchema::date_, dates)
));
} catch (std::exception &e) {
LOG(DEBUG) << e.what();
throw e;
HandleException(e);
}
return Status::OK();
}
@ -175,12 +183,12 @@ Status DBMetaImpl::CreateTable(TableSchema &table_schema) {
auto total_time = METRICS_MICROSECONDS(start_time, end_time);
server::Metrics::GetInstance().MetaAccessDurationSecondsHistogramObserve(total_time);
auto group_path = GetTablePath(table_schema.table_id_);
if (!boost::filesystem::is_directory(group_path)) {
auto ret = boost::filesystem::create_directories(group_path);
auto table_path = GetTablePath(table_schema.table_id_);
table_schema.location_ = table_path;
if (!boost::filesystem::is_directory(table_path)) {
auto ret = boost::filesystem::create_directories(table_path);
if (!ret) {
LOG(ERROR) << "Create directory " << group_path << " Error";
ENGINE_LOG_ERROR << "Create directory " << table_path << " Error";
}
assert(ret);
}
@ -188,6 +196,21 @@ Status DBMetaImpl::CreateTable(TableSchema &table_schema) {
return Status::OK();
}
Status DBMetaImpl::DeleteTable(const std::string& table_id) {
try {
//drop the table from meta
auto tables = ConnectorPtr->select(columns(&TableSchema::id_),
where(c(&TableSchema::table_id_) == table_id));
for (auto &table : tables) {
ConnectorPtr->remove<TableSchema>(std::get<0>(table));
}
} catch (std::exception &e) {
HandleException(e);
}
return Status::OK();
}
Status DBMetaImpl::DescribeTable(TableSchema &table_schema) {
try {
server::Metrics::GetInstance().MetaAccessTotalIncrement();
@ -212,9 +235,12 @@ Status DBMetaImpl::DescribeTable(TableSchema &table_schema) {
} else {
return Status::NotFound("Table " + table_schema.table_id_ + " not found");
}
auto table_path = GetTablePath(table_schema.table_id_);
table_schema.location_ = table_path;
} catch (std::exception &e) {
LOG(DEBUG) << e.what();
throw e;
HandleException(e);
}
return Status::OK();
@ -237,12 +263,42 @@ Status DBMetaImpl::HasTable(const std::string &table_id, bool &has_or_not) {
has_or_not = false;
}
} catch (std::exception &e) {
LOG(DEBUG) << e.what();
throw e;
HandleException(e);
}
return Status::OK();
}
Status DBMetaImpl::AllTables(std::vector<TableSchema>& table_schema_array) {
try {
server::Metrics::GetInstance().MetaAccessTotalIncrement();
auto start_time = METRICS_NOW_TIME;
auto selected = ConnectorPtr->select(columns(&TableSchema::id_,
&TableSchema::table_id_,
&TableSchema::files_cnt_,
&TableSchema::dimension_,
&TableSchema::engine_type_,
&TableSchema::store_raw_data_));
auto end_time = METRICS_NOW_TIME;
auto total_time = METRICS_MICROSECONDS(start_time, end_time);
server::Metrics::GetInstance().MetaAccessDurationSecondsHistogramObserve(total_time);
for (auto &table : selected) {
TableSchema schema;
schema.id_ = std::get<0>(table);
schema.table_id_ = std::get<1>(table);
schema.files_cnt_ = std::get<2>(table);
schema.dimension_ = std::get<3>(table);
schema.engine_type_ = std::get<4>(table);
schema.store_raw_data_ = std::get<5>(table);
table_schema_array.emplace_back(schema);
}
} catch (std::exception &e) {
HandleException(e);
}
return Status::OK();
}
Status DBMetaImpl::CreateTableFile(TableFileSchema &file_schema) {
if (file_schema.date_ == EmptyDate) {
file_schema.date_ = Meta::GetDate();
@ -282,7 +338,7 @@ Status DBMetaImpl::CreateTableFile(TableFileSchema &file_schema) {
if (!boost::filesystem::is_directory(partition_path)) {
auto ret = boost::filesystem::create_directory(partition_path);
if (!ret) {
LOG(ERROR) << "Create directory " << partition_path << " Error";
ENGINE_LOG_ERROR << "Create directory " << partition_path << " Error";
}
assert(ret);
}
@ -336,8 +392,7 @@ Status DBMetaImpl::FilesToIndex(TableFilesSchema &files) {
files.push_back(table_file);
}
} catch (std::exception &e) {
LOG(DEBUG) << e.what();
throw e;
HandleException(e);
}
return Status::OK();
@ -438,8 +493,7 @@ Status DBMetaImpl::FilesToSearch(const std::string &table_id,
}
} catch (std::exception &e) {
LOG(DEBUG) << e.what();
throw e;
HandleException(e);
}
return Status::OK();
@ -488,8 +542,79 @@ Status DBMetaImpl::FilesToMerge(const std::string &table_id,
files[table_file.date_].push_back(table_file);
}
} catch (std::exception &e) {
LOG(DEBUG) << e.what();
throw e;
HandleException(e);
}
return Status::OK();
}
Status DBMetaImpl::FilesToDelete(const std::string& table_id,
const DatesT& partition,
DatePartionedTableFilesSchema& files) {
auto now = utils::GetMicroSecTimeStamp();
try {
if(partition.empty()) {
//step 1: get table files by dates
auto selected = ConnectorPtr->select(columns(&TableFileSchema::id_,
&TableFileSchema::table_id_,
&TableFileSchema::file_id_,
&TableFileSchema::size_,
&TableFileSchema::date_),
where(c(&TableFileSchema::file_type_) !=
(int) TableFileSchema::TO_DELETE
and c(&TableFileSchema::table_id_) == table_id));
//step 2: erase table files from meta
for (auto &file : selected) {
TableFileSchema table_file;
table_file.id_ = std::get<0>(file);
table_file.table_id_ = std::get<1>(file);
table_file.file_id_ = std::get<2>(file);
table_file.size_ = std::get<3>(file);
table_file.date_ = std::get<4>(file);
GetTableFilePath(table_file);
auto dateItr = files.find(table_file.date_);
if (dateItr == files.end()) {
files[table_file.date_] = TableFilesSchema();
}
files[table_file.date_].push_back(table_file);
ConnectorPtr->remove<TableFileSchema>(std::get<0>(file));
}
} else {
//step 1: get all table files
auto selected = ConnectorPtr->select(columns(&TableFileSchema::id_,
&TableFileSchema::table_id_,
&TableFileSchema::file_id_,
&TableFileSchema::size_,
&TableFileSchema::date_),
where(c(&TableFileSchema::file_type_) !=
(int) TableFileSchema::TO_DELETE
and in(&TableFileSchema::date_, partition)
and c(&TableFileSchema::table_id_) == table_id));
//step 2: erase table files from meta
for (auto &file : selected) {
TableFileSchema table_file;
table_file.id_ = std::get<0>(file);
table_file.table_id_ = std::get<1>(file);
table_file.file_id_ = std::get<2>(file);
table_file.size_ = std::get<3>(file);
table_file.date_ = std::get<4>(file);
GetTableFilePath(table_file);
auto dateItr = files.find(table_file.date_);
if (dateItr == files.end()) {
files[table_file.date_] = TableFilesSchema();
}
files[table_file.date_].push_back(table_file);
ConnectorPtr->remove<TableFileSchema>(std::get<0>(file));
}
}
} catch (std::exception &e) {
HandleException(e);
}
return Status::OK();
@ -520,8 +645,7 @@ Status DBMetaImpl::GetTableFile(TableFileSchema &file_schema) {
" File:" + file_schema.file_id_ + " not found");
}
} catch (std::exception &e) {
LOG(DEBUG) << e.what();
throw e;
HandleException(e);
}
return Status::OK();
@ -550,12 +674,11 @@ Status DBMetaImpl::Archive() {
c(&TableFileSchema::file_type_) != (int) TableFileSchema::TO_DELETE
));
} catch (std::exception &e) {
LOG(DEBUG) << e.what();
throw e;
HandleException(e);
}
}
if (criteria == "disk") {
long sum = 0;
uint64_t sum = 0;
Size(sum);
auto to_delete = (sum - limit * G);
@ -566,7 +689,7 @@ Status DBMetaImpl::Archive() {
return Status::OK();
}
Status DBMetaImpl::Size(long &result) {
Status DBMetaImpl::Size(uint64_t &result) {
result = 0;
try {
auto selected = ConnectorPtr->select(columns(sum(&TableFileSchema::size_)),
@ -578,11 +701,10 @@ Status DBMetaImpl::Size(long &result) {
if (!std::get<0>(sub_query)) {
continue;
}
result += (long) (*std::get<0>(sub_query));
result += (uint64_t) (*std::get<0>(sub_query));
}
} catch (std::exception &e) {
LOG(DEBUG) << e.what();
throw e;
HandleException(e);
}
return Status::OK();
@ -609,7 +731,8 @@ Status DBMetaImpl::DiscardFiles(long to_discard_size) {
table_file.id_ = std::get<0>(file);
table_file.size_ = std::get<1>(file);
ids.push_back(table_file.id_);
LOG(DEBUG) << "Discard table_file.id=" << table_file.file_id_ << " table_file.size=" << table_file.size_;
ENGINE_LOG_DEBUG << "Discard table_file.id=" << table_file.file_id_
<< " table_file.size=" << table_file.size_;
to_discard_size -= table_file.size_;
}
@ -626,11 +749,9 @@ Status DBMetaImpl::DiscardFiles(long to_discard_size) {
));
} catch (std::exception &e) {
LOG(DEBUG) << e.what();
throw e;
HandleException(e);
}
return DiscardFiles(to_discard_size);
}
@ -644,9 +765,8 @@ Status DBMetaImpl::UpdateTableFile(TableFileSchema &file_schema) {
auto total_time = METRICS_MICROSECONDS(start_time, end_time);
server::Metrics::GetInstance().MetaAccessDurationSecondsHistogramObserve(total_time);
} catch (std::exception &e) {
LOG(DEBUG) << e.what();
LOG(DEBUG) << "table_id= " << file_schema.table_id_ << " file_id=" << file_schema.file_id_;
throw e;
ENGINE_LOG_DEBUG << "table_id= " << file_schema.table_id_ << " file_id=" << file_schema.file_id_;
HandleException(e);
}
return Status::OK();
}
@ -669,8 +789,7 @@ Status DBMetaImpl::UpdateTableFiles(TableFilesSchema &files) {
return Status::DBTransactionError("Update files Error");
}
} catch (std::exception &e) {
LOG(DEBUG) << e.what();
throw e;
HandleException(e);
}
return Status::OK();
}
@ -708,8 +827,7 @@ Status DBMetaImpl::CleanUpFilesWithTTL(uint16_t seconds) {
/* LOG(DEBUG) << "Removing deleted id=" << table_file.id << " location=" << table_file.location << std::endl; */
}
} catch (std::exception &e) {
LOG(DEBUG) << e.what();
throw e;
HandleException(e);
}
return Status::OK();
@ -747,14 +865,13 @@ Status DBMetaImpl::CleanUp() {
/* LOG(DEBUG) << "Removing id=" << table_file.id << " location=" << table_file.location << std::endl; */
}
} catch (std::exception &e) {
LOG(DEBUG) << e.what();
throw e;
HandleException(e);
}
return Status::OK();
}
Status DBMetaImpl::Count(const std::string &table_id, long &result) {
Status DBMetaImpl::Count(const std::string &table_id, uint64_t &result) {
try {
@ -785,10 +902,10 @@ Status DBMetaImpl::Count(const std::string &table_id, long &result) {
}
result /= table_schema.dimension_;
result /= sizeof(float);
} catch (std::exception &e) {
LOG(DEBUG) << e.what();
throw e;
HandleException(e);
}
return Status::OK();
}

10
cpp/src/db/DBMetaImpl.h
View File

@ -20,8 +20,10 @@ public:
DBMetaImpl(const DBMetaOptions& options_);
virtual Status CreateTable(TableSchema& table_schema) override;
virtual Status DeleteTable(const std::string& table_id) override;
virtual Status DescribeTable(TableSchema& group_info_) override;
virtual Status HasTable(const std::string& table_id, bool& has_or_not) override;
virtual Status AllTables(std::vector<TableSchema>& table_schema_array) override;
virtual Status CreateTableFile(TableFileSchema& file_schema) override;
virtual Status DropPartitionsByDates(const std::string& table_id,
@ -40,11 +42,15 @@ public:
virtual Status FilesToMerge(const std::string& table_id,
DatePartionedTableFilesSchema& files) override;
virtual Status FilesToDelete(const std::string& table_id,
const DatesT& partition,
DatePartionedTableFilesSchema& files) override;
virtual Status FilesToIndex(TableFilesSchema&) override;
virtual Status Archive() override;
virtual Status Size(long& result) override;
virtual Status Size(uint64_t& result) override;
virtual Status CleanUp() override;
@ -52,7 +58,7 @@ public:
virtual Status DropAll() override;
virtual Status Count(const std::string& table_id, long& result) override;
virtual Status Count(const std::string& table_id, uint64_t& result) override;
virtual ~DBMetaImpl();

5
cpp/src/db/FaissExecutionEngine.cpp
View File

@ -131,8 +131,11 @@ Status FaissExecutionEngine::Search(long n,
long k,
float *distances,
long *labels) const {
auto start_time = METRICS_NOW_TIME;
pIndex_->search(n, data, k, distances, labels);
auto end_time = METRICS_NOW_TIME;
auto total_time = METRICS_MICROSECONDS(start_time,end_time);
server::Metrics::GetInstance().QueryIndexTypePerSecondSet(build_index_type_, double(n)/double(total_time));
return Status::OK();
}

6
cpp/src/db/MemManager.cpp
View File

@ -27,9 +27,14 @@ MemVectors::MemVectors(const std::shared_ptr<meta::Meta>& meta_ptr,
pEE_(EngineFactory::Build(schema_.dimension_, schema_.location_, (EngineType)schema_.engine_type_)) {
}
void MemVectors::Add(size_t n_, const float* vectors_, IDNumbers& vector_ids_) {
auto start_time = METRICS_NOW_TIME;
pIdGenerator_->GetNextIDNumbers(n_, vector_ids_);
pEE_->AddWithIds(n_, vectors_, vector_ids_.data());
auto end_time = METRICS_NOW_TIME;
auto total_time = METRICS_MICROSECONDS(start_time, end_time);
server::Metrics::GetInstance().AddVectorsPerSecondGaugeSet(static_cast<int>(n_), static_cast<int>(schema_.dimension_), total_time);
}
size_t MemVectors::Total() const {
@ -97,6 +102,7 @@ Status MemManager::InsertVectors(const std::string& table_id_,
const float* vectors_,
IDNumbers& vector_ids_) {
std::unique_lock<std::mutex> lock(mutex_);
return InsertVectorsNoLock(table_id_, n_, vectors_, vector_ids_);
}

10
cpp/src/db/Meta.h
View File

@ -24,8 +24,10 @@ public:
using Ptr = std::shared_ptr<Meta>;
virtual Status CreateTable(TableSchema& table_schema) = 0;
virtual Status DeleteTable(const std::string& table_id) = 0;
virtual Status DescribeTable(TableSchema& table_schema) = 0;
virtual Status HasTable(const std::string& table_id, bool& has_or_not) = 0;
virtual Status AllTables(std::vector<TableSchema>& table_schema_array) = 0;
virtual Status CreateTableFile(TableFileSchema& file_schema) = 0;
virtual Status DropPartitionsByDates(const std::string& table_id,
@ -43,7 +45,11 @@ public:
virtual Status FilesToMerge(const std::string& table_id,
DatePartionedTableFilesSchema& files) = 0;
virtual Status Size(long& result) = 0;
virtual Status FilesToDelete(const std::string& table_id,
const DatesT& partition,
DatePartionedTableFilesSchema& files) = 0;
virtual Status Size(uint64_t& result) = 0;
virtual Status Archive() = 0;
@ -54,7 +60,7 @@ public:
virtual Status DropAll() = 0;
virtual Status Count(const std::string& table_id, long& result) = 0;
virtual Status Count(const std::string& table_id, uint64_t& result) = 0;
static DateT GetDate(const std::time_t& t, int day_delta = 0);
static DateT GetDate();

2
cpp/src/db/scheduler/SearchTaskQueue.cpp
View File

@ -111,7 +111,7 @@ bool SearchTask::DoSearch() {
return false;
}
server::TimeRecorder rc("DoSearch");
server::TimeRecorder rc("DoSearch index(" + std::to_string(index_id_) + ")");
std::vector<long> output_ids;
std::vector<float> output_distence;

11
cpp/src/metrics/MetricBase.h
View File

@ -8,6 +8,7 @@
#include "utils/Error.h"
#include "server/ServerConfig.h"
#include "SystemInfo.h"
namespace zilliz {
namespace vecwise {
@ -71,6 +72,16 @@ class MetricsBase{
virtual void AddVectorsFailGaugeSet(double value) {};
virtual void QueryVectorResponseSummaryObserve(double value, int count = 1) {};
virtual void QueryVectorResponsePerSecondGaugeSet(double value) {};
virtual void CPUUsagePercentSet() {};
virtual void RAMUsagePercentSet() {};
virtual void QueryResponsePerSecondGaugeSet(double value) {};
virtual void GPUPercentGaugeSet() {};
virtual void GPUMemoryUsageGaugeSet() {};
virtual void AddVectorsPerSecondGaugeSet(int num_vector, int dim, double time) {};
virtual void QueryIndexTypePerSecondSet(std::string type, double value) {};
virtual void ConnectionGaugeIncrement() {};
virtual void ConnectionGaugeDecrement() {};
virtual void KeepingAliveCounterIncrement(double value = 1) {};
};

102
cpp/src/metrics/PrometheusMetrics.cpp
View File

@ -6,6 +6,8 @@
#include "PrometheusMetrics.h"
#include "utils/Log.h"
#include "SystemInfo.h"
namespace zilliz {
namespace vecwise {
@ -32,8 +34,108 @@ PrometheusMetrics::Init() {
}
return SERVER_SUCCESS;
}
void
PrometheusMetrics::CPUUsagePercentSet() {
if(!startup_) return ;
double usage_percent = server::SystemInfo::GetInstance().CPUPercent();
CPU_usage_percent_.Set(usage_percent);
}
void
PrometheusMetrics::RAMUsagePercentSet() {
if(!startup_) return ;
double usage_percent = server::SystemInfo::GetInstance().MemoryPercent();
RAM_usage_percent_.Set(usage_percent);
}
void
PrometheusMetrics::GPUPercentGaugeSet() {
if(!startup_) return;
int numDevide = server::SystemInfo::GetInstance().num_device();
std::vector<unsigned int> values = server::SystemInfo::GetInstance().GPUPercent();
// for (int i = 0; i < numDevide; ++i) {
// GPU_percent_gauges_[i].Set(static_cast<double>(values[i]));
// }
if(numDevide >= 1) GPU0_percent_gauge_.Set(static_cast<double>(values[0]));
if(numDevide >= 2) GPU1_percent_gauge_.Set(static_cast<double>(values[1]));
if(numDevide >= 3) GPU2_percent_gauge_.Set(static_cast<double>(values[2]));
if(numDevide >= 4) GPU3_percent_gauge_.Set(static_cast<double>(values[3]));
if(numDevide >= 5) GPU4_percent_gauge_.Set(static_cast<double>(values[4]));
if(numDevide >= 6) GPU5_percent_gauge_.Set(static_cast<double>(values[5]));
if(numDevide >= 7) GPU6_percent_gauge_.Set(static_cast<double>(values[6]));
if(numDevide >= 8) GPU7_percent_gauge_.Set(static_cast<double>(values[7]));
// to do
}
void PrometheusMetrics::GPUMemoryUsageGaugeSet() {
if(!startup_) return;
int numDevide = server::SystemInfo::GetInstance().num_device();
std::vector<unsigned long long> values = server::SystemInfo::GetInstance().GPUMemoryUsed();
constexpr unsigned long long MtoB = 1024*1024;
int numDevice = values.size();
// for (int i = 0; i < numDevice; ++i) {
// GPU_memory_usage_gauges_[i].Set(values[i]/MtoB);
// }
if(numDevice >=1) GPU0_memory_usage_gauge_.Set(values[0]/MtoB);
if(numDevice >=2) GPU1_memory_usage_gauge_.Set(values[1]/MtoB);
if(numDevice >=3) GPU2_memory_usage_gauge_.Set(values[2]/MtoB);
if(numDevice >=4) GPU3_memory_usage_gauge_.Set(values[3]/MtoB);
if(numDevice >=5) GPU4_memory_usage_gauge_.Set(values[4]/MtoB);
if(numDevice >=6) GPU5_memory_usage_gauge_.Set(values[5]/MtoB);
if(numDevice >=7) GPU6_memory_usage_gauge_.Set(values[6]/MtoB);
if(numDevice >=8) GPU7_memory_usage_gauge_.Set(values[7]/MtoB);
// to do
}
void PrometheusMetrics::AddVectorsPerSecondGaugeSet(int num_vector, int dim, double time) {
// MB/s
if(!startup_) return;
long long MtoB = 1024*1024;
long long size = num_vector * dim * 4;
add_vectors_per_second_gauge_.Set(size/time/MtoB);
}
void PrometheusMetrics::QueryIndexTypePerSecondSet(std::string type, double value) {
if(!startup_) return;
if(type == "IVF"){
query_index_IVF_type_per_second_gauge_.Set(value);
} else if(type == "IDMap"){
query_index_IDMAP_type_per_second_gauge_.Set(value);
}
}
void PrometheusMetrics::ConnectionGaugeIncrement() {
if(!startup_) return;
connection_gauge_.Increment();
}
void PrometheusMetrics::ConnectionGaugeDecrement() {
if(!startup_) return;
connection_gauge_.Decrement();
}
//void PrometheusMetrics::GpuPercentInit() {
// int num_device = SystemInfo::GetInstance().num_device();
// constexpr char device_number[] = "DeviceNum";
// for(int i = 0; i < num_device; ++ i) {
// GPU_percent_gauges_.emplace_back(GPU_percent_.Add({{device_number,std::to_string(i)}}));
// }
//
//}
//void PrometheusMetrics::GpuMemoryInit() {
// int num_device = SystemInfo::GetInstance().num_device();
// constexpr char device_number[] = "DeviceNum";
// for(int i = 0; i < num_device; ++ i) {
// GPU_memory_usage_gauges_.emplace_back(GPU_memory_usage_.Add({{device_number,std::to_string(i)}}));
// }
//}
}
}
}

132
cpp/src/metrics/PrometheusMetrics.h
View File

@ -49,6 +49,8 @@ class PrometheusMetrics: public MetricsBase {
std::shared_ptr<prometheus::Exposer> exposer_ptr_;
std::shared_ptr<prometheus::Registry> registry_ = std::make_shared<prometheus::Registry>();
bool startup_ = false;
// void GpuPercentInit();
// void GpuMemoryInit();
public:
void AddGroupSuccessTotalIncrement(double value = 1.0) override { if(startup_) add_group_success_total_.Increment(value);};
@ -104,51 +106,20 @@ class PrometheusMetrics: public MetricsBase {
void AddVectorsFailGaugeSet(double value) override { if(startup_) add_vectors_fail_gauge_.Set(value);};
void QueryVectorResponseSummaryObserve(double value, int count = 1) override { if (startup_) for(int i = 0 ; i < count ; ++i) query_vector_response_summary_.Observe(value);};
void QueryVectorResponsePerSecondGaugeSet(double value) override {if (startup_) query_vector_response_per_second_gauge_.Set(value);};
void CPUUsagePercentSet() override ;
void RAMUsagePercentSet() override ;
void QueryResponsePerSecondGaugeSet(double value) override {if(startup_) query_response_per_second_gauge.Set(value);};
void GPUPercentGaugeSet() override ;
void GPUMemoryUsageGaugeSet() override ;
void AddVectorsPerSecondGaugeSet(int num_vector, int dim, double time) override ;
void QueryIndexTypePerSecondSet(std::string type, double value) override ;
void ConnectionGaugeIncrement() override ;
void ConnectionGaugeDecrement() override ;
void KeepingAliveCounterIncrement(double value = 1) override {if(startup_) keeping_alive_counter_.Increment(value);};
// prometheus::Counter &connection_total() {return connection_total_; }
//
// prometheus::Counter &add_group_success_total() { return add_group_success_total_; }
// prometheus::Counter &add_group_fail_total() { return add_group_fail_total_; }
//
// prometheus::Counter &get_group_success_total() { return get_group_success_total_;}
// prometheus::Counter &get_group_fail_total() { return get_group_fail_total_;}
//
// prometheus::Counter &has_group_success_total() { return has_group_success_total_;}
// prometheus::Counter &has_group_fail_total() { return has_group_fail_total_;}
//
// prometheus::Counter &get_group_files_success_total() { return get_group_files_success_total_;};
// prometheus::Counter &get_group_files_fail_total() { return get_group_files_fail_total_;}
//
// prometheus::Counter &add_vectors_success_total() { return add_vectors_success_total_; }
// prometheus::Counter &add_vectors_fail_total() { return add_vectors_fail_total_; }
//
// prometheus::Histogram &add_vectors_duration_histogram() { return add_vectors_duration_histogram_;}
//
// prometheus::Counter &search_success_total() { return search_success_total_; }
// prometheus::Counter &search_fail_total() { return search_fail_total_; }
//
// prometheus::Histogram &search_duration_histogram() { return search_duration_histogram_; }
// prometheus::Histogram &raw_files_size_histogram() { return raw_files_size_histogram_; }
// prometheus::Histogram &index_files_size_histogram() { return index_files_size_histogram_; }
//
// prometheus::Histogram &build_index_duration_seconds_histogram() { return build_index_duration_seconds_histogram_; }
//
// prometheus::Histogram &all_build_index_duration_seconds_histogram() { return all_build_index_duration_seconds_histogram_; }
//
// prometheus::Gauge &cache_usage_gauge() { return cache_usage_gauge_; }
//
// prometheus::Counter &meta_visit_total() { return meta_visit_total_; }
//
// prometheus::Histogram &meta_visit_duration_seconds_histogram() { return meta_visit_duration_seconds_histogram_; }
//
// prometheus::Gauge &mem_usage_percent_gauge() { return mem_usage_percent_gauge_; }
//
// prometheus::Gauge &mem_usage_total_gauge() { return mem_usage_total_gauge_; }
std::shared_ptr<prometheus::Exposer> &exposer_ptr() {return exposer_ptr_; }
@ -273,7 +244,7 @@ class PrometheusMetrics: public MetricsBase {
.Name("build_index_duration_microseconds")
.Help("histogram of processing time for building index")
.Register(*registry_);
prometheus::Histogram &build_index_duration_seconds_histogram_ = build_index_duration_seconds_.Add({}, BucketBoundaries{2e6, 4e6, 6e6, 8e6, 1e7});
prometheus::Histogram &build_index_duration_seconds_histogram_ = build_index_duration_seconds_.Add({}, BucketBoundaries{5e5, 2e6, 4e6, 6e6, 8e6, 1e7});
//record processing time for all building index
@ -414,6 +385,12 @@ class PrometheusMetrics: public MetricsBase {
.Register(*registry_);
prometheus::Gauge &query_vector_response_per_second_gauge_ = query_vector_response_per_second_.Add({});
prometheus::Family<prometheus::Gauge> &query_response_per_second_ = prometheus::BuildGauge()
.Name("query_response_per_microsecond")
.Help("the number of queries can be processed every microsecond")
.Register(*registry_);
prometheus::Gauge &query_response_per_second_gauge = query_response_per_second_.Add({});
prometheus::Family<prometheus::Gauge> &disk_store_IO_speed_ = prometheus::BuildGauge()
.Name("disk_store_IO_speed_bytes_per_microseconds")
.Help("disk_store_IO_speed")
@ -433,6 +410,77 @@ class PrometheusMetrics: public MetricsBase {
prometheus::Gauge &add_vectors_success_gauge_ = add_vectors_.Add({{"outcome", "success"}});
prometheus::Gauge &add_vectors_fail_gauge_ = add_vectors_.Add({{"outcome", "fail"}});
prometheus::Family<prometheus::Gauge> &add_vectors_per_second_ = prometheus::BuildGauge()
.Name("add_vectors_throughput_per_microsecond")
.Help("add vectors throughput per microsecond")
.Register(*registry_);
prometheus::Gauge &add_vectors_per_second_gauge_ = add_vectors_per_second_.Add({});
prometheus::Family<prometheus::Gauge> &CPU_ = prometheus::BuildGauge()
.Name("CPU_usage_percent")
.Help("CPU usage percent by this this process")
.Register(*registry_);
prometheus::Gauge &CPU_usage_percent_ = CPU_.Add({});
prometheus::Family<prometheus::Gauge> &RAM_ = prometheus::BuildGauge()
.Name("RAM_usage_percent")
.Help("RAM usage percent by this process")
.Register(*registry_);
prometheus::Gauge &RAM_usage_percent_ = RAM_.Add({});
//GPU Usage Percent
prometheus::Family<prometheus::Gauge> &GPU_percent_ = prometheus::BuildGauge()
.Name("Gpu_usage_percent")
.Help("GPU_usage_percent ")
.Register(*registry_);
prometheus::Gauge &GPU0_percent_gauge_ = GPU_percent_.Add({{"DeviceNum", "0"}});
prometheus::Gauge &GPU1_percent_gauge_ = GPU_percent_.Add({{"DeviceNum", "1"}});
prometheus::Gauge &GPU2_percent_gauge_ = GPU_percent_.Add({{"DeviceNum", "2"}});
prometheus::Gauge &GPU3_percent_gauge_ = GPU_percent_.Add({{"DeviceNum", "3"}});
prometheus::Gauge &GPU4_percent_gauge_ = GPU_percent_.Add({{"DeviceNum", "4"}});
prometheus::Gauge &GPU5_percent_gauge_ = GPU_percent_.Add({{"DeviceNum", "5"}});
prometheus::Gauge &GPU6_percent_gauge_ = GPU_percent_.Add({{"DeviceNum", "6"}});
prometheus::Gauge &GPU7_percent_gauge_ = GPU_percent_.Add({{"DeviceNum", "7"}});
// std::vector<prometheus::Gauge> GPU_percent_gauges_;
//GPU Mempry used
prometheus::Family<prometheus::Gauge> &GPU_memory_usage_ = prometheus::BuildGauge()
.Name("GPU_memory_usage_total")
.Help("GPU memory usage total ")
.Register(*registry_);
prometheus::Gauge &GPU0_memory_usage_gauge_ = GPU_memory_usage_.Add({{"DeviceNum", "0"}});
prometheus::Gauge &GPU1_memory_usage_gauge_ = GPU_memory_usage_.Add({{"DeviceNum", "1"}});
prometheus::Gauge &GPU2_memory_usage_gauge_ = GPU_memory_usage_.Add({{"DeviceNum", "2"}});
prometheus::Gauge &GPU3_memory_usage_gauge_ = GPU_memory_usage_.Add({{"DeviceNum", "3"}});
prometheus::Gauge &GPU4_memory_usage_gauge_ = GPU_memory_usage_.Add({{"DeviceNum", "4"}});
prometheus::Gauge &GPU5_memory_usage_gauge_ = GPU_memory_usage_.Add({{"DeviceNum", "5"}});
prometheus::Gauge &GPU6_memory_usage_gauge_ = GPU_memory_usage_.Add({{"DeviceNum", "6"}});
prometheus::Gauge &GPU7_memory_usage_gauge_ = GPU_memory_usage_.Add({{"DeviceNum", "7"}});
// std::vector<prometheus::Gauge> GPU_memory_usage_gauges_;
prometheus::Family<prometheus::Gauge> &query_index_type_per_second_ = prometheus::BuildGauge()
.Name("query_index_throughtout_per_microsecond")
.Help("query index throughtout per microsecond")
.Register(*registry_);
prometheus::Gauge &query_index_IVF_type_per_second_gauge_ = query_index_type_per_second_.Add({{"IndexType","IVF"}});
prometheus::Gauge &query_index_IDMAP_type_per_second_gauge_ = query_index_type_per_second_.Add({{"IndexType","IDMAP"}});
prometheus::Family<prometheus::Gauge> &connection_ = prometheus::BuildGauge()
.Name("connection_number")
.Help("the number of connections")
.Register(*registry_);
prometheus::Gauge &connection_gauge_ = connection_.Add({});
prometheus::Family<prometheus::Counter> &keeping_alive_ = prometheus::BuildCounter()
.Name("keeping_alive_seconds_total")
.Help("total seconds of the serve alive")
.Register(*registry_);
prometheus::Counter &keeping_alive_counter_ = keeping_alive_.Add({});
};

207
cpp/src/metrics/SystemInfo.cpp Normal file
View File

@ -0,0 +1,207 @@
/*******************************************************************************
* Copyright 上海赜睿信息科技有限公司(Zilliz) - All Rights Reserved
* Unauthorized copying of this file, via any medium is strictly prohibited.
* Proprietary and confidential.
******************************************************************************/
#include "SystemInfo.h"
#include <sys/types.h>
#include <unistd.h>
#include <iostream>
#include <fstream>
#include "nvml.h"
//#include <mutex>
//
//std::mutex mutex;
namespace zilliz {
namespace vecwise {
namespace server {
void SystemInfo::Init() {
if(initialized_) return;
initialized_ = true;
// initialize CPU information
FILE* file;
struct tms time_sample;
char line[128];
last_cpu_ = times(&time_sample);
last_sys_cpu_ = time_sample.tms_stime;
last_user_cpu_ = time_sample.tms_utime;
file = fopen("/proc/cpuinfo", "r");
num_processors_ = 0;
while(fgets(line, 128, file) != NULL){
if (strncmp(line, "processor", 9) == 0) num_processors_++;
}
total_ram_ = GetPhysicalMemory();
fclose(file);
//initialize GPU information
nvmlReturn_t nvmlresult;
nvmlresult = nvmlInit();
if(NVML_SUCCESS != nvmlresult) {
printf("System information initilization failed");
return ;
}
nvmlresult = nvmlDeviceGetCount(&num_device_);
if(NVML_SUCCESS != nvmlresult) {
printf("Unable to get devidce number");
return ;
}
}
long long
SystemInfo::ParseLine(char *line) {
// This assumes that a digit will be found and the line ends in " Kb".
int i = strlen(line);
const char *p = line;
while (*p < '0' || *p > '9') p++;
line[i - 3] = '\0';
i = atoi(p);
return static_cast<long long>(i);
}
unsigned long
SystemInfo::GetPhysicalMemory() {
struct sysinfo memInfo;
sysinfo (&memInfo);
unsigned long totalPhysMem = memInfo.totalram;
//Multiply in next statement to avoid int overflow on right hand side...
totalPhysMem *= memInfo.mem_unit;
return totalPhysMem;
}
unsigned long
SystemInfo::GetProcessUsedMemory() {
//Note: this value is in KB!
FILE* file = fopen("/proc/self/status", "r");
constexpr int64_t line_length = 128;
long long result = -1;
constexpr int64_t KB_SIZE = 1024;
char line[line_length];
while (fgets(line, line_length, file) != NULL){
if (strncmp(line, "VmRSS:", 6) == 0){
result = ParseLine(line);
break;
}
}
fclose(file);
// return value in Byte
return (result*KB_SIZE);
}
double
SystemInfo::MemoryPercent() {
if (!initialized_) Init();
return GetProcessUsedMemory()*100/total_ram_;
}
double
SystemInfo::CPUPercent() {
if (!initialized_) Init();
struct tms time_sample;
clock_t now;
double percent;
now = times(&time_sample);
if (now <= last_cpu_ || time_sample.tms_stime < last_sys_cpu_ ||
time_sample.tms_utime < last_user_cpu_){
//Overflow detection. Just skip this value.
percent = -1.0;
}
else{
percent = (time_sample.tms_stime - last_sys_cpu_) +
(time_sample.tms_utime - last_user_cpu_);
percent /= (now - last_cpu_);
percent /= num_processors_;
percent *= 100;
}
last_cpu_ = now;
last_sys_cpu_ = time_sample.tms_stime;
last_user_cpu_ = time_sample.tms_utime;
return percent;
}
//std::unordered_map<int,std::vector<double>>
//SystemInfo::GetGPUMemPercent(){
// // return GPUID: MEM%
//
// //write GPU info to a file
// system("nvidia-smi pmon -c 1 > GPUInfo.txt");
// int pid = (int)getpid();
//
// //parse line
// std::ifstream read_file;
// read_file.open("GPUInfo.txt");
// std::string line;
// while(getline(read_file, line)){
// std::vector<std::string> words = split(line);
// // 0 1 2 3 4 5 6 7
// //words stand for gpuindex, pid, type, sm, mem, enc, dec, command respectively
// if(std::stoi(words[1]) != pid) continue;
// int GPUindex = std::stoi(words[0]);
// double sm_percent = std::stod(words[3]);
// double mem_percent = std::stod(words[4]);
//
// }
//
//}
//std::vector<std::string>
//SystemInfo::split(std::string input) {
// std::vector<std::string> words;
// input += " ";
// int word_start = 0;
// for (int i = 0; i < input.size(); ++i) {
// if(input[i] != ' ') continue;
// if(input[i] == ' ') {
// word_start = i + 1;
// continue;
// }
// words.push_back(input.substr(word_start,i-word_start));
// }
// return words;
//}
std::vector<unsigned int>
SystemInfo::GPUPercent() {
// get GPU usage percent
if(!initialized_) Init();
std::vector<unsigned int> result;
nvmlUtilization_t utilization;
for (int i = 0; i < num_device_; ++i) {
nvmlDevice_t device;
nvmlDeviceGetHandleByIndex(i, &device);
nvmlDeviceGetUtilizationRates(device, &utilization);
result.push_back(utilization.gpu);
}
return result;
}
std::vector<unsigned long long>
SystemInfo::GPUMemoryUsed() {
// get GPU memory used
if(!initialized_) Init();
std::vector<unsigned long long int> result;
nvmlMemory_t nvmlMemory;
for (int i = 0; i < num_device_; ++i) {
nvmlDevice_t device;
nvmlDeviceGetHandleByIndex(i, &device);
nvmlDeviceGetMemoryInfo(device, &nvmlMemory);
result.push_back(nvmlMemory.used);
}
return result;
}
}
}
}

60
cpp/src/metrics/SystemInfo.h Normal file
View File

@ -0,0 +1,60 @@
/*******************************************************************************
* Copyright 上海赜睿信息科技有限公司(Zilliz) - All Rights Reserved
* Unauthorized copying of this file, via any medium is strictly prohibited.
* Proprietary and confidential.
******************************************************************************/
#pragma once
#include "sys/types.h"
#include "sys/sysinfo.h"
#include "stdlib.h"
#include "stdio.h"
#include "string.h"
#include "sys/times.h"
#include "sys/vtimes.h"
#include <unordered_map>
#include <vector>
namespace zilliz {
namespace vecwise {
namespace server {
class SystemInfo {
private:
unsigned long total_ram_ = 0;
clock_t last_cpu_ = clock_t();
clock_t last_sys_cpu_ = clock_t();
clock_t last_user_cpu_ = clock_t();
int num_processors_ = 0;
//number of GPU
unsigned int num_device_ = 0;
bool initialized_ = false;
public:
static SystemInfo &
GetInstance(){
static SystemInfo instance;
return instance;
}
void Init();
int num_device() const {return num_device_;};
long long ParseLine(char* line);
unsigned long GetPhysicalMemory();
unsigned long GetProcessUsedMemory();
double MemoryPercent();
double CPUPercent();
// std::unordered_map<int,std::vector<double>> GetGPUMemPercent() {};
// std::vector<std::string> split(std::string input) {};
std::vector<unsigned int> GPUPercent();
std::vector<unsigned long long> GPUMemoryUsed();
};
}
}
}

1
cpp/src/sdk/CMakeLists.txt
View File

@ -26,7 +26,6 @@ add_library(megasearch_sdk STATIC
${service_files}
)
link_directories(../../third_party/build/lib)
target_link_libraries(megasearch_sdk
${third_party_libs}
)

14
cpp/src/sdk/examples/simple/src/ClientTest.cpp
View File

@ -28,7 +28,7 @@ namespace {
std::cout << "Table name: " << tb_schema.table_name << std::endl;
std::cout << "Table index type: " << (int)tb_schema.index_type << std::endl;
std::cout << "Table dimension: " << tb_schema.dimension << std::endl;
std::cout << "Table store raw data: " << tb_schema.store_raw_vector << std::endl;
std::cout << "Table store raw data: " << (tb_schema.store_raw_vector ? "true" : "false") << std::endl;
BLOCK_SPLITER
}
@ -148,7 +148,9 @@ ClientTest::Test(const std::string& address, const std::string& port) {
std::cout << "ShowTables function call status: " << stat.ToString() << std::endl;
std::cout << "All tables: " << std::endl;
for(auto& table : tables) {
std::cout << "\t" << table << std::endl;
int64_t row_count = 0;
stat = conn->GetTableRowCount(table, row_count);
std::cout << "\t" << table << "(" << row_count << " rows)" << std::endl;
}
}
@ -192,10 +194,10 @@ ClientTest::Test(const std::string& address, const std::string& port) {
PrintSearchResult(topk_query_result_array);
}
// {//delete table
// Status stat = conn->DeleteTable(TABLE_NAME);
// std::cout << "DeleteTable function call status: " << stat.ToString() << std::endl;
// }
{//delete table
Status stat = conn->DeleteTable(TABLE_NAME);
std::cout << "DeleteTable function call status: " << stat.ToString() << std::endl;
}
{//server status
std::string status = conn->ServerStatus();

4
cpp/src/sdk/include/Status.h
View File

@ -72,7 +72,7 @@ class Status {
* @return, the status is assigned.
*
*/
inline Status &operator=(const Status &s);
Status &operator=(const Status &s);
/**
* @brief Status
@ -93,7 +93,7 @@ class Status {
* @return, the status is moved.
*
*/
inline Status &operator=(Status &&s) noexcept;
Status &operator=(Status &&s) noexcept;
/**
* @brief Status

4
cpp/src/sdk/src/client/ClientProxy.cpp
View File

@ -77,7 +77,7 @@ ClientProxy::Disconnect() {
std::string
ClientProxy::ClientVersion() const {
return std::string("v1.0");
return "";
}
Status
@ -221,6 +221,8 @@ ClientProxy::DescribeTable(const std::string &table_name, TableSchema &table_sch
table_schema.table_name = thrift_schema.table_name;
table_schema.index_type = (IndexType)thrift_schema.index_type;
table_schema.dimension = thrift_schema.dimension;
table_schema.store_raw_vector = thrift_schema.store_raw_vector;
} catch ( std::exception& ex) {
return Status(StatusCode::UnknownError, "failed to describe table: " + std::string(ex.what()));

3
cpp/src/sdk/src/interface/ConnectionImpl.cpp
View File

@ -4,6 +4,7 @@
* Proprietary and confidential.
******************************************************************************/
#include "ConnectionImpl.h"
#include "version.h"
namespace megasearch {
@ -47,7 +48,7 @@ ConnectionImpl::Disconnect() {
std::string
ConnectionImpl::ClientVersion() const {
return client_proxy_->ClientVersion();
return MEGASEARCH_VERSION;
}
Status

3
cpp/src/server/MegasearchServer.cpp
View File

@ -8,6 +8,7 @@
#include "megasearch_types.h"
#include "megasearch_constants.h"
#include "ServerConfig.h"
#include "MegasearchThreadPoolServer.h"
#include <thrift/protocol/TBinaryProtocol.h>
#include <thrift/protocol/TJSONProtocol.h>
@ -76,7 +77,7 @@ MegasearchServer::StartService() {
threadManager->threadFactory(threadFactory);
threadManager->start();
s_server.reset(new TThreadPoolServer(processor,
s_server.reset(new MegasearchThreadPoolServer(processor,
server_transport,
transport_factory,
protocol_factory,

123
cpp/src/server/MegasearchTask.cpp
View File

@ -21,9 +21,6 @@ static const std::string DQL_TASK_GROUP = "dql";
static const std::string DDL_DML_TASK_GROUP = "ddl_dml";
static const std::string PING_TASK_GROUP = "ping";
static const std::string VECTOR_UID = "uid";
static const uint64_t USE_MT = 5000;
using DB_META = zilliz::vecwise::engine::meta::Meta;
using DB_DATE = zilliz::vecwise::engine::meta::DateT;
@ -76,6 +73,20 @@ namespace {
return map_type[type];
}
int IndexType(engine::EngineType type) {
static std::map<engine::EngineType, int> map_type = {
{engine::EngineType::INVALID, 0},
{engine::EngineType::FAISS_IDMAP, 1},
{engine::EngineType::FAISS_IVFFLAT, 2},
};
if(map_type.find(type) == map_type.end()) {
return 0;
}
return map_type[type];
}
ServerError
ConvertRowRecordToFloatArray(const std::vector<thrift::RowRecord>& record_array,
uint64_t dimension,
@ -174,16 +185,17 @@ ServerError CreateTableTask::OnExecute() {
//step 2: create table
engine::Status stat = DB()->CreateTable(table_info);
if(!stat.ok()) {//table could exist
error_code_ = SERVER_UNEXPECTED_ERROR;
error_msg_ = "Engine failed: " + stat.ToString();
SERVER_LOG_ERROR << error_msg_;
return SERVER_SUCCESS;
return error_code_;
}
} catch (std::exception& ex) {
error_code_ = SERVER_UNEXPECTED_ERROR;
error_msg_ = ex.what();
SERVER_LOG_ERROR << error_msg_;
return SERVER_UNEXPECTED_ERROR;
return error_code_;
}
rc.Record("done");
@ -215,10 +227,13 @@ ServerError DescribeTableTask::OnExecute() {
error_msg_ = "Engine failed: " + stat.ToString();
SERVER_LOG_ERROR << error_msg_;
return error_code_;
} else {
}
schema_.table_name = table_info.table_id_;
schema_.index_type = IndexType((engine::EngineType)table_info.engine_type_);
schema_.dimension = table_info.dimension_;
schema_.store_raw_vector = table_info.store_raw_data_;
} catch (std::exception& ex) {
error_code_ = SERVER_UNEXPECTED_ERROR;
error_msg_ = ex.what();
@ -243,16 +258,53 @@ BaseTaskPtr DeleteTableTask::Create(const std::string& group_id) {
}
ServerError DeleteTableTask::OnExecute() {
error_code_ = SERVER_NOT_IMPLEMENT;
error_msg_ = "delete table not implemented";
SERVER_LOG_ERROR << error_msg_;
try {
TimeRecorder rc("DeleteTableTask");
return SERVER_NOT_IMPLEMENT;
//step 1: check validation
if (table_name_.empty()) {
error_code_ = SERVER_INVALID_ARGUMENT;
error_msg_ = "Table name cannot be empty";
SERVER_LOG_ERROR << error_msg_;
return error_code_;
}
//step 2: check table existence
engine::meta::TableSchema table_info;
table_info.table_id_ = table_name_;
engine::Status stat = DB()->DescribeTable(table_info);
if(!stat.ok()) {
error_code_ = SERVER_TABLE_NOT_EXIST;
error_msg_ = "Engine failed: " + stat.ToString();
SERVER_LOG_ERROR << error_msg_;
return error_code_;
}
rc.Record("check validation");
//step 3: delete table
std::vector<DB_DATE> dates;
stat = DB()->DeleteTable(table_name_, dates);
if(!stat.ok()) {
SERVER_LOG_ERROR << "Engine failed: " << stat.ToString();
return SERVER_UNEXPECTED_ERROR;
}
rc.Record("deleta table");
rc.Elapse("totally cost");
} catch (std::exception& ex) {
error_code_ = SERVER_UNEXPECTED_ERROR;
error_msg_ = ex.what();
SERVER_LOG_ERROR << error_msg_;
return error_code_;
}
return SERVER_SUCCESS;
}
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
ShowTablesTask::ShowTablesTask(std::vector<std::string>& tables)
: BaseTask(PING_TASK_GROUP),
: BaseTask(DQL_TASK_GROUP),
tables_(tables) {
}
@ -262,6 +314,19 @@ BaseTaskPtr ShowTablesTask::Create(std::vector<std::string>& tables) {
}
ServerError ShowTablesTask::OnExecute() {
std::vector<engine::meta::TableSchema> schema_array;
engine::Status stat = DB()->AllTables(schema_array);
if(!stat.ok()) {
error_code_ = SERVER_UNEXPECTED_ERROR;
error_msg_ = "Engine failed: " + stat.ToString();
SERVER_LOG_ERROR << error_msg_;
return error_code_;
}
tables_.clear();
for(auto& schema : schema_array) {
tables_.push_back(schema.table_id_);
}
return SERVER_SUCCESS;
}
@ -468,17 +533,39 @@ BaseTaskPtr GetTableRowCountTask::Create(const std::string& table_name, int64_t&
}
ServerError GetTableRowCountTask::OnExecute() {
if(table_name_.empty()) {
try {
TimeRecorder rc("GetTableRowCountTask");
//step 1: check validation
if (table_name_.empty()) {
error_code_ = SERVER_INVALID_ARGUMENT;
error_msg_ = "Table name cannot be empty";
SERVER_LOG_ERROR << error_msg_;
return error_code_;
}
//step 2: get row count
uint64_t row_count = 0;
engine::Status stat = DB()->GetTableRowCount(table_name_, row_count);
if (!stat.ok()) {
error_code_ = SERVER_UNEXPECTED_ERROR;
error_msg_ = "Engine failed: " + stat.ToString();
SERVER_LOG_ERROR << error_msg_;
return error_code_;
}
row_count_ = (int64_t) row_count;
rc.Elapse("totally cost");
} catch (std::exception& ex) {
error_code_ = SERVER_UNEXPECTED_ERROR;
error_msg_ = "Table name cannot be empty";
error_msg_ = ex.what();
SERVER_LOG_ERROR << error_msg_;
return error_code_;
}
error_code_ = SERVER_NOT_IMPLEMENT;
error_msg_ = "Not implemented";
SERVER_LOG_ERROR << error_msg_;
return error_code_;
return SERVER_SUCCESS;
}
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////

34
cpp/src/server/MegasearchThreadPoolServer.cpp Normal file
View File

@ -0,0 +1,34 @@
/*******************************************************************************
* Copyright 上海赜睿信息科技有限公司(Zilliz) - All Rights Reserved
* Unauthorized copying of this file, via any medium is strictly prohibited.
* Proprietary and confidential.
******************************************************************************/
#include "metrics/Metrics.h"
#include "MegasearchThreadPoolServer.h"
namespace zilliz {
namespace vecwise {
namespace server {
void
MegasearchThreadPoolServer::onClientConnected(const std::shared_ptr<apache::thrift::server::TConnectedClient> &pClient) {
server::Metrics::GetInstance().ConnectionGaugeIncrement();
TThreadPoolServer::onClientConnected(pClient);
}
void
MegasearchThreadPoolServer::onClientDisconnected(apache::thrift::server::TConnectedClient *pClient) {
server::Metrics::GetInstance().ConnectionGaugeDecrement();
TThreadPoolServer::onClientDisconnected(pClient);
}
zilliz::vecwise::server::MegasearchThreadPoolServer::MegasearchThreadPoolServer(const std::shared_ptr<apache::thrift::TProcessor> &processor,
const std::shared_ptr<apache::thrift::transport::TServerTransport> &serverTransport,
const std::shared_ptr<apache::thrift::transport::TTransportFactory> &transportFactory,
const std::shared_ptr<apache::thrift::protocol::TProtocolFactory> &protocolFactory,
const std::shared_ptr<apache::thrift::concurrency::ThreadManager> &threadManager)
: TThreadPoolServer(processor, serverTransport, transportFactory, protocolFactory, threadManager) {
}
}
}
}

33
cpp/src/server/MegasearchThreadPoolServer.h Normal file
View File

@ -0,0 +1,33 @@
/*******************************************************************************
* Copyright 上海赜睿信息科技有限公司(Zilliz) - All Rights Reserved
* Unauthorized copying of this file, via any medium is strictly prohibited.
* Proprietary and confidential.
******************************************************************************/
#pragma once
#include <thrift/server/TThreadPoolServer.h>
namespace zilliz {
namespace vecwise {
namespace server {
class MegasearchThreadPoolServer : public apache::thrift::server::TThreadPoolServer {
public:
MegasearchThreadPoolServer(
const std::shared_ptr<apache::thrift::TProcessor>& processor,
const std::shared_ptr<apache::thrift::transport::TServerTransport>& serverTransport,
const std::shared_ptr<apache::thrift::transport::TTransportFactory>& transportFactory,
const std::shared_ptr<apache::thrift::protocol::TProtocolFactory>& protocolFactory,
const std::shared_ptr<apache::thrift::concurrency::ThreadManager>& threadManager
= apache::thrift::concurrency::ThreadManager::newSimpleThreadManager());
protected:
void onClientConnected(const std::shared_ptr<apache::thrift::server::TConnectedClient>& pClient) override ;
void onClientDisconnected(apache::thrift::server::TConnectedClient* pClient) override ;
};
}
}
}

1
cpp/src/server/Server.cpp
View File

@ -173,6 +173,7 @@ Server::Start() {
signal(SIGHUP, SignalUtil::HandleSignal);
signal(SIGTERM, SignalUtil::HandleSignal);
server::Metrics::GetInstance().Init();
server::SystemInfo::GetInstance().Init();
SERVER_LOG_INFO << "Vecwise server is running...";
StartService();

4
cpp/thirdparty/versions.txt vendored
View File

@ -31,7 +31,7 @@ GTEST_VERSION=1.8.1
LAPACK_VERSION=v3.8.0
LZ4_VERSION=v1.9.1
OPENBLAS_VERSION=v0.3.6
PROMETHEUS_VERSION=master
PROMETHEUS_VERSION=v0.7.0
ROCKSDB_VERSION=v6.0.2
SNAPPY_VERSION=1.1.7
SQLITE_VERSION=3280000
@ -54,7 +54,7 @@ DEPENDENCIES=(
"MEGASEARCH_LAPACK_URL lapack-${LAPACK_VERSION}.tar.gz https://github.com/Reference-LAPACK/lapack/archive/${LAPACK_VERSION}.tar.gz
"MEGASEARCH_LZ4_URL lz4-${LZ4_VERSION}.tar.gz https://github.com/lz4/lz4/archive/${LZ4_VERSION}.tar.gz"
"MEGASEARCH_OPENBLAS_URL openblas-${OPENBLAS_VERSION}.tar.gz https://github.com/xianyi/OpenBLAS/archive/${OPENBLAS_VERSION}.tar.gz"
"MEGASEARCH_PROMETHEUS_URL prometheus-${PROMETHEUS_VERSION}.tar.gz https://github.com/JinHai-CN/prometheus-cpp/archive/${PROMETHEUS_VERSION}.tar.gz"
"MEGASEARCH_PROMETHEUS_URL https://github.com/jupp0r/prometheus-cpp.git"
"MEGASEARCH_ROCKSDB_URL rocksdb-${ROCKSDB_VERSION}.tar.gz https://github.com/facebook/rocksdb/archive/${ROCKSDB_VERSION}.tar.gz"
"MEGASEARCH_SNAPPY_URL snappy-${SNAPPY_VERSION}.tar.gz https://github.com/google/snappy/archive/${SNAPPY_VERSION}.tar.gz"
"MEGASEARCH_SQLITE_URL sqlite-autoconf-${SQLITE_VERSION}.tar.gz https://www.sqlite.org/2019/sqlite-autoconf-${SQLITE_VERSION}.tar.gz"

2
cpp/unittest/CMakeLists.txt
View File

@ -29,9 +29,9 @@ set(unittest_libs
prometheus-cpp-pull
prometheus-cpp-push
prometheus-cpp-core
civetweb
dl
z
${CUDA_TOOLKIT_ROOT_DIR}/lib64/stubs/libnvidia-ml.so
)
add_subdirectory(server)

6
cpp/unittest/db/db_tests.cpp
View File

@ -64,7 +64,7 @@ TEST_F(DBTest2, ARHIVE_DISK_CHECK) {
static const std::string group_name = "test_group";
static const int group_dim = 256;
long size;
uint64_t size;
engine::meta::TableSchema group_info;
group_info.dimension_ = group_dim;
@ -149,8 +149,8 @@ TEST_F(DBTest, DB_TEST) {
INIT_TIMER;
std::stringstream ss;
long count = 0;
long prev_count = -1;
uint64_t count = 0;
uint64_t prev_count = 0;
for (auto j=0; j<10; ++j) {
ss.str("");

4
cpp/unittest/metrics/metrics_test.cpp
View File

@ -72,8 +72,8 @@ TEST_F(DBTest, Metric_Tes) {
INIT_TIMER;
std::stringstream ss;
long count = 0;
long prev_count = -1;
uint64_t count = 0;
uint64_t prev_count = 0;
for (auto j=0; j<10; ++j) {
ss.str("");