chaoyli commented on a change in pull request #688: Add path GC for BE
URL: https://github.com/apache/incubator-doris/pull/688#discussion_r263261811
##########
File path: be/src/olap/storage_engine.cpp
##########
@@ -1021,4 +987,232 @@ OLAPStatus StorageEngine::execute_task(EngineTask*
task) {
}
}
+void StorageEngine::add_check_paths(std::set<std::string> paths) {
+ _check_path_mutex.wrlock();
+ _all_check_paths.insert(paths.begin(), paths.end());
+ _check_path_mutex.unlock();
+}
+
+void StorageEngine::remove_check_paths(std::set<std::string> paths) {
+ _check_path_mutex.wrlock();
+ _remove_check_paths_no_lock(paths);
+ _check_path_mutex.unlock();
+}
+
+void StorageEngine::add_pending_paths(int64_t id, std::set<std::string> paths)
{
+ WriteLock wr_lock(&_pending_path_mutex);
+ auto pending_paths= _pending_paths[id];
+ pending_paths.insert(paths.begin(), paths.end());
+}
+
+void StorageEngine::remove_pending_paths(int64_t id) {
+ WriteLock wr_lock(&_pending_path_mutex);
+ _pending_paths.erase(id);
+}
+
+bool StorageEngine::check_path_in_pending_paths(std::string path) {
+ ReadLock rd_lock(&_pending_path_mutex);
+ for (auto id_pending_paths : _pending_paths) {
+ if (id_pending_paths.second.find(path) !=
id_pending_paths.second.end()) {
+ return true;
+ }
+ }
+ return false;
+}
+
+void StorageEngine::process_garbage_path(std::string path) {
+ if (check_dir_existed(path)) {
+ LOG(INFO) << "collect garbage dir path:" << path;
+ OLAPStatus status = remove_all_dir(path);
+ if (status != OLAP_SUCCESS) {
+ LOG(WARNING) << "remove garbage dir path:" << path << " failed";
+ }
+ }
+}
+
+void StorageEngine::_perform_path_gc(void* arg) {
+ // init the set of valid path
+ // validate the path in data dir
+ LOG(INFO) << "start to path gc.";
+ int start = 0;
+ int step = config::path_gc_check_step;
+ while (true) {
+ _check_path_mutex.wrlock();
+ for (int index = start; index < start + step;) {
+ auto path_iter = std::next(_all_check_paths.begin(), index);
+ if (path_iter == _all_check_paths.end()) {
+ break;
+ }
+ std::string path = *path_iter;
+ TTabletId tablet_id = -1;
+ TSchemaHash schema_hash = -1;
+ bool is_valid =
TabletManager::instance()->get_tablet_id_and_schema_hash_from_path(path,
+ &tablet_id, &schema_hash);
+ std::set<std::string> paths;
+ paths.insert(path);
+ if (!is_valid) {
+ LOG(WARNING) << "unknow path:" << path;
+ _remove_check_paths_no_lock(paths);
+ continue;
+ } else {
+ if (tablet_id >0 && schema_hash >0) {
+ // tablet schema hash path or rowset file path
+ TabletSharedPtr tablet =
TabletManager::instance()->get_tablet(tablet_id, schema_hash);
+ if (tablet == nullptr) {
+ bool exist_in_pending =
check_path_in_pending_paths(path);
+ if (!exist_in_pending) {
+ process_garbage_path(path);
+ _remove_check_paths_no_lock(paths);
+ continue;
+ }
+ } else {
+ bool valid = tablet->check_path(path);
+ if (!valid) {
+ bool exist_in_pending =
check_path_in_pending_paths(path);
+ if (!exist_in_pending) {
+ process_garbage_path(path);
+ _remove_check_paths_no_lock(paths);
+ continue;
+ }
+ }
+ }
+ } else if (tablet_id >0 && schema_hash <= 0) {
+ // tablet id path
+ if (FileUtils::is_dir(path)) {
+ bool exist =
TabletManager::instance()->check_tablet_id_exist(tablet_id);
+ if (!exist) {
+ bool exist_in_pending =
check_path_in_pending_paths(path);
+ if (!exist_in_pending) {
+ process_garbage_path(path);
+ _remove_check_paths_no_lock(paths);
+ continue;
+ }
+ }
+ } else {
+ LOG(WARNING) << "unknown path:" << path;
+ _remove_check_paths_no_lock(paths);
+ continue;
+ }
+ } else {
+ LOG(WARNING) << "unknown path:" << path;
+ _remove_check_paths_no_lock(paths);
+ continue;
+ }
+ }
+ ++index;
+ }
+
+ start += step;
+ if (start >= _all_check_paths.size()) {
+ _check_path_mutex.unlock();
+ break;
+ }
+ _check_path_mutex.unlock();
+ sleep(config::path_gc_check_step_interval_ms);
+ }
+ LOG(INFO) << "finished one time path gc.";
+}
+
+void* StorageEngine::_path_gc_thread_callback(void* arg) {
+#ifdef GOOGLE_PROFILER
+ ProfilerRegisterThread();
+#endif
+
+ LOG(INFO) << "try to start path gc thread!";
+ uint32_t interval = config::path_gc_check_interval_second;
+ if (interval <= 0) {
+ LOG(WARNING) << "path gc thread check interval config is illegal:" <<
interval
+ << "will be forced set to half hour";
+ interval = 1800; // 0.5 hour
+ }
+
+ while (true) {
+ LOG(INFO) << "try to perform path gc!";
+ CgroupsMgr::apply_system_cgroup();
Review comment:
no need to use cgroup
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
