Waiting for tasks

A task is an object that you own and the scheduler runs. Most programs need to know when a task has finished: to read its result, to start the next step, or to destroy the object. ConcurrentTasks offers two ways to wait: WaitableTask for a single task and TaskGroup for any number of tasks (fork/join). Both suspend the waiting task, so its thread runs other tasks meanwhile. This guide shows both, together with Task::Create for tasks built from lambdas and the lifetime rules that come with them. The code is taken from one program, guide-waiting-for-tasks.cpp, which checks its own results outside the parts shown here.

Wait for one task

Derive from WaitableTask_s (or WaitableTask) and implement Run:

// Sums the numbers in [begin, end). One other task (or converted thread) waits for
// it.
class SumTask : public TC::Tasks::WaitableTask_s
{
public:
    SumTask(std::uint64_t begin, std::uint64_t end) noexcept
        : begin_(begin),
          end_(end)
    {}

    void Run() noexcept override
    {
        for (auto i = begin_; i < end_; ++i)
            sum_ += i;
    }

    std::uint64_t Sum() const noexcept { return sum_; }

private:
    std::uint64_t begin_;
    std::uint64_t end_;
    std::uint64_t sum_ = 0;
};

Wait suspends the caller until the task's Run has returned and the core has called OnTaskExited; it returns at once if that has happened already. Everything the task wrote before it exited is visible to the waiter afterwards.

SumTask task(0, 1'000'000);
scheduler->AddTask(
    &task);  // from a converted thread: AddTask on the scheduler object
task.Wait(); // suspends main until Run has returned

The caller of Wait must be a task, or an OS thread that called ConvertThreadToTask, as main does here. To check without waiting, call TryWait.

The two variants differ in the number of waiters:

Choose WaitableTask_s whenever one party waits. Several waiters need WaitableTask:

// One task loads the settings, three workers wait for them: WaitableTask allows
// any number of waiters.
struct LoadSettings : TC::Tasks::WaitableTask {
    void Run() noexcept override { blockSize = 64 * 1024; }
    int  blockSize = 0;
} settings;

auto worker = [&settings]() noexcept {
    settings.Wait(); // passes once the settings are loaded
    Process(settings.blockSize);
};
auto w1 = TC::Tasks::Task::Create(worker);
auto w2 = TC::Tasks::Task::Create(worker);
auto w3 = TC::Tasks::Task::Create(worker);

TC::Synchronization::TaskGroup_s workers;
TC::Tasks::Task*                 workerTasks[] = {&w1, &w2, &w3};
for (TC::Tasks::Task* t : workerTasks) {
    workers.AddTask(t);
    scheduler->AddTask(t);
}
scheduler->AddTask(&settings);
workers.Wait();

Wait for a group of tasks

TaskGroup counts tasks: AddTask counts a task in, the task's exit counts it out, and Wait passes once the count is 0. The group learns about the exit by becoming the task's exit callback (SetNotifyCallback), so the tasks can be plain Task objects:

// A plain Task: its end is observed through the group it belongs to.
class RangeTask : public TC::Tasks::Task
{
public:
    void Set(std::uint64_t begin, std::uint64_t end) noexcept
    {
        begin_ = begin;
        end_   = end;
        sum_   = 0;
    }

    void Run() noexcept override
    {
        for (auto i = begin_; i < end_; ++i)
            sum_ += i;
    }

    std::uint64_t Sum() const noexcept { return sum_; }

private:
    std::uint64_t begin_ = 0;
    std::uint64_t end_   = 0;
    std::uint64_t sum_   = 0;
};

Fork eight of them and join:

// Fork/join from main: eight tasks, one group.
constexpr std::uint64_t          n = 8'000'000;
RangeTask                        parts[8];
TC::Synchronization::TaskGroup_s group;

for (std::uint64_t i = 0; i < 8; ++i) {
    parts[i].Set(n * i / 8, n * (i + 1) / 8);
    group.AddTask(&parts[i]); // before the task is added to the scheduler
    scheduler->AddTask(&parts[i]);
}
group.Wait(); // passes once all eight have exited

std::uint64_t total = 0;
for (auto& part : parts)
    total += part.Sum();

TaskGroup allows any number of waiting tasks; TaskGroup_s allows one, is cheaper and also offers Wait(Duration timeout). TryWait tells without waiting whether the count is 0.

