YT-27634: Annotate DCAS with TSAN intrinsics

commit_hash:7b0bb805a82d9829ea93f5b867962e77a2c56244

2 files changed, 97 insertions, 3 deletions

diff --git a/library/cpp/yt/memory/free_list-inl.h b/library/cpp/yt/memory/free_list-inl.h
index 135eb5421b6..a7adbb0e238 100644
--- a/library/cpp/yt/memory/free_list-inl.h
+++ b/library/cpp/yt/memory/free_list-inl.h
@@ -4,6 +4,8 @@
 #include "free_list.h"
 #endif
 
+#include <util/system/sanitizers.h>
+
 namespace NYT {
 
 ////////////////////////////////////////////////////////////////////////////////
@@ -23,6 +25,16 @@ Y_FORCE_INLINE bool CasFreeListPackedPair(
     T1 new1,
     T2 new2)
 {
+#if defined(_tsan_enabled_)
+    __tsan_acquire(const_cast<unsigned __int128*>(atomic));
+    __tsan_release(const_cast<unsigned __int128*>(atomic));
+    // NB(sabdenovch): release must happen before cmpxchg to avoid the situation
+    // where producer writes to non-atomic memory does CAS,
+    // sleeps without letting thread sanitizer know about release,
+    // then consumer does CAS, informs thread sanitizer about acquire and reads non-atomic memory.
+    // No real race exists anyway, this is a workaround for TSAN false positives.
+#endif
+
 #if defined(__x86_64__)
     bool success;
     __asm__ __volatile__
@@ -153,7 +165,6 @@ TFreeList<TItem>::~TFreeList()
 
 template <class TItem>
 template <class TPredicate>
-Y_NO_SANITIZE("thread")
 bool TFreeList<TItem>::PutIf(TItem* head, TItem* tail, TPredicate predicate)
 {
     auto* current = Head_.Pointer.load(std::memory_order::relaxed);
@@ -172,7 +183,6 @@ bool TFreeList<TItem>::PutIf(TItem* head, TItem* tail, TPredicate predicate)
 }
 
 template <class TItem>
-Y_NO_SANITIZE("thread")
 void TFreeList<TItem>::Put(TItem* head, TItem* tail)
 {
     auto* current = Head_.Pointer.load(std::memory_order::relaxed);
@@ -190,7 +200,6 @@ void TFreeList<TItem>::Put(TItem* item)
 }
 
 template <class TItem>
-Y_NO_SANITIZE("thread")
 TItem* TFreeList<TItem>::Extract()
 {
     auto* current = Head_.Pointer.load(std::memory_order::relaxed);
diff --git a/library/cpp/yt/memory/unittests/free_list_ut.cpp b/library/cpp/yt/memory/unittests/free_list_ut.cpp
index 81f60b694fe..371dc6a2da8 100644
--- a/library/cpp/yt/memory/unittests/free_list_ut.cpp
+++ b/library/cpp/yt/memory/unittests/free_list_ut.cpp
@@ -155,5 +155,90 @@ INSTANTIATE_TEST_SUITE_P(
 
 ////////////////////////////////////////////////////////////////////////////////
 
+struct TFreeListItem
+    : public TFreeListItemBase<TFreeListItem>
+{
+    explicit TFreeListItem(i64 value)
+        : Value(value)
+    { }
+    i64 Value;
+};
+
+static std::chrono::steady_clock::time_point Now()
+{
+    return std::chrono::steady_clock::now();
+}
+
+void ProducerThread(TFreeList<TFreeListItem>* freeList, i64 startValue, std::chrono::steady_clock::time_point until)
+{
+    i64 v = startValue;
+    while (Now() < until) {
+        freeList->Put(new TFreeListItem(v));
+        v += 2;
+    }
+}
+
+void ConsumerThread(TFreeList<TFreeListItem>* freeList, std::chrono::steady_clock::time_point until)
+{
+    i64 latestEven = -2;
+    i64 latestOdd = -1;
+    std::vector<i64> buffer;
+    while (Now() < until) {
+        buffer.clear();
+        auto* extractedList = freeList->ExtractAll();
+        while (extractedList) {
+            buffer.push_back(extractedList->Value);
+            auto* next = extractedList->Next.load(std::memory_order::acquire);
+            delete extractedList;
+            extractedList = next;
+        }
+
+        for (auto it = buffer.rbegin(); it != buffer.rend(); ++it) {
+            auto value = *it;
+            if (value % 2 == 0) {
+                ASSERT_GT(value, latestEven);
+                latestEven = value;
+            } else {
+                ASSERT_GT(value, latestOdd);
+                latestOdd = value;
+            }
+        }
+    }
+}
+
+void Cleanup(TFreeList<TFreeListItem>* freeList)
+{
+    auto* node = freeList->ExtractAll();
+    while (node) {
+        auto* next = node->Next.load(std::memory_order::acquire);
+        delete node;
+        node = next;
+    }
+}
+
+TEST(TFreeListTest, ProducerConsumer)
+{
+    auto now = Now();
+    auto until = now + std::chrono::seconds(15);
+
+    auto freeList = TFreeList<TFreeListItem>();
+
+    std::vector<std::thread> threads;
+
+    threads.emplace_back(ProducerThread, &freeList, 0, until);
+    threads.emplace_back(ProducerThread, &freeList, 1, until);
+
+    threads.emplace_back(ConsumerThread, &freeList, until);
+    threads.emplace_back(ConsumerThread, &freeList, until);
+
+    for (auto& thread : threads) {
+        thread.join();
+    }
+
+    Cleanup(&freeList);
+}
+
+////////////////////////////////////////////////////////////////////////////////
+
 } // namespace
 } // namespace NYT