refactor utest: rm GetThreadCount

* Removed "magical" `GetThreadCount()` API in @ref UTEST macro and friends. Use @ref engine::GetWorkerCount + @ref engine::current_task::GetTaskProcessor instead.
commit_hash:29bf0432022f4cec7e3d0c71c807cddd761869ba

Author: Anton3

core/include/userver/engine/task/current_task.hpp

@@ -28,6 +28,14 @@ bool IsTaskProcessorThread() noexcept;
 /// Returns reference to the task processor executing the caller
 TaskProcessor& GetTaskProcessor();
 
+/// @brief Get the number of worker threads in the current task processor.
+///
+/// The value is configured with the `task_processors.*NAME*.worker_threads`
+/// static option of @ref components::ManagerControllerComponent.
+///
+/// @see @ref engine::GetWorkerCount
+std::size_t GetWorkerCount();
+
 /// Returns reference to the blocking task processor
 TaskProcessor& GetBlockingTaskProcessor();
 

core/include/userver/engine/task/task_processor_fwd.hpp

@@ -21,6 +21,11 @@ class TaskProcessor;
 std::size_t GetQueueSize(const TaskProcessor& task_processor) noexcept;
 
 /// @brief Get the number of worker threads in `task_processor`.
+///
+/// The value is configured with the `task_processors.*NAME*.worker_threads`
+/// static option of @ref components::ManagerControllerComponent.
+///
+/// @see @ref engine::current_task::GetWorkerCount
 std::size_t GetWorkerCount(const TaskProcessor& task_processor) noexcept;
 
 /// @brief Register a function that runs on all threads on task processor

core/src/concurrent/impl/intrusive_stack_test.cpp

@@ -8,6 +8,7 @@
 
 #include <userver/engine/async.hpp>
 #include <userver/engine/sleep.hpp>
+#include <userver/engine/task/current_task.hpp>
 #include <userver/engine/task/task_with_result.hpp>
 #include <userver/utest/utest.hpp>
 #include <userver/utils/assert.hpp>