A task of the group may add further tasks to the same group while it runs: the group cannot complete before that task has exited. From outside the group, add tasks only while the group cannot complete, or before anyone waits for the round they belong to.

Reusing a group

Once the count has reached 0, the group can be used for the next round. The task objects can be reused too, because a task may be added again once its OnTaskExited has been called, and the group's Wait passes only after that:

// The count is back at 0: the same group and the same task objects run a second
// round.
for (std::uint64_t i = 0; i < 8; ++i) {
    parts[i].Set(2 * n * i / 8, 2 * n * (i + 1) / 8);
    group.AddTask(&parts[i]);
    scheduler->AddTask(&parts[i]);
}
group.Wait();

total = 0;
for (auto& part : parts)
    total += part.Sum();

Tasks from lambdas

Task::Create wraps a callable in a LambdaTask and returns it by value. The task is not scheduled yet: keep it in a variable that outlives the task, then add it to a group and a scheduler, as the workers in the example above do:

auto w1 = TC::Tasks::Task::Create(worker);

The callable is copied or moved into the task object. References it captures must stay valid until the task has exited. The callable must not throw: an exception that escapes Run ends the process. Task objects cannot be copied or moved, since the scheduler holds their address.

Schedule or AddTask

There are two ways to start a task:

• ITaskScheduler::AddTask on the scheduler object. It works from any thread or task, also before Start (the task then runs once the scheduler has started). main uses it, because a converted thread does not belong to a scheduler.
• Task::Schedule adds the task to the scheduler of the calling task. Call it from a task running on a scheduler.

Both take a TaskPriority. TaskPriority::Default inherits the priority of the calling task, or means Normal when the caller is not a task. Neither call blocks. The scheduler's queues have a fixed capacity (SchedulerInfo::taskQueueCapacity), which must cover the tasks that can be runnable at the same time.

A task that forks and joins children from inside Run looks like this:

// A task that forks children onto its own scheduler with Schedule and joins them.
// The children and the group are members, so they live as long as the parent
// object.
class ParentTask : public TC::Tasks::WaitableTask_s
{
public:
    void Run() noexcept override
    {
        constexpr std::uint64_t n = 4'000'000;
        for (std::uint64_t i = 0; i < Children; ++i) {
            children_[i].Set(n * i / Children, n * (i + 1) / Children);
            group_.AddTask(&children_[i]); // count in before scheduling
            children_[i].Schedule(TC::TaskPriority::High);
        }
        group_.Wait(); // this task suspends, its thread runs the children

        for (auto& child : children_)
            total_ += child.Sum();
    }

    std::uint64_t Total() const noexcept { return total_; }

private:
    static constexpr std::uint64_t   Children = 4;
    RangeTask                        children_[Children];
    TC::Synchronization::TaskGroup_s group_;
    std::uint64_t                    total_ = 0;
};

The parent suspends in Wait, so the scheduler thread it ran on is free to run the children. main starts the parent like any other task and waits for it:

ParentTask parent;
scheduler->AddTask(&parent);
parent.Wait();

Task object lifetimes

The objects you wait on have rules of their own:

• A WaitableTask_s may be destroyed as soon as its single waiter's Wait has returned.
• A WaitableTask (several waiters) must not be destroyed while a waiter may still be suspending on it or its exit is still resuming the other waiters. The return of one Wait alone does not guarantee that.
• A TaskGroup or TaskGroup_s must exist as long as tasks can still report to it or wait on it, including the exit notification of the last task, which may still be running when a Wait returns. This applies to both variants.

The example program declares every object in main and shuts down only after all tasks have exited, so every object outlives its tasks:

scheduler->Stop(); // every task has exited; objects above still exist
scheduler->Destroy();
TC::ConvertTaskToThread();

Stop does not run queued or suspended tasks to completion, so wait for your tasks first. Call it from outside the scheduler's threads.

The program prints:

one task:      499999500000
three workers: 196608
eight tasks:   31999996000000
second round:  127999992000000
parent/child:  7999998000000
OK

See also

• WaitableTask, WaitableTask_s, TaskGroup, TaskGroup_s, Task, LambdaTask, ITaskScheduler
• Tasks, Lifetimes and sizing, Single-waiter variants, Threads as tasks, Priorities
• Samples: Hello task, Wait for a group