find disk damage, report immediately

be/src/agent/task_worker_pool.cpp

@@ -71,6 +71,8 @@ map<TTaskType::type, map<string, uint32_t>> TaskWorkerPool::_s_running_task_user
 map<TTaskType::type, map<string, uint32_t>> TaskWorkerPool::_s_total_task_user_count;
 map<TTaskType::type, uint32_t> TaskWorkerPool::_s_total_task_count;
 FrontendServiceClientCache TaskWorkerPool::_master_service_client_cache;
+boost::mutex TaskWorkerPool::_disk_broken_lock;
+boost::posix_time::time_duration TaskWorkerPool::_wait_duration;
 
 TaskWorkerPool::TaskWorkerPool(
         const TaskWorkerType task_worker_type,
@@ -146,10 +148,12 @@ void TaskWorkerPool::start() {
         _callback_function = _report_task_worker_thread_callback;
         break;
     case TaskWorkerType::REPORT_DISK_STATE:
+        _wait_duration = boost::posix_time::time_duration(0, 0, config::report_disk_state_interval_seconds, 0);
         _worker_count = REPORT_DISK_STATE_WORKER_COUNT;
         _callback_function = _report_disk_state_worker_thread_callback;
         break;
     case TaskWorkerType::REPORT_OLAP_TABLE:
+        _wait_duration = boost::posix_time::time_duration(0, 0, config::report_disk_state_interval_seconds, 0);
         _worker_count = REPORT_OLAP_TABLE_WORKER_COUNT;
         _callback_function = _report_olap_table_worker_thread_callback;
         break;
@@ -1525,15 +1529,6 @@ void* TaskWorkerPool::_report_disk_state_worker_thread_callback(void* arg_this)
 
         OLAPStatus get_all_root_path_stat =
                 worker_pool_this->_command_executor->get_all_root_path_stat(&root_paths_stat);
-        if (get_all_root_path_stat != OLAPStatus::OLAP_SUCCESS) {
-            OLAP_LOG_WARNING("fail to get all root path stat.");
-#ifndef BE_TEST
-            sleep(config::report_disk_state_interval_seconds);
-            continue;
-#else
-            return (void*)0;
-#endif
-        }
 
         map<string, TDisk> disks;
         for (auto root_path_state : root_paths_stat) {
@@ -1558,7 +1553,15 @@ void* TaskWorkerPool::_report_disk_state_worker_thread_callback(void* arg_this)
         }
 
 #ifndef BE_TEST
-        sleep(config::report_disk_state_interval_seconds);
+        {
+            // wait disk_broken_cv awaken
+            // if awaken, set is_report_disk_state_already to true, it will not notify again
+            // if overtime, while will go to next cycle
+            boost::unique_lock<boost::mutex> lk(_disk_broken_lock);
+            if (OLAPRootPath::get_instance()->disk_broken_cv.timed_wait(lk, _wait_duration)) {
+                OLAPRootPath::get_instance()->is_report_disk_state_already = true;
+            }
+        }
     }
 #endif
 
@@ -1588,7 +1591,13 @@ void* TaskWorkerPool::_report_olap_table_worker_thread_callback(void* arg_this)
             OLAP_LOG_WARNING("report get all tablets info failed. status: %d",
                              report_all_tablets_info_status);
 #ifndef BE_TEST
-            sleep(config::report_olap_table_interval_seconds);
+            // wait disk_broken_cv awaken
+            // if awaken, set is_report_olap_table_already to true, it will not notify again
+            // if overtime, while will go to next cycle
+            boost::unique_lock<boost::mutex> lk(_disk_broken_lock);
+            if (OLAPRootPath::get_instance()->disk_broken_cv.timed_wait(lk, _wait_duration)) {
+                OLAPRootPath::get_instance()->is_report_olap_table_already =  true;
+            }
             continue;
 #else
             return (void*)0;
@@ -1606,7 +1615,13 @@ void* TaskWorkerPool::_report_olap_table_worker_thread_callback(void* arg_this)
         }
 
 #ifndef BE_TEST
-        sleep(config::report_olap_table_interval_seconds);
+        // wait disk_broken_cv awaken
+        // if awaken, set is_report_olap_table_already to true, it will not notify again
+        // if overtime, while will go to next cycle
+        boost::unique_lock<boost::mutex> lk(_disk_broken_lock);
+        if (OLAPRootPath::get_instance()->disk_broken_cv.timed_wait(lk, _wait_duration)) {
+            OLAPRootPath::get_instance()->is_report_olap_table_already =  true;
+        }
     }
 #endif
 

