From ba1964d0e844dc5899cc765ae3e5a5cb63cc1241 Mon Sep 17 00:00:00 2001 From: xuyongfei Date: Mon, 25 Jan 2021 10:12:27 +0800 Subject: [PATCH] Serving, support restful float16 output support --- build.sh | 4 +- cmake/external_libs/eigen.cmake | 21 +++ cmake/mind_expression.cmake | 10 +- mindspore_serving/CMakeLists.txt | 1 + mindspore_serving/ccsrc/common/float16.h | 33 ++++ mindspore_serving/ccsrc/common/thread_pool.cc | 7 +- .../ccsrc/master/restful/http_process.cc | 171 ++++++++++-------- .../ccsrc/master/restful/http_process.h | 4 + .../ccsrc/worker/predict_thread.cc | 32 ++-- .../ccsrc/worker/predict_thread.h | 3 +- mindspore_serving/ccsrc/worker/task_queue.cc | 20 +- tests/ut/CMakeLists.txt | 1 + ...parse_restful.cc => test_parse_restful.cc} | 83 ++++++--- .../test_start_preprocess_postprocess.cc | 1 - tests/ut/python/tests/common.py | 4 +- tests/ut/python/tests/test_grpc_request.py | 9 + .../python/tests/test_restful_base64_data.py | 42 ----- 17 files changed, 260 insertions(+), 186 deletions(-) create mode 100644 cmake/external_libs/eigen.cmake create mode 100644 mindspore_serving/ccsrc/common/float16.h rename tests/ut/cpp/tests/{parse_restful.cc => test_parse_restful.cc} (94%) diff --git a/build.sh b/build.sh index 987b3c1..5c70d9f 100644 --- a/build.sh +++ b/build.sh @@ -7,7 +7,7 @@ export BUILD_PATH="${PROJECTPATH}/build/" usage() { echo "Usage:" - echo "bash build.sh [-d] [-v] [-c] [-a on|off] [-j[n]] [-p]" + echo "bash build.sh [-d] [-v] [-c on|off] [-a on|off] [-j[n]] [-p]" echo "" echo "Options:" echo " -d Debug model" @@ -47,7 +47,7 @@ checkopts() RUN_TESTCASES="off" # Process the options - while getopts 'dvcj:a:p:e:V:t:' opt + while getopts 'dvc:j:a:p:e:V:t:' opt do LOW_OPTARG=$(echo ${OPTARG} | tr '[A-Z]' '[a-z]') diff --git a/cmake/external_libs/eigen.cmake b/cmake/external_libs/eigen.cmake new file mode 100644 index 0000000..1b4824b --- /dev/null +++ b/cmake/external_libs/eigen.cmake @@ -0,0 +1,21 @@ +set(Eigen3_CXXFLAGS "-D_FORTIFY_SOURCE=2 -O2") +set(Eigen3_CFLAGS "-D_FORTIFY_SOURCE=2 -O2") + +if(ENABLE_GITEE) + set(REQ_URL "https://gitee.com/mirrors/eigen-git-mirrorsource/repository/archive/3.3.7.tar.gz") + set(MD5 "cf6552a5d90c1aca4b5e0b011f65ea93") +else() + set(REQ_URL "https://gitlab.com/libeigen/eigen/-/archive/3.3.7/eigen-3.3.7.tar.gz") + set(MD5 "9e30f67e8531477de4117506fe44669b") +endif() + +mindspore_add_pkg(Eigen3 + VER 3.3.7 + URL ${REQ_URL} + MD5 ${MD5} + CMAKE_OPTION -DBUILD_TESTING=OFF) +find_package(Eigen3 3.3.7 REQUIRED ${MS_FIND_NO_DEFAULT_PATH}) +include_directories(${Eigen3_INC}) +include_directories(${EIGEN3_INCLUDE_DIR}) +set_property(TARGET Eigen3::Eigen PROPERTY IMPORTED_GLOBAL TRUE) +add_library(mindspore_serving::eigen ALIAS Eigen3::Eigen) diff --git a/cmake/mind_expression.cmake b/cmake/mind_expression.cmake index 2c37bf5..5986e1d 100644 --- a/cmake/mind_expression.cmake +++ b/cmake/mind_expression.cmake @@ -1,13 +1,13 @@ set(SECURE_CXX_FLAGS "") -if ("${CMAKE_CXX_COMPILER_ID}" STREQUAL "GNU") - set(SECURE_CXX_FLAGS "-fstack-protector-all -Wl,-z,relro,-z,now,-z,noexecstack") +if("${CMAKE_CXX_COMPILER_ID}" STREQUAL "GNU") + set(SECURE_CXX_FLAGS "-fstack-protector-all -Wl,-z,relro,-z,now,-z,noexecstack") endif() set(_ms_tmp_CMAKE_CXX_FLAGS_F ${CMAKE_CXX_FLAGS}) set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -fvisibility=hidden") # define third party library download function include(cmake/utils.cmake) - +include(${CMAKE_SOURCE_DIR}/cmake/external_libs/eigen.cmake) include(${CMAKE_SOURCE_DIR}/cmake/dependency_securec.cmake) include(${CMAKE_SOURCE_DIR}/cmake/external_libs/json.cmake) include(${CMAKE_SOURCE_DIR}/cmake/external_libs/protobuf.cmake) @@ -28,5 +28,5 @@ include(${CMAKE_SOURCE_DIR}/cmake/external_libs/glog.cmake) set(CMAKE_CXX_FLAGS ${_ms_tmp_CMAKE_CXX_FLAGS_F}) if(MS_BACKEND) - include(${CMAKE_SOURCE_DIR}/cmake/dependency_ms.cmake) -endif() \ No newline at end of file + include(${CMAKE_SOURCE_DIR}/cmake/dependency_ms.cmake) +endif() diff --git a/mindspore_serving/CMakeLists.txt b/mindspore_serving/CMakeLists.txt index f5d7f1c..2b1db25 100644 --- a/mindspore_serving/CMakeLists.txt +++ b/mindspore_serving/CMakeLists.txt @@ -94,6 +94,7 @@ set_property(TARGET serving_common PROPERTY POSITION_INDEPENDENT_CODE TRUE) target_link_libraries(serving_common PRIVATE PROTO_SRC_LIB mindspore_serving::event mindspore_serving::event_pthreads) target_link_libraries(serving_common PRIVATE ${_REFLECTION} ${_GRPC_GRPCPP} ${_PROTOBUF_LIBPROTOBUF} pthread) +target_link_libraries(serving_common PRIVATE mindspore_serving::eigen) # python add_compile_definitions(ENABLE_PYTHON) diff --git a/mindspore_serving/ccsrc/common/float16.h b/mindspore_serving/ccsrc/common/float16.h new file mode 100644 index 0000000..6f75329 --- /dev/null +++ b/mindspore_serving/ccsrc/common/float16.h @@ -0,0 +1,33 @@ +/** + * Copyright 2020 Huawei Technologies Co., Ltd + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +#ifndef MINDSPORE_SERVING_COMMON_FLOAT16_H_ +#define MINDSPORE_SERVING_COMMON_FLOAT16_H_ + +#if defined(ENABLE_ARM32) || defined(ENABLE_ARM64) +// Built for lite and ARM +#include + +using float16 = float16_t; +inline float half_to_float(float16 h) { return static_cast(h); } +#else +#include +#include "Eigen/Core" + +using float16 = Eigen::half; +using HalfToFloat = std::function; +const inline HalfToFloat half_to_float = Eigen::half_impl::half_to_float; +#endif +#endif // MINDSPORE_SERVING_COMMON_FLOAT16_H_ diff --git a/mindspore_serving/ccsrc/common/thread_pool.cc b/mindspore_serving/ccsrc/common/thread_pool.cc index d86da22..cd3caa6 100644 --- a/mindspore_serving/ccsrc/common/thread_pool.cc +++ b/mindspore_serving/ccsrc/common/thread_pool.cc @@ -31,8 +31,11 @@ ThreadPool::ThreadPool(uint32_t size) : is_stoped_(false), idle_thrd_num_(size < } ThreadPool::~ThreadPool() { - is_stoped_.store(true); - cond_var_.notify_all(); + { + std::unique_lock lock{m_lock_}; + is_stoped_.store(true); + cond_var_.notify_all(); + } for (std::thread &thd : pool_) { if (thd.joinable()) { diff --git a/mindspore_serving/ccsrc/master/restful/http_process.cc b/mindspore_serving/ccsrc/master/restful/http_process.cc index f271db4..ed8ed2b 100644 --- a/mindspore_serving/ccsrc/master/restful/http_process.cc +++ b/mindspore_serving/ccsrc/master/restful/http_process.cc @@ -23,6 +23,7 @@ #include #include "common/serving_common.h" #include "master/restful/http_handle.h" +#include "common/float16.h" using mindspore::serving::proto::Instance; using mindspore::serving::proto::PredictReply; @@ -304,12 +305,8 @@ Status RestfulService::CheckObj(const json &js) { int shape_count = 0; int type_count = 0; for (auto item = js.begin(); item != js.end(); ++item) { - auto key = item.key(); + const auto &key = item.key(); auto value = item.value(); - if (key != kB64 && key != kType && key != kShape) { - return INFER_STATUS_LOG_ERROR(INVALID_INPUTS) - << "json object, key is not ['b64', 'type', 'shape'], fail key:" << key; - } if (key == kB64) { b64_count++; } else if (key == kType) { @@ -346,6 +343,9 @@ Status RestfulService::CheckObj(const json &js) { } } shape_count++; + } else { + return INFER_STATUS_LOG_ERROR(INVALID_INPUTS) + << "json object, key is not ['b64', 'type', 'shape'], fail key:" << key; } } @@ -364,8 +364,7 @@ Status RestfulService::CheckObj(const json &js) { return SUCCESS; } -// 1. parse request common func -Status RestfulService::ParseItem(const json &value, ProtoTensor *const pb_tensor) { +Status RestfulService::ParseItemScalar(const json &value, ProtoTensor *const pb_tensor) { Status status(SUCCESS); std::vector scalar_shape = {}; if (value.is_number_integer()) { @@ -391,87 +390,98 @@ Status RestfulService::ParseItem(const json &value, ProtoTensor *const pb_tensor pb_tensor->set_data_type(type); pb_tensor->set_shape(scalar_shape); status = GetScalarByType(type, value, 0, pb_tensor); - } else if (value.is_object()) { - status = CheckObj(value); - if (status != SUCCESS) { - return status; - } - - DataType type = GetObjDataType(value); - if (type == kMSI_Unknown) { - return INFER_STATUS_LOG_ERROR(INVALID_INPUTS) << "json object, type is unknown"; - } + } else if (value.is_null()) { + return INFER_STATUS_LOG_ERROR(INVALID_INPUTS) << "json value is null, it is not supported"; + } else if (value.is_discarded()) { + return INFER_STATUS_LOG_ERROR(INVALID_INPUTS) << "json value is discarded type, it is not supported"; + } else { + return INFER_STATUS_LOG_ERROR(INVALID_INPUTS) << "json value type is unregistered"; + } + return status; +} - std::vector shape = GetObjShape(value); - bool is_tensor = false; - if (type != kMSI_String && type != kMSI_Bytes) { - is_tensor = true; - } - if (is_tensor) { - size_t shape_size = std::accumulate(shape.begin(), shape.end(), 1LL, std::multiplies()); - size_t type_size = pb_tensor->GetTypeSize(type); - pb_tensor->resize_data(shape_size * type_size); - } +Status RestfulService::ParseItemObject(const json &value, ProtoTensor *const pb_tensor) { + auto status = CheckObj(value); + if (status != SUCCESS) { + return status; + } - status = CheckObjTypeMatchShape(type, shape); - if (status != SUCCESS) { - return status; - } - pb_tensor->set_data_type(type); - pb_tensor->set_shape(shape); - status = GetScalarByType(serving::kMSI_Bytes, value[kB64], 0, pb_tensor); - } else if (value.is_array()) { - HTTP_DATA_TYPE type_format = HTTP_DATA_NONE; - auto shape = GetArrayShape(value); - if (shape.empty()) { - return INFER_STATUS_LOG_ERROR(INVALID_INPUTS) << "json array, shape is empty"; - } - DataType data_type = GetArrayDataType(value, &type_format); - if (data_type == kMSI_Unknown) { - return INFER_STATUS_LOG_ERROR(INVALID_INPUTS) << "json array, data type is unknown"; - } + DataType type = GetObjDataType(value); + if (type == kMSI_Unknown) { + return INFER_STATUS_LOG_ERROR(INVALID_INPUTS) << "json object, type is unknown"; + } - bool is_tensor = false; - if (data_type != kMSI_String && data_type != kMSI_Bytes) { - is_tensor = true; - } + std::vector shape = GetObjShape(value); + bool is_tensor = false; + if (type != kMSI_String && type != kMSI_Bytes) { + is_tensor = true; + } + if (is_tensor) { + size_t shape_size = std::accumulate(shape.begin(), shape.end(), 1LL, std::multiplies()); + size_t type_size = pb_tensor->GetTypeSize(type); + pb_tensor->resize_data(shape_size * type_size); + } + status = CheckObjTypeMatchShape(type, shape); + if (status != SUCCESS) { + return status; + } + pb_tensor->set_data_type(type); + pb_tensor->set_shape(shape); + status = GetScalarByType(serving::kMSI_Bytes, value[kB64], 0, pb_tensor); + return status; +} - // instances mode:only support one item - if (request_type_ == kInstanceType) { - if (!is_tensor) { - size_t elements_nums = std::accumulate(shape.begin(), shape.end(), 1LL, std::multiplies()); - if (elements_nums != 1) { - return INFER_STATUS_LOG_ERROR(INVALID_INPUTS) << "json array, string or bytes type only support one item"; - } +Status RestfulService::ParseItemArray(const json &value, ProtoTensor *const pb_tensor) { + HTTP_DATA_TYPE type_format = HTTP_DATA_NONE; + auto shape = GetArrayShape(value); + if (shape.empty()) { + return INFER_STATUS_LOG_ERROR(INVALID_INPUTS) << "json array, shape is empty"; + } + DataType data_type = GetArrayDataType(value, &type_format); + if (data_type == kMSI_Unknown) { + return INFER_STATUS_LOG_ERROR(INVALID_INPUTS) << "json array, data type is unknown"; + } + bool is_tensor = false; + if (data_type != kMSI_String && data_type != kMSI_Bytes) { + is_tensor = true; + } + // instances mode:only support one item + if (request_type_ == kInstanceType) { + if (!is_tensor) { + size_t elements_nums = std::accumulate(shape.begin(), shape.end(), 1LL, std::multiplies()); + if (elements_nums != 1) { + return INFER_STATUS_LOG_ERROR(INVALID_INPUTS) << "json array, string or bytes type only support one item"; } } + } + // set real data type + pb_tensor->set_data_type(data_type); + pb_tensor->set_shape(shape); - // set real data type - pb_tensor->set_data_type(data_type); - pb_tensor->set_shape(shape); + if (is_tensor) { + size_t shape_size = std::accumulate(shape.begin(), shape.end(), 1LL, std::multiplies()); + size_t type_size = pb_tensor->GetTypeSize(data_type); + pb_tensor->resize_data(shape_size * type_size); + } - if (is_tensor) { - size_t shape_size = std::accumulate(shape.begin(), shape.end(), 1LL, std::multiplies()); - size_t type_size = pb_tensor->GetTypeSize(data_type); - pb_tensor->resize_data(shape_size * type_size); + if (type_format == HTTP_DATA_OBJ) { + if (data_type != kMSI_Bytes && data_type != kMSI_String) { + return INFER_STATUS_LOG_ERROR(INVALID_INPUTS) + << "json array, item is object type, object only support string or bytes type"; } + } + return RecursiveGetArray(value, 0, 0, type_format, pb_tensor); +} - if (type_format == HTTP_DATA_OBJ) { - if (data_type != kMSI_Bytes && data_type != kMSI_String) { - return INFER_STATUS_LOG_ERROR(INVALID_INPUTS) - << "json array, item is object type, object only support string or bytes type"; - } - } - status = RecursiveGetArray(value, 0, 0, type_format, pb_tensor); - if (status != SUCCESS) { - return status; - } - } else if (value.is_null()) { - return INFER_STATUS_LOG_ERROR(INVALID_INPUTS) << "json value is null, it is not supported"; - } else if (value.is_discarded()) { - return INFER_STATUS_LOG_ERROR(INVALID_INPUTS) << "json value is discarded type, it is not supported"; +// 1. parse request common func +Status RestfulService::ParseItem(const json &value, ProtoTensor *const pb_tensor) { + if (value.is_object()) { + return ParseItemObject(value, pb_tensor); + } else if (value.is_array()) { + return ParseItemArray(value, pb_tensor); + } else { + return ParseItemScalar(value, pb_tensor); } - return status; } Status RestfulService::RecursiveGetArray(const json &json_data, size_t depth, size_t data_index, @@ -892,9 +902,12 @@ Status RestfulService::ParseScalar(const ProtoTensor &pb_tensor, size_t index, j case kMSI_Uint64: status = ParseScalarData(pb_tensor, false, index, js); break; - case kMSI_Float16: - status = INFER_STATUS_LOG_ERROR(INVALID_INPUTS) << "float16 reply is not supported"; + case kMSI_Float16: { + const float16 *data = reinterpret_cast(pb_tensor.data()) + index; + float value = half_to_float(*data); + *js = value; break; + } case kMSI_Float32: status = ParseScalarData(pb_tensor, false, index, js); break; diff --git a/mindspore_serving/ccsrc/master/restful/http_process.h b/mindspore_serving/ccsrc/master/restful/http_process.h index 4163e4d..be7e668 100644 --- a/mindspore_serving/ccsrc/master/restful/http_process.h +++ b/mindspore_serving/ccsrc/master/restful/http_process.h @@ -76,6 +76,10 @@ class RestfulService { Status GetInstancesType(const json &instances); Status ParseKeyInstances(const json &instances, proto::PredictRequest *const request); Status PaserKeyOneInstance(const json &instance_msg, proto::PredictRequest *const request); + + Status ParseItemScalar(const json &value, ProtoTensor *const pb_tensor); + Status ParseItemArray(const json &value, ProtoTensor *const pb_tensor); + Status ParseItemObject(const json &value, ProtoTensor *const pb_tensor); Status ParseItem(const json &value, ProtoTensor *const pb_tensor); Status ParseRequest(const std::shared_ptr &restful_request, proto::PredictRequest *const request); diff --git a/mindspore_serving/ccsrc/worker/predict_thread.cc b/mindspore_serving/ccsrc/worker/predict_thread.cc index 134dc79..b0ccc96 100644 --- a/mindspore_serving/ccsrc/worker/predict_thread.cc +++ b/mindspore_serving/ccsrc/worker/predict_thread.cc @@ -27,7 +27,7 @@ serving::PredictThread::PredictThread() {} PredictThread::~PredictThread() { Stop(); } Status PredictThread::PushPredictTask(const std::vector &inputs) { - if (!has_started) { + if (!is_running_) { MSI_LOG_EXCEPTION << "Predict task thread has not been started"; } std::unique_lock lock{m_lock_}; @@ -44,33 +44,39 @@ Status PredictThread::PushPredictTask(const std::vector &inputs) { void PredictThread::ThreadFunc(PredictThread *queue) { queue->Predict(); } void PredictThread::Predict() { - while (!is_stoped_) { - std::vector intances; + while (true) { + std::vector instances; { std::unique_lock lock{m_lock_}; + if (!is_running_) { + break; + } if (predict_buffer_.empty()) { - cond_var_.wait(lock, [this] { return is_stoped_.load() || !predict_buffer_.empty(); }); - if (is_stoped_) { + cond_var_.wait(lock, [this] { return !is_running_ || !predict_buffer_.empty(); }); + if (!is_running_) { return; } } for (uint32_t i = 0; i < batch_size_ && !predict_buffer_.empty(); i++) { - intances.push_back(predict_buffer_.front()); + instances.push_back(predict_buffer_.front()); predict_buffer_.pop(); } } MSI_TIME_STAMP_START(InvokePredict) - predict_fun_(intances); + predict_fun_(instances); MSI_TIME_STAMP_END(InvokePredict) } } void PredictThread::Stop() { - if (is_stoped_) { - return; + { + std::unique_lock lock{m_lock_}; + if (!is_running_) { + return; + } + is_running_ = false; + cond_var_.notify_all(); } - is_stoped_.store(true); - cond_var_.notify_all(); if (predict_thread_.joinable()) { try { predict_thread_.join(); @@ -81,13 +87,13 @@ void PredictThread::Stop() { } void PredictThread::Start(PredictFun predict_fun, uint32_t batch_size) { - if (has_started) { + if (is_running_) { return; } predict_fun_ = std::move(predict_fun); batch_size_ = batch_size; predict_thread_ = std::thread(ThreadFunc, this); - has_started = true; + is_running_ = true; } } // namespace mindspore::serving diff --git a/mindspore_serving/ccsrc/worker/predict_thread.h b/mindspore_serving/ccsrc/worker/predict_thread.h index ea6e15c..ba455c8 100644 --- a/mindspore_serving/ccsrc/worker/predict_thread.h +++ b/mindspore_serving/ccsrc/worker/predict_thread.h @@ -44,8 +44,7 @@ class MS_API PredictThread { std::mutex m_lock_; std::condition_variable cond_var_; - std::atomic is_stoped_{false}; - std::atomic has_started = false; + std::atomic is_running_ = false; static void ThreadFunc(PredictThread *queue); void Predict(); diff --git a/mindspore_serving/ccsrc/worker/task_queue.cc b/mindspore_serving/ccsrc/worker/task_queue.cc index fc32d1f..43fe511 100644 --- a/mindspore_serving/ccsrc/worker/task_queue.cc +++ b/mindspore_serving/ccsrc/worker/task_queue.cc @@ -67,11 +67,11 @@ void TaskQueue::PushTask(const std::string &task_name, uint64_t worker_id, const void TaskQueue::PushTaskResult(uint64_t worker_id, const Instance &input, const ResultInstance &output) { auto it = callback_map_.find(worker_id); if (it == callback_map_.end()) { - MSI_LOG_ERROR << "Worker serivce " << worker_id << " has not specificated callback"; + MSI_LOG_ERROR << "Worker service " << worker_id << " has not specified callback"; return; } if (it->second == nullptr) { - MSI_LOG_ERROR << "Worker serivce " << worker_id << " has not specify callback preprocess"; + MSI_LOG_ERROR << "Worker service " << worker_id << " has not specify callback preprocess"; return; } it->second({input}, {output}); @@ -81,11 +81,11 @@ void TaskQueue::PushTaskResult(uint64_t worker_id, const std::vector & const std::vector &outputs) { auto it = callback_map_.find(worker_id); if (it == callback_map_.end()) { - MSI_LOG_ERROR << "Worker serivce " << worker_id << " has not specificated callback"; + MSI_LOG_ERROR << "Worker service " << worker_id << " has not specified callback"; return; } if (it->second == nullptr) { - MSI_LOG_ERROR << "Worker serivce " << worker_id << " has not specify callback preprocess"; + MSI_LOG_ERROR << "Worker service " << worker_id << " has not specify callback preprocess"; return; } it->second(inputs, outputs); @@ -114,11 +114,11 @@ Status TaskQueue::PushTaskPyResult(const std::vector &outputs) { void TaskQueue::PopTask(TaskItem *task_item) { MSI_EXCEPTION_IF_NULL(task_item); + std::unique_lock lock{*lock_}; if (!is_running) { // before start, or after stop task_item->task_type = kTaskTypeStop; return; } - std::unique_lock lock{*lock_}; while (true) { if (task_priority_list_.empty()) { cond_var_->wait(lock, [this] { return !is_running || !task_priority_list_.empty(); }); @@ -127,6 +127,10 @@ void TaskQueue::PopTask(TaskItem *task_item) { return; } } + if (task_priority_list_.empty()) { + MSI_LOG_EXCEPTION << "task_priority_list_.empty(), is_running " << is_running << ", task_priority_list_ size " + << task_priority_list_.size(); + } auto task_item_info = task_priority_list_.front(); task_priority_list_.pop(); auto &cur_task = task_map_[task_item_info]; @@ -142,11 +146,11 @@ void TaskQueue::PopTask(TaskItem *task_item) { void TaskQueue::TryPopTask(TaskItem *task_item) { MSI_EXCEPTION_IF_NULL(task_item); + std::unique_lock lock{*lock_}; if (!is_running) { // before start, or after stop task_item->task_type = kTaskTypeStop; return; } - std::unique_lock lock{*lock_}; while (true) { if (task_priority_list_.empty()) { task_item->task_type = kTaskTypeEmpty; @@ -174,6 +178,7 @@ void TaskQueue::TryPopPyTask(TaskItem *task_item) { } void TaskQueue::Start() { + std::unique_lock lock{*lock_}; if (is_running) { return; } @@ -185,6 +190,7 @@ void TaskQueue::Start() { } void TaskQueue::Stop() { + std::unique_lock lock{*lock_}; if (!is_running) { return; } @@ -286,10 +292,10 @@ void TaskQueueThreadPool::Start(uint32_t size) { return; } is_running = true; + task_queue_->Start(); for (uint32_t i = 0; i < size; ++i) { pool_.emplace_back(ThreadFunc, this); } - task_queue_->Start(); } void TaskQueueThreadPool::Stop() { diff --git a/tests/ut/CMakeLists.txt b/tests/ut/CMakeLists.txt index a4d4611..97e13e5 100644 --- a/tests/ut/CMakeLists.txt +++ b/tests/ut/CMakeLists.txt @@ -88,6 +88,7 @@ add_library(serving_ut_common STATIC ${UT_SERVING_COMMON}) target_link_libraries(serving_ut_common PRIVATE ${_REFLECTION} ${_GRPC_GRPCPP} ${_PROTOBUF_LIBPROTOBUF}) target_link_libraries(serving_ut_common PRIVATE mindspore_serving::event mindspore_serving::event_pthreads) target_link_libraries(serving_ut_common PRIVATE pthread mindspore_serving::glog) +target_link_libraries(serving_ut_common PRIVATE mindspore_serving::eigen) target_link_libraries(serving_ut_common PRIVATE ${SECUREC_LIBRARY}) # copy libevent lib diff --git a/tests/ut/cpp/tests/parse_restful.cc b/tests/ut/cpp/tests/test_parse_restful.cc similarity index 94% rename from tests/ut/cpp/tests/parse_restful.cc rename to tests/ut/cpp/tests/test_parse_restful.cc index e6f59e0..452deac 100644 --- a/tests/ut/cpp/tests/parse_restful.cc +++ b/tests/ut/cpp/tests/test_parse_restful.cc @@ -65,7 +65,8 @@ TEST_F(TestParseInput, test_parse_SUCCESS) { } )"_json; - struct evhttp_request *request = new evhttp_request(); + struct evhttp_request request_local = {}; + struct evhttp_request *request = &request_local; int size = 100; std::shared_ptr request_msg = std::make_shared(request, size); request_msg->request_message_ = js; @@ -76,10 +77,10 @@ TEST_F(TestParseInput, test_parse_SUCCESS) { Status status = restful_service.ParseRequest(restful_request, &predict_request); ASSERT_EQ(status.StatusCode(), SUCCESS); ASSERT_EQ(predict_request.instances().size(), 3); - for (int32_t i = 0; i < predict_request.instances().size(); i++) { - auto &cur_instance = predict_request.instances(i); + for (int32_t k = 0; k < predict_request.instances().size(); k++) { + auto &cur_instance = predict_request.instances(k); auto &items = cur_instance.items(); - if (i == 0) { + if (k == 0) { ASSERT_EQ(items.size(), 6); for (const auto &item : items) { ProtoTensor pb_tensor(const_cast(&item.second)); @@ -119,7 +120,7 @@ TEST_F(TestParseInput, test_parse_SUCCESS) { ASSERT_EQ(value, "ut_test"); } } - } else if (i == 1) { + } else if (k == 1) { ASSERT_EQ(items.size(), 6); for (const auto &item : items) { ProtoTensor pb_tensor(const_cast(&item.second)); @@ -183,7 +184,7 @@ TEST_F(TestParseInput, test_parse_SUCCESS) { ASSERT_EQ(value, "ut_test"); } } - } else if (i == 2) { + } else if (k == 2) { ASSERT_EQ(items.size(), 6); for (const auto &item : items) { ProtoTensor pb_tensor(const_cast(&item.second)); @@ -259,7 +260,8 @@ TEST_F(TestParseInput, test_instances_empty_FAIL) { } )"_json; - struct evhttp_request *request = new evhttp_request(); + struct evhttp_request request_local = {}; + struct evhttp_request *request = &request_local; int size = 100; std::shared_ptr request_msg = std::make_shared(request, size); request_msg->request_message_ = js; @@ -285,7 +287,8 @@ TEST_F(TestParseInput, test_instances_incorrect_FAIL) { } )"_json; - struct evhttp_request *request = new evhttp_request(); + struct evhttp_request request_local = {}; + struct evhttp_request *request = &request_local; int size = 100; std::shared_ptr request_msg = std::make_shared(request, size); request_msg->request_message_ = js; @@ -311,7 +314,8 @@ TEST_F(TestParseInput, test_key_empty_FAIL) { } )"_json; - struct evhttp_request *request = new evhttp_request(); + struct evhttp_request request_local = {}; + struct evhttp_request *request = &request_local; int size = 100; std::shared_ptr request_msg = std::make_shared(request, size); request_msg->request_message_ = js; @@ -337,7 +341,8 @@ TEST_F(TestParseInput, test_value_empty_SUCCESS) { } )"_json; - struct evhttp_request *request = new evhttp_request(); + struct evhttp_request request_local = {}; + struct evhttp_request *request = &request_local; int size = 100; std::shared_ptr request_msg = std::make_shared(request, size); request_msg->request_message_ = js; @@ -363,7 +368,8 @@ TEST_F(TestParseInput, test_obj_unknown_key_FAIL) { } )"_json; - struct evhttp_request *request = new evhttp_request(); + struct evhttp_request request_local = {}; + struct evhttp_request *request = &request_local; int size = 100; std::shared_ptr request_msg = std::make_shared(request, size); request_msg->request_message_ = js; @@ -389,7 +395,8 @@ TEST_F(TestParseInput, test_obj_nob64_key_FAIL) { } )"_json; - struct evhttp_request *request = new evhttp_request(); + struct evhttp_request request_local = {}; + struct evhttp_request *request = &request_local; int size = 100; std::shared_ptr request_msg = std::make_shared(request, size); request_msg->request_message_ = js; @@ -415,7 +422,8 @@ TEST_F(TestParseInput, test_obj_illegal_b64value_FAIL) { } )"_json; - struct evhttp_request *request = new evhttp_request(); + struct evhttp_request request_local = {}; + struct evhttp_request *request = &request_local; int size = 100; std::shared_ptr request_msg = std::make_shared(request, size); request_msg->request_message_ = js; @@ -441,7 +449,8 @@ TEST_F(TestParseInput, test_obj_unknown_type_FAIL) { } )"_json; - struct evhttp_request *request = new evhttp_request(); + struct evhttp_request request_local = {}; + struct evhttp_request *request = &request_local; int size = 100; std::shared_ptr request_msg = std::make_shared(request, size); request_msg->request_message_ = js; @@ -467,7 +476,8 @@ TEST_F(TestParseInput, test_obj_error_shape_format_FAIL) { } )"_json; - struct evhttp_request *request = new evhttp_request(); + struct evhttp_request request_local = {}; + struct evhttp_request *request = &request_local; int size = 100; std::shared_ptr request_msg = std::make_shared(request, size); request_msg->request_message_ = js; @@ -493,7 +503,8 @@ TEST_F(TestParseInput, test_obj_error_shape_format2_FAIL) { } )"_json; - struct evhttp_request *request = new evhttp_request(); + struct evhttp_request request_local = {}; + struct evhttp_request *request = &request_local; int size = 100; std::shared_ptr request_msg = std::make_shared(request, size); request_msg->request_message_ = js; @@ -519,7 +530,8 @@ TEST_F(TestParseInput, test_obj_error_shape_value_FAIL) { } )"_json; - struct evhttp_request *request = new evhttp_request(); + struct evhttp_request request_local = {}; + struct evhttp_request *request = &request_local; int size = 100; std::shared_ptr request_msg = std::make_shared(request, size); request_msg->request_message_ = js; @@ -545,7 +557,8 @@ TEST_F(TestParseInput, test_obj_error_shape_value2_FAIL) { } )"_json; - struct evhttp_request *request = new evhttp_request(); + struct evhttp_request request_local = {}; + struct evhttp_request *request = &request_local; int size = 100; std::shared_ptr request_msg = std::make_shared(request, size); request_msg->request_message_ = js; @@ -571,7 +584,8 @@ TEST_F(TestParseInput, test_obj_error_shape_value3_FAIL) { } )"_json; - struct evhttp_request *request = new evhttp_request(); + struct evhttp_request request_local = {}; + struct evhttp_request *request = &request_local; int size = 100; std::shared_ptr request_msg = std::make_shared(request, size); request_msg->request_message_ = js; @@ -597,7 +611,8 @@ TEST_F(TestParseInput, test_tensor_value_empty_FAIL) { } )"_json; - struct evhttp_request *request = new evhttp_request(); + struct evhttp_request request_local = {}; + struct evhttp_request *request = &request_local; int size = 100; std::shared_ptr request_msg = std::make_shared(request, size); request_msg->request_message_ = js; @@ -623,7 +638,8 @@ TEST_F(TestParseInput, test_tensor_value_diff_type_FAIL) { } )"_json; - struct evhttp_request *request = new evhttp_request(); + struct evhttp_request request_local = {}; + struct evhttp_request *request = &request_local; int size = 100; std::shared_ptr request_msg = std::make_shared(request, size); request_msg->request_message_ = js; @@ -649,7 +665,8 @@ TEST_F(TestParseInput, test_tensor_value_diff_dimention_FAIL) { } )"_json; - struct evhttp_request *request = new evhttp_request(); + struct evhttp_request request_local = {}; + struct evhttp_request *request = &request_local; int size = 100; std::shared_ptr request_msg = std::make_shared(request, size); request_msg->request_message_ = js; @@ -675,7 +692,8 @@ TEST_F(TestParseInput, test_tensor_multi_object_FAIL) { } )"_json; - struct evhttp_request *request = new evhttp_request(); + struct evhttp_request request_local = {}; + struct evhttp_request *request = &request_local; int size = 100; std::shared_ptr request_msg = std::make_shared(request, size); request_msg->request_message_ = js; @@ -709,7 +727,8 @@ TEST_F(TestParseReply, test_reply_SUCCESS) { } )"_json; - struct evhttp_request *request = new evhttp_request(); + struct evhttp_request request_local = {}; + struct evhttp_request *request = &request_local; int size = 100; std::shared_ptr request_msg = std::make_shared(request, size); request_msg->request_message_ = js; @@ -950,7 +969,8 @@ TEST_F(TestParseReply, test_reply_instances_num_not_match_FAIL) { } )"_json; - struct evhttp_request *request = new evhttp_request(); + struct evhttp_request request_local = {}; + struct evhttp_request *request = &request_local; int size = 100; std::shared_ptr request_msg = std::make_shared(request, size); request_msg->request_message_ = js; @@ -1099,7 +1119,8 @@ TEST_F(TestParseReply, test_reply_error_num_not_match_FAIL) { } )"_json; - struct evhttp_request *request = new evhttp_request(); + struct evhttp_request request_local = {}; + struct evhttp_request *request = &request_local; int size = 100; std::shared_ptr request_msg = std::make_shared(request, size); request_msg->request_message_ = js; @@ -1138,7 +1159,8 @@ TEST_F(TestParseReply, test_reply_type_not_set_FAIL) { } )"_json; - struct evhttp_request *request = new evhttp_request(); + struct evhttp_request request_local = {}; + struct evhttp_request *request = &request_local; int size = 100; std::shared_ptr request_msg = std::make_shared(request, size); request_msg->request_message_ = js; @@ -1168,7 +1190,7 @@ TEST_F(TestParseReply, test_reply_type_not_set_FAIL) { ASSERT_NE(status2.StatusCode(), SUCCESS); } -TEST_F(TestParseReply, test_reply_type_fp16_FAIL) { +TEST_F(TestParseReply, test_reply_type_fp16_SUCCESS) { nlohmann::json js = R"( {"instances":[ { @@ -1183,7 +1205,8 @@ TEST_F(TestParseReply, test_reply_type_fp16_FAIL) { } )"_json; - struct evhttp_request *request = new evhttp_request(); + struct evhttp_request request_local = {}; + struct evhttp_request *request = &request_local; int size = 100; std::shared_ptr request_msg = std::make_shared(request, size); request_msg->request_message_ = js; @@ -1210,7 +1233,7 @@ TEST_F(TestParseReply, test_reply_type_fp16_FAIL) { map_item["key_float16"] = tensor_float; Status status2 = restful_service.ParseReply(reply, &out_js); - ASSERT_NE(status2.StatusCode(), SUCCESS); + ASSERT_EQ(status2.StatusCode(), SUCCESS); } } // namespace serving } // namespace mindspore diff --git a/tests/ut/cpp/tests/test_start_preprocess_postprocess.cc b/tests/ut/cpp/tests/test_start_preprocess_postprocess.cc index 716f6ca..9149120 100644 --- a/tests/ut/cpp/tests/test_start_preprocess_postprocess.cc +++ b/tests/ut/cpp/tests/test_start_preprocess_postprocess.cc @@ -56,7 +56,6 @@ TEST_F(TestPreprocessPostprocess, test_master_worker_with_preproces_and_postproc // start_servable Status status = StartServable("test_servable_dir", "test_servable", 0); EXPECT_TRUE(status.IsSuccess()); - // run servable proto::PredictRequest request; size_t instances_count = 3; diff --git a/tests/ut/python/tests/common.py b/tests/ut/python/tests/common.py index fb6ac43..e78c70c 100644 --- a/tests/ut/python/tests/common.py +++ b/tests/ut/python/tests/common.py @@ -253,9 +253,7 @@ def int_plus_1(int_val): return value def float_postprocess(float_val): - value = float_val + 1 - if value.dtype == np.float16: - value = value.astype(np.float32) + value = (float_val + 1).astype(float_val.dtype) # also support float16 input and output return value @register.register_method(output_names=["value"]) diff --git a/tests/ut/python/tests/test_grpc_request.py b/tests/ut/python/tests/test_grpc_request.py index 2592369..2eb6816 100644 --- a/tests/ut/python/tests/test_grpc_request.py +++ b/tests/ut/python/tests/test_grpc_request.py @@ -510,6 +510,9 @@ def common_test_grpc_request_np_float_type_scalar_input_output_success(dtype): client = create_client("localhost", 5500, base.servable_name, "float_plus_1") result = client.infer(instances) + assert result[0]["value"].dtype == dtype + assert result[1]["value"].dtype == dtype + assert result[2]["value"].dtype == dtype assert result[0]["value"].tolist() == y_data_list[0] assert result[1]["value"].tolist() == y_data_list[1] assert result[2]["value"].tolist() == y_data_list[2] @@ -544,6 +547,9 @@ def common_test_grpc_request_np_float_type_1d_array_input_output_success(dtype): client = create_client("localhost", 5500, base.servable_name, "float_plus_1") result = client.infer(instances) + assert result[0]["value"].dtype == dtype + assert result[1]["value"].dtype == dtype + assert result[2]["value"].dtype == dtype assert result[0]["value"].tolist() == y_data_list[0] assert result[1]["value"].tolist() == y_data_list[1] assert result[2]["value"].tolist() == y_data_list[2] @@ -580,6 +586,9 @@ def common_test_grpc_request_np_float_type_2d_array_input_output_success(dtype): client = create_client("localhost", 5500, base.servable_name, "float_plus_1") result = client.infer(instances) + assert result[0]["value"].dtype == dtype + assert result[1]["value"].dtype == dtype + assert result[2]["value"].dtype == dtype assert result[0]["value"].tolist() == y_data_list[0] assert result[1]["value"].tolist() == y_data_list[1] assert result[2]["value"].tolist() == y_data_list[2] diff --git a/tests/ut/python/tests/test_restful_base64_data.py b/tests/ut/python/tests/test_restful_base64_data.py index 487a0bb..c783298 100644 --- a/tests/ut/python/tests/test_restful_base64_data.py +++ b/tests/ut/python/tests/test_restful_base64_data.py @@ -579,48 +579,6 @@ def mix_all_type(bool_val, int_val, float_val, str_val, bytes_val): assert b64_decode_to_str(result[i]["bytes_val"]) == "DEF456" -@serving_test -def test_restful_base64_float16_2d_array_not_support_fp16_output_failed(): - base = ServingTestBase() - servable_content = servable_config_import - servable_content += servable_config_declare_servable - servable_content += r""" -def preprocess(float_val): - return np.ones([2,2], np.float32), np.ones([2,2], np.float32) - -def postprocess(float_val): - return float_val + 1 - -@register.register_method(output_names=["value"]) -def float_plus_1(float_val): - x1, x2 = register.call_preprocess(preprocess, float_val) - y = register.call_servable(x1, x2) - value = register.call_postprocess(postprocess, float_val) - return value -""" - base.init_servable_with_servable_config(1, servable_content) - worker.start_servable_in_master(base.servable_dir, base.servable_name) - master.start_restful_server("0.0.0.0", 5500) - # Client - instances = [{}, {}, {}] - - dtype = np.float16 - dtype_str_map = {np.float16: "fp16", np.float32: "fp32", np.float64: "fp64"} - assert dtype in dtype_str_map - - y_data_list = [] - for i, instance in enumerate(instances): - val = i * 2.2 * (-1 if i % 2 == 0 else 1) # 0, 2.2 ,-4.4 - val = [[val] * (i + 1)] * (i + 1) - val = np.array(val).astype(dtype) - y_data_list.append(val + 1) - instance["float_val"] = {"b64": base64.b64encode(val.tobytes()).decode(), 'type': dtype_str_map[dtype], - 'shape': [i + 1, i + 1]} - - result = post_restful("localhost", 5500, base.servable_name, "float_plus_1", instances) - assert "float16 reply is not supported" in result["error_msg"] - - @serving_test def test_restful_base64_without_b64_key_failed(): base = start_bool_int_float_restful_server()