Merge pull request #44352 from ydb-platform/merge-rightlib-260624-0119

6 files changed, 385 insertions, 38 deletions

diff --git a/library/cpp/threading/atomic_shared_ptr/atomic_shared_ptr.h b/library/cpp/threading/atomic_shared_ptr/atomic_shared_ptr.h
index a991eb3bccc..fbc9b9576b8 100644
--- a/library/cpp/threading/atomic_shared_ptr/atomic_shared_ptr.h
+++ b/library/cpp/threading/atomic_shared_ptr/atomic_shared_ptr.h
@@ -36,11 +36,11 @@ namespace NPrivate {
         }
 
         ui64 Ref(ui64 add = USE_INCREMENT) noexcept {
-            return ref_count_.fetch_add(add, std::memory_order_relaxed) + add;
+            return ref_count_.fetch_add(add, std::memory_order_acquire) + add;
         }
 
         ui64 Unref(ui64 sub = USE_INCREMENT) noexcept {
-            return ref_count_.fetch_sub(sub, std::memory_order_seq_cst) - sub;
+            return ref_count_.fetch_sub(sub, std::memory_order_acq_rel) - sub;
         }
 
         void UnrefAndDelete(ui64 sub = USE_INCREMENT) {
@@ -48,8 +48,7 @@ namespace NPrivate {
                 return;
             ui64 expect = 0;
             bool flag_is_set = ref_count_.compare_exchange_strong(
-                expect, DESTROYED_FLAG,
-                std::memory_order_seq_cst, std::memory_order_relaxed);
+                expect, DESTROYED_FLAG, std::memory_order_acq_rel, std::memory_order_relaxed);
             if (!flag_is_set)
                 return;
             DestroyPayload();
@@ -61,8 +60,7 @@ namespace NPrivate {
                 return nullptr;
             ui64 expect = 0;
             bool flag_is_set = ref_count_.compare_exchange_strong(
-                expect, DESTROYED_FLAG,
-                std::memory_order_seq_cst, std::memory_order_relaxed);
+                expect, DESTROYED_FLAG, std::memory_order_acq_rel, std::memory_order_relaxed);
             if (!flag_is_set)
                 return nullptr;
             void* result = GetPtr();
@@ -79,17 +77,16 @@ namespace NPrivate {
         }
 
         ui64 RefWeak(ui64 add = 1) noexcept {
-            return weak_count_.fetch_add(add, std::memory_order_relaxed) + add;
+            return weak_count_.fetch_add(add, std::memory_order_acquire) + add;
         }
 
         ui64 UnrefWeak(ui64 sub = 1) noexcept {
-            return weak_count_.fetch_sub(sub, std::memory_order_seq_cst) - sub;
+            return weak_count_.fetch_sub(sub, std::memory_order_acq_rel) - sub;
         }
 
         void UnrefWeakAndDelete(ui64 sub = 1) noexcept {
-            if (UnrefWeak(sub) != 0)
-                return;
-            delete this;
+            if (UnrefWeak(sub) == 0)
+              delete this;
         }
     };
 
