Function RunWaitTest

library/cpp/threading/future/future_mt_ut.cpp:76–140

    // needed synchronization
    template <class TFactory>
    void RunWaitTest(TFactory global) {
        auto pool = MakePool();

        const auto exception = std::make_exception_ptr(42);

        for (auto numPromises : xrange(1, 5)) {
            for (auto loopIter : xrange(1024 * 64)) {
                const auto numParticipants = numPromises + 1;

                TRelaxedBarrier barrier{numParticipants};

                std::atomic<i64> started = 0;
                std::atomic<i64> startedException = 0;
                std::atomic<i64> completed = 0;

                TVector<TPromise<void>> promises;
                for (auto i : xrange(numPromises)) {
                    Y_UNUSED(i);
                    promises.push_back(NewPromise());
                }

                const auto futures = ToFutures(promises);

                auto snapshotter = [&] {
                    return TStateSnapshot{
                        .Started = started.load(std::memory_order_relaxed),
                        .StartedException = startedException.load(std::memory_order_relaxed),
                        .Futures = &futures,
                    };
                };

                for (auto i : xrange(numPromises)) {
                    pool->SafeAddFunc([&, i] {
                        barrier.Arrive();

                        // subscribers must observe effects of this operation
                        // after .Set*
                        started.fetch_add(1, std::memory_order_relaxed);

                        if ((loopIter % 4 == 0) && i == 0) {
                            startedException.fetch_add(1, std::memory_order_relaxed);
                            promises[i].SetException(exception);
                        } else {
                            promises[i].SetValue();
                        }

                        completed.fetch_add(1, std::memory_order_release);
                    });
                }

                pool->SafeAddFunc([&] {
                    auto local = global(snapshotter);

                    barrier.Arrive();

                    local();

                    completed.fetch_add(1, std::memory_order_release);