提交 208f4d8b 编写于 作者: J jinhai

Merge branch 'branch-0.4.0' into 'branch-0.4.0'

grpc and thrift server run concurrently

See merge request megasearch/milvus!327

Former-commit-id: fca6ff245b7873689288d74e28e97080aa739de9
......@@ -60,6 +60,7 @@ Please mark all change in change log and use the ticket from JIRA.
- MS-310 - Add milvus CPU utilization ratio and CPU/GPU temperature metrics
- MS-324 - Show error when there is not enough gpu memory to build index
- MS-328 - Check metric type on server start
- MS-332 - Set grpc and thrift server run concurrently
## New Feature
- MS-180 - Add new mem manager
......
......@@ -90,7 +90,7 @@ define_option(MILVUS_WITH_SQLITE_ORM "Build with SQLite ORM library" ON)
define_option(MILVUS_WITH_MYSQLPP "Build with MySQL++" ON)
define_option(MILVUS_WITH_THRIFT "Build with Apache Thrift library" OFF)
define_option(MILVUS_WITH_THRIFT "Build with Apache Thrift library" ON)
define_option(MILVUS_WITH_YAMLCPP "Build with yaml-cpp library" ON)
......
......@@ -2679,7 +2679,7 @@ macro(build_grpc)
add_dependencies(grpc_protoc grpc_ep)
endmacro()
if(NOT MILVUS_WITH_THRIFT STREQUAL "ON")
#if(NOT MILVUS_WITH_THRIFT STREQUAL "ON")
resolve_dependency(GRPC)
get_target_property(GRPC_INCLUDE_DIR grpc INTERFACE_INCLUDE_DIRECTORIES)
......@@ -2690,4 +2690,4 @@ if(NOT MILVUS_WITH_THRIFT STREQUAL "ON")
include_directories(SYSTEM ${GRPC_THIRD_PARTY_DIR}/protobuf/src)
link_directories(SYSTEM ${GRPC_PROTOBUF_LIB_DIR})
endif()
#endif()
......@@ -4,7 +4,6 @@
# Proprietary and confidential.
#-------------------------------------------------------------------------------
aux_source_directory(cache cache_files)
aux_source_directory(config config_files)
aux_source_directory(server server_files)
......@@ -78,17 +77,17 @@ include_directories(/usr/include/mysql)
include_directories(grpc/gen-status)
include_directories(grpc/gen-milvus)
if (MILVUS_WITH_THRIFT STREQUAL "ON")
set(client_lib
#if (MILVUS_WITH_THRIFT STREQUAL "ON")
set(client_thrift_lib
thrift)
else()
set(client_lib
#else()
set(client_grpc_lib
grpcpp_channelz
grpc++
grpc
grpc_protobuf
grpc_protoc)
endif()
#endif()
set(third_party_libs
knowhere
......@@ -100,7 +99,8 @@ set(third_party_libs
lapack
easyloggingpp
sqlite
${client_lib}
${client_thrift_lib}
${client_grpc_lib}
yaml-cpp
prometheus-cpp-push
prometheus-cpp-pull
......@@ -197,7 +197,7 @@ set(knowhere_libs
tbb
)
if (MILVUS_WITH_THRIFT STREQUAL "ON")
#if (MILVUS_WITH_THRIFT STREQUAL "ON")
add_executable(milvus_thrift_server
${config_files}
${server_files}
......@@ -206,7 +206,7 @@ if (MILVUS_WITH_THRIFT STREQUAL "ON")
${thrift_service_files}
${metrics_files}
)
else()
#else()
add_executable(milvus_grpc_server
${config_files}
${server_files}
......@@ -215,7 +215,16 @@ else()
${grpc_service_files}
${metrics_files}
)
endif()
#endif()
add_executable(milvus_server
${config_files}
${server_files}
${thriftserver_files}
${grpcserver_files}
${utils_files}
${thrift_service_files}
${grpc_service_files}
${metrics_files})
if (ENABLE_LICENSE STREQUAL "ON")
add_executable(get_sys_info ${get_sys_info_files})
......@@ -224,25 +233,28 @@ if (ENABLE_LICENSE STREQUAL "ON")
target_link_libraries(get_sys_info ${license_libs} license_check ${third_party_libs})
target_link_libraries(license_generator ${license_libs} ${third_party_libs})
if(MILVUS_WITH_THRIFT STREQUAL "ON")
# if(MILVUS_WITH_THRIFT STREQUAL "ON")
target_link_libraries(milvus_thrift_server ${server_libs} license_check ${knowhere_libs} ${third_party_libs})
else()
# else()
target_link_libraries(milvus_grpc_server ${server_libs} license_check ${knowhere_libs} ${third_party_libs})
endif()
# endif()
target_link_libraries(milvus_server ${server_libs} license_check ${knowhere_libs} ${third_party_libs})
else ()
if(MILVUS_WITH_THRIFT STREQUAL "ON")
# if(MILVUS_WITH_THRIFT STREQUAL "ON")
target_link_libraries(milvus_thrift_server ${server_libs} ${knowhere_libs} ${third_party_libs})
else()
# else()
target_link_libraries(milvus_grpc_server ${server_libs} ${knowhere_libs} ${third_party_libs})
endif()
# endif()
target_link_libraries(milvus_server ${server_libs} ${knowhere_libs} ${third_party_libs})
endif()
if (MILVUS_WITH_THRIFT STREQUAL "ON")
#if (MILVUS_WITH_THRIFT STREQUAL "ON")
install(TARGETS milvus_thrift_server DESTINATION bin)
else()
#else()
install(TARGETS milvus_grpc_server DESTINATION bin)
endif()
#endif()
install(TARGETS milvus_server DESTINATION bin)
install(FILES
${KNOWHERE_BUILD_DIR}/lib/${CMAKE_SHARED_LIBRARY_PREFIX}tbb${CMAKE_SHARED_LIBRARY_SUFFIX}
......
......@@ -12,7 +12,7 @@ include_directories(/usr/include)
include_directories(include)
include_directories(/usr/local/include)
if (MILVUS_WITH_THRIFT STREQUAL "ON")
#if (MILVUS_WITH_THRIFT STREQUAL "ON")
aux_source_directory(thrift thrift_client_files)
include_directories(thrift)
include_directories(${CMAKE_SOURCE_DIR}/src/thrift/gen-cpp)
......@@ -34,7 +34,7 @@ if (MILVUS_WITH_THRIFT STREQUAL "ON")
${third_party_libs}
)
install(TARGETS milvus_thrift_sdk DESTINATION lib)
else()
#else()
aux_source_directory(grpc grpc_client_files)
include_directories(${CMAKE_SOURCE_DIR}/src/grpc/gen-milvus)
......@@ -58,6 +58,6 @@ else()
${third_party_libs}
)
install(TARGETS milvus_grpc_sdk DESTINATION lib)
endif()
#endif()
add_subdirectory(examples)
......@@ -4,8 +4,8 @@
# Proprietary and confidential.
#-------------------------------------------------------------------------------
if (MILVUS_WITH_THRIFT STREQUAL "ON")
#if (MILVUS_WITH_THRIFT STREQUAL "ON")
add_subdirectory(thriftsimple)
else()
#else()
add_subdirectory(grpcsimple)
endif()
\ No newline at end of file
#endif()
\ No newline at end of file
......@@ -25,7 +25,7 @@ main(int argc, char *argv[]) {
{NULL, 0, 0, 0}};
int option_index = 0;
std::string address = "127.0.0.1", port = "19530";
std::string address = "127.0.0.1", port = "19531";
app_name = argv[0];
int value;
......
......@@ -3,13 +3,14 @@
// Unauthorized copying of this file, via any medium is strictly prohibited.
// Proprietary and confidential.
////////////////////////////////////////////////////////////////////////////////
#include <thread>
#include "Server.h"
//#include "ServerConfig.h"
#ifdef MILVUS_ENABLE_THRIFT
//#ifdef MILVUS_ENABLE_THRIFT
#include "server/thrift_impl/MilvusServer.h"
#else
#include "server/grpc_impl/MilvusServer.h"
#endif
//#else
#include "server/grpc_impl/GrpcMilvusServer.h"
//#endif
#include "utils/Log.h"
#include "utils/SignalUtil.h"
......@@ -224,12 +225,19 @@ Server::LoadConfig() {
void
Server::StartService() {
MilvusServer::StartService();
std::thread thrift_thread = std::thread(&MilvusServer::StartService);
std::thread grpc_thread = std::thread(&grpc::GrpcMilvusServer::StartService);
thrift_thread.join();
grpc_thread.join();
//
// MilvusServer::StartService();
// grpc::GrpcMilvusServer::StartService();
}
void
Server::StopService() {
MilvusServer::StopService();
MilvusServer::StartService();
grpc::GrpcMilvusServer::StopService();
}
}
......
/*******************************************************************************
* Copyright 上海赜睿信息科技有限公司(Zilliz) - All Rights Reserved
* Unauthorized copying of this file, via any medium is strictly prohibited.
* Proprietary and confidential.
******************************************************************************/
* Copyright 上海赜睿信息科技有限公司(Zilliz) - All Rights Reserved
* Unauthorized copying of this file, via any medium is strictly prohibited.
* Proprietary and confidential.
******************************************************************************/
#include "milvus.grpc.pb.h"
#include "MilvusServer.h"
#include "GrpcMilvusServer.h"
#include "../ServerConfig.h"
#include "../DBWrapper.h"
#include "utils/Log.h"
#include "faiss/utils.h"
#include "RequestHandler.h"
#include "GrpcRequestHandler.h"
#include <chrono>
#include <iostream>
......@@ -28,14 +28,15 @@
namespace zilliz {
namespace milvus {
namespace server {
namespace grpc {
static std::unique_ptr<grpc::Server> server;
static std::unique_ptr<::grpc::Server> server;
constexpr long MESSAGE_SIZE = -1;
void
MilvusServer::StartService() {
if (server != nullptr){
GrpcMilvusServer::StartService() {
if (server != nullptr) {
std::cout << "stopservice!\n";
StopService();
}
......@@ -50,9 +51,9 @@ MilvusServer::StartService() {
DBWrapper::DB();//initialize db
std::string server_address(address + ":" + std::to_string(port));
std::string server_address(address + ":" + std::to_string(port + 1));
grpc::ServerBuilder builder;
::grpc::ServerBuilder builder;
builder.SetMaxReceiveMessageSize(MESSAGE_SIZE); //default 4 * 1024 * 1024
builder.SetMaxSendMessageSize(MESSAGE_SIZE);
......@@ -60,9 +61,9 @@ MilvusServer::StartService() {
builder.SetDefaultCompressionAlgorithm(GRPC_COMPRESS_STREAM_GZIP);
builder.SetDefaultCompressionLevel(GRPC_COMPRESS_LEVEL_HIGH);
RequestHandler service;
GrpcRequestHandler service;
builder.AddListeningPort(server_address, grpc::InsecureServerCredentials());
builder.AddListeningPort(server_address, ::grpc::InsecureServerCredentials());
builder.RegisterService(&service);
server = builder.BuildAndStart();
......@@ -71,12 +72,13 @@ MilvusServer::StartService() {
}
void
MilvusServer::StopService() {
GrpcMilvusServer::StopService() {
if (server != nullptr) {
server->Shutdown();
}
}
}
}
}
}
\ No newline at end of file
/*******************************************************************************
* Copyright 上海赜睿信息科技有限公司(Zilliz) - All Rights Reserved
* Unauthorized copying of this file, via any medium is strictly prohibited.
* Proprietary and confidential.
******************************************************************************/
* Copyright 上海赜睿信息科技有限公司(Zilliz) - All Rights Reserved
* Unauthorized copying of this file, via any medium is strictly prohibited.
* Proprietary and confidential.
******************************************************************************/
#pragma once
#include <cstdint>
......@@ -11,7 +11,9 @@
namespace zilliz {
namespace milvus {
namespace server {
class MilvusServer {
namespace grpc {
class GrpcMilvusServer {
public:
static void
StartService();
......@@ -23,3 +25,4 @@ public:
}
}
}
}
/*******************************************************************************
* Copyright 上海赜睿信息科技有限公司(Zilliz) - All Rights Reserved
* Unauthorized copying of this file, via any medium is strictly prohibited.
* Proprietary and confidential.
******************************************************************************/
* Copyright 上海赜睿信息科技有限公司(Zilliz) - All Rights Reserved
* Unauthorized copying of this file, via any medium is strictly prohibited.
* Proprietary and confidential.
******************************************************************************/
#include "RequestHandler.h"
#include "RequestTask.h"
#include "GrpcRequestHandler.h"
#include "GrpcRequestTask.h"
#include "utils/TimeRecorder.h"
namespace zilliz {
namespace milvus {
namespace server {
namespace grpc {
::grpc::Status
RequestHandler::CreateTable(::grpc::ServerContext *context,
const ::milvus::grpc::TableSchema *request,
::milvus::grpc::Status *response) {
GrpcRequestHandler::CreateTable(::grpc::ServerContext *context,
const ::milvus::grpc::TableSchema *request,
::milvus::grpc::Status *response) {
BaseTaskPtr task_ptr = CreateTableTask::Create(*request);
RequestScheduler::ExecTask(task_ptr, response);
GrpcRequestScheduler::ExecTask(task_ptr, response);
return ::grpc::Status::OK;
}
::grpc::Status
RequestHandler::HasTable(::grpc::ServerContext *context,
const ::milvus::grpc::TableName *request,
::milvus::grpc::BoolReply *response) {
GrpcRequestHandler::HasTable(::grpc::ServerContext *context,
const ::milvus::grpc::TableName *request,
::milvus::grpc::BoolReply *response) {
bool has_table = false;
BaseTaskPtr task_ptr = HasTableTask::Create(request->table_name(), has_table);
::milvus::grpc::Status grpc_status;
RequestScheduler::ExecTask(task_ptr, &grpc_status);
GrpcRequestScheduler::ExecTask(task_ptr, &grpc_status);
response->set_bool_reply(has_table);
response->mutable_status()->set_reason(grpc_status.reason());
response->mutable_status()->set_error_code(grpc_status.error_code());
......@@ -38,47 +39,47 @@ RequestHandler::HasTable(::grpc::ServerContext *context,
}
::grpc::Status
RequestHandler::DropTable(::grpc::ServerContext* context,
const ::milvus::grpc::TableName* request,
::milvus::grpc::Status* response) {
GrpcRequestHandler::DropTable(::grpc::ServerContext *context,
const ::milvus::grpc::TableName *request,
::milvus::grpc::Status *response) {
BaseTaskPtr task_ptr = DropTableTask::Create(request->table_name());
RequestScheduler::ExecTask(task_ptr, response);
GrpcRequestScheduler::ExecTask(task_ptr, response);
return ::grpc::Status::OK;
}
::grpc::Status
RequestHandler::BuildIndex(::grpc::ServerContext* context,
const ::milvus::grpc::TableName* request,
::milvus::grpc::Status* response) {
GrpcRequestHandler::BuildIndex(::grpc::ServerContext *context,
const ::milvus::grpc::TableName *request,
::milvus::grpc::Status *response) {
BaseTaskPtr task_ptr = BuildIndexTask::Create(request->table_name());
RequestScheduler::ExecTask(task_ptr, response);
GrpcRequestScheduler::ExecTask(task_ptr, response);
return ::grpc::Status::OK;
}
::grpc::Status
RequestHandler::InsertVector(::grpc::ServerContext* context,
const ::milvus::grpc::InsertInfos* request,
::milvus::grpc::VectorIds* response) {
GrpcRequestHandler::InsertVector(::grpc::ServerContext *context,
const ::milvus::grpc::InsertInfos *request,
::milvus::grpc::VectorIds *response) {
BaseTaskPtr task_ptr = InsertVectorTask::Create(*request, *response);
::milvus::grpc::Status grpc_status;
RequestScheduler::ExecTask(task_ptr, &grpc_status);
GrpcRequestScheduler::ExecTask(task_ptr, &grpc_status);
response->mutable_status()->set_reason(grpc_status.reason());
response->mutable_status()->set_error_code(grpc_status.error_code());
return ::grpc::Status::OK;
}
::grpc::Status
RequestHandler::SearchVector(::grpc::ServerContext* context,
const ::milvus::grpc::SearchVectorInfos* request,
::grpc::ServerWriter<::milvus::grpc::TopKQueryResult>* writer) {
GrpcRequestHandler::SearchVector(::grpc::ServerContext *context,
const ::milvus::grpc::SearchVectorInfos *request,
::grpc::ServerWriter<::milvus::grpc::TopKQueryResult> *writer) {
std::vector<std::string> file_id_array;
BaseTaskPtr task_ptr = SearchVectorTask::Create(*request, file_id_array, *writer);
::milvus::grpc::Status grpc_status;
RequestScheduler::ExecTask(task_ptr, &grpc_status);
GrpcRequestScheduler::ExecTask(task_ptr, &grpc_status);
if (grpc_status.error_code() != SERVER_SUCCESS) {
::grpc::Status status(::grpc::INVALID_ARGUMENT, grpc_status.reason());
return status;
......@@ -88,14 +89,14 @@ RequestHandler::SearchVector(::grpc::ServerContext* context,
}
::grpc::Status
RequestHandler::SearchVectorInFiles(::grpc::ServerContext* context,
const ::milvus::grpc::SearchVectorInFilesInfos* request,
::grpc::ServerWriter<::milvus::grpc::TopKQueryResult>* writer) {
GrpcRequestHandler::SearchVectorInFiles(::grpc::ServerContext *context,
const ::milvus::grpc::SearchVectorInFilesInfos *request,
::grpc::ServerWriter<::milvus::grpc::TopKQueryResult> *writer) {
std::vector<std::string> file_id_array;
BaseTaskPtr task_ptr = SearchVectorTask::Create(request->search_vector_infos(), file_id_array, *writer);
::milvus::grpc::Status grpc_status;
RequestScheduler::ExecTask(task_ptr, &grpc_status);
GrpcRequestScheduler::ExecTask(task_ptr, &grpc_status);
if (grpc_status.error_code() != SERVER_SUCCESS) {
::grpc::Status status(::grpc::INVALID_ARGUMENT, grpc_status.reason());
return status;
......@@ -105,27 +106,27 @@ RequestHandler::SearchVectorInFiles(::grpc::ServerContext* context,
}
::grpc::Status
RequestHandler::DescribeTable(::grpc::ServerContext* context,
const ::milvus::grpc::TableName* request,
::milvus::grpc::TableSchema* response) {
GrpcRequestHandler::DescribeTable(::grpc::ServerContext *context,
const ::milvus::grpc::TableName *request,
::milvus::grpc::TableSchema *response) {
BaseTaskPtr task_ptr = DescribeTableTask::Create(request->table_name(), *response);
::milvus::grpc::Status grpc_status;
RequestScheduler::ExecTask(task_ptr, &grpc_status);
GrpcRequestScheduler::ExecTask(task_ptr, &grpc_status);
response->mutable_table_name()->mutable_status()->set_error_code(grpc_status.error_code());
response->mutable_table_name()->mutable_status()->set_reason(grpc_status.reason());
return ::grpc::Status::OK;
}
::grpc::Status
RequestHandler::GetTableRowCount(::grpc::ServerContext* context,
const ::milvus::grpc::TableName* request,
::milvus::grpc::TableRowCount* response) {
GrpcRequestHandler::GetTableRowCount(::grpc::ServerContext *context,
const ::milvus::grpc::TableName *request,
::milvus::grpc::TableRowCount *response) {
int64_t row_count = 0;
BaseTaskPtr task_ptr = GetTableRowCountTask::Create(request->table_name(), row_count);
::milvus::grpc::Status grpc_status;
RequestScheduler::ExecTask(task_ptr, &grpc_status);
GrpcRequestScheduler::ExecTask(task_ptr, &grpc_status);
response->set_table_row_count(row_count);
response->mutable_status()->set_reason(grpc_status.reason());
response->mutable_status()->set_error_code(grpc_status.error_code());
......@@ -133,13 +134,13 @@ RequestHandler::GetTableRowCount(::grpc::ServerContext* context,
}
::grpc::Status
RequestHandler::ShowTables(::grpc::ServerContext* context,
const ::milvus::grpc::Command* request,
::grpc::ServerWriter<::milvus::grpc::TableName>* writer) {
GrpcRequestHandler::ShowTables(::grpc::ServerContext *context,
const ::milvus::grpc::Command *request,
::grpc::ServerWriter<::milvus::grpc::TableName> *writer) {
BaseTaskPtr task_ptr = ShowTablesTask::Create(*writer);
::milvus::grpc::Status grpc_status;
RequestScheduler::ExecTask(task_ptr, &grpc_status);
GrpcRequestScheduler::ExecTask(task_ptr, &grpc_status);
if (grpc_status.error_code() != SERVER_SUCCESS) {
::grpc::Status status(::grpc::UNKNOWN, grpc_status.reason());
return status;
......@@ -149,21 +150,21 @@ RequestHandler::ShowTables(::grpc::ServerContext* context,
}
::grpc::Status
RequestHandler::Ping(::grpc::ServerContext* context,
const ::milvus::grpc::Command* request,
::milvus::grpc::ServerStatus* response) {
GrpcRequestHandler::Ping(::grpc::ServerContext *context,
const ::milvus::grpc::Command *request,
::milvus::grpc::ServerStatus *response) {
std::string result;
BaseTaskPtr task_ptr = PingTask::Create(request->cmd(), result);
::milvus::grpc::Status grpc_status;
RequestScheduler::ExecTask(task_ptr, &grpc_status);
GrpcRequestScheduler::ExecTask(task_ptr, &grpc_status);
response->set_info(result);
response->mutable_status()->set_reason(grpc_status.reason());
response->mutable_status()->set_error_code(grpc_status.error_code());
return ::grpc::Status::OK;
}
}
}
}
}
\ No newline at end of file
/*******************************************************************************
* Copyright 上海赜睿信息科技有限公司(Zilliz) - All Rights Reserved
* Unauthorized copying of this file, via any medium is strictly prohibited.
* Proprietary and confidential.
******************************************************************************/
* Copyright 上海赜睿信息科技有限公司(Zilliz) - All Rights Reserved
* Unauthorized copying of this file, via any medium is strictly prohibited.
* Proprietary and confidential.
******************************************************************************/
#pragma once
#include <cstdint>
......@@ -14,7 +14,8 @@
namespace zilliz {
namespace milvus {
namespace server {
class RequestHandler final : public ::milvus::grpc::MilvusService::Service {
namespace grpc {
class GrpcRequestHandler final : public ::milvus::grpc::MilvusService::Service {
public:
/**
* @brief Create table method
......@@ -32,8 +33,8 @@ public:
* @param context
*/
::grpc::Status
CreateTable(::grpc::ServerContext* context,
const ::milvus::grpc::TableSchema* request, ::milvus::grpc::Status* response) override ;
CreateTable(::grpc::ServerContext *context,
const ::milvus::grpc::TableSchema *request, ::milvus::grpc::Status *response) override;
/**
* @brief Test table existence method
......@@ -51,8 +52,8 @@ public:
* @param context
*/
::grpc::Status
HasTable(::grpc::ServerContext* context,
const ::milvus::grpc::TableName* request, ::milvus::grpc::BoolReply* response) override ;
HasTable(::grpc::ServerContext *context,
const ::milvus::grpc::TableName *request, ::milvus::grpc::BoolReply *response) override;
/**
* @brief Drop table method
......@@ -70,8 +71,8 @@ public:
* @param context
*/
::grpc::Status
DropTable(::grpc::ServerContext* context,
const ::milvus::grpc::TableName* request, ::milvus::grpc::Status* response) override;
DropTable(::grpc::ServerContext *context,
const ::milvus::grpc::TableName *request, ::milvus::grpc::Status *response) override;
/**
* @brief build index by table method
......@@ -89,8 +90,8 @@ public:
* @param context
*/
::grpc::Status
BuildIndex(::grpc::ServerContext* context,
const ::milvus::grpc::TableName* request, ::milvus::grpc::Status* response) override;
BuildIndex(::grpc::ServerContext *context,
const ::milvus::grpc::TableName *request, ::milvus::grpc::Status *response) override;
/**
......@@ -109,8 +110,9 @@ public:
* @param response
*/
::grpc::Status
InsertVector(::grpc::ServerContext* context,
const ::milvus::grpc::InsertInfos* request, ::milvus::grpc::VectorIds* response) override;
InsertVector(::grpc::ServerContext *context,
const ::milvus::grpc::InsertInfos *request,
::milvus::grpc::VectorIds *response) override;
/**
* @brief Query vector
......@@ -133,8 +135,9 @@ public:
* @param writer
*/
::grpc::Status
SearchVector(::grpc::ServerContext* context,
const ::milvus::grpc::SearchVectorInfos* request, ::grpc::ServerWriter<::milvus::grpc::TopKQueryResult>* writer) override;
SearchVector(::grpc::ServerContext *context,
const ::milvus::grpc::SearchVectorInfos *request,
::grpc::ServerWriter<::milvus::grpc::TopKQueryResult> *writer) override;
/**
* @brief Internal use query interface
......@@ -157,8 +160,9 @@ public:
* @param writer
*/
::grpc::Status
SearchVectorInFiles(::grpc::ServerContext* context,
const ::milvus::grpc::SearchVectorInFilesInfos* request, ::grpc::ServerWriter<::milvus::grpc::TopKQueryResult>* writer) override;
SearchVectorInFiles(::grpc::ServerContext *context,
const ::milvus::grpc::SearchVectorInFilesInfos *request,
::grpc::ServerWriter<::milvus::grpc::TopKQueryResult> *writer) override;
/**
* @brief Get table schema
......@@ -176,8 +180,9 @@ public:
* @param response
*/
::grpc::Status
DescribeTable(::grpc::ServerContext* context,
const ::milvus::grpc::TableName* request, ::milvus::grpc::TableSchema* response) override;
DescribeTable(::grpc::ServerContext *context,
const ::milvus::grpc::TableName *request,
::milvus::grpc::TableSchema *response) override;
/**
* @brief Get table row count
......@@ -195,8 +200,9 @@ public:
* @param context
*/
::grpc::Status
GetTableRowCount(::grpc::ServerContext* context,
const ::milvus::grpc::TableName* request, ::milvus::grpc::TableRowCount* response) override;
GetTableRowCount(::grpc::ServerContext *context,
const ::milvus::grpc::TableName *request,
::milvus::grpc::TableRowCount *response) override;
/**
* @brief List all tables in database
......@@ -214,8 +220,9 @@ public:
* @param writer
*/
::grpc::Status
ShowTables(::grpc::ServerContext* context,
const ::milvus::grpc::Command* request, ::grpc::ServerWriter< ::milvus::grpc::TableName>* writer) override;
ShowTables(::grpc::ServerContext *context,
const ::milvus::grpc::Command *request,
::grpc::ServerWriter<::milvus::grpc::TableName> *writer) override;
/**
* @brief Give the server status
......@@ -233,13 +240,12 @@ public:
* @param response
*/
::grpc::Status
Ping(::grpc::ServerContext* context,
const ::milvus::grpc::Command* request, ::milvus::grpc::ServerStatus* response) override;
Ping(::grpc::ServerContext *context,
const ::milvus::grpc::Command *request, ::milvus::grpc::ServerStatus *response) override;
};
}
}
}
}
/*******************************************************************************
* Copyright 上海赜睿信息科技有限公司(Zilliz) - All Rights Reserved
* Unauthorized copying of this file, via any medium is strictly prohibited.
* Proprietary and confidential.
******************************************************************************/
#include "RequestScheduler.h"
* Copyright 上海赜睿信息科技有限公司(Zilliz) - All Rights Reserved
* Unauthorized copying of this file, via any medium is strictly prohibited.
* Proprietary and confidential.
******************************************************************************/
#include "GrpcRequestScheduler.h"
#include "utils/Log.h"
#include "src/grpc/gen-status/status.pb.h"
......@@ -11,11 +11,12 @@
namespace zilliz {
namespace milvus {
namespace server {
namespace grpc {
using namespace ::milvus;
namespace {
const std::map<ServerError, ::milvus::grpc::ErrorCode > &ErrorMap() {
const std::map<ServerError, ::milvus::grpc::ErrorCode> &ErrorMap() {
static const std::map<ServerError, ::milvus::grpc::ErrorCode> code_map = {
{SERVER_UNEXPECTED_ERROR, ::milvus::grpc::ErrorCode::UNEXPECTED_ERROR},
{SERVER_UNSUPPORTED_ERROR, ::milvus::grpc::ErrorCode::UNEXPECTED_ERROR},
......@@ -50,7 +51,7 @@ namespace {
}
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
BaseTask::BaseTask(const std::string& task_group, bool async)
GrpcBaseTask::GrpcBaseTask(const std::string &task_group, bool async)
: task_group_(task_group),
async_(async),
done_(false),
......@@ -58,12 +59,12 @@ BaseTask::BaseTask(const std::string& task_group, bool async)
}
BaseTask::~BaseTask() {
GrpcBaseTask::~GrpcBaseTask() {
WaitToFinish();
}
ServerError
BaseTask::Execute() {
GrpcBaseTask::Execute() {
error_code_ = OnExecute();
done_ = true;
finish_cond_.notify_all();
......@@ -71,7 +72,7 @@ BaseTask::Execute() {
}
ServerError
BaseTask::SetError(ServerError error_code, const std::string& error_msg) {
GrpcBaseTask::SetError(ServerError error_code, const std::string &error_msg) {
error_code_ = error_code;
error_msg_ = error_msg;
......@@ -80,33 +81,33 @@ BaseTask::SetError(ServerError error_code, const std::string& error_msg) {
}
ServerError
BaseTask::WaitToFinish() {
std::unique_lock <std::mutex> lock(finish_mtx_);
GrpcBaseTask::WaitToFinish() {
std::unique_lock<std::mutex> lock(finish_mtx_);
finish_cond_.wait(lock, [this] { return done_; });
return error_code_;
}
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
RequestScheduler::RequestScheduler()
GrpcRequestScheduler::GrpcRequestScheduler()
: stopped_(false) {
Start();
}
RequestScheduler::~RequestScheduler() {
GrpcRequestScheduler::~GrpcRequestScheduler() {
Stop();
}
void
RequestScheduler::ExecTask(BaseTaskPtr& task_ptr, ::milvus::grpc::Status *grpc_status) {
if(task_ptr == nullptr) {
GrpcRequestScheduler::ExecTask(BaseTaskPtr &task_ptr, ::milvus::grpc::Status *grpc_status) {
if (task_ptr == nullptr) {
return;
}
RequestScheduler& scheduler = RequestScheduler::GetInstance();
GrpcRequestScheduler &scheduler = GrpcRequestScheduler::GetInstance();
scheduler.ExecuteTask(task_ptr);
if(!task_ptr->IsAsync()) {
if (!task_ptr->IsAsync()) {
task_ptr->WaitToFinish();
ServerError err = task_ptr->ErrorCode();
if (err != SERVER_SUCCESS) {
......@@ -117,8 +118,8 @@ RequestScheduler::ExecTask(BaseTaskPtr& task_ptr, ::milvus::grpc::Status *grpc_s
}
void
RequestScheduler::Start() {
if(!stopped_) {
GrpcRequestScheduler::Start() {
if (!stopped_) {
return;
}
......@@ -126,23 +127,23 @@ RequestScheduler::Start() {
}
void
RequestScheduler::Stop() {
if(stopped_) {
GrpcRequestScheduler::Stop() {
if (stopped_) {
return;
}
SERVER_LOG_INFO << "Scheduler gonna stop...";
{
std::lock_guard<std::mutex> lock(queue_mtx_);
for(auto iter : task_groups_) {
if(iter.second != nullptr) {
for (auto iter : task_groups_) {
if (iter.second != nullptr) {
iter.second->Put(nullptr);
}
}
}
for(auto iter : execute_threads_) {
if(iter == nullptr)
for (auto iter : execute_threads_) {
if (iter == nullptr)
continue;
iter->join();
......@@ -152,18 +153,18 @@ RequestScheduler::Stop() {
}
ServerError
RequestScheduler::ExecuteTask(const BaseTaskPtr& task_ptr) {
if(task_ptr == nullptr) {
GrpcRequestScheduler::ExecuteTask(const BaseTaskPtr &task_ptr) {
if (task_ptr == nullptr) {
return SERVER_NULL_POINTER;
}
ServerError err = PutTaskToQueue(task_ptr);
if(err != SERVER_SUCCESS) {
SERVER_LOG_ERROR << "Put task to queue failed with code: " << err ;
ServerError err = PutTaskToQueue(task_ptr);
if (err != SERVER_SUCCESS) {
SERVER_LOG_ERROR << "Put task to queue failed with code: " << err;
return err;
}
if(task_ptr->IsAsync()) {
if (task_ptr->IsAsync()) {
return SERVER_SUCCESS;//async execution, caller need to call WaitToFinish at somewhere
}
......@@ -172,11 +173,11 @@ RequestScheduler::ExecuteTask(const BaseTaskPtr& task_ptr) {
namespace {
void TakeTaskToExecute(TaskQueuePtr task_queue) {
if(task_queue == nullptr) {
if (task_queue == nullptr) {
return;
}
while(true) {
while (true) {
BaseTaskPtr task = task_queue->Take();
if (task == nullptr) {
SERVER_LOG_ERROR << "Take null from task queue, stop thread";
......@@ -185,22 +186,22 @@ namespace {
try {
ServerError err = task->Execute();
if(err != SERVER_SUCCESS) {
if (err != SERVER_SUCCESS) {
SERVER_LOG_ERROR << "Task failed with code: " << err;
}
} catch (std::exception& ex) {
} catch (std::exception &ex) {
SERVER_LOG_ERROR << "Task failed to execute: " << ex.what();
}
}
}
}
ServerError
RequestScheduler::PutTaskToQueue(const BaseTaskPtr& task_ptr) {
ServerError
GrpcRequestScheduler::PutTaskToQueue(const BaseTaskPtr &task_ptr) {
std::lock_guard<std::mutex> lock(queue_mtx_);
std::string group_name = task_ptr->TaskGroup();
if(task_groups_.count(group_name) > 0) {
if (task_groups_.count(group_name) > 0) {
task_groups_[group_name]->Put(task_ptr);
} else {
TaskQueuePtr queue = std::make_shared<TaskQueue>();
......@@ -219,3 +220,4 @@ RequestScheduler::PutTaskToQueue(const BaseTaskPtr& task_ptr) {
}
}
}
}
/*******************************************************************************
* Copyright 上海赜睿信息科技有限公司(Zilliz) - All Rights Reserved
* Unauthorized copying of this file, via any medium is strictly prohibited.
* Proprietary and confidential.
******************************************************************************/
* Copyright 上海赜睿信息科技有限公司(Zilliz) - All Rights Reserved
* Unauthorized copying of this file, via any medium is strictly prohibited.
* Proprietary and confidential.
******************************************************************************/
#pragma once
#include "utils/BlockingQueue.h"
......@@ -16,11 +16,13 @@
namespace zilliz {
namespace milvus {
namespace server {
namespace grpc {
class BaseTask {
class GrpcBaseTask {
protected:
BaseTask(const std::string& task_group, bool async = false);
virtual ~BaseTask();
GrpcBaseTask(const std::string &task_group, bool async = false);
virtual ~GrpcBaseTask();
public:
ServerError
......@@ -46,7 +48,7 @@ protected:
OnExecute() = 0;
ServerError
SetError(ServerError error_code, const std::string& msg);
SetError(ServerError error_code, const std::string &msg);
protected:
mutable std::mutex finish_mtx_;
......@@ -59,33 +61,35 @@ protected:
std::string error_msg_;
};
using BaseTaskPtr = std::shared_ptr<BaseTask>;
using BaseTaskPtr = std::shared_ptr<GrpcBaseTask>;
using TaskQueue = BlockingQueue<BaseTaskPtr>;
using TaskQueuePtr = std::shared_ptr<TaskQueue>;
using ThreadPtr = std::shared_ptr<std::thread>;
class RequestScheduler {
class GrpcRequestScheduler {
public:
static RequestScheduler& GetInstance() {
static RequestScheduler scheduler;
static GrpcRequestScheduler &GetInstance() {
static GrpcRequestScheduler scheduler;
return scheduler;
}
void Start();
void Stop();
ServerError
ExecuteTask(const BaseTaskPtr& task_ptr);
ExecuteTask(const BaseTaskPtr &task_ptr);
static void
ExecTask(BaseTaskPtr& task_ptr, ::milvus::grpc::Status* grpc_status);
ExecTask(BaseTaskPtr &task_ptr, ::milvus::grpc::Status *grpc_status);
protected:
RequestScheduler();
virtual ~RequestScheduler();
GrpcRequestScheduler();
virtual ~GrpcRequestScheduler();
ServerError
PutTaskToQueue(const BaseTaskPtr& task_ptr);
PutTaskToQueue(const BaseTaskPtr &task_ptr);
private:
mutable std::mutex queue_mtx_;
......@@ -97,7 +101,7 @@ private:
bool stopped_;
};
}
}
}
}
/*******************************************************************************
* Copyright 上海赜睿信息科技有限公司(Zilliz) - All Rights Reserved
* Unauthorized copying of this file, via any medium is strictly prohibited.
* Proprietary and confidential.
******************************************************************************/
* Copyright 上海赜睿信息科技有限公司(Zilliz) - All Rights Reserved
* Unauthorized copying of this file, via any medium is strictly prohibited.
* Proprietary and confidential.
******************************************************************************/
#pragma once
#include "RequestScheduler.h"
#include "GrpcRequestScheduler.h"
#include "utils/Error.h"
#include "db/Types.h"
......@@ -17,16 +17,17 @@
namespace zilliz {
namespace milvus {
namespace server {
namespace grpc {
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
class CreateTableTask : public BaseTask {
class CreateTableTask : public GrpcBaseTask {
public:
static BaseTaskPtr
Create(const ::milvus::grpc::TableSchema& schema);
Create(const ::milvus::grpc::TableSchema &schema);
protected:
explicit
CreateTableTask(const ::milvus::grpc::TableSchema& request);
CreateTableTask(const ::milvus::grpc::TableSchema &request);
ServerError
OnExecute() override;
......@@ -36,13 +37,13 @@ private:
};
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
class HasTableTask : public BaseTask {
class HasTableTask : public GrpcBaseTask {
public:
static BaseTaskPtr
Create(const std::string& table_name, bool& has_table);
Create(const std::string &table_name, bool &has_table);
protected:
HasTableTask(const std::string& request, bool& has_table);
HasTableTask(const std::string &request, bool &has_table);
ServerError
OnExecute() override;
......@@ -50,17 +51,17 @@ protected:
private:
std::string table_name_;
bool& has_table_;
bool &has_table_;
};
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
class DescribeTableTask : public BaseTask {
class DescribeTableTask : public GrpcBaseTask {
public:
static BaseTaskPtr
Create(const std::string& table_name, ::milvus::grpc::TableSchema& schema);
Create(const std::string &table_name, ::milvus::grpc::TableSchema &schema);
protected:
DescribeTableTask(const std::string& table_name, ::milvus::grpc::TableSchema& schema);
DescribeTableTask(const std::string &table_name, ::milvus::grpc::TableSchema &schema);
ServerError
OnExecute() override;
......@@ -68,18 +69,18 @@ protected:
private:
std::string table_name_;
::milvus::grpc::TableSchema& schema_;
::milvus::grpc::TableSchema &schema_;
};
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
class DropTableTask : public BaseTask {
class DropTableTask : public GrpcBaseTask {
public:
static BaseTaskPtr
Create(const std::string& table_name);
Create(const std::string &table_name);
protected:
explicit
DropTableTask(const std::string& table_name);
DropTableTask(const std::string &table_name);
ServerError
OnExecute() override;
......@@ -90,14 +91,14 @@ private:
};
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
class BuildIndexTask : public BaseTask {
class BuildIndexTask : public GrpcBaseTask {
public:
static BaseTaskPtr
Create(const std::string& table_name);
Create(const std::string &table_name);
protected:
explicit
BuildIndexTask(const std::string& table_name);
BuildIndexTask(const std::string &table_name);
ServerError
OnExecute() override;
......@@ -108,53 +109,53 @@ private:
};
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
class ShowTablesTask : public BaseTask {
class ShowTablesTask : public GrpcBaseTask {
public:
static BaseTaskPtr
Create(::grpc::ServerWriter< ::milvus::grpc::TableName>& writer);
Create(::grpc::ServerWriter<::milvus::grpc::TableName> &writer);
protected:
explicit
ShowTablesTask(::grpc::ServerWriter< ::milvus::grpc::TableName>& writer);
ShowTablesTask(::grpc::ServerWriter<::milvus::grpc::TableName> &writer);
ServerError
OnExecute() override;
private:
::grpc::ServerWriter< ::milvus::grpc::TableName> writer_;
::grpc::ServerWriter<::milvus::grpc::TableName> writer_;
};
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
class InsertVectorTask : public BaseTask {
class InsertVectorTask : public GrpcBaseTask {
public:
static BaseTaskPtr
Create(const ::milvus::grpc::InsertInfos& insert_infos,
::milvus::grpc::VectorIds& record_ids_);
Create(const ::milvus::grpc::InsertInfos &insert_infos,
::milvus::grpc::VectorIds &record_ids_);
protected:
InsertVectorTask(const ::milvus::grpc::InsertInfos& insert_infos,
::milvus::grpc::VectorIds& record_ids_);
InsertVectorTask(const ::milvus::grpc::InsertInfos &insert_infos,
::milvus::grpc::VectorIds &record_ids_);
ServerError
OnExecute() override;
private:
const ::milvus::grpc::InsertInfos insert_infos_;
::milvus::grpc::VectorIds& record_ids_;
::milvus::grpc::VectorIds &record_ids_;
};
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
class SearchVectorTask : public BaseTask {
class SearchVectorTask : public GrpcBaseTask {
public:
static BaseTaskPtr
Create(const ::milvus::grpc::SearchVectorInfos& searchVectorInfos,
const std::vector<std::string>& file_id_array,
::grpc::ServerWriter<::milvus::grpc::TopKQueryResult>& writer);
Create(const ::milvus::grpc::SearchVectorInfos &searchVectorInfos,
const std::vector<std::string> &file_id_array,
::grpc::ServerWriter<::milvus::grpc::TopKQueryResult> &writer);
protected:
SearchVectorTask(const ::milvus::grpc::SearchVectorInfos& searchVectorInfos,
const std::vector<std::string>& file_id_array,
::grpc::ServerWriter<::milvus::grpc::TopKQueryResult>& writer);
SearchVectorTask(const ::milvus::grpc::SearchVectorInfos &searchVectorInfos,
const std::vector<std::string> &file_id_array,
::grpc::ServerWriter<::milvus::grpc::TopKQueryResult> &writer);
ServerError
OnExecute() override;
......@@ -166,39 +167,39 @@ private:
};
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
class GetTableRowCountTask : public BaseTask {
class GetTableRowCountTask : public GrpcBaseTask {
public:
static BaseTaskPtr
Create(const std::string& table_name, int64_t& row_count);
Create(const std::string &table_name, int64_t &row_count);
protected:
GetTableRowCountTask(const std::string& table_name, int64_t& row_count);
GetTableRowCountTask(const std::string &table_name, int64_t &row_count);
ServerError
OnExecute() override;
private:
std::string table_name_;
int64_t& row_count_;
int64_t &row_count_;
};
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
class PingTask : public BaseTask {
class PingTask : public GrpcBaseTask {
public:
static BaseTaskPtr
Create(const std::string& cmd, std::string& result);
Create(const std::string &cmd, std::string &result);
protected:
PingTask(const std::string& cmd, std::string& result);
PingTask(const std::string &cmd, std::string &result);
ServerError
OnExecute() override;
private:
std::string cmd_;
std::string& result_;
std::string &result_;
};
}
}
}
}
\ No newline at end of file
......@@ -47,18 +47,18 @@ RequestHandler::BuildIndex(const std::string &table_name) {
void
RequestHandler::AddVector(std::vector<int64_t> &_return,
const std::string &table_name,
const std::vector<thrift::RowRecord> &record_array) {
const std::string &table_name,
const std::vector<thrift::RowRecord> &record_array) {
BaseTaskPtr task_ptr = AddVectorTask::Create(table_name, record_array, _return);
RequestScheduler::ExecTask(task_ptr);
}
void
RequestHandler::SearchVector(std::vector<thrift::TopKQueryResult> &_return,
const std::string &table_name,
const std::vector<thrift::RowRecord> &query_record_array,
const std::vector<thrift::Range> &query_range_array,
const int64_t topk) {
const std::string &table_name,
const std::vector<thrift::RowRecord> &query_record_array,
const std::vector<thrift::Range> &query_range_array,
const int64_t topk) {
// SERVER_LOG_DEBUG << "Entering RequestHandler::SearchVector";
BaseTaskPtr task_ptr = SearchVectorTask1::Create(table_name, std::vector<std::string>(), query_record_array,
query_range_array, topk, _return);
......@@ -67,10 +67,10 @@ RequestHandler::SearchVector(std::vector<thrift::TopKQueryResult> &_return,
void
RequestHandler::SearchVector2(std::vector<thrift::TopKQueryBinResult> & _return,
const std::string& table_name,
const std::vector<thrift::RowRecord> & query_record_array,
const std::vector<thrift::Range> & query_range_array,
const int64_t topk) {
const std::string& table_name,
const std::vector<thrift::RowRecord> & query_record_array,
const std::vector<thrift::Range> & query_range_array,
const int64_t topk) {
BaseTaskPtr task_ptr = SearchVectorTask2::Create(table_name, std::vector<std::string>(), query_record_array,
query_range_array, topk, _return);
RequestScheduler::ExecTask(task_ptr);
......@@ -78,11 +78,11 @@ RequestHandler::SearchVector2(std::vector<thrift::TopKQueryBinResult> & _return,
void
RequestHandler::SearchVectorInFiles(std::vector<::milvus::thrift::TopKQueryResult> &_return,
const std::string& table_name,
const std::vector<std::string> &file_id_array,
const std::vector<::milvus::thrift::RowRecord> &query_record_array,
const std::vector<::milvus::thrift::Range> &query_range_array,
const int64_t topk) {
const std::string& table_name,
const std::vector<std::string> &file_id_array,
const std::vector<::milvus::thrift::RowRecord> &query_record_array,
const std::vector<::milvus::thrift::Range> &query_range_array,
const int64_t topk) {
// SERVER_LOG_DEBUG << "Entering RequestHandler::SearchVectorInFiles. file_id_array size = " << std::to_string(file_id_array.size());
BaseTaskPtr task_ptr = SearchVectorTask1::Create(table_name, file_id_array, query_record_array,
query_range_array, topk, _return);
......
#!/bin/bash
./cmake_build/src/milvus_grpc_server -c ./conf/server_config.yaml -l ./conf/log_config.conf &
./cmake_build/src/milvus_server -c ./conf/server_config.yaml -l ./conf/log_config.conf &
Markdown is supported
0% .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑!
想要评论请 注册