Ordered output

A producer task builds a file of 2,000 records, about 62 MB, with an OrderedFileWriter. Each record header is appended at once; each payload is a deferred copy that one of three copy tasks makes later. A writer task writes the front of the file as soon as it is complete, while the producer keeps appending. At the end the producer patches the record count into the file header, closes the file and lets the copy tasks end. The program reads the file back, checks every record and deletes it.

The producer

// The producer: appends with Write (copied at once) and WriteDeferred (copied later
// by a copy task), ends each batch, patches the header and closes the file.
bool Produce(OrderedFileWriter&            writer,
             const std::vector<std::byte>& source) noexcept
{
    const FileHeader header{{'R', 'E', 'C', 'O', 'R', 'D', 'S', '\0'}, 0};
    writer.Write(&header, sizeof(header));

    for (std::uint32_t i = 0; i < Records; ++i) {
        const RecordHeader record{i, PayloadSize(i)};
        writer.Write(&record, sizeof(record));
        // Borrowed, not copied: source must stay unchanged until the copy has run.
        writer.WriteDeferred(source.data() + PayloadOffset(i), record.size);
        if ((i + 1) % RecordsPerBatch == 0)
            writer.EndBatch(); // hands the copies to the copy tasks, shows the
                               // writer the finished front
    }

    const std::uint64_t records = Records;
    writer.WriteAt(offsetof(FileHeader, records), &records, sizeof(records));
    return writer.Close(); // finishes the copies, waits for the writer task to
                           // write everything
}

Write copies the bytes into the file's memory at once; the source may be reused right after the call.
WriteDeferred only reserves room and records where the bytes come from. A copy task copies them after the next EndBatch.
EndBatch hands the recorded copies to the copy tasks and tells the writer task how far the file is complete.
WriteAt overwrites bytes appended earlier – here the record count in the header, which is only known at the end.
Close finishes all copies, waits until the writer task has written everything and applies the patches. The writer task may still be finishing when Close returns, so the writer object is destroyed only after its tasks have exited.

The file receives the bytes exactly in append order, whatever order the copy tasks finish in.

Setting it up

// The writer, its tasks and the group outlive every task: they go away
// after Stop.
OrderedFileWriter           writer;
OrderedFileWriter::CopyTask copyTasks[CopyTasks]
    = {OrderedFileWriter::CopyTask(writer),
       OrderedFileWriter::CopyTask(writer),
       OrderedFileWriter::CopyTask(writer)};
TC::Synchronization::TaskGroup_s group;

// The bound only reserves address space; the memory is committed and
// decommitted segment by segment as the file grows and gets written.
const std::uint64_t bound = sizeof(FileHeader)
    + std::uint64_t{Records} * (sizeof(RecordHeader) + MaxPayload);
const TC::Windows::ErrorCode error = writer.Open(path.c_str(), bound);

auto producer = TC::Tasks::Task::Create([&] {
    closed = Produce(writer, source);
    writer.StopCopyTasks(CopyTasks); // after Close: lets the copy tasks end
});

if (TC::Windows::Error::Succeeded(error)) {
    // Writer and copy tasks after Open, the producer last.
    group.AddTask(&writer.Writer());
    scheduler->AddTask(&writer.Writer());
    for (auto& copyTask : copyTasks) {
        group.AddTask(&copyTask);
        scheduler->AddTask(&copyTask);
    }
    group.AddTask(&producer);
    scheduler->AddTask(&producer);
    group.Wait(); // all tasks have ended
    stats = writer.Stats();
}

The protocol, in order:

Open creates the file and reserves address space for bound bytes – memory is committed only while a part of the file is being built.
Schedule the writer task (Writer) once and any number of copy tasks (CopyTask), after Open.
One producer appends with Write, WriteDeferred and EndBatch, then calls Close.
StopCopyTasks with the number of copy tasks lets them end.
Wait until all tasks have ended; only then read Stats and let the writer go.

Full program

// Ordered output: a producer task builds a file of records with an
// OrderedFileWriter. Record headers are appended at once, payloads are deferred
// copies done by copy tasks, and the writer task writes the finished front of the
// file while the producer goes on.
#define NOMINMAX
#include <Windows.h>

