Tutorial 4: Sizing the work

On this page

Where we are

Chapter 3 encodes every block in its own task and writes the blocks in order. It is several times faster than the sequential reference.

The problem

Chapter 3 took the scheduler's defaults and a block size of 1 MB without asking whether they fit. Try other block sizes (-b, in KB) on the 256 MB test input, each time with a reference packed with the same block size:

Very small blocks (a few KB) make tens of thousands of tasks. Each task costs a little to schedule, and each block adds a header to the output. Worse: chapter 3 adds all tasks at once, and with 4 KB blocks that is 65,536 runnable tasks. The scheduler's task queues hold 4,096 by default. Overfilling them is invalid use, and a release build does not check it.
Very large blocks (16 MB and more) make few tasks. With 16 blocks on a machine with 32 threads, half the threads never get a block, and at the end every thread waits for the slowest block.

Between the two lies a wide range in which the block size hardly matters. The program has to stay inside it, and the scheduler has to be sized for the work it gets.

The building block

SchedulerInfo describes the scheduler that CreateTaskScheduler makes. Two of its fields matter here:

numberOfThreads: the number of scheduler threads; 0 means one per logical processor. Set it lower to share the machine with other work, or to see how the program scales (-t).
taskQueueCapacity: how many tasks can be runnable at the same time; 0 means DefaultTaskQueueCapacity (4,096). It is rounded up to a power of two and may not exceed MaximumTaskQueueCapacity. Suspended tasks do not count.

Why doesn't the queue just grow? Growing means allocating and locking on the path every task takes. ConcurrentTasks puts performance first: the queues are fixed, and their size is the program's decision. See Performance first.

Important

Contract. The task queue capacity must cover the most tasks the program can have runnable at the same time. Filling a queue is invalid use; a release build does not check it. See Lifetimes and sizing.

The change

The Packer is the same as in chapter 3. main now works out how many blocks the file has before it creates the scheduler, and sizes the scheduler for them:

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

    // Every block becomes a task, and all of them are runnable at once: the
    // task queues must hold that many.
    const std::uint64_t fileSize = File::OpenRead(arguments.input).Size();
    const std::uint64_t blocks
        = (fileSize + arguments.blockSize - 1) / arguments.blockSize;
    if (blocks > TC::MaximumTaskQueueCapacity) {
        std::printf("%llu blocks are too many; choose larger blocks\n",
                    static_cast<unsigned long long>(blocks));
        return 1;
    }

    TC::ConvertThreadToTask();

    TC::SchedulerInfo info{};
    info.numberOfThreads   = arguments.threads; // 0: one per logical processor
    info.taskQueueCapacity = std::max(static_cast<std::uint32_t>(blocks),
                                      TC::DefaultTaskQueueCapacity);
    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 (%llu blocks of %zu KB, %u threads)\n", total.Seconds(),
        static_cast<unsigned long long>(blocks), arguments.blockSize / 1024,
        arguments.threads != 0 ? arguments.threads
                               : GetActiveProcessorCount(ALL_PROCESSOR_GROUPS));

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

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

Every block becomes a task, and chapter 3 adds them all at once, so the capacity must be at least the number of blocks. The program refuses block counts above the maximum.
-t sets the number of threads; the output line shows the blocks and threads of the run.

The result

blockpack input.bin output.bpk -r reference.bpk
time: … s (256 blocks of 1024 KB, … threads)
OK

Run it with -t 1, -t 2, -t 4 and so on up to the number of logical processors. The time drops almost in proportion at first and flattens out towards the end – partly because the machine's cores share caches and memory bandwidth, partly because the whole file is still read before the first block is encoded.

Rules of thumb for the size of a piece of work:

Long enough to pay for scheduling it: tens of microseconds or more. Encoding a 1 MB block takes milliseconds, far above that.
Short enough that there are several pieces per thread, so that the load balances itself.

What limits us now

The capacity grows with the file. That works for this program, but a design whose number of runnable tasks depends on the input size is fragile.
Nothing is encoded until the whole file has been read.
The whole file and every encoded block are in memory at the same time.

The next chapter starts encoding while the file is still being read.

Learn more

Show the whole program

// Tutorial chapter 4: sizing the work - the number of scheduler threads and the
// capacity of the task queues are chosen for the work instead of taken as given.
#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, as in chapter 3.
class EncodeTask : public TC::Tasks::WaitableTask
{
public:
    void Run() noexcept override { EncodeBlock(input, encoded); }

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

// Forks one task per block and writes the blocks in order, as in chapter 3.
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;

        const std::size_t blockSize = arguments.blockSize;
        const std::size_t count     = (data_.size() + blockSize - 1) / blockSize;
        tasks_                      = std::make_unique<EncodeTask[]>(count);
        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]);
        }

        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;

    // Every block becomes a task, and all of them are runnable at once: the
    // task queues must hold that many.
    const std::uint64_t fileSize = File::OpenRead(arguments.input).Size();
    const std::uint64_t blocks
        = (fileSize + arguments.blockSize - 1) / arguments.blockSize;
    if (blocks > TC::MaximumTaskQueueCapacity) {
        std::printf("%llu blocks are too many; choose larger blocks\n",
                    static_cast<unsigned long long>(blocks));
        return 1;
    }

    TC::ConvertThreadToTask();

    TC::SchedulerInfo info{};
    info.numberOfThreads   = arguments.threads; // 0: one per logical processor
    info.taskQueueCapacity = std::max(static_cast<std::uint32_t>(blocks),
                                      TC::DefaultTaskQueueCapacity);
    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 (%llu blocks of %zu KB, %u threads)\n", total.Seconds(),
        static_cast<unsigned long long>(blocks), arguments.blockSize / 1024,
        arguments.threads != 0 ? arguments.threads
                               : GetActiveProcessorCount(ALL_PROCESSOR_GROUPS));

    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 5: Reading without blocking