How do I start N tasks on a thread pool?

[popizdeh]

	auto do_n_jobs = [](auto, int size) -> lf::task<> {
		printf("Is do_n_jobs called more than once???\n");
		for (unsigned i = 0; i < size; ++i) {
			co_await lf::fork(lf::lift)(printf, "%d\n", i);
		}
		co_await lf::join;
	};

	auto cycle = [do_n_jobs](auto, int size) -> lf::task<> {
		printf("Is cycle called more than once???\n");
		co_await lf::call(do_n_jobs)(size);
	};

	lf::lazy_pool pool{std::thread::hardware_concurrency()};
	lf::sync_wait(pool, cycle, 10);

I've modified lazy_pool test code to get this but I'm seeing cycle called 3 times for some reason? I just want to run N tasks executing a global function with different arguments for each call and have them execute in parallel, what am I doing wrong here?

[ConorWilliams]

Hi there, thank you for trying out libfork!

In your example both cycle and do_n_jobs should be called only once. However, your example contains a few bits of UB(call without join, wrong printf specifier, a coroutine lambda with a capture). Here is the output with these fixes. Please note that this is quite an inefficient way to launch n-tasks as you are serializing the launching of the tasks in do_n_jobs, libfork provides the for_each algorithm for this common use-case, here is your program using for_each. The for_each algorithm can also accept an async_invocable to enable recursive launching of tasks.

[ConorWilliams]

Here is the code using for_each:

inline constexpr auto print = [](int i){
    printf("%u\n", i);
};

int main() {

    lf::lazy_pool pool{4};

    auto is = std::ranges::iota_view{0, 1000};

    lf::sync_wait(pool, lf::for_each, is, print);
}

[popizdeh]

Thanks for the quick response Conor.

I think the issue I'm seeing is related to Google benchmark that I'm running libfork from, it looks like the benchmark itself is executed more than once!

Few more comments/questions:

• If I understand correctly each lf::call should have a matching lf::join, I copied all of that code from test/source/core/lazy_pool.cpp so I think that test also has UB
• What exactly is serialised in do_n_jobs, I'm guessing that call to lf::join blocks until all tasks complete but I don't see how this is inefficient?
• I guess I should've told you what I'm trying to do exactly. I have a "worker" function (this is what printf in my example is simulating) that I'm calling from multiple nested loops with different arguments. I'm not sure for_each is the best fit here as I'm dealing with a bit more state, here's some pseudo-code to illustrate what I'm doing. This executes on multiple threads using Taskflow and I'm trying to get the same semantics using libfork.

tf::Executor executor;
tf::Taskflow taskflow;
for (int y = 0; y < height; ++y)
{
    auto y_state = get_y_state(y);
    for (int x = 0; x < width; ++x)
    {
        auto x_state = get_x_state(x);
        taskflow.emplace([=]{
            do_something(x_state, y_state);
        });
    }
}
exector.run(taskflow);
executor.wait_for_all();

[ConorWilliams]

I think the issue I'm seeing is related to Google benchmark that I'm running libfork from, it looks like the benchmark itself is executed more than once!

That makes sense, google benchmark will internally rerun the benchmark enough times to get good statistics

Few more comments/questions:

• If I understand correctly each lf::call should have a matching lf::join, I copied all of that code from test/source/core/lazy_pool.cpp so I think that test also has UB

The test code is testing some internal behaviour, in this case the cycle function never forks so not calling join will not cause any problems. However, this is could be UB for users if an exception was thrown later down the call-stack. Although this is safe for-now in the future this might change which is why I said it was UB.

• What exactly is serialised in do_n_jobs, I'm guessing that call to lf::join blocks until all tasks complete but I don't see how this is inefficient?

Each task is launched one after the other in a loop. This does not mean each task starts after the previous one completes i.e. the tasks can run in parallel.

• I guess I should've told you what I'm trying to do exactly. I have a "worker" function (this is what printf in my example is simulating) that I'm calling from multiple nested loops with different arguments. I'm not sure for_each is the best fit here as I'm dealing with a bit more state, here's some pseudo-code to illustrate what I'm doing. This executes on multiple threads using Taskflow and I'm trying to get the same semantics using libfork.

tf::Executor executor;
tf::Taskflow taskflow;
for (int y = 0; y < height; ++y)
{
    auto y_state = get_y_state(y);
    for (int x = 0; x < width; ++x)
    {
        auto x_state = get_x_state(x);
        taskflow.emplace([=]{
            do_something(x_state, y_state);
        });
    }
}
exector.run(taskflow);
executor.wait_for_all();

Assuming get_y_state and get_x_state can run concurrently then this will run every (x,y) pair concurrently:

#include <ranges>

#include "https://raw.githubusercontent.com/ConorWilliams/libfork/v3.7.1/single_header/libfork.hpp"

namespace {

using lf::just;
using lf::task;
using lf::for_each;

namespace ranges = std::ranges;

int get_y_state(int y) { return 2 * y; }

int get_x_state(int x) { return 2 * x; }

void do_something(int x, int y) {
    printf("x_state=%d, y_state=%d\n", x, y);
}

inline constexpr auto with_for_each = [](auto, int height, int width) -> task<> {

    // Repeat "width" height times
    auto hs = ranges::views::repeat(width, height);
    auto ys = ranges::views::enumerate(hs);

    co_await just[for_each](ys, [](auto, auto yp) -> task<> {

        auto [y, width] = yp;
        
        auto y_state = get_y_state(y);
        
        co_await just[for_each](ranges::iota_view{0, width}, [&](int x) {
            do_something(get_x_state(x), y_state);
        });
    });
};

}

int main() {
	lf::sync_wait(lf::lazy_pool{}, with_for_each, 3, 3); //
}

[ConorWilliams]

If get_y_state and get_x_state must be run in serial before any tasks are launched (like in the taskflow example) then the best way to do this would probably be more like this

[ConorWilliams]

How did libfork compare in your benchmarks in the end?

[popizdeh]

I got the same numbers as with Taskflow, I guess the work that's being done dominates... Or maybe 300 tasks I'm starting is simply not enough to show any difference.

…

On Tue, 19 Mar 2024, 12:49 am Conor Williams, ***@***.***> wrote: How did libfork compare in your benchmarks in the end? — Reply to this email directly, view it on GitHub <#11 (comment)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AAMIXUFFCE54DUMOI7OY6L3YY3WEZAVCNFSM6AAAAABERT4Q72VHI2DSMVQWIX3LMV43SRDJONRXK43TNFXW4Q3PNVWWK3TUHM4DQMRYGI4TG> . You are receiving this because you modified the open/close state.Message ID: ***@***.*** com>

[ConorWilliams]

Thanks for letting me know, I would image a bit of both is at play