Tutorial 3: One task per block

On this page

Where we are

Chapter 2 runs the sequential loop as one task. The output is correct, but only one core works.

The problem

Encoding takes about 90 % of the time, and no block depends on another. A single task cannot use more than one core; the encoding has to be split into tasks that can run at the same time.

The building block

One WaitableTask per block. You know the type from chapter 2; what is new is the pattern, fork/join: start many tasks, then wait for their results.

Fork: AddTask for every block. The scheduler threads take the tasks from the queue as they become free.
Join in order: main waits for block 0, writes it, waits for block 1, writes it, and so on. While it writes one block, the blocks after it are still being encoded. A block that finishes early waits in its task object until it is its turn.

Writing in block order is what keeps the output identical to chapter 1, even though the blocks are encoded in any order on any thread.

Why not one task per core, each with an equal share of the file? Blocks differ in how long they take, and threads are interrupted by other work on the machine. With a fixed share per core, the fastest cores finish early and sit idle while the slowest one finishes its share. With many small tasks, a free thread simply takes the next block, and the load balances itself.

Important

Contract. Every task object must outlive its task, up to and including its exit. A returned Wait is not enough to destroy a WaitableTask: its exit may still be finishing. The program keeps all task objects until the scheduler has stopped. See Lifetimes and sizing.

The change

The task encodes one block. Input and output are members, so the result stays in the object until main writes it:

// Encodes one block. Blocks are independent, so these tasks can run in any
// order and on any thread.
class EncodeTask : public TC::Tasks::WaitableTask
{
public:
    void Run() noexcept override { EncodeBlock(input, encoded); }

    std::span<const std::byte> input;
    EncodedBlock               encoded;
};

A Packer reads the whole file into memory, forks one task per block and joins them in order:

// Holds the file and the tasks. It must outlive every task it started, so main
// destroys it only after the scheduler has stopped.
class Packer
{
public:
    bool Pack(TC::ITaskScheduler& scheduler, const Arguments& arguments)
    {
        File input  = File::OpenRead(arguments.input);
        File output = File::Create(arguments.output);
        if (!input.IsOpen() || !output.IsOpen() || !input.ReadAll(data_))
            return false;

        // Fork: one task per block.
        const std::size_t blockSize = arguments.blockSize;
        const std::size_t count     = (data_.size() + blockSize - 1) / blockSize;
        tasks_ = std::make_unique<EncodeTask[]>(count); // fixed addresses
        for (std::size_t i = 0; i < count; ++i) {
            const std::size_t offset = i * blockSize;
            tasks_[i].input          = std::span(data_).subspan(
                offset, std::min(blockSize, data_.size() - offset));
            scheduler.AddTask(&tasks_[i]);
        }

        // Join in order: block i is written as soon as it is ready, while the
        // blocks after it are still being encoded.
        std::uint32_t crc = 0;
        bool          ok  = true;
        for (std::size_t i = 0; i < count; ++i) {
            tasks_[i].Wait();
            crc = WriteBlock(output, tasks_[i].encoded, crc, ok);
        }
        return WriteFooter(output, crc) && ok;
    }

private:
    std::vector<std::byte>        data_;
    std::unique_ptr<EncodeTask[]> tasks_;
};

The tasks live in an array that is allocated once (std::make_unique<EncodeTask[]>). The scheduler holds their addresses, so they must never move – a std::vector that grows would move them.
Packer keeps the file data and the tasks. main creates it before the first task is added and destroys it after Stop, as the contract asks:

int wmain(int argc, wchar_t* argv[])
{
    Arguments arguments;
    if (!ParseArguments(argc, argv, arguments))
        return arguments.exitCode;

    TC::ConvertThreadToTask();

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

    Packer          packer; // lives until after Stop
    const Stopwatch total;
    const bool      packed = packer.Pack(*scheduler, arguments);
    std::printf("time: %.2f s\n", total.Seconds());

    scheduler->Stop();
    scheduler->Destroy();
    TC::ConvertTaskToThread();

    if (!packed) {
        std::printf("cannot pack the file\n");
        return 1;
    }
    return CheckOutput(arguments) ? 0 : 1;
}

