Wait for a group

Fork/join in its simplest form: the program splits an array of 16 million values into 256 parts, sums each part in its own task and waits for all of them with one TaskGroup. Then it reuses the group for a second round of two tasks built from lambdas with Task::Create. Both results are checked against a plain loop.

A task class for each part

// One part of the work: sums a slice of the array. The result stays in the object,
// so the object has to live at least until the group has been waited for.
class SumTask : public TC::Tasks::Task
{
public:
    void Run() noexcept override
    {
        sum = std::accumulate(part.begin(), part.end(), std::uint64_t{0});
    }

    std::span<const std::uint32_t> part;
    std::uint64_t                  sum = 0;
};

The result stays in the task object. That is the simplest way to get results out of tasks – as long as the object outlives the wait.

Fork and join

// Declared before the tasks are added and destroyed only after Stop: the
// group and the task objects must outlive the tasks, including the last
// exit notification.
TC::Synchronization::TaskGroup group;
std::vector<SumTask>           tasks(
    Parts); // not movable: the scheduler holds their addresses

const std::size_t partSize = Values / Parts;
for (std::size_t i = 0; i < Parts; ++i) {
    tasks[i].part = std::span(values).subspan(i * partSize, partSize);
    group.AddTask(&tasks[i]); // count the task in before it can exit
    scheduler->AddTask(&tasks[i]);
}
group.Wait(); // suspends main until every SumTask has exited

std::uint64_t total = 0;
for (const SumTask& task : tasks)
    total += task.sum;

AddTask counts a task in and makes the group its exit notification; the task's exit counts it out. Wait passes once the count is back at 0.

Lambda tasks and reuse

// The group is reusable once its count is back at 0. Task::Create returns
// the LambdaTask by value; the variables keep it alive.
const auto    half    = values.begin() + Values / 2;
std::uint32_t lowMax  = 0;
std::uint32_t highMax = 0;
auto          lowTask = TC::Tasks::Task::Create(
    [&] { lowMax = *std::max_element(values.begin(), half); });
auto highTask = TC::Tasks::Task::Create(
    [&] { highMax = *std::max_element(half, values.end()); });

group.AddTask(&lowTask);
group.AddTask(&highTask);
scheduler->AddTask(&lowTask);
scheduler->AddTask(&highTask);
group.Wait();

const std::uint32_t maximum = std::max(lowMax, highMax);

Task::Create returns a LambdaTask by value. It cannot be copied or moved – the scheduler holds its address – so it lives in a local variable for as long as it runs. The lambdas capture by reference, which is fine for the same reason: everything they capture outlives the tasks.

Full program

// Wait for a group: fork many short tasks, join them with a TaskGroup, then reuse
// the group for a second round of tasks built from lambdas.
#define NOMINMAX
#include <Windows.h>

#include <algorithm>
#include <cstddef>
#include <cstdint>
#include <cstdio>
#include <numeric>
#include <span>
#include <vector>

#include <ThinkMeta.ConcurrentTasks/Core.h>
import ThinkMeta.ConcurrentTasks.Synchronization.TaskGroup;
import ThinkMeta.ConcurrentTasks.Tasks.Task;

namespace TC = ThinkMeta::ConcurrentTasks;

// One part of the work: sums a slice of the array. The result stays in the object,
// so the object has to live at least until the group has been waited for.
class SumTask : public TC::Tasks::Task
{
public:
    void Run() noexcept override
    {
        sum = std::accumulate(part.begin(), part.end(), std::uint64_t{0});
    }

    std::span<const std::uint32_t> part;
    std::uint64_t                  sum = 0;
};