@@ -125,7 +122,7 @@ namespace NPrivate {
         explicit TSharedBasePtr(TRefCounter* counter_ptr) noexcept
             : ptr_((uintptr_t)counter_ptr)
         {
-            Y_ABORT_UNLESS((ptr_.load(std::memory_order_relaxed) & ~PTR_MASK) == 0,
+            Y_ABORT_UNLESS((ptr_.load(std::memory_order_acquire) & ~PTR_MASK) == 0,
                      "you must provide a clean ptr");
         }
 
@@ -140,21 +137,32 @@ namespace NPrivate {
 
         TRefCounter* ConcurrentAcquire() noexcept {
             auto result =
-                ptr_.fetch_add(CONCURRENT_INCREMENT, std::memory_order_seq_cst);
+                ptr_.fetch_add(CONCURRENT_INCREMENT, std::memory_order_acquire);
             auto ptr_result = CleanUpPtr(result);
-            if (ptr_result)
-                ptr_result->Ref(CONCURRENT_INCREMENT + TRefCounter::USE_INCREMENT);
+            if (!ptr_result)
+                return nullptr;
+            ptr_result->Ref(CONCURRENT_INCREMENT + TRefCounter::USE_INCREMENT);
             return ptr_result;
         }
 
         TRefCounter* ConcurrentWeakAcquire() noexcept {
             auto result =
-                ptr_.fetch_add(CONCURRENT_INCREMENT, std::memory_order_seq_cst);
+                ptr_.fetch_add(CONCURRENT_INCREMENT, std::memory_order_acquire);
+            auto ptr_result = CleanUpPtr(result);
+            if (!ptr_result)
+                return nullptr;
+            ptr_result->RefWeak();
+            ptr_result->Ref(CONCURRENT_INCREMENT);
+            return ptr_result;
+        }
+
+        TRefCounter* ConcurrentWeakAcquireFromWeak() noexcept {
+            auto result =
+                ptr_.fetch_add(CONCURRENT_INCREMENT, std::memory_order_acquire);
             auto ptr_result = CleanUpPtr(result);
-            if (ptr_result) {
-                ptr_result->RefWeak();
-                ptr_result->Ref(CONCURRENT_INCREMENT);
-            }
+            if (!ptr_result)
+                return nullptr;
+            ptr_result->RefWeak(CONCURRENT_INCREMENT + 1);
             return ptr_result;
         }
 
@@ -176,35 +184,35 @@ namespace NPrivate {
 
         static void DestroyWeakPtr(TRefCounter* ptr) noexcept {
             auto clean_ptr = CleanUpPtr(ptr);
-            if (clean_ptr) {
-                ui64 cnt = GetCounter(ptr);
-                clean_ptr->UnrefWeakAndDelete(cnt);
-            }
+            if (!clean_ptr)
+                return;
+            ui64 cnt = GetCounter(ptr);
+            clean_ptr->UnrefWeakAndDelete(cnt + 1);
         }
 
         static TRefCounter* CleanUpPtr(TRefCounter* ptr) noexcept {
-            return (TRefCounter*)((uintptr_t)ptr & PTR_MASK);
+            return CleanUpPtr(reinterpret_cast<uintptr_t>(ptr));
         }
 
         static TRefCounter* CleanUpPtr(uintptr_t ptr) noexcept {
-            return (TRefCounter*)(ptr & PTR_MASK);
+            return reinterpret_cast<TRefCounter*>(ptr & PTR_MASK);
         }
 
         static uintptr_t GetCounter(TRefCounter* ptr) noexcept {
-            return (uintptr_t)ptr & ~PTR_MASK;
+            return reinterpret_cast<uintptr_t>(ptr) & ~PTR_MASK;
         }
 
         TRefCounter* GetClean() const noexcept {
-            return CleanUpPtr(ptr_.load(std::memory_order_relaxed));
+            return CleanUpPtr(ptr_.load(std::memory_order_acquire));
         }
 
         TRefCounter* GetRaw() const noexcept {
-            return (TRefCounter*)ptr_.load(std::memory_order_relaxed);
+            return reinterpret_cast<TRefCounter*>(ptr_.load(std::memory_order_acquire));
         }
 
         TRefCounter* Swap(TRefCounter* other) noexcept {
-            return (TRefCounter*)ptr_.exchange(
-                (uintptr_t)other, std::memory_order_seq_cst);
+            return reinterpret_cast<TRefCounter*>(
+                ptr_.exchange(reinterpret_cast<uintptr_t>(other)));
         }
 
         size_t UseCount() const noexcept {
@@ -212,7 +220,7 @@ namespace NPrivate {
             auto clean_ptr = CleanUpPtr(ptr);
             if (!clean_ptr)
                 return 0;
-            ui64 result = clean_ptr->ref_count_.load(std::memory_order_relaxed);
+            ui64 result = clean_ptr->ref_count_.load(std::memory_order_acquire);
             result -= GetCounter(ptr);
             result /= TRefCounter::USE_INCREMENT;
             return result;
@@ -368,17 +376,36 @@ public:
     }
 
     TTrueAtomicWeakPtr(const TTrueAtomicWeakPtr& other) noexcept
-        : ptr_(other.ptr_.ConcurrentWeakAcquire())
+        : ptr_(other.ptr_.ConcurrentWeakAcquireFromWeak())
     {
     }
 
-    TTrueAtomicWeakPtr& operator=(const TTrueAtomicWeakPtr& other) noexcept {
+
+    TTrueAtomicWeakPtr(TTrueAtomicWeakPtr&& other) noexcept
+        : ptr_(other.ptr_.Swap(nullptr))
+    {
+    }
+
+    ~TTrueAtomicWeakPtr()
+    {
+      auto ptr = ptr_.GetRaw();
+      NPrivate::TSharedBasePtr::DestroyWeakPtr(ptr);
+    }
+
+    TTrueAtomicWeakPtr& operator=(const TTrueAtomicSharedPtr<PayloadType>& other) noexcept {
         auto new_ptr = other.ptr_.ConcurrentWeakAcquire();
         auto back_ptr = ptr_.Swap(new_ptr);
         NPrivate::TSharedBasePtr::DestroyWeakPtr(back_ptr);
         return *this;
     }
 
+    TTrueAtomicWeakPtr& operator=(const TTrueAtomicWeakPtr& other) noexcept {
+        auto new_ptr = other.ptr_.ConcurrentWeakAcquireFromWeak();
+        auto back_ptr = ptr_.Swap(new_ptr);
+        NPrivate::TSharedBasePtr::DestroyWeakPtr(back_ptr);
+        return *this;
+    }
+
     TTrueAtomicWeakPtr& operator=(TTrueAtomicWeakPtr&& other) noexcept {
         auto new_ptr = other.ptr_.Swap(nullptr);
         auto back_ptr = ptr_.Swap(new_ptr);
@@ -389,9 +416,10 @@ public:
     TTrueAtomicSharedPtr<PayloadType> lock() noexcept {
         // create local TTrueAtomicWeakPtr to avoid concurrent changes of this
         TTrueAtomicWeakPtr<PayloadType> local(*this);
-        if (!local.ptr_.GetRaw()->RefFromWeak())
-            return TTrueAtomicSharedPtr<PayloadType>();
-        return TTrueAtomicSharedPtr<PayloadType>(ptr_.GetClean());
+        if (auto raw_ptr = local.ptr_.GetRaw())
+          if (raw_ptr->RefFromWeak())
+            return TTrueAtomicSharedPtr<PayloadType>(ptr_.GetClean());
+        return TTrueAtomicSharedPtr<PayloadType>();
     }
 
     void reset() noexcept {
@@ -403,7 +431,7 @@ public:
     // Having A with possible accesses from multiple threads and B with exclusive
     // use from a single thread, then calling A.swap(B) is atomic.
     void swap(TTrueAtomicWeakPtr& other) noexcept {
-        auto copy = other.ptr_.ConcurrentWeakAcquire();
+        auto copy = other.ptr_.ConcurrentWeakAcquireFromWeak();
         auto back_ptr = ptr_.Swap(copy);
         back_ptr = other.ptr_.Swap(back_ptr);
         NPrivate::TSharedBasePtr::DestroyWeakPtr(back_ptr);
diff --git a/library/cpp/yt/string/stream.cpp b/library/cpp/yt/string/stream.cpp
new file mode 100644
index 00000000000..4bf8dcd599c
--- /dev/null
+++ b/library/cpp/yt/string/stream.cpp
@@ -0,0 +1,144 @@
+#include "stream.h"
+
+#include <util/generic/bitops.h>
+#include <util/generic/string.h>
+#include <util/generic/utility.h>
+#include <util/system/yassert.h>
+
+namespace NYT {
+
+////////////////////////////////////////////////////////////////////////////////
+
+static constexpr size_t MinBufferGrowSize = 16;
+
+////////////////////////////////////////////////////////////////////////////////
+
+TStdStringOutput::TStdStringOutput(std::string& string) noexcept
+    : String_(&string)
+{ }
+
+TStdStringOutput::~TStdStringOutput() = default;
+
+void TStdStringOutput::Reserve(size_t size)
+{
+    String_->reserve(String_->size() + size);
+}
+
+void TStdStringOutput::Swap(TStdStringOutput& other) noexcept
+{
+    DoSwap(String_, other.String_);
+}
+
+size_t TStdStringOutput::DoNext(void** ptr)
+{
+    if (String_->size() == String_->capacity()) {
+        String_->reserve(FastClp2(String_->capacity() + MinBufferGrowSize));
+    }
+    size_t previousSize = String_->size();
+    ResizeUninitialized(*String_, String_->capacity());
+    *ptr = String_->data() + previousSize;
+    return String_->size() - previousSize;
+}
+
+void TStdStringOutput::DoUndo(size_t len)
+{
+    Y_ABORT_UNLESS(len <= String_->size(), "trying to undo more bytes than actually written");
+    String_->resize(String_->size() - len);
+}
+
+void TStdStringOutput::DoWrite(const void* buf, size_t len)
+{
+    String_->append(static_cast<const char*>(buf), len);
+}
+
+void TStdStringOutput::DoWriteC(char c)
+{
+    String_->push_back(c);
+}
+
+////////////////////////////////////////////////////////////////////////////////
+
+TStdStringStream::TStdStringStream()
+    : TEmbeddedString()
+    , TStdStringOutput(*TEmbeddedString::Ptr())
+{ }
+
+TStdStringStream::TStdStringStream(const std::string& string)
+    : TEmbeddedString(string)
+    , TStdStringOutput(*TEmbeddedString::Ptr())
+{ }
+
+TStdStringStream::TStdStringStream(std::string&& string)
+    : TEmbeddedString(std::move(string))
+    , TStdStringOutput(*TEmbeddedString::Ptr())
+{ }
+
+TStdStringStream::TStdStringStream(const TStdStringStream& other)
+    : TEmbeddedString(other.Str())
+    , TStdStringOutput(*TEmbeddedString::Ptr())
+{ }
+
+TStdStringStream::TStdStringStream(TStdStringStream&& other) noexcept
+    : TEmbeddedString(std::move(other).Str())
+    , TStdStringOutput(*TEmbeddedString::Ptr())
+{ }
+
+TStdStringStream& TStdStringStream::operator=(const TStdStringStream& other)
+{
+    // The embedded string reference remains valid; only its contents change.
+    Str() = other.Str();
+    return *this;
+}
+
+TStdStringStream& TStdStringStream::operator=(TStdStringStream&& other) noexcept
+{
+    // The embedded string reference remains valid; only its contents change.
+    Str() = std::move(other).Str();
+    return *this;
+}
+
+TStdStringStream::~TStdStringStream() = default;
+
+TStdStringStream::operator bool() const noexcept
+{
+    return !Empty();
+}
+
+std::string& TStdStringStream::Str() & noexcept
+{
+    return *Ptr();
+}
+
+const std::string& TStdStringStream::Str() const& noexcept
+{
+    return *Ptr();
+}
+
+std::string&& TStdStringStream::Str() && noexcept
+{
+    return std::move(*Ptr());
+}
+
+const char* TStdStringStream::Data() const noexcept
+{
+    return Ptr()->data();
+}
+
+size_t TStdStringStream::Size() const noexcept
+{
+    return Ptr()->size();
+}
+
+bool TStdStringStream::Empty() const noexcept
+{
+    return Ptr()->empty();
+}
+
+void TStdStringStream::Clear()
+{
+    Str().clear();
+}
+
+////////////////////////////////////////////////////////////////////////////////
+
+} // namespace NYT
diff --git a/library/cpp/yt/string/stream.h b/library/cpp/yt/string/stream.h
new file mode 100644
index 00000000000..b041a5e0e49
--- /dev/null
+++ b/library/cpp/yt/string/stream.h
@@ -0,0 +1,107 @@
+#pragma once
+
+#include <util/generic/store_policy.h>
+
+#include <util/stream/zerocopy_output.h>
+
+#include <util/system/compiler.h>
+
+#include <string>
+
+namespace NYT {
+
+////////////////////////////////////////////////////////////////////////////////
+
+//! Drop-in, |std::string|-backed replacements for util's |TStringOutput| /
+//! |TStringStream| (util/stream/str.h).
+//!
+//! The util adapters are hard-wired to |TString| (they take a |TString&| / own
+//! a |TString|), which forces any string they touch to stay |TString|. These
+//! mirror the same |IOutputStream| / zero-copy interfaces over |std::string|,
+//! so a buffer that previously had to be |TString| solely to feed one of those
+//! streams can become |std::string|.
+//!
+//! Only the output/owning variants live here: for *reading* from a
+//! |std::string| / |TStringBuf|, util's |TMemoryInput| already works (it views
+//! raw bytes and involves no |TString|).
+
+//! An output stream that appends to a referenced |std::string|.
+class TStdStringOutput
+    : public IZeroCopyOutput
+{
+public:
+    //! Constructs an output stream appending to \p string.
+    /*!
+     *  As with |TStringOutput|, the stream keeps a reference to \p string, so
+     *  the caller must keep it alive while the stream is in use.
+     */
+    explicit TStdStringOutput(std::string& string Y_LIFETIME_BOUND) noexcept;
+
+    TStdStringOutput(TStdStringOutput&&) noexcept = default;
+
+    ~TStdStringOutput() override;
+
+    //! Reserves \p size additional characters in the output string.
+    void Reserve(size_t size);
+
+    void Swap(TStdStringOutput& other) noexcept;
+
+protected:
+    size_t DoNext(void** ptr) override;
+    void DoUndo(size_t len) override;
+    void DoWrite(const void* buf, size_t len) override;
+    void DoWriteC(char c) override;
+
+private:
+    std::string* String_;
+};
+
+////////////////////////////////////////////////////////////////////////////////
+
+//! An output stream that owns a |std::string| (the writing half of util's
+//! |TStringStream|). Read the accumulated bytes back via |Str()|, or wrap it in
+//! a |TMemoryInput|.
+class TStdStringStream
+    : private TEmbedPolicy<std::string>
+    , public TStdStringOutput
+{
+    using TEmbeddedString = TEmbedPolicy<std::string>;
+
+public:
+    TStdStringStream();
+    explicit TStdStringStream(const std::string& string);
+    explicit TStdStringStream(std::string&& string);
+    TStdStringStream(const TStdStringStream& other);
+    TStdStringStream(TStdStringStream&& other) noexcept;
+
+    TStdStringStream& operator=(const TStdStringStream& other);
+    TStdStringStream& operator=(TStdStringStream&& other) noexcept;
+
+    ~TStdStringStream() override;
+
+    //! Returns whether this stream contains any data.
+    explicit operator bool() const noexcept;
+
+    //! Returns the string this stream is writing into.
+    std::string& Str() & noexcept;
+    const std::string& Str() const& noexcept;
+    std::string&& Str() && noexcept;
+
+    //! Returns a pointer to the (null-terminated) character data.
+    const char* Data() const noexcept;
+
+    //! Returns the total number of characters in the stream.
+    size_t Size() const noexcept;
+
+    //! Returns whether the underlying string is empty.
+    Y_PURE_FUNCTION bool Empty() const noexcept;
+
+    using TStdStringOutput::Reserve;
+
+    //! Clears the underlying string.
+    void Clear();
+};
+
+////////////////////////////////////////////////////////////////////////////////
+
+} // namespace NYT
diff --git a/library/cpp/yt/string/unittests/stream_ut.cpp b/library/cpp/yt/string/unittests/stream_ut.cpp
new file mode 100644
index 00000000000..5f46105373c
--- /dev/null
+++ b/library/cpp/yt/string/unittests/stream_ut.cpp
@@ -0,0 +1,66 @@
+#include <library/cpp/testing/gtest/gtest.h>
+
+#include <library/cpp/yt/string/stream.h>
+
+#include <util/stream/str.h>
+
+namespace NYT {
+namespace {
+
+////////////////////////////////////////////////////////////////////////////////
+
+TEST(TStdStringOutputTest, AppendsToReferencedString)
+{
+    std::string buffer = "abc";
+    {
+        TStdStringOutput output(buffer);
+        output << "de" << 'f' << 42;
+    }
+    EXPECT_EQ(buffer, "abcdef42");
+}
+
+TEST(TStdStringOutputTest, MatchesTStringOutput)
+{
+    std::string stdBuffer;
+    TString utilBuffer;
+    TStdStringOutput stdOutput(stdBuffer);
+    TStringOutput utilOutput(utilBuffer);
+    for (int i = 0; i < 1000; ++i) {
+        stdOutput << i << ',';
+        utilOutput << i << ',';
+    }
+    EXPECT_EQ(stdBuffer, std::string(utilBuffer));
+}
+
+TEST(TStdStringStreamTest, WriteAndStr)
+{
+    TStdStringStream stream;
+    EXPECT_FALSE(static_cast<bool>(stream));
+    stream << "value=" << 123;
+    EXPECT_TRUE(static_cast<bool>(stream));
+    EXPECT_EQ(stream.Str(), "value=123");
+    EXPECT_EQ(stream.Size(), 9u);
+}
+
+TEST(TStdStringStreamTest, ClearAndReuse)
+{
+    TStdStringStream stream("seed");
+    EXPECT_EQ(stream.Str(), "seed");
+    stream.Clear();
+    EXPECT_TRUE(stream.Empty());
+    stream << "again";
+    EXPECT_EQ(stream.Str(), "again");
+}
+
+TEST(TStdStringStreamTest, MoveOutBuffer)
+{
+    TStdStringStream stream;
+    stream << "movable";
+    std::string extracted = std::move(stream).Str();
+    EXPECT_EQ(extracted, "movable");
+}
+
+////////////////////////////////////////////////////////////////////////////////
+
+} // namespace
+} // namespace NYT
diff --git a/library/cpp/yt/string/unittests/ya.make b/library/cpp/yt/string/unittests/ya.make
index 8fd64d36627..5f8d7867673 100644
--- a/library/cpp/yt/string/unittests/ya.make
+++ b/library/cpp/yt/string/unittests/ya.make
@@ -6,6 +6,7 @@ SRCS(
     enum_ut.cpp
     format_ut.cpp
     guid_ut.cpp
+    stream_ut.cpp
     string_ut.cpp
 )
 
diff --git a/library/cpp/yt/string/ya.make b/library/cpp/yt/string/ya.make
index bea7df0b1bf..bb146b64726 100644
--- a/library/cpp/yt/string/ya.make
+++ b/library/cpp/yt/string/ya.make
@@ -9,6 +9,7 @@ SRCS(
     format_string.cpp
     format.cpp
     string_builder.cpp
+    stream.cpp
 )
 
 PEERDIR(