The result

blockpack input.bin output.bpk -r reference.bpk
time: … s
OK

The output is still byte for byte the same as chapter 1's, and the time drops by a factor that grows with the number of cores. The scheduler threads now encode at the same time while main writes behind them:

main      │ read whole file │ wait         │W│W│W│W│ wait │W│W│W│W│ …
thread 1  │ idle            │ encode 0     │ encode 4     │ …
thread 2  │ idle            │ encode 1     │ encode 5     │ …
thread 3  │ idle            │ encode 2     │ encode 6     │ …
thread 4  │ idle            │ encode 3     │ encode 7     │ …

What limits us now

Size. Every block is a task. With a large file or small blocks (-b) the program starts thousands of tasks at once – more than the scheduler's task queues are made for. How large should a block be, and how many tasks may be waiting to run?
Reading. Nothing is encoded until the whole file has been read.
Memory. The whole file and all encoded blocks are in memory at the same time.

The next chapter starts with the first point: sizing the work.

Learn more

Waiting for tasks – waiting for many tasks at once
Wait for a group – fork/join when the order does not matter
Tasks, Lifetimes and sizing
WaitableTask, AddTask

Show the whole program

// Tutorial chapter 3: one task per block - the blocks are encoded in parallel,
// main writes them in order as they become ready.
#include "tutorial-common.h"

#include <memory>

#include <ThinkMeta.ConcurrentTasks/Core.h>
import ThinkMeta.ConcurrentTasks.Tasks.WaitableTask;

namespace TC = ThinkMeta::ConcurrentTasks;
using namespace Tutorial;

// Encodes one block. Blocks are independent, so these tasks can run in any
// order and on any thread.
class EncodeTask : public TC::Tasks::WaitableTask
{
public:
    void Run() noexcept override { EncodeBlock(input, encoded); }

    std::span<const std::byte> input;
    EncodedBlock               encoded;
};

// Holds the file and the tasks. It must outlive every task it started, so main
// destroys it only after the scheduler has stopped.
class Packer
{
public:
    bool Pack(TC::ITaskScheduler& scheduler, const Arguments& arguments)
    {
        File input  = File::OpenRead(arguments.input);
        File output = File::Create(arguments.output);
        if (!input.IsOpen() || !output.IsOpen() || !input.ReadAll(data_))
            return false;

        // Fork: one task per block.
        const std::size_t blockSize = arguments.blockSize;
        const std::size_t count     = (data_.size() + blockSize - 1) / blockSize;
        tasks_ = std::make_unique<EncodeTask[]>(count); // fixed addresses
        for (std::size_t i = 0; i < count; ++i) {
            const std::size_t offset = i * blockSize;
            tasks_[i].input          = std::span(data_).subspan(
                offset, std::min(blockSize, data_.size() - offset));
            scheduler.AddTask(&tasks_[i]);
        }

        // Join in order: block i is written as soon as it is ready, while the
        // blocks after it are still being encoded.
        std::uint32_t crc = 0;
        bool          ok  = true;
        for (std::size_t i = 0; i < count; ++i) {
            tasks_[i].Wait();
            crc = WriteBlock(output, tasks_[i].encoded, crc, ok);
        }
        return WriteFooter(output, crc) && ok;
    }

private:
    std::vector<std::byte>        data_;
    std::unique_ptr<EncodeTask[]> tasks_;
};

int wmain(int argc, wchar_t* argv[])
{
    Arguments arguments;
    if (!ParseArguments(argc, argv, arguments))
        return arguments.exitCode;

    TC::ConvertThreadToTask();

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

    Packer          packer; // lives until after Stop
    const Stopwatch total;
    const bool      packed = packer.Pack(*scheduler, arguments);
    std::printf("time: %.2f s\n", total.Seconds());

    scheduler->Stop();
    scheduler->Destroy();
    TC::ConvertTaskToThread();

    if (!packed) {
        std::printf("cannot pack the file\n");
        return 1;
    }
    return CheckOutput(arguments) ? 0 : 1;
}

Source and project on GitHub

Tutorial 4: Sizing the work