Motivation
The initial requirement came from an HTTP/TFTP gateway.
A local HTTP client downloads a file from a remote server, while a TFTP server simultaneously sends the same content to a network client. Since TFTP works with fixed-size blocks and acknowledgments, while HTTP typically delivers larger chunks asynchronously, some form of synchronization layer is required.
The traditional approaches all have drawbacks:
- Buffer the whole file in memory before transmission.
- Write to a temporary file and read it back.
- Use operating-system pipes or socket pairs.
- Build a custom queue of byte buffers.
What was needed was a simple object exposing the familiar TStream API:
HttpClientStream.CopyTo(PipeStream);
in one thread, while another thread performs:
n := PipeStream.Read(TftpBuffer, SizeOf(TftpBuffer));
with automatic synchronization.
Architecture
TPipeStream is implemented as a bounded ring buffer.
Internally it maintains:
- a power-of-two memory buffer;
- a read position;
- a write position;
- a pending byte count;
- two synchronization events.
Only one producer thread and one consumer thread are supported, which allows a very compact implementation.
The ring buffer uses power-of-two sizing so that wrap-around is performed with a simple:
Position := (Position + Count) and (BufferSize - 1);
instead of an expensive modulo operation.
The producer writes into the buffer until it becomes full.
The consumer reads from the buffer until it becomes empty.
When necessary, the corresponding thread blocks until progress can continue.
Efficient Synchronization
The implementation uses mORMot's lightweight synchronization primitives:
TLightLockTSynEvent
The lock protects the buffer state, while the events are used only as wake-up mechanisms.
A writer waiting for space is awakened only when the buffer transitions from:
full -> non-full
A reader waiting for data is awakened only when the buffer transitions from:
empty -> non-empty
This transition-based signaling minimizes unnecessary wakeups and context switches.
During development, moving from "buffer empty" wakeups to proper "buffer became non-full" wakeups doubled the measured throughput under sustained load.
Why Not Use Operating-System Pipes?
Most operating systems already provide pipes.
However, system pipes require kernel transitions for every blocking operation:
Write() -> kernel -> scheduler -> wake reader -> kernel -> user mode
By contrast, TPipeStream remains entirely in user space.
Data is copied directly into the ring buffer, and synchronization only occurs when producer and consumer need to wait for each other.
This significantly reduces system call overhead and scheduler activity, especially when transferring large streams of data inside the same process.
For local inter-thread communication, it can therefore be substantially faster than an operating-system pipe.
Stream-Oriented Design
A major advantage of TPipeStream is that it is a standard TStream.
Existing code often works immediately:
Source.CopyTo(PipeStream);
or:
n := PipeStream.Read(Buffer, SizeOf(Buffer));
without introducing custom queue APIs or callback mechanisms.
Timeouts are also supported for both reads and writes, making the class suitable for network-oriented workloads where indefinite blocking is undesirable.
Cross-Platform
The implementation relies only on mORMot synchronization primitives and standard memory management.
As a result, the same code works on:
- Windows
- Linux (using very efficient eventfd)
- macOS
- FreeBSD
and on both:
- Delphi
- Free Pascal
without platform-specific code.
A Missing Building Block
Interestingly, neither Delphi's RTL nor Free Pascal's FCL provides an equivalent class.
There are thread-safe queues, buffers, channels, socket pairs and operating-system pipes, but there is no standard in-memory blocking TStream designed specifically for producer/consumer streaming.
Similar concepts exist in other ecosystems:
- POSIX pipes
- Java's
PipedInputStreamandPipedOutputStream - .NET pipelines
- Go channels
Yet Delphi developers typically end up building ad-hoc solutions whenever they need to stream data between threads.
TPipeStream fills this gap with a compact, efficient and portable implementation.
Typical Use Cases
Typical applications include:
- HTTP download → TFTP upload
- HTTP download → decompression thread
- HTTP download → decryption thread
- producer thread → parser thread
- network receiver → processing pipeline
- streaming transformations without temporary files
Whenever two threads need to exchange a continuous stream of bytes, TPipeStream provides a simple and efficient solution while preserving the familiar TStream programming model.
Read More
You can look at the corresponding commit code on github.
As with most mORMot classes, you will see that the implementation remains intentionally compact: a lightweight building block designed to solve a common problem with minimal overhead and maximum portability.
To be fair, debugging this class was not obvious, so we wrote some dedicated regression tests in the mORMot 2 test suite, to ensure it remains stable and truly cross-platform.
Any feedback is welcome, as usual!
The core synchronization seems ready, but there is some work to be done on this class. Feedback has been given on DelphiPraxis: you can join with your ideas!
