Task graphs

Most programs that use every core are not a single loop split into equal parts. They are a graph: stages that must run in order (reading a file, writing a bitstream, carrying state from one frame to the next) and stages whose pieces are independent (transforming blocks, analyzing frames). This guide shows how to map such a graph onto tasks with the primitives of ConcurrentTasks, using a three-stage pipeline that reads a file, transforms its blocks in parallel and writes them in order. The example is guide-task-graphs.cpp, a complete program that checks its own output.

Stages as tasks

Draw the computation first. Each box is a piece of work, each arrow a dependency. Then:

A serial stage, whose steps must happen one after the other, becomes one long-running task that loops over its steps. It keeps its state in local variables, and nothing else needs a lock to touch that state.
A parallel stage becomes one task per independent piece, or a fixed set of worker tasks that take pieces from a queue. Tasks are cheap: a task per block or per frame is fine.
Each arrow becomes a wait: a TaskGroup where one piece joins several, a semaphore where a ring of buffers connects two stages, an event where one task tells another that something is ready.

A waiting task is suspended, not blocked, so a stage that waits for its input does not take a thread away from the stages that produce it.

The example: read, transform, write

reader (serial) --> transform (parallel, one task per block) --> writer (serial, in order)
         \______________ ring of 8 slots ______________________/

The pipeline moves the file in blocks of 1 MB, and the ring has 8 slots, an arbitrary number that keeps several blocks in flight. Transform stands for the per-byte work of the parallel stage:

constexpr DWORD    BlockSize = 1 << 20; // 1 MB per block
constexpr unsigned Slots     = 8; // blocks in flight between reader and writer

// The work of the parallel stage, for one byte at a file position.
std::uint8_t Transform(std::uint8_t value, std::uint64_t position) noexcept;

Each block travels through one slot of a ring: the reader fills the slot's input buffer and forks a transform task for it, the transform fills the output buffer, the writer writes it at the block's file offset and frees the slot. A slot carries its own task object and a TaskGroup_s that joins it:

// One slot of the ring: a block on its way from the reader through a transform task
// to the writer.
struct Slot;

class TransformTask : public TC::Tasks::Task
{
public:
    Slot* slot = nullptr;
    void  Run() noexcept override;
};

struct Slot {
    std::vector<std::uint8_t> input  = std::vector<std::uint8_t>(BlockSize);
    std::vector<std::uint8_t> output = std::vector<std::uint8_t>(BlockSize);
    std::uint64_t             offset = 0;
    DWORD                     size   = 0; // 0: the end of the input
    TransformTask             transform;
    TC::Synchronization::TaskGroup_s
        transformed; // the writer waits here for this slot's block
};

void TransformTask::Run() noexcept
{
    for (DWORD i = 0; i < slot->size; ++i)
        slot->output[i] = Transform(slot->input[i], slot->offset + i);
}

The pipeline holds the ring and two semaphores:

class Pipeline
{
public:
    Pipeline(const wchar_t* inputPath, const wchar_t* outputPath) noexcept
        : inputPath_(inputPath),
          outputPath_(outputPath)
    {
        for (Slot& slot : slots_)
            slot.transform.slot = &slot;
    }

    // Serial stage 1: read block after block into free slots and fork a transform
    // for each. An empty slot marks the end of the input for the writer.
    void Read() noexcept
    {
        AsyncFile  file;
        const bool open = Error::Succeeded(
            file.Create(inputPath_, GENERIC_READ, FILE_SHARE_READ, OPEN_EXISTING,
                        FILE_FLAG_SEQUENTIAL_SCAN));
        readFailed_ = !open;
        for (unsigned block = 0;; ++block) {
            freeSlots_
                .Wait(); // at most Slots blocks in flight: bounds memory and tasks
            Slot& slot  = slots_[block % Slots];
            slot.offset = std::uint64_t{BlockSize} * block;
            slot.size   = 0;
            if (open) {
                const auto error = file.Read(slot.input.data(), BlockSize,
                                             slot.size, slot.offset);
                if (Error::Failed(error) && error != Error::EndOfFile)
                    readFailed_ = true; // slot.size is 0: the pipeline ends here
            }
            if (slot.size == 0) {
                readySlots_.Release(); // the end mark
                return;
            }
            slot.transformed.AddTask(&slot.transform);
            slot.transform.Schedule(
                TC::TaskPriority::Normal); // Default would inherit High
            readySlots_.Release();         // the writer may wait for this block now
        }
    }

