Function test_scq_sync_multi_producer

src/core/test/utils/thread.cpp:25–67

template <int producer_sleep, int consumer_sleep>
void test_scq_sync_multi_producer() {
    size_t nr_worker_call = 0;
    SCQueueSynchronizer sync(0);
    auto worker = [&]() {
        RNGxorshf rng{next_rand_seed()};
        while (auto nr = sync.consumer_fetch(1)) {
            nr_worker_call += nr;
            ASSERT_EQ(1u, nr);
            if (consumer_sleep) {
                std::this_thread::sleep_for(
                        std::chrono::microseconds(rng() % consumer_sleep));
            }
            sync.consumer_commit(nr);
        }
    };
    sync.start_worker(std::thread{worker});

    constexpr size_t N = 500, M = 8;
    std::atomic_size_t nr_worker_started{0};
    auto producer_impl = [&]() {
        RNGxorshf rng{next_rand_seed()};
        ++nr_worker_started;
        while (nr_worker_started.load() != M)
            ;
        for (size_t i = 0; i < N; ++i) {
            if (producer_sleep) {
                std::this_thread::sleep_for(
                        std::chrono::microseconds(rng() % producer_sleep));
            }
            sync.producer_add();
            if (i % 4 == 0)
                sync.producer_wait();
        }
    };
    std::vector<std::thread> producer_threads;
    for (size_t i = 0; i < M; ++i) {
        producer_threads.emplace_back(producer_impl);
    }
    for (auto&& i : producer_threads)
        i.join();
    sync.producer_wait();
    ASSERT_EQ(N * M, nr_worker_call);
}
}  // namespace

TEST(TestAsyncQueue, Synchronizer) {