Merge branch 'branch-0.5.0-yk' into 'branch-0.5.0'

MS-639 SQ8H index created failed and server hang

See merge request megasearch/milvus!704

Former-commit-id: 7ea975cd9cb71317c70b07471cc0c79bc9723c30

cpp/CHANGELOG.md

@@ -16,6 +16,7 @@ Please mark all change in change log and use the ticket from JIRA.
 - MS-637 - out of memory when load too many tasks
 - MS-640 - Cache object size calculate incorrect
 - MS-641 - Segment fault(signal 11) in PickToLoad
+- MS-639 - SQ8H index created failed and server hang
 
 ## Improvement
 - MS-552 - Add and change the easylogging library

cpp/src/scheduler/BuildMgr.cpp

@@ -0,0 +1,22 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you 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.
+
+#include "scheduler/BuildMgr.h"
+
+namespace milvus {
+namespace scheduler {}  // namespace scheduler
+}  // namespace milvus

cpp/src/scheduler/BuildMgr.h

@@ -0,0 +1,63 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you 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.
+
+#pragma once
+
+#include <atomic>
+#include <condition_variable>
+#include <deque>
+#include <list>
+#include <memory>
+#include <mutex>
+#include <queue>
+#include <string>
+#include <thread>
+#include <unordered_map>
+#include <vector>
+
+namespace milvus {
+namespace scheduler {
+
+class BuildMgr {
+ public:
+    explicit BuildMgr(int64_t numoftasks) : numoftasks_(numoftasks) {
+    }
+
+ public:
+    void
+    Put() {
+        ++numoftasks_;
+    }
+
+    void
+    take() {
+        --numoftasks_;
+    }
+
+    int64_t
+    numoftasks() {
+        return (int64_t)numoftasks_;
+    }
+
+ private:
+    std::atomic_long numoftasks_;
+};
+
+using BuildMgrPtr = std::shared_ptr<BuildMgr>;
+
+}  // namespace scheduler
+}  // namespace milvus

cpp/src/scheduler/SchedInst.cpp

@@ -41,6 +41,9 @@ std::mutex JobMgrInst::mutex_;
 OptimizerPtr OptimizerInst::instance = nullptr;
 std::mutex OptimizerInst::mutex_;
 
+BuildMgrPtr BuildMgrInst::instance = nullptr;
+std::mutex BuildMgrInst::mutex_;
+
 void
 load_simple_config() {
     server::Config& config = server::Config::GetInstance();

cpp/src/scheduler/SchedInst.h

@@ -17,6 +17,7 @@
 
 #pragma once
 
+#include "BuildMgr.h"
 #include "JobMgr.h"
 #include "ResourceMgr.h"
 #include "Scheduler.h"
@@ -106,6 +107,24 @@ class OptimizerInst {
     static std::mutex mutex_;
 };
 
+class BuildMgrInst {
+ public:
+    static BuildMgrPtr
+    GetInstance() {
+        if (instance == nullptr) {
+            std::lock_guard<std::mutex> lock(mutex_);
+            if (instance == nullptr) {
+                instance = std::make_shared<BuildMgr>(4);
+            }
+        }
+        return instance;
+    }
+
+ private:
+    static BuildMgrPtr instance;
+    static std::mutex mutex_;
+};
+
 void
 StartSchedulerService();
 

cpp/src/scheduler/TaskTable.cpp

@@ -18,6 +18,7 @@
 #include "scheduler/TaskTable.h"
 #include "Utils.h"
 #include "event/TaskTableUpdatedEvent.h"
+#include "scheduler/SchedInst.h"
 #include "utils/Log.h"
 
 #include <ctime>
@@ -164,6 +165,13 @@ TaskTable::PickToLoad(uint64_t limit) {
         if (not table_[j]) {
             SERVER_LOG_WARNING << "table[" << j << "] is nullptr";
         }
+
+        if (table_[j]->task->path().Current() == "cpu") {
+            if (table_[j]->task->Type() == TaskType::BuildIndexTask && BuildMgrInst::GetInstance()->numoftasks() < 1) {
+                return std::vector<uint64_t>();
+            }
+        }
+
         if (table_[j]->state == TaskTableItemState::LOADED) {
             ++count;
             if (count > 2)
@@ -177,9 +185,21 @@ TaskTable::PickToLoad(uint64_t limit) {
         if (not cross && table_[i]->IsFinish()) {
             last_finish_ = i;
         } else if (table_[i]->state == TaskTableItemState::START) {
-            cross = true;
-            indexes.push_back(i);
-            ++count;
+            auto task = table_[i]->task;
+            if (task->Type() == TaskType::BuildIndexTask && task->path().Current() == "cpu") {
+                if (BuildMgrInst::GetInstance()->numoftasks() == 0) {
+                    break;
+                } else {
+                    cross = true;
+                    indexes.push_back(i);
+                    ++count;
+                    BuildMgrInst::GetInstance()->take();
+                }
+            } else {
+                cross = true;
+                indexes.push_back(i);
+                ++count;
+            }
         }
     }
     return indexes;

cpp/src/scheduler/resource/Resource.cpp

@@ -16,6 +16,7 @@
 // under the License.
 
 #include "scheduler/resource/Resource.h"
+#include "scheduler/SchedInst.h"
 #include "scheduler/Utils.h"
 
 #include <iostream>
@@ -175,6 +176,12 @@ Resource::executor_function() {
             total_cost_ += finish - start;
 
             task_item->Executed();
+
+            if (task_item->task->Type() == TaskType::BuildIndexTask) {
+                BuildMgrInst::GetInstance()->Put();
+                ResMgrInst::GetInstance()->GetResource("cpu")->WakeupLoader();
+            }
+
             if (subscriber_) {
                 auto event = std::make_shared<FinishTaskEvent>(shared_from_this(), task_item);
                 subscriber_(std::static_pointer_cast<Event>(event));

cpp/src/scheduler/task/Path.h

@@ -40,13 +40,22 @@ class Path {
         return path_;
     }
 
+    std::string
+    Current() {
+        if (!path_.empty() && path_.size() > index_) {
+            return path_[index_];
+        } else {
+            return "";
+        }
+    }
+
     std::string
     Next() {
         if (index_ > 0 && !path_.empty()) {
             --index_;
             return path_[index_];
         } else {
-            return nullptr;
+            return "";
         }
     }