int main()
{
    constexpr std::size_t Values = 16 * 1024 * 1024;
    constexpr std::size_t Parts  = 256;

    std::vector<std::uint32_t> values(Values);
    for (std::size_t i = 0; i < Values; ++i)
        values[i] = static_cast<std::uint32_t>(
            i * 2654435761u); // spread over the whole range

    TC::ConvertThreadToTask();

    TC::SchedulerInfo   info{};
    TC::ITaskScheduler* scheduler
        = TC::CreateTaskScheduler(TC::SchedulerKind::Default, &info);
    if (scheduler == nullptr)
        return 1;
    scheduler->Start();

    bool ok = true;
    {
        // Declared before the tasks are added and destroyed only after Stop: the
        // group and the task objects must outlive the tasks, including the last
        // exit notification.
        TC::Synchronization::TaskGroup group;
        std::vector<SumTask>           tasks(
            Parts); // not movable: the scheduler holds their addresses

        const std::size_t partSize = Values / Parts;
        for (std::size_t i = 0; i < Parts; ++i) {
            tasks[i].part = std::span(values).subspan(i * partSize, partSize);
            group.AddTask(&tasks[i]); // count the task in before it can exit
            scheduler->AddTask(&tasks[i]);
        }
        group.Wait(); // suspends main until every SumTask has exited

        std::uint64_t total = 0;
        for (const SumTask& task : tasks)
            total += task.sum;

        const std::uint64_t expected
            = std::accumulate(values.begin(), values.end(), std::uint64_t{0});
        ok = ok && total == expected;
        std::printf("Sum of %zu values in %zu tasks: %llu (%s)\n", Values, Parts,
                    static_cast<unsigned long long>(total),
                    total == expected ? "ok" : "MISMATCH");

        // The group is reusable once its count is back at 0. Task::Create returns
        // the LambdaTask by value; the variables keep it alive.
        const auto    half    = values.begin() + Values / 2;
        std::uint32_t lowMax  = 0;
        std::uint32_t highMax = 0;
        auto          lowTask = TC::Tasks::Task::Create(
            [&] { lowMax = *std::max_element(values.begin(), half); });
        auto highTask = TC::Tasks::Task::Create(
            [&] { highMax = *std::max_element(half, values.end()); });

        group.AddTask(&lowTask);
        group.AddTask(&highTask);
        scheduler->AddTask(&lowTask);
        scheduler->AddTask(&highTask);
        group.Wait();

        const std::uint32_t maximum = std::max(lowMax, highMax);

        const std::uint32_t expectedMax
            = *std::max_element(values.begin(), values.end());
        ok = ok && maximum == expectedMax;
        std::printf("Maximum from 2 tasks: %u (%s)\n", maximum,
                    maximum == expectedMax ? "ok" : "MISMATCH");

        scheduler->Stop(); // the tasks and the group are destroyed after this, at
                           // the end of the scope
    }

    scheduler->Destroy();
    TC::ConvertTaskToThread();
    return ok ? 0 : 1;
}

Source and project on GitHub

Build and run

A console program. With the core's props file imported and the library referenced (see Set up a project) it needs nothing else. It prints:

Sum of 16777216 values in 256 tasks: 36028801976631296 (ok)
Maximum from 2 tasks: 4294967208 (ok)

What to notice

Important

The group and the task objects must outlive the tasks. Wait returns once the count is 0, but the exit notification of the last task may still be running inside the group at that moment. That is why group and tasks are destroyed only after Stop, at the end of the scope. See Lifetimes.

Count in before starting. group.AddTask comes before scheduler->AddTask. The other way round, a short task could exit before it is counted, and Wait could pass too early.
Task objects do not move. std::vector<SumTask> tasks(Parts) constructs the objects in place and never grows; tasks cannot be pushed into a vector one by one.
One waiter or many. Only main waits here, so TaskGroup_s would do as well and also offers a wait with a timeout. TaskGroup allows any number of waiting tasks.
Queue capacity. All 256 tasks are runnable at once. The default capacity of the scheduler's queues is 4096 per queue (SchedulerInfo); a program that starts more tasks at once has to raise taskQueueCapacity, since a full queue is invalid use.