Stream Buffers

Modules
	Core Functions
	Producer Functions
	Consumer Functions

Typedefs
typedef size_t(*	sbufPullCB) (stvlist *cvars, StreamBuffer *sb, uint8 *buf, size_t sz)
typedef void(*	sbufPushCB) (stvlist *cvars, StreamBuffer *sb, const uint8 *buf, size_t sz)
typedef void(*	sbufNotifyCB) (stvlist *cvars, StreamBuffer *sb, size_t sz)
typedef void(*	sbufResumeCB) (stvlist *cvars, StreamBuffer *sb)

Detailed Description

A stream buffer carries bytes from a producer to a consumer. One side drives and the other is called back; which is which is the buffer's mode:

Push mode: the producer drives. It calls sbufPWrite() whenever it has data, and the consumer's callback runs to take it.

Pull mode: the consumer drives. It calls sbufCRead() whenever it wants data, and the producer's callback runs to supply it.

Only the side that gets called back registers, and that registration picks the mode: registering a consumer callback makes the buffer push mode, registering a producer callback makes it pull mode. The driving side has nothing to register – it just calls the read or write functions.

Those functions are the same in both modes. sbufCRead() hands back buffered data in push mode and calls the producer in pull mode, so code that only reads, or only writes, does not have to know which mode it is in.

The registered side is a typed closure (see Typed closures), so whatever the callback needs is captured alongside it and released when the registration goes away.

Push mode:

static void onData(stvlist *cvars, StreamBuffer *sb, size_t sz) {
    Parser *p = stvlAtPtr(cvars, 0);
    // read from sb with sbufCRead() or sbufCSend()
}
 
StreamBuffer *sb = sbufCreate(4096);
sbufCRegisterPush(sb, closureCreateAs(sbufNotifyCB, onData, stvar(ptr, parser)));
 
sbufPWrite(sb, data, size);                           // producer drives
sbufFinish(&sb);                                      // close and release

Pull mode:

StreamBuffer *sb = sbufCreate(4096);
sbufPRegisterPull(sb, closureCreateAs(sbufPullCB, onPull, stvar(object, file)));
 
size_t bytesread;
while (sbufCRead(sb, buffer, sizeof(buffer), &bytesread)) {   // consumer drives
    // process buffer
}
sbufFinish(&sb);                                      // close and release

Lifetime: stream buffers are reference counted. sbufCreate() returns one reference, sbufAcquire() takes another, and sbufRelease() gives one back. Releasing the last reference frees the buffer, and is the only thing that can. Registering takes a reference of its own, so hold your own reference – acquire one if you did not create the buffer – for as long as you keep the pointer.

Closing a stream: the driving side calls sbufClose() once the stream is over: the producer in push mode, the consumer in pull mode. That says something about the stream rather than about either party, so there is only one such call and only the driving side makes it. Writes stop working, a consumer may still drain whatever is already buffered, and the registered side gets one last callback with sz == 0. Anything still registered after that callback is detached, since nothing can call it again.

Closing does not release your own reference. sbufFinish() does both, and is what the driving side normally calls when it is finished with the buffer.

Leaving a role: a registered party that is simply done calls sbufPUnregister() or sbufCUnregister() instead. That empties the slot without ending the stream, so a replacement can register and carry on – a pull producer that ran out of bytes, or a log file rotated out from under a writer. Until someone new attaches, a reader gets short reads and a writer's bytes pile up in the buffer. Unregistering destroys that registration's closure and gives back the reference the registration took.

Whoever holds a reference may unregister a slot, not only the party that filled it, which is what lets an application swap a sink out from under a stream. Flush first with sbufPFlush() so the outgoing sink gets the bytes that were meant for it.

Errors: sbufError() reports that something went wrong. Reads and writes fail while the error stands, so a failure cannot be quietly written over, but the stream is not over. The driving side finds out on its next call and decides: give up with sbufClose(), or unregister whoever failed, call sbufClearError() and attach a replacement.

Threads: a stream buffer is single-threaded by default. The producer and the consumer are expected to run on the same thread, taking turns through the callbacks. Pass SBUF_Locked to sbufCreate() when they must live on different threads; that adds a lock around every operation and is the only supported way to share a stream buffer.

Flow control: by default the buffer grows without limit, so a producer that outruns its consumer uses as much memory as it writes. Call sbufSetWatermark() to cap it. Once the buffered data reaches the high mark the producer is held until the consumer drains it back to the low mark. A producer that passes SBUF_Wait waits inside sbufPWrite() until that happens; otherwise sbufPWrite() returns false right away and the resume callback set with sbufPSetResume() says when to try again.

StreamBuffer *sb = sbufCreate(4096, SBUF_Locked);
sbufSetWatermark(sb, 65536, 16384);
sbufPWrite(sb, data, size, SBUF_Wait);   // waits at the mark rather than failing

Typedef Documentation

sbufNotifyCB

typedef void(* sbufNotifyCB) (stvlist *cvars, StreamBuffer *sb, size_t sz)

Notify callback, for a consumer registered with sbufCRegisterPush()

Create it with closureCreateAs(sbufNotifyCB, func, ...). Data is available; read all or part of it with the sbufC* functions.

A consumer that no longer wants the stream calls sbufCUnregister(), which leaves the stream open for another consumer. A call with sz of 0 is a status check rather than an offer of data: the stream has closed or failed. A closed stream detaches the consumer on its own once this returns.

Parameters

cvars	Captured variables of the registered closure
sb	The stream buffer
sz	Number of bytes available

Definition at line 173 of file streambuf.h.

sbufPullCB

typedef size_t(* sbufPullCB) (stvlist *cvars, StreamBuffer *sb, uint8 *buf, size_t sz)

Pull callback, for a producer registered with sbufPRegisterPull()

Create it with closureCreateAs(sbufPullCB, func, ...). Fill buf with up to sz bytes and return how many were written. Returning 0 means no data is ready yet; the callback will likely be called again right away, so waiting for data here is reasonable. A producer with more than sz bytes ready may instead call sbufPWrite() with all of it and return 0.

A producer that has run out of data calls sbufPUnregister(), which leaves the stream open for another producer. A call with sz of 0 is a status check rather than a request for data: the stream has closed or failed. A closed stream detaches the producer on its own once this returns.

Parameters

cvars	Captured variables of the registered closure
sb	The stream buffer
buf	Where to write the data
sz	Most bytes that may be written

Returns

Number of bytes written

Definition at line 132 of file streambuf.h.

sbufPushCB

typedef void(* sbufPushCB) (stvlist *cvars, StreamBuffer *sb, const uint8 *buf, size_t sz)

Push callback, for a consumer registered with sbufCRegisterPushDirect()

Create it with closureCreateAs(sbufPushCB, func, ...). The data is handed over once and must all be taken, or it is lost. A call with sz of 0 means the stream has closed or failed.

Parameters

cvars	Captured variables of the registered closure
sb	The stream buffer
buf	The data
sz	Number of bytes

Definition at line 144 of file streambuf.h.

sbufResumeCB

typedef void(* sbufResumeCB) (stvlist *cvars, StreamBuffer *sb)

Resume callback, set with sbufPSetResume()

Create it with closureCreateAs(sbufResumeCB, func, ...). Tells a producer that was refused at the high watermark that the buffer has drained back to the low mark and writing may continue. It runs on whichever thread drained the buffer, after that thread has released the buffer's lock, so it may call sbufPWrite().

Parameters

cvars	Captured variables of the registered closure
sb	The stream buffer

Definition at line 184 of file streambuf.h.