fix(db): db desctruction bug fix

Former-commit-id: e456a0d76c2b1401107d47600c34d4622ca8cd8a

cpp/src/db/DBImpl.cpp

@@ -400,6 +400,7 @@ DBImpl<EngineT>::~DBImpl() {
     }
     std::vector<std::string> ids;
     _pMemMgr->serialize(ids);
+    _env->Stop();
 }
 
 /*

cpp/src/db/Env.cpp

@@ -3,18 +3,23 @@
  * Unauthorized copying of this file, via any medium is strictly prohibited.
  * Proprietary and confidential.
  ******************************************************************************/
+#include <easylogging++.h>
 #include <assert.h>
+#include <atomic>
 #include "Env.h"
 
 namespace zilliz {
 namespace vecwise {
 namespace engine {
 
-Env::Env() : _bg_work_started(false) {
+Env::Env()
+    : _bg_work_started(false),
+      _shutting_down(false) {
 }
 
 void Env::schedule(void (*function_)(void* arg_), void* arg_) {
     std::unique_lock<std::mutex> lock(_bg_work_mutex);
+    if (_shutting_down) return;
 
     if (!_bg_work_started) {
         _bg_work_started = true;
@@ -30,12 +35,14 @@ void Env::schedule(void (*function_)(void* arg_), void* arg_) {
 }
 
 void Env::backgroud_thread_main() {
-    while (true) {
+    while (!_shutting_down) {
         std::unique_lock<std::mutex> lock(_bg_work_mutex);
-        while (_bg_work_queue.empty()) {
+        while (_bg_work_queue.empty() && !_shutting_down) {
             _bg_work_cv.wait(lock);
         }
 
+        if (_shutting_down) break;
+
         assert(!_bg_work_queue.empty());
         auto bg_function = _bg_work_queue.front()._function;
         void* bg_arg = _bg_work_queue.front()._arg;
@@ -44,6 +51,28 @@ void Env::backgroud_thread_main() {
         lock.unlock();
         bg_function(bg_arg);
     }
+
+    std::unique_lock<std::mutex> lock(_bg_work_mutex);
+    _bg_work_started = false;
+    _bg_work_cv.notify_all();
+}
+
+void Env::Stop() {
+    {
+        std::unique_lock<std::mutex> lock(_bg_work_mutex);
+        if (_shutting_down || !_bg_work_started) return;
+    }
+    _shutting_down = true;
+    {
+        std::unique_lock<std::mutex> lock(_bg_work_mutex);
+        if (_bg_work_queue.empty()) {
+            _bg_work_cv.notify_one();
+        }
+        while (_bg_work_started) {
+            _bg_work_cv.wait(lock);
+        }
+    }
+    _shutting_down = false;
 }
 
 Env::~Env() {}

cpp/src/db/Env.h

@@ -9,6 +9,7 @@
 #include <thread>
 #include <mutex>
 #include <queue>
+#include <atomic>
 
 namespace zilliz {
 namespace vecwise {
@@ -23,6 +24,8 @@ public:
 
     void schedule(void (*function_)(void* arg_), void* arg_);
 
+    virtual void Stop();
+
     virtual ~Env();
 
     static Env* Default();
@@ -45,6 +48,7 @@ protected:
     std::condition_variable _bg_work_cv;
     std::queue<BGWork> _bg_work_queue;
     bool _bg_work_started;
+    std::atomic<bool> _shutting_down;
 
 }; // Env