7#include <moodycamel/concurrentqueue.h>
9#include <userver/concurrent/impl/semaphore_capacity_control.hpp>
10#include <userver/concurrent/queue_helpers.hpp>
11#include <userver/engine/deadline.hpp>
12#include <userver/engine/semaphore.hpp>
13#include <userver/engine/single_consumer_event.hpp>
14#include <userver/engine/task/cancel.hpp>
15#include <userver/utils/assert.hpp>
16#include <userver/utils/atomic.hpp>
18USERVER_NAMESPACE_BEGIN
24template <
bool MultipleProducer,
bool MultipleConsumer>
25struct SimpleQueuePolicy {
27 static constexpr std::size_t GetElementSize(
const T&) {
31 static constexpr bool kIsMultipleProducer{MultipleProducer};
32 static constexpr bool kIsMultipleConsumer{MultipleConsumer};
35template <
bool MultipleProducer,
bool MultipleConsumer>
36struct ContainerQueuePolicy {
38 static std::size_t GetElementSize(
const T& value) {
39 return std::size(value);
42 static constexpr bool kIsMultipleProducer{MultipleProducer};
43 static constexpr bool kIsMultipleConsumer{MultipleConsumer};
51template <
typename T,
typename QueuePolicy>
52class GenericQueue
final
53 :
public std::enable_shared_from_this<GenericQueue<T, QueuePolicy>> {
54 struct EmplaceEnabler
final {
56 explicit EmplaceEnabler() =
default;
60 std::conditional_t<QueuePolicy::kIsMultipleProducer,
61 moodycamel::ProducerToken, impl::NoToken>;
63 std::conditional_t<QueuePolicy::kIsMultipleProducer,
64 moodycamel::ConsumerToken, impl::NoToken>;
65 using MultiProducerToken = impl::MultiToken;
66 using MultiConsumerToken =
67 std::conditional_t<QueuePolicy::kIsMultipleProducer, impl::MultiToken,
70 using SingleProducerToken =
71 std::conditional_t<!QueuePolicy::kIsMultipleProducer,
72 moodycamel::ProducerToken, impl::NoToken>;
74 friend class Producer<GenericQueue, ProducerToken, EmplaceEnabler>;
75 friend class Producer<GenericQueue, MultiProducerToken, EmplaceEnabler>;
76 friend class Consumer<GenericQueue, ConsumerToken, EmplaceEnabler>;
77 friend class Consumer<GenericQueue, MultiConsumerToken, EmplaceEnabler>;
83 concurrent::Producer<GenericQueue, ProducerToken, EmplaceEnabler>;
85 concurrent::Consumer<GenericQueue, ConsumerToken, EmplaceEnabler>;
87 concurrent::Producer<GenericQueue, MultiProducerToken, EmplaceEnabler>;
89 concurrent::Consumer<GenericQueue, MultiConsumerToken, EmplaceEnabler>;
91 static constexpr std::size_t kUnbounded =
92 std::numeric_limits<std::size_t>::max() / 4;
96 explicit GenericQueue(std::size_t max_size, EmplaceEnabler )
98 single_producer_token_(queue_),
99 producer_side_(*
this, std::min(max_size, kUnbounded)),
100 consumer_side_(*
this) {}
103 UASSERT(consumers_count_ == kCreatedAndDead || !consumers_count_);
104 UASSERT(producers_count_ == kCreatedAndDead || !producers_count_);
106 if (producers_count_ == kCreatedAndDead) {
108 consumer_side_.ResumeBlockingOnPop();
113 ConsumerToken token{queue_};
114 while (consumer_side_.PopNoblock(token, value)) {
118 GenericQueue(GenericQueue&&) =
delete;
119 GenericQueue(
const GenericQueue&) =
delete;
120 GenericQueue& operator=(GenericQueue&&) =
delete;
121 GenericQueue& operator=(
const GenericQueue&) =
delete;
125 static std::shared_ptr<GenericQueue>
Create(
126 std::size_t max_size = kUnbounded) {
127 return std::make_shared<GenericQueue>(max_size, EmplaceEnabler{});
138 return Producer(
this->shared_from_this(), EmplaceEnabler{});
151 static_assert(QueuePolicy::kIsMultipleProducer,
152 "Trying to obtain MultiProducer for a single-producer queue");
154 return MultiProducer(
this->shared_from_this(), EmplaceEnabler{});
165 return Consumer(
this->shared_from_this(), EmplaceEnabler{});
178 static_assert(QueuePolicy::kIsMultipleConsumer,
179 "Trying to obtain MultiConsumer for a single-consumer queue");
181 return MultiConsumer(
this->shared_from_this(), EmplaceEnabler{});
187 producer_side_.SetSoftMaxSize(std::min(max_size, kUnbounded));
191 std::size_t
GetSoftMaxSize()
const {
return producer_side_.GetSoftMaxSize(); }
195 return producer_side_.GetSizeApproximate();
199 class SingleProducerSide;
200 class MultiProducerSide;
201 class SingleConsumerSide;
202 class MultiConsumerSide;
207 std::conditional_t<QueuePolicy::kIsMultipleProducer, MultiProducerSide,
213 std::conditional_t<QueuePolicy::kIsMultipleConsumer, MultiConsumerSide,
216 template <
typename Token>
217 [[nodiscard]]
bool Push(Token& token, T&& value, engine::Deadline deadline) {
218 return producer_side_.Push(token, std::move(value), deadline);
221 template <
typename Token>
222 [[nodiscard]]
bool PushNoblock(Token& token, T&& value) {
223 return producer_side_.PushNoblock(token, std::move(value));
226 template <
typename Token>
227 [[nodiscard]]
bool Pop(Token& token, T& value, engine::Deadline deadline) {
228 return consumer_side_.Pop(token, value, deadline);
231 template <
typename Token>
232 [[nodiscard]]
bool PopNoblock(Token& token, T& value) {
233 return consumer_side_.PopNoblock(token, value);
236 void PrepareProducer() {
237 std::size_t old_producers_count{};
238 utils::AtomicUpdate(producers_count_, [&](
auto old_value) {
239 UINVARIANT(QueuePolicy::kIsMultipleProducer || old_value != 1,
240 "Incorrect usage of queue producers");
241 old_producers_count = old_value;
242 return old_value == kCreatedAndDead ? 1 : old_value + 1;
245 if (old_producers_count == kCreatedAndDead) {
246 consumer_side_.ResumeBlockingOnPop();
250 void PrepareConsumer() {
251 std::size_t old_consumers_count{};
252 utils::AtomicUpdate(consumers_count_, [&](
auto old_value) {
253 UINVARIANT(QueuePolicy::kIsMultipleConsumer || old_value != 1,
254 "Incorrect usage of queue consumers");
255 old_consumers_count = old_value;
256 return old_value == kCreatedAndDead ? 1 : old_value + 1;
259 if (old_consumers_count == kCreatedAndDead) {
260 producer_side_.ResumeBlockingOnPush();
264 void MarkConsumerIsDead() {
265 const auto new_consumers_count =
266 utils::AtomicUpdate(consumers_count_, [](
auto old_value) {
267 return old_value == 1 ? kCreatedAndDead : old_value - 1;
269 if (new_consumers_count == kCreatedAndDead) {
270 producer_side_.StopBlockingOnPush();
274 void MarkProducerIsDead() {
275 const auto new_producers_count =
276 utils::AtomicUpdate(producers_count_, [](
auto old_value) {
277 return old_value == 1 ? kCreatedAndDead : old_value - 1;
279 if (new_producers_count == kCreatedAndDead) {
280 consumer_side_.StopBlockingOnPop();
286 bool NoMoreConsumers()
const {
return consumers_count_ == kCreatedAndDead; }
288 bool NoMoreProducers()
const {
return producers_count_ == kCreatedAndDead; }
292 template <
typename Token>
293 void DoPush(Token& token, T&& value) {
294 if constexpr (std::is_same_v<Token, moodycamel::ProducerToken>) {
295 static_assert(QueuePolicy::kIsMultipleProducer);
296 queue_.enqueue(token, std::move(value));
297 }
else if constexpr (std::is_same_v<Token, MultiProducerToken>) {
298 static_assert(QueuePolicy::kIsMultipleProducer);
299 queue_.enqueue(std::move(value));
301 static_assert(std::is_same_v<Token, impl::NoToken>);
302 static_assert(!QueuePolicy::kIsMultipleProducer);
303 queue_.enqueue(single_producer_token_, std::move(value));
306 consumer_side_.OnElementPushed();
309 template <
typename Token>
310 [[nodiscard]]
bool DoPop(Token& token, T& value) {
313 if constexpr (std::is_same_v<Token, moodycamel::ConsumerToken>) {
314 static_assert(QueuePolicy::kIsMultipleProducer);
315 success = queue_.try_dequeue(token, value);
316 }
else if constexpr (std::is_same_v<Token, impl::MultiToken>) {
317 static_assert(QueuePolicy::kIsMultipleProducer);
318 success = queue_.try_dequeue(value);
320 static_assert(std::is_same_v<Token, impl::NoToken>);
321 static_assert(!QueuePolicy::kIsMultipleProducer);
322 success = queue_.try_dequeue_from_producer(single_producer_token_, value);
326 producer_side_.OnElementPopped(QueuePolicy::GetElementSize(value));
333 moodycamel::ConcurrentQueue<T> queue_{1};
334 std::atomic<std::size_t> consumers_count_{0};
335 std::atomic<std::size_t> producers_count_{0};
337 SingleProducerToken single_producer_token_;
339 ProducerSide producer_side_;
340 ConsumerSide consumer_side_;
342 static constexpr std::size_t kCreatedAndDead =
343 std::numeric_limits<std::size_t>::max();
344 static constexpr std::size_t kSemaphoreUnlockValue =
345 std::numeric_limits<std::size_t>::max() / 2;
349template <
typename T,
typename QueuePolicy>
350class GenericQueue<T, QueuePolicy>::SingleProducerSide
final {
352 explicit SingleProducerSide(GenericQueue& queue, std::size_t capacity)
353 : queue_(queue), used_capacity_(0), total_capacity_(capacity) {}
357 template <
typename Token>
358 [[nodiscard]]
bool Push(Token& token, T&& value, engine::Deadline deadline) {
359 if (DoPush(token, std::move(value))) {
363 return non_full_event_.WaitForEventUntil(deadline) &&
365 DoPush(token, std::move(value));
368 template <
typename Token>
369 [[nodiscard]]
bool PushNoblock(Token& token, T&& value) {
370 return DoPush(token, std::move(value));
373 void OnElementPopped(std::size_t released_capacity) {
374 used_capacity_.fetch_sub(released_capacity);
375 non_full_event_.Send();
378 void StopBlockingOnPush() {
379 total_capacity_ += kSemaphoreUnlockValue;
380 non_full_event_.Send();
383 void ResumeBlockingOnPush() { total_capacity_ -= kSemaphoreUnlockValue; }
385 void SetSoftMaxSize(std::size_t new_capacity) {
386 const auto old_capacity = total_capacity_.exchange(new_capacity);
387 if (new_capacity > old_capacity) non_full_event_.Send();
390 std::size_t GetSoftMaxSize()
const noexcept {
return total_capacity_.load(); }
392 std::size_t GetSizeApproximate()
const noexcept {
393 return used_capacity_.load();
397 template <
typename Token>
398 [[nodiscard]]
bool DoPush(Token& token, T&& value) {
399 const std::size_t value_size = QueuePolicy::GetElementSize(value);
400 if (queue_.NoMoreConsumers() ||
401 used_capacity_.load() + value_size > total_capacity_.load()) {
405 used_capacity_.fetch_add(value_size);
406 queue_.DoPush(token, std::move(value));
407 non_full_event_.Reset();
411 GenericQueue& queue_;
412 engine::SingleConsumerEvent non_full_event_;
413 std::atomic<std::size_t> used_capacity_;
414 std::atomic<std::size_t> total_capacity_;
418template <
typename T,
typename QueuePolicy>
419class GenericQueue<T, QueuePolicy>::MultiProducerSide
final {
421 explicit MultiProducerSide(GenericQueue& queue, std::size_t capacity)
423 remaining_capacity_(capacity),
424 remaining_capacity_control_(remaining_capacity_) {}
428 template <
typename Token>
429 [[nodiscard]]
bool Push(Token& token, T&& value, engine::Deadline deadline) {
430 const std::size_t value_size = QueuePolicy::GetElementSize(value);
431 return remaining_capacity_.try_lock_shared_until_count(deadline,
433 DoPush(token, std::move(value));
436 template <
typename Token>
437 [[nodiscard]]
bool PushNoblock(Token& token, T&& value) {
438 const std::size_t value_size = QueuePolicy::GetElementSize(value);
439 return remaining_capacity_.try_lock_shared_count(value_size) &&
440 DoPush(token, std::move(value));
443 void OnElementPopped(std::size_t value_size) {
444 remaining_capacity_.unlock_shared_count(value_size);
447 void StopBlockingOnPush() {
448 remaining_capacity_control_.SetCapacityOverride(0);
451 void ResumeBlockingOnPush() {
452 remaining_capacity_control_.RemoveCapacityOverride();
455 void SetSoftMaxSize(std::size_t count) {
456 remaining_capacity_control_.SetCapacity(count);
459 std::size_t GetSizeApproximate()
const noexcept {
460 return remaining_capacity_.UsedApprox();
463 std::size_t GetSoftMaxSize()
const noexcept {
464 return remaining_capacity_control_.GetCapacity();
468 template <
typename Token>
469 [[nodiscard]]
bool DoPush(Token& token, T&& value) {
470 const std::size_t value_size = QueuePolicy::GetElementSize(value);
472 if (queue_.NoMoreConsumers()) {
473 remaining_capacity_.unlock_shared_count(value_size);
477 queue_.DoPush(token, std::move(value));
481 GenericQueue& queue_;
482 engine::CancellableSemaphore remaining_capacity_;
483 concurrent::impl::SemaphoreCapacityControl remaining_capacity_control_;
487template <
typename T,
typename QueuePolicy>
488class GenericQueue<T, QueuePolicy>::SingleConsumerSide
final {
490 explicit SingleConsumerSide(GenericQueue& queue)
491 : queue_(queue), element_count_(0) {}
494 template <
typename Token>
495 [[nodiscard]]
bool Pop(Token& token, T& value, engine::Deadline deadline) {
496 while (!DoPop(token, value)) {
497 if (queue_.NoMoreProducers() ||
498 !nonempty_event_.WaitForEventUntil(deadline)) {
502 return DoPop(token, value);
508 template <
typename Token>
509 [[nodiscard]]
bool PopNoblock(Token& token, T& value) {
510 return DoPop(token, value);
513 void OnElementPushed() {
515 nonempty_event_.Send();
518 void StopBlockingOnPop() { nonempty_event_.Send(); }
520 void ResumeBlockingOnPop() {}
522 std::size_t GetElementCount()
const {
return element_count_; }
525 template <
typename Token>
526 [[nodiscard]]
bool DoPop(Token& token, T& value) {
527 if (queue_.DoPop(token, value)) {
529 nonempty_event_.Reset();
535 GenericQueue& queue_;
536 engine::SingleConsumerEvent nonempty_event_;
537 std::atomic<std::size_t> element_count_;
541template <
typename T,
typename QueuePolicy>
542class GenericQueue<T, QueuePolicy>::MultiConsumerSide
final {
544 explicit MultiConsumerSide(GenericQueue& queue)
546 element_count_(kUnbounded),
547 element_count_control_(element_count_) {
548 const bool success = element_count_.try_lock_shared_count(kUnbounded);
552 ~MultiConsumerSide() { element_count_.unlock_shared_count(kUnbounded); }
555 template <
typename Token>
556 [[nodiscard]]
bool Pop(Token& token, T& value, engine::Deadline deadline) {
557 return element_count_.try_lock_shared_until(deadline) &&
561 template <
typename Token>
562 [[nodiscard]]
bool PopNoblock(Token& token, T& value) {
563 return element_count_.try_lock_shared() && DoPop(token, value);
566 void OnElementPushed() { element_count_.unlock_shared(); }
568 void StopBlockingOnPop() {
569 element_count_control_.SetCapacityOverride(kUnbounded +
570 kSemaphoreUnlockValue);
573 void ResumeBlockingOnPop() {
574 element_count_control_.RemoveCapacityOverride();
577 std::size_t GetElementCount()
const {
578 const std::size_t cur_element_count = element_count_.RemainingApprox();
579 if (cur_element_count < kUnbounded) {
580 return cur_element_count;
581 }
else if (cur_element_count <= kSemaphoreUnlockValue) {
584 return cur_element_count - kSemaphoreUnlockValue;
588 template <
typename Token>
589 [[nodiscard]]
bool DoPop(Token& token, T& value) {
591 if (queue_.DoPop(token, value)) {
594 if (queue_.NoMoreProducers()) {
595 element_count_.unlock_shared();
603 GenericQueue& queue_;
604 engine::CancellableSemaphore element_count_;
605 concurrent::impl::SemaphoreCapacityControl element_count_control_;
623using NonFifoMpmcQueue = GenericQueue<T, impl::SimpleQueuePolicy<
true,
true>>;
632using NonFifoMpscQueue = GenericQueue<T, impl::SimpleQueuePolicy<
true,
false>>;
640using SpmcQueue = GenericQueue<T, impl::SimpleQueuePolicy<
false,
true>>;
648using SpscQueue = GenericQueue<T, impl::SimpleQueuePolicy<
false,
false>>;
656using StringStreamQueue =
657 GenericQueue<std::string, impl::ContainerQueuePolicy<
false,
false>>;