#include <cstddef>
#include <cstdint>
#include <cstdio>
#include <cstring>
#include <filesystem>
#include <fstream>
#include <vector>

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

namespace TC = ThinkMeta::ConcurrentTasks;
using TC::AsyncIo::OrderedFileWriter;

constexpr std::uint32_t Records         = 2000;
constexpr std::uint32_t MaxPayload      = 64 * 1024;
constexpr std::uint32_t RecordsPerBatch = 50;
constexpr std::uint32_t CopyTasks       = 3;

struct FileHeader {
    char          magic[8];
    std::uint64_t records; // patched with WriteAt when the count is known
};

struct RecordHeader {
    std::uint32_t index;
    std::uint32_t size;
};

// The payload of record i: a slice of the source buffer.
std::uint32_t PayloadSize(std::uint32_t i) noexcept
{
    return 1 + (i * 7919u) % MaxPayload;
}
std::size_t PayloadOffset(std::uint32_t i) noexcept
{
    return (static_cast<std::size_t>(i) * 4099u) % MaxPayload;
}

// The producer: appends with Write (copied at once) and WriteDeferred (copied later
// by a copy task), ends each batch, patches the header and closes the file.
bool Produce(OrderedFileWriter&            writer,
             const std::vector<std::byte>& source) noexcept
{
    const FileHeader header{{'R', 'E', 'C', 'O', 'R', 'D', 'S', '\0'}, 0};
    writer.Write(&header, sizeof(header));

    for (std::uint32_t i = 0; i < Records; ++i) {
        const RecordHeader record{i, PayloadSize(i)};
        writer.Write(&record, sizeof(record));
        // Borrowed, not copied: source must stay unchanged until the copy has run.
        writer.WriteDeferred(source.data() + PayloadOffset(i), record.size);
        if ((i + 1) % RecordsPerBatch == 0)
            writer.EndBatch(); // hands the copies to the copy tasks, shows the
                               // writer the finished front
    }

    const std::uint64_t records = Records;
    writer.WriteAt(offsetof(FileHeader, records), &records, sizeof(records));
    return writer.Close(); // finishes the copies, waits for the writer task to
                           // write everything
}

// Reads the file back and checks every record.
bool Verify(const std::filesystem::path& path, const std::vector<std::byte>& source,
            std::uint64_t expectedSize)
{
    std::error_code     error;
    const std::uint64_t size = std::filesystem::file_size(path, error);
    if (error || size != expectedSize || size < sizeof(FileHeader))
        return false;
    std::vector<char> data(static_cast<std::size_t>(size));
    std::ifstream     file(path, std::ios::binary);
    if (!file.read(data.data(), static_cast<std::streamsize>(data.size())))
        return false;

    FileHeader header;
    std::memcpy(&header, data.data(), sizeof(header));
    if (std::memcmp(header.magic, "RECORDS", 8) != 0 || header.records != Records)
        return false;

    std::size_t pos = sizeof(FileHeader);
    for (std::uint32_t i = 0; i < Records; ++i) {
        RecordHeader record;
        if (data.size() - pos < sizeof(record))
            return false;
        std::memcpy(&record, data.data() + pos, sizeof(record));
        pos += sizeof(record);
        if (record.index != i || record.size != PayloadSize(i)
            || data.size() - pos < record.size)
            return false;
        if (std::memcmp(data.data() + pos, source.data() + PayloadOffset(i),
                        record.size)
            != 0)
            return false;
        pos += record.size;
    }
    return pos == data.size();
}

