Tutorial 9: Errors and shutdown

On this page

Where we are

Chapter 8 runs the reader at High priority. The pipeline is complete: reader, parallel encoding with a CRC per block, writer in order.

The problem

Until now the program assumed that writing works. Suppose it fails at block 10 – a full disk, a removed drive. What should the writer do?

Simply return? The reader may be suspended in freeSlots_.Wait(), waiting for a slot that the writer will never free. main would then wait for the reader forever.
Stop the scheduler? Encode tasks may still be running in their slots, and the reader may be waiting. Stop is only allowed when no task runs or waits any more, and the Packer – slots, groups, semaphores – must not be destroyed while a task can still use it.

The same applies to Ctrl+C: the user wants the program to end soon, not to leave half a file and not to crash.

The building block: the shutdown contract

No new type in this chapter. What is new is the rule that every pipeline must follow:

Important

Contract. Shut a pipeline down the way it runs: stop the source, let everything in flight drain, and destroy the objects only after the last task that can touch them has exited. Nothing checks this in a release build; Debug builds assert some of these rules. See Lifetimes and sizing and Debugging.

For blockpack that means:

A stop request that the reader looks at before each read. When it is set, the reader passes on the end mark instead of reading – the same empty slot that ends a normal run.
The writer drains. After an error it stops writing but goes on taking slots: it waits for each slot's encode task and frees the slot, so the reader is never stuck, until the end mark arrives.
The order of the shutdown stays as before: Pack returns only when the reader and every encode task have finished; then Stop, then the Packer is destroyed.

Why a plain std::atomic<bool> and not a ManualResetEvent? An event is for a signal that tasks wait for: they are suspended until it is set. Here nobody waits for the stop request; the reader only looks at it between two reads. A flag is enough and costs nothing. Use the event when a task has nothing else to do until the signal comes.

The change

The stop request is a global flag – global, because Ctrl+C can come at any moment, also before the Packer exists or after it is gone:

// Set to ask the pipeline to stop: the reader reads no further block. A plain
// flag, because nobody waits for it - the reader only looks at it before each
// read. Global, because Ctrl+C may set it at any time, even when no packer
// exists.
std::atomic<bool> stopRequested{false};

The reader passes on the end mark when a stop is requested:

class Packer
{
    // ...
    void Read() noexcept
    {
        for (std::size_t block = 0;; ++block) {
            freeSlots_.Wait();
            Slot& slot = slots_[block % slotCount_];
            DWORD read = 0; // DWORD: Read's out parameter
            if (!stopRequested) {
                const std::uint64_t offset
                    = std::uint64_t{slot.buffer.size()} * block;
                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) { // the end of the file, an error or a stop
                readySlots_.Release();
                return;
            }
            slot.task.input = std::span(slot.buffer).first(slot.size);
            slot.encoded.AddTask(&slot.task);
            scheduler_->AddTask(&slot.task, TC::TaskPriority::Normal);
            readySlots_.Release();
        }
    }
    // ...
};

The writer stops writing after an error, requests the stop and drains the ring. -v fail-write simulates a write error at block 10:

class Packer
{
    // ...
    bool Pack(TC::ITaskScheduler& scheduler, const Arguments& arguments,
              std::size_t slotCount)
    {
        // ...
        scheduler.AddTask(&reader_, TC::TaskPriority::High);
        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; // the end mark: the reader has finished
            // Wait for the encoding even after an error: the task must have
            // exited before its slot is reused or destroyed.
            slot.encoded.Wait();
            if (ok) {
                const EncodedBlock& encoded = slot.task.encoded;
                ok = !(failWrite && block == 10) && WriteBlockData(output, encoded);
                crc = CombineCrc32(crc, slot.task.crc, BlockBytes(encoded));
                if (!ok)
                    stopRequested
                        = true; // no more reading; drain what is in flight
            }
            freeSlots_.Release(); // also after an error: the reader may wait
        }
        reader_.Wait();
        input_.Close();
        const bool stopped = stopRequested;
        return WriteFooter(output, crc) && ok && readOk_ && !stopped;
    }
    // ...
};