@@ -61,7 +62,7 @@ TEST(IntrusiveStack, CanHoldSingle) {
 
 UTEST_MT(IntrusiveStack, TortureTest, 12) {
     // 'nodes' must outlive 'stack'
-    utils::FixedArray<CheckedInt> nodes(GetThreadCount() * 2);
+    utils::FixedArray<CheckedInt> nodes(engine::current_task::GetWorkerCount() * 2);
 
     CheckedIntStack stack;
     for (auto& node : nodes) {
@@ -70,9 +71,9 @@ UTEST_MT(IntrusiveStack, TortureTest, 12) {
 
     std::atomic<bool> keep_running{true};
     std::vector<engine::TaskWithResult<void>> tasks;
-    tasks.reserve(GetThreadCount() - 1);
+    tasks.reserve(engine::current_task::GetWorkerCount() - 1);
 
-    for (std::size_t i = 0; i < GetThreadCount() - 1; ++i) {
+    for (std::size_t i = 0; i < engine::current_task::GetWorkerCount() - 1; ++i) {
         tasks.push_back(engine::AsyncNoTracing([&] {
             std::vector<CheckedInt*> our_nodes;
             our_nodes.reserve(nodes.size());

core/src/concurrent/impl/striped_intrusive_pool_test.cpp

@@ -11,6 +11,7 @@
 
 #include <userver/engine/async.hpp>
 #include <userver/engine/sleep.hpp>
+#include <userver/engine/task/current_task.hpp>
 #include <userver/engine/task/task_with_result.hpp>
 #include <userver/utest/utest.hpp>
 #include <userver/utils/algo.hpp>
@@ -96,16 +97,16 @@ TEST(StripedIntrusivePool, WalkUnsafe) {
 
 UTEST_MT(StripedIntrusivePool, TortureTest, 12) {
     // 'nodes' must outlive 'stack'
-    utils::FixedArray<CheckedInt> nodes(GetThreadCount() * 2);
+    utils::FixedArray<CheckedInt> nodes(engine::current_task::GetWorkerCount() * 2);
 
     constexpr std::size_t kMinNodesPerTask = 2;
     CheckedIntPool pool;
 
     std::atomic<bool> keep_running{true};
     std::vector<engine::TaskWithResult<std::size_t>> tasks;
-    tasks.reserve(GetThreadCount() - 1);
+    tasks.reserve(engine::current_task::GetWorkerCount() - 1);
 
-    for (std::size_t i = 0; i < GetThreadCount() - 1; ++i) {
+    for (std::size_t i = 0; i < engine::current_task::GetWorkerCount() - 1; ++i) {
         tasks.push_back(engine::AsyncNoTracing([&] {
             std::size_t nodes_created = 0;
             std::vector<std::unique_ptr<CheckedInt>> nodes_we_could_pop;

core/src/concurrent/intrusive_walkable_pool_test.cpp

@@ -6,6 +6,7 @@
 #include <vector>
 
 #include <userver/engine/async.hpp>
+#include <userver/engine/task/current_task.hpp>
 #include <userver/engine/task/task_with_result.hpp>
 #include <userver/utest/utest.hpp>
 #include <userver/utils/assert.hpp>
@@ -81,9 +82,9 @@ UTEST_MT(IntrusiveWalkablePool, TortureTest, 4) {
     // 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);
+    tasks.reserve(engine::current_task::GetWorkerCount() - 1);
 
-    for (std::size_t i = 0; i < GetThreadCount() - 2; ++i) {
+    for (std::size_t i = 0; i < engine::current_task::GetWorkerCount() - 2; ++i) {
         tasks.push_back(engine::AsyncNoTracing([&] {
             CheckedInt* nodes[kNodesPerTask]{};
 

core/src/concurrent/mpsc_queue_test.cpp

@@ -5,6 +5,7 @@
 #include <userver/engine/async.hpp>
 #include <userver/engine/exception.hpp>
 #include <userver/engine/task/cancel.hpp>
+#include <userver/engine/task/current_task.hpp>
 #include <userver/logging/log.hpp>
 #include <userver/tracing/span.hpp>
 #include <userver/utest/utest.hpp>
@@ -176,7 +177,7 @@ void FooProcessor::DoProcess(const FooItem& item) { foo_items.push_back(item); }
 }  // namespace
 
 UTEST(MpscQueue, ProcessingRemainingItemsSample) {
-    ASSERT_EQ(GetThreadCount(), 1)
+    ASSERT_EQ(engine::current_task::GetWorkerCount(), 1)
         << "In this test we can observe the exact moments of task switching, because there "
            "is a single TaskProcessor thread. We also don't need protecting 'foo_items'";
     foo_items.clear();
@@ -208,7 +209,7 @@ UTEST(MpscQueue, ProcessingRemainingItemsSample) {
 }
 
 UTEST(MpscQueue, ProcessingRemainingItemsCancelled) {
-    ASSERT_EQ(GetThreadCount(), 1)
+    ASSERT_EQ(engine::current_task::GetWorkerCount(), 1)
         << "In this test we can observe the exact moments of task switching, because there "
            "is a single TaskProcessor thread. We also don't need protecting 'foo_items'";
     foo_items.clear();

core/src/concurrent/mutex_set_test.cpp

@@ -3,6 +3,7 @@
 #include <userver/concurrent/mutex_set.hpp>
 #include <userver/engine/async.hpp>
 #include <userver/engine/sleep.hpp>
+#include <userver/engine/task/current_task.hpp>
 
 USERVER_NAMESPACE_BEGIN
 
@@ -118,7 +119,7 @@ UTEST(MutexSet, Sample) {
 }
 
 UTEST_MT(MutexSet, HighContention, 4) {
-    const auto concurrent_jobs = GetThreadCount();
+    const auto concurrent_jobs = engine::current_task::GetWorkerCount();
     concurrent::MutexSet<int> ms;
 
     std::vector<engine::Task> tasks;

core/src/engine/impl/async_flat_combining_queue_test.cpp

@@ -8,6 +8,7 @@
 #include <userver/engine/async.hpp>
 #include <userver/engine/sleep.hpp>
 #include <userver/engine/task/cancel.hpp>
+#include <userver/engine/task/current_task.hpp>
 #include <userver/utest/utest.hpp>
 #include <userver/utils/fixed_array.hpp>
 
@@ -150,7 +151,7 @@ TEST(AsyncFlatCombiningQueueNoCoro, StressSync) {
 }
 
 UTEST_MT(AsyncFlatCombiningQueue, StressAsync, 3) {
-    const auto producers_count = GetThreadCount() - 1;
+    const auto producers_count = engine::current_task::GetWorkerCount() - 1;
     const auto test_deadline = engine::Deadline::FromDuration(100ms);
     constexpr auto kTimeForModeChange = 1ms;
 

core/src/engine/semaphore_test.cpp

@@ -4,6 +4,7 @@
 #include <userver/engine/deadline.hpp>
 #include <userver/engine/single_consumer_event.hpp>
 #include <userver/engine/sleep.hpp>
+#include <userver/engine/task/current_task.hpp>
 #include <userver/utest/utest.hpp>
 #include <userver/utils/async.hpp>
 #include <userver/utils/fast_scope_guard.hpp>
@@ -207,8 +208,8 @@ UTEST_MT(Semaphore, LockPassing, 4) {
 
     while (!test_deadline.IsReached()) {
         std::vector<engine::TaskWithResult<void>> tasks;
-        tasks.reserve(GetThreadCount());
-        for (size_t i = 0; i < GetThreadCount(); ++i) {
+        tasks.reserve(engine::current_task::GetWorkerCount());
+        for (size_t i = 0; i < engine::current_task::GetWorkerCount(); ++i) {
             tasks.push_back(engine::AsyncNoTracing(work));
         }
 
@@ -222,9 +223,9 @@ UTEST_MT(Semaphore, LockFastPathRace, 5) {
     const auto test_deadline = engine::Deadline::FromDuration(100ms);
     engine::Semaphore sem{-1UL};
     std::vector<engine::TaskWithResult<void>> tasks;
-    tasks.reserve(GetThreadCount());
+    tasks.reserve(engine::current_task::GetWorkerCount());
 
-    for (std::size_t i = 0; i < GetThreadCount(); ++i) {
+    for (std::size_t i = 0; i < engine::current_task::GetWorkerCount(); ++i) {
         tasks.push_back(engine::AsyncNoTracing([&] {
             std::size_t locks_taken = 0;
             const utils::FastScopeGuard unlock([&]() noexcept { sem.unlock_shared_count(locks_taken); });

core/src/engine/single_consumer_event_test.cpp

@@ -6,6 +6,7 @@
 #include <userver/engine/async.hpp>
 #include <userver/engine/sleep.hpp>
 #include <userver/engine/task/cancel.hpp>
+#include <userver/engine/task/current_task.hpp>
 #include <userver/engine/wait_any.hpp>
 #include <userver/logging/log.hpp>
 #include <userver/utest/utest.hpp>
@@ -214,7 +215,10 @@ UTEST(SingleConsumerEvent, AwaitableTokenHasSpuriousWakeups) {
     // In this example, we actually want to wait on the predicate `count >= 3`, using SCE like a condition variable.
     // This test shows how it can be done without race conditions.
 
-    ASSERT_EQ(GetThreadCount(), 1) << "This test relies on Yield passing execution to the other task";
+    ASSERT_EQ(engine::current_task::GetWorkerCount(), 1)
+        << "This test relies on Yield "
+           "passing execution to the other "
+           "task";
 
     std::atomic<int> count{0};
     int wait_iterations{0};
@@ -296,7 +300,7 @@ UTEST_MT(SingleConsumerEvent, AsConditionVariable, 4) {
     engine::SingleConsumerEvent event;
     /// [CV init]
 
-    auto incrementors = utils::GenerateFixedArray(GetThreadCount() - 1, [&](std::size_t) {
+    auto incrementors = utils::GenerateFixedArray(engine::current_task::GetWorkerCount() - 1, [&](std::size_t) {
         return engine::CriticalAsyncNoTracing([&count, &event] {
             while (!engine::current_task::ShouldCancel()) {
                 /// [CV notifier]