How to parallelism

[n0lavar]

When developing at some point I came to the necessity of paralleling the loading process.
In the readme I found an example for the simplest case, but I was not able to adapt it to my own case.
This is what the process looks like at the moment (all code is pseudocode):

concurrencpp::result<void> ChunksManager::LoadChunks()
{
    ...
    for (auto& chunk : chunksToLoad)
        co_await chunk.Load();
    ...
}

concurrencpp::result<void> Chunk::Load()
{
    co_await concurrencpp::resume_on(thread_pool_executor());
    ...

    auto resources = co_await resourceManager.LoadChunkResouces();

    co_await concurrencpp::resume_on(thread_pool_executor());
    ....
}

concurrencpp::result<T> ResourceManager::LoadChunkResouces()
{
    co_await concurrencpp::resume_on(io_thread());
    ...
    co_return ...;
}

So basically it's:

// chunk 1
co_await chunk.Load();
    co_await concurrencpp::resume_on(thread_pool_executor());
    ...

    auto resources = co_await resourceManager.LoadChunkResouces();
        co_await concurrencpp::resume_on(io_thread());
        ...

    co_await concurrencpp::resume_on(thread_pool_executor());
    ...


// chunk 2
co_await chunk.Load();
    co_await concurrencpp::resume_on(thread_pool_executor());
    ...

    auto resources = co_await resourceManager.LoadChunkResouces();
        co_await concurrencpp::resume_on(io_thread());
        ...

    co_await concurrencpp::resume_on(thread_pool_executor());
    ...

// etc

The problem is that in the example from the readme is a synchronous task, while I have a sequence of asynchronous tasks, and somewhere in the middle the type of executor changes several times. I need the part that runs on the thread pool executor to run in parallel.

My attempts looked like this:

concurrencpp::result<void> ChunksManager::LoadChunks()
{
    std::vector<concurencpp::result<void>> results;
    for (auto& chunk : chunksToLoad)
        results.push_back(chunk.Load());

    co_await concurencpp::when_all(thread_pool_executor(), results.begin(), results.end());
}

concurencpp::result<void> LoadChinks_(
    concurrencpp::executor_tag,
    std::shared_ptr<concurrencpp::executor> pExecutor,
    ...)
{
    for (auto& chunk : chunksToLoad)
        co_await chunk.Load();
}

concurrencpp::result<void> ChunksManager::LoadChunks()
{
    ...
    co_await LoadChinks_(...);
    ...
}

concurencpp::result<void> LoadChinks_(
    std::shared_ptr<concurrencpp::executor> pExecutor,
    ...)
{
    for (auto& chunk : chunksToLoad)
        co_await pExecutor->submit([](){ chunk.Load(); });
}

concurrencpp::result<void> ChunksManager::LoadChunks()
{
    ...
    co_await LoadChinks_(...);
    ...
}

concurencpp::result<void> LoadChinks_(
    std::shared_ptr<concurrencpp::executor> pExecutor,
    ...)
{
    std::vector<concurencpp::result<void>> results;
    for (auto& chunk : chunksToLoad)
        results.push_back(pExecutor->submit([](){ chunk.Load(); }));

    co_await concurencpp::when_all(pExecutor, results.begin(), results.end());
}

concurrencpp::result<void> ChunksManager::LoadChunks()
{
    ...
    co_await LoadChinks_(...);
    ...
}

What am I doing wrong? Is there a possibility to do it right?

[n0lavar]

I am not sure if a repo owner receives a notification so I am tagging you personally :) @David-Haim

[David-Haim]

your first attempt should work. why do you think it's not parallelized? every chunk runs independently (it's an eager task) and you join all of them with when_all.

[n0lavar]

I can see it in profiler, there are only a couple of active threads

My current execution looks like
1 2 3 1 2 3 1 2 3
While I expect it to be more like this

[n0lavar]

Well I did some research and found out that the lib works just fine, which means the problem is in my code
Maybe you will be interested in these beautiful pictures

#include <tracy/Tracy.hpp>
#include <concurrencpp/concurrencpp.h>

using namespace std::chrono_literals;

void RegisterThread(std::string_view svThreadName)
{
    thread_local std::string sThreadName(svThreadName);
    tracy::SetThreadName(sThreadName.c_str());
}

std::atomic_bool g_bFinish { false };
std::shared_ptr<concurrencpp::executor> g_pThreadPoolExecutor;
std::shared_ptr<concurrencpp::executor> g_pIoThreadExecutor;

concurrencpp::result<void> ProceedChunkResourced()
{
    co_await concurrencpp::resume_on(g_pThreadPoolExecutor);
    ZoneScopedC(tracy::Color::PaleTurquoise);
    std::this_thread::sleep_for(3s);
}

concurrencpp::result<void> LoadChunkResources()
{
    co_await concurrencpp::resume_on(g_pIoThreadExecutor);
    ZoneScopedC(tracy::Color::PaleGoldenrod);
    std::this_thread::sleep_for(1s);
}

concurrencpp::result<void> PrepareChunk()
{
    co_await concurrencpp::resume_on(g_pThreadPoolExecutor);
    ZoneScopedC(tracy::Color::PaleGreen);
    std::this_thread::sleep_for(2s);
}

concurrencpp::result<void> LoadChunk()
{
    co_await PrepareChunk();
    co_await LoadChunkResources();
    co_await ProceedChunkResourced();
}

concurrencpp::result<void> LoadChunks()
{
    std::vector<concurrencpp::result<void>> results;
    for (size_t i = 0; i < 9; ++i)
        results.push_back(LoadChunk());

    co_await concurrencpp::when_all(g_pThreadPoolExecutor, results.begin(), results.end());
    g_bFinish = true;
}

int main()
{
    RegisterThread("Main thread");

    concurrencpp::runtime_options options;
    options.max_cpu_threads = 4;
    options.thread_started_callback = RegisterThread;

    concurrencpp::runtime runtime(options);
    g_pThreadPoolExecutor = runtime.thread_pool_executor();
    g_pIoThreadExecutor = runtime.make_executor<concurrencpp::worker_thread_executor>(
        [](std::string_view)
        {
            RegisterThread("IO thread");
        });

    LoadChunks();

    while (!g_bFinish)
    {
        FrameMarkNamed(nullptr);

        ZoneScopedC(tracy::Color::PaleVioletRed);
        std::this_thread::sleep_for(16.666ms);
    }
}

when_all.zip

Thank you for your replies and your work