Tutorial 10: Ready-made stages

On this page

Where we are

Chapter 9 completes the pipeline: a reader, parallel encoding with a CRC per block, a writer in order, and a clean shutdown after an error or Ctrl+C.

The problem

The pipeline works, but a good part of it is plumbing that every file pipeline needs again:

The reader reads into slot buffers with AsyncFile, block by block, and the encode task then reads the block from that buffer.
The writer calls a blocking WriteFile for every block. main waits for the disk each time, and while it waits it neither takes the next block nor frees a slot.

Now that you know what a reader and a writer have to do – read ahead but not too far, keep the order, free memory when it is no longer used, stop cleanly – you can use the two stages that ConcurrentTasks provides for exactly this.

The building blocks

StreamedFileReader streams a file into one reserved address range. Every byte sits at its file position, as in a mapped file: block n starts at Data() + n * blockSize. Its own task reads segment after segment, at most readAhead bytes ahead of the oldest byte still in use.

Wait(end) suspends the calling task until every byte before end is loaded.
Release(user, below) says that a user is done with every byte before below; whole segments below that are decommitted, and the reader may read further.

OrderedFileWriter builds the output front to back in memory. One producer appends with Write, which copies the bytes at once; its own writer task writes complete segments to the file behind the producer, with several writes in flight. WriteDeferred goes further and lets copy tasks copy the bytes later; it is not used here, because a slot's encoded block is reused as soon as the slot is free.

Why not keep the hand-written stages? They work, and you now understand what the ready-made ones do inside. But the ready-made stages read straight into the place where the data is used, write without making the producer wait for the disk, bound their memory by setting rather than by code, and have been measured and tested. Use them unless your stage needs something they do not offer.

Important

Contract. StreamedFileReader: Wait before touching bytes, Release when done with them, and every user releases everything at the end (~0); a read error is kept in ReadError and the affected bytes read as zero, so check it. OrderedFileWriter: one producer; schedule Writer(); Close, then StopCopyTasks with the number of copy tasks. Destroy either object only after its tasks have ended.

The change

The program imports the two stages instead of AsyncFile:

#include "tutorial-common.h"

#include <atomic>
#include <memory>

#include <ThinkMeta.ConcurrentTasks/Core.h>
import ThinkMeta.ConcurrentTasks.AsyncIo.OrderedFileWriter;
import ThinkMeta.ConcurrentTasks.AsyncIo.StreamedFileReader;
import ThinkMeta.ConcurrentTasks.Synchronization.Semaphore;
import ThinkMeta.ConcurrentTasks.Synchronization.TaskGroup;
import ThinkMeta.ConcurrentTasks.Tasks.Task;
import ThinkMeta.ConcurrentTasks.Windows.Error;

namespace TC = ThinkMeta::ConcurrentTasks;
using TC::AsyncIo::OrderedFileWriter;
using TC::AsyncIo::StreamedFileReader;
using TC::Windows::Error;
using namespace Tutorial;

A slot no longer holds a buffer, only the end of its block in the file:

// A slot no longer holds a buffer: the encode task reads the block where the
// stream has put it.
struct Slot {
    std::uint64_t                    end = 0; // 0: the end of the input
    EncodeTask                       task;
    TC::Synchronization::TaskGroup_s encoded;
};

The Packer holds the stream and the writer. The reader task of chapter 9 becomes a fork task: it no longer reads, it waits for the stream:

class Packer
{
    // ...
    TC::ITaskScheduler*              scheduler_ = nullptr;
    std::size_t                      blockSize_ = 0;
    std::size_t                      slotCount_ = 0;
    StreamedFileReader               input_;
    OrderedFileWriter                output_;
    std::unique_ptr<Slot[]>          slots_;
    TC::Synchronization::Semaphore_s freeSlots_;
    TC::Synchronization::Semaphore_s readySlots_;
    ForkTask                         fork_{*this};
    TC::Synchronization::TaskGroup_s stages_;
};

Pack opens both stages and schedules their tasks, together with the fork task, at High. A task group tells when all three have ended:

class Packer
{
    // ...
    bool Pack(TC::ITaskScheduler& scheduler, const Arguments& arguments,
              std::size_t slotCount)
    {
        blockSize_ = arguments.blockSize;
        slotCount_ = slotCount;

        // The input: streamed in segments of one block, at most one ring ahead
        // of the oldest block still in use.
        TC::AsyncIo::StreamedFileReaderSettings readSettings;
        readSettings.segmentSize = static_cast<std::uint32_t>(blockSize_);
        readSettings.readAhead   = slotCount * blockSize_;
        if (Error::Failed(input_.Open(arguments.input, readSettings)))
            return false;

        // The output: the encoded size is not known in advance, so reserve
        // address space for the worst case of the encoder (twice the input).
        const std::uint64_t blocks = input_.Size() / blockSize_ + 1;
        const std::uint64_t bound  = 2 * input_.Size() + 16 * blocks + 16;
        const bool opened = Error::Succeeded(output_.Open(arguments.output, bound));

        // The stages run as tasks at High; the group tells when all have ended.
        TC::Tasks::Task* stages[] = {&input_, &output_.Writer(), &fork_};
        for (TC::Tasks::Task* stage : stages) {
            stages_.AddTask(stage);
            scheduler.AddTask(stage, TC::TaskPriority::High);
        }
        scheduler_ = &scheduler;
        slots_     = std::make_unique<Slot[]>(slotCount);
        if (!opened)
            stopRequested = true; // let the stages end, then report the error
        freeSlots_.Release(slotCount);
    // ...
};

The read-ahead is one ring of blocks: the stream keeps the blocks of every slot loaded, no more.
The output's final size is not known in advance. bound reserves address space – not memory – for the worst case of the encoder.

The fork task waits for each block in the stream and encodes it where it is, without a copy:

class Packer
{
    // ...
    // Takes block after block from the stream and forks its encoding. Wait
    // suspends this task until the stream has loaded the block.
    void Fork() noexcept
    {
        const std::uint64_t size = input_.Size();
        for (std::size_t block = 0;; ++block) {
            freeSlots_.Wait();
            Slot&               slot  = slots_[block % slotCount_];
            const std::uint64_t begin = std::uint64_t{blockSize_} * block;
            const std::uint64_t end
                = std::min<std::uint64_t>(begin + blockSize_, size);
            slot.end = 0;
            if (begin < size && !stopRequested && input_.Wait(end) >= end)
                slot.end = end;
            if (slot.end == 0) { // the end of the file, or a stop
                readySlots_.Release();
                return;
            }
            slot.task.input = std::span(input_.Data() + begin, end - begin);
            slot.encoded.AddTask(&slot.task);
            scheduler_->AddTask(&slot.task, TC::TaskPriority::Normal);
            readySlots_.Release();
        }
    }
    // ...
};

main appends the blocks in order and releases each block of the input when it is written:

class Packer
{
    // ...
    bool Pack(TC::ITaskScheduler& scheduler, const Arguments& arguments,
              std::size_t slotCount)
    {
        // ...
        // main is the producer of the output: it appends the blocks in order.
        std::uint32_t crc = 0;
        for (std::size_t block = 0;; ++block) {
            readySlots_.Wait();
            Slot& slot = slots_[block % slotCount];
            if (slot.end == 0)
                break;
            slot.encoded.Wait();
            const EncodedBlock& encoded = slot.task.encoded;
            const auto          header  = BlockHeader(encoded);
            output_.Write(header.data(), sizeof(header)); // copied at once
            output_.Write(encoded.data.data(), encoded.data.size());
            output_.EndBatch(); // the writer task may write up to here
            crc = CombineCrc32(crc, slot.task.crc, BlockBytes(encoded));
            if (output_.Failed())
                stopRequested = true;
            input_.Release(0, slot.end); // the stream may drop the block
            freeSlots_.Release();
        }
        output_.Write(&crc, sizeof(crc));
        const bool closed = output_.Close(); // waits for the last write
        input_.Stop();                       // ends the stream after a stop
        input_.Release(0, ~std::uint64_t{0});
        output_.StopCopyTasks(0); // no copy tasks: every Write copies at once
        stages_.Wait();           // the stream, the writer and the fork task
        return closed && Error::Succeeded(input_.ReadError()) && !stopRequested;
    }
    // ...
};

EndBatch publishes the progress: the writer task may write everything appended so far.
Release(0, slot.end) lets the stream drop the block and read further.
At the end: Close waits for the last write, Stop and Release(0, ~0) end the stream also after a stop, StopCopyTasks(0) releases the writer (no copy tasks were started), and the group waits for all stage tasks. Only then may the Packer be destroyed.

The result

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

The output is still byte for byte the reference. The program is shorter where it matters: no buffers per slot, no file reads in the pipeline, no blocking writes. main no longer waits for the disk, and the input is never copied – the encode tasks read it where the stream has put it.

What limits us now

Nothing that this tutorial set out to fix. The last page sums up the method and points to what comes next.

Learn more

Streaming large files – both stages in detail, copy tasks, WriteAt
Hashing, Ordered output
StreamedFileReader, OrderedFileWriter

Show the whole program

// Tutorial chapter 10: ready-made stages - StreamedFileReader streams the input
// into memory and OrderedFileWriter writes the output behind the producer; the
// hand-written reading, slot buffers and blocking writes are gone.
#include "tutorial-common.h"

#include <atomic>
#include <memory>

#include <ThinkMeta.ConcurrentTasks/Core.h>
import ThinkMeta.ConcurrentTasks.AsyncIo.OrderedFileWriter;
import ThinkMeta.ConcurrentTasks.AsyncIo.StreamedFileReader;
import ThinkMeta.ConcurrentTasks.Synchronization.Semaphore;
import ThinkMeta.ConcurrentTasks.Synchronization.TaskGroup;
import ThinkMeta.ConcurrentTasks.Tasks.Task;
import ThinkMeta.ConcurrentTasks.Windows.Error;

namespace TC = ThinkMeta::ConcurrentTasks;
using TC::AsyncIo::OrderedFileWriter;
using TC::AsyncIo::StreamedFileReader;
using TC::Windows::Error;
using namespace Tutorial;

// Set by Ctrl+C, as in chapter 9.
std::atomic<bool> stopRequested{false};

// Encodes one block and computes its CRC, as in chapter 7.
class EncodeTask : public TC::Tasks::Task
{
public:
    void Run() noexcept override
    {
        EncodeBlock(input, encoded);
        crc = BlockCrc(encoded);
    }

