1
0
mirror of https://github.com/facebook/rocksdb.git synced 2026-10-23 08:09:02 +03:00
Files
rocksdb/monitoring/statistics_test.cc
Yoshinori Matsunobu d6a7847d37 Add a DB option to log long-held global mutexes and release sites (#15436)
Summary:
Pull Request resolved: https://github.com/facebook/rocksdb/pull/15436

## Why
Long-held global mutexes can stall foreground and background work. Existing mutex statistics measure acquisition waits, not continuous ownership. This adds opt-in logging to identify the mutex, owning thread, operation and release site.

## Enable for an investigation
Provide an INFO logger and shared `Statistics` object when opening the DB, then:
- Set the shared statistics level to `StatsLevel::kAll`.
- Set `mutex_hold_log_threshold_us` through `DB::SetDBOptions`; `100000` means 100 ms.
- Restore threshold 0 and the previous statistics level afterward.

The threshold defaults to **0 (disabled)**. `kAll` alone does not enable these hold probes; threshold 0 leaves existing statistics timers unchanged.

## Output
Illustrative records, not production observations:
```
SLOW_GLOBAL_MUTEX mutex=DBImpl::mutex_ mutex_addr=0x12340000 owner_thread_id=12345 thread_operation=Compaction hold_us=200000 threshold_us=100000 observed_end_us=1792205400000000 buffer_dropped_records=0 unlock=db_impl_compaction_flush.cc:4660 BackgroundCallCompaction()
SLOW_GLOBAL_MUTEX mutex=DBImpl::wal_write_mutex_ mutex_addr=0x12340100 owner_thread_id=12346 thread_operation= hold_us=35000 threshold_us=25000 observed_end_us=1792205401000000 buffer_dropped_records=0 scope_exit=db_impl_write.cc:2250 PreprocessWrite()
```
The first proves a completed 200 ms DB-mutex ownership interval on a thread tagged `Compaction`. The second uses a 25 ms threshold; an empty operation tag means none was available. Neither record reports acquisition wait or proves the cause of a stall.

Release attribution uses compiler-embedded strings available in optimized, stripped builds, with a GCC/Clang fallback where standard source locations are unavailable. Direct unlock and condition wait identify their call sites; scope exit and temporary unlock identify guard declarations, not closing braces. No stack unwinding or symbol lookup.

## Scope and safety
- Track six selected mutexes: DB, WAL, point-lock wait graph, two prepare-log bookkeeping mutexes and transaction name map. This is not comprehensive synchronization tracing.
- Acquisition waits and condition-variable release gaps are excluded from hold duration. Configuration is pre-first-use; repeated setup retains the first valid context.
- Sample at most once per mutex per second, with eight deferred records per thread. `buffer_dropped_records` counts capacity overflow only; rate limiting, pause and emission contention can omit samples.
- Emit after covered outer lock scopes release. Emission is nonblocking best-effort; shutdown drains/pauses emission and stops diagnostics before the owned logger closes. Closing from that logger's own callback is unsupported.
- No data-file format change. The mutable option is persisted in OPTIONS and wired into bench/stress tools. Typed C binding is deferred; string DB-option APIs are available.

Disabled bookkeeping still has a cost. Release A/B measurements found about 3 ns added per selected mutex pair at level 1, and about 70 ns for hold timing over `kAll`/0. Get and reused-iterator Seek showed no resolved small probe effect; fresh iterator+Seek at 16 threads showed +8.7% CPU/query and -9.4% QPS. No logs were emitted in those benchmarks; logging cost and broader workload qualification remain open.

Reviewed By: xingbowang

Differential Revision: D125790955

fbshipit-source-id: 1096163f3bd25ffee585dc6ec8ebfd7f1759851a
2026-10-22 09:50:20 -07:00

844 lines
28 KiB
C++

// 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 "rocksdb/statistics.h"
#include <array>
#include <atomic>
#include <cinttypes>
#include <condition_variable>
#include <cstdio>
#include <functional>
#include <limits>
#include <mutex>
#include <string>
#include <thread>
#include <vector>
#include "monitoring/instrumented_mutex.h"
#include "monitoring/thread_status_util.h"
#include "port/stack_trace.h"
#include "rocksdb/convenience.h"
#include "rocksdb/perf_level.h"
#include "rocksdb/utilities/options_type.h"
#include "test_util/testharness.h"
#include "test_util/testutil.h"
namespace ROCKSDB_NAMESPACE {
class StatisticsTest : public testing::Test {};
// Sanity check to make sure that contents and order of TickersNameMap
// match Tickers enum
TEST_F(StatisticsTest, SanityTickers) {
EXPECT_EQ(static_cast<size_t>(Tickers::TICKER_ENUM_MAX),
TickersNameMap.size());
for (uint32_t t = 0; t < Tickers::TICKER_ENUM_MAX; t++) {
auto pair = TickersNameMap[static_cast<size_t>(t)];
ASSERT_EQ(pair.first, t) << "Miss match at " << pair.second;
}
}
// Sanity check to make sure that contents and order of HistogramsNameMap
// match Tickers enum
TEST_F(StatisticsTest, SanityHistograms) {
EXPECT_EQ(static_cast<size_t>(Histograms::HISTOGRAM_ENUM_MAX),
HistogramsNameMap.size());
for (uint32_t h = 0; h < Histograms::HISTOGRAM_ENUM_MAX; h++) {
auto pair = HistogramsNameMap[static_cast<size_t>(h)];
ASSERT_EQ(pair.first, h) << "Miss match at " << pair.second;
}
}
TEST_F(StatisticsTest, NoNameStats) {
static std::unordered_map<std::string, OptionTypeInfo> no_name_opt_info = {
{"inner",
OptionTypeInfo::AsCustomSharedPtr<Statistics>(
0, OptionVerificationType::kByName,
OptionTypeFlags::kAllowNull | OptionTypeFlags::kCompareNever)},
};
class DefaultNameStatistics : public Statistics {
public:
DefaultNameStatistics(const std::shared_ptr<Statistics>& stats = nullptr)
: inner(stats) {
RegisterOptions("", &inner, &no_name_opt_info);
}
uint64_t getTickerCount(uint32_t /*tickerType*/) const override {
return 0;
}
void histogramData(uint32_t /*type*/,
HistogramData* const /*data*/) const override {}
void recordTick(uint32_t /*tickerType*/, uint64_t /*count*/) override {}
void setTickerCount(uint32_t /*tickerType*/, uint64_t /*count*/) override {}
uint64_t getAndResetTickerCount(uint32_t /*tickerType*/) override {
return 0;
}
std::shared_ptr<Statistics> inner;
};
ConfigOptions options;
options.ignore_unsupported_options = false;
auto stats = std::make_shared<DefaultNameStatistics>();
ASSERT_STREQ(stats->Name(), "");
ASSERT_EQ("", stats->ToString(
options)); // A stats with no name with have no options...
ASSERT_OK(stats->ConfigureFromString(options, "inner="));
ASSERT_EQ("", stats->ToString(
options)); // A stats with no name with have no options...
ASSERT_NE(stats->inner, nullptr);
ASSERT_NE("", stats->inner->ToString(options)); // ... even if it does...
}
class SlowMutexClock : public SystemClockWrapper {
public:
SlowMutexClock() : SystemClockWrapper(SystemClock::Default()) {}
const char* Name() const override { return "SlowMutexClock"; }
uint64_t NowNanos() override {
++reads;
return nanos.load();
}
void Advance(uint64_t micros) { nanos.fetch_add(micros * 1000); }
std::atomic<uint64_t> nanos{0};
std::atomic<uint64_t> reads{0};
};
class SlowMutexLogger : public Logger {
public:
using Logger::Logv;
void Logv(const char* format, va_list ap) override {
char message[1024];
vsnprintf(message, sizeof(message), format, ap);
if (std::string(message).find("SLOW_GLOBAL_MUTEX") == std::string::npos) {
return;
}
if (before_log) {
before_log();
}
std::lock_guard<std::mutex> lock(mutex);
entries.emplace_back(message);
}
std::function<void()> before_log;
std::mutex mutex;
std::vector<std::string> entries;
};
class SlowMutexTest : public testing::Test {
public:
void SetUp() override {
previous_level = GetPerfLevel();
SetPerfLevel(PerfLevel::kEnableCount);
stats->set_stats_level(StatsLevel::kExceptTimeForMutex);
}
void TearDown() override { SetPerfLevel(previous_level); }
void Configure(InstrumentedMutex* mutex, const char* name) {
mutex->ConfigureSlowHoldLogging(context, name);
}
void Hold(InstrumentedMutex* mutex, uint64_t micros) {
mutex->Lock();
clock.Advance(micros);
mutex->Unlock();
}
void CheckConditionWait(bool timed) {
InstrumentedMutex mutex;
Configure(&mutex, "condition_mutex");
InstrumentedCondVar cv(&mutex);
mutex.Lock();
clock.Advance(80000);
uint64_t worker_id = 0;
std::thread worker([&] {
mutex.Lock();
worker_id = Env::Default()->GetThreadID();
clock.Advance(500000);
cv.Signal();
mutex.Unlock();
});
if (timed) {
EXPECT_FALSE(cv.TimedWait(Env::Default()->NowMicros() + 30000000));
} else {
cv.Wait();
}
clock.Advance(80000);
mutex.Unlock();
worker.join();
ASSERT_EQ(logger->entries.size(), 1);
EXPECT_NE(logger->entries.front().find("hold_us=500000"),
std::string::npos);
EXPECT_NE(logger->entries.front().find("owner_thread_id=" +
std::to_string(worker_id)),
std::string::npos);
logger->entries.clear();
}
void CheckReleaseLocation(const std::string& entry,
const MutexSourceLocation& expected,
uint32_t line_offset, const char* kind) {
std::string file(expected.file_name());
const size_t separator = file.find_last_of("/\\");
if (separator != std::string::npos) {
file.erase(0, separator + 1);
}
const std::string site = std::string(kind) + "=" + file + ":" +
std::to_string(expected.line() + line_offset) +
" TestBody()";
EXPECT_NE(entry.find(site), std::string::npos);
EXPECT_EQ(entry.find("release_line="), std::string::npos);
EXPECT_EQ(entry.find("release_function="), std::string::npos);
}
std::shared_ptr<Statistics> stats = CreateDBStatistics();
SlowMutexClock clock;
std::atomic<uint64_t> threshold_us{100000};
std::shared_ptr<SlowMutexLogger> logger = std::make_shared<SlowMutexLogger>();
std::shared_ptr<SlowMutexLoggingContext> context =
std::make_shared<SlowMutexLoggingContext>(
stats.get(), &clock, Env::Default(), logger, &threshold_us);
PerfLevel previous_level = PerfLevel::kDisable;
};
TEST_F(SlowMutexTest, DefaultThresholdDisablesHoldTimingAtKAll) {
stats->set_stats_level(StatsLevel::kAll);
const DBOptions options;
ASSERT_EQ(options.mutex_hold_log_threshold_us, 0);
threshold_us.store(options.mutex_hold_log_threshold_us);
InstrumentedMutex mutex;
Configure(&mutex, "default_disabled_mutex");
Hold(&mutex, 200000);
EXPECT_EQ(clock.reads.load(), 0);
EXPECT_TRUE(logger->entries.empty());
threshold_us.store(100000);
Hold(&mutex, 200000);
ASSERT_EQ(logger->entries.size(), 1);
}
TEST_F(SlowMutexTest, DisabledLevelsDoNotReadClock) {
InstrumentedMutex mutex;
Configure(&mutex, "disabled_mutex");
const std::array<StatsLevel, 5> levels = {
StatsLevel::kDisableAll, StatsLevel::kExceptHistogramOrTimers,
StatsLevel::kExceptTimers, StatsLevel::kExceptDetailedTimers,
StatsLevel::kExceptTimeForMutex};
for (StatsLevel level : levels) {
stats->set_stats_level(level);
Hold(&mutex, 200000);
}
EXPECT_EQ(clock.reads.load(), 0);
EXPECT_TRUE(logger->entries.empty());
}
TEST_F(SlowMutexTest, RecordsOwnerAndActualHoldAfterUnlock) {
stats->set_stats_level(StatsLevel::kAll);
InstrumentedMutex mutex;
Configure(&mutex, "test_mutex");
logger->before_log = [&] {
ASSERT_TRUE(mutex.TryLock());
mutex.Unlock();
};
Hold(&mutex, 200000);
ASSERT_EQ(logger->entries.size(), 1);
const std::string& entry = logger->entries.front();
EXPECT_NE(entry.find("mutex=test_mutex"), std::string::npos);
EXPECT_NE(entry.find("hold_us=200000"), std::string::npos);
EXPECT_NE(entry.find("owner_thread_id=" +
std::to_string(Env::Default()->GetThreadID())),
std::string::npos);
EXPECT_EQ(clock.reads.load(), 2);
}
TEST_F(SlowMutexTest, DirectUnlockRecordsActualCaller) {
stats->set_stats_level(StatsLevel::kAll);
InstrumentedMutex mutex;
Configure(&mutex, "direct_release");
mutex.Lock();
clock.Advance(200000);
const MutexSourceLocation expected = MutexSourceLocation::current();
mutex.Unlock();
ASSERT_EQ(logger->entries.size(), 1);
CheckReleaseLocation(logger->entries.front(), expected, 1, "unlock");
}
TEST_F(SlowMutexTest, ScopeExitRecordsGuardDeclaration) {
stats->set_stats_level(StatsLevel::kAll);
InstrumentedMutex mutex;
Configure(&mutex, "scoped_release");
MutexSourceLocation expected;
{
expected = MutexSourceLocation::current();
InstrumentedMutexLock guard(&mutex);
clock.Advance(200000);
}
ASSERT_EQ(logger->entries.size(), 1);
CheckReleaseLocation(logger->entries.front(), expected, 1, "scope_exit");
}
TEST_F(SlowMutexTest, StdMutexScopeExitRecordsGuardDeclaration) {
stats->set_stats_level(StatsLevel::kAll);
std::mutex mutex;
SlowMutexTimer timer(context, "std_scoped_release", &mutex);
MutexSourceLocation expected;
{
expected = MutexSourceLocation::current();
SlowMutexLock guard(&mutex, &timer);
clock.Advance(200000);
}
ASSERT_EQ(logger->entries.size(), 1);
CheckReleaseLocation(logger->entries.front(), expected, 1, "scope_exit");
}
TEST_F(SlowMutexTest, TemporaryUnlockRecordsReleaseCaller) {
stats->set_stats_level(StatsLevel::kAll);
InstrumentedMutex mutex;
Configure(&mutex, "temporary_release");
mutex.Lock();
clock.Advance(200000);
MutexSourceLocation expected;
{
expected = MutexSourceLocation::current();
InstrumentedMutexUnlock unlock(&mutex);
}
mutex.Unlock();
ASSERT_EQ(logger->entries.size(), 1);
CheckReleaseLocation(logger->entries.front(), expected, 1,
"temporary_unlock");
}
TEST_F(SlowMutexTest, ConditionWaitRecordsReleaseCaller) {
stats->set_stats_level(StatsLevel::kAll);
InstrumentedMutex mutex;
Configure(&mutex, "condition_release");
InstrumentedCondVar cv(&mutex);
mutex.Lock();
clock.Advance(200000);
const MutexSourceLocation expected = MutexSourceLocation::current();
EXPECT_TRUE(cv.TimedWait(1));
mutex.Unlock();
ASSERT_EQ(logger->entries.size(), 1);
CheckReleaseLocation(logger->entries.front(), expected, 1, "condition_wait");
}
TEST_F(SlowMutexTest, DeferredGuardExplicitUnlockRecordsReleaseCaller) {
stats->set_stats_level(StatsLevel::kAll);
InstrumentedMutex mutex;
Configure(&mutex, "deferred_release");
DeferredInstrumentedMutexLock guard(&mutex, true);
clock.Advance(200000);
const MutexSourceLocation expected = MutexSourceLocation::current();
guard.Unlock();
ASSERT_EQ(logger->entries.size(), 1);
CheckReleaseLocation(logger->entries.front(), expected, 1, "unlock");
}
TEST_F(SlowMutexTest, UntimedConditionWaitRecordsReleaseCaller) {
stats->set_stats_level(StatsLevel::kAll);
InstrumentedMutex mutex;
Configure(&mutex, "untimed_condition_release");
InstrumentedCondVar cv(&mutex);
mutex.Lock();
clock.Advance(200000);
std::thread worker([&] {
mutex.Lock();
cv.Signal();
mutex.Unlock();
});
const MutexSourceLocation expected = MutexSourceLocation::current();
cv.Wait();
mutex.Unlock();
worker.join();
ASSERT_EQ(logger->entries.size(), 1);
CheckReleaseLocation(logger->entries.front(), expected, 1, "condition_wait");
}
TEST_F(SlowMutexTest, DeferredGuardScopeExitRecordsDeclaration) {
stats->set_stats_level(StatsLevel::kAll);
InstrumentedMutex mutex;
Configure(&mutex, "deferred_scope_release");
MutexSourceLocation expected;
{
expected = MutexSourceLocation::current();
DeferredInstrumentedMutexLock guard(&mutex, true);
clock.Advance(200000);
}
ASSERT_EQ(logger->entries.size(), 1);
CheckReleaseLocation(logger->entries.front(), expected, 1, "scope_exit");
}
TEST_F(SlowMutexTest, MissingReleaseSiteIsExplicitlyUnavailable) {
stats->set_stats_level(StatsLevel::kAll);
InstrumentedMutex mutex;
Configure(&mutex, "unavailable_site");
mutex.Lock();
clock.Advance(200000);
mutex.Unlock(MutexSourceLocation{});
ASSERT_EQ(logger->entries.size(), 1);
const std::string& entry = logger->entries.front();
EXPECT_NE(entry.find("unlock=(unavailable):0 unavailable()"),
std::string::npos);
}
TEST_F(SlowMutexTest, RepeatedConfigurationKeepsFirstContext) {
stats->set_stats_level(StatsLevel::kAll);
const long initial_owners = context.use_count();
{
InstrumentedMutex mutex;
Configure(&mutex, "first_configuration");
EXPECT_EQ(context.use_count(), initial_owners + 1);
Configure(&mutex, "ignored_configuration");
EXPECT_EQ(context.use_count(), initial_owners + 1);
std::shared_ptr<SlowMutexLoggingContext> other_context =
std::make_shared<SlowMutexLoggingContext>(
stats.get(), &clock, Env::Default(), logger, &threshold_us);
mutex.ConfigureSlowHoldLogging(other_context, "ignored_context");
EXPECT_EQ(other_context.use_count(), 1);
Hold(&mutex, 200000);
ASSERT_EQ(logger->entries.size(), 1);
EXPECT_NE(logger->entries.front().find("mutex=first_configuration "),
std::string::npos);
EXPECT_EQ(logger->entries.front().find("ignored_"), std::string::npos);
}
EXPECT_EQ(context.use_count(), initial_owners);
}
TEST_F(SlowMutexTest, PreservesConstructorStatisticsAccounting) {
stats->set_stats_level(StatsLevel::kAll);
InstrumentedMutex mutex(stats.get(), SystemClock::Default().get(),
DB_MUTEX_WAIT_MICROS);
Configure(&mutex, "constructor_mutex");
#ifndef NPERF_CONTEXT
const uint64_t before = stats->getTickerCount(DB_MUTEX_WAIT_MICROS);
#endif
Hold(&mutex, 200000);
#ifndef NPERF_CONTEXT
EXPECT_GT(stats->getTickerCount(DB_MUTEX_WAIT_MICROS), before);
#endif
ASSERT_EQ(logger->entries.size(), 1);
EXPECT_NE(logger->entries.front().find("hold_us=200000"), std::string::npos);
}
TEST_F(SlowMutexTest, ThresholdAndPerMutexRateLimit) {
stats->set_stats_level(StatsLevel::kAll);
InstrumentedMutex mutex;
Configure(&mutex, "threshold_mutex");
Hold(&mutex, 100000);
EXPECT_TRUE(logger->entries.empty());
Hold(&mutex, 100001);
Hold(&mutex, 200000);
EXPECT_EQ(logger->entries.size(), 1);
clock.Advance(2000000);
Hold(&mutex, 200000);
EXPECT_EQ(logger->entries.size(), 2);
}
TEST_F(SlowMutexTest, ConfigurableThresholdAndRuntimeChanges) {
stats->set_stats_level(StatsLevel::kAll);
InstrumentedMutex mutex;
Configure(&mutex, "configurable_mutex");
threshold_us.store(500000);
Hold(&mutex, 200000);
EXPECT_TRUE(logger->entries.empty());
mutex.Lock();
clock.Advance(200000);
threshold_us.store(25000);
mutex.Unlock();
ASSERT_EQ(logger->entries.size(), 1);
EXPECT_NE(logger->entries.front().find("threshold_us=25000"),
std::string::npos);
clock.Advance(2000000);
threshold_us.store(500000);
Hold(&mutex, 200000);
EXPECT_EQ(logger->entries.size(), 1);
}
TEST_F(SlowMutexTest, ZeroThresholdDisablesHoldTiming) {
stats->set_stats_level(StatsLevel::kAll);
InstrumentedMutex mutex;
Configure(&mutex, "zero_mutex");
threshold_us.store(0);
Hold(&mutex, 200000);
EXPECT_EQ(clock.reads.load(), 0);
EXPECT_TRUE(logger->entries.empty());
threshold_us.store(100000);
Hold(&mutex, 200000);
ASSERT_EQ(logger->entries.size(), 1);
mutex.Lock();
clock.Advance(2000000);
threshold_us.store(0);
mutex.Unlock();
EXPECT_EQ(logger->entries.size(), 1);
}
TEST_F(SlowMutexTest, LargeThresholdDoesNotOverflow) {
stats->set_stats_level(StatsLevel::kAll);
InstrumentedMutex mutex;
Configure(&mutex, "large_threshold_mutex");
threshold_us.store(std::numeric_limits<uint64_t>::max());
Hold(&mutex, 200000);
EXPECT_TRUE(logger->entries.empty());
threshold_us.store(1);
Hold(&mutex, 1);
EXPECT_TRUE(logger->entries.empty());
Hold(&mutex, 2);
ASSERT_EQ(logger->entries.size(), 1);
}
TEST_F(SlowMutexTest, SharedThresholdControlsBackgroundOwner) {
stats->set_stats_level(StatsLevel::kAll);
InstrumentedMutex foreground;
InstrumentedMutex background;
Configure(&foreground, "foreground");
Configure(&background, "background");
threshold_us.store(25000);
Hold(&foreground, 30000);
std::thread worker([&] { Hold(&background, 30000); });
worker.join();
ASSERT_EQ(logger->entries.size(), 2);
threshold_us.store(500000);
clock.Advance(2000000);
std::thread suppressed([&] { Hold(&background, 200000); });
suppressed.join();
EXPECT_EQ(logger->entries.size(), 2);
}
TEST_F(SlowMutexTest, FailedTryLockDoesNotOverwriteOwner) {
stats->set_stats_level(StatsLevel::kAll);
InstrumentedMutex mutex;
Configure(&mutex, "try_mutex");
mutex.Lock();
std::thread contender([&] { EXPECT_FALSE(mutex.TryLock()); });
contender.join();
clock.Advance(300000);
mutex.Unlock();
ASSERT_EQ(logger->entries.size(), 1);
EXPECT_NE(logger->entries.front().find("hold_us=300000"), std::string::npos);
EXPECT_NE(
logger->entries.front().find(
"owner_thread_id=" + std::to_string(Env::Default()->GetThreadID())),
std::string::npos);
}
TEST_F(SlowMutexTest, ConditionWaitsExcludeReleasedIntervals) {
stats->set_stats_level(StatsLevel::kAll);
CheckConditionWait(false);
CheckConditionWait(true);
}
TEST_F(SlowMutexTest, TemporaryUnlockSplitsOwnership) {
stats->set_stats_level(StatsLevel::kAll);
InstrumentedMutex mutex;
Configure(&mutex, "temporary_mutex");
{
InstrumentedMutexLock lock(&mutex);
clock.Advance(80000);
{
InstrumentedMutexUnlock unlock(&mutex);
clock.Advance(500000);
}
clock.Advance(80000);
}
EXPECT_TRUE(logger->entries.empty());
}
TEST_F(SlowMutexTest, RuntimeSwitchDoesNotInventEarlierHistory) {
InstrumentedMutex mutex;
Configure(&mutex, "runtime_mutex");
mutex.Lock();
clock.Advance(200000);
stats->set_stats_level(StatsLevel::kAll);
mutex.Unlock();
EXPECT_EQ(clock.reads.load(), 0);
EXPECT_TRUE(logger->entries.empty());
Hold(&mutex, 200000);
ASSERT_EQ(logger->entries.size(), 1);
mutex.Lock();
clock.Advance(2000000);
stats->set_stats_level(StatsLevel::kExceptTimeForMutex);
mutex.Unlock();
Hold(&mutex, 200000);
EXPECT_EQ(logger->entries.size(), 1);
}
TEST_F(SlowMutexTest, StdMutexLogWaitsForDisabledOuterMutexRelease) {
InstrumentedMutex outer;
Configure(&outer, "outer_mutex");
std::mutex inner;
std::shared_ptr<Statistics> inner_stats = CreateDBStatistics();
inner_stats->set_stats_level(StatsLevel::kAll);
std::shared_ptr<SlowMutexLoggingContext> inner_context =
std::make_shared<SlowMutexLoggingContext>(
inner_stats.get(), &clock, Env::Default(), logger, &threshold_us);
SlowMutexTimer timer(inner_context, "inner_mutex", &inner);
logger->before_log = [&] {
ASSERT_TRUE(outer.TryLock());
outer.Unlock();
ASSERT_TRUE(inner.try_lock());
inner.unlock();
};
outer.Lock();
{
SlowMutexLock lock(&inner, &timer);
clock.Advance(200000);
}
EXPECT_TRUE(logger->entries.empty());
outer.Unlock();
ASSERT_EQ(logger->entries.size(), 1);
EXPECT_NE(logger->entries.front().find("mutex=inner_mutex"),
std::string::npos);
}
TEST_F(SlowMutexTest, DefersStdMutexRecordPastUnmeasuredStripe) {
stats->set_stats_level(StatsLevel::kAll);
std::mutex stripe;
std::mutex graph;
SlowMutexTimer timer(context, "graph_mutex", &graph);
logger->before_log = [&] {
ASSERT_TRUE(stripe.try_lock());
stripe.unlock();
};
{
SlowMutexLogScope defer_logs(true);
{
std::lock_guard<std::mutex> stripe_lock(stripe);
{
SlowMutexLock graph_lock(&graph, &timer);
clock.Advance(200000);
}
EXPECT_TRUE(logger->entries.empty());
}
EXPECT_TRUE(logger->entries.empty());
}
ASSERT_EQ(logger->entries.size(), 1);
}
TEST_F(SlowMutexTest, DeferredNativeMutexGuardReleasesBeforeLogging) {
stats->set_stats_level(StatsLevel::kAll);
InstrumentedMutex outer;
InstrumentedMutex inner;
Configure(&inner, "guarded_inner");
logger->before_log = [&] {
ASSERT_TRUE(outer.TryLock());
outer.Unlock();
};
for (bool explicit_unlock : {false, true}) {
clock.Advance(2000000);
{
DeferredInstrumentedMutexLock guard(&outer, true);
Hold(&inner, 200000);
EXPECT_EQ(logger->entries.size(), explicit_unlock ? 1 : 0);
if (explicit_unlock) {
guard.Unlock();
}
}
EXPECT_EQ(logger->entries.size(), explicit_unlock ? 2 : 1);
}
}
TEST_F(SlowMutexTest, MissingStatisticsOrLoggerDoesNotReadClock) {
InstrumentedMutex mutex;
std::shared_ptr<SlowMutexLoggingContext> missing_stats =
std::make_shared<SlowMutexLoggingContext>(nullptr, &clock, Env::Default(),
logger, &threshold_us);
mutex.ConfigureSlowHoldLogging(missing_stats, "no_statistics");
Hold(&mutex, 200000);
stats->set_stats_level(StatsLevel::kAll);
std::shared_ptr<SlowMutexLoggingContext> missing_logger =
std::make_shared<SlowMutexLoggingContext>(
stats.get(), &clock, Env::Default(), nullptr, &threshold_us);
mutex.ConfigureSlowHoldLogging(missing_logger, "no_logger");
Hold(&mutex, 200000);
EXPECT_EQ(clock.reads.load(), 0);
EXPECT_TRUE(logger->entries.empty());
}
TEST_F(SlowMutexTest, TimedOutWaitSplitsOwnership) {
stats->set_stats_level(StatsLevel::kAll);
InstrumentedMutex mutex;
Configure(&mutex, "timeout_mutex");
InstrumentedCondVar cv(&mutex);
mutex.Lock();
clock.Advance(80000);
EXPECT_TRUE(cv.TimedWait(1));
clock.Advance(80000);
mutex.Unlock();
EXPECT_TRUE(logger->entries.empty());
}
#ifndef NROCKSDB_THREAD_STATUS
TEST_F(SlowMutexTest, PreservesCompactionOperationAcrossCleanup) {
stats->set_stats_level(StatsLevel::kAll);
ThreadStatusUtil::RegisterThread(Env::Default(), ThreadStatus::LOW_PRIORITY);
ThreadStatusUtil::SetEnableTracking(true);
ThreadStatusUtil::SetThreadOperation(ThreadStatus::OP_COMPACTION);
InstrumentedMutex mutex;
Configure(&mutex, "compaction_mutex");
mutex.Lock();
clock.Advance(200000);
ThreadStatusUtil::SetThreadOperation(ThreadStatus::OP_UNKNOWN);
mutex.Unlock();
ThreadStatusUtil::UnregisterThread();
ASSERT_EQ(logger->entries.size(), 1);
EXPECT_NE(logger->entries.front().find(
"thread_operation=" +
ThreadStatus::GetOperationName(ThreadStatus::OP_COMPACTION)),
std::string::npos);
}
#endif
TEST_F(SlowMutexTest, PreservesGenericMutexLayouts) {
struct PreviousMutexLayout {
port::Mutex mutex;
Statistics* stats;
SystemClock* clock;
int stats_code;
#ifdef COERCE_CONTEXT_SWITCH
InstrumentedCondVar* bg_cv;
#endif
};
struct PreviousCondVarLayout {
port::CondVar cond;
Statistics* stats;
SystemClock* clock;
int stats_code;
};
const size_t previous_aligned =
(sizeof(PreviousMutexLayout) + CACHE_LINE_SIZE - 1) / CACHE_LINE_SIZE *
CACHE_LINE_SIZE;
EXPECT_EQ(sizeof(CacheAlignedInstrumentedMutex), previous_aligned);
if (sizeof(void*) == 8) {
EXPECT_EQ(sizeof(InstrumentedMutex), sizeof(PreviousMutexLayout));
}
EXPECT_EQ(sizeof(InstrumentedCondVar), sizeof(PreviousCondVarLayout));
printf(
"mutex_bytes=%zu previous_mutex_bytes=%zu aligned_mutex_bytes=%zu "
"previous_aligned_bytes=%zu condvar_bytes=%zu "
"previous_condvar_bytes=%zu\n",
sizeof(InstrumentedMutex), sizeof(PreviousMutexLayout),
sizeof(CacheAlignedInstrumentedMutex), previous_aligned,
sizeof(InstrumentedCondVar), sizeof(PreviousCondVarLayout));
}
#ifndef NDEBUG
TEST_F(SlowMutexTest, PauseDoesNotBlockBackgroundEmitter) {
stats->set_stats_level(StatsLevel::kAll);
InstrumentedMutex mutex;
Configure(&mutex, "background_race_mutex");
std::mutex sync_mutex;
std::condition_variable cv;
bool reached = false;
bool resume = false;
SyncPoint::GetInstance()->SetCallBack(
"SlowMutexTimer::FlushLogs:BeforeEmissionLock", [&](void*) {
std::unique_lock<std::mutex> lock(sync_mutex);
reached = true;
cv.notify_all();
cv.wait(lock, [&] { return resume; });
});
SyncPoint::GetInstance()->EnableProcessing();
std::thread worker([&] { Hold(&mutex, 200000); });
{
std::unique_lock<std::mutex> lock(sync_mutex);
cv.wait(lock, [&] { return reached; });
}
{
SlowMutexLoggingPause pause(context);
{
std::lock_guard<std::mutex> lock(sync_mutex);
resume = true;
cv.notify_all();
}
worker.join();
}
SyncPoint::GetInstance()->DisableProcessing();
SyncPoint::GetInstance()->ClearAllCallBacks();
EXPECT_TRUE(logger->entries.empty());
}
TEST_F(SlowMutexTest, FinalUnlockDoesNotReadDestroyedMutex) {
stats->set_stats_level(StatsLevel::kAll);
InstrumentedMutex* mutex = new InstrumentedMutex();
Configure(mutex, "destroyed_mutex");
SyncPoint::GetInstance()->SetCallBack("InstrumentedMutex::Unlock:AfterUnlock",
[&](void* pointer) {
if (pointer == mutex) {
delete mutex;
mutex = nullptr;
}
});
SyncPoint::GetInstance()->EnableProcessing();
Hold(mutex, 200000);
SyncPoint::GetInstance()->DisableProcessing();
SyncPoint::GetInstance()->ClearAllCallBacks();
EXPECT_EQ(mutex, nullptr);
ASSERT_EQ(logger->entries.size(), 1);
}
#endif
TEST_F(SlowMutexTest, PausedContextDropsPendingRecordsAndResumes) {
stats->set_stats_level(StatsLevel::kAll);
InstrumentedMutex mutex;
Configure(&mutex, "paused_mutex");
{
SlowMutexLogScope defer_logs(true);
Hold(&mutex, 200000);
SlowMutexLoggingPause pause(context);
defer_logs.Release();
EXPECT_TRUE(logger->entries.empty());
clock.Advance(2000000);
Hold(&mutex, 200000);
EXPECT_TRUE(logger->entries.empty());
}
EXPECT_FALSE(context->paused.load());
Hold(&mutex, 200000);
ASSERT_EQ(logger->entries.size(), 1);
}
TEST_F(SlowMutexTest, StoppedContextDropsDeferredRecords) {
stats->set_stats_level(StatsLevel::kAll);
InstrumentedMutex mutex;
Configure(&mutex, "stopped_mutex");
{
SlowMutexLogScope defer_logs(true);
Hold(&mutex, 200000);
{
SlowMutexLoggingPause pause(context);
context->stopped.store(true);
}
}
EXPECT_TRUE(logger->entries.empty());
clock.Advance(2000000);
Hold(&mutex, 200000);
EXPECT_TRUE(logger->entries.empty());
EXPECT_EQ(stats->get_stats_level(), StatsLevel::kAll);
EXPECT_EQ(threshold_us.load(), 100000);
}
TEST_F(SlowMutexTest, RecordBufferIsBounded) {
stats->set_stats_level(StatsLevel::kAll);
std::array<std::unique_ptr<InstrumentedMutex>, 9> mutexes;
for (std::unique_ptr<InstrumentedMutex>& mutex : mutexes) {
mutex.reset(new InstrumentedMutex());
Configure(mutex.get(), "nested_mutex");
mutex->Lock();
}
clock.Advance(200000);
for (size_t i = mutexes.size(); i > 0; --i) {
mutexes[i - 1]->Unlock();
}
ASSERT_EQ(logger->entries.size(), 8);
EXPECT_NE(logger->entries.front().find("buffer_dropped_records=1"),
std::string::npos);
clock.Advance(2000000);
Hold(mutexes.front().get(), 200000);
ASSERT_EQ(logger->entries.size(), 9);
EXPECT_NE(logger->entries.back().find("buffer_dropped_records=0"),
std::string::npos);
}
} // namespace ROCKSDB_NAMESPACE
int main(int argc, char** argv) {
ROCKSDB_NAMESPACE::port::InstallStackTraceHandler();
::testing::InitGoogleTest(&argc, argv);
return RUN_ALL_TESTS();
}