Tutorial 7: State that flows

On this page

Where we are

Chapter 6 moves the blocks through a ring of slots: a reader, parallel encode tasks and a writer that writes in order and computes the CRC over everything it writes.

The problem

The CRC is state that flows: the CRC after block 5 depends on the CRC after block 4, and so on. The writer computes it alone, over every byte of the output. With few cores that is cheap. With many cores the encode tasks deliver blocks faster, and the writer's CRC becomes a serial stage that everything else waits for.

The obvious idea: let the encode tasks do the CRC work. Each task updates one shared CRC as soon as its block is encoded, under a lock, so that no two tasks change it at the same time.

The first attempt: a lock

CriticalSection lets one task at a time into a section of code. A task that finds the section taken is suspended, not spinning or blocking its thread, and resumed when the section is free. CriticalSectionLock enters in its constructor and leaves in its destructor.

// The first attempt: one CRC for the whole output, shared by all encode tasks
// and guarded by a critical section.
struct SharedCrc {
    TC::Synchronization::CriticalSection lock;
    std::uint32_t                        crc = 0;
};

The encode task continues the shared CRC under the lock (-v lock):

class EncodeTask : public TC::Tasks::Task
{
public:
    void Run() noexcept override
    {
        EncodeBlock(input, encoded);
        if (shared != nullptr) {
            // Wrong: continues the CRC in the order in which the blocks finish.
            TC::Synchronization::CriticalSectionLock lock(shared->lock);
            shared->crc = ContinueCrc(shared->crc, encoded);
        } else
            crc = BlockCrc(encoded); // this block's own CRC, in parallel
    }

    std::span<const std::byte> input;
    EncodedBlock               encoded;
    std::uint32_t              crc    = 0;
    SharedCrc*                 shared = nullptr; // set for -v lock
};

Run it:

blockpack input.bin output.bpk -r reference.bpk -v lock -t 1
time: … s
OK
blockpack input.bin output.bpk -r reference.bpk -v lock
time: … s
MISMATCH

With one thread the output is right; with several it is wrong. The lock did its job – no two tasks changed the CRC at the same time – but it protects against simultaneous access, not against the wrong order. A CRC over block 7 followed by block 6 is a different CRC. The blocks finish in whatever order the threads happen to run them, and with one thread that happens to be the file order. A test on one core passes; that is why every chapter compares its output with the reference, on all cores.

CriticalSection remains the right tool where the order does not matter: counters, statistics, a shared list of results.

The building block: split the state

Serial state cannot be made parallel by locking it. It can often be split into a part that does not depend on the state, which runs in parallel, and a small part that does, which runs in order.