int main()
{
    const std::filesystem::path path = std::filesystem::temp_directory_path()
        / L"ConcurrentTasks-ordered-output.bin";

    // The payload source: borrowed by WriteDeferred, so it outlives Close.
    std::vector<std::byte> source(2 * MaxPayload);
    for (std::size_t i = 0; i < source.size(); ++i)
        source[i] = static_cast<std::byte>(i * 131 + i / 251);

    TC::ConvertThreadToTask();
    if (!TC::AsyncIo::EnableAsyncIo())
        return 1;
    TC::SchedulerInfo   info{};
    TC::ITaskScheduler* scheduler
        = TC::CreateTaskScheduler(TC::SchedulerKind::Default, &info);
    if (scheduler == nullptr)
        return 1;
    scheduler->Start();

    bool                                closed = false;
    TC::AsyncIo::OrderedFileWriterStats stats;
    {
        // The writer, its tasks and the group outlive every task: they go away
        // after Stop.
        OrderedFileWriter           writer;
        OrderedFileWriter::CopyTask copyTasks[CopyTasks]
            = {OrderedFileWriter::CopyTask(writer),
               OrderedFileWriter::CopyTask(writer),
               OrderedFileWriter::CopyTask(writer)};
        TC::Synchronization::TaskGroup_s group;

        // The bound only reserves address space; the memory is committed and
        // decommitted segment by segment as the file grows and gets written.
        const std::uint64_t bound = sizeof(FileHeader)
            + std::uint64_t{Records} * (sizeof(RecordHeader) + MaxPayload);
        const TC::Windows::ErrorCode error = writer.Open(path.c_str(), bound);

        auto producer = TC::Tasks::Task::Create([&] {
            closed = Produce(writer, source);
            writer.StopCopyTasks(CopyTasks); // after Close: lets the copy tasks end
        });

        if (TC::Windows::Error::Succeeded(error)) {
            // Writer and copy tasks after Open, the producer last.
            group.AddTask(&writer.Writer());
            scheduler->AddTask(&writer.Writer());
            for (auto& copyTask : copyTasks) {
                group.AddTask(&copyTask);
                scheduler->AddTask(&copyTask);
            }
            group.AddTask(&producer);
            scheduler->AddTask(&producer);
            group.Wait(); // all tasks have ended
            stats = writer.Stats();
        }
        scheduler->Stop();
    }
    scheduler->Destroy();
    TC::AsyncIo::DisableAsyncIo();
    TC::ConvertTaskToThread();

    const bool ok = closed && Verify(path, source, stats.bytes);
    std::printf("%u records, %llu bytes, %llu deferred copies (%llu bytes)\n",
                Records, static_cast<unsigned long long>(stats.bytes),
                static_cast<unsigned long long>(stats.deferredCopies),
                static_cast<unsigned long long>(stats.deferredBytes));
    std::printf("producer waited %.3f s, writer waited %.3f s (%s)\n",
                stats.producerWaited, stats.writerWaited, ok ? "ok" : "MISMATCH");

    std::error_code ignored;
    std::filesystem::remove(path, ignored);
    return ok ? 0 : 1;
}

Source and project on GitHub

Build and run

A console program. With the core's props file imported and the library referenced (see Set up a project) it needs nothing else. It writes and deletes a file in %TEMP% and prints the writer's counters; the waiting times depend on the disk:

2000 records, 65348680 bytes, 2000 deferred copies (65332664 bytes)
producer waited 0.000 s, writer waited 0.000 s (ok)

What to notice

Important

A source passed to WriteDeferred is borrowed: it must stay valid and unchanged until LowestPendingSource no longer covers it, at the latest until Close returns. Here the source buffer lives in main and outlives the writer.

One producer. Write, WriteDeferred, EndBatch, WriteAt and Close are not thread-safe among themselves; one task (or a converted thread) calls them all. Some of them suspend the producer – when a ring is full or in Close – so it must be a task.
The bound is a hard limit. Open reserves the address space for bound bytes; an append beyond it fails, and Failed and Close report it. Compute it from the largest file the program can produce.
The writer outlives its tasks. The writer task and the copy tasks refer to the OrderedFileWriter. It is destroyed only after the group has seen all of them exit and the scheduler has stopped.
The writer task must run. Close waits for it, so it has to be scheduled; without StopCopyTasks the copy tasks would wait for work forever.
Batches decide the overlap. Without EndBatch the writer task sees no progress until Close. Ending a batch every few records lets copying and writing run while the producer appends.
Asynchronous I/O must be on. EnableAsyncIo before Open, DisableAsyncIo after all tasks have ended.