be/src/agent/task_worker_pool.h

@@ -17,6 +17,9 @@
 #define BDG_PALO_BE_SRC_TASK_WORKER_POOL_H
 
 #include <atomic>
+#include <boost/date_time/posix_time/posix_time.hpp>
+#include <boost/thread/condition_variable.hpp>
+#include <boost/thread/mutex.hpp>
 #include <deque>
 #include <utility>
 #include <vector>
@@ -27,6 +30,7 @@
 #include "gen_cpp/HeartbeatService_types.h"
 #include "olap/command_executor.h"
 #include "olap/olap_define.h"
+#include "olap/olap_rootpath.h"
 #include "olap/utils.h"
 
 namespace palo {
@@ -147,6 +151,9 @@ class TaskWorkerPool {
     static MutexLock _s_running_task_user_count_lock;
     static FrontendServiceClientCache _master_service_client_cache;
 
+    static boost::mutex _disk_broken_lock;
+    static boost::posix_time::time_duration _wait_duration;
+
     DISALLOW_COPY_AND_ASSIGN(TaskWorkerPool);
 };  // class TaskWorkerPool
 }  // namespace palo

be/src/olap/olap_rootpath.cpp

@@ -76,7 +76,10 @@ OLAPRootPath::OLAPRootPath() :
         _total_storage_medium_type_count(0),
         _available_storage_medium_type_count(0),
         _effective_cluster_id(-1),
-        _is_all_cluster_id_exist(true) {}
+        _is_all_cluster_id_exist(true),
+        _is_drop_tables(false),
+        is_report_disk_state_already(false),
+        is_report_olap_table_already(false) {}
 
 OLAPRootPath::~OLAPRootPath() {
     clear();
@@ -435,8 +438,25 @@ void OLAPRootPath::start_disk_stat_monitor() {
     _start_check_disks();
     _detect_unused_flag();
     _delete_tables_on_unused_root_path();
+
+    // if drop tables
+    // notify disk_state_worker_thread and olap_table_worker_thread until they received
+    if (_is_drop_tables) {
+        disk_broken_cv.notify_all();
+
+        bool is_report_disk_state_expected = true;
+        bool is_report_olap_table_expected = true;
+        bool is_report_disk_state_exchanged = 
+                is_report_disk_state_already.compare_exchange_strong(is_report_disk_state_expected, false);
+        bool is_report_olap_table_exchanged =
+                is_report_olap_table_already.compare_exchange_strong(is_report_olap_table_expected, false);
+        if (is_report_disk_state_exchanged && is_report_olap_table_exchanged) {
+            _is_drop_tables = false;
+        }
+    }
 }
 
+
 void OLAPRootPath::_start_check_disks() {
     OLAPRootPath::RootPathVec all_available_root_path;
     get_all_available_root_path(&all_available_root_path);
@@ -940,6 +960,10 @@ void OLAPRootPath::_delete_tables_on_unused_root_path() {
         exit(0);
     }
 
+    if (!table_info_vec.empty()) {
+        _is_drop_tables = true;
+    }
+
     OLAPEngine::get_instance()->drop_tables_on_error_root_path(table_info_vec);
 }
 

be/src/olap/olap_rootpath.h

@@ -16,6 +16,8 @@
 #ifndef BDG_PALO_BE_SRC_OLAP_OLAP_ROOTPATH_H
 #define BDG_PALO_BE_SRC_OLAP_OLAP_ROOTPATH_H
 
+#include <atomic>
+#include <boost/thread/condition_variable.hpp>
 #include <list>
 #include <memory>
 #include <queue>
@@ -130,6 +132,10 @@ class OLAPRootPath {
             const std::vector<std::string>& root_path_vec,
             const std::vector<bool>& is_accessable_vec);
 
+    boost::condition_variable disk_broken_cv;
+    std::atomic_bool is_report_disk_state_already;
+    std::atomic_bool is_report_olap_table_already;
+
 private:
     struct RootPathInfo {
         RootPathInfo():
@@ -221,6 +227,7 @@ class OLAPRootPath {
 
     int32_t _effective_cluster_id;
     bool _is_all_cluster_id_exist;
+    bool _is_drop_tables;
 
     // 错误磁盘所在百分比，超过设定的值，则engine需要退出运行
     uint32_t _min_percentage_of_error_disk;