Asynchronous files

AsyncFile is a file opened for overlapped I/O on the core's I/O completion port. A task that reads or writes is suspended until the operation has completed, so its thread runs other tasks while the disk works. AsyncFile offers three levels: waiting calls for one operation (Read, Write), starting calls that leave the waiting to you (ReadAsync, WriteAsync), and whole-file transfers into or out of sequential memory (ReadFile, WriteFile). The examples are small functions from guide-async-files.cpp, a complete program that uses them on a file in %TEMP% and checks the results.

Setup

if (!TC::AsyncIo::EnableAsyncIo() || !TC::EnableTimers())
    return 1;

EnableAsyncIo starts the core's completion port and the I/O thread that drains it. Calls are counted: only the first starts them, so a library and the application can both enable asynchronous I/O. Timers are needed only for the timeouts below.

Create passes its parameters to CreateFileW unchanged, adds FILE_FLAG_OVERLAPPED and associates the handle with the completion port:

AsyncFile file;
if (Error::Failed(file.Create(path_, GENERIC_READ | GENERIC_WRITE, 0,
                              CREATE_ALWAYS, FILE_ATTRIBUTE_NORMAL)))
    return false;

After any error no file is open. The object owns the handle and closes it on destruction.

Important

Call EnableAsyncIo before Create. Close or destroy the file only when no operation on it is pending, and call the last DisableAsyncIo only when no operation of the process is pending any more.

Waiting reads and writes

Read and Write look like ordinary blocking calls, but they suspend only the calling task until the operation has completed. The handle has no file pointer, so every call carries its file offset (here 0):

// Writes data at the start of the file and reads it back. Each call suspends the
// calling task until the operation has completed; its thread runs other tasks
// meanwhile.
static bool WriteAndReadBack(AsyncFile& file, const std::vector<std::byte>& data,
                             std::vector<std::byte>& copy) noexcept
{
    DWORD written = 0;
    if (Error::Failed(
            file.Write(data.data(), static_cast<DWORD>(data.size()), written, 0)))
        return false;

    DWORD read = 0;
    if (Error::Failed(
            file.Read(copy.data(), static_cast<DWORD>(copy.size()), read, 0)))
        return false;
    return read == copy.size();
}

The number of bytes read may be smaller than requested at the end of the file; it is 0 on error.
When the call returns, the operation is over and the buffer may be reused.

An optional last parameter limits the wait:

// The same read, but giving up after five seconds (needs EnableTimers).
static bool ReadWithTimeout(AsyncFile&              file,
                            std::vector<std::byte>& buffer) noexcept
{
    DWORD      read  = 0;
    const auto error = file.Read(buffer.data(), static_cast<DWORD>(buffer.size()),
                                 read, 0, TC::Duration::Seconds(5));
    if (error == ERROR_TIMEOUT)
        std::printf("the disk did not answer in time\n");
    return Error::Succeeded(error);
}

An operation still pending after the timeout is cancelled with CancelIoEx and the call returns ERROR_TIMEOUT, unless the operation completed meanwhile; the call still waits until the cancellation is through. A timeout needs EnableTimers; without it the wait has no limit. The default is Duration::Infinite.

These calls suspend the task, so call them from a task or from a thread after ConvertThreadToTask.

Several operations in flight

A single waiting call keeps one request in flight. To keep the disk busy, start several operations and wait for them afterwards. ReadAsync and WriteAsync never suspend; each takes a request that carries the offset, and after completion the error code and the transferred bytes. WaitableAsyncFileIoRequest_s adds an event that a task can wait on:

// Reads the first four blocks with four reads in flight at once: start them all,
// then wait.
static bool ReadFourBlocks(AsyncFile& file,
                           std::vector<std::byte> (&buffers)[4]) noexcept
{
    TC::AsyncIo::WaitableAsyncFileIoRequest_s requests[4];

    int started = 0;
    for (; started < 4; ++started) {
        requests[started].SetOffset(std::uint64_t{BlockSize} * started);
        const auto error
            = file.ReadAsync(buffers[started].data(), BlockSize, requests[started]);
        if (Error::Failed(error) && !Error::IsPending(error))
            break; // not started: never wait for this request
    }

    // Every request that was started must complete before it goes out of scope.
    bool ok = started == 4;
    for (int i = 0; i < started; ++i) {
        requests[i].Wait(); // returns at once if the read has completed already
        ok = ok && Error::Succeeded(requests[i].GetError());
    }
    return ok;
}

The return value of the starting call decides what happens next:

Result	Meaning
`ERROR_SUCCESS`	Completed at once; the request is already completed (on the calling thread).
`ERROR_IO_PENDING`	In flight; the request completes later on the core's I/O thread.
any other code	Not started; the request will never complete. Do not wait for it.

After completion, GetError returns the operation's Win32 error code and GetTransferredBytes the bytes transferred. Set the offset with SetOffset before the call; to reuse a request, call Reset first (it zeroes the offset), then SetOffset.

Important

The request and the buffer belong to you and must stay valid and untouched until the request has completed: the kernel writes into the request's OVERLAPPED and the I/O thread reports the result to it. Never destroy a request while its operation is pending. Once Wait has returned, the request may be destroyed at once, so requests can live in a task's stack frame.

The _s request allows one waiting task and a Wait(Duration timeout). A timeout of that wait does not cancel the operation: it stays pending and the request must stay alive. Cancel it with CancelIoEx on the file handle and wait again, as Read does internally. WaitableAsyncFileIoRequest allows any number of waiting tasks and no timeout.

Error codes

Every AsyncFile function returns a Win32 error code, ERROR_SUCCESS (0) on success; nothing throws. Error names the common codes and has the predicates Succeeded, Failed and IsPending. A read that starts at or beyond the end of the file fails with ERROR_HANDLE_EOF (Error::EndOfFile) instead of returning 0 bytes:

// Reading at or beyond the end of the file is an error, not an empty read.
static bool IsEndOfFile(AsyncFile& file, std::uint64_t offset) noexcept
{
    std::byte buffer[512];
    DWORD     read = 0;
    return file.Read(buffer, sizeof(buffer), read, offset)
        == Error::EndOfFile; // ERROR_HANDLE_EOF
}

ERROR_INVALID_USER_BUFFER and ERROR_NOT_ENOUGH_MEMORY from a starting call mean that too many overlapped requests are outstanding; Error::TooManyAsynchronousRequests tests for both. Retry later.

Whole files

ReadFile opens a file, reads it from BulkIoOptions::startOffset to the end into an IWriteSequentialMemory with several reads in flight, and closes it. WriteFile does the reverse from an IReadSequentialMemory, always from offset 0. Each finished block goes to the memory at once and in file order, so a consumer can start on the first blocks while the rest is on its way. Here the memory is a SpscVirtualMemoryPipe, and a second task takes the data out as it arrives:

// Takes the data out of the pipe as it arrives; returns the number of bytes.
static std::uint64_t Consume(TC::Memory::SpscVirtualMemoryPipe& pipe) noexcept
{
    std::uint64_t total = 0;
    for (;;) {
        std::size_t      size = 128 * 1024;
        const std::byte* data
            = pipe.RequestReadMemory(size); // suspends until data or the end
        if (data == nullptr)
            return total;
        total += size; // ... process data[0 .. size) here ...
        pipe.ReleaseReadMemory(size);
    }
}

// Read the whole file into a pipe with up to 8 reads of 256 KB in flight,
// while a second task consumes the data as it arrives.
TC::Memory::SpscVirtualMemoryPipe pipe;
std::uint64_t                     bytes    = 0;
auto                              consumer = TC::Tasks::Task::Create(
    [&pipe, &bytes]() noexcept { bytes = Consume(pipe); });
helpers_.AddTask(&consumer);
consumer.Schedule();

TC::AsyncIo::BulkIoOptions options;
options.blockSize      = 256 * 1024;
options.blocksInFlight = 8;
const auto error       = AsyncFile::ReadFile(
    path_, GENERIC_READ, FILE_SHARE_READ, OPEN_EXISTING,
    FILE_FLAG_SEQUENTIAL_SCAN, &pipe, options);
helpers_.Wait(); // the consumer has taken everything

BulkIoOptions sets the shape of the I/O. What counts is the amount in flight, blockSize × blocksInFlight: below about 1 MB a fast disk waits for requests. The default is 5 blocks of 1 MB. Out-of-range values are corrected: a blockSize of 0 means the default, and blocksInFlight is clamped to 1 to 64.

Both functions suspend the calling task while they wait, return the first error, and return only when no request is in flight any more, also after an error. ReadFile always ends the stream with SetEndOfStream, also on errors, so a consumer never waits forever. WriteFile does not truncate the file: with OPEN_EXISTING or OPEN_ALWAYS, older data beyond the written size stays. See Pipes for the memory side.

Shutting down

TC::DisableTimers();
TC::AsyncIo::DisableAsyncIo(); // no operation may be pending any more

The program prints:

written and read back: ok
read past the end:     ERROR_HANDLE_EOF
four reads in flight:  ok
whole file via pipe:   262144 bytes
OK