For a CRC this works exactly: the CRC of A followed by B can be computed from the CRC of A, the CRC of B and the length of B. CombineCrc32 in tutorial-common.h does that (it is zlib's crc32_combine) in a few thousand operations, whatever the length.

In parallel: each encode task computes the CRC of its own block, as if the block were the whole file (BlockCrc). That is the else branch of Run above.
In order: the writer combines the block CRCs one after the other – one cheap step per block instead of a pass over every byte:

class Packer
{
    // ...
    bool Pack(TC::ITaskScheduler& scheduler, const Arguments& arguments,
              std::size_t slotCount)
    {
        // ...
        scheduler.AddTask(&reader_);
        std::uint32_t crc = 0;
        bool          ok  = true;
        for (std::size_t block = 0;; ++block) {
            readySlots_.Wait();
            Slot& slot = slots_[block % slotCount];
            if (slot.size == 0)
                break;
            slot.encoded.Wait();
            const EncodedBlock& encoded = slot.task.encoded;
            ok                          = ok && WriteBlockData(output, encoded);
            // The serial part shrinks to a combine per block.
            crc = CombineCrc32(crc, slot.task.crc, BlockBytes(encoded));
            freeSlots_.Release();
        }
        reader_.Wait();
        input_.Close();
        if (useLock)
            crc = sharedCrc_.crc; // every encode task has exited
        return WriteFooter(output, crc) && ok && readOk_;
    }
    // ...
};

Why not a separate CRC task between the encoding and the writer? It would still touch every byte in order. Moving serial work to another task does not shorten it; only splitting it does.

The same pattern – the bulk of the work in parallel, independent of the state, and a short step in order that carries the state – is how the case studies below deal with the state that runs through their files.

The result

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

The output is the same as chapter 1's on any number of threads. The serial work per block is now a write and a combine; the CRC itself is computed in parallel.

What limits us now

The reader is the stage that feeds everything else. When its read completes, it becomes runnable again – and waits in the same queue as dozens of encode tasks.

Learn more

Semaphores and locks – CriticalSection and its lock types
Task graphs
Case studies: ctLame, ctDeflate
CriticalSection

Show the whole program

// Tutorial chapter 7: state that flows - the CRC over the output depends on
// every block before. -v lock tries to update it from the encode tasks under a
// lock and gets it wrong; the default splits it into a CRC per block, computed
// in parallel, and a cheap combine in order.
#include "tutorial-common.h"

#include <cwchar>
#include <memory>

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

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

// The first attempt: one CRC for the whole output, shared by all encode tasks
// and guarded by a critical section.
struct SharedCrc {
    TC::Synchronization::CriticalSection lock;
    std::uint32_t                        crc = 0;
};

class EncodeTask : public TC::Tasks::Task
{
public:
    void Run() noexcept override
    {
        EncodeBlock(input, encoded);
        if (shared != nullptr) {
            // Wrong: continues the CRC in the order in which the blocks finish.
            TC::Synchronization::CriticalSectionLock lock(shared->lock);
            shared->crc = ContinueCrc(shared->crc, encoded);
        } else
            crc = BlockCrc(encoded); // this block's own CRC, in parallel
    }

    std::span<const std::byte> input;
    EncodedBlock               encoded;
    std::uint32_t              crc    = 0;
    SharedCrc*                 shared = nullptr; // set for -v lock
};

// A slot of the ring, as in chapter 6.
struct Slot {
    std::vector<std::byte>           buffer;
    std::size_t                      size = 0;
    EncodeTask                       task;
    TC::Synchronization::TaskGroup_s encoded;
};

class Packer;

class ReaderTask : public TC::Tasks::WaitableTask
{
public:
    explicit ReaderTask(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)
    {
        if (Error::Failed(input_.Create(arguments.input, GENERIC_READ,
                                        FILE_SHARE_READ, OPEN_EXISTING,
                                        FILE_FLAG_SEQUENTIAL_SCAN)))
            return false;
        File output = File::Create(arguments.output);
        if (!output.IsOpen())
            return false;

        const bool useLock = std::wcscmp(arguments.variant, L"lock") == 0;
        scheduler_         = &scheduler;
        slotCount_         = slotCount;
        slots_             = std::make_unique<Slot[]>(slotCount);
        for (std::size_t i = 0; i < slotCount; ++i) {
            slots_[i].buffer.resize(arguments.blockSize);
            slots_[i].task.shared = useLock ? &sharedCrc_ : nullptr;
        }
        freeSlots_.Release(slotCount);

        scheduler.AddTask(&reader_);
        std::uint32_t crc = 0;
        bool          ok  = true;
        for (std::size_t block = 0;; ++block) {
            readySlots_.Wait();
            Slot& slot = slots_[block % slotCount];
            if (slot.size == 0)
                break;
            slot.encoded.Wait();
            const EncodedBlock& encoded = slot.task.encoded;
            ok                          = ok && WriteBlockData(output, encoded);
            // The serial part shrinks to a combine per block.
            crc = CombineCrc32(crc, slot.task.crc, BlockBytes(encoded));
            freeSlots_.Release();
        }
        reader_.Wait();
        input_.Close();
        if (useLock)
            crc = sharedCrc_.crc; // every encode task has exited
        return WriteFooter(output, crc) && ok && readOk_;
    }

    // The reader of chapter 6.
    void Read() noexcept
    {
        for (std::size_t block = 0;; ++block) {
            freeSlots_.Wait();
            Slot&               slot   = slots_[block % slotCount_];
            const std::uint64_t offset = std::uint64_t{slot.buffer.size()} * block;
            DWORD               read   = 0; // DWORD: Read's out parameter
            const auto          error
                = input_.Read(slot.buffer.data(),
                              static_cast<DWORD>(slot.buffer.size()), read, offset);
            if (Error::Failed(error) && error != Error::EndOfFile)
                readOk_ = false;
            slot.size = read;
            if (slot.size == 0) {
                readySlots_.Release();
                return;
            }
            slot.task.input = std::span(slot.buffer).first(slot.size);
            slot.encoded.AddTask(&slot.task);
            scheduler_->AddTask(&slot.task);
            readySlots_.Release();
        }
    }

private:
    TC::ITaskScheduler*              scheduler_ = nullptr;
    AsyncFile                        input_;
    std::size_t                      slotCount_ = 0;
    std::unique_ptr<Slot[]>          slots_;
    TC::Synchronization::Semaphore_s freeSlots_;
    TC::Synchronization::Semaphore_s readySlots_;
    SharedCrc                        sharedCrc_;
    ReaderTask                       reader_{*this};
    bool                             readOk_ = true;
};

void ReaderTask::Run() noexcept
{
    packer_.Read();
}

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;

    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) {
        std::printf("cannot pack the file\n");
        return 1;
    }
    return CheckOutput(arguments) ? 0 : 1;
}

Source and project on GitHub

Tutorial 8: The critical path