stream.h

Go to the documentation of this file.

#ifndef NEXUS_IO_STREAM_H
#define NEXUS_IO_STREAM_H
 
#include "nexus/core/status.h"
 
#ifdef __cplusplus
extern "C" {
#endif
 
#define NX_STREAM_BOUNDARY_IDLE    1U
#define NX_STREAM_BOUNDARY_TRIGGER 2U
#define NX_STREAM_BOUNDARY_LOSS    4U
#define NX_STREAM_BOUNDARY_ERROR   8U
 
typedef enum {
    NX_STREAM_SLOT_FREE,
    NX_STREAM_SLOT_FILLING,
    NX_STREAM_SLOT_READY,
    NX_STREAM_SLOT_BORROWED
} nx_stream_slot_state_t;

typedef struct {
    uint8_t* data;
    size_t capacity;
    size_t length;
    uint64_t epoch;
    uint32_t flags;
    uint32_t lost_blocks;
    nx_stream_slot_state_t state;
} nx_stream_slot_t;

typedef struct {
    nx_stream_slot_t* slots;
    size_t count;
    size_t producer;
    size_t consumer;
    size_t outstanding;
    uint64_t epoch;
    uint32_t losses;
    bool stopping;
} nx_stream_t;

typedef struct {
    uint8_t* data;
    size_t capacity;
    size_t slot;
    uint64_t epoch;
} nx_stream_fill_t;

typedef struct {
    nx_stream_t* stream;
    size_t cursor;
    size_t pending;
} nx_stream_producer_t;

typedef struct {
    const uint8_t* data;
    size_t length;
    size_t slot;
    uint64_t epoch;
    uint32_t flags;
    uint32_t lost_blocks;
} nx_stream_block_t;

nx_result_t nx_stream_initialize(nx_stream_t* stream, nx_stream_slot_t* slots,
                                 size_t count);

nx_result_t nx_stream_reserve(nx_stream_t* stream, nx_stream_fill_t* fill);
bool nx_stream_can_reserve(const nx_stream_t* stream);
void nx_stream_note_loss(nx_stream_t* stream, uint32_t count);

nx_result_t nx_stream_publish(nx_stream_t* stream, const nx_stream_fill_t* fill,
                              size_t length, uint32_t flags);

nx_result_t nx_stream_acquire(nx_stream_t* stream, nx_stream_block_t* block);

nx_result_t nx_stream_release(nx_stream_t* stream,
                              const nx_stream_block_t* block);

nx_result_t nx_stream_abort(nx_stream_t* stream, const nx_stream_fill_t* fill,
                            bool memory_quiesced);

nx_result_t nx_stream_stop(nx_stream_t* stream);

nx_result_t nx_stream_post_initialize(nx_stream_producer_t* producer,
                                      nx_stream_t* stream);
bool nx_stream_post_can_reserve(const nx_stream_producer_t* producer);
nx_result_t nx_stream_post_reserve(nx_stream_producer_t* producer,
                                   nx_stream_fill_t* fill);
nx_result_t nx_stream_post_publish(nx_stream_producer_t* producer,
                                   const nx_stream_fill_t* fill, size_t length,
                                   uint32_t flags);
nx_result_t nx_stream_post_abort(nx_stream_producer_t* producer,
                                 const nx_stream_fill_t* fill,
                                 bool memory_quiesced);
 
#ifdef __cplusplus
}
#endif
 
#endif