Streaming large files

Two library classes process files that are larger than you want to hold in memory. StreamedFileReader reads a file ahead into one reserved address range, so consumers address every byte by its file position, like a mapped file, while only a window of the file is committed. OrderedFileWriter builds an output file front to back in memory, lets copy tasks fill in parts of it and has a writer task write the finished front with overlapped writes. Both use AsyncFile underneath and need EnableAsyncIo. The examples come from guide-streaming-files.cpp, which exports items into a file of records with OrderedFileWriter, reads it back with StreamedFileReader and checks the records.

Writing with OrderedFileWriter

The example file

The file starts with a header and holds one record per exported item: its id followed by its payload. The payloads stay in buffers that belong to the items:

// The file: a header, then one record per item, its 8-byte id followed by its
// payload.
struct Header {
    std::uint64_t magic;
    std::uint64_t count;
};
constexpr std::uint64_t Magic       = 0x5345525547544321; // any marker
constexpr std::size_t   PayloadSize = 60'000;
constexpr std::size_t   RecordSize  = sizeof(std::uint64_t) + PayloadSize;

// An item to export. Its payload belongs to someone else; nullptr: nothing to
// export.
struct Item {
    std::uint64_t    id;
    const std::byte* payload; // PayloadSize bytes
};

Setup

The writer works with two kinds of tasks: one writer task (Writer) and any number of CopyTask objects:

OrderedFileWriter                      writer;
OrderedFileWriter::CopyTask            copy1(writer);
OrderedFileWriter::CopyTask            copy2(writer);
TC::AsyncIo::OrderedFileWriterSettings settings;
settings.segmentSize = 1 << 20; // write and decommit in 1 MB steps

const std::uint64_t bound = sizeof(Header)
    + items.size() * RecordSize; // the largest size the file may reach
if (TC::Windows::Error::Failed(writer.Open(path, bound, settings)))
    return 1;

TC::Tasks::Task* writerTasks[] = {&writer.Writer(), &copy1, &copy2};
for (TC::Tasks::Task* task : writerTasks) {
    tasks.AddTask(task);
    scheduler->AddTask(task);
}

Open creates or truncates the file (CREATE_ALWAYS) and reserves address space for bound bytes, the largest size the file may reach; memory is committed only as the producer appends. Call it once, before any append, from the producer. Schedule the writer task once (before or after Open) and the copy tasks after Open. With no copy task at all the producer runs the copies itself when a ring is full and in Close.

OrderedFileWriterSettings sets the segment size (the unit of writing, committing and decommitting, a multiple of 64 KB, default 4 MB), the writes in flight (default 4, at most 16) and the sizes of the copy queues. Out-of-range values are corrected, not rejected.

Producing

One producer, a task or a converted thread, appends the file in order:

// The single producer appends the file in order. The payloads are not copied here:
// copy tasks copy them into place later, so they must stay unchanged until Close
// has returned.
bool WriteItems(OrderedFileWriter& writer, std::span<const Item> items) noexcept
{
    Header header{}; // the count is known only at the end: patched below
    writer.Write(&header, sizeof(header));

    std::uint64_t count = 0;
    for (const Item& item : items) {
        if (item.payload == nullptr)
            continue;                            // nothing to export
        writer.Write(&item.id, sizeof(item.id)); // copied at once
        writer.WriteDeferred(item.payload,
                             PayloadSize); // copied later by a copy task
        if (++count % 16 == 0)
            writer
                .EndBatch(); // hand the copies to the copy tasks, publish progress
    }

    header = {Magic, count};
    writer.WriteAt(0, &header,
                   sizeof(header)); // in memory, or a patch if already written
    return writer.Close();          // finishes copies and writes, applies the patch
}

Once the producer has closed the file, let the copy tasks end and wait for all tasks:

const bool closed = WriteItems(
    writer, items);      // main is the producer (a converted thread may be one)
writer.StopCopyTasks(2); // two copy tasks were scheduled
tasks.Wait();            // writer task and copy tasks have exited

• Write copies the bytes at once; the data may be reused after the call. nullptr appends zeros. It never suspends.
• WriteDeferred appends room for bytes that a copy task copies from source later. The bytes land at the position they were appended at, whatever order the copies finish in.
• EndBatch hands the copies since the last batch to the copy tasks and tells the writer task how far the file is complete. Without EndBatch the writer sees no progress until Close.
• WriteAt overwrites bytes appended earlier, here the header once the record count is known; items without a payload are skipped, so it is not known before. If the bytes have not been issued to the file yet, it patches them in memory; otherwise Close applies the patch after the last write. Either way the file ends up with the patched bytes.
• Position is the file offset of the next append.

The writer task writes only the prefix in which every deferred copy has finished, segment by segment in rising offset order, with several writes overlapping, and decommits each written segment. Memory use therefore follows the distance between the producer and the finished front, not the file size.

Errors and statistics

Open returns a Win32 error code. Write, WriteDeferred and WriteAt report nothing: a failed append (address range exhausted, commit failed, WriteAt beyond Position) sets Failed, after which appends are dropped. A failed or short write stops further writes. Close returns false for any of these; the file is not deleted. Stats returns counters such as the file size and the number of deferred copies once Close has returned.

