diff options
Diffstat (limited to 'library')
48 files changed, 839 insertions, 197 deletions
diff --git a/library/cpp/blockcodecs/codecs.h b/library/cpp/blockcodecs/codecs.h index fd499b54b0..43a9244465 100644 --- a/library/cpp/blockcodecs/codecs.h +++ b/library/cpp/blockcodecs/codecs.h @@ -1,3 +1,3 @@ #pragma once -#include <library/cpp/blockcodecs/core/codecs.h> +#include <library/cpp/blockcodecs/core/codecs.h> // IWYU pragma: export diff --git a/library/cpp/containers/dense_hash/dense_hash.h b/library/cpp/containers/dense_hash/dense_hash.h index 739479c25a..b5feb16eef 100644 --- a/library/cpp/containers/dense_hash/dense_hash.h +++ b/library/cpp/containers/dense_hash/dense_hash.h @@ -168,14 +168,10 @@ public: } else { initSize = FastClp2(initSize); } - BucketMask = initSize - 1; + Buckets.clear(); + BucketMask = 0; NumFilled = 0; - TVector<value_type> tmp; - for (size_type i = 0; i < initSize; ++i) { - tmp.emplace_back(EmptyMarker, mapped_type{}); - } - tmp.swap(Buckets); - GrowThreshold = Max<size_type>(1, initSize * MaxLoadFactor / 100) - 1; + Grow(initSize); } template <class K> diff --git a/library/cpp/getopt/small/last_getopt_opt.h b/library/cpp/getopt/small/last_getopt_opt.h index 8754ebb7ee..de3b10bade 100644 --- a/library/cpp/getopt/small/last_getopt_opt.h +++ b/library/cpp/getopt/small/last_getopt_opt.h @@ -13,7 +13,6 @@ #include <util/string/join.h> #include <optional> -#include <stdarg.h> namespace NLastGetopt { enum EHasArg { diff --git a/library/cpp/json/json_value.h b/library/cpp/json/json_value.h index f70f4d2ee8..47273f6b13 100644 --- a/library/cpp/json/json_value.h +++ b/library/cpp/json/json_value.h @@ -1,3 +1,3 @@ #pragma once -#include <library/cpp/json/writer/json_value.h> +#include <library/cpp/json/writer/json_value.h> // IWYU pragma: export diff --git a/library/cpp/lwtrace/mon/analytics/util.h b/library/cpp/lwtrace/mon/analytics/util.h index e07d06cc43..fd2ec572af 100644 --- a/library/cpp/lwtrace/mon/analytics/util.h +++ b/library/cpp/lwtrace/mon/analytics/util.h @@ -4,6 +4,7 @@ #include <util/generic/algorithm.h> #include <util/generic/hash_set.h> #include <util/string/vector.h> +#include <cmath> namespace NAnalytics { diff --git a/library/cpp/monlib/consumers/collecting_consumer.h b/library/cpp/monlib/consumers/collecting_consumer.h index 0b061fce0b..9f9631bef0 100644 --- a/library/cpp/monlib/consumers/collecting_consumer.h +++ b/library/cpp/monlib/consumers/collecting_consumer.h @@ -3,6 +3,7 @@ #include <library/cpp/monlib/metrics/labels.h> #include <library/cpp/monlib/metrics/metric_value.h> #include <library/cpp/monlib/metrics/metric_consumer.h> +#include <library/cpp/monlib/metrics/metric_registry.h> #include <util/datetime/base.h> @@ -26,6 +27,7 @@ namespace NMonitoring { // TODO(ivanzhukov@): rename to Type NMonitoring::EMetricType Kind{NMonitoring::EMetricType::UNKNOWN}; THolder<NMonitoring::TMetricTimeSeries> Values; + NMonitoring::TMetricOpts Opts; }; template <typename TLabelsImpl> @@ -96,6 +98,10 @@ namespace NMonitoring { val->Add(time, snapshot.Get()); } + virtual void OnMemOnly(bool isMemOnly) override{ + Metrics.back().Opts.MemOnly = isMemOnly; + } + bool DoMergeCommonLabels{false}; TVector<TMetricData> Metrics; TLabelsImpl CommonLabels; diff --git a/library/cpp/monlib/dynamic_counters/page.cpp b/library/cpp/monlib/dynamic_counters/page.cpp index 5cd750026f..73b1309d81 100644 --- a/library/cpp/monlib/dynamic_counters/page.cpp +++ b/library/cpp/monlib/dynamic_counters/page.cpp @@ -4,6 +4,7 @@ #include <library/cpp/monlib/service/pages/templates.h> #include <library/cpp/string_utils/quote/quote.h> +#include <util/string/builder.h> #include <util/string/split.h> #include <util/system/tls.h> @@ -26,6 +27,19 @@ TMaybe<EFormat> ParseFormat(TStringBuf str) { } } +namespace { + +TStringBuf GetParams(NMonitoring::IMonHttpRequest& request) { + TStringBuf uri = request.GetUri(); + TStringBuf params = uri.After('?'); + if (params.Size() == uri.Size()) { + params.Clear(); + } + return params; +} + +} + void TDynamicCountersPage::Output(NMonitoring::IMonHttpRequest& request) { if (OutputCallback) { OutputCallback(); @@ -37,28 +51,51 @@ void TDynamicCountersPage::Output(NMonitoring::IMonHttpRequest& request) { }; TVector<TStringBuf> parts; - StringSplitter(request.GetPathInfo()) - .Split('/') - .SkipEmpty() - .Collect(&parts); - - TMaybe<EFormat> format = !parts.empty() ? ParseFormat(parts.back()) : Nothing(); - if (format) { - parts.pop_back(); - } + TMaybe<EFormat> format; + TStringBuf params = GetParams(request); + + if (request.GetPathInfo().empty() && !params.empty()) { + StringSplitter(params).Split('&').SkipEmpty().Consume([&](TStringBuf part) { + TStringBuf name; + TStringBuf value; + part.Split('=', name, value); + if (name.StartsWith("@")) { + if (name == "@format") { + format = ParseFormat(value); + } else if (name == "@name_label") { + nameLabel = value; + } else if (name == "@private") { + visibility = TCountableBase::EVisibility::Private; + } + } else { + parts.push_back(part); + } + return true; + }); + } else { + StringSplitter(request.GetPathInfo()) + .Split('/') + .SkipEmpty() + .Collect(&parts); + + format = !parts.empty() ? ParseFormat(parts.back()) : Nothing(); + if (format) { + parts.pop_back(); + } - if (!parts.empty() && parts.back().StartsWith(TStringBuf("name_label="))) { - TVector<TString> labels; - StringSplitter(parts.back()).Split('=').SkipEmpty().Collect(&labels); - if (labels.size() == 2U) { - nameLabel = labels.back(); + if (!parts.empty() && parts.back().StartsWith(TStringBuf("name_label="))) { + TVector<TString> labels; + StringSplitter(parts.back()).Split('=').SkipEmpty().Collect(&labels); + if (labels.size() == 2U) { + nameLabel = labels.back(); + } + parts.pop_back(); } - parts.pop_back(); - } - if (!parts.empty() && parts.back() == TStringBuf("private")) { - visibility = TCountableBase::EVisibility::Private; - parts.pop_back(); + if (!parts.empty() && parts.back() == TStringBuf("private")) { + visibility = TCountableBase::EVisibility::Private; + parts.pop_back(); + } } auto counters = Counters; @@ -121,9 +158,15 @@ void TDynamicCountersPage::HandleAbsentSubgroup(IMonHttpRequest& request) { void TDynamicCountersPage::BeforePre(IMonHttpRequest& request) { IOutputStream& out = request.Output(); + TStringBuf params = GetParams(request); + TStringBuilder base; + base << Path << '?'; + if (!params.empty()) { + base << params << '&'; + } HTML(out) { DIV() { - out << "<a href='" << request.GetPath() << "/json'>Counters as JSON</a>"; + out << "<a href='" << base << "@format=json'>Counters as JSON</a>"; out << " for Solomon"; } @@ -133,9 +176,11 @@ void TDynamicCountersPage::BeforePre(IMonHttpRequest& request) { UL() { currentCounters->EnumerateSubgroups([&](const TString& name, const TString& value) { LI() { - TString pathPart = name + "=" + value; - Quote(pathPart, ""); - out << "\n<a href='" << request.GetPath() << "/" << pathPart << "'>" << name << " " << value << "</a>"; + auto escName = name; + auto escValue = value; + Quote(escName); + Quote(escValue); + out << "\n<a href='" << base << escName << '=' << escValue << "'>" << name << " " << value << "</a>"; } }); } diff --git a/library/cpp/monlib/metrics/fake.h b/library/cpp/monlib/metrics/fake.h index b01ff2505a..a058e1d99a 100644 --- a/library/cpp/monlib/metrics/fake.h +++ b/library/cpp/monlib/metrics/fake.h @@ -82,6 +82,8 @@ namespace NMonitoring { i64 Get() const noexcept override { return 0; } + + void Reset() noexcept override {} }; struct TFakeRate final: public TFakeAcceptor<IRate> { @@ -102,6 +104,8 @@ namespace NMonitoring { ui64 Get() const noexcept override { return 0; } + + void Reset() noexcept override {} }; struct TFakeGauge final: public TFakeAcceptor<IGauge> { @@ -117,12 +121,16 @@ namespace NMonitoring { double Get() const noexcept override { return 0; } + + void Reset() noexcept override {} }; struct TFakeLazyGauge final: public TFakeAcceptor<ILazyGauge> { double Get() const noexcept override { return 0; } + + void Reset() noexcept override {} }; struct TFakeHistogram final: public IHistogram { @@ -169,5 +177,7 @@ namespace NMonitoring { ui64 Get() const noexcept override { return 0; } + + void Reset() noexcept override {} }; } // namespace NMonitoring diff --git a/library/cpp/monlib/metrics/metric.h b/library/cpp/monlib/metrics/metric.h index 2f7d9de687..bb5fda322e 100644 --- a/library/cpp/monlib/metrics/metric.h +++ b/library/cpp/monlib/metrics/metric.h @@ -15,6 +15,7 @@ namespace NMonitoring { virtual EMetricType Type() const noexcept = 0; virtual void Accept(TInstant time, IMetricConsumer* consumer) const = 0; + virtual void Reset() noexcept = 0; }; using IMetricPtr = TIntrusivePtr<IMetric>; @@ -28,7 +29,7 @@ namespace NMonitoring { virtual double Add(double n) noexcept = 0; virtual void Set(double n) noexcept = 0; virtual double Get() const noexcept = 0; - virtual void Reset() noexcept { + void Reset() noexcept override { Set(0); } }; @@ -58,7 +59,7 @@ namespace NMonitoring { virtual void Set(i64 value) noexcept = 0; virtual i64 Get() const noexcept = 0; - virtual void Reset() noexcept { + void Reset() noexcept override { Set(0); } }; @@ -84,7 +85,6 @@ namespace NMonitoring { virtual ui64 Add(ui64 n) noexcept = 0; virtual ui64 Get() const noexcept = 0; - virtual void Reset() noexcept = 0; }; class ILazyCounter: public IMetric { @@ -108,7 +108,6 @@ namespace NMonitoring { virtual ui64 Add(ui64 n) noexcept = 0; virtual ui64 Get() const noexcept = 0; - virtual void Reset() noexcept = 0; }; class ILazyRate: public IMetric { @@ -134,7 +133,6 @@ namespace NMonitoring { virtual void Record(double value) noexcept = 0; virtual void Record(double value, ui32 count) noexcept = 0; virtual IHistogramSnapshotPtr TakeSnapshot() const = 0; - virtual void Reset() noexcept = 0; protected: const bool IsRate_; @@ -194,6 +192,8 @@ namespace NMonitoring { consumer->OnDouble(time, Get()); } + void Reset() noexcept override {} + private: std::function<double()> Supplier_; }; @@ -245,6 +245,8 @@ namespace NMonitoring { consumer->OnInt64(time, Get()); } + void Reset() noexcept override {} + private: std::function<i64()> Supplier_; }; @@ -296,6 +298,8 @@ namespace NMonitoring { consumer->OnUint64(time, Get()); } + void Reset() noexcept override {} + private: std::function<ui64()> Supplier_; }; @@ -347,6 +351,8 @@ namespace NMonitoring { consumer->OnUint64(time, Get()); } + void Reset() noexcept override {} + private: std::function<ui64()> Supplier_; }; diff --git a/library/cpp/monlib/metrics/metric_registry.cpp b/library/cpp/monlib/metrics/metric_registry.cpp index 245f65702d..dbbea603c1 100644 --- a/library/cpp/monlib/metrics/metric_registry.cpp +++ b/library/cpp/monlib/metrics/metric_registry.cpp @@ -226,29 +226,7 @@ namespace NMonitoring { void TMetricRegistry::Reset() { TWriteGuard g{*Lock_}; for (auto& [label, metricValue] : Metrics_) { - auto metric = metricValue.Metric; - switch (metric->Type()) { - case EMetricType::GAUGE: - static_cast<TGauge*>(metric.Get())->Set(.0); - break; - case EMetricType::IGAUGE: - static_cast<TIntGauge*>(metric.Get())->Set(0); - break; - case EMetricType::COUNTER: - static_cast<TCounter*>(metric.Get())->Reset(); - break; - case EMetricType::RATE: - static_cast<TRate*>(metric.Get())->Reset(); - break; - case EMetricType::HIST: - case EMetricType::HIST_RATE: - static_cast<THistogram*>(metric.Get())->Reset(); - break; - case EMetricType::UNKNOWN: - case EMetricType::DSUMMARY: - case EMetricType::LOGHIST: - break; - } + metricValue.Metric->Reset(); } } @@ -257,40 +235,6 @@ namespace NMonitoring { Metrics_.clear(); } - template <typename TMetric, EMetricType type, typename TLabelsType, typename... Args> - TMetric* TMetricRegistry::Metric(TLabelsType&& labels, TMetricOpts&& opts, Args&&... args) { - { - TReadGuard g{*Lock_}; - - auto it = Metrics_.find(labels); - if (it != Metrics_.end()) { - Y_ENSURE(it->second.Metric->Type() == type, "cannot create metric " << labels - << " with type " << MetricTypeToStr(type) - << ", because registry already has same metric with type " << MetricTypeToStr(it->second.Metric->Type())); - Y_ENSURE(it->second.Opts.MemOnly == opts.MemOnly,"cannot create metric " << labels - << " with memOnly=" << opts.MemOnly - << ", because registry already has same metric with memOnly=" << it->second.Opts.MemOnly); - return static_cast<TMetric*>(it->second.Metric.Get()); - } - } - - { - IMetricPtr metric = MakeIntrusive<TMetric>(std::forward<Args>(args)...); - - TWriteGuard g{*Lock_}; - // decltype(Metrics_)::iterator breaks build on windows - THashMap<ILabelsPtr, TMetricValue>::iterator it; - TMetricValue metricValue = {metric, opts}; - if constexpr (!std::is_convertible_v<TLabelsType, ILabelsPtr>) { - it = Metrics_.emplace(new TLabels{std::forward<TLabelsType>(labels)}, std::move(metricValue)).first; - } else { - it = Metrics_.emplace(std::forward<TLabelsType>(labels), std::move(metricValue)).first; - } - - return static_cast<TMetric*>(it->second.Metric.Get()); - } - } - void TMetricRegistry::RemoveMetric(const ILabels& labels) noexcept { TWriteGuard g{*Lock_}; Metrics_.erase(labels); diff --git a/library/cpp/monlib/metrics/metric_registry.h b/library/cpp/monlib/metrics/metric_registry.h index f60467cf91..7669a8c088 100644 --- a/library/cpp/monlib/metrics/metric_registry.h +++ b/library/cpp/monlib/metrics/metric_registry.h @@ -274,13 +274,45 @@ namespace NMonitoring { TMetricOpts Opts; }; + protected: + template <typename TMetric, EMetricType type, typename TLabelsType, typename... Args> + TMetric* Metric(TLabelsType&& labels, TMetricOpts&& opts, Args&&... args) { + { + TReadGuard g{*Lock_}; + + auto it = Metrics_.find(labels); + if (it != Metrics_.end()) { + Y_ENSURE(it->second.Metric->Type() == type, "cannot create metric " << labels + << " with type " << MetricTypeToStr(type) + << ", because registry already has same metric with type " << MetricTypeToStr(it->second.Metric->Type())); + Y_ENSURE(it->second.Opts.MemOnly == opts.MemOnly,"cannot create metric " << labels + << " with memOnly=" << opts.MemOnly + << ", because registry already has same metric with memOnly=" << it->second.Opts.MemOnly); + return static_cast<TMetric*>(it->second.Metric.Get()); + } + } + + { + IMetricPtr metric = MakeIntrusive<TMetric>(std::forward<Args>(args)...); + + TWriteGuard g{*Lock_}; + // decltype(Metrics_)::iterator breaks build on windows + THashMap<ILabelsPtr, TMetricValue>::iterator it; + TMetricValue metricValue = {metric, opts}; + if constexpr (!std::is_convertible_v<TLabelsType, ILabelsPtr>) { + it = Metrics_.emplace(new TLabels{std::forward<TLabelsType>(labels)}, std::move(metricValue)).first; + } else { + it = Metrics_.emplace(std::forward<TLabelsType>(labels), std::move(metricValue)).first; + } + + return static_cast<TMetric*>(it->second.Metric.Get()); + } + } + private: THolder<TRWMutex> Lock_ = MakeHolder<TRWMutex>(); THashMap<ILabelsPtr, TMetricValue> Metrics_; - template <typename TMetric, EMetricType type, typename TLabelsType, typename... Args> - TMetric* Metric(TLabelsType&& labels, TMetricOpts&& opts, Args&&... args); - TLabels CommonLabels_; }; diff --git a/library/cpp/protobuf/json/proto2json_printer.cpp b/library/cpp/protobuf/json/proto2json_printer.cpp index a9f8c3fce9..706b3b8c62 100644 --- a/library/cpp/protobuf/json/proto2json_printer.cpp +++ b/library/cpp/protobuf/json/proto2json_printer.cpp @@ -253,7 +253,7 @@ namespace NProtobufJson { bool inProtoMap) { Y_ABORT_UNLESS(!field.is_repeated(), "field is repeated."); - if (!key) { + if (!key && !inProtoMap) { key = MakeKey(field); } diff --git a/library/cpp/protobuf/json/ut/proto2json_ut.cpp b/library/cpp/protobuf/json/ut/proto2json_ut.cpp index f5bcfac49d..ce7d119be7 100644 --- a/library/cpp/protobuf/json/ut/proto2json_ut.cpp +++ b/library/cpp/protobuf/json/ut/proto2json_ut.cpp @@ -968,10 +968,10 @@ Y_UNIT_TEST(TestMapAsObject) { auto& items = *proto.MutableItems(); items["key1"] = "value1"; - items["key2"] = "value2"; + items[""] = "value2"; items["key3"] = "value3"; - TString modelStr(R"_({"Items":{"key3":"value3","key2":"value2","key1":"value1"}})_"); + TString modelStr(R"_({"Items":{"key3":"value3","":"value2","key1":"value1"}})_"); TStringStream jsonStr; TProto2JsonConfig config; diff --git a/library/cpp/protobuf/yql/descriptor.cpp b/library/cpp/protobuf/yql/descriptor.cpp index e5e1c8e28b..3a0d0d8b47 100644 --- a/library/cpp/protobuf/yql/descriptor.cpp +++ b/library/cpp/protobuf/yql/descriptor.cpp @@ -232,6 +232,9 @@ TString GenerateProtobufTypeConfig( case ERecursionTraits::Bytes: ret["view"]["recursion"] = "bytes"; break; + case ERecursionTraits::BytesV2: + ret["view"]["recursion"] = "bytesV2"; + break; } if (options.YtMode) { @@ -314,6 +317,8 @@ TProtoTypeConfig ParseTypeConfig(const TStringBuf& config) { result.Recursion = ERecursionTraits::Ignore; } else if (recursion == "bytes") { result.Recursion = ERecursionTraits::Bytes; + } else if (recursion == "bytesV2") { + result.Recursion = ERecursionTraits::BytesV2; } else { ythrow yexception() << "unsupported recursion trait " << recursion; diff --git a/library/cpp/protobuf/yql/descriptor.h b/library/cpp/protobuf/yql/descriptor.h index f5f51add0b..3a864f2026 100644 --- a/library/cpp/protobuf/yql/descriptor.h +++ b/library/cpp/protobuf/yql/descriptor.h @@ -27,6 +27,10 @@ enum class ERecursionTraits { Ignore = 1, //! Возвращать поля с рекурсивным типом в виде сериализованной строки Bytes = 2, + //! Возвращать поля с рекурсивным типом в виде сериализованной строки. + //! В этом режиме пофиксен баг, в котором + //! |optional RecursiveMessage field = 1| превращалось в байтовую строку, а не в опциональную байтовую строку. + BytesV2 = 3, }; struct TProtoTypeConfig { diff --git a/library/cpp/svnversion/svnversion.h b/library/cpp/svnversion/svnversion.h index b99615daa9..b1f8de5723 100644 --- a/library/cpp/svnversion/svnversion.h +++ b/library/cpp/svnversion/svnversion.h @@ -10,4 +10,4 @@ #include <util/system/compiler.h> // Automatically generated functions. -#include "build/scripts/c_templates/svnversion.h" +#include <build/scripts/c_templates/svnversion.h> // IWYU pragma: export diff --git a/library/cpp/testing/common/env.cpp b/library/cpp/testing/common/env.cpp index 1440186d78..41bf2b20ab 100644 --- a/library/cpp/testing/common/env.cpp +++ b/library/cpp/testing/common/env.cpp @@ -141,7 +141,6 @@ namespace NPrivate { void TTestEnv::ReInitialize() { IsRunningFromTest = false; - ArcadiaTestsDataDir = ""; SourceRoot = ""; BuildRoot = ""; WorkPath = ""; @@ -171,11 +170,6 @@ namespace NPrivate { BuildRoot = value->GetStringSafe(""); } - value = context.GetValueByPath("runtime.atd_root"); - if (value) { - ArcadiaTestsDataDir = value->GetStringSafe(""); - } - value = context.GetValueByPath("runtime.work_path"); if (value) { WorkPath = value->GetStringSafe(""); @@ -249,10 +243,6 @@ namespace NPrivate { BuildRoot = GetEnv("ARCADIA_BUILD_ROOT"); } - if (!ArcadiaTestsDataDir) { - ArcadiaTestsDataDir = GetEnv("ARCADIA_TESTS_DATA_DIR"); - } - if (!WorkPath) { WorkPath = GetEnv("TEST_WORK_PATH"); } diff --git a/library/cpp/testing/common/env.h b/library/cpp/testing/common/env.h index 6cd840442f..f9a5aff5d3 100644 --- a/library/cpp/testing/common/env.h +++ b/library/cpp/testing/common/env.h @@ -69,7 +69,6 @@ namespace NPrivate { void AddTestParam(TStringBuf name, TStringBuf value); bool IsRunningFromTest; - TString ArcadiaTestsDataDir; TString SourceRoot; TString BuildRoot; TString WorkPath; diff --git a/library/cpp/testing/gtest/gtest.h b/library/cpp/testing/gtest/gtest.h index a4d0c775d0..258a4459a9 100644 --- a/library/cpp/testing/gtest/gtest.h +++ b/library/cpp/testing/gtest/gtest.h @@ -1,11 +1,13 @@ #pragma once +// IWYU pragma: begin_exports #include <library/cpp/testing/gtest/matchers.h> #include <library/cpp/testing/gtest_extensions/gtest_extensions.h> #include <gtest/gtest.h> #include <gmock/gmock.h> +// IWYU pragma: end_exports #include <optional> #include <string_view> diff --git a/library/cpp/testing/gtest_extensions/gtest_extensions.h b/library/cpp/testing/gtest_extensions/gtest_extensions.h index e20532241e..e4f336fd54 100644 --- a/library/cpp/testing/gtest_extensions/gtest_extensions.h +++ b/library/cpp/testing/gtest_extensions/gtest_extensions.h @@ -1,6 +1,8 @@ #pragma once +// IWYU pragma: begin_exports #include "assertions.h" #include "matchers.h" #include "pretty_printers.h" #include "probe.h" +// IWYU pragma: end_exports diff --git a/library/cpp/testing/unittest/registar.cpp b/library/cpp/testing/unittest/registar.cpp index d48f13813e..5c2e9343f9 100644 --- a/library/cpp/testing/unittest/registar.cpp +++ b/library/cpp/testing/unittest/registar.cpp @@ -5,7 +5,6 @@ #include <util/generic/yexception.h> #include <util/random/fast.h> -#include <util/string/printf.h> #include <util/system/backtrace.h> #include <util/system/guard.h> #include <util/system/tls.h> @@ -134,8 +133,8 @@ struct TTraceDiffFormatter { } }; -TString NUnitTest::GetFormatTag(const char* name) { - return Sprintf("[[%s]]", name); +TString NUnitTest::GetFormatTag(TStringBuf name) { + return TString::Join("[[", name, "]]"); } TString NUnitTest::GetResetTag() { @@ -156,7 +155,7 @@ TString NUnitTest::ColoredDiff(TStringBuf s1, TStringBuf s2, const TString& deli } static TString MakeTestName(const NUnitTest::ITestSuiteProcessor::TTest& test) { - return TStringBuilder() << test.unit->name << "::" << test.name; + return TString::Join(test.unit->name, "::", test.name); } static size_t CountTests(const TMap<TString, size_t>& testErrors, bool succeeded) { diff --git a/library/cpp/testing/unittest/registar.h b/library/cpp/testing/unittest/registar.h index c65c79e79f..c6c1281b4d 100644 --- a/library/cpp/testing/unittest/registar.h +++ b/library/cpp/testing/unittest/registar.h @@ -42,7 +42,7 @@ namespace NUnitTest { extern bool ShouldColorizeDiff; TString ColoredDiff(TStringBuf s1, TStringBuf s2, const TString& delims = TString(), bool reverse = false); - TString GetFormatTag(const char* name); + TString GetFormatTag(TStringBuf name); TString GetResetTag(); // Raise error handler diff --git a/library/cpp/threading/future/future.h b/library/cpp/threading/future/future.h index 35db9abbe2..91ade3f6e5 100644 --- a/library/cpp/threading/future/future.h +++ b/library/cpp/threading/future/future.h @@ -1,4 +1,6 @@ #pragma once +// IWYU pragma: begin_exports #include "core/future.h" #include "wait/wait.h" +// IWYU pragma: end_exports diff --git a/library/cpp/threading/local_executor/README.md b/library/cpp/threading/local_executor/README.md index aaad2e2986..9a2831815f 100644 --- a/library/cpp/threading/local_executor/README.md +++ b/library/cpp/threading/local_executor/README.md @@ -1,37 +1,39 @@ -# Library for parallel task execution in thread pool +# Library for parallel task execution in a thread pool -This library allows easy parallelization of existing code and cycles. +This library allows easy parallelization of the existing code, particularly loops. It provides `NPar::TLocalExecutor` class and `NPar::LocalExecutor()` singleton accessor. -At start, `TLocalExecutor` has no threads in thread pool and all async tasks will be queued for later execution when extra threads appear. -All tasks should be `NPar::ILocallyExecutable` child class or function equal to `std::function<void(int)>` +At the start, `TLocalExecutor` has no threads in the thread pool and all async tasks will be queued for later execution when extra threads appear. +All tasks should be either derived from `NPar::ILocallyExecutable` or be of type `std::function<void(int)>`. ## TLocalExecutor methods -`TLocalExecutor::Run(int threadcount)` - add threads to thread pool (**WARNING!** `Run(threadcount)` will *add* `threadcount` threads to pool) +`TLocalExecutor::Run(int threadcount)` - add threads to the thread pool (**WARNING!** `Run(threadcount)` will *add* `threadcount` threads to pool) -`void TLocalExecutor::Exec(TLocallyExecutableFunction exec, int id, int flags)` - run one task and pass id as task function input, flags - bitmask composition of: +`void TLocalExecutor::Exec(TLocallyExecutableFunction exec, int id, int flags)` - run a single task and pass `id` as a task function argument, flags - bitmask that can contain: -- `TLocalExecutor::HIGH_PRIORITY = 0` - put task in high priority queue -- `TLocalExecutor::MED_PRIORITY = 1` - put task in medium priority queue -- `TLocalExecutor::LOW_PRIORITY = 2` - put task in low priority queue -- `TLocalExecutor::WAIT_COMPLETE = 4` - wait for task completion +- `TLocalExecutor::HIGH_PRIORITY = 0` - put the task in the high priority queue +- `TLocalExecutor::MED_PRIORITY = 1` - put the task in the medium priority queue +- `TLocalExecutor::LOW_PRIORITY = 2` - put the task in the low priority queue +- `TLocalExecutor::WAIT_COMPLETE = 4` - wait for the task completion -`void TLocalExecutor::ExecRange(TLocallyExecutableFunction exec, TExecRangeParams blockParams, int flags);` - run range of tasks `[TExecRangeParams::FirstId, TExecRangeParams::LastId).` +`void TLocalExecutor::ExecRange(TLocallyExecutableFunction exec, TExecRangeParams blockParams, int flags);` - run a range of tasks with ids `[TExecRangeParams::FirstId, TExecRangeParams::LastId).` `flags` is the same as for `TLocalExecutor::Exec`. -`TExecRangeParams` is a structure that describes the range. -By default each task is executed separately. Threads from thread pool are taking -the tasks in the manner first come first serve. +By default each task for each `id` is executed separately. Threads from the thread pool are taking the tasks in the FIFO manner. -It is also possible to partition range of tasks in consequtive blocks and execute each block as a bigger task. -`TExecRangeParams::SetBlockCountToThreadCount()` will result in thread count tasks, - where thread count is the count of threads in thread pool. - each thread will execute approximately equal count of tasks from range. +It is also possible to partition a range of tasks to consecutive blocks and execute each block as a bigger task. -`TExecRangeParams::SetBlockSize()` and `TExecRangeParams::SetBlockCount()` will partition -the range of tasks into consequtive blocks of approximately given size, or of size calculated - by partitioning the range into approximately equal size blocks of given count. +`TExecRangeParams` is a structure that is used for that. + +`TExecRangeParams::SetBlockCountToThreadCount()` will partition +the range of tasks into consecutive blocks with the number of tasks equivalent to the number of threads in the execution pool. The intent is that each thread will take an exactly single block from this partition, although it is not guaranteed, especially if the thread pool is already busy. + +`TExecRangeParams::SetBlockSize(TBlockSize blockSize)` will partition +the range of tasks into consecutive blocks of the size approximately equal to `blockSize`. + +`TExecRangeParams::SetBlockCount(TBlockCount blockCount)` will partition +the range of tasks into consecutive `blockCount` blocks with the approximately equal size. ## Examples @@ -51,7 +53,7 @@ SomeOtherCode(); event.WaitI(); ``` -### Execute task range and wait completion +### Execute a task range and wait for completion ```cpp using namespace NPar; @@ -64,11 +66,11 @@ LocalExecutor().ExecRange([](int id) { ### Exception handling -By default if a not caught exception arise in a task which runs through the Local Executor, then std::terminate() will be called immediately. The exception will be printed to stderr before the termination. Best practice is to handle exception within a task, or avoid throwing exceptions at all for performance reasons. +By default if an uncaught exception is thrown in a task that runs through the Local Executor, then `std::terminate()` will be called immediately. Best practice is to handle exception within a task, or avoid throwing exceptions at all for performance reasons. -However, if you'd like to handle and/or rethrow exceptions outside of a range, you can use ExecRangeWithFuture(). -It returns vector [0 .. LastId-FirstId] elements, where i-th element is a TFuture corresponding to task with id = (FirstId + i). -Use method .HasValue() of the element to check in Async mode if the corresponding task is complete. -Use .GetValue() or .GetValueSync() to wait for completion of the corresponding task. GetValue() and GetValueSync() will also rethrow an exception if it appears during execution of the task. +However, if you'd like to get exceptions that might have occured during the tasks execution instead, you can use `ExecRangeWithFutures()`. +It returns a vector of [0 .. LastId-FirstId] elements, where i-th element is a `TFuture` corresponding to the task with `id = (FirstId + i)`. +Use a method `.HasValue()` of the element to check in Async mode if the corresponding task is complete. +Use `.GetValue()` or `.GetValueSync()` to wait for completion of the corresponding task. `GetValue()` and `GetValueSync()` will also rethrow an exception if it has been thrown during the execution of the task. -You may also use ExecRangeWithThrow() to just receive an exception from a range if it appears. It rethrows an exception from a task with minimal id if such an exception exists, and guarantees normal flow if no exception arise. +You may also use `ExecRangeWithThrow()` to just receive an exception from a range if it has been thrown from at least one task. It rethrows an exception from a task with the minimal `id` from all the tasks where exceptions have been thrown or just continues as normal of there were no exceptions. diff --git a/library/cpp/threading/local_executor/local_executor.cpp b/library/cpp/threading/local_executor/local_executor.cpp index 4b4f69c0cb..107f981ff7 100644 --- a/library/cpp/threading/local_executor/local_executor.cpp +++ b/library/cpp/threading/local_executor/local_executor.cpp @@ -57,7 +57,7 @@ namespace { void LocalExec(int id) override { Y_ASSERT(FirstId <= id && id < LastId); - NThreading::NImpl::SetValue(Promises[id - FirstId], [=] { Exec(id); }); + NThreading::NImpl::SetValue(Promises[id - FirstId], [this, id] { Exec(id); }); } TVector<NThreading::TFuture<void>> GetFutures() const { diff --git a/library/cpp/threading/local_executor/local_executor.h b/library/cpp/threading/local_executor/local_executor.h index c1c824f67c..4eb67b034e 100644 --- a/library/cpp/threading/local_executor/local_executor.h +++ b/library/cpp/threading/local_executor/local_executor.h @@ -54,8 +54,12 @@ namespace NPar { // @param flags Same as for `Exec`. virtual void ExecRange(TIntrusivePtr<ILocallyExecutable> exec, int firstId, int lastId, int flags) = 0; - // 0-based ILocalExecutor worker thread identification + // returns: + // 0 for for a thread outside the internal pool + // (because ILocalExecutor is allowed to use a calling thread to execute tasks as well), + // 1 ... GetThreadCount() for a thread inside the internal pool virtual int GetWorkerThreadId() const noexcept = 0; + virtual int GetThreadCount() const noexcept = 0; // Describes a range of tasks with parameters from integer range [FirstId, LastId). @@ -237,7 +241,6 @@ namespace NPar { int GetLPQueueSize() const noexcept; void ClearLPQueue(); - // 0-based TLocalExecutor worker thread identification int GetWorkerThreadId() const noexcept override; int GetThreadCount() const noexcept override; diff --git a/library/cpp/threading/local_executor/tbb_local_executor.cpp b/library/cpp/threading/local_executor/tbb_local_executor.cpp index 65d6659443..cad425a1a6 100644 --- a/library/cpp/threading/local_executor/tbb_local_executor.cpp +++ b/library/cpp/threading/local_executor/tbb_local_executor.cpp @@ -14,9 +14,21 @@ int NPar::TTbbLocalExecutor<RespectTls>::GetThreadCount() const noexcept { template <bool RespectTls> int NPar::TTbbLocalExecutor<RespectTls>::GetWorkerThreadId() const noexcept { - return TbbArena.execute([] { - return tbb::this_task_arena::current_thread_index(); - }); + static thread_local int WorkerThreadId = -1; + if (WorkerThreadId == -1) { + // Can't rely on return value except checking that it is 'not_initialized' because of + // "Since a thread may exit the arena at any time if it does not execute a task, the index of + // a thread may change between any two tasks" + // (https://oneapi-spec.uxlfoundation.org/specifications/oneapi/latest/elements/onetbb/source/task_scheduler/task_arena/this_task_arena_ns#_CPPv4N3tbb15this_task_arena20current_thread_indexEv) + const auto tbbThreadIndex = tbb::this_task_arena::current_thread_index(); + if (tbbThreadIndex == tbb::task_arena::not_initialized) { + // This thread does not belong to TBB worker threads + WorkerThreadId = 0; + } else { + WorkerThreadId = ++RegisteredThreadCounter; + } + } + return WorkerThreadId; } template <bool RespectTls> @@ -24,7 +36,7 @@ void NPar::TTbbLocalExecutor<RespectTls>::Exec(TIntrusivePtr<ILocallyExecutable> if (flags & WAIT_COMPLETE) { exec->LocalExec(id); } else { - TbbArena.execute([=] { + TbbArena.execute([this, exec, id] { SubmitAsyncTasks([=] (int id) { exec->LocalExec(id); }, id, id + 1); }); } @@ -43,8 +55,8 @@ void NPar::TTbbLocalExecutor<RespectTls>::ExecRange(TIntrusivePtr<ILocallyExecut } }); } else { - TbbArena.execute([=] { - SubmitAsyncTasks([=] (int id) { exec->LocalExec(id); }, firstId, lastId); + TbbArena.execute([this, exec, firstId, lastId] { + SubmitAsyncTasks([exec] (int id) { exec->LocalExec(id); }, firstId, lastId); }); } } diff --git a/library/cpp/threading/local_executor/tbb_local_executor.h b/library/cpp/threading/local_executor/tbb_local_executor.h index 8d790db18c..f67c07349d 100644 --- a/library/cpp/threading/local_executor/tbb_local_executor.h +++ b/library/cpp/threading/local_executor/tbb_local_executor.h @@ -9,6 +9,9 @@ #include <contrib/libs/tbb/include/tbb/task_arena.h> #include <contrib/libs/tbb/include/tbb/task_group.h> +#include <atomic> + + namespace NPar { template <bool RespectTls = false> class TTbbLocalExecutor final: public ILocalExecutor { @@ -16,10 +19,11 @@ namespace NPar { TTbbLocalExecutor(int nThreads) : ILocalExecutor() , TbbArena(nThreads) - , NumberOfTbbThreads(nThreads) {} + , NumberOfTbbThreads(nThreads) + , RegisteredThreadCounter(0) + {} ~TTbbLocalExecutor() noexcept override {} - // 0-based ILocalExecutor worker thread identification virtual int GetWorkerThreadId() const noexcept override; virtual int GetThreadCount() const noexcept override; @@ -45,5 +49,7 @@ namespace NPar { mutable tbb::task_arena TbbArena; tbb::task_group Group; int NumberOfTbbThreads; + + mutable std::atomic_int RegisteredThreadCounter; }; } diff --git a/library/cpp/tld/tlds-alpha-by-domain.txt b/library/cpp/tld/tlds-alpha-by-domain.txt index a118629445..8d976d0d26 100644 --- a/library/cpp/tld/tlds-alpha-by-domain.txt +++ b/library/cpp/tld/tlds-alpha-by-domain.txt @@ -1,4 +1,4 @@ -# Version 2025033101, Last Updated Tue Apr 1 07:07:01 2025 UTC +# Version 2025041900, Last Updated Sat Apr 19 07:07:01 2025 UTC AAA AARP ABB diff --git a/library/cpp/yt/logging/logger-inl.h b/library/cpp/yt/logging/logger-inl.h index 958ede8cda..c9a98cb3b4 100644 --- a/library/cpp/yt/logging/logger-inl.h +++ b/library/cpp/yt/logging/logger-inl.h @@ -35,21 +35,35 @@ void TLogger::AddStructuredTag(TStringBuf key, TType value) } template <class... TArgs> -TLogger TLogger::WithTag(const char* format, TArgs&&... args) const +TLogger TLogger::WithTag(const char* format, TArgs&&... args) const & { auto result = *this; result.AddTag(format, std::forward<TArgs>(args)...); return result; } +template <class... TArgs> +TLogger TLogger::WithTag(const char* format, TArgs&&... args) && +{ + AddTag(format, std::forward<TArgs>(args)...); + return std::move(*this); +} + template <class TType> -TLogger TLogger::WithStructuredTag(TStringBuf key, TType value) const +TLogger TLogger::WithStructuredTag(TStringBuf key, TType value) const & { auto result = *this; result.AddStructuredTag(key, value); return result; } +template <class TType> +TLogger TLogger::WithStructuredTag(TStringBuf key, TType value) && +{ + AddStructuredTag(key, value); + return std::move(*this); +} + Y_FORCE_INLINE ELogLevel TLogger::GetEffectiveLoggingLevel(ELogLevel level, const TLoggingAnchor& anchor) { // Check if anchor is suppressed. diff --git a/library/cpp/yt/logging/logger.cpp b/library/cpp/yt/logging/logger.cpp index 682231d489..38fd5b6e13 100644 --- a/library/cpp/yt/logging/logger.cpp +++ b/library/cpp/yt/logging/logger.cpp @@ -226,34 +226,52 @@ void TLogger::AddRawTag(const std::string& tag) state->Tag += tag; } -TLogger TLogger::WithRawTag(const std::string& tag) const +TLogger TLogger::WithRawTag(const std::string& tag) const & { auto result = *this; result.AddRawTag(tag); return result; } -TLogger TLogger::WithEssential(bool essential) const +TLogger TLogger::WithRawTag(const std::string& tag) && +{ + AddRawTag(tag); + return std::move(*this); +} + +TLogger TLogger::WithEssential(bool essential) const & { auto result = *this; result.Essential_ = essential; return result; } +TLogger TLogger::WithEssential(bool essential) && +{ + Essential_ = essential; + return std::move(*this); +} + void TLogger::AddStructuredValidator(TStructuredValidator validator) { auto* state = GetMutableCoWState(); state->StructuredValidators.push_back(std::move(validator)); } -TLogger TLogger::WithStructuredValidator(TStructuredValidator validator) const +TLogger TLogger::WithStructuredValidator(TStructuredValidator validator) const & { auto result = *this; result.AddStructuredValidator(std::move(validator)); return result; } -TLogger TLogger::WithMinLevel(ELogLevel minLevel) const +TLogger TLogger::WithStructuredValidator(TStructuredValidator validator) && +{ + AddStructuredValidator(std::move(validator)); + return std::move(*this); +} + +TLogger TLogger::WithMinLevel(ELogLevel minLevel) const & { auto result = *this; if (result) { @@ -262,6 +280,14 @@ TLogger TLogger::WithMinLevel(ELogLevel minLevel) const return result; } +TLogger TLogger::WithMinLevel(ELogLevel minLevel) && +{ + if (*this) { + MinLevel_ = minLevel; + } + return std::move(*this); +} + const std::string& TLogger::GetTag() const { static const std::string emptyResult; diff --git a/library/cpp/yt/logging/logger.h b/library/cpp/yt/logging/logger.h index 53bb13e705..dfa1b79bac 100644 --- a/library/cpp/yt/logging/logger.h +++ b/library/cpp/yt/logging/logger.h @@ -181,7 +181,9 @@ public: TLogger() = default; TLogger(const TLogger& other) = default; + TLogger(TLogger&& other) = default; TLogger& operator=(const TLogger& other) = default; + TLogger& operator=(TLogger&& other) = default; TLogger(ILogManager* logManager, TStringBuf categoryName); explicit TLogger(TStringBuf categoryName); @@ -224,18 +226,26 @@ public: void AddStructuredValidator(TStructuredValidator validator); - TLogger WithRawTag(const std::string& tag) const; + TLogger WithRawTag(const std::string& tag) const &; + TLogger WithRawTag(const std::string& tag) &&; template <class... TArgs> - TLogger WithTag(const char* format, TArgs&&... args) const; + TLogger WithTag(const char* format, TArgs&&... args) const &; + template <class... TArgs> + TLogger WithTag(const char* format, TArgs&&... args) &&; template <class TType> - TLogger WithStructuredTag(TStringBuf key, TType value) const; + TLogger WithStructuredTag(TStringBuf key, TType value) const &; + template <class TType> + TLogger WithStructuredTag(TStringBuf key, TType value) &&; - TLogger WithStructuredValidator(TStructuredValidator validator) const; + TLogger WithStructuredValidator(TStructuredValidator validator) const &; + TLogger WithStructuredValidator(TStructuredValidator validator) &&; - TLogger WithMinLevel(ELogLevel minLevel) const; + TLogger WithMinLevel(ELogLevel minLevel) const &; + TLogger WithMinLevel(ELogLevel minLevel) &&; - TLogger WithEssential(bool essential = true) const; + TLogger WithEssential(bool essential = true) const &; + TLogger WithEssential(bool essential = true) &&; const std::string& GetTag() const; const TStructuredTags& GetStructuredTags() const; diff --git a/library/cpp/yt/memory/chunked_memory_pool.cpp b/library/cpp/yt/memory/chunked_memory_pool.cpp index d8673f5d6c..69794dcee1 100644 --- a/library/cpp/yt/memory/chunked_memory_pool.cpp +++ b/library/cpp/yt/memory/chunked_memory_pool.cpp @@ -58,6 +58,7 @@ TChunkedMemoryPool::TChunkedMemoryPool( , ChunkProviderHolder_(std::move(chunkProvider)) , ChunkProvider_(ChunkProviderHolder_.Get()) { + YT_VERIFY(ChunkProviderHolder_); Initialize(startChunkSize); } diff --git a/library/cpp/yt/string/format-inl.h b/library/cpp/yt/string/format-inl.h index 22da423f89..5b49fc5925 100644 --- a/library/cpp/yt/string/format-inl.h +++ b/library/cpp/yt/string/format-inl.h @@ -253,7 +253,7 @@ void FormatCompactIntervalRange( auto first = range.begin(); auto last = first; - auto current = first + 1; + auto current = std::next(first); while (current != range.end()) { if (valueGetter(current) != valueGetter(last) + 1) { @@ -316,7 +316,7 @@ typename TFormattableView<TRange, TFormatter>::TEnd TFormattableView<TRange, TFo } template <class TRange, class TFormatter> -TFormattableView<TRange, TFormatter> MakeFormattableView( +TFormattableView<TRange, std::decay_t<TFormatter>> MakeFormattableView( const TRange& range, TFormatter&& formatter) { @@ -324,7 +324,7 @@ TFormattableView<TRange, TFormatter> MakeFormattableView( } template <class TRange, class TFormatter> -TFormattableView<TRange, TFormatter> MakeShrunkFormattableView( +TFormattableView<TRange, std::decay_t<TFormatter>> MakeShrunkFormattableView( const TRange& range, TFormatter&& formatter, size_t limit) diff --git a/library/cpp/yt/string/format.h b/library/cpp/yt/string/format.h index d15127fae0..fa45f39e5a 100644 --- a/library/cpp/yt/string/format.h +++ b/library/cpp/yt/string/format.h @@ -91,12 +91,12 @@ struct TFormattableView //! Annotates a given #range with #formatter to be applied to each item. template <class TRange, class TFormatter> -TFormattableView<TRange, TFormatter> MakeFormattableView( +TFormattableView<TRange, std::decay_t<TFormatter>> MakeFormattableView( const TRange& range, TFormatter&& formatter); template <class TRange, class TFormatter> -TFormattableView<TRange, TFormatter> MakeShrunkFormattableView( +TFormattableView<TRange, std::decay_t<TFormatter>> MakeShrunkFormattableView( const TRange& range, TFormatter&& formatter, size_t limit); diff --git a/library/cpp/yt/string/unittests/format_ut.cpp b/library/cpp/yt/string/unittests/format_ut.cpp index e2e23c737c..a6c5ef6837 100644 --- a/library/cpp/yt/string/unittests/format_ut.cpp +++ b/library/cpp/yt/string/unittests/format_ut.cpp @@ -267,6 +267,19 @@ TEST(TFormatTest, LazyMultiValueFormatter) EXPECT_EQ("int: 1, string: hello, range: [1, 2, 3]", Format("%v", lazyFormatter)); } +TEST(TFormatTest, ReusableLambdaFormatter) +{ + auto formatter = [&] (auto* builder, int value) { + builder->AppendFormat("%v", value); + }; + + std::vector<int> range1{1, 2, 3}; + EXPECT_EQ("[1, 2, 3]", Format("%v", MakeFormattableView(range1, formatter))); + + std::vector<int> range2{4, 5, 6}; + EXPECT_EQ("[4, 5, 6]", Format("%v", MakeFormattableView(range2, formatter))); +} + TEST(TFormatTest, VectorArg) { std::vector<TString> params = {"a", "b", "c"}; diff --git a/library/cpp/yt/threading/atomic_object.h b/library/cpp/yt/threading/atomic_object.h index 8b642c0f4f..a77ade0a00 100644 --- a/library/cpp/yt/threading/atomic_object.h +++ b/library/cpp/yt/threading/atomic_object.h @@ -1,6 +1,6 @@ #pragma once -#include <library/cpp/yt/threading/rw_spin_lock.h> +#include <library/cpp/yt/threading/writer_starving_rw_spin_lock.h> #include <concepts> diff --git a/library/cpp/yt/threading/rw_spin_lock-inl.h b/library/cpp/yt/threading/rw_spin_lock-inl.h index 779de1b64a..0a31b1d9de 100644 --- a/library/cpp/yt/threading/rw_spin_lock-inl.h +++ b/library/cpp/yt/threading/rw_spin_lock-inl.h @@ -31,7 +31,7 @@ inline void TReaderWriterSpinLock::AcquireReaderForkFriendly() noexcept inline void TReaderWriterSpinLock::ReleaseReader() noexcept { auto prevValue = Value_.fetch_sub(ReaderDelta, std::memory_order::release); - Y_ASSERT((prevValue & ~WriterMask) != 0); + Y_ASSERT((prevValue & ~(WriterMask | WriterReadyMask)) != 0); NDetail::RecordSpinLockReleased(); } @@ -45,14 +45,14 @@ inline void TReaderWriterSpinLock::AcquireWriter() noexcept inline void TReaderWriterSpinLock::ReleaseWriter() noexcept { - auto prevValue = Value_.fetch_and(~WriterMask, std::memory_order::release); + auto prevValue = Value_.fetch_and(~(WriterMask | WriterReadyMask), std::memory_order::release); Y_ASSERT(prevValue & WriterMask); NDetail::RecordSpinLockReleased(); } inline bool TReaderWriterSpinLock::IsLocked() const noexcept { - return Value_.load() != UnlockedValue; + return (Value_.load() & ~WriterReadyMask) != 0; } inline bool TReaderWriterSpinLock::IsLockedByReader() const noexcept @@ -68,7 +68,7 @@ inline bool TReaderWriterSpinLock::IsLockedByWriter() const noexcept inline bool TReaderWriterSpinLock::TryAcquireReader() noexcept { auto oldValue = Value_.fetch_add(ReaderDelta, std::memory_order::acquire); - if ((oldValue & WriterMask) != 0) { + if ((oldValue & (WriterMask | WriterReadyMask)) != 0) { Value_.fetch_sub(ReaderDelta, std::memory_order::relaxed); return false; } @@ -79,7 +79,7 @@ inline bool TReaderWriterSpinLock::TryAcquireReader() noexcept inline bool TReaderWriterSpinLock::TryAndTryAcquireReader() noexcept { auto oldValue = Value_.load(std::memory_order::relaxed); - if ((oldValue & WriterMask) != 0) { + if ((oldValue & (WriterMask | WriterReadyMask)) != 0) { return false; } return TryAcquireReader(); @@ -88,7 +88,7 @@ inline bool TReaderWriterSpinLock::TryAndTryAcquireReader() noexcept inline bool TReaderWriterSpinLock::TryAcquireReaderForkFriendly() noexcept { auto oldValue = Value_.load(std::memory_order::relaxed); - if ((oldValue & WriterMask) != 0) { + if ((oldValue & (WriterMask | WriterReadyMask)) != 0) { return false; } auto newValue = oldValue + ReaderDelta; @@ -98,22 +98,35 @@ inline bool TReaderWriterSpinLock::TryAcquireReaderForkFriendly() noexcept return acquired; } -inline bool TReaderWriterSpinLock::TryAcquireWriter() noexcept +inline bool TReaderWriterSpinLock::TryAcquireWriterWithExpectedValue(TValue expected) noexcept { - auto expected = UnlockedValue; - - bool acquired = Value_.compare_exchange_weak(expected, WriterMask, std::memory_order::acquire); + bool acquired = Value_.compare_exchange_weak(expected, WriterMask, std::memory_order::acquire); NDetail::RecordSpinLockAcquired(acquired); return acquired; } +inline bool TReaderWriterSpinLock::TryAcquireWriter() noexcept +{ + // NB(pavook): we cannot expect writer ready to be set, as this method + // might be called without indicating writer readiness and we cannot + // indicate readiness on the hot path. This means that code calling + // TryAcquireWriter will spin against code calling AcquireWriter. + return TryAcquireWriterWithExpectedValue(UnlockedValue); +} + inline bool TReaderWriterSpinLock::TryAndTryAcquireWriter() noexcept { auto oldValue = Value_.load(std::memory_order::relaxed); - if (oldValue != UnlockedValue) { + + if ((oldValue & WriterReadyMask) == 0) { + oldValue = Value_.fetch_or(WriterReadyMask, std::memory_order::relaxed); + } + + if ((oldValue & (~WriterReadyMask)) != 0) { return false; } - return TryAcquireWriter(); + + return TryAcquireWriterWithExpectedValue(WriterReadyMask); } //////////////////////////////////////////////////////////////////////////////// diff --git a/library/cpp/yt/threading/rw_spin_lock.h b/library/cpp/yt/threading/rw_spin_lock.h index a915e677e8..64a241bb6b 100644 --- a/library/cpp/yt/threading/rw_spin_lock.h +++ b/library/cpp/yt/threading/rw_spin_lock.h @@ -16,8 +16,23 @@ namespace NYT::NThreading { //! Single-writer multiple-readers spin lock. /*! - * Reader-side calls are pretty cheap. - * The lock is unfair. + * Reader-side acquires are pretty cheap, and readers don't spin unless writers + * are present. + * + * The lock is unfair, but writers are prioritized over readers, that is, + * if AcquireWriter() is called at some time, then some writer + * (not necessarily the same one that called AcquireWriter) will succeed + * in the next time. This is implemented by an additional flag "WriterReady", + * that writers set on arrival. No readers can proceed until this flag is reset. + * + * WARNING: You probably should not use this lock if forks are possible: see + * fork_aware_rw_spin_lock.h for a proper fork-safe lock which does the housekeeping for you. + * + * WARNING: This lock is not recursive: you can't call AcquireReader() twice in the same + * thread, as that may lead to a deadlock. For the same reason you shouldn't do WaitFor or any + * other context switch under lock. + * + * See tla+/spinlock.tla for the formally verified lock's properties. */ class TReaderWriterSpinLock : public TSpinLockBase @@ -29,18 +44,26 @@ public: /*! * Optimized for the case of read-intensive workloads. * Cheap (just one atomic increment and no spinning if no writers are present). - * Don't use this call if forks are possible: forking at some + * + * WARNING: Don't use this call if forks are possible: forking at some * intermediate point inside #AcquireReader may corrupt the lock state and - * leave lock forever stuck for the child process. + * leave the lock stuck forever for the child process. + * + * WARNING: The lock is not recursive/reentrant, i.e. it assumes that no thread calls + * AcquireReader() if the reader is already acquired for it. */ void AcquireReader() noexcept; //! Acquires the reader lock. /*! * A more expensive version of #AcquireReader (includes at least * one atomic load and CAS; also may spin even if just readers are present). + * * In contrast to #AcquireReader, this method can be used in the presence of forks. - * Note that fork-friendliness alone does not provide fork-safety: additional - * actions must be performed to release the lock after a fork. + * + * WARNING: fork-friendliness alone does not provide fork-safety: additional + * actions must be performed to release the lock after a fork. This means you + * probably should NOT use this lock in the presence of forks, consider + * fork_aware_rw_spin_lock.h instead as a proper fork-safe lock. */ void AcquireReaderForkFriendly() noexcept; //! Tries acquiring the reader lock; see #AcquireReader. @@ -94,10 +117,12 @@ private: using TValue = ui32; static constexpr TValue UnlockedValue = 0; static constexpr TValue WriterMask = 1; - static constexpr TValue ReaderDelta = 2; + static constexpr TValue WriterReadyMask = 2; + static constexpr TValue ReaderDelta = 4; std::atomic<TValue> Value_ = UnlockedValue; + bool TryAcquireWriterWithExpectedValue(TValue expected) noexcept; bool TryAndTryAcquireReader() noexcept; bool TryAndTryAcquireWriter() noexcept; diff --git a/library/cpp/yt/threading/unittests/rw_spin_lock_ut.cpp b/library/cpp/yt/threading/unittests/rw_spin_lock_ut.cpp new file mode 100644 index 0000000000..653772604c --- /dev/null +++ b/library/cpp/yt/threading/unittests/rw_spin_lock_ut.cpp @@ -0,0 +1,56 @@ +#include <library/cpp/testing/gtest/gtest.h> + +#include <library/cpp/yt/threading/rw_spin_lock.h> + +#include <util/thread/pool.h> + +#include <latch> +#include <thread> + +namespace NYT::NThreading { +namespace { + +//////////////////////////////////////////////////////////////////////////////// + +TEST(TReaderWriterSpinLockTest, WriterPriority) +{ + int readerThreads = 10; + std::latch latch(readerThreads + 1); + std::atomic<size_t> finishedCount = {0}; + + TReaderWriterSpinLock lock; + + volatile std::atomic<ui32> x = {0}; + + auto readerTask = [&latch, &lock, &finishedCount, &x] () { + latch.arrive_and_wait(); + while (true) { + { + auto guard = ReaderGuard(lock); + // do some stuff + for (ui32 i = 0; i < 10'000u; ++i) { + x.fetch_add(i); + } + } + if (finishedCount.fetch_add(1) > 20'000) { + break; + } + } + }; + + auto readerPool = CreateThreadPool(readerThreads); + for (int i = 0; i < readerThreads; ++i) { + readerPool->SafeAddFunc(readerTask); + } + + latch.arrive_and_wait(); + while (finishedCount.load() == 0); + auto guard = WriterGuard(lock); + EXPECT_LE(finishedCount.load(), 1'000u); + DoNotOptimizeAway(x); +} + +//////////////////////////////////////////////////////////////////////////////// + +} // namespace +} // namespace NYT::NConcurrency diff --git a/library/cpp/yt/threading/unittests/spin_lock_fork_ut.cpp b/library/cpp/yt/threading/unittests/spin_lock_fork_ut.cpp new file mode 100644 index 0000000000..26e58fff74 --- /dev/null +++ b/library/cpp/yt/threading/unittests/spin_lock_fork_ut.cpp @@ -0,0 +1,160 @@ +#include <library/cpp/testing/gtest/gtest.h> + +#include <library/cpp/yt/threading/rw_spin_lock.h> +#include <library/cpp/yt/threading/fork_aware_spin_lock.h> + +#include <util/thread/pool.h> + +#include <sys/wait.h> + +namespace NYT::NThreading { +namespace { + +//////////////////////////////////////////////////////////////////////////////// + +TEST(TReaderWriterSpinLockTest, ForkFriendlyness) +{ + std::atomic<bool> stopped = {false}; + YT_DECLARE_SPIN_LOCK(TReaderWriterSpinLock, lock); + + auto readerTask = [&lock, &stopped] () { + while (!stopped.load()) { + ForkFriendlyReaderGuard(lock); + } + }; + + auto tryReaderTask = [&lock, &stopped] () { + while (!stopped.load()) { + // NB(pavook): TryAcquire instead of Acquire to minimize checks. + bool acquired = lock.TryAcquireReaderForkFriendly(); + if (acquired) { + lock.ReleaseReader(); + } + } + }; + + auto tryWriterTask = [&lock, &stopped] () { + while (!stopped.load()) { + Sleep(TDuration::MicroSeconds(1)); + bool acquired = lock.TryAcquireWriter(); + if (acquired) { + lock.ReleaseWriter(); + } + } + }; + + auto writerTask = [&lock, &stopped] () { + while (!stopped.load()) { + Sleep(TDuration::MicroSeconds(1)); + WriterGuard(lock); + } + }; + + int readerCount = 20; + int writerCount = 10; + + auto reader = CreateThreadPool(readerCount); + auto writer = CreateThreadPool(writerCount); + + for (int i = 0; i < readerCount / 2; ++i) { + reader->SafeAddFunc(readerTask); + reader->SafeAddFunc(tryReaderTask); + } + for (int i = 0; i < writerCount / 2; ++i) { + writer->SafeAddFunc(writerTask); + writer->SafeAddFunc(tryWriterTask); + } + + // And let the chaos begin! + int forkCount = 2000; + for (int iter = 1; iter <= forkCount; ++iter) { + pid_t pid; + { + auto guard = WriterGuard(lock); + pid = fork(); + } + + YT_VERIFY(pid >= 0); + + // NB(pavook): check different orders to maximize chaos. + if (iter % 2 == 0) { + ReaderGuard(lock); + } + WriterGuard(lock); + ReaderGuard(lock); + if (pid == 0) { + // NB(pavook): thread pools are no longer with us. + _exit(0); + } + } + + for (int i = 1; i <= forkCount; ++i) { + int status; + YT_VERIFY(waitpid(0, &status, 0) > 0); + YT_VERIFY(WIFEXITED(status) && WEXITSTATUS(status) == 0); + } + + stopped.store(true); +} + +//////////////////////////////////////////////////////////////////////////////// + +TEST(TForkAwareSpinLockTest, ForkSafety) +{ + std::atomic<bool> stopped = {false}; + YT_DECLARE_SPIN_LOCK(TForkAwareSpinLock, lock); + + auto acquireTask = [&lock, &stopped] () { + while (!stopped.load()) { + Guard(lock); + } + }; + + // NB(pavook): TryAcquire instead of Acquire to minimize checks. + auto tryAcquireTask = [&lock, &stopped] () { + while (!stopped.load()) { + bool acquired = lock.TryAcquire(); + if (acquired) { + lock.Release(); + } + } + }; + + int workerCount = 20; + + auto worker = CreateThreadPool(workerCount); + + for (int i = 0; i < workerCount / 2; ++i) { + worker->SafeAddFunc(acquireTask); + worker->SafeAddFunc(tryAcquireTask); + } + + // And let the chaos begin! + int forkCount = 2000; + for (int iter = 1; iter <= forkCount; ++iter) { + pid_t pid = fork(); + + YT_VERIFY(pid >= 0); + + Guard(lock); + Guard(lock); + + if (pid == 0) { + // NB(pavook): thread pools are no longer with us. + _exit(0); + } + } + + for (int i = 1; i <= forkCount; ++i) { + int status; + YT_VERIFY(waitpid(0, &status, 0) > 0); + YT_VERIFY(WIFEXITED(status) && WEXITSTATUS(status) == 0); + } + + stopped.store(true); +} + +//////////////////////////////////////////////////////////////////////////////// + +} // namespace +} // namespace NYT::NConcurrency diff --git a/library/cpp/yt/threading/unittests/ya.make b/library/cpp/yt/threading/unittests/ya.make index ef9b5d2995..da006012c0 100644 --- a/library/cpp/yt/threading/unittests/ya.make +++ b/library/cpp/yt/threading/unittests/ya.make @@ -5,9 +5,14 @@ INCLUDE(${ARCADIA_ROOT}/library/cpp/yt/ya_cpp.make.inc) SRCS( count_down_latch_ut.cpp recursive_spin_lock_ut.cpp + rw_spin_lock_ut.cpp spin_wait_ut.cpp ) +IF (NOT OS_WINDOWS) + SRC(spin_lock_fork_ut.cpp) +ENDIF() + PEERDIR( library/cpp/yt/assert library/cpp/yt/threading 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 new file mode 100644 index 0000000000..cf8bde715c --- /dev/null +++ b/library/cpp/yt/threading/writer_starving_rw_spin_lock-inl.h @@ -0,0 +1,101 @@ +#pragma once +#ifndef WRITER_STARVING_RW_SPIN_LOCK_INL_H_ +#error "Direct inclusion of this file is not allowed, include rw_spin_lock.h" +// 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" + +namespace NYT::NThreading { + +//////////////////////////////////////////////////////////////////////////////// + +inline void TWriterStarvingRWSpinLock::AcquireReader() noexcept +{ + if (TryAcquireReader()) { + return; + } + AcquireReaderSlow(); +} + +inline void TWriterStarvingRWSpinLock::ReleaseReader() noexcept +{ + auto prevValue = Value_.fetch_sub(ReaderDelta, std::memory_order::release); + Y_ASSERT((prevValue & ~WriterMask) != 0); + NDetail::RecordSpinLockReleased(); +} + +inline void TWriterStarvingRWSpinLock::AcquireWriter() noexcept +{ + if (TryAcquireWriter()) { + return; + } + AcquireWriterSlow(); +} + +inline void TWriterStarvingRWSpinLock::ReleaseWriter() noexcept +{ + auto prevValue = Value_.fetch_and(~WriterMask, std::memory_order::release); + Y_ASSERT(prevValue & WriterMask); + NDetail::RecordSpinLockReleased(); +} + +inline bool TWriterStarvingRWSpinLock::IsLocked() const noexcept +{ + return Value_.load() != UnlockedValue; +} + +inline bool TWriterStarvingRWSpinLock::IsLockedByReader() const noexcept +{ + return Value_.load() >= ReaderDelta; +} + +inline bool TWriterStarvingRWSpinLock::IsLockedByWriter() const noexcept +{ + return (Value_.load() & WriterMask) != 0; +} + +inline bool TWriterStarvingRWSpinLock::TryAcquireReader() noexcept +{ + auto oldValue = Value_.fetch_add(ReaderDelta, std::memory_order::acquire); + if ((oldValue & WriterMask) != 0) { + Value_.fetch_sub(ReaderDelta, std::memory_order::relaxed); + return false; + } + NDetail::RecordSpinLockAcquired(); + return true; +} + +inline bool TWriterStarvingRWSpinLock::TryAndTryAcquireReader() noexcept +{ + auto oldValue = Value_.load(std::memory_order::relaxed); + if ((oldValue & WriterMask) != 0) { + return false; + } + return TryAcquireReader(); +} + +inline bool TWriterStarvingRWSpinLock::TryAcquireWriter() noexcept +{ + auto expected = UnlockedValue; + + bool acquired = Value_.compare_exchange_weak(expected, WriterMask, std::memory_order::acquire); + NDetail::RecordSpinLockAcquired(acquired); + return acquired; +} + +inline bool TWriterStarvingRWSpinLock::TryAndTryAcquireWriter() noexcept +{ + auto oldValue = Value_.load(std::memory_order::relaxed); + if (oldValue != UnlockedValue) { + return false; + } + return TryAcquireWriter(); +} + +//////////////////////////////////////////////////////////////////////////////// + +} // namespace NYT::NThreading + diff --git a/library/cpp/yt/threading/writer_starving_rw_spin_lock.cpp b/library/cpp/yt/threading/writer_starving_rw_spin_lock.cpp new file mode 100644 index 0000000000..74c9f59db1 --- /dev/null +++ b/library/cpp/yt/threading/writer_starving_rw_spin_lock.cpp @@ -0,0 +1,25 @@ +#include "writer_starving_rw_spin_lock.h" + +namespace NYT::NThreading { + +//////////////////////////////////////////////////////////////////////////////// + +void TWriterStarvingRWSpinLock::AcquireReaderSlow() noexcept +{ + TSpinWait spinWait(Location_, ESpinLockActivityKind::Read); + while (!TryAndTryAcquireReader()) { + spinWait.Wait(); + } +} + +void TWriterStarvingRWSpinLock::AcquireWriterSlow() noexcept +{ + TSpinWait spinWait(Location_, ESpinLockActivityKind::Write); + while (!TryAndTryAcquireWriter()) { + spinWait.Wait(); + } +} + +//////////////////////////////////////////////////////////////////////////////// + +} // namespace NYT::NThreading diff --git a/library/cpp/yt/threading/writer_starving_rw_spin_lock.h b/library/cpp/yt/threading/writer_starving_rw_spin_lock.h new file mode 100644 index 0000000000..8a456afe21 --- /dev/null +++ b/library/cpp/yt/threading/writer_starving_rw_spin_lock.h @@ -0,0 +1,115 @@ +#pragma once + +#include "public.h" +#include "rw_spin_lock.h" +#include "spin_lock_base.h" +#include "spin_lock_count.h" + +#include <library/cpp/yt/memory/public.h> + +#include <util/system/rwlock.h> + +#include <atomic> + +namespace NYT::NThreading { + +//////////////////////////////////////////////////////////////////////////////// + +// TODO(pavook): deprecate it. + +//! Single-writer multiple-readers spin lock. +/*! + * Reader-side calls are pretty cheap. + * WARNING: The lock is unfair, and readers can starve writers. See rw_spin_lock.h for a writer-prioritized lock. + * WARNING: Never use the bare lock if forks are possible: see fork_aware_rw_spin_lock.h for a fork-safe lock. + * Unlike rw_spin_lock.h, reader-side is reentrant here: it is possible to acquire the **reader** lock multiple times + * even in the single thread. + * This doesn't mean you should do it: in fact, you shouldn't: use separate locks for separate entities. + * If you see this class in your code, try migrating to the proper rw_spin_lock.h after ensuring you don't rely on + * reentrant locking. + */ +class TWriterStarvingRWSpinLock + : public TSpinLockBase +{ +public: + using TSpinLockBase::TSpinLockBase; + + //! Acquires the reader lock. + /*! + * Optimized for the case of read-intensive workloads. + * Cheap (just one atomic increment and no spinning if no writers are present). + * Don't use this call if forks are possible: forking at some + * intermediate point inside #AcquireReader may corrupt the lock state and + * leave lock forever stuck for the child process. + */ + void AcquireReader() noexcept; + //! Tries acquiring the reader lock; see #AcquireReader. + //! Returns |true| on success. + bool TryAcquireReader() noexcept; + //! Releases the reader lock. + /*! + * Cheap (just one atomic decrement). + */ + void ReleaseReader() noexcept; + + //! Acquires the writer lock. + /*! + * Rather cheap (just one CAS). + */ + void AcquireWriter() noexcept; + //! Tries acquiring the writer lock; see #AcquireWriter. + //! Returns |true| on success. + bool TryAcquireWriter() noexcept; + //! Releases the writer lock. + /*! + * Cheap (just one atomic store). + */ + void ReleaseWriter() noexcept; + + //! Returns true if the lock is taken (either by a reader or writer). + /*! + * This is inherently racy. + * Only use for debugging and diagnostic purposes. + */ + bool IsLocked() const noexcept; + + //! Returns true if the lock is taken by reader. + /*! + * This is inherently racy. + * Only use for debugging and diagnostic purposes. + */ + bool IsLockedByReader() const noexcept; + + //! Returns true if the lock is taken by writer. + /*! + * This is inherently racy. + * Only use for debugging and diagnostic purposes. + */ + bool IsLockedByWriter() const noexcept; + +private: + using TValue = ui32; + static constexpr TValue UnlockedValue = 0; + static constexpr TValue WriterMask = 1; + static constexpr TValue ReaderDelta = 2; + + std::atomic<TValue> Value_ = UnlockedValue; + + + bool TryAndTryAcquireReader() noexcept; + bool TryAndTryAcquireWriter() noexcept; + + void AcquireReaderSlow() noexcept; + void AcquireWriterSlow() noexcept; +}; + +REGISTER_TRACKED_SPIN_LOCK_CLASS(TWriterStarvingRWSpinLock) + +//////////////////////////////////////////////////////////////////////////////// + +} // namespace NYT::NThreading + +#define WRITER_STARVING_RW_SPIN_LOCK_INL_H_ +#include "writer_starving_rw_spin_lock-inl.h" +#undef WRITER_STARVING_RW_SPIN_LOCK_INL_H_ + diff --git a/library/cpp/yt/threading/ya.make b/library/cpp/yt/threading/ya.make index cc11e7974e..d25f0a7068 100644 --- a/library/cpp/yt/threading/ya.make +++ b/library/cpp/yt/threading/ya.make @@ -18,6 +18,7 @@ SRCS( spin_lock.cpp spin_wait.cpp spin_wait_hook.cpp + writer_starving_rw_spin_lock.cpp ) PEERDIR( diff --git a/library/cpp/yt/ya_cpp.make.inc b/library/cpp/yt/ya_cpp.make.inc index 81c522e0d6..59dfb6e1c0 100644 --- a/library/cpp/yt/ya_cpp.make.inc +++ b/library/cpp/yt/ya_cpp.make.inc @@ -1,8 +1,18 @@ # This file should be included in all YT projects (including YT ORM installations). -IF (NOT MSVC) +IF (CLANG) CXXFLAGS( -Wdeprecated-this-capture -Wimplicit-fallthrough + -Wparentheses + -Wno-logical-op-parentheses + -Wno-bitwise-op-parentheses ) + + IF (MUSL) + CXXFLAGS( + --system-header-prefix=endian.h + --system-header-prefix=byteswap.h + ) + ENDIF() ENDIF() diff --git a/library/python/filelock/ut/ya.make b/library/python/filelock/ut/ya.make index 60108f73c6..62910fd331 100644 --- a/library/python/filelock/ut/ya.make +++ b/library/python/filelock/ut/ya.make @@ -13,12 +13,14 @@ IF (OS_DARWIN) TAG( ya:fat ya:exotic_platform + ya:large_tests_on_single_slots ) ELSEIF (OS_WINDOWS) SIZE(LARGE) TAG( ya:fat - sb:ssd&~MULTISLOT&WINDOWS + sb:ssd&WINDOWS + ya:large_tests_on_single_slots ) ENDIF() |