feat engine: return FutureStatus from SingleConsumerEvent::WaitUntil

Breaking change:

* @ref engine::SingleConsumerEvent::WaitUntil now returns @ref engine::FutureStatus instead of `bool`, allowing to distinguish failure reason precisely.
commit_hash:14c9786711ae95baf432063ef2317df52c68b8ef

Author: Anton3

core/include/userver/concurrent/mpsc_queue.hpp

@@ -223,7 +223,7 @@ bool MpscQueue<T>::DoPush(ProducerToken& /*unused*/, T&& value) {
 template <typename T>
 bool MpscQueue<T>::Pop(ConsumerToken& token, T& value, engine::Deadline deadline) {
     bool no_more_producers = false;
-    const bool success = nonempty_event_.WaitUntil(deadline, [&] {
+    const auto wait_status = nonempty_event_.WaitUntil(deadline, [&] {
         // kWeak is OK here, because if there is another push operation in process,
         // they will notify us after pushing.
         if (DoPop(token, value, impl::IntrusiveMpscQueueImpl::PopMode::kWeak)) {
@@ -240,7 +240,7 @@ bool MpscQueue<T>::Pop(ConsumerToken& token, T& value, engine::Deadline deadline
         }
         return false;
     });
-    return success && !no_more_producers;
+    return (wait_status == engine::FutureStatus::kReady) && !no_more_producers;
 }
 
 template <typename T>

core/include/userver/concurrent/queue.hpp

@@ -418,7 +418,7 @@ class GenericQueue<T, QueuePolicy>::SingleProducerSide final {
     template <typename Token>
     [[nodiscard]] bool Push(Token& token, T&& value, engine::Deadline deadline, std::size_t value_size) {
         bool no_more_consumers = false;
-        const bool success = non_full_event_.WaitUntil(deadline, [&] {
+        const auto wait_status = non_full_event_.WaitUntil(deadline, [&] {
             if (queue_.NoMoreConsumers()) {
                 no_more_consumers = true;
                 return true;
@@ -428,7 +428,7 @@ class GenericQueue<T, QueuePolicy>::SingleProducerSide final {
             }
             return false;
         });
-        return success && !no_more_consumers;
+        return (wait_status == engine::FutureStatus::kReady) && !no_more_consumers;
     }
 
     template <typename Token>
@@ -579,7 +579,7 @@ class GenericQueue<T, QueuePolicy>::SingleConsumerSide final {
     template <typename Token>
     [[nodiscard]] bool Pop(Token& token, T& value, engine::Deadline deadline) {
         bool no_more_producers = false;
-        const bool success = nonempty_event_.WaitUntil(deadline, [&] {
+        const auto wait_status = nonempty_event_.WaitUntil(deadline, [&] {
             if (DoPop(token, value)) {
                 return true;
             }
@@ -594,7 +594,7 @@ class GenericQueue<T, QueuePolicy>::SingleConsumerSide final {
             }
             return false;
         });
-        return success && !no_more_producers;
+        return (wait_status == engine::FutureStatus::kReady) && !no_more_producers;
     }
 
     template <typename Token>

core/include/userver/engine/single_consumer_event.hpp

@@ -89,8 +89,11 @@ class SingleConsumerEvent final {
     ///
     /// Waiter side:
     /// @snippet engine/single_consumer_event_test.cpp  CV waiter
+    ///
+    /// @return `FutureStatus::kReady` if @a stop_waiting became `true`, `FutureStatus::kCancelled` if the current
+    /// task was cancelled, `FutureStatus::kTimeout` if the deadline was reached.
     template <typename Predicate>
-    [[nodiscard]] bool WaitUntil(Deadline, Predicate stop_waiting);
+    [[nodiscard]] FutureStatus WaitUntil(Deadline, Predicate stop_waiting);
 
     /// Resets the signal flag, if there is any existing event. Guarantees at least 'acquire' and 'release'
     /// memory ordering. Must only be called by the waiting task.
@@ -139,7 +142,7 @@ bool SingleConsumerEvent::WaitForEventUntil(std::chrono::time_point<Clock, Durat
 }
 
 template <typename Predicate>
-bool SingleConsumerEvent::WaitUntil(Deadline deadline, Predicate stop_waiting) {
+FutureStatus SingleConsumerEvent::WaitUntil(Deadline deadline, Predicate stop_waiting) {
     // If the state, according to what we've been previously notified of via
     // 'Send', is OK, then return right away. Fresh state updates can also
     // leak to us here, but we should not rely on it.
@@ -151,8 +154,8 @@ bool SingleConsumerEvent::WaitUntil(Deadline deadline, Predicate stop_waiting) {
         // We may also receive false signals from cases when we are allowed
         // and unallowed to make progress in a rapid sequence, or when the notifier
         // thinks that we might be happy with the state, but we aren't.
-        if (!WaitForEventUntil(deadline)) {
-            return false;
+        if (const auto status = WaitUntil(deadline); status != FutureStatus::kReady) {
+            return status;
         }
 
         if (!IsAutoReset()) {
@@ -163,7 +166,7 @@ bool SingleConsumerEvent::WaitUntil(Deadline deadline, Predicate stop_waiting) {
         }
     }
 
-    return true;
+    return FutureStatus::kReady;
 }
 
 }  // namespace engine

core/src/engine/single_consumer_event_test.cpp

@@ -337,14 +337,14 @@ UTEST_MT(SingleConsumerEvent, AsConditionVariable, 4) {
 
     /// [CV waiter]
     std::uint64_t count_acquired{};
-    const bool success = event.WaitUntil({}, [&] {
+    const auto wait_status = event.WaitUntil({}, [&] {
         // Operations must be atomic, can be std::memory_order_relaxed.
         count_acquired = count.load(std::memory_order_relaxed);
         return count_acquired % 2 == 0 && count.compare_exchange_strong(count_acquired, 0, std::memory_order_relaxed);
     });
     /// [CV waiter]
 
-    EXPECT_TRUE(success);
+    EXPECT_EQ(wait_status, engine::FutureStatus::kReady);
     EXPECT_TRUE(count_acquired != 0);
     EXPECT_TRUE(count_acquired % 2 == 0);
 

core/src/engine/wait_all_checked.cpp

@@ -160,12 +160,12 @@ FutureStatus WaitAllCheckedContext::Impl::WaitUntil(Deadline deadline) {
         }
     }
 
-    if (!result_ready_.WaitUntil(deadline, [this]() {
-            return exception_source_index_.load(std::memory_order_relaxed) != kNoExceptionSource ||
-                   pending_notifications_.load(std::memory_order_relaxed) == 0;
-        }))
-    {
-        return current_task::ShouldCancel() ? FutureStatus::kCancelled : FutureStatus::kTimeout;
+    const auto wait_status = result_ready_.WaitUntil(deadline, [this] {
+        return exception_source_index_.load(std::memory_order_relaxed) != kNoExceptionSource ||
+               pending_notifications_.load(std::memory_order_relaxed) == 0;
+    });
+    if (wait_status != FutureStatus::kReady) {
+        return wait_status;
     }
     auto exception_source_index = exception_source_index_.exchange(kNoExceptionSource, std::memory_order_relaxed);
     if (exception_source_index != kNoExceptionSource) {

core/src/server/net/listener_impl.cpp

@@ -82,7 +82,7 @@ void ListenerImpl::StopListening() {
 bool ListenerImpl::WaitForNoConnections(engine::Deadline deadline) const {
     return endpoint_info_->no_connections_event.WaitUntil(deadline, [this] {
         return endpoint_info_->connection_count == 0;
-    });
+    }) == engine::FutureStatus::kReady;
 }
 
 void ListenerImpl::AcceptConnection(engine::io::Socket& request_socket, const PortConfig& port_config) {

core/src/utils/impl/wait_token_storage.cpp

@@ -109,10 +109,10 @@ void WaitTokenStorage::WaitForAllTokens() noexcept {
     }
 
     const engine::TaskCancellationBlocker cancel_blocker;
-    const bool wait_success = impl_.tokens_is_free_event.WaitUntil(engine::Deadline{}, [this] {
+    const auto wait_status = impl_.tokens_is_free_event.WaitUntil(engine::Deadline{}, [this] {
         return impl_.tokens.IsFree();
     });
-    UASSERT(wait_success);
+    UASSERT(wait_status == engine::FutureStatus::kReady);
 }
 
 void WaitTokenStorage::DoLock() {