    std::span<const std::byte> input;
    EncodedBlock               encoded;
    std::uint32_t              crc = 0;
};

// A slot no longer holds a buffer: the encode task reads the block where the
// stream has put it.
struct Slot {
    std::uint64_t                    end = 0; // 0: the end of the input
    EncodeTask                       task;
    TC::Synchronization::TaskGroup_s encoded;
};

class Packer;

// Runs Packer::Fork as a task.
class ForkTask : public TC::Tasks::Task
{
public:
    explicit ForkTask(Packer& packer) noexcept : packer_(packer) {}

    void Run() noexcept override;

private:
    Packer& packer_;
};

class Packer
{
public:
    bool Pack(TC::ITaskScheduler& scheduler, const Arguments& arguments,
              std::size_t slotCount)
    {
        blockSize_ = arguments.blockSize;
        slotCount_ = slotCount;

        // The input: streamed in segments of one block, at most one ring ahead
        // of the oldest block still in use.
        TC::AsyncIo::StreamedFileReaderSettings readSettings;
        readSettings.segmentSize = static_cast<std::uint32_t>(blockSize_);
        readSettings.readAhead   = slotCount * blockSize_;
        if (Error::Failed(input_.Open(arguments.input, readSettings)))
            return false;

        // The output: the encoded size is not known in advance, so reserve
        // address space for the worst case of the encoder (twice the input).
        const std::uint64_t blocks = input_.Size() / blockSize_ + 1;
        const std::uint64_t bound  = 2 * input_.Size() + 16 * blocks + 16;
        const bool opened = Error::Succeeded(output_.Open(arguments.output, bound));

        // The stages run as tasks at High; the group tells when all have ended.
        TC::Tasks::Task* stages[] = {&input_, &output_.Writer(), &fork_};
        for (TC::Tasks::Task* stage : stages) {
            stages_.AddTask(stage);
            scheduler.AddTask(stage, TC::TaskPriority::High);
        }
        scheduler_ = &scheduler;
        slots_     = std::make_unique<Slot[]>(slotCount);
        if (!opened)
            stopRequested = true; // let the stages end, then report the error
        freeSlots_.Release(slotCount);

        // main is the producer of the output: it appends the blocks in order.
        std::uint32_t crc = 0;
        for (std::size_t block = 0;; ++block) {
            readySlots_.Wait();
            Slot& slot = slots_[block % slotCount];
            if (slot.end == 0)
                break;
            slot.encoded.Wait();
            const EncodedBlock& encoded = slot.task.encoded;
            const auto          header  = BlockHeader(encoded);
            output_.Write(header.data(), sizeof(header)); // copied at once
            output_.Write(encoded.data.data(), encoded.data.size());
            output_.EndBatch(); // the writer task may write up to here
            crc = CombineCrc32(crc, slot.task.crc, BlockBytes(encoded));
            if (output_.Failed())
                stopRequested = true;
            input_.Release(0, slot.end); // the stream may drop the block
            freeSlots_.Release();
        }
        output_.Write(&crc, sizeof(crc));
        const bool closed = output_.Close(); // waits for the last write
        input_.Stop();                       // ends the stream after a stop
        input_.Release(0, ~std::uint64_t{0});
        output_.StopCopyTasks(0); // no copy tasks: every Write copies at once
        stages_.Wait();           // the stream, the writer and the fork task
        return closed && Error::Succeeded(input_.ReadError()) && !stopRequested;
    }

