diff options
Diffstat (limited to 'library/cpp')
64 files changed, 2972 insertions, 358 deletions
diff --git a/library/cpp/containers/paged_vector/README.md b/library/cpp/containers/paged_vector/README.md index a3d881acb86..1f67a0d1f7a 100644 --- a/library/cpp/containers/paged_vector/README.md +++ b/library/cpp/containers/paged_vector/README.md @@ -46,6 +46,7 @@ Iterators are random-access and are implemented as an *(owner pointer, offset)* - Iterators are not invalidated by `push_back`/`emplace_back` (an `end()` iterator taken earlier keeps pointing to the same logical position). - Dereferencing goes through the vector, so an iterator is only valid while its source container is alive. +- To get the current index of an element from an iterator, call `it.GetIndex()` — it returns the offset of the pointed-to element within the container (equivalent to `it - begin()`). ## Complexity diff --git a/library/cpp/containers/paged_vector/paged_vector.h b/library/cpp/containers/paged_vector/paged_vector.h index a02c7ce1d18..5bb8d8eb521 100644 --- a/library/cpp/containers/paged_vector/paged_vector.h +++ b/library/cpp/containers/paged_vector/paged_vector.h @@ -18,7 +18,7 @@ namespace NPagedVector { friend class TPagedVector<TT, PageSize>; using TVec = TPagedVector<TT, PageSize>; using TSelf = TPagedVectorIterator<T, TT, PageSize>; - size_t Offset_; + size_t Index_; TVec* Vector_; template <class T1, class TT1, ui32 PageSize1> @@ -26,26 +26,26 @@ namespace NPagedVector { public: TPagedVectorIterator() - : Offset_() + : Index_() , Vector_() { } - TPagedVectorIterator(TVec* vector, size_t offset) - : Offset_(offset) + TPagedVectorIterator(TVec* vector, size_t index) + : Index_(index) , Vector_(vector) { } template <class T1, class TT1, ui32 PageSize1> TPagedVectorIterator(const TPagedVectorIterator<T1, TT1, PageSize1>& it) - : Offset_(it.Offset_) + : Index_(it.Index_) , Vector_(it.Vector_) { } T& operator*() const { - return (*Vector_)[Offset_]; + return (*Vector_)[Index_]; } T* operator->() const { @@ -54,7 +54,7 @@ namespace NPagedVector { template <class T1, class TT1, ui32 PageSize1> bool operator==(const TPagedVectorIterator<T1, TT1, PageSize1>& it) const { - return Offset_ == it.Offset_; + return Index_ == it.Index_; } template <class T1, class TT1, ui32 PageSize1> @@ -64,12 +64,12 @@ namespace NPagedVector { template <class T1, class TT1, ui32 PageSize1> bool operator<(const TPagedVectorIterator<T1, TT1, PageSize1>& it) const { - return Offset_ < it.Offset_; + return Index_ < it.Index_; } template <class T1, class TT1, ui32 PageSize1> bool operator<=(const TPagedVectorIterator<T1, TT1, PageSize1>& it) const { - return Offset_ <= it.Offset_; + return Index_ <= it.Index_; } template <class T1, class TT1, ui32 PageSize1> @@ -84,11 +84,11 @@ namespace NPagedVector { template <class T1, class TT1, ui32 PageSize1> ptrdiff_t operator-(const TPagedVectorIterator<T1, TT1, PageSize1>& it) const { - return Offset_ - it.Offset_; + return Index_ - it.Index_; } TSelf& operator+=(ptrdiff_t off) { - Offset_ += off; + Index_ += off; return *this; } @@ -126,8 +126,8 @@ namespace NPagedVector { return this->operator+(-off); } - size_t GetOffset() const { - return Offset_; + [[nodiscard]] size_t GetIndex() const { + return Index_; } }; } // namespace NPrivate @@ -362,8 +362,8 @@ namespace NPagedVector { CurrentPageSize_ = Pages_.empty() ? 0 : PageSize; } - size_t pidx = InPageIndex(it.Offset_); - for (size_t pnum = PageNumber(it.Offset_);; ++pnum) { + size_t pidx = InPageIndex(it.Index_); + for (size_t pnum = PageNumber(it.Index_);; ++pnum) { TPage& page = *Pages_[pnum]; if (pnum + 1 == Pages_.size()) { std::shift_left(page.data() + pidx, page.data() + CurrentPageSize_, 1); diff --git a/library/cpp/containers/paged_vector/ut/paged_vector_ut.cpp b/library/cpp/containers/paged_vector/ut/paged_vector_ut.cpp index 7205fb06064..4863227e7f1 100644 --- a/library/cpp/containers/paged_vector/ut/paged_vector_ut.cpp +++ b/library/cpp/containers/paged_vector/ut/paged_vector_ut.cpp @@ -28,6 +28,7 @@ class TPagedVectorTest: public TTestBase { UNIT_TEST(TestEmplaceBackNoncopyable) UNIT_TEST(TestClear) UNIT_TEST(TestBack) + UNIT_TEST(TestIterator) UNIT_TEST_SUITE_END(); private: @@ -613,6 +614,37 @@ private: v.pop_back(); UNIT_ASSERT_VALUES_EQUAL(v.back(), "2"); } + + void TestIterator() { + using NPagedVector::TPagedVector; + TPagedVector<TString, 5> v; + for (int i = 0; i < 11; ++i) { + v.push_back(ToString(i)); + } + + v.emplace_back("Hello"); + v.emplace_back("world"); + + auto it = v.begin(); + + UNIT_ASSERT_VALUES_EQUAL(it.GetIndex(), 0); + UNIT_ASSERT_VALUES_EQUAL(*it, "0"); + + ++it; + + UNIT_ASSERT_VALUES_EQUAL(it.GetIndex(), 1); + UNIT_ASSERT_VALUES_EQUAL(*it, "1"); + + it += 5; + + UNIT_ASSERT_VALUES_EQUAL(it.GetIndex(), 6); + UNIT_ASSERT_VALUES_EQUAL(*it, "6"); + + it = v.erase(it); + + UNIT_ASSERT_VALUES_EQUAL(it.GetIndex(), 6); + UNIT_ASSERT_VALUES_EQUAL(*it, "7"); + } }; UNIT_TEST_SUITE_REGISTRATION(TPagedVectorTest); diff --git a/library/cpp/dot_product/dot_product.cpp b/library/cpp/dot_product/dot_product.cpp index 003dbe88335..d3f2b31b2d4 100644 --- a/library/cpp/dot_product/dot_product.cpp +++ b/library/cpp/dot_product/dot_product.cpp @@ -1,6 +1,7 @@ #include "dot_product.h" #include "dot_product_sse.h" #include "dot_product_avx2.h" +#include "dot_product_vnni.h" #include "dot_product_simple.h" #include <library/cpp/sse/sse.h> @@ -31,7 +32,6 @@ namespace NDotProductImpl { namespace { [[maybe_unused]] const int _ = [] { if (!FromYaTest() && GetEnv("Y_NO_AVX_IN_DOT_PRODUCT") == "" && NX86::HaveAVX2() && NX86::HaveFMA()) { - DotProductI8Impl = &DotProductAvx2; DotProductUi8Impl = &DotProductAvx2; DotProductI32Impl = &DotProductAvx2; DotProductFloatImpl = &DotProductAvx2; @@ -40,6 +40,12 @@ namespace NDotProductImpl { TriWayDotProductImpl = &TriWayDotProductAvx2; TriWayDotProductFloatI8Impl = &TriWayDotProductFloatI8Avx2; TriWayDotProductI8Impl = &TriWayDotProductI8Avx2; + + if (GetEnv("Y_NO_VNNI_IN_DOT_PRODUCT") == "" && NX86::HaveAVX512VNNI() && NX86::HaveAVX512BW()) { + DotProductI8Impl = &DotProductVnni; + } else { + DotProductI8Impl = &DotProductAvx2; + } } else { #ifdef ARCADIA_SSE DotProductI8Impl = &DotProductSse; diff --git a/library/cpp/dot_product/dot_product_ut.cpp b/library/cpp/dot_product/dot_product_ut.cpp index 6d07d1a3a31..8ba451f4964 100644 --- a/library/cpp/dot_product/dot_product_ut.cpp +++ b/library/cpp/dot_product/dot_product_ut.cpp @@ -1,6 +1,7 @@ #include "dot_product.h" #include "dot_product_simple.h" #include "dot_product_avx2.h" +#include "dot_product_vnni.h" #include <library/cpp/testing/unittest/registar.h> @@ -48,16 +49,46 @@ Y_UNIT_TEST_SUITE(TDocProductTestSuite) { return sum; } + bool HaveVnniDotProduct() { + return NX86::HaveAVX2() && NX86::HaveFMA() && NX86::HaveAVX512BW() && NX86::HaveAVX512VNNI(); + } + Y_UNIT_TEST(TestDotProduct8) { TVector<i8> a(100); FillWithRandomNumbers(a.data(), 179, 100); TVector<i8> b(100); FillWithRandomNumbers(b.data(), 239, 100); + const bool haveVnni = HaveVnniDotProduct(); for (size_t i = 0; i < 30; ++i) { for (size_t length = 1; length + i + 1 < a.size(); ++length) { - UNIT_ASSERT_EQUAL(DotProduct(a.data() + i, b.data() + i, length), (SimpleDotProduct<i32, i8>(a.data() + i, b.data() + i, length))); - UNIT_ASSERT_EQUAL(DotProductSimple(a.data() + i, b.data() + i, length), (SimpleDotProduct<i32, i8>(a.data() + i, b.data() + i, length))); + const i32 expected = SimpleDotProduct<i32, i8>(a.data() + i, b.data() + i, length); + UNIT_ASSERT_EQUAL(DotProduct(a.data() + i, b.data() + i, length), expected); + UNIT_ASSERT_EQUAL(DotProductSimple(a.data() + i, b.data() + i, length), expected); + if (haveVnni) { + UNIT_ASSERT_EQUAL(DotProductVnni(a.data() + i, b.data() + i, length), expected); + } + } + } + } + + Y_UNIT_TEST(TestDotProduct8VnniEdges) { + if (!HaveVnniDotProduct()) { + return; + } + + TVector<i8> a(512); + TVector<i8> b(512); + for (size_t i = 0; i < a.size(); ++i) { + a[i] = static_cast<i8>((i * 37) % 256 - 128); + b[i] = static_cast<i8>((i * 53 + 17) % 256 - 128); + } + + for (size_t offset = 0; offset < 16; ++offset) { + for (size_t length = 0; length + offset <= a.size(); ++length) { + UNIT_ASSERT_EQUAL( + DotProductVnni(a.data() + offset, b.data() + offset, length), + (SimpleDotProduct<i32, i8>(a.data() + offset, b.data() + offset, length))); } } } diff --git a/library/cpp/dot_product/dot_product_vnni.cpp b/library/cpp/dot_product/dot_product_vnni.cpp new file mode 100644 index 00000000000..4b3c52def49 --- /dev/null +++ b/library/cpp/dot_product/dot_product_vnni.cpp @@ -0,0 +1,125 @@ +#include "dot_product_vnni.h" +#include "dot_product_avx2.h" + +#include <util/system/compiler.h> +#include <util/system/platform.h> + +#if defined(__AVX512VNNI__) && defined(__AVX512BW__) && defined(__AVX512F__) + +#include <immintrin.h> + +namespace { + Y_FORCE_INLINE __m512i Load512i(const void* ptr) { + return _mm512_loadu_si512(ptr); + } + + Y_FORCE_INLINE __m512i CorrectSignedLhsDotProduct( + const __m512i biasedDotProduct, + const __m512i rhsSum) noexcept + { + return _mm512_sub_epi32(biasedDotProduct, _mm512_slli_epi32(rhsSum, 7)); + } + + Y_FORCE_INLINE void AccumulateDotProductI8Vnni( + __m512i& dotProduct, + const i8* lhs, + const i8* rhs, + const __m512i zero, + const __m512i signBit, + const __m512i ones) noexcept + { + const __m512i l = Load512i(lhs); + const __m512i r = Load512i(rhs); + const __m512i unsignedL = _mm512_xor_si512(l, signBit); + + // AVX512 VNNI has u8*i8 dot product. For signed lhs we compute + // (lhs + 128) * rhs and immediately subtract 128 * sum(rhs). + // Correcting each 64-byte chunk keeps the accumulator in the same + // range as a regular signed i8 dot product accumulator instead of + // accumulating a much larger biased intermediate value. + const __m512i biasedDotProduct = _mm512_dpbusd_epi32(zero, unsignedL, r); + const __m512i rhsSum = _mm512_dpbusd_epi32(zero, ones, r); + dotProduct = _mm512_add_epi32(dotProduct, CorrectSignedLhsDotProduct(biasedDotProduct, rhsSum)); + } + + Y_FORCE_INLINE void AccumulateTailDotProductI8Vnni( + __m512i& dotProduct, + const i8* lhs, + const i8* rhs, + const size_t length, + const __m512i zero, + const __m512i signBit, + const __m512i ones) noexcept + { + const __mmask64 mask = (ui64(1) << length) - 1; + const __m512i l = _mm512_maskz_loadu_epi8(mask, lhs); + const __m512i r = _mm512_maskz_loadu_epi8(mask, rhs); + const __m512i unsignedL = _mm512_xor_si512(l, signBit); + + const __m512i biasedDotProduct = _mm512_dpbusd_epi32(zero, unsignedL, r); + const __m512i rhsSum = _mm512_dpbusd_epi32(zero, ones, r); + dotProduct = _mm512_add_epi32(dotProduct, CorrectSignedLhsDotProduct(biasedDotProduct, rhsSum)); + } + + i32 HsumI32(const __m512i v) noexcept { + __m128i sum = _mm512_castsi512_si128(v); + sum = _mm_add_epi32(sum, _mm512_extracti32x4_epi32(v, 1)); + sum = _mm_add_epi32(sum, _mm512_extracti32x4_epi32(v, 2)); + sum = _mm_add_epi32(sum, _mm512_extracti32x4_epi32(v, 3)); + const __m128i hi64 = _mm_unpackhi_epi64(sum, sum); + const __m128i sum64 = _mm_add_epi32(hi64, sum); + const __m128i hi32 = _mm_shuffle_epi32(sum64, _MM_SHUFFLE(2, 3, 0, 1)); + const __m128i sum32 = _mm_add_epi32(sum64, hi32); + return _mm_cvtsi128_si32(sum32); + } +} + +i32 DotProductVnni(const i8* lhs, const i8* rhs, size_t length) noexcept { + if (length < 64) { + return DotProductAvx2(lhs, rhs, length); + } + + const __m512i zero = _mm512_setzero_si512(); + const __m512i signBit = _mm512_set1_epi8(-128); + const __m512i ones = _mm512_set1_epi8(1); + + __m512i dotProduct0 = zero; + __m512i dotProduct1 = zero; + __m512i dotProduct2 = zero; + __m512i dotProduct3 = zero; + + while (length >= 256) { + AccumulateDotProductI8Vnni(dotProduct0, lhs, rhs, zero, signBit, ones); + AccumulateDotProductI8Vnni(dotProduct1, lhs + 64, rhs + 64, zero, signBit, ones); + AccumulateDotProductI8Vnni(dotProduct2, lhs + 128, rhs + 128, zero, signBit, ones); + AccumulateDotProductI8Vnni(dotProduct3, lhs + 192, rhs + 192, zero, signBit, ones); + lhs += 256; + rhs += 256; + length -= 256; + } + + while (length >= 64) { + AccumulateDotProductI8Vnni(dotProduct0, lhs, rhs, zero, signBit, ones); + lhs += 64; + rhs += 64; + length -= 64; + } + + if (length > 0) { + AccumulateTailDotProductI8Vnni(dotProduct0, lhs, rhs, length, zero, signBit, ones); + } + + const __m512i dotProduct = _mm512_add_epi32( + _mm512_add_epi32(dotProduct0, dotProduct1), + _mm512_add_epi32(dotProduct2, dotProduct3)); + + return HsumI32(dotProduct); +} + +#else + +i32 DotProductVnni(const i8* lhs, const i8* rhs, size_t length) noexcept { + return DotProductAvx2(lhs, rhs, length); +} + +#endif diff --git a/library/cpp/dot_product/dot_product_vnni.h b/library/cpp/dot_product/dot_product_vnni.h new file mode 100644 index 00000000000..7949c28cd37 --- /dev/null +++ b/library/cpp/dot_product/dot_product_vnni.h @@ -0,0 +1,9 @@ +#pragma once + +#include <util/system/types.h> +#include <util/system/compiler.h> + +#include <stddef.h> + +Y_PURE_FUNCTION +i32 DotProductVnni(const i8* lhs, const i8* rhs, size_t length) noexcept; diff --git a/library/cpp/dot_product/ya.make b/library/cpp/dot_product/ya.make index fea9ba8152c..64d78d7bbdc 100644 --- a/library/cpp/dot_product/ya.make +++ b/library/cpp/dot_product/ya.make @@ -8,8 +8,10 @@ SRCS( IF (USE_SSE4 == "yes" AND OS_LINUX == "yes") SRC_C_AVX2(dot_product_avx2.cpp -mfma) + SRC_C_AVX512(dot_product_vnni.cpp -mavx512vnni) ELSE() SRC(dot_product_avx2.cpp) + SRC(dot_product_vnni.cpp) ENDIF() PEERDIR( diff --git a/library/cpp/getopt/small/last_getopt_opts.h b/library/cpp/getopt/small/last_getopt_opts.h index fd0ef978c1f..868477ec79e 100644 --- a/library/cpp/getopt/small/last_getopt_opts.h +++ b/library/cpp/getopt/small/last_getopt_opts.h @@ -463,6 +463,24 @@ namespace NLastGetopt { } /** + * Add section with examples. + * + * @param examples text of this section + */ + void SetExamples(std::string_view examples) { + SetExamples(TString(examples)); + } + + /** + * Add section with examples. + * + * @param examples text of this section + */ + void SetExamples(const char* examples) { + SetExamples(TString(examples)); + } + + /** * Set minimal number of free args * * @param min new value diff --git a/library/cpp/http/misc/parsed_request.cpp b/library/cpp/http/misc/parsed_request.cpp index e332a24e917..5ce6d3ddc6c 100644 --- a/library/cpp/http/misc/parsed_request.cpp +++ b/library/cpp/http/misc/parsed_request.cpp @@ -4,16 +4,16 @@ #include <util/generic/yexception.h> #include <util/string/cast.h> -static inline TStringBuf StripLeft(const TStringBuf& s) noexcept { - const char* b = s.begin(); - const char* e = s.end(); +static TStringBuf StripLeft(const TStringBuf s Y_LIFETIME_BOUND) noexcept { + auto b = s.begin(); + auto e = s.end(); StripRangeBegin(b, e); return TStringBuf(b, e); } -TParsedHttpRequest::TParsedHttpRequest(const TStringBuf& str) { +TParsedHttpRequest::TParsedHttpRequest(const TStringBuf str Y_LIFETIME_BOUND) { TStringBuf tmp; if (!StripLeft(str).TrySplit(' ', Method, tmp)) { @@ -27,6 +27,6 @@ TParsedHttpRequest::TParsedHttpRequest(const TStringBuf& str) { Proto = StripLeft(Proto); } -TParsedHttpLocation::TParsedHttpLocation(const TStringBuf& req) { +TParsedHttpLocation::TParsedHttpLocation(const TStringBuf req Y_LIFETIME_BOUND) { req.Split('?', Path, Cgi); } diff --git a/library/cpp/http/misc/parsed_request.h b/library/cpp/http/misc/parsed_request.h index d4df7054951..27e8c37c0ee 100644 --- a/library/cpp/http/misc/parsed_request.h +++ b/library/cpp/http/misc/parsed_request.h @@ -3,7 +3,7 @@ #include <util/generic/strbuf.h> struct TParsedHttpRequest { - TParsedHttpRequest(const TStringBuf& str); + TParsedHttpRequest(const TStringBuf str Y_LIFETIME_BOUND); TStringBuf Method; TStringBuf Request; @@ -11,14 +11,14 @@ struct TParsedHttpRequest { }; struct TParsedHttpLocation { - TParsedHttpLocation(const TStringBuf& req); + TParsedHttpLocation(const TStringBuf req Y_LIFETIME_BOUND); TStringBuf Path; TStringBuf Cgi; }; struct TParsedHttpFull: public TParsedHttpRequest, public TParsedHttpLocation { - inline TParsedHttpFull(const TStringBuf& line) + TParsedHttpFull(const TStringBuf line Y_LIFETIME_BOUND) : TParsedHttpRequest(line) , TParsedHttpLocation(Request) { diff --git a/library/cpp/scheme/tests/ut/scheme_ut.cpp b/library/cpp/scheme/tests/ut/scheme_ut.cpp index 97cd4524354..42a7bc7d718 100644 --- a/library/cpp/scheme/tests/ut/scheme_ut.cpp +++ b/library/cpp/scheme/tests/ut/scheme_ut.cpp @@ -838,10 +838,10 @@ Y_UNIT_TEST_SUITE(TSchemeTest) { { NSc::TValue va = NSc::TValue::FromJson("{\"x\":\"ab\",\"y\":{\"p\":\"cd\",\"q\":\"ef\"}}"); const NSc::TValue& vb = va.Get("y"); - va = vb; TString sa = "{\"p\":\"cd\",\"q\":\"ef\"}"; - UNIT_ASSERT_VALUES_EQUAL(va.ToJson(), sa); UNIT_ASSERT_VALUES_EQUAL(vb.ToJson(), sa); + va = vb; + UNIT_ASSERT_VALUES_EQUAL(va.ToJson(), sa); } } diff --git a/library/cpp/unified_agent_client/client_impl.cpp b/library/cpp/unified_agent_client/client_impl.cpp index ffe16d8da37..dc4241344ff 100644 --- a/library/cpp/unified_agent_client/client_impl.cpp +++ b/library/cpp/unified_agent_client/client_impl.cpp @@ -8,6 +8,7 @@ #include <contrib/libs/grpc/src/core/lib/iomgr/executor.h> #include <util/charset/utf8.h> +#include <util/generic/scope.h> #include <util/generic/size_literals.h> #include <util/system/env.h> @@ -513,14 +514,22 @@ namespace NUnifiedAgent::NPrivate { } void TClientSession::CheckGrpcCallInactivity() { - with_lock(Lock) { - if (Closed || !Started) { - return; - } - const auto timeout = Client->GetParameters().GrpcCallInactivityTimeout; - if (timeout == TDuration::Zero()) { - return; + if (!Lock.TryAcquire()) { + TSpinWait sw; + + while (Lock.IsLocked() || !Lock.TryAcquire()) { + if (ForkInProgress.load()) { + return; + } + sw.Sleep(); } + } + Y_DEFER { + Lock.Release(); + }; + + const auto timeout = Client->GetParameters().GrpcCallInactivityTimeout; + if (!Closed && Started && timeout != TDuration::Zero()) { if (NegotiatedProtocol.Defined() && *NegotiatedProtocol > 0 && ActiveGrpcCall && !CloseStarted && diff --git a/library/cpp/yt/coding/bit_io-inl.h b/library/cpp/yt/coding/bit_io-inl.h new file mode 100644 index 00000000000..89d69be6251 --- /dev/null +++ b/library/cpp/yt/coding/bit_io-inl.h @@ -0,0 +1,69 @@ +#ifndef BIT_IO_INL_H_ +#error "Direct inclusion of this file is not allowed, include bit_io.h" +// For the sake of sane code completion. +#include "bit_io.h" +#endif + +#include <util/system/compiler.h> +#include <util/system/unaligned_mem.h> + +namespace NYT { + +//////////////////////////////////////////////////////////////////////////////// + +inline TBitWriter::TBitWriter(char* ptr) + : Ptr_(ptr) +{ } + +Y_FORCE_INLINE void TBitWriter::WriteBits(ui32 value, int width) +{ + // Flush a whole 32-bit word at once (a single unaligned store) instead of + // looping over individual bytes with a data-dependent trip count. + Accumulator_ = (Accumulator_ << width) | value; + BitCount_ += width; + if (BitCount_ >= 32) { + BitCount_ -= 32; + ui32 word = __builtin_bswap32(static_cast<ui32>(Accumulator_ >> BitCount_)); + WriteUnaligned<ui32>(Ptr_, word); + Ptr_ += sizeof(word); + } +} + +inline char* TBitWriter::Finish() +{ + while (BitCount_ >= 8) { + BitCount_ -= 8; + *Ptr_++ = static_cast<char>((Accumulator_ >> BitCount_) & 0xff); + } + if (BitCount_ > 0) { + *Ptr_++ = static_cast<char>((Accumulator_ << (8 - BitCount_)) & 0xff); + BitCount_ = 0; + } + Accumulator_ = 0; + return Ptr_; +} + +//////////////////////////////////////////////////////////////////////////////// + +inline TBitReader::TBitReader(const char* ptr) + : Ptr_(reinterpret_cast<const ui8*>(ptr)) +{ } + +Y_FORCE_INLINE ui32 TBitReader::ReadBits(int width) +{ + while (BitCount_ < width) { + Accumulator_ = (Accumulator_ << 8) | *Ptr_++; + BitCount_ += 8; + } + BitCount_ -= width; + return static_cast<ui32>((Accumulator_ >> BitCount_) & ((1ull << width) - 1)); +} + +inline const char* TBitReader::Finish() +{ + return reinterpret_cast<const char*>(Ptr_); +} + +//////////////////////////////////////////////////////////////////////////////// + +} // namespace NYT diff --git a/library/cpp/yt/coding/bit_io.h b/library/cpp/yt/coding/bit_io.h new file mode 100644 index 00000000000..18900d69f85 --- /dev/null +++ b/library/cpp/yt/coding/bit_io.h @@ -0,0 +1,63 @@ +#pragma once + +#include <util/system/types.h> + +namespace NYT { + +//////////////////////////////////////////////////////////////////////////////// + +//! MSB-first bit writer over a caller-owned buffer. +/*! + * Bits are flushed to the buffer in 4-byte chunks, so the buffer must have room + * for up to 3 bytes beyond the last logically-written byte before #Finish is + * called. + */ +class TBitWriter +{ +public: + explicit TBitWriter(char* ptr); + + //! Appends the #width low bits of #value. Requires 0 <= #width <= 32 and + //! #value < 2^#width (for #width == 32 any value is accepted). + void WriteBits(ui32 value, int width); + + //! Pads the last partial byte with zero low bits and returns the + //! one-past-end pointer. + char* Finish(); + +private: + char* Ptr_; + ui64 Accumulator_ = 0; + int BitCount_ = 0; +}; + +//! MSB-first bit reader. +/*! + * Reads bits written by #TBitWriter. Assumes up to 8 bytes past the logical end + * of the stream are safe to read. + */ +class TBitReader +{ +public: + explicit TBitReader(const char* ptr); + + //! Reads and returns #width bits (0 <= #width <= 32). + ui32 ReadBits(int width); + + //! Returns the one-past-end pointer; the (< 8) buffered sub-byte bits, which + //! are the writer's zero padding, are dropped. + const char* Finish(); + +private: + const ui8* Ptr_; + ui64 Accumulator_ = 0; + int BitCount_ = 0; +}; + +//////////////////////////////////////////////////////////////////////////////// + +} // namespace NYT + +#define BIT_IO_INL_H_ +#include "bit_io-inl.h" +#undef BIT_IO_INL_H_ diff --git a/library/cpp/yt/coding/interpolative-inl.h b/library/cpp/yt/coding/interpolative-inl.h new file mode 100644 index 00000000000..5eadfdd483e --- /dev/null +++ b/library/cpp/yt/coding/interpolative-inl.h @@ -0,0 +1,182 @@ +#ifndef INTERPOLATIVE_INL_H_ +#error "Direct inclusion of this file is not allowed, include interpolative.h" +// For the sake of sane code completion. +#include "interpolative.h" +#endif + +#include "bit_io.h" + +#include <library/cpp/yt/assert/assert.h> + +#include <library/cpp/yt/memory/range.h> + +#include <util/system/compiler.h> + +#include <array> +#include <bit> +#include <concepts> + +namespace NYT { + +//////////////////////////////////////////////////////////////////////////////// + +namespace NInterpolativeCodingDetail { + +// Truncated-binary (minimal) code for a value in [0, rangeSize): the first +// Cutoff values take LowWidth bits, the rest one more. +struct TTruncatedBinaryParams +{ + int LowWidth; // floor(log2(rangeSize)) + ui32 Cutoff; // 2^(LowWidth + 1) - rangeSize +}; + +Y_FORCE_INLINE TTruncatedBinaryParams GetTruncatedBinaryParams(ui32 rangeSize) +{ + int lowWidth = std::bit_width(rangeSize) - 1; + ui32 cutoff = (2u << lowWidth) - rangeSize; + return {lowWidth, cutoff}; +} + +// Writes #value in [0, #rangeSize) with the entropy-optimal integer code for a +// uniform value. rangeSize == 1 yields lowWidth 0 / cutoff 1 and emits a +// zero-width code (a no-op), so the singleton case needs no branch. +Y_FORCE_INLINE void WriteTruncatedBinary(TBitWriter* writer, ui32 value, ui32 rangeSize) +{ + auto [lowWidth, cutoff] = GetTruncatedBinaryParams(rangeSize); + // Branchless: the (value >= cutoff) predicate would mispredict on nearly + // every element, so fold it into the emitted codeword and width instead. + ui32 isLong = value >= cutoff ? 1 : 0; + writer->WriteBits(value + (cutoff & (0u - isLong)), lowWidth + static_cast<int>(isLong)); +} + +Y_FORCE_INLINE ui32 ReadTruncatedBinary(TBitReader* reader, ui32 rangeSize) +{ + auto [lowWidth, cutoff] = GetTruncatedBinaryParams(rangeSize); + ui32 high = reader->ReadBits(lowWidth); + if (high < cutoff) { + return high; + } + ui32 low = reader->ReadBits(1); + return ((high << 1) | low) - cutoff; +} + +// A subrange [BeginIndex, EndIndex) of the value array together with the +// inclusive value bounds [Lo, Hi] that its elements must fall in. The median +// values[m] is the only array element touched per node, which matters because +// the traversal walks a large sequence in a cache-unfriendly tree order. +struct TInterpolativeFrame +{ + int BeginIndex; + int EndIndex; + ui32 Lo; + ui32 Hi; +}; + +} // namespace NInterpolativeCodingDetail + +//////////////////////////////////////////////////////////////////////////////// + +inline size_t GetInterpolativeMaxByteSize(int count, ui32 lo, ui32 hi) +{ + // Each element is coded in at most ceil(log2(hi - lo + 1)) bits; the trailing + // word covers the writer's 4-byte flush store. + int maxBitWidth = std::bit_width(hi - lo); + return (static_cast<size_t>(count) * maxBitWidth + 7) / 8 + sizeof(ui32); +} + +//////////////////////////////////////////////////////////////////////////////// + +// Both traversals descend the left child in place and stack only the (non-empty) +// right child, so the stack sees ~count/2 pushes instead of ~count. Left-first +// order matches between encoder and decoder. +template <std::unsigned_integral T> +void InterpolativeEncode(TBitWriter* writer, TRange<T> values, ui32 lo, ui32 hi) +{ + using namespace NInterpolativeCodingDetail; + + int count = std::ssize(values); + if (count == 0) { + return; + } + + // Right children stack up along the leftmost path => depth <= ceil(log2(count)). + std::array<TInterpolativeFrame, 48> stack; + int top = 0; + int beginIndex = 0; + int endIndex = count; + ui32 l = lo; + ui32 h = hi; + for (;;) { + // beginIndex is invariant while descending left; only endIndex/l/h change. + while (beginIndex < endIndex) { + int half = (endIndex - beginIndex) / 2; + int m = beginIndex + half; + // The next descent step reads values[beginIndex + half/2]; prefetch it + // to hide the cache-scattered tree walk on large sequences. + Y_PREFETCH_READ(values.data() + beginIndex + (half >> 1), 0); + ui32 value = static_cast<ui32>(values[m]); + // rangeSize = (upperBound - lowerBound + 1) simplifies to this; + // lowerBound = l + half. + ui32 rangeSize = (h - l) - static_cast<ui32>(endIndex - beginIndex) + 2; + WriteTruncatedBinary(writer, value - l - static_cast<ui32>(half), rangeSize); + if (m + 1 < endIndex) { + YT_ASSERT(top < std::ssize(stack)); + stack[top++] = {m + 1, endIndex, value + 1, h}; + } + endIndex = m; + h = value - 1; + } + if (top == 0) { + break; + } + auto f = stack[--top]; + beginIndex = f.BeginIndex; + endIndex = f.EndIndex; + l = f.Lo; + h = f.Hi; + } +} + +template <std::unsigned_integral T> +void InterpolativeDecode(TBitReader* reader, TMutableRange<T> values, ui32 lo, ui32 hi) +{ + using namespace NInterpolativeCodingDetail; + + int count = std::ssize(values); + if (count == 0) { + return; + } + + std::array<TInterpolativeFrame, 48> stack; + int top = 0; + int beginIndex = 0; + int endIndex = count; + ui32 l = lo; + ui32 h = hi; + for (;;) { + while (beginIndex < endIndex) { + int half = (endIndex - beginIndex) / 2; + int m = beginIndex + half; + ui32 rangeSize = (h - l) - static_cast<ui32>(endIndex - beginIndex) + 2; + ui32 value = l + static_cast<ui32>(half) + ReadTruncatedBinary(reader, rangeSize); + values[m] = static_cast<T>(value); + if (m + 1 < endIndex) { + stack[top++] = {m + 1, endIndex, value + 1, h}; + } + endIndex = m; + h = value - 1; + } + if (top == 0) { + break; + } + auto f = stack[--top]; + beginIndex = f.BeginIndex; + endIndex = f.EndIndex; + l = f.Lo; + h = f.Hi; + } +} + +//////////////////////////////////////////////////////////////////////////////// + +} // namespace NYT diff --git a/library/cpp/yt/coding/interpolative.h b/library/cpp/yt/coding/interpolative.h new file mode 100644 index 00000000000..88e415e754c --- /dev/null +++ b/library/cpp/yt/coding/interpolative.h @@ -0,0 +1,52 @@ +#pragma once + +#include "bit_io.h" + +#include <library/cpp/yt/memory/range.h> + +#include <util/system/types.h> + +#include <concepts> + +namespace NYT { + +//////////////////////////////////////////////////////////////////////////////// +// +// Binary interpolative coding for a sorted, strictly increasing sequence of +// integers drawn from a known range [lo, hi]. The value domain is 32-bit: lo, hi +// and hence every value fit in ui32, independent of the element type T (which is +// merely the container's width). +// +// It recursively encodes the median element within the range implied by its +// position and its already-coded neighbors, so clustered sequences compress far +// below a flat log2 per element and no per-element headers are needed. Each +// element is stored with a truncated-binary (minimal) code, which spends the +// fractional part of log2(range) instead of rounding every element up to a whole +// bit. +// +// The bit stream is MSB-first (see bit_io.h). +// +//////////////////////////////////////////////////////////////////////////////// + +//! Encodes #values, which must be strictly increasing and all within [#lo, #hi], +//! with binary interpolative coding. An empty range emits nothing; the length is +//! not stored and must be conveyed out of band (e.g. as a varint prefix). +template <std::unsigned_integral T> +void InterpolativeEncode(TBitWriter* writer, TRange<T> values, ui32 lo, ui32 hi); + +//! Decodes a sequence written by #InterpolativeEncode into #values, whose size +//! must equal the encoded element count. #lo and #hi must match the encoder. +template <std::unsigned_integral T> +void InterpolativeDecode(TBitReader* reader, TMutableRange<T> values, ui32 lo, ui32 hi); + +//! An upper bound on the buffer size #InterpolativeEncode needs to encode #count +//! values over [#lo, #hi], including the slack the writer requires. +size_t GetInterpolativeMaxByteSize(int count, ui32 lo, ui32 hi); + +//////////////////////////////////////////////////////////////////////////////// + +} // namespace NYT + +#define INTERPOLATIVE_INL_H_ +#include "interpolative-inl.h" +#undef INTERPOLATIVE_INL_H_ diff --git a/library/cpp/yt/coding/unittests/bit_io_ut.cpp b/library/cpp/yt/coding/unittests/bit_io_ut.cpp new file mode 100644 index 00000000000..5d61b166f0a --- /dev/null +++ b/library/cpp/yt/coding/unittests/bit_io_ut.cpp @@ -0,0 +1,69 @@ +#include <library/cpp/testing/gtest/gtest.h> + +#include <library/cpp/yt/coding/bit_io.h> + +#include <utility> +#include <vector> + +namespace NYT { +namespace { + +//////////////////////////////////////////////////////////////////////////////// + +TEST(TBitIOTest, RoundTrip) +{ + std::vector<std::pair<ui32, int>> items = { + {0, 0}, {1, 1}, {0, 1}, {5, 3}, {0, 8}, {255, 8}, {12345, 14}, + {0, 32}, {0xffffffffu, 32}, {0x12345, 17}, {7, 3}, {1u << 30, 31}, + }; + std::vector<char> buffer(256, 0); + + TBitWriter writer(buffer.data()); + for (auto [value, width] : items) { + writer.WriteBits(value, width); + } + writer.Finish(); + + TBitReader reader(buffer.data()); + for (auto [value, width] : items) { + EXPECT_EQ(reader.ReadBits(width), value) << "width=" << width; + } +} + +TEST(TBitIOTest, ZeroWidthIsNoop) +{ + std::vector<char> buffer(16, 0); + TBitWriter writer(buffer.data()); + writer.WriteBits(0, 0); + writer.WriteBits(1, 1); + writer.WriteBits(0, 0); + char* end = writer.Finish(); + EXPECT_EQ(end - buffer.data(), 1); // a single bit occupies one byte + + TBitReader reader(buffer.data()); + EXPECT_EQ(reader.ReadBits(0), 0u); + EXPECT_EQ(reader.ReadBits(1), 1u); + EXPECT_EQ(reader.ReadBits(0), 0u); +} + +TEST(TBitIOTest, FlushBoundary) +{ + // Many 17-bit writes repeatedly cross the internal 32-bit flush boundary. + constexpr int Count = 1000; + std::vector<char> buffer(Count * 3 + 16, 0); + TBitWriter writer(buffer.data()); + for (int i = 0; i < Count; ++i) { + writer.WriteBits(static_cast<ui32>(i) & 0x1ffff, 17); + } + writer.Finish(); + + TBitReader reader(buffer.data()); + for (int i = 0; i < Count; ++i) { + EXPECT_EQ(reader.ReadBits(17), static_cast<ui32>(i) & 0x1ffff) << "i=" << i; + } +} + +//////////////////////////////////////////////////////////////////////////////// + +} // namespace +} // namespace NYT diff --git a/library/cpp/yt/coding/unittests/interpolative_ut.cpp b/library/cpp/yt/coding/unittests/interpolative_ut.cpp new file mode 100644 index 00000000000..a63b12f6eac --- /dev/null +++ b/library/cpp/yt/coding/unittests/interpolative_ut.cpp @@ -0,0 +1,221 @@ +#include <library/cpp/testing/gtest/gtest.h> + +#include <library/cpp/yt/coding/interpolative.h> +#include <library/cpp/yt/coding/varint.h> + +#include <algorithm> +#include <bit> +#include <numeric> +#include <random> +#include <set> +#include <vector> + +namespace NYT { +namespace { + +//////////////////////////////////////////////////////////////////////////////// +// Truncated binary + +TEST(TTruncatedBinaryTest, Exhaustive) +{ + for (ui32 rangeSize = 1; rangeSize <= 2050; ++rangeSize) { + for (ui32 value = 0; value < rangeSize; ++value) { + std::vector<char> buffer(16, 0); + TBitWriter writer(buffer.data()); + NInterpolativeCodingDetail::WriteTruncatedBinary(&writer, value, rangeSize); + writer.Finish(); + + TBitReader reader(buffer.data()); + EXPECT_EQ(NInterpolativeCodingDetail::ReadTruncatedBinary(&reader, rangeSize), value) + << "rangeSize=" << rangeSize << " value=" << value; + } + } +} + +TEST(TTruncatedBinaryTest, MinimalLength) +{ + // Every codeword is floor(log2(rangeSize)) or ceil(log2(rangeSize)) bits. + for (ui32 rangeSize = 1; rangeSize <= 1000; ++rangeSize) { + int lowWidth = std::bit_width(rangeSize) - 1; + for (ui32 value = 0; value < rangeSize; ++value) { + std::vector<char> buffer(16, 0); + TBitWriter writer(buffer.data()); + NInterpolativeCodingDetail::WriteTruncatedBinary(&writer, value, rangeSize); + char* end = writer.Finish(); + + TBitReader reader(buffer.data()); + NInterpolativeCodingDetail::ReadTruncatedBinary(&reader, rangeSize); + const char* readEnd = reader.Finish(); + i64 bytes = end - buffer.data(); + // Read must consume no more bytes than were written. + EXPECT_LE(readEnd - buffer.data(), bytes); + EXPECT_LE(bytes, (lowWidth + 1 + 7) / 8 + 1); + } + } +} + +//////////////////////////////////////////////////////////////////////////////// +// Interpolative coding + +template <class T> +std::vector<char> Encode(const std::vector<T>& values, ui32 lo, ui32 hi) +{ + std::vector<char> buffer(values.size() * sizeof(ui32) + 16, 0); + TBitWriter writer(buffer.data()); + InterpolativeEncode(&writer, TRange(values), lo, hi); + char* end = writer.Finish(); + buffer.resize(end - buffer.data()); + return buffer; +} + +template <class T> +std::vector<T> Decode(std::vector<char> buffer, int count, ui32 lo, ui32 hi) +{ + buffer.resize(buffer.size() + 8, 0); // reader may over-read up to 8 bytes + std::vector<T> values(count); + TBitReader reader(buffer.data()); + InterpolativeDecode(&reader, TMutableRange(values), lo, hi); + return values; +} + +template <class T> +void ExpectRoundTrip(const std::vector<T>& values, ui32 lo, ui32 hi) +{ + auto decoded = Decode<T>(Encode(values, lo, hi), std::ssize(values), lo, hi); + EXPECT_EQ(decoded, values); +} + +TEST(TInterpolativeCodingTest, Empty) +{ + ExpectRoundTrip<ui32>({}, 0, 100); +} + +TEST(TInterpolativeCodingTest, Single) +{ + ExpectRoundTrip<ui32>({0}, 0, 0); + ExpectRoundTrip<ui32>({42}, 0, 100); + ExpectRoundTrip<ui32>({100}, 0, 100); +} + +TEST(TInterpolativeCodingTest, SmallCases) +{ + ExpectRoundTrip<ui32>({3, 7}, 0, 10); + ExpectRoundTrip<ui32>({0, 1, 2}, 0, 2); // full, zero bits + ExpectRoundTrip<ui32>({0, 5, 10}, 0, 10); // boundaries present + ExpectRoundTrip<ui32>({1, 2, 3, 4, 5}, 0, 6); +} + +TEST(TInterpolativeCodingTest, FullRange) +{ + // Every value present => every range collapses to a singleton (zero bits). + std::vector<ui32> values(500); + std::iota(values.begin(), values.end(), 7); + auto encoded = Encode(values, 7, 506); + EXPECT_TRUE(encoded.empty()); + EXPECT_EQ(Decode<ui32>(encoded, 500, 7, 506), values); +} + +TEST(TInterpolativeCodingTest, RandomRoundTrip) +{ + std::mt19937 rng(12345); + for (int iteration = 0; iteration < 500; ++iteration) { + ui32 universe = 1 + rng() % 200'000; + int count = std::min<ui32>(universe, 1 + rng() % 2000); + std::set<ui32> unique; + std::uniform_int_distribution<ui32> dist(0, universe - 1); + while (std::ssize(unique) < count) { + unique.insert(dist(rng)); + } + std::vector<ui32> values(unique.begin(), unique.end()); + ExpectRoundTrip<ui32>(values, 0, universe - 1); + } +} + +TEST(TInterpolativeCodingTest, NonZeroLowerBound) +{ + std::mt19937 rng(999); + for (int iteration = 0; iteration < 200; ++iteration) { + ui32 lo = rng() % 100'000; + ui32 span = 1 + rng() % 100'000; + ui32 hi = lo + span; + int count = std::min<ui32>(span + 1, 1 + rng() % 500); + std::set<ui32> unique; + std::uniform_int_distribution<ui32> dist(lo, hi); + while (std::ssize(unique) < count) { + unique.insert(dist(rng)); + } + std::vector<ui32> values(unique.begin(), unique.end()); + ExpectRoundTrip<ui32>(values, lo, hi); + } +} + +TEST(TInterpolativeCodingTest, Ui64Values) +{ + std::vector<ui64> values = {0, 1, 100, 1000, 50'000, 200'000}; + ExpectRoundTrip<ui64>(values, 0, 200'000); +} + +TEST(TInterpolativeCodingTest, MaxByteSize) +{ + std::mt19937 rng(555); + for (int iteration = 0; iteration < 300; ++iteration) { + ui32 universe = 1 + rng() % 200'000; + int count = std::min<ui32>(universe, 1 + rng() % 1000); + std::set<ui32> unique; + std::uniform_int_distribution<ui32> dist(0, universe - 1); + while (std::ssize(unique) < count) { + unique.insert(dist(rng)); + } + std::vector<ui32> values(unique.begin(), unique.end()); + + size_t bound = GetInterpolativeMaxByteSize(count, 0, universe - 1); + std::vector<char> buffer(bound, 0); + TBitWriter writer(buffer.data()); + InterpolativeEncode(&writer, TRange(values), 0, universe - 1); + EXPECT_LE(static_cast<size_t>(writer.Finish() - buffer.data()), bound); + } +} + +TEST(TInterpolativeCodingTest, MultipleListsInOneBuffer) +{ + // Mirrors real usage: each list is prefixed with a varint length and is + // byte-aligned, so lists can be concatenated and read back in sequence. + std::mt19937 rng(7); + ui32 hi = 199999; + std::vector<std::vector<ui32>> lists; + for (int listIndex = 0; listIndex < 50; ++listIndex) { + int count = 1 + rng() % 300; + std::set<ui32> unique; + std::uniform_int_distribution<ui32> dist(0, hi); + while (std::ssize(unique) < count) { + unique.insert(dist(rng)); + } + lists.emplace_back(unique.begin(), unique.end()); + } + + std::vector<char> buffer(200'000, 0); + char* ptr = buffer.data(); + for (const auto& list : lists) { + ptr += WriteVarUint32(ptr, static_cast<ui32>(list.size())); + TBitWriter writer(ptr); + InterpolativeEncode(&writer, TRange(list), 0, hi); + ptr = writer.Finish(); + } + + const char* readPtr = buffer.data(); + for (const auto& list : lists) { + ui32 count; + readPtr += ReadVarUint32(readPtr, &count); + ASSERT_EQ(count, list.size()); + std::vector<ui32> decoded(count); + TBitReader reader(readPtr); + InterpolativeDecode(&reader, TMutableRange(decoded), 0, hi); + readPtr = reader.Finish(); + EXPECT_EQ(decoded, list); + } +} + +//////////////////////////////////////////////////////////////////////////////// + +} // namespace +} // namespace NYT diff --git a/library/cpp/yt/coding/unittests/ya.make b/library/cpp/yt/coding/unittests/ya.make index ab94ee8796a..945e63ab506 100644 --- a/library/cpp/yt/coding/unittests/ya.make +++ b/library/cpp/yt/coding/unittests/ya.make @@ -3,6 +3,8 @@ GTEST() INCLUDE(${ARCADIA_ROOT}/library/cpp/yt/ya_cpp.make.inc) SRCS( + bit_io_ut.cpp + interpolative_ut.cpp zig_zag_ut.cpp varint_ut.cpp ) diff --git a/library/cpp/yt/coding/ya.make b/library/cpp/yt/coding/ya.make index 639d94e755c..e5ea301cf90 100644 --- a/library/cpp/yt/coding/ya.make +++ b/library/cpp/yt/coding/ya.make @@ -6,7 +6,9 @@ SRCS( ) PEERDIR( + library/cpp/yt/assert library/cpp/yt/exception + library/cpp/yt/memory ) END() diff --git a/library/cpp/yt/coding/zig_zag-inl.h b/library/cpp/yt/coding/zig_zag-inl.h index c611f7e1d46..9df9a2cc5f4 100644 --- a/library/cpp/yt/coding/zig_zag-inl.h +++ b/library/cpp/yt/coding/zig_zag-inl.h @@ -3,7 +3,6 @@ // For the sake of sane code completion. #include "zig_zag.h" #endif -#undef ZIG_ZAG_INL_H_ namespace NYT { diff --git a/library/cpp/yt/compact_containers/compact_vector-inl.h b/library/cpp/yt/compact_containers/compact_vector-inl.h index cec19960e19..1070e6b1c91 100644 --- a/library/cpp/yt/compact_containers/compact_vector-inl.h +++ b/library/cpp/yt/compact_containers/compact_vector-inl.h @@ -3,7 +3,6 @@ // For the sake of sane code completion. #include "compact_vector.h" #endif -#undef COMPACT_VECTOR_INL_H_ #include <library/cpp/yt/assert/assert.h> diff --git a/library/cpp/yt/error/error-inl.h b/library/cpp/yt/error/error-inl.h index 1542d008f58..85751e8a3cf 100644 --- a/library/cpp/yt/error/error-inl.h +++ b/library/cpp/yt/error/error-inl.h @@ -604,4 +604,24 @@ void FormatValue(TStringBuilderBase* builder, const TErrorOr<T>& error, TStringB //////////////////////////////////////////////////////////////////////////////// +template <class F, class... As> +auto RunNoExcept(F&& functor, As&&... args) noexcept -> decltype(functor(std::forward<As>(args)...)) +{ + return functor(std::forward<As>(args)...); +} + +//////////////////////////////////////////////////////////////////////////////// + +inline TStringBuf GetWellKnownLoggingTag(const std::exception&) +{ + return "Error"_sb; +} + +inline TStringBuf GetWellKnownLoggingTag(const TError&) +{ + return "Error"_sb; +} + +//////////////////////////////////////////////////////////////////////////////// + } // namespace NYT diff --git a/library/cpp/yt/error/error.cpp b/library/cpp/yt/error/error.cpp index d0a5489ab22..dc5aca7ccd2 100644 --- a/library/cpp/yt/error/error.cpp +++ b/library/cpp/yt/error/error.cpp @@ -5,6 +5,8 @@ #include <library/cpp/yt/error/error_attributes.h> #include <library/cpp/yt/error/origin_attributes.h> +#include <library/cpp/yt/memory/leaky_singleton.h> + #include <library/cpp/yt/string/string.h> #include <library/cpp/yt/system/proc.h> @@ -29,8 +31,20 @@ void FormatValue(TStringBuilderBase* builder, TErrorCode code, TStringBuf spec) constexpr TStringBuf ErrorMessageTruncatedSuffix = "...<message truncated>"; -TError::TEnricher TError::Enricher_; -TError::TFromExceptionEnricher TError::FromExceptionEnricher_; +namespace { + +struct TEnricherStorage +{ + static TEnricherStorage* Get() + { + return LeakySingleton<TEnricherStorage>(); + } + + TError::TEnricher Enricher; + TError::TFromExceptionEnricher FromExceptionEnricher; +}; + +} // namespace //////////////////////////////////////////////////////////////////////////////// @@ -636,11 +650,12 @@ void TError::RegisterEnricher(TEnricher enricher) { // NB: This daisy-chaining strategy is optimal when there's O(1) callbacks. Convert to a vector // if the number grows. - if (!Enricher_) { - Enricher_ = std::move(enricher); + auto* storage = TEnricherStorage::Get(); + if (!storage->Enricher) { + storage->Enricher = std::move(enricher); return; } - Enricher_ = [first = std::move(Enricher_), second = std::move(enricher)] (TError* error) { + storage->Enricher = [first = std::move(storage->Enricher), second = std::move(enricher)] (TError* error) { first(error); second(error); }; @@ -650,12 +665,13 @@ void TError::RegisterFromExceptionEnricher(TFromExceptionEnricher enricher) { // NB: This daisy-chaining strategy is optimal when there's O(1) callbacks. Convert to a vector // if the number grows. - if (!FromExceptionEnricher_) { - FromExceptionEnricher_ = std::move(enricher); + auto* storage = TEnricherStorage::Get(); + if (!storage->FromExceptionEnricher) { + storage->FromExceptionEnricher = std::move(enricher); return; } - FromExceptionEnricher_ = [ - first = std::move(FromExceptionEnricher_), + storage->FromExceptionEnricher = [ + first = std::move(storage->FromExceptionEnricher), second = std::move(enricher) ] (TError* error, const std::exception& exception) { first(error, exception); @@ -676,15 +692,15 @@ void TError::MakeMutable() void TError::Enrich() { - if (Enricher_) { - Enricher_(this); + if (const auto& enricher = TEnricherStorage::Get()->Enricher) { + enricher(this); } } void TError::EnrichFromException(const std::exception& exception) { - if (FromExceptionEnricher_) { - FromExceptionEnricher_(this, exception); + if (const auto& enricher = TEnricherStorage::Get()->FromExceptionEnricher) { + enricher(this, exception); } } diff --git a/library/cpp/yt/error/error.h b/library/cpp/yt/error/error.h index b2d0abacfe8..1e5c2147d82 100644 --- a/library/cpp/yt/error/error.h +++ b/library/cpp/yt/error/error.h @@ -253,9 +253,6 @@ private: void EnrichFromException(const std::exception& exception); friend class TErrorAttributes; - - static TEnricher Enricher_; - static TFromExceptionEnricher FromExceptionEnricher_; }; //////////////////////////////////////////////////////////////////////////////// @@ -478,10 +475,15 @@ void FormatValue(TStringBuilderBase* builder, const TErrorOr<T>& error, TStringB //////////////////////////////////////////////////////////////////////////////// template <class F, class... As> -auto RunNoExcept(F&& functor, As&&... args) noexcept -> decltype(functor(std::forward<As>(args)...)) -{ - return functor(std::forward<As>(args)...); -} +auto RunNoExcept(F&& functor, As&&... args) noexcept -> decltype(functor(std::forward<As>(args)...)); + +//////////////////////////////////////////////////////////////////////////////// + +//! Registers errors as a well-known logging tag (ADL customization point for +//! library/cpp/yt/logging), so that |YT_TLOG_*(...).With(error)| attaches them under +//! the "Error" key, rendered after the message in plain text. +TStringBuf GetWellKnownLoggingTag(const std::exception&); +TStringBuf GetWellKnownLoggingTag(const TError&); //////////////////////////////////////////////////////////////////////////////// diff --git a/library/cpp/yt/logging/backends/arcadia/backend.cpp b/library/cpp/yt/logging/backends/arcadia/backend.cpp index 3c6ff9f5f58..5878ba52be3 100644 --- a/library/cpp/yt/logging/backends/arcadia/backend.cpp +++ b/library/cpp/yt/logging/backends/arcadia/backend.cpp @@ -57,7 +57,7 @@ public: // Use low-level api, because it is more convinient here. auto loggingContext = GetLoggingContext(); auto event = NDetail::CreateLogEvent(loggingContext, Logger_, logLevel); - event.MessageRef = NDetail::BuildLogMessage(loggingContext, Logger_, message).MessageRef; + event.Payload = NDetail::BuildLogMessage(loggingContext, Logger_, message).Payload; event.Family = ELogFamily::PlainText; Logger_.Write(std::move(event)); } diff --git a/library/cpp/yt/logging/benchmark/logging.cpp b/library/cpp/yt/logging/benchmark/logging.cpp new file mode 100644 index 00000000000..8745511f2bd --- /dev/null +++ b/library/cpp/yt/logging/benchmark/logging.cpp @@ -0,0 +1,167 @@ +#include <benchmark/benchmark.h> + +#include <library/cpp/yt/logging/logger.h> + +#include <atomic> + +namespace NYT::NLogging { +namespace { + +//////////////////////////////////////////////////////////////////////////////// + +// A log manager whose Enqueue is essentially free, so that the benchmark measures +// only the producer-side cost (building the event payload), not the logging thread. +class TBenchmarkLogManager + : public ILogManager +{ +public: + void RegisterStaticAnchor(TLoggingAnchor* anchor, ::TSourceLocation /*sourceLocation*/, TStringBuf /*message*/) override + { + anchor->Registered.store(true); + } + + void UpdateAnchor(TLoggingAnchor* anchor) override + { + anchor->CurrentVersion.store(ActualVersion_.load()); + } + + void Enqueue(TLogEvent&& event) override + { + // Keep the produced payload observable so the build is not optimized away; + // the event is then destroyed (freeing its chunk slice) -- same for both APIs. + benchmark::DoNotOptimize(std::get<TTaggedLogEventPayload>(event.Payload).Underlying().Begin()); + } + + const TLoggingCategory* GetCategory(TStringBuf /*categoryName*/) override + { + return &Category_; + } + + void UpdateCategory(TLoggingCategory* category) override + { + category->CurrentVersion.store(ActualVersion_.load()); + } + + bool GetAbortOnAlert() const override + { + return false; + } + +private: + std::atomic<int> ActualVersion_ = 1; + TLoggingCategory Category_ = { + .Name = "Bench", + .MinPlainTextLevel = ELogLevel::Minimum, + .CurrentVersion = 0, + .ActualVersion = &ActualVersion_, + }; +}; + +//////////////////////////////////////////////////////////////////////////////// +// No tags. + +void BM_Log_NoTags(benchmark::State& state) +{ + TBenchmarkLogManager manager; + TLogger Logger(&manager, "Bench"); + for (auto _ : state) { + YT_LOG_INFO("This is a log message of a typical length without tags"); + } +} +BENCHMARK(BM_Log_NoTags); + +void BM_TLog_NoTags(benchmark::State& state) +{ + TBenchmarkLogManager manager; + TLogger Logger(&manager, "Bench"); + for (auto _ : state) { + YT_TLOG_INFO("This is a log message of a typical length without tags"); + } +} +BENCHMARK(BM_TLog_NoTags); + +//////////////////////////////////////////////////////////////////////////////// +// One int tag. + +void BM_Log_OneTag(benchmark::State& state) +{ + TBenchmarkLogManager manager; + TLogger Logger(&manager, "Bench"); + for (auto _ : state) { + YT_LOG_INFO("This is a log message of a typical length (TagName: %v)", + 123); + } +} +BENCHMARK(BM_Log_OneTag); + +void BM_TLog_OneTag(benchmark::State& state) +{ + TBenchmarkLogManager manager; + TLogger Logger(&manager, "Bench"); + for (auto _ : state) { + YT_TLOG_INFO("This is a log message of a typical length") + .With("TagName", 123); + } +} +BENCHMARK(BM_TLog_OneTag); + +//////////////////////////////////////////////////////////////////////////////// +// Two tags (int + string). + +void BM_Log_TwoTags(benchmark::State& state) +{ + TBenchmarkLogManager manager; + TLogger Logger(&manager, "Bench"); + for (auto _ : state) { + YT_LOG_INFO("This is a log message of a typical length (TagName1: %v, TagName2: %v)", + 123, + "this-is-a-string"); + } +} +BENCHMARK(BM_Log_TwoTags); + +void BM_TLog_TwoTags(benchmark::State& state) +{ + TBenchmarkLogManager manager; + TLogger Logger(&manager, "Bench"); + for (auto _ : state) { + YT_TLOG_INFO("This is a log message of a typical length") + .With("TagName1", 123) + .With("TagName2", "this-is-a-string"); + } +} +BENCHMARK(BM_TLog_TwoTags); + +//////////////////////////////////////////////////////////////////////////////// +// Three tags (int + string + double). + +void BM_Log_ThreeTags(benchmark::State& state) +{ + TBenchmarkLogManager manager; + TLogger Logger(&manager, "Bench"); + for (auto _ : state) { + YT_LOG_INFO("This is a log message of a typical length (TagName1: %v, TagName2: %v, TagName3: %v)", + 123, + "this-is-a-string", + 3.14); + } +} +BENCHMARK(BM_Log_ThreeTags); + +void BM_TLog_ThreeTags(benchmark::State& state) +{ + TBenchmarkLogManager manager; + TLogger Logger(&manager, "Bench"); + for (auto _ : state) { + YT_TLOG_INFO("This is a log message of a typical length") + .With("TagName1", 123) + .With("TagName2", "this-is-a-string") + .With("TagName3", 3.14); + } +} +BENCHMARK(BM_TLog_ThreeTags); + +//////////////////////////////////////////////////////////////////////////////// + +} // namespace +} // namespace NYT::NLogging diff --git a/library/cpp/yt/logging/benchmark/ya.make b/library/cpp/yt/logging/benchmark/ya.make index 61fd0b49276..a7a23128cdf 100644 --- a/library/cpp/yt/logging/benchmark/ya.make +++ b/library/cpp/yt/logging/benchmark/ya.make @@ -4,6 +4,7 @@ INCLUDE(${ARCADIA_ROOT}/library/cpp/yt/ya_cpp.make.inc) SRCS( logger_tag.cpp + logging.cpp ) PEERDIR( diff --git a/library/cpp/yt/logging/logger-inl.h b/library/cpp/yt/logging/logger-inl.h index 87672ec6f6a..ff4d6d81bbe 100644 --- a/library/cpp/yt/logging/logger-inl.h +++ b/library/cpp/yt/logging/logger-inl.h @@ -3,7 +3,8 @@ // For the sake of sane code completion. #include "logger.h" #endif -#undef LOGGER_INL_H_ + +#include "tagged_payload.h" #include <library/cpp/yt/yson_string/convert.h> #include <library/cpp/yt/yson_string/string.h> @@ -108,33 +109,6 @@ Y_FORCE_INLINE const TLogger& TLogger::operator()() const 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: - TSharedMutableRef Buffer_; - - static constexpr size_t ChunkSize = 128_KB - 64; -}; - inline bool HasMessageTags( const TLoggingContext& loggingContext, const TLogger& logger) @@ -163,14 +137,14 @@ inline void AppendMessageTags( } if (const auto& traceLoggingTag = loggingContext.TraceLoggingTag; !traceLoggingTag.empty()) { if (printComma) { - builder->AppendString(TStringBuf(", ")); + builder->AppendString(", "_sb); } builder->AppendString(traceLoggingTag); printComma = true; } if (const auto& threadMessageTag = GetThreadMessageTag(); !threadMessageTag.empty()) { if (printComma) { - builder->AppendString(TStringBuf(", ")); + builder->AppendString(", "_sb); } builder->AppendString(threadMessageTag); printComma = true; @@ -186,10 +160,10 @@ inline void AppendLogMessage( if (HasMessageTags(loggingContext, logger)) { if (message.Size() >= 1 && message[message.Size() - 1] == ')') { builder->AppendString(TStringBuf(message.Begin(), message.Size() - 1)); - builder->AppendString(TStringBuf(", ")); + builder->AppendString(", "_sb); } else { builder->AppendString(TStringBuf(message.Begin(), message.Size())); - builder->AppendString(TStringBuf(" (")); + builder->AppendString(" ("_sb); } AppendMessageTags(builder, loggingContext, logger); builder->AppendChar(')'); @@ -209,10 +183,10 @@ void AppendLogMessageWithFormat( if (HasMessageTags(loggingContext, logger)) { if (format.size() >= 2 && format[format.size() - 1] == ')') { builder->AppendFormat(TRuntimeFormat{format.substr(0, format.size() - 1)}, std::forward<TArgs>(args)...); - builder->AppendString(TStringBuf(", ")); + builder->AppendString(", "_sb); } else { builder->AppendFormat(TRuntimeFormat{format}, std::forward<TArgs>(args)...); - builder->AppendString(TStringBuf(" (")); + builder->AppendString(" ("_sb); } AppendMessageTags(builder, loggingContext, logger); builder->AppendChar(')'); @@ -223,7 +197,7 @@ void AppendLogMessageWithFormat( struct TLogMessage { - TSharedRef MessageRef; + TTaggedLogEventPayload Payload; TStringBuf Anchor; }; @@ -234,9 +208,10 @@ TLogMessage BuildLogMessage( TFormatString<TArgs...> format, TArgs&&... args) { - TMessageStringBuilder builder; - AppendLogMessageWithFormat(&builder, loggingContext, logger, format.Get(), std::forward<TArgs>(args)...); - return {builder.Flush(), format.Get()}; + TTaggedPayloadWriter writer; + AppendLogMessageWithFormat(writer.BeginMessage(), loggingContext, logger, format.Get(), std::forward<TArgs>(args)...); + writer.EndMessage(); + return {writer.Finish(), format.Get()}; } template <CFormattable T> @@ -246,13 +221,15 @@ TLogMessage BuildLogMessage( const TLogger& logger, const T& obj) { - TMessageStringBuilder builder; - FormatValue(&builder, obj, TStringBuf("v")); + TTaggedPayloadWriter writer; + auto* builder = writer.BeginMessage(); + FormatValue(builder, obj, "v"_sb); if (HasMessageTags(loggingContext, logger)) { - builder.AppendString(TStringBuf(" (")); - AppendMessageTags(&builder, loggingContext, logger); - builder.AppendChar(')'); + builder->AppendString(" ("_sb); + AppendMessageTags(builder, loggingContext, logger); + builder->AppendChar(')'); } + writer.EndMessage(); if constexpr (std::same_as<TStringBuf, std::remove_cvref_t<T>>) { // NB(arkady-e1ppa): This is the overload where TStringBuf @@ -263,9 +240,9 @@ TLogMessage BuildLogMessage( // us having overload for TStringBuf (both have implicit ctors from // string literals) thus we have to accommodate TStringBuf specifics // in this if constexpr part. - return {builder.Flush(), obj}; + return {writer.Finish(), obj}; } else { - return {builder.Flush(), TStringBuf()}; + return {writer.Finish(), TStringBuf()}; } } @@ -296,13 +273,10 @@ inline TLogMessage BuildLogMessage( 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()}; - } + TTaggedPayloadWriter writer; + AppendLogMessage(writer.BeginMessage(), loggingContext, logger, message); + writer.EndMessage(); + return {writer.Finish(), TStringBuf()}; } inline TLogEvent CreateLogEvent( @@ -311,10 +285,10 @@ inline TLogEvent CreateLogEvent( ELogLevel level) { TLogEvent event; - event.Instant = loggingContext.Instant; event.Category = logger.GetCategory(); - event.Essential = logger.IsEssential(); event.Level = level; + event.Essential = logger.IsEssential(); + event.Instant = loggingContext.Instant; event.ThreadId = loggingContext.ThreadId; event.ThreadName = loggingContext.ThreadName; event.FiberId = loggingContext.FiberId; @@ -333,15 +307,24 @@ inline void LogEventImpl( ELogLevel level, ::TSourceLocation sourceLocation, TLoggingAnchor* anchor, - TSharedRef message) + TTaggedLogEventPayload payload) { - 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; - event.Anchor = anchor; + auto event = TLogEvent{ + .Category = logger.GetCategory(), + .Level = level, + .Family = ELogFamily::PlainText, + .Essential = logger.IsEssential(), + .Payload = std::move(payload), + .Instant = loggingContext.Instant, + .ThreadId = loggingContext.ThreadId, + .ThreadName = loggingContext.ThreadName, + .FiberId = loggingContext.FiberId, + .TraceId = loggingContext.TraceId, + .RequestId = loggingContext.RequestId, + .SourceFile = sourceLocation.File, + .SourceLine = sourceLocation.Line, + .Anchor = anchor, + }; if (Y_UNLIKELY(event.Level >= ELogLevel::Alert)) { logger.Write(TLogEvent(event)); OnCriticalLogEvent(logger, event); @@ -350,6 +333,299 @@ inline void LogEventImpl( } } +//////////////////////////////////////////////////////////////////////////////// + +//! References the per-call-site static anchor and its one-shot registration flag. +//! Produced by the lambda embedded in the fluent |YT_TLOG_*| macros. +struct TStaticAnchorRef +{ + TLoggingAnchor* Anchor; + std::atomic<bool>* Registered; +}; + +//! Wraps the format spec passed to |TTaggedLoggingGuard::With| and validates at compile +//! time that it is a |%|-prefixed string literal (e.g. |"%v"|, |"%08x"|). The stored +//! spec has the leading |%| stripped, as expected by |FormatValue|. +class TLoggingTagSpec +{ +public: + template <size_t N> + consteval TLoggingTagSpec(const char (&spec)[N]) + : Spec_(spec + 1, N - 2) + { + static_assert(N >= 2, "Logging tag format spec must be a non-empty string literal"); + if (spec[0] != '%') { + TheLoggingTagFormatSpecMustStartWithPercentSign(); + } + } + + TStringBuf Get() const + { + return Spec_; + } + +private: + const TStringBuf Spec_; + + // Undefined on purpose: calling it from the |consteval| ctor turns a missing + // leading |%| into a compile error that names the violated rule. + static void TheLoggingTagFormatSpecMustStartWithPercentSign(); +}; + +class TWellKnownTaggedLoggingGuard; + +//! Accumulates a tagged log message via a fluent |.With| chain and emits the event in +//! its destructor. Instantiated by the fluent |YT_TLOG_*| macros, which guarantee that +//! the chain is reached only when the level is enabled (so tag value expressions are +//! not evaluated otherwise). +//! +//! The user message -- with the logger's contextual (logger/trace/thread) tags folded +//! in -- goes to the payload message field; each |.With(key, value)| becomes a +//! structured payload tag (see #TTaggedPayloadWriter). +class TTaggedLoggingGuard +{ +public: + TTaggedLoggingGuard( + const TLogger& logger, + ELogLevel level, + ::TSourceLocation sourceLocation, + TStaticAnchorRef anchorRef, + TStringBuf message) + : TTaggedLoggingGuard( + logger, + level, + sourceLocation, + anchorRef, + message, + /*alwaysBuildMessage*/ false) + { } + + TTaggedLoggingGuard(const TTaggedLoggingGuard&) = delete; + TTaggedLoggingGuard& operator=(const TTaggedLoggingGuard&) = delete; + + bool IsEnabled() const + { + return Enabled_; + } + + template <class TValue> + TTaggedLoggingGuard& With(TStringBuf tag, const TValue& value) & + { + return DoWith(tag, value, "v"_sb); + } + + template <class TValue> + TTaggedLoggingGuard& With(TStringBuf tag, const TValue& value, TLoggingTagSpec spec) & + { + return DoWith(tag, value, spec.Get()); + } + + //! Attaches a well-known tag whose key is resolved from #value's type via the + //! |GetWellKnownLoggingTag| ADL point (the type must opt in, e.g. errors). + //! + //! Returns a #TWellKnownTaggedLoggingGuard, which exposes only further well-known + //! tags: the payload contract requires well-known tags to come last (so + //! #FormatTaggedPayload can stay single-pass), so a keyed |.With(key, value)| after a + //! well-known tag must not compile. + template <class TValue> + TWellKnownTaggedLoggingGuard With(const TValue& value) &; + + ~TTaggedLoggingGuard() + { + if (!Enabled_) { + return; + } + LogEventImpl( + LoggingContext_, + Logger_, + EffectiveLevel_, + SourceLocation_, + Anchor_, + Writer_.Finish()); + } + +protected: + const TLogger& Logger_; + const ::TSourceLocation SourceLocation_; + TLoggingAnchor* const Anchor_; + + bool Enabled_ = false; + ELogLevel EffectiveLevel_ = ELogLevel::Minimum; + TLoggingContext LoggingContext_; + TTaggedPayloadWriter Writer_; + + //! Shared constructor. When #alwaysBuildMessage is set the payload message is built + //! even if the level is disabled (so a terminal guard can still recover it); #Enabled_ + //! continues to gate whether the destructor emits the event. + TTaggedLoggingGuard( + const TLogger& logger, + ELogLevel level, + ::TSourceLocation sourceLocation, + TStaticAnchorRef anchorRef, + TStringBuf message, + bool alwaysBuildMessage) + : Logger_(logger) + , SourceLocation_(sourceLocation) + , Anchor_(anchorRef.Anchor) + { + if (!Logger_.IsAnchorUpToDate(*Anchor_)) [[unlikely]] { + Logger_.UpdateStaticAnchor(Anchor_, anchorRef.Registered, sourceLocation, message); + } + + EffectiveLevel_ = TLogger::GetEffectiveLoggingLevel(level, *Anchor_); + Enabled_ = Logger_.IsLevelEnabled(EffectiveLevel_); + if (!Enabled_ && !alwaysBuildMessage) { + return; + } + + LoggingContext_ = GetLoggingContext(); + + auto* builder = Writer_.BeginMessage(); + builder->AppendString(message); + if (HasMessageTags(LoggingContext_, Logger_)) { + builder->AppendString(" ("_sb); + AppendMessageTags(builder, LoggingContext_, Logger_); + builder->AppendChar(')'); + } + Writer_.EndMessage(); + } + +private: + template <class TValue> + TTaggedLoggingGuard& DoWith(TStringBuf tag, const TValue& value, TStringBuf spec) & + { + // Format the value straight into the payload buffer; no temporary. + FormatValue(Writer_.BeginTag(tag), value, spec); + Writer_.EndTag(); + return *this; + } +}; + +//! Restricts the fluent |.With| chain once a well-known tag has been attached. The +//! payload contract requires well-known tags to come last (#FormatTaggedPayload is +//! single-pass), so only further well-known |.With(value)| calls are exposed -- a keyed +//! |.With(key, value)| after a well-known tag fails to compile. +class TWellKnownTaggedLoggingGuard +{ +public: + explicit TWellKnownTaggedLoggingGuard(TTaggedLoggingGuard& guard) + : Guard_(guard) + { } + + template <class TValue> + TWellKnownTaggedLoggingGuard With(const TValue& value) && + { + return Guard_.With(value); + } + +private: + TTaggedLoggingGuard& Guard_; +}; + +template <class TValue> +TWellKnownTaggedLoggingGuard TTaggedLoggingGuard::With(const TValue& value) & +{ + FormatValue(Writer_.BeginWellKnownTag(GetWellKnownLoggingTag(value)), value, "v"_sb); + Writer_.EndTag(); + return TWellKnownTaggedLoggingGuard(*this); +} + +//! Terminal guard for the fluent |YT_TLOG_FATAL| macros. Builds the message +//! unconditionally and, once the |.With| chain completes, emits the event at |Fatal| +//! level -- which aborts the process. The enclosing |for| invokes #Commit in its step. +class TTaggedFatalLoggingGuard + : public TTaggedLoggingGuard +{ +public: + TTaggedFatalLoggingGuard( + const TLogger& logger, + ::TSourceLocation sourceLocation, + TStaticAnchorRef anchorRef, + TStringBuf message) + : TTaggedLoggingGuard(logger, ELogLevel::Fatal, sourceLocation, anchorRef, message, /*alwaysBuildMessage*/ true) + { } + + //! Returns true exactly once, so the enclosing |for| runs the |.With| chain a single + //! time before its step expression commits the event. + bool TryEnter() + { + bool pending = Pending_; + Pending_ = false; + return pending; + } + + //! Emits the event at |Fatal| level; the log manager aborts the process. + [[noreturn]] void Commit() & + { + Enabled_ = false; // The event is emitted here, not from the base destructor. + LogEventImpl(LoggingContext_, Logger_, ELogLevel::Fatal, SourceLocation_, Anchor_, Writer_.Finish()); + Y_UNREACHABLE(); + } + +private: + bool Pending_ = true; +}; + +//! Terminal guard for the fluent |YT_TLOG_ALERT_AND_THROW| macros. Builds the message +//! unconditionally; once the |.With| chain completes, #Commit emits the event at |Alert| +//! level (when enabled) and returns the message so the macro can attach it to the thrown +//! error. The throw lives in the macro -- the logging library must not depend on the +//! error library, and a destructor must not throw. +class TTaggedThrowingLoggingGuard + : public TTaggedLoggingGuard +{ +public: + TTaggedThrowingLoggingGuard( + const TLogger& logger, + ::TSourceLocation sourceLocation, + TStaticAnchorRef anchorRef, + TStringBuf message) + : TTaggedLoggingGuard(logger, ELogLevel::Alert, sourceLocation, anchorRef, message, /*alwaysBuildMessage*/ true) + { } + + //! Returns true exactly once, so the enclosing |for| runs the |.With| chain a single + //! time before its step expression commits the event and throws. + bool TryEnter() + { + bool pending = Pending_; + Pending_ = false; + return pending; + } + + //! Emits the alert event (when enabled) and returns its message for the error payload. + std::string Commit() & + { + auto payload = Writer_.Finish(); + std::string message(GetMessageFromTaggedPayload(payload)); + if (Enabled_) { + Enabled_ = false; // The event is emitted here, not from the base destructor. + LogEventImpl(LoggingContext_, Logger_, EffectiveLevel_, SourceLocation_, Anchor_, std::move(payload)); + } + return message; + } + +private: + bool Pending_ = true; +}; + +//! A no-op stand-in for #TTaggedLoggingGuard used by compile-time-disabled trace logging: +//! it swallows the |.With| chain without evaluating it. +class TNullTaggedLoggingGuard +{ +public: + template <class... TArgs> + TNullTaggedLoggingGuard& With(TArgs&&...) + { + return *this; + } +}; + +template <class TMessage> +TNullTaggedLoggingGuard MakeNullTaggedLoggingGuard(const TMessage&) +{ + return {}; +} + } // namespace NDetail //////////////////////////////////////////////////////////////////////////////// diff --git a/library/cpp/yt/logging/logger.cpp b/library/cpp/yt/logging/logger.cpp index 64d8b06fa7e..5359c1cd141 100644 --- a/library/cpp/yt/logging/logger.cpp +++ b/library/cpp/yt/logging/logger.cpp @@ -1,9 +1,13 @@ #include "logger.h" +#include "structured_payload.h" + #include <library/cpp/yt/assert/assert.h> #include <library/cpp/yt/cpu_clock/clock.h> +#include <library/cpp/yt/memory/leaky_singleton.h> + #include <library/cpp/yt/system/thread_name.h> #include <util/system/compiler.h> @@ -23,98 +27,13 @@ void OnCriticalLogEvent( event.Level == ELogLevel::Alert && logger.GetAbortOnAlert()) { fprintf(stderr, "*** Aborting on critical log event\n"); - fwrite(event.MessageRef.begin(), 1, event.MessageRef.size(), stderr); + auto message = FormatTaggedPayload(std::get<TTaggedLogEventPayload>(event.Payload)); + fwrite(message.data(), 1, message.size(), stderr); fprintf(stderr, "\n"); YT_ABORT(); } } -TSharedRef TMessageStringBuilder::Flush() -{ - return Buffer_.Slice(0, GetLength()); -} - -void TMessageStringBuilder::DoReset() -{ - Buffer_.Reset(); -} - -struct TPerThreadCache; - -YT_DEFINE_THREAD_LOCAL(TPerThreadCache*, Cache); -YT_DEFINE_THREAD_LOCAL(bool, CacheDestroyed); - -struct TPerThreadCache -{ - TSharedMutableRef Chunk; - size_t ChunkOffset = 0; - - ~TPerThreadCache() - { - TMessageStringBuilder::DisablePerThreadCache(); - } - - static YT_PREVENT_TLS_CACHING TPerThreadCache* GetCache() - { - auto& cache = Cache(); - if (Y_LIKELY(cache)) { - return cache; - } - if (CacheDestroyed()) { - return nullptr; - } - static thread_local TPerThreadCache CacheData; - cache = &CacheData; - return cache; - } -}; - -void TMessageStringBuilder::DisablePerThreadCache() -{ - Cache() = nullptr; - CacheDestroyed() = true; -} - -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 = TPerThreadCache::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; -} - } // namespace NDetail //////////////////////////////////////////////////////////////////////////////// @@ -149,16 +68,34 @@ ELogLevel GetThreadMinLogLevel() //////////////////////////////////////////////////////////////////////////////// -YT_DEFINE_THREAD_LOCAL(std::string, ThreadMessageTag); +YT_DEFINE_THREAD_LOCAL(bool, ThreadMessageTagDestroyed, false); + +struct TThreadMessageTagStorage +{ + std::string Tag; + + ~TThreadMessageTagStorage() + { + ThreadMessageTagDestroyed() = true; + } +}; + +YT_DEFINE_THREAD_LOCAL(TThreadMessageTagStorage, ThreadMessageTag); void SetThreadMessageTag(std::string messageTag) { - ThreadMessageTag() = std::move(messageTag); + if (Y_UNLIKELY(ThreadMessageTagDestroyed())) { + return; + } + ThreadMessageTag().Tag = std::move(messageTag); } std::string& GetThreadMessageTag() { - return ThreadMessageTag(); + if (Y_UNLIKELY(ThreadMessageTagDestroyed())) { + return *LeakySingleton<std::string>(); + } + return ThreadMessageTag().Tag; } //////////////////////////////////////////////////////////////////////////////// @@ -350,8 +287,7 @@ void LogStructuredEvent( loggingContext, logger, level); - event.MessageKind = ELogMessageKind::Structured; - event.MessageRef = message.ToSharedRef(); + event.Payload = MakeStructuredPayloadFromYson(message); event.Family = ELogFamily::Structured; logger.Write(std::move(event)); } diff --git a/library/cpp/yt/logging/logger.h b/library/cpp/yt/logging/logger.h index d781326e88d..0f29b6f7376 100644 --- a/library/cpp/yt/logging/logger.h +++ b/library/cpp/yt/logging/logger.h @@ -80,11 +80,6 @@ using TRequestId = TGuid; //////////////////////////////////////////////////////////////////////////////// -DEFINE_ENUM(ELogMessageKind, - (Unstructured) - (Structured) -); - struct TLogEvent { const TLoggingCategory* Category = nullptr; @@ -92,8 +87,9 @@ struct TLogEvent ELogFamily Family = ELogFamily::PlainText; bool Essential = false; - ELogMessageKind MessageKind = ELogMessageKind::Unstructured; - TSharedRef MessageRef; + //! The active alternative identifies the event kind (tagged plain-text vs + //! structured) and determines how the bytes are decoded. + TLogEventPayload Payload; TCpuInstant Instant = 0; @@ -327,10 +323,10 @@ void LogStructuredEvent( #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 { \ +#define YT_LOG_FATAL(...) \ + do { \ YT_LOG_EVENT(Logger, ::NYT::NLogging::ELogLevel::Fatal, __VA_ARGS__); \ - Y_UNREACHABLE(); \ + Y_UNREACHABLE(); \ } 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__) @@ -346,126 +342,233 @@ void LogStructuredEvent( * 4. The exception is raised regardless of the logging level configured for the |Logger| and * and the currently active message suppressions. */ -#define YT_LOG_ALERT_AND_THROW(...) \ - do { \ - /* NOLINTBEGIN(bugprone-reserved-identifier, readability-identifier-naming) */ \ - const auto& logger__ = Logger(); \ - auto level__ = ::NYT::NLogging::ELogLevel::Alert; \ - auto location__ = __LOCATION__; \ - \ - auto loggingContext__ = ::NYT::NLogging::GetLoggingContext(); \ - auto message__ = ::NYT::NLogging::NDetail::BuildLogMessage(loggingContext__, logger__, __VA_ARGS__); \ - auto messageStr__ = ::std::string(message__.MessageRef.ToStringBuf()); \ - \ - static ::NYT::TLeakyStorage<::NYT::NLogging::TLoggingAnchor> anchorStorage__; \ - auto* anchor__ = anchorStorage__.Get(); \ - \ - bool anchorUpToDate__ = logger__.IsAnchorUpToDate(*anchor__); \ - if (!anchorUpToDate__) [[unlikely]] { \ - static std::atomic<bool> anchorRegistered__; \ - logger__.UpdateStaticAnchor(anchor__, &anchorRegistered__, location__, message__.Anchor); \ - } \ - \ - auto effectiveLevel__ = ::NYT::NLogging::TLogger::GetEffectiveLoggingLevel(level__, *anchor__); \ - if (logger__.IsLevelEnabled(effectiveLevel__)) { \ - ::NYT::NLogging::NDetail::LogEventImpl( \ - loggingContext__, \ - logger__, \ - effectiveLevel__, \ - location__, \ - anchor__, \ - std::move(message__.MessageRef)); \ - } \ - \ - THROW_ERROR_EXCEPTION( \ - ::NYT::EErrorCode::Fatal, \ - "Malformed request or incorrect state detected") \ - << ::NYT::TErrorAttribute("message", std::move(messageStr__)); \ - /* NOLINTEND(bugprone-reserved-identifier, readability-identifier-naming) */ \ +#define YT_LOG_ALERT_AND_THROW(...) \ + do { \ + /* NOLINTBEGIN(bugprone-reserved-identifier, readability-identifier-naming) */ \ + const auto& logger__ = Logger(); \ + auto level__ = ::NYT::NLogging::ELogLevel::Alert; \ + auto location__ = __LOCATION__; \ + \ + auto loggingContext__ = ::NYT::NLogging::GetLoggingContext(); \ + auto message__ = ::NYT::NLogging::NDetail::BuildLogMessage(loggingContext__, logger__, __VA_ARGS__); \ + /* Copy the message out before the payload is moved into the log event below. */ \ + auto messageStr__ = ::std::string(::NYT::NLogging::GetMessageFromTaggedPayload(message__.Payload)); \ + \ + static ::NYT::TLeakyStorage<::NYT::NLogging::TLoggingAnchor> anchorStorage__; \ + auto* anchor__ = anchorStorage__.Get(); \ + \ + bool anchorUpToDate__ = logger__.IsAnchorUpToDate(*anchor__); \ + if (!anchorUpToDate__) [[unlikely]] { \ + static std::atomic<bool> anchorRegistered__; \ + logger__.UpdateStaticAnchor(anchor__, &anchorRegistered__, location__, message__.Anchor); \ + } \ + \ + auto effectiveLevel__ = ::NYT::NLogging::TLogger::GetEffectiveLoggingLevel(level__, *anchor__); \ + if (logger__.IsLevelEnabled(effectiveLevel__)) { \ + ::NYT::NLogging::NDetail::LogEventImpl( \ + loggingContext__, \ + logger__, \ + effectiveLevel__, \ + location__, \ + anchor__, \ + std::move(message__.Payload)); \ + } \ + \ + THROW_ERROR_EXCEPTION( \ + ::NYT::EErrorCode::Fatal, \ + "Malformed request or incorrect state detected") \ + << ::NYT::TErrorAttribute("message", std::move(messageStr__)); \ + /* NOLINTEND(bugprone-reserved-identifier, readability-identifier-naming) */ \ } while (false) #define YT_LOG_ALERT_AND_THROW_IF(condition, ...) \ - if (Y_UNLIKELY(condition)) { \ - YT_LOG_ALERT_AND_THROW(__VA_ARGS__); \ - } \ + if (Y_UNLIKELY(condition)) { \ + YT_LOG_ALERT_AND_THROW(__VA_ARGS__); \ + } \ static_assert(true) #define YT_LOG_ALERT_AND_THROW_UNLESS(condition, ...) \ - if (!Y_UNLIKELY(condition)) { \ - YT_LOG_ALERT_AND_THROW(__VA_ARGS__); \ - } \ + if (!Y_UNLIKELY(condition)) { \ + YT_LOG_ALERT_AND_THROW(__VA_ARGS__); \ + } \ static_assert(true) -#define YT_LOG_EVENT(logger, level, ...) \ - do { \ - /* NOLINTBEGIN(bugprone-reserved-identifier, readability-identifier-naming) */ \ - const auto& logger__ = (logger)(); \ - auto level__ = (level); \ - auto location__ = __LOCATION__; \ - static ::NYT::TLeakyStorage<::NYT::NLogging::TLoggingAnchor> anchorStorage__; \ - auto* anchor__ = anchorStorage__.Get(); \ - \ - bool anchorUpToDate__ = logger__.IsAnchorUpToDate(*anchor__); \ - if (anchorUpToDate__) [[likely]] { \ - auto effectiveLevel__ = ::NYT::NLogging::TLogger::GetEffectiveLoggingLevel(level__, *anchor__); \ - if (!logger__.IsLevelEnabled(effectiveLevel__)) { \ - break; \ - } \ - } \ - \ - auto loggingContext__ = ::NYT::NLogging::GetLoggingContext(); \ +#define YT_LOG_EVENT(logger, level, ...) \ + do { \ + /* NOLINTBEGIN(bugprone-reserved-identifier, readability-identifier-naming) */ \ + const auto& logger__ = (logger)(); \ + auto level__ = (level); \ + auto location__ = __LOCATION__; \ + static ::NYT::TLeakyStorage<::NYT::NLogging::TLoggingAnchor> anchorStorage__; \ + auto* anchor__ = anchorStorage__.Get(); \ + \ + bool anchorUpToDate__ = logger__.IsAnchorUpToDate(*anchor__); \ + if (anchorUpToDate__) [[likely]] { \ + auto effectiveLevel__ = ::NYT::NLogging::TLogger::GetEffectiveLoggingLevel(level__, *anchor__); \ + if (!logger__.IsLevelEnabled(effectiveLevel__)) { \ + break; \ + } \ + } \ + \ + auto loggingContext__ = ::NYT::NLogging::GetLoggingContext(); \ auto message__ = ::NYT::NLogging::NDetail::BuildLogMessage(loggingContext__, logger__, __VA_ARGS__); \ - \ - if (!anchorUpToDate__) [[unlikely]] { \ - static std::atomic<bool> anchorRegistered__; \ - logger__.UpdateStaticAnchor(anchor__, &anchorRegistered__, location__, message__.Anchor); \ - } \ - \ - auto effectiveLevel__ = ::NYT::NLogging::TLogger::GetEffectiveLoggingLevel(level__, *anchor__); \ - if (!logger__.IsLevelEnabled(effectiveLevel__)) { \ - break; \ - } \ - \ - ::NYT::NLogging::NDetail::LogEventImpl( \ - loggingContext__, \ - logger__, \ - effectiveLevel__, \ - location__, \ - anchor__, \ - std::move(message__.MessageRef)); \ - /* NOLINTEND(bugprone-reserved-identifier, readability-identifier-naming) */ \ + \ + if (!anchorUpToDate__) [[unlikely]] { \ + static std::atomic<bool> anchorRegistered__; \ + logger__.UpdateStaticAnchor(anchor__, &anchorRegistered__, location__, message__.Anchor); \ + } \ + \ + auto effectiveLevel__ = ::NYT::NLogging::TLogger::GetEffectiveLoggingLevel(level__, *anchor__); \ + if (!logger__.IsLevelEnabled(effectiveLevel__)) { \ + break; \ + } \ + \ + ::NYT::NLogging::NDetail::LogEventImpl( \ + loggingContext__, \ + logger__, \ + effectiveLevel__, \ + location__, \ + anchor__, \ + std::move(message__.Payload)); \ + /* NOLINTEND(bugprone-reserved-identifier, readability-identifier-naming) */ \ } while (false) -#define YT_LOG_EVENT_WITH_DYNAMIC_ANCHOR(logger, level, anchor, ...) \ - do { \ - const auto& logger__ = (logger)(); \ - auto level__ = (level); \ - auto location__ = __LOCATION__; \ - auto* anchor__ = (anchor); \ - \ - bool anchorUpToDate__ = logger__.IsAnchorUpToDate(*anchor__); \ - if (!anchorUpToDate__) [[unlikely]] { \ - logger__.UpdateDynamicAnchor(anchor__); \ - } \ - \ - auto effectiveLevel__ = ::NYT::NLogging::TLogger::GetEffectiveLoggingLevel(level__, *anchor__); \ - if (!logger__.IsLevelEnabled(effectiveLevel__)) { \ - break; \ - } \ - \ - auto loggingContext__ = ::NYT::NLogging::GetLoggingContext(); \ +#define YT_LOG_EVENT_WITH_DYNAMIC_ANCHOR(logger, level, anchor, ...) \ + do { \ + const auto& logger__ = (logger)(); \ + auto level__ = (level); \ + auto location__ = __LOCATION__; \ + auto* anchor__ = (anchor); \ + \ + bool anchorUpToDate__ = logger__.IsAnchorUpToDate(*anchor__); \ + if (!anchorUpToDate__) [[unlikely]] { \ + logger__.UpdateDynamicAnchor(anchor__); \ + } \ + \ + auto effectiveLevel__ = ::NYT::NLogging::TLogger::GetEffectiveLoggingLevel(level__, *anchor__); \ + if (!logger__.IsLevelEnabled(effectiveLevel__)) { \ + break; \ + } \ + \ + auto loggingContext__ = ::NYT::NLogging::GetLoggingContext(); \ auto message__ = ::NYT::NLogging::NDetail::BuildLogMessage(loggingContext__, logger__, __VA_ARGS__); \ - \ - ::NYT::NLogging::NDetail::LogEventImpl( \ - loggingContext__, \ - logger__, \ - effectiveLevel__, \ - location__, \ - anchor__, \ - std::move(message__.MessageRef)); \ + \ + ::NYT::NLogging::NDetail::LogEventImpl( \ + loggingContext__, \ + logger__, \ + effectiveLevel__, \ + location__, \ + anchor__, \ + std::move(message__.Payload)); \ } while (false) //////////////////////////////////////////////////////////////////////////////// +// Tagged logging +// +// Tags are supplied via a fluent |.With(name, value)| (or |.With(name, value, "%spec")|) +// chain; they are carried as structured key/value pairs in the event payload. A single- +// argument |.With(value)| attaches the value under a statically known key resolved by ADL +// (e.g. |.With(error)| under the "Error" key): +// +// YT_TLOG_INFO("Message") +// .With("Key", value) +// .With("Count", count, "%08x") +// .With(error); +// +// If the message is not logged then the |.With| chain is not evaluated, so tag value +// expressions cost nothing. + +//! Yields a #TStaticAnchorRef for the expansion site: a per-call-site leaky anchor and +//! its one-shot registration flag, produced via an immediately-invoked lambda. +#define YT_TLOG_STATIC_ANCHOR_REF() \ + [] { \ + /* NOLINTBEGIN(bugprone-reserved-identifier, readability-identifier-naming) */ \ + static ::NYT::TLeakyStorage<::NYT::NLogging::TLoggingAnchor> anchorStorage__; \ + static std::atomic<bool> anchorRegistered__; \ + return ::NYT::NLogging::NDetail::TStaticAnchorRef{anchorStorage__.Get(), &anchorRegistered__}; \ + /* NOLINTEND(bugprone-reserved-identifier, readability-identifier-naming) */ \ + }() + +#define YT_TLOG_EVENT_FLUENT(logger, level, message) \ + if (::NYT::NLogging::NDetail::TTaggedLoggingGuard loggingGuard__( \ + (logger)(), \ + (level), \ + __LOCATION__, \ + YT_TLOG_STATIC_ANCHOR_REF(), \ + (message)); \ + !loggingGuard__.IsEnabled()) \ + { } else \ + loggingGuard__ + +#ifdef YT_ENABLE_TRACE_LOGGING +#define YT_TLOG_TRACE(message) YT_TLOG_EVENT_FLUENT(Logger, ::NYT::NLogging::ELogLevel::Trace, message) +#define YT_TLOG_TRACE_IF(condition, message) if (condition) YT_TLOG_TRACE(message) +#define YT_TLOG_TRACE_UNLESS(condition, message) if (!(condition)) YT_TLOG_TRACE(message) +#else +#define YT_TLOG_UNUSED(message) if (true) { } else ::NYT::NLogging::NDetail::MakeNullTaggedLoggingGuard(message) +#define YT_TLOG_TRACE(message) YT_TLOG_UNUSED(message) +#define YT_TLOG_TRACE_IF(condition, message) YT_TLOG_UNUSED(message) +#define YT_TLOG_TRACE_UNLESS(condition, message) YT_TLOG_UNUSED(message) +#endif + +#define YT_TLOG_DEBUG(message) YT_TLOG_EVENT_FLUENT(Logger, ::NYT::NLogging::ELogLevel::Debug, message) +#define YT_TLOG_DEBUG_IF(condition, message) if (condition) YT_TLOG_DEBUG(message) +#define YT_TLOG_DEBUG_UNLESS(condition, message) if (!(condition)) YT_TLOG_DEBUG(message) + +#define YT_TLOG_INFO(message) YT_TLOG_EVENT_FLUENT(Logger, ::NYT::NLogging::ELogLevel::Info, message) +#define YT_TLOG_INFO_IF(condition, message) if (condition) YT_TLOG_INFO(message) +#define YT_TLOG_INFO_UNLESS(condition, message) if (!(condition)) YT_TLOG_INFO(message) + +#define YT_TLOG_WARNING(message) YT_TLOG_EVENT_FLUENT(Logger, ::NYT::NLogging::ELogLevel::Warning, message) +#define YT_TLOG_WARNING_IF(condition, message) if (condition) YT_TLOG_WARNING(message) +#define YT_TLOG_WARNING_UNLESS(condition, message) if (!(condition)) YT_TLOG_WARNING(message) + +#define YT_TLOG_ERROR(message) YT_TLOG_EVENT_FLUENT(Logger, ::NYT::NLogging::ELogLevel::Error, message) +#define YT_TLOG_ERROR_IF(condition, message) if (condition) YT_TLOG_ERROR(message) +#define YT_TLOG_ERROR_UNLESS(condition, message) if (!(condition)) YT_TLOG_ERROR(message) + +#define YT_TLOG_ALERT(message) YT_TLOG_EVENT_FLUENT(Logger, ::NYT::NLogging::ELogLevel::Alert, message) +#define YT_TLOG_ALERT_IF(condition, message) if (condition) YT_TLOG_ALERT(message) +#define YT_TLOG_ALERT_UNLESS(condition, message) if (!(condition)) YT_TLOG_ALERT(message) + +// The terminal action of |YT_TLOG_FATAL|/|YT_TLOG_ALERT_AND_THROW| (aborting or throwing) +// must run *after* the |.With| chain, which the user appends to the macro. A guard cannot +// do it in its destructor -- that would run mid-chain on a temporary, and a throwing +// destructor terminates. So both expand to a single-iteration |for| whose step expression +// fires once the chain (the loop body) has completed. + +#define YT_TLOG_FATAL(message) \ + for (::NYT::NLogging::NDetail::TTaggedFatalLoggingGuard loggingGuard__( \ + Logger(), \ + __LOCATION__, \ + YT_TLOG_STATIC_ANCHOR_REF(), \ + (message)); \ + loggingGuard__.TryEnter(); \ + loggingGuard__.Commit()) \ + loggingGuard__ +#define YT_TLOG_FATAL_IF(condition, message) if (condition) [[unlikely]] YT_TLOG_FATAL(message) +#define YT_TLOG_FATAL_UNLESS(condition, message) if (!(condition)) [[unlikely]] YT_TLOG_FATAL(message) + +// See #YT_LOG_ALERT_AND_THROW for the rationale. The throw lives here -- not in the guard +// -- because the logging library must not depend on the error library. The guard's +// |Commit| logs the alert (when enabled) and returns the message for the |"message"| +// attribute. +#define YT_TLOG_ALERT_AND_THROW(message) \ + for (::NYT::NLogging::NDetail::TTaggedThrowingLoggingGuard loggingGuard__( \ + Logger(), \ + __LOCATION__, \ + YT_TLOG_STATIC_ANCHOR_REF(), \ + (message)); \ + loggingGuard__.TryEnter(); \ + THROW_ERROR_EXCEPTION( \ + ::NYT::EErrorCode::Fatal, \ + "Malformed request or incorrect state detected") \ + << ::NYT::TErrorAttribute("message", loggingGuard__.Commit())) \ + loggingGuard__ +#define YT_TLOG_ALERT_AND_THROW_IF(condition, message) if (condition) [[unlikely]] YT_TLOG_ALERT_AND_THROW(message) +#define YT_TLOG_ALERT_AND_THROW_UNLESS(condition, message) if (!(condition)) [[unlikely]] YT_TLOG_ALERT_AND_THROW(message) + +//////////////////////////////////////////////////////////////////////////////// } // namespace NYT::NLogging diff --git a/library/cpp/yt/logging/plain_text_formatter/formatter.cpp b/library/cpp/yt/logging/plain_text_formatter/formatter.cpp index ab2181113dd..85f0889fa9b 100644 --- a/library/cpp/yt/logging/plain_text_formatter/formatter.cpp +++ b/library/cpp/yt/logging/plain_text_formatter/formatter.cpp @@ -1,9 +1,13 @@ #include "formatter.h" +#include <library/cpp/yt/logging/structured_payload.h> + #include <library/cpp/yt/cpu_clock/clock.h> #include <library/cpp/yt/misc/port.h> +#include <variant> + #ifdef YT_USE_SSE42 #include <emmintrin.h> #include <pmmintrin.h> @@ -146,6 +150,37 @@ void FormatMessage(TBaseFormatter* out, TStringBuf message) } } +// Formats |Message (Key: Value, ...)|, with well-known tags (e.g. an error) appended +// after the |(...)| group. Well-known tags are always written last, so a single pass +// suffices. Every piece -- message, tag keys/values, and the newline separating a +// well-known tag -- goes through FormatMessage and is escaped, so the rendered payload +// stays on a single physical line (a newline is emitted as the literal "\n"). +void FormatPayload(TBaseFormatter* out, const TTaggedLogEventPayload& payload) +{ + TTaggedPayloadReader reader(payload); + FormatMessage(out, reader.ReadMessage()); + bool parenOpen = false; + while (auto tag = reader.TryReadTag()) { + if (tag->IsWellKnown) { + if (parenOpen) { + out->AppendChar(')'); + parenOpen = false; + } + FormatMessage(out, "\n"_sb); + FormatMessage(out, tag->Value); + } else { + out->AppendString(parenOpen ? ", "_sb : " ("_sb); + parenOpen = true; + FormatMessage(out, tag->Key); + out->AppendString(": "_sb); + FormatMessage(out, tag->Value); + } + } + if (parenOpen) { + out->AppendChar(')'); + } +} + //////////////////////////////////////////////////////////////////////////////// void TCachingDateFormatter::Format(TBaseFormatter* buffer, TInstant dateTime, bool printMicroseconds) @@ -186,7 +221,13 @@ void TPlainTextEventFormatter::Format(TBaseFormatter* buffer, const TLogEvent& e buffer->AppendChar('\t'); - FormatMessage(buffer, event.MessageRef.ToStringBuf()); + if (const auto* tagged = std::get_if<TTaggedLogEventPayload>(&event.Payload)) { + FormatPayload(buffer, *tagged); + } else { + // A structured event routed to a plain-text writer: emit its raw YSON fragment + // (escaped, so the record stays a single physical line). + FormatMessage(buffer, GetYsonFromStructuredPayload(std::get<TStructuredLogEventPayload>(event.Payload)).AsStringBuf()); + } buffer->AppendChar('\t'); diff --git a/library/cpp/yt/logging/private.h b/library/cpp/yt/logging/private.h new file mode 100644 index 00000000000..b80f0488048 --- /dev/null +++ b/library/cpp/yt/logging/private.h @@ -0,0 +1,15 @@ +#pragma once + +#include "public.h" + +namespace NYT::NLogging::NDetail { + +//////////////////////////////////////////////////////////////////////////////// + +//! Ref-counted memory tag for payload buffers; used for memory accounting. +struct TMessageBufferTag +{ }; + +//////////////////////////////////////////////////////////////////////////////// + +} // namespace NYT::NLogging::NDetail diff --git a/library/cpp/yt/logging/public.h b/library/cpp/yt/logging/public.h index 1e2b59ca0d3..17f8998f1af 100644 --- a/library/cpp/yt/logging/public.h +++ b/library/cpp/yt/logging/public.h @@ -1,6 +1,11 @@ #pragma once #include <library/cpp/yt/misc/enum.h> +#include <library/cpp/yt/misc/strong_typedef.h> + +#include <library/cpp/yt/memory/ref.h> + +#include <variant> namespace NYT::NLogging { @@ -36,4 +41,18 @@ struct ILogManager; //////////////////////////////////////////////////////////////////////////////// +//! Opaque payload of a plain-text log event: a message plus optional key/value tags, +//! framed by #TTaggedPayloadWriter. Produced by the YT_LOG_*/YT_TLOG_* macros. +YT_DEFINE_STRONG_TYPEDEF(TTaggedLogEventPayload, TSharedRef); + +//! Opaque payload of a structured log event: a raw YSON map fragment. +//! Produced by #LogStructuredEvent. +YT_DEFINE_STRONG_TYPEDEF(TStructuredLogEventPayload, TSharedRef); + +//! The payload carried by a #TLogEvent: exactly one of the encodings above. The active +//! alternative both identifies the event kind and determines how the payload is decoded. +using TLogEventPayload = std::variant<TTaggedLogEventPayload, TStructuredLogEventPayload>; + +//////////////////////////////////////////////////////////////////////////////// + } // namespace NYT::NLogging diff --git a/library/cpp/yt/logging/structured_payload.cpp b/library/cpp/yt/logging/structured_payload.cpp new file mode 100644 index 00000000000..1a6ce5ea370 --- /dev/null +++ b/library/cpp/yt/logging/structured_payload.cpp @@ -0,0 +1,19 @@ +#include "structured_payload.h" + +namespace NYT::NLogging { + +//////////////////////////////////////////////////////////////////////////////// + +TStructuredLogEventPayload MakeStructuredPayloadFromYson(const NYson::TYsonString& message) +{ + return TStructuredLogEventPayload(message.ToSharedRef()); +} + +NYson::TYsonStringBuf GetYsonFromStructuredPayload(const TStructuredLogEventPayload& payload) +{ + return NYson::TYsonStringBuf(payload.Underlying().ToStringBuf(), NYson::EYsonType::MapFragment); +} + +//////////////////////////////////////////////////////////////////////////////// + +} // namespace NYT::NLogging diff --git a/library/cpp/yt/logging/structured_payload.h b/library/cpp/yt/logging/structured_payload.h new file mode 100644 index 00000000000..81464f4aa92 --- /dev/null +++ b/library/cpp/yt/logging/structured_payload.h @@ -0,0 +1,25 @@ +#pragma once + +#include "public.h" + +#include <library/cpp/yt/memory/ref.h> + +#include <library/cpp/yt/yson_string/string.h> + +namespace NYT::NLogging { + +//////////////////////////////////////////////////////////////////////////////// + +// Structured events carry the raw YSON map fragment as their opaque payload, with no +// framing. These helpers isolate that representation from callers. + +//! Producer: wraps #message into a payload (zero-copy). +TStructuredLogEventPayload MakeStructuredPayloadFromYson(const NYson::TYsonString& message); + +//! Consumer: views the payload as the YSON map fragment it carries. The result views +//! into #payload, which must outlive it. +NYson::TYsonStringBuf GetYsonFromStructuredPayload(const TStructuredLogEventPayload& payload); + +//////////////////////////////////////////////////////////////////////////////// + +} // namespace NYT::NLogging diff --git a/library/cpp/yt/logging/tagged_payload-inl.h b/library/cpp/yt/logging/tagged_payload-inl.h new file mode 100644 index 00000000000..4b71a5bebef --- /dev/null +++ b/library/cpp/yt/logging/tagged_payload-inl.h @@ -0,0 +1,108 @@ +#ifndef TAGGED_PAYLOAD_INL_H_ +#error "Direct inclusion of this file is not allowed, include tagged_payload.h" +// For the sake of sane code completion. +#include "tagged_payload.h" +#endif + +#include <library/cpp/yt/assert/assert.h> + +#include <cstring> + +namespace NYT::NLogging { + +//////////////////////////////////////////////////////////////////////////////// + +inline TSharedRef TTaggedPayloadBuilder::Flush() +{ + return Buffer_.Slice(0, GetLength()); +} + +template <class T> +void TTaggedPayloadBuilder::AppendPod(const T& value) +{ + ::memcpy(Preallocate(sizeof(value)), &value, sizeof(value)); + Advance(sizeof(value)); +} + +template <class T> +void TTaggedPayloadBuilder::WritePodAt(size_t offset, const T& value) +{ + YT_ASSERT(offset + sizeof(value) <= GetLength()); + ::memcpy(Begin_ + offset, &value, sizeof(value)); +} + +//////////////////////////////////////////////////////////////////////////////// + +inline TStringBuilderBase* TTaggedPayloadWriter::BeginMessage() & +{ + YT_ASSERT(!MessageStarted_); + MessageStarted_ = true; + ReserveLengthPrefix(); + return &Builder_; +} + +inline TTaggedPayloadWriter& TTaggedPayloadWriter::EndMessage() & +{ + YT_ASSERT(MessageStarted_ && !MessageEnded_); + MessageEnded_ = true; + BackpatchLengthPrefix(); + return *this; +} + +inline TStringBuilderBase* TTaggedPayloadWriter::BeginTag(TStringBuf key) & +{ + return DoBeginTag(key, /*wellKnown*/ false); +} + +inline TStringBuilderBase* TTaggedPayloadWriter::BeginWellKnownTag(TStringBuf key) & +{ + return DoBeginTag(key, /*wellKnown*/ true); +} + +inline TStringBuilderBase* TTaggedPayloadWriter::DoBeginTag(TStringBuf key, bool wellKnown) +{ + YT_ASSERT(MessageEnded_ && !InTag_); + // High bit reserved for the well-known flag. + YT_ASSERT(key.size() < WellKnownTagFlag); + InTag_ = true; + // Reserve the key-size prefix, the key bytes, and the value-size prefix in a single + // capacity check; the value is then formatted in place and EndTag backpatches its size. + auto keySize = static_cast<ui32>(key.size()) | (wellKnown ? WellKnownTagFlag : 0); + char* ptr = Builder_.Preallocate(2 * sizeof(ui32) + key.size()); + PrefixOffset_ = Builder_.GetLength() + sizeof(ui32) + key.size(); + ::memcpy(ptr, &keySize, sizeof(keySize)); + ::memcpy(ptr + sizeof(keySize), key.data(), key.size()); + // The value-size prefix is left uninitialized; EndTag overwrites it. + Builder_.Advance(2 * sizeof(ui32) + key.size()); + return &Builder_; +} + +inline TTaggedPayloadWriter& TTaggedPayloadWriter::EndTag() & +{ + YT_ASSERT(InTag_); + InTag_ = false; + BackpatchLengthPrefix(); + return *this; +} + +inline TTaggedLogEventPayload TTaggedPayloadWriter::Finish() & +{ + YT_ASSERT(MessageEnded_ && !InTag_); + return TTaggedLogEventPayload(Builder_.Flush()); +} + +inline void TTaggedPayloadWriter::ReserveLengthPrefix() +{ + PrefixOffset_ = Builder_.GetLength(); + // Reserve a zeroed ui32; backpatched by BackpatchLengthPrefix. + Builder_.AppendPod<ui32>(0); +} + +inline void TTaggedPayloadWriter::BackpatchLengthPrefix() +{ + Builder_.WritePodAt<ui32>(PrefixOffset_, Builder_.GetLength() - PrefixOffset_ - sizeof(ui32)); +} + +//////////////////////////////////////////////////////////////////////////////// + +} // namespace NYT::NLogging diff --git a/library/cpp/yt/logging/tagged_payload.cpp b/library/cpp/yt/logging/tagged_payload.cpp new file mode 100644 index 00000000000..4133cc0bd55 --- /dev/null +++ b/library/cpp/yt/logging/tagged_payload.cpp @@ -0,0 +1,214 @@ +#include "tagged_payload.h" +#include "private.h" + +#include <library/cpp/yt/assert/assert.h> + +#include <library/cpp/yt/misc/tls.h> + +#include <util/system/compiler.h> +#include <util/system/types.h> + +#include <util/generic/bitops.h> + +#include <algorithm> + +namespace NYT::NLogging { + +//////////////////////////////////////////////////////////////////////////////// + +namespace { + +struct TPerThreadCache; + +YT_DEFINE_THREAD_LOCAL(TPerThreadCache*, Cache); +YT_DEFINE_THREAD_LOCAL(bool, CacheDestroyed); + +struct TPerThreadCache +{ + TSharedMutableRef Chunk; + size_t ChunkOffset = 0; + + ~TPerThreadCache() + { + TTaggedPayloadBuilder::DisablePerThreadCache(); + } + + static YT_PREVENT_TLS_CACHING TPerThreadCache* GetCache() + { + auto& cache = Cache(); + [[likely]] if (cache) { + return cache; + } + if (CacheDestroyed()) { + return nullptr; + } + static thread_local TPerThreadCache CacheData; + cache = &CacheData; + return cache; + } +}; + +TSharedMutableRef AllocateChunk(size_t size) +{ + return TSharedMutableRef::Allocate<::NYT::NLogging::NDetail::TMessageBufferTag>(size, {.InitializeStorage = false}); +} + +} // namespace + +//////////////////////////////////////////////////////////////////////////////// + +void TTaggedPayloadBuilder::DisablePerThreadCache() +{ + Cache() = nullptr; + CacheDestroyed() = true; +} + +void TTaggedPayloadBuilder::DoReset() +{ + Buffer_.Reset(); +} + +void TTaggedPayloadBuilder::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 = TPerThreadCache::GetCache(); + [[likely]] if (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; + } + + [[unlikely]] if (cache->ChunkOffset + newCapacity > cache->Chunk.Size()) { + cache->Chunk = AllocateChunk(newChunkSize); + cache->ChunkOffset = 0; + } + + Buffer_ = cache->Chunk.Slice(cache->ChunkOffset, cache->ChunkOffset + newCapacity); + cache->ChunkOffset += newCapacity; + } else { + Buffer_ = AllocateChunk(newChunkSize); + newCapacity = newChunkSize; + } + if (oldLength > 0) { + ::memcpy(Buffer_.Begin(), Begin_, oldLength); + } + Begin_ = Buffer_.Begin(); + End_ = Begin_ + newCapacity; +} + +//////////////////////////////////////////////////////////////////////////////// + +void TTaggedPayloadWriter::DisablePerThreadCache() +{ + TTaggedPayloadBuilder::DisablePerThreadCache(); +} + +//////////////////////////////////////////////////////////////////////////////// + +TTaggedPayloadReader::TTaggedPayloadReader(const TTaggedLogEventPayload& payload) + : Current_(payload.Underlying().Begin()) + , End_(payload.Underlying().End()) +{ } + +TStringBuf TTaggedPayloadReader::ReadMessage() +{ + YT_ASSERT(!MessageRead_); + MessageRead_ = true; + return ReadString(); +} + +std::optional<TTaggedPayloadReader::TTag> TTaggedPayloadReader::TryReadTag() +{ + YT_ASSERT(MessageRead_); + if (Current_ == End_) { + return std::nullopt; + } + auto keySize = ReadLength(); + bool wellKnown = (keySize & WellKnownTagFlag) != 0; + auto key = ReadBytes(keySize & ~WellKnownTagFlag); + auto value = ReadString(); + return TTag{.IsWellKnown = wellKnown, .Key = key, .Value = value}; +} + +ui32 TTaggedPayloadReader::ReadLength() +{ + YT_ASSERT(Current_ + sizeof(ui32) <= End_); + ui32 size; + ::memcpy(&size, Current_, sizeof(size)); + Current_ += sizeof(size); + return size; +} + +TStringBuf TTaggedPayloadReader::ReadBytes(ui32 size) +{ + YT_ASSERT(Current_ + size <= End_); + TStringBuf result(Current_, size); + Current_ += size; + return result; +} + +TStringBuf TTaggedPayloadReader::ReadString() +{ + return ReadBytes(ReadLength()); +} + +//////////////////////////////////////////////////////////////////////////////// + +TTaggedLogEventPayload MakeTaggedPayloadFromMessage(TStringBuf message) +{ + TTaggedPayloadWriter writer; + writer.BeginMessage()->AppendString(message); + writer.EndMessage(); + return writer.Finish(); +} + +TStringBuf GetMessageFromTaggedPayload(const TTaggedLogEventPayload& payload) +{ + return TTaggedPayloadReader(payload).ReadMessage(); +} + +std::string FormatTaggedPayload(const TTaggedLogEventPayload& payload) +{ + TTaggedPayloadReader reader(payload); + std::string result(reader.ReadMessage()); + // Well-known tags (e.g. an error) are always written last -- the fluent |.With| chain + // enforces this at the type level (see #TWellKnownTaggedLoggingGuard) -- so they render + // after the |(...)| group on trailing lines in a single pass. + bool parenOpen = false; + while (auto tag = reader.TryReadTag()) { + if (tag->IsWellKnown) { + if (parenOpen) { + result += ')'; + parenOpen = false; + } + result += '\n'; + result += tag->Value; + } else { + result += parenOpen ? ", " : " ("; + parenOpen = true; + result += tag->Key; + result += ": "; + result += tag->Value; + } + } + if (parenOpen) { + result += ')'; + } + return result; +} + +//////////////////////////////////////////////////////////////////////////////// + +} // namespace NYT::NLogging diff --git a/library/cpp/yt/logging/tagged_payload.h b/library/cpp/yt/logging/tagged_payload.h new file mode 100644 index 00000000000..da46f30f42a --- /dev/null +++ b/library/cpp/yt/logging/tagged_payload.h @@ -0,0 +1,201 @@ +#pragma once + +#include "public.h" + +#include <library/cpp/yt/string/string_builder.h> + +#include <library/cpp/yt/memory/ref.h> + +#include <util/generic/strbuf.h> +#include <util/generic/size_literals.h> + +#include <util/system/types.h> + +#include <optional> +#include <string> + +namespace NYT::NLogging { + +//////////////////////////////////////////////////////////////////////////////// + +// Plain-text events carry their message together with structured key/value tags in a +// flat, length-prefixed payload, deferring tag rendering to the consumer (the logging +// thread). These helpers own the wire format and isolate it from callers. + +//! A per-thread chunk-cached string builder backing payload buffers. +/*! + * Generic: it knows nothing about the payload layout (see #TTaggedPayloadWriter). + */ +class TTaggedPayloadBuilder + : public TStringBuilderBase +{ +public: + TSharedRef Flush(); + + //! Appends a trivially-copyable value as raw bytes (the ui32 length prefix). + template <class T> + void AppendPod(const T& value); + + //! Overwrites a trivially-copyable value at #offset. + //! #offset + sizeof(value) must not exceed the current length. + template <class T> + void WritePodAt(size_t offset, const T& value); + + //! For testing only. + static void DisablePerThreadCache(); + +private: + TSharedMutableRef Buffer_; + + static constexpr size_t ChunkSize = 128_KB - 64; + + void DoReset() override; + void DoReserve(size_t newCapacity) override; +}; + +//////////////////////////////////////////////////////////////////////////////// + +//! A log message together with its tag key/value pairs, serialized into a flat, +//! opaque byte payload that #TLogEvent hands to the logging thread. +/*! + * Layout (all sizes are host-endian |ui32|; the transport is in-process): + * \code + * [message size][message bytes] + * [key size][key bytes][value size][value bytes] x N + * \endcode + * + * A "well-known" tag sets the high bit of its key-size field; the low 31 bits stay the + * key length and the key bytes are still written. Consumers may render such tags + * specially (see #GetWellKnownLoggingTag). + * + * Both the producer (#TTaggedPayloadWriter) and the consumer + * (#TTaggedPayloadReader) own this single definition of the layout. + */ + +//! The high bit of a tag's key-size field, marking a well-known tag. +constexpr ui32 WellKnownTagFlag = 1u << 31; + +//////////////////////////////////////////////////////////////////////////////// + +//! Producer side: serializes the message and its tags into a payload. +/*! + * Owns a per-thread chunk-cached buffer; both the message text (#BeginMessage) and + * tag values (#BeginTag) are appended through the returned #TStringBuilderBase, so a + * formatted value is built directly into the payload without an intermediate + * allocation/copy. The length-prefix framing lives entirely here -- the underlying + * builder is format-agnostic. + */ +class TTaggedPayloadWriter +{ +public: + //! Constructs an empty writer; the message is built incrementally through the + //! builder returned by #BeginMessage. + TTaggedPayloadWriter() = default; + + TTaggedPayloadWriter(const TTaggedPayloadWriter&) = delete; + TTaggedPayloadWriter& operator=(const TTaggedPayloadWriter&) = delete; + + //! Begins the message field and returns a builder for appending its text. + //! Must be called exactly once, first. + TStringBuilderBase* BeginMessage() &; + + //! Ends the message field, filling in its reserved length prefix. Must be + //! called exactly once, after the message text and before any tag/#Finish. + TTaggedPayloadWriter& EndMessage() &; + + //! Begins a tag: writes #key, then reserves the value's length prefix and returns + //! a builder for appending the value. Lets the value be formatted directly into the + //! payload buffer. Pair with #EndTag. Must follow #EndMessage. + TStringBuilderBase* BeginTag(TStringBuf key) &; + + //! Like #BeginTag, but marks the tag well-known (sets #WellKnownTagFlag). + TStringBuilderBase* BeginWellKnownTag(TStringBuf key) &; + + //! Ends the current tag, filling in the value's reserved length prefix. + TTaggedPayloadWriter& EndTag() &; + + //! Returns the serialized payload. Must follow #EndMessage. + TTaggedLogEventPayload Finish() &; + + //! For testing only. + static void DisablePerThreadCache(); + +private: + TTaggedPayloadBuilder Builder_; + bool MessageStarted_ = false; + bool MessageEnded_ = false; + bool InTag_ = false; + //! Offset of the length prefix currently being filled (message or tag value). + size_t PrefixOffset_ = 0; + + //! Shared implementation of #BeginTag and #BeginWellKnownTag. + TStringBuilderBase* DoBeginTag(TStringBuf key, bool wellKnown); + //! Reserves a length prefix at the current position, remembering its offset. + void ReserveLengthPrefix(); + //! Backpatches the reserved prefix with the length appended after it. + void BackpatchLengthPrefix(); +}; + +//////////////////////////////////////////////////////////////////////////////// + +//! Consumer side: parses a payload produced by #TTaggedPayloadWriter. +/*! + * A single-pass cursor; the returned views point into #payload, which must outlive + * the reader. + * + * Call order: #ReadMessage must be called exactly once, first; then #TryReadTag + * must be called repeatedly until it returns |std::nullopt|, yielding the tags in + * write order. + */ +class TTaggedPayloadReader +{ +public: + struct TTag + { + bool IsWellKnown = false; + TStringBuf Key; + TStringBuf Value; + }; + + explicit TTaggedPayloadReader(const TTaggedLogEventPayload& payload); + + //! Reads the message field. Must be called exactly once, before any #TryReadTag. + TStringBuf ReadMessage(); + + //! Reads the next tag; returns |std::nullopt| once the tags are exhausted. + //! Must follow #ReadMessage. + std::optional<TTag> TryReadTag(); + +private: + const char* Current_; + const char* const End_; + bool MessageRead_ = false; + + //! Reads a ui32 length word. + ui32 ReadLength(); + //! Reads #size bytes as a view. + TStringBuf ReadBytes(ui32 size); + //! Reads a length-prefixed string (a length word followed by its bytes). + TStringBuf ReadString(); +}; + +//////////////////////////////////////////////////////////////////////////////// + +//! Producer convenience: builds a payload carrying only #message and no tags. +TTaggedLogEventPayload MakeTaggedPayloadFromMessage(TStringBuf message); + +//! Consumer convenience: returns the message, disregarding any tags. The result +//! views into #payload, which must outlive it. +TStringBuf GetMessageFromTaggedPayload(const TTaggedLogEventPayload& payload); + +//! Consumer convenience: renders the payload as |Message (Key: Value, ...)|, or just +//! the message if it carries no tags. +std::string FormatTaggedPayload(const TTaggedLogEventPayload& payload); + +//////////////////////////////////////////////////////////////////////////////// + +} // namespace NYT::NLogging + +#define TAGGED_PAYLOAD_INL_H_ +#include "tagged_payload-inl.h" +#undef TAGGED_PAYLOAD_INL_H_ diff --git a/library/cpp/yt/logging/unittests/helpers.cpp b/library/cpp/yt/logging/unittests/helpers.cpp new file mode 100644 index 00000000000..2a2b20120cc --- /dev/null +++ b/library/cpp/yt/logging/unittests/helpers.cpp @@ -0,0 +1,27 @@ +#include "helpers.h" + +namespace NYT::NLogging { + +//////////////////////////////////////////////////////////////////////////////// + +void WriteMessage(TTaggedPayloadWriter* writer, TStringBuf message) +{ + writer->BeginMessage()->AppendString(message); + writer->EndMessage(); +} + +void WriteTag(TTaggedPayloadWriter* writer, TStringBuf key, TStringBuf value) +{ + writer->BeginTag(key)->AppendString(value); + writer->EndTag(); +} + +void WriteWellKnownTag(TTaggedPayloadWriter* writer, TStringBuf key, TStringBuf value) +{ + writer->BeginWellKnownTag(key)->AppendString(value); + writer->EndTag(); +} + +//////////////////////////////////////////////////////////////////////////////// + +} // namespace NYT::NLogging diff --git a/library/cpp/yt/logging/unittests/helpers.h b/library/cpp/yt/logging/unittests/helpers.h new file mode 100644 index 00000000000..6785a65f99b --- /dev/null +++ b/library/cpp/yt/logging/unittests/helpers.h @@ -0,0 +1,22 @@ +#pragma once + +#include <library/cpp/yt/logging/tagged_payload.h> + +#include <util/generic/strbuf.h> + +namespace NYT::NLogging { + +//////////////////////////////////////////////////////////////////////////////// + +//! Writes the message (equivalent to BeginMessage, an append, and EndMessage). +void WriteMessage(TTaggedPayloadWriter* writer, TStringBuf message); + +//! Writes a tag with a ready string value (equivalent to BeginTag, an append, and EndTag). +void WriteTag(TTaggedPayloadWriter* writer, TStringBuf key, TStringBuf value); + +//! Writes a well-known tag with a ready string value (see BeginWellKnownTag). +void WriteWellKnownTag(TTaggedPayloadWriter* writer, TStringBuf key, TStringBuf value); + +//////////////////////////////////////////////////////////////////////////////// + +} // namespace NYT::NLogging diff --git a/library/cpp/yt/logging/unittests/logger_ut.cpp b/library/cpp/yt/logging/unittests/logger_ut.cpp index 410a56fc309..db176ca5394 100644 --- a/library/cpp/yt/logging/unittests/logger_ut.cpp +++ b/library/cpp/yt/logging/unittests/logger_ut.cpp @@ -1,15 +1,117 @@ #include <library/cpp/testing/gtest/gtest.h> #include <library/cpp/yt/logging/logger.h> +#include <library/cpp/yt/logging/tagged_payload.h> +#include <library/cpp/yt/logging/structured_payload.h> #include <library/cpp/yt/error/error.h> +#include <library/cpp/yt/yson_string/string.h> + +#include <atomic> +#include <string> +#include <utility> +#include <vector> + namespace NYT::NLogging { namespace { +using namespace NDetail; + +//////////////////////////////////////////////////////////////////////////////// + +//! A log manager that captures enqueued events for inspection in tests. +class TMockLogManager + : public ILogManager +{ +public: + explicit TMockLogManager(ELogLevel minLevel = ELogLevel::Minimum) + { + Category_.MinPlainTextLevel.store(minLevel); + } + + void RegisterStaticAnchor(TLoggingAnchor* anchor, ::TSourceLocation /*sourceLocation*/, TStringBuf /*message*/) override + { + anchor->Registered.store(true); + } + + void UpdateAnchor(TLoggingAnchor* anchor) override + { + anchor->CurrentVersion.store(ActualVersion_.load()); + } + + void Enqueue(TLogEvent&& event) override + { + Events_.push_back(std::move(event)); + } + + const TLoggingCategory* GetCategory(TStringBuf /*categoryName*/) override + { + return &Category_; + } + + void UpdateCategory(TLoggingCategory* category) override + { + category->CurrentVersion.store(ActualVersion_.load()); + } + + bool GetAbortOnAlert() const override + { + return false; + } + + const std::vector<TLogEvent>& GetEvents() const + { + return Events_; + } + +private: + std::vector<TLogEvent> Events_; + std::atomic<int> ActualVersion_ = 1; + TLoggingCategory Category_ = { + .Name = "Test", + .MinPlainTextLevel = ELogLevel::Minimum, + .CurrentVersion = 0, + .ActualVersion = &ActualVersion_, + }; +}; + +struct TDecodedEvent +{ + std::string Message; + std::vector<std::pair<std::string, std::string>> Tags; +}; + +TDecodedEvent DecodeEvent(const TLogEvent& event) +{ + TTaggedPayloadReader reader(std::get<TTaggedLogEventPayload>(event.Payload)); + TDecodedEvent result; + result.Message = reader.ReadMessage(); + while (auto tag = reader.TryReadTag()) { + result.Tags.emplace_back(tag->Key, tag->Value); + } + return result; +} + +TDecodedEvent DecodeSingleEvent(const TMockLogManager& manager) +{ + EXPECT_EQ(manager.GetEvents().size(), 1u); + return DecodeEvent(manager.GetEvents()[0]); +} + +NYson::TYsonString MakeMapFragment(TStringBuf yson) +{ + return NYson::TYsonString(yson, NYson::EYsonType::MapFragment); +} + +TStringBuf GetStructuredYson(const TLogEvent& event) +{ + return GetYsonFromStructuredPayload(std::get<TStructuredLogEventPayload>(event.Payload)).AsStringBuf(); +} + //////////////////////////////////////////////////////////////////////////////// -TEST(TLogger, NullByDefault) +TEST(TLoggerTest, NullByDefault) { { TLogger logger; @@ -23,7 +125,7 @@ TEST(TLogger, NullByDefault) } } -TEST(TLogger, CopyOfNullLogger) +TEST(TLoggerTest, CopyOfNullLogger) { TLogger nullLogger{/*logManager*/ nullptr, "Category"}; ASSERT_FALSE(nullLogger); @@ -34,7 +136,7 @@ TEST(TLogger, CopyOfNullLogger) EXPECT_FALSE(logger.IsLevelEnabled(ELogLevel::Fatal)); } -TEST(TLogger, LogAlertAndThrowMessage) +TEST(TLoggerTest, LogAlertAndThrowMessage) { try { TLogger Logger; @@ -52,5 +154,228 @@ TEST(TLogger, LogAlertAndThrowMessage) //////////////////////////////////////////////////////////////////////////////// +TEST(TTaggedApiTest, MessageOnly) +{ + TMockLogManager manager; + TLogger Logger(&manager, "Test"); + YT_TLOG_INFO("Message"); + + auto decoded = DecodeSingleEvent(manager); + EXPECT_EQ(decoded.Message, "Message"); + EXPECT_TRUE(decoded.Tags.empty()); +} + +TEST(TTaggedApiTest, Tags) +{ + TMockLogManager manager; + TLogger Logger(&manager, "Test"); + YT_TLOG_INFO("Message") + .With("Arg1", 123) + .With("Arg2", "test"); + + auto decoded = DecodeSingleEvent(manager); + EXPECT_EQ(decoded.Message, "Message"); + ASSERT_EQ(decoded.Tags.size(), 2u); + EXPECT_EQ(decoded.Tags[0], std::pair(std::string("Arg1"), std::string("123"))); + EXPECT_EQ(decoded.Tags[1], std::pair(std::string("Arg2"), std::string("test"))); +} + +TEST(TTaggedApiTest, CustomSpec) +{ + TMockLogManager manager; + TLogger Logger(&manager, "Test"); + YT_TLOG_INFO("Message") + .With("Arg1", 256, "%x"); + + auto decoded = DecodeSingleEvent(manager); + ASSERT_EQ(decoded.Tags.size(), 1u); + EXPECT_EQ(decoded.Tags[0], std::pair(std::string("Arg1"), std::string("100"))); +} + +TEST(TTaggedApiTest, LoggerTagFoldedIntoMessage) +{ + TMockLogManager manager; + auto Logger = TLogger(&manager, "Test") + .WithTag("LoggingTag: %v", 555); + YT_TLOG_INFO("Message") + .With("Arg1", 123); + + auto decoded = DecodeSingleEvent(manager); + EXPECT_EQ(decoded.Message, "Message (LoggingTag: 555)"); + ASSERT_EQ(decoded.Tags.size(), 1u); + EXPECT_EQ(decoded.Tags[0], std::pair(std::string("Arg1"), std::string("123"))); +} + +TEST(TTaggedApiTest, DisabledDoesNotEvaluateTags) +{ + TMockLogManager manager(/*minLevel*/ ELogLevel::Warning); + TLogger Logger(&manager, "Test"); + + int evaluated = 0; + auto evaluate = [&] { + ++evaluated; + return 123; + }; + YT_TLOG_INFO("Message") + .With("Arg1", evaluate()); + + EXPECT_EQ(evaluated, 0); + EXPECT_TRUE(manager.GetEvents().empty()); +} + +TEST(TTaggedApiTest, WellKnownErrorTag) +{ + TMockLogManager manager; + TLogger Logger(&manager, "Test"); + auto error = TError("boom"); + YT_TLOG_INFO("Message") + .With("Tag", 1) + .With(error); + + ASSERT_EQ(manager.GetEvents().size(), 1u); + TTaggedPayloadReader reader(std::get<TTaggedLogEventPayload>(manager.GetEvents()[0].Payload)); + EXPECT_EQ(reader.ReadMessage(), "Message"); + + auto regular = reader.TryReadTag(); + ASSERT_TRUE(regular); + EXPECT_EQ(regular->Key, "Tag"); + EXPECT_EQ(regular->Value, "1"); + EXPECT_FALSE(regular->IsWellKnown); + + // |.With(error)| attaches the error under the well-known "Error" key (resolved via + // GetWellKnownLoggingTag), with the formatted error as the value. + auto errorTag = reader.TryReadTag(); + ASSERT_TRUE(errorTag); + EXPECT_EQ(errorTag->Key, "Error"); + EXPECT_TRUE(errorTag->IsWellKnown); + EXPECT_NE(errorTag->Value.find("boom"), TStringBuf::npos); + + EXPECT_FALSE(reader.TryReadTag()); +} + +// The single-pass formatter assumes well-known tags come last, so the fluent API must +// forbid a keyed tag after a well-known one (e.g. |.With(error).With("Key", 1)|) at +// compile time. +template <class TGuard> +concept CAllowsKeyedTagAfterWellKnown = requires (TGuard guard, TError error) { + guard.With(error).With("Key", 1); +}; + +// A further well-known tag after a well-known one stays allowed. +template <class TGuard> +concept CAllowsWellKnownTagAfterWellKnown = requires (TGuard guard, TError error) { + guard.With(error).With(error); +}; + +static_assert(!CAllowsKeyedTagAfterWellKnown<TTaggedLoggingGuard>); +static_assert(CAllowsWellKnownTagAfterWellKnown<TTaggedLoggingGuard>); + +TEST(TTaggedApiTest, AlertAndThrow) +{ + TMockLogManager manager; + TLogger Logger(&manager, "Test"); + + try { + YT_TLOG_ALERT_AND_THROW("Alert message") + .With("Arg1", 1) + .With("Arg2", 2); + EXPECT_TRUE(false); + } catch (const TErrorException& ex) { + const auto& error = ex.Error(); + EXPECT_EQ(error.GetCode(), NYT::EErrorCode::Fatal); + EXPECT_EQ(error.GetMessage(), "Malformed request or incorrect state detected"); + EXPECT_EQ(error.Attributes().Get<std::string>("message"), "Alert message"); + } + + // The alert is also logged, carrying the structured tags. + auto decoded = DecodeSingleEvent(manager); + EXPECT_EQ(decoded.Message, "Alert message"); + ASSERT_EQ(decoded.Tags.size(), 2u); + EXPECT_EQ(decoded.Tags[0], std::pair(std::string("Arg1"), std::string("1"))); + EXPECT_EQ(decoded.Tags[1], std::pair(std::string("Arg2"), std::string("2"))); +} + +TEST(TTaggedApiTest, AlertAndThrowDisabledStillThrows) +{ + TMockLogManager manager(/*minLevel*/ ELogLevel::Fatal); + TLogger Logger(&manager, "Test"); + + // The level is disabled, so nothing is logged, but the exception (with its message) + // is still raised. + EXPECT_THROW( + YT_TLOG_ALERT_AND_THROW("Alert message") + .With("Arg1", 1), + TErrorException); + EXPECT_TRUE(manager.GetEvents().empty()); +} + +TEST(TTaggedApiDeathTest, Fatal) +{ + EXPECT_DEATH({ + TMockLogManager manager; + TLogger Logger(&manager, "Test"); + YT_TLOG_FATAL("Fatal message") + .With("Arg1", 1); + }, "Fatal message \\(Arg1: 1\\)"); +} + +//////////////////////////////////////////////////////////////////////////////// + +TEST(TStructuredApiTest, StructuredEvent) +{ + TMockLogManager manager; + TLogger Logger(&manager, "Test"); + + LogStructuredEvent(Logger, MakeMapFragment("\"key\"=\"value\""), ELogLevel::Info); + + ASSERT_EQ(manager.GetEvents().size(), 1u); + const auto& event = manager.GetEvents()[0]; + EXPECT_TRUE(std::holds_alternative<TStructuredLogEventPayload>(event.Payload)); + EXPECT_EQ(event.Family, ELogFamily::Structured); + EXPECT_EQ(event.Level, ELogLevel::Info); + // The payload is the raw YSON map fragment. + EXPECT_EQ(GetStructuredYson(event), "\"key\"=\"value\""); +} + +TEST(TStructuredApiTest, PreservesLevel) +{ + TMockLogManager manager; + TLogger Logger(&manager, "Test"); + + LogStructuredEvent(Logger, MakeMapFragment("\"a\"=1;\"b\"=2"), ELogLevel::Warning); + + ASSERT_EQ(manager.GetEvents().size(), 1u); + EXPECT_EQ(manager.GetEvents()[0].Level, ELogLevel::Warning); + EXPECT_EQ(GetStructuredYson(manager.GetEvents()[0]), "\"a\"=1;\"b\"=2"); +} + +TEST(TStructuredApiTest, EmptyFragment) +{ + TMockLogManager manager; + TLogger Logger(&manager, "Test"); + + LogStructuredEvent(Logger, MakeMapFragment(""), ELogLevel::Info); + + ASSERT_EQ(manager.GetEvents().size(), 1u); + EXPECT_EQ(GetStructuredYson(manager.GetEvents()[0]), ""); +} + +TEST(TStructuredApiTest, MultipleEvents) +{ + TMockLogManager manager; + TLogger Logger(&manager, "Test"); + + LogStructuredEvent(Logger, MakeMapFragment("\"i\"=0"), ELogLevel::Debug); + LogStructuredEvent(Logger, MakeMapFragment("\"i\"=1"), ELogLevel::Info); + + ASSERT_EQ(manager.GetEvents().size(), 2u); + EXPECT_EQ(manager.GetEvents()[0].Level, ELogLevel::Debug); + EXPECT_EQ(GetStructuredYson(manager.GetEvents()[0]), "\"i\"=0"); + EXPECT_EQ(manager.GetEvents()[1].Level, ELogLevel::Info); + EXPECT_EQ(GetStructuredYson(manager.GetEvents()[1]), "\"i\"=1"); +} + +//////////////////////////////////////////////////////////////////////////////// + } // namespace } // namespace NYT::NLogging diff --git a/library/cpp/yt/logging/unittests/structured_payload_ut.cpp b/library/cpp/yt/logging/unittests/structured_payload_ut.cpp new file mode 100644 index 00000000000..f15970dcb23 --- /dev/null +++ b/library/cpp/yt/logging/unittests/structured_payload_ut.cpp @@ -0,0 +1,36 @@ +#include <library/cpp/testing/gtest/gtest.h> + +#include <library/cpp/yt/logging/structured_payload.h> + +#include <library/cpp/yt/yson_string/string.h> + +namespace NYT::NLogging { +namespace { + +using namespace NYson; + +//////////////////////////////////////////////////////////////////////////////// + +TYsonString MakeMapFragment(TStringBuf yson) +{ + return TYsonString(yson, EYsonType::MapFragment); +} + +TEST(TStructuredPayloadTest, RoundTrip) +{ + auto payload = MakeStructuredPayloadFromYson(MakeMapFragment("\"key\"=\"value\"")); + auto view = GetYsonFromStructuredPayload(payload); + EXPECT_EQ(view.GetType(), EYsonType::MapFragment); + EXPECT_EQ(view.AsStringBuf(), "\"key\"=\"value\""); +} + +TEST(TStructuredPayloadTest, EmptyFragment) +{ + auto payload = MakeStructuredPayloadFromYson(MakeMapFragment("")); + EXPECT_EQ(GetYsonFromStructuredPayload(payload).AsStringBuf(), ""); +} + +//////////////////////////////////////////////////////////////////////////////// + +} // namespace +} // namespace NYT::NLogging diff --git a/library/cpp/yt/logging/unittests/tagged_payload_ut.cpp b/library/cpp/yt/logging/unittests/tagged_payload_ut.cpp new file mode 100644 index 00000000000..57d9ac168f4 --- /dev/null +++ b/library/cpp/yt/logging/unittests/tagged_payload_ut.cpp @@ -0,0 +1,146 @@ +#include "helpers.h" + +#include <library/cpp/testing/gtest/gtest.h> + +#include <library/cpp/yt/logging/tagged_payload.h> + +#include <string> +#include <vector> + +namespace NYT::NLogging { +namespace { + +using namespace NDetail; + +//////////////////////////////////////////////////////////////////////////////// + +struct TDecodedPayload +{ + std::string Message; + std::vector<std::pair<std::string, std::string>> Tags; +}; + +TTaggedLogEventPayload Encode(TStringBuf message, const std::vector<std::pair<TStringBuf, TStringBuf>>& tags = {}) +{ + TTaggedPayloadWriter writer; + WriteMessage(&writer, message); + for (auto [key, value] : tags) { + WriteTag(&writer, key, value); + } + return writer.Finish(); +} + +TDecodedPayload Decode(const TTaggedLogEventPayload& payload) +{ + TTaggedPayloadReader reader(payload); + TDecodedPayload result; + result.Message = reader.ReadMessage(); + while (auto tag = reader.TryReadTag()) { + result.Tags.emplace_back(tag->Key, tag->Value); + } + return result; +} + +TEST(TTaggedPayloadTest, MessageOnly) +{ + auto decoded = Decode(Encode("Hello")); + EXPECT_EQ(decoded.Message, "Hello"); + EXPECT_TRUE(decoded.Tags.empty()); +} + +TEST(TTaggedPayloadTest, EmptyMessage) +{ + auto decoded = Decode(Encode("")); + EXPECT_EQ(decoded.Message, ""); + EXPECT_TRUE(decoded.Tags.empty()); +} + +TEST(TTaggedPayloadTest, OneTag) +{ + auto decoded = Decode(Encode("Message", {{"Key", "Value"}})); + EXPECT_EQ(decoded.Message, "Message"); + ASSERT_EQ(decoded.Tags.size(), 1u); + EXPECT_EQ(decoded.Tags[0].first, "Key"); + EXPECT_EQ(decoded.Tags[0].second, "Value"); +} + +TEST(TTaggedPayloadTest, ManyTags) +{ + auto decoded = Decode(Encode("Message", {{"Arg1", "123"}, {"Arg2", "test"}, {"Arg3", ""}})); + EXPECT_EQ(decoded.Message, "Message"); + ASSERT_EQ(decoded.Tags.size(), 3u); + EXPECT_EQ(decoded.Tags[0], std::pair(std::string("Arg1"), std::string("123"))); + EXPECT_EQ(decoded.Tags[1], std::pair(std::string("Arg2"), std::string("test"))); + EXPECT_EQ(decoded.Tags[2], std::pair(std::string("Arg3"), std::string(""))); +} + +TEST(TTaggedPayloadTest, BinarySafeValues) +{ + // Keys/values may carry arbitrary bytes, including embedded NULs and delimiters. + std::string message("a\0b ()", 6); + std::string key("k\0:", 3); + std::string value("v\0, ", 4); + + auto decoded = Decode(Encode(message, {{key, value}})); + EXPECT_EQ(decoded.Message, message); + ASSERT_EQ(decoded.Tags.size(), 1u); + EXPECT_EQ(decoded.Tags[0].first, key); + EXPECT_EQ(decoded.Tags[0].second, value); +} + +TEST(TTaggedPayloadTest, ReaderViewsPointIntoPayload) +{ + auto payload = Encode("Message", {{"Key", "Value"}}); + + auto pointsInto = [&] (TStringBuf view) { + return view.data() >= payload.Underlying().Begin() && view.data() + view.size() <= payload.Underlying().End(); + }; + + TTaggedPayloadReader reader(payload); + EXPECT_TRUE(pointsInto(reader.ReadMessage())); + auto tag = reader.TryReadTag(); + ASSERT_TRUE(tag.has_value()); + EXPECT_TRUE(pointsInto(tag->Key)); + EXPECT_TRUE(pointsInto(tag->Value)); +} + +TEST(TTaggedPayloadTest, WellKnownTag) +{ + TTaggedPayloadWriter writer; + WriteMessage(&writer, "Message"); + WriteTag(&writer, "Key", "Value"); + WriteWellKnownTag(&writer, "Error", "boom"); + auto payload = writer.Finish(); + + TTaggedPayloadReader reader(payload); + EXPECT_EQ(reader.ReadMessage(), "Message"); + + auto regular = reader.TryReadTag(); + ASSERT_TRUE(regular); + EXPECT_EQ(regular->Key, "Key"); + EXPECT_EQ(regular->Value, "Value"); + EXPECT_FALSE(regular->IsWellKnown); + + auto wellKnown = reader.TryReadTag(); + ASSERT_TRUE(wellKnown); + EXPECT_EQ(wellKnown->Key, "Error"); + EXPECT_EQ(wellKnown->Value, "boom"); + EXPECT_TRUE(wellKnown->IsWellKnown); + + EXPECT_FALSE(reader.TryReadTag()); +} + +TEST(TTaggedPayloadTest, FormatWellKnownTagTrailing) +{ + TTaggedPayloadWriter writer; + WriteMessage(&writer, "Message"); + WriteTag(&writer, "Key", "Value"); + WriteWellKnownTag(&writer, "Error", "boom"); + // Regular tags stay inline; the well-known tag is appended after the |(...)| group. + EXPECT_EQ(FormatTaggedPayload(writer.Finish()), "Message (Key: Value)\nboom"); +} + +//////////////////////////////////////////////////////////////////////////////// + +} // namespace +} // namespace NYT::NLogging diff --git a/library/cpp/yt/logging/unittests/ya.make b/library/cpp/yt/logging/unittests/ya.make index 01946699bb2..cc068fa24c6 100644 --- a/library/cpp/yt/logging/unittests/ya.make +++ b/library/cpp/yt/logging/unittests/ya.make @@ -3,7 +3,10 @@ GTEST(unittester-library-logging) INCLUDE(${ARCADIA_ROOT}/library/cpp/yt/ya_cpp.make.inc) SRCS( + helpers.cpp logger_ut.cpp + tagged_payload_ut.cpp + structured_payload_ut.cpp static_analysis_ut.cpp ) diff --git a/library/cpp/yt/logging/ya.make b/library/cpp/yt/logging/ya.make index b9cf5c6e715..d839a6f07b7 100644 --- a/library/cpp/yt/logging/ya.make +++ b/library/cpp/yt/logging/ya.make @@ -4,6 +4,8 @@ INCLUDE(${ARCADIA_ROOT}/library/cpp/yt/ya_cpp.make.inc) SRCS( logger.cpp + tagged_payload.cpp + structured_payload.cpp ) PEERDIR( diff --git a/library/cpp/yt/memory/atomic_intrusive_ptr-inl.h b/library/cpp/yt/memory/atomic_intrusive_ptr-inl.h index d91d20f2fd6..4211806880f 100644 --- a/library/cpp/yt/memory/atomic_intrusive_ptr-inl.h +++ b/library/cpp/yt/memory/atomic_intrusive_ptr-inl.h @@ -3,7 +3,6 @@ // For the sake of sane code completion. #include "atomic_intrusive_ptr.h" #endif -#undef ATOMIC_INTRUSIVE_PTR_INL_H_ #include <util/system/spinlock.h> diff --git a/library/cpp/yt/memory/ref.cpp b/library/cpp/yt/memory/ref.cpp index 8601c2ba3e1..eddae004ca6 100644 --- a/library/cpp/yt/memory/ref.cpp +++ b/library/cpp/yt/memory/ref.cpp @@ -380,11 +380,6 @@ TSharedRef TSharedRef::FromStringImpl(TString str, TRefCountedTypeCookie tagCook return TSharedRef(ref, std::move(holder)); } -TSharedRef TSharedRef::FromString(const char* str) -{ - return FromString(std::string(str)); -} - TSharedRef TSharedRef::FromBlob(TBlob&& blob) { auto ref = TRef::FromBlob(blob); diff --git a/library/cpp/yt/memory/ref.h b/library/cpp/yt/memory/ref.h index 1fc451f50aa..91bce3fde6a 100644 --- a/library/cpp/yt/memory/ref.h +++ b/library/cpp/yt/memory/ref.h @@ -162,9 +162,6 @@ public: //! Same as above but the memory tag is specified in #tagCookie. static TSharedRef FromString(std::string str, TRefCountedTypeCookie tagCookie); - //! Creates a TSharedRef from a zero-terminated C string. - static TSharedRef FromString(const char* str); - //! Creates a TSharedRef for a given blob taking ownership of its content. static TSharedRef FromBlob(TBlob&& blob); diff --git a/library/cpp/yt/memory/ref_counted-inl.h b/library/cpp/yt/memory/ref_counted-inl.h index 94acfc6669b..ce3c686ee28 100644 --- a/library/cpp/yt/memory/ref_counted-inl.h +++ b/library/cpp/yt/memory/ref_counted-inl.h @@ -124,6 +124,10 @@ Y_FORCE_INLINE void DestroyRefCountedImpl(T* obj) // No virtual call when T is final. obj->~T(); + // The ref-counter is a base subobject of *obj, so obj->~T() above poisons it + // under -fsanitize-memory-use-after-dtor. Unpoison before accessing it. + NSan::Unpoison(refCounter, sizeof(TRefCounter)); + // Fast path. Weak refs cannot appear if there are neither strong nor weak refs. if (refCounter->GetWeakRefCount() == 1) { NYT::NDetail::TMemoryReleaser<T>::Do(obj, offset); diff --git a/library/cpp/yt/memory/unittests/ref_ut.cpp b/library/cpp/yt/memory/unittests/ref_ut.cpp index 734fc030d29..175ca2bb00e 100644 --- a/library/cpp/yt/memory/unittests/ref_ut.cpp +++ b/library/cpp/yt/memory/unittests/ref_ut.cpp @@ -13,7 +13,7 @@ namespace { TEST(TSharedRefTest, Save) { - const TSharedRef expected = TSharedRef::FromString("My tests data"); + const TSharedRef expected = TSharedRef::FromString(std::string("My tests data")); TStringStream s; ::Save(&s, expected); // only Save supported for TSharedRef. You can ::Load serialized data to vector. } diff --git a/library/cpp/yt/memory/weak_ptr-inl.h b/library/cpp/yt/memory/weak_ptr-inl.h index 11026dc5fc6..0e5ad795774 100644 --- a/library/cpp/yt/memory/weak_ptr-inl.h +++ b/library/cpp/yt/memory/weak_ptr-inl.h @@ -4,6 +4,8 @@ #include "weak_ptr.h" #endif +#include <util/system/sanitizers.h> + namespace NYT { //////////////////////////////////////////////////////////////////////////////// @@ -16,8 +18,13 @@ template <class T> TWeakPtr<T>::TWeakPtr(T* p) noexcept : T_(p) { -#if defined(_tsan_enabled_) +#if defined(_tsan_enabled_) || defined(_msan_enabled_) if (T_) { + // p may legitimately point to an object whose destructor has already run + // while its ref-counter is still alive (weak count > 0). When T derives + // from TRefCounted virtually, the upcast below reads p's vptr, poisoned + // by -fsanitize-memory-use-after-dtor. Unpoison it (no-op for a live object). + NSan::Unpoison(T_, sizeof(void*)); RefCounter_ = GetRefCounter(T_); } #endif @@ -42,7 +49,7 @@ TWeakPtr<T>::TWeakPtr(const TIntrusivePtr<U>& ptr) noexcept template <class T> TWeakPtr<T>::TWeakPtr(const TWeakPtr& other) noexcept : T_(other.T_) -#if defined(_tsan_enabled_) +#if defined(_tsan_enabled_) || defined(_msan_enabled_) , RefCounter_(other.RefCounter_) #endif { @@ -77,7 +84,7 @@ TWeakPtr<T>::TWeakPtr(TWeakPtr<U>&& other) noexcept T_ = other.T_; other.T_ = nullptr; -#if defined(_tsan_enabled_) +#if defined(_tsan_enabled_) || defined(_msan_enabled_) RefCounter_ = other.RefCounter_; other.RefCounter_ = nullptr; #endif @@ -163,7 +170,7 @@ template <class T> void TWeakPtr<T>::Swap(TWeakPtr& other) noexcept { DoSwap(T_, other.T_); -#if defined(_tsan_enabled_) +#if defined(_tsan_enabled_) || defined(_msan_enabled_) DoSwap(RefCounter_, other.RefCounter_); #endif } @@ -208,12 +215,17 @@ void TWeakPtr<T>::ReleaseRef() if (T_) { // Support incomplete type. if (GetRefCounterImpl()->WeakUnref()) { + // The object's destructor has fully completed at this point + // and has poisoned T_'s vptr word under -fsanitize-memory-use-after-dtor. + // The upcast inside DeallocateRefCounted still needs that word + // to locate the virtual base. + NSan::Unpoison(T_, sizeof(void*)); DeallocateRefCounted(T_); } } } -#if defined(_tsan_enabled_) +#if defined(_tsan_enabled_) || defined(_msan_enabled_) template <class T> const TRefCounter* TWeakPtr<T>::GetRefCounterImpl() const { diff --git a/library/cpp/yt/memory/weak_ptr.h b/library/cpp/yt/memory/weak_ptr.h index d3598f62c74..c98abe15194 100644 --- a/library/cpp/yt/memory/weak_ptr.h +++ b/library/cpp/yt/memory/weak_ptr.h @@ -101,7 +101,7 @@ private: friend class TWeakPtr; T* T_ = nullptr; -#if defined(_tsan_enabled_) +#if defined(_tsan_enabled_) || defined(_msan_enabled_) const TRefCounter* RefCounter_ = nullptr; #endif diff --git a/library/cpp/yt/system/cpu_id-inl.h b/library/cpp/yt/system/cpu_id-inl.h index e336ba5a2ce..804995f5973 100644 --- a/library/cpp/yt/system/cpu_id-inl.h +++ b/library/cpp/yt/system/cpu_id-inl.h @@ -3,7 +3,6 @@ // For the sake of sane code completion. #include "cpu_id.h" #endif -#undef CPU_ID_INL_H_ #ifdef __linux__ #include <library/cpp/yt/rseq/rseq.h> diff --git a/library/cpp/yt/threading/event_count-inl.h b/library/cpp/yt/threading/event_count-inl.h index 18f7bfec52f..b366f93bb0f 100644 --- a/library/cpp/yt/threading/event_count-inl.h +++ b/library/cpp/yt/threading/event_count-inl.h @@ -3,7 +3,6 @@ // For the sake of sane code completion. #include "event_count.h" #endif -#undef EVENT_COUNT_INL_H_ #include <library/cpp/yt/assert/assert.h> diff --git a/library/cpp/yt/threading/recursive_spin_lock-inl.h b/library/cpp/yt/threading/recursive_spin_lock-inl.h index f692999f0b1..28b233efa94 100644 --- a/library/cpp/yt/threading/recursive_spin_lock-inl.h +++ b/library/cpp/yt/threading/recursive_spin_lock-inl.h @@ -3,7 +3,6 @@ // For the sake of sane code completion. #include "recursive_spin_lock.h" #endif -#undef RECURSIVE_SPIN_LOCK_INL_H_ #include "spin_wait.h" diff --git a/library/cpp/yt/threading/rw_spin_lock-inl.h b/library/cpp/yt/threading/rw_spin_lock-inl.h index c6043e8f23a..167c95edfc5 100644 --- a/library/cpp/yt/threading/rw_spin_lock-inl.h +++ b/library/cpp/yt/threading/rw_spin_lock-inl.h @@ -4,7 +4,6 @@ // For the sake of sane code completion. #include "rw_spin_lock.h" #endif -#undef RW_SPIN_LOCK_INL_H_ #include "spin_wait.h" diff --git a/library/cpp/yt/threading/spin_lock-inl.h b/library/cpp/yt/threading/spin_lock-inl.h index eedb1e52856..7e629ae3381 100644 --- a/library/cpp/yt/threading/spin_lock-inl.h +++ b/library/cpp/yt/threading/spin_lock-inl.h @@ -4,7 +4,6 @@ // For the sake of sane code completion. #include "spin_lock.h" #endif -#undef SPIN_LOCK_INL_H_ #include "spin_wait.h" diff --git a/library/cpp/yt/threading/spin_lock_base-inl.h b/library/cpp/yt/threading/spin_lock_base-inl.h index 7135d081112..2e16a4e01f9 100644 --- a/library/cpp/yt/threading/spin_lock_base-inl.h +++ b/library/cpp/yt/threading/spin_lock_base-inl.h @@ -4,7 +4,6 @@ // For the sake of sane code completion. #include "spin_lock_base.h" #endif -#undef SPIN_LOCK_BASE_INL_H_ #include <library/cpp/yt/misc/source_location.h> diff --git a/library/cpp/yt/threading/spin_lock_count-inl.h b/library/cpp/yt/threading/spin_lock_count-inl.h index 6ba1782052f..8100f2f3e73 100644 --- a/library/cpp/yt/threading/spin_lock_count-inl.h +++ b/library/cpp/yt/threading/spin_lock_count-inl.h @@ -2,7 +2,6 @@ #ifndef SPIN_LOCK_COUNT_INL_H_ #error "Direct inclusion of this file is not allowed, include spin_lock_count.h" #endif -#undef SPIN_LOCK_COUNT_INL_H_ #include <util/system/compiler.h> diff --git a/library/cpp/yt/threading/writer_starving_rw_spin_lock-inl.h b/library/cpp/yt/threading/writer_starving_rw_spin_lock-inl.h index b880135f244..590e3238a9d 100644 --- a/library/cpp/yt/threading/writer_starving_rw_spin_lock-inl.h +++ b/library/cpp/yt/threading/writer_starving_rw_spin_lock-inl.h @@ -4,7 +4,6 @@ // For the sake of sane code completion. #include "writer_starving_rw_spin_lock.h" #endif -#undef WRITER_STARVING_RW_SPIN_LOCK_INL_H_ #include "spin_wait.h" diff --git a/library/cpp/yt/yson_string/string.cpp b/library/cpp/yt/yson_string/string.cpp index 2680cdbcdcf..7c7713b0d9a 100644 --- a/library/cpp/yt/yson_string/string.cpp +++ b/library/cpp/yt/yson_string/string.cpp @@ -156,8 +156,8 @@ TSharedRef TYsonString::ToSharedRef() const [&] (const TSharedRangeHolderPtr& holder) { return TSharedRef(Begin_, Size_, holder); }, - [] (const TCowString& payload) { - return TSharedRef::FromString(payload); + [&] (const TCowString& payload) { + return TSharedRef(Begin_, Size_, MakeSharedRangeHolder(payload)); }); } diff --git a/library/cpp/yt/yson_string/unittests/saveload_ut.cpp b/library/cpp/yt/yson_string/unittests/saveload_ut.cpp index a7700fb2ff6..bcb1bfa196b 100644 --- a/library/cpp/yt/yson_string/unittests/saveload_ut.cpp +++ b/library/cpp/yt/yson_string/unittests/saveload_ut.cpp @@ -31,7 +31,7 @@ TEST(TYsonStringTest, SaveLoadString) TEST(TYsonStringTest, SaveLoadSharedRef) { - auto ref = TSharedRef::FromString("My tests data"); + auto ref = TSharedRef::FromString(std::string("My tests data")); const TYsonString expected(ref); TStringStream s; ::Save(&s, expected); |