Reading with StreamedFileReader

Setup

StreamedFileReader                      reader;
TC::AsyncIo::StreamedFileReaderSettings readSettings;
readSettings.segmentSize = 1 << 20; // read, commit and decommit in 1 MB steps
readSettings.readAhead   = 4
    << 20; // read at most 4 MB past the oldest byte in use
readSettings.readsInFlight = 4;

if (TC::Windows::Error::Failed(reader.Open(path, readSettings, 2))) // two users
    return 1;
tasks.AddTask(&reader);
scheduler->AddTask(&reader); // the reader is a task itself

Open opens the file (OPEN_EXISTING, shared for reading) and reserves address space for all of it; nothing is read yet. The reader object is itself the reader task: schedule it once after Open. It ends at the end of the file, or after Stop once its reads in flight are done. StreamedFileReaderSettings sets the segment size (default 4 MB), the read-ahead (default 256 MB) and the reads in flight (default 4, at most 16); out-of-range values are corrected.

Users

A user is a slot that reports how far it is done with the file. Open takes the number of slots (1 to MaxUsers, 8). Consumers follow one protocol:

• Wait(end) before touching bytes below end. It returns at once if they are loaded; otherwise it asks the reader to read up to end, even beyond the read-ahead, and suspends the calling task. Any number of tasks may wait at once. It returns how far the file is loaded, which is less than end only after Stop.
• Release(user, below) when the user is done with the bytes before below. Whole segments below the lowest position of all slots are decommitted, which lets the reader read further ahead. ~0 releases everything. Release never suspends.

Data is byte 0 of the file; a byte is valid between a Wait that covers it and its release. Here two users share the file: user 1 reads the header, user 0 walks the records and releases each one when it is done with it:

// User 1 reads the header and releases everything at once.
Header ReadHeader(StreamedFileReader& reader) noexcept
{
    Header header{};
    if (reader.Wait(sizeof(Header)) >= sizeof(Header)) // less only after Stop
        std::memcpy(&header, reader.Data(), sizeof(header));
    reader.Release(1, ~std::uint64_t{0}); // done with everything
    return header;
}

// User 0 walks the records: Wait before touching bytes, Release when done with
// them.
void ReadRecords(StreamedFileReader& reader, std::uint64_t count) noexcept
{
    const std::byte* data = reader.Data(); // byte 0 of the file
    std::uint64_t    pos  = sizeof(Header);
    for (std::uint64_t i = 0; i < count; ++i, pos += RecordSize) {
        if (reader.Wait(pos + RecordSize) < pos + RecordSize)
            break; // only after Stop
        std::uint64_t id;
        std::memcpy(&id, data + pos, sizeof(id));
        Process(id, data + pos + sizeof(id));
        reader.Release(0, pos + RecordSize); // user 0 is done below this position
    }
    reader.Release(0, ~std::uint64_t{0}); // done with the file
}

Each user runs in a task of its own. The record walk needs the count from the header, so the header task starts it:

// The record walk starts only after user 1 has released: until every user has
// released, nothing can be decommitted.
Header header{};
auto   walkRecords = TC::Tasks::Task::Create([&reader, &header]() noexcept {
    if (header.magic == Magic)
        ReadRecords(reader, header.count);
    else
        reader.Release(
            0, ~std::uint64_t{0}); // not our file: release it all the same
});
auto   readHeader
    = TC::Tasks::Task::Create([&reader, &header, &walkRecords]() noexcept {
          header = ReadHeader(reader);
          walkRecords.Schedule();
      });

tasks.AddTask(&walkRecords); // both users count in before the first one starts
tasks.AddTask(&readHeader);
scheduler->AddTask(&readHeader);
tasks.Wait(); // both users and the reader task have exited

How much memory

The reader reads at most readAhead past the lowest released position, so with one user that keeps releasing, the committed window stays at about readAhead plus one segment, however large the file is. PeakWindow reports the largest window of a run; the example checks it against that bound.

Two things widen the window: every slot starts at position 0, so nothing is decommitted until each slot in use has released; and a Wait beyond the read-ahead is still served. A fast user that waits far ahead while a slow user still holds the front makes the reader commit everything in between. That is why the example starts the record walk only after user 1 has released, and why every user releases everything when it is done, also when it stops early. Give each slot to a consumer that releases steadily, and keep consumers at similar positions.

The program prints:

written:  392 records, close ok
read:     392 records, bad 0, read error 0
memory:   peak window within read-ahead + 1 segment: yes
OK

When to use which

• One pass over a file by one consumer: AsyncFile::ReadFile into a pipe is simpler.
• Random access within a moving window, several consumers, or data that must stay addressable by file position: StreamedFileReader.
• Output assembled from many pieces, some of which are copied from buffers that other stages still own, with patches to earlier bytes: OrderedFileWriter.

See also

• StreamedFileReader, StreamedFileReaderSettings, OrderedFileWriter, OrderedFileWriterSettings, OrderedFileWriterStats, AsyncFile
• Asynchronous I/O, Lifetimes and sizing
• Asynchronous files, Task graphs
• Samples: Ordered output, Hashing