Tutorial 2: A scheduler and a task

Where we are

Chapter 1 packs the file sequentially and shows that encoding takes almost all of the time.

The problem

To encode on all cores, the program needs something that runs work on them. In this chapter the work does not change at all: the loop of chapter 1 runs as a single task. That isolates the new building blocks – the scheduler and a task – from the parallelization, which comes in the next chapter.

The building blocks

The scheduler owns one thread per logical processor and runs tasks on them. A task is an object with a Run method. When a task waits – for another task, a semaphore, a file read – it is suspended, and its thread runs other tasks in the meantime. See The scheduler and Tasks.

WaitableTask is a task that others can wait for. Its Wait suspends the caller until the task's Run has returned.

ConvertThreadToTask lets main wait like a task. Waiting is only allowed in a task; this call gives the main thread what it needs to be suspended and resumed.

Why not std::thread? A thread per piece of work is expensive to create and blocks while it waits. Tasks are cheap – a task per block, as in the next chapter, is fine – and a waiting task does not hold on to a thread. One scheduler serves every part of the program.

The change

The program now includes the core header and imports the one module it needs:

#include "tutorial-common.h"

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

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

Pack is the loop of chapter 1 without the stopwatches. A task wraps it; the result stays in the task object:

// The whole pack loop as one task. WaitableTask adds Wait, so main can wait for
// it; the result stays in the object.
class PackTask : public TC::Tasks::WaitableTask
{
public:
    explicit PackTask(const Arguments& arguments) noexcept : arguments_(arguments)
    {}

    void Run() noexcept override { packed = Pack(arguments_); }

    bool packed = false;

private:
    const Arguments& arguments_;
};

main converts its thread, creates and starts a scheduler, adds the task and waits for it:

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

    // main becomes a task too, so it may wait for one.
    TC::ConvertThreadToTask();

    // One scheduler thread per logical processor.
    TC::SchedulerInfo   info{};
    TC::ITaskScheduler* scheduler
        = TC::CreateTaskScheduler(TC::SchedulerKind::Default, &info);
    if (scheduler == nullptr)
        return 1;
    scheduler->Start();

    PackTask        task(arguments); // lives until after Stop
    const Stopwatch total;
    scheduler->AddTask(&task);
    task.Wait(); // main is suspended until Run has returned
    std::printf("time: %.2f s\n", total.Seconds());

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

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

1. ConvertThreadToTask comes first, so that task.Wait() is allowed on the main thread.
2. CreateTaskScheduler with a zero-initialized SchedulerInfo means one thread per logical processor. Start starts the threads.
3. AddTask queues the task; a scheduler thread calls its Run.
4. Wait suspends main until Run has returned.
5. Stop, Destroy and ConvertTaskToThread shut down in reverse order. task is destroyed at the end of main, after Stop, so it outlives everything the scheduler does with it.

The result

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

The output is the same as in chapter 1, and so is the time. One task runs on one thread at a time, so it uses one core – the other scheduler threads are idle:

thread 1  │R│ encode │W│R│ encode │W│R│ encode │W│ …
thread 2  idle
thread 3  idle

Pack still calls ReadFile and WriteFile directly. They block the scheduler thread while the disk works. With one task that does not matter; chapter 5 replaces them.

What limits us now

A single task uses a single core. The blocks are independent, so each one can be its own task.

Learn more

• Your first program – the same frame, step by step
• The scheduler, Tasks, Threads as tasks
• WaitableTask, CreateTaskScheduler, SchedulerInfo

// Tutorial chapter 2: a scheduler and one task - the sequential pack loop runs
// unchanged as a single task.
#include "tutorial-common.h"

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

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

// The loop of chapter 1, without the stage times.
bool Pack(const Arguments& arguments)
{
    File input  = File::OpenRead(arguments.input);
    File output = File::Create(arguments.output);
    if (!input.IsOpen() || !output.IsOpen())
        return false;

    std::vector<std::byte> buffer(arguments.blockSize);
    EncodedBlock           block;
    std::uint32_t          crc = 0;
    bool                   ok  = true;
    for (;;) {
        const std::size_t size = input.Read(buffer);
        if (size == 0)
            break;
        EncodeBlock(std::span(buffer).first(size), block);
        crc = WriteBlock(output, block, crc, ok);
    }
    return WriteFooter(output, crc) && ok;
}

// The whole pack loop as one task. WaitableTask adds Wait, so main can wait for
// it; the result stays in the object.
class PackTask : public TC::Tasks::WaitableTask
{
public:
    explicit PackTask(const Arguments& arguments) noexcept : arguments_(arguments)
    {}

    void Run() noexcept override { packed = Pack(arguments_); }

    bool packed = false;

private:
    const Arguments& arguments_;
};

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

    // main becomes a task too, so it may wait for one.
    TC::ConvertThreadToTask();

    // One scheduler thread per logical processor.
    TC::SchedulerInfo   info{};
    TC::ITaskScheduler* scheduler
        = TC::CreateTaskScheduler(TC::SchedulerKind::Default, &info);
    if (scheduler == nullptr)
        return 1;
    scheduler->Start();

    PackTask        task(arguments); // lives until after Stop
    const Stopwatch total;
    scheduler->AddTask(&task);
    task.Wait(); // main is suspended until Run has returned
    std::printf("time: %.2f s\n", total.Seconds());

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

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

Next

• Tutorial 3: One task per block