diff options
author | qrort <qrort@yandex-team.com> | 2022-11-30 23:47:12 +0300 |
---|---|---|
committer | qrort <qrort@yandex-team.com> | 2022-11-30 23:47:12 +0300 |
commit | 22f8ae0e3f5d68b92aecccdf96c1d841a0334311 (patch) | |
tree | bffa27765faf54126ad44bcafa89fadecb7a73d7 /library/cpp/yt | |
parent | 332b99e2173f0425444abb759eebcb2fafaa9209 (diff) | |
download | ydb-22f8ae0e3f5d68b92aecccdf96c1d841a0334311.tar.gz |
validate canons without yatest_common
Diffstat (limited to 'library/cpp/yt')
24 files changed, 2216 insertions, 0 deletions
diff --git a/library/cpp/yt/containers/sharded_set-inl.h b/library/cpp/yt/containers/sharded_set-inl.h new file mode 100644 index 00000000000..67d5be58c62 --- /dev/null +++ b/library/cpp/yt/containers/sharded_set-inl.h @@ -0,0 +1,217 @@ +#ifndef SHARDED_SET_INL_H_ +#error "Direct inclusion of this file is not allowed, include sharded_set.h" +// For the sake of sane code completion. +#include "sharded_set.h" +#endif + +#include <library/cpp/yt/assert/assert.h> + +namespace NYT { + +//////////////////////////////////////////////////////////////////////////////// + +template <class T, int N, class F, class S> +class TShardedSet<T, N, F, S>::const_iterator +{ +private: + friend class TShardedSet<T, N, F, S>; + + using TOwner = TShardedSet<T, N, F, S>; + using TShardIterator = typename S::const_iterator; + + const TOwner* const Owner_; + + int ShardIndex_; + TShardIterator ShardIterator_; + + const_iterator( + const TOwner* owner, + int shardIndex, + TShardIterator shardIterator) + : Owner_(owner) + , ShardIndex_(shardIndex) + , ShardIterator_(shardIterator) + { } + + bool IsValid() const + { + return ShardIterator_ != Owner_->Shards_[ShardIndex_].end(); + } + + void FastForward() + { + while (ShardIndex_ != N - 1 && !IsValid()) { + ++ShardIndex_; + ShardIterator_ = Owner_->Shards_[ShardIndex_].begin(); + } + } + +public: + using difference_type = typename std::iterator_traits<TShardIterator>::difference_type; + using value_type = typename std::iterator_traits<TShardIterator>::value_type; + using pointer = typename std::iterator_traits<TShardIterator>::pointer; + using reference = typename std::iterator_traits<TShardIterator>::reference; + using iterator_category = std::forward_iterator_tag; + + const_iterator& operator++() + { + ++ShardIterator_; + FastForward(); + + return *this; + } + + const_iterator operator++(int) + { + auto result = *this; + + ++ShardIterator_; + FastForward(); + + return result; + } + + bool operator==(const const_iterator& rhs) const + { + return + ShardIndex_ == rhs.ShardIndex_ && + ShardIterator_ == rhs.ShardIterator_; + } + + bool operator!=(const const_iterator& rhs) const + { + return !(*this == rhs); + } + + const T& operator*() const + { + return *ShardIterator_; + } + + const T* operator->() const + { + return &operator*(); + } +}; + +//////////////////////////////////////////////////////////////////////////////// + +template <class T, int N, class F, class S> +TShardedSet<T, N, F, S>::TShardedSet(F elementToShard) + : ElementToShard_(elementToShard) +{ } + +template <class T, int N, class F, class S> +bool TShardedSet<T, N, F, S>::empty() const +{ + return size() == 0; +} + +template <class T, int N, class F, class S> +typename TShardedSet<T, N, F, S>::size_type TShardedSet<T, N, F, S>::size() const +{ + size_type result = 0; + for (const auto& shard : Shards_) { + result += shard.size(); + } + + return result; +} + +template <class T, int N, class F, class S> +const T& TShardedSet<T, N, F, S>::front() const +{ + return *begin(); +} + +template <class T, int N, class F, class S> +typename TShardedSet<T, N, F, S>::size_type TShardedSet<T, N, F, S>::count(const T& value) const +{ + return GetShard(value).count(value); +} + +template <class T, int N, class F, class S> +bool TShardedSet<T, N, F, S>::contains(const T& value) const +{ + return GetShard(value).contains(value); +} + +template <class T, int N, class F, class S> +std::pair<typename TShardedSet<T, N, F, S>::const_iterator, bool> TShardedSet<T, N, F, S>::insert(const T& value) +{ + auto shardIndex = ElementToShard_(value); + auto& shard = Shards_[shardIndex]; + auto [shardIterator, inserted] = shard.insert(value); + + const_iterator iterator(this, shardIndex, shardIterator); + return {iterator, inserted}; +} + +template <class T, int N, class F, class S> +bool TShardedSet<T, N, F, S>::erase(const T& value) +{ + return GetShard(value).erase(value); +} + +template <class T, int N, class F, class S> +void TShardedSet<T, N, F, S>::clear() +{ + for (auto& shard : Shards_) { + shard.clear(); + } +} + +template <class T, int N, class F, class S> +typename TShardedSet<T, N, F, S>::const_iterator TShardedSet<T, N, F, S>::begin() const +{ + const_iterator iterator(this, /*shardIndex*/ 0, /*shardIterator*/ Shards_[0].begin()); + iterator.FastForward(); + + return iterator; +} + +template <class T, int N, class F, class S> +typename TShardedSet<T, N, F, S>::const_iterator TShardedSet<T, N, F, S>::cbegin() const +{ + return begin(); +} + +template <class T, int N, class F, class S> +typename TShardedSet<T, N, F, S>::const_iterator TShardedSet<T, N, F, S>::end() const +{ + return const_iterator(this, /*shardIndex*/ N - 1, /*shardIterator*/ Shards_[N - 1].end()); +} + +template <class T, int N, class F, class S> +typename TShardedSet<T, N, F, S>::const_iterator TShardedSet<T, N, F, S>::cend() const +{ + return end(); +} + +template <class T, int N, class F, class S> +const S& TShardedSet<T, N, F, S>::Shard(int shardIndex) const +{ + return Shards_[shardIndex]; +} + +template <class T, int N, class F, class S> +S& TShardedSet<T, N, F, S>::MutableShard(int shardIndex) +{ + return Shards_[shardIndex]; +} + +template <class T, int N, class F, class S> +S& TShardedSet<T, N, F, S>::GetShard(const T& value) +{ + return Shards_[ElementToShard_(value)]; +} + +template <class T, int N, class F, class S> +const S& TShardedSet<T, N, F, S>::GetShard(const T& value) const +{ + return Shards_[ElementToShard_(value)]; +} + +//////////////////////////////////////////////////////////////////////////////// + +} // namespace NYT diff --git a/library/cpp/yt/containers/sharded_set.h b/library/cpp/yt/containers/sharded_set.h new file mode 100644 index 00000000000..fa24893aa44 --- /dev/null +++ b/library/cpp/yt/containers/sharded_set.h @@ -0,0 +1,69 @@ +#pragma once + +#include <util/generic/hash_set.h> + +#include <array> +#include <cstddef> +#include <utility> + +namespace NYT { + +//////////////////////////////////////////////////////////////////////////////// + +//! A set that stores elements divided into fixed amount of shards. +//! Provides access to whole set and particular shards. +//! The interface is pretty minimalistic, feel free to extend it when needed. +template <class T, int N, class F, class S = THashSet<T>> +class TShardedSet +{ +public: + using size_type = size_t; + using difference_type = ptrdiff_t; + + using value_type = T; + + class const_iterator; + + explicit TShardedSet(F elementToShard = F()); + + [[nodiscard]] bool empty() const; + + size_type size() const; + + const T& front() const; + + size_type count(const T& value) const; + + bool contains(const T& value) const; + + std::pair<const_iterator, bool> insert(const T& value); + + bool erase(const T& value); + + void clear(); + + const_iterator begin() const; + const_iterator cbegin() const; + + const_iterator end() const; + const_iterator cend() const; + + const S& Shard(int shardIndex) const; + S& MutableShard(int shardIndex); + +private: + std::array<S, N> Shards_; + + const F ElementToShard_; + + S& GetShard(const T& value); + const S& GetShard(const T& value) const; +}; + +//////////////////////////////////////////////////////////////////////////////// + +} // namespace NYT + +#define SHARDED_SET_INL_H_ +#include "sharded_set-inl.h" +#undef SHARDED_SET_INL_H_ diff --git a/library/cpp/yt/containers/unittests/sharded_set_ut.cpp b/library/cpp/yt/containers/unittests/sharded_set_ut.cpp new file mode 100644 index 00000000000..2c4f8c59356 --- /dev/null +++ b/library/cpp/yt/containers/unittests/sharded_set_ut.cpp @@ -0,0 +1,121 @@ +#include <library/cpp/yt/containers/sharded_set.h> + +#include <library/cpp/testing/gtest/gtest.h> + +#include <random> + +namespace NYT { +namespace { + +//////////////////////////////////////////////////////////////////////////////// + +struct TIntToShard +{ + int operator()(int value) const + { + return value % 16; + } +}; + +using TSet = TShardedSet<int, 16, TIntToShard>; + +//////////////////////////////////////////////////////////////////////////////// + +TEST(CompactSetTest, Insert) +{ + TSet set; + + for (int i = 0; i < 4; i++) { + set.insert(i); + } + + for (int i = 0; i < 4; i++) { + set.insert(i); + } + + EXPECT_EQ(4u, set.size()); + + for (int i = 0; i < 4; i++) + EXPECT_EQ(1u, set.count(i)); + + EXPECT_EQ(0u, set.count(4)); +} + +TEST(CompactSetTest, Erase) +{ + TSet set; + + for (int i = 0; i < 8; i++) { + set.insert(i); + } + + EXPECT_EQ(8u, set.size()); + + // Remove elements one by one and check if all other elements are still there. + for (int i = 0; i < 8; i++) { + EXPECT_EQ(1u, set.count(i)); + EXPECT_TRUE(set.erase(i)); + EXPECT_EQ(0u, set.count(i)); + EXPECT_EQ(8u - i - 1, set.size()); + for (int j = i + 1; j < 8; j++) { + EXPECT_EQ(1u, set.count(j)); + } + } + + EXPECT_EQ(0u, set.count(8)); +} + +TEST(CompactSetTest, StressTest) +{ + TSet set; + + constexpr int Iterations = 1'000'000; + constexpr int Values = 128; + + THashSet<int> values; + + auto checkEverything = [&] { + EXPECT_EQ(values.size(), set.size()); + EXPECT_EQ(values.empty(), set.empty()); + EXPECT_EQ(values, THashSet<int>(set.begin(), set.end())); + + std::array<THashSet<int>, 16> shards; + for (int value : values) { + shards[value % 16].insert(value); + } + for (int shardIndex = 0; shardIndex < 16; ++shardIndex) { + EXPECT_EQ(shards[shardIndex], set.Shard(shardIndex)); + } + + for (int value = 0; value < Values; ++value) { + EXPECT_EQ(values.contains(value), set.contains(value)); + EXPECT_EQ(values.count(value), set.count(value)); + } + }; + + std::mt19937_64 rng(42); + + for (int iteration = 0; iteration < Iterations; ++iteration) { + if (rng() % 100 == 0) { + set.clear(); + values.clear(); + checkEverything(); + } + + int value = rng() % Values; + if (rng() % 2 == 0) { + set.insert(value); + values.insert(value); + } else { + set.erase(value); + values.erase(value); + } + + checkEverything(); + } +} + +//////////////////////////////////////////////////////////////////////////////// + +} // namespace +} // namespace NYT diff --git a/library/cpp/yt/cpu_clock/benchmark/benchmark.cpp b/library/cpp/yt/cpu_clock/benchmark/benchmark.cpp new file mode 100644 index 00000000000..e326da68379 --- /dev/null +++ b/library/cpp/yt/cpu_clock/benchmark/benchmark.cpp @@ -0,0 +1,32 @@ +#include "benchmark/benchmark.h" +#include <benchmark/benchmark.h> + +#include <library/cpp/yt/cpu_clock/clock.h> + +namespace NYT { +namespace { + +//////////////////////////////////////////////////////////////////////////////// + +void BenchmarkCpuTime(benchmark::State& state) +{ + for (auto _ : state) { + benchmark::DoNotOptimize(GetCpuInstant()); + } +} + +BENCHMARK(BenchmarkCpuTime); + +void BenchmarkInstantNow(benchmark::State& state) +{ + for (auto _ : state) { + benchmark::DoNotOptimize(TInstant::Now()); + } +} + +BENCHMARK(BenchmarkInstantNow); + +//////////////////////////////////////////////////////////////////////////////// + +} // namespace +} // namespace NYT diff --git a/library/cpp/yt/cpu_clock/unittests/clock_ut.cpp b/library/cpp/yt/cpu_clock/unittests/clock_ut.cpp new file mode 100644 index 00000000000..bd9cb6d4be1 --- /dev/null +++ b/library/cpp/yt/cpu_clock/unittests/clock_ut.cpp @@ -0,0 +1,46 @@ +#include <gtest/gtest.h> + +#include <library/cpp/yt/cpu_clock/clock.h> + +namespace NYT { +namespace { + +//////////////////////////////////////////////////////////////////////////////// + +template <class T> +i64 DiffMS(T a, T b) +{ + return a >= b + ? static_cast<i64>(a.MilliSeconds()) - static_cast<i64>(b.MilliSeconds()) + : DiffMS(b, a); +} + +TEST(TTimingTest, GetInstant) +{ + GetInstant(); + + EXPECT_LE(DiffMS(GetInstant(), TInstant::Now()), 10); +} + +TEST(TTimingTest, InstantVSCpuInstant) +{ + auto instant1 = TInstant::Now(); + auto cpuInstant = InstantToCpuInstant(instant1); + auto instant2 = CpuInstantToInstant(cpuInstant); + EXPECT_LE(DiffMS(instant1, instant2), 10); +} + +TEST(TTimingTest, DurationVSCpuDuration) +{ + auto cpuInstant1 = GetCpuInstant(); + constexpr auto duration1 = TDuration::MilliSeconds(100); + Sleep(duration1); + auto cpuInstant2 = GetCpuInstant(); + auto duration2 = CpuDurationToDuration(cpuInstant2 - cpuInstant1); + EXPECT_LE(DiffMS(duration1, duration2), 10); +} + +//////////////////////////////////////////////////////////////////////////////// + +} // namespace +} // namespace NYT diff --git a/library/cpp/yt/farmhash/farm_hash.h b/library/cpp/yt/farmhash/farm_hash.h new file mode 100644 index 00000000000..fe4c8193a0a --- /dev/null +++ b/library/cpp/yt/farmhash/farm_hash.h @@ -0,0 +1,63 @@ +#pragma once + +#include <contrib/libs/farmhash/farmhash.h> + +#include <util/system/types.h> + +#include <util/generic/strbuf.h> + +namespace NYT { + +//////////////////////////////////////////////////////////////////////////////// + +using TFingerprint = ui64; + +static inline TFingerprint FarmHash(ui64 value) +{ + return ::util::Fingerprint(value); +} + +static inline TFingerprint FarmHash(const void* buf, size_t len) +{ + return ::util::Hash64(static_cast<const char*>(buf), len); +} + +static inline TFingerprint FarmHash(const void* buf, size_t len, ui64 seed) +{ + return ::util::Hash64WithSeed(static_cast<const char*>(buf), len, seed); +} + +static inline TFingerprint FarmFingerprint(ui64 value) +{ + return ::util::Fingerprint(value); +} + +static inline TFingerprint FarmFingerprint(const void* buf, size_t len) +{ + return ::util::Fingerprint64(static_cast<const char*>(buf), len); +} + +static inline TFingerprint FarmFingerprint(TStringBuf buf) +{ + return FarmFingerprint(buf.Data(), buf.Size()); +} + +static inline TFingerprint FarmFingerprint(ui64 first, ui64 second) +{ + return ::util::Fingerprint(::util::Uint128(first, second)); +} + +// Forever-fixed Google FarmHash fingerprint. +template <class T> +TFingerprint FarmFingerprint(const T* begin, const T* end) +{ + ui64 result = 0xdeadc0de; + for (const auto* value = begin; value < end; ++value) { + result = FarmFingerprint(result, FarmFingerprint(*value)); + } + return result ^ (end - begin); +} + +//////////////////////////////////////////////////////////////////////////////// + +} // namespace NYT diff --git a/library/cpp/yt/logging/logger-inl.h b/library/cpp/yt/logging/logger-inl.h new file mode 100644 index 00000000000..6118aa00476 --- /dev/null +++ b/library/cpp/yt/logging/logger-inl.h @@ -0,0 +1,296 @@ +#ifndef LOGGER_INL_H_ +#error "Direct inclusion of this file is not allowed, include logger.h" +// For the sake of sane code completion. +#include "logger.h" +#endif +#undef LOGGER_INL_H_ + +#include <library/cpp/yt/yson_string/convert.h> +#include <library/cpp/yt/yson_string/string.h> + +namespace NYT::NLogging { + +//////////////////////////////////////////////////////////////////////////////// + +inline bool TLogger::IsAnchorUpToDate(const TLoggingAnchor& position) const +{ + return + !Category_ || + position.CurrentVersion == Category_->ActualVersion->load(std::memory_order::relaxed); +} + +template <class... TArgs> +void TLogger::AddTag(const char* format, TArgs&&... args) +{ + AddRawTag(Format(format, std::forward<TArgs>(args)...)); +} + +template <class TType> +void TLogger::AddStructuredTag(TStringBuf key, TType value) +{ + StructuredTags_.emplace_back(key, NYson::ConvertToYsonString(value)); +} + +template <class... TArgs> +TLogger TLogger::WithTag(const char* format, TArgs&&... args) const +{ + auto result = *this; + result.AddTag(format, std::forward<TArgs>(args)...); + return result; +} + +template <class TType> +TLogger TLogger::WithStructuredTag(TStringBuf key, TType value) const +{ + auto result = *this; + result.AddStructuredTag(key, value); + return result; +} + +Y_FORCE_INLINE bool TLogger::IsLevelEnabled(ELogLevel level) const +{ + // This is the first check which is intended to be inlined next to + // logging invocation point. Check below is almost zero-cost due + // to branch prediction (which requires inlining for proper work). + if (level < MinLevel_) { + return false; + } + + // Next check is heavier and requires full log manager definition which + // is undesirable in -inl.h header file. This is why we extract it + // to a separate method which is implemented in cpp file. + return IsLevelEnabledHeavy(level); +} + +//////////////////////////////////////////////////////////////////////////////// + +namespace NDetail { + +struct TMessageStringBuilderContext +{ + TSharedMutableRef Chunk; +}; + +struct TMessageBufferTag +{ }; + +class TMessageStringBuilder + : public TStringBuilderBase +{ +public: + TSharedRef Flush(); + + // For testing only. + static void DisablePerThreadCache(); + +protected: + void DoReset() override; + void DoReserve(size_t newLength) override; + +private: + struct TPerThreadCache + { + ~TPerThreadCache(); + + TSharedMutableRef Chunk; + size_t ChunkOffset = 0; + }; + + TSharedMutableRef Buffer_; + + static thread_local TPerThreadCache* Cache_; + static thread_local bool CacheDestroyed_; + static TPerThreadCache* GetCache(); + + static constexpr size_t ChunkSize = 128_KB - 64; +}; + +inline bool HasMessageTags( + const TLoggingContext& loggingContext, + const TLogger& logger) +{ + if (logger.GetTag()) { + return true; + } + if (loggingContext.TraceLoggingTag) { + return true; + } + return false; +} + +inline void AppendMessageTags( + TStringBuilderBase* builder, + const TLoggingContext& loggingContext, + const TLogger& logger) +{ + bool printComma = false; + if (const auto& loggerTag = logger.GetTag()) { + builder->AppendString(loggerTag); + printComma = true; + } + if (auto traceLoggingTag = loggingContext.TraceLoggingTag) { + if (printComma) { + builder->AppendString(TStringBuf(", ")); + } + builder->AppendString(traceLoggingTag); + } +} + +inline void AppendLogMessage( + TStringBuilderBase* builder, + const TLoggingContext& loggingContext, + const TLogger& logger, + TRef message) +{ + if (HasMessageTags(loggingContext, logger)) { + if (message.Size() >= 1 && message[message.Size() - 1] == ')') { + builder->AppendString(TStringBuf(message.Begin(), message.Size() - 1)); + builder->AppendString(TStringBuf(", ")); + } else { + builder->AppendString(TStringBuf(message.Begin(), message.Size())); + builder->AppendString(TStringBuf(" (")); + } + AppendMessageTags(builder, loggingContext, logger); + builder->AppendChar(')'); + } else { + builder->AppendString(TStringBuf(message.Begin(), message.Size())); + } +} + +template <class... TArgs> +void AppendLogMessageWithFormat( + TStringBuilderBase* builder, + const TLoggingContext& loggingContext, + const TLogger& logger, + TStringBuf format, + TArgs&&... args) +{ + if (HasMessageTags(loggingContext, logger)) { + if (format.size() >= 2 && format[format.size() - 1] == ')') { + builder->AppendFormat(format.substr(0, format.size() - 1), std::forward<TArgs>(args)...); + builder->AppendString(TStringBuf(", ")); + } else { + builder->AppendFormat(format, std::forward<TArgs>(args)...); + builder->AppendString(TStringBuf(" (")); + } + AppendMessageTags(builder, loggingContext, logger); + builder->AppendChar(')'); + } else { + builder->AppendFormat(format, std::forward<TArgs>(args)...); + } +} + +struct TLogMessage +{ + TSharedRef MessageRef; + TStringBuf Anchor; +}; + +template <size_t Length, class... TArgs> +TLogMessage BuildLogMessage( + const TLoggingContext& loggingContext, + const TLogger& logger, + const char (&format)[Length], + TArgs&&... args) +{ + TMessageStringBuilder builder; + AppendLogMessageWithFormat(&builder, loggingContext, logger, format, std::forward<TArgs>(args)...); + return {builder.Flush(), format}; +} + +template <class T> +TLogMessage BuildLogMessage( + const TLoggingContext& loggingContext, + const TLogger& logger, + const T& obj) +{ + TMessageStringBuilder builder; + FormatValue(&builder, obj, TStringBuf()); + if (HasMessageTags(loggingContext, logger)) { + builder.AppendString(TStringBuf(" (")); + AppendMessageTags(&builder, loggingContext, logger); + builder.AppendChar(')'); + } + return {builder.Flush(), TStringBuf()}; +} + +inline TLogMessage BuildLogMessage( + const TLoggingContext& loggingContext, + const TLogger& logger, + TStringBuf message) +{ + TMessageStringBuilder builder; + builder.AppendString(message); + if (HasMessageTags(loggingContext, logger)) { + builder.AppendString(TStringBuf(" (")); + AppendMessageTags(&builder, loggingContext, logger); + builder.AppendChar(')'); + } + return {builder.Flush(), message}; +} + +template <size_t Length> +TLogMessage BuildLogMessage( + const TLoggingContext& loggingContext, + const TLogger& logger, + const char (&message)[Length]) +{ + return BuildLogMessage( + loggingContext, + logger, + TStringBuf(message)); +} + +inline TLogMessage BuildLogMessage( + const TLoggingContext& loggingContext, + const TLogger& logger, + TSharedRef&& message) +{ + if (HasMessageTags(loggingContext, logger)) { + TMessageStringBuilder builder; + AppendLogMessage(&builder, loggingContext, logger, message); + return {builder.Flush(), TStringBuf()}; + } else { + return {std::move(message), TStringBuf()}; + } +} + +inline TLogEvent CreateLogEvent( + const TLoggingContext& loggingContext, + const TLogger& logger, + ELogLevel level) +{ + TLogEvent event; + event.Instant = loggingContext.Instant; + event.Category = logger.GetCategory(); + event.Essential = logger.IsEssential(); + event.Level = level; + event.ThreadId = loggingContext.ThreadId; + event.ThreadName = loggingContext.ThreadName; + event.FiberId = loggingContext.FiberId; + event.TraceId = loggingContext.TraceId; + event.RequestId = loggingContext.RequestId; + return event; +} + +inline void LogEventImpl( + const TLoggingContext& loggingContext, + const TLogger& logger, + ELogLevel level, + ::TSourceLocation sourceLocation, + TSharedRef message) +{ + auto event = CreateLogEvent(loggingContext, logger, level); + event.MessageKind = ELogMessageKind::Unstructured; + event.MessageRef = std::move(message); + event.Family = ELogFamily::PlainText; + event.SourceFile = sourceLocation.File; + event.SourceLine = sourceLocation.Line; + logger.Write(std::move(event)); +} + +} // namespace NDetail + +//////////////////////////////////////////////////////////////////////////////// + +} // namespace NYT::NLogging diff --git a/library/cpp/yt/logging/logger.cpp b/library/cpp/yt/logging/logger.cpp new file mode 100644 index 00000000000..7f6f217d860 --- /dev/null +++ b/library/cpp/yt/logging/logger.cpp @@ -0,0 +1,249 @@ +#include "logger.h" + +#include <library/cpp/yt/assert/assert.h> + +#include <library/cpp/yt/cpu_clock/clock.h> + +#include <util/system/compiler.h> +#include <util/system/thread.h> + +namespace NYT::NLogging { + +//////////////////////////////////////////////////////////////////////////////// + +namespace NDetail { + +TSharedRef TMessageStringBuilder::Flush() +{ + return Buffer_.Slice(0, GetLength()); +} + +void TMessageStringBuilder::DisablePerThreadCache() +{ + Cache_ = nullptr; + CacheDestroyed_ = true; +} + +void TMessageStringBuilder::DoReset() +{ + Buffer_.Reset(); +} + +void TMessageStringBuilder::DoReserve(size_t newCapacity) +{ + auto oldLength = GetLength(); + newCapacity = FastClp2(newCapacity); + + auto newChunkSize = std::max(ChunkSize, newCapacity); + // Hold the old buffer until the data is copied. + auto oldBuffer = std::move(Buffer_); + auto* cache = GetCache(); + if (Y_LIKELY(cache)) { + auto oldCapacity = End_ - Begin_; + auto deltaCapacity = newCapacity - oldCapacity; + if (End_ == cache->Chunk.Begin() + cache->ChunkOffset && + cache->ChunkOffset + deltaCapacity <= cache->Chunk.Size()) + { + // Resize inplace. + Buffer_ = cache->Chunk.Slice(cache->ChunkOffset - oldCapacity, cache->ChunkOffset + deltaCapacity); + cache->ChunkOffset += deltaCapacity; + End_ = Begin_ + newCapacity; + return; + } + + if (Y_UNLIKELY(cache->ChunkOffset + newCapacity > cache->Chunk.Size())) { + cache->Chunk = TSharedMutableRef::Allocate<TMessageBufferTag>(newChunkSize, {.InitializeStorage = false}); + cache->ChunkOffset = 0; + } + + Buffer_ = cache->Chunk.Slice(cache->ChunkOffset, cache->ChunkOffset + newCapacity); + cache->ChunkOffset += newCapacity; + } else { + Buffer_ = TSharedMutableRef::Allocate<TMessageBufferTag>(newChunkSize, {.InitializeStorage = false}); + newCapacity = newChunkSize; + } + if (oldLength > 0) { + ::memcpy(Buffer_.Begin(), Begin_, oldLength); + } + Begin_ = Buffer_.Begin(); + End_ = Begin_ + newCapacity; +} + +TMessageStringBuilder::TPerThreadCache* TMessageStringBuilder::GetCache() +{ + if (Y_LIKELY(Cache_)) { + return Cache_; + } + if (CacheDestroyed_) { + return nullptr; + } + static thread_local TPerThreadCache Cache; + Cache_ = &Cache; + return Cache_; +} + +TMessageStringBuilder::TPerThreadCache::~TPerThreadCache() +{ + TMessageStringBuilder::DisablePerThreadCache(); +} + +thread_local TMessageStringBuilder::TPerThreadCache* TMessageStringBuilder::Cache_; +thread_local bool TMessageStringBuilder::CacheDestroyed_; + +} // namespace NDetail + +//////////////////////////////////////////////////////////////////////////////// + +Y_WEAK TLoggingContext GetLoggingContext() +{ + return { + .Instant = GetCpuInstant(), + .ThreadId = TThread::CurrentThreadId(), + .ThreadName = GetLoggingThreadName(), + }; +} + +Y_WEAK ILogManager* GetDefaultLogManager() +{ + return nullptr; +} + +//////////////////////////////////////////////////////////////////////////////// + +TLoggingThreadName GetLoggingThreadName() +{ + static thread_local TLoggingThreadName loggingThreadName; + if (loggingThreadName.Length == 0) { + if (auto name = TThread::CurrentThreadName()) { + auto length = std::min(TLoggingThreadName::BufferCapacity - 1, static_cast<int>(name.length())); + loggingThreadName.Length = length; + ::memcpy(loggingThreadName.Buffer.data(), name.data(), length); + } + } + return loggingThreadName; +} + +//////////////////////////////////////////////////////////////////////////////// + +TLogger::TLogger(ILogManager* logManager, TStringBuf categoryName) + : LogManager_(logManager) + , Category_(LogManager_ ? LogManager_->GetCategory(categoryName) : nullptr) + , MinLevel_(LogManager_ ? LoggerDefaultMinLevel : NullLoggerMinLevel) +{ } + +TLogger::TLogger(TStringBuf categoryName) + : TLogger(GetDefaultLogManager(), categoryName) +{ } + +TLogger::operator bool() const +{ + return LogManager_; +} + +const TLoggingCategory* TLogger::GetCategory() const +{ + return Category_; +} + +bool TLogger::IsLevelEnabledHeavy(ELogLevel level) const +{ + // Note that we managed to reach this point, i.e. level >= MinLevel_, + // which implies that MinLevel_ != ELogLevel::Maximum, so this logger was not + // default constructed, thus it has non-trivial category. + YT_ASSERT(Category_); + + if (Category_->CurrentVersion != Category_->ActualVersion->load(std::memory_order::relaxed)) { + LogManager_->UpdateCategory(const_cast<TLoggingCategory*>(Category_)); + } + + return level >= Category_->MinPlainTextLevel; +} + +bool TLogger::GetAbortOnAlert() const +{ + return LogManager_->GetAbortOnAlert(); +} + +bool TLogger::IsEssential() const +{ + return Essential_; +} + +void TLogger::UpdateAnchor(TLoggingAnchor* anchor) const +{ + LogManager_->UpdateAnchor(anchor); +} + +void TLogger::RegisterStaticAnchor(TLoggingAnchor* anchor, ::TSourceLocation sourceLocation, TStringBuf message) const +{ + LogManager_->RegisterStaticAnchor(anchor, sourceLocation, message); +} + +void TLogger::Write(TLogEvent&& event) const +{ + LogManager_->Enqueue(std::move(event)); +} + +void TLogger::AddRawTag(const TString& tag) +{ + if (!Tag_.empty()) { + Tag_ += ", "; + } + Tag_ += tag; +} + +TLogger TLogger::WithRawTag(const TString& tag) const +{ + auto result = *this; + result.AddRawTag(tag); + return result; +} + +TLogger TLogger::WithEssential(bool essential) const +{ + auto result = *this; + result.Essential_ = essential; + return result; +} + +TLogger TLogger::WithMinLevel(ELogLevel minLevel) const +{ + auto result = *this; + if (result) { + result.MinLevel_ = minLevel; + } + return result; +} + +const TString& TLogger::GetTag() const +{ + return Tag_; +} + +const TLogger::TStructuredTags& TLogger::GetStructuredTags() const +{ + return StructuredTags_; +} + +//////////////////////////////////////////////////////////////////////////////// + +void LogStructuredEvent( + const TLogger& logger, + NYson::TYsonString message, + ELogLevel level) +{ + YT_VERIFY(message.GetType() == NYson::EYsonType::MapFragment); + auto loggingContext = GetLoggingContext(); + auto event = NDetail::CreateLogEvent( + loggingContext, + logger, + level); + event.MessageKind = ELogMessageKind::Structured; + event.MessageRef = message.ToSharedRef(); + event.Family = ELogFamily::Structured; + logger.Write(std::move(event)); +} + +//////////////////////////////////////////////////////////////////////////////// + +} // namespace NYT::NLogging diff --git a/library/cpp/yt/logging/logger.h b/library/cpp/yt/logging/logger.h new file mode 100644 index 00000000000..26b415a6b66 --- /dev/null +++ b/library/cpp/yt/logging/logger.h @@ -0,0 +1,345 @@ +#pragma once + +#include "public.h" + +#include <library/cpp/yt/string/format.h> + +#include <library/cpp/yt/memory/ref.h> + +#include <library/cpp/yt/cpu_clock/public.h> + +#include <library/cpp/yt/yson_string/string.h> + +#include <library/cpp/yt/misc/guid.h> + +#include <library/cpp/yt/memory/leaky_singleton.h> + +#include <util/system/src_location.h> + +#include <util/generic/size_literals.h> + +#include <atomic> + +namespace NYT::NLogging { + +//////////////////////////////////////////////////////////////////////////////// + +struct TLoggingCategory +{ + TString Name; + //! This value is used for early dropping of plaintext events in order + //! to reduce load on logging thread for events which are definitely going + //! to be dropped due to rule setup. + //! NB: this optimization is used only for plaintext events since structured + //! logging rate is negligible comparing to the plaintext logging rate. + std::atomic<ELogLevel> MinPlainTextLevel; + std::atomic<int> CurrentVersion; + std::atomic<int>* ActualVersion; +}; + +//////////////////////////////////////////////////////////////////////////////// + +struct TLoggingAnchor +{ + std::atomic<bool> Registered = false; + ::TSourceLocation SourceLocation = {TStringBuf{}, 0}; + TString AnchorMessage; + TLoggingAnchor* NextAnchor = nullptr; + + std::atomic<int> CurrentVersion = 0; + std::atomic<bool> Enabled = false; + + struct TCounter + { + std::atomic<i64> Current = 0; + i64 Previous = 0; + }; + + TCounter MessageCounter; + TCounter ByteCounter; +}; + +//////////////////////////////////////////////////////////////////////////////// + +struct TLoggingThreadName +{ + static constexpr int BufferCapacity = 16; // including zero terminator + std::array<char, BufferCapacity> Buffer; // zero-terminated + int Length; // not including zero terminator +}; + +TLoggingThreadName GetLoggingThreadName(); + +//////////////////////////////////////////////////////////////////////////////// +// Declare some type aliases to avoid circular dependencies. +using TThreadId = size_t; +using TFiberId = size_t; +using TTraceId = TGuid; +using TRequestId = TGuid; + +//////////////////////////////////////////////////////////////////////////////// + +DEFINE_ENUM(ELogMessageKind, + (Unstructured) + (Structured) +); + +struct TLogEvent +{ + const TLoggingCategory* Category = nullptr; + ELogLevel Level = ELogLevel::Minimum; + ELogFamily Family = ELogFamily::PlainText; + bool Essential = false; + + ELogMessageKind MessageKind = ELogMessageKind::Unstructured; + TSharedRef MessageRef; + + TCpuInstant Instant = 0; + + TThreadId ThreadId = {}; + TLoggingThreadName ThreadName = {}; + + TFiberId FiberId = {}; + + TTraceId TraceId; + TRequestId RequestId; + + TStringBuf SourceFile; + int SourceLine = -1; +}; + +//////////////////////////////////////////////////////////////////////////////// + +struct ILogManager +{ + virtual ~ILogManager() = default; + + virtual void RegisterStaticAnchor( + TLoggingAnchor* position, + ::TSourceLocation sourceLocation, + TStringBuf anchorMessage) = 0; + virtual void UpdateAnchor(TLoggingAnchor* position) = 0; + + virtual void Enqueue(TLogEvent&& event) = 0; + + virtual const TLoggingCategory* GetCategory(TStringBuf categoryName) = 0; + virtual void UpdateCategory(TLoggingCategory* category) = 0; + + virtual bool GetAbortOnAlert() const = 0; +}; + +ILogManager* GetDefaultLogManager(); + +//////////////////////////////////////////////////////////////////////////////// + +struct TLoggingContext +{ + TCpuInstant Instant; + TThreadId ThreadId; + TLoggingThreadName ThreadName; + TFiberId FiberId; + TTraceId TraceId; + TRequestId RequestId; + TStringBuf TraceLoggingTag; +}; + +TLoggingContext GetLoggingContext(); + +//////////////////////////////////////////////////////////////////////////////// + +static constexpr auto NullLoggerMinLevel = ELogLevel::Maximum; + +// Min level for non-null logger depends on whether we are in debug or release build. +// - For release mode default behavior is to omit trace logging, +// this is done by setting logger min level to Debug by default. +// - For debug mode logger min level is set to trace by default, so that trace logging is +// allowed by logger, but still may be discarded by category min level. +#ifdef NDEBUG +static constexpr auto LoggerDefaultMinLevel = ELogLevel::Debug; +#else +static constexpr auto LoggerDefaultMinLevel = ELogLevel::Trace; +#endif + +class TLogger +{ +public: + using TStructuredTag = std::pair<TString, NYson::TYsonString>; + // TODO(max42): switch to TCompactVector after YT-15430. + using TStructuredTags = std::vector<TStructuredTag>; + + TLogger() = default; + TLogger(const TLogger& other) = default; + TLogger& operator=(const TLogger& other) = default; + + TLogger(ILogManager* logManager, TStringBuf categoryName); + explicit TLogger(TStringBuf categoryName); + + explicit operator bool() const; + + const TLoggingCategory* GetCategory() const; + + //! Validate that level is admitted by logger's own min level + //! and by category's min level. + bool IsLevelEnabled(ELogLevel level) const; + + bool GetAbortOnAlert() const; + + bool IsEssential() const; + + bool IsAnchorUpToDate(const TLoggingAnchor& anchor) const; + void UpdateAnchor(TLoggingAnchor* anchor) const; + void RegisterStaticAnchor(TLoggingAnchor* anchor, ::TSourceLocation sourceLocation, TStringBuf message) const; + + void Write(TLogEvent&& event) const; + + void AddRawTag(const TString& tag); + template <class... TArgs> + void AddTag(const char* format, TArgs&&... args); + + template <class TType> + void AddStructuredTag(TStringBuf key, TType value); + + TLogger WithRawTag(const TString& tag) const; + template <class... TArgs> + TLogger WithTag(const char* format, TArgs&&... args) const; + + template <class TType> + TLogger WithStructuredTag(TStringBuf key, TType value) const; + + TLogger WithMinLevel(ELogLevel minLevel) const; + + TLogger WithEssential(bool essential = true) const; + + const TString& GetTag() const; + const TStructuredTags& GetStructuredTags() const; + +protected: + // These fields are set only during logger creation, so they are effectively const + // and accessing them is thread-safe. + ILogManager* LogManager_ = nullptr; + const TLoggingCategory* Category_ = nullptr; + bool Essential_ = false; + ELogLevel MinLevel_ = NullLoggerMinLevel; + TString Tag_; + TStructuredTags StructuredTags_; + +private: + //! This method checks level against category's min level. + //! Refer to comment in TLogger::IsLevelEnabled for more details. + bool IsLevelEnabledHeavy(ELogLevel level) const; +}; + +//////////////////////////////////////////////////////////////////////////////// + +void LogStructuredEvent( + const TLogger& logger, + NYson::TYsonString message, + ELogLevel level); + +//////////////////////////////////////////////////////////////////////////////// + +#ifdef YT_ENABLE_TRACE_LOGGING +#define YT_LOG_TRACE(...) YT_LOG_EVENT(Logger, ::NYT::NLogging::ELogLevel::Trace, __VA_ARGS__) +#define YT_LOG_TRACE_IF(condition, ...) if (condition) YT_LOG_TRACE(__VA_ARGS__) +#define YT_LOG_TRACE_UNLESS(condition, ...) if (!(condition)) YT_LOG_TRACE(__VA_ARGS__) +#else +#define YT_LOG_UNUSED(...) if (true) { } else { YT_LOG_DEBUG(__VA_ARGS__); } +#define YT_LOG_TRACE(...) YT_LOG_UNUSED(__VA_ARGS__) +#define YT_LOG_TRACE_IF(condition, ...) YT_LOG_UNUSED(__VA_ARGS__) +#define YT_LOG_TRACE_UNLESS(condition, ...) YT_LOG_UNUSED(__VA_ARGS__) +#endif + +#define YT_LOG_DEBUG(...) YT_LOG_EVENT(Logger, ::NYT::NLogging::ELogLevel::Debug, __VA_ARGS__) +#define YT_LOG_DEBUG_IF(condition, ...) if (condition) YT_LOG_DEBUG(__VA_ARGS__) +#define YT_LOG_DEBUG_UNLESS(condition, ...) if (!(condition)) YT_LOG_DEBUG(__VA_ARGS__) + +#define YT_LOG_INFO(...) YT_LOG_EVENT(Logger, ::NYT::NLogging::ELogLevel::Info, __VA_ARGS__) +#define YT_LOG_INFO_IF(condition, ...) if (condition) YT_LOG_INFO(__VA_ARGS__) +#define YT_LOG_INFO_UNLESS(condition, ...) if (!(condition)) YT_LOG_INFO(__VA_ARGS__) + +#define YT_LOG_WARNING(...) YT_LOG_EVENT(Logger, ::NYT::NLogging::ELogLevel::Warning, __VA_ARGS__) +#define YT_LOG_WARNING_IF(condition, ...) if (condition) YT_LOG_WARNING(__VA_ARGS__) +#define YT_LOG_WARNING_UNLESS(condition, ...) if (!(condition)) YT_LOG_WARNING(__VA_ARGS__) + +#define YT_LOG_ERROR(...) YT_LOG_EVENT(Logger, ::NYT::NLogging::ELogLevel::Error, __VA_ARGS__) +#define YT_LOG_ERROR_IF(condition, ...) if (condition) YT_LOG_ERROR(__VA_ARGS__) +#define YT_LOG_ERROR_UNLESS(condition, ...) if (!(condition)) YT_LOG_ERROR(__VA_ARGS__) + +#define YT_LOG_ALERT(...) \ + do { \ + YT_LOG_EVENT(Logger, ::NYT::NLogging::ELogLevel::Alert, __VA_ARGS__); \ + YT_VERIFY(!Logger.GetAbortOnAlert()); \ + } while(false) +#define YT_LOG_ALERT_IF(condition, ...) if (condition) YT_LOG_ALERT(__VA_ARGS__) +#define YT_LOG_ALERT_UNLESS(condition, ...) if (!(condition)) YT_LOG_ALERT(__VA_ARGS__) + +#define YT_LOG_FATAL(...) \ + do { \ + YT_LOG_EVENT(Logger, ::NYT::NLogging::ELogLevel::Fatal, __VA_ARGS__); \ + YT_ABORT(); \ + } while(false) +#define YT_LOG_FATAL_IF(condition, ...) if (Y_UNLIKELY(condition)) YT_LOG_FATAL(__VA_ARGS__) +#define YT_LOG_FATAL_UNLESS(condition, ...) if (!Y_LIKELY(condition)) YT_LOG_FATAL(__VA_ARGS__) + +#define YT_LOG_EVENT(logger, level, ...) \ + YT_LOG_EVENT_WITH_ANCHOR(logger, level, nullptr, __VA_ARGS__) + +#define YT_LOG_EVENT_WITH_ANCHOR(logger, level, anchor, ...) \ + do { \ + const auto& logger__##__LINE__ = (logger); \ + auto level__##__LINE__ = (level); \ + \ + if (!logger__##__LINE__.IsLevelEnabled(level__##__LINE__)) { \ + break; \ + } \ + \ + auto location__##__LINE__ = __LOCATION__; \ + \ + ::NYT::NLogging::TLoggingAnchor* anchor__##__LINE__ = (anchor); \ + if (!anchor__##__LINE__) { \ + static ::NYT::TLeakyStorage<::NYT::NLogging::TLoggingAnchor> staticAnchor__##__LINE__; \ + anchor__##__LINE__ = staticAnchor__##__LINE__.Get(); \ + } \ + \ + bool anchorUpToDate__##__LINE__ = logger__##__LINE__.IsAnchorUpToDate(*anchor__##__LINE__); \ + if (anchorUpToDate__##__LINE__ && !anchor__##__LINE__->Enabled.load(std::memory_order::relaxed)) { \ + break; \ + } \ + \ + auto loggingContext__##__LINE__ = ::NYT::NLogging::GetLoggingContext(); \ + auto message__##__LINE__ = ::NYT::NLogging::NDetail::BuildLogMessage(loggingContext__##__LINE__, logger__##__LINE__, __VA_ARGS__); \ + \ + if (!anchorUpToDate__##__LINE__) { \ + logger__##__LINE__.RegisterStaticAnchor(anchor__##__LINE__, location__##__LINE__, message__##__LINE__.Anchor); \ + logger__##__LINE__.UpdateAnchor(anchor__##__LINE__); \ + } \ + \ + if (!anchor__##__LINE__->Enabled.load(std::memory_order::relaxed)) { \ + break; \ + } \ + \ + static thread_local i64 localByteCounter__##__LINE__; \ + static thread_local ui8 localMessageCounter__##__LINE__; \ + \ + localByteCounter__##__LINE__ += message__##__LINE__.MessageRef.Size(); \ + if (Y_UNLIKELY(++localMessageCounter__##__LINE__ == 0)) { \ + anchor__##__LINE__->MessageCounter.Current += 256; \ + anchor__##__LINE__->ByteCounter.Current += localByteCounter__##__LINE__; \ + localByteCounter__##__LINE__ = 0; \ + } \ + \ + ::NYT::NLogging::NDetail::LogEventImpl( \ + loggingContext__##__LINE__, \ + logger__##__LINE__, \ + level__##__LINE__, \ + location__##__LINE__, \ + std::move(message__##__LINE__.MessageRef)); \ + } while (false) + +//////////////////////////////////////////////////////////////////////////////// + +} // namespace NYT::NLogging + +#define LOGGER_INL_H_ +#include "logger-inl.h" +#undef LOGGER_INL_H_ diff --git a/library/cpp/yt/logging/public.h b/library/cpp/yt/logging/public.h new file mode 100644 index 00000000000..1e2b59ca0d3 --- /dev/null +++ b/library/cpp/yt/logging/public.h @@ -0,0 +1,39 @@ +#pragma once + +#include <library/cpp/yt/misc/enum.h> + +namespace NYT::NLogging { + +//////////////////////////////////////////////////////////////////////////////// + +// Any change to this enum must be also propagated to FormatLevel. +DEFINE_ENUM(ELogLevel, + (Minimum) + (Trace) + (Debug) + (Info) + (Warning) + (Error) + (Alert) + (Fatal) + (Maximum) +); + +DEFINE_ENUM(ELogFamily, + (PlainText) + (Structured) +); + +//////////////////////////////////////////////////////////////////////////////// + +struct TLoggingCategory; +struct TLoggingAnchor; +struct TLogEvent; +struct TLoggingContext; + +class TLogger; +struct ILogManager; + +//////////////////////////////////////////////////////////////////////////////// + +} // namespace NYT::NLogging diff --git a/library/cpp/yt/logging/unittests/logger_ut.cpp b/library/cpp/yt/logging/unittests/logger_ut.cpp new file mode 100644 index 00000000000..7696ea4a83b --- /dev/null +++ b/library/cpp/yt/logging/unittests/logger_ut.cpp @@ -0,0 +1,38 @@ +#include <library/cpp/testing/gtest/gtest.h> + +#include <library/cpp/yt/logging/logger.h> + +namespace NYT::NLogging { +namespace { + +//////////////////////////////////////////////////////////////////////////////// + +TEST(TLogger, NullByDefault) +{ + { + TLogger logger; + EXPECT_FALSE(logger); + EXPECT_FALSE(logger.IsLevelEnabled(ELogLevel::Fatal)); + } + { + TLogger logger{"Category"}; + EXPECT_FALSE(logger); + EXPECT_FALSE(logger.IsLevelEnabled(ELogLevel::Fatal)); + } +} + +TEST(TLogger, CopyOfNullLogger) +{ + TLogger nullLogger{/*logManager*/ nullptr, "Category"}; + ASSERT_FALSE(nullLogger); + + auto logger = nullLogger.WithMinLevel(ELogLevel::Debug); + + EXPECT_FALSE(logger); + EXPECT_FALSE(logger.IsLevelEnabled(ELogLevel::Fatal)); +} + +//////////////////////////////////////////////////////////////////////////////// + +} // namespace +} // namespace NYT::NLogging diff --git a/library/cpp/yt/memory/leaky_ref_counted_singleton-inl.h b/library/cpp/yt/memory/leaky_ref_counted_singleton-inl.h new file mode 100644 index 00000000000..1fba63c427a --- /dev/null +++ b/library/cpp/yt/memory/leaky_ref_counted_singleton-inl.h @@ -0,0 +1,43 @@ +#ifndef LEAKY_REF_COUNTED_SINGLETON_INL_H_ +#error "Direct inclusion of this file is not allowed, include leaky_ref_counted_singleton.h" +// For the sake of sane code completion. +#include "leaky_ref_counted_singleton.h" +#endif + +#include "new.h" + +#include <atomic> +#include <mutex> + +#include <util/system/compiler.h> +#include <util/system/sanitizers.h> + +namespace NYT { + +//////////////////////////////////////////////////////////////////////////////// + +template <class T, class... TArgs> +TIntrusivePtr<T> LeakyRefCountedSingleton(TArgs&&... args) +{ + static std::atomic<T*> Ptr; + auto* ptr = Ptr.load(std::memory_order::acquire); + if (Y_LIKELY(ptr)) { + return ptr; + } + + static std::once_flag Initialized; + std::call_once(Initialized, [&] { + auto ptr = New<T>(std::forward<TArgs>(args)...); + Ref(ptr.Get()); + Ptr.store(ptr.Get()); +#if defined(_asan_enabled_) + NSan::MarkAsIntentionallyLeaked(ptr.Get()); +#endif + }); + + return Ptr.load(); +} + +//////////////////////////////////////////////////////////////////////////////// + +} // namespace NYT diff --git a/library/cpp/yt/memory/leaky_ref_counted_singleton.h b/library/cpp/yt/memory/leaky_ref_counted_singleton.h new file mode 100644 index 00000000000..4ad5b5f0528 --- /dev/null +++ b/library/cpp/yt/memory/leaky_ref_counted_singleton.h @@ -0,0 +1,18 @@ +#pragma once + +#include "intrusive_ptr.h" + +namespace NYT { + +//////////////////////////////////////////////////////////////////////////////// + +template <class T, class... TArgs> +TIntrusivePtr<T> LeakyRefCountedSingleton(TArgs&&... args); + +//////////////////////////////////////////////////////////////////////////////// + +} // namespace NYT + +#define LEAKY_REF_COUNTED_SINGLETON_INL_H_ +#include "leaky_ref_counted_singleton-inl.h" +#undef LEAKY_REF_COUNTED_SINGLETON_INL_H_ diff --git a/library/cpp/yt/misc/property.h b/library/cpp/yt/misc/property.h new file mode 100644 index 00000000000..bef8024ae15 --- /dev/null +++ b/library/cpp/yt/misc/property.h @@ -0,0 +1,289 @@ +#pragma once + +//////////////////////////////////////////////////////////////////////////////// + +//! Declares a trivial public read-write property that is passed by reference. +#define DECLARE_BYREF_RW_PROPERTY(type, name) \ +public: \ + type& name(); \ + const type& name() const; + +//! Defines a trivial public read-write property that is passed by reference. +//! All arguments after name are used as default value (via braced-init-list). +#define DEFINE_BYREF_RW_PROPERTY(type, name, ...) \ +protected: \ + type name##_ { __VA_ARGS__ }; \ + \ +public: \ + Y_FORCE_INLINE type& name() \ + { \ + return name##_; \ + } \ + \ + Y_FORCE_INLINE const type& name() const \ + { \ + return name##_; \ + } + +//! Defines a trivial public read-write property that is passed by reference +//! and is not inline-initialized. +#define DEFINE_BYREF_RW_PROPERTY_NO_INIT(type, name) \ +protected: \ + type name##_; \ + \ +public: \ + Y_FORCE_INLINE type& name() \ + { \ + return name##_; \ + } \ + \ + Y_FORCE_INLINE const type& name() const \ + { \ + return name##_; \ + } + +//! Forwards a trivial public read-write property that is passed by reference. +#define DELEGATE_BYREF_RW_PROPERTY(declaringType, type, name, delegateTo) \ + type& declaringType::name() \ + { \ + return (delegateTo).name(); \ + } \ + \ + const type& declaringType::name() const \ + { \ + return (delegateTo).name(); \ + } + +//////////////////////////////////////////////////////////////////////////////// + +//! Declares a trivial public read-only property that is passed by reference. +#define DECLARE_BYREF_RO_PROPERTY(type, name) \ +public: \ + const type& name() const; + +//! Defines a trivial public read-only property that is passed by reference. +//! All arguments after name are used as default value (via braced-init-list). +#define DEFINE_BYREF_RO_PROPERTY(type, name, ...) \ +protected: \ + type name##_ { __VA_ARGS__ }; \ + \ +public: \ + Y_FORCE_INLINE const type& name() const \ + { \ + return name##_; \ + } + +//! Defines a trivial public read-only property that is passed by reference +//! and is not inline-initialized. +#define DEFINE_BYREF_RO_PROPERTY_NO_INIT(type, name) \ +protected: \ + type name##_; \ + \ +public: \ + Y_FORCE_INLINE const type& name() const \ + { \ + return name##_; \ + } + +//! Forwards a trivial public read-only property that is passed by reference. +#define DELEGATE_BYREF_RO_PROPERTY(declaringType, type, name, delegateTo) \ + const type& declaringType::name() const \ + { \ + return (delegateTo).name(); \ + } + +//////////////////////////////////////////////////////////////////////////////// + +//! Declares a trivial public read-write property that is passed by value. +#define DECLARE_BYVAL_RW_PROPERTY(type, name) \ +public: \ + type Get##name() const; \ + void Set##name(type value); + +//! Defines a trivial public read-write property that is passed by value. +//! All arguments after name are used as default value (via braced-init-list). +#define DEFINE_BYVAL_RW_PROPERTY(type, name, ...) \ +protected: \ + type name##_ { __VA_ARGS__ }; \ + \ +public: \ + Y_FORCE_INLINE type Get##name() const \ + { \ + return name##_; \ + } \ + \ + Y_FORCE_INLINE void Set##name(type value) \ + { \ + name##_ = value; \ + } \ + +//! Defines a trivial public read-write property that is passed by value. +//! All arguments after name are used as default value (via braced-init-list). +#define DEFINE_BYVAL_RW_PROPERTY_WITH_FLUENT_SETTER(declaringType, type, name, ...) \ +protected: \ + type name##_ { __VA_ARGS__ }; \ + \ +public: \ + Y_FORCE_INLINE type Get##name() const \ + { \ + return name##_; \ + } \ + \ + Y_FORCE_INLINE void Set##name(type value) &\ + { \ + name##_ = value; \ + } \ + \ + Y_FORCE_INLINE declaringType&& Set##name(type value) &&\ + { \ + name##_ = value; \ + return std::move(*this); \ + } \ + +//! Defines a trivial public read-write property that is passed by value +//! and is not inline-initialized. +#define DEFINE_BYVAL_RW_PROPERTY_NO_INIT(type, name, ...) \ +protected: \ + type name##_; \ + \ +public: \ + Y_FORCE_INLINE type Get##name() const \ + { \ + return name##_; \ + } \ + \ + Y_FORCE_INLINE void Set##name(type value) \ + { \ + name##_ = value; \ + } \ + +//! Forwards a trivial public read-write property that is passed by value. +#define DELEGATE_BYVAL_RW_PROPERTY(declaringType, type, name, delegateTo) \ + type declaringType::Get##name() const \ + { \ + return (delegateTo).Get##name(); \ + } \ + \ + void declaringType::Set##name(type value) \ + { \ + (delegateTo).Set##name(value); \ + } + +//////////////////////////////////////////////////////////////////////////////// + +//! Declares a trivial public read-only property that is passed by value. +#define DECLARE_BYVAL_RO_PROPERTY(type, name) \ +public: \ + type Get##name() const; + +//! Defines a trivial public read-only property that is passed by value. +//! All arguments after name are used as default value (via braced-init-list). +#define DEFINE_BYVAL_RO_PROPERTY(type, name, ...) \ +protected: \ + type name##_ { __VA_ARGS__ }; \ + \ +public: \ + Y_FORCE_INLINE type Get##name() const \ + { \ + return name##_; \ + } + + +//! Defines a trivial public read-only property that is passed by value +//! and is not inline-initialized. +#define DEFINE_BYVAL_RO_PROPERTY_NO_INIT(type, name) \ +protected: \ + type name##_; \ + \ +public: \ + Y_FORCE_INLINE type Get##name() const \ + { \ + return name##_; \ + } + +//! Forwards a trivial public read-only property that is passed by value. +#define DELEGATE_BYVAL_RO_PROPERTY(declaringType, type, name, delegateTo) \ + type declaringType::Get##name() \ + { \ + return (delegateTo).Get##name(); \ + } + +//////////////////////////////////////////////////////////////////////////////// + +//! Below are macro helpers for extra properties. +//! Extra properties should be used for lazy memory allocation for properties that +//! hold default values for the majority of objects. + +//! Initializes extra property holder if it is not initialized. +#define INITIALIZE_EXTRA_PROPERTY_HOLDER(holder) \ + if (!holder##_) { \ + holder##_.reset(new decltype(holder##_)::element_type()); \ + } + +//! Declares an extra property holder. Holder contains extra properties values. +//! Holder is not created until some property is set with a non-default value. +//! If there is no holder property getter returns default value. +#define DECLARE_EXTRA_PROPERTY_HOLDER(type, holder) \ +public: \ + Y_FORCE_INLINE bool HasCustom##holder() const \ + { \ + return static_cast<bool>(holder##_); \ + } \ + Y_FORCE_INLINE const type* GetCustom##holder() const \ + { \ + return holder##_.get(); \ + } \ + Y_FORCE_INLINE type* GetCustom##holder() \ + { \ + return holder##_.get(); \ + } \ + Y_FORCE_INLINE void InitializeCustom##holder() \ + { \ + INITIALIZE_EXTRA_PROPERTY_HOLDER(holder) \ + } \ +private: \ + std::unique_ptr<type> holder##_; \ + static const type Default##holder##_; + +//! Defines a storage for extra properties default values. +#define DEFINE_EXTRA_PROPERTY_HOLDER(class, type, holder) \ + const type class::Default##holder##_; + +//! Defines a public read-write extra property that is passed by value. +#define DEFINE_BYVAL_RW_EXTRA_PROPERTY(holder, name) \ +public: \ + Y_FORCE_INLINE decltype(holder##_->name) Get##name() const \ + { \ + if (!holder##_) { \ + return Default##holder##_.name; \ + } \ + return holder##_->name; \ + } \ + Y_FORCE_INLINE void Set##name(decltype(holder##_->name) val) \ + { \ + if (!holder##_) { \ + if (val == Default##holder##_.name) { \ + return; \ + } \ + INITIALIZE_EXTRA_PROPERTY_HOLDER(holder); \ + } \ + holder##_->name = val; \ + } + +//! Defines a public read-write extra property that is passed by reference. +#define DEFINE_BYREF_RW_EXTRA_PROPERTY(holder, name) \ +public: \ + Y_FORCE_INLINE const decltype(holder##_->name)& name() const \ + { \ + if (!holder##_) { \ + return Default##holder##_.name; \ + } \ + return holder##_->name; \ + } \ + Y_FORCE_INLINE decltype(holder##_->name)& Mutable##name() \ + { \ + INITIALIZE_EXTRA_PROPERTY_HOLDER(holder); \ + return holder##_->name; \ + } + +//////////////////////////////////////////////////////////////////////////////// diff --git a/library/cpp/yt/mlock/README.md b/library/cpp/yt/mlock/README.md new file mode 100644 index 00000000000..b61b6072c4d --- /dev/null +++ b/library/cpp/yt/mlock/README.md @@ -0,0 +1,11 @@ +# mlock + +MlockFileMappings подгружает и лочит в память все страницы исполняемого файла. + +В отличии от вызова mlockall, функция не лочит другие страницы процесса. +mlockall явно выделяет физическую память под все vma. Типичный процесс сначала +стартует и инициализирует аллокатор, а потом уже вызывает функцию для mlock страниц. +Аллокатор при старте выделяет большие диапазоны через mmap, но реально их не использует. +Поэтому mlockall приводит в повышенному потреблению памяти. + +Также, в отличии от mlockall, функция может подгрузить страницы в память сразу. diff --git a/library/cpp/yt/mlock/mlock.h b/library/cpp/yt/mlock/mlock.h new file mode 100644 index 00000000000..035fc47e373 --- /dev/null +++ b/library/cpp/yt/mlock/mlock.h @@ -0,0 +1,11 @@ +#pragma once + +namespace NYT { + +//////////////////////////////////////////////////////////////////////////////// + +bool MlockFileMappings(bool populate = true); + +//////////////////////////////////////////////////////////////////////////////// + +} // namespace NYT diff --git a/library/cpp/yt/mlock/mlock_linux.cpp b/library/cpp/yt/mlock/mlock_linux.cpp new file mode 100644 index 00000000000..8791869f951 --- /dev/null +++ b/library/cpp/yt/mlock/mlock_linux.cpp @@ -0,0 +1,89 @@ +#include "mlock.h" + +#include <stdio.h> +#include <sys/mman.h> +#include <stdint.h> +#include <inttypes.h> + +namespace NYT { + +//////////////////////////////////////////////////////////////////////////////// + +void PopulateFile(void* ptr, size_t size) +{ + constexpr size_t PageSize = 4096; + + auto* begin = static_cast<volatile char*>(ptr); + for (auto* current = begin; current < begin + size; current += PageSize) { + *current; + } +} + +bool MlockFileMappings(bool populate) +{ + auto* file = ::fopen("/proc/self/maps", "r"); + if (!file) { + return false; + } + + // Each line of /proc/<pid>/smaps has the following format: + // address perms offset dev inode path + // E.g. + // 08048000-08056000 r-xp 00000000 03:0c 64593 /usr/sbin/gpm + + bool failed = false; + while (true) { + char line[1024]; + if (!fgets(line, sizeof(line), file)) { + break; + } + + char addressStr[64]; + char permsStr[64]; + char offsetStr[64]; + char devStr[64]; + int inode; + if (sscanf(line, "%s %s %s %s %d", + addressStr, + permsStr, + offsetStr, + devStr, + &inode) != 5) + { + continue; + } + + if (inode == 0) { + continue; + } + + if (permsStr[0] != 'r') { + continue; + } + + uintptr_t startAddress; + uintptr_t endAddress; + if (sscanf(addressStr, "%" PRIx64 "-%" PRIx64, + &startAddress, + &endAddress) != 2) + { + continue; + } + + if (::mlock(reinterpret_cast<const void*>(startAddress), endAddress - startAddress) != 0) { + failed = true; + continue; + } + + if (populate) { + PopulateFile(reinterpret_cast<void*>(startAddress), endAddress - startAddress); + } + } + + ::fclose(file); + return !failed; +} + +//////////////////////////////////////////////////////////////////////////////// + +} // namespace NYT diff --git a/library/cpp/yt/mlock/mlock_other.cpp b/library/cpp/yt/mlock/mlock_other.cpp new file mode 100644 index 00000000000..269c5c3cb98 --- /dev/null +++ b/library/cpp/yt/mlock/mlock_other.cpp @@ -0,0 +1,14 @@ +#include "mlock.h" + +namespace NYT { + +//////////////////////////////////////////////////////////////////////////////// + +bool MlockFileMappings(bool /* populate */) +{ + return false; +} + +//////////////////////////////////////////////////////////////////////////////// + +} // namespace NYT diff --git a/library/cpp/yt/stockpile/README.md b/library/cpp/yt/stockpile/README.md new file mode 100644 index 00000000000..6ee4cd1b1f1 --- /dev/null +++ b/library/cpp/yt/stockpile/README.md @@ -0,0 +1,12 @@ +# stockpile + +При приближении к лимиту памяти в memory cgroup, linux запускает механизм direct reclaim, +чтобы освободить свободную память. По опыту YT, direct reclaim очень сильно замедляет работу +всего процесса. + +Проблема возникает не только, когда память занята анонимными страницами. 50% памяти контейнера +может быть занято не dirty страницами page cache, но проблема всёравно будет проявляться. Например, +если процесс активно читает файлы с диска без O_DIRECT, вся память очень быстро будет забита. + +Чтобы бороться с этой проблемой, в яндексовом ядре добавлена ручка `madvise(MADV_STOCKPILE)`. +Больше подробностей в https://st.yandex-team.ru/KERNEL-186
\ No newline at end of file diff --git a/library/cpp/yt/stockpile/stockpile.h b/library/cpp/yt/stockpile/stockpile.h new file mode 100644 index 00000000000..2b1b53e9b57 --- /dev/null +++ b/library/cpp/yt/stockpile/stockpile.h @@ -0,0 +1,24 @@ +#pragma once + +#include <util/system/types.h> + +#include <util/generic/size_literals.h> + +#include <util/datetime/base.h> + +namespace NYT { + +//////////////////////////////////////////////////////////////////////////////// + +struct TStockpileOptions +{ + i64 BufferSize = 4_GB; + int ThreadCount = 4; + TDuration Period = TDuration::MilliSeconds(10); +}; + +void StockpileMemory(const TStockpileOptions& options); + +//////////////////////////////////////////////////////////////////////////////// + +} // namespace NYT diff --git a/library/cpp/yt/stockpile/stockpile_linux.cpp b/library/cpp/yt/stockpile/stockpile_linux.cpp new file mode 100644 index 00000000000..9be33e3811e --- /dev/null +++ b/library/cpp/yt/stockpile/stockpile_linux.cpp @@ -0,0 +1,42 @@ +#include "stockpile.h" + +#include <thread> +#include <mutex> + +#include <sys/mman.h> + +#include <util/system/thread.h> + +namespace NYT { + +//////////////////////////////////////////////////////////////////////////////// + +namespace { + +void RunStockpile(const TStockpileOptions& options) +{ + TThread::SetCurrentThreadName("Stockpile"); + + constexpr int MADV_STOCKPILE = 0x59410004; + + while (true) { + ::madvise(nullptr, options.BufferSize, MADV_STOCKPILE); + Sleep(options.Period); + } +} + +} // namespace + +void StockpileMemory(const TStockpileOptions& options) +{ + static std::once_flag OnceFlag; + std::call_once(OnceFlag, [options] { + for (int i = 0; i < options.ThreadCount; i++) { + std::thread(RunStockpile, options).detach(); + } + }); +} + +//////////////////////////////////////////////////////////////////////////////// + +} // namespace NYT diff --git a/library/cpp/yt/stockpile/stockpile_other.cpp b/library/cpp/yt/stockpile/stockpile_other.cpp new file mode 100644 index 00000000000..913ad88b206 --- /dev/null +++ b/library/cpp/yt/stockpile/stockpile_other.cpp @@ -0,0 +1,12 @@ +#include "stockpile.h" + +namespace NYT { + +//////////////////////////////////////////////////////////////////////////////// + +void StockpileMemory(const TStockpileOptions& /*options*/) +{ } + +//////////////////////////////////////////////////////////////////////////////// + +} // namespace NYT diff --git a/library/cpp/yt/threading/unittests/recursive_spin_lock_ut.cpp b/library/cpp/yt/threading/unittests/recursive_spin_lock_ut.cpp new file mode 100644 index 00000000000..9c2d8f16cbf --- /dev/null +++ b/library/cpp/yt/threading/unittests/recursive_spin_lock_ut.cpp @@ -0,0 +1,88 @@ +#include <library/cpp/testing/gtest/gtest.h> + +#include <library/cpp/yt/threading/recursive_spin_lock.h> +#include <library/cpp/yt/threading/event_count.h> + +#include <thread> + +namespace NYT::NThreading { +namespace { + +//////////////////////////////////////////////////////////////////////////////// + +TEST(TRecursiveSpinLockTest, SingleThread) +{ + TRecursiveSpinLock lock; + EXPECT_FALSE(lock.IsLocked()); + EXPECT_TRUE(lock.TryAcquire()); + EXPECT_TRUE(lock.IsLocked()); + EXPECT_TRUE(lock.TryAcquire()); + EXPECT_TRUE(lock.IsLocked()); + lock.Release(); + EXPECT_TRUE(lock.IsLocked()); + lock.Release(); + EXPECT_FALSE(lock.IsLocked()); + EXPECT_TRUE(lock.TryAcquire()); + EXPECT_TRUE(lock.IsLocked()); + lock.Release(); + lock.Acquire(); + lock.Release(); +} + +TEST(TRecursiveSpinLockTest, TwoThreads) +{ + TRecursiveSpinLock lock; + TEvent e1, e2, e3, e4, e5, e6, e7; + + std::thread t1([&] { + e1.Wait(); + EXPECT_TRUE(lock.IsLocked()); + EXPECT_FALSE(lock.IsLockedByCurrentThread()); + EXPECT_FALSE(lock.TryAcquire()); + e2.NotifyOne(); + e3.Wait(); + EXPECT_TRUE(lock.IsLocked()); + EXPECT_FALSE(lock.IsLockedByCurrentThread()); + EXPECT_FALSE(lock.TryAcquire()); + e4.NotifyOne(); + e5.Wait(); + EXPECT_FALSE(lock.IsLocked()); + EXPECT_FALSE(lock.IsLockedByCurrentThread()); + EXPECT_TRUE(lock.TryAcquire()); + e6.NotifyOne(); + e7.Wait(); + lock.Release(); + }); + + std::thread t2([&] { + EXPECT_FALSE(lock.IsLocked()); + EXPECT_TRUE(lock.TryAcquire()); + EXPECT_TRUE(lock.IsLockedByCurrentThread()); + e1.NotifyOne(); + e2.Wait(); + EXPECT_TRUE(lock.TryAcquire()); + EXPECT_TRUE(lock.IsLockedByCurrentThread()); + e3.NotifyOne(); + e4.Wait(); + lock.Release(); + lock.Release(); + EXPECT_FALSE(lock.IsLocked()); + e5.NotifyOne(); + e6.Wait(); + EXPECT_TRUE(lock.IsLocked()); + EXPECT_FALSE(lock.IsLockedByCurrentThread()); + e7.NotifyOne(); + lock.Acquire(); + lock.Acquire(); + lock.Release(); + lock.Release(); + }); + + t1.join(); + t2.join(); +} + +//////////////////////////////////////////////////////////////////////////////// + +} // namespace +} // namespace NYT::NThreading diff --git a/library/cpp/yt/threading/unittests/spin_wait_ut.cpp b/library/cpp/yt/threading/unittests/spin_wait_ut.cpp new file mode 100644 index 00000000000..8469634f346 --- /dev/null +++ b/library/cpp/yt/threading/unittests/spin_wait_ut.cpp @@ -0,0 +1,48 @@ +#include <library/cpp/testing/gtest/gtest.h> + +#include <library/cpp/yt/threading/spin_wait.h> +#include <library/cpp/yt/threading/spin_wait_hook.h> + +#include <thread> +#include <mutex> + +namespace NYT::NThreading { +namespace { + +//////////////////////////////////////////////////////////////////////////////// + +bool SpinWaitSlowPathHookInvoked; + +void SpinWaitSlowPathHook( + TCpuDuration cpuDelay, + const TSourceLocation& /*location*/, + ESpinLockActivityKind /*activityKind*/) +{ + SpinWaitSlowPathHookInvoked = true; + auto delay = CpuDurationToDuration(cpuDelay); + EXPECT_GE(delay, TDuration::Seconds(1)); + EXPECT_LE(delay, TDuration::Seconds(5)); +} + +TEST(TSpinWaitTest, SlowPathHook) +{ + static std::once_flag registerFlag; + std::call_once( + registerFlag, + [] { + RegisterSpinWaitSlowPathHook(SpinWaitSlowPathHook); + }); + SpinWaitSlowPathHookInvoked = false; + { + TSpinWait spinWait(__LOCATION__, ESpinLockActivityKind::ReadWrite); + for (int i = 0; i < 1'000'000; ++i) { + spinWait.Wait(); + } + } + EXPECT_TRUE(SpinWaitSlowPathHookInvoked); +} + +//////////////////////////////////////////////////////////////////////////////// + +} // namespace +} // namespace NYT::NThreading |