streambuf.h

Go to the documentation of this file.

 
 
#pragma once
 
#include <cx/buffer/bufring.h>
#include <cx/closure/closure.h>
#include <cx/stype/stype.h>
#include <cx/thread/condvar.h>
#include <cx/thread/mutex.h>
 
CX_C_BEGIN
 
typedef struct StreamBuffer StreamBuffer;
 
typedef size_t (*sbufPullCB)(stvlist* cvars, _Pre_valid_ StreamBuffer* sb,
                             _Out_writes_bytes_(sz) uint8* buf, size_t sz);
 
typedef void (*sbufPushCB)(stvlist* cvars, _Pre_valid_ StreamBuffer* sb,
                           _In_reads_bytes_(sz) const uint8* buf, size_t sz);
 
// Send callback
// This callback is used with sbufCSend. It may be called multiple times with varying
// offsets. The offset passed is always from the start of the available bytes in the
// buffer.
// Consumption is all-or-nothing across the whole sbufCSend call: if every invocation
// returns true the bytes are consumed and removed from the buffer, and if any one of them
// returns false the buffer keeps all of them, like the peek functions. A callback that can
// accept some of the data but not the rest should therefore return false every time and
// have the caller sbufCSkip() exactly what it took.
// This callback MUST NOT call any sbuf function on the buffer it was passed, with the single
// exception of sbufError(), which a sink that failed partway through uses to report the failure.
typedef bool (*sbufSendCB)(_Pre_valid_ StreamBuffer* sb, _In_reads_bytes_(sz) const uint8* buf,
                           size_t off, size_t sz, _Pre_opt_valid_ void* ctx);
 
typedef void (*sbufNotifyCB)(stvlist* cvars, _Pre_valid_ StreamBuffer* sb, size_t sz);
 
typedef void (*sbufResumeCB)(stvlist* cvars, _Pre_valid_ StreamBuffer* sb);
 
 
enum STREAM_BUFFER_FLAGS_ENUM {
    SBUF_Push   = 0x0001,
    SBUF_Pull   = 0x0002,
    SBUF_Direct = 0x0010,
    SBUF_Error  = 0x0800,
    SBUF_Closed = 0x1000,
};
 
enum STREAM_BUFFER_OPT_FLAGS {
    SBUF_Locked = 0x0004,
 
    SBUF_Wait = 0x0008,
};
 
// Internal state, not for callers.
enum STREAM_BUFFER_STATE_ENUM {
    SBUF_PHeld       = 0x0020,   // producer is held at the high watermark
    SBUF_PResumeOwed = 0x0040,   // a write was refused; owe the producer a resume callback
};
 
typedef struct StreamBuffer {
    BufRing buf;                     
    size_t targetsz;                 
 
    closure producerPull;            
    struct SbufResume* producerResume;   
 
    closure consumerNotify;          
    closure consumerPush;            
 
    // Closures of roles that unregistered from inside a callback, destroyed once the stack has
    // unwound back out of the buffer.
    closure pendingP;
    closure pendingC;
 
    size_t high;                     
    size_t low;                      
 
    Mutex lock;                      
    CondVar ready;                   
    CondVar flushed;                 
    atomic(intptr) owner;            
    uint32 depth;                    
 
    int refcount;                    
    bool locked;                     
    bool resumePending;              
    bool destroyPending;             
    bool walking;                    
    atomic(uint32) flags;            
} StreamBuffer;
 
// Internal function - use sbufCreate() macro instead
_Ret_valid_ StreamBuffer* _sbufCreate(size_t targetsz, flags_t flags);
 
#define sbufCreate(targetsz, ...) _sbufCreate(targetsz, opt_flags(__VA_ARGS__))
 
_Ret_valid_ StreamBuffer* sbufAcquire(_Inout_ StreamBuffer* sb);
 
_At_(*sb, _Pre_maybenull_ _Post_null_) void sbufRelease(_Inout_ StreamBuffer** sb);
 
void sbufClose(_Inout_opt_ StreamBuffer* sb);
 
_At_(*sb, _Pre_maybenull_ _Post_null_) void sbufFinish(_Inout_ StreamBuffer** sb);
 
void sbufError(_Inout_ StreamBuffer* sb);
 
void sbufClearError(_Inout_ StreamBuffer* sb);
 
void sbufSetWatermark(_Inout_ StreamBuffer* sb, size_t high, size_t low);
 
_meta_inline bool sbufIsLocked(_In_ StreamBuffer* sb)
{
    return sb->locked;
}
 
_meta_inline bool sbufIsPull(_In_ StreamBuffer* sb)
{
    return (atomicLoad(uint32, &sb->flags, Relaxed) & SBUF_Pull) != 0;
}
 
_meta_inline bool sbufIsPush(_In_ StreamBuffer* sb)
{
    return (atomicLoad(uint32, &sb->flags, Relaxed) & SBUF_Push) != 0;
}
 
_meta_inline bool sbufIsError(_In_ StreamBuffer* sb)
{
    return (atomicLoad(uint32, &sb->flags, Relaxed) & SBUF_Error) != 0;
}
 
_meta_inline bool sbufIsClosed(_In_ StreamBuffer* sb)
{
    return (atomicLoad(uint32, &sb->flags, Relaxed) & SBUF_Closed) != 0;
}
 
bool sbufCMore(_Inout_ StreamBuffer* sb);
 
 
 
_Check_return_ bool sbufPRegisterPull(_Inout_ StreamBuffer* sb, _In_ closure ppull);
 
void sbufPUnregister(_Inout_ StreamBuffer* sb);
 
bool sbufPAttached(_Inout_ StreamBuffer* sb);
 
void sbufPSetResume(_Inout_ StreamBuffer* sb, _In_opt_ closure resume);
 
bool sbufPIsHeld(_Inout_ StreamBuffer* sb);
 
size_t sbufPAvail(_Inout_ StreamBuffer* sb);
 
// Internal function - use sbufPWrite() macro instead
bool _sbufPWrite(_Inout_ StreamBuffer* sb, _In_reads_bytes_(sz) const uint8* buf, size_t sz,
                 flags_t flags);
 
#define sbufPWrite(sb, buf, sz, ...) _sbufPWrite(sb, buf, sz, opt_flags(__VA_ARGS__))
 
// Internal function - use sbufPWriteStr() macro instead
bool _sbufPWriteStr(_Inout_ StreamBuffer* sb, _In_opt_ strref str, flags_t flags);
 
#define sbufPWriteStr(sb, str, ...) _sbufPWriteStr(sb, str, opt_flags(__VA_ARGS__))
 
// Internal function - use sbufPWriteLine() macro instead
bool _sbufPWriteLine(_Inout_ StreamBuffer* sb, _In_opt_ strref str, flags_t flags);
 
#define sbufPWriteLine(sb, str, ...) _sbufPWriteLine(sb, str, opt_flags(__VA_ARGS__))
 
// Internal function - use sbufPWriteEOL() macro instead
bool _sbufPWriteEOL(_Inout_ StreamBuffer* sb, flags_t flags);
 
#define sbufPWriteEOL(sb, ...) _sbufPWriteEOL(sb, opt_flags(__VA_ARGS__))
 
bool sbufPFlush(_Inout_ StreamBuffer* sb);
 
 
 
_Check_return_ bool sbufCRegisterPush(_Inout_ StreamBuffer* sb, _In_ closure cnotify);
 
_Check_return_ bool sbufCRegisterPushDirect(_Inout_ StreamBuffer* sb, _In_ closure cpush);
 
void sbufCUnregister(_Inout_ StreamBuffer* sb);
 
bool sbufCAttached(_Inout_ StreamBuffer* sb);
 
size_t sbufCAvail(_Inout_ StreamBuffer* sb);
 
_Success_(return) bool
sbufCRead(_Inout_ StreamBuffer* sb, _Out_writes_bytes_to_(sz, *bytesread) uint8* buf, size_t sz,
          _Out_ _Deref_out_range_(0, sz) size_t* bytesread);
 
_Success_(return > 0) bool
sbufCPeek(_Inout_ StreamBuffer* sb, _Out_writes_bytes_(sz) uint8* buf, size_t off, size_t sz);
 
bool sbufCFeed(_Inout_ StreamBuffer* sb, size_t minsz);
 
bool sbufCSend(_Inout_ StreamBuffer* sb, _In_ sbufSendCB func, size_t sz, _Inout_opt_ void* ctx);
 
bool sbufCSkip(_Inout_ StreamBuffer* sb, size_t bytes);
 
 
CX_C_END