fix conversation of PR296

core/cube/cube-api/include/cube_api.h

@@ -99,6 +99,13 @@ class CubeAPI {
                std::function<void(DictValue*, size_t)> parse,
                std::string* version);
  
+  /**
+   * @brief: get all table names from cube server, thread safe.
+   * @param [out] vals: vector of table names
+   *
+   */ 
+  std::vector<std::string>  get_table_names();  
+
  public:
   static const char* error_msg(int error_code);
 

core/cube/cube-api/src/cube_api.cpp

@@ -682,5 +682,13 @@ int CubeAPI::opt_seek(const std::string& dict_name,
   return ret;
 }
 
+std::vector<std::string> CubeAPI::get_table_names() {
+  const std::vector<const MetaInfo*> metas = _meta->metas();
+  std::vector<std::string> table_names;
+  for (auto itr = metas.begin(); itr != metas.end(); ++itr) {
+    table_names.push_back((*itr)->dict_name);
+  }
+  return table_names;
+}
 }  // namespace mcube
 }  // namespace rec

core/general-server/op/general_dist_kv_infer_op.cpp

@@ -22,6 +22,7 @@
 #include "core/predictor/framework/memory.h"
 #include "core/predictor/framework/resource.h"
 #include "core/util/include/timer.h"
+#include <utility>
 
 namespace baidu {
 namespace paddle_serving {
@@ -55,6 +56,8 @@ int GeneralDistKVInferOp::inference() {
   std::vector<rec::mcube::CubeValue> values;
   int sparse_count = 0;
   int dense_count = 0;
+  std::vector<std::pair<int64_t*, size_t>> dataptr_size_pairs;
+  size_t key_len = 0;
   for (size_t i = 0; i < in->size(); ++i) {
     if (in->at(i).dtype != paddle::PaddleDType::INT64) {
       ++dense_count;
@@ -65,22 +68,29 @@ int GeneralDistKVInferOp::inference() {
     for (size_t s = 0; s < in->at(i).shape.size(); ++s) {
       elem_num *= in->at(i).shape[s];
     }
+    key_len += elem_num;
     int64_t *data_ptr = static_cast<int64_t *>(in->at(i).data.data());
-    for (size_t j = 0; j < elem_num; ++j) {
-      keys.push_back(data_ptr[j]);
+    dataptr_size_pairs.push_back(std::make_pair(data_ptr, elem_num));
   }
+  keys.resize(key_len);
+  int key_idx = 0;
+  for (size_t i = 0; i < dataptr_size_pairs.size(); ++i) {
+     std::copy(dataptr_size_pairs[i].first, dataptr_size_pairs[i].first +  dataptr_size_pairs[i].second, keys.begin() + key_idx);
+     key_idx += dataptr_size_pairs[i].second;
   }
-  // TODO: Add Seek CubeValues Here, and replace EMBEDDING_SIZE with variable.
   rec::mcube::CubeAPI *cube = rec::mcube::CubeAPI::instance();
   // TODO: temp hard code "test_dict" here, fix this with next commit
   // related to cube conf
-  std::string table_name = "test_dict";
-  int ret = cube->seek(table_name, keys, &values);
+  std::vector<std::string> table_names = cube->get_table_names();
+  if (table_names.size() == 0) {
+    LOG(ERROR) << "cube init error or cube config not given.";
+  }
+  int ret = cube->seek(table_names[0], keys, &values);
 
   if (values.size() != keys.size() || values[0].buff.size() == 0) {
     LOG(ERROR) << "cube value return null";
   }
-  size_t EMBEDDING_SIZE = values[0].buff.size() / 4;
+  size_t EMBEDDING_SIZE = values[0].buff.size() / sizeof(float);
   TensorVector sparse_out;
   sparse_out.resize(sparse_count);
   TensorVector dense_out;

core/general-server/op/general_dist_kv_infer_op.h

-// Copyright (c) 2019 PaddlePaddle Authors. All Rights Reserved.
+// 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.

core/predictor/CMakeLists.txt

@@ -3,7 +3,7 @@ include(common/CMakeLists.txt)
 include(op/CMakeLists.txt)
 include(mempool/CMakeLists.txt)
 include(framework/CMakeLists.txt)
-#include(plugin/CMakeLists.txt)
+include(tools/CMakeLists.txt)
 include(src/CMakeLists.txt)
 
 include_directories(SYSTEM  ${CMAKE_CURRENT_LIST_DIR}/../kvdb/include)

core/predictor/common/CMakeLists.txt

 FILE(GLOB common_srcs ${CMAKE_CURRENT_LIST_DIR}/constant.cpp)
 LIST(APPEND pdserving_srcs ${common_srcs})
 
-set(seq_gen_src ${CMAKE_CURRENT_LIST_DIR}/seq_generator.cpp  ${CMAKE_CURRENT_LIST_DIR}/seq_file.cpp)
-LIST(APPEND seq_gen_src ${PROTO_SRCS})
-add_executable(seq_generator ${seq_gen_src})
-target_link_libraries(seq_generator protobuf -lpthread)

core/predictor/framework/resource.cpp

@@ -170,7 +170,11 @@ int Resource::general_model_initialize(const std::string& path,
   if (this->cube_config_fullpath.size() != 0) {
     LOG(INFO) << "init cube by config file : " << this->cube_config_fullpath;
     rec::mcube::CubeAPI* cube = rec::mcube::CubeAPI::instance();
-    cube->init(this->cube_config_fullpath.c_str());
+    int ret = cube->init(this->cube_config_fullpath.c_str());
+    if (ret != 0) {
+      LOG(ERROR) << "cube init error";
+      return -1;
+    }
   }
   VLOG(2) << "general model path: " << path;
   VLOG(2) << "general model file: " << file;

core/predictor/tools/CMakeLists.txt

+set(seq_gen_src ${CMAKE_CURRENT_LIST_DIR}/seq_generator.cpp  ${CMAKE_CURRENT_LIST_DIR}/seq_file.cpp)
+LIST(APPEND seq_gen_src ${PROTO_SRCS})
+add_executable(seq_generator ${seq_gen_src})
+target_link_libraries(seq_generator protobuf -lpthread)

core/predictor/common/seq_file.cpp → core/predictor/tools/seq_file.cpp

-// Copyright (c) 2019 PaddlePaddle Authors. All Rights Reserved.
+// 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.

core/predictor/common/seq_file.h → core/predictor/tools/seq_file.h

-// Copyright (c) 2019 PaddlePaddle Authors. All Rights Reserved.
+// 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.

core/predictor/common/seq_generator.cpp → core/predictor/tools/seq_generator.cpp

-// Copyright (c) 2019 PaddlePaddle Authors. All Rights Reserved.
+// 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.

python/examples/criteo_ctr_with_cube/cube/conf/transfer.conf

-[default]
-dict_name: test_dict
-mode: base_only
-download_mode: http
-wget_port: 8098
-buildtool_local: /Serving/python/examples/criteo_ctr/cube/cube-builder
-donefile_address: http://127.0.0.1:8080/output/ctr_cube/donefile
-output_address: /Serving/python/examples/criteo_ctr/cube/output
-tmp_address: /Serving/python/examples/criteo_ctr/cube/output
-shard_num: 1
-copy_num: 1
-deploy_path: /Serving/python/examples/criteo_ctr/cube/test_dict
-transfer_address: 127.0.0.1
-
-[cube_agent]
-agent0_0: 127.0.0.1:8001
-cube0_0: 127.0.0.1:8027:/Serving/python/examples/criteo_ctr/cube

python/examples/criteo_ctr_with_cube/cube_prepare.sh

@@ -2,6 +2,6 @@
 mkdir -p cube_model
 mkdir -p cube/data
 ./seq_generator ctr_serving_model/SparseFeatFactors ./cube_model/feature
-/Serving/python/examples/criteo_ctr/cube/cube-builder -dict_name=test_dict -job_mode=base -last_version=0 -cur_version=0 -depend_version=0 -input_path=/Serving/python/examples/criteo_ctr/cube_model -output_path=/Serving/python/examples/criteo_ctr/cube/data -shard_num=1  -only_build=false
-mv /Serving/python/examples/criteo_ctr/cube/data/0_0/test_dict_part0/* /Serving/python/examples/criteo_ctr/cube/data/
+./cube/cube-builder -dict_name=test_dict -job_mode=base -last_version=0 -cur_version=0 -depend_version=0 -input_path=./cube_model -output_path=./cube/data -shard_num=1  -only_build=false
+mv ./cube/data/0_0/test_dict_part0/* ./cube/data/
 cd cube && ./cube 

python/examples/criteo_ctr_with_cube/local_train.py

@@ -66,7 +66,7 @@ def train():
     ]
 
     print(whole_filelist)
-    dataset.set_filelist(whole_filelist[:thread_num])
+    dataset.set_filelist(whole_filelist[:100])
     dataset.load_into_memory()
     fluid.layers.Print(auc_var)
     epochs = 1

python/examples/criteo_ctr_with_cube/test_client.py

@@ -29,15 +29,14 @@ buf_size = 100
 dataset = criteo.CriteoDataset()
 dataset.setup(1000001)
 test_filelists = [
-    "{}/part-%d".format(sys.argv[2]) % x
-    for x in range(len(os.listdir(sys.argv[2])))
+    "{}/part-test".format(sys.argv[2])
 ]
-reader = dataset.infer_reader(test_filelists[len(test_filelists) - 40:], batch,
+reader = dataset.infer_reader(test_filelists, batch,
                               buf_size)
 label_list = []
 prob_list = []
 start = time.time()
-for ei in range(1000):
+for ei in range(10000):
     data = reader().next()
     feed_dict = {}
     feed_dict['dense_input'] = data[0][0]