Pipes

A pipe passes a stream of bytes from one producer task to one consumer task. With SpscVirtualMemoryPipe both sides get pointers into one contiguous range of virtual memory: the producer writes the bytes in place, the consumer reads them in place, and no side copies. The pipe implements two small interfaces, IWriteSequentialMemory for the producer and IReadSequentialMemory for the consumer, which AsyncFile::ReadFile and AsyncFile::WriteFile also use. The examples come from guide-pipes.cpp, which writes a stream of measurements through a pipe into a file, reads it back through a second pipe, and checks the result.

The producer side

The protocol of IWriteSequentialMemory:

Initialize once with the total size of the stream.
For each block: RequestWriteMemory, fill it, ReleaseWriteMemory.
SetEndOfStream at the end, also after an error.

The example stream is a sequence of samples:

// One measurement; the stream is a sequence of them.
struct Sample {
    std::uint64_t time;
    double        value;
};

The producer writes them in blocks of 4096 samples, directly into the memory the pipe hands out:

// Writes count samples into the stream, block by block, straight into the stream's
// memory.
void WriteSamples(IWriteSequentialMemory& out, std::size_t count) noexcept
{
    out.Initialize(
        count
        * sizeof(Sample)); // announce the size first (also for an empty stream)
    constexpr std::size_t PerBlock = 4096; // samples per block
    for (std::size_t first = 0; first < count; first += PerBlock) {
        const std::size_t n    = std::min(PerBlock, count - first);
        std::byte*        data = out.RequestWriteMemory(
            n * sizeof(Sample)); // the next bytes of the stream
        if (data == nullptr)
            break;
        for (std::size_t i = 0; i < n; ++i) {
            const Sample sample = TakeSample(first + i);
            std::memcpy(data + i * sizeof(Sample), &sample, sizeof(Sample));
        }
        out.ReleaseWriteMemory(n * sizeof(Sample)); // publish them to the consumer
    }
    out.SetEndOfStream(); // always, also after an error
}

TakeSample stands for wherever the data comes from.

Each requested block directly follows the previous one. Several blocks may be requested before the first is released; they are released in the order they were requested, and ReleaseWriteMemory publishes the next size bytes of them to the consumer.

The consumer side

The protocol of IReadSequentialMemory: for each block RequestReadMemory, use the bytes, ReleaseReadMemory. GetSize returns the total size.

// Reads the samples block by block, in place, and adds up their values. Returns
// false if the stream ended early.
bool SumSamples(IReadSequentialMemory& in, double& sum) noexcept
{
    const std::size_t total
        = in.GetSize(); // waits until the producer has called Initialize
    std::size_t read = 0;
    sum              = 0;
    for (;;) {
        std::size_t size = 4096 * sizeof(Sample); // wanted; less only at the end
        const std::byte* data = in.RequestReadMemory(size);
        if (data == nullptr)
            break; // end of the stream
        for (std::size_t offset = 0; offset < size; offset += sizeof(Sample)) {
            Sample sample;
            std::memcpy(&sample, data + offset, sizeof(Sample));
            sum += sample.value;
        }
        read += size;
        in.ReleaseReadMemory(size); // the memory may be reused now
    }
    return read == total;
}

RequestReadMemory takes the wanted size and returns the size it handed out, which is smaller only at the end of the stream; once the stream is exhausted it returns nullptr with size 0. Because the wanted size is a multiple of the sample size, every block holds whole samples. A block stays valid until it is released, and several blocks may be held at once, for example to look back at the previous block.

SpscVirtualMemoryPipe

The pipe reserves address space for the whole announced size (rounded up to 64 KB) in Initialize. It commits memory in 64 KB steps as the producer requests blocks and decommits it in 64 KB steps once the consumer has released them. The committed memory is therefore about the distance between the two sides, not the size of the stream.

The two sides behave differently:

The consumer waits: GetSize suspends until the producer has called Initialize, and RequestReadMemory suspends until enough bytes have been released or the stream has ended.
The producer never waits. If it runs far ahead of the consumer, the distance grows and so does the committed memory, up to the whole stream.

Important

One producer and one consumer. Both sides must be tasks or converted threads.
The producer must call Initialize also for an empty stream (the consumer waits for it in GetSize) and must not write more than it announced there.
The pipe must outlive both sides. A failure to reserve, commit or decommit memory ends the process.

When the producer is much faster than the consumer and the stream is large, bound the distance yourself, for example with a Semaphore that the consumer releases per block and the producer waits on.

Combining with AsyncFile

AsyncFile::WriteFile is a ready-made consumer: it takes blocks with RequestReadMemory until the end of the stream and writes them with several overlapped writes in flight. Here a producer task fills the pipe while main writes it to a file:

// A producer task fills the pipe; main writes it to a file while it fills.
constexpr std::size_t             count = 2'000'000; // 32 MB of samples
TC::Memory::SpscVirtualMemoryPipe toFile;
auto                              producer = TC::Tasks::Task::Create(
    [&toFile]() noexcept { WriteSamples(toFile, count); });
helpers.AddTask(&producer);
scheduler->AddTask(&producer);

const auto written = TC::AsyncIo::AsyncFile::WriteFile(
    path, GENERIC_WRITE, 0, CREATE_ALWAYS, FILE_ATTRIBUTE_NORMAL, &toFile);
helpers.Wait();

AsyncFile::ReadFile is a ready-made producer: it calls Initialize with the file size, requests and releases one block per read in file order, and always ends with SetEndOfStream, also on errors:

// main reads the file into a second pipe; a consumer task adds up the samples
// as they arrive.
TC::Memory::SpscVirtualMemoryPipe fromFile;
double                            sum      = 0;
bool                              complete = false;
auto                              consumer
    = TC::Tasks::Task::Create([&fromFile, &sum, &complete]() noexcept {
          complete = SumSamples(fromFile, sum);
      });
helpers.AddTask(&consumer);
scheduler->AddTask(&consumer);

TC::AsyncIo::BulkIoOptions options; // 5 reads of 1 MB in flight by default
const auto                 read = TC::AsyncIo::AsyncFile::ReadFile(
    path, GENERIC_READ, FILE_SHARE_READ, OPEN_EXISTING,
    FILE_FLAG_SEQUENTIAL_SCAN, &fromFile, options);
helpers.Wait();

The consumer starts on the first block while later blocks are still being read, so reading and processing overlap. BulkIoOptions sets the block size and the number of operations in flight; see Asynchronous files.

The program prints:

pipe to file:  ok
file to pipe:  ok, all samples read back: yes
OK

Pipes

The producer side

The consumer side

SpscVirtualMemoryPipe

Combining with AsyncFile

See also