diff --git a/core/cube/cube-api/src/cube_cli.cpp b/core/cube/cube-api/src/cube_cli.cpp index 7f45e436a821af3db06d4a90fe4cc8cc4a4936b6..eee4b0c31ad83ca69d242e81bae3ce4ecfb5bf1a 100644 --- a/core/cube/cube-api/src/cube_cli.cpp +++ b/core/cube/cube-api/src/cube_cli.cpp @@ -217,6 +217,7 @@ int run_m(int argc, char** argv) { LOG(INFO) << " total_request = " << std::to_string(request_num) << " speed = " << std::to_string(1000000 * thread_num / mean_time) // mean_time us << " query per second"; + return 0; } } // namespace mcube diff --git a/core/general-client/include/general_model.h b/core/general-client/include/general_model.h index 703593f8333c75386ab8a16fc1cf71b803158d9e..9d0cc8b66cfe2e3fe2f4d012c7920f518d32ef5a 100644 --- a/core/general-client/include/general_model.h +++ b/core/general-client/include/general_model.h @@ -21,6 +21,7 @@ #include #include #include +#include // move #include #include "core/sdk-cpp/builtin_format.pb.h" @@ -39,12 +40,32 @@ namespace baidu { namespace paddle_serving { namespace general_model { -class PredictorRes { - public: - PredictorRes() {} - ~PredictorRes() {} - +class ModelRes { public: + ModelRes() {} + ModelRes(const ModelRes& res) { + _engine_name = res._engine_name; + _int64_value_map.insert(res._int64_value_map.begin(), + res._int64_value_map.end()); + _float_value_map.insert(res._float_value_map.begin(), + res._float_value_map.end()); + _shape_map.insert(res._shape_map.begin(), res._shape_map.end()); + _lod_map.insert(res._lod_map.begin(), res._lod_map.end()); + } + ModelRes(ModelRes&& res) { + _engine_name = std::move(res._engine_name); + _int64_value_map.insert( + std::make_move_iterator(std::begin(res._int64_value_map)), + std::make_move_iterator(std::end(res._int64_value_map))); + _float_value_map.insert( + std::make_move_iterator(std::begin(res._float_value_map)), + std::make_move_iterator(std::end(res._float_value_map))); + _shape_map.insert(std::make_move_iterator(std::begin(res._shape_map)), + std::make_move_iterator(std::end(res._shape_map))); + _lod_map.insert(std::make_move_iterator(std::begin(res._lod_map)), + std::make_move_iterator(std::end(res._lod_map))); + } + ~ModelRes() {} const std::vector& get_int64_by_name(const std::string& name) { return _int64_value_map[name]; } @@ -57,19 +78,75 @@ class PredictorRes { const std::vector& get_lod(const std::string& name) { return _lod_map[name]; } - void set_variant_tag(const std::string& variant_tag) { - _variant_tag = variant_tag; + void set_engine_name(const std::string& engine_name) { + _engine_name = engine_name; + } + const std::string& engine_name() { return _engine_name; } + ModelRes& operator=(ModelRes&& res) { + if (this != &res) { + _engine_name = std::move(res._engine_name); + _int64_value_map.insert( + std::make_move_iterator(std::begin(res._int64_value_map)), + std::make_move_iterator(std::end(res._int64_value_map))); + _float_value_map.insert( + std::make_move_iterator(std::begin(res._float_value_map)), + std::make_move_iterator(std::end(res._float_value_map))); + _shape_map.insert(std::make_move_iterator(std::begin(res._shape_map)), + std::make_move_iterator(std::end(res._shape_map))); + _lod_map.insert(std::make_move_iterator(std::begin(res._lod_map)), + std::make_move_iterator(std::end(res._lod_map))); + } + return *this; } - const std::string& variant_tag() { return _variant_tag; } public: + std::string _engine_name; std::map> _int64_value_map; std::map> _float_value_map; std::map> _shape_map; std::map> _lod_map; +}; + +class PredictorRes { + public: + PredictorRes() {} + ~PredictorRes() {} + + public: + void clear() { + _models.clear(); + _engine_names.clear(); + } + const std::vector& get_int64_by_name(const int model_idx, + const std::string& name) { + return _models[model_idx].get_int64_by_name(name); + } + const std::vector& get_float_by_name(const int model_idx, + const std::string& name) { + return _models[model_idx].get_float_by_name(name); + } + const std::vector& get_shape(const int model_idx, + const std::string& name) { + return _models[model_idx].get_shape(name); + } + const std::vector& get_lod(const int model_idx, + const std::string& name) { + return _models[model_idx].get_lod(name); + } + void add_model_res(ModelRes&& res) { + _engine_names.push_back(res.engine_name()); + _models.emplace_back(std::move(res)); + } + void set_variant_tag(const std::string& variant_tag) { + _variant_tag = variant_tag; + } + const std::string& variant_tag() { return _variant_tag; } + const std::vector& get_engine_names() { return _engine_names; } private: + std::vector _models; std::string _variant_tag; + std::vector _engine_names; }; class PredictorClient { diff --git a/core/general-client/src/general_model.cpp b/core/general-client/src/general_model.cpp index 0270caf91441745ab02d460de8c6396462a3a3fe..65fa6587ecb68f18b72a03c7f54433252ea1608a 100644 --- a/core/general-client/src/general_model.cpp +++ b/core/general-client/src/general_model.cpp @@ -111,6 +111,7 @@ void PredictorClient::set_predictor_conf(const std::string &conf_path, int PredictorClient::destroy_predictor() { _api.thrd_finalize(); _api.destroy(); + return 0; } int PredictorClient::create_predictor_by_desc(const std::string &sdk_desc) { @@ -119,6 +120,7 @@ int PredictorClient::create_predictor_by_desc(const std::string &sdk_desc) { return -1; } _api.thrd_initialize(); + return 0; } int PredictorClient::create_predictor() { @@ -129,6 +131,7 @@ int PredictorClient::create_predictor() { return -1; } _api.thrd_initialize(); + return 0; } int PredictorClient::batch_predict( @@ -143,10 +146,7 @@ int PredictorClient::batch_predict( const int &pid) { int batch_size = std::max(float_feed_batch.size(), int_feed_batch.size()); - predict_res_batch._int64_value_map.clear(); - predict_res_batch._float_value_map.clear(); - predict_res_batch._shape_map.clear(); - predict_res_batch._lod_map.clear(); + predict_res_batch.clear(); Timer timeline; int64_t preprocess_start = timeline.TimeStampUS(); @@ -189,11 +189,11 @@ int PredictorClient::batch_predict( Tensor *tensor = tensor_vec[idx]; VLOG(2) << "prepare float feed " << name << " shape size " << float_shape[vec_idx].size(); - for (int j = 0; j < float_shape[vec_idx].size(); ++j) { + for (uint32_t j = 0; j < float_shape[vec_idx].size(); ++j) { tensor->add_shape(float_shape[vec_idx][j]); } tensor->set_elem_type(1); - for (int j = 0; j < float_feed[vec_idx].size(); ++j) { + for (uint32_t j = 0; j < float_feed[vec_idx].size(); ++j) { tensor->add_float_data(float_feed[vec_idx][j]); } vec_idx++; @@ -208,13 +208,13 @@ int PredictorClient::batch_predict( Tensor *tensor = tensor_vec[idx]; VLOG(2) << "prepare int feed " << name << " shape size " << int_shape[vec_idx].size(); - for (int j = 0; j < int_shape[vec_idx].size(); ++j) { + for (uint32_t j = 0; j < int_shape[vec_idx].size(); ++j) { tensor->add_shape(int_shape[vec_idx][j]); } tensor->set_elem_type(0); VLOG(3) << "feed var name " << name << " index " << vec_idx << "first data " << int_feed[vec_idx][0]; - for (int j = 0; j < int_feed[vec_idx].size(); ++j) { + for (uint32_t j = 0; j < int_feed[vec_idx].size(); ++j) { tensor->add_int64_data(int_feed[vec_idx][j]); } vec_idx++; @@ -248,51 +248,59 @@ int PredictorClient::batch_predict( client_infer_end = timeline.TimeStampUS(); postprocess_start = client_infer_end; - for (auto &name : fetch_name) { - // int idx = _fetch_name_to_idx[name]; - int idx = 0; - int shape_size = res.insts(0).tensor_array(idx).shape_size(); - VLOG(2) << "fetch var " << name << " index " << idx << " shape size " - << shape_size; - predict_res_batch._shape_map[name].resize(shape_size); - for (int i = 0; i < shape_size; ++i) { - predict_res_batch._shape_map[name][i] = - res.insts(0).tensor_array(idx).shape(i); - } - int lod_size = res.insts(0).tensor_array(idx).lod_size(); - if (lod_size > 0) { - predict_res_batch._lod_map[name].resize(lod_size); - for (int i = 0; i < lod_size; ++i) { - predict_res_batch._lod_map[name][i] = - res.insts(0).tensor_array(idx).lod(i); + uint32_t model_num = res.outputs_size(); + for (uint32_t m_idx = 0; m_idx < model_num; ++m_idx) { + VLOG(2) << "process model output index: " << m_idx; + auto output = res.outputs(m_idx); + ModelRes model; + model.set_engine_name(output.engine_name()); + + for (auto &name : fetch_name) { + // int idx = _fetch_name_to_idx[name]; + int idx = 0; + int shape_size = output.insts(0).tensor_array(idx).shape_size(); + VLOG(2) << "fetch var " << name << " index " << idx << " shape size " + << shape_size; + model._shape_map[name].resize(shape_size); + for (int i = 0; i < shape_size; ++i) { + model._shape_map[name][i] = + output.insts(0).tensor_array(idx).shape(i); } + int lod_size = output.insts(0).tensor_array(idx).lod_size(); + if (lod_size > 0) { + model._lod_map[name].resize(lod_size); + for (int i = 0; i < lod_size; ++i) { + model._lod_map[name][i] = output.insts(0).tensor_array(idx).lod(i); + } + } + idx += 1; } - idx += 1; - } - for (auto &name : fetch_name) { - // int idx = _fetch_name_to_idx[name]; - int idx = 0; - if (_fetch_name_to_type[name] == 0) { - VLOG(2) << "ferch var " << name << "type int"; - predict_res_batch._int64_value_map[name].resize( - res.insts(0).tensor_array(idx).int64_data_size()); - int size = res.insts(0).tensor_array(idx).int64_data_size(); - for (int i = 0; i < size; ++i) { - predict_res_batch._int64_value_map[name][i] = - res.insts(0).tensor_array(idx).int64_data(i); - } - } else { - VLOG(2) << "fetch var " << name << "type float"; - predict_res_batch._float_value_map[name].resize( - res.insts(0).tensor_array(idx).float_data_size()); - int size = res.insts(0).tensor_array(idx).float_data_size(); - for (int i = 0; i < size; ++i) { - predict_res_batch._float_value_map[name][i] = - res.insts(0).tensor_array(idx).float_data(i); + for (auto &name : fetch_name) { + // int idx = _fetch_name_to_idx[name]; + int idx = 0; + if (_fetch_name_to_type[name] == 0) { + VLOG(2) << "ferch var " << name << "type int"; + model._int64_value_map[name].resize( + output.insts(0).tensor_array(idx).int64_data_size()); + int size = output.insts(0).tensor_array(idx).int64_data_size(); + for (int i = 0; i < size; ++i) { + model._int64_value_map[name][i] = + output.insts(0).tensor_array(idx).int64_data(i); + } + } else { + VLOG(2) << "fetch var " << name << "type float"; + model._float_value_map[name].resize( + output.insts(0).tensor_array(idx).float_data_size()); + int size = output.insts(0).tensor_array(idx).float_data_size(); + for (int i = 0; i < size; ++i) { + model._float_value_map[name][i] = + output.insts(0).tensor_array(idx).float_data(i); + } } + idx += 1; } - idx += 1; + predict_res_batch.add_model_res(std::move(model)); } postprocess_end = timeline.TimeStampUS(); } @@ -305,7 +313,6 @@ int PredictorClient::batch_predict( << "prepro_1:" << preprocess_end << " " << "client_infer_0:" << client_infer_start << " " << "client_infer_1:" << client_infer_end << " "; - if (FLAGS_profile_server) { int op_num = res.profile_time_size() / 2; for (int i = 0; i < op_num; ++i) { diff --git a/core/general-client/src/pybind_general_model.cpp b/core/general-client/src/pybind_general_model.cpp index f30a3a859cb562e1f477defaeb23b5b8c7866cb0..066a2cfbe7af64807d4be1982a8822f93a6c32ec 100644 --- a/core/general-client/src/pybind_general_model.cpp +++ b/core/general-client/src/pybind_general_model.cpp @@ -31,27 +31,28 @@ PYBIND11_MODULE(serving_client, m) { py::class_(m, "PredictorRes", py::buffer_protocol()) .def(py::init()) .def("get_int64_by_name", - [](PredictorRes &self, std::string &name) { - return self.get_int64_by_name(name); + [](PredictorRes &self, int model_idx, std::string &name) { + return self.get_int64_by_name(model_idx, name); }, py::return_value_policy::reference) .def("get_float_by_name", - [](PredictorRes &self, std::string &name) { - return self.get_float_by_name(name); + [](PredictorRes &self, int model_idx, std::string &name) { + return self.get_float_by_name(model_idx, name); }, py::return_value_policy::reference) .def("get_shape", - [](PredictorRes &self, std::string &name) { - return self.get_shape(name); + [](PredictorRes &self, int model_idx, std::string &name) { + return self.get_shape(model_idx, name); }, py::return_value_policy::reference) .def("get_lod", - [](PredictorRes &self, std::string &name) { - return self.get_lod(name); + [](PredictorRes &self, int model_idx, std::string &name) { + return self.get_lod(model_idx, name); }, py::return_value_policy::reference) - .def("variant_tag", - [](PredictorRes &self) { return self.variant_tag(); }); + .def("variant_tag", [](PredictorRes &self) { return self.variant_tag(); }) + .def("get_engine_names", + [](PredictorRes &self) { return self.get_engine_names(); }); py::class_(m, "PredictorClient", py::buffer_protocol()) .def(py::init()) diff --git a/core/general-server/op/general_copy_op.cpp b/core/general-server/op/general_copy_op.cpp index a7f7d2904343bb9203fada6872994ed8916346bd..322bcc07795f1b053847991eae17cb3922dd7a7b 100644 --- a/core/general-server/op/general_copy_op.cpp +++ b/core/general-server/op/general_copy_op.cpp @@ -35,8 +35,17 @@ using baidu::paddle_serving::predictor::PaddleGeneralModelConfig; int GeneralCopyOp::inference() { // reade request from client - const GeneralBlob *input_blob = get_depend_argument(pre_name()); - VLOG(2) << "precedent name: " << pre_name(); + const std::vector pre_node_names = pre_names(); + if (pre_node_names.size() != 1) { + LOG(ERROR) << "This op(" << op_name() + << ") can only have one predecessor op, but received " + << pre_node_names.size(); + return -1; + } + const std::string pre_name = pre_node_names[0]; + + const GeneralBlob *input_blob = get_depend_argument(pre_name); + VLOG(2) << "precedent name: " << pre_name; const TensorVector *in = &input_blob->tensor_vector; VLOG(2) << "input size: " << in->size(); int batch_size = input_blob->GetBatchSize(); diff --git a/core/general-server/op/general_dist_kv_infer_op.cpp b/core/general-server/op/general_dist_kv_infer_op.cpp old mode 100755 new mode 100644 index ac4e7bb23e9410aede4fd353099d3c90ce91bcd3..9c6c70352b5387fab95acd16cdf79aa2b46f6122 --- a/core/general-server/op/general_dist_kv_infer_op.cpp +++ b/core/general-server/op/general_dist_kv_infer_op.cpp @@ -40,12 +40,21 @@ using baidu::paddle_serving::predictor::PaddleGeneralModelConfig; int GeneralDistKVInferOp::inference() { VLOG(2) << "Going to run inference"; - const GeneralBlob *input_blob = get_depend_argument(pre_name()); - VLOG(2) << "Get precedent op name: " << pre_name(); + const std::vector pre_node_names = pre_names(); + if (pre_node_names.size() != 1) { + LOG(ERROR) << "This op(" << op_name() + << ") can only have one predecessor op, but received " + << pre_node_names.size(); + return -1; + } + const std::string pre_name = pre_node_names[0]; + + const GeneralBlob *input_blob = get_depend_argument(pre_name); + VLOG(2) << "Get precedent op name: " << pre_name; GeneralBlob *output_blob = mutable_data(); if (!input_blob) { - LOG(ERROR) << "Failed mutable depended argument, op:" << pre_name(); + LOG(ERROR) << "Failed mutable depended argument, op:" << pre_name; return -1; } @@ -149,8 +158,8 @@ int GeneralDistKVInferOp::inference() { timeline.Start(); if (InferManager::instance().infer( - GENERAL_MODEL_NAME, &infer_in, out, batch_size)) { - LOG(ERROR) << "Failed do infer in fluid model: " << GENERAL_MODEL_NAME; + engine_name().c_str(), &infer_in, out, batch_size)) { + LOG(ERROR) << "Failed do infer in fluid model: " << engine_name(); return -1; } diff --git a/core/general-server/op/general_dist_kv_quant_infer_op.cpp b/core/general-server/op/general_dist_kv_quant_infer_op.cpp index 583384b79ed5ec69d14cb31b7c8239c3f786c33d..8752e8a72085c946b097cecf62a0bdbf90d682c4 100644 --- a/core/general-server/op/general_dist_kv_quant_infer_op.cpp +++ b/core/general-server/op/general_dist_kv_quant_infer_op.cpp @@ -41,12 +41,21 @@ using baidu::paddle_serving::predictor::PaddleGeneralModelConfig; int GeneralDistKVQuantInferOp::inference() { VLOG(2) << "Going to run inference"; - const GeneralBlob *input_blob = get_depend_argument(pre_name()); - VLOG(2) << "Get precedent op name: " << pre_name(); + const std::vector pre_node_names = pre_names(); + if (pre_node_names.size() != 1) { + LOG(ERROR) << "This op(" << op_name() + << ") can only have one predecessor op, but received " + << pre_node_names.size(); + return -1; + } + const std::string pre_name = pre_node_names[0]; + + const GeneralBlob *input_blob = get_depend_argument(pre_name); + VLOG(2) << "Get precedent op name: " << pre_name; GeneralBlob *output_blob = mutable_data(); if (!input_blob) { - LOG(ERROR) << "Failed mutable depended argument, op:" << pre_name(); + LOG(ERROR) << "Failed mutable depended argument, op:" << pre_name; return -1; } @@ -180,8 +189,8 @@ int GeneralDistKVQuantInferOp::inference() { timeline.Start(); if (InferManager::instance().infer( - GENERAL_MODEL_NAME, &infer_in, out, batch_size)) { - LOG(ERROR) << "Failed do infer in fluid model: " << GENERAL_MODEL_NAME; + engine_name().c_str(), &infer_in, out, batch_size)) { + LOG(ERROR) << "Failed do infer in fluid model: " << engine_name(); return -1; } diff --git a/core/general-server/op/general_infer_helper.h b/core/general-server/op/general_infer_helper.h index b09ef6d55b8aef3dc54dd5ce2921c27eb6ad86c5..6a6b24329ed73cbb10e467f366ff04c4e2ac8031 100644 --- a/core/general-server/op/general_infer_helper.h +++ b/core/general-server/op/general_infer_helper.h @@ -31,8 +31,6 @@ namespace baidu { namespace paddle_serving { namespace serving { -static const char* GENERAL_MODEL_NAME = "general_model"; - struct GeneralBlob { std::vector tensor_vector; int64_t time_stamp[20]; diff --git a/core/general-server/op/general_infer_op.cpp b/core/general-server/op/general_infer_op.cpp index 6cec9d1cb4d87e13c566c24c90cef11f148749c8..a9ff2e7226b25842889e391d82217b3b6a140170 100644 --- a/core/general-server/op/general_infer_op.cpp +++ b/core/general-server/op/general_infer_op.cpp @@ -37,12 +37,21 @@ using baidu::paddle_serving::predictor::PaddleGeneralModelConfig; int GeneralInferOp::inference() { VLOG(2) << "Going to run inference"; - const GeneralBlob *input_blob = get_depend_argument(pre_name()); - VLOG(2) << "Get precedent op name: " << pre_name(); + const std::vector pre_node_names = pre_names(); + if (pre_node_names.size() != 1) { + LOG(ERROR) << "This op(" << op_name() + << ") can only have one predecessor op, but received " + << pre_node_names.size(); + return -1; + } + const std::string pre_name = pre_node_names[0]; + + const GeneralBlob *input_blob = get_depend_argument(pre_name); + VLOG(2) << "Get precedent op name: " << pre_name; GeneralBlob *output_blob = mutable_data(); if (!input_blob) { - LOG(ERROR) << "Failed mutable depended argument, op:" << pre_name(); + LOG(ERROR) << "Failed mutable depended argument, op:" << pre_name; return -1; } @@ -59,8 +68,9 @@ int GeneralInferOp::inference() { int64_t start = timeline.TimeStampUS(); timeline.Start(); - if (InferManager::instance().infer(GENERAL_MODEL_NAME, in, out, batch_size)) { - LOG(ERROR) << "Failed do infer in fluid model: " << GENERAL_MODEL_NAME; + if (InferManager::instance().infer( + engine_name().c_str(), in, out, batch_size)) { + LOG(ERROR) << "Failed do infer in fluid model: " << engine_name().c_str(); return -1; } diff --git a/core/general-server/op/general_response_op.cpp b/core/general-server/op/general_response_op.cpp index b84cb199001a864ba55de7fe278262c650e98c22..e00984f42b36d708825622b39fef63a449ba147e 100644 --- a/core/general-server/op/general_response_op.cpp +++ b/core/general-server/op/general_response_op.cpp @@ -33,23 +33,17 @@ using baidu::paddle_serving::predictor::general_model::Tensor; using baidu::paddle_serving::predictor::general_model::Response; using baidu::paddle_serving::predictor::general_model::Request; using baidu::paddle_serving::predictor::general_model::FetchInst; +using baidu::paddle_serving::predictor::general_model::ModelOutput; using baidu::paddle_serving::predictor::InferManager; using baidu::paddle_serving::predictor::PaddleGeneralModelConfig; int GeneralResponseOp::inference() { - const GeneralBlob *input_blob = get_depend_argument(pre_name()); - - if (!input_blob) { - LOG(ERROR) << "Failed mutable depended argument, op: " << pre_name(); - return -1; - } - - const TensorVector *in = &input_blob->tensor_vector; - int batch_size = input_blob->GetBatchSize(); - - VLOG(2) << "input batch size: " << batch_size; + const std::vector pre_node_names = pre_names(); + VLOG(2) << "pre node names size: " << pre_node_names.size(); const Request *req = dynamic_cast(get_request_message()); + // response inst with only fetch_var_names + Response *res = mutable_data(); Timer timeline; // double response_time = 0.0; @@ -73,77 +67,107 @@ int GeneralResponseOp::inference() { model_config->_fetch_alias_name_to_index[req->fetch_var_names(i)]; } - // response inst with only fetch_var_names - Response *res = mutable_data(); - FetchInst *fetch_inst = res->add_insts(); - for (auto &idx : fetch_index) { - Tensor *tensor = fetch_inst->add_tensor_array(); - tensor->set_elem_type(1); - if (model_config->_is_lod_fetch[idx]) { - VLOG(2) << "out[" << idx << "] is lod_tensor"; - for (int k = 0; k < in->at(idx).shape.size(); ++k) { - VLOG(2) << "shape[" << k << "]: " << in->at(idx).shape[k]; - tensor->add_shape(in->at(idx).shape[k]); - } - } else { - VLOG(2) << "out[" << idx << "] is tensor"; - for (int k = 0; k < in->at(idx).shape.size(); ++k) { - VLOG(2) << "shape[" << k << "]: " << in->at(idx).shape[k]; - tensor->add_shape(in->at(idx).shape[k]); - } + const GeneralBlob *input_blob; + for (uint32_t pi = 0; pi < pre_node_names.size(); ++pi) { + const std::string &pre_name = pre_node_names[pi]; + VLOG(2) << "pre names[" << pi << "]: " << pre_name << " (" + << pre_node_names.size() << ")"; + input_blob = get_depend_argument(pre_name); + // fprintf(stderr, "input(%s) blob address %x\n", pre_names.c_str(), + // input_blob); + if (!input_blob) { + LOG(ERROR) << "Failed mutable depended argument, op: " << pre_name; + return -1; } - } - int var_idx = 0; - for (auto &idx : fetch_index) { - int cap = 1; - for (int j = 0; j < in->at(idx).shape.size(); ++j) { - cap *= in->at(idx).shape[j]; - } - if (in->at(idx).dtype == paddle::PaddleDType::INT64) { - int64_t *data_ptr = static_cast(in->at(idx).data.data()); + const TensorVector *in = &input_blob->tensor_vector; + + ModelOutput *output = res->add_outputs(); + // To get the order of model return values + output->set_engine_name(pre_name); + FetchInst *fetch_inst = output->add_insts(); + for (auto &idx : fetch_index) { + Tensor *tensor = fetch_inst->add_tensor_array(); + tensor->set_elem_type(1); if (model_config->_is_lod_fetch[idx]) { - FetchInst *fetch_p = res->mutable_insts(0); - for (int j = 0; j < in->at(idx).lod[0].size(); ++j) { - fetch_p->mutable_tensor_array(var_idx)->add_lod( - in->at(idx).lod[0][j]); - } - for (int j = 0; j < cap; ++j) { - fetch_p->mutable_tensor_array(var_idx)->add_int64_data(data_ptr[j]); + VLOG(2) << "out[" << idx << "] is lod_tensor"; + for (int k = 0; k < in->at(idx).shape.size(); ++k) { + VLOG(2) << "shape[" << k << "]: " << in->at(idx).shape[k]; + tensor->add_shape(in->at(idx).shape[k]); } } else { - FetchInst *fetch_p = res->mutable_insts(0); - for (int j = 0; j < cap; ++j) { - fetch_p->mutable_tensor_array(var_idx)->add_float_data(data_ptr[j]); + VLOG(2) << "out[" << idx << "] is tensor"; + for (int k = 0; k < in->at(idx).shape.size(); ++k) { + VLOG(2) << "shape[" << k << "]: " << in->at(idx).shape[k]; + tensor->add_shape(in->at(idx).shape[k]); } } - var_idx++; - } else if (in->at(idx).dtype == paddle::PaddleDType::FLOAT32) { - float *data_ptr = static_cast(in->at(idx).data.data()); - if (model_config->_is_lod_fetch[idx]) { - FetchInst *fetch_p = res->mutable_insts(0); - for (int j = 0; j < in->at(idx).lod[0].size(); ++j) { - fetch_p->mutable_tensor_array(var_idx)->add_lod( - in->at(idx).lod[0][j]); - } - for (int j = 0; j < cap; ++j) { - fetch_p->mutable_tensor_array(var_idx)->add_float_data(data_ptr[j]); + } + + int var_idx = 0; + for (auto &idx : fetch_index) { + int cap = 1; + for (int j = 0; j < in->at(idx).shape.size(); ++j) { + cap *= in->at(idx).shape[j]; + } + if (in->at(idx).dtype == paddle::PaddleDType::INT64) { + int64_t *data_ptr = static_cast(in->at(idx).data.data()); + if (model_config->_is_lod_fetch[idx]) { + FetchInst *fetch_p = output->mutable_insts(0); + for (int j = 0; j < in->at(idx).lod[0].size(); ++j) { + fetch_p->mutable_tensor_array(var_idx)->add_lod( + in->at(idx).lod[0][j]); + } + for (int j = 0; j < cap; ++j) { + fetch_p->mutable_tensor_array(var_idx)->add_int64_data(data_ptr[j]); + } + } else { + FetchInst *fetch_p = output->mutable_insts(0); + for (int j = 0; j < cap; ++j) { + fetch_p->mutable_tensor_array(var_idx)->add_float_data(data_ptr[j]); + } } - } else { - FetchInst *fetch_p = res->mutable_insts(0); - for (int j = 0; j < cap; ++j) { - fetch_p->mutable_tensor_array(var_idx)->add_float_data(data_ptr[j]); + var_idx++; + } else if (in->at(idx).dtype == paddle::PaddleDType::FLOAT32) { + float *data_ptr = static_cast(in->at(idx).data.data()); + if (model_config->_is_lod_fetch[idx]) { + FetchInst *fetch_p = output->mutable_insts(0); + for (int j = 0; j < in->at(idx).lod[0].size(); ++j) { + fetch_p->mutable_tensor_array(var_idx)->add_lod( + in->at(idx).lod[0][j]); + } + for (int j = 0; j < cap; ++j) { + fetch_p->mutable_tensor_array(var_idx)->add_float_data(data_ptr[j]); + } + } else { + FetchInst *fetch_p = output->mutable_insts(0); + for (int j = 0; j < cap; ++j) { + fetch_p->mutable_tensor_array(var_idx)->add_float_data(data_ptr[j]); + } } + var_idx++; } - var_idx++; } } if (req->profile_server()) { int64_t end = timeline.TimeStampUS(); - VLOG(2) << "p size for input blob: " << input_blob->p_size; - for (int i = 0; i < input_blob->p_size; ++i) { - res->add_profile_time(input_blob->time_stamp[i]); + // TODO(barriery): multi-model profile_time. + // At present, only the response_op is multi-input, so here we get + // the profile_time by hard coding. It needs to be replaced with + // a more elegant way. + for (uint32_t pi = 0; pi < pre_node_names.size(); ++pi) { + input_blob = get_depend_argument(pre_node_names[pi]); + VLOG(2) << "p size for input blob: " << input_blob->p_size; + int profile_time_idx = -1; + if (pi == 0) { + profile_time_idx = 0; + } else { + profile_time_idx = input_blob->p_size - 2; + } + for (; profile_time_idx < input_blob->p_size; ++profile_time_idx) { + res->add_profile_time(input_blob->time_stamp[profile_time_idx]); + } } // TODO(guru4elephant): find more elegant way to do this res->add_profile_time(start); diff --git a/core/general-server/op/general_text_response_op.cpp b/core/general-server/op/general_text_response_op.cpp index 43c7af774fd939a8fa1ca14456285cc75dbd7f8d..ae194119f1fc3edad01662041035f7011873998a 100644 --- a/core/general-server/op/general_text_response_op.cpp +++ b/core/general-server/op/general_text_response_op.cpp @@ -32,22 +32,18 @@ using baidu::paddle_serving::predictor::general_model::Tensor; using baidu::paddle_serving::predictor::general_model::Response; using baidu::paddle_serving::predictor::general_model::Request; using baidu::paddle_serving::predictor::general_model::FetchInst; +using baidu::paddle_serving::predictor::general_model::ModelOutput; using baidu::paddle_serving::predictor::InferManager; using baidu::paddle_serving::predictor::PaddleGeneralModelConfig; int GeneralTextResponseOp::inference() { - const GeneralBlob *input_blob = get_depend_argument(pre_name()); + VLOG(2) << "Going to run inference"; + const std::vector pre_node_names = pre_names(); + VLOG(2) << "pre node names size: " << pre_node_names.size(); - if (!input_blob) { - LOG(ERROR) << "Failed mutable depended argument, op: " << pre_name(); - return -1; - } - - const TensorVector *in = &input_blob->tensor_vector; - int batch_size = input_blob->GetBatchSize(); - - VLOG(2) << "infer batch size: " << batch_size; const Request *req = dynamic_cast(get_request_message()); + // response inst with only fetch_var_names + Response *res = mutable_data(); Timer timeline; int64_t start = timeline.TimeStampUS(); @@ -67,59 +63,90 @@ int GeneralTextResponseOp::inference() { model_config->_fetch_alias_name_to_index[req->fetch_var_names(i)]; } - // response inst with only fetch_var_names - Response *res = mutable_data(); + const GeneralBlob *input_blob; + for (uint32_t pi = 0; pi < pre_node_names.size(); ++pi) { + const std::string &pre_name = pre_node_names[pi]; + VLOG(2) << "pre names[" << pi << "]: " << pre_name << " (" + << pre_node_names.size() << ")"; + input_blob = get_depend_argument(pre_name); + if (!input_blob) { + LOG(ERROR) << "Failed mutable depended argument, op: " << pre_name; + return -1; + } - for (int i = 0; i < batch_size; ++i) { - FetchInst *fetch_inst = res->add_insts(); - for (auto &idx : fetch_index) { - Tensor *tensor = fetch_inst->add_tensor_array(); - // currently only response float tensor or lod_tensor - tensor->set_elem_type(1); - if (model_config->_is_lod_fetch[idx]) { - VLOG(2) << "out[" << idx << " is lod_tensor"; - tensor->add_shape(-1); - } else { - VLOG(2) << "out[" << idx << "] is tensor"; - for (int k = 1; k < in->at(idx).shape.size(); ++k) { - VLOG(2) << "shape[" << k - 1 << "]: " << in->at(idx).shape[k]; - tensor->add_shape(in->at(idx).shape[k]); + const TensorVector *in = &input_blob->tensor_vector; + int batch_size = input_blob->GetBatchSize(); + VLOG(2) << "input batch size: " << batch_size; + + ModelOutput *output = res->add_outputs(); + output->set_engine_name( + pre_name); // To get the order of model return values + for (int i = 0; i < batch_size; ++i) { + FetchInst *fetch_inst = output->add_insts(); + for (auto &idx : fetch_index) { + Tensor *tensor = fetch_inst->add_tensor_array(); + // currently only response float tensor or lod_tensor + tensor->set_elem_type(1); + if (model_config->_is_lod_fetch[idx]) { + VLOG(2) << "out[" << idx << " is lod_tensor"; + tensor->add_shape(-1); + } else { + VLOG(2) << "out[" << idx << "] is tensor"; + for (int k = 1; k < in->at(idx).shape.size(); ++k) { + VLOG(2) << "shape[" << k - 1 << "]: " << in->at(idx).shape[k]; + tensor->add_shape(in->at(idx).shape[k]); + } } } } - } - int var_idx = 0; - for (auto &idx : fetch_index) { - float *data_ptr = static_cast(in->at(idx).data.data()); - int cap = 1; - for (int j = 1; j < in->at(idx).shape.size(); ++j) { - cap *= in->at(idx).shape[j]; - } - if (model_config->_is_lod_fetch[idx]) { - for (int j = 0; j < batch_size; ++j) { - for (int k = in->at(idx).lod[0][j]; k < in->at(idx).lod[0][j + 1]; - k++) { - res->mutable_insts(j)->mutable_tensor_array(var_idx)->add_float_data( - data_ptr[k]); - } + int var_idx = 0; + for (auto &idx : fetch_index) { + float *data_ptr = static_cast(in->at(idx).data.data()); + int cap = 1; + for (int j = 1; j < in->at(idx).shape.size(); ++j) { + cap *= in->at(idx).shape[j]; } - } else { - for (int j = 0; j < batch_size; ++j) { - for (int k = j * cap; k < (j + 1) * cap; ++k) { - res->mutable_insts(j)->mutable_tensor_array(var_idx)->add_float_data( - data_ptr[k]); + if (model_config->_is_lod_fetch[idx]) { + for (int j = 0; j < batch_size; ++j) { + for (int k = in->at(idx).lod[0][j]; k < in->at(idx).lod[0][j + 1]; + k++) { + output->mutable_insts(j) + ->mutable_tensor_array(var_idx) + ->add_float_data(data_ptr[k]); + } + } + } else { + for (int j = 0; j < batch_size; ++j) { + for (int k = j * cap; k < (j + 1) * cap; ++k) { + output->mutable_insts(j) + ->mutable_tensor_array(var_idx) + ->add_float_data(data_ptr[k]); + } } } + var_idx++; } - var_idx++; } if (req->profile_server()) { int64_t end = timeline.TimeStampUS(); - - for (int i = 0; i < input_blob->p_size; ++i) { - res->add_profile_time(input_blob->time_stamp[i]); + // TODO(barriery): multi-model profile_time. + // At present, only the response_op is multi-input, so here we get + // the profile_time by hard coding. It needs to be replaced with + // a more elegant way. + for (uint32_t pi = 0; pi < pre_node_names.size(); ++pi) { + input_blob = get_depend_argument(pre_node_names[pi]); + VLOG(2) << "p size for input blob: " << input_blob->p_size; + int profile_time_idx = -1; + if (pi == 0) { + profile_time_idx = 0; + } else { + profile_time_idx = input_blob->p_size - 2; + } + for (; profile_time_idx < input_blob->p_size; ++profile_time_idx) { + res->add_profile_time(input_blob->time_stamp[profile_time_idx]); + } } // TODO(guru4elephant): find more elegant way to do this res->add_profile_time(start); diff --git a/core/general-server/proto/general_model_service.proto b/core/general-server/proto/general_model_service.proto index ad1128c373491ac82e3d7042f0693aa574ac92c5..8581ecb2a2e10deced910a20ce26c2beaca956fa 100644 --- a/core/general-server/proto/general_model_service.proto +++ b/core/general-server/proto/general_model_service.proto @@ -40,10 +40,15 @@ message Request { }; message Response { - repeated FetchInst insts = 1; + repeated ModelOutput outputs = 1; repeated int64 profile_time = 2; }; +message ModelOutput { + repeated FetchInst insts = 1; + optional string engine_name = 2; +} + service GeneralModelService { rpc inference(Request) returns (Response); rpc debug(Request) returns (Response); diff --git a/core/predictor/common/macros.h b/core/predictor/common/macros.h index d25c530fd40b85d0893fbabf572197032cc55a24..fa4a068668cb1a37c37a2726634c24be26a3fb40 100644 --- a/core/predictor/common/macros.h +++ b/core/predictor/common/macros.h @@ -27,6 +27,10 @@ namespace predictor { } #endif +#ifdef WITH_GPU +#define USE_PTHREAD +#endif + #ifdef USE_PTHREAD #define THREAD_T pthread_t diff --git a/core/predictor/framework/dag.cpp b/core/predictor/framework/dag.cpp index 9aea595e1281ccac9367f89acde3f8b19b98cc5e..f039ac70ffe2e55a59f926d754ca411a034058f4 100644 --- a/core/predictor/framework/dag.cpp +++ b/core/predictor/framework/dag.cpp @@ -14,6 +14,7 @@ #include "core/predictor/framework/dag.h" #include +#include // make_pair #include #include "core/predictor/common/inner_common.h" #include "core/predictor/framework/predictor_metric.h" // PredictorMetric @@ -199,25 +200,81 @@ const DagStage* Dag::stage_by_index(uint32_t index) { return _stages[index]; } int Dag::topo_sort() { std::stringstream ss; - for (uint32_t nid = 0; nid < _index_nodes.size(); nid++) { - DagStage* stage = new (std::nothrow) DagStage(); - if (stage == NULL) { - LOG(ERROR) << "Invalid stage!"; - return ERR_MEM_ALLOC_FAILURE; + uint32_t nodes_size = _index_nodes.size(); + std::vector in_degree(nodes_size, 0); + std::vector> in_egde(nodes_size); + for (uint32_t nid = 0; nid < nodes_size; nid++) { + in_degree[nid] += _index_nodes[nid]->depends.size(); + for (auto it = _index_nodes[nid]->depends.begin(); + it != _index_nodes[nid]->depends.end(); + ++it) { + uint32_t pnid = Dag::node_by_name(it->first)->id - + 1; // 0 is reserved for begginer-op + in_egde[pnid].push_back(nid); + } + } + for (int i = 0; i < in_degree.size(); ++i) { + LOG(INFO) << "(" << _index_nodes[i]->name << ") in_degree[" << i + << "]: " << in_degree[i]; + } + int sorted_num = 0; + DagStage* stage = new (std::nothrow) DagStage(); + if (stage == NULL) { + LOG(ERROR) << "Invalid stage!"; + return ERR_MEM_ALLOC_FAILURE; + } + ss.str(""); + ss << _stages.size(); + stage->name = ss.str(); + stage->full_name = full_name() + NAME_DELIMITER + stage->name; + for (uint32_t nid = 0; nid < nodes_size; ++nid) { + if (in_degree[nid] == 0) { + ++sorted_num; + stage->nodes.push_back(_index_nodes[nid]); + // assign stage number after stage created + _index_nodes[nid]->stage = _stages.size(); + // assign dag node full name after stage created + _index_nodes[nid]->full_name = + stage->full_name + NAME_DELIMITER + _index_nodes[nid]->name; } - stage->nodes.push_back(_index_nodes[nid]); + } + + if (stage->nodes.size() == 0) { + LOG(ERROR) << "Invalid Dag!"; + return ERR_INTERNAL_FAILURE; + } + _stages.push_back(stage); + + while (sorted_num < nodes_size) { + auto pre_nodes = _stages.back()->nodes; + DagStage* stage = new (std::nothrow) DagStage(); ss.str(""); ss << _stages.size(); stage->name = ss.str(); stage->full_name = full_name() + NAME_DELIMITER + stage->name; + for (uint32_t pi = 0; pi < pre_nodes.size(); ++pi) { + uint32_t pnid = pre_nodes[pi]->id - 1; + for (uint32_t ei = 0; ei < in_egde[pnid].size(); ++ei) { + uint32_t nid = in_egde[pnid][ei]; + --in_degree[nid]; + if (in_degree[nid] == 0) { + ++sorted_num; + stage->nodes.push_back(_index_nodes[nid]); + // assign stage number after stage created + _index_nodes[nid]->stage = _stages.size(); + // assign dag node full name after stage created + _index_nodes[nid]->full_name = + stage->full_name + NAME_DELIMITER + _index_nodes[nid]->name; + } + } + } + if (stage->nodes.size() == 0) { + LOG(ERROR) << "Invalid Dag!"; + return ERR_INTERNAL_FAILURE; + } _stages.push_back(stage); - - // assign stage number after stage created - _index_nodes[nid]->stage = nid; - // assign dag node full name after stage created - _index_nodes[nid]->full_name = - stage->full_name + NAME_DELIMITER + _index_nodes[nid]->name; } + return ERR_OK; } diff --git a/core/predictor/framework/dag_view.cpp b/core/predictor/framework/dag_view.cpp index 743e73418b535776e651bd1500509ffcad0e0618..bde8084b41fee00bc95d2a35444a15258d2a12a8 100644 --- a/core/predictor/framework/dag_view.cpp +++ b/core/predictor/framework/dag_view.cpp @@ -76,19 +76,34 @@ int DagView::init(Dag* dag, const std::string& service_name) { } op->set_full_name(service_name + NAME_DELIMITER + node->full_name); + + // Set the name of the Op as the key of the matching engine. + VLOG(2) << "op->set_engine_name(" << node->name.c_str() << ")"; + op->set_engine_name(node->name); + vnode->conf = node; vnode->op = op; + // Add depends + for (auto it = vnode->conf->depends.begin(); + it != vnode->conf->depends.end(); + ++it) { + std::string pre_node_name = it->first; + VLOG(2) << "add op pre name: \n" + << "current op name: " << vnode->op->op_name() + << ", previous op name: " << pre_node_name; + vnode->op->add_pre_node_name(pre_node_name); + } vstage->nodes.push_back(vnode); } // TODO(guru4elephant): this seems buggy, please review later - if (si > 0) { - VLOG(2) << "set op pre name: \n" - << "current op name: " << vstage->nodes.back()->op->op_name() - << " previous op name: " - << _view[si - 1]->nodes.back()->op->op_name(); - vstage->nodes.back()->op->set_pre_node_name( - _view[si - 1]->nodes.back()->op->op_name()); - } + /*if (si > 0) {*/ + // VLOG(2) << "set op pre name: \n" + //<< "current op name: " << vstage->nodes.back()->op->op_name() + //<< " previous op name: " + //<< _view[si - 1]->nodes.back()->op->op_name(); + // vstage->nodes.back()->op->set_pre_node_name( + //_view[si - 1]->nodes.back()->op->op_name()); + /*}*/ _view.push_back(vstage); } @@ -139,6 +154,7 @@ int DagView::execute_one_stage(ViewStage* vstage, butil::IOBufBuilder* debug_os) { butil::Timer stage_time(butil::Timer::STARTED); uint32_t node_size = vstage->nodes.size(); + VLOG(2) << "vstage->nodes.size(): " << node_size; for (uint32_t ni = 0; ni < node_size; ni++) { ViewNode* vnode = vstage->nodes[ni]; DagNode* conf = vnode->conf; diff --git a/core/predictor/framework/infer.h b/core/predictor/framework/infer.h index f8bd7843706187c13d8f38c7d33c87b51369e5a0..4bb3be9ad2c3dc7ef94a32200b014325aceedf45 100644 --- a/core/predictor/framework/infer.h +++ b/core/predictor/framework/infer.h @@ -765,6 +765,8 @@ class InferManager { } size_t engine_num = model_toolkit_conf.engines_size(); for (size_t ei = 0; ei < engine_num; ++ei) { + LOG(INFO) << "model_toolkit_conf.engines(" << ei + << ").name: " << model_toolkit_conf.engines(ei).name(); std::string engine_name = model_toolkit_conf.engines(ei).name(); VersionedInferEngine* engine = new (std::nothrow) VersionedInferEngine(); if (!engine) { diff --git a/core/predictor/framework/memory.cpp b/core/predictor/framework/memory.cpp index ec5967c85630c549fdc19eb6ce1773bfd541a05c..9e7f028e80d42ad855de006131540fece845afe3 100644 --- a/core/predictor/framework/memory.cpp +++ b/core/predictor/framework/memory.cpp @@ -56,11 +56,11 @@ int MempoolWrapper::thread_initialize() { im::fugue::memory::Region* region = new im::fugue::memory::Region(); region->init(); im::Mempool* mempool = new (std::nothrow) im::Mempool(region); - MempoolRegion* mempool_region = new MempoolRegion(region, mempool); if (mempool == NULL) { LOG(ERROR) << "Failed create thread mempool"; return -1; } + MempoolRegion* mempool_region = new MempoolRegion(region, mempool); if (THREAD_SETSPECIFIC(_bspec_key, mempool_region) != 0) { LOG(ERROR) << "unable to set the thrd_data"; diff --git a/core/predictor/op/op.cpp b/core/predictor/op/op.cpp index d2e512eb69af0b70cbe07b5bc75c3acb88fea918..59ef6aed71977a3f762ff4fbe9480db19cb4057e 100644 --- a/core/predictor/op/op.cpp +++ b/core/predictor/op/op.cpp @@ -60,6 +60,7 @@ int Op::init(Bus* bus, return -1; } + _pre_node_names.clear(); return custom_init(); } diff --git a/core/predictor/op/op.h b/core/predictor/op/op.h index 84bcf44575826a6ab00e037ce57e119ffbe4f3f3..ae52975fe6f2506fb0bf483318f607df137c8a96 100644 --- a/core/predictor/op/op.h +++ b/core/predictor/op/op.h @@ -14,7 +14,9 @@ #pragma once #include // bvar::LatencyRecorder +#include #include +#include #include "core/predictor/common/inner_common.h" #include "core/predictor/framework/channel.h" #include "core/predictor/framework/op_repository.h" @@ -132,18 +134,28 @@ class Op { const std::string& full_name() const { return _full_name; } - const std::string& pre_name() const { return _pre_node_name; } + const std::vector& pre_names() const { return _pre_node_names; } void set_full_name(const std::string full_name) { _full_name = full_name; } - void set_pre_node_name(const std::string pre_name) { - _pre_node_name = pre_name; + void add_pre_node_name(const std::string pre_name) { + _pre_node_names.push_back(pre_name); } const std::string& type() const; uint32_t id() const; + // Set the name of the Op as the key of the matching engine. + // Notes that this key is only used by infer_op (only the + // infer_op needs to find the corresponding engine). + // At present, there is only general_infer_op. + void set_engine_name(const std::string engine_name) { + _engine_name = engine_name; + } + + const std::string& engine_name() const { return _engine_name; } + // --------------- Default implements ---------------- virtual int custom_init() { return 0; } @@ -189,13 +201,14 @@ class Op { Bus* _bus; Dag* _dag; uint32_t _id; - std::string _pre_node_name; // only for sequential execution + std::vector _pre_node_names; // for DAG execution std::string _name; std::string _full_name; // service_workflow_stageindex_opname std::string _type; bool _has_calc; bool _has_init; TimerFlow* _timer; + std::string _engine_name; // only for infer_op }; template @@ -215,7 +228,10 @@ class OpWithChannel : public Op { return _channel; } - _channel = butil::get_object(); + // TODO(barriery): There are some problems in using butil::get_object + // _channel = butil::get_object(); + _channel = new ChannelType(); + if (!_channel) { LOG(ERROR) << "Failed mutable channel of type:" << typeid(T).name(); return NULL; @@ -229,8 +245,14 @@ class OpWithChannel : public Op { int release_channel() { if (_channel) { _channel->deinit(); - butil::return_object(_channel); + delete _channel; } + // TODO(barriery): There are some problems in using butil::get_object + /* + if (_channel) { + _channel->deinit(); + butil::return_object(_channel); + } */ _channel = NULL; return 0; diff --git a/core/sdk-cpp/proto/general_model_service.proto b/core/sdk-cpp/proto/general_model_service.proto index 39c09f09f1fa56fcc8572dde9981b48b76214917..51c0335a9db896e1260e83915de81e51451a904b 100644 --- a/core/sdk-cpp/proto/general_model_service.proto +++ b/core/sdk-cpp/proto/general_model_service.proto @@ -40,10 +40,15 @@ message Request { }; message Response { - repeated FetchInst insts = 1; + repeated ModelOutput outputs = 1; repeated int64 profile_time = 2; }; +message ModelOutput { + repeated FetchInst insts = 1; + optional string engine_name = 2; +} + service GeneralModelService { rpc inference(Request) returns (Response); rpc debug(Request) returns (Response); diff --git a/doc/INFERNCE_TO_SERVING.md b/doc/INFERNCE_TO_SERVING.md new file mode 100644 index 0000000000000000000000000000000000000000..8334159ea255ca65241a2b567e43682a148bb775 --- /dev/null +++ b/doc/INFERNCE_TO_SERVING.md @@ -0,0 +1,14 @@ +# How to Convert Paddle Inference Model To Paddle Serving Format + +([简体中文](./INFERENCE_TO_SERVING_CN.md)|English) + +## Example + +``` python +from paddle_serving_client.io import inference_model_to_serving +inference_model_dir = "your_inference_model" +serving_client_dir = "serving_client_dir" +serving_server_dir = "serving_server_dir" +feed_var_names, fetch_var_names = inference_model_to_serving( + inference_model_dir, serving_client_dir, serving_server_dir) +``` diff --git a/doc/INFERNCE_TO_SERVING_CN.md b/doc/INFERNCE_TO_SERVING_CN.md new file mode 100644 index 0000000000000000000000000000000000000000..94d1def424db467e200020c69fbd6d1599a5ffde --- /dev/null +++ b/doc/INFERNCE_TO_SERVING_CN.md @@ -0,0 +1,14 @@ +# 如何从Paddle保存的预测模型转为Paddle Serving格式可部署的模型 + +([English](./INFERENCE_TO_SERVING.md)|简体中文) + +## 示例 + +``` python +from paddle_serving_client.io import inference_model_to_serving +inference_model_dir = "your_inference_model" +serving_client_dir = "serving_client_dir" +serving_server_dir = "serving_server_dir" +feed_var_names, fetch_var_names = inference_model_to_serving( + inference_model_dir, serving_client_dir, serving_server_dir) +``` diff --git a/doc/MODEL_ENSEMBLE_IN_PADDLE_SERVING.md b/doc/MODEL_ENSEMBLE_IN_PADDLE_SERVING.md new file mode 100644 index 0000000000000000000000000000000000000000..7f583342cf2437b29916f6711c7bd0701206bf8d --- /dev/null +++ b/doc/MODEL_ENSEMBLE_IN_PADDLE_SERVING.md @@ -0,0 +1,121 @@ +# Model Ensemble in Paddle Serving + +([简体中文](MODEL_ENSEMBLE_IN_PADDLE_SERVING_CN.md)|English) + +In some scenarios, multiple models with the same input may be used to predict in parallel and integrate predicted results for better prediction effect. Paddle Serving also supports this feature. + +Next, we will take the text classification task as an example to show model ensemble in Paddle Serving (This feature is still serial prediction for the time being. We will support parallel prediction as soon as possible). + +## Simple example + +In this example (see the figure below), the server side predict the bow and CNN models with the same input in a service in parallel, The client side fetchs the prediction results of the two models, and processes the prediction results to get the final predict results. + +![simple example](model_ensemble_example.png) + +It should be noted that at present, only multiple models with the same format input and output in the same service are supported. In this example, the input and output formats of CNN and BOW model are the same. + +The code used in the example is saved in the `python/examples/imdb` path: + +```shell +. +├── get_data.sh +├── imdb_reader.py +├── test_ensemble_client.py +└── test_ensemble_server.py +``` + +### Prepare data + +Get the pre-trained CNN and BOW models by the following command (you can also run the `get_data.sh` script): + +```shell +wget --no-check-certificate https://fleet.bj.bcebos.com/text_classification_data.tar.gz +wget --no-check-certificate https://paddle-serving.bj.bcebos.com/imdb-demo/imdb_model.tar.gz +tar -zxvf text_classification_data.tar.gz +tar -zxvf imdb_model.tar.gz +``` + +### Start server + +Start server by the following Python code (you can also run the `test_ensemble_server.py` script): + +```python +from paddle_serving_server import OpMaker +from paddle_serving_server import OpGraphMaker +from paddle_serving_server import Server + +op_maker = OpMaker() +read_op = op_maker.create('general_reader') +cnn_infer_op = op_maker.create( + 'general_infer', engine_name='cnn', inputs=[read_op]) +bow_infer_op = op_maker.create( + 'general_infer', engine_name='bow', inputs=[read_op]) +response_op = op_maker.create( + 'general_response', inputs=[cnn_infer_op, bow_infer_op]) + +op_graph_maker = OpGraphMaker() +op_graph_maker.add_op(read_op) +op_graph_maker.add_op(cnn_infer_op) +op_graph_maker.add_op(bow_infer_op) +op_graph_maker.add_op(response_op) + +server = Server() +server.set_op_graph(op_graph_maker.get_op_graph()) +model_config = {cnn_infer_op: 'imdb_cnn_model', bow_infer_op: 'imdb_bow_model'} +server.load_model_config(model_config) +server.prepare_server(workdir="work_dir1", port=9393, device="cpu") +server.run_server() +``` + +Different from the normal prediction service, here we need to use DAG to describe the logic of the server side. + +When creating an Op, you need to specify the predecessor of the current Op (in this example, the predecessor of `cnn_infer_op` and `bow_infer_op` is `read_op`, and the predecessor of `response_op` is `cnn_infer_op` and `bow_infer_op`. For the infer Op `infer_op`, you need to define the prediction engine name `engine_name` (You can also use the default value. It is recommended to set the value to facilitate the client side to obtain the order of prediction results). + +At the same time, when configuring the model path, you need to create a model configuration dictionary with the infer Op as the key and the corresponding model path as value to inform Serving which model each infer OP uses. + +### Start client + +Start client by the following Python code (you can also run the `test_ensemble_client.py` script): + +```python +from paddle_serving_client import Client +from imdb_reader import IMDBDataset + +client = Client() +# If you have more than one model, make sure that the input +# and output of more than one model are the same. +client.load_client_config('imdb_bow_client_conf/serving_client_conf.prototxt') +client.connect(["127.0.0.1:9393"]) + +# you can define any english sentence or dataset here +# This example reuses imdb reader in training, you +# can define your own data preprocessing easily. +imdb_dataset = IMDBDataset() +imdb_dataset.load_resource('imdb.vocab') + +for i in range(3): + line = 'i am very sad | 0' + word_ids, label = imdb_dataset.get_words_and_label(line) + feed = {"words": word_ids} + fetch = ["acc", "cost", "prediction"] + fetch_maps = client.predict(feed=feed, fetch=fetch) + if len(fetch_maps) == 1: + print("step: {}, res: {}".format(i, fetch_maps['prediction'][0][1])) + else: + for model, fetch_map in fetch_maps.items(): + print("step: {}, model: {}, res: {}".format(i, model, fetch_map[ + 'prediction'][0][1])) +``` + +Compared with the normal prediction service, the client side has not changed much. When multiple model predictions are used, the prediction service will return a dictionary with engine name `engine_name`(the value is defined on the server side) as the key, and the corresponding model prediction results as the value. + +### Expected result + +```shell +step: 0, model: cnn, res: 0.560272455215 +step: 0, model: bow, res: 0.633530199528 +step: 1, model: cnn, res: 0.560272455215 +step: 1, model: bow, res: 0.633530199528 +step: 2, model: cnn, res: 0.560272455215 +step: 2, model: bow, res: 0.633530199528 +``` diff --git a/doc/MODEL_ENSEMBLE_IN_PADDLE_SERVING_CN.md b/doc/MODEL_ENSEMBLE_IN_PADDLE_SERVING_CN.md new file mode 100644 index 0000000000000000000000000000000000000000..ee816aa5a441610e845a933217041a774fad8129 --- /dev/null +++ b/doc/MODEL_ENSEMBLE_IN_PADDLE_SERVING_CN.md @@ -0,0 +1,121 @@ +# Paddle Serving中的集成预测 + +(简体中文|[English](MODEL_ENSEMBLE_IN_PADDLE_SERVING.md)) + +在一些场景中，可能使用多个相同输入的模型并行集成预测以获得更好的预测效果，Paddle Serving提供了这项功能。 + +下面将以文本分类任务为例，来展示Paddle Serving的集成预测功能（暂时还是串行预测，我们会尽快支持并行化）。 + +## 集成预测样例 + +该样例中（见下图），Server端在一项服务中并行预测相同输入的BOW和CNN模型，Client端获取两个模型的预测结果并进行后处理，得到最终的预测结果。 + +![simple example](model_ensemble_example.png) + +需要注意的是，目前只支持在同一个服务中使用多个相同格式输入输出的模型。在该例子中，CNN模型和BOW模型的输入输出格式是相同的。 + +样例中用到的代码保存在`python/examples/imdb`路径下： + +```shell +. +├── get_data.sh +├── imdb_reader.py +├── test_ensemble_client.py +└── test_ensemble_server.py +``` + +### 数据准备 + +通过下面命令获取预训练的CNN和BOW模型（您也可以直接运行`get_data.sh`脚本）： + +```shell +wget --no-check-certificate https://fleet.bj.bcebos.com/text_classification_data.tar.gz +wget --no-check-certificate https://paddle-serving.bj.bcebos.com/imdb-demo/imdb_model.tar.gz +tar -zxvf text_classification_data.tar.gz +tar -zxvf imdb_model.tar.gz +``` + +### 启动Server + +通过下面的Python代码启动Server端（您也可以直接运行`test_ensemble_server.py`脚本）： + +```python +from paddle_serving_server import OpMaker +from paddle_serving_server import OpGraphMaker +from paddle_serving_server import Server + +op_maker = OpMaker() +read_op = op_maker.create('general_reader') +cnn_infer_op = op_maker.create( + 'general_infer', engine_name='cnn', inputs=[read_op]) +bow_infer_op = op_maker.create( + 'general_infer', engine_name='bow', inputs=[read_op]) +response_op = op_maker.create( + 'general_response', inputs=[cnn_infer_op, bow_infer_op]) + +op_graph_maker = OpGraphMaker() +op_graph_maker.add_op(read_op) +op_graph_maker.add_op(cnn_infer_op) +op_graph_maker.add_op(bow_infer_op) +op_graph_maker.add_op(response_op) + +server = Server() +server.set_op_graph(op_graph_maker.get_op_graph()) +model_config = {cnn_infer_op: 'imdb_cnn_model', bow_infer_op: 'imdb_bow_model'} +server.load_model_config(model_config) +server.prepare_server(workdir="work_dir1", port=9393, device="cpu") +server.run_server() +``` + +与普通预测服务不同的是，这里我们需要用DAG来描述Server端的运行逻辑。 + +在创建Op的时候需要指定当前Op的前继（在该例子中，`cnn_infer_op`与`bow_infer_op`的前继均是`read_op`，`response_op`的前继是`cnn_infer_op`和`bow_infer_op`），对于预测Op`infer_op`还需要定义预测引擎名称`engine_name`（也可以使用默认值，建议设置该值方便Client端获取预测结果）。 + +同时在配置模型路径时，需要以预测Op为key，对应的模型路径为value，创建模型配置字典，来告知Serving每个预测Op使用哪个模型。 + +### 启动Client + +通过下面的Python代码运行Client端（您也可以直接运行`test_ensemble_client.py`脚本）： + +```python +from paddle_serving_client import Client +from imdb_reader import IMDBDataset + +client = Client() +# If you have more than one model, make sure that the input +# and output of more than one model are the same. +client.load_client_config('imdb_bow_client_conf/serving_client_conf.prototxt') +client.connect(["127.0.0.1:9393"]) + +# you can define any english sentence or dataset here +# This example reuses imdb reader in training, you +# can define your own data preprocessing easily. +imdb_dataset = IMDBDataset() +imdb_dataset.load_resource('imdb.vocab') + +for i in range(3): + line = 'i am very sad | 0' + word_ids, label = imdb_dataset.get_words_and_label(line) + feed = {"words": word_ids} + fetch = ["acc", "cost", "prediction"] + fetch_maps = client.predict(feed=feed, fetch=fetch) + if len(fetch_maps) == 1: + print("step: {}, res: {}".format(i, fetch_maps['prediction'][0][1])) + else: + for model, fetch_map in fetch_maps.items(): + print("step: {}, model: {}, res: {}".format(i, model, fetch_map[ + 'prediction'][0][1])) +``` + +Client端与普通预测服务没有发生太大的变化。当使用多个模型预测时，预测服务将返回一个key为Server端定义的引擎名称`engine_name`，value为对应的模型预测结果的字典。 + +### 预期结果 + +```txt +step: 0, model: cnn, res: 0.560272455215 +step: 0, model: bow, res: 0.633530199528 +step: 1, model: cnn, res: 0.560272455215 +step: 1, model: bow, res: 0.633530199528 +step: 2, model: cnn, res: 0.560272455215 +step: 2, model: bow, res: 0.633530199528 +``` diff --git a/doc/SERVER_DAG.md b/doc/SERVER_DAG.md index 5a5c851efacc28e5419d262ca671c83ec61e2015..dbf277ccbccc2a06838d65bfbf75e514b4d9a1ed 100644 --- a/doc/SERVER_DAG.md +++ b/doc/SERVER_DAG.md @@ -14,13 +14,19 @@ Deep neural nets often have some preprocessing steps on input data, and postproc ## How to define Node +### Simple series structure + PaddleServing has some predefined Computation Node in the framework. A very commonly used Computation Graph is the simple reader-inference-response mode that can cover most of the single model inference scenarios. A example graph and the corresponding DAG definition code is as follows. +
``` python import paddle_serving_server as serving +from paddle_serving_server import OpMaker +from paddle_serving_server import OpSeqMaker + op_maker = serving.OpMaker() read_op = op_maker.create('general_reader') general_infer_op = op_maker.create('general_infer') @@ -32,18 +38,54 @@ op_seq_maker.add_op(general_infer_op) op_seq_maker.add_op(general_response_op) ``` +For simple series logic, we simplify it and build it with `OpSeqMaker`. You can determine the successor by default according to the order of joining `OpSeqMaker` without specifying the successor of each node. + Since the code will be commonly used and users do not have to change the code, PaddleServing releases a easy-to-use launching command for service startup. An example is as follows: ``` python python -m paddle_serving_server.serve --model uci_housing_model --thread 10 --port 9292 ``` +### Nodes with multiple inputs + +An example containing multiple input nodes is given in the [MODEL_ENSEMBLE_IN_PADDLE_SERVING](MODEL_ENSEMBLE_IN_PADDLE_SERVING.md). A example graph and the corresponding DAG definition code is as follows. + +
+ +
+ +```python +from paddle_serving_server import OpMaker +from paddle_serving_server import OpGraphMaker +from paddle_serving_server import Server + +op_maker = OpMaker() +read_op = op_maker.create('general_reader') +cnn_infer_op = op_maker.create( + 'general_infer', engine_name='cnn', inputs=[read_op]) +bow_infer_op = op_maker.create( + 'general_infer', engine_name='bow', inputs=[read_op]) +response_op = op_maker.create( + 'general_response', inputs=[cnn_infer_op, bow_infer_op]) + +op_graph_maker = OpGraphMaker() +op_graph_maker.add_op(read_op) +op_graph_maker.add_op(cnn_infer_op) +op_graph_maker.add_op(bow_infer_op) +op_graph_maker.add_op(response_op) +``` + +For a graph with multiple input nodes, we need to use `OpGraphMaker` to build it, and you must give the predecessor of each node. + ## More Examples If a user has sparse features as inputs, and the model will do embedding lookup for each feature, we can do distributed embedding lookup operation which is not in the Paddle training computation graph. An example is as follows: ``` python import paddle_serving_server as serving +from paddle_serving_server import OpMaker +from paddle_serving_server import OpSeqMaker + op_maker = serving.OpMaker() read_op = op_maker.create('general_reader') dist_kv_op = op_maker.create('general_dist_kv') diff --git a/doc/SERVER_DAG_CN.md b/doc/SERVER_DAG_CN.md index 3bf42ef8e3fbcb8c509a69bfe6aea12f78dc4567..80d01f0287c5f721f093e96c7bcd1827f0601496 100644 --- a/doc/SERVER_DAG_CN.md +++ b/doc/SERVER_DAG_CN.md @@ -14,6 +14,8 @@ ## 如何定义节点 +### 简单的串联结构 + PaddleServing在框架中具有一些预定义的计算节点。 一种非常常用的计算图是简单的reader-infer-response模式，可以涵盖大多数单一模型推理方案。 示例图和相应的DAG定义代码如下。
@@ -21,6 +23,9 @@ PaddleServing在框架中具有一些预定义的计算节点。 一种非常常 ``` python import paddle_serving_server as serving +from paddle_serving_server import OpMaker +from paddle_serving_server import OpSeqMaker + op_maker = serving.OpMaker() read_op = op_maker.create('general_reader') general_infer_op = op_maker.create('general_infer') @@ -32,18 +37,54 @@ op_seq_maker.add_op(general_infer_op) op_seq_maker.add_op(general_response_op) ``` +对于简单的串联逻辑，我们将其简化为`Sequence`，使用`OpSeqMaker`进行构建。用户可以不指定每个节点的前继，默认按加入`OpSeqMaker`的顺序来确定前继。 + 由于该代码在大多数情况下都会被使用，并且用户不必更改代码，因此PaddleServing会发布一个易于使用的启动命令来启动服务。 示例如下： ``` python python -m paddle_serving_server.serve --model uci_housing_model --thread 10 --port 9292 ``` +### 包含多个输入的节点 + +在[Paddle Serving中的集成预测](MODEL_ENSEMBLE_IN_PADDLE_SERVING_CN.md)文档中给出了一个包含多个输入节点的样例，示意图和代码如下。 + +
+ +
+ +```python +from paddle_serving_server import OpMaker +from paddle_serving_server import OpGraphMaker +from paddle_serving_server import Server + +op_maker = OpMaker() +read_op = op_maker.create('general_reader') +cnn_infer_op = op_maker.create( + 'general_infer', engine_name='cnn', inputs=[read_op]) +bow_infer_op = op_maker.create( + 'general_infer', engine_name='bow', inputs=[read_op]) +response_op = op_maker.create( + 'general_response', inputs=[cnn_infer_op, bow_infer_op]) + +op_graph_maker = OpGraphMaker() +op_graph_maker.add_op(read_op) +op_graph_maker.add_op(cnn_infer_op) +op_graph_maker.add_op(bow_infer_op) +op_graph_maker.add_op(response_op) +``` + +对于含有多输入节点的计算图，需要使用`OpGraphMaker`来构建，同时必须给出每个节点的前继。 + ## 更多示例 如果用户将稀疏特征作为输入，并且模型将对每个特征进行嵌入查找，则我们可以进行分布式嵌入查找操作，该操作不在Paddle训练计算图中。 示例如下： ``` python import paddle_serving_server as serving +from paddle_serving_server import OpMaker +from paddle_serving_server import OpSeqMaker + op_maker = serving.OpMaker() read_op = op_maker.create('general_reader') dist_kv_op = op_maker.create('general_dist_kv') diff --git a/doc/complex_dag.png b/doc/complex_dag.png new file mode 100644 index 0000000000000000000000000000000000000000..4e844d9fc3915579ec44bb981e9e2bfc3e4f7675 Binary files /dev/null and b/doc/complex_dag.png differ diff --git a/doc/model_ensemble_example.png b/doc/model_ensemble_example.png new file mode 100644 index 0000000000000000000000000000000000000000..823e91ee9ea6e2b10c3bd2c0ca119f088582c685 Binary files /dev/null and b/doc/model_ensemble_example.png differ diff --git a/python/examples/imdb/test_ensemble_client.py b/python/examples/imdb/test_ensemble_client.py new file mode 100644 index 0000000000000000000000000000000000000000..6cafb3389fff5a25103bcb2b3a867b73b35b9e8e --- /dev/null +++ b/python/examples/imdb/test_ensemble_client.py @@ -0,0 +1,42 @@ +# Copyright (c) 2020 PaddlePaddle Authors. All Rights Reserved. +# +# 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. +# pylint: disable=doc-string-missing + +from paddle_serving_client import Client +from imdb_reader import IMDBDataset + +client = Client() +# If you have more than one model, make sure that the input +# and output of more than one model are the same. +client.load_client_config('imdb_bow_client_conf/serving_client_conf.prototxt') +client.connect(["127.0.0.1:9393"]) + +# you can define any english sentence or dataset here +# This example reuses imdb reader in training, you +# can define your own data preprocessing easily. +imdb_dataset = IMDBDataset() +imdb_dataset.load_resource('imdb.vocab') + +for i in range(3): + line = 'i am very sad | 0' + word_ids, label = imdb_dataset.get_words_and_label(line) + feed = {"words": word_ids} + fetch = ["acc", "cost", "prediction"] + fetch_maps = client.predict(feed=feed, fetch=fetch) + if len(fetch_maps) == 1: + print("step: {}, res: {}".format(i, fetch_maps['prediction'][0][1])) + else: + for model, fetch_map in fetch_maps.items(): + print("step: {}, model: {}, res: {}".format(i, model, fetch_map[ + 'prediction'][0][1])) diff --git a/python/examples/imdb/test_ensemble_server.py b/python/examples/imdb/test_ensemble_server.py new file mode 100644 index 0000000000000000000000000000000000000000..464288a0a167d8487f787d12c4b44a138da86f88 --- /dev/null +++ b/python/examples/imdb/test_ensemble_server.py @@ -0,0 +1,40 @@ +# Copyright (c) 2020 PaddlePaddle Authors. All Rights Reserved. +# +# 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. +# pylint: disable=doc-string-missing + +from paddle_serving_server import OpMaker +from paddle_serving_server import OpGraphMaker +from paddle_serving_server import Server + +op_maker = OpMaker() +read_op = op_maker.create('general_reader') +cnn_infer_op = op_maker.create( + 'general_infer', engine_name='cnn', inputs=[read_op]) +bow_infer_op = op_maker.create( + 'general_infer', engine_name='bow', inputs=[read_op]) +response_op = op_maker.create( + 'general_response', inputs=[cnn_infer_op, bow_infer_op]) + +op_graph_maker = OpGraphMaker() +op_graph_maker.add_op(read_op) +op_graph_maker.add_op(cnn_infer_op) +op_graph_maker.add_op(bow_infer_op) +op_graph_maker.add_op(response_op) + +server = Server() +server.set_op_graph(op_graph_maker.get_op_graph()) +model_config = {cnn_infer_op: 'imdb_cnn_model', bow_infer_op: 'imdb_bow_model'} +server.load_model_config(model_config) +server.prepare_server(workdir="work_dir1", port=9393, device="cpu") +server.run_server() diff --git a/python/paddle_serving_client/__init__.py b/python/paddle_serving_client/__init__.py index 98d233f059a1ad0b588bce5bf3ef831d783c3a44..8aeb22c92c781a4fb27b70403537f7016f05940d 100644 --- a/python/paddle_serving_client/__init__.py +++ b/python/paddle_serving_client/__init__.py @@ -264,28 +264,45 @@ class Client(object): if res == -1: return None - result_map_batch = [] - result_map = {} - # result map needs to be a numpy array - for i, name in enumerate(fetch_names): - if self.fetch_names_to_type_[name] == int_type: - result_map[name] = result_batch.get_int64_by_name(name) - shape = result_batch.get_shape(name) - result_map[name] = np.array(result_map[name]) - result_map[name].shape = shape - if name in self.lod_tensor_set: - result_map["{}.lod".format(name)] = result_batch.get_lod( - name) - elif self.fetch_names_to_type_[name] == float_type: - result_map[name] = result_batch.get_float_by_name(name) - shape = result_batch.get_shape(name) - result_map[name] = np.array(result_map[name]) - result_map[name].shape = shape - if name in self.lod_tensor_set: - result_map["{}.lod".format(name)] = result_batch.get_lod( - name) - - return result_map + multi_result_map = [] + model_engine_names = result_batch.get_engine_names() + for mi, engine_name in enumerate(model_engine_names): + result_map = {} + # result map needs to be a numpy array + for i, name in enumerate(fetch_names): + if self.fetch_names_to_type_[name] == int_type: + result_map[name] = result_batch.get_int64_by_name(mi, name) + shape = result_batch.get_shape(mi, name) + result_map[name] = np.array(result_map[name]) + result_map[name].shape = shape + if name in self.lod_tensor_set: + result_map["{}.lod".format( + name)] = result_batch.get_lod(mi, name) + elif self.fetch_names_to_type_[name] == float_type: + result_map[name] = result_batch.get_float_by_name(mi, name) + shape = result_batch.get_shape(mi, name) + result_map[name] = np.array(result_map[name]) + result_map[name].shape = shape + if name in self.lod_tensor_set: + result_map["{}.lod".format( + name)] = result_batch.get_lod(mi, name) + multi_result_map.append(result_map) + + ret = None + if len(model_engine_names) == 1: + # If only one model result is returned, the format of ret is result_map + ret = multi_result_map[0] + else: + # If multiple model results are returned, the format of ret is {name: result_map} + ret = { + engine_name: multi_result_map[mi] + for mi, engine_name in enumerate(model_engine_names) + } + + # When using the A/B test, the tag of variant needs to be returned + return ret if not need_variant_tag else [ + ret, self.result_handle_.variant_tag() + ] def release(self): self.client_handle_.destroy_predictor() diff --git a/python/paddle_serving_client/io/__init__.py b/python/paddle_serving_client/io/__init__.py index d723795f214e22957bff49f0ddf8fd42086b8a7e..74a6ca871b5c1e32b3c1ecbc6656c95d7c78a399 100644 --- a/python/paddle_serving_client/io/__init__.py +++ b/python/paddle_serving_client/io/__init__.py @@ -20,6 +20,7 @@ from paddle.fluid.framework import default_main_program from paddle.fluid.framework import Program from paddle.fluid import CPUPlace from paddle.fluid.io import save_inference_model +import paddle.fluid as fluid from ..proto import general_model_config_pb2 as model_conf import os @@ -100,3 +101,20 @@ def save_model(server_model_folder, with open("{}/serving_server_conf.stream.prototxt".format( server_model_folder), "wb") as fout: fout.write(config.SerializeToString()) + + +def inference_model_to_serving(infer_model, serving_client, serving_server): + place = fluid.CPUPlace() + exe = fluid.Executor(place) + inference_program, feed_target_names, fetch_targets = \ + fluid.io.load_inference_model(dirname=infer_model, executor=exe) + feed_dict = { + x: inference_program.global_block().var(x) + for x in feed_target_names + } + fetch_dict = {x.name: x for x in fetch_targets} + save_model(serving_client, serving_server, feed_dict, fetch_dict, + inference_program) + feed_names = feed_dict.keys() + fetch_names = fetch_dict.keys() + return feed_names, fetch_names diff --git a/python/paddle_serving_server/__init__.py b/python/paddle_serving_server/__init__.py index 6da0e6597e69d605b7172139d32c00b77d538386..8062a7c83d99c0bed712ff46840b81f4557a353d 100644 --- a/python/paddle_serving_server/__init__.py +++ b/python/paddle_serving_server/__init__.py @@ -11,6 +11,7 @@ # 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. +# pylint: disable=doc-string-missing import os from .proto import server_configure_pb2 as server_sdk @@ -21,6 +22,7 @@ import socket import paddle_serving_server as paddle_serving_server from .version import serving_server_version from contextlib import closing +import collections class OpMaker(object): @@ -36,17 +38,35 @@ class OpMaker(object): "general_dist_kv_quant_infer": "GeneralDistKVQuantInferOp", "general_copy": "GeneralCopyOp" } + self.node_name_suffix_ = collections.defaultdict(int) - # currently, inputs and outputs are not used - # when we have OpGraphMaker, inputs and outputs are necessary - def create(self, name, inputs=[], outputs=[]): - if name not in self.op_dict: - raise Exception("Op name {} is not supported right now".format( - name)) + def create(self, node_type, engine_name=None, inputs=[], outputs=[]): + if node_type not in self.op_dict: + raise Exception("Op type {} is not supported right now".format( + node_type)) node = server_sdk.DAGNode() - node.name = "{}_op".format(name) - node.type = self.op_dict[name] - return node + # node.name will be used as the infer engine name + if engine_name: + node.name = engine_name + else: + node.name = '{}_{}'.format(node_type, + self.node_name_suffix_[node_type]) + self.node_name_suffix_[node_type] += 1 + + node.type = self.op_dict[node_type] + if inputs: + for dep_node_str in inputs: + dep_node = server_sdk.DAGNode() + google.protobuf.text_format.Parse(dep_node_str, dep_node) + dep = server_sdk.DAGNodeDependency() + dep.name = dep_node.name + dep.mode = "RO" + node.dependencies.extend([dep]) + # Because the return value will be used as the key value of the + # dict, and the proto object is variable which cannot be hashed, + # so it is processed into a string. This has little effect on + # overall efficiency. + return google.protobuf.text_format.MessageToString(node) class OpSeqMaker(object): @@ -55,12 +75,25 @@ class OpSeqMaker(object): self.workflow.name = "workflow1" self.workflow.workflow_type = "Sequence" - def add_op(self, node): + def add_op(self, node_str): + node = server_sdk.DAGNode() + google.protobuf.text_format.Parse(node_str, node) + if len(node.dependencies) > 1: + raise Exception( + 'Set more than one predecessor for op in OpSeqMaker is not allowed.' + ) if len(self.workflow.nodes) >= 1: - dep = server_sdk.DAGNodeDependency() - dep.name = self.workflow.nodes[-1].name - dep.mode = "RO" - node.dependencies.extend([dep]) + if len(node.dependencies) == 0: + dep = server_sdk.DAGNodeDependency() + dep.name = self.workflow.nodes[-1].name + dep.mode = "RO" + node.dependencies.extend([dep]) + elif len(node.dependencies) == 1: + if node.dependencies[0].name != self.workflow.nodes[-1].name: + raise Exception( + 'You must add op in order in OpSeqMaker. The previous op is {}, but the current op is followed by {}.'. + format(node.dependencies[0].name, self.workflow.nodes[ + -1].name)) self.workflow.nodes.extend([node]) def get_op_sequence(self): @@ -69,13 +102,30 @@ class OpSeqMaker(object): return workflow_conf +class OpGraphMaker(object): + def __init__(self): + self.workflow = server_sdk.Workflow() + self.workflow.name = "workflow1" + # Currently, SDK only supports "Sequence" + self.workflow.workflow_type = "Sequence" + + def add_op(self, node_str): + node = server_sdk.DAGNode() + google.protobuf.text_format.Parse(node_str, node) + self.workflow.nodes.extend([node]) + + def get_op_graph(self): + workflow_conf = server_sdk.WorkflowConf() + workflow_conf.workflows.extend([self.workflow]) + return workflow_conf + + class Server(object): def __init__(self): self.server_handle_ = None self.infer_service_conf = None self.model_toolkit_conf = None self.resource_conf = None - self.engine = None self.memory_optimization = False self.model_conf = None self.workflow_fn = "workflow.prototxt" @@ -94,6 +144,7 @@ class Server(object): self.cur_path = os.getcwd() self.use_local_bin = False self.mkl_flag = False + self.model_config_paths = None # for multi-model in a workflow def set_max_concurrency(self, concurrency): self.max_concurrency = concurrency @@ -118,6 +169,9 @@ class Server(object): def set_op_sequence(self, op_seq): self.workflow_conf = op_seq + def set_op_graph(self, op_graph): + self.workflow_conf = op_graph + def set_memory_optimize(self, flag=False): self.memory_optimization = flag @@ -126,32 +180,30 @@ class Server(object): self.use_local_bin = True self.bin_path = os.environ["SERVING_BIN"] - def _prepare_engine(self, model_config_path, device): + def _prepare_engine(self, model_config_paths, device): if self.model_toolkit_conf == None: self.model_toolkit_conf = server_sdk.ModelToolkitConf() - if self.engine == None: - self.engine = server_sdk.EngineDesc() - - self.model_config_path = model_config_path - self.engine.name = "general_model" - self.engine.reloadable_meta = model_config_path + "/fluid_time_file" - os.system("touch {}".format(self.engine.reloadable_meta)) - self.engine.reloadable_type = "timestamp_ne" - self.engine.runtime_thread_num = 0 - self.engine.batch_infer_size = 0 - self.engine.enable_batch_align = 0 - self.engine.model_data_path = model_config_path - self.engine.enable_memory_optimization = self.memory_optimization - self.engine.static_optimization = False - self.engine.force_update_static_cache = False - - if device == "cpu": - self.engine.type = "FLUID_CPU_ANALYSIS_DIR" - elif device == "gpu": - self.engine.type = "FLUID_GPU_ANALYSIS_DIR" - - self.model_toolkit_conf.engines.extend([self.engine]) + for engine_name, model_config_path in model_config_paths.items(): + engine = server_sdk.EngineDesc() + engine.name = engine_name + engine.reloadable_meta = model_config_path + "/fluid_time_file" + os.system("touch {}".format(engine.reloadable_meta)) + engine.reloadable_type = "timestamp_ne" + engine.runtime_thread_num = 0 + engine.batch_infer_size = 0 + engine.enable_batch_align = 0 + engine.model_data_path = model_config_path + engine.enable_memory_optimization = self.memory_optimization + engine.static_optimization = False + engine.force_update_static_cache = False + + if device == "cpu": + engine.type = "FLUID_CPU_ANALYSIS_DIR" + elif device == "gpu": + engine.type = "FLUID_GPU_ANALYSIS_DIR" + + self.model_toolkit_conf.engines.extend([engine]) def _prepare_infer_service(self, port): if self.infer_service_conf == None: @@ -184,10 +236,49 @@ class Server(object): with open(filepath, "w") as fout: fout.write(str(pb_obj)) - def load_model_config(self, path): - self.model_config_path = path + def load_model_config(self, model_config_paths): + # At present, Serving needs to configure the model path in + # the resource.prototxt file to determine the input and output + # format of the workflow. To ensure that the input and output + # of multiple models are the same. + workflow_oi_config_path = None + if isinstance(model_config_paths, str): + # If there is only one model path, use the default infer_op. + # Because there are several infer_op type, we need to find + # it from workflow_conf. + default_engine_names = [ + 'general_infer_0', 'general_dist_kv_infer_0', + 'general_dist_kv_quant_infer_0' + ] + engine_name = None + for node in self.workflow_conf.workflows[0].nodes: + if node.name in default_engine_names: + engine_name = node.name + break + if engine_name is None: + raise Exception( + "You have set the engine_name of Op. Please use the form {op: model_path} to configure model path" + ) + self.model_config_paths = {engine_name: model_config_paths} + workflow_oi_config_path = self.model_config_paths[engine_name] + elif isinstance(model_config_paths, dict): + self.model_config_paths = {} + for node_str, path in model_config_paths.items(): + node = server_sdk.DAGNode() + google.protobuf.text_format.Parse(node_str, node) + self.model_config_paths[node.name] = path + print("You have specified multiple model paths, please ensure " + "that the input and output of multiple models are the same.") + workflow_oi_config_path = self.model_config_paths.items()[0][1] + else: + raise Exception("The type of model_config_paths must be str or " + "dict({op: model_path}), not {}.".format( + type(model_config_paths))) + self.model_conf = m_config.GeneralModelConfig() - f = open("{}/serving_server_conf.prototxt".format(path), 'r') + f = open( + "{}/serving_server_conf.prototxt".format(workflow_oi_config_path), + 'r') self.model_conf = google.protobuf.text_format.Merge( str(f.read()), self.model_conf) # check config here @@ -258,7 +349,7 @@ class Server(object): if not self.port_is_available(port): raise SystemExit("Prot {} is already used".format(port)) self._prepare_resource(workdir) - self._prepare_engine(self.model_config_path, device) + self._prepare_engine(self.model_config_paths, device) self._prepare_infer_service(port) self.workdir = workdir diff --git a/python/paddle_serving_server/monitor.py b/python/paddle_serving_server/monitor.py index 11fe393bc07c4cff0a1c9667a05cc5c598f06608..3f1ff6436917b8ae7ff4ea06fcae1f55bd65e887 100644 --- a/python/paddle_serving_server/monitor.py +++ b/python/paddle_serving_server/monitor.py @@ -91,6 +91,7 @@ class Monitor(object): model_name)) return model_name tar_model_path = os.path.join(local_tmp_path, model_name) + _LOGGER.info("try to unpack remote file({})".format(tar_model_path)) if not tarfile.is_tarfile(tar_model_path): raise Exception('not a tar packaged file type. {}'.format( self._check_param_help('remote_model_name', model_name))) @@ -105,10 +106,11 @@ class Monitor(object): self._check_param_help('local_tmp_path', local_tmp_path))) finally: os.remove(tar_model_path) - _LOGGER.debug('remove packed file({}).'.format(model_name)) + _LOGGER.debug('remove packed file({}).'.format(tar_model_path)) _LOGGER.info('using unpacked filename: {}.'.format( unpacked_filename)) - if not os.path.exists(unpacked_filename): + if not os.path.exists( + os.path.join(local_tmp_path, unpacked_filename)): raise Exception('file not exist. {}'.format( self._check_param_help('unpacked_filename', unpacked_filename))) @@ -124,13 +126,14 @@ class Monitor(object): '_local_tmp_path', '_interval' ] self._print_params(params) - if not os.path.exists(self._local_tmp_path): - _LOGGER.info('mkdir: {}'.format(self._local_tmp_path)) - os.makedirs(self._local_tmp_path) + local_tmp_path = os.path.join(self._local_path, self._local_tmp_path) + _LOGGER.info('local_tmp_path: {}'.format(local_tmp_path)) + if not os.path.exists(local_tmp_path): + _LOGGER.info('mkdir: {}'.format(local_tmp_path)) + os.makedirs(local_tmp_path) while True: [flag, timestamp] = self._exist_remote_file( - self._remote_path, self._remote_donefile_name, - self._local_tmp_path) + self._remote_path, self._remote_donefile_name, local_tmp_path) if flag: if self._remote_donefile_timestamp is None or \ timestamp != self._remote_donefile_timestamp: @@ -139,15 +142,15 @@ class Monitor(object): self._remote_donefile_timestamp = timestamp self._pull_remote_dir(self._remote_path, self._remote_model_name, - self._local_tmp_path) + local_tmp_path) _LOGGER.info('pull remote model({}).'.format( self._remote_model_name)) unpacked_filename = self._decompress_model_file( - self._local_tmp_path, self._remote_model_name, + local_tmp_path, self._remote_model_name, self._unpacked_filename) - self._update_local_model( - self._local_tmp_path, unpacked_filename, - self._local_path, self._local_model_name) + self._update_local_model(local_tmp_path, unpacked_filename, + self._local_path, + self._local_model_name) _LOGGER.info('update local model({}).'.format( self._local_model_name)) self._update_local_donefile(self._local_path, @@ -220,7 +223,12 @@ class HadoopMonitor(Monitor): local_dirpath = os.path.join(local_tmp_path, dirname) if os.path.exists(local_dirpath): _LOGGER.info('remove old temporary model file({}).'.format(dirname)) - shutil.rmtree(local_dirpath) + if self._unpacked_filename is None: + # the remote file is model folder. + shutil.rmtree(local_dirpath) + else: + # the remote file is a packed model file + os.remove(local_dirpath) remote_dirpath = os.path.join(remote_path, dirname) cmd = '{} -get {} {} 2>/dev/null'.format(self._cmd_prefix, remote_dirpath, local_dirpath) @@ -301,8 +309,8 @@ class FTPMonitor(Monitor): os.path.join(remote_path, remote_dirname), name, os.path.join(local_tmp_path, remote_dirname), overwrite) else: - self._download_remote_file(remote_dirname, name, - local_tmp_path, overwrite) + self._download_remote_file(remote_dirpath, name, + local_dirpath, overwrite) except ftplib.error_perm: _LOGGER.debug('{} is file.'.format(remote_dirname)) self._download_remote_file(remote_path, remote_dirname, @@ -325,17 +333,17 @@ class GeneralMonitor(Monitor): def _get_local_file_timestamp(self, filename): return os.path.getmtime(filename) - def _exist_remote_file(self, path, filename, local_tmp_path): - remote_filepath = os.path.join(path, filename) + def _exist_remote_file(self, remote_path, filename, local_tmp_path): + remote_filepath = os.path.join(remote_path, filename) url = '{}/{}'.format(self._general_host, remote_filepath) _LOGGER.debug('remote file url: {}'.format(url)) - cmd = 'wget -N -P {} {} &>/dev/null'.format(local_tmp_path, url) + # only for check donefile, which is not a folder. + cmd = 'wget -nd -N -P {} {} &>/dev/null'.format(local_tmp_path, url) _LOGGER.debug('wget cmd: {}'.format(cmd)) if os.system(cmd) != 0: - _LOGGER.debug('remote file({}) not exist.'.format(filename)) + _LOGGER.debug('remote file({}) not exist.'.format(remote_filepath)) return [False, None] else: - _LOGGER.debug('download remote file({}).'.format(filename)) timestamp = self._get_local_file_timestamp( os.path.join(local_tmp_path, filename)) return [True, timestamp] @@ -344,7 +352,13 @@ class GeneralMonitor(Monitor): remote_dirpath = os.path.join(remote_path, dirname) url = '{}/{}'.format(self._general_host, remote_dirpath) _LOGGER.debug('remote file url: {}'.format(url)) - cmd = 'wget -nH -r -P {} {} &>/dev/null'.format(local_tmp_path, url) + if self._unpacked_filename is None: + # the remote file is model folder. + cmd = 'wget -nH -r -P {} {} &>/dev/null'.format( + os.path.join(local_tmp_path, dirname), url) + else: + # the remote file is a packed model file + cmd = 'wget -nd -N -P {} {} &>/dev/null'.format(local_tmp_path, url) _LOGGER.debug('wget cmd: {}'.format(cmd)) if os.system(cmd) != 0: raise Exception('pull remote dir failed. {}'.format( @@ -352,7 +366,11 @@ class GeneralMonitor(Monitor): def parse_args(): - ''' parse args. ''' + """ parse args. + + Returns: + parser.parse_args(). + """ parser = argparse.ArgumentParser(description="Monitor") parser.add_argument( "--type", type=str, default='general', help="Type of remote server") diff --git a/python/paddle_serving_server/web_service.py b/python/paddle_serving_server/web_service.py index ca43426c2a82a0c8be296c8410361acbf498fc5c..4a033cbcf1d32a55eaacbe9c0f6704e304e127b3 100755 --- a/python/paddle_serving_server/web_service.py +++ b/python/paddle_serving_server/web_service.py @@ -62,22 +62,14 @@ class WebService(object): abort(400) try: feed, fetch = self.preprocess(request.json, request.json["fetch"]) - if isinstance(feed, list): - fetch_map_batch = self.client_service.predict( - feed_batch=feed, fetch=fetch) - fetch_map_batch = self.postprocess( - feed=request.json, fetch=fetch, fetch_map=fetch_map_batch) - for key in fetch_map_batch: - fetch_map_batch[key] = fetch_map_batch[key].tolist() - result = {"result": fetch_map_batch} - elif isinstance(feed, dict): - if "fetch" in feed: - del feed["fetch"] - fetch_map = self.client_service.predict(feed=feed, fetch=fetch) - for key in fetch_map: - fetch_map[key] = fetch_map[key][0].tolist() - result = self.postprocess( - feed=request.json, fetch=fetch, fetch_map=fetch_map) + if isinstance(feed, dict) and "fetch" in feed: + del feed["fetch"] + fetch_map = self.client_service.predict(feed=feed, fetch=fetch) + for key in fetch_map: + fetch_map[key] = fetch_map[key][0].tolist() + result = self.postprocess( + feed=request.json, fetch=fetch, fetch_map=fetch_map) + result = {"result": result} except ValueError: result = {"result": "Request Value Error"} return result diff --git a/python/paddle_serving_server_gpu/__init__.py b/python/paddle_serving_server_gpu/__init__.py index 0e5a49c4870956557c99fdf8abf08edf47bc4aa6..45e71a383b4fe0e5ca3a5284985b702cd815f18c 100644 --- a/python/paddle_serving_server_gpu/__init__.py +++ b/python/paddle_serving_server_gpu/__init__.py @@ -24,6 +24,7 @@ import time from .version import serving_server_version from contextlib import closing import argparse +import collections def serve_args(): @@ -66,17 +67,35 @@ class OpMaker(object): "general_dist_kv_infer": "GeneralDistKVInferOp", "general_dist_kv": "GeneralDistKVOp" } + self.node_name_suffix_ = collections.defaultdict(int) - # currently, inputs and outputs are not used - # when we have OpGraphMaker, inputs and outputs are necessary - def create(self, name, inputs=[], outputs=[]): - if name not in self.op_dict: - raise Exception("Op name {} is not supported right now".format( - name)) + def create(self, node_type, engine_name=None, inputs=[], outputs=[]): + if node_type not in self.op_dict: + raise Exception("Op type {} is not supported right now".format( + node_type)) node = server_sdk.DAGNode() - node.name = "{}_op".format(name) - node.type = self.op_dict[name] - return node + # node.name will be used as the infer engine name + if engine_name: + node.name = engine_name + else: + node.name = '{}_{}'.format(node_type, + self.node_name_suffix_[node_type]) + self.node_name_suffix_[node_type] += 1 + + node.type = self.op_dict[node_type] + if inputs: + for dep_node_str in inputs: + dep_node = server_sdk.DAGNode() + google.protobuf.text_format.Parse(dep_node_str, dep_node) + dep = server_sdk.DAGNodeDependency() + dep.name = dep_node.name + dep.mode = "RO" + node.dependencies.extend([dep]) + # Because the return value will be used as the key value of the + # dict, and the proto object is variable which cannot be hashed, + # so it is processed into a string. This has little effect on + # overall efficiency. + return google.protobuf.text_format.MessageToString(node) class OpSeqMaker(object): @@ -85,12 +104,25 @@ class OpSeqMaker(object): self.workflow.name = "workflow1" self.workflow.workflow_type = "Sequence" - def add_op(self, node): + def add_op(self, node_str): + node = server_sdk.DAGNode() + google.protobuf.text_format.Parse(node_str, node) + if len(node.dependencies) > 1: + raise Exception( + 'Set more than one predecessor for op in OpSeqMaker is not allowed.' + ) if len(self.workflow.nodes) >= 1: - dep = server_sdk.DAGNodeDependency() - dep.name = self.workflow.nodes[-1].name - dep.mode = "RO" - node.dependencies.extend([dep]) + if len(node.dependencies) == 0: + dep = server_sdk.DAGNodeDependency() + dep.name = self.workflow.nodes[-1].name + dep.mode = "RO" + node.dependencies.extend([dep]) + elif len(node.dependencies) == 1: + if node.dependencies[0].name != self.workflow.nodes[-1].name: + raise Exception( + 'You must add op in order in OpSeqMaker. The previous op is {}, but the current op is followed by {}.'. + format(node.dependencies[0].name, self.workflow.nodes[ + -1].name)) self.workflow.nodes.extend([node]) def get_op_sequence(self): @@ -99,13 +131,30 @@ class OpSeqMaker(object): return workflow_conf +class OpGraphMaker(object): + def __init__(self): + self.workflow = server_sdk.Workflow() + self.workflow.name = "workflow1" + # Currently, SDK only supports "Sequence" + self.workflow.workflow_type = "Sequence" + + def add_op(self, node_str): + node = server_sdk.DAGNode() + google.protobuf.text_format.Parse(node_str, node) + self.workflow.nodes.extend([node]) + + def get_op_graph(self): + workflow_conf = server_sdk.WorkflowConf() + workflow_conf.workflows.extend([self.workflow]) + return workflow_conf + + class Server(object): def __init__(self): self.server_handle_ = None self.infer_service_conf = None self.model_toolkit_conf = None self.resource_conf = None - self.engine = None self.memory_optimization = False self.model_conf = None self.workflow_fn = "workflow.prototxt" @@ -125,6 +174,7 @@ class Server(object): self.check_cuda() self.use_local_bin = False self.gpuid = 0 + self.model_config_paths = None # for multi-model in a workflow def set_max_concurrency(self, concurrency): self.max_concurrency = concurrency @@ -149,6 +199,9 @@ class Server(object): def set_op_sequence(self, op_seq): self.workflow_conf = op_seq + def set_op_graph(self, op_graph): + self.workflow_conf = op_graph + def set_memory_optimize(self, flag=False): self.memory_optimization = flag @@ -167,33 +220,31 @@ class Server(object): def set_gpuid(self, gpuid=0): self.gpuid = gpuid - def _prepare_engine(self, model_config_path, device): + def _prepare_engine(self, model_config_paths, device): if self.model_toolkit_conf == None: self.model_toolkit_conf = server_sdk.ModelToolkitConf() - if self.engine == None: - self.engine = server_sdk.EngineDesc() - - self.model_config_path = model_config_path - self.engine.name = "general_model" - #self.engine.reloadable_meta = model_config_path + "/fluid_time_file" - self.engine.reloadable_meta = self.workdir + "/fluid_time_file" - os.system("touch {}".format(self.engine.reloadable_meta)) - self.engine.reloadable_type = "timestamp_ne" - self.engine.runtime_thread_num = 0 - self.engine.batch_infer_size = 0 - self.engine.enable_batch_align = 0 - self.engine.model_data_path = model_config_path - self.engine.enable_memory_optimization = self.memory_optimization - self.engine.static_optimization = False - self.engine.force_update_static_cache = False - - if device == "cpu": - self.engine.type = "FLUID_CPU_ANALYSIS_DIR" - elif device == "gpu": - self.engine.type = "FLUID_GPU_ANALYSIS_DIR" - - self.model_toolkit_conf.engines.extend([self.engine]) + for engine_name, model_config_path in model_config_paths.items(): + engine = server_sdk.EngineDesc() + engine.name = engine_name + # engine.reloadable_meta = model_config_path + "/fluid_time_file" + engine.reloadable_meta = self.workdir + "/fluid_time_file" + os.system("touch {}".format(engine.reloadable_meta)) + engine.reloadable_type = "timestamp_ne" + engine.runtime_thread_num = 0 + engine.batch_infer_size = 0 + engine.enable_batch_align = 0 + engine.model_data_path = model_config_path + engine.enable_memory_optimization = self.memory_optimization + engine.static_optimization = False + engine.force_update_static_cache = False + + if device == "cpu": + engine.type = "FLUID_CPU_ANALYSIS_DIR" + elif device == "gpu": + engine.type = "FLUID_GPU_ANALYSIS_DIR" + + self.model_toolkit_conf.engines.extend([engine]) def _prepare_infer_service(self, port): if self.infer_service_conf == None: @@ -225,10 +276,49 @@ class Server(object): with open(filepath, "w") as fout: fout.write(str(pb_obj)) - def load_model_config(self, path): - self.model_config_path = path + def load_model_config(self, model_config_paths): + # At present, Serving needs to configure the model path in + # the resource.prototxt file to determine the input and output + # format of the workflow. To ensure that the input and output + # of multiple models are the same. + workflow_oi_config_path = None + if isinstance(model_config_paths, str): + # If there is only one model path, use the default infer_op. + # Because there are several infer_op type, we need to find + # it from workflow_conf. + default_engine_names = [ + 'general_infer_0', 'general_dist_kv_infer_0', + 'general_dist_kv_quant_infer_0' + ] + engine_name = None + for node in self.workflow_conf.workflows[0].nodes: + if node.name in default_engine_names: + engine_name = node.name + break + if engine_name is None: + raise Exception( + "You have set the engine_name of Op. Please use the form {op: model_path} to configure model path" + ) + self.model_config_paths = {engine_name: model_config_paths} + workflow_oi_config_path = self.model_config_paths[engine_name] + elif isinstance(model_config_paths, dict): + self.model_config_paths = {} + for node_str, path in model_config_paths.items(): + node = server_sdk.DAGNode() + google.protobuf.text_format.Parse(node_str, node) + self.model_config_paths[node.name] = path + print("You have specified multiple model paths, please ensure " + "that the input and output of multiple models are the same.") + workflow_oi_config_path = self.model_config_paths.items()[0][1] + else: + raise Exception("The type of model_config_paths must be str or " + "dict({op: model_path}), not {}.".format( + type(model_config_paths))) + self.model_conf = m_config.GeneralModelConfig() - f = open("{}/serving_server_conf.prototxt".format(path), 'r') + f = open( + "{}/serving_server_conf.prototxt".format(workflow_oi_config_path), + 'r') self.model_conf = google.protobuf.text_format.Merge( str(f.read()), self.model_conf) # check config here @@ -291,7 +381,7 @@ class Server(object): self.set_port(port) self._prepare_resource(workdir) - self._prepare_engine(self.model_config_path, device) + self._prepare_engine(self.model_config_paths, device) self._prepare_infer_service(port) self.workdir = workdir diff --git a/python/paddle_serving_server_gpu/monitor.py b/python/paddle_serving_server_gpu/monitor.py new file mode 100644 index 0000000000000000000000000000000000000000..3f1ff6436917b8ae7ff4ea06fcae1f55bd65e887 --- /dev/null +++ b/python/paddle_serving_server_gpu/monitor.py @@ -0,0 +1,504 @@ +# Copyright (c) 2020 PaddlePaddle Authors. All Rights Reserved. +# +# 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. +""" +Usage: + Start monitor with one line command + Example: + python -m paddle_serving_server.monitor +""" +import os +import time +import argparse +import commands +import datetime +import shutil +import tarfile +import logging + +_LOGGER = logging.getLogger(__name__) + + +class Monitor(object): + ''' + Monitor base class. It is used to monitor the remote model, pull and update the local model. + ''' + + def __init__(self, interval): + self._remote_path = None + self._remote_model_name = None + self._remote_donefile_name = None + self._local_path = None + self._local_model_name = None + self._local_timestamp_file = None + self._interval = interval + self._remote_donefile_timestamp = None + self._local_tmp_path = None + self._unpacked_filename = None + + def set_remote_path(self, remote_path): + self._remote_path = remote_path + + def set_remote_model_name(self, model_name): + self._remote_model_name = model_name + + def set_remote_donefile_name(self, donefile_name): + self._remote_donefile_name = donefile_name + + def set_local_path(self, local_path): + self._local_path = local_path + + def set_local_model_name(self, model_name): + self._local_model_name = model_name + + def set_local_timestamp_file(self, timestamp_file): + self._local_timestamp_file = timestamp_file + + def set_local_tmp_path(self, tmp_path): + self._local_tmp_path = tmp_path + + def set_unpacked_filename(self, unpacked_filename): + self._unpacked_filename = unpacked_filename + + def _check_param_help(self, param_name, param_value): + return "Please check the {}({}) parameter.".format(param_name, + param_value) + + def _check_params(self, params): + for param in params: + if getattr(self, param, None) is None: + raise Exception('{} not set.'.format(param)) + + def _print_params(self, params_name): + self._check_params(params_name) + for name in params_name: + _LOGGER.info('{}: {}'.format(name, getattr(self, name))) + + def _decompress_model_file(self, local_tmp_path, model_name, + unpacked_filename): + if unpacked_filename is None: + _LOGGER.debug('remote file({}) is already unpacked.'.format( + model_name)) + return model_name + tar_model_path = os.path.join(local_tmp_path, model_name) + _LOGGER.info("try to unpack remote file({})".format(tar_model_path)) + if not tarfile.is_tarfile(tar_model_path): + raise Exception('not a tar packaged file type. {}'.format( + self._check_param_help('remote_model_name', model_name))) + try: + _LOGGER.info('unpack remote file({}).'.format(model_name)) + tar = tarfile.open(tar_model_path) + tar.extractall(local_tmp_path) + tar.close() + except: + raise Exception( + 'Decompressing failed, maybe no disk space left. {}'.foemat( + self._check_param_help('local_tmp_path', local_tmp_path))) + finally: + os.remove(tar_model_path) + _LOGGER.debug('remove packed file({}).'.format(tar_model_path)) + _LOGGER.info('using unpacked filename: {}.'.format( + unpacked_filename)) + if not os.path.exists( + os.path.join(local_tmp_path, unpacked_filename)): + raise Exception('file not exist. {}'.format( + self._check_param_help('unpacked_filename', + unpacked_filename))) + return unpacked_filename + + def run(self): + ''' + Monitor the remote model by polling and update the local model. + ''' + params = [ + '_remote_path', '_remote_model_name', '_remote_donefile_name', + '_local_model_name', '_local_path', '_local_timestamp_file', + '_local_tmp_path', '_interval' + ] + self._print_params(params) + local_tmp_path = os.path.join(self._local_path, self._local_tmp_path) + _LOGGER.info('local_tmp_path: {}'.format(local_tmp_path)) + if not os.path.exists(local_tmp_path): + _LOGGER.info('mkdir: {}'.format(local_tmp_path)) + os.makedirs(local_tmp_path) + while True: + [flag, timestamp] = self._exist_remote_file( + self._remote_path, self._remote_donefile_name, local_tmp_path) + if flag: + if self._remote_donefile_timestamp is None or \ + timestamp != self._remote_donefile_timestamp: + _LOGGER.info('doneilfe({}) changed.'.format( + self._remote_donefile_name)) + self._remote_donefile_timestamp = timestamp + self._pull_remote_dir(self._remote_path, + self._remote_model_name, + local_tmp_path) + _LOGGER.info('pull remote model({}).'.format( + self._remote_model_name)) + unpacked_filename = self._decompress_model_file( + local_tmp_path, self._remote_model_name, + self._unpacked_filename) + self._update_local_model(local_tmp_path, unpacked_filename, + self._local_path, + self._local_model_name) + _LOGGER.info('update local model({}).'.format( + self._local_model_name)) + self._update_local_donefile(self._local_path, + self._local_model_name, + self._local_timestamp_file) + _LOGGER.info('update model timestamp({}).'.format( + self._local_timestamp_file)) + else: + _LOGGER.info('remote({}) has no donefile.'.format( + self._remote_path)) + _LOGGER.info('sleep {}s.'.format(self._interval)) + time.sleep(self._interval) + + def _exist_remote_file(self, path, filename, local_tmp_path): + raise Exception('This function must be inherited.') + + def _pull_remote_dir(self, remote_path, dirname, local_tmp_path): + raise Exception('This function must be inherited.') + + def _update_local_model(self, local_tmp_path, remote_model_name, local_path, + local_model_name): + tmp_model_path = os.path.join(local_tmp_path, remote_model_name) + local_model_path = os.path.join(local_path, local_model_name) + cmd = 'cp -r {}/* {}'.format(tmp_model_path, local_model_path) + _LOGGER.debug('update model cmd: {}'.format(cmd)) + if os.system(cmd) != 0: + raise Exception('update local model failed.') + + def _update_local_donefile(self, local_path, local_model_name, + local_timestamp_file): + donefile_path = os.path.join(local_path, local_model_name, + local_timestamp_file) + cmd = 'touch {}'.format(donefile_path) + _LOGGER.debug('update timestamp cmd: {}'.format(cmd)) + if os.system(cmd) != 0: + raise Exception('update local donefile failed.') + + +class HadoopMonitor(Monitor): + ''' Monitor HDFS or AFS by Hadoop-client. ''' + + def __init__(self, hadoop_bin, fs_name='', fs_ugi='', interval=10): + super(HadoopMonitor, self).__init__(interval) + self._hadoop_bin = hadoop_bin + self._fs_name = fs_name + self._fs_ugi = fs_ugi + self._print_params(['_hadoop_bin', '_fs_name', '_fs_ugi']) + self._cmd_prefix = '{} fs '.format(self._hadoop_bin) + if self._fs_name: + self._cmd_prefix += '-D fs.default.name={} '.format(self._fs_name) + if self._fs_ugi: + self._cmd_prefix += '-D hadoop.job.ugi={} '.format(self._fs_ugi) + _LOGGER.info('Hadoop prefix cmd: {}'.format(self._cmd_prefix)) + + def _exist_remote_file(self, path, filename, local_tmp_path): + remote_filepath = os.path.join(path, filename) + cmd = '{} -ls {} 2>/dev/null'.format(self._cmd_prefix, remote_filepath) + _LOGGER.debug('check cmd: {}'.format(cmd)) + [status, output] = commands.getstatusoutput(cmd) + _LOGGER.debug('resp: {}'.format(output)) + if status == 0: + [_, _, _, _, _, mdate, mtime, _] = output.split('\n')[-1].split() + timestr = mdate + mtime + return [True, timestr] + else: + return [False, None] + + def _pull_remote_dir(self, remote_path, dirname, local_tmp_path): + # remove old file before pull remote dir + local_dirpath = os.path.join(local_tmp_path, dirname) + if os.path.exists(local_dirpath): + _LOGGER.info('remove old temporary model file({}).'.format(dirname)) + if self._unpacked_filename is None: + # the remote file is model folder. + shutil.rmtree(local_dirpath) + else: + # the remote file is a packed model file + os.remove(local_dirpath) + remote_dirpath = os.path.join(remote_path, dirname) + cmd = '{} -get {} {} 2>/dev/null'.format(self._cmd_prefix, + remote_dirpath, local_dirpath) + _LOGGER.debug('pull cmd: {}'.format(cmd)) + if os.system(cmd) != 0: + raise Exception('pull remote dir failed. {}'.format( + self._check_param_help('remote_model_name', dirname))) + + +class FTPMonitor(Monitor): + ''' FTP Monitor. ''' + + def __init__(self, host, port, username="", password="", interval=10): + super(FTPMonitor, self).__init__(interval) + import ftplib + self._ftp = ftplib.FTP() + self._ftp_host = host + self._ftp_port = port + self._ftp_username = username + self._ftp_password = password + self._ftp.connect(self._ftp_host, self._ftp_port) + self._ftp.login(self._ftp_username, self._ftp_password) + self._print_params( + ['_ftp_host', '_ftp_port', '_ftp_username', '_ftp_password']) + + def _exist_remote_file(self, path, filename, local_tmp_path): + import ftplib + try: + _LOGGER.debug('cwd: {}'.format(path)) + self._ftp.cwd(path) + timestamp = self._ftp.voidcmd('MDTM {}'.format(filename))[4:].strip( + ) + return [True, timestamp] + except ftplib.error_perm: + _LOGGER.debug('remote file({}) not exist.'.format(filename)) + return [False, None] + + def _download_remote_file(self, + remote_path, + remote_filename, + local_tmp_path, + overwrite=True): + local_fullpath = os.path.join(local_tmp_path, remote_filename) + if not overwrite and os.path.isfile(fullpath): + return + else: + with open(local_fullpath, 'wb') as f: + _LOGGER.debug('cwd: {}'.format(remote_path)) + self._ftp.cwd(remote_path) + _LOGGER.debug('download remote file({})'.format( + remote_filename)) + self._ftp.retrbinary('RETR {}'.format(remote_filename), f.write) + + def _download_remote_files(self, + remote_path, + remote_dirname, + local_tmp_path, + overwrite=True): + import ftplib + remote_dirpath = os.path.join(remote_path, remote_dirname) + # Check whether remote_dirpath is a file or a folder + try: + _LOGGER.debug('cwd: {}'.format(remote_dirpath)) + self._ftp.cwd(remote_dirpath) + _LOGGER.debug('{} is folder.'.format(remote_dirname)) + + local_dirpath = os.path.join(local_tmp_path, remote_dirname) + if not os.path.exists(local_dirpath): + _LOGGER.info('mkdir: {}'.format(local_dirpath)) + os.mkdir(local_dirpath) + + output = [] + self._ftp.dir(output.append) + for line in output: + [attr, _, _, _, _, _, _, _, name] = line.split() + if attr[0] == 'd': + self._download_remote_files( + os.path.join(remote_path, remote_dirname), name, + os.path.join(local_tmp_path, remote_dirname), overwrite) + else: + self._download_remote_file(remote_dirpath, name, + local_dirpath, overwrite) + except ftplib.error_perm: + _LOGGER.debug('{} is file.'.format(remote_dirname)) + self._download_remote_file(remote_path, remote_dirname, + local_tmp_path, overwrite) + return + + def _pull_remote_dir(self, remote_path, dirname, local_tmp_path): + self._download_remote_files( + remote_path, dirname, local_tmp_path, overwrite=True) + + +class GeneralMonitor(Monitor): + ''' General Monitor. ''' + + def __init__(self, host, interval=10): + super(GeneralMonitor, self).__init__(interval) + self._general_host = host + self._print_params(['_general_host']) + + def _get_local_file_timestamp(self, filename): + return os.path.getmtime(filename) + + def _exist_remote_file(self, remote_path, filename, local_tmp_path): + remote_filepath = os.path.join(remote_path, filename) + url = '{}/{}'.format(self._general_host, remote_filepath) + _LOGGER.debug('remote file url: {}'.format(url)) + # only for check donefile, which is not a folder. + cmd = 'wget -nd -N -P {} {} &>/dev/null'.format(local_tmp_path, url) + _LOGGER.debug('wget cmd: {}'.format(cmd)) + if os.system(cmd) != 0: + _LOGGER.debug('remote file({}) not exist.'.format(remote_filepath)) + return [False, None] + else: + timestamp = self._get_local_file_timestamp( + os.path.join(local_tmp_path, filename)) + return [True, timestamp] + + def _pull_remote_dir(self, remote_path, dirname, local_tmp_path): + remote_dirpath = os.path.join(remote_path, dirname) + url = '{}/{}'.format(self._general_host, remote_dirpath) + _LOGGER.debug('remote file url: {}'.format(url)) + if self._unpacked_filename is None: + # the remote file is model folder. + cmd = 'wget -nH -r -P {} {} &>/dev/null'.format( + os.path.join(local_tmp_path, dirname), url) + else: + # the remote file is a packed model file + cmd = 'wget -nd -N -P {} {} &>/dev/null'.format(local_tmp_path, url) + _LOGGER.debug('wget cmd: {}'.format(cmd)) + if os.system(cmd) != 0: + raise Exception('pull remote dir failed. {}'.format( + self._check_param_help('remote_model_name', dirname))) + + +def parse_args(): + """ parse args. + + Returns: + parser.parse_args(). + """ + parser = argparse.ArgumentParser(description="Monitor") + parser.add_argument( + "--type", type=str, default='general', help="Type of remote server") + parser.add_argument( + "--remote_path", + type=str, + required=True, + help="The base path for the remote") + parser.add_argument( + "--remote_model_name", + type=str, + required=True, + help="The model name to be pulled from the remote") + parser.add_argument( + "--remote_donefile_name", + type=str, + required=True, + help="The donefile name that marks the completion of the remote model update" + ) + parser.add_argument( + "--local_path", type=str, required=True, help="Local work path") + parser.add_argument( + "--local_model_name", type=str, required=True, help="Local model name") + parser.add_argument( + "--local_timestamp_file", + type=str, + default='fluid_time_file', + help="The timestamp file used locally for hot loading, The file is considered to be placed in the `local_path/local_model_name` folder." + ) + parser.add_argument( + "--local_tmp_path", + type=str, + default='_serving_monitor_tmp', + help="The path of the folder where temporary files are stored locally. If it does not exist, it will be created automatically" + ) + parser.add_argument( + "--unpacked_filename", + type=str, + default=None, + help="If the model of the remote production is a packaged file, the unpacked file name should be set. Currently, only tar packaging format is supported." + ) + parser.add_argument( + "--interval", + type=int, + default=10, + help="The polling interval in seconds") + parser.add_argument( + "--debug", action='store_true', help="If set, output more details") + parser.set_defaults(debug=False) + # general monitor + parser.add_argument("--general_host", type=str, help="General remote host") + # ftp monitor + parser.add_argument("--ftp_host", type=str, help="FTP remote host") + parser.add_argument("--ftp_port", type=int, help="FTP remote port") + parser.add_argument( + "--ftp_username", + type=str, + default='', + help="FTP username. Not used if anonymous access.") + parser.add_argument( + "--ftp_password", + type=str, + default='', + help="FTP password. Not used if anonymous access") + # afs/hdfs monitor + parser.add_argument( + "--hadoop_bin", type=str, help="Path of Hadoop binary file") + parser.add_argument( + "--fs_name", + type=str, + default='', + help="AFS/HDFS fs_name. Not used if set in Hadoop-client.") + parser.add_argument( + "--fs_ugi", + type=str, + default='', + help="AFS/HDFS fs_ugi, Not used if set in Hadoop-client") + return parser.parse_args() + + +def get_monitor(mtype): + """ generator monitor instance. + + Args: + mtype: type of monitor + + Returns: + monitor instance. + """ + if mtype == 'ftp': + return FTPMonitor( + args.ftp_host, + args.ftp_port, + username=args.ftp_username, + password=args.ftp_password, + interval=args.interval) + elif mtype == 'general': + return GeneralMonitor(args.general_host, interval=args.interval) + elif mtype == 'afs' or mtype == 'hdfs': + return HadoopMonitor( + args.hadoop_bin, args.fs_name, args.fs_ugi, interval=args.interval) + else: + raise Exception('unsupport type.') + + +def start_monitor(monitor, args): + monitor.set_remote_path(args.remote_path) + monitor.set_remote_model_name(args.remote_model_name) + monitor.set_remote_donefile_name(args.remote_donefile_name) + monitor.set_local_path(args.local_path) + monitor.set_local_model_name(args.local_model_name) + monitor.set_local_timestamp_file(args.local_timestamp_file) + monitor.set_local_tmp_path(args.local_tmp_path) + monitor.set_unpacked_filename(args.unpacked_filename) + monitor.run() + + +if __name__ == "__main__": + args = parse_args() + if args.debug: + logging.basicConfig( + format='%(asctime)s %(levelname)-8s [%(filename)s:%(lineno)d] %(message)s', + datefmt='%Y-%m-%d %H:%M', + level=logging.DEBUG) + else: + logging.basicConfig( + format='%(asctime)s %(levelname)-8s [%(filename)s:%(lineno)d] %(message)s', + datefmt='%Y-%m-%d %H:%M', + level=logging.INFO) + monitor = get_monitor(args.type) + start_monitor(monitor, args) diff --git a/python/paddle_serving_server_gpu/web_service.py b/python/paddle_serving_server_gpu/web_service.py index b4c911b7d805dab376a9709ce1ca5d719ea49794..cb833ba32b20edeb22efd5b772506d32e05e4497 100644 --- a/python/paddle_serving_server_gpu/web_service.py +++ b/python/paddle_serving_server_gpu/web_service.py @@ -11,6 +11,7 @@ # 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. +# pylint: disable=doc-string-missing from flask import Flask, request, abort from paddle_serving_server_gpu import OpMaker, OpSeqMaker, Server @@ -103,13 +104,18 @@ class WebService(object): abort(400) if "fetch" not in request.json: abort(400) - feed, fetch = self.preprocess(request.json, request.json["fetch"]) - fetch_map_batch = self.client.predict(feed=feed, fetch=fetch) - fetch_map_batch = self.postprocess( - feed=request.json, fetch=fetch, fetch_map=fetch_map_batch) - for key in fetch_map_batch: - fetch_map_batch[key] = fetch_map_batch[key].tolist() - result = {"result": fetch_map_batch} + try: + feed, fetch = self.preprocess(request.json, request.json["fetch"]) + if isinstance(feed, dict) and "fetch" in feed: + del feed["fetch"] + fetch_map = self.client.predict(feed=feed, fetch=fetch) + for key in fetch_map: + fetch_map[key] = fetch_map[key][0].tolist() + result = self.postprocess( + feed=request.json, fetch=fetch, fetch_map=fetch_map) + result = {"result": result} + except ValueError: + result = {"result": "Request Value Error"} return result def run_server(self): diff --git a/tools/serving_build.sh b/tools/serving_build.sh index e4bf6ece3a9df1808b9190e9e77d8d2e8aba62c0..1e47b8f4fe26c689b5d6680c1478740201b335b9 100644 --- a/tools/serving_build.sh +++ b/tools/serving_build.sh @@ -323,6 +323,9 @@ function python_test_bert() { echo "bert RPC inference pass" ;; *) + echo "error type" + exit 1 + ;; esac echo "test bert $TYPE finished as expected." unset SERVING_BIN @@ -357,6 +360,9 @@ function python_test_imdb() { echo "imdb ignore GPU test" ;; *) + echo "error type" + exit 1 + ;; esac echo "test imdb $TYPE finished as expected." unset SERVING_BIN @@ -389,6 +395,9 @@ function python_test_lac() { echo "lac ignore GPU test" ;; *) + echo "error type" + exit 1 + ;; esac echo "test lac $TYPE finished as expected." unset SERVING_BIN @@ -408,6 +417,248 @@ function python_run_test() { cd ../.. # pwd: /Serving } +function monitor_test() { + local TYPE=$1 # pwd: /Serving + mkdir _monitor_test && cd _monitor_test # pwd: /Serving/_monitor_test + case $TYPE in + CPU): + pip install pyftpdlib + mkdir remote_path + mkdir local_path + cd remote_path # pwd: /Serving/_monitor_test/remote_path + check_cmd "python -m pyftpdlib -p 8000 &>/dev/null &" + cd .. # pwd: /Serving/_monitor_test + + # type: ftp + # remote_path: / + # remote_model_name: uci_housing.tar.gz + # local_tmp_path: ___tmp + # local_path: local_path + cd remote_path # pwd: /Serving/_monitor_test/remote_path + wget --no-check-certificate https://paddle-serving.bj.bcebos.com/uci_housing.tar.gz + touch donefile + cd .. # pwd: /Serving/_monitor_test + mkdir -p local_path/uci_housing_model + python -m paddle_serving_server.monitor \ + --type='ftp' --ftp_host='127.0.0.1' --ftp_port='8000' \ + --remote_path='/' --remote_model_name='uci_housing.tar.gz' \ + --remote_donefile_name='donefile' --local_path='local_path' \ + --local_model_name='uci_housing_model' --local_timestamp_file='fluid_time_file' \ + --local_tmp_path='___tmp' --unpacked_filename='uci_housing_model' \ + --interval='1' >/dev/null & + sleep 10 + if [ ! -f "local_path/uci_housing_model/fluid_time_file" ]; then + echo "local_path/uci_housing_model/fluid_time_file not exist." + exit 1 + fi + ps -ef | grep "monitor" | grep -v grep | awk '{print $2}' | xargs kill + rm -rf remote_path/* + rm -rf local_path/* + + # type: ftp + # remote_path: /tmp_dir + # remote_model_name: uci_housing_model + # local_tmp_path: ___tmp + # local_path: local_path + mkdir -p remote_path/tmp_dir && cd remote_path/tmp_dir # pwd: /Serving/_monitor_test/remote_path/tmp_dir + wget --no-check-certificate https://paddle-serving.bj.bcebos.com/uci_housing.tar.gz + tar -xzf uci_housing.tar.gz + touch donefile + cd ../.. # pwd: /Serving/_monitor_test + mkdir -p local_path/uci_housing_model + python -m paddle_serving_server.monitor \ + --type='ftp' --ftp_host='127.0.0.1' --ftp_port='8000' \ + --remote_path='/tmp_dir' --remote_model_name='uci_housing_model' \ + --remote_donefile_name='donefile' --local_path='local_path' \ + --local_model_name='uci_housing_model' --local_timestamp_file='fluid_time_file' \ + --local_tmp_path='___tmp' --interval='1' >/dev/null & + sleep 10 + if [ ! -f "local_path/uci_housing_model/fluid_time_file" ]; then + echo "local_path/uci_housing_model/fluid_time_file not exist." + exit 1 + fi + ps -ef | grep "monitor" | grep -v grep | awk '{print $2}' | xargs kill + rm -rf remote_path/* + rm -rf local_path/* + + # type: general + # remote_path: / + # remote_model_name: uci_housing.tar.gz + # local_tmp_path: ___tmp + # local_path: local_path + cd remote_path # pwd: /Serving/_monitor_test/remote_path + wget --no-check-certificate https://paddle-serving.bj.bcebos.com/uci_housing.tar.gz + touch donefile + cd .. # pwd: /Serving/_monitor_test + mkdir -p local_path/uci_housing_model + python -m paddle_serving_server.monitor \ + --type='general' --general_host='ftp://127.0.0.1:8000' \ + --remote_path='/' --remote_model_name='uci_housing.tar.gz' \ + --remote_donefile_name='donefile' --local_path='local_path' \ + --local_model_name='uci_housing_model' --local_timestamp_file='fluid_time_file' \ + --local_tmp_path='___tmp' --unpacked_filename='uci_housing_model' \ + --interval='1' >/dev/null & + sleep 10 + if [ ! -f "local_path/uci_housing_model/fluid_time_file" ]; then + echo "local_path/uci_housing_model/fluid_time_file not exist." + exit 1 + fi + ps -ef | grep "monitor" | grep -v grep | awk '{print $2}' | xargs kill + rm -rf remote_path/* + rm -rf local_path/* + + # type: general + # remote_path: /tmp_dir + # remote_model_name: uci_housing_model + # local_tmp_path: ___tmp + # local_path: local_path + mkdir -p remote_path/tmp_dir && cd remote_path/tmp_dir # pwd: /Serving/_monitor_test/remote_path/tmp_dir + wget --no-check-certificate https://paddle-serving.bj.bcebos.com/uci_housing.tar.gz + tar -xzf uci_housing.tar.gz + touch donefile + cd ../.. # pwd: /Serving/_monitor_test + mkdir -p local_path/uci_housing_model + python -m paddle_serving_server.monitor \ + --type='general' --general_host='ftp://127.0.0.1:8000' \ + --remote_path='/tmp_dir' --remote_model_name='uci_housing_model' \ + --remote_donefile_name='donefile' --local_path='local_path' \ + --local_model_name='uci_housing_model' --local_timestamp_file='fluid_time_file' \ + --local_tmp_path='___tmp' --interval='1' >/dev/null & + sleep 10 + if [ ! -f "local_path/uci_housing_model/fluid_time_file" ]; then + echo "local_path/uci_housing_model/fluid_time_file not exist." + exit 1 + fi + ps -ef | grep "monitor" | grep -v grep | awk '{print $2}' | xargs kill + rm -rf remote_path/* + rm -rf local_path/* + + ps -ef | grep "pyftpdlib" | grep -v grep | awk '{print $2}' | xargs kill + ;; + GPU): + pip install pyftpdlib + mkdir remote_path + mkdir local_path + cd remote_path # pwd: /Serving/_monitor_test/remote_path + check_cmd "python -m pyftpdlib -p 8000 &>/dev/null &" + cd .. # pwd: /Serving/_monitor_test + + # type: ftp + # remote_path: / + # remote_model_name: uci_housing.tar.gz + # local_tmp_path: ___tmp + # local_path: local_path + cd remote_path # pwd: /Serving/_monitor_test/remote_path + wget --no-check-certificate https://paddle-serving.bj.bcebos.com/uci_housing.tar.gz + touch donefile + cd .. # pwd: /Serving/_monitor_test + mkdir -p local_path/uci_housing_model + python -m paddle_serving_server_gpu.monitor \ + --type='ftp' --ftp_host='127.0.0.1' --ftp_port='8000' \ + --remote_path='/' --remote_model_name='uci_housing.tar.gz' \ + --remote_donefile_name='donefile' --local_path='local_path' \ + --local_model_name='uci_housing_model' --local_timestamp_file='fluid_time_file' \ + --local_tmp_path='___tmp' --unpacked_filename='uci_housing_model' \ + --interval='1' >/dev/null & + sleep 10 + if [ ! -f "local_path/uci_housing_model/fluid_time_file" ]; then + echo "local_path/uci_housing_model/fluid_time_file not exist." + exit 1 + fi + ps -ef | grep "monitor" | grep -v grep | awk '{print $2}' | xargs kill + rm -rf remote_path/* + rm -rf local_path/* + + # type: ftp + # remote_path: /tmp_dir + # remote_model_name: uci_housing_model + # local_tmp_path: ___tmp + # local_path: local_path + mkdir -p remote_path/tmp_dir && cd remote_path/tmp_dir # pwd: /Serving/_monitor_test/remote_path/tmp_dir + wget --no-check-certificate https://paddle-serving.bj.bcebos.com/uci_housing.tar.gz + tar -xzf uci_housing.tar.gz + touch donefile + cd ../.. # pwd: /Serving/_monitor_test + mkdir -p local_path/uci_housing_model + python -m paddle_serving_server_gpu.monitor \ + --type='ftp' --ftp_host='127.0.0.1' --ftp_port='8000' \ + --remote_path='/tmp_dir' --remote_model_name='uci_housing_model' \ + --remote_donefile_name='donefile' --local_path='local_path' \ + --local_model_name='uci_housing_model' --local_timestamp_file='fluid_time_file' \ + --local_tmp_path='___tmp' --interval='1' >/dev/null & + sleep 10 + if [ ! -f "local_path/uci_housing_model/fluid_time_file" ]; then + echo "local_path/uci_housing_model/fluid_time_file not exist." + exit 1 + fi + ps -ef | grep "monitor" | grep -v grep | awk '{print $2}' | xargs kill + rm -rf remote_path/* + rm -rf local_path/* + + # type: general + # remote_path: / + # remote_model_name: uci_housing.tar.gz + # local_tmp_path: ___tmp + # local_path: local_path + cd remote_path # pwd: /Serving/_monitor_test/remote_path + wget --no-check-certificate https://paddle-serving.bj.bcebos.com/uci_housing.tar.gz + touch donefile + cd .. # pwd: /Serving/_monitor_test + mkdir -p local_path/uci_housing_model + python -m paddle_serving_server_gpu.monitor \ + --type='general' --general_host='ftp://127.0.0.1:8000' \ + --remote_path='/' --remote_model_name='uci_housing.tar.gz' \ + --remote_donefile_name='donefile' --local_path='local_path' \ + --local_model_name='uci_housing_model' --local_timestamp_file='fluid_time_file' \ + --local_tmp_path='___tmp' --unpacked_filename='uci_housing_model' \ + --interval='1' >/dev/null & + sleep 10 + if [ ! -f "local_path/uci_housing_model/fluid_time_file" ]; then + echo "local_path/uci_housing_model/fluid_time_file not exist." + exit 1 + fi + ps -ef | grep "monitor" | grep -v grep | awk '{print $2}' | xargs kill + rm -rf remote_path/* + rm -rf local_path/* + + # type: general + # remote_path: /tmp_dir + # remote_model_name: uci_housing_model + # local_tmp_path: ___tmp + # local_path: local_path + mkdir -p remote_path/tmp_dir && cd remote_path/tmp_dir # pwd: /Serving/_monitor_test/remote_path/tmp_dir + wget --no-check-certificate https://paddle-serving.bj.bcebos.com/uci_housing.tar.gz + tar -xzf uci_housing.tar.gz + touch donefile + cd ../.. # pwd: /Serving/_monitor_test + mkdir -p local_path/uci_housing_model + python -m paddle_serving_server_gpu.monitor \ + --type='general' --general_host='ftp://127.0.0.1:8000' \ + --remote_path='/tmp_dir' --remote_model_name='uci_housing_model' \ + --remote_donefile_name='donefile' --local_path='local_path' \ + --local_model_name='uci_housing_model' --local_timestamp_file='fluid_time_file' \ + --local_tmp_path='___tmp' --interval='1' >/dev/null & + sleep 10 + if [ ! -f "local_path/uci_housing_model/fluid_time_file" ]; then + echo "local_path/uci_housing_model/fluid_time_file not exist." + exit 1 + fi + ps -ef | grep "monitor" | grep -v grep | awk '{print $2}' | xargs kill + rm -rf remote_path/* + rm -rf local_path/* + + ps -ef | grep "pyftpdlib" | grep -v grep | awk '{print $2}' | xargs kill + ;; + *) + echo "error type" + exit 1 + ;; + esac + cd .. # pwd: /Serving + rm -rf _monitor_test + echo "test monitor $TYPE finished as expected." +} + function main() { local TYPE=$1 # pwd: / init # pwd: /Serving @@ -415,6 +666,7 @@ function main() { build_server $TYPE # pwd: /Serving build_app $TYPE # pwd: /Serving python_run_test $TYPE # pwd: /Serving + monitor_test $TYPE # pwd: /Serving echo "serving $TYPE part finished as expected." }