Function DequeueAllRunner

util/thread/lfqueue_ut.cpp:192–267

    }

    void DequeueAllRunner(TLockFreeQueue<int>& queue, bool singleConsumer) {
        size_t threadsNum = 4;
        size_t enqueuesPerThread = 10'000;
        TThreadPool p;
        p.Start(threadsNum, 0);

        TVector<NThreading::TFuture<void>> futures;

        for (size_t i = 0; i < threadsNum; ++i) {
            NThreading::TPromise<void> promise = NThreading::NewPromise();
            futures.emplace_back(promise.GetFuture());

            p.SafeAddFunc([enqueuesPerThread, &queue, promise]() mutable {
                for (size_t i = 0; i != enqueuesPerThread; ++i) {
                    queue.Enqueue(i);
                }

                promise.SetValue();
            });
        }

        std::atomic<size_t> elementsLeft = threadsNum * enqueuesPerThread;

        ui64 numOfConsumers = singleConsumer ? 1 : threadsNum;

        TVector<TVector<int>> dataBuckets(numOfConsumers);

        for (size_t i = 0; i < numOfConsumers; ++i) {
            NThreading::TPromise<void> promise = NThreading::NewPromise();
            futures.emplace_back(promise.GetFuture());

            p.SafeAddFunc([&queue, &elementsLeft, promise, consumerData{&dataBuckets[i]}]() mutable {
                TVector<int> vec;
                while (static_cast<i64>(elementsLeft.load()) > 0) {
                    for (size_t i = 0; i != 100; ++i) {
                        vec.clear();
                        queue.DequeueAll(&vec);

                        elementsLeft -= vec.size();
                        consumerData->insert(consumerData->end(), vec.begin(), vec.end());
                    }
                }

                promise.SetValue();
            });
        }

        NThreading::WaitExceptionOrAll(futures).GetValueSync();
        p.Stop();

        TVector<int> left;
        queue.DequeueAll(&left);

        UNIT_ASSERT(left.empty());

        TVector<int> data;
        for (auto& dataBucket : dataBuckets) {
            data.insert(data.end(), dataBucket.begin(), dataBucket.end());