 |
Sane C++ Libraries
C++ Platform Abstraction Libraries
|
|
Sane C++ Libraries
C++ Platform Abstraction Libraries
|
🟨 Bounded byte-stream pipelines with explicit backpressure
SaneCppAsyncStreams.h provides allocation-free readable, writable, transform, and pipeline state machines for moving bytes between asynchronous producers and consumers.
Async Streams is for workloads where data arrives or leaves over time and should be processed without first collecting the entire payload in memory. A source can read a file or socket into a bounded pool of buffers, transforms can process those buffers incrementally, and one or more sinks can consume them concurrently.
The core library defines the stream and pipeline state machines without depending on Async, File, or Socket. AsyncRequestStreams.h supplies templates that adapt compatible asynchronous request and event-loop types; the file and socket compositions in this repository instantiate them with Async.
Use Async Streams when you need:
It is not a general container for arbitrary objects, a pull-based read() interface, or an unbounded buffering layer. There is currently no object mode, and consumers receive pushed data events rather than a Node.js-style readable event.
An Async Streams graph has four layers:
AsyncBuffersPool names and tracks caller-provided AsyncBufferView objects.AsyncReadableStream produces buffer IDs; AsyncWritableStream consumes them.AsyncDuplexStream combines both directions, while AsyncTransformStream consumes input and produces transformed output.AsyncPipeline connects one source through an optional transform chain to one or more sinks.Streams exchange buffer IDs instead of owning byte allocations. The pool resolves an ID to its readable or writable span and reference-counts that view while it is in flight. A reusable view returns to the pool when its references reach zero. Child views can expose a slice without copying while retaining the parent buffer.
Read and write request queues are also supplied by the caller. The internal circular queues reserve one slot to distinguish full from empty, so an array of N + 1 requests provides N usable queued requests.
This compiled test divides one caller-owned byte allocation into two reusable views. Real applications choose both the number and size of buffers from their expected concurrency and chunk size.
The pool never grows. When a readable cannot obtain a sufficiently large free buffer, getBufferOrPause() pauses it. When downstream processing releases a buffer, the pipeline can resume the source. This makes memory exhaustion part of the flow-control model instead of an implicit allocation.
The ready-made request streams adapt file and socket descriptors to the core stream model. After constructing the caller-owned queues and associating descriptors with an event loop, a basic file copy is a source connected directly to a sink:
The source, every transform, and every sink in one pipeline must use the same AsyncBuffersPool. pipe() validates the layout and registers event listeners; start() begins source production. Completion remains asynchronous: the event loop must continue running until the streams finish or are destroyed.
AsyncReadableFileStream and AsyncWritableFileStream accept FileDescriptor and PipeDescriptor, so anonymous and named pipes from File compose without manually extracting their read or write endpoint.
Backpressure can come from either side of the graph:
Result when its bounded write queue is full;Readable implementations acquire a buffer, fill it, call push(bufferID, size), and release their own reference. The stream retains the buffer for queued delivery. Writable implementations finish each accepted operation through finishedWriting(), which advances the queue and releases the stream's reference. Custom streams must respect this protocol; releasing a view too early invalidates downstream data, while failing to finish an operation prevents drain and end propagation.
eventData listeners should treat the buffer as borrowed for the callback unless they explicitly add a reference. Buffers can be read by multiple sinks because the pool tracks references, but writable access is only meaningful where the underlying view permits it.
An AsyncTransformStream is a duplex stream: it consumes buffers through its writable side and emits new buffers from its readable side. SyncZLibTransformStream runs compression work synchronously; AsyncZLibTransformStreamT schedules the same work through a compatible event loop and can be directed to a caller-selected thread pool.
A pipeline has a fixed layout rather than a dynamically growing graph. It supports at most eight transforms and eight sinks. Every sink receives the source data, with buffer references held until each write completes. This is useful for bounded fan-out, but the slowest sink applies backpressure to the shared source. If consumers need independent pacing or unbounded history, that storage and policy belongs outside this library.
CryptographyTransformStreams.h provides two header-only adapters:
AsyncCipherTransformStreamT<T_Cipher> turns a compatible incremental cipher into a transform;AsyncHmacWritableStreamT<T_Hmac> authenticates every byte written to it and is especially useful as a fan-out sink.The header includes no Cryptography interface. A translation unit that wants the native implementation includes both libraries and chooses the concrete session at instantiation time:
The adapters forward a single constructor argument to the concrete session, so an explicitly selected backend remains possible without teaching Async Streams about Cryptography:
source, encrypt, and destination must use the same pool. Every reusable writable view that the pool may return to the cipher must be at least 16 bytes. Larger buffers are more efficient: the adapter reserves up to 15 bytes of capacity for output caused by input retained from an earlier update. Decryption uses the same setup with Operation::Decrypt.
This cipher is AES-CBC with PKCS#7 padding. CBC does not authenticate ciphertext and is suitable only when an existing protocol already supplies correct authentication, IV rules, framing, and failure handling. Stream errors include native cipher failures, invalid ciphertext or padding, and undersized output buffers. Decryption may emit earlier plaintext blocks before finish() validates the final padding, so the surrounding protocol must authenticate the ciphertext before any plaintext is released to an application.
An HMAC sink can share source buffers with a storage or network sink through normal pipeline fan-out:
Retrieve the MAC inside eventFinish. Streams auto-destroy immediately after that event returns, and destruction resets the HMAC session so key-dependent native state is released. The fan-out does not copy the source buffer: the pool keeps the same view referenced until both sinks finish their writes.
There is intentionally no AES-GCM byte-stream adapter. Secure AEAD decryption must authenticate a complete bounded record before releasing any plaintext, which requires explicit record framing, nonce derivation, tag placement, AAD, and storage policy. Those protocol decisions do not belong in an apparently generic byte transform.
Stream errors are delivered through eventError, and AsyncPipeline::eventError forwards failures from connected sources, transforms, and sinks. pushEnd() ends a readable after queued data is delivered; end() lets a writable finish queued and in-flight writes before emitting its finish event.
destroy() is the forceful, idempotent path for stopping a stream and its pending asynchronous work. Custom implementations that override asynchronous destruction must eventually call the corresponding finishedDestroyingReadable() or finishedDestroyingWritable() hook. Request-stream adapters do not close their file or socket descriptor by default; call setAutoCloseDescriptor(true) only when the stream owns that descriptor. Streams auto-destroy after normal completion by default, and this can be disabled when the surrounding owner needs a different lifetime policy.
Async Streams performs no hidden dynamic allocation. The caller supplies:
AsyncBufferView slots in each AsyncBuffersPool;Some examples use Buffer, String, or Vector to conveniently own test memory. Those allocations belong to the caller and are not required by the stream machinery; fixed spans can be used instead. Capacity failures are observable as pauses or invalid Result values and must be handled as normal control flow.
For the complete API reference, see Async Streams. The tests under Tests/Libraries/AsyncStreams contain working file, pipe, socket, fan-out, backpressure, child-view, cryptography-adapter, and synchronous/asynchronous zlib compositions.
🟨 MVP
The core file, pipe, socket, pipeline, and zlib transform paths are implemented and exercised by tests. The pipeline automatically pauses and resumes streams as part of backpressure handling, but it does not yet expose explicit pipeline-level pause, resume, or cancellation controls. The API remains an MVP: some setup is verbose, and examples outside the test suite are limited.
This is the list of videos that have been recorded showing some of the internal thoughts that have been going into this library:
Some relevant blog posts are:
🟩 Usable features:
🟦 Complete Features:
💡 Unplanned Features:
LOC counts exclude comments. Library counts files physically under Libraries/AsyncStreams. Single File counts SaneCppAsyncStreams.h. Standalone counts SaneCppAsyncStreamsStandalone.h and intentionally includes dependency payloads.
| Metric | Header | Source | Sum |
|---|---|---|---|
| Library | 1037 | 1799 | 2836 |
| Single File | 1820 | 2046 | 3866 |
| Standalone | 1820 | 2046 | 3866 |
1.12.0