Ctrl+C runs a handler on a thread that Windows creates – neither a task nor a scheduler thread. It does nothing but set the flag:

// Ctrl+C runs this on a thread that Windows creates - no task, no scheduler
// thread. It only sets the flag; the pipeline then shuts down as after an
// error.
BOOL WINAPI OnConsoleCtrl(DWORD ctrlType) noexcept
{
    if (ctrlType != CTRL_C_EVENT)
        return FALSE;
    stopRequested = true;
    return TRUE;
}

main installs the handler and deletes the output when the pack failed:

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());
        // Pack returns only after the reader and every encode task have
        // finished, so stopping the scheduler now is safe.
        scheduler->Stop();
    }

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

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

The result

A normal run is unchanged:

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

With a write error, or after Ctrl+C, the program ends within a few blocks, leaves no output file and returns an error:

blockpack input.bin output.bpk -v fail-write
time: … s
cannot pack the file

What limits us now

Nothing in the program's behaviour – but a lot in its code. The reader with its ring, semaphores, end mark and error handling, and the writer with its ordering and draining, are written by hand. ConcurrentTasks has both stages ready-made.

Learn more

Show the whole program

// Tutorial chapter 9: errors and shutdown - a write error or Ctrl+C stops the
// pipeline without leaving a task waiting or an object destroyed too early.
// -v fail-write simulates a write error at block 10.
#include "tutorial-common.h"

#include <atomic>
#include <cwchar>
#include <memory>

#include <ThinkMeta.ConcurrentTasks/Core.h>
import ThinkMeta.ConcurrentTasks.AsyncIo.AsyncFile;
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;

// Set to ask the pipeline to stop: the reader reads no further block. A plain
// flag, because nobody waits for it - the reader only looks at it before each
// read. Global, because Ctrl+C may set it at any time, even when no packer
// exists.
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;
};

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 failWrite = std::wcscmp(arguments.variant, L"fail-write") == 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);
        freeSlots_.Release(slotCount);

        scheduler.AddTask(&reader_, TC::TaskPriority::High);
        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; // the end mark: the reader has finished
            // Wait for the encoding even after an error: the task must have
            // exited before its slot is reused or destroyed.
            slot.encoded.Wait();
            if (ok) {
                const EncodedBlock& encoded = slot.task.encoded;
                ok = !(failWrite && block == 10) && WriteBlockData(output, encoded);
                crc = CombineCrc32(crc, slot.task.crc, BlockBytes(encoded));
                if (!ok)
                    stopRequested
                        = true; // no more reading; drain what is in flight
            }
            freeSlots_.Release(); // also after an error: the reader may wait
        }
        reader_.Wait();
        input_.Close();
        const bool stopped = stopRequested;
        return WriteFooter(output, crc) && ok && readOk_ && !stopped;
    }

    void Read() noexcept
    {
        for (std::size_t block = 0;; ++block) {
            freeSlots_.Wait();
            Slot& slot = slots_[block % slotCount_];
            DWORD read = 0; // DWORD: Read's out parameter
            if (!stopRequested) {
                const std::uint64_t offset
                    = std::uint64_t{slot.buffer.size()} * block;
                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) { // the end of the file, an error or a stop
                readySlots_.Release();
                return;
            }
            slot.task.input = std::span(slot.buffer).first(slot.size);
            slot.encoded.AddTask(&slot.task);
            scheduler_->AddTask(&slot.task, TC::TaskPriority::Normal);
            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_;
    ReaderTask                       reader_{*this};
    bool                             readOk_ = true;
};

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

// Ctrl+C runs this on a thread that Windows creates - no task, no scheduler
// thread. It only sets the flag; the pipeline then shuts down as after an
// error.
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());
        // Pack returns only after the reader and every encode task have
        // finished, so stopping the scheduler now is safe.
        scheduler->Stop();
    }

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

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

Source and project on GitHub

Tutorial 10: Ready-made stages