Async file copy

A task creates an 8 MB source file, copies it and checks both files byte for byte; then the program deletes them. The copy uses two buffers: while one block is being written, the next one is already being read. Write and Read suspend the task until the operation has completed; the copy starts its writes with WriteAsync instead, each with a WaitableAsyncFileIoRequest_s, so that the task can read the next block while the write is in flight.

Writing with explicit offsets

// Writes the source file block by block. Write suspends the calling task until the
// write has completed; every call names its file offset, since the handle has no
// file pointer.
ErrorCode CreateSource(const wchar_t* path) noexcept
{
    AsyncFile file;
    ErrorCode error
        = file.Create(path, GENERIC_WRITE, 0, CREATE_ALWAYS, FILE_ATTRIBUTE_NORMAL);
    std::vector<std::byte> block(BlockSize);
    for (std::uint64_t offset = 0; Error::Succeeded(error) && offset < FileSize;
         offset += BlockSize) {
        const auto bytes = static_cast<DWORD>(
            std::min<std::uint64_t>(BlockSize, FileSize - offset));
        for (DWORD i = 0; i < bytes; ++i)
            block[i] = Pattern(offset + i);
        DWORD written = 0;
        error         = file.Write(block.data(), bytes, written, offset);
        if (Error::Succeeded(error) && written != bytes)
            error = ERROR_WRITE_FAULT;
    }
    return error;
}

An AsyncFile has no file pointer: every operation names its offset. Write suspends only the calling task; its thread runs other tasks until the write has completed.

Overlapping reads and writes

// Two buffers: while the block in one buffer is being written, the next block is
// read into the other one.
ErrorCode Copy(const wchar_t* sourcePath, const wchar_t* targetPath) noexcept
{
    AsyncFile source;
    AsyncFile target;
    ErrorCode error = source.Create(sourcePath, GENERIC_READ, FILE_SHARE_READ,
                                    OPEN_EXISTING, FILE_FLAG_SEQUENTIAL_SCAN);
    if (Error::Succeeded(error))
        error = target.Create(targetPath, GENERIC_WRITE, 0, CREATE_ALWAYS,
                              FILE_ATTRIBUTE_NORMAL);
    AsyncFile::FileSize size = 0;
    if (Error::Succeeded(error))
        error = source.GetSize(size);
    if (Error::Failed(error))
        return error;

    std::vector<std::byte> buffers[2]
        = {std::vector<std::byte>(BlockSize), std::vector<std::byte>(BlockSize)};
    WaitableAsyncFileIoRequest_s writes[2];
    bool writing[2] = {false, false}; // started and not yet waited for

    for (std::uint64_t offset = 0, block = 0; offset < size;
         offset += BlockSize, ++block) {
        const std::size_t i = block % 2;
        const auto        bytes
            = static_cast<DWORD>(std::min<std::uint64_t>(BlockSize, size - offset));

        // The buffer is free once the write from it has completed.
        if (writing[i]) {
            writes[i].Wait();
            writing[i] = false;
            error      = writes[i].GetError();
            if (Error::Failed(error))
                break;
        }

        // Suspends this task until the block is read; the write from the other
        // buffer goes on meanwhile.
        DWORD read = 0;
        error      = source.Read(buffers[i].data(), bytes, read, offset);
        if (Error::Succeeded(error) && read != bytes)
            error = ERROR_HANDLE_EOF;
        if (Error::Failed(error))
            break;

        writes[i].Reset(); // also clears the offset
        writes[i].SetOffset(offset);
        const ErrorCode started
            = target.WriteAsync(buffers[i].data(), bytes, writes[i]);
        if (Error::Failed(started) && !Error::IsPending(started)) {
            error = started; // not started: never wait for this request
            break;
        }
        writing[i] = true;
    }

    // No write may be pending when its buffer, its request or its file goes away -
    // also after an error.
    for (std::size_t i = 0; i < 2; ++i) {
        if (writing[i]) {
            writes[i].Wait();
            if (Error::Succeeded(error))
                error = writes[i].GetError();
        }
    }
    return error;
}

Each buffer has its own write request. WriteAsync returns ERROR_SUCCESS when the write completed at once and ERROR_IO_PENDING when it is in flight; in both cases the request completes and its Wait returns. Any other result means the write never started – waiting for it would never end. Before a buffer is reused for the next read, the write from it must have completed. ReadAsync works the same way for reads.

The task and main

