fix concurrent: fix PopNoblock sometimes not giving out last item

`PopNoblock` would sometimes not give out the last remaining item. This happened due to a misunderstanding of the semantics of `TryPopWeak`. The bug surfaced in the following test:

```
[ RUN      ] MpscQueue.MultiProducer
userver/core/src/concurrent/mpsc_queue_test.cpp:232: Failure
Value of: consumer.PopNoblock(value_2)
  Actual: false
  Expected: true
[  FAILED  ] MpscQueue.MultiProducer (14 ms)
```
commit_hash:e0a4d77e412d434ff2e592af92bf50cfb4941bc5

Author: Anton3

core/include/userver/concurrent/impl/intrusive_mpsc_queue.hpp

@@ -29,6 +29,13 @@ class IntrusiveMpscQueueImpl final {
     using NodePtr = SinglyLinkedBaseHook*;
     using NodeRef = utils::NotNull<NodePtr>;
 
+    enum class PopMode {
+        // See TryPopBlocking.
+        kRarelyBlocking,
+        // See TryPopWeak.
+        kWeak,
+    };
+
     constexpr IntrusiveMpscQueueImpl() = default;
 
     IntrusiveMpscQueueImpl(IntrusiveMpscQueueImpl&&) = delete;
@@ -53,7 +60,7 @@ class IntrusiveMpscQueueImpl final {
     // Can only be called from one thread at a time.
     NodePtr TryPopBlocking() noexcept;
 
-    // Returns the oldest pushed not, or `nullptr` if the queue "seems to be"
+    // Returns the oldest pushed node, or `nullptr` if the queue "seems to be"
     // empty. Can only be called from one thread at a time.
     // Unlike TryPopBlocking, never blocks, but this comes at a cost: it might
     // not return an item that has been completely pushed (happens-before).
@@ -64,6 +71,9 @@ class IntrusiveMpscQueueImpl final {
     // responsible for pushing both items, and the other producer walks away,
     // and for them the push operation is essentially pushed asynchronously.
     //
+    // In addition to that, the last remaining item in the queue may not be
+    // returned if the push of a second item has started.
+    //
     // If the producers always notify the consumer after pushing,
     // then TryPopWeak is enough: the consumer will be notified of all
     // the pushed items by some of the producers.
@@ -74,16 +84,12 @@ class IntrusiveMpscQueueImpl final {
     // upon the notification.
     NodePtr TryPopWeak() noexcept;
 
-private:
-    enum class PopMode {
-        kRarelyBlocking,
-        kWeak,
-    };
+    // See TryPopBlocking and TryPopWeak.
+    NodePtr TryPop(PopMode pop_mode) noexcept;
 
+private:
     static std::atomic<NodePtr>& GetNext(NodeRef node) noexcept;
 
-    NodePtr DoTryPop(PopMode) noexcept;
-
     // This node is put into the queue when it would otherwise be empty.
     SinglyLinkedBaseHook stub_;
     // Points to the oldest node not yet popped by the consumer,
@@ -109,6 +115,8 @@ class IntrusiveMpscQueue final {
 
     T* TryPopWeak() noexcept { return static_cast<T*>(impl_.TryPopWeak()); }
 
+    T* TryPop(IntrusiveMpscQueueImpl::PopMode pop_mode) noexcept { return static_cast<T*>(impl_.TryPop(pop_mode)); }
+
 private:
     IntrusiveMpscQueueImpl impl_;
 };

core/include/userver/concurrent/mpsc_queue.hpp

@@ -125,7 +125,7 @@ class MpscQueue final : public std::enable_shared_from_this<MpscQueue<T>> {
 
     bool Pop(ConsumerToken&, T&, engine::Deadline);
     bool PopNoblock(ConsumerToken&, T&);
-    bool DoPop(ConsumerToken&, T&);
+    bool DoPop(ConsumerToken&, T&, impl::IntrusiveMpscQueueImpl::PopMode);
 
     void MarkConsumerIsDead();
     void MarkProducerIsDead();
@@ -224,14 +224,16 @@ 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, [&] {
-        if (DoPop(token, value)) {
+        // 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)) {
             return true;
         }
         if (NoMoreProducers()) {
             // Producer might have pushed something in queue between .pop()
             // and !producer_is_created_and_dead_ check. Check twice to avoid
             // TOCTOU.
-            if (!DoPop(token, value)) {
+            if (!DoPop(token, value, impl::IntrusiveMpscQueueImpl::PopMode::kRarelyBlocking)) {
                 no_more_producers = true;
             }
             return true;
@@ -243,12 +245,18 @@ bool MpscQueue<T>::Pop(ConsumerToken& token, T& value, engine::Deadline deadline
 
 template <typename T>
 bool MpscQueue<T>::PopNoblock(ConsumerToken& token, T& value) {
-    return DoPop(token, value);
+    // kRarelyBlocking is required here, because with kWeak we sometimes would miss an item if another push
+    // is in process, and there is no guarantee that the user will retry PopNoblock.
+    //
+    // If there was a high-level consumer API that is not affected by kWeak (e.g. some batching API),
+    // then it could be used there.
+    // As it stands, it would be too bug-prone to provide weak guarantees in PopNoblock.
+    return DoPop(token, value, impl::IntrusiveMpscQueueImpl::PopMode::kRarelyBlocking);
 }
 
 template <typename T>
-bool MpscQueue<T>::DoPop(ConsumerToken& /*unused*/, T& value) {
-    if (const auto node = std::unique_ptr<Node>{queue_.TryPopWeak()}) {
+bool MpscQueue<T>::DoPop(ConsumerToken& /*unused*/, T& value, impl::IntrusiveMpscQueueImpl::PopMode pop_mode) {
+    if (const auto node = std::unique_ptr<Node>{queue_.TryPop(pop_mode)}) {
         value = std::move(node->value);
 
         --size_;

core/src/concurrent/impl/intrusive_mpsc_queue.cpp

@@ -54,12 +54,12 @@ bool IntrusiveMpscQueueImpl::PushIfEmpty(NodeRef node) noexcept {
 }
 
 IntrusiveMpscQueueImpl::NodePtr IntrusiveMpscQueueImpl::TryPopBlocking() noexcept {
-    return DoTryPop(PopMode::kRarelyBlocking);
+    return TryPop(PopMode::kRarelyBlocking);
 }
 
-IntrusiveMpscQueueImpl::NodePtr IntrusiveMpscQueueImpl::TryPopWeak() noexcept { return DoTryPop(PopMode::kWeak); }
+IntrusiveMpscQueueImpl::NodePtr IntrusiveMpscQueueImpl::TryPopWeak() noexcept { return TryPop(PopMode::kWeak); }
 
-IntrusiveMpscQueueImpl::NodePtr IntrusiveMpscQueueImpl::DoTryPop(PopMode mode) noexcept {
+IntrusiveMpscQueueImpl::NodePtr IntrusiveMpscQueueImpl::TryPop(PopMode mode) noexcept {
     UASSERT_MSG(!is_consuming_.exchange(true), "Multiple concurrent consumers detected");
     const utils::FastScopeGuard guard([this]() noexcept {
         UASSERT_MSG(is_consuming_.exchange(false), "Multiple concurrent consumers detected");