cc tests: make userver tests friendly to thread-pinning scheduler

commit_hash:179b671cc60f2b270fadd79dad3b5d18cac253b7

Author: Anton3

core/src/concurrent/impl/striped_read_indicator_test.cpp

@@ -1,16 +1,17 @@
 #include <userver/concurrent/impl/striped_read_indicator.hpp>
 
+#include <chrono>
 #include <thread>
+#include <vector>
 
-#include <userver/compiler/impl/tsan.hpp>
 #include <userver/engine/async.hpp>
+#include <userver/engine/deadline.hpp>
 #include <userver/engine/mutex.hpp>
-#include <userver/engine/sleep.hpp>
+#include <userver/engine/task/task_with_result.hpp>
 #include <userver/utest/utest.hpp>
 
 USERVER_NAMESPACE_BEGIN
 
-#if !USERVER_IMPL_HAS_TSAN
 namespace {
 constexpr std::size_t kReadersCount = 3;
 constexpr std::size_t kCheckersCount = 1;
@@ -21,13 +22,13 @@ UTEST_MT(StripedReadIndicator, LockPassingStress, kReadersCount + kCheckersCount
     concurrent::impl::StripedReadIndicatorLock indicator_lock = indicator.GetLock();
     engine::Mutex ping_pong_mutex;
 
-    std::atomic<bool> keep_running{true};
+    const auto test_deadline = engine::Deadline::FromDuration(std::chrono::milliseconds{100});
     std::vector<engine::TaskWithResult<void>> tasks;
     tasks.reserve(kReadersCount + kCheckersCount);
 
     for (std::size_t i = 0; i < kReadersCount; ++i) {
         tasks.push_back(engine::AsyncNoTracing([&] {
-            while (keep_running) {
+            while (!test_deadline.IsReached()) {
                 const std::lock_guard ping_pong_mutex_lock{ping_pong_mutex};
                 auto lock_copy = indicator_lock;
                 indicator_lock = std::move(lock_copy);
@@ -39,22 +40,19 @@ UTEST_MT(StripedReadIndicator, LockPassingStress, kReadersCount + kCheckersCount
 
     for (std::size_t i = 0; i < kCheckersCount; ++i) {
         tasks.push_back(engine::AsyncNoTracing([&] {
-            while (keep_running) {
+            while (!test_deadline.IsReached()) {
                 ASSERT_FALSE(indicator.IsFree());
             }
         }));
     }
 
-    engine::SleepFor(std::chrono::milliseconds{100});
-    keep_running = false;
     for (auto& task : tasks) {
         task.Get();
     }
 
     indicator_lock = concurrent::impl::StripedReadIndicatorLock{};
     EXPECT_TRUE(indicator.IsFree());
 }
-#endif
 
 UTEST(StripedReadIndicator, Metrics) {
     concurrent::impl::StripedReadIndicator indicator;

core/src/concurrent/intrusive_walkable_pool_test.cpp

@@ -1,13 +1,11 @@
 #include <concurrent/intrusive_walkable_pool.hpp>
 
-#include <atomic>
 #include <cstddef>
 #include <thread>
 #include <unordered_set>
 #include <vector>
 
 #include <userver/engine/async.hpp>
-#include <userver/engine/sleep.hpp>
 #include <userver/engine/task/task_with_result.hpp>
 #include <userver/utest/utest.hpp>
 #include <userver/utils/assert.hpp>
@@ -79,15 +77,17 @@ UTEST_MT(IntrusiveWalkablePool, TortureTest, 4) {
     constexpr std::size_t kNodesPerTask = 3;
     CheckedIntPool pool;
 
-    std::atomic<bool> keep_running{true};
+    // Let worker tasks stop by their own deadline: with a pinning queue the
+    // main test task may not resume promptly enough to flip a shared stop flag.
+    const auto test_deadline = engine::Deadline::FromDuration(50ms);
     std::vector<engine::TaskWithResult<void>> tasks;
     tasks.reserve(GetThreadCount() - 1);
 
     for (std::size_t i = 0; i < GetThreadCount() - 2; ++i) {
         tasks.push_back(engine::AsyncNoTracing([&] {
             CheckedInt* nodes[kNodesPerTask]{};
 
-            while (keep_running) {
+            while (!test_deadline.IsReached()) {
                 for (auto*& node : nodes) {
                     node = &pool.Acquire();
                     node->CheckAlive();
@@ -101,7 +101,7 @@ UTEST_MT(IntrusiveWalkablePool, TortureTest, 4) {
     }
 
     tasks.push_back(engine::AsyncNoTracing([&] {
-        while (keep_running) {
+        while (!test_deadline.IsReached()) {
             std::size_t node_count = 0;
             pool.Walk([&](CheckedInt& node) {
                 node.CheckAlive();
@@ -111,8 +111,6 @@ UTEST_MT(IntrusiveWalkablePool, TortureTest, 4) {
         }
     }));
 
-    engine::SleepFor(50ms);
-    keep_running = false;
     for (auto& task : tasks) {
         task.Get();
     }

core/src/engine/errno_test.cpp

@@ -1,14 +1,12 @@
 #include <userver/utest/utest.hpp>
 
-#include <atomic>
 #include <cerrno>
 #include <chrono>
 #include <condition_variable>
 #include <mutex>
 #include <thread>
 #include <vector>
 
-#include <userver/compiler/impl/tsan.hpp>
 #include <userver/engine/async.hpp>
 #include <userver/engine/deadline.hpp>
 #include <userver/engine/sleep.hpp>
@@ -17,14 +15,14 @@
 
 USERVER_NAMESPACE_BEGIN
 
-// Test is not ready to TSan non-migrating scheduler
-#if !USERVER_IMPL_HAS_TSAN
 namespace {
-constexpr std::size_t kNumThreads = 2;
+constexpr std::size_t kNumThreads = 3;
 }  // namespace
 
 UTEST_MT(Errno, IsCoroLocal, kNumThreads) {
-    const auto deadline = engine::Deadline::FromDuration(utest::kMaxTestWaitTime);
+    // A pinning queue is allowed to keep a coroutine on its original worker.
+    // The test should verify errno after an actual switch, not require one.
+    const auto deadline = engine::Deadline::FromDuration(std::chrono::milliseconds{100});
 
     std::size_t threads_started{0};
     std::size_t threads_switched{0};
@@ -61,14 +59,13 @@ UTEST_MT(Errno, IsCoroLocal, kNumThreads) {
                     engine::Yield();
                 }
             }
-            return false;
+            return true;
         }));
     }
 
     for (auto& task : tasks) {
         EXPECT_TRUE(task.Get());
     }
 }
-#endif
 
 USERVER_NAMESPACE_END

core/src/engine/single_use_event_test.cpp

@@ -75,7 +75,6 @@ UTEST(SingleUseEvent, Sample) {
     sender.Get();
 }
 
-#if !USERVER_IMPL_HAS_TSAN
 UTEST_MT(SingleUseEvent, SimpleTaskQueue, 5) {
     struct SimpleTask final {
         std::uint64_t request;
@@ -85,17 +84,19 @@ UTEST_MT(SingleUseEvent, SimpleTaskQueue, 5) {
 
     boost::lockfree::queue<SimpleTask*> task_queue{1};
 
-    std::atomic<bool> keep_running_clients{true};
     std::atomic<bool> keep_running_server{true};
 
+    // Clients must not rely on the main test task resuming to flip a stop flag:
+    // pinning queues can keep the main task behind a busy worker-local queue.
+    const auto test_deadline = engine::Deadline::FromDuration(50ms);
     std::vector<engine::TaskWithResult<void>> client_tasks;
     client_tasks.reserve(GetThreadCount() - 1);
 
     for (std::size_t i = 0; i < GetThreadCount() - 1; ++i) {
         client_tasks.push_back(utils::Async("client", [&, i] {
             std::uint64_t request = i;
 
-            while (keep_running_clients) {
+            while (!test_deadline.IsReached()) {
                 SimpleTask task{request, {}, {}};
                 task_queue.push(&task);
                 task.completion.WaitNonCancellable();
@@ -110,6 +111,9 @@ UTEST_MT(SingleUseEvent, SimpleTaskQueue, 5) {
         while (keep_running_server) {
             SimpleTask* task{};
             if (!task_queue.pop(task)) {
+                // Avoid burning the worker forever while pinned client tasks are
+                // waiting for their own worker-local queues to make progress.
+                engine::Yield();
                 continue;
             }
 
@@ -118,16 +122,12 @@ UTEST_MT(SingleUseEvent, SimpleTaskQueue, 5) {
         }
     });
 
-    engine::SleepFor(50ms);
-
-    keep_running_clients = false;
     for (auto& task : client_tasks) {
         task.Get();
     }
     keep_running_server = false;
     server_task.Get();
 }
-#endif
 
 UTEST(SingleUseEvent, Cancellation) {
     engine::SingleUseEvent event;

core/src/rcu/rcu_map_test.cpp

@@ -8,7 +8,7 @@
 #include <mutex>
 #include <thread>
 
-#include <userver/compiler/impl/tsan.hpp>
+#include <userver/engine/deadline.hpp>
 #include <userver/engine/sleep.hpp>
 #include <userver/utest/utest.hpp>
 #include <userver/utils/algo.hpp>
@@ -161,16 +161,15 @@ UTEST(RcuMap, Snapshot) {
     EXPECT_EQ(2, *second_snap.at("b"));
 }
 
-#if !USERVER_IMPL_HAS_TSAN
 UTEST_MT(RcuMap, ConcurrentUpdates, 4) {
     rcu::RcuMap<int, std::atomic<uint32_t>> map;
     std::array<engine::TaskWithResult<void>, 4> workers;
-    std::atomic<bool> stop_flag{false};
+    const auto test_deadline = engine::Deadline::FromDuration(std::chrono::milliseconds{100});
 
     for (size_t i = 0; i < workers.size(); ++i) {
-        workers[i] = utils::Async("writer", [i, &map, &stop_flag] {
+        workers[i] = utils::Async("writer", [i, &map, &test_deadline] {
             const uint32_t mask = 0xFFu << (i * 8);
-            while (!stop_flag) {
+            while (!test_deadline.IsReached()) {
                 *map[i << 8] = -1;
                 for (uint8_t v = 1; v != 0; ++v) {
                     ASSERT_TRUE(map.Get((i << 8) + v - 1));
@@ -191,8 +190,6 @@ UTEST_MT(RcuMap, ConcurrentUpdates, 4) {
         });
     }
 
-    engine::SleepFor(std::chrono::milliseconds(100));
-    stop_flag = true;
     for (auto& w : workers) {
         w.Get();
     }
@@ -214,12 +211,12 @@ UTEST_MT(RcuMap, ConcurrentTryEmplace, 16) {
         tasks.reserve(kTaskCount);
         for (std::size_t i = 0; i < kTaskCount; ++i) {
             tasks.push_back(engine::AsyncNoTracing([&map, &insertions, i] {
-                auto key = std::string(20 + i / 2, 'x');
+                auto key = std::string(20 + (i / 2), 'x');
                 auto res = map.TryEmplace(key, i);
                 if (res.inserted) {
                     ++insertions;
                 }
-                EXPECT_EQ(*res.value / 2, i / 2);
+                EXPECT_EQ((*res.value) / 2, i / 2);
             }));
         }
         for (auto& task : tasks) {
@@ -228,7 +225,6 @@ UTEST_MT(RcuMap, ConcurrentTryEmplace, 16) {
         EXPECT_EQ(insertions, kTaskCount / 2);
     }
 }
-#endif
 
 UTEST(RcuMap, IterStability) {
     rcu::RcuMap<int, int> map;

core/src/rcu/rcu_test.cpp

@@ -418,17 +418,19 @@ constexpr std::size_t kTotalTasks = kReadablePtrPingPongTasks + kReadingTasks +
 
 UTEST_MT(Rcu, TortureTest, kTotalTasks) {
     rcu::Variable<CleaningUpInt> data{1};
-    std::atomic<bool> keep_running{true};
 
     engine::Mutex ping_pong_mutex;
     rcu::ReadablePtr<CleaningUpInt> ptr = data.Read();
 
+    // Pinning queues may delay the main test task while worker-local queues are
+    // busy, so the stress tasks must be able to finish without an external flag.
+    const auto test_deadline = engine::Deadline::FromDuration(std::chrono::milliseconds{100});
     std::vector<engine::TaskWithResult<void>> tasks;
     tasks.reserve(kTotalTasks - 1);
 
     for (std::size_t i = 0; i < kReadablePtrPingPongTasks; ++i) {
         tasks.push_back(engine::AsyncNoTracing([&] {
-            while (keep_running) {
+            while (!test_deadline.IsReached()) {
                 {
                     const std::lock_guard lock(ping_pong_mutex);
                     // copy a ptr created by another thread
@@ -442,7 +444,7 @@ UTEST_MT(Rcu, TortureTest, kTotalTasks) {
 
     for (std::size_t i = 0; i < kReadingTasks; ++i) {
         tasks.push_back(engine::AsyncNoTracing([&] {
-            while (keep_running) {
+            while (!test_deadline.IsReached()) {
                 const auto local_ptr = data.Read();
                 ASSERT_GT(local_ptr->value, 0);
             }
@@ -451,15 +453,16 @@ UTEST_MT(Rcu, TortureTest, kTotalTasks) {
 
     for (std::size_t i = 0; i < kWritingTasks; ++i) {
         tasks.push_back(engine::AsyncNoTracing([&] {
-            while (keep_running) {
+            while (!test_deadline.IsReached()) {
                 const auto old = data.Read();
                 data.Assign(CleaningUpInt{old->value + 1});
             }
         }));
     }
 
-    engine::SleepFor(std::chrono::milliseconds{100});
-    keep_running = false;
+    for (auto& task : tasks) {
+        task.Get();
+    }
 }
 
 UTEST(Rcu, WritablePtrUnlocksInCommit) {