    // Takes block after block from the stream and forks its encoding. Wait
    // suspends this task until the stream has loaded the block.
    void Fork() noexcept
    {
        const std::uint64_t size = input_.Size();
        for (std::size_t block = 0;; ++block) {
            freeSlots_.Wait();
            Slot&               slot  = slots_[block % slotCount_];
            const std::uint64_t begin = std::uint64_t{blockSize_} * block;
            const std::uint64_t end
                = std::min<std::uint64_t>(begin + blockSize_, size);
            slot.end = 0;
            if (begin < size && !stopRequested && input_.Wait(end) >= end)
                slot.end = end;
            if (slot.end == 0) { // the end of the file, or a stop
                readySlots_.Release();
                return;
            }
            slot.task.input = std::span(input_.Data() + begin, end - begin);
            slot.encoded.AddTask(&slot.task);
            scheduler_->AddTask(&slot.task, TC::TaskPriority::Normal);
            readySlots_.Release();
        }
    }

private:
    TC::ITaskScheduler*              scheduler_ = nullptr;
    std::size_t                      blockSize_ = 0;
    std::size_t                      slotCount_ = 0;
    StreamedFileReader               input_;
    OrderedFileWriter                output_;
    std::unique_ptr<Slot[]>          slots_;
    TC::Synchronization::Semaphore_s freeSlots_;
    TC::Synchronization::Semaphore_s readySlots_;
    ForkTask                         fork_{*this};
    TC::Synchronization::TaskGroup_s stages_;
};

void ForkTask::Run() noexcept
{
    packer_.Fork();
}

BOOL WINAPI OnConsoleCtrl(DWORD ctrlType) noexcept
{
    if (ctrlType != CTRL_C_EVENT)
        return FALSE;
    stopRequested = true;
    return TRUE;
}

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

    const std::uint32_t threads   = arguments.threads != 0
        ? arguments.threads
        : GetActiveProcessorCount(ALL_PROCESSOR_GROUPS);
    const std::size_t   slotCount = std::size_t{4} * threads;

    SetConsoleCtrlHandler(OnConsoleCtrl, TRUE);
    TC::ConvertThreadToTask();
    if (!TC::AsyncIo::EnableAsyncIo())
        return 1;

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

    bool packed = false;
    {
        Packer          packer;
        const Stopwatch total;
        packed = packer.Pack(*scheduler, arguments, slotCount);
        std::printf("time: %.2f s\n", total.Seconds());
        scheduler->Stop();
    }

    scheduler->Destroy();
    TC::AsyncIo::DisableAsyncIo();
    TC::ConvertTaskToThread();

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

Source and project on GitHub

Tutorial 11: Where to go from here