Optimize snapshot GC executor (#3528)

* Optimize gc event executor (#3465)
Signed-off-by: Nyinghao.zou <yinghao.zou@zilliz.com>

CHANGELOG.md

@@ -61,6 +61,7 @@ Please mark all changes in change log and use the issue from GitHub
 -   \#2884 Using BlockingQueue in JobMgr
 -   \#3220 Enable -Werror to improve code quality
 -   \#3449 Upgrade master version to v0.11.0
+-   \#3465 Optimize gc event executor
 
 ## Task
 

core/src/db/Constants.h

@@ -33,5 +33,7 @@ constexpr int64_t MAX_SCRIPT_FILE_SIZE = 256 * MB;          // max file size of
 
 constexpr int64_t BUILD_INEDX_RETRY_TIMES = 3;  // retry times if build index failed
 
+constexpr const char* DB_FOLDER = "/db";
+
 }  // namespace engine
 }  // namespace milvus

core/src/db/DBImpl.cpp

@@ -90,15 +90,6 @@ DBImpl::Start() {
         return Status::OK();
     }
 
-    // snapshot
-    auto store = snapshot::Store::Build(options_.meta_.backend_uri_, options_.meta_.path_,
-                                        codec::Codec::instance().GetSuffixSet());
-    snapshot::OperationExecutor::Init(store);
-    snapshot::OperationExecutor::GetInstance().Start();
-    snapshot::EventExecutor::Init(store);
-    snapshot::EventExecutor::GetInstance().Start();
-    snapshot::Snapshots::GetInstance().Init(store);
-
     knowhere::enable_faiss_logging();
 
     // LOG_ENGINE_TRACE_ << "DB service start";
@@ -155,9 +146,6 @@ DBImpl::Stop() {
         bg_metric_thread_.join();
     }
 
-    snapshot::EventExecutor::GetInstance().Stop();
-    snapshot::OperationExecutor::GetInstance().Stop();
-
     // LOG_ENGINE_TRACE_ << "DB service stop";
     return Status::OK();
 }

core/src/db/snapshot/CompoundOperations.cpp

