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

Fix bug in OnCopyCompleted and Add DumpTaskTables in sdk

See merge request megasearch/milvus!449

Former-commit-id: 3837ff45b43f246a3b411fdce082744157ef3b87

cpp/CHANGELOG.md

@@ -66,6 +66,7 @@ Please mark all change in change log and use the ticket from JIRA.
 - MS-421 - Add TaskLabel in scheduler
 - MS-422 - Support DeleteTask in Multi-GpuResource case
 - MS-428 - Add PushTaskByDataLocality in scheduler
+- MS-440 - Add DumpTaskTables in sdk
 
 ## New Feature
 - MS-343 - Implement ResourceMgr

cpp/conf/server_config.template

@@ -70,15 +70,15 @@ resource_config:
       type: GPU
       memory: 6
       device_id: 0
-      enable_loader: false
-      enable_executor: false
+      enable_loader: true
+      enable_executor: true
 
     gtx1660:
       type: GPU
       memory: 6
       device_id: 1
-      enable_loader: false
-      enable_executor: false
+      enable_loader: true
+      enable_executor: true
 
   # connection list, length: 0~N
   # format: -${resource_name}===${resource_name}

cpp/src/scheduler/Scheduler.cpp

@@ -120,15 +120,20 @@ Scheduler::OnCopyCompleted(const EventPtr &event) {
                     auto task = load_completed_event->task_table_item_->task;
                     auto search_task = std::static_pointer_cast<XSearchTask>(task);
                     auto location = search_task->index_engine_->GetLocation();
+                    bool moved = false;
 
                     for (auto i = 0; i < res_mgr_.lock()->GetNumGpuResource(); ++i) {
                         auto index = zilliz::milvus::cache::GpuCacheMgr::GetInstance(i)->GetIndex(location);
                         if (index != nullptr) {
+                            moved = true;
                             auto dest_resource = res_mgr_.lock()->GetResource(ResourceType::GPU, i);
                             Action::PushTaskToResource(load_completed_event->task_table_item_->task, dest_resource);
+                            break;
                         }
                     }
-
+                    if (not moved) {
+                        Action::PushTaskToNeighbourRandomly(task, resource);
+                    }
                 }
                 break;
             }

cpp/src/scheduler/action/PushTaskToNeighbour.cpp

@@ -21,6 +21,8 @@ get_neighbours(const ResourcePtr &self) {
         if (not node) continue;
 
         auto resource = std::static_pointer_cast<Resource>(node);
+//        if (not resource->HasExecutor()) continue;
+
         neighbours.emplace_back(resource);
     }
     return neighbours;
@@ -31,11 +33,16 @@ void
 Action::PushTaskToNeighbourRandomly(const TaskPtr &task,
                                     const ResourcePtr &self) {
     auto neighbours = get_neighbours(self);
-    std::random_device rd;
-    std::mt19937 mt(rd());
-    std::uniform_int_distribution<uint64_t> dist(0, neighbours.size() - 1);
+    if (not neighbours.empty()) {
+        std::random_device rd;
+        std::mt19937 mt(rd());
+        std::uniform_int_distribution<uint64_t> dist(0, neighbours.size() - 1);
+
+        neighbours[dist(mt)]->task_table().Put(task);
+    } else {
+        //TODO: process
+    }
 
-    neighbours[dist(mt)]->task_table().Put(task);
 }
 
 void

cpp/src/sdk/examples/grpcsimple/src/ClientTest.cpp

@@ -23,11 +23,11 @@ namespace {
     const std::string TABLE_NAME = GetTableName();
     constexpr int64_t TABLE_DIMENSION = 512;
     constexpr int64_t TABLE_INDEX_FILE_SIZE = 768;
-    constexpr int64_t BATCH_ROW_COUNT = 100000;
-    constexpr int64_t NQ = 10;
+    constexpr int64_t BATCH_ROW_COUNT = 1000000;
+    constexpr int64_t NQ = 100;
     constexpr int64_t TOP_K = 10;
     constexpr int64_t SEARCH_TARGET = 5000; //change this value, result is different
-    constexpr int64_t ADD_VECTOR_LOOP = 10;
+    constexpr int64_t ADD_VECTOR_LOOP = 1;
     constexpr int64_t SECONDS_EACH_HOUR = 3600;
 
 #define BLOCK_SPLITER std::cout << "===========================================" << std::endl;
@@ -172,15 +172,20 @@ namespace {
             record_array.push_back(pair.second);
         }
 
-        std::vector<TopKQueryResult> topk_query_result_array;
-        {
-            TimeRecorder rc(phase_name);
-            Status stat = conn->Search(TABLE_NAME, record_array, query_range_array, TOP_K, 10, topk_query_result_array);
-            std::cout << "SearchVector function call status: " << stat.ToString() << std::endl;
+        auto start = std::chrono::high_resolution_clock::now();
+        for (auto i = 0; i < 5; ++i) {
+            std::vector<TopKQueryResult> topk_query_result_array;
+            {
+                TimeRecorder rc(phase_name);
+                Status stat = conn->Search(TABLE_NAME, record_array, query_range_array, TOP_K, 32, topk_query_result_array);
+                std::cout << "SearchVector function call status: " << stat.ToString() << std::endl;
+            }
         }
+        auto finish = std::chrono::high_resolution_clock::now();
+        std::cout << "SEARCHVECTOR COST: " << std::chrono::duration_cast<std::chrono::duration<double>>(finish - start).count() << "s\n";
 
-        PrintSearchResult(search_record_array, topk_query_result_array);
-        CheckResult(search_record_array, topk_query_result_array);
+//        PrintSearchResult(search_record_array, topk_query_result_array);
+//        CheckResult(search_record_array, topk_query_result_array);
     }
 }
 
@@ -287,7 +292,7 @@ ClientTest::Test(const std::string& address, const std::string& port) {
         IndexParam index;
         index.table_name = TABLE_NAME;
         index.index_type = IndexType::gpu_ivfflat;
-        index.nlist = 1000;
+        index.nlist = 16384;
         index.metric_type = 1;
         Status stat = conn->CreateIndex(index);
         std::cout << "CreateIndex function call status: " << stat.ToString() << std::endl;
@@ -304,6 +309,7 @@ ClientTest::Test(const std::string& address, const std::string& port) {
 
     {//search vectors after build index finish
         DoSearch(conn, search_record_array, "Search after build index finish");
+//        std::cout << conn->DumpTaskTables() << std::endl;
     }
 
     {//delete index

cpp/src/sdk/grpc/ClientProxy.cpp

@@ -329,6 +329,21 @@ ClientProxy::ServerStatus() const {
     }
 }
 
+std::string
+ClientProxy::DumpTaskTables() const {
+    if (channel_ == nullptr) {
+        return "not connected to server";
+    }
+
+    try {
+        std::string dummy;
+        Status status = client_ptr_->Cmd(dummy, "tasktable");
+        return dummy;
+    } catch (std::exception &ex) {
+        return "connection lost";
+    }
+}
+
 Status
 ClientProxy::DeleteByRange(milvus::Range &range, const std::string &table_name) {
     try {

cpp/src/sdk/grpc/ClientProxy.h

@@ -68,6 +68,9 @@ public:
     virtual std::string
     ServerStatus() const override;
 
+    virtual std::string
+    DumpTaskTables() const override;
+
     virtual Status
     DeleteByRange(Range &range,
                   const std::string &table_name) override;

cpp/src/sdk/include/MilvusApi.h

@@ -318,6 +318,9 @@ class Connection {
     virtual std::string
     ServerStatus() const = 0;
 
+    virtual std::string
+    DumpTaskTables() const = 0;
+
     /**
      * @brief delete tables by range
      *

cpp/src/sdk/interface/ConnectionImpl.cpp

@@ -114,6 +114,11 @@ ConnectionImpl::ServerStatus() const {
     return client_proxy_->ServerStatus();
 }
 
+std::string
+ConnectionImpl::DumpTaskTables() const {
+    return client_proxy_->DumpTaskTables();
+}
+
 Status
 ConnectionImpl::DeleteByRange(Range &range,
               const std::string &table_name) {

cpp/src/sdk/interface/ConnectionImpl.h

@@ -70,6 +70,9 @@ public:
     virtual std::string
     ServerStatus() const override;
 
+    virtual std::string
+    DumpTaskTables() const override;
+
     virtual Status
     DeleteByRange(Range &range,
                   const std::string &table_name) override;

cpp/unittest/db/utils.cpp

@@ -64,9 +64,12 @@ void DBTest::SetUp() {
     res_mgr->Clear();
     res_mgr->Add(engine::ResourceFactory::Create("disk", "DISK", 0, true, false));
     res_mgr->Add(engine::ResourceFactory::Create("cpu", "CPU", 0, true, true));
+    res_mgr->Add(engine::ResourceFactory::Create("gtx1660", "GPU", 0, true, true));
 
     auto default_conn = engine::Connection("IO", 500.0);
+    auto PCIE = engine::Connection("IO", 11000.0);
     res_mgr->Connect("disk", "cpu", default_conn);
+    res_mgr->Connect("cpu", "gtx1660", PCIE);
     res_mgr->Start();
     engine::SchedInst::GetInstance()->Start();