This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-doris.git
The following commit(s) were added to refs/heads/master by this push:
new c69dd54 [refactor](mutex) Use std::mutex to replace Mutex and
refactor some lock logic (#8452)
c69dd54 is described below
commit c69dd54116bee30f93566cb693a9604eac0d1399
Author: caiconghui <[email protected]>
AuthorDate: Thu Mar 24 14:50:02 2022 +0800
[refactor](mutex) Use std::mutex to replace Mutex and refactor some lock
logic (#8452)
---
be/src/http/action/compaction_action.cpp | 8 ++---
be/src/olap/base_compaction.cpp | 8 ++---
be/src/olap/cumulative_compaction.cpp | 8 ++---
be/src/olap/delta_writer.cpp | 2 +-
be/src/olap/lru_cache.cpp | 12 +++----
be/src/olap/lru_cache.h | 2 +-
be/src/olap/push_handler.cpp | 9 ++---
be/src/olap/schema_change.cpp | 11 ++----
be/src/olap/storage_engine.cpp | 8 ++---
be/src/olap/storage_engine.h | 2 +-
be/src/olap/tablet.h | 24 ++++---------
be/src/olap/tablet_manager.cpp | 8 ++---
be/src/olap/task/engine_clone_task.cpp | 12 +++----
be/src/olap/task/engine_storage_migration_task.cpp | 6 +---
be/src/util/mutex.cpp | 40 ----------------------
15 files changed, 48 insertions(+), 112 deletions(-)
diff --git a/be/src/http/action/compaction_action.cpp
b/be/src/http/action/compaction_action.cpp
index 157e7cc..e6fb790 100644
--- a/be/src/http/action/compaction_action.cpp
+++ b/be/src/http/action/compaction_action.cpp
@@ -184,8 +184,8 @@ Status
CompactionAction::_handle_run_status_compaction(HttpRequest* req, std::st
{
// use try lock to check this tablet is running cumulative
compaction
- MutexLock lock_cumulative(tablet->get_cumulative_lock(), TRY_LOCK);
- if (!lock_cumulative.own_lock()) {
+ std::unique_lock<std::mutex>
lock_cumulative(tablet->get_cumulative_compaction_lock(), std::try_to_lock);
+ if (!lock_cumulative.owns_lock()) {
msg = "compaction task for this tablet is running";
compaction_type = "cumulative";
run_status = 1;
@@ -197,8 +197,8 @@ Status
CompactionAction::_handle_run_status_compaction(HttpRequest* req, std::st
{
// use try lock to check this tablet is running base compaction
- MutexLock lock_base(tablet->get_base_lock(), TRY_LOCK);
- if (!lock_base.own_lock()) {
+ std::unique_lock<std::mutex>
lock_base(tablet->get_base_compaction_lock(), std::try_to_lock);
+ if (!lock_base.owns_lock()) {
msg = "compaction task for this tablet is running";
compaction_type = "base";
run_status = 1;
diff --git a/be/src/olap/base_compaction.cpp b/be/src/olap/base_compaction.cpp
index 986044e..d379957 100644
--- a/be/src/olap/base_compaction.cpp
+++ b/be/src/olap/base_compaction.cpp
@@ -33,8 +33,8 @@ OLAPStatus BaseCompaction::prepare_compact() {
return OLAP_ERR_INPUT_PARAMETER_ERROR;
}
- MutexLock lock(_tablet->get_base_lock(), TRY_LOCK);
- if (!lock.own_lock()) {
+ std::unique_lock<std::mutex> lock(_tablet->get_base_compaction_lock(),
std::try_to_lock);
+ if (!lock.owns_lock()) {
LOG(WARNING) << "another base compaction is running. tablet=" <<
_tablet->full_name();
return OLAP_ERR_BE_TRY_BE_LOCK_ERROR;
}
@@ -50,8 +50,8 @@ OLAPStatus BaseCompaction::prepare_compact() {
}
OLAPStatus BaseCompaction::execute_compact_impl() {
- MutexLock lock(_tablet->get_base_lock(), TRY_LOCK);
- if (!lock.own_lock()) {
+ std::unique_lock<std::mutex> lock(_tablet->get_base_compaction_lock(),
std::try_to_lock);
+ if (!lock.owns_lock()) {
LOG(WARNING) << "another base compaction is running. tablet=" <<
_tablet->full_name();
return OLAP_ERR_BE_TRY_BE_LOCK_ERROR;
}
diff --git a/be/src/olap/cumulative_compaction.cpp
b/be/src/olap/cumulative_compaction.cpp
index 0e721a8..94e0a83 100644
--- a/be/src/olap/cumulative_compaction.cpp
+++ b/be/src/olap/cumulative_compaction.cpp
@@ -33,8 +33,8 @@ OLAPStatus CumulativeCompaction::prepare_compact() {
return OLAP_ERR_CUMULATIVE_INVALID_PARAMETERS;
}
- MutexLock lock(_tablet->get_cumulative_lock(), TRY_LOCK);
- if (!lock.own_lock()) {
+ std::unique_lock<std::mutex>
lock(_tablet->get_cumulative_compaction_lock(), std::try_to_lock);
+ if (!lock.owns_lock()) {
LOG(INFO) << "The tablet is under cumulative compaction. tablet=" <<
_tablet->full_name();
return OLAP_ERR_CE_TRY_CE_LOCK_ERROR;
}
@@ -56,8 +56,8 @@ OLAPStatus CumulativeCompaction::prepare_compact() {
}
OLAPStatus CumulativeCompaction::execute_compact_impl() {
- MutexLock lock(_tablet->get_cumulative_lock(), TRY_LOCK);
- if (!lock.own_lock()) {
+ std::unique_lock<std::mutex>
lock(_tablet->get_cumulative_compaction_lock(), std::try_to_lock);
+ if (!lock.owns_lock()) {
LOG(INFO) << "The tablet is under cumulative compaction. tablet=" <<
_tablet->full_name();
return OLAP_ERR_CE_TRY_CE_LOCK_ERROR;
}
diff --git a/be/src/olap/delta_writer.cpp b/be/src/olap/delta_writer.cpp
index 2080aa3..7dc985f 100644
--- a/be/src/olap/delta_writer.cpp
+++ b/be/src/olap/delta_writer.cpp
@@ -109,7 +109,7 @@ OLAPStatus DeltaWriter::init() {
if (!base_migration_rlock.owns_lock()) {
return OLAP_ERR_RWLOCK_ERROR;
}
- MutexLock push_lock(_tablet->get_push_lock());
+ std::lock_guard<std::mutex> push_lock(_tablet->get_push_lock());
RETURN_NOT_OK(_storage_engine->txn_manager()->prepare_txn(_req.partition_id,
_tablet,
_req.txn_id,
_req.load_id));
}
diff --git a/be/src/olap/lru_cache.cpp b/be/src/olap/lru_cache.cpp
index e494c54..5d56499 100644
--- a/be/src/olap/lru_cache.cpp
+++ b/be/src/olap/lru_cache.cpp
@@ -209,7 +209,7 @@ void LRUCache::_lru_append(LRUHandle* list, LRUHandle* e) {
}
Cache::Handle* LRUCache::lookup(const CacheKey& key, uint32_t hash) {
- MutexLock l(&_mutex);
+ std::lock_guard<std::mutex> l(_mutex);
++_lookup_count;
LRUHandle* e = _table.lookup(key, hash);
if (e != nullptr) {
@@ -232,7 +232,7 @@ void LRUCache::release(Cache::Handle* handle) {
LRUHandle* e = reinterpret_cast<LRUHandle*>(handle);
bool last_ref = false;
{
- MutexLock l(&_mutex);
+ std::lock_guard<std::mutex> l(_mutex);
last_ref = _unref(e);
if (last_ref) {
_usage -= e->total_size;
@@ -309,7 +309,7 @@ Cache::Handle* LRUCache::insert(const CacheKey& key,
uint32_t hash, void* value,
memcpy(e->key_data, key.data(), key.size());
LRUHandle* to_remove_head = nullptr;
{
- MutexLock l(&_mutex);
+ std::lock_guard<std::mutex> l(_mutex);
// Free the space following strict LRU policy until enough space
// is freed or the lru list is empty
@@ -348,7 +348,7 @@ void LRUCache::erase(const CacheKey& key, uint32_t hash,
MemTracker* tracker) {
LRUHandle* e = nullptr;
bool last_ref = false;
{
- MutexLock l(&_mutex);
+ std::lock_guard<std::mutex> l(_mutex);
e = _table.remove(key, hash);
if (e != nullptr) {
last_ref = _unref(e);
@@ -371,7 +371,7 @@ void LRUCache::erase(const CacheKey& key, uint32_t hash,
MemTracker* tracker) {
int64_t LRUCache::prune() {
LRUHandle* to_remove_head = nullptr;
{
- MutexLock l(&_mutex);
+ std::lock_guard<std::mutex> l(_mutex);
while (_lru_normal.next != &_lru_normal) {
LRUHandle* old = _lru_normal.next;
_evict_one_entry(old);
@@ -398,7 +398,7 @@ int64_t LRUCache::prune() {
int64_t LRUCache::prune_if(CacheValuePredicate pred) {
LRUHandle* to_remove_head = nullptr;
{
- MutexLock l(&_mutex);
+ std::lock_guard<std::mutex> l(_mutex);
LRUHandle* p = _lru_normal.next;
while (p != &_lru_normal) {
LRUHandle* next = p->next;
diff --git a/be/src/olap/lru_cache.h b/be/src/olap/lru_cache.h
index fd81cd8..9a20843 100644
--- a/be/src/olap/lru_cache.h
+++ b/be/src/olap/lru_cache.h
@@ -339,7 +339,7 @@ private:
size_t _capacity = 0;
// _mutex protects the following state.
- Mutex _mutex;
+ std::mutex _mutex;
size_t _usage = 0;
// Dummy head of LRU list.
diff --git a/be/src/olap/push_handler.cpp b/be/src/olap/push_handler.cpp
index 9d7855e..c420bcb 100644
--- a/be/src/olap/push_handler.cpp
+++ b/be/src/olap/push_handler.cpp
@@ -91,13 +91,14 @@ OLAPStatus
PushHandler::_do_streaming_ingestion(TabletSharedPtr tablet, const TP
if (!base_migration_rlock.owns_lock()) {
return OLAP_ERR_RWLOCK_ERROR;
}
- tablet->obtain_push_lock();
PUniqueId load_id;
load_id.set_hi(0);
load_id.set_lo(0);
- RETURN_NOT_OK(StorageEngine::instance()->txn_manager()->prepare_txn(
- request.partition_id, tablet, request.transaction_id, load_id));
- tablet->release_push_lock();
+ {
+ std::lock_guard<std::mutex> push_lock(tablet->get_push_lock());
+ RETURN_NOT_OK(StorageEngine::instance()->txn_manager()->prepare_txn(
+ request.partition_id, tablet, request.transaction_id,
load_id));
+ }
if (tablet_vars->size() == 1) {
tablet_vars->resize(2);
diff --git a/be/src/olap/schema_change.cpp b/be/src/olap/schema_change.cpp
index bc88142..caf9e96 100644
--- a/be/src/olap/schema_change.cpp
+++ b/be/src/olap/schema_change.cpp
@@ -1477,9 +1477,9 @@ OLAPStatus
SchemaChangeHandler::_do_process_alter_tablet_v2(const TAlterTabletRe
// begin to find deltas to convert from base tablet to new tablet so that
// obtain base tablet and new tablet's push lock and header write lock to
prevent loading data
- base_tablet->obtain_push_lock();
- new_tablet->obtain_push_lock();
{
+ std::lock_guard<std::mutex>
base_tablet_lock(base_tablet->get_push_lock());
+ std::lock_guard<std::mutex>
new_tablet_lock(new_tablet->get_push_lock());
WriteLock base_tablet_rdlock(base_tablet->get_header_lock());
WriteLock new_tablet_rdlock(new_tablet->get_header_lock());
// check if the tablet has alter task
@@ -1589,8 +1589,6 @@ OLAPStatus
SchemaChangeHandler::_do_process_alter_tablet_v2(const TAlterTabletRe
} while (0);
}
- new_tablet->release_push_lock();
- base_tablet->release_push_lock();
do {
if (res != OLAP_SUCCESS) {
@@ -1911,11 +1909,10 @@ OLAPStatus
SchemaChangeHandler::_convert_historical_rowsets(const SchemaChangePa
rowset_writer->rowset_id().to_string());
// Add the new version of the data to the header
// In order to prevent the occurrence of deadlock, we must first lock
the old table, and then lock the new table
- sc_params.new_tablet->obtain_push_lock();
+ std::lock_guard<std::mutex>
lock(sc_params.new_tablet->get_push_lock());
RowsetSharedPtr new_rowset = rowset_writer->build();
if (new_rowset == nullptr) {
LOG(WARNING) << "failed to build rowset, exit alter process";
- sc_params.new_tablet->release_push_lock();
goto PROCESS_ALTER_EXIT;
}
res = sc_params.new_tablet->add_rowset(new_rowset, false);
@@ -1931,14 +1928,12 @@ OLAPStatus
SchemaChangeHandler::_convert_historical_rowsets(const SchemaChangePa
<< ", version=" << rs_reader->version().first << "-"
<< rs_reader->version().second;
StorageEngine::instance()->add_unused_rowset(new_rowset);
- sc_params.new_tablet->release_push_lock();
goto PROCESS_ALTER_EXIT;
} else {
VLOG_NOTICE << "register new version. tablet=" <<
sc_params.new_tablet->full_name()
<< ", version=" << rs_reader->version().first << "-"
<< rs_reader->version().second;
}
- sc_params.new_tablet->release_push_lock();
VLOG_TRACE << "succeed to convert a history version."
<< " version=" << rs_reader->version().first << "-"
diff --git a/be/src/olap/storage_engine.cpp b/be/src/olap/storage_engine.cpp
index e831d02..dc54fdb 100644
--- a/be/src/olap/storage_engine.cpp
+++ b/be/src/olap/storage_engine.cpp
@@ -136,7 +136,7 @@ StorageEngine::StorageEngine(const EngineOptions& options)
_s_instance = this;
}
REGISTER_HOOK_METRIC(unused_rowsets_count, [this]() {
- MutexLock lock(&_gc_mutex);
+ std::lock_guard<std::mutex> lock(_gc_mutex);
return _unused_rowsets.size();
});
REGISTER_HOOK_METRIC(compaction_mem_consumption, [this]() {
@@ -871,7 +871,7 @@ void StorageEngine::_parse_default_rowset_type() {
}
void StorageEngine::start_delete_unused_rowset() {
- MutexLock lock(&_gc_mutex);
+ std::lock_guard<std::mutex> lock(_gc_mutex);
for (auto it = _unused_rowsets.begin(); it != _unused_rowsets.end();) {
if (it->second.use_count() != 1) {
++it;
@@ -898,7 +898,7 @@ void StorageEngine::add_unused_rowset(RowsetSharedPtr
rowset) {
auto rowset_id = rowset->rowset_id().to_string();
- MutexLock lock(&_gc_mutex);
+ std::lock_guard<std::mutex> lock(_gc_mutex);
auto it = _unused_rowsets.find(rowset_id);
if (it == _unused_rowsets.end()) {
rowset->set_need_delete_file();
@@ -1074,7 +1074,7 @@ OLAPStatus StorageEngine::execute_task(EngineTask* task) {
// check whether any unused rowsets's id equal to rowset_id
bool StorageEngine::check_rowset_id_in_unused_rowsets(const RowsetId&
rowset_id) {
- MutexLock lock(&_gc_mutex);
+ std::lock_guard<std::mutex> lock(_gc_mutex);
auto search = _unused_rowsets.find(rowset_id.to_string());
return search != _unused_rowsets.end();
}
diff --git a/be/src/olap/storage_engine.h b/be/src/olap/storage_engine.h
index 0cd1675..5dbb569 100644
--- a/be/src/olap/storage_engine.h
+++ b/be/src/olap/storage_engine.h
@@ -322,7 +322,7 @@ private:
static StorageEngine* _s_instance;
- Mutex _gc_mutex;
+ std::mutex _gc_mutex;
// map<rowset_id(str), RowsetSharedPtr>, if we use RowsetId as the key, we
need custom hash func
std::unordered_map<std::string, RowsetSharedPtr> _unused_rowsets;
diff --git a/be/src/olap/tablet.h b/be/src/olap/tablet.h
index 9571e74..a506ae9 100644
--- a/be/src/olap/tablet.h
+++ b/be/src/olap/tablet.h
@@ -141,21 +141,9 @@ public:
// meta lock
inline std::shared_mutex& get_header_lock() { return _meta_lock; }
-
- // ingest lock
- inline void obtain_push_lock() { _ingest_lock.lock(); }
- inline void release_push_lock() { _ingest_lock.unlock(); }
- inline Mutex* get_push_lock() { return &_ingest_lock; }
-
- // base lock
- inline void obtain_base_compaction_lock() { _base_lock.lock(); }
- inline void release_base_compaction_lock() { _base_lock.unlock(); }
- inline Mutex* get_base_lock() { return &_base_lock; }
-
- // cumulative lock
- inline void obtain_cumulative_lock() { _cumulative_lock.lock(); }
- inline void release_cumulative_lock() { _cumulative_lock.unlock(); }
- inline Mutex* get_cumulative_lock() { return &_cumulative_lock; }
+ inline std::mutex& get_push_lock() { return _ingest_lock; }
+ inline std::mutex& get_base_compaction_lock() { return
_base_compaction_lock; }
+ inline std::mutex& get_cumulative_compaction_lock() { return
_cumulative_compaction_lock; }
inline std::shared_mutex& get_migration_lock() { return _migration_lock; }
@@ -303,9 +291,9 @@ private:
// meta store lock is used for prevent 2 threads do checkpoint concurrently
// it will be used in econ-mode in the future
std::shared_mutex _meta_store_lock;
- Mutex _ingest_lock;
- Mutex _base_lock;
- Mutex _cumulative_lock;
+ std::mutex _ingest_lock;
+ std::mutex _base_compaction_lock;
+ std::mutex _cumulative_compaction_lock;
std::mutex _schema_change_lock;
std::shared_mutex _migration_lock;
diff --git a/be/src/olap/tablet_manager.cpp b/be/src/olap/tablet_manager.cpp
index ed1a711..5e48f62 100644
--- a/be/src/olap/tablet_manager.cpp
+++ b/be/src/olap/tablet_manager.cpp
@@ -618,14 +618,14 @@ TabletSharedPtr
TabletManager::find_best_tablet_to_compaction(
}
if (compaction_type == CompactionType::BASE_COMPACTION) {
- MutexLock lock(tablet_ptr->get_base_lock(), TRY_LOCK);
- if (!lock.own_lock()) {
+ std::unique_lock<std::mutex>
lock(tablet_ptr->get_base_compaction_lock(), std::try_to_lock);
+ if (!lock.owns_lock()) {
LOG(INFO) << "can not get base lock: " <<
tablet_ptr->tablet_id();
continue;
}
} else {
- MutexLock lock(tablet_ptr->get_cumulative_lock(), TRY_LOCK);
- if (!lock.own_lock()) {
+ std::unique_lock<std::mutex>
lock(tablet_ptr->get_cumulative_compaction_lock(), std::try_to_lock);
+ if (!lock.owns_lock()) {
LOG(INFO) << "can not get cumu lock: " <<
tablet_ptr->tablet_id();
continue;
}
diff --git a/be/src/olap/task/engine_clone_task.cpp
b/be/src/olap/task/engine_clone_task.cpp
index 6231a96..b33b1c9 100644
--- a/be/src/olap/task/engine_clone_task.cpp
+++ b/be/src/olap/task/engine_clone_task.cpp
@@ -525,11 +525,11 @@ OLAPStatus EngineCloneTask::_finish_clone(Tablet* tablet,
const string& clone_di
OLAPStatus res = OLAP_SUCCESS;
std::vector<string> linked_success_files;
// clone and compaction operation should be performed sequentially
- tablet->obtain_base_compaction_lock();
- tablet->obtain_cumulative_lock();
- tablet->set_clone_occurred(true);
- tablet->obtain_push_lock();
{
+ std::lock_guard<std::mutex>
base_compaction_lock(tablet->get_base_compaction_lock());
+ std::lock_guard<std::mutex>
cumulative_compaction_lock(tablet->get_cumulative_compaction_lock());
+ tablet->set_clone_occurred(true);
+ std::lock_guard<std::mutex> push_lock(tablet->get_push_lock());
WriteLock wrlock(tablet->get_header_lock());
do {
// check clone dir existed
@@ -618,10 +618,6 @@ OLAPStatus EngineCloneTask::_finish_clone(Tablet* tablet,
const string& clone_di
FileUtils::remove_paths(linked_success_files);
}
}
- tablet->release_push_lock();
- tablet->release_cumulative_lock();
- tablet->release_base_compaction_lock();
-
// clear clone dir
std::filesystem::path clone_dir_path(clone_dir);
std::filesystem::remove_all(clone_dir_path);
diff --git a/be/src/olap/task/engine_storage_migration_task.cpp
b/be/src/olap/task/engine_storage_migration_task.cpp
index 8f0d708..486730a 100644
--- a/be/src/olap/task/engine_storage_migration_task.cpp
+++ b/be/src/olap/task/engine_storage_migration_task.cpp
@@ -58,8 +58,7 @@ OLAPStatus EngineStorageMigrationTask::_migrate() {
return OLAP_ERR_HEADER_HAS_PENDING_DATA;
}
- _tablet->obtain_push_lock();
-
+ std::lock_guard<std::mutex> lock(_tablet->get_push_lock());
// TODO(ygl): the tablet should not under schema change or rollup or load
do {
std::vector<RowsetSharedPtr> consistent_rowsets;
@@ -165,9 +164,6 @@ OLAPStatus EngineStorageMigrationTask::_migrate() {
break;
}
} while (0);
-
- _tablet->release_push_lock();
-
return res;
}
diff --git a/be/src/util/mutex.cpp b/be/src/util/mutex.cpp
index d8d38b4..921f7a2 100644
--- a/be/src/util/mutex.cpp
+++ b/be/src/util/mutex.cpp
@@ -55,46 +55,6 @@ namespace doris {
} \
} while (0)
-#define PTHREAD_RWLOCK_INIT_WITH_LOG(lockptr, param) \
- do { \
- int lock_ret = 0; \
- if (0 != (lock_ret = pthread_rwlock_init(lockptr, param))) { \
- LOG(FATAL) << "fail to init rwlock. err=" << strerror(lock_ret); \
- } \
- } while (0)
-
-#define PTHREAD_RWLOCK_DESTROY_WITH_LOG(lockptr)
\
- do {
\
- int lock_ret = 0;
\
- if (0 != (lock_ret = pthread_rwlock_destroy(lockptr))) {
\
- LOG(FATAL) << "fail to destroy rwlock. err=" <<
strerror(lock_ret); \
- }
\
- } while (0)
-
-#define PTHREAD_RWLOCK_RDLOCK_WITH_LOG(lockptr)
\
- do {
\
- int lock_ret = 0;
\
- if (0 != (lock_ret = pthread_rwlock_rdlock(lockptr))) {
\
- LOG(FATAL) << "fail to lock reader lock. err=" <<
strerror(lock_ret); \
- }
\
- } while (0)
-
-#define PTHREAD_RWLOCK_WRLOCK_WITH_LOG(lockptr)
\
- do {
\
- int lock_ret = 0;
\
- if (0 != (lock_ret = pthread_rwlock_wrlock(lockptr))) {
\
- LOG(FATAL) << "fail to lock writer lock. err=" <<
strerror(lock_ret); \
- }
\
- } while (0)
-
-#define PTHREAD_RWLOCK_UNLOCK_WITH_LOG(lockptr)
\
- do {
\
- int lock_ret = 0;
\
- if (0 != (lock_ret = pthread_rwlock_unlock(lockptr))) {
\
- LOG(FATAL) << "fail to unlock rwlock. err=" << strerror(lock_ret);
\
- }
\
- } while (0)
-
Mutex::Mutex() {
PTHREAD_MUTEX_INIT_WITH_LOG(&_lock, nullptr);
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