@@ -958,7 +958,7 @@ CreateCollectionOperation::DoExecute(StorePtr store) {
     auto status = store->CreateResource<Collection>(
         Collection(c_context_.collection->GetName(), c_context_.collection->GetParams()), collection);
     if (!status.ok()) {
-        std::cerr << status.ToString() << std::endl;
+        LOG_ENGINE_ERROR_ << status.ToString();
         return status;
     }
     auto c_ctx_p = ResourceContextBuilder<Collection>().SetOp(meta::oUpdate).CreatePtr();

core/src/db/snapshot/EventExecutor.h

@@ -13,18 +13,22 @@
 
 #include <memory>
 #include <mutex>
+#include <queue>
 #include <thread>
+#include <utility>
 
+#include "db/SimpleWaitNotify.h"
 #include "db/snapshot/MetaEvent.h"
 #include "utils/BlockingQueue.h"
 
-namespace milvus {
-namespace engine {
-namespace snapshot {
+namespace milvus::engine::snapshot {
 
 using EventPtr = std::shared_ptr<MetaEvent>;
 using ThreadPtr = std::shared_ptr<std::thread>;
-using EventQueue = BlockingQueue<EventPtr>;
+using EventQueue = std::queue<EventPtr>;
+
+constexpr size_t EVENT_QUEUE_SIZE = 4096;
+constexpr size_t EVENT_TIMING_INTERVAL = 5;
 
 class EventExecutor {
  public:
@@ -52,35 +56,51 @@ class EventExecutor {
     }
 
     Status
-    Submit(const EventPtr& evt) {
+    Submit(const EventPtr& evt, bool flush = false) {
         if (evt == nullptr) {
             return Status(SS_INVALID_ARGUMENT_ERROR, "Invalid Resource");
         }
-        Enqueue(evt);
+        Enqueue(evt, flush);
         return Status::OK();
     }
 
     void
     Start() {
-        if (thread_ptr_ == nullptr) {
-            thread_ptr_ = std::make_shared<std::thread>(&EventExecutor::ThreadMain, this);
+        if (running_.exchange(true)) {
+            return;
+        }
+
+        if (gc_thread_ptr_ == nullptr) {
+            gc_thread_ptr_ = std::make_shared<std::thread>(&EventExecutor::GCThread, this);
+        }
+
+        if (timing_thread_ptr_ == nullptr) {
+            timing_thread_ptr_ = std::make_shared<std::thread>(&EventExecutor::TimingThread, this);
         }
     }
 
     void
     Stop() {
-        if (thread_ptr_ != nullptr) {
-            Enqueue(nullptr);
-            thread_ptr_->join();
-            thread_ptr_ = nullptr;
-            std::cout << "EventExecutor Stopped" << std::endl;
+        if (!running_.exchange(false)) {
+            //             executor has been stopped, just return
+            return;
+        }
+
+        timing_.Notify();
+
+        if (timing_thread_ptr_ != nullptr) {
+            timing_thread_ptr_->join();
+            timing_thread_ptr_ = nullptr;
+        }
+
+        if (gc_thread_ptr_ != nullptr) {
+            gc_thread_ptr_->join();
+            gc_thread_ptr_ = nullptr;
         }
     }
 
  private:
-    EventExecutor() {
-        queue_.SetCapacity(10000);
-    }
+    EventExecutor() = default;
 
     static EventExecutor&
     GetInstanceImpl() {
@@ -89,24 +109,79 @@ class EventExecutor {
     }
 
     void
-    ThreadMain() {
+    GCThread() {
         Status status;
         while (true) {
-            EventPtr evt = queue_.Take();
-            if (evt == nullptr) {
+            auto front = cache_queues_.Take();
+
+            while (front && !front->empty()) {
+                EventPtr event_front(front->front());
+                front->pop();
+                if (event_front == nullptr) {
+                    continue;
+                }
+
+                /* std::cout << std::this_thread::get_id() << " Dequeue Event " << std::endl; */
+                status = event_front->Process(store_);
+                if (!status.ok()) {
+                    LOG_ENGINE_ERROR_ << "EventExecutor Handle Event Error: " << status.ToString();
+                }
+            }
+
+            if ((!initialized_.load() || !running_.load()) && cache_queues_.Empty()) {
                 break;
             }
-            /* std::cout << std::this_thread::get_id() << " Dequeue Event " << std::endl; */
-            status = evt->Process(store_);
-            if (!status.ok()) {
-                std::cout << "EventExecutor Handle Event Error: " << status.ToString() << std::endl;
+        }
+    }
+
+    void
+    TimingThread() {
+        while (true) {
+            timing_.Wait_For(std::chrono::seconds(EVENT_TIMING_INTERVAL));
+
+            std::shared_ptr<EventQueue> queue;
+            {
+                std::unique_lock<std::mutex> lock(mutex_);
+                if (queue_ != nullptr && !queue_->empty()) {
+                    queue = std::move(queue_);
+                    queue_ = nullptr;
+                }
+            }
+
+            if (queue) {
+                cache_queues_.Put(queue);
+            }
+
+            if (!running_.load() || !initialized_.load()) {
+                cache_queues_.Put(nullptr);
+                break;
             }
         }
     }
 
     void
-    Enqueue(const EventPtr& evt) {
-        queue_.Put(evt);
+    Enqueue(const EventPtr& evt, bool flush = false) {
+        if (!initialized_.load() || !running_.load()) {
+            LOG_ENGINE_WARNING_ << "GcEvent exiting ...";
+            return;
+        }
+
+        bool need_notify = flush;
+        {
+            std::unique_lock<std::mutex> lock(mutex_);
+            if (queue_ == nullptr) {
+                queue_ = std::make_shared<EventQueue>();
+            }
+
+            queue_->push(evt);
+            if (queue_->size() >= EVENT_QUEUE_SIZE) {
+                need_notify = true;
+            }
+        }
+
+        if (!initialized_.load() || !running_.load() || need_notify) {
+            timing_.Notify();
+        }
         if (evt != nullptr) {
             /* std::cout << std::this_thread::get_id() << " Enqueue Event " << std::endl; */
         }
@@ -114,12 +189,15 @@ class EventExecutor {
 
     EventExecutor(const EventExecutor&) = delete;
 
-    ThreadPtr thread_ptr_ = nullptr;
-    EventQueue queue_;
+    ThreadPtr timing_thread_ptr_ = nullptr;
+    ThreadPtr gc_thread_ptr_ = nullptr;
+    SimpleWaitNotify timing_;
+    std::mutex mutex_;
+    std::shared_ptr<EventQueue> queue_;
+    BlockingQueue<std::shared_ptr<EventQueue>> cache_queues_;
     std::atomic_bool initialized_ = false;
+    std::atomic_bool running_ = false;
     StorePtr store_;
 };
 
-}  // namespace snapshot
-}  // namespace engine
-}  // namespace milvus
+}  // namespace milvus::engine::snapshot

core/src/db/snapshot/InActiveResourcesGCEvent.h

@@ -42,7 +42,7 @@ class InActiveResourcesGCEvent : public GCEvent, public Operations {
 
     Status
     OnExecute(StorePtr store) override {
-        std::cout << "Executing InActiveResourcesGCEvent" << std::endl;
+        LOG_ENGINE_INFO_ << "Executing InActiveResourcesGCEvent";
 
         STATUS_CHECK(ClearInActiveResources<Collection>(store));
         STATUS_CHECK(ClearInActiveResources<CollectionCommit>(store));
@@ -78,7 +78,7 @@ class InActiveResourcesGCEvent : public GCEvent, public Operations {
                 /* std::cout << "[GC] Remove file " << res_->ToString() << " " << res_path << " " << ok << std::endl; */
             } else {
                 RemoveWithSuffix<ResourceT>(res, res_path, store->GetSuffixSet());
-                std::cout << "[GC] Remove stale " << res_path << " for " << res->ToString() << std::endl;
+                LOG_ENGINE_DEBUG_ << "[GC] Remove stale " << res_path << " for " << res->ToString();
             }
 
             /* remove resource from meta */

core/src/db/snapshot/MetaEvent.h

@@ -40,11 +40,11 @@ class GCEvent : virtual public MetaEvent {
             auto adjusted = path + suffix;
             if (std::experimental::filesystem::is_regular_file(adjusted)) {
                 auto ok = std::experimental::filesystem::remove(adjusted);
-                std::cout << "[GC] Remove FILE " << res->ToString() << " " << adjusted << " " << ok << std::endl;
+                LOG_ENGINE_DEBUG_ << "[GC] Remove FILE " << res->ToString() << " " << adjusted << " " << ok;
                 return;
             }
         }
-        std::cout << "[GC] Remove STALE OBJECT " << path << " for " << res->ToString() << std::endl;
+        LOG_ENGINE_DEBUG_ << "[GC] Remove STALE OBJECT " << path << " for " << res->ToString();
     }
 };
 

core/src/db/snapshot/OperationExecutor.h

@@ -73,7 +73,7 @@ class OperationExecutor {
             Enqueue(nullptr);
             thread_ptr_->join();
             thread_ptr_ = nullptr;
-            std::cout << "OperationExecutor Stopped" << std::endl;
+            LOG_ENGINE_INFO_ << "OperationExecutor Stopped";
         }
     }
 

core/src/db/snapshot/Operations.cpp

@@ -21,9 +21,7 @@
 #include "db/snapshot/ResourceGCEvent.h"
 #include "db/snapshot/Snapshots.h"
 
-namespace milvus {
-namespace engine {
-namespace snapshot {
+namespace milvus::engine::snapshot {
 
 static ID_TYPE UID = 1;
 
@@ -114,7 +112,7 @@ Operations::Done(StorePtr store) {
         /*                 context_.new_collection_commit->GetCollectionId(), holder.rbegin()->GetID()); */
         /*     } */
         /* } */
-        std::cout << ToString() << std::endl;
+        LOG_ENGINE_DEBUG_ << ToString();
     }
     finish_cond_.notify_all();
 }
@@ -255,7 +253,7 @@ Operations::OnApplyTimeoutCallback(StorePtr store) {
 
     auto status = store->ApplyOperation(*this, context);
     while (status.code() == SS_TIMEOUT && !HasAborted()) {
-        std::cout << GetName() << " Timeout! Try " << ++try_times << std::endl;
+        LOG_ENGINE_WARNING_ << GetName() << " Timeout! Try " << ++try_times;
         std::this_thread::sleep_for(std::chrono::milliseconds(1));
         status = store->ApplyOperation(*this, context);
     }
@@ -290,7 +288,7 @@ ApplyRollBack(std::set<std::shared_ptr<ResourceContext<ResourceT>>>& step_contex
         auto res = step_context->Resource();
         auto evt_ptr = std::make_shared<ResourceGCEvent<ResourceT>>(res);
         EventExecutor::GetInstance().Submit(evt_ptr);
-        std::cout << "Rollback " << typeid(ResourceT).name() << ": " << res->GetID() << std::endl;
+        LOG_ENGINE_DEBUG_ << "Rollback " << typeid(ResourceT).name() << ": " << res->GetID();
     }
 }
 
@@ -305,6 +303,4 @@ Operations::~Operations() {
     }
 }
 
-}  // namespace snapshot
-}  // namespace engine
-}  // namespace milvus
+}  // namespace milvus::engine::snapshot

core/src/db/snapshot/ResourceGCEvent.h

@@ -48,16 +48,16 @@ class ResourceGCEvent : public GCEvent {
                 if (res_path.empty()) {
                     /* std::cout << "[GC] No remove action for " << res_->ToString() << std::endl; */
                 } else if (std::experimental::filesystem::is_directory(res_path)) {
-                    std::cout << "[GC] Remove DIR " << res_->ToString() << " " << res_path << std::endl;
+                    LOG_ENGINE_DEBUG_ << "[GC] Remove DIR " << res_->ToString() << " " << res_path;
                     auto ok = std::experimental::filesystem::remove_all(res_path);
                 } else if (std::experimental::filesystem::is_regular_file(res_path)) {
-                    std::cout << "[GC] Remove FILE " << res_->ToString() << " " << res_path << std::endl;
+                    LOG_ENGINE_DEBUG_ << "[GC] Remove FILE " << res_->ToString() << " " << res_path;
                     auto ok = std::experimental::filesystem::remove(res_path);
                 } else {
                     RemoveWithSuffix<ResourceT>(res_, res_path, store->GetSuffixSet());
                 }
             } catch (const std::experimental::filesystem::filesystem_error& er) {
-                std::cout << "[GC] " << er.what() << std::endl;
+                LOG_SERVER_ERROR_ << "[GC] Error when removing path " << res_path << ": " << er.what();
                 if (do_throw) {
                     throw;
                 }
@@ -73,7 +73,7 @@ class ResourceGCEvent : public GCEvent {
                 break;
             }
             std::this_thread::sleep_for(std::chrono::milliseconds(1));
-            std::cout << "[GC] Retry remove: " << res_path << std::endl;
+            LOG_ENGINE_DEBUG_ << "[GC] Retry remove: " << res_path;
         }
 
         auto hard_delete_meta = [](ID_TYPE id, StorePtr store) -> Status {
@@ -82,7 +82,7 @@ class ResourceGCEvent : public GCEvent {
         };
 
         if (!status.ok()) {
-            std::cout << "[GC] Stale Resource: " << res_path << " need to be cleanup later" << std::endl;
+            LOG_ENGINE_ERROR_ << "[GC] Stale Resource: " << res_path << " need to be cleanup later";
             return status;
         }
 

core/src/db/snapshot/ResourceHolder.h

@@ -105,12 +105,12 @@ class ResourceHolder {
     virtual void
     Dump(const std::string& tag = "") {
         std::unique_lock<std::mutex> lock(mutex_);
-        std::cout << typeid(*this).name() << " Dump Start [" << tag << "]:" << id_map_.size() << std::endl;
+        LOG_ENGINE_DEBUG_ << typeid(*this).name() << " Dump Start [" << tag << "]:" << id_map_.size();
         for (auto& kv : id_map_) {
             /* std::cout << "\t" << kv.second->ToString() << std::endl; */
-            std::cout << "\t" << kv.first << " RefCnt " << kv.second->ref_count() << std::endl;
+            LOG_ENGINE_DEBUG_ << "\t" << kv.first << " RefCnt " << kv.second->ref_count();
         }
-        std::cout << typeid(*this).name() << " Dump   End [" << tag << "]" << std::endl;
+        LOG_ENGINE_DEBUG_ << typeid(*this).name() << " Dump   End [" << tag << "]";
     }
 
  private:

core/src/db/snapshot/Snapshot.h

@@ -272,13 +272,13 @@ class Snapshot : public ReferenceProxy {
     void
     DumpResource(const std::string& tag = "") const {
         auto& resources = GetResources<ResourceT>();
-        std::cout << typeid(*this).name() << " Dump " << GetID() << " " << ResourceT::Name << " Start [" << tag
-                  << "]:" << resources.size() << std::endl;
+        LOG_ENGINE_DEBUG_ << typeid(*this).name() << " Dump " << GetID() << " " << ResourceT::Name << " Start [" << tag
+                          << "]:" << resources.size();
         for (auto& kv : resources) {
-            std::cout << "\t" << kv.second->ToString() << std::endl;
+            LOG_ENGINE_DEBUG_ << "\t" << kv.second->ToString();
         }
-        std::cout << typeid(*this).name() << " Dump " << GetID() << " " << ResourceT::Name << "  End [" << tag
-                  << "]:" << resources.size() << std::endl;
+        LOG_ENGINE_DEBUG_ << typeid(*this).name() << " Dump " << GetID() << " " << ResourceT::Name << "  End [" << tag
+                          << "]:" << resources.size();
     }
 
     template <typename T>

core/src/db/snapshot/Snapshots.cpp

@@ -10,13 +10,16 @@
 // or implied. See the License for the specific language governing permissions and limitations under the License.
 
 #include "db/snapshot/Snapshots.h"
+
+#include "config/ServerConfig.h"
+#include "db/Constants.h"
 #include "db/snapshot/CompoundOperations.h"
 #include "db/snapshot/EventExecutor.h"
 #include "db/snapshot/InActiveResourcesGCEvent.h"
+#include "db/snapshot/OperationExecutor.h"
+#include "utils/CommonUtil.h"
 
-namespace milvus {
-namespace engine {
-namespace snapshot {
+namespace milvus::engine::snapshot {
 
 /* Status */
 /* Snapshots::DropAll() { */
@@ -128,7 +131,7 @@ Snapshots::LoadNoLock(StorePtr store, ID_TYPE collection_id, SnapshotHolderPtr&
 Status
 Snapshots::Init(StorePtr store) {
     auto event = std::make_shared<InActiveResourcesGCEvent>();
-    EventExecutor::GetInstance().Submit(event);
+    EventExecutor::GetInstance().Submit(event, true);
     STATUS_CHECK(event->WaitToFinish());
     auto op = std::make_shared<GetCollectionIDsOperation>();
     STATUS_CHECK((*op)(store));
@@ -209,10 +212,34 @@ void
 Snapshots::SnapshotGCCallback(Snapshot::Ptr ss_ptr) {
     /* to_release_.push_back(ss_ptr); */
     ss_ptr->UnRef();
-    std::cout << "Snapshot " << ss_ptr->GetID() << " ref_count = " << ss_ptr->ref_count() << " To be removed"
-              << std::endl;
+    LOG_ENGINE_DEBUG_ << "Snapshot " << ss_ptr->GetID() << " ref_count = " << ss_ptr->ref_count() << " To be removed";
+}
+
+Status
+Snapshots::StartService() {
+    auto meta_path = config.storage.path() + DB_FOLDER;
+
+    // create db root path
+    auto s = CommonUtil::CreateDirectory(meta_path);
+    if (!s.ok()) {
+        std::cerr << "Error: Failed to create database primary path: " << meta_path
+                  << ". Possible reason: db_config.primary_path is wrong in milvus.yaml or not available." << std::endl;
+        kill(0, SIGUSR1);
+    }
+
+    auto store = snapshot::Store::Build(config.general.meta_uri(), meta_path, codec::Codec::instance().GetSuffixSet());
+    snapshot::OperationExecutor::Init(store);
+    snapshot::OperationExecutor::GetInstance().Start();
+    snapshot::EventExecutor::Init(store);
+    snapshot::EventExecutor::GetInstance().Start();
+    return snapshot::Snapshots::GetInstance().Init(store);
+}
+
+Status
+Snapshots::StopService() {
+    snapshot::EventExecutor::GetInstance().Stop();
+    snapshot::OperationExecutor::GetInstance().Stop();
+    return Status::OK();
 }
 
-}  // namespace snapshot
-}  // namespace engine
-}  // namespace milvus
+}  // namespace milvus::engine::snapshot

core/src/db/snapshot/Snapshots.h

@@ -23,9 +23,7 @@
 #include "db/snapshot/Store.h"
 #include "utils/Status.h"
 
-namespace milvus {
-namespace engine {
-namespace snapshot {
+namespace milvus::engine::snapshot {
 
 class Snapshots {
  public:
@@ -66,6 +64,13 @@ class Snapshots {
 
     Status Init(StorePtr);
 
+ public:
+    Status
+    StartService();
+
+    Status
+    StopService();
+
  private:
     void
     SnapshotGCCallback(Snapshot::Ptr ss_ptr);
@@ -84,6 +89,4 @@ class Snapshots {
     std::vector<Snapshot::Ptr> to_release_;
 };
 
-}  // namespace snapshot
-}  // namespace engine
-}  // namespace milvus
+}  // namespace milvus::engine::snapshot

core/src/db/snapshot/Store.h

@@ -244,7 +244,7 @@ class Store : public std::enable_shared_from_this<Store> {
     DoReset() {
         auto status = adapter_->TruncateAll();
         if (!status.ok()) {
-            std::cout << "TruncateAll failed: " << status.ToString() << std::endl;
+            LOG_ENGINE_ERROR_ << "TruncateAll failed: " << status.ToString();
         }
     }
 

core/src/server/DBWrapper.cpp

@@ -17,6 +17,7 @@
 #include <vector>
 
 #include "config/ServerConfig.h"
+#include "db/Constants.h"
 #include "db/DBFactory.h"
 #include "db/snapshot/OperationExecutor.h"
 #include "utils/CommonUtil.h"
@@ -35,7 +36,7 @@ DBWrapper::StartService() {
     opt.meta_.backend_uri_ = config.general.meta_uri();
 
     std::string path = config.storage.path();
-    opt.meta_.path_ = path + "/db";
+    opt.meta_.path_ = path + engine::DB_FOLDER;
 
     opt.auto_flush_interval_ = config.storage.auto_flush_interval();
     opt.metric_enable_ = config.metric.enable();

core/src/server/Server.cpp

@@ -18,6 +18,7 @@
 #include <unordered_map>
 
 #include "config/ServerConfig.h"
+#include "db/snapshot/Snapshots.h"
 #include "index/archive/KnowhereResource.h"
 #include "log/LogMgr.h"
 #include "metrics/Metrics.h"
@@ -250,6 +251,8 @@ Server::StartService() {
         goto FAIL;
     }
 
+    engine::snapshot::Snapshots::GetInstance().StartService();
+
     scheduler::StartSchedulerService();
 
     stat = DBWrapper::GetInstance().StartService();
@@ -280,6 +283,7 @@ Server::StopService() {
     grpc::GrpcServer::GetInstance().Stop();
     DBWrapper::GetInstance().StopService();
     scheduler::StopSchedulerService();
+    engine::snapshot::Snapshots::GetInstance().StopService();
     engine::KnowhereResource::Finalize();
 }
 

core/unittest/db/test_db.cpp

@@ -24,12 +24,14 @@
 #include "db/SnapshotUtils.h"
 #include "db/SnapshotVisitor.h"
 #include "db/snapshot/IterateHandler.h"
+#include "db/snapshot/InActiveResourcesGCEvent.h"
 #include "db/snapshot/ResourceHelper.h"
 #include "db/utils.h"
 #include "knowhere/index/vector_index/helpers/IndexParameter.h"
 #include "segment/Segment.h"
 
 using SegmentVisitor = milvus::engine::SegmentVisitor;
+using InActiveResourcesGCEvent = milvus::engine::snapshot::InActiveResourcesGCEvent;
 
 namespace {
 const char* VECTOR_FIELD_NAME = "vector";
@@ -284,7 +286,7 @@ TEST_F(DBTest, CollectionTest) {
     };
     std::string c1 = "c1";
     auto status = CreateCollection(db_, c1, next_lsn());
-    ASSERT_TRUE(status.ok());
+    ASSERT_TRUE(status.ok()) << status.ToString();
 
     ScopedSnapshotT ss;
     status = Snapshots::GetInstance().GetSnapshot(ss, c1);
@@ -672,6 +674,9 @@ TEST_F(DBTest, MergeTest) {
 
     // wait to merge finished
     sleep(2);
+    auto event = std::make_shared<InActiveResourcesGCEvent>();
+    milvus::engine::snapshot::EventExecutor::GetInstance().Submit(event, true);
+    event->WaitToFinish();
 
     // validate entities count
     int64_t row_count = 0;

core/unittest/db/test_ss_event.cpp

@@ -27,6 +27,9 @@ using FType = milvus::engine::DataType;
 using FEType = milvus::engine::FieldElementType;
 
 using InActiveResourcesGCEvent = milvus::engine::snapshot::InActiveResourcesGCEvent;
+template<typename T>
+using ResourceGCEvent = milvus::engine::snapshot::ResourceGCEvent<T>;
+using EventExecutor = milvus::engine::snapshot::EventExecutor;
 
 TEST_F(EventTest, TestInActiveResGcEvent) {
     CollectionPtr collection;
@@ -104,8 +107,6 @@ TEST_F(EventTest, TestInActiveResGcEvent) {
 
     auto event = std::make_shared<InActiveResourcesGCEvent>();
     status = event->Process(store_);
-//    milvus::engine::snapshot::EventExecutor::GetInstance().Submit(event);
-//    status = event->WaitToFinish();
     ASSERT_TRUE(status.ok()) << status.ToString();
 
     std::vector<FieldElementPtr> field_elements;
@@ -152,3 +153,47 @@ TEST_F(EventTest, TestInActiveResGcEvent) {
     ASSERT_TRUE(store_->GetInActiveResources<CollectionCommit>(collection_commits).ok());
     ASSERT_TRUE(collection_commits.empty());
 }
+
+TEST_F(EventTest, GcTest) {
+    std::string collection_name = "event_test_gc_test";
+
+    CollectionPtr collection;
+    auto cc = Collection(collection_name);
+    cc.Activate();
+    auto status = store_->CreateResource(std::move(cc), collection);
+    ASSERT_TRUE(status.ok()) << status.ToString();
+
+    CollectionPtr collection2;
+    status = store_->GetResource<Collection>(collection->GetID(), collection2);
+    ASSERT_TRUE(status.ok()) << status.ToString();
+
+    auto event = std::make_shared<ResourceGCEvent<Collection>>(collection);
+    status = event->Process(store_);
+    ASSERT_TRUE(status.ok()) << status.ToString();
+
+    CollectionPtr rcollection;
+    status = store_->GetResource<Collection>(collection->GetID(),rcollection);
+    ASSERT_FALSE(status.ok());
+}
+
+TEST_F(EventTest, GcBlockingTest) {
+    size_t max_count = 1000;
+
+    std::vector<CollectionPtr> collections;
+    for (size_t i = 0; i < max_count; i++) {
+        CollectionPtr collection;
+        std::string name = "test_gc_c_" + std::to_string(i);
+        auto status = store_->CreateResource(Collection(name), collection);
+        ASSERT_TRUE(status.ok()) << status.ToString();
+        collections.push_back(collection);
+    }
+
+    auto start = std::chrono::system_clock::now();
+    for (auto& collection : collections) {
+        auto gc_event = std::make_shared<ResourceGCEvent<Collection>>(collection);
+        EventExecutor::GetInstance().Submit(gc_event);
+    }
+    auto end = std::chrono::system_clock::now();
+    auto count = std::chrono::duration_cast<std::chrono::milliseconds>(end - start).count();
+    ASSERT_LT(count, 1000);
+}

core/unittest/db/utils.cpp

@@ -137,7 +137,7 @@ void
 SnapshotTest::SetUp() {
     BaseTest::SetUp();
     DBOptions options;
-    options.meta_.path_ = "/tmp/milvus_ss";
+    options.meta_.path_ = "/tmp/milvus_ss/db";
     options.meta_.backend_uri_ = "mock://:@:/";
     options.wal_enable_ = false;
     BaseTest::SnapshotStart(true, options);
@@ -155,7 +155,7 @@ DBTest::GetOptions() {
     milvus::cache::CpuCacheMgr::GetInstance().SetCapacity(256 * milvus::engine::MB);
 
     auto options = DBOptions();
-    options.meta_.path_ = "/tmp/milvus_ss";
+    options.meta_.path_ = "/tmp/milvus_ss/db";
     options.meta_.backend_uri_ = "mock://:@:/";
     options.wal_enable_ = false;
     options.auto_flush_interval_ = 1;
@@ -165,7 +165,9 @@ DBTest::GetOptions() {
 void
 DBTest::SetUp() {
     BaseTest::SetUp();
-    BaseTest::SnapshotStart(false, GetOptions());
+    auto options = GetOptions();
+    std::experimental::filesystem::create_directories(options.meta_.path_);
+    BaseTest::SnapshotStart(false, options);
 
     dummy_context_ = std::make_shared<milvus::server::Context>("dummy_request_id");
     opentracing::mocktracer::MockTracerOptions tracer_options;
@@ -175,9 +177,6 @@ DBTest::SetUp() {
     auto trace_context = std::make_shared<milvus::tracing::TraceContext>(mock_span);
     dummy_context_->SetTraceContext(trace_context);
 
-    db_ = milvus::engine::DBFactory::BuildDB(GetOptions());
-    db_->Start();
-
     auto res_mgr = milvus::scheduler::ResMgrInst::GetInstance();
     res_mgr->Clear();
     res_mgr->Add(milvus::scheduler::ResourceFactory::Create("disk", "DISK", 0, false));
@@ -194,6 +193,9 @@ DBTest::SetUp() {
     milvus::scheduler::SchedInst::GetInstance()->Start();
     milvus::scheduler::JobMgrInst::GetInstance()->Start();
     milvus::scheduler::CPUBuilderInst::GetInstance()->Start();
+
+    db_ = milvus::engine::DBFactory::BuildDB(GetOptions());
+    db_->Start();
 }
 
 void
@@ -220,7 +222,7 @@ void
 SegmentTest::SetUp() {
     BaseTest::SetUp();
     DBOptions options;
-    options.meta_.path_ = "/tmp/milvus_ss";
+    options.meta_.path_ = "/tmp/milvus_ss/db";
     options.meta_.backend_uri_ = "mock://:@:/";
     options.wal_enable_ = false;
     BaseTest::SnapshotStart(false, options);
@@ -231,9 +233,9 @@ SegmentTest::SetUp() {
 
 void
 SegmentTest::TearDown() {
-    BaseTest::SnapshotStop();
     db_->Stop();
     db_ = nullptr;
+    BaseTest::SnapshotStop();
     BaseTest::TearDown();
 }
 
@@ -257,7 +259,7 @@ void
 SchedulerTest::SetUp() {
     BaseTest::SetUp();
     DBOptions options;
-    options.meta_.path_ = "/tmp/milvus_ss";
+    options.meta_.path_ = "/tmp/milvus_ss/db";
     options.meta_.backend_uri_ = "mock://:@:/";
     options.wal_enable_ = false;
     BaseTest::SnapshotStart(true, options);
@@ -303,10 +305,17 @@ EventTest::SetUp() {
     auto uri = "mock://:@:/";
     store_ = Store::Build(uri, "/tmp/milvus_ss/db");
     store_->DoReset();
+
+    milvus::engine::snapshot::OperationExecutor::Init(store_);
+    milvus::engine::snapshot::OperationExecutor::GetInstance().Start();
+    milvus::engine::snapshot::EventExecutor::Init(store_);
+    milvus::engine::snapshot::EventExecutor::GetInstance().Start();
 }
 
 void
 EventTest::TearDown() {
+    milvus::engine::snapshot::EventExecutor::GetInstance().Stop();
+    milvus::engine::snapshot::OperationExecutor::GetInstance().Stop();
 }
 
 /////////////////////////////////////////////////////////////////////////////////////////////////////////////////

core/unittest/server/test_web.cpp

@@ -26,6 +26,7 @@
 #include "config/ConfigMgr.h"
 #include "db/snapshot/EventExecutor.h"
 #include "db/snapshot/OperationExecutor.h"
+#include "db/snapshot/Snapshots.h"
 #include "scheduler/ResourceFactory.h"
 #include "scheduler/SchedInst.h"
 #include "server/DBWrapper.h"
@@ -158,11 +159,11 @@ static const char* CONTROLLER_TEST_VALID_CONFIG_STR =
     "\n"
     "general:\n"
     "  timezone: UTC+8\n"
-    "  meta_uri: sqlite://:@:/\n"
+    "  meta_uri: mock://:@:/\n"
     "\n"
     "network:\n"
     "  bind.address: 0.0.0.0\n"
-    "  bind.port: 19530\n"
+    "  bind.port: 19540\n"
     "  http.enable: true\n"
     "  http.port: 29999\n"
     "\n"
@@ -323,7 +324,18 @@ class WebControllerTest : public ::testing::Test {
         fs.close();
 
         milvus::ConfigMgr::GetInstance().Init();
-        milvus::ConfigMgr::GetInstance().Load(config_path);
+//        milvus::ConfigMgr::GetInstance().Set("general.meta_uri", "mock://:@:/");
+//        milvus::ConfigMgr::GetInstance().Set("storage.path", CONTROLLER_TEST_CONFIG_DIR);
+//        milvus::ConfigMgr::GetInstance().Set("network.http.enable", "true");
+//        milvus::ConfigMgr::GetInstance().Set("network.http.port", "20121");
+
+        auto& config = milvus::ConfigMgr::GetInstance();
+
+//        milvus::ConfigMgr::GetInstance().Init();
+        config.Load(config_path);
+//        milvus::ConfigMgr::GetInstance().Set("general.meta_uri", "mock://:@:/");
+
+        milvus::engine::snapshot::Snapshots::GetInstance().StartService();
 
         auto res_mgr = milvus::scheduler::ResMgrInst::GetInstance();
         res_mgr->Clear();
@@ -358,7 +370,6 @@ class WebControllerTest : public ::testing::Test {
         fs.flush();
         fs.close();
 
-        //        milvus::ConfigMgr::GetInstance().Init();
         milvus::ConfigMgr::GetInstance().Load(config_path);
 
         OATPP_COMPONENT(std::shared_ptr<oatpp::network::ClientConnectionProvider>, clientConnectionProvider);
@@ -382,6 +393,8 @@ class WebControllerTest : public ::testing::Test {
         milvus::scheduler::ResMgrInst::GetInstance()->Stop();
         milvus::scheduler::ResMgrInst::GetInstance()->Clear();
 
+        milvus::engine::snapshot::Snapshots::GetInstance().StopService();
+
         boost::filesystem::remove_all(CONTROLLER_TEST_CONFIG_DIR);
     }