// The file operations suspend their caller, so they run in a task. WaitableTask_s
// lets main wait for it.
class CopyTask : public TC::Tasks::WaitableTask_s
{
public:
    CopyTask(const wchar_t* source, const wchar_t* target) noexcept
        : source_(source),
          target_(target)
    {}

    void Run() noexcept override
    {
        error = CreateSource(source_);
        if (Error::Succeeded(error))
            error = Copy(source_, target_);
        verified = Error::Succeeded(error) && Verify(source_) && Verify(target_);
    }

    ErrorCode error    = Error::Success;
    bool      verified = false;

private:
    const wchar_t* source_;
    const wchar_t* target_;
};

// Before the first AsyncFile is created: it associates its handle with the
// core's completion port.
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();

CopyTask task(source.c_str(), target.c_str());
scheduler->AddTask(&task);
task.Wait();
scheduler->Stop();
scheduler->Destroy();

TC::AsyncIo::DisableAsyncIo(); // no operation is pending any more: every file
                               // is closed

Full program

// Async file copy: a task copies a file with overlapped reads and writes on the
// core's I/O completion port - the read of the next block runs while the current
// block is being written.
#define NOMINMAX
#include <Windows.h>

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

#include <ThinkMeta.ConcurrentTasks/Core.h>
import ThinkMeta.ConcurrentTasks.AsyncIo.AsyncFile;
import ThinkMeta.ConcurrentTasks.Tasks.WaitableTask;
import ThinkMeta.ConcurrentTasks.Windows.Error;

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

constexpr std::uint64_t FileSize
    = 8 * 1024 * 1024 + 123; // not a multiple of the block size
constexpr DWORD BlockSize = 1024 * 1024;

std::byte Pattern(std::uint64_t offset) noexcept
{
    return static_cast<std::byte>((offset * 7 + offset / 4096) & 0xFF);
}

// Writes the source file block by block. Write suspends the calling task until the
// write has completed; every call names its file offset, since the handle has no
// file pointer.
ErrorCode CreateSource(const wchar_t* path) noexcept
{
    AsyncFile file;
    ErrorCode error
        = file.Create(path, GENERIC_WRITE, 0, CREATE_ALWAYS, FILE_ATTRIBUTE_NORMAL);
    std::vector<std::byte> block(BlockSize);
    for (std::uint64_t offset = 0; Error::Succeeded(error) && offset < FileSize;
         offset += BlockSize) {
        const auto bytes = static_cast<DWORD>(
            std::min<std::uint64_t>(BlockSize, FileSize - offset));
        for (DWORD i = 0; i < bytes; ++i)
            block[i] = Pattern(offset + i);
        DWORD written = 0;
        error         = file.Write(block.data(), bytes, written, offset);
        if (Error::Succeeded(error) && written != bytes)
            error = ERROR_WRITE_FAULT;
    }
    return error;
}

// Two buffers: while the block in one buffer is being written, the next block is
// read into the other one.
ErrorCode Copy(const wchar_t* sourcePath, const wchar_t* targetPath) noexcept
{
    AsyncFile source;
    AsyncFile target;
    ErrorCode error = source.Create(sourcePath, GENERIC_READ, FILE_SHARE_READ,
                                    OPEN_EXISTING, FILE_FLAG_SEQUENTIAL_SCAN);
    if (Error::Succeeded(error))
        error = target.Create(targetPath, GENERIC_WRITE, 0, CREATE_ALWAYS,
                              FILE_ATTRIBUTE_NORMAL);
    AsyncFile::FileSize size = 0;
    if (Error::Succeeded(error))
        error = source.GetSize(size);
    if (Error::Failed(error))
        return error;

    std::vector<std::byte> buffers[2]
        = {std::vector<std::byte>(BlockSize), std::vector<std::byte>(BlockSize)};
    WaitableAsyncFileIoRequest_s writes[2];
    bool writing[2] = {false, false}; // started and not yet waited for

    for (std::uint64_t offset = 0, block = 0; offset < size;
         offset += BlockSize, ++block) {
        const std::size_t i = block % 2;
        const auto        bytes
            = static_cast<DWORD>(std::min<std::uint64_t>(BlockSize, size - offset));

        // The buffer is free once the write from it has completed.
        if (writing[i]) {
            writes[i].Wait();
            writing[i] = false;
            error      = writes[i].GetError();
            if (Error::Failed(error))
                break;
        }

        // Suspends this task until the block is read; the write from the other
        // buffer goes on meanwhile.
        DWORD read = 0;
        error      = source.Read(buffers[i].data(), bytes, read, offset);
        if (Error::Succeeded(error) && read != bytes)
            error = ERROR_HANDLE_EOF;
        if (Error::Failed(error))
            break;

        writes[i].Reset(); // also clears the offset
        writes[i].SetOffset(offset);
        const ErrorCode started
            = target.WriteAsync(buffers[i].data(), bytes, writes[i]);
        if (Error::Failed(started) && !Error::IsPending(started)) {
            error = started; // not started: never wait for this request
            break;
        }
        writing[i] = true;
    }

    // No write may be pending when its buffer, its request or its file goes away -
    // also after an error.
    for (std::size_t i = 0; i < 2; ++i) {
        if (writing[i]) {
            writes[i].Wait();
            if (Error::Succeeded(error))
                error = writes[i].GetError();
        }
    }
    return error;
}

// Reads both files with Read (suspending, explicit offsets) and compares them with
// the pattern.
bool Verify(const wchar_t* path) noexcept
{
    AsyncFile           file;
    AsyncFile::FileSize size = 0;
    if (Error::Failed(file.Create(path, GENERIC_READ, FILE_SHARE_READ,
                                  OPEN_EXISTING, FILE_ATTRIBUTE_NORMAL))
        || Error::Failed(file.GetSize(size)) || size != FileSize)
        return false;
    std::vector<std::byte> block(BlockSize);
    for (std::uint64_t offset = 0; offset < size; offset += BlockSize) {
        const auto bytes
            = static_cast<DWORD>(std::min<std::uint64_t>(BlockSize, size - offset));
        DWORD read = 0;
        if (Error::Failed(file.Read(block.data(), bytes, read, offset))
            || read != bytes)
            return false;
        for (DWORD i = 0; i < bytes; ++i) {
            if (block[i] != Pattern(offset + i))
                return false;
        }
    }
    return true;
}

// The file operations suspend their caller, so they run in a task. WaitableTask_s
// lets main wait for it.
class CopyTask : public TC::Tasks::WaitableTask_s
{
public:
    CopyTask(const wchar_t* source, const wchar_t* target) noexcept
        : source_(source),
          target_(target)
    {}

    void Run() noexcept override
    {
        error = CreateSource(source_);
        if (Error::Succeeded(error))
            error = Copy(source_, target_);
        verified = Error::Succeeded(error) && Verify(source_) && Verify(target_);
    }

    ErrorCode error    = Error::Success;
    bool      verified = false;

private:
    const wchar_t* source_;
    const wchar_t* target_;
};

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

    TC::ConvertThreadToTask();
    // Before the first AsyncFile is created: it associates its handle with the
    // core's completion port.
    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();

    CopyTask task(source.c_str(), target.c_str());
    scheduler->AddTask(&task);
    task.Wait();
    scheduler->Stop();
    scheduler->Destroy();

    TC::AsyncIo::DisableAsyncIo(); // no operation is pending any more: every file
                                   // is closed
    TC::ConvertTaskToThread();

    std::printf("Copied %llu bytes: %s, %s\n", FileSize,
                task.error == Error::Success ? "no error" : "FAILED",
                task.verified ? "verified" : "NOT VERIFIED");
    std::error_code ignored;
    std::filesystem::remove(source, ignored);
    std::filesystem::remove(target, ignored);
    return task.error == Error::Success && task.verified ? 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 two files in %TEMP% and prints:

Copied 8388731 bytes: no error, verified

What to notice

Important

While an operation is pending, its request and its buffer must stay alive and in place, and the file must stay open. That is why Copy waits for every started write before it returns – also after an error. See Lifetimes.

Asynchronous I/O must be on. EnableAsyncIo before the first file is created, DisableAsyncIo only when no operation is pending any more.
One request, one operation. A request is Reset before each use; Reset also clears the offset, so SetOffset comes after it.
One waiter per request. Only the copying task waits on each request, so the _s variant fits. Once its Wait has returned, the request may be destroyed.
Which thread may call what. Read, Write and Wait suspend, so they run in a task (or on a thread after ConvertThreadToTask). ReadAsync and WriteAsync never suspend.
Known size, no reads past the end. The copy reads exactly the file's size, block by block; an overlapped read at the end of a file fails with ERROR_HANDLE_EOF instead of returning 0 bytes.