Refactor optimizer

core/src/scheduler/SchedInst.h

@@ -24,9 +24,10 @@
 #include "Utils.h"
 #include "optimizer/BuildIndexPass.h"
 #include "optimizer/FallbackPass.h"
-#include "optimizer/HybridPass.h"
-#include "optimizer/LargeSQ8HPass.h"
-#include "optimizer/OnlyCPUPass.h"
+#include "optimizer/FaissFlatPass.h"
+#include "optimizer/FaissIVFFlatPass.h"
+#include "optimizer/FaissIVFSQ8Pass.h"
+#include "optimizer/FaissIVFSQ8HPass.h"
 #include "optimizer/Optimizer.h"
 #include "server/Config.h"
 
@@ -100,15 +101,12 @@ class OptimizerInst {
             std::lock_guard<std::mutex> lock(mutex_);
             if (instance == nullptr) {
                 std::vector<PassPtr> pass_list;
-                pass_list.push_back(std::make_shared<LargeSQ8HPass>());
-                pass_list.push_back(std::make_shared<HybridPass>());
-#ifdef MILVUS_CPU_VERSION
-                pass_list.push_back(std::make_shared<OnlyCPUPass>());
-#else
-                server::Config& config = server::Config::GetInstance();
-                std::vector<int32_t> build_resources;
-                config.GetGpuResourceConfigBuildIndexResources(build_resources);
-                pass_list.push_back(std::make_shared<BuildIndexPass>(build_resources));
+#ifdef MILVUS_GPU_VERSION
+                pass_list.push_back(std::make_shared<BuildIndexPass>());
+                pass_list.push_back(std::make_shared<FaissFlatPass>());
+                pass_list.push_back(std::make_shared<FaissIVFFlatPass>());
+                pass_list.push_back(std::make_shared<FaissIVFSQ8Pass>());
+                pass_list.push_back(std::make_shared<FaissIVFSQ8HPass>());
 #endif
                 pass_list.push_back(std::make_shared<FallbackPass>());
                 instance = std::make_shared<Optimizer>(pass_list);

core/src/scheduler/optimizer/BuildIndexPass.cpp

@@ -23,11 +23,14 @@
 namespace milvus {
 namespace scheduler {
 
-BuildIndexPass::BuildIndexPass(std::vector<int32_t>& build_gpu_ids) : build_gpu_ids_(build_gpu_ids) {
-}
-
 void
 BuildIndexPass::Init() {
+    server::Config& config = server::Config::GetInstance();
+    std::vector<int32_t> build_resources;
+    Status s = config.GetGpuResourceConfigBuildIndexResources(build_resources);
+    if (!s.ok()) {
+        throw;
+    }
 }
 
 bool

core/src/scheduler/optimizer/BuildIndexPass.h

@@ -34,7 +34,7 @@ namespace scheduler {
 
 class BuildIndexPass : public Pass {
  public:
-    explicit BuildIndexPass(std::vector<int32_t>& build_gpu_id);
+    BuildIndexPass() = default;
 
  public:
     void

core/src/scheduler/optimizer/FaissFlatPass.cpp

@@ -0,0 +1,69 @@
+// 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/optimizer/FaissFlatPass.h"
+#include "cache/GpuCacheMgr.h"
+#include "scheduler/SchedInst.h"
+#include "scheduler/Utils.h"
+#include "scheduler/task/SearchTask.h"
+#include "scheduler/tasklabel/SpecResLabel.h"
+#include "server/Config.h"
+#include "utils/Log.h"
+
+namespace milvus {
+namespace scheduler {
+
+void
+FaissFlatPass::Init() {
+    server::Config& config = server::Config::GetInstance();
+    Status s = config.GetEngineConfigGpuSearchThreshold(threshold_);
+    if (!s.ok()) {
+        threshold_ = std::numeric_limits<int32_t>::max();
+    }
+    s = config.GetGpuResourceConfigSearchResources(gpus);
+    if (!s.ok()) {
+        throw;
+    }
+}
+
+bool
+FaissFlatPass::Run(const TaskPtr& task) {
+    if (task->Type() != TaskType::SearchTask) {
+        return false;
+    }
+
+    auto search_task = std::static_pointer_cast<XSearchTask>(task);
+    if (search_task->file_->engine_type_ != (int)engine::EngineType::FAISS_IDMAP) {
+        return false;
+    }
+
+    auto search_job = std::static_pointer_cast<SearchJob>(search_task->job_.lock());
+    ResourcePtr res_ptr;
+    if (search_job->nq() < threshold_) {
+        res_ptr = ResMgrInst::GetInstance()->GetResource("cpu");
+    } else {
+        auto best_device_id = count_ % gpus.size();
+        count_++;
+        res_ptr = ResMgrInst::GetInstance()->GetResource(ResourceType::GPU, best_device_id);
+    }
+    auto label = std::make_shared<SpecResLabel>(res_ptr);
+    task->label() = label;
+    return true;
+}
+
+}  // namespace scheduler
+}  // namespace milvus

core/src/scheduler/optimizer/FaissFlatPass.h

@@ -33,9 +33,9 @@
 namespace milvus {
 namespace scheduler {
 
-class LargeSQ8HPass : public Pass {
+class FaissFlatPass : public Pass {
  public:
-    LargeSQ8HPass() = default;
+    FaissFlatPass() = default;
 
  public:
     void
@@ -50,7 +50,7 @@ class LargeSQ8HPass : public Pass {
     std::vector<int32_t> gpus;
 };
 
-using LargeSQ8HPassPtr = std::shared_ptr<LargeSQ8HPass>;
+using FaissFlatPassPtr = std::shared_ptr<FaissFlatPass>;
 
 }  // namespace scheduler
 }  // namespace milvus

core/src/scheduler/optimizer/FaissIVFFlatPass.cpp

@@ -15,31 +15,52 @@
 // specific language governing permissions and limitations
 // under the License.
 
-#include "scheduler/optimizer/OnlyCPUPass.h"
+#include "scheduler/optimizer/FaissIVFFlatPass.h"
+#include "cache/GpuCacheMgr.h"
 #include "scheduler/SchedInst.h"
 #include "scheduler/Utils.h"
 #include "scheduler/task/SearchTask.h"
 #include "scheduler/tasklabel/SpecResLabel.h"
+#include "server/Config.h"
+#include "utils/Log.h"
 
 namespace milvus {
 namespace scheduler {
 
 void
-OnlyCPUPass::Init() {
+FaissIVFFlatPass::Init() {
+    server::Config& config = server::Config::GetInstance();
+    Status s = config.GetEngineConfigGpuSearchThreshold(threshold_);
+    if (!s.ok()) {
+        threshold_ = std::numeric_limits<int32_t>::max();
+    }
+    s = config.GetGpuResourceConfigSearchResources(gpus);
+    if (!s.ok()) {
+        throw;
+    }
 }
 
 bool
-OnlyCPUPass::Run(const TaskPtr& task) {
-    if (task->Type() != TaskType::SearchTask)
-        return false;
-    auto search_task = std::static_pointer_cast<XSearchTask>(task);
-    if (search_task->file_->engine_type_ != (int)engine::EngineType::FAISS_IVFSQ8 &&
-        search_task->file_->engine_type_ != (int)engine::EngineType::FAISS_IVFFLAT) {
+FaissIVFFlatPass::Run(const TaskPtr& task) {
+    if (task->Type() != TaskType::SearchTask) {
         return false;
     }
 
-    ResourcePtr res_ptr = ResMgrInst::GetInstance()->GetResource("cpu");
-    auto label = std::make_shared<SpecResLabel>(std::weak_ptr<Resource>(res_ptr));
+    auto search_task = std::static_pointer_cast<XSearchTask>(task);
+    if (search_task->file_->engine_type_ != (int)engine::EngineType::FAISS_IVFFLAT) {
+        return false;
+    }
+
+    auto search_job = std::static_pointer_cast<SearchJob>(search_task->job_.lock());
+    ResourcePtr res_ptr;
+    if (search_job->nq() < threshold_) {
+        res_ptr = ResMgrInst::GetInstance()->GetResource("cpu");
+    } else {
+        auto best_device_id = count_ % gpus.size();
+        count_++;
+        res_ptr = ResMgrInst::GetInstance()->GetResource(ResourceType::GPU, best_device_id);
+    }
+    auto label = std::make_shared<SpecResLabel>(res_ptr);
     task->label() = label;
     return true;
 }

core/src/scheduler/optimizer/FaissIVFFlatPass.h

@@ -18,6 +18,7 @@
 
 #include <condition_variable>
 #include <deque>
+#include <limits>
 #include <list>
 #include <memory>
 #include <mutex>
@@ -32,9 +33,9 @@
 namespace milvus {
 namespace scheduler {
 
-class HybridPass : public Pass {
+class FaissIVFFlatPass : public Pass {
  public:
-    HybridPass() = default;
+    FaissIVFFlatPass() = default;
 
  public:
     void
@@ -42,9 +43,14 @@ class HybridPass : public Pass {
 
     bool
     Run(const TaskPtr& task) override;
+
+ private:
+    int32_t threshold_ = std::numeric_limits<int32_t>::max();
+    int64_t count_ = 0;
+    std::vector<int32_t> gpus;
 };
 
-using HybridPassPtr = std::shared_ptr<HybridPass>;
+using FaissIVFFlatPassPtr = std::shared_ptr<FaissIVFFlatPass>;
 
 }  // namespace scheduler
 }  // namespace milvus

core/src/scheduler/optimizer/FaissIVFSQ8HPass.cpp

@@ -15,7 +15,7 @@
 // specific language governing permissions and limitations
 // under the License.
 
-#include "scheduler/optimizer/LargeSQ8HPass.h"
+#include "scheduler/optimizer/FaissIVFSQ8HPass.h"
 #include "cache/GpuCacheMgr.h"
 #include "scheduler/SchedInst.h"
 #include "scheduler/Utils.h"
@@ -28,7 +28,7 @@ namespace milvus {
 namespace scheduler {
 
 void
-LargeSQ8HPass::Init() {
+FaissIVFSQ8HPass::Init() {
     server::Config& config = server::Config::GetInstance();
     Status s = config.GetEngineConfigGpuSearchThreshold(threshold_);
     if (!s.ok()) {
@@ -38,7 +38,7 @@ LargeSQ8HPass::Init() {
 }
 
 bool
-LargeSQ8HPass::Run(const TaskPtr& task) {
+FaissIVFSQ8HPass::Run(const TaskPtr& task) {
     if (task->Type() != TaskType::SearchTask) {
         return false;
     }
@@ -49,36 +49,16 @@ LargeSQ8HPass::Run(const TaskPtr& task) {
     }
 
     auto search_job = std::static_pointer_cast<SearchJob>(search_task->job_.lock());
-
-    // TODO: future, Index::IVFSQ8H, if nq < threshold set cpu, else set gpu
-
+    ResourcePtr res_ptr;
     if (search_job->nq() < threshold_) {
-        return false;
+        res_ptr = ResMgrInst::GetInstance()->GetResource("cpu");
+    } else {
+        auto best_device_id = count_ % gpus.size();
+        count_++;
+        res_ptr = ResMgrInst::GetInstance()->GetResource(ResourceType::GPU, best_device_id);
     }
-
-    //    std::vector<int64_t> all_free_mem;
-    //    for (auto& gpu : gpus) {
-    //        auto cache = cache::GpuCacheMgr::GetInstance(gpu);
-    //        auto free_mem = cache->CacheCapacity() - cache->CacheUsage();
-    //        all_free_mem.push_back(free_mem);
-    //    }
-    //
-    //    auto max_e = std::max_element(all_free_mem.begin(), all_free_mem.end());
-    //    auto best_index = std::distance(all_free_mem.begin(), max_e);
-    //    auto best_device_id = gpus[best_index];
-    auto best_device_id = count_ % gpus.size();
-    count_++;
-
-    ResourcePtr res_ptr = ResMgrInst::GetInstance()->GetResource(ResourceType::GPU, best_device_id);
-    if (not res_ptr) {
-        SERVER_LOG_ERROR << "GpuResource " << best_device_id << " invalid.";
-        // TODO: throw critical error and exit
-        return false;
-    }
-
-    auto label = std::make_shared<SpecResLabel>(std::weak_ptr<Resource>(res_ptr));
+    auto label = std::make_shared<SpecResLabel>(res_ptr);
     task->label() = label;
-
     return true;
 }
 

core/src/scheduler/optimizer/FaissIVFSQ8HPass.h

@@ -0,0 +1,56 @@
+// 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 <condition_variable>
+#include <deque>
+#include <limits>
+#include <list>
+#include <memory>
+#include <mutex>
+#include <queue>
+#include <string>
+#include <thread>
+#include <unordered_map>
+#include <vector>
+
+#include "Pass.h"
+
+namespace milvus {
+namespace scheduler {
+
+class FaissIVFSQ8HPass : public Pass {
+ public:
+    FaissIVFSQ8HPass() = default;
+
+ public:
+    void
+    Init() override;
+
+    bool
+    Run(const TaskPtr& task) override;
+
+ private:
+    int32_t threshold_ = std::numeric_limits<int32_t>::max();
+    int64_t count_ = 0;
+    std::vector<int32_t> gpus;
+};
+
+using FaissIVFSQ8HPassPtr = std::shared_ptr<FaissIVFSQ8HPass>;
+
+}  // namespace scheduler
+}  // namespace milvus

core/src/scheduler/optimizer/FaissIVFSQ8Pass.cpp

@@ -0,0 +1,69 @@
+// 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/optimizer/FaissIVFSQ8Pass.h"
+#include "cache/GpuCacheMgr.h"
+#include "scheduler/SchedInst.h"
+#include "scheduler/Utils.h"
+#include "scheduler/task/SearchTask.h"
+#include "scheduler/tasklabel/SpecResLabel.h"
+#include "server/Config.h"
+#include "utils/Log.h"
+
+namespace milvus {
+namespace scheduler {
+
+void
+FaissIVFSQ8Pass::Init() {
+    server::Config& config = server::Config::GetInstance();
+    Status s = config.GetEngineConfigGpuSearchThreshold(threshold_);
+    if (!s.ok()) {
+        threshold_ = std::numeric_limits<int32_t>::max();
+    }
+    s = config.GetGpuResourceConfigSearchResources(gpus);
+    if (!s.ok()) {
+        throw;
+    }
+}
+
+bool
+FaissIVFSQ8Pass::Run(const TaskPtr& task) {
+    if (task->Type() != TaskType::SearchTask) {
+        return false;
+    }
+
+    auto search_task = std::static_pointer_cast<XSearchTask>(task);
+    if (search_task->file_->engine_type_ != (int)engine::EngineType::FAISS_IVFSQ8) {
+        return false;
+    }
+
+    auto search_job = std::static_pointer_cast<SearchJob>(search_task->job_.lock());
+    ResourcePtr res_ptr;
+    if (search_job->nq() < threshold_) {
+        res_ptr = ResMgrInst::GetInstance()->GetResource("cpu");
+    } else {
+        auto best_device_id = count_ % gpus.size();
+        count_++;
+        res_ptr = ResMgrInst::GetInstance()->GetResource(ResourceType::GPU, best_device_id);
+    }
+    auto label = std::make_shared<SpecResLabel>(res_ptr);
+    task->label() = label;
+    return true;
+}
+
+}  // namespace scheduler
+}  // namespace milvus

core/src/scheduler/optimizer/FaissIVFSQ8Pass.h

@@ -18,6 +18,7 @@
 
 #include <condition_variable>
 #include <deque>
+#include <limits>
 #include <list>
 #include <memory>
 #include <mutex>
@@ -32,9 +33,9 @@
 namespace milvus {
 namespace scheduler {
 
-class OnlyCPUPass : public Pass {
+class FaissIVFSQ8Pass : public Pass {
  public:
-    OnlyCPUPass() = default;
+    FaissIVFSQ8Pass() = default;
 
  public:
     void
@@ -42,9 +43,14 @@ class OnlyCPUPass : public Pass {
 
     bool
     Run(const TaskPtr& task) override;
+
+ private:
+    int32_t threshold_ = std::numeric_limits<int32_t>::max();
+    int64_t count_ = 0;
+    std::vector<int32_t> gpus;
 };
 
-using OnlyCPUPassPtr = std::shared_ptr<OnlyCPUPass>;
+using FaissIVFSQ8PassPtr = std::shared_ptr<FaissIVFSQ8Pass>;
 
 }  // namespace scheduler
 }  // namespace milvus

core/src/scheduler/optimizer/HybridPass.cpp

@@ -1,47 +0,0 @@
-// 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/optimizer/HybridPass.h"
-#include "scheduler/SchedInst.h"
-#include "scheduler/task/SearchTask.h"
-#include "scheduler/tasklabel/SpecResLabel.h"
-
-namespace milvus {
-namespace scheduler {
-
-void
-HybridPass::Init() {
-}
-
-bool
-HybridPass::Run(const TaskPtr& task) {
-    // TODO: future, Index::IVFSQ8H, if nq < threshold set cpu, else set gpu
-    if (task->Type() != TaskType::SearchTask)
-        return false;
-    auto search_task = std::static_pointer_cast<XSearchTask>(task);
-    if (search_task->file_->engine_type_ == (int)engine::EngineType::FAISS_IVFSQ8H) {
-        // TODO: remove "cpu" hardcode
-        ResourcePtr res_ptr = ResMgrInst::GetInstance()->GetResource("cpu");
-        auto label = std::make_shared<SpecResLabel>(std::weak_ptr<Resource>(res_ptr));
-        task->label() = label;
-        return true;
-    }
-    return false;
-}
-
-}  // namespace scheduler
-}  // namespace milvus