Merge branch 'new_orm' into jinhai

Former-commit-id: 3cc778a3f03cd9d71f91e0bb8b12f16f0aea7fbc

cpp/conf/server_config.yaml

@@ -5,6 +5,7 @@ server_config:
   server_mode: thread_pool    #optional: simple, thread_pool
   db_path: /tmp/vecwise
   db_backend_url: http://127.0.0.1
+  db_flush_interval: 10   #unit: second
 
 log_config:
   global:

cpp/src/server/ServerConfig.h

@@ -21,6 +21,7 @@ static const std::string CONFIG_SERVER_PROTOCOL = "transfer_protocol";
 static const std::string CONFIG_SERVER_MODE = "server_mode";
 static const std::string CONFIG_SERVER_DB_URL = "db_backend_url";
 static const std::string CONFIG_SERVER_DB_PATH = "db_path";
+static const std::string CONFIG_SERVER_DB_FLUSH_INTERVAL = "db_flush_interval";
 
 static const std::string CONFIG_LOG = "log_config";
 

cpp/src/server/VecServiceTask.cpp

@@ -27,6 +27,7 @@ namespace {
             ConfigNode& config = ServerConfig::GetInstance().GetConfig(CONFIG_SERVER);
             opt.meta.backend_uri = config.GetValue(CONFIG_SERVER_DB_URL);
             std::string db_path = config.GetValue(CONFIG_SERVER_DB_PATH);
+            opt.memory_sync_interval = (uint16_t)config.GetInt32Value(CONFIG_SERVER_DB_FLUSH_INTERVAL, 10);
             opt.meta.path = db_path + "/db";
 
             CommonUtil::CreateDirectory(opt.meta.path);
@@ -155,9 +156,29 @@ BaseTaskPtr AddSingleVectorTask::Create(const std::string& group_id,
 
 ServerError AddSingleVectorTask::OnExecute() {
     try {
+        engine::meta::GroupSchema group_info;
+        group_info.group_id = group_id_;
+        engine::Status stat = DB()->get_group(group_info);
+        if(!stat.ok()) {
+            SERVER_LOG_ERROR << "Engine failed: " << stat.ToString();
+            return SERVER_INVALID_ARGUMENT;
+        }
+
+        uint64_t vec_dim = group_info.dimension;
+        if(vec_dim != tensor_.tensor.size()) {
+            SERVER_LOG_ERROR << "Invalid vector dimension: " << tensor_.tensor.size()
+                             << " vs. group dimension:" << vec_dim;
+            return SERVER_INVALID_ARGUMENT;
+        }
+
+        std::vector<float> vec_f;
+        vec_f.resize(vec_dim);
+        for(uint64_t d = 0; d < vec_dim; d++) {
+            vec_f[d] = (float)(tensor_.tensor[d]);
+        }
+
         engine::IDNumbers vector_ids;
-        std::vector<float> vec_f(tensor_.tensor.begin(), tensor_.tensor.end());
-        engine::Status stat = DB()->add_vectors(group_id_, 1, vec_f.data(), vector_ids);
+        stat = DB()->add_vectors(group_id_, 1, vec_f.data(), vector_ids);
         if(!stat.ok()) {
             SERVER_LOG_ERROR << "Engine failed: " << stat.ToString();
             return SERVER_UNEXPECTED_ERROR;
@@ -213,10 +234,10 @@ ServerError AddBatchVectorTask::OnExecute() {
         vec_f.resize(vec_count*vec_dim);//allocate enough memory
         for(uint64_t i = 0; i < vec_count; i ++) {
             const std::vector<double>& tensor = tensor_list_.tensor_list[i].tensor;
-            if(tensor.size() != group_info.dimension) {
-                SERVER_LOG_ERROR << "Invalid vector data size: " << tensor.size()
-                                 << " vs. group dimension:" << group_info.dimension;
-                return SERVER_UNEXPECTED_ERROR;
+            if(tensor.size() != vec_dim) {
+                SERVER_LOG_ERROR << "Invalid vector dimension: " << tensor.size()
+                                 << " vs. group dimension:" << vec_dim;
+                return SERVER_INVALID_ARGUMENT;
             }
 
             for(uint64_t d = 0; d < vec_dim; d++) {

cpp/test_client/src/ClientApp.cpp

@@ -41,6 +41,7 @@ void ClientApp::Run(const std::string &config_file) {
     int32_t port = server_config.GetInt32Value(server::CONFIG_SERVER_PORT, 33001);
     std::string protocol = server_config.GetValue(server::CONFIG_SERVER_PROTOCOL, "binary");
     //std::string mode = server_config.GetValue(server::CONFIG_SERVER_MODE, "thread_pool");
+    int32_t flush_interval = server_config.GetInt32Value(server::CONFIG_SERVER_DB_FLUSH_INTERVAL);
 
     CLIENT_LOG_INFO << "Connect to server: " << address << ":" << std::to_string(port);
 
@@ -65,7 +66,7 @@ void ClientApp::Run(const std::string &config_file) {
                 for (int32_t i = 0; i < dim; i++) {
                     tensor.tensor.push_back((double) (i + k));
                 }
-                tensor.uid = "vec_" + std::to_string(k);
+                tensor.uid = "s_vec_" + std::to_string(k);
 
                 session.interface()->add_vector(group.id, tensor);
 
@@ -83,14 +84,14 @@ void ClientApp::Run(const std::string &config_file) {
                 for (int32_t i = 0; i < dim; i++) {
                     tensor.tensor.push_back((double) (i + k));
                 }
-                tensor.uid = "vec_" + std::to_string(k);
+                tensor.uid = "m_vec_" + std::to_string(k);
                 vec_list.tensor_list.push_back(tensor);
             }
             session.interface()->add_vector_batch(group.id, vec_list);
             rc.Elapse("done!");
         }
 
-        sleep(20);
+        sleep(flush_interval);
         
         //search vector
         {