corochanneltest.cpp

Go to the documentation of this file.

/**********************************************************************
 * LeechCraft - modular cross-platform feature rich internet client.
 * Copyright (C) 2006-2014  Georg Rudoy
 *
 * Distributed under the Boost Software License, Version 1.0.
 * (See accompanying file LICENSE or copy at https://www.boost.org/LICENSE_1_0.txt)
 **********************************************************************/

#include "corochanneltest.h"
#include <QtConcurrentRun>
#include <QtTest>
#include "coro.h"
#include "coro/channel.h"
#include "coro/channelutils.h"
#include "coro/getresult.h"

QTEST_GUILESS_MAIN (LC::Util::CoroChannelTest)

namespace LC::Util
{
    void CoroChannelTest::testSingleRecv ()
    {
        using namespace std::chrono_literals;

        constexpr auto producersCount = 32;
        constexpr auto repCount = 100;
        constexpr auto sleepLength = 1ms;

        Channel<int> ch { this };

        std::vector<std::thread> threads;
        std::atomic_int expected;
        for (int i = 0; i < producersCount; ++i)
            threads.emplace_back ([&, i]
                    {
                        for (int j = 0; j < repCount; ++j)
                        {
                            const auto val = j * producersCount + i;
                            ch.Send (val);
                            expected.fetch_add (val, std::memory_order::relaxed);
                            std::this_thread::sleep_for (sleepLength);
                        }
                    });

        auto mainThread = std::this_thread::get_id ();
        auto reader = [] (auto mainThread, Channel<int> *ch) -> Task<int, ThreadSafetyExtension>
        {
            int sum = 0;
            while (auto next = co_await ch->Receive ())
            {
                [=]
                {
                    auto thisThread = std::this_thread::get_id ();
                    QCOMPARE (thisThread, mainThread);
                } ();

                sum += *next;
            }
            co_return sum;
        } (mainThread, &ch);

        for (auto& thread : threads)
            thread.join ();

        ch.Close ();

        auto result = GetTaskResult (reader);
        QCOMPARE (result, expected);
    }

    void CoroChannelTest::testManyRecvs ()
    {
        using namespace std::chrono_literals;

        constexpr auto producersCount = 32;
        constexpr auto consumersCount = 8;
        constexpr auto repCount = 100;
        constexpr auto sleepLength = 1ms;

        Channel<int> ch;

        std::vector<std::thread> producers;
        std::atomic_int expected;
        for (int i = 0; i < producersCount; ++i)
            producers.emplace_back ([&, i]
                    {
                        for (int j = 0; j < repCount; ++j)
                        {
                            const auto val = j * producersCount + i;
                            ch.Send (val);
                            expected.fetch_add (val, std::memory_order::relaxed);
                            std::this_thread::sleep_for (sleepLength);
                        }
                    });

        std::atomic_int sum;
        std::vector<std::thread> consumers;
        for (int i = 0; i < consumersCount; ++i)
            consumers.emplace_back ([&]
            {
                auto reader = [] (Channel<int> *ch) -> Task<int, ThreadSafetyExtension>
                {
                    int sum = 0;
                    while (auto next = co_await ch->Receive ())
                        sum += *next;
                    co_return sum;
                } (&ch);
                sum.fetch_add (GetTaskResult (reader));
            });

        for (auto& thread : producers)
            thread.join ();

        ch.Close ();

        for (auto& thread : consumers)
            thread.join ();

        QCOMPARE (sum, expected);
    }

    void CoroChannelTest::testSingleThreaded ()
    {
        constexpr auto iterations = 1000;

        Channel<int> ch;

        auto reader = [] (Channel<int> *ch) -> Task<int, ThreadSafetyExtension>
        {
            int sum = 0;
            while (auto next = co_await ch->Receive ())
                sum += *next;
            co_return sum;
        } (&ch);

        int expected = 0;
        for (int i = 0; i < iterations; ++i)
        {
            expected += i;
            ch.Send (i);
        }

        ch.Close ();

        const auto result = GetTaskResult (reader);
        QCOMPARE (result, expected);
    }

    void CoroChannelTest::testSingleThreadedTimered ()
    {
        using namespace std::chrono_literals;

        constexpr auto iterations = 100;
        constexpr auto interval = 1ms;

        Channel<int> ch;

        auto reader = [] (Channel<int> *ch) -> Task<int, ThreadSafetyExtension>
        {
            int sum = 0;
            while (auto next = co_await ch->Receive ())
                sum += *next;
            co_return sum;
        } (&ch);

        int expected = 0;

        QTimer timer;
        timer.callOnTimeout ([&, i = 0] mutable
                {
                    expected += i;
                    ch.Send (i);
                    if (++i == iterations)
                    {
                        timer.stop ();
                        ch.Close ();
                    }
                });
        timer.start (interval);

        const auto result = GetTaskResult (reader);
        QCOMPARE (result, expected);
    }

    void CoroChannelTest::testMerge ()
    {
        using namespace std::chrono_literals;

        constexpr auto numChannels = 100;
        constexpr auto iterations = 100;
        constexpr auto interval = 1ms;

        QVector<Channel_ptr<int>> channels;
        std::generate_n (std::back_inserter (channels), numChannels, [] { return std::make_shared<Channel<int>> (); });

        auto merged = MergeChannels (channels);

        auto reader = [] (Channel<int> *ch) -> Task<int, ThreadSafetyExtension>
        {
            int sum = 0;
            while (auto next = co_await ch->Receive ())
                sum += *next;
            co_return sum;
        } (merged.get ());

        int expected = 0;

        QTimer timer;
        timer.callOnTimeout ([&, i = 0] mutable
                {
                    for (int j = 0; j < numChannels; ++j)
                    {
                        auto value = i * numChannels + j;
                        expected += value;
                        channels [j]->Send (value);
                    }
                    if (++i == iterations)
                    {
                        timer.stop ();
                        for (auto chan : channels)
                            chan->Close ();
                    }
                });
        timer.start (interval);

        const auto result = GetTaskResult (reader);
        QCOMPARE (result, expected);
    }
}