From 10a7c0378fdc76cdd5c82b497fcf0cf1c536793f Mon Sep 17 00:00:00 2001 From: xmx8 Date: Thu, 12 Jun 2025 09:07:23 +0800 Subject: [PATCH] =?UTF-8?q?monitoring=E7=9B=AE=E5=BD=95=E6=96=B0=E5=A2=9Et?= =?UTF-8?q?hread=5Fstatus=5Futil=E5=8A=9F=E8=83=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../rocksdb/monitoring/thread_status_util.cc | 206 ++++++++++++++++++ .../rocksdb/monitoring/thread_status_util.h | 134 ++++++++++++ .../monitoring/thread_status_util_debug.cc | 32 +++ 3 files changed, 372 insertions(+) create mode 100644 storage/rocksdb/rocksdb/monitoring/thread_status_util.cc create mode 100644 storage/rocksdb/rocksdb/monitoring/thread_status_util.h create mode 100644 storage/rocksdb/rocksdb/monitoring/thread_status_util_debug.cc diff --git a/storage/rocksdb/rocksdb/monitoring/thread_status_util.cc b/storage/rocksdb/rocksdb/monitoring/thread_status_util.cc new file mode 100644 index 000000000..13a79163c --- /dev/null +++ b/storage/rocksdb/rocksdb/monitoring/thread_status_util.cc @@ -0,0 +1,206 @@ +// Copyright (c) 2011-present, Facebook, Inc. All rights reserved. +// This source code is licensed under both the GPLv2 (found in the +// COPYING file in the root directory) and Apache 2.0 License +// (found in the LICENSE.Apache file in the root directory). + +#include "monitoring/thread_status_util.h" + +#include "monitoring/thread_status_updater.h" +#include "rocksdb/env.h" + +namespace ROCKSDB_NAMESPACE { + +#ifdef ROCKSDB_USING_THREAD_STATUS +__thread ThreadStatusUpdater* ThreadStatusUtil::thread_updater_local_cache_ = + nullptr; +__thread bool ThreadStatusUtil::thread_updater_initialized_ = false; + +void ThreadStatusUtil::RegisterThread(const Env* env, + ThreadStatus::ThreadType thread_type) { + if (!MaybeInitThreadLocalUpdater(env)) { + return; + } + assert(thread_updater_local_cache_); + thread_updater_local_cache_->RegisterThread(thread_type, env->GetThreadID()); +} + +void ThreadStatusUtil::UnregisterThread() { + thread_updater_initialized_ = false; + if (thread_updater_local_cache_ != nullptr) { + thread_updater_local_cache_->UnregisterThread(); + thread_updater_local_cache_ = nullptr; + } +} + +void ThreadStatusUtil::SetColumnFamily(const ColumnFamilyData* cfd, + const Env* env, + bool enable_thread_tracking) { + if (!MaybeInitThreadLocalUpdater(env)) { + return; + } + assert(thread_updater_local_cache_); + if (cfd != nullptr && enable_thread_tracking) { + thread_updater_local_cache_->SetColumnFamilyInfoKey(cfd); + } else { + // When cfd == nullptr or enable_thread_tracking == false, we set + // ColumnFamilyInfoKey to nullptr, which makes SetThreadOperation + // and SetThreadState become no-op. + thread_updater_local_cache_->SetColumnFamilyInfoKey(nullptr); + } +} + +void ThreadStatusUtil::SetThreadOperation(ThreadStatus::OperationType op) { + if (thread_updater_local_cache_ == nullptr) { + // thread_updater_local_cache_ must be set in SetColumnFamily + // or other ThreadStatusUtil functions. + return; + } + + if (op != ThreadStatus::OP_UNKNOWN) { + uint64_t current_time = Env::Default()->NowMicros(); + thread_updater_local_cache_->SetOperationStartTime(current_time); + } else { + // TDOO(yhchiang): we could report the time when we set operation to + // OP_UNKNOWN once the whole instrumentation has been done. + thread_updater_local_cache_->SetOperationStartTime(0); + } + thread_updater_local_cache_->SetThreadOperation(op); +} + +ThreadStatus::OperationStage ThreadStatusUtil::SetThreadOperationStage( + ThreadStatus::OperationStage stage) { + if (thread_updater_local_cache_ == nullptr) { + // thread_updater_local_cache_ must be set in SetColumnFamily + // or other ThreadStatusUtil functions. + return ThreadStatus::STAGE_UNKNOWN; + } + + return thread_updater_local_cache_->SetThreadOperationStage(stage); +} + +void ThreadStatusUtil::SetThreadOperationProperty(int code, uint64_t value) { + if (thread_updater_local_cache_ == nullptr) { + // thread_updater_local_cache_ must be set in SetColumnFamily + // or other ThreadStatusUtil functions. + return; + } + + thread_updater_local_cache_->SetThreadOperationProperty(code, value); +} + +void ThreadStatusUtil::IncreaseThreadOperationProperty(int code, + uint64_t delta) { + if (thread_updater_local_cache_ == nullptr) { + // thread_updater_local_cache_ must be set in SetColumnFamily + // or other ThreadStatusUtil functions. + return; + } + + thread_updater_local_cache_->IncreaseThreadOperationProperty(code, delta); +} + +void ThreadStatusUtil::SetThreadState(ThreadStatus::StateType state) { + if (thread_updater_local_cache_ == nullptr) { + // thread_updater_local_cache_ must be set in SetColumnFamily + // or other ThreadStatusUtil functions. + return; + } + + thread_updater_local_cache_->SetThreadState(state); +} + +void ThreadStatusUtil::ResetThreadStatus() { + if (thread_updater_local_cache_ == nullptr) { + return; + } + thread_updater_local_cache_->ResetThreadStatus(); +} + +void ThreadStatusUtil::NewColumnFamilyInfo(const DB* db, + const ColumnFamilyData* cfd, + const std::string& cf_name, + const Env* env) { + if (!MaybeInitThreadLocalUpdater(env)) { + return; + } + assert(thread_updater_local_cache_); + if (thread_updater_local_cache_) { + thread_updater_local_cache_->NewColumnFamilyInfo(db, db->GetName(), cfd, + cf_name); + } +} + +void ThreadStatusUtil::EraseColumnFamilyInfo(const ColumnFamilyData* cfd) { + if (thread_updater_local_cache_ == nullptr) { + return; + } + thread_updater_local_cache_->EraseColumnFamilyInfo(cfd); +} + +void ThreadStatusUtil::EraseDatabaseInfo(const DB* db) { + ThreadStatusUpdater* thread_updater = db->GetEnv()->GetThreadStatusUpdater(); + if (thread_updater == nullptr) { + return; + } + thread_updater->EraseDatabaseInfo(db); +} + +bool ThreadStatusUtil::MaybeInitThreadLocalUpdater(const Env* env) { + if (!thread_updater_initialized_ && env != nullptr) { + thread_updater_initialized_ = true; + thread_updater_local_cache_ = env->GetThreadStatusUpdater(); + } + return (thread_updater_local_cache_ != nullptr); +} + +AutoThreadOperationStageUpdater::AutoThreadOperationStageUpdater( + ThreadStatus::OperationStage stage) { + prev_stage_ = ThreadStatusUtil::SetThreadOperationStage(stage); +} + +AutoThreadOperationStageUpdater::~AutoThreadOperationStageUpdater() { + ThreadStatusUtil::SetThreadOperationStage(prev_stage_); +} + +#else + +ThreadStatusUpdater* ThreadStatusUtil::thread_updater_local_cache_ = nullptr; +bool ThreadStatusUtil::thread_updater_initialized_ = false; + +bool ThreadStatusUtil::MaybeInitThreadLocalUpdater(const Env* /*env*/) { + return false; +} + +void ThreadStatusUtil::SetColumnFamily(const ColumnFamilyData* /*cfd*/, + const Env* /*env*/, + bool /*enable_thread_tracking*/) {} + +void ThreadStatusUtil::SetThreadOperation(ThreadStatus::OperationType /*op*/) {} + +void ThreadStatusUtil::SetThreadOperationProperty(int /*code*/, + uint64_t /*value*/) {} + +void ThreadStatusUtil::IncreaseThreadOperationProperty(int /*code*/, + uint64_t /*delta*/) {} + +void ThreadStatusUtil::SetThreadState(ThreadStatus::StateType /*state*/) {} + +void ThreadStatusUtil::NewColumnFamilyInfo(const DB* /*db*/, + const ColumnFamilyData* /*cfd*/, + const std::string& /*cf_name*/, + const Env* /*env*/) {} + +void ThreadStatusUtil::EraseColumnFamilyInfo(const ColumnFamilyData* /*cfd*/) {} + +void ThreadStatusUtil::EraseDatabaseInfo(const DB* /*db*/) {} + +void ThreadStatusUtil::ResetThreadStatus() {} + +AutoThreadOperationStageUpdater::AutoThreadOperationStageUpdater( + ThreadStatus::OperationStage /*stage*/) {} + +AutoThreadOperationStageUpdater::~AutoThreadOperationStageUpdater() {} + +#endif // ROCKSDB_USING_THREAD_STATUS + +} // namespace ROCKSDB_NAMESPACE diff --git a/storage/rocksdb/rocksdb/monitoring/thread_status_util.h b/storage/rocksdb/rocksdb/monitoring/thread_status_util.h new file mode 100644 index 000000000..b4d97b0b6 --- /dev/null +++ b/storage/rocksdb/rocksdb/monitoring/thread_status_util.h @@ -0,0 +1,134 @@ +// Copyright (c) 2011-present, Facebook, Inc. All rights reserved. +// This source code is licensed under both the GPLv2 (found in the +// COPYING file in the root directory) and Apache 2.0 License +// (found in the LICENSE.Apache file in the root directory). + +#pragma once + +#include + +#include "monitoring/thread_status_updater.h" +#include "rocksdb/db.h" +#include "rocksdb/env.h" +#include "rocksdb/thread_status.h" + +namespace ROCKSDB_NAMESPACE { + +class ColumnFamilyData; + +// The static utility class for updating thread-local status. +// +// The thread-local status is updated via the thread-local cached +// pointer thread_updater_local_cache_. During each function call, +// when ThreadStatusUtil finds thread_updater_local_cache_ is +// left uninitialized (determined by thread_updater_initialized_), +// it will tries to initialize it using the return value of +// Env::GetThreadStatusUpdater(). When thread_updater_local_cache_ +// is initialized by a non-null pointer, each function call will +// then update the status of the current thread. Otherwise, +// all function calls to ThreadStatusUtil will be no-op. +class ThreadStatusUtil { + public: + // Register the current thread for tracking. + static void RegisterThread( + const Env* env, ThreadStatus::ThreadType thread_type); + + // Unregister the current thread. + static void UnregisterThread(); + + // Create an entry in the global ColumnFamilyInfo table for the + // specified column family. This function should be called only + // when the current thread does not hold db_mutex. + static void NewColumnFamilyInfo(const DB* db, const ColumnFamilyData* cfd, + const std::string& cf_name, const Env* env); + + // Erase the ConstantColumnFamilyInfo that is associated with the + // specified ColumnFamilyData. This function should be called only + // when the current thread does not hold db_mutex. + static void EraseColumnFamilyInfo(const ColumnFamilyData* cfd); + + // Erase all ConstantColumnFamilyInfo that is associated with the + // specified db instance. This function should be called only when + // the current thread does not hold db_mutex. + static void EraseDatabaseInfo(const DB* db); + + // Update the thread status to indicate the current thread is doing + // something related to the specified column family. + static void SetColumnFamily(const ColumnFamilyData* cfd, const Env* env, + bool enable_thread_tracking); + + static void SetThreadOperation(ThreadStatus::OperationType type); + + static ThreadStatus::OperationStage SetThreadOperationStage( + ThreadStatus::OperationStage stage); + + static void SetThreadOperationProperty( + int code, uint64_t value); + + static void IncreaseThreadOperationProperty( + int code, uint64_t delta); + + static void SetThreadState(ThreadStatus::StateType type); + + static void ResetThreadStatus(); + +#ifndef NDEBUG + static void TEST_SetStateDelay( + const ThreadStatus::StateType state, int micro); + static void TEST_StateDelay(const ThreadStatus::StateType state); +#endif + + protected: + // Initialize the thread-local ThreadStatusUpdater when it finds + // the cached value is nullptr. Returns true if it has cached + // a non-null pointer. + static bool MaybeInitThreadLocalUpdater(const Env* env); + +#ifdef ROCKSDB_USING_THREAD_STATUS + // A boolean flag indicating whether thread_updater_local_cache_ + // is initialized. It is set to true when an Env uses any + // ThreadStatusUtil functions using the current thread other + // than UnregisterThread(). It will be set to false when + // UnregisterThread() is called. + // + // When this variable is set to true, thread_updater_local_cache_ + // will not be updated until this variable is again set to false + // in UnregisterThread(). + static __thread bool thread_updater_initialized_; + + // The thread-local cached ThreadStatusUpdater that caches the + // thread_status_updater_ of the first Env that uses any ThreadStatusUtil + // function other than UnregisterThread(). This variable will + // be cleared when UnregisterThread() is called. + // + // When this variable is set to a non-null pointer, then the status + // of the current thread will be updated when a function of + // ThreadStatusUtil is called. Otherwise, all functions of + // ThreadStatusUtil will be no-op. + // + // When thread_updater_initialized_ is set to true, this variable + // will not be updated until this thread_updater_initialized_ is + // again set to false in UnregisterThread(). + static __thread ThreadStatusUpdater* thread_updater_local_cache_; +#else + static bool thread_updater_initialized_; + static ThreadStatusUpdater* thread_updater_local_cache_; +#endif +}; + +// A helper class for updating thread state. It will set the +// thread state according to the input parameter in its constructor +// and set the thread state to the previous state in its destructor. +class AutoThreadOperationStageUpdater { + public: + explicit AutoThreadOperationStageUpdater( + ThreadStatus::OperationStage stage); + ~AutoThreadOperationStageUpdater(); + +#ifdef ROCKSDB_USING_THREAD_STATUS + private: + ThreadStatus::OperationStage prev_stage_; +#endif +}; + +} // namespace ROCKSDB_NAMESPACE diff --git a/storage/rocksdb/rocksdb/monitoring/thread_status_util_debug.cc b/storage/rocksdb/rocksdb/monitoring/thread_status_util_debug.cc new file mode 100644 index 000000000..375fe8c0a --- /dev/null +++ b/storage/rocksdb/rocksdb/monitoring/thread_status_util_debug.cc @@ -0,0 +1,32 @@ +// Copyright (c) 2011-present, Facebook, Inc. All rights reserved. +// This source code is licensed under both the GPLv2 (found in the +// COPYING file in the root directory) and Apache 2.0 License +// (found in the LICENSE.Apache file in the root directory). + +#include + +#include "monitoring/thread_status_updater.h" +#include "monitoring/thread_status_util.h" +#include "rocksdb/env.h" + +namespace ROCKSDB_NAMESPACE { + +#ifndef NDEBUG +// the delay for debugging purpose. +static std::atomic states_delay[ThreadStatus::NUM_STATE_TYPES]; + +void ThreadStatusUtil::TEST_SetStateDelay( + const ThreadStatus::StateType state, int micro) { + states_delay[state].store(micro, std::memory_order_relaxed); +} + +void ThreadStatusUtil::TEST_StateDelay(const ThreadStatus::StateType state) { + auto delay = states_delay[state].load(std::memory_order_relaxed); + if (delay > 0) { + Env::Default()->SleepForMicroseconds(delay); + } +} + +#endif // !NDEBUG + +} // namespace ROCKSDB_NAMESPACE -- Gitee