    // Serial stage 3: write the blocks in order, each as soon as its transform has
    // exited.
    void Write() noexcept
    {
        AsyncFile file;
        writeFailed_ = Error::Failed(file.Create(
            outputPath_, GENERIC_WRITE, 0, CREATE_ALWAYS, FILE_ATTRIBUTE_NORMAL));
        for (unsigned block = 0;; ++block) {
            readySlots_.Wait(); // the reader has counted the transform in, or ended
            Slot& slot = slots_[block % Slots];
            if (slot.size == 0)
                return;              // the end mark
            slot.transformed.Wait(); // the join of this block's transform
            DWORD written = 0;
            if (!writeFailed_
                && (Error::Failed(file.Write(slot.output.data(), slot.size, written,
                                             slot.offset))
                    || written != slot.size))
                writeFailed_ = true;
            freeSlots_.Release(); // the slot can take the next block
        }
    }

    bool Failed() const noexcept { return readFailed_ || writeFailed_; }

private:
    const wchar_t*                   inputPath_;
    const wchar_t*                   outputPath_;
    Slot                             slots_[Slots];
    TC::Synchronization::Semaphore_s freeSlots_{
        Slots}; // free slots; one waiter: the reader
    TC::Synchronization::Semaphore_s readySlots_{
        0}; // forked blocks; one waiter: the writer
    bool readFailed_  = false;
    bool writeFailed_ = false;
};

freeSlots_ starts at the number of slots. The reader takes a permit per block and the writer gives it back once the block is written. This bounds the blocks in flight, and with them the memory and the number of transform tasks.
readySlots_ counts blocks that the reader has forked. The writer waits for it before it waits on the slot's group: a group whose count is 0 lets Wait pass at once, so the writer must not wait for a round before the reader has added the task to it.
At the end of the file, or after a read error, the reader passes on an empty slot without a transform. The writer stops when it reaches it, so neither stage needs to know the number of blocks in advance.
The writer handles the blocks strictly in order, so the output is in order no matter in which order the transforms finish. A block that finishes early simply waits in its slot.

Each semaphore has exactly one waiter (the reader, the writer), so the cheaper Semaphore_s is enough. A slot's task object is reused for the next block only after the writer's Wait on the slot's group has passed, that is, after the previous run has exited.

Priorities for the critical path

The serial stages decide how long the whole run takes: if the reader is late, the transforms starve; if the writer is late, the slots fill up and the reader stops. Give them TaskPriority::High and the parallel work Normal:

static Pipeline pipeline(input, output); // large: keep it off the stack
auto reader = TC::Tasks::Task::Create([]() noexcept { pipeline.Read(); });
auto writer = TC::Tasks::Task::Create([]() noexcept { pipeline.Write(); });

TC::Synchronization::TaskGroup_s stages;
stages.AddTask(&reader);
stages.AddTask(&writer);
scheduler->AddTask(
    &reader, TC::TaskPriority::High); // the serial stages are the critical path
scheduler->AddTask(&writer, TC::TaskPriority::High);
stages.Wait(); // the writer waits for every transform, so all have run

A High task is always taken before Normal work, so a reader or writer that is resumed after its I/O runs as soon as a thread is free instead of queuing behind dozens of transforms. Two details:

Task::Schedule and AddTask with TaskPriority::Default inherit the priority of the calling task. The reader runs at High, so it passes TaskPriority::Normal explicitly when it forks the transforms.
There is no aging: a steady stream of High work starves Normal tasks. Keep High for the few tasks on the critical path, which spend most of their time waiting anyway.

YieldTask with YieldFlags::HigherPriority lets a long-running Normal task give way to waiting High work at points of its choosing. See Priorities and Yielding.

Avoiding oversubscription

The scheduler runs one thread per logical processor by default. More runnable tasks than threads do not make the work faster; they only use memory and queue slots. Keep the graph bounded:

Bound the work in flight with a semaphore over a ring of buffers, as above. A ring of two to four times the number of threads keeps every thread busy while the serial stages catch up.
The scheduler's task queues have a fixed capacity (SchedulerInfo::taskQueueCapacity, default DefaultTaskQueueCapacity). It must cover the tasks that can be runnable at the same time; a bounded graph makes that number known.
Size pieces so that a task runs long enough to pay for scheduling it, roughly tens of microseconds or more, and short enough that there are several pieces per thread.
Do not block a scheduler thread: use AsyncFile, sockets and the library's primitives, which suspend the task instead.

Important

Every object in the graph has to outlive the tasks that use it: slots, groups, semaphores and the task objects themselves. The example keeps the pipeline in static storage and stops the scheduler only after the serial stages, and with them all transforms, have finished. See Lifetimes and sizing.

The program prints:

output:  in order and correct
OK

Real-world graphs

The same building blocks scale to complex codecs. Each of these case studies shows, step by step, how a sequential reference implementation became a task graph on ConcurrentTasks that produces byte-identical output:

ctLame: an MP3 encoder in which every frame depends on the state of the frame before it.
ctJpeg: a JPEG encoder built as a sequence of fork/join stages, each handing hundreds of independent jobs to the threads.
ctOpus: an Opus encoder that ends up computing future frames ahead on speculation.
ctDeflate: a gzip compressor in which block boundaries and